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