view release on metacpan or  search on metacpan

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

        if ( is_interactive() ) {
            printf "Do you want to kill the duplicate coordinator(s)? Ids: '%s' [yN]: \n",
                        join( q{', '}, @other_instances ),
            ;
            $yesno = <STDIN>;
            chomp $yesno;
        }
        else {
            $logger->warn( 'Not running interactively and there are other instances. The next calls will fail.' );
        }

        my $outbuffer;
        if ( lc $yesno eq 'y' ) {
            for (@other_instances) {
                $logger->info( "Killing oozie coordinator $_" );

                my $command = [
                    ( $self->secure_cluster
                        ? ()
                        : ( $self->execute_as_someone_else
                                            ? ( qw[ sudo -u ], $self->username )
                                            : ()
                            )
                    ),
                    $self->oozie_cli,
                    ($self->secure_cluster ? ('-Doozie.auth.token.cache=false') : ()),
                    job => -kill => $_,
                    (-oozie => $self->oozie_uri),
                    ($self->doas ? (-doas => $self->doas) : ()), #impersonation
                ];

                IPC::Cmd::run(
                    command => $command,
                    verbose => $self->verbose,
                    buffer  => \$outbuffer,
                    timeout => $self->timeout,
                ) or do {
                    $logger->error( "Error encountered when trying to kill old instance $_" );
                    die sprintf 'Error: %s', $outbuffer // '[unknown]';
                };
                $is_killed++;
            }
            $logger->info( 'Coordinator(s) are now killed' );
        }
    }

    push @{ $self->errors }, 'At least one coordinator running under the same name' if ! $is_killed;

    return;
}

sub check_coordinator_function_calls {
    my $self = shift;
    my $skip = shift;
    my $logger = $self->logger;
    # try to predict the value of such things?
    # examples:
    #    ${coord:formatTime(coord:nominalTime(), 'yyyy-MM-dd')}
    #    ${coord:formatTime(coord:nominalTime(), 'HH')}
    #
    my $looks_like_coord_conf = qr< [$][{]coord[:] >xms;
    my %missing;
    my $collector = sub {
        my($h, $key) = @_;
        return if $key ne 'property';
        my $slot = $h->{ $key };
        #
        # this is possibly broken for some cases
        # as it tries to locate stuff in the hairy xml
        # which can be defined in different ways.
        # need to be fixed / extended per wf
        #
        foreach my $name ( keys %{ $slot } ) {
            if ( ! is_hashref $slot->{ $name } ) {
                if (   exists $slot->{name}
                    && exists $slot->{value}
                    && $slot->{value} =~ $looks_like_coord_conf
                ) {
                    $missing{ $slot->{name} } = $slot->{value};
                }
                next;
            }

            next if exists $slot->{ $name }{ action };

            if ( my $val = $slot->{ $name }{ value } ) {
                next if $val !~ $looks_like_coord_conf;
                $missing{ $name } = $val;
                next;
            }
        }
    };

    my $loop_xml_conf_hash;
    $loop_xml_conf_hash = sub {
        my $hash = shift;
        my $cb   = shift;
        foreach my $key ( keys %{ $hash } ) {
            $cb->( $hash, $key );
            my $value = $hash->{ $key };
            $loop_xml_conf_hash->( $value, $cb ) if is_hashref $value;
        }
        return;
    };

    foreach my $conf ( qw(
        workflow.xml
        coordinator.xml
    )) {
        my $abs_path = File::Spec->catfile($self->path, $conf );
        my $raw;
        eval {
            $raw = $self->hdfs->read( $abs_path );
            if ( ! $raw ) {
                my $msg = 'Could not read the workflow file in HDFS: '
                        . 'did you do the deploy first? No data for: %s'
                        ;
                $logger->logdie( sprintf $msg, $abs_path );
            }
            my $xs = XML::LibXML::Simple->new;
            my $oozie_conf = $xs->XMLin( \$raw );
            $loop_xml_conf_hash->( $oozie_conf, $collector );
            1;
        } or do {
            my $eval_error = $@ || 'Zombie error';
            my $log_level = $conf =~ m{ workflow }xms ? 'logdie' : 'warn';
            $logger->$log_level( $eval_error );
            1;
        };
    }

    return if ! %missing;

    my @vars = sort {  lc $a cmp lc $b } grep { ! $skip->{ $_ } } keys %missing;

    if ( ! @vars ) {
        my $what = join ', ', sort keys %missing;
        $self->logger->info( sprintf 'The missing coordinator variables (%s) were manually defined. Skipping ...', $what );
        return;
    }

    my $fyi = join SPACE_CHAR, map { "--define '$_=value'" } @vars;

    print <<"DEFINE";
The oozie workflow you are trying to run has several coordinator function
dependencies in it's configuration, but since you've wanted to execute it as
as type=workflow, they won't be defined and the oozie job will either fail