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                 error("--- 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 my $parse_res = GetOptions(
 350         'help|?' => sub { pod2usage(-verbose => 1, -exitval => 0); },
 351         'quiet|q' => \$quiet,
 352         'progress|P' => \$progress,
 353         'kill-after|k=i' => \$kill_after,
 354         'max-parallel|p=i' => \$max_par,
 355         'max-intensive-parallel|i=i' => \$max_par_intensive,
 356         'load-triggers=f{2}' => \@load_triggers,
 357         'restart-delay|d=i' => \$restart_delay,
 358         'lockfile|l=s' => \$lockfile,
 359         'all-once|a' => \$all_once,
 360         'one|o=s' => \$one,
 361 ) || pod2usage(2);
 362 fatal("Error: can only use one out of --all-once and --one")
 363         if ($all_once && $one);
 364
 365 unless ($quiet) {
 366         $ENV{'show_progress'} = '1';
 367         $progress = 1;
 368 }
 369
 370 @load_triggers = (0, 0) if (!-f '/proc/loadavg');
 371 ($load_trig, $load_untrig) = @load_triggers;
 372
 373 if ($one) {
 374         queue_one($one);
 375         run_queue();
 376         exit;
 377 }
 378
 379 if ($all_once) {
 380         queue_all();
 381         run_queue();
 382         exit;
 383 }
 384
 385 run_perpetually();
 386
 387 ########## Documentation {{{1
 388
 389 __END__
 390
 391 =head1 NAME
 392
 393 jobd - Perform Girocco maintenance jobs
 394
 395 =head1 SYNOPSIS
 396
 397 jobd [options]
 398
 399  Options:
 400    -h | --help                           detailed instructions
 401    -q | --quiet                          run quietly
 402    -P | --progress                       show occasional status updates
 403    -k SECONDS | --kill-after SECONDS     how long to wait before killing jobs
 404    -p NUM | --max-parallel NUM           how many jobs to run at the same time
 405    -i NUM | --max-intensive-parallel NUM how many resource-hungry jobs to run
 406                                          at the same time
 407    --load-triggers TRIG UNTRIG           stop queueing jobs at load above
 408                                          TRIG and resume at load below UNTRIG
 409    -d NUM | --restart-delay SECONDS      wait for this many seconds between
 410                                          queue runs
 411    -l FILE | --lockfile FILE             create a lockfile in the given
 412                                          location
 413    -a | --all-once                       process the list only once
 414    -o PRJNAME | --one PRJNAME            process only one project
 415
 416 =head1 OPTIONS
 417
 418 =over 8
 419
 420 =item B<--help>
 421
 422 Print the full description of jobd's options.
 423
 424 =item B<--quiet>
 425
 426 Suppress non-error messages, e.g. for use when running this task as a cronjob.
 427
 428 =item B<--progress>
 429
 430 Show information about the current status of the job queue occasionally. This
 431 is automatically enabled if --quiet is not given.
 432
 433 =item B<--kill-after SECONDS>
 434
 435 Kill supervised jobs after a certain time to avoid hanging the daemon.
 436
 437 =item B<--max-parallel NUM>
 438
 439 Run no more than that many jobs at the same time.
 440
 441 =item B<--max-intensive-parallel NUM>
 442
 443 Run no more than that many resource-hungry jobs at the same time. Right now,
 444 this refers to repacking jobs.
 445
 446 =item B<--load-triggers TRIG UNTRIG>
 447
 448 If the first system load average (1 minute average) exceeds TRIG, don't queue
 449 any more jobs until it goes below UNTRIG. This is currently only supported on
 450 Linux.
 451
 452 If both values are zero, load checks are disabled. Note that this is not the
 453 default.
 454
 455 =item B<--restart-delay NUM>
 456
 457 After processing the queue, wait this many seconds until the queue is
 458 restarted.
 459
 460 =item B<--lockfile FILE>
 461
 462 For perpetual operation, create a lockfile in that place and clean it up after
 463 finishing/aborting.
 464
 465 =item B<--all-once>
 466
 467 Instead of perpetuously processing all projects over and over again, process
 468 them just once and then exit.
 469
 470 =item B<--one PRJNAME>
 471
 472 Process only the given project (given as just the project name without C<.git>
 473 suffix) and then exit.
 474
 475 =back
 476
 477 =head1 DESCRIPTION
 478
 479 jobd is Girocco's repositories maintenance servant; it periodically checks all
 480 the repositories and updates mirrored repositories and repacks push-mode
 481 repositories when needed.
 482
 483 =cut