jobd/jobd.pl

   1 #!/usr/bin/perl
   2 #
   3 # jobd - perform Girocco maintenance jobs
   4 #
   5 # Run with --help for details
   6
   7 use strict;
   8 use warnings;
   9
  10 use Getopt::Long;
  11 use Pod::Usage;
  12
  13 use Girocco::Config;
  14 use Girocco::Project;
  15 use Girocco::User;
  16
  17 # Options
  18 my $quiet;
  19 my $progress;
  20 my $kill_after = 900;
  21 my $max_par = 3;
  22 my $lockfile = "/tmp/jobd.lock";
  23 my $all_once;
  24 my $one;
  25
  26 ######### Jobs {{{1
  27
  28 sub update_project {
  29         my $job = shift;
  30         my $p = $job->{'project'};
  31         check_project_exists($job) || return;
  32         (-e "$Girocco::Config::reporoot/$p.git/.nofetch") && do {
  33                 job_skip($job);
  34                 return;
  35         };
  36         exec_job_command($job, ["$Girocco::Config::basedir/jobd/update.sh", $p], $quiet);
  37 }
  38
  39 sub gc_project {
  40         my $job = shift;
  41         my $p = $job->{'project'};
  42         check_project_exists($job) || return;
  43         exec_job_command($job, ["$Girocco::Config::basedir/jobd/gc.sh", $p], $quiet);
  44 }
  45
  46 sub setup_gc {
  47         my $job = shift;
  48         queue_job(
  49                 project => $job->{'project'},
  50                 type => 'gc',
  51                 command => \&gc_project,
  52         );
  53 }
  54
  55 sub check_project_exists {
  56         my $job = shift;
  57         my $p = $job->{'project'};
  58         if (!-d "$Girocco::Config::reporoot/$p.git") {
  59                 error("Warning: skipping non-existent project: $job->{project}")
  60                         unless $quiet;
  61                 job_skip();
  62                 return 0;
  63         }
  64         1;
  65 }
  66
  67 sub queue_one {
  68         my $project = shift;
  69         queue_job(
  70                 project => $project,
  71                 type => 'update',
  72                 command => \&update_project,
  73                 on_success => \&setup_gc,
  74                 on_error => \&setup_gc,
  75         );
  76 }
  77
  78 sub queue_all {
  79         queue_one($_) for (Girocco::Project->get_full_list());
  80 }
  81
  82 ######### Daemon operation {{{1
  83
  84 my @queue;
  85 my @running;
  86 my $perpetual = 1;
  87 my $locked = 0;
  88 my $jobs_executed;
  89 my @jobs_killed;
  90
  91 sub handle_softexit {
  92         error("Waiting for outstanding jobs to finish... ".
  93                 "^C again to exit immediately");
  94         @queue = ();
  95         $perpetual = 0;
  96         $SIG{'INT'} = \&handle_exit;
  97 }
  98
  99 sub handle_exit {
 100         error("Killing outstanding jobs...");
 101         $SIG{'CHLD'} = 'IGNORE';
 102         $SIG{'TERM'} = 'IGNORE';
 103         for (@running) {
 104                 kill 'KILL', $_->{'pid'};
 105         }
 106         unlink $lockfile if ($locked);
 107         exit(0);
 108 }
 109
 110 sub handle_childgone {
 111         my $pid = wait;
 112         if ($pid != -1) {
 113                 my $child = grep { $_->{'pid'} == $pid } @running;
 114                 if ($?) {
 115                         # XXX- we currently don't care
 116                 }
 117                 $child->{'finished'} = 2;
 118                 $jobs_executed++;
 119         }
 120         # Just to be safe
 121         $SIG{'CHLD'} = \&handle_childgone;
 122 }
 123
 124 sub queue_job {
 125         my %opts = @_;
 126         $opts{'queued_at'} = time;
 127         push @queue, \%opts;
 128 }
 129
 130 sub run_job {
 131         my $job = shift;
 132
 133         push @running, $job;
 134         $job->{'command'}->($job);
 135 }
 136
 137 sub _job_name {
 138         my $job = shift;
 139         "[".$job->{'type'}."::".$job->{'project'}."]";
 140 }
 141
 142 # Only one of those per job!
 143 sub exec_job_command {
 144         my ($job, $command, $err_only) = @_;
 145
 146         my $pid;
 147         if (!defined($pid = fork)) {
 148                 error(_job_name($job) ." Can't fork job: $!");
 149                 $job->{'finished'} = 1;
 150                 return;
 151         }
 152         if (!$pid) {
 153                 if ($err_only) {
 154                         open STDOUT, '>/dev/null' || do {
 155                                 error(_job_name($job) ." Can't write to /dev/null: $!");
 156                                 $job->{'finished'} = 1;
 157                                 return;
 158                         };
 159                 }
 160                 exec @$command;
 161                 exit $?;
 162         }
 163         $job->{'pid'} = $pid;
 164         $job->{'finished'} = 0;
 165         $job->{'started_at'} = time;
 166 }
 167
 168 sub job_skip {
 169         my $job = shift;
 170         exec_job_command($job, ['/bin/false']);
 171 }
 172
 173 sub reap_hanging_jobs {
 174         for (@running) {
 175                 if ((time - $_->{'started_at'}) > $kill_after) {
 176                         $_->{'finished'} = 1;
 177                         kill 'KILL', $_->{'pid'};
 178                         print STDERR _job_name($_) ." KILLED due to timeout\n";
 179                         push @jobs_killed, _job_name($_);
 180                 }
 181         }
 182 }
 183
 184 sub reap_finished_jobs {
 185         for (@running) {
 186                 my $status = $_->{'finished'};
 187                 if ($status == 0) { next; }
 188                 elsif ($status == 1 && defined($_->{'on_error'})) {
 189                         $_->{'on_error'}->($_);
 190                 } elsif ($status == 2 && defined($_->{'on_success'})) {
 191                         $_->{'on_success'}->($_);
 192                 }
 193         }
 194         @running = grep { $_->{'finished'} == 0 } @running;
 195 }
 196
 197 sub run_queue {
 198         my $queue_steps = 0;
 199         $jobs_executed = 0;
 200         @jobs_killed = ();
 201         unless ($quiet) {
 202                 printf STDERR "--- Processing %d queued jobs\n", scalar(@queue);
 203         }
 204         $SIG{'CHLD'} = \&handle_childgone;
 205         while (@queue || @running) {
 206                 reap_hanging_jobs();
 207                 reap_finished_jobs();
 208                 # Back off if we're too busy
 209                 if (@running >= $max_par) {
 210                         sleep 10;
 211                         $queue_steps++;
 212                         unless (($quiet && !$progress) || ($queue_steps % 10)) {
 213                                 printf STDERR "STATUS: %d queued, %d running, %d finished, %d killed\n", scalar(@queue), scalar(@running), $jobs_executed, scalar(@jobs_killed);
 214                         }
 215                         next;
 216                 }
 217                 # Run next
 218                 run_job(shift(@queue)) if @queue;
 219         }
 220         unless ($quiet) {
 221                 printf STDERR "--- Queue processed. %d jobs executed, %d killed due to timeouts. Now restarting.\n", $jobs_executed, scalar(@jobs_killed);
 222         }
 223 }
 224
 225 sub run_perpetually {
 226         if (-e $lockfile) {
 227                 die "Lockfile exists. Please make sure no other instance of jobd is running.";
 228         }
 229         open LOCK, '>', $lockfile || die "Cannot create lockfile $lockfile: $!";
 230         print LOCK $$;
 231         close LOCK;
 232         $locked = 1;
 233
 234         while ($perpetual) {
 235                 queue_all();
 236                 run_queue();
 237         }
 238         unlink $lockfile;
 239 }
 240
 241 ######### Helpers {{{1
 242
 243 sub error($) {
 244         print STDERR shift()."\n";
 245 }
 246 sub fatal($) {
 247         error(shift);
 248         exit 1;
 249 }
 250
 251 ######### Main {{{1
 252
 253 # Parse options
 254 Getopt::Long::Configure('bundling');
 255 my $parse_res = GetOptions(
 256         'help|?' => sub { pod2usage(-verbose => 1, -exitval => 0); },
 257         'quiet|q' => \$quiet,
 258         'progress|P' => \$progress,
 259         'kill-after|k=i' => \$kill_after,
 260         'max-parallel|p=i' => \$max_par,
 261         'lockfile|l=s' => \$lockfile,
 262         'all-once|a' => \$all_once,
 263         'one|o=s' => \$one,
 264 ) || pod2usage(2);
 265 fatal("Error: can only use one out of --all-once and --one")
 266         if ($all_once && $one);
 267
 268 unless ($quiet) {
 269         $ENV{'show_progress'} = '1';
 270 }
 271
 272 if ($one) {
 273         queue_one($one);
 274         run_queue();
 275         exit;
 276 }
 277
 278 if ($all_once) {
 279         queue_all();
 280         run_queue();
 281         exit;
 282 }
 283
 284 run_perpetually();
 285
 286 ########## Documentation {{{1
 287
 288 __END__
 289
 290 =head1 NAME
 291
 292 jobd - Perform Girocco maintenance jobs
 293
 294 =head1 SYNOPSIS
 295
 296 jobd [options]
 297
 298  Options:
 299    -h | --help                         detailed instructions
 300    -q | --quiet                        run quietly
 301    -P | --progress                     show occasional status updates
 302    -k SECONDS | --kill-after=SECONDS   how long to wait before killing jobs
 303    -p NUM | --max-parallel=NUM         how many jobs to run at the same time
 304    -l FILE | --lockfile=FILE           create a lockfile in the given location
 305    -a | --all-once                     process the list only once
 306    -o PRJNAME | --one=PRJNAME          process only one project
 307
 308 =head1 OPTIONS
 309
 310 =over 8
 311
 312 =item B<--help>
 313
 314 Print the full description of jobd's options.
 315
 316 =item B<--quiet>
 317
 318 Suppress non-error messages, e.g. for use when running this task as a cronjob.
 319
 320 =item B<--progress>
 321
 322 Show information about the current status of the job queue occasionally. This
 323 is automatically enabled if --quiet is not given.
 324
 325 =item B<--kill-after=SECONDS>
 326
 327 Kill supervised jobs after a certain time to avoid hanging the daemon.
 328
 329 =item B<--max-parallel=NUM>
 330
 331 Run no more than that many jobs at the same time.
 332
 333 =item B<--lockfile=FILE>
 334
 335 For perpetual operation, create a lockfile in that place and clean it up after
 336 finishing/aborting.
 337
 338 =item B<--all-once>
 339
 340 Instead of perpetuously processing all projects over and over again, process
 341 them just once and then exit.
 342
 343 =item B<--one=PRJNAME>
 344
 345 Process only the given project (given as just the project name without C<.git>
 346 suffix) and then exit.
 347
 348 =back
 349
 350 =head1 DESCRIPTION
 351
 352 jobd is Girocco's repositories maintenance servant; it periodically checks all
 353 the repositories and updates mirrored repositories and repacks push-mode
 354 repositories when needed.
 355
 356 =cut