view release on metacpan or  search on metacpan

lib/Net/Hadoop/YARN/ResourceManager/Scheduler/UserApps.pm  view on Meta::CPAN

            $app->{ $resource } = 0 if $app->{ $resource } eq '-1';
        }

        if ( $app->{allocatedMB} ) {
            $app->{allocatedMB_fmt} = $self->format_bytes( $app->{allocatedMB} * 1024**2 );
        }

        if ( $app->{allocatedVCores} ) {
            $app->{allocatedVCores_fmt} = sprintf '%s vCore%s',
                                                    $app->{allocatedVCores},
                                                    $app->{allocatedVCores} > 1 ? 's' : '',
            ;
        }

        # TODO
        # [STRING]"applicationTags"
        # the value is something like "oozie-59a27f107d250c9822fd45e87fd40db8"
        # which is not the job id.

        foreach my $hash_or_string ( qw(
            diagnostics
            applicationTags
        )) {
            next if ! exists $app->{ $hash_or_string };
            # This is a bug in the REST layer
            if (   Ref::Util::is_hashref $app->{ $hash_or_string }
                && ! keys %{ $app->{ $hash_or_string } }
            ) {
                $app->{ $hash_or_string } = '';
            }
        }

        # https://www.cloudera.com/documentation/enterprise/latest/topics/cm_dg_yarn_applications.html
        foreach my $duration_field ( qw(
            vcoreSeconds
            elapsedTime
            memorySeconds
        ) ) {
            next if ! exists $app->{ $duration_field };
            if ( $app->{ $duration_field } ) {
                $app->{ $duration_field . '_fmt' } = Time::Duration::duration(
                                                         $duration_field eq 'elapsedTime'
                                                            ? $app->{ $duration_field } / 1000
                                                            : $app->{ $duration_field }
                                                    );
            }
        }

        foreach my $time_field ( qw(
            finishedTime
            startedTime
        ) ) {
            next if ! exists $app->{ $time_field };
            if ( $app->{ $time_field } ) {
                $app->{ $time_field . '_fmt' } = $format_epoch->( $app->{ $time_field } / 1000);
            }
        }

        if ( $app->{name} =~ m{ \Q-oozie-oozi-W\E \z }xms ) {
            my %name = map  { @{ $_ } > 1 ? @{ $_ } : ( $_->[0] => 1 ) }
                        map { [ split m{ [=] }xms, $_, 2 ] }
                        split m{ [:] }xms, $app->{name};
            $name{workflow_name} = delete $name{W} if $name{W};
            $name{action_name}   = delete $name{A} if $name{A};
            $name{action_type}   = delete $name{T} if $name{T};
            $name{id}            = delete $name{ID} if $name{ID};
            $app->{oozie_meta} = \%name;
            $app->{oozie_id}   = $name{id} if $name{id};
        }

        push @{ $apps_by_state{ $app->{state} } ||= [] }, $app;
    }

    my %total_res;
    foreach my $app ( @{ $apps_by_state{RUNNING} }) {
        $total_res{allocatedMB}     += $app->{allocatedMB};
        $total_res{allocatedVCores} += $app->{allocatedVCores};
    }

    if ( $total_res{allocatedMB} ) {
        $total_res{allocatedMB_fmt} = $self->format_bytes( $total_res{allocatedMB} * 1024**2 );
    }

    if ( $total_res{allocatedVCores} ) {
        $total_res{allocatedVCores_fmt} = sprintf '%s vCore%s',
                                                $total_res{allocatedVCores},
                                                $total_res{allocatedVCores} > 1 ? 's' : '',
        ;
    }

    my @grouped;
    foreach my $ordered_state (qw(
        RUNNING
        ACCEPTED
        FINISHED
        KILLED
        FAILED
    )) {
        push @grouped, {
            state     => lc( $ordered_state ),
            state_fmt => ucfirst( lc $ordered_state ),
            apps      => delete( $apps_by_state{$ordered_state} )  || [],
        },
    }

    # TODO: possibly needs to be removed if we are sure that the code above
    # is handling all of the possible states. So, this is a "just in case" part
    #
    push @grouped, {
        state => 'rest',
        apps  => [ map { @{ $_ } } values %apps_by_state ],
    };

    # Spark jobs are returned like this for whatever reason.
    if ( my $apps = $grouped[-1]->{apps} ) {
        if ( Ref::Util::is_arrayref $apps && Ref::Util::is_arrayref $apps->[0] ) {
            $grouped[-1]->{apps} = [ @{ $apps->[0] } ];
        }
    }

    return {
        grouped_apps => [ grep { @{ $_->{apps} } > 0 } @grouped ],