FS/bin/freeside-queued

   1 #!/usr/bin/perl -w
   2
   3 use strict;
   4 use vars qw( $log_file $sigterm $sigint $kids $max_kids );
   5 use subs qw( _die _logmsg );
   6 use Fcntl qw(:flock);
   7 use POSIX qw(setsid);
   8 use Date::Format;
   9 use IO::File;
  10 use FS::UID qw(adminsuidsetup forksuidsetup driver_name dbh);
  11 use FS::Record qw(qsearch qsearchs);
  12 use FS::queue;
  13 use FS::queue_depend;
  14
  15 # no autoloading just yet
  16 use FS::cust_main;
  17 use FS::svc_acct;
  18 use Net::SSH 0.06;
  19 use FS::part_export;
  20
  21 $max_kids = '10'; #guess it should be a config file...
  22 $kids = 0;
  23
  24 my $user = shift or die &usage;
  25
  26 #my $pid_file = "/var/run/freeside-queued.$user.pid";
  27 my $pid_file = "/var/run/freeside-queued.pid";
  28
  29 &daemonize1;
  30
  31 sub REAPER { my $pid = wait; $SIG{CHLD} = \&REAPER; $kids--; }
  32 $SIG{CHLD} =  \&REAPER;
  33
  34 $sigterm = 0;
  35 $sigint = 0;
  36 $SIG{INT} = sub { warn "SIGINT received; shutting down\n"; $sigint++; };
  37 $SIG{TERM} = sub { warn "SIGTERM received; shutting down\n"; $sigterm++; };
  38
  39 $< = $FS::UID::freeside_uid;
  40
  41 #freebsd is sofa king broken, won't setuid()
  42 $> = $FS::UID::freeside_uid;
  43 ($<,$>) = ($>,$<);
  44 $> = $FS::UID::freeside_uid;
  45
  46 $ENV{HOME} = (getpwuid($>))[7]; #for ssh
  47 adminsuidsetup $user;
  48
  49 $log_file = "/usr/local/etc/freeside/queuelog.". $FS::UID::datasrc;
  50
  51 &daemonize2;
  52
  53 $SIG{__DIE__} = \&_die;
  54 $SIG{__WARN__} = \&_logmsg;
  55
  56 warn "freeside-queued starting\n";
  57
  58 my $warnkids=0;
  59 while (1) {
  60
  61   #prevent runaway forking
  62   if ( $kids >= $max_kids ) {
  63     warn "WARNING: maximum $kids children reached\n" unless $warnkids++;
  64     sleep 1; #waiting for signals is cheap
  65     next;
  66   }
  67   $warnkids=0;
  68
  69   my $nodepend = driver_name eq 'mysql'
  70    ? ''
  71    : 'AND 0 = ( SELECT COUNT(*) FROM queue_depend'.
  72      ' WHERE queue_depend.jobnum = queue.jobnum ) ';
  73
  74   #my($job, $ljob);
  75   #{
  76   #  my $oldAutoCommit = $FS::UID::AutoCommit;
  77   #  local $FS::UID::AutoCommit = 0;
  78   $FS::UID::AutoCommit = 0;
  79   my $dbh = dbh;
  80
  81   my $job = qsearchs(
  82     'queue',
  83     { 'status' => 'new' },
  84     '',
  85     driver_name eq 'mysql'
  86       ? "$nodepend ORDER BY jobnum LIMIT 1 FOR UPDATE"
  87       : "$nodepend ORDER BY jobnum FOR UPDATE LIMIT 1"
  88   ) or do {
  89     $dbh->commit or die $dbh->errstr; #if $oldAutoCommit;
  90     sleep 5; #connecting to db is expensive
  91     next;
  92   };
  93
  94   if ( driver_name eq 'mysql'
  95        && qsearch('queue_depend', { 'jobnum' => $job->jobnum } ) ) {
  96     $dbh->commit or die $dbh->errstr; #if $oldAutoCommit;
  97     sleep 5; #would be better if mysql could do everything in query above
  98     next;
  99   }
 100
 101   my %hash = $job->hash;
 102   $hash{'status'} = 'locked';
 103   my $ljob = new FS::queue ( \%hash );
 104   my $error = $ljob->replace($job);
 105   die $error if $error;
 106
 107   $dbh->commit or die $dbh->errstr; #if $oldAutoCommit;
 108
 109   $FS::UID::AutoCommit = 1;
 110   #}
 111
 112   my @args = $ljob->args;
 113
 114   defined( my $pid = fork ) or do {
 115     warn "WARNING: can't fork: $!\n";
 116     my %hash = $job->hash;
 117     $hash{'status'} = 'failed';
 118     $hash{'statustext'} = "[freeside-queued] can't fork: $!";
 119     my $ljob = new FS::queue ( \%hash );
 120     my $error = $ljob->replace($job);
 121     die $error if $error;
 122     next; #don't increment the kid counter
 123   };
 124
 125   if ( $pid ) {
 126     $kids++;
 127   } else { #kid time
 128
 129     #get new db handle
 130     $FS::UID::dbh->{InactiveDestroy} = 1;
 131
 132     forksuidsetup($user);
 133
 134     #auto-use export classes...
 135     if ( $ljob->job =~ /(FS::part_export::\w+)::/ ) {
 136       my $class = $1;
 137       eval "use $class;";
 138       if ( $@ ) {
 139         warn "job use $class failed";
 140         my %hash = $ljob->hash;
 141         $hash{'status'} = 'failed';
 142         $hash{'statustext'} = $@;
 143         my $fjob = new FS::queue( \%hash );
 144         my $error = $fjob->replace($ljob);
 145         die $error if $error;
 146         exit; #end-of-kid
 147       };
 148     }
 149
 150     my $eval = "&". $ljob->job. '(@args);';
 151     warn "running $eval";
 152     eval $eval; #throw away return value?  suppose so
 153     if ( $@ ) {
 154       warn "job $eval failed";
 155       my %hash = $ljob->hash;
 156       $hash{'status'} = 'failed';
 157       $hash{'statustext'} = $@;
 158       my $fjob = new FS::queue( \%hash );
 159       my $error = $fjob->replace($ljob);
 160       die $error if $error;
 161     } else {
 162       $ljob->delete;
 163     }
 164
 165     exit;
 166     #end-of-kid
 167   }
 168
 169 } continue {
 170   if ( $sigterm ) {
 171     warn "received TERM signal; exiting\n";
 172     exit;
 173   }
 174   if ( $sigint ) {
 175     warn "received INT signal; exiting\n";
 176     exit;
 177   }
 178 }
 179
 180 sub usage {
 181   die "Usage:\n\n  freeside-queued user\n";
 182 }
 183
 184 sub _die {
 185   my $msg = shift;
 186   unlink $pid_file if -e $pid_file;
 187   _logmsg($msg);
 188 }
 189
 190 sub _logmsg {
 191   chomp( my $msg = shift );
 192   my $log = new IO::File ">>$log_file";
 193   flock($log, LOCK_EX);
 194   seek($log, 0, 2);
 195   print $log "[". time2str("%a %b %e %T %Y",time). "] [$$] $msg\n";
 196   flock($log, LOCK_UN);
 197   close $log;
 198 }
 199
 200 sub daemonize1 {
 201
 202   chdir "/" or die "Can't chdir to /: $!";
 203   open STDIN, '/dev/null'   or die "Can't read /dev/null: $!";
 204   defined(my $pid = fork) or die "Can't fork: $!";
 205   if ( $pid ) {
 206     print "freeside-queued started with pid $pid\n"; #logging to $log_file\n";
 207     exit unless $pid_file;
 208     my $pidfh = new IO::File ">$pid_file" or exit;
 209     print $pidfh "$pid\n";
 210     exit;
 211   }
 212   #open STDOUT, '>/dev/null'
 213   #                          or die "Can't write to /dev/null: $!";
 214   #setsid                  or die "Can't start a new session: $!";
 215   #open STDERR, '>&STDOUT' or die "Can't dup stdout: $!";
 216
 217 }
 218
 219 sub daemonize2 {
 220   open STDOUT, '>/dev/null'
 221                             or die "Can't write to /dev/null: $!";
 222   setsid                  or die "Can't start a new session: $!";
 223   open STDERR, '>&STDOUT' or die "Can't dup stdout: $!";
 224 }
 225
 226 =head1 NAME
 227
 228 freeside-queued - Job queue daemon
 229
 230 =head1 SYNOPSIS
 231
 232   freeside-queued user
 233
 234 =head1 DESCRIPTION
 235
 236 Job queue daemon.  Should be running at all times.
 237
 238 user: from the mapsecrets file - see config.html from the base documentation
 239
 240 =head1 VERSION
 241
 242 =head1 BUGS
 243
 244 =head1 SEE ALSO
 245
 246 =cut
 247