view release on metacpan or  search on metacpan

lib/App/Oozie/Rerun.pm  view on Meta::CPAN

    my $jobs  = $oozie->jobs(
                    filter => {
                        status => $self->status,
                    },
                    len => $self->maxjobs,
                );

    return if ! $jobs || ! $jobs->{workflows};

    my @candidates = grep {
                    $when <= (
                        $_->{lastModTime_epoch}
                     || $_->{lastModifiedTime_epoch}
                    )
                } @{ $jobs->{workflows} };

    return if ! @candidates;

    my(%seen, %coord_cache);

    my %is_status = map { $_ => 1 } @{ $self->status };
    my $reruns    = {};
    $re_name      = qr{ $re_name }xms if $re_name;

    for my $fail ( @candidates ) {

        my $name = $fail->{appName};
        my $id   = $fail->{id};
        my $cid  = $fail->{parentId}
                    ? ( split RE_AT, $fail->{parentId} )[0]
                    : undef
                    ;

        if (   ! $fail->{parentId} # Standalone WF, it can't be re-run.
            || $seen{ $cid }
            || ( $re_name && $fail->{appName} !~ $re_name )
        ) {
            if ( $verbose ) {
                $logger->debug(
                    sprintf 'Skipping %s [%s] -> does not match the criterias',
                                $id || 'N/A',
                                $name,
                );
            }
            next;
        }

        my $coord = $coord_cache{ $cid } ||= $oozie->job( $cid );
        my $job   = $oozie->job( $fail->{parentId} );

        if ( $coord->{status} eq 'KILLED' && ! $resurrect ) {
            if ( ! $seen{ $cid }++ ) {
                $logger->warn( "Coordinator $name ( $cid ) is dead. Skipping (see --help)" );
            }
            next;
        }

        if ( ! $job->{coordJobId} || ! $is_status{ uc $job->{status} } ) {
            if ( $verbose ) {
                $logger->debug(
                    sprintf 'Skipping %s %s [%s] -> either not a corodinator action or not matching the status list',
                                $id || 'N/A',
                                $job->{status},
                                $name,
                );
            }
            next;
        }

        my $key = $job->{coordJobId} . q{#} . $job->{actionNumber};

        # keep them in a hash, we will sort the keys so the actions are in
        # asccending order for a coordinator when issuing the bash commands; only
        # keep the most recent failure for a given coord+action

        my $last_mtime =  $job->{lastModTime_epoch} || $job->{lastModifiedTime_epoch};

        if (   ! $reruns->{$key}
            || $last_mtime > $reruns->{ $key }{last_mtime_epoch}
        ) {
            my $cmd = $job->{status} =~ m{ susp }xmsi
                    ? '-resume'
                    : '-refresh -rerun'
                    ;

            $reruns->{ $key } = {
                action_number    => $job->{actionNumber}     || EMPTY_STRING,
                cmd              => $cmd,
                coord_job_id     => $job->{coordJobId}       || EMPTY_STRING,
                id               => $id                      || EMPTY_STRING,
                last_mtime       => $job->{lastModifiedTime} || EMPTY_STRING,
                last_mtime_epoch => $last_mtime,
                name             => $name                    || EMPTY_STRING,
                nominal_time     => $job->{nominalTime}      || EMPTY_STRING,
            };
        }
    }

    return if ! %{ $reruns };

    return $reruns;
}

1;

__END__

=pod

=encoding UTF-8

=head1 NAME

App::Oozie::Rerun

=head1 VERSION

version 0.020

=head1 SYNOPSIS