App-Oozie
view release on metacpan or search on metacpan
lib/App/Oozie/Deploy.pm view on Meta::CPAN
package App::Oozie::Deploy;
use 5.014;
use strict;
use warnings;
our $VERSION = '0.020'; # VERSION
use namespace::autoclean -except => [qw/_options_data _options_config/];
use App::Oozie::Constants qw(
DEFAULT_DIR_MODE
DEFAULT_FILE_MODE
EMPTY_STRING
FILE_FIND_FOLLOW_SKIP_IGNORE_DUPLICATES
MILISEC_DIV
MODE_BITSHIFT_READ
SPACE_CHAR
STAT_MODE
TERMINAL_INFO_LINE_LEN
WEBHDFS_CREATE_CHUNK_SIZE
);
use Cwd 'abs_path';
use Moo;
use MooX::Options prefer_commandline => 0,
protect_argv => 0,
usage_string => <<'USAGE',
Usage: %c %o
Deploys workflows to HDFS. Specifying names as final arguments will upload only those
USAGE
;
use App::Oozie::Deploy::Template;
use App::Oozie::Deploy::Validate::Spec;
use App::Oozie::Types::Common qw( IsDir IsFile );
use App::Oozie::Util::Misc qw( resolve_tmp_dir trim_slashes );
use App::Oozie::Constants qw( OOZIE_STATES_RUNNING );
use Carp ();
use Config::Properties;
use Config::General ();
use DateTime::Format::Strptime;
use DateTime;
use Email::Valid;
use Fcntl qw( :mode );
use File::Basename qw( basename dirname );
use File::Find ();
use File::Find::Rule;
use File::Spec;
use File::Temp ();
use List::MoreUtils qw( uniq );
use List::Util qw( max );
use Path::Tiny qw( path );
use Ref::Util qw( is_arrayref is_hashref );
use Sys::Hostname ();
use Template;
use Text::Glob qw(
match_glob
glob_to_regex
);
use Time::Duration qw( duration_exact );
use Time::HiRes qw( time );
use Types::Standard qw(
ArrayRef
CodeRef
StrictNum
Str
);
with qw(
App::Oozie::Role::Log
App::Oozie::Role::Fields::Common
App::Oozie::Role::NameNode
App::Oozie::Role::Git
App::Oozie::Role::Meta
App::Oozie::Role::Info
);
option write_ownership_to_workflow_xml => (
is => 'rw',
default => sub { 1 },
doc => 'Populate the meta file into workflow.xml? This option is temporary while testing',
);
option hdfs_dest => (
is => 'rw',
format => 's',
doc => 'HDFS destination (default is <default_hdfs_destination>/<name>)',
);
lib/App/Oozie/Deploy.pm view on Meta::CPAN
my $logger = $self->logger;
my $oozie_base_dir = $self->local_oozie_code_path;
my $ttlib_base_dir = $self->ttlib_base_dir;
my $verbose = $self->verbose;
my $is_file = IsFile->library->get_type( IsFile );
foreach my $file ( @{ $self->required_tt_files } ) {
my $absolute_path = File::Spec->catfile( $ttlib_base_dir, $file );
if ( $verbose ) {
$logger->debug("Assert file: $absolute_path");
}
# assert_valid() does not display the error message, hence the manual check
my $error = $is_file->validate( $absolute_path ) || next;
$logger->logdie( sprintf 'required_tt_files(): %s', $error );
}
if ( $verbose ) {
$logger->debug( join q{=}, $_, $self->$_ ) for qw(
local_oozie_code_path
ttlib_base_dir
);
}
if ( $self->dump_xml_to_json && ! $self->dryrun ) {
$self->logger->info( 'dump_xml_to_json is enabled without a dryrun. Enabling dryrun as well.' );
$self->dryrun( 1 );
}
return;
}
sub run {
my $self = shift;
my $workflows = shift;
my $logger = $self->logger;
my $config = $self->internal_conf;
my $dryrun = $self->dryrun;
my $verbose = $self->verbose;
my $run_start_epoch = time;
my $log_marker = q{#} x TERMINAL_INFO_LINE_LEN;
$logger->info(
sprintf '%s Starting deployment in %s%s %s',
$log_marker,
$self->cluster_name,
$verbose ? EMPTY_STRING : '. Enable --verbose to see the underlying commands',
$log_marker,
);
$self->log_versions if $verbose;
my($update_coord) = $self->_verify_and_compile_all_workflows( $workflows );
if (!$self->secure_cluster) {
# Left in place for historial reasons.
# All clusters should be under Kerberos.
# Possible removal in a future version.
#
# unsafe, but needed when uploading with mapred's uid or hdfs dfs cannot see the files
chmod oct( DEFAULT_FILE_MODE ), $config->{base_dest};
}
my $success = $self->upload_to_hdfs;
$self->maybe_update_coordinators( $update_coord ) if @{ $update_coord };
if ($self->prune) {
$logger->info( '--prune is set, checking workflow directories for old files' );
for my $workflow ( @{ $workflows } ) {
$self->prune_path(
File::Spec->catdir(
$config->{hdfs_dest},
basename $workflow
)
);
}
}
$logger->info(
sprintf '%s Completed successfully in %s (took %s) %s',
$log_marker,
sprintf( '%s%s', $self->cluster_name, ( $dryrun ? ' (dryrun is set)' : EMPTY_STRING ) ),
duration_exact( time - $run_start_epoch ),
$log_marker,
);
return $success;
}
sub _verify_and_compile_all_workflows {
my $self = shift;
my $workflows = shift;
my $logger = $self->logger;
if ( ! is_arrayref $workflows || ! @{ $workflows } ) {
$logger->logdie( 'Please give one or several workflow name(s) on the command line (glob pattern accepted). Also see --help' );
}
$self->pre_verification( $workflows );
$self->verify_temp_dir;
if ( $self->gitfeatures
&& ! $self->gitforce
) {
$self->verify_git_tag;
}
my $wfs = $self->collect_names_to_deploy( $workflows );
my($total_errors, $validation_errors);
my @update_coord;
for my $workflow ( @{ $wfs } ) {
my($t_validation_errors, $t_total_errors, $dest, $cvc) = $self->process_workflow( $workflow );
push @update_coord, $self->guess_running_coordinator( $workflow, $cvc, $dest );
$total_errors += $t_validation_errors;
$validation_errors += $t_total_errors;
}
if ($total_errors) {
$logger->fatal( sprintf 'ERROR: %s errors were encountered during this run. Please fix it!', $total_errors );
$logger->fatal( 'The --force option has been disabled, as not enough really paid attention.' );
$logger->fatal( 'Fixing the errors is really your best and easiest option.' );
lib/App/Oozie/Deploy.pm view on Meta::CPAN
my @firstLevelWorkflows =
File::Find::Rule->directory
->maxdepth( 1 )
->mindepth( 1 )
->extras({
follow => 1,
follow_skip => FILE_FIND_FOLLOW_SKIP_IGNORE_DUPLICATES,
})
->name(@firstLevelMatchingPatterns)
->in( $owf_base );
#Don't want to be matching the 'workflows' part in workflows/stuff/workflow
my $workflowFolderPrefixLength = length( $owf_base ) + 1;
my @secondLevelWorkflows =
File::Find::Rule
->directory
->maxdepth( 2 )
->mindepth( 2 )
->extras({
follow => 1,
follow_skip => FILE_FIND_FOLLOW_SKIP_IGNORE_DUPLICATES,
})
->exec(
sub{
my $str = substr $_[2], $workflowFolderPrefixLength;
# might be a good idea to limit matching
# globs to the last level of folder structure
# only (e.g. no "f*g/k*s")
return grep { match_glob($_, $str) }
@secondLevelMatchingPatterns
}
)
->in( $owf_base );
for my $i ( 0..$#firstLevelWorkflows ) {
my $workflowFileLocationGuess = File::Spec->rel2abs(
$firstLevelWorkflows[$i].'/workflow.xml'
);
my $bundleFileLocationGuess = File::Spec->rel2abs($firstLevelWorkflows[$i].'/bundle.xml');
if (! -f $workflowFileLocationGuess) {
my $msg = q{It doesn't look like there's a workflow at `%s`. }
. q{I will process its subfolders, if any, instead.};
$logger->info( sprintf $msg, $firstLevelWorkflows[$i] );
my @subs = File::Find::Rule->directory
->maxdepth(1)
->mindepth(1)
->in( $firstLevelWorkflows[$i] )
;
if (@subs) {
for my $subx ( @subs ) {
$logger->debug(
sprintf 'I will additionally look for workflows in the following sub folder: %s',
$subx,
);
}
}
push @secondLevelWorkflows, @subs;
# When deploying a wf/coord, it makes no sense to upload what we have in parent,
# otherwise, yes, we'll need to update bundle.xml along with any other file on it
if (! -f $bundleFileLocationGuess) {
splice @firstLevelWorkflows, $i, 1;
$i--;
}
else {
$self->logger->debug( 'We have identified this a a bundle.' );
}
}
}
@firstLevelWorkflows = uniq @firstLevelWorkflows;
@secondLevelWorkflows = uniq @secondLevelWorkflows;
my $num_workflows = @secondLevelWorkflows + @firstLevelWorkflows;
if ( ! $num_workflows ) {
die "Exiting: found no workflows to deploy under `$owf_base`.";
}
if ($workflowPatternCount > $num_workflows) {
my @uniq_wfs = (@secondLevelWorkflows, @firstLevelWorkflows);
my @params = (
$num_workflows,
$workflowPatternCount,
join(qq{\n\t}, @{ $names } ),
join(qq{\n\t}, @uniq_wfs),
);
my $msg = sprintf <<"ERROR", @params;
Exiting: only %s workflow folders found when we expected at least %s (from the number of command-line arguments).
Expected workflows:
\t%s
Computed:
\t%s
Hint: you might have a character which could look like a dash but it is not in your arguments.
If this is the case, such an argument will be treated as a workflow name.
ERROR
;
$logger->logdie( $msg );
}
@workflow = ( @firstLevelWorkflows, @secondLevelWorkflows );
for my $wf ( @workflow ) {
$logger->info( "Workflow to be deployed: $wf" );
}
return \@workflow;
}
sub collect_data_for_deployment_meta_file {
my $self = shift;
my $workflow = shift;
my $total_errors = shift;
my $use_git = $self->gitfeatures;
lib/App/Oozie/Deploy.pm view on Meta::CPAN
$!,
);
}
return;
}
sub prune_path {
my $self = shift;
my $path = shift || die 'No path was specified';
my $files = $self->hdfs->list($path);
my $total_files = scalar @{ $files };
my $deleted_files = 0;
my $deploy_start = $self->deploy_start;
my $dryrun = $self->dryrun;
for my $file ( @{ $files } ) {
#next if $file->{pathSuffix} =~ /^(\.deployment|coordinator\.xml)$/;
if ( $file->{type} eq 'FILE'
&& $file->{modificationTime} / MILISEC_DIV < $deploy_start
) {
my $msg = sprintf 'Old file found in destination: %s (mtime %s) -> %s',
$file->{pathSuffix},
$self->date->epoch_yyyy_mm_dd_hh_mm_ss(
int( $file->{modificationTime} / MILISEC_DIV )
),
$dryrun ? 'would have deleted if dryrun was not specified' : 'is now deleted',
;
$self->logger->info( $msg );
$self->hdfs->delete("$path/$file->{pathSuffix}") if ! $dryrun;
$deleted_files++;
}
# check directories regardless of age
if( $file->{type} eq 'DIRECTORY' ) {
my $msg = sprintf 'Directory found in destination: %s (mtime %s) -> checking contents',
$file->{pathSuffix},
$self->date->epoch_yyyy_mm_dd_hh_mm_ss(
int( $file->{modificationTime} / MILISEC_DIV )
),
;
$self->logger->info( $msg );
#recurse down to check lower directories
my $empty = $self->prune_path("$path/$file->{pathSuffix}");
if( $empty ) {
$self->logger->info( "$file->{pathSuffix} is empty, " . ( $dryrun ? 'would have deleted if dryrun was not specified' : 'deleting' ) );
$self->hdfs->delete("$path/$file->{pathSuffix}") if ! $dryrun;
$deleted_files++;
} else {
$self->logger->info( "$file->{pathSuffix} has current files, keeping it" );
}
}
}
return ($total_files == $deleted_files);
}
sub upload_to_hdfs {
my $self = shift;
my $config = $self->internal_conf;
if ( $self->dryrun ) {
$self->logger->warn(
sprintf 'Skipping upload to HDFS as dryrun was set. Would have uploaded from %s to %s',
$config->{base_dest},
$config->{hdfs_dest},
);
return 1;
}
my $success = $self->_copy_to_hdfs_with_webhdfs($config->{base_dest}, $config->{hdfs_dest});
return $success;
}
sub _hdfs_exists_no_exception {
my $self = shift;
my $path = shift;
my $hdfs = $self->hdfs;
my $rv;
eval {
$rv = $hdfs->exists( $path );
1;
} or do {
my $eval_error = $@ || 'Zombie error';
if ( $self->verbose ) {
$self->logger->debug(
sprintf 'WebHDFS exists() failed with exception, however since this is a silent call, it is ignored: %s',
$eval_error,
)
}
};
return $rv;
}
sub _copy_to_hdfs_with_webhdfs {
my $self = shift;
my $sourceFolder = shift;
my $destFolder = shift;
my $hdfs = $self->hdfs;
my $logger = $self->logger;
my $verbose = $self->verbose;
$logger->info(
sprintf 'copying from `%s` to `%s`',
$sourceFolder,
$destFolder,
);
if ( ! $self->_hdfs_exists_no_exception( $destFolder ) ) {
if ( $verbose ) {
$logger->debug(
sprintf 'HDFS destination %s does not exist',
$destFolder,
);
}
my(undef, @paths) = File::Spec->splitpath( $destFolder );
my $remote_base;
for my $chunk ( @paths ) {
if ( $remote_base ) {
$remote_base = File::Spec->catdir( $remote_base, $chunk);
lib/App/Oozie/Deploy.pm view on Meta::CPAN
1;
__END__
=pod
=encoding UTF-8
=head1 NAME
App::Oozie::Deploy
=head1 VERSION
version 0.020
=head1 SYNOPSIS
use App::Oozie::Deploy;
App::Oozie::Deploy->new_with_options->run;
=head1 DESCRIPTION
This is an action/program in the Oozie Tooling.
=for Pod::Coverage BUILD
=head1 NAME
App::Oozie::Deploy - The program to deploy Oozie workflows.
=head1 Methods
=head2 collect_data_for_deployment_meta_file
=head2 collect_names_to_deploy
=head2 compile_templates
=head2 create_deployment_meta_file
=head2 destination_path
=head2 guess_running_coordinator
=head2 max_wf_xml_length
=head2 maybe_update_coordinators
=head2 pre_verification
=head2 process_templates
=head2 process_workflow
=head2 prune_path
=head2 run
=head2 upload_to_hdfs
=head2 validate_meta_file
=head2 verify_temp_dir
=head2 write_deployment_meta_file
=head1 Accessors
=head2 Overridable from cli
=head3 dump_xml_to_json
=head3 hdfs_dest
=head3 keep_deploy_path
=head3 oozie_workflows_base
=head3 prune
=head3 sla
=head3 write_ownership_to_workflow_xml
=head2 Overridable from sub-classes
=head3 configuration_files
=head3 deploy_start
=head3 deployment_meta_file_name
=head3 email_validator
=head3 internal_conf
=head3 max_node_name_len
=head3 process_coord_directive_varname
=head3 required_tt_files
=head3 spec_queue_is_missing_message
=head3 ttlib_base_dir
=head3 ttlib_dynamic_base_dir_name
=head1 SEE ALSO
L<App::Oozie>.
=head1 AUTHORS
=over 4
=item *
David Morel
( run in 2.415 seconds using v1.01-cache-2.11-cpan-b16cb0d3907 )