Helios::Service - base class for services in the Helios job processing system

 get/setConfig()
 get/setHostname()
 get/setIniFile()
 get/setJob()
 get/setJobType()
 errstr()
 debug()

 log_priority_threshold=6   

package Helios::Service;

use 5.008;
use base qw( TheSchwartz::Worker );
use strict;
use warnings;
use File::Spec;
use Sys::Hostname;
use Config::IniFiles;
use DBI;
use Data::ObjectDriver::Driver::DBI;
use Error qw(:try);
use TheSchwartz;
use TheSchwartz::Job;
require XML::Simple;
$XML::Simple::PREFERRED_PARSER = 'XML::Parser';

use Helios::Error;
use Helios::Job;
use Helios::ConfigParam;
use Helios::LogEntry;
use Helios::LogEntry::Levels qw(:all);

our $VERSION = '2.22';

our $CACHED_CONFIG;
our $CACHED_CONFIG_RETRIEVAL_COUNT = 0;
our $WORKER_START_TIME = 0;

sub max_retries { return MaxRetries(); }
sub retry_delay { return RetryInterval(); }

sub work {
	my $class = shift;
	my $schwartz_job = shift;
	my $job = $class->JobClass() ? $class->JobClass()->new($schwartz_job) : Helios::Job->new($schwartz_job);
	$WORKER_START_TIME = $WORKER_START_TIME ? $WORKER_START_TIME : time();     # for WORKER_MAX_TTL 
	my $return_code;
	my $args;

	# instantiate the service class into a worker
	my $self = new $class;
	try {
	    # if we've previously retrieved a config
        # AND OVERDRIVE is enabled (1) 
        # AND LAZY_CONFIG_UPDATE is enabled (1),
        # AND we're not servicing the 10th job (or technically a multiple of ten)
        # THEN just retrieve the pre-existing config
        
        #[]
        print "CACHED_CONFIG=",$CACHED_CONFIG,"\n";
        print "CACHED_CONFIG_RETRIEVAL_COUNT=",$CACHED_CONFIG_RETRIEVAL_COUNT,"\n";
        
        if ( defined($CACHED_CONFIG) && 
                $CACHED_CONFIG->{LAZY_CONFIG_UPDATE} == 1 &&
                $CACHED_CONFIG->{OVERDRIVE} == 1 &&
                $CACHED_CONFIG_RETRIEVAL_COUNT % 10 != 0 
            ) {
            $self->prep($CACHED_CONFIG);
            $CACHED_CONFIG_RETRIEVAL_COUNT++;
            if ($self->debug) { $self->logMsg(LOG_DEBUG,"Retrieved config params from in-memory cache"); }    #[] take this out before release
        } else {
            $self->prep();
            if ( defined($self->getConfig()->{LAZY_CONFIG_UPDATE}) && 
                    $self->getConfig()->{LAZY_CONFIG_UPDATE} == 1 ) {
                $CACHED_CONFIG = $self->getConfig();
                $CACHED_CONFIG_RETRIEVAL_COUNT = 1;     # "prime the pump"
            }	    
        }
	    	    
		$job->setConfig($self->getConfig());
		$job->debug( $self->debug );
		$args = $job->parseArgs();
	} catch Helios::Error::InvalidArg with {
		my $e = shift;
		$self->logMsg($job, LOG_ERR, "Invalid arguments: ".$e->text);
		$job->failedNoRetry($e->text);			
		exit(1);
	} catch Helios::Error::DatabaseError with {
		my $e = shift;
		$self->logMsg($job, LOG_ERR, "Database error: ".$e->text);
		$job->failed($e->text);
		exit(1);
	} otherwise {
		my $e = shift;
		$self->logMsg($job, LOG_ERR, "Unexpected error: ".$e->text);
		$job->failed($e->text);
		exit(1);
	};

	# if this is a meta job, we need to burst it apart 
	# if it isn't, we should call the service's run() method 
	$self->setJob($job);
	if ( defined($args->{metajob}) && $args->{metajob} == 1 ) {
		$self->logMsg($job, LOG_NOTICE, 'Job '.$job->getJobid.' is a metajob.  Bursting...');
		try {
			my $numberOfJobs = $self->burstJob($job);
			$self->logMsg($job, LOG_NOTICE, 'Job '.$job->getJobid.' burst into '.$numberOfJobs.' jobs');
		} otherwise {
			my $e = shift;
			$self->logMsg($job, LOG_ERR, 'Job '.$job->getJobid.' burst failure: '.$e->text);
		};
	} else {
		if ($self->debug) { print "RUNNING JOB\n"; }
		$return_code = $self->run($job);
		if ($self->debug) { print "RUN RETURN CODE: ", $return_code,"\n"; }
	}

	# either run() or burstJob() should have marked the job as completed or failed
	# now we have to decide whether to exit or not

	# if run() returned a nonzero, we're assuming run() ended badly, 
	# and we need to exit the process to be safe
	if ($return_code != 0) { exit(1); }

	# if we're not in OVERDRIVE, the worker process will exit as soon as work() returns anyway 
	#    (calling shouldExitOverdrive will be a noop)
	# if we're in OVERDRIVE, work() will exit and the worker process will call it again with another job
	# if we were in OVERDRIVE, but now we're NOT, we should explicitly exit() to accomplish the downshift
	if ( $self->shouldExitOverdrive() ) {
		$self->logMsg(LOG_NOTICE,"Class $class exited (downshift)");
		exit(0);
	}
}

sub setJob { $_[0]->{job} = $_[1]; }
sub getJob { return $_[0]->{job}; }

# need for helios.pl logging	
sub setJobType { $_[0]->{jobType} = $_[1]; }
sub getJobType { return $_[0]->{jobType}; }

sub setConfig { $_[0]->{config} = $_[1]; }
sub getConfig { return $_[0]->{config}; }

sub setFuncid { $_[0]->{funcid} = $_[1]; }
sub getFuncid { return $_[0]->{funcid}; }

sub setIniFile { $_[0]->{inifile} = $_[1]; }
sub getIniFile { return $_[0]->{inifile}; }

sub setHostname { $_[0]->{hostname} = $_[1]; }
sub getHostname { return $_[0]->{hostname}; }

sub errstr { my $self = shift; @_ ? $self->{errstr} = shift : $self->{errstr}; }
sub debug { my $self = shift; @_ ? $self->{debug} = shift : $self->{debug}; }

sub new {
	my $caller = shift;
	my $class = ref($caller) || $caller;
#	my $self = $class->SUPER::new(@_);
	my $self = {};
	bless $self, $class;

	# init fields
	my $jobtype = $caller;
	$self->setJobType($jobtype);

	return $self;
}

sub prep {
	my $self = shift;
	my $cached_config = shift;

	# pull params from environment
	$self->setHostname(hostname);
	if ( defined($ENV{HELIOS_DEBUG}) ) {
		$self->debug($ENV{HELIOS_DEBUG});
	}
	if ( defined($ENV{HELIOS_INI}) && 
		!defined($self->getIniFile) ) {
		$self->setIniFile($ENV{HELIOS_INI});
	}

    if ( defined($cached_config) ) {
        $self->setConfig($cached_config);
        return 1;        
    }

	# now get the Helios conf params from INI and db
	# (these may throw their own errors that the calling routine
	#  will have to catch)
	$self->getConfigFromIni();
	$self->getConfigFromDb();

	return 1;
}

sub getConfigFromIni {
	my $self = shift;
	my $inifile = shift;
	my $jobtype = $self->getJobType();
	my %params;

	# use the object's INI file if we weren't given one explicitly
	# or use the contents of the HELIOS_INI env var
	unless ($inifile) {
		if ( $self->getIniFile() ) {
			$inifile = $self->getIniFile();
		} elsif ( defined($ENV{HELIOS_INI}) ) {
			$inifile = $ENV{HELIOS_INI};
		} elsif (-r File::Spec->catfile(File::Spec->curdir,'helios.ini') ) {
			$inifile = File::Spec->catfile(File::Spec->curdir,'helios.ini');
		} else {
			throw Helios::Error::InvalidArg("INI configuration file not specified.");
		}
	}

	if ( $self->debug() ) { print "inifile: $inifile\nclass:$jobtype\n"; }		
	unless (-r $inifile) { $self->errstr("INI read error: $!"); return undef; }

	my $ini = new Config::IniFiles( -file => $inifile );
	unless ( defined($ini) ) { throw Helios::Error::Fatal("Invalid INI file; check configuration"); }

	# global must exist; it's where the helios db is declared
	if ($ini->SectionExists("global") ) {
		foreach ( $ini->Parameters("global") ) {
			$params{$_} = $ini->val("global", $_);
		}
	} else {
		throw Helios::Error::InvalidArg("Section [global] doesn't exist in config file $inifile");
	}
	
	# if there's a section specifically for this service class, read it too
	# (it will effectively override the global section, BTW)
	if ( $ini->SectionExists($jobtype) ) {
		foreach ( $ini->Parameters($jobtype) ) {
			$params{$_} = $ini->val($jobtype, $_);
		}
	}

	$self->setConfig(\%params);
	return %params;
}

sub getConfigFromDb {
	my $self = shift;
	my $params = $self->getConfig();
	my $hostname = $self->getHostname();
	my $jobtype = $self->getJobType();
	my @cps;
	my $cp;

	if ($self->debug) { print "Retrieving params for ".$self->getJobType()." on ".$self->getHostname()."\n"; }

	try {
		my $driver = $self->getDriver();
		@cps = $driver->search('Helios::ConfigParam' => {
				worker_class => $jobtype,
				host         => '*',
			}
		);
		foreach $cp (@cps) {
			if ($self->debug) { 
				print $cp->param(),'=',$cp->value(),"\n";
			}
			$params->{$cp->param()} = $cp->value();
		}
		@cps = $driver->search('Helios::ConfigParam' => {
				worker_class => $jobtype,
				host         => $hostname,
			}
		);
		foreach $cp (@cps) {
			if ($self->debug) { 
				print $cp->param(),'=',$cp->value(),"\n";
			}
			$params->{$cp->param()} = $cp->value();
		}
		
	} otherwise {
		my $e = shift;
		throw Helios::Error::DatabaseError($e->text);
	};

	$self->setConfig($params);
	return %{$params};
}

sub getFuncidFromDb {
    my $self = shift;
    my $params = $self->getConfig();
    my $jobtype = $self->getJobType();
    my @funcids;

    if ($self->debug) { print "Retrieving funcid for ".$self->getJobType()."\n"; }

    try {
        my $driver = $self->getDriver();
        # also get the funcid 
        my @funcids = $driver->search('TheSchwartz::FuncMap' => {
                funcname => $jobtype 
            }
        );
        if ( scalar(@funcids) > 0 ) {
            $self->setFuncid( $funcids[0]->funcid() );
        }
		
	} otherwise {
		my $e = shift;
		throw Helios::Error::DatabaseError($e->text);
	};
    return $self->getFuncid();	
}

sub jobsWaiting {
	my $self = shift;
	my $params = $self->getConfig();
	my $jobType = $self->getJobType();
	my $funcid;

	try {

		my $dbh = $self->dbConnect($params->{dsn}, $params->{user}, $params->{password});
		unless ($dbh) { throw Helios::Error::DatabaseError($self->errstr); }

		# get the funcid if we don't already have it
		if ( !defined($self->getFuncid()) ) {
		    $funcid = $self->getFuncidFromDb();
		} else {
		    $funcid = $self->getFuncid();
		}

		my $sql2 = <<JWSQL2;
SELECT COUNT(*)
FROM job
WHERE funcid = ?
	AND (run_after < ?)
	AND (grabbed_until < ?)
JWSQL2
		my $sth2 = $dbh->prepare($sql2);

		my $current_time = time();
		$sth2->execute($funcid, $current_time, $current_time);
		my $result2 = $sth2->fetchrow_arrayref();
		unless ( defined($result2->[0]) ) {
			$self->errstr("Received NULL value in jobsWaiting() for funcname $jobType!"); return undef; 
		}
		$sth2->finish();
		$dbh->disconnect();
		return $result2->[0];

	} otherwise {
		throw Helios::Error::DatabaseError($DBI::errstr);
	};
	
}

sub getDriver {
	my $self = shift;
	my $config = $self->getConfig();
	my $driver = Data::ObjectDriver::Driver::DBI->new(
	    dsn      => $config->{dsn},
	    username => $config->{user},
	    password => $config->{password}
	);	
	return $driver;	
}

sub shouldExitOverdrive {
	my $self = shift;
	my $params = $self->getConfig();
	if ( defined($params->{HALT}) ) { return 1; }
	if ( defined($params->{HOLD}) && $params->{HOLD} == 1) { return 1; }
	if ( defined($params->{WORKER_MAX_TTL}) && $params->{WORKER_MAX_TTL} > 0 && 
	       time() > $WORKER_START_TIME + $params->{WORKER_MAX_TTL} ) {
        return 1;
    }
	return 0;
}

sub dbConnect {
	my $self = shift;
	my $dsn = shift;
	my $user = shift;
	my $password = shift;
	my $options = shift;
	my $params = $self->getConfig();

	# if we weren't given params
	unless ($dsn) {
		$dsn = $params->{dsn};
		$user = $params->{user};
		$password = $params->{password};
		$options = $params->{options};
	}

	try {

		my $dbh;
		if ($options) {
			my $o = eval "{$options}";
			if ($@) { $self->errstr($@); return undef;	}
			if ($self->debug) { print "dsn=$dsn\nuser=$user\npass=$password\noptions=$options\n"; }	
			$dbh = DBI->connect_cached($dsn, $user, $password, $o);	
		} else {
			if ($self->debug) { print "dsn=$dsn\nuser=$user\npass=$password\n";	} 
			$dbh = DBI->connect_cached($dsn, $user, $password);
		}
#[]		if ( $DBI::errstr ) { 
        unless ( defined($dbh) ) {
			$self->errstr("DB ERROR: ".$DBI::errstr); 
			throw Helios::Error::DatabaseError($DBI::errstr);
		}
		$dbh->{RaiseError} = 1;
		return $dbh;

	} otherwise {
		throw Helios::Error::DatabaseError($DBI::errstr);
	};

}

sub logMsg {
	my $self = shift;
	my $priority;
	my $msg;
	my $job;

	# if the first parameter is a Helios::Job object, log extra info
	if ( ref($_[0]) && $_[0]->isa('Helios::Job') ) {
		$job = shift;
	}

	# if we were given 2 params, the first is priority, second message
	# if only one, it is the message, default to LOG_INFO priority
	if ( defined($_[0]) && defined($_[1]) ) {
		$priority = shift;
		$msg = shift;
	} else {
		$priority = LOG_INFO;
		$msg = shift;
	}
	
	my $params = $self->getConfig();
	my $jobType = $self->getJobType();
	my $hostname = $self->getHostname();

    # if this log message's priority is lower than log_priority_threshold,
    # don't bother logging the message
    if ( defined($params->{log_priority_threshold}) &&
        $priority > $params->{log_priority_threshold} ) {
        return 1;
    }

    # Sys::Syslog support
	# log the message to syslog IF log_facility was set
	#[]? removed in favor of HeliosX::Logger::Syslog

	# log to database
	my $retries = 0;
	my $retry_limit = 3;
	RETRY: {
		try{
			my $driver = $self->getDriver();
			my $log_entry;
			if ( defined($job) ) {
				$log_entry = Helios::LogEntry->new(
					log_time   => time(),
					host       => $self->getHostname,
					process_id => $$,
					jobid      => $job->getJobid,
					funcid     => $job->getFuncid,
					job_class  => $jobType,
					priority   => $priority,
					message    => $msg
				);
			} else {
				$log_entry = Helios::LogEntry->new(
					log_time   => time(),
					host       => $self->getHostname,
					process_id => $$,
					jobid      => undef,
					funcid     => undef,
					job_class  => $jobType,
					priority   => $priority,
					message    => $msg
				);
			}
			$driver->insert($log_entry);		

		} otherwise {
			my $e = shift;
			if ($retries > $retry_limit) {
				throw Helios::Error::DatabaseError($e->text());
			} else {
				# we're going to give it another shot
				$retries++;
				sleep 10;
				next RETRY;
			}
		};
	} # retry block end
	return 1;
}

sub parseArgXML {
	my $self = shift;
	my $xml = shift;

	my $xs = XML::Simple->new(SuppressEmpty => undef);
	my $args;
	try {
		$args = $xs->XMLin($xml);
	} otherwise {
		throw Helios::Error::InvalidArg($!);
	};
	return $args;
}

sub getJobArgs {
	my $self = shift;
	my $job = shift;
	return $job->getArgs() ? $job->getArgs() : $job->parseArgs();
}

sub completedJob {
	my $self = shift;
	my $job = shift;
	return $job->completed();
}

sub failedJob {
	my $self = shift;
	my $job = shift;
	my $error = shift;
	my $exitstatus = shift;
	return $job->failed($error, $exitstatus);
}

sub failedJobPermanent {
	my $self = shift;
	my $job = shift;
	my $error = shift;
	my $exitstatus = shift;
	return $job->failedNoRetry($error, $exitstatus);
}

sub burstJob {
	my $self = shift;
	my $job = shift;
	my $jobnumber = $job->burst();	
	return $jobnumber;
}

sub run {
    throw Helios::Error::FatalNoRetry($_[0]->getJobType.': run() method not implemented!'); 
}

sub MaxRetries { return undef; }
sub RetryInterval { return undef; }

sub JobClass { return undef; }


1;
__END__