view release on metacpan or  search on metacpan

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

        $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
            }),
        );
    }

    if (@{ $self->errors } ) {
        $logger->error(
            'Overridable errors encountered',
            ( $self->force ? EMPTY_STRING : ' (relaunch using --force to proceed)' )
        );
        $logger->error( '- ' . $_ ) for @{ $self->errors };
        die if !$self->force && !$self->dryrun;
    }

    my $hash_to_def = sub {
        my($h, $no_quote) = @_;
        my $tmpl = $no_quote ? q{-D%s=%s} : q{-D'%s=%s'};
        map { sprintf $tmpl, $_, $h->{$_} } keys %{ $h }
    };

    # IMPORTANT ! keep this in sync with the sudoers file,
    # if you have a corresponding setting in such a place

    my %def = (
        appName                             => '[% app_name      %]',
        startTime                           => '[% start_time    %]',
        endTime                             => '[% end_time      %]',
        workflowPath                        => '[% workflow_path %]',
        'oozie.[% type %].application.path' => '[% workflow_path %]',
        path                                => '[% path %]',
        nameNode                            => '[% name_node     %]',
        'oozie.use.system.libpath'          => 'true',
        ( @define ? ( map { split m{ [=] }xms, $_, 2 } @define ) : () ),
    );

    my %prop  = $self->probe_settings;
    my %owner = $self->probe_meta;

    $self->logger->info( 'Combining owner info into job.properties' );

    my $override_file = File::Temp->new(
                            SUFFIX => '.properties',
                            DIR    => resolve_tmp_dir(),
                        );
    my $original = EMPTY_STRING;
    my $orig_filename = 'job.properties';

    if ( open my $ORIG_FH, '<', $orig_filename ) {
        local $/;
        $original = <$ORIG_FH>;
        if ( ! close $ORIG_FH ) {
            $logger->warn(
                sprintf 'Failed to close %s: %s',
                            $orig_filename,
                            $!,
            );
        }
    }

    $override_file->print( $original, "\n\n" );

    my $c = Config::Properties->new( be_like_java => 1 );
    for my $var ( keys %owner ) {
        $c->setProperty( $var => $owner{ $var } );
    }

    $override_file->print( $c->saveToString );

    my @args = (
        $hash_to_def->( \%def ),
        $hash_to_def->( \%extra_def ),
        @extra_oozie_args,
        '-config' => $override_file,
        ($self->doas ? (-doas => $self->doas) : ()), #impersonation
    );

    # These can't be set in the list above, because of Oozie requiring them
    # to be set before the sub-command as otherwise they become no-op.
    # You gotta love the Hadoop stack.
    #
    # This in turn also requires yet another sudoers template, so be careful
    # with this and contact the BigData Team if you need something to be
    # changed in this section.
    #

    # IMPORTANT! the order matters for this section for sudoers if you have a corresponding setting.
    my @username_override = $prop{username_override}
        ? (
            $hash_to_def->({ 'oozie.auth.token.cache' => 'false'                  }, 1),
            $hash_to_def->({ 'user.name'              => $prop{username_override} }, 1),
            )
        : ();

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

    }

    return %prop;
}

sub ask {
    my $self = shift;
    my $var = shift;
    my $msg = "Please enter the new value for `$var` based on the definition above";
    my($input, $count);
    while ( 1 ) {
        if ( ++$count > MAX_RETRY ) {
            print "\tYou didn't specify anything 3 times, so I give up! ($var=undef)\n";
            last;
        }
        print "$msg: ";
        chomp($input = <STDIN>);
        if ( $input eq EMPTY_STRING ) {
            print "Nothing specified!\n";
            next;
        }
        last;
    }

    if ( $input ne EMPTY_STRING ) {
        print "\t$var will be set to '$input'\n";
    }

    return $input;
}

sub execute {
    my $self    = shift;
    my $command = shift;
    my $logger  = $self->logger;
    my $verbose = $self->verbose;
    my $outbuffer;

    if ( ! $self->dryrun ) {
        $logger->info( 'Executing the command to schedule' );

        my ($ok, $err, $full_buf, $stdout_buff, $stderr_buff);
        ($ok, $err, $full_buf, $stdout_buff, $stderr_buff)  = IPC::Cmd::run(
            command => $command,
            verbose => 1,
            buffer  => \$outbuffer,
            timeout => $self->timeout,
        );

        if ( ! $ok ) {
            $logger->fatal( $err );
            return 0;
        }

        if ( is_arrayref $stderr_buff && @{ $stderr_buff } ) {
            $logger->warn( join "\n", @{ $stderr_buff } );
        }

        ($outbuffer) = reverse
                        map {
                            split m{ \n }xms, $_
                        }
                        @{ $stdout_buff }
                        ;

        if (
            # TODO: move to constant
            $outbuffer =~ m{
                ([0-9-]+ oozie-oozi- [CWB])
            }xms
        ) {
            my $job_id =  $1;
            $self->log_console_url( $job_id );
        }
        else {
            $logger->warn( 'Failed to locate the Oozie job id from the system call' );
        }

        return 1;
    }

    my $type = $self->type;
    my @info = (
        'Running oozie --dryrun for your command',
        sprintf( 'This will dryrun or test run a %s job, no job will be queued or scheduled ', $type ),
    );
    push @info, 'In case of --type=wf you will only get an OK in case of success' if $type eq 'wf';
    push @info, 'oozie --dryrun Output START';

    $logger->info( $_ ) for @info;

    IPC::Cmd::run(
        command => $command,
        verbose => 1,
        buffer  => \$outbuffer,
        timeout => $self->timeout,
    ) or do {
        $logger->fatal( sprintf 'Error encountered when trying to dryrun the new %s!!!', $type );
        $logger->fatal( 'oozie response: ' . $outbuffer );
        $logger->info( 'oozie --dryrun Output END' );
        return 0;
    };

    $logger->info( 'oozie response: ' . $outbuffer );
    $logger->info( 'oozie --dryrun Output END' );

    return 1;
}

sub log_console_url {
    my $self   = shift;
    my $job_id = shift;
    $self->logger->info( sprintf 'Console URL: %s?job=%s', $self->oozie_uri, $job_id );
    return;
}

1;

__END__

=pod