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 myconnect);
  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
  55 $@ = 'not connected';
  56 while ( $@ ) {
  57   eval { adminsuidsetup $user; };
  58   if ( $@ ) {
  59     warn $@;
  60     warn "sleeping for reconnect...\n";
  61     sleep 5;
  62   }
  63 }
  64
  65 $log_file = "/usr/local/etc/freeside/queuelog.". $FS::UID::datasrc;
  66
  67 &daemonize2;
  68
  69 $SIG{__DIE__} = \&_die;
  70 $SIG{__WARN__} = \&_logmsg;
  71
  72 warn "freeside-queued starting\n";
  73
  74 my $warnkids=0;
  75 while (1) {
  76
  77   &reap_kids;
  78   #prevent runaway forking
  79   if ( $kids >= $max_kids ) {
  80     warn "WARNING: maximum $kids children reached\n" unless $warnkids++;
  81     &reap_kids;
  82     sleep 1; #waiting for signals is cheap
  83     next;
  84   }
  85   $warnkids=0;
  86
  87   unless ( dbh && dbh->ping ) {
  88     warn "WARNING: connection to database lost, reconnecting...\n";
  89
  90     eval { $FS::UID::dbh = myconnect; };
  91
  92     unless ( !$@ && dbh && dbh->ping ) {
  93       warn "WARNING: still no connection to database, sleeping for retry...\n";
  94       sleep 10;
  95       next;
  96     } else {
  97       warn "WARNING: reconnected to database\n";
  98     }
  99   }
 100
 101   #my($job, $ljob);
 102   #{
 103   #  my $oldAutoCommit = $FS::UID::AutoCommit;
 104   #  local $FS::UID::AutoCommit = 0;
 105   $FS::UID::AutoCommit = 0;
 106
 107   #assuming mysql 4.1 w/subqueries now
 108   #my $nodepend = driver_name eq 'mysql'
 109   # ? ''
 110   # : 'AND 0 = ( SELECT COUNT(*) FROM queue_depend'.
 111   #   ' WHERE queue_depend.jobnum = queue.jobnum ) ';
 112   my $nodepend = 'AND 0 = ( SELECT COUNT(*) FROM queue_depend'.
 113                  '           WHERE queue_depend.jobnum = queue.jobnum ) ';
 114
 115   my $job = qsearchs(
 116     'queue',
 117     { 'status' => 'new' },
 118     '',
 119     driver_name eq 'mysql'
 120       ? "$nodepend ORDER BY jobnum LIMIT 1 FOR UPDATE"
 121       : "$nodepend ORDER BY jobnum FOR UPDATE LIMIT 1"
 122   ) or do {
 123     # if $oldAutoCommit {
 124     dbh->commit or do {
 125       warn "WARNING: database error, closing connection: ". dbh->errstr;
 126       undef $FS::UID::dbh;
 127       next;
 128     };
 129     # }
 130     sleep 5; #connecting to db is expensive
 131     next;
 132   };
 133
 134   #assuming mysql 4.1 w/subqueries now
 135   #if ( driver_name eq 'mysql'
 136   #     && qsearch('queue_depend', { 'jobnum' => $job->jobnum } ) ) {
 137   #  dbh->commit or die dbh->errstr; #if $oldAutoCommit;
 138   #  sleep 5; #would be better if mysql could do everything in query above
 139   #  next;
 140   #}
 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   # if $oldAutoCommit {
 154   dbh->commit or do {
 155     warn "WARNING: database error, closing connection: ". dbh->errstr;
 156     undef $FS::UID::dbh;
 157     next;
 158   };
 159   # }
 160
 161   $FS::UID::AutoCommit = 1;
 162   #}
 163
 164   my @args = $ljob->args;
 165   splice @args, 0, 1, $ljob if $args[0] eq '_JOB';
 166
 167   defined( my $pid = fork ) or do {
 168     warn "WARNING: can't fork: $!\n";
 169     my %hash = $job->hash;
 170     $hash{'status'} = 'failed';
 171     $hash{'statustext'} = "[freeside-queued] can't fork: $!";
 172     my $ljob = new FS::queue ( \%hash );
 173     my $error = $ljob->replace($job);
 174     die $error if $error;
 175     next; #don't increment the kid counter
 176   };
 177
 178   if ( $pid ) {
 179     $kids++;
 180     $kids{$pid} = 1;
 181   } else { #kid time
 182
 183     #get new db handle
 184     $FS::UID::dbh->{InactiveDestroy} = 1;
 185
 186     forksuidsetup($user);
 187
 188     #auto-use classes...
 189     #if ( $ljob->job =~ /(FS::part_export::\w+)::/ ) {
 190     if (    $ljob->job =~ /(FS::part_export::\w+)::/
 191          || $ljob->job =~ /(FS::\w+)::/
 192        )
 193     {
 194       my $class = $1;
 195       eval "use $class;";
 196       if ( $@ ) {
 197         warn "job use $class failed";
 198         my %hash = $ljob->hash;
 199         $hash{'status'} = 'failed';
 200         $hash{'statustext'} = $@;
 201         my $fjob = new FS::queue( \%hash );
 202         my $error = $fjob->replace($ljob);
 203         die $error if $error;
 204         exit; #end-of-kid
 205       };
 206     }
 207
 208     my $eval = "&". $ljob->job. '(@args);';
 209     warn "running $eval";
 210     eval $eval; #throw away return value?  suppose so
 211     if ( $@ ) {
 212       warn "job $eval failed";
 213       my %hash = $ljob->hash;
 214       $hash{'status'} = 'failed';
 215       $hash{'statustext'} = $@;
 216       my $fjob = new FS::queue( \%hash );
 217       my $error = $fjob->replace($ljob);
 218       die $error if $error;
 219     } else {
 220       $ljob->delete;
 221     }
 222
 223     exit;
 224     #end-of-kid
 225   }
 226
 227 } continue {
 228   if ( $sigterm ) {
 229     warn "received TERM signal; exiting\n";
 230     exit;
 231   }
 232   if ( $sigint ) {
 233     warn "received INT signal; exiting\n";
 234     exit;
 235   }
 236 }
 237
 238 sub usage {
 239   die "Usage:\n\n  freeside-queued user\n";
 240 }
 241
 242 sub _die {
 243   my $msg = shift;
 244   unlink $pid_file if -e $pid_file;
 245   _logmsg($msg);
 246 }
 247
 248 sub _logmsg {
 249   chomp( my $msg = shift );
 250   my $log = new IO::File ">>$log_file";
 251   flock($log, LOCK_EX);
 252   seek($log, 0, 2);
 253   print $log "[". time2str("%a %b %e %T %Y",time). "] [$$] $msg\n";
 254   flock($log, LOCK_UN);
 255   close $log;
 256 }
 257
 258 sub daemonize1 {
 259
 260   chdir "/" or die "Can't chdir to /: $!";
 261   open STDIN, '/dev/null'   or die "Can't read /dev/null: $!";
 262   defined(my $pid = fork) or die "Can't fork: $!";
 263   if ( $pid ) {
 264     print "freeside-queued started with pid $pid\n"; #logging to $log_file\n";
 265     exit unless $pid_file;
 266     my $pidfh = new IO::File ">$pid_file" or exit;
 267     print $pidfh "$pid\n";
 268     exit;
 269   }
 270   #open STDOUT, '>/dev/null'
 271   #                          or die "Can't write to /dev/null: $!";
 272   #setsid                  or die "Can't start a new session: $!";
 273   #open STDERR, '>&STDOUT' or die "Can't dup stdout: $!";
 274
 275 }
 276
 277 sub daemonize2 {
 278   open STDOUT, '>/dev/null'
 279                             or die "Can't write to /dev/null: $!";
 280   setsid                  or die "Can't start a new session: $!";
 281   open STDERR, '>&STDOUT' or die "Can't dup stdout: $!";
 282 }
 283
 284 sub reap_kids {
 285   foreach my $pid ( keys %kids ) {
 286     my $kid = waitpid($pid, WNOHANG);
 287     if ( $kid > 0 ) {
 288       $kids--;
 289       delete $kids{$kid};
 290     }
 291   }
 292 }
 293
 294 =head1 NAME
 295
 296 freeside-queued - Job queue daemon
 297
 298 =head1 SYNOPSIS
 299
 300   freeside-queued user
 301
 302 =head1 DESCRIPTION
 303
 304 Job queue daemon.  Should be running at all times.
 305
 306 user: from the mapsecrets file - see config.html from the base documentation
 307
 308 =head1 VERSION
 309
 310 =head1 BUGS
 311
 312 =head1 SEE ALSO
 313
 314 =cut
 315