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 use POSIX ":sys_wait_h";
  13
  14 use Girocco::Config;
  15 use Girocco::Project;
  16 use Girocco::User;
  17
  18 # Options
  19 my $quiet;
  20 my $progress;
  21 my $kill_after = 900;
  22 my $max_par = 20;
  23 my $max_par_intensive = 1;
  24 my $lockfile = "/tmp/jobd.lock";
  25 my $restart_delay = 60;
  26 my $all_once;
  27 my $one;
  28
  29 ######### Jobs {{{1
  30
  31 sub update_project {
  32         my $job = shift;
  33         my $p = $job->{'project'};
  34         check_project_exists($job) || return;
  35         if (-e get_project_path($p).".nofetch") {
  36                 job_skip($job);
  37                 return setup_gc($job);
  38         }
  39         if (-e get_project_path($p).".clone_in_progress") {
  40                 job_skip($job, "initial mirroring not complete yet");
  41                 return;
  42         }
  43         if (my $ts = is_operation_uptodate($p, 'lastrefresh', $Girocco::Config::min_mirror_interval)) {
  44                 job_skip($job, "not needed right now, last run at $ts");
  45                 setup_gc($job);
  46                 return;
  47         };
  48         exec_job_command($job, ["$Girocco::Config::basedir/jobd/update.sh", $p], $quiet);
  49 }
  50
  51 sub gc_project {
  52         my $job = shift;
  53         my $p = $job->{'project'};
  54         check_project_exists($job) || return;
  55         if (my $ts = is_operation_uptodate($p, 'lastgc', $Girocco::Config::min_gc_interval)) {
  56                 job_skip($job, "not needed right now, last run at $ts");
  57                 return;
  58         }
  59         exec_job_command($job, ["$Girocco::Config::basedir/jobd/gc.sh", $p], $quiet);
  60 }
  61
  62 sub setup_gc {
  63         my $job = shift;
  64         queue_job(
  65                 project => $job->{'project'},
  66                 type => 'gc',
  67                 command => \&gc_project,
  68                 intensive => 1,
  69         );
  70 }
  71
  72 sub check_project_exists {
  73         my $job = shift;
  74         my $p = $job->{'project'};
  75         if (!-d get_project_path($p)) {
  76                 job_skip($job, "non-existent project");
  77                 return 0;
  78         }
  79         1;
  80 }
  81
  82 sub get_project_path {
  83         "$Girocco::Config::reporoot/".shift().".git/";
  84 }
  85
  86 sub is_operation_uptodate {
  87         my ($project, $which, $threshold) = @_;
  88         my $path = get_project_path($project);
  89         my $timestamp = `GIT_DIR="$path" $Girocco::Config::git_bin config "gitweb.$which"`;
  90         my $unix_ts = `date +%s -d "$timestamp"`;
  91         (time - $unix_ts) <= $threshold ? $timestamp : undef;
  92 }
  93
  94 sub queue_one {
  95         my $project = shift;
  96         queue_job(
  97                 project => $project,
  98                 type => 'update',
  99                 command => \&update_project,
 100                 on_success => \&setup_gc,
 101                 on_error => \&setup_gc,
 102         );
 103 }
 104
 105 sub queue_all {
 106         queue_one($_) for (Girocco::Project->get_full_list());
 107 }
 108
 109 ######### Daemon operation {{{1
 110
 111 my @queue;
 112 my @running;
 113 my $perpetual = 1;
 114 my $locked = 0;
 115 my $jobs_executed;
 116 my $jobs_skipped;
 117 my @jobs_killed;
 118
 119 sub handle_softexit {
 120         error("Waiting for outstanding jobs to finish... ".
 121                 "^C again to exit immediately");
 122         @queue = ();
 123         $perpetual = 0;
 124         $SIG{'INT'} = \&handle_exit;
 125 }
 126
 127 sub handle_exit {
 128         error("Killing outstanding jobs...");
 129         $SIG{'TERM'} = 'IGNORE';
 130         for (@running) {
 131                 kill 'KILL', -($_->{'pid'});
 132         }
 133         unlink $lockfile if ($locked);
 134         exit(0);
 135 }
 136
 137 sub queue_job {
 138         my %opts = @_;
 139         $opts{'queued_at'} = time;
 140         $opts{'dont_run'} = 0;
 141         $opts{'intensive'} = 0 unless exists $opts{'intensive'};
 142         push @queue, \%opts;
 143 }
 144
 145 sub run_job {
 146         my $job = shift;
 147
 148         push @running, $job;
 149         $job->{'command'}->($job);
 150         if ($job->{'dont_run'}) {
 151                 pop @running;
 152                 $jobs_skipped++;
 153                 return;
 154         }
 155 }
 156
 157 sub _job_name {
 158         my $job = shift;
 159         "[".$job->{'type'}."::".$job->{'project'}."]";
 160 }
 161
 162 # Only one of those per job!
 163 sub exec_job_command {
 164         my ($job, $command, $err_only) = @_;
 165
 166         my $pid;
 167         if (!defined($pid = fork)) {
 168                 error(_job_name($job) ." Can't fork job: $!");
 169                 $job->{'finished'} = 1;
 170                 return;
 171         }
 172         if (!$pid) {
 173                 open STDIN, '/dev/null' || do {
 174                         error(_job_name($job) ."Can't read from /dev/null: $!");
 175                         $job->{'finished'} = 1;
 176                         return;
 177                 };
 178                 if ($err_only) {
 179                         open STDOUT, '>/dev/null' || do {
 180                                 error(_job_name($job) ." Can't write to /dev/null: $!");
 181                                 $job->{'finished'} = 1;
 182                                 return;
 183                         };
 184                 }
 185                 # New process group so we can keep track of all of its children
 186                 if (!defined(POSIX::setpgid(0, 0))) {
 187                         error(_job_name($job) ." Can't create process group: $!");
 188                         $job->{'finished'} = 1;
 189                         return;
 190                 }
 191                 # "Prevent" races
 192                 select(undef, undef, undef, 0.1);
 193                 exec @$command;
 194                 # Stop perl from complaining
 195                 exit $?;
 196         }
 197         $job->{'pid'} = $pid;
 198         $job->{'finished'} = 0;
 199         $job->{'started_at'} = time;
 200 }
 201
 202 sub job_skip {
 203         my ($job, $msg) = @_;
 204         $job->{'dont_run'} = 1;
 205         error(_job_name($job) ." Skipping job: $msg") unless $quiet || !$msg;
 206 }
 207
 208 sub reap_hanging_jobs {
 209         for (@running) {
 210                 if (defined($_->{'started_at'}) && (time - $_->{'started_at'}) > $kill_after) {
 211                         $_->{'finished'} = 1;
 212                         kill 'KILL', -($_->{'pid'});
 213                         error(_job_name($_) ." KILLED due to timeout");
 214                         push @jobs_killed, _job_name($_);
 215                 }
 216         }
 217 }
 218
 219 sub reap_finished_jobs {
 220         my $pid;
 221         my $finished_any = 0;
 222         while (1) {
 223                 $pid = waitpid(-1, WNOHANG);
 224                 last if $pid < 1;
 225                 $finished_any = 1;
 226
 227                 my @child = grep { $_->{'pid'} && $_->{'pid'} == $pid } @running;
 228                 if ($?) {
 229                         # XXX- we currently don't care
 230                 }
 231                 if (@child && !$child[0]->{'finished'}) {
 232                         $child[0]->{'on_success'}->($child[0]) if defined($child[0]->{'on_success'});
 233                         $child[0]->{'finished'} = 1;
 234                         $jobs_executed++;
 235                 } elsif (@child) {
 236                         $child[0]->{'on_error'}->($child[0]) if defined($child[0]->{'on_error'});
 237                 }
 238         }
 239         @running = grep { $_->{'finished'} == 0 } @running;
 240         $finished_any;
 241 }
 242
 243 sub have_intensive_jobs {
 244         grep { $_->{'intensive'} == 1 } @running;
 245 }
 246
 247 sub ts {
 248         "[". scalar(localtime) ."] ";
 249 }
 250
 251 sub run_queue {
 252         my $last_progress = time;
 253         $jobs_executed = 0;
 254         $jobs_skipped = 0;
 255         @jobs_killed = ();
 256         if ($progress) {
 257                 error("--- Processing %d queued jobs", scalar(@queue));
 258         }
 259         $SIG{'INT'} = \&handle_softexit;
 260         $SIG{'TERM'} = \&handle_exit;
 261         while (@queue || @running) {
 262                 reap_hanging_jobs();
 263                 my $proceed_immediately = reap_finished_jobs();
 264                 # Status output
 265                 if ($progress && (time - $last_progress) >= 60) {
 266                         ferror("STATUS: %d queued, %d running, %d finished, %d skipped, %d killed", scalar(@queue), scalar(@running), $jobs_executed, $jobs_skipped, scalar(@jobs_killed));
 267                         if (@running) {
 268                                 my @run_status;
 269                                 for (@running) {
 270                                         push @run_status, _job_name($_)." ". (time - $_->{'started_at'}) ."s";
 271                                 }
 272                                 error("STATUS: currently running: ". join(', ', @run_status));
 273                         }
 274                         $last_progress = time;
 275                 }
 276                 # Back off if we're too busy
 277                 if (@running >= $max_par || have_intensive_jobs() >= $max_par_intensive || !@queue) {
 278                         sleep 1 unless $proceed_immediately;
 279                         next;
 280                 }
 281                 # Run next
 282                 run_job(shift(@queue)) if @queue;
 283         }
 284         if ($progress) {
 285                 ferror("--- Queue processed. %d jobs executed, %d skipped, %d killed.", $jobs_executed, $jobs_skipped, scalar(@jobs_killed));
 286         }
 287 }
 288
 289 sub run_perpetually {
 290         if (-e $lockfile) {
 291                 die "Lockfile exists. Please make sure no other instance of jobd is running.";
 292         }
 293         open LOCK, '>', $lockfile || die "Cannot create lockfile $lockfile: $!";
 294         print LOCK $$;
 295         close LOCK;
 296         $locked = 1;
 297
 298         while ($perpetual) {
 299                 queue_all();
 300                 run_queue();
 301                 sleep($restart_delay) if $perpetual; # Let the system breathe for a moment
 302         }
 303         unlink $lockfile;
 304 }
 305
 306 ######### Helpers {{{1
 307
 308 sub error($) {
 309         print STDERR ts().shift()."\n";
 310 }
 311 sub ferror(@) {
 312         error(sprintf(@_));
 313 }
 314 sub fatal($) {
 315         error(shift);
 316         exit 1;
 317 }
 318
 319 ######### Main {{{1
 320
 321 # Parse options
 322 my $parse_res = GetOptions(
 323         'help|?' => sub { pod2usage(-verbose => 1, -exitval => 0); },
 324         'quiet|q' => \$quiet,
 325         'progress|P' => \$progress,
 326         'kill-after|k=i' => \$kill_after,
 327         'max-parallel|p=i' => \$max_par,
 328         'max-intensive-parallel|i=i' => \$max_par_intensive,
 329         'restart-delay|d=i' => \$restart_delay,
 330         'lockfile|l=s' => \$lockfile,
 331         'all-once|a' => \$all_once,
 332         'one|o=s' => \$one,
 333 ) || pod2usage(2);
 334 fatal("Error: can only use one out of --all-once and --one")
 335         if ($all_once && $one);
 336
 337 unless ($quiet) {
 338         $ENV{'show_progress'} = '1';
 339         $progress = 1;
 340 }
 341
 342 if ($one) {
 343         queue_one($one);
 344         run_queue();
 345         exit;
 346 }
 347
 348 if ($all_once) {
 349         queue_all();
 350         run_queue();
 351         exit;
 352 }
 353
 354 run_perpetually();
 355
 356 ########## Documentation {{{1
 357
 358 __END__
 359
 360 =head1 NAME
 361
 362 jobd - Perform Girocco maintenance jobs
 363
 364 =head1 SYNOPSIS
 365
 366 jobd [options]
 367
 368  Options:
 369    -h | --help                           detailed instructions
 370    -q | --quiet                          run quietly
 371    -P | --progress                       show occasional status updates
 372    -k SECONDS | --kill-after SECONDS     how long to wait before killing jobs
 373    -p NUM | --max-parallel NUM           how many jobs to run at the same time
 374    -i NUM | --max-intensive-parallel NUM how many resource-hungry jobs to run
 375                                          at the same time
 376    -d NUM | --restart-delay SECONDS      wait for this many seconds between
 377                                          queue runs
 378    -l FILE | --lockfile FILE             create a lockfile in the given
 379                                          location
 380    -a | --all-once                       process the list only once
 381    -o PRJNAME | --one PRJNAME            process only one project
 382
 383 =head1 OPTIONS
 384
 385 =over 8
 386
 387 =item B<--help>
 388
 389 Print the full description of jobd's options.
 390
 391 =item B<--quiet>
 392
 393 Suppress non-error messages, e.g. for use when running this task as a cronjob.
 394
 395 =item B<--progress>
 396
 397 Show information about the current status of the job queue occasionally. This
 398 is automatically enabled if --quiet is not given.
 399
 400 =item B<--kill-after SECONDS>
 401
 402 Kill supervised jobs after a certain time to avoid hanging the daemon.
 403
 404 =item B<--max-parallel NUM>
 405
 406 Run no more than that many jobs at the same time.
 407
 408 =item B<--max-intensive-parallel NUM>
 409
 410 Run no more than that many resource-hungry jobs at the same time. Right now,
 411 this refers to repacking jobs.
 412
 413 =item B<--restart-delay NUM>
 414
 415 After processing the queue, wait this many seconds until the queue is
 416 restarted.
 417
 418 =item B<--lockfile FILE>
 419
 420 For perpetual operation, create a lockfile in that place and clean it up after
 421 finishing/aborting.
 422
 423 =item B<--all-once>
 424
 425 Instead of perpetuously processing all projects over and over again, process
 426 them just once and then exit.
 427
 428 =item B<--one PRJNAME>
 429
 430 Process only the given project (given as just the project name without C<.git>
 431 suffix) and then exit.
 432
 433 =back
 434
 435 =head1 DESCRIPTION
 436
 437 jobd is Girocco's repositories maintenance servant; it periodically checks all
 438 the repositories and updates mirrored repositories and repacks push-mode
 439 repositories when needed.
 440
 441 =cut