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