DBIx::JCL - Job Control Library for database load tasks.
Index
Code Index:
NAME
DBIx::JCL - Job Control Library for database load tasks.
SYNOPSIS
# file: test_job.pl
use strict;
use warnings;
use DBIx::JCL qw( :all );
my $jobname = 'name_of_job';
sys_init( $jobname );
# perform database tasks calling DBIx-JCL functions
# ...
sys_end();
exit sys_get_errorlevel();
DESCRIPTION
This documentation describes the perl module DBIx-JCL.pm and the use of
standardized perl scripts which together provide a common job execution
environment to support database backend load and maintenance tasks.
RATIONALE
Provide a suite of standard functions that can be shared across all batch
job scripts used to support database back end tasks. Provide a standardized
approach for the development of all back end database job scripts.
Centralize the administration and access to configuration data. Enforce
coding standards and documentation. Abstract the sql used to support back
end processes from the task processing logic, by placing all sqlinto an sql
library. This will make maintenance of back end sql a trivial task. Provide
generalized logging, notification, and system information functions.
If you want to write a robust database extract and load job with complete
support for logging and error notification, and do it in 25 lines of code,
read on.
OPTIONS
Database maintenance and load jobs written using DBIx-JCL support the following
options out-of-the-box, with no additional work required on your part.
Job Options:
| -r | Run job
| -rb | Run job in the background
| -rs | Run job at requested start time
| -rr | Restart job after failure
| -rde | Run using specified DE number
| -x | Pass extra parameters to job script
| -c | Specify database connections
| -v | Verbose
| -vv | Very Verbose
| -ng | No greeting
| -tc | Test database connections
Logging Options:
| -lf | Log filename
| -lg | Log generations
| -ll | Log log levels
| -lp | Log file prefix
| -lr | Log archive file radix
| -cl | Log console levels
Notificaiton Options:
| -ne | Notify email on completion
| -np | Notify pager on completion
| -et | Email notification to list
| -el | Email notification levels
| -pt | Pager notification to list
| -pl | Pager notification levels
Information Options:
| -dp | Display job parameters
| -dq | Display job querys
| -dd | Display job documentation
| -dl | Display last log file
| -da | Display archived log files
| -dj | Display a list of job scripts
| -dja | Diaplay jobs active in the system
Utility Options:
| -se | Send email message
| -sp | Send pager message
| -um | Util no move files
| -h | Help
| -ha | Help on option arguments
Please see ADDITIONAL INFORMATION below.
CAPABILITIES
The DBIx-JCL modules provides many capabilities commonly needed in support of
database maintenance jobs designed to run in a production environment. Below
is a summary list of features and the types of functions provided to support
those features.
Features
The following features have been designed in to the DBIx-JCL module:
- * Logging support with log file rotation
- * Notification support
- * Simplified DBI interface
- * Configuration data stored externally
- * High level functions not available in the DBI
- * SQL stored in "SQL books"
- * Job documentation enforced
- * Job control functions
- * Plugin support
Implementation
The features listed above have been implemented by providing [many] functions
for use by your database mantenance jobs:
- * Functions for command line interaction
- * Functions for initialization, monitoring, and control
- * Functions for database interaction
- * Functions for log file access and maintenance
- * Functions for file manipulation
Please see ADDITIONAL INFORMATION below.
EXAMPLE JOB
Shown below is the standard approach to writing job scripts.
##@@name_of_script.pl,bin
##$$Description of this job
use strict;
use warnings;
use DBIx::JCL qw( :all );
# initialize
# -------------------------------------------------------------------------
my $jobname = 'name_of_script';
sys_init( $jobname );
my $dbenv1 = 'mydb1';
my $mysql1 = sys_get_sql( 'query_number_1' );
# main
# -------------------------------------------------------------------------
log_info( sys_get_dbdescr( $dbenv1 ) );
db_connect( $dbenv1 );
# do more db stuff here
# end
# -------------------------------------------------------------------------
=begin wiki
!1 NAME
Name of script
----
!1 DESCRIPTION
Describe the job script here.
----
!1 RECOVERY NOTES
Document recovery notes here.
----
!1 DEPENDENCIES
Document dependencies here.
=cut
__END__
Please see ADDITIONAL INFORMATION below.
Please see the documentation embedded in this source file for [LOTS!] of
additional details on how to use JCL.pm. You can view this documentation using
WikiText.pm module to format the WikiText content in this file. Hint: download
and install WikiText.pm.
Thank you!
COPYRIGHT
Copyright 2008 Brad Adkins <dbijcl@gmail.com>.
Permission is granted to copy, distribute and/or modify this document under the
terms of the GNU Free Documentation License, published by the Free Software
Foundation; with no Invariant Sections, with no Front-Cover Texts, and with no
Back-Cover Texts.
AUTHOR
Brad Adkins, dbijcl@gmail.com
##@@JCL.pm,dbixlib
##$$Job Control Library for Data Management Tasks
##author:Brad Adkins
##format:codehtml
##outfile:JCL.html
##title:Job Control Library
##toc:yes
##header:<h1>DBIx-JCL</h1>
# package
# ------------------------------------------------------------------------------
package DBIx::JCL;
use strict;
use warnings;
# package exports
# ------------------------------------------------------------------------------
require Exporter;
use base qw( Exporter );
our @EXPORT_OK = qw(
sys_init
sys_init_setuser
sys_end
sys_init_plugin
sys_get_sql
sys_get_item
sys_get_hash
sys_get_array
sys_get_common_sql
sys_get_run_control
sys_get_dbdescr
sys_get_dbinst
sys_set_restart
sys_load_library
sys_set_verbose
sys_die
sys_warn
sys_info
sys_ctime2str
sys_disp_active_jobs
sys_run_job
sys_run_job_background
sys_run_job_wait
sys_run_job_maxrc
sys_run_job_reset
sys_get_path_bin_dir
sys_get_path_lib_dir
sys_get_path_log_dir
sys_get_path_load_dir
sys_get_path_extr_dir
sys_get_path_scripts_dir
sys_get_path_plugin_dir
sys_get_path_prev_dir
sys_get_mail_server
sys_get_mail_from
sys_get_mail_emailto
sys_get_mail_pagerto
sys_get_mail_email_levels
sys_get_mail_pager_levels
sys_get_log_file
sys_get_log_filefull
sys_get_log_logging_levels
sys_get_log_console_levels
sys_get_log_gdg
sys_get_dataenvr
sys_get_errorlevel
sys_get_conf_dir
sys_get_email_levels
sys_get_pager_levels
sys_get_logging_levels
sys_get_console_levels
sys_get_commandline
sys_get_commandline_opt
sys_get_commandline_val
sys_get_script_file
sys_get_user
sys_get_util_move
sys_get_maxval
sys_set_errorlevel
sys_set_die
sys_set_warn
sys_set_conf_file
sys_set_email_levels
sys_set_pager_levels
sys_set_mail_emailto
sys_set_logging_levels
sys_set_console_levels
sys_set_script_file
sys_set_path_log_dir
sys_set_path_plugin_dir
sys_set_maxval
sys_check_dataenvr
sys_timer
sys_wait
sys_disp_doc
log_fatal
log_error
log_warn
log_info
log_debug
log_close
log_write_log
log_write_screen
db_init
db_connect
db_nil
db_finish
db_disconnect
db_prepare
db_execute
db_commit
db_get_sth
db_get_defenvr
db_pef
db_pef_list
db_fetchrow
db_bindcols
db_rollback
db_insert_from_file
db_query_to_file
db_dump_query
db_dump_table
db_grant
db_func
db_proc
db_proc_in
db_proc_out
db_proc_inout
db_rowcount_query
db_sanity_check
db_rowcount_table
db_truncate
db_dbms_output_enable
db_dbms_output_disable
db_dbms_output_get
db_drop_index
db_drop_table
db_drop_procedure
db_drop_function
db_drop_package
db_rename_index
db_rename_table
db_purge_table
db_purge_index
db_update_statistics
db_sqlloader
db_sqlloaderx
db_sqlloaderx_parse_logfile
db_sqlloaderx_read
db_sqlloaderx_skipped
db_sqlloaderx_rejected
db_sqlloaderx_discarded
db_sqlloaderx_elapsed_time
db_sqlloaderx_cpu_time
db_index_rebuild
db_exchange_partition
util_get_filename_load
util_get_filename_extr
util_get_filename_log
util_read_header
util_read_footer
util_read_file
util_write_header
util_write_footer
util_move
util_trim
util_zsdf
test_init
test_ok
test_results
test_harness_init
test_harness_run
test_harness_results
$VERSION
$SQLLDR_SUCC
$SQLLDR_WARN
$SQLLDR_FAIL
$SQLLDR_FTL
);
our %EXPORT_TAGS = (
all => [
@EXPORT_OK
],
sys => [ qw(
sys_init
sys_init_setuser
sys_end
sys_init_plugin
sys_get_sql
sys_get_item
sys_get_hash
sys_get_array
sys_get_common_sql
sys_get_run_control
sys_get_dbdescr
sys_get_dbinst
sys_set_restart
sys_load_library
sys_set_verbose
sys_die
sys_warn
sys_info
sys_ctime2str
sys_disp_active_jobs
sys_run_job
sys_run_job_background
sys_run_job_wait
sys_run_job_maxrc
sys_run_job_reset
sys_get_path_bin_dir
sys_get_path_lib_dir
sys_get_path_log_dir
sys_get_path_load_dir
sys_get_path_extr_dir
sys_get_path_prev_dir
sys_get_path_scripts_dir
sys_get_mail_server
sys_get_mail_from
sys_get_mail_emailto
sys_get_mail_pagerto
sys_get_mail_email_levels
sys_get_mail_pager_levels
sys_get_log_file
sys_get_log_filefull
sys_get_log_logging_levels
sys_get_log_console_levels
sys_get_log_gdg
sys_get_dataenvr
sys_get_errorlevel
sys_get_conf_dir
sys_get_email_levels
sys_get_pager_levels
sys_get_logging_levels
sys_get_console_levels
sys_get_commandline
sys_get_commandline_opt
sys_get_commandline_val
sys_get_script_file
sys_get_path_plugin_dir
sys_get_util_move
sys_get_user
sys_get_maxval
sys_set_errorlevel
sys_set_die
sys_set_warn
sys_set_email_levels
sys_set_pager_levels
sys_set_mail_emailto
sys_set_logging_levels
sys_set_console_levels
sys_set_script_file
sys_set_conf_file
sys_set_path_log_dir
sys_set_path_plugin_dir
sys_set_maxval
sys_check_dataenvr
sys_timer
sys_wait
sys_disp_doc
) ],
log => [ qw(
log_fatal
log_error
log_warn
log_info
log_debug
log_close
log_write_log
log_write_screen
) ],
db => [ qw(
db_init
db_connect
db_nil
db_finish
db_disconnect
db_prepare
db_execute
db_commit
db_get_sth
db_get_defenvr
db_pef
db_pef_list
db_fetchrow
db_bindcols
db_rollback
db_insert_from_file
db_query_to_file
db_dump_query
db_dump_table
db_grant
db_func
db_proc
db_proc_in
db_proc_out
db_proc_inout
db_rowcount_query
db_sanity_check
db_rowcount_table
db_truncate
db_dbms_output_enable
db_dbms_output_disable
db_dbms_output_get
db_drop_index
db_drop_table
db_drop_procedure
db_drop_function
db_drop_package
db_rename_index
db_rename_table
db_purge_table
db_purge_index
db_update_statistics
db_sqlloader
db_sqlloaderx
db_sqlloaderx_parse_logfile
db_sqlloaderx_read
db_sqlloaderx_skipped
db_sqlloaderx_rejected
db_sqlloaderx_discarded
db_sqlloaderx_elapsed_time
db_sqlloaderx_cpu_time
db_index_rebuild
db_exchange_partition
) ],
util => [ qw(
util_get_filename_load
util_get_filename_extr
util_get_filename_log
util_read_header
util_read_footer
util_read_file
util_write_header
util_write_footer
util_move
util_trim
util_zsdf
) ],
test => [ qw(
test_init
test_ok
test_results
test_harness_init
test_harness_run
test_harness_results
) ],
const => [ qw(
$SQLLDR_SUCC
$SQLLDR_WARN
$SQLLDR_FAIL
$SQLLDR_FTL
) ],
);
# package imports
# ------------------------------------------------------------------------------
use English qw( -no_match_vars );
use Getopt::Long;
use Config::IniFiles;
use Pod::WikiText;
use IO::File;
use IO::Handle;
use IO::LockedFile;
use Fcntl qw(:flock);
use File::Copy;
use File::Bidirectional;
use File::Basename;
use MIME::Lite;
use Date::Format;
use DBI;
#|++ ## flush print buffer on write
# version
# ------------------------------------------------------------------------------
our $VERSION = "0.12";
# const exports
# ------------------------------------------------------------------------------
our $SQLLDR_SUCC = 0;
our $SQLLDR_WARN = 2;
our $SQLLDR_FAIL = 1;
our $SQLLDR_FTL = 3;
# state variables
# ------------------------------------------------------------------------------
my $path_bin_dir = '';
my $path_lib_dir = '';
my $path_log_dir = '';
my $path_load_dir = '';
my $path_extr_dir = '';
my $path_prev_dir = '';
my $path_scripts_dir = '';
my $mail_server = '';
my $mail_from = '';
my $mail_emailto = '';
my $mail_pagerto = '';
my $mail_email_levels = '';
my $mail_pager_levels = '';
my $log_file = '';
my $log_filefull = '';
my $log_logging_levels = '';
my $log_console_levels = '';
my $dataenvr = '';
my $log_gdg = 0;
my $log_prefix = '';
my $log_radix = 2;
my $errorlevel = 0;
my $util_move = 1;
# command line variables
# ------------------------------------------------------------------------------
my $opt_run = 0;
my $opt_run_background = 0;
my $opt_run_scheduled = '';
my $opt_run_restart = '';
my $opt_connection = '';
my $opt_run_de = '';
my $opt_commandline_ext = '';
my $opt_verbose = 0;
my $opt_very_verbose = 0;
my $opt_no_greeting = 0;
my $opt_test_dbcon = '';
my $opt_log_file = '';
my $opt_logging_levels = '';
my $opt_console_levels = '';
my $opt_log_gdg = 0;
my $opt_log_prefix = '';
my $opt_log_radix = 0;
my $opt_notify_email_oncomp = 0;
my $opt_notify_pager_oncomp = 0;
my $opt_notify_email_tolist = '';
my $opt_notify_pager_tolist = '';
my $opt_notify_email_levels = '';
my $opt_notify_pager_levels = '';
my $opt_disp_params = 0;
my $opt_disp_sql = 0;
my $opt_disp_doc = 0;
my $opt_disp_sysdoc = 0;
my $opt_disp_logprev = 0;
my $opt_disp_logarch = 0;
my $opt_disp_jobs = 0;
my $opt_disp_active_jobs = 0;
my $opt_disp_exec = 0;
my $opt_send_email = '';
my $opt_send_pager = '';
my $opt_util_move = 0;
my $opt_help = 0;
my $opt_help_args = 0;
my $opt_commandline = join ' ', @ARGV;
# module variables
# ------------------------------------------------------------------------------
use constant QUOTE => q{"};
use constant SPACE => q{ };
my $RC_FATAL = 32;
my $RC_ERROR = 16;
my $RC_WARN = 8;
my %MONTHS = (
Jan => 0, Feb => 1, Mar => 2, Apr => 3, May => 4, Jun => 5,
Jul => 6, Aug => 7, Sep => 8, Oct => 9, Nov => 10, Dec=> 11,
);
my $jobname = ''; # name used to identify job script
my $pid = 0; # os process id number
my %pidlib = (); # hash of info about background jobs
my $pidcnt = 0; # count of child pids
my $maxrc = 0; # max return code for foreground jobs
my $osuser = ''; # os username
my $commandline_ext = ''; # extended command line
my @plugins = (); # loaded plugin information
my %timers = (); # hash of timers
my %function_params = (); # hash of stored function params
my $wt_seconds = 0; # wait seconds
my $wt_start = time; # init wait start time
my %maxval = (); # hash of max values
my $t_num = 0; # test script
my $t_ok = 0; # test script
my $t_notok = 0; # test script
my $th_num = 0; # test harness
my $th_error = 0; # test harness
my $sys_dbms_output = 0; # has dbms_output been enabled
my $sys_log_open = 0; # is log file open
my $sys_stderr_redirected = 0; # STDERR has been redirected to /dev/null
my $sys_jobconf_override = 0; # using override job conf file
my $sys_jobconf_file = ''; # override jobconf filename
my $path_plugin_dir = ''; # path to plugin directory
my $path_conf_dir = ''; # path to conf file directory
my %sqlloader_results = (); # hash of SQL*Loader results
my %log_level_opts = (); # hash of logging options
my (%conf_data, %conf_log, %conf_mail, %conf_query, %conf_job, %conf_util);
my (%conf_system, %conf_de, %conf_rcontrols);
my (@databases, @dat_envrs, @job_acros);
my (%dbname, %dbdefenvr, %dbinst, %dbconn, %dbhandles);
my $script_file = $PROGRAM_NAME;
my $script_filefull = $script_file;
my $log_ext = '.log';
my $dbitrace_base = 'dbitrace';
my $dbitrace_file = $dbitrace_base . $log_ext;
my $dbitrace_filefull = '';
$script_file =~ s{^/.*/}{};
$path_conf_dir = $ENV{JCLCONF} || '';
if ( ! defined $path_conf_dir ) {
sys_die( 'Environment variable JCLCONF not set', 0 );
}
if ( $path_conf_dir =~ m/(.*)\/$/ ) { $path_conf_dir = $1; }
my %db_func_params = (
db_insert_from_file => {
TrimLead => 'no',
TrimFieldLead => 'no',
TrimFieldTrail => 'no',
CommentChar => '#',
SkipComments => 'no',
SkipLastField => 'no',
UseRegex => 'no',
},
db_insert_from_conf => {
TrimLead => 'no',
TrimFieldLead => 'no',
TrimFieldTrail => 'no',
CommentChar => '#',
SkipComments => 'no',
SkipLastField => 'no',
UseRegex => 'no',
},
db_sqlloader => {
DatFilePath => '',
DbEnvr => '',
NetService => '',
},
);
# public methods
# ------------------------------------------------------------------------------
sub sys_init {
my ($jn, @cl) = @_;
$jobname = $jn;
foreach my $opt ( @cl ) {
push @ARGV, $opt; # add additional command line option
}
unless ( $jobname ) {
sys_die( 'Please specify jobname when initializing', 0 );
}
_sys_init_vars();
$log_file = $jobname . $log_ext;
$log_filefull = $path_log_dir.$log_file;
push @ARGV, '-r' if $jobname eq "JCL"; # for convenience
$sys_jobconf_file = _sys_check_de_override( $jobname );
$sys_jobconf_file .= ".conf";
_sys_read_conf( $sys_jobconf_file ); # tie %conf_job to job's conf file
_sys_read_job(); # read job specific settings from %conf_job
GetOptions( "r" => \$opt_run,
"rb" => \$opt_run_background,
"rs=s" => \$opt_run_scheduled,
"rr=s" => \$opt_run_restart,
"rde=s" => \$opt_run_de,
"x=s" => \$opt_commandline_ext,
"c=s" => \$opt_connection,
"v" => \$opt_verbose,
"vv" => \$opt_very_verbose,
"ng" => \$opt_no_greeting,
"tc=s" => \$opt_test_dbcon,
"lf=s" => \$opt_log_file,
"lg=i" => \$opt_log_gdg,
"lp=s" => \$opt_log_prefix,
"lr=i" => \$opt_log_radix,
"ll=s" => \$opt_logging_levels,
"cl=s" => \$opt_console_levels,
"ne" => \$opt_notify_email_oncomp,
"np" => \$opt_notify_pager_oncomp,
"et=s" => \$opt_notify_email_tolist,
"el=s" => \$opt_notify_email_levels,
"pt=s" => \$opt_notify_pager_tolist,
"pl=s" => \$opt_notify_pager_levels,
"dp" => \$opt_disp_params,
"dq" => \$opt_disp_sql,
"dd" => \$opt_disp_doc,
"dl" => \$opt_disp_logprev,
"da" => \$opt_disp_logarch,
"dj" => \$opt_disp_jobs,
"dja" => \$opt_disp_active_jobs,
"se=s" => \$opt_send_email,
"sp=s" => \$opt_send_pager,
"um" => \$opt_util_move,
"h" => \$opt_help,
"ha" => \$opt_help_args,
) || _sys_help(0);
if ( $opt_connection ) {
foreach my $connectdef ( split m/,/, $opt_connection ) {
my ($db, $inst) = split m/:/, $connectdef;
_check_array_val( $db, \@databases )
|| sys_die( "Invalid database: [$db]", 0 );
_check_array_val( $inst, [split m/,/, $dbinst{$db}] )
|| sys_die( "Invalid database instance: [$db.$inst]", 0 );
## update default connection data
$dbdefenvr{$db} = $inst;
}
}
# create dbitrace file if not found
if ( ! -e $dbitrace_filefull ) {
open my $fh, ">", $dbitrace_filefull
|| sys_die( 'Unable to open dbitrace file', 0 );
close $fh;
}
if ( $opt_help ) {
_sys_help( 1 ); }
if ( $opt_help_args ) {
_sys_help( 2 ); }
if ( $opt_run_background ) {
_sys_run_background(); }
if ( $opt_run_scheduled ) {
_sys_run_scheduled(); }
if ( $opt_run_de ) {
_sys_run_de( $opt_run_de ); }
if ( $opt_run_restart ) {
_sys_run_restart(); }
if ( $opt_test_dbcon ) {
_sys_test_dbcon( $opt_test_dbcon); }
if ( $opt_commandline_ext ) {
$commandline_ext = $opt_commandline_ext; }
if ( $opt_logging_levels ) {
$log_logging_levels = _sys_check_severity_levels( $opt_logging_levels ); }
if ( $opt_console_levels ) {
$log_console_levels = _sys_check_severity_levels( $opt_console_levels ); }
if ( $opt_log_gdg ) {
$log_gdg = _sys_check_log_gdg( $opt_log_gdg ); }
if ( $opt_log_prefix ) {
$log_prefix = $opt_log_prefix; }
if ( $opt_log_radix ) {
$log_radix = _sys_check_log_radix( $opt_log_radix ); }
if ( $opt_notify_email_tolist ) {
$mail_emailto = $opt_notify_email_tolist; }
if ( $opt_notify_pager_tolist ) {
$mail_pagerto = $opt_notify_pager_tolist; }
if ( $opt_notify_email_levels ) {
$mail_email_levels = _sys_check_severity_levels( $opt_notify_email_levels ); }
if ( $opt_notify_pager_levels ) {
$mail_pager_levels = _sys_check_severity_levels( $opt_notify_pager_levels ); }
if ( $opt_disp_logprev ) {
_sys_disp_logprev(); }
if ( $opt_disp_logarch ) {
_sys_disp_logarch(); }
if ( $opt_disp_exec ) {
_sys_disp_exec(); }
if ( $opt_disp_sql ) {
_sys_disp_sql(); }
if ( $opt_disp_params ) {
_sys_disp_params(); }
if ( $opt_disp_doc ) {
_sys_disp_doc(); }
if ( $opt_disp_jobs ) {
_sys_disp_jobs(); }
if ( $opt_disp_active_jobs ) {
_sys_disp_active_jobs( 0 ); }
if ( $opt_send_email ) {
_sys_send_email_message($opt_send_email); }
if ( $opt_send_pager ) {
_sys_send_pager_message($opt_send_pager); }
if ( $opt_util_move ) {
$util_move = 0; }
# must have a Run option to continue
if ( ! $opt_run ) {
_sys_help(1);
}
$log_file = $log_prefix . $jobname . $log_ext; # default
if ( $osuser ) { # custom
$log_file = $log_prefix . $jobname . '_' . $osuser . $log_ext;
}
$log_filefull = $path_log_dir . $log_file;
if ( $opt_log_file ) { # override
$log_file = $opt_log_file;
$log_filefull = $path_log_dir . $log_file;
}
_log_init_log_file(); # log rotation handler
# validate script name using configured acros
my ($base, $path, $type) = fileparse( $script_file );
if ( $base =~ m/^([a-z]+_)/x ) { ## acro + underscore
$base = $1;
}
_check_array_val($base, \@job_acros) || sys_die( "Not a valid job acro", 0 );
_sys_init_source_validation();
sys_timer( 'start', '__default_timer' );
log_info( "Start: $jobname" ) unless $opt_no_greeting;
if ( $opt_very_verbose ) { $opt_verbose = 1; }
if ( $opt_verbose ) {
log_info( 'Running in verbose mode' );
log_info( "Process: $pid" );
log_info( "Options: $opt_commandline" );
}
if ( $sys_jobconf_override ) {
log_info( "Jobconf override: $sys_jobconf_file" );
}
_sys_job_init();
return 0;
}
sub sys_init_setuser {
my ($jn, @cl) = @_;
$osuser = getlogin || 'unknown';
sys_init( $jn, @cl );
return 0;
}
sub sys_end {
_sys_job_end();
if ( $opt_no_greeting ) { return 0; }
sys_timer( 'stop', '__default_timer' );
log_info( "Errorlevel: $errorlevel" );
log_info( "Elapsed time: " . sys_timer( 'elapsed', '__default_timer' ) );
log_info( "End: $jobname" ) unless $opt_no_greeting;
return 0;
}
sub sys_load_library {
my $conf_filename = shift;
## load a conf file replacing the contents of sys_common.conf
tie %conf_query, 'Config::IniFiles', ( -file => $path_conf_dir.'/'.$conf_filename )
or sys_die( "Unable to load conf file $conf_filename", 0 );
return 0;
}
sub sys_init_plugin {
my ($plugin_file, $package_name) = @_;
my $plugin_filefull = $path_plugin_dir.$plugin_file.'.pm';
unless ( -f $plugin_filefull ) { sys_die( "Plugin not found: $plugin_file", 0 ); }
require $plugin_filefull;
push @plugins, join '~', ($package_name, $plugin_file, $plugin_filefull);
$package_name->start($path_conf_dir, $path_plugin_dir, $dataenvr);
return $package_name->can('plugin_main'); ## deep magic
}
sub sys_ctime2str {
my $format = shift;
return time2str($format, time);
}
sub sys_die {
my ($message, $notify) = @_;
$notify = 0 unless defined $notify;
$errorlevel = $RC_FATAL;
_log_write_to_screen( 'FATAL', $notify, $message );
if ( $sys_log_open ) {
_log_write_to_log( 'FATAL', $notify, $message );
}
## save a call if possible
if ( $notify ) { _log_send_notifications( 'FATAL', $notify, $message ); }
_sys_job_end();
exit $errorlevel;
}
sub sys_warn {
my ($message, $notify) = @_;
$notify = 1 unless defined $notify;
$errorlevel = $RC_WARN;
## force write to screen
_log_write_to_screen( 'WARN', 1, $message );
## force write to log if log is open
if ( $sys_log_open ) {
_log_write_to_log( 'WARN', 1, $message );
}
## force notifications if notification requested
if ( $notify ) { _log_send_notifications( 'WARN', 1, $message ); }
return $errorlevel;
}
sub sys_info {
my ($message, $extmsg, $notify, $nolog) = @_;
$notify = 1 unless defined $notify;
$nolog = 0 unless defined $nolog;
## get destination email address from job conf
my $emailto = sys_get_item( 'sys_info_emailto' );
my $mail_emailto_save = $mail_emailto;
$mail_emailto = $emailto;
log_info( $message, $extmsg, $nolog );
_log_send_notifications( 'INFO', 1, $message ) if $notify;
$mail_emailto = $mail_emailto_save;
return 0;
}
sub sys_disp_active_jobs {
_sys_disp_active_jobs( 1 );
return 0;
}
sub sys_run_job {
my ($jobname, $job_maxrc, @params) = @_;
my @args = ($jobname, @params);
system(@args);
my $childrc = $CHILD_ERROR >> 8;
if ( $childrc > $job_maxrc ) {
sys_die( "Process failed with return code $childrc" );
}
if ( $job_maxrc > $maxrc ) { $maxrc = $job_maxrc; }
return $childrc;
}
sub sys_run_job_background {
my ($jobname, $maxrc, @params) = @_;
$maxrc = 0 unless $maxrc;
my $pid = _sys_forkexec( $jobname, @params );
$pidlib{$pid} = { jobname => $jobname,
maxrc => $maxrc,
retcd => 0
};
$pidcnt++;
return $pid;
}
sub sys_run_job_wait {
return 0 if $pidcnt < 1;
while (1) {
my $pid = _sys_reap_child();
$pidcnt--;
my $childrc = $pidlib{$pid}{retcd};
my $msg = "Complete $pidlib{$pid}{jobname}. Return code: $childrc.";
if ( $childrc > $pidlib{$pid}{maxrc} ) {
## log_warn sets errorlevel
log_warn( "$msg Max allowed: $pidlib{$pid}{maxrc}." );
} else {
log_info( $msg );
}
last if $pidcnt < 1;
}
return $pidcnt;
}
sub sys_run_job_maxrc {
## return the max of either the current background max return code or the
## current foreground max return code
my $tmprc = 0;
foreach my $pid ( keys %pidlib ) {
if ( $pidlib{$pid}{retcd} > $tmprc ) { $tmprc = $pidlib{$pid}{retcd}; }
}
( $tmprc >= $maxrc ) ? return $tmprc : return $maxrc;
}
sub sys_run_job_reset {
$pidcnt = 0; ## reset background jobs count
%pidlib = (); ## reset background jobs info hash
$maxrc = 0; ## reset foreground jobs max return code
return 0;
}
sub sys_get_path_bin_dir {
return $path_bin_dir;
}
sub sys_get_path_lib_dir {
return $path_lib_dir;
}
sub sys_get_path_log_dir {
return $path_log_dir;
}
sub sys_get_path_load_dir {
return $path_load_dir;
}
sub sys_get_path_extr_dir {
return $path_extr_dir;
}
sub sys_get_path_prev_dir {
return $path_prev_dir;
}
sub sys_get_path_scripts_dir {
return $path_scripts_dir;
}
sub sys_get_path_plugin_dir {
return $path_plugin_dir;
}
sub sys_get_mail_server {
return $mail_server;
}
sub sys_get_mail_from {
return $mail_from;
}
sub sys_get_mail_emailto {
return $mail_emailto;
}
sub sys_get_mail_pagerto {
return $mail_pagerto;
}
sub sys_get_mail_email_levels {
return $mail_email_levels;
}
sub sys_get_mail_pager_levels {
return $mail_pager_levels;
}
sub sys_get_log_file {
return $log_file;
}
sub sys_get_log_filefull {
return $log_filefull;
}
sub sys_get_log_logging_levels {
return $log_logging_levels;
}
sub sys_get_log_console_levels {
return $log_console_levels;
}
sub sys_get_log_gdg {
return $log_gdg;
}
sub sys_get_dataenvr {
return $dataenvr;
}
sub sys_get_errorlevel {
return $errorlevel;
}
sub sys_get_dbdescr {
my $dbacro = shift;
my $dbdescr = 'Database: acronym not found';
foreach my $acro ( @databases ) {
if ( $acro eq $dbacro ) {
$dbdescr = 'Database Connection: ' . $dbname{$dbacro} . ' (' .
$dbacro . '/' . $dbdefenvr{$dbacro} . ')';
}
}
return $dbdescr;
}
sub sys_get_dbinst {
my $dbacro = shift;
my $dbdescr = 'Database: instance not found';
foreach my $acro ( @databases ) {
if ( $acro eq $dbacro ) {
$dbdescr = $dbacro . '/' . $dbdefenvr{$dbacro};
}
}
return uc($dbdescr);
}
sub sys_get_conf_dir {
return $path_conf_dir . '/';
}
sub sys_get_sql {
my ($sqlname, $altsection) = @_;
my $section = $altsection || 'sql';
if ( ! $conf_job{$section}{$sqlname} ) {
$sqlname = 'sql:'.$sqlname;
if ( ! $conf_job{$section}{$sqlname} ) {
sys_die( "The job conf file does not contain a query named [$sqlname]", 0 );
}
}
return $conf_job{$section}{$sqlname};
}
sub sys_get_item {
my ($item, $altsection) = @_;
my $section = $altsection || 'job';
my $value = $conf_job{$section}{$item};
if ( ! defined $value ) {
sys_die( "Job conf missing entry [$item] in section [$section]", 0 );
}
if ( $value eq '0' ) {
return $conf_job{$section}{$item};
}
return $value;
}
sub sys_get_hash {
my ($section, $entry, $delim) = @_;
$delim = ':' unless $delim;
my ($pseudo, %hash);
if ( $conf_job{$section}{$entry} ) {
$pseudo = $conf_job{$section}{$entry};
} else {
sys_die( "No job conf entry found for $entry in section $section" );
}
## construct a real hash from the pseudo hash
foreach my $item ( split "\n", $pseudo ) {
my ($key, $value) = split m/$delim/, $item;
$hash{$key} = $value;
}
return \%hash; ## ref to hash
}
sub sys_get_array {
my ($section, $entry, $delim) = @_;
$delim = ':' unless $delim;
my ($pseudo, @array);
if ( $conf_job{$section}{$entry} ) {
$pseudo = $conf_job{$section}{$entry};
} else {
sys_die( "No job conf entry found for $entry in section $section" );
}
## construct a real array from the pseudo array
foreach my $item ( split "\n", $pseudo ) {
push @array, $item;
}
return \@array; ## ref to an array
}
sub sys_get_common_sql {
my ($sqlname, $altsection) = @_;
my $section = $altsection || 'sql';
if ( ! $conf_query{$section}{$sqlname} ) {
$sqlname = 'sql:'.$sqlname;
if ( ! $conf_query{$section}{$sqlname} ) {
sys_die( 'Common sql conf missing query by that name', 0 );
}
}
return $conf_query{$section}{$sqlname};
}
sub sys_get_run_control {
my ($jobname, $section, $default) = @_;
my $rcontrol = $default || 0;
if ( ! $conf_rcontrols{$section}{$jobname} ) {
return $rcontrol;
}
return $conf_rcontrols{$section}{$jobname};
}
sub sys_get_email_levels {
return $mail_email_levels;
}
sub sys_get_pager_levels {
return $mail_pager_levels;
}
sub sys_get_logging_levels {
return $log_logging_levels;
}
sub sys_get_console_levels {
return $log_console_levels;
}
sub sys_get_commandline {
return join ' ', @ARGV;
}
sub sys_get_commandline_opt {
my $target_opt = shift;
foreach my $option ( @ARGV ) {
my ($opt,$val) = split m/=/, $option;
$opt =~ s/^-\s*//x;
$opt =~ s/\s+$//x;
if ( $opt =~ m/^$target_opt$/ix ) {
return 1;
}
}
return 0;
}
sub sys_get_commandline_val {
my ($target_opt,$default_value) = @_;
## handle:
## >script.pl -r -- -batchsize=10
foreach my $option ( @ARGV ) {
$option =~ s/\s+=/=/x;
$option =~ s/=\s+/=/x;
my ($opt,$val) = split m/=/, $option;
$opt =~ s/^-\s*//x;
$opt =~ s/\s+$//x;
if ( $opt =~ m/^$target_opt$/ix ) {
#$val =~ s/^\s*//;
#$val =~ s/\s*$//;
return $val;
}
}
return $default_value;
}
sub sys_get_script_file {
return $script_file;
}
sub sys_get_util_move {
return $util_move;
}
sub sys_get_user {
return getlogin || 'unknown';
}
sub sys_get_maxval {
my $key = shift;
return $maxval{$key} || 0;
}
sub sys_set_restart {
my $restart_option = shift;
if ( $restart_option !~ m/^\d+/x ) {
sys_die( 'Restart option is not numeric', 0 );
return 1;
}
my $rtconf = $path_conf_dir.'/'.$jobname.'.running';
my $conf = new Config::IniFiles( -file => $rtconf );
unless ( defined $conf ) { sys_die( "Error opening runtime jobconf file", 0 ); }
$conf->setval( 'restart', 'restart', $restart_option );
$conf->RewriteConfig;
return 0;
}
sub sys_set_verbose {
$opt_verbose = 1;
return 0;
}
sub sys_set_errorlevel {
my $errlvl = shift;
if ( $errlvl !~ /^\d+$/ ) {
sys_die( "Invalid value passed to sys_set_errorlevel()" );
}
my $save_errlvl = $errorlevel;
$errorlevel = $errlvl;
return $save_errlvl;
}
sub sys_set_warn {
$errorlevel = $RC_WARN;
return $RC_WARN;
}
sub sys_set_die {
$errorlevel = $RC_FATAL;
return $RC_FATAL;
}
sub sys_set_email_levels {
my $email_levels = shift || "FATAL";
$mail_email_levels = _sys_check_severity_levels( $email_levels );
return $mail_email_levels;
}
sub sys_set_pager_levels {
my $pager_levels = shift || "FATAL";
$mail_pager_levels = _sys_check_severity_levels( $pager_levels );
return $mail_pager_levels;
}
sub sys_set_mail_emailto {
my $new_emailto = shift;
my $old_emailto = $mail_emailto;
$mail_emailto = $new_emailto;
return $old_emailto;
}
sub sys_set_logging_levels {
my $logging_levels = shift || "FATAL,ERROR,WARN,INFO";
$log_logging_levels = _sys_check_severity_levels( $logging_levels );
return $log_logging_levels;
}
sub sys_set_console_levels {
my $console_levels = shift || "FATAL,ERROR,WARN,INFO";
$log_console_levels = _sys_check_severity_levels( $console_levels );
return $log_console_levels;
}
sub sys_set_script_file {
my $file = shift || $script_file;
$script_file = $file;
return $script_file;
}
sub sys_set_conf_file {
my $jobconf = shift || '';
if ( $jobconf ) {
## change jobconf file and read
$sys_jobconf_file = $jobconf . '.conf';
_sys_read_conf( $sys_jobconf_file ); ## tie %conf_job to job conf file
_sys_read_job(); ## read job specific settings from %conf_job
} else {
## reset jobconf file to default and reread
$sys_jobconf_file = _sys_check_de_override( $jobname . '.conf' );
_sys_read_conf( $sys_jobconf_file ); ## tie %conf_job to job conf file
_sys_read_job(); ## read job specific settings from %conf_job
}
return 0;
}
sub sys_set_path_log_dir {
my $path = shift || $path_log_dir;
$path_log_dir = $path;
return $path_log_dir;
}
sub sys_set_path_plugin_dir {
my $path = shift || $path_plugin_dir;
$path_plugin_dir = $path;
return $path_plugin_dir;
}
sub sys_set_maxval {
my ($key, $val) = @_;
if ( $maxval{$key} ) {
if ( $val > $maxval{$key} ) {
$maxval{$key} = $val;
}
return $val;
}
$maxval{$key} = $val;
return $val;
}
sub sys_check_dataenvr {
my $data_envrs = shift;
my @check_envrs;
if ( ref $data_envrs eq 'ARRAY' ) {
@check_envrs = map { $_ } @{$data_envrs};
} else {
push @check_envrs, $data_envrs; ## single entry
}
## is current data environment in the list of acceptable environments
if ( grep { $_ eq $dataenvr } @check_envrs ) {
return 1;
}
return 0;
}
sub sys_disp_doc {
return _sys_disp_doc();
}
sub sys_timer {
my ($opt, $timer_name) = @_;
$timer_name = 't1' unless $timer_name;
if ( $opt =~ m/start/ix ) {
$timers{$timer_name.'_start'} = time;
return $timers{$timer_name.'_start'};
}
if ( $opt =~ m/stop/ix ) {
$timers{$timer_name.'_stop'} = time;
return $timers{$timer_name.'_stop'};
}
if ( $opt =~ m/elapsed/ix ) {
my $estart = $timers{$timer_name.'_start'};
my $estop = $timers{$timer_name.'_stop'};
my $eelapsed = $estop - $estart;
my $ehours = int $eelapsed / 3600;
my $emins = int $eelapsed / 60 % 60;
my $esecs = int $eelapsed % 60;
return sprintf "%02d:%02d:%02d", $ehours, $emins, $esecs;
}
if ( $opt =~ /elapsed_seconds/i ) {
my $sstart = $timers{$timer_name.'_start'};
my $sstop = $timers{$timer_name.'_stop'};
my $selapsed = $sstop - $sstart;
return $selapsed;
}
return 'TIMER ERROR';
}
sub sys_wait {
my ($action, $minutes) = @_;
if ( $action =~ /^init$/i ) {
$wt_start = time;
$wt_seconds = 0;
return 0 unless $minutes =~ /^\d+$/;
$wt_seconds = $minutes * 60;
}
if ( $action =~ /^wait$/i ) {
while ( 1 ) {
my $currtime = time;
my $elapsedt = $currtime - $wt_start;
log_info( "Waiting $wt_seconds, Elapsed: $elapsedt" );
if ( ($currtime - $wt_start) < $wt_seconds ) {
sleep 10;
} else {
last;
}
}
}
return 0;
}
sub log_fatal {
my ($message, $extmsg) = @_;
$errorlevel = $RC_FATAL;
_log_write_to_log( 'FATAL', 0, $message, $extmsg);
_log_write_to_screen( 'FATAL', 0, $message, $extmsg);
return $errorlevel;
}
sub log_error {
my ($message, $extmsg) = @_;
$errorlevel = $RC_ERROR;
_log_write_to_log( 'ERROR', 0, $message, $extmsg);
_log_write_to_screen( 'ERROR', 0, $message, $extmsg);
return $errorlevel;
}
sub log_warn {
my ($message, $extmsg) = @_;
$errorlevel = $RC_WARN;
_log_write_to_log( 'WARN', 0, $message, $extmsg);
_log_write_to_screen( 'WARN', 0, $message, $extmsg);
return $errorlevel;
}
sub log_info {
my ($message, $extmsg, $nolog) = @_;
$nolog = 0 unless $nolog;
return 0 if $nolog;
_log_write_to_log( 'INFO', 0, $message, $extmsg);
_log_write_to_screen( 'INFO', 0, $message, $extmsg);
return 0;
}
sub log_debug {
my ($message, $extmsg) = @_;
_log_write_to_log( 'DEBUG', 0, $message, $extmsg);
_log_write_to_screen( 'DEBUG', 0, $message, $extmsg);
return 0;
}
sub log_close {
my ($message, $extmsg) = @_;
_log_write_to_log( 'INFO', 0, $message, $extmsg);
_log_write_to_screen( 'INFO', 0, $message, $extmsg);
$sys_log_open = 0;
return 0;
}
sub log_write_screen {
my $message = shift;
_log_write_to_screen( 'INFO', 1, $message);
return 0;
}
sub log_write_log {
my $message = shift;
_log_write_to_log( 'INFO', 1, $message);
return 0;
}
sub db_init {
my ($id, %params) = @_;
if ( ! defined $db_func_params{$id} ) {
sys_die( "Param $id to db_init is invalid")
}
foreach my $key ( keys %params ) {
if ( ! defined $db_func_params{$id}{$key} ) {
sys_die( "Param $key to db_init is invalid" );
}
$db_func_params{$id}{$key} = $params{$key};
}
return 0;
}
sub db_connect {
my ($vdn, %connect_params) = @_;
my ($starttime, $dbh, $instance);
## time increment is secs, action is either 'run' or 'fail'
my $dependent_jobname = $connect_params{dependent_jobname} || '';
my $wait_duration = $connect_params{wait_duration} || 60;
my $wait_max_secs = $connect_params{wait_max_secs} || 60*60;
my $wait_action = $connect_params{wait_action} || 'fail';
my $retry_duration = $connect_params{retry_duration} || 0;
my $retry_max_secs = $connect_params{retry_max_secs} || 0;
if ( $vdn =~ m/:/x ) { ## vdn contains instance definiton
my ($db, $inst) = split m/:/, $vdn;
_check_array_val( $db, \@databases )
|| sys_die( "Invalid database: [$db]", 0 );
_check_array_val( $inst, [split m/,/, $dbinst{$db}] )
|| sys_die( "Invalid database instance: [$db.$inst]", 0 );
$dbdefenvr{$db} = $inst; ## update default connection data
$vdn = $db; ## vdn gets true vdn
}
## check for dependent job
_db_connect_check_dependent(
$dependent_jobname, $wait_duration, $wait_max_secs, $wait_action
);
## get database parameters
my ($db, $un, $pw) = _db_vdn('connect', $vdn);
DBI->trace( 1, $dbitrace_filefull );
open STDERR, '>', '/dev/null' unless $opt_very_verbose;
## connect with retry
$dbh = _db_connect_retry(
$db, $un, $pw, $retry_duration, $retry_max_secs
);
## connection established
$dbhandles{$vdn}{'dbh'} = $dbh; ## store handle for cleanup on exit
db_nil( $vdn );
return 0;
}
sub db_nil {
my $vdn = shift;
my ($dbh, $sth) = _db_vdn( 'nil', $vdn);
return 0;
}
sub db_disconnect {
my $vdn = shift;
my ($dbh, $sth) = _db_vdn( 'disconnect', $vdn);
if ( $dbh ) {
$dbh->disconnect;
if ( DBI->errstr ) {
log_warn( DBI->errstr );
return 1;
}
}
$dbhandles{$vdn}{'dbh'} = 0;
return 0;
}
sub db_finish {
my $vdn = shift;
my ($dbh, $sth) = _db_vdn( 'finish', $vdn);
if ( $sth ) {
$sth->finish;
if ( DBI->errstr ) {
log_warn( DBI->errstr );
return 1;
}
}
$dbhandles{$vdn}{'sth'} = 0;
return 0;
}
sub db_prepare {
my ($vdn, $sql, $longrlen) = @_;
$longrlen = 0 unless $longrlen;
my $sth_name = 'sth_default'; ## default statement handle name
if ( $vdn =~ m/\./x ) {
($vdn, $sth_name) = split m/\./x, $vdn;
if ( $sth_name eq 'sth_default' ) {
sys_die( 'Invalid statement handle name', 0 );
}
}
my ($dbh, $sth) = _db_vdn('prepare', $vdn);
if ( $longrlen > 0 ) { $dbh->{LongReadLen} = $longrlen; }
$sth = $dbh->prepare( $sql )
or sys_die( $dbh->errstr );
## store statement handle for this vdn
$dbhandles{$vdn}{$sth_name} = $sth;
return 0;
}
sub db_truncate {
my ($vdn, $table_name) = @_;
my ($dbh, $sth) = _db_vdn('truncate', $vdn);
my $sql = "truncate table $table_name";
$dbh->do( $sql )
or sys_die( DBI->errstr );
return 0;
}
sub db_execute {
my ($vdn, @params) = @_;
my ($dbh, $sth) = _db_vdn('execute', $vdn);
$sth->execute( @params )
or sys_die( $sth->errstr );
return 0;
}
sub db_get_sth {
my $vdn = shift;
my $sth_name = 'sth_default'; ## default statement handle name
if ( $vdn =~ m/\./x ) {
($vdn, $sth_name) = split m/\./x, $vdn;
}
return $dbhandles{$vdn}{$sth_name};
}
sub db_get_defenvr {
my $vdn = shift;
if ( $dbdefenvr{$vdn} ) {
return $dbdefenvr{$vdn};
}
return '';
}
sub db_bindcols {
#
# interface:
# interface to sth->bind_columns()
#
# accepts:
# 1st position
# a raw statement handle
# a vdn which is used to obtain a default statment handle (one per vdn)
# a vdn, named statement handle pair in the form vdn||nsth
# remaining
# any number of references to scalars
#
# returns:
# 0 = success
# errors handled internally
#
my ($vdn,@colrefs) = @_;
my $sth;
if ( ref $vdn ) {
$sth = $vdn; ## received a raw sth
} else {
my $sth_name = 'sth_default'; ## default statement handle name
if ( $vdn =~ m/\./x ) { ## dot notation vdn.sthn
($vdn, $sth_name) = split m/\./x, $vdn;
}
$sth = $dbhandles{$vdn}{$sth_name};
}
foreach my $colref ( @colrefs ) {
if ( ! ref $colref ) { sys_die( "Received bad ref in db_bindcols" ); }
}
$sth->bind_columns( @colrefs );
return 0;
}
sub db_pef {
my ($vdn, $sqlname, @params) = @_;
my $sql = sys_get_sql( $sqlname );
db_prepare( $vdn, $sql );
db_execute( $vdn, @params );
my $row = db_fetchrow( $vdn );
return @{$row}[0];
}
sub db_pef_list {
my ($vdn, $sqlname, @params) = @_;
my @rsalist;
my $sql = sys_get_sql( $sqlname );
db_prepare( $vdn, $sql );
db_execute( $vdn, @params );
while ( my $row = db_fetchrow( $vdn ) ) {
push @rsalist, @{$row}[0];
}
return \@rsalist; ## return result set asa list
}
sub db_fetchrow {
#
# interface:
# interface to sth->fetchrow_arrayref()
#
# accepts:
# a raw statement handle
# a vdn which is used to obtain a default statment handle (one per vdn)
# a vdn, named statement handle pair in the form vdn||nsth
#
# note:
# If you are going to make lots of calls to db_fetchrow for the
# same execute cycle, you will get better performance using a raw
# statement handle over a statement handle name
#
# returns:
# reference to an array
#
my $vdn = shift;
my $sth;
if ( ref $vdn ) {
$sth = $vdn; ## received a raw sth
} else {
my $sth_name = 'sth_default'; ## default statement handle name
if ( $vdn =~ m/\./x ) {
($vdn, $sth_name) = split m/\./x, $vdn;
}
$sth = $dbhandles{$vdn}{$sth_name};
}
return $sth->fetchrow_arrayref();
}
sub db_commit {
my ($vdn) = shift;
my ($dbh, $sth) = _db_vdn('commit', $vdn);
$dbh->commit;
if ( DBI->errstr ) {
sys_die( DBI->errstr );
return 1; ## test harness returns from sys_die
}
return 0;
}
sub db_rollback {
my ($vdn) = shift;
my ($dbh, $sth) = _db_vdn('rollback', $vdn);
$dbh->rollback;
if ( DBI->errstr ) {
sys_die( DBI->errstr );
return 1; ## test harness returns from sys_die
}
return 0;
}
sub db_rowcount_table {
my ($vdn, $table_name) = @_;
my ($dbh, $sth) = _db_vdn('rowcount_table', $vdn);
my $sql = "select count(*) from $table_name";
my $count = $dbh->selectrow_array( $sql );
return $count;
}
sub db_rowcount_query {
my ($vdn, $sql, @params ) = @_;
my ($dbh, $sth) = _db_vdn('rowcount_query', $vdn);
if ( @params ) {
my $tmp_sth = $dbh->prepare( $sql )
or sys_die( $dbh->errstr );
$tmp_sth->execute( @params )
or sys_die( $sth->errstr );
my @row = $tmp_sth->fetchrow_array();
return $row[0];
} else {
my $count = $dbh->selectrow_array( $sql );
return $count;
}
}
sub db_sanity_check {
my ($vdn, $query_name, $notify) = @_;
$notify = 0 unless $notify;
my $warnings = 0;
my $lead = "Sanity check:";
my $okay = " Ok ";
my $outofbounds = " Out Of Bounds ";
my $disabled = " Disabled ";
## get checkpoints
my $checkpoints;
my $conf_entry = sys_get_dataenvr . '_checkpoints';
if ( $conf_job{threshold}{$conf_entry} ) {
$checkpoints = $conf_job{threshold}{$conf_entry};
} else {
log_warn( "No threshold checkpoints found in job conf for: $conf_entry" );
return 1;
}
## prepare range limit query
my $query = sys_get_sql( $query_name );
db_prepare( $vdn, $query );
log_info( "$lead Status [Test] Expected/Actual/Threshold(%)/Threshold(#)" );
## perform checkpoint tests
foreach my $chkpt ( split "\n", $checkpoints ) {
my ($param,$rest) = split m/=/, $chkpt;
my ($exp,$range) = split m/:/, $rest;
$param = _trim($param); ## col to check
$exp = _trim($exp); ## expected value
$range = _trim($range); ## range/tolerance
db_execute( $vdn, $param );
my $row = db_fetchrow( $vdn );
my $act = @{$row}[0]; ## actual value
my $dev = int $exp * ( $range / 100 ); ## deviation as a percent
my $status = "[$param] $exp/$act/$range/$dev ";
if ( $exp == 0 ) { ## checking has been disabled
log_info( $lead . $disabled . $status );
next;
}
if ( $range == 0 ) { ## any positive value for actual is acceptable
if ( $act > 0 ) {
log_info( $lead . $okay . $status );
next;
}
$warnings++;
log_info( $lead . $outofbounds . $status );
next;
}
if ( $act < $exp ) { ## actual is below threshold
if ( $act < $exp - $dev ) {
log_info( $lead . $outofbounds . $status );
$warnings++;
next;
}
}
if ( $act > $exp ) { ## actual is above threshold
if ( $act > $exp + $dev ) {
log_info( $lead . $outofbounds . $status );
$warnings++;
next;
}
}
log_info( $lead . $okay . $status );
}
## send out notifications if there are warnings
if ( $warnings && $notify ) {
_log_send_notifications( "WARN", 1, "Sanity check threshold exceeded" );
}
return 0;
}
sub db_drop_index {
my ($vdn, $index_name) = @_;
my ($dbh, $sth) = _db_vdn('drop_index', $vdn);
my $tmp_sth = $dbh->prepare("drop index $index_name")
or sys_die( DBI->errstr );
$tmp_sth->execute;
if ( DBI->err && DBI->err != 1418 ) { ## ORA-00942: specified index does not exist
sys_die( DBI->errstr );
}
return 0;
}
sub db_drop_table {
my ($vdn, $table_name) = @_;
my ($dbh, $sth) = _db_vdn('drop_table', $vdn);
my $tmp_sth = $dbh->prepare("drop table $table_name" )
or sys_die( DBI->errstr );
$tmp_sth->execute;
if ( DBI->err && DBI->err != 942 ) { ## ORA-00942: specified table does not exist
sys_die( DBI->errstr );
}
$tmp_sth->finish;
return 0;
}
sub db_drop_procedure {
my ($vdn, $procedure_name) = @_;
my ($dbh, $sth) = _db_vdn('drop_procedure', $vdn);
my $tmp_sth = $dbh->prepare("drop procedure $procedure_name")
or sys_die( DBI->errstr );
$tmp_sth->execute;
if ( DBI->err && DBI->err != 4043 ) { ## ORA-04043: object does not exist
sys_die( DBI->errstr );
}
$tmp_sth->finish;
return 0;
}
sub db_drop_function {
my ($vdn, $function_name) = @_;
my ($dbh, $sth) = _db_vdn('drop_function', $vdn);
my $tmp_sth = $dbh->prepare("drop function $function_name")
or sys_die( DBI->errstr );
$tmp_sth->execute;
if ( DBI->err && DBI->err != 4043 ) { ## ORA-04043: object does not exist
sys_die( DBI->errstr );
}
$tmp_sth->finish;
return 0;
}
sub db_drop_package {
my ($vdn, $package_name) = @_;
my ($dbh, $sth) = _db_vdn('drop_package', $vdn);
my $tmp_sth = $dbh->prepare("drop package $package_name")
or sys_die( DBI->errstr );
$tmp_sth->execute;
if ( DBI->err && DBI->err != 4043 ) { ## ORA-04043: object does not exist
sys_die( DBI->errstr );
}
$tmp_sth->finish;
return 0;
}
sub db_rename_index {
my ($vdn, $oldname, $newname) = @_;
my ($dbh, $sth) = _db_vdn('rename_index', $vdn);
my $tmp_sth = $dbh->prepare("alter index $oldname rename to $newname")
or sys_die( DBI->errstr );
$tmp_sth->execute;
if ( DBI->err ) {
sys_die( DBI->errstr );
}
return 0;
}
sub db_rename_table {
my ($vdn, $oldname, $newname) = @_;
my ($dbh, $sth) = _db_vdn('rename_table', $vdn);
my $tmp_sth = $dbh->prepare("alter table $oldname rename to $newname" )
or sys_die( DBI->errstr );
$tmp_sth->execute;
if ( DBI->err ) {
sys_die( DBI->errstr );
}
$tmp_sth->finish;
return 0;
}
sub db_purge_table {
my ($vdn, $table_name) = @_;
my ($dbh, $sth) = _db_vdn('purge_table', $vdn);
my $tmp_sth = $dbh->prepare("purge table $table_name" )
or sys_die( DBI->errstr );
$tmp_sth->execute;
if ( DBI->err && DBI->err != 38307 ) { ## ORA-38307: object not in recycle bin
sys_die( DBI->errstr );
}
$tmp_sth->finish;
return 0;
}
sub db_purge_index {
my ($vdn, $table_name) = @_;
my ($dbh, $sth) = _db_vdn('purge_index', $vdn);
my $tmp_sth = $dbh->prepare("purge index $table_name")
or sys_die( DBI->errstr );
$tmp_sth->execute;
if ( DBI->err && DBI->err != 38307 ) { ## ORA-38307: object not in recycle bin
sys_die( DBI->errstr );
}
return 0;
}
sub db_grant {
my ($vdn, $priv, $objname, $ag) = @_;
my ($dbh, $sth) = _db_vdn('grant', $vdn);
unless ( $priv =~ m/^r$|^u$/x ) {
log_warn( "Privilege to db_grant must be either 'r' or 'u'" );
return 1;
}
my $sql;
if ( $priv eq 'r' ) {
$sql = qq{begin execute immediate 'grant select on $objname to $ag'; end;};
}
if ( $priv eq 'u' ) {
$sql = qq{begin execute immediate 'grant update, insert, delete on $objname to $ag'; end;};
}
my $tmp_sth = $dbh->prepare( $sql )
or sys_die( DBI->errstr );
$tmp_sth->execute
or sys_die( DBI->errstr );
$tmp_sth->finish;
return 0;
}
sub db_update_statistics {
my ($vdn, $table_name) = @_;
my ($dbh, $sth) = _db_vdn('update_statistics', $vdn);
unless ( _db_is_oracle($vdn) ) {
sys_die( 'Not an Oracle database connection in db_update_statistics', 0 );
}
my $sql = "BEGIN dbms_stats.gather_table_stats('','"
. "$table_name',NULL,NULL,FALSE,'FOR ALL COLUMNS SIZE 1'"
. ",NULL,'DEFAULT',TRUE); END;";
my $tmp_sth = $dbh->prepare( $sql );
if ( DBI->errstr ) { sys_die( DBI->errstr ); }
$tmp_sth->execute;
if ( DBI->errstr ) { sys_die( DBI->errstr ); }
$tmp_sth->finish;
return 0;
}
sub db_insert_from_file {
my ($vdn, $file_name, $delim) = @_;
my ($dbh, $sth) = _db_vdn('insert_from_file', $vdn);
my $id = 'db_insert_from_file';
my $TrimLead = _is_yes($db_func_params{$id}{'TrimLead'});
my $TrimFieldLead = _is_yes($db_func_params{$id}{'TrimFieldLead'});
my $TrimFieldTrail = _is_yes($db_func_params{$id}{'TrimFieldTrail'});
my $SkipComments = _is_yes($db_func_params{$id}{'SkipComments'});
my $SkipLastField = _is_yes($db_func_params{$id}{'SkipLastField'});
my $UseRegex = _is_yes($db_func_params{$id}{'UseRegex'});
my $CommentChar = $db_func_params{$id}{'CommentChar'};
my ($count, @row);
open my $fh, "<", $file_name or sys_die( "Error opening $file_name" );
my $regex = "\Q$delim\E"; # escape regex meta chars
if ( $UseRegex ) {
$regex = $delim; # do escaping meta chars
}
while ( <$fh> ) {
my $line = $_;
chomp $line;
if ( $TrimLead ) {
$line = _trim_lead($line);
}
if ( $SkipComments ) {
if ( substr($line,0,1) eq $CommentChar ) { next; }
}
@row = split($regex,$line,-1); # -1 preserves trailing null fields
if ( $SkipLastField ){
pop @row;
}
if ( $TrimFieldLead ) {
for (my $i=0;$i<@row;$i++) {
$row[$i]=_trim_lead($row[$i]);
}
}
if ( $TrimFieldTrail ) {
for (my $i=0;$i<@row;$i++) {
$row[$i]=_trim_trail($row[$i]);
}
}
$sth->execute( @row );
if ( DBI->errstr ) {
print DBI->errstr;
log_warn( DBI->errstr );
my $errrec = 'RECORD: ' . join "~", @row;
log_warn( $errrec );
sys_die( 'Aborting' );
}
$count++;
}
db_commit( $vdn );
close $fh or sys_die( "Error closing $file_name" );
return $count;
}
sub db_insert_from_query {
my ($src_vdn, $des_vdn, $plugin) = @_;
$plugin = 0 unless $plugin;
## set up array of plugins
my @plugins;
if ( ref $plugin eq 'ARRAY' ) {
@plugins = map { $_ } @{$plugin}; ## copy plugin list to plugin array
} else {
push @plugins, $plugin; ## copy single plugin entry to plugin array
}
my ($src_dbh, $src_sth) = _db_vdn('insert_from_query', $src_vdn);
my ($des_dbh, $des_sth) = _db_vdn('insert_from_query', $des_vdn);
my $count = 0;
while ( my $row = $src_sth->fetchrow_arrayref() ) { ## fetch insert loop
my @tmprow = @{$row};
my $plugin_result = 0;
foreach my $plugin ( @plugins ) { ## call each plugin
my $result = $plugin->( \@tmprow ) if $plugin;
if ( $result > 1000 ) { $plugin_result = 1; } ## plugin bad return
}
next if $plugin_result; ## if any plugin complains, skip the record
$des_sth->execute( @tmprow );
if ( DBI->errstr ) {
log_warn( DBI->errstr );
my $errrec = 'RECORD: ' . join "~", @{$row};
log_warn( $errrec );
sys_die( 'Aborting' );
}
$count++;
}
return $count;
}
sub db_query_to_file {
my ($vdn, $file_name, $delim, $append, $plugin, $protect) = @_;
$delim = '~' unless $delim;
$append = 0 unless $append;
$plugin = 0 unless $plugin; ## unblessed ref to a plugin or ref to array
$protect = 0 unless $protect; ## ref to array of cols to protect
## set up array of plugins
my @plugins;
if ( ref $plugin eq 'ARRAY' ) {
@plugins = map { $_ } @{$plugin}; ## copy plugin list to plugin array
} else {
push @plugins, $plugin; ## copy single plugin entry to plugin array
}
my ($dbh, $sth) = _db_vdn('query_to_file', $vdn);
my $mode;
if ( $append ) {
$mode = '>>';
} else {
$mode = '>';
}
my $count = 0;
open my $fh, $mode, $file_name or sys_die( "Error opening $file_name" );
while ( my $row = $sth->fetchrow_arrayref() ) {
my @outrow = @{$row};
my $plugin_result = 0;
foreach my $plugin ( @plugins ) { ## call each plugin in turn
my $result = $plugin->( \@outrow ) if $plugin;
if ( $result > 1000 ) { $plugin_result = 1; } ## bypass this record
}
next if $plugin_result;
_db_query_to_file_protect( \@outrow, $protect ) if $protect;
print {$fh} join $delim, @outrow;
print {$fh} "\n";
$count++;
}
close $fh or sys_die( "Error closing $file_name" );
return $count;
}
sub db_dump_query {
my ($vdn, $cols) = @_;
my ($dbh, $sth) = _db_vdn('dump_query', $vdn);
while ( my @row = $sth->fetchrow_array() ) {
print "RECORD:\n";
for my $i ( 0 .. $#row ) {
print "\t", $cols->[$i], '=', _db_null( $row[$i] ), "\n";
}
}
return 0;
}
sub db_dump_table {
my ($vdn, $table_name, $max_rows) = @_;
my ($dbh, $sth) = _db_vdn('dump_table', $vdn);
$max_rows = 999_999 unless defined $max_rows;
$table_name = uc $table_name;
my $col_sql = "select column_name " .
" from all_tab_columns " .
" where table_name = '$table_name'";
my ( $tmp_sth, @cols );
$tmp_sth = $dbh->prepare( $col_sql )
or sys_die( DBI->errstr );
$tmp_sth->execute
or sys_die( DBI->errstr );
while ( my @row = $tmp_sth->fetchrow_array() ) {
push @cols, $row[0];
}
$tmp_sth->finish;
my $columns = join ', ', @cols;
my $tab_sql = "select $columns " .
" from $table_name";
$tmp_sth = $dbh->prepare( $tab_sql )
or sys_die( DBI->errstr );
$tmp_sth->execute
or sys_die( DBI->errstr );
my $row_count = 0;
while ( my @row = $tmp_sth->fetchrow_array() ) {
print "RECORD:\n";
for my $i ( 0 .. $#row ) {
print "\t", $cols[$i], "=", _db_null( $row[$i] ), "\n";
}
last if ++$row_count >= $max_rows;
}
$tmp_sth->finish;
return 0;
}
sub db_sqlloader {
my ($vdn, $datfile, $ctlname, $maxerrors) = @_;
my $id = 'db_sqlloader';
my $datfilepath = $db_func_params{$id}{DatFilePath};
my $dbenvr = $db_func_params{$id}{DbEnvr};
my $netservice = $db_func_params{$id}{NetService};
my $datfilefull = $datfilepath . $datfile;
my ($sqlldr_retcd, $sqlldr_result);
log_info( "Executing SQLLoader" );
if ( $dbenvr =~ /$netservice/ ) {
log_info( "Using netservice db connection symantics" );
$sqlldr_retcd = db_sqlloaderx( "$vdn:$dbenvr", $datfilefull, $ctlname, $maxerrors );
} else {
log_info( "Using local db connection symantics" );
$sqlldr_retcd = db_sqlloaderx( $vdn, $datfilefull, $ctlname, $maxerrors );
}
$sqlldr_result = db_sqlloaderx_parse_logfile( $datfilefull );
log_info( "SQLLoader Output:", $sqlldr_result );
if ( $sqlldr_retcd == $SQLLDR_SUCC ) {
log_info( "Load data file $datfile completed successfully" );
}
if ( $sqlldr_retcd == $SQLLDR_WARN ) {
log_warn( "Load data file $datfile completed with warnings" );
}
if ( $sqlldr_retcd == $SQLLDR_FTL || $sqlldr_retcd == $SQLLDR_FAIL ) {
$sqlldr_retcd = $SQLLDR_FAIL;
log_warn( "Load data file $datfile failed" );
}
my $rej_count = db_sqlloaderx_rejected();
if ( $rej_count > 0 ) {
log_warn( "SQLLoader rejected $rej_count records loading $datfile to " . sys_get_dbinst( $vdn ) );
}
if ( $rej_count > $maxerrors ) {
log_warn( "SQLLoader failed loading $datfile to " . sys_get_dbinst( $vdn ) . " due to max rejected records" );
}
return $sqlldr_retcd;
}
sub db_sqlloaderx {
my ($vdn, $datfile, $ctlname, $maxerrors) = @_;
my $defenvr = $dbdefenvr{$vdn};
my $netservice = _db_netservice( $vdn );
my ($db, $un, $pw) = _db_vdn('connect', $vdn);
$maxerrors = $maxerrors || 50;
## validate the data file exists
if ( ! -e $datfile ) { sys_die( "Data file $datfile not found" ); }
## get control file input from job conf
my $key = $ctlname;
my $section = 'sqlloader';
if ( ! $conf_job{$section}{$key} ) {
$key = 'control_file:' . $key;
if ( ! $conf_job{$section}{$key} ) {
sys_die( "No loader definition found in [$section] for key [$ctlname]", 0 );
}
}
my $control = $conf_job{$section}{$key};
my ($base,$path,$type) = fileparse($datfile,qr{\.dat|\.txt});
my $ctlfile = $path.$base.'.ctl';
my $parfile = $path.$base.'.par';
my $badfile = $path.$base.'.bad';
my $disfile = $path.$base.'.dis';
my $outfile = $path.$base.'.out';
## build control file
open my $fh, ">", $ctlfile || sys_die( 'Unable to create SQLLoader ctlfile', 0 );
print $fh $control;
close $fh;
## build params file
open $fh, ">", $parfile || sys_die( 'Unable to create SQLLoader parfile', 0 );
print $fh "userid=$un/$pw$netservice\n";
print $fh "control=$ctlfile\n";
print $fh "silent=(all)\n";
print $fh "data=$datfile\n";
print $fh "log=$outfile\n";
print $fh "bad=$badfile\n";
print $fh "discard=$disfile\n";
close $fh;
my @args = ("sqlldr", "PARFILE=$parfile errors=$maxerrors");
system @args;
my $sqlldr_retcd = $CHILD_ERROR >> 8;
## Normalize os dependent return codes. Why Oracle returns an os dependent
## return code from a cross-platform product is a mystery to me...
if ( $OSNAME eq 'MSWin32' ) {
if ( $sqlldr_retcd == 3 ) { $sqlldr_retcd = 1; }
if ( $sqlldr_retcd == 4 ) { $sqlldr_retcd = 3; }
}
unlink $parfile;
unlink $ctlfile;
return $sqlldr_retcd;
}
sub db_sqlloaderx_parse_logfile {
my $datfile = shift;
my ($base,$path,$type) = fileparse($datfile,qr{\.dat|\.txt});
my $outfile = $path.$base.'.out';
return _db_sqlloaderx_parse_logfile( $outfile );
}
sub db_sqlloaderx_skipped {
if ( defined $sqlloader_results{'skipped'} ) {
return $sqlloader_results{'skipped'}
} else {
return -1;
}
}
sub db_sqlloaderx_read {
if ( defined $sqlloader_results{'read'} ) {
return $sqlloader_results{'read'}
} else {
return -1;
}
}
sub db_sqlloaderx_rejected {
if ( defined $sqlloader_results{'rejected'} ) {
return $sqlloader_results{'rejected'}
} else {
return -1;
}
}
sub db_sqlloaderx_discarded {
if ( defined $sqlloader_results{'discarded'} ) {
return $sqlloader_results{'discarded'}
} else {
return -1;
}
}
sub db_sqlloaderx_elapsed_time {
if ( defined $sqlloader_results{'elapsed_time'} ) {
return $sqlloader_results{'elapsed_time'}
} else {
return 'error';
}
}
sub db_sqlloaderx_cpu_time {
if ( defined $sqlloader_results{'cpu_time'} ) {
return $sqlloader_results{'cpu_time'}
} else {
return 'error';
}
}
sub db_func {
my ($vdn, $package, $proc_name) = @_;
my ($dbh, $sth) = _db_vdn('funcx', $vdn);
unless ( _db_is_oracle($vdn) ) {
sys_die( 'Not an Oracle database connection in db_funcx' );
}
if ( $package ) { $proc_name = $package. '.' .$proc_name; }
my $sql = 'BEGIN :result := ' . $proc_name . '; END;';
my $result;
my $tmp_sth = $dbh->prepare( $sql );
if ( DBI->errstr ) { sys_die( DBI->errstr ); }
$tmp_sth->bind_param_inout( ':result', \$result, 100 );
$tmp_sth->execute;
if ( DBI->errstr ) { sys_die( DBI->errstr ); }
$tmp_sth->finish;
return $result;
}
sub db_proc {
my ($vdn, $package, $proc_name) = @_;
my ($dbh, $sth) = _db_vdn('procx', $vdn);
unless ( _db_is_oracle($vdn) ) {
sys_die( 'Not an Oracle database connection in db_procx' );
}
if ( $package ) { $proc_name = $package . '.' . $proc_name; }
my $sql = 'BEGIN ' . $proc_name . '; END;';
my $tmp_sth = $dbh->prepare( $sql );
if ( DBI->errstr ) { sys_die( DBI->errstr ); }
$tmp_sth->execute;
if ( DBI->errstr ) { sys_die( DBI->errstr ); }
$tmp_sth->finish;
return 0;
}
sub db_proc_in {
my ($vdn, $package, $proc_name, $params) = @_;
unless ( ref $params eq 'ARRAY' ) {
sys_die( 'Invalid type in call to db_procx_in' );
}
my ($dbh, $sth) = _db_vdn('procx_in', $vdn);
unless ( _db_is_oracle($vdn) ) {
sys_die( 'Not an Oracle database connection in db_procx_in' );
}
my $sql = _db_proc_build_sql( $package, $proc_name, $params );
my $tmp_sth = $dbh->prepare( $sql );
if ( DBI->errstr ) { sys_die( DBI->errstr ); }
$tmp_sth = _db_proc_bind_inparams( $tmp_sth, $params );
$tmp_sth->execute;
if ( DBI->errstr ) { sys_die( DBI->errstr ); }
$tmp_sth->finish;
return 0;
}
sub db_proc_out {
my ($vdn, $package, $proc_name, $params) = @_;
unless ( ref $params eq 'ARRAY' ) {
sys_die( 'Invalid type in call to db_procx_out' );
}
my ($dbh, $sth) = _db_vdn('procx_out', $vdn);
unless ( _db_is_oracle($vdn) ) {
sys_die( 'Not an Oracle database connection in db_procx_out' );
}
my $sql = _db_proc_build_sql( $package, $proc_name, $params );
my $tmp_sth = $dbh->prepare( $sql );
if ( DBI->errstr ) { sys_die( DBI->errstr ); }
$tmp_sth = _db_proc_bind_outparams( $tmp_sth, $params);
$tmp_sth->execute;
if ( DBI->errstr ) { sys_die( DBI->errstr ); }
$tmp_sth->finish;
return 0;
}
sub db_proc_inout {
my ($vdn, $package, $proc_name, $params) = @_;
unless ( ref $params eq 'ARRAY' ) {
sys_die( 'Invalid type in call to db_procx_inout' );
}
my ($dbh, $sth) = _db_vdn('procx_inout', $vdn);
unless ( _db_is_oracle($vdn) ) {
sys_die( 'Not an Oracle database connection in db_procx_inout' );
}
my $sql = _db_proc_build_sql( $package, $proc_name, $params );
my $tmp_sth = $dbh->prepare( $sql );
if ( DBI->errstr ) { sys_die( DBI->errstr ); }
$tmp_sth = _db_proc_bind_inoutparams( $tmp_sth, $params);
$tmp_sth->execute;
if ( DBI->errstr ) { sys_die( DBI->errstr ); }
$tmp_sth->finish;
return 0;
}
sub db_dbms_output_enable {
my ($vdn, $bufsize) = shift;
my ($dbh, $sth) = _db_vdn('enable_dbms_output', $vdn);
unless ( _db_is_oracle($vdn) ) {
sys_die( 'Not an Oracle database connection in db_dbms_output_get' );
}
$sys_dbms_output = 1;
$bufsize = 1_000_000 unless $bufsize;
$dbh->func($bufsize, 'dbms_output_enable');
if ( DBI->errstr ) { log_warn( DBI->errstr ); return 1; }
return 0;
}
sub db_dbms_output_disable {
my $vdn = shift;
unless ( _db_is_oracle($vdn) ) {
sys_die( 'Not an Oracle database connection in db_dbms_output_get' );
}
$sys_dbms_output = 0;
return 0;
}
sub db_dbms_output_get {
my $vdn = shift;
my ($dbh, $sth) = _db_vdn('get_dbms_output', $vdn);
unless ( _db_is_oracle($vdn) ) {
sys_die( 'Not an Oracle database connection in db_dbms_output_get' );
}
my @arr;
unless ( $sys_dbms_output ) {
log_warn( 'Output option has not been enabled' );
return \@arr;
}
@arr = $dbh->func('dbms_output_get');
if ( DBI->errstr ) { log_warn( DBI->errstr ); }
return \@arr;
}
sub db_index_rebuild {
my ($vdn, $index_name) = @_;
my ($dbh, $sth) = _db_vdn('ora_index_rebuild', $vdn);
unless ( _db_is_oracle($vdn) ) {
sys_die( 'Not an Oracle database connection in function index_rebuild', 0 );
}
my $sql = "ALTER INDEX $index_name REBUILD";
my $tmp_sth = $dbh->prepare( $sql );
if ( DBI->errstr ) { sys_die( DBI->errstr ); }
$tmp_sth->execute;
if ( DBI->errstr ) { sys_die( DBI->errstr ); }
$tmp_sth->finish;
return 0;
}
sub db_exchange_partition {
my ($vdn, $to_table, $from_table, $partition) = @_;
my ($dbh, $sth) = _db_vdn('ora_swap_partition', $vdn);
unless ( _db_is_oracle($vdn) ) {
sys_die( 'Not an Oracle database connection in function swap_partition', 0 );
}
## REPAIR REQUIRED need to figure out why this is required...
db_commit( $vdn );
sleep 3;
my $sql = "ALTER TABLE $to_table "
. "EXCHANGE PARTITION $partition "
. "WITH TABLE $from_table "
. "INCLUDING INDEXES "
. "WITH VALIDATION";
my $tmp_sth = $dbh->prepare( $sql );
if ( DBI->errstr ) { sys_die( DBI->errstr ); }
$tmp_sth->execute;
if ( DBI->errstr ) { sys_die( DBI->errstr ); }
$tmp_sth->finish;
return 0;
}
sub util_get_filename_load {
my ($base, $ext) = @_;
my $filename = $base . '.' . $ext;
if ( $osuser ) {
$filename = $base . '_' . $osuser . '.' . $ext;
}
return $path_load_dir . $filename;
}
sub util_get_filename_extr {
my ($base, $ext) = @_;
my $filename = $base . '.' . $ext;
if ( $osuser ) {
$filename = $base . '_' . $osuser . '.' . $ext;
}
return $path_extr_dir . $filename;
}
sub util_get_filename_log {
my $base = shift;
return $path_log_dir . $base . $log_ext;
}
sub util_read_header {
my ($filename, $format) = @_;
my $fh = File::Bidirectional->new($filename, {origin => 1} )
or sys_die( "Unable to open file $filename" );
my $head = $fh->readline();
$fh->close;
return $head;
}
sub util_read_footer {
my ($filename, $format) = @_;
my $fh = File::Bidirectional->new($filename, {origin => -1} )
or sys_die( "Unable to open file $filename" );
my $foot = $fh->readline();
$fh->close;
return $foot;
}
sub util_read_file {
my $file = shift;
open( my $fh, $file ) or return 0;
my $text = do { local( $/ ) ; <$fh> } ;
return \$text;
}
sub util_write_header {
my ($filename, $header, $append) = @_;
$header = 'HEADER' unless $header;
my $mode = ">>";
$mode = ">" unless $append;
open my $fh, $mode, $filename or sys_die( "Error writing header to $filename" );
print {$fh} "$header\n";
close $fh or sys_die( "Error closing $filename" );
return 0;
}
sub util_write_footer {
my ($filename, $footer) = @_;
$footer = 'FOOTER' unless $footer;
open my $fh, ">>", $filename or sys_die( "Error writing footer to $filename" );
print {$fh} "$footer\n";
close $fh or sys_die( "Error closing $filename" );
return 0;
}
sub util_move {
my ($from, $to) = @_;
return 0 unless $util_move;
my $result = move($from, $to);
return $result;
}
sub util_trim {
my $str = shift;
$str =~ s/^\s+//;
$str =~ s/\s+$//;
return $str;
}
sub util_zsdf {
my ($number, $width) = @_;
$number =~ s/(?<=\d)(?=(\d{3})+(?!\d))/,/g;
return sprintf '%*s', $width, $number;
}
sub test_init {
$t_ok = 0;
$t_notok = 0;
return 0;
}
sub test_ok {
my ($actual,$expected,$description) = @_;
$t_num++;
if ($actual eq $expected) {
$t_ok++;
log_info("ok $t_num");
} else {
$t_notok++;
sys_set_errorlevel(sys_get_errorlevel()+1);
log_info("not ok $t_num - $description");
}
return 0;
}
sub test_results {
log_info("Test script: passed $t_ok, failed $t_notok");
if ( $t_notok == 0 ) {
log_info("Test script: PASS");
} else {
log_info("Test script: FAIL");
}
return 0;
}
sub test_harness_init {
$th_num = 0;
return 0;
}
sub test_harness_run {
my $test_scripts = shift;
foreach my $ts ( @{$test_scripts} ) {
$th_num++;
log_info("Test script: $ts");
my $retcd = sys_run_job($ts, 8, '-r', '-v');
if ( $retcd > 0 ) {
sys_set_errorlevel( sys_get_errorlevel() + $retcd );
}
}
return 0;
}
sub test_harness_results {
my $test_scripts = shift;
my ($ts_passed, $ts_failed);
my $th_result = 'PASS';
my $th_passed = 0;
my $th_failed = 0;
foreach my $ts ( @{$test_scripts} ) {
$ts =~ s/\.pl$//;
my $tsfull = util_get_filename_log( $ts );
my $log = util_read_file( $tsfull );
if ( ! $log ) {
log_info( "Error reading log for test script: $ts" );
next;
}
$ts_passed = 0;
$ts_failed = 0;
$th_num++;
$$log =~ m#.{19,19} Test script: (PASS|FAIL|DUBIOUS)#;
my $ts_result = $1;
$$log =~ m#.{19,19} Test script: passed (\d+), failed (\d+)#;
$ts_passed = $1;
$ts_failed = $2;
if ( $ts_result eq 'PASS' ) {
$th_passed++;
}
if ( $ts_result eq 'FAIL' ) {
$th_failed++;
$th_result = 'FAIL';
}
log_info( "Test harness: script $ts, passed $ts_passed, failed $ts_failed, $ts_result" );
}
log_info( "Test harness: passed $th_passed, failed $th_failed" );
log_info( "Test harness: $th_result" );
return 0;
}
sub test_harness_summary {
my $test_harnesses = shift;
foreach my $th ( @{$test_harnesses} ) {
$th =~ s/\.pl$//;
my $thfull = util_get_filename_log( $th );
my $log = util_read_file( $thfull );
if ( ! $log ) {
log_info( "Error reading log for test harness: $th" );
next;
}
log_info( "Test harness summary: $th" );
}
return 0;
}
# private methods
# -----------------------------------------------------------------------------
sub _sys_init_vars {
$pid = $PROCESS_ID;
$errorlevel = 0;
@plugins = ();
$sys_dbms_output = 1;
$sys_log_open = 0;
$sys_jobconf_override = 0;
$sys_jobconf_file = '';
%log_level_opts = (
FATAL => 'FATAL',
ERROR => 'FATAL,ERROR',
WARN => 'FATAL,ERROR,WARN',
INFO => 'FATAL,ERROR,WARN,INFO',
DEBUG => 'FATAL,ERROR,WARN,INFO,DEBUG',
NONE => 'NONE',
);
_sys_read_conf( 'sys_data.conf' );
_sys_read_conf( 'sys_log.conf' );
_sys_read_conf( 'sys_mail.conf' );
_sys_read_conf( 'sys_common.conf' );
_sys_read_conf( 'sys_util.conf' );
_sys_read_conf( 'sys_environment.conf' );
_sys_read_conf( 'sys_de.conf');
_sys_read_conf( 'sys_run_controls.conf');
my $envvar = uc $conf_system{'system'}{'envvar'};
$dataenvr = lc $ENV{$envvar};
if ( ! defined $dataenvr ) {
sys_die( "Environment variable $dataenvr not set", 0 );
}
$path_bin_dir = $conf_system{"$OSNAME directory bin"}{$dataenvr};
$path_lib_dir = $conf_system{"$OSNAME directory lib"}{$dataenvr};
$path_log_dir = $conf_system{"$OSNAME directory log"}{$dataenvr};
$path_load_dir = $conf_system{"$OSNAME directory load"}{$dataenvr};
$path_extr_dir = $conf_system{"$OSNAME directory extr"}{$dataenvr};
$path_prev_dir = $conf_system{"$OSNAME directory prev"}{$dataenvr};
$path_scripts_dir = $conf_system{"$OSNAME directory scripts"}{$dataenvr};
$mail_server = $conf_mail{'mail'}{'server'};
$mail_from = $conf_mail{'mail'}{'from'};
$mail_emailto = $conf_mail{'mail'}{'emailto'};
$mail_pagerto = $conf_mail{'mail'}{'pagerto'};
$mail_email_levels = $conf_mail{'mail'}{'email_levels'} || "FATAL";
$mail_pager_levels = $conf_mail{'mail'}{'pager_levels'} || "FATAL";
$log_file = $conf_log{'log'}{'default_logfile'};
$log_filefull = $path_log_dir . $log_file;
$log_logging_levels = $conf_log{'log'}{'logging_levels'} || "FATAL,ERROR,WARN,INFO";
$log_console_levels = $conf_log{'log'}{'console_levels'} || "FATAL,ERROR,WARN,INFO";
$log_gdg = $conf_log{'log'}{'gdg'} || 5;
$path_plugin_dir = $conf_system{"$OSNAME directory plugin"}{$dataenvr};
if ( $osuser ) {
$dbitrace_file = $dbitrace_base . '_' . $osuser . $log_ext;
}
$dbitrace_filefull = $path_log_dir.$dbitrace_file;
## load data structures
@databases = split m/,/, $conf_data{'databases'}{'databases'};
@dat_envrs = split m/,/, $conf_system{'system'}{'dat_envrs'};
@job_acros = split m/,/, $conf_system{'system'}{'job_acros'};
foreach my $db ( @databases ) {
$dbname{$db} = $conf_data{'names'}{$db};
}
foreach my $db ( @databases ) {
$dbdefenvr{$db} = $conf_data{'default '.$dataenvr}{$db};
}
foreach my $db ( @databases ) {
$dbhandles{$db}{'dbh'} = 0;
$dbhandles{$db}{'sth'} = 0;
}
foreach my $db ( @databases ) {
$dbinst{$db} = $conf_data{'instances'}{$db};
}
foreach my $db ( @databases ) {
foreach my $inst ( split m/,/, $conf_data{'instances'}{$db} ) {
$dbconn{$db}{$inst}{'netservice'} = $conf_data{"$db $inst"}{'netservice'};
$dbconn{$db}{$inst}{'database' } = $conf_data{"$db $inst"}{'database'};
$dbconn{$db}{$inst}{'username' } = $conf_data{"$db $inst"}{'username'};
$dbconn{$db}{$inst}{'password' } = $conf_data{"$db $inst"}{'password'};
}
}
return 0;
}
sub _sys_job_init {
my $rtconf = $path_conf_dir.'/'.$jobname.'.'.$pid.'.running';
## create runtime conf file
open my $cfile, '>', $rtconf or sys_die( "Error creating runtime jobconf file" );
close $cfile;
my $conf = new Config::IniFiles( -file => $rtconf );
unless ( defined $conf ) { sys_die( "Error opening runtime jobconf file" ); }
my $starttime = time;
$conf->newval( 'pid', 'pid', $pid );
$conf->newval( 'starttime', 'starttime', $starttime );
$conf->newval( 'restart', 'restart', 0 );
$conf->RewriteConfig;
return 0;
}
sub _sys_job_end {
my $rtconf = $path_conf_dir.'/'.$jobname.'.'.$pid.'.running';
if ( -e $rtconf ) {
unlink $rtconf;
}
return 0;
}
sub _sys_job_dependent {
my $dependent_jobname = shift;
return 0 unless $dependent_jobname;
my $conf = new Config::IniFiles( -file => $path_conf_dir.'/sys_environment.conf' );
unless ( defined $conf ) { sys_die( "Error opening sys_environment.conf (4)" ); }
my $params = join '~', $conf->Parameters( 'jobs' );
if ( $params =~ m/$dependent_jobname/x ) { ## case sensitive
## one or more instances of dependent job is currently running
log_info( "Job name $dependent_jobname is active in the system, waiting" );
return 1;
}
return 0;
}
sub _sys_read_conf {
my $conf = shift;
my $conf_filefull = $path_conf_dir . '/' . $conf;
my $msg1 = "Probably syntax error, unable to load";
if ( $conf =~ m/^sys_data/x ) {
tie %conf_data, 'Config::IniFiles', ( -file => $conf_filefull )
or sys_die( "$msg1 data conf: $conf", 0 );
}
if ( $conf =~ m/^sys_log/x ) {
tie %conf_log, 'Config::IniFiles', ( -file => $conf_filefull )
or sys_die( "$msg1 log conf: $conf", 0 );
}
if ( $conf =~ m/^sys_mail/x ) {
tie %conf_mail, 'Config::IniFiles', ( -file => $conf_filefull )
or sys_die( "$msg1 mail conf: $conf", 0 );
}
if ( $conf =~ m/^sys_common/x ) {
tie %conf_query, 'Config::IniFiles', ( -file => $conf_filefull )
or sys_die( "$msg1 query conf: $conf", 0 );
}
if ( $conf =~ m/^sys_util/x ) {
tie %conf_util, 'Config::IniFiles', ( -file => $conf_filefull )
or sys_die( "$msg1 util conf: $conf", 0 );
}
if ( $conf =~ m/^sys_environment/x ) {
tie %conf_system, 'Config::IniFiles', ( -file => $conf_filefull )
or sys_die( "$msg1 environment conf: $conf", 0 );
}
if ( $conf =~ m/^sys_test/x ) {
tie %conf_job, 'Config::IniFiles', ( -file => $conf_filefull )
or sys_die( "$msg1 test conf: $conf", 0 );
}
if ( $conf =~ m/^sys_de/x ) {
tie %conf_de, 'Config::IniFiles', ( -file => $conf_filefull )
or sys_die( "$msg1 de conf: $conf", 0 );
}
if ( $conf =~ m/^sys_run_controls/x ) {
tie %conf_rcontrols, 'Config::IniFiles', ( -file => $conf_filefull )
or sys_die( "$msg1 run controls conf: $conf", 0 );
}
## job specific conf file
if ( $conf !~ m/^sys_/x ) {
tie %conf_job, 'Config::IniFiles', ( -file => $conf_filefull )
or sys_die( "$msg1 job conf: $conf", 0 );
}
return 0;
}
sub _sys_read_job {
if ( $conf_job{job}{'logfile'} ) {
$log_file = $conf_job{job}{'logfile'};
}
if ( $conf_job{job}{'logging_levels'} ) {
$log_logging_levels = $conf_job{job}{'logging_levels'};
}
if ( $conf_job{job}{'console_levels'} ) {
$log_console_levels = $conf_job{job}{'console_levels'};
}
if ( $conf_job{job}{'log_gdg'} ) {
$log_gdg = $conf_job{job}{'log_gdg'};
}
if ( $conf_job{job}{'log_prefix'} ) {
$log_prefix = $conf_job{job}{'log_prefix'};
}
if ( $conf_job{job}{'emailto'} ) {
$mail_emailto = $conf_job{job}{'emailto'};
}
if ( $conf_job{job}{'pagerto'} ) {
$mail_pagerto = $conf_job{job}{'pagerto'};
}
if ( $conf_job{job}{'email_levels'} ) {
$mail_email_levels = $conf_job{job}{'email_levels'};
}
if ( $conf_job{job}{'pager_levels'} ) {
$mail_pager_levels = $conf_job{job}{'pager_levels'};
}
return 0;
}
sub _sys_init_source_validation {
open my $fh, "<", $script_filefull
|| sys_die( "Unable to open $script_file for validatation", 0 );
my @r = <$fh>;
close $fh;
my $source = join '', @r;
my $errm1 = "$script_file failed source validation, id tag ";
my $errm2 = "$script_file failed source validation, pod section ";
my $errm3 = " is missing or invalid";
my $checkfor;
$checkfor = "FILENAME";
$source =~ m/^\#\#@@.*/m
or sys_die( $errm1.$checkfor.$errm3, 0 );
$checkfor = "SOURCETITLE";
$source =~ m/^\#\#\$\$.*/m
or sys_die( $errm1.$checkfor.$errm3, 0 );
$checkfor = "NAME";
$source =~ m/^!1 $checkfor\n\n[A-Za-z]/m
or sys_die( $errm2.$checkfor.$errm3, 1 );
$checkfor = "DESCRIPTION";
$source =~ m/^!1 $checkfor\n\n[A-Za-z]/m
or sys_die( $errm2.$checkfor.$errm3, 1 );
$checkfor = "RECOVERY NOTES";
$source =~ m/^!1 $checkfor\n\n[A-Za-z]/m
or sys_die( $errm2.$checkfor.$errm3, 1 );
$checkfor = "ENVIRONMENT NOTES";
$source =~ m/^!1 $checkfor\n\n[A-Za-z]/m
or sys_die( $errm2.$checkfor.$errm3, 1 );
$checkfor = "DEPENDENCIES";
$source =~ m/^!1 $checkfor\n\n[A-Za-z]/m
or sys_die( $errm2.$checkfor.$errm3, 1 );
$checkfor = "HISTORY";
$source =~ m/^!1 $checkfor\n\n[A-Za-z0-9\*]/m
or sys_die( $errm2.$checkfor.$errm3, 1 );
return 0;
}
sub _sys_run_background {
if ( $OSNAME eq 'MSWin32' ) {
sys_die( 'Background run mode not available on Windows', 0 );
}
$opt_commandline =~ s{-rb }{-r };
$opt_commandline =~ s{-rb$}{-r};
print "$script_filefull $opt_commandline".' &';
exit 0;
}
sub _sys_run_scheduled {
## this die is temporary should use sys_die
die "Not yet implemented\n\n";
}
sub _sys_run_de {
my $de = shift;
my $conf_file = $jobname . '.' . $de . '.conf';
_sys_read_conf( $conf_file ); ## tie %conf_job to job specific conf file
_sys_read_job(); ## read job specific settings from %conf_job
return 0;
}
sub _sys_run_restart {
## this die is temporary should use sys_die
die "Not yet implemented\n\n";
}
sub _sys_forkexec {
my ($jobname, @params) = @_;
my $pid;
if ( $pid = fork ) {
return $pid;
## this is the parent, so return the pid, everything below here is
## either the child or a major system failure
}
elsif ( defined $pid ) {
exec $jobname, @params;
## shouldn't reach this unless exec fails, we exit here (not return)
## becuase we are in the child
exit 0;
} else {
log_warn( "Could not fork $!" );
return 0;
}
}
sub _sys_reap_child {
my $pid = 0;
if ( ($pid = waitpid(-1, 0)) > 0 ) {
$pidlib{$pid}{retcd} = $? >> 8;
}
return $pid;
}
sub _sys_test_dbcon {
my $connections = shift;
## open dbi trace file
DBI->trace(1, $dbitrace_filefull );
foreach my $connectdef ( split m/,/, $connections ) {
my ($db, $inst) = split m/:/, $connectdef;
_check_array_val( $db, \@databases )
|| sys_die( "Invalid database: [$db]", 0 );
_check_array_val( $inst, [split m/,/, $dbinst{$db}] )
|| sys_die( "Invalid database instance: [$db.$inst]", 0 );
my $database = $dbconn{$db}{$inst}{'database'};
my $username = $dbconn{$db}{$inst}{'username'};
my $password = $dbconn{$db}{$inst}{'password'};
print "Connecting to: $db/$inst\n";
my $dbh = DBI->connect( $database, $username, $password, { RaiseError => 0, AutoCommit => 0 } )
or sys_die( DBI->errstr, 0 );
## push resulting handle onto handle stack for cleanup on exit
$dbhandles{$db}{'dbh'} = $dbh;
print "Success\n\n";
}
exit 0;
}
sub _sys_check_severity_levels {
my $lvls_str = shift;
## levls_str can be either a single value or a comma delimited list
if ( $lvls_str =~ /,/ ) {
## received a list of severity levels
my @loglvls = split m/,/, $lvls_str;
foreach my $level ( @loglvls ) {
if ( $level !~ /FATAL|ERROR|WARN|INFO|DEBUG|NONE/ ) {
sys_die( 'Invalid logging/notification severity list', 0 );
}
}
return $lvls_str;
} else {
## received a single severity level to be translated to a list
if ( $lvls_str =~ /^FATAL$/i ) {
$lvls_str = 'FATAL';
return $lvls_str;
}
if ( $lvls_str =~ /^ERROR$/i ) {
$lvls_str = 'FATAL,ERROR';
return $lvls_str;
}
if ( $lvls_str =~ /^WARN$/i ) {
$lvls_str = 'FATAL,ERROR,WARN';
return $lvls_str;
}
if ( $lvls_str =~ /^INFO$/i ) {
$lvls_str = 'FATAL,ERROR,WARN,INFO';
return $lvls_str;
}
if ( $lvls_str =~ /^DEBUG$/i ) {
$lvls_str = 'FATAL,ERROR,WARN,INFO,DEBUG';
return $lvls_str;
}
if ( $lvls_str =~ /^NONE$/i ) {
$lvls_str = '';
return $lvls_str;
}
sys_die( 'Invalid logging/notification severity level', 0 );
}
return 0;
}
sub _sys_check_log_gdg {
if ( $opt_log_gdg =~ /[0-9]{1,3}/ ) {
sys_die( 'Invalid log gdg specified', 0 );
}
return $opt_log_gdg;
}
sub _sys_check_log_radix {
if ( $opt_log_radix < 1 || $opt_log_radix > 4 ) {
sys_die( 'Invalid log radix specified', 0 );
}
return $opt_log_radix;
}
sub _sys_check_de_override {
my $tmp_jobname = shift;
my $tmp_jobconf_file = $tmp_jobname;
my $delist = $conf_de{jobname}{$tmp_jobname};
if ( $delist ) { ## possible override of job conf
my $de = '0000';
if ( $delist =~ /(\d\d\d\d\d)\s?$/ ) {
$de = $1;
}
my $overenvs = $conf_de{$de}{'env'};
if ( $overenvs =~ /$dataenvr/i ) {
## as a side-effect, sys_jobconf_override gets set here...
$sys_jobconf_override = 1; ## so we know override is effective
$tmp_jobconf_file .= ".$de";
}
}
return $tmp_jobconf_file;
}
sub _sys_disp_logprev {
if ( $opt_log_file ) { $log_file = $opt_log_file; }
$log_filefull = $path_log_dir . $log_file;
if ( -e $log_filefull ) {
print "Log: $log_filefull\n";
system "cat $log_filefull";
print "\n";
exit 0;
}
print "No previous log file found\n\n";
return 0;
}
sub _sys_disp_logarch {
if ( $opt_log_file ) { $log_file = $opt_log_file; }
$log_filefull = $path_log_dir . $log_file;
my @logs = glob $log_filefull . '.*';
if ( @logs ) {
foreach my $log ( sort @logs ) {
print "Log: $log\n";
system "cat $log";
}
print "\n";
exit 0;
}
print "No archived log files found\n\n";
return 0;
}
sub _sys_disp_jobs {
my @jobs = glob $path_bin_dir.'*.pl';
if ( @jobs ) {
foreach my $job ( sort @jobs ) {
my $description = 'No description found';
open my $fh, "<", $job or sys_die( "Unable to open $job", 0 );
while ( <$fh> ) {
chomp;
if ( /^\#\#\$\$/ ) {
$description = substr $_, 4;
}
}
close $fh;
$job =~ s{^\/.*\/}{};
print "Job: $job\n";
print " $description\n";
}
print "\n";
exit 0;
}
print "No archived job files found\n\n";
return 0;
}
sub _sys_disp_active_jobs {
my $logging = shift; ## needs implementing
my @actjobs = glob $path_conf_dir.'/*.running';
print 'Jobs currently active: ' . scalar @actjobs . "\n";
if ( @actjobs ) {
foreach my $job ( sort @actjobs ) {
my $conf = new Config::IniFiles( -file => $job );
unless ( defined $conf ) { sys_die( "Error opening $job" ); }
my $pid = $conf->val( 'pid', 'pid' );
## NOTE: use Unix::PID to determine if pid is still runninng...
## If pid is no longer running, replace "Job:" with "???:".
my $starttime = $conf->val( 'starttime', 'starttime' );
my $fmtdtime = time2str( '%Y/%m/%d %T', $starttime );
$job =~ s{^\/.*\/}{};
$job =~ s{\.\d+\.running$}{};
print "Job: $job\n";
print " pid=$pid\n";
print " starttime=$fmtdtime\n";
$conf = undef;
}
}
print "\n";
exit 0;
}
sub _sys_disp_doc {
if ( -e $script_filefull ) {
my %podparams = (
infile => $script_filefull,
outfile => "STDOUT",
);
wikipod2text( %podparams );
} else {
print "File not found $script_filefull\n\n";
}
exit 0;
}
sub _sys_disp_sql {
my @query_names = keys %{$conf_query{$jobname}};
if ( @query_names ) {
foreach my $query_name ( sort @query_names ) {
my $query = $conf_query{$jobname}{$query_name};
print "Query: $query_name\n";
print $query;
print "\n\n";
}
} else {
print "No querys found\n\n";
}
exit 0;
}
sub _sys_disp_params {
my $dblen = 0;
foreach my $db ( @databases ) {
if ( length $dbname{$db} > $dblen ) { $dblen = length $dbname{$db}; }
}
print "\n" . uc($dataenvr) . " Database Connections:\n";
foreach my $db ( @databases ) {
my $dbstr = sprintf "%-${dblen}s", $dbname{$db};
$dbstr .= ' = ' . $db . '/' . $dbdefenvr{$db};
print " $dbstr\n",;
}
print "\n" . uc($dataenvr) . " Job Settings:\n";
print " Job Name = ", $jobname, "\n";
print " Log File = ", $log_file, "\n";
print " Log Logging Levels = ", $log_logging_levels, "\n";
print " Log Console Levels = ", $log_console_levels, "\n";
print " Log Gdg = ", $log_gdg, "\n";
print " Path Bin Dir = ", $path_bin_dir, "\n";
print " Path Log Dir = ", $path_log_dir, "\n";
print " Path Lib Dir = ", $path_lib_dir, "\n";
print " Path Conf Dir = ", $path_conf_dir, "\n";
print " Path Plugin Dir = ", $path_plugin_dir, "\n";
print " Path Load Dir = ", $path_load_dir, "\n";
print " path Extract Dir = ", $path_extr_dir, "\n";
print " path Prev Dir = ", $path_prev_dir, "\n";
print " path Scripts Dir = ", $path_scripts_dir, "\n";
print " Mail Server = ", $mail_server, "\n";
print " Mail Email From = ", $mail_from, "\n";
print " Mail Email To = ", $mail_emailto, "\n";
print " Mail Pager To = ", $mail_pagerto, "\n";
print " Mail Email Levels = ", $mail_email_levels, "\n";
print " Mail Pager Levels = ", $mail_pager_levels, "\n";
print "\n";
exit 0;
}
sub _sys_send_email_message {
my $params = shift;
my ($addrlist, $message) = split m/~/, $params;
$mail_emailto = $addrlist;
_log_send_mail($message, 'MESSAGE');
exit 0;
}
sub _sys_send_pager_message {
my $params = shift;
my ($addrlist, $message) = split m/~/, $params;
$mail_pagerto = $addrlist;
_log_send_page($message, 'MESSAGE');
exit 0;
}
sub _sys_help {
my $verbose = shift;
$verbose = 0 unless $verbose;
my $section;
if ( $verbose == 0 ) {
print "\nUSAGE\n $script_file [options]\n\n";
print "Use option -h for help with options\n";
print "Use option -hp for help with option parameters\n";
print "Use option -man for system documentation\n";
exit 1;
}
if ( $verbose == 1 ) { $section = 'OPTIONS'; };
if ( $verbose == 2 ) { $section = 'ARGUMENTS'; };
print "\n";
my %podparams = (
infile => $path_lib_dir."DBIx/JCL.pm",
outfile => "STDOUT",
section => $section,
);
wikipod2text( %podparams );
exit 1;
}
sub _log_init_log_file {
## log file rotation if generations > 0
if ( -e $log_filefull && $log_gdg > 0 ) {
_log_rotate();
}
## create new locked log file
## if the file is already locked, will wait until the file is unlocked
my $fh = new IO::LockedFile(">$log_filefull")
or sys_die( 'Failed opening log file', 0 );
## close and unlock the file
$fh->close();
$sys_log_open = 1;
return 0;
}
sub _log_write_to_log {
my ($level, $force, $msg, $exmsg) = @_;
my ($message,$exmessage);
if ( ref $exmsg eq 'ARRAY' ) {
my $lead = ' ' x 18;
$lead .= '+ ';
my @output = map { $lead . $_ . "\n" } @{$exmsg};
my $exmessage = join '', @output;
$exmessage =~ s/\n$//ms;
$message = $msg . "\n" . $exmessage;
} else {
$message = $msg;
$message =~ s/\n/ /g;
}
if ( $log_logging_levels =~ /$level/ || $force ) {
_log_print_log( $level, $message );
}
_log_send_notifications( $level, $force, $msg );
return 0;
}
sub _log_write_to_screen {
my ($level, $force, $msg, $exmsg) = @_;
my ($message,$exmessage);
if ( ref $exmsg eq 'ARRAY' ) {
my $lead = ' ' x 18;
$lead .= '+ ';
my @output = map { $lead . $_ . "\n" } @{$exmsg};
my $exmessage = join '', @output;
$message = $msg . "\n" . $exmessage;
} else {
$message = $msg;
$message =~ s/\n/ /g;
}
$message = _log_trim_msg( $message );
if ( $opt_verbose ) {
print "$message\n";
} else {
if ( $log_console_levels =~ /$level/ || $force ) {
print "$message\n";
}
}
return 0;
}
sub _log_print_log {
my ($level, $message) = @_;
my $preamble = time2str( '%Y/%m/%d %T', time );
if ( $level eq 'FATAL' ) { $preamble .= ' FATAL'; }
if ( $level eq 'ERROR' ) { $preamble .= ' ERROR'; }
if ( $level eq 'WARN' ) { $preamble .= ' WARNING'; }
## open locked log file for appending
## if the file is already locked, will wait until the file is unlocked
my $fh = new IO::LockedFile(">>$log_filefull")
or sys_die( 'Failed opening log file', 0 );
print {$fh} "$preamble $message\n";
## close and unlock the file
$fh->close();
return 0;
}
sub _log_trim_msg {
my $msg = shift;
my $trimmed = '';
if ( $msg =~ /\n/ms ) { ## trim leading spaces from multi-line messages
foreach my $m ( split m/\n/, $msg ) {
$m =~ s/^\s+//;
$trimmed .= $m."\n";
}
$trimmed =~ s/\n$//ms;
} else {
$trimmed = $msg;
}
return $trimmed;
}
sub _log_send_notifications {
my ($level, $force, $message) = @_;
# if ( $tst_harness ) {
# return 0;
# }
if ( $mail_email_levels =~ /$level/ || $force ) {
_log_send_mail( $message, $level );
}
if ( $mail_pager_levels =~ /$level/ || $force ) {
_log_send_page( $message, $level );
}
return 0;
}
sub _log_send_mail {
my ($message, $severity) = @_;
return 0 unless $mail_emailto;
return 0 if $mail_emailto =~ /NONE/i;
my ($subject, $job);
if ( $severity eq 'MESSAGE' ) {
$subject = 'Message from ' . uc $dataenvr;
} else {
$subject = uc($dataenvr). ' Batch Notice';
$message = time2str("%Y/%m/%d %H:%M:%S : ", time) . uc($severity) . " : $script_file : $message";
}
## get the log file contents and append to message
if ( ! $severity eq 'MESSAGE' ) {
if ( -e $log_filefull ) {
$message .= "\nLog Entries:\n";
open my $fh, "<", $log_filefull;
while ( <$fh> ) {
$message .= $_;
}
close $fh;
}
}
MIME::Lite->send('smtp', $mail_server, Timeout => 60);
my $msg = MIME::Lite->new(
From => $mail_from,
To => $mail_emailto,
Subject => $subject,
Data => $message
);
$msg->send;
return 0;
}
sub _log_send_page {
my ($message, $severity) = @_;
return 0 unless $mail_pagerto;
return 0 if $mail_pagerto =~ /NONE/i;
my ($subject, $job);
if ( $severity eq 'MESSAGE' ) {
$subject = 'Message from ' . uc $dataenvr;
} else {
my $subject = uc($dataenvr). ' Batch Notice';
$message = time2str("%Y/%m/%d %H:%M:%S : ", time) . uc($severity) . " : $script_file : $message";
}
MIME::Lite->send('smtp', $mail_server, Timeout => 60);
my $msg = MIME::Lite->new(
From => $mail_from,
To => $mail_pagerto,
Subject => $subject,
Data => $message
);
$msg->send;
return 0;
}
sub _log_rotate {
my ($prev,$next,$i,$j);
my $curr = $log_filefull;
my $currn = $curr;
for ($i = $log_gdg; $i > 1; $i--) {
$j = $i - 1;
my $nextgen = sprintf("%0${log_radix}d", $i);
my $prevgen = sprintf("%0${log_radix}d", $j);
$next = "${currn}." . $nextgen; ##. $ext;
$prev = "${currn}." . $prevgen; ##. $ext;
if ( -r $prev && -f $prev ) {
move($prev,$next) or sys_die( "Log move failed: ($prev,$next)" );
}
}
## copy current to next incremental
my $nextgen = sprintf("%0${log_radix}d", 1);
$next = "${currn}." . $nextgen;
copy($curr, $next);
## preserve permissions and status
my @stat = stat $curr;
chmod( $stat[2], $next ) or sys_warn( "log chmod failed: ($next)" );
utime( $stat[8], $stat[9], $next ) or sys_warn( "log utime failed: ($next)" );
chown( $stat[4], $stat[5], $next ) or sys_warn( "log chown failed: ($next)" );
## now truncate the file
truncate $curr, 0 or sys_die( "Could not truncate $curr" );
return 0;
}
sub _db_connect_check_dependent {
my ($dependent_jobname,$wait_duration,$wait_max_secs,$wait_action) = @_;
my $starttime = time;
while ( 1 ) {
if ( _sys_job_dependent($dependent_jobname) ) {
sleep $wait_duration;
my $curtime = time;
if ( $curtime - $starttime > $wait_max_secs ) {
if ( $wait_action =~ m/^run$/ix ) {
log_info( "Maximum dependent job wait time exceeded, starting" );
last;
} else {
sys_die( "Maximum dependent job wait time exceeded, aborting" );
return 1; ## reachable if $sys_test_harness
}
}
} else {
last;
}
}
return 0;
}
sub _db_connect_retry {
my ($db,$un,$pw,$retry_duration,$retry_max_secs) = @_;
my $dbh = 0;
my $starttime = time;
while ( 1 ) {
$dbh = DBI->connect( $db, $un, $pw, { RaiseError => 0, AutoCommit => 0 } );
if ( DBI->errstr ) {
if ( $retry_max_secs < 1 ) {
sys_die( DBI->errstr );
return 1; ## reachable if $sys_test_harness
}
if ( DBI->err == 1017 ) { ## ora invalid account or password
sys_die( DBI->errstr );
return 1; ## reachable if $sys_test_harness
}
log_info( DBI->errstr );
log_info( "Connection retry requested, waiting" );
sleep $retry_duration;
my $curtime = time;
if ( $curtime - $starttime > $retry_max_secs ) {
sys_die( "Maximum connection retry time exceeded, aborting" );
return 1; ## reachable if $sys_test_harness
}
} else {
last;
}
}
return $dbh;
}
sub _db_vdn {
my ($caller, $vdn) = @_;
my $sth_name = 'sth_default'; ## default statement handle name
## does vdn contains explicit statement handle?
if ( $vdn =~ /\./ ) {
($vdn, $sth_name) = split /\./, $vdn;
}
my ($this_db, $this_inst);
if ( $vdn =~ m/:/x ) { ## does vdn contain explicit instance?
($this_db, $this_inst) = split m/:/, $vdn;
} else {
$this_db = $vdn;
$this_inst = $dbdefenvr{$vdn};
}
if ( ! $dbname{$this_db} ) {
sys_die( "Virtual database name [$vdn] is invalid" );
}
## special return values if caller is 'connect'
if ( $caller eq 'connect' ) {
my $database = $dbconn{$this_db}{$this_inst}{'database'};
my $username = $dbconn{$this_db}{$this_inst}{'username'};
my $password = $dbconn{$this_db}{$this_inst}{'password'};
return ($database, $username, $password);
}
# ## shutdown gracefully if running under the 'test connections' flag
# if ( $opt_test ) {
# log_close( "End connection test: $jobname" );
# sys_end();
# exit 0;
# }
## return database and statement handles for this vdn
my $dbh = $dbhandles{$this_db}{'dbh'};
my $sth = $dbhandles{$vdn}{$sth_name};
return ($dbh, $sth);
}
sub _db_netservice {
my ($vdni) = shift;
my $netservice = '';
if ( $vdni =~ m/:/x ) { ## vdn contains instance definiton
my ($db, $inst) = split m/:/, $vdni;
_check_array_val( $db, \@databases )
|| sys_die( "Invalid database: [$db]", 0 );
_check_array_val( $inst, [split m/,/, $dbinst{$db}] )
|| sys_die( "Invalid database instance: [$db.$inst]", 0 );
$netservice = $dbconn{$db}{$inst}{netservice};
}
return $netservice;
}
sub _db_proc_build_sql {
my ($package, $proc_name, $params) = @_;
my $numparams = scalar @{$params};
if ( $package ) { $proc_name = $package . '.' . $proc_name; }
my $sql = 'BEGIN ' . $proc_name . '(';
for my $i ( 0 .. $numparams - 1 ) {
$sql .= ':p'.$i;
if ( $i < $numparams - 1 ) { $sql .= ','; }
}
$sql .= '); END;';
return $sql;
}
sub _db_sqlloaderx_parse_logfile {
my $logfile = shift;
%sqlloader_results = (); ## hash of SQL*Loader results
## default values
$sqlloader_results{'skipped'} = "Problem obtaining value";
$sqlloader_results{'read'} = $sqlloader_results{'skipped'};
$sqlloader_results{'rejected'} = $sqlloader_results{'skipped'};
$sqlloader_results{'discarded'} = $sqlloader_results{'skipped'};
$sqlloader_results{'elapsed_time'} = $sqlloader_results{'skipped'};
$sqlloader_results{'cpu_time'} = $sqlloader_results{'skipped'};
my $log = new IO::File "<$logfile";
if (! defined $log) {
sys_warn( "Failed to open SQL*Loader log file $logfile" );
return 1;
}
## skip the first line, check the second for the SQL*Loader declaration
my $line = <$log>;
$line = <$log>;
unless ($line =~ /^SQL\*Loader/) {
sys_warn( 'File does not appear to be a valid SQL*Loader log file' );
return 1;
}
while (<$log>) {
chomp;
if ( m/^Total logical records skipped:\s+(\d+)/ ) {
$sqlloader_results{'skipped'} = $1;
next;
}
if ( m/^Total logical records read:\s+(\d+)/ ) {
$sqlloader_results{'read'} = $1;
next;
}
if ( m/^Total logical records rejected:\s+(\d+)/ ) {
$sqlloader_results{'rejected'} = $1;
next;
}
if ( m/^Total logical records discarded:\s+(\d+)/ ) {
$sqlloader_results{'discarded'} = $1;
next;
}
if( m/^Elapsed time was:\s+(.+)/ ) {
$sqlloader_results{'elapsed_time'} = $1;
next;
}
if( m/^CPU time was:\s+(.+)/ ) {
$sqlloader_results{'cpu_time'} = $1;
next;
}
}
$log->close;
my @results;
push @results, "Skipped: " . $sqlloader_results{'skipped'};
push @results, "Read: " . $sqlloader_results{'read'};
push @results, "Rejected: " . $sqlloader_results{'rejected'};
push @results, "Discarded: " . $sqlloader_results{'discarded'};
push @results, "Elapsed Time: " . $sqlloader_results{'elapsed_time'};
push @results, "CPU Time: " . $sqlloader_results{'cpu_time'};
## return ref to array of results
return \@results;
}
sub _db_proc_bind_inparams {
my ($sth, $params) = @_;
my $numparams = scalar @{$params};
for my $i ( 0 .. $numparams - 1 ) {
my $var = ':p'.$i;
$sth->bind_param( $var, ${$params}[$i] );
}
return $sth;
}
sub _db_proc_bind_outparams {
my ($sth, $params) = @_;
my $numparams = scalar @{$params};
for my $i ( 0 .. $numparams - 1 ) {
my $var = ':p'.$i;
$sth->bind_param_inout( $var, @{$params}[$i], 100 );
}
return $sth;
}
sub _db_proc_bind_inoutparams {
my ($sth, $params) = @_;
my $numparams = scalar @{$params};
for my $i ( 0 .. $numparams - 1 ) {
my $var = ':p'.$i;
if ( ref @{$params}[$i] eq 'SCALAR' ) {
$sth->bind_param_inout( $var, @{$params}[$i], 100 );
} else {
$sth->bind_param( $var, ${$params}[$i] );
}
}
return $sth;
}
sub _db_is_oracle {
my $vdn = shift;
my $inst = $dbdefenvr{$vdn};
my $database = $dbconn{$vdn}{$inst}{'database'}; ## e.g., dbi:Oracle:myinst
if ( $database=~ /^dbi:Oracle:/ ) {
return 1;
}
return 0;
}
sub _db_null {
my $val = shift;
return '<NULL>' unless defined $val;
return $val;
}
sub _db_query_to_file_protect {
my ($row, $protect) = @_;
return 0 if scalar @{$protect} < 1;
foreach my $i ( @{$protect} ) {
my $len = length @{$row}[$i];
my $fil = '*'x$len;
@{$row}[$i] = $fil;
}
return 0;
}
sub _check_array_val {
my ($val, $arr) = @_;
if ( grep { $_ eq $val } @{$arr} ) {
return 1;
}
return 0;
}
sub _trim {
my $str = shift;
$str =~ s/^\s+//;
$str =~ s/\s+$//;
return $str;
}
sub _trim_lead {
my $str = shift;
$str =~ s/^\s+//;
return $str;
}
sub _trim_trail {
my $str = shift;
$str =~ s/\s+$//;
return $str;
}
sub _is_yes {
my $str = shift;
if ( $str =~ /^y$|^yes$/i ) { return 1; }
return 0;
}
sub _is_no {
my $str = shift;
if ( $str =~ /^n$|^no$/i ) { return 1; }
return 0;
}
sub END {
## remove job information from sys_environment.conf
_sys_job_end();
## disconnect any open database handles
foreach my $vdn ( keys %dbhandles ) {
my $dbh = $dbhandles{$vdn}{'dbh'};
my $sth = $dbhandles{$vdn}{'sth'};
if ( defined $sth && $sth ) { $sth->finish; }
if ( defined $dbh && $dbh ) { $dbh->disconnect; }
}
## call plugin end functions
while ( my $pluginf = pop @plugins ) {
my ($pp, $pf, $pff) = split m/~/, $pluginf;
$pp->end();
}
## send completion notifications
unless ( defined $jobname ) { $jobname = '?'; }
my $msg = "Job $jobname ($script_file) has completed ($errorlevel).";
if ( $opt_notify_email_oncomp ) {
_log_send_mail($msg, 'MESSAGE' );
}
if ( $opt_notify_pager_oncomp ) {
_log_send_page($msg, 'MESSAGE' );
}
}
1;