view release on metacpan or  search on metacpan

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

                  protect_argv       => 0,
                  usage_string       => <<'USAGE',
Usage: %c %o [options] workflow-name
USAGE
;

with qw(
    App::Oozie::Role::Log
    App::Oozie::Role::Fields::Common
    App::Oozie::Role::Fields::Path
    App::Oozie::Role::Meta
    App::Oozie::Role::Info
);

#------------------------------------------------------------------------------#

option appname => (
    is     => 'rw',
    format => 's',
    doc    => remove_newline( <<'DOC' ),
Workflow name, useful if you want to run different instances of the same
workflow with different parameters. If not set, then this will default to
the workflow basename.
DOC
);

my $__JOB_TYPES = eval {
    sprintf ' Valid values are %s',
            join q{, },
            @{ IsJobType->parent->values },
    ;
} || EMPTY_STRING;

option type => (
    default => sub { 'coord' },
    is      => 'rw',
    isa     => IsJobType,
    format  => 's',
    doc     => remove_newline( sprintf <<'DOC', $__JOB_TYPES ),
Defines the type of job the user needs to launch. If nothing is specified,
the script will check the existence of a coordinator.xml file, to determine
whether this should be launched as a coordinator or a single workflow.%s
DOC
);

option notify => (
    is      => 'rw',
    default => sub { 1 },
);

option define => (
    is      => 'rw',
    format  => 's@',
    default => sub { [] },
    doc     => q{Define extra parameters for oozie, like "--define 'foo=bar'"},
);

option path => (
    is      => 'rw',
    format  => 's',
    default => \&_option_build_guess_wf_path,
    lazy    => 1,
    doc     => 'HDFS location for the workflow. Defaults to <default_hdfs_destination>/<workflow-basename>',
);

option sla_duration => (
    is      => 'rw',
    isa     => Int,
    format  => 'i',
    doc     => remove_newline( <<'DOC' ),
the runtime, in minutes, after which a workflow deployed using the --sla
switch should send an sla-duration-miss email to the errorEmailTo
recipient(s). This may be adjusted dynamically and automatically after
several runs have happened to provide enough statistics, but is for now
required to deploy a new workflow needing SLA events
DOC
);

option starthour => (
    is      => 'rw',
    format  => 's',
    default => sub { 0 },
    isa     => IsHour,
    doc     => remove_newline( <<'DOC' ),
hour of day (0 to 23) for the 1st coordinator run. Applies to all
coordinators, even hourly ones. Defaults to midnight (in UTC)
DOC
);

option startmin => (
    is      => 'rw',
    format  => 's',
    default => sub { 0 },
    isa     => IsMinute,
    doc     => remove_newline( <<'DOC' ),
minute within starthour 00 to 59 for the coordinator run. Applies to all
coordinators. Defaults to 00
DOC
);

option endhour => (
    is      => 'rw',
    format  => 's',
    default => sub { 0 },
    isa     => IsHour,
    doc     => remove_newline( <<'DOC' ),
hour of day (0 to 23) for the last coordinator run. Applies to all
coordinators, even hourly ones. Defaults to midnight (in UTC)
DOC
);

option endmin => (
    is      => 'rw',
    format  => 's',
    default => sub { 0 },
    isa     => IsMinute,
    doc     => remove_newline( <<'DOC' ),
minute within endhour 00 to 59 for the coordinator run. Applies to all
coordinators. Defaults to 00
DOC
);

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

        if ( $is_shortcut{ $startdate } ) {
            $startdate = $date->$startdate();
        }
        else {
            my $intersects = $date->intersection(
                $startdate,
                $startdate,
                $date->move( $date->today, -DEFAULT_START_DATE_DAY_FRAME ),
                $date->move( $date->today,  DEFAULT_START_DATE_DAY_FRAME ),
            );

            if ( ! $intersects ) {
                push @{ $self->errors },
                    sprintf 'Start date is out of normal bounds (%s days in the past or in the future)',
                                DEFAULT_START_DATE_DAY_FRAME,
                    ;
            }
        }
    }
    else {
        $startdate = $date->tomorrow;
    }

    $self->startdate( $startdate );
    my $enddate = $self->enddate;

    $enddate //= $date->move( $date->today, DEFAULT_END_DATE_DAYS );

    if ( $is_shortcut_date->{ $enddate } ) {
        $enddate = $date->$enddate();
    }

    if ( $enddate lt $startdate ) {
        die 'End date should be later than start date';
    }

    if ( ! $self->force
        && abs $date->diff($enddate, $date->today) > DEFAULT_END_DATE_DAYS
    ) {
        die sprintf 'End date should not be later than %s days from today',
                        DEFAULT_END_DATE_DAYS,
        ;
    }

    $self->enddate( $enddate );

    return;
}

#------------------------------------------------------------------------------#

has basedir => (
    is => 'rw',
);

has errors => (
    is      => 'rw',
    default => sub { [] },
);

sub _option_build_guess_wf_path {
    my $self    = shift;
    my $wf_dir = $self->basedir;

    my $rv;
    # Should be the same on local file system and HDFS
    my $relativePath;
    my $local_wf_basedir = '/workflows/';

    if (File::Spec->file_name_is_absolute($wf_dir)) {
        my $workflowsPartIndex = rindex($wf_dir, $local_wf_basedir);
        if ( $workflowsPartIndex != INDEX_NOT_FOUND ) {
            $relativePath = substr $wf_dir, $workflowsPartIndex + length($local_wf_basedir);
        }
    }
    else {
        $relativePath = $wf_dir;
    }

    if ( $relativePath ) {
        $rv = File::Spec->catfile(
                    $self->oozie_basepath,
                    trim_slashes( $relativePath ),
                );
    }
    else {
        die 'Failed to guess the workflow path!';
    }

    return $rv;
}

sub run {
    my $self   = shift;
    my $wf_dir = shift || $self->logger->logdie( 'Please specify a workflow/coordinator/bundle to run' );

    my $logger  = $self->logger;

    my $run_start_epoch = time;

    for my $huh ( @_ ) {
        $logger->warn( sprintf 'Unknown parameter: %s', $huh // '[undefined]');
    }

    my $verbose = $self->verbose;

    $logger->info( 'Starting' . ( $verbose ? EMPTY_STRING : '. Enable --verbose to see the underlying commands' ) );

    $self->log_versions if $self->verbose;

    $self->basedir( $wf_dir );

    my $CWD = getcwd() || die "Can't happen: unable to get cwd: $!";
    if ( ! chdir $wf_dir ) {
        die sprintf 'Cannot chdir to %s: %s -- Current dir: %s', $wf_dir, $!, $CWD;
    }

    if ( ! $self->appname ) {
        my $guess = basename getcwd;
        $logger->info( 'appname is not set, using the basedir=' . $guess );
        $self->appname( $guess );
    }

    # move to constructor?
    (my $appname = $self->appname) =~ s{ [/]+ \z }{}xms;
    $self->appname( $appname );

    $self->logger->info( sprintf 'Job name: %s',            $self->appname );
    $self->logger->info( sprintf 'Job path (HDFS dir): %s', $self->path    );

    $self->setup_dates;

    my($cmd_tmpl, $cmd_param) = $self->collect_oozie_cmd_args;

    # Are we alone or do we need to kill our brothers and sisters?
    $self->check_current_instances;

    Template->new
            ->process(
                \join( SPACE_CHAR, @{ $cmd_tmpl } ),
                $cmd_param,
                \my $command,
            );
    my $success = $self->execute( $command );
    if(!$success){
        return $success;
    }
    # go back where we started!
    chdir $CWD if $CWD;


    $logger->info(
        sprintf 'Completed successfully in %s (took %s)',
                    sprintf( '%s%s', $self->cluster_name, ( $self->dryrun ? ' (dryrun is set)' : EMPTY_STRING ) ),
                    duration_exact( time - $run_start_epoch ),
    );

    return $success;
}

sub collect_oozie_cmd_args {
    my $self   = shift;
    my $logger = $self->logger;

    my @extra_oozie_args;

    my @define = @{ $self->define };

    my %extra_def = ();
    # We are not supporting sla for bundles (yet)
    if ( !($self->type eq 'bundle') ) {
      %extra_def = (
          $self->verify_sla,
      );
    }

    if ( $self->type eq 'wf' ) {
        %extra_def = (
            %extra_def,
            $self->check_coordinator_function_calls({
                map { (split m{ [=] }xms, $_)[0] => 1 } @define