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