FS/bin/freeside-queued

   1 #!/usr/bin/perl -w
   2
   3 use strict;
   4 use vars qw( $DEBUG $kids $max_kids $sleep_time %kids );
   5 use POSIX qw(:sys_wait_h);
   6 use IO::File;
   7 use Getopt::Std;
   8 use FS::UID qw(adminsuidsetup forksuidsetup driver_name dbh myconnect);
   9 use FS::Daemon qw(daemonize1 drop_root logfile daemonize2 sigint sigterm);
  10 use FS::Conf;
  11 use FS::Record qw(qsearch);
  12 use FS::queue;
  13 use FS::queue_depend;
  14 use FS::Log;
  15
  16 # no autoloading for non-FS classes...
  17 use Net::SSH 0.07;
  18
  19 $DEBUG = 0;
  20
  21 $kids = 0;
  22
  23 &untaint_argv;  #what it sounds like  (eww)
  24 use vars qw(%opt);
  25 getopts('sn', \%opt );
  26
  27 my $user = shift or die &usage;
  28
  29 warn "starting daemonization (forking)\n" if $DEBUG;
  30 #daemonize1('freeside-queued',$user); #to keep pid files unique w/multi installs
  31 daemonize1('freeside-queued');
  32
  33 warn "dropping privledges\n" if $DEBUG;
  34 drop_root();
  35
  36 $ENV{HOME} = (getpwuid($>))[7]; #for ssh
  37
  38 warn "connecting to database\n" if $DEBUG;
  39 $@ = 'not connected';
  40 while ( $@ ) {
  41   eval { adminsuidsetup $user; };
  42   if ( $@ ) {
  43     warn $@;
  44     warn "sleeping for reconnect...\n";
  45     sleep 5;
  46   }
  47 }
  48
  49 my $log = FS::Log->new('queue');
  50 logfile( "%%%FREESIDE_LOG%%%/queuelog.". $FS::UID::datasrc );
  51
  52 warn "completing daemonization (detaching))\n" if $DEBUG;
  53 daemonize2();
  54
  55 #--
  56
  57 my $conf = new FS::Conf;
  58 $max_kids = $conf->config('queued-max_kids') || 10;
  59 $sleep_time = $conf->config('queued-sleep_time') || 10;
  60
  61 my $warnkids=0;
  62 while (1) {
  63
  64   &reap_kids;
  65   #prevent runaway forking
  66   if ( $kids >= $max_kids ) {
  67     warn "WARNING: maximum $kids children reached\n" unless $warnkids++;
  68     &reap_kids;
  69     sleep 1; #waiting for signals is cheap
  70     next;
  71   }
  72   $warnkids=0;
  73
  74   unless ( dbh && dbh->ping ) {
  75     warn "WARNING: connection to database lost, reconnecting...\n";
  76
  77     eval { $FS::UID::dbh = myconnect; };
  78
  79     unless ( !$@ && dbh && dbh->ping ) {
  80       warn "WARNING: still no connection to database, sleeping for retry...\n";
  81       sleep 10;
  82       next;
  83     } else {
  84       warn "WARNING: reconnected to database\n";
  85     }
  86   }
  87
  88   #my($job, $ljob);
  89   #{
  90   #  my $oldAutoCommit = $FS::UID::AutoCommit;
  91   #  local $FS::UID::AutoCommit = 0;
  92   $FS::UID::AutoCommit = 0;
  93
  94   my $nodepend = 'AND NOT EXISTS( SELECT 1 FROM queue_depend'.
  95                  '           WHERE queue_depend.jobnum = queue.jobnum )';
  96
  97   #anything with a priority goes after stuff without one
  98   my $order_by = ' ORDER BY COALESCE(priority,0) ASC, jobnum ASC ';
  99
 100   my $limit = $max_kids - $kids;
 101
 102   $order_by .= ( driver_name eq 'mysql'
 103                    ? " LIMIT $limit FOR UPDATE "
 104                    : " FOR UPDATE LIMIT $limit " );
 105
 106   my $hashref = { 'status' => 'new' };
 107   if ( $opt{'s'} ) {
 108     $hashref->{'secure'} = 'Y';
 109   } elsif ( $opt{'n'} ) {
 110     $hashref->{'secure'} = '';
 111   }
 112
 113   #qsearch dies when the db goes away
 114   my @jobs = eval {
 115     qsearch({
 116       'table'     => 'queue',
 117       'hashref'   => $hashref,
 118       'extra_sql' => $nodepend,
 119       'order_by'  => $order_by,
 120     });
 121   };
 122   if ( $@ ) {
 123     warn "WARNING: error searching for jobs, closing connection: $@";
 124     undef $FS::UID::dbh;
 125     next;
 126   }
 127
 128   unless ( @jobs ) {
 129     dbh->commit or do {
 130       warn "WARNING: database error, closing connection: ". dbh->errstr;
 131       undef $FS::UID::dbh;
 132       next;
 133     };
 134     sleep $sleep_time;
 135     next;
 136   }
 137
 138   foreach my $job ( @jobs ) {
 139
 140     $log->debug('locking queue job', object => $job);
 141
 142     my %hash = $job->hash;
 143     $hash{'status'} = 'locked';
 144     my $ljob = new FS::queue ( \%hash );
 145     my $error = $ljob->replace($job);
 146     if ( $error ) {
 147       warn "WARNING: database error locking job, closing connection: ".
 148            dbh->errstr;
 149       undef $FS::UID::dbh;
 150       next;
 151     }
 152
 153     dbh->commit or do {
 154       warn "WARNING: database error, closing connection: ". dbh->errstr;
 155       undef $FS::UID::dbh;
 156       next;
 157     };
 158
 159     $FS::UID::AutoCommit = 1;
 160
 161     my @args = eval { $ljob->args; };
 162     if ( $@ ) {
 163       warn "WARNING: error retrieving job arguments, closing connection: $@";
 164       undef $FS::UID::dbh;
 165       next;
 166     }
 167     splice @args, 0, 1, $ljob if $args[0] eq '_JOB';
 168
 169     defined( my $pid = fork ) or do {
 170       warn "WARNING: can't fork: $!\n";
 171       my %hash = $job->hash;
 172       $hash{'status'} = 'failed';
 173       $hash{'statustext'} = "[freeside-queued] can't fork: $!";
 174       my $ljob = new FS::queue ( \%hash );
 175       my $error = $ljob->replace($job);
 176       die $error if $error; #XXX still dying if we can't fork AND we can't connect to the db
 177       next; #don't increment the kid counter
 178     };
 179
 180     if ( $pid ) {
 181       $kids++;
 182       $kids{$pid} = 1;
 183     } else { #kid time
 184
 185       #get new db handle
 186       $FS::UID::dbh->{InactiveDestroy} = 1;
 187
 188       forksuidsetup($user);
 189
 190       dbh->{'private_profile'} = {} if UNIVERSAL::can(dbh, 'sprintProfile');
 191
 192       #auto-use classes...
 193       if (    $ljob->job =~ /(FS::(part_export|cust_main|cust_pkg|Cron)::\w+)::/
 194            || $ljob->job =~ /(FS::\w+)::/
 195          )
 196       {
 197         my $class = $1;
 198         eval "use $class;";
 199         if ( $@ ) {
 200           warn "job use $class failed";
 201           my %hash = $ljob->hash;
 202           $hash{'status'} = 'failed';
 203           $hash{'statustext'} = $@;
 204           my $fjob = new FS::queue( \%hash );
 205           my $error = $fjob->replace($ljob);
 206           die $error if $error;
 207           exit; #end-of-kid
 208         };
 209       }
 210
 211       my $eval = "&". $ljob->job. '(@args);';
 212       # don't put @args in the log, may expose passwords
 213       $log->info('starting job ('.$ljob->job.')');
 214       warn 'running "&'. $ljob->job. '('. join(', ', @args). ")\n" if $DEBUG;
 215       local $FS::UID::AutoCommit = 0; # so that we can clean up failures
 216       eval $eval; #throw away return value?  suppose so
 217       if ( $@ ) {
 218         dbh->rollback;
 219         my %hash = $ljob->hash;
 220         $hash{'statustext'} = $@;
 221         if ( $hash{'statustext'} =~ /\/misc\/queued_report/ ) { #use return?
 222           $hash{'status'} = 'done';
 223         } else {
 224           $hash{'status'} = 'failed';
 225           warn "job $eval failed";
 226         }
 227         my $fjob = new FS::queue( \%hash );
 228         my $error = $fjob->replace($ljob);
 229         die $error if $error;
 230         dbh->commit; # for the status change only
 231       } else {
 232         $ljob->delete;
 233         dbh->commit; # for the job itself
 234       }
 235
 236       if ( UNIVERSAL::can(dbh, 'sprintProfile') ) {
 237         open(PROFILE,">%%%FREESIDE_LOG%%%/queueprofile.$$.".time)
 238           or die "can't open profile file: $!";
 239         print PROFILE dbh->sprintProfile();
 240         close PROFILE or die "can't close profile file: $!";
 241       }
 242
 243       exit;
 244       #end-of-kid
 245     }
 246
 247   } #foreach my $job
 248
 249 } continue {
 250   if ( sigterm() ) {
 251     warn "received TERM signal; exiting\n";
 252     exit;
 253   }
 254   if ( sigint() ) {
 255     warn "received INT signal; exiting\n";
 256     exit;
 257   }
 258 }
 259
 260 sub untaint_argv {
 261   foreach $_ ( $[ .. $#ARGV ) { #untaint @ARGV
 262     #$ARGV[$_] =~ /^([\w\-\/]*)$/ || die "Illegal arguement \"$ARGV[$_]\"";
 263     # Date::Parse
 264     $ARGV[$_] =~ /^(.*)$/ || die "Illegal arguement \"$ARGV[$_]\"";
 265     $ARGV[$_]=$1;
 266   }
 267 }
 268
 269 sub usage {
 270   die "Usage:\n\n  freeside-queued user\n";
 271 }
 272
 273 sub reap_kids {
 274   foreach my $pid ( keys %kids ) {
 275     my $kid = waitpid($pid, WNOHANG);
 276     if ( $kid > 0 ) {
 277       $kids--;
 278       delete $kids{$kid};
 279     }
 280   }
 281 }
 282
 283 =head1 NAME
 284
 285 freeside-queued - Job queue daemon
 286
 287 =head1 SYNOPSIS
 288
 289   freeside-queued [ -s | -n ] user
 290
 291 =head1 DESCRIPTION
 292
 293 Job queue daemon.  Should be running at all times.
 294
 295 -s: "secure" jobs only (queued billing jobs)
 296
 297 -n: non-"secure" jobs only (other jobs)
 298
 299 user: Typically "fs_queue"
 300
 301 =head1 VERSION
 302
 303 =head1 BUGS
 304
 305 =head1 SEE ALSO
 306
 307 =cut
 308