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