FS/bin/freeside-queued

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