taskd/taskd.pl

   1 #!/usr/bin/perl
   2 #
   3 # taskd - Clone repositories on request
   4 #
   5 # taskd is Girocco mirroring servant; it processes requests for clones
   6 # of given URLs received over its socket.
   7 #
   8 # When a request is received, new process is spawned that sets up
   9 # the repository and reports further progress
  10 # to .clonelog within the repository. In case the clone fails,
  11 # .clone_failed is touched and .clone_in_progress is removed.
  12
  13 # Clone protocol:
  14 # Alice sets up repository and touches .cloning
  15 # Alice opens connection to Bob
  16 # Alice sends project name through the connection
  17 # Bob opens the repository and sends error code if there is a problem
  18 # Bob closes connection
  19 # Alice polls .clonelog in case of success.
  20 # If Alice reads "@OVER@" from .clonelog, it stops polling.
  21
  22 # Ref-change protocol:
  23 # Alice opens connection to Bob
  24 # Alice sends ref-change command for each changed ref
  25 # Alice closes connection
  26 # Bob sends out notifications
  27
  28 # Based on perlipc example.
  29
  30 use strict;
  31 use warnings;
  32
  33 use Getopt::Long;
  34 use Pod::Usage;
  35 use Socket;
  36 use Errno;
  37 use POSIX ":sys_wait_h";
  38 use File::Basename;
  39
  40 use lib dirname($0);
  41 use Girocco::Config;
  42 use Girocco::Notify;
  43 use Girocco::Project;
  44 use Girocco::User;
  45 use Girocco::Util qw(noFatalsToBrowser);
  46 BEGIN {noFatalsToBrowser}
  47
  48 # Options
  49 my $quiet;
  50 my $progress;
  51 my $syslog;
  52 my $stderr;
  53 my $inetd;
  54 my $idle_timeout;
  55 my $abbrev = 8;
  56
  57 $| = 1;
  58
  59 my $progname = basename($0);
  60 my $children = 0;
  61 my $idlestart = time;
  62
  63 sub logmsg { print '['.(scalar localtime)."] $progname $$: @_\n" }
  64 sub statmsg { print STDERR "[@{[scalar localtime]}] $progname $$: @_\n" if $progress }
  65
  66 sub REAPER {
  67         local $!;
  68         my $child;
  69         my $waitedpid;
  70         while (($waitedpid = waitpid(-1, WNOHANG)) > 0) {
  71                 $idlestart = time if !--$children;
  72                 logmsg "reaped $waitedpid" . ($? ? " with exit $?" : '');
  73         }
  74         $SIG{CHLD} = \&REAPER;  # loathe sysV
  75 }
  76
  77 $SIG{CHLD} = \&REAPER;  # Apollo 440
  78
  79 sub spawn {
  80         my $coderef = shift;
  81
  82         my $pid = fork;
  83         if (not defined $pid) {
  84                 logmsg "cannot fork: $!";
  85                 return;
  86         } elsif ($pid) {
  87                 $idlestart = time if !++$children;
  88                 logmsg "begat $pid";
  89                 return; # I'm the parent
  90         }
  91
  92         $SIG{CHLD} = 'DEFAULT';
  93
  94         local *STDIN;
  95         open STDIN, "+<&Client" or die "can't dup client to stdin";
  96         exit &$coderef();
  97 }
  98
  99 sub clone {
 100         my ($name) = @_;
 101         Girocco::Project::does_exist($name, 1) or die "no such project: $name";
 102         my $proj = Girocco::Project->load($name);
 103         $proj or die "failed to load project $name";
 104         $proj->{clone_in_progress} or die "project $name is not marked for cloning";
 105         $proj->{clone_logged} and die "project $name is already being cloned";
 106         statmsg "cloning $name";
 107         open STDOUT, '>', "$Girocco::Config::reporoot/$name.git/.clonelog" or die "cannot open clonelog: $!";
 108         open STDERR, ">&STDOUT";
 109         open STDIN, '<', '/dev/null';
 110         exec "$Girocco::Config::basedir/taskd/clone.sh", "$name.git" or die "exec failed: $!";
 111 }
 112
 113 sub ref_change {
 114         my ($arg) = @_;
 115         my ($username, $name, $oldrev, $newrev, $ref) = split(/\s+/, $arg);
 116         $username && $name && $oldrev && $newrev && $ref or return 0;
 117         $oldrev =~ /^[0-9a-f]{40}$/ && $newrev =~ /^[0-9a-f]{40}$/ && $ref =~ m{^refs/} or return 0;
 118         $newrev ne $oldrev or return 0;
 119
 120         Girocco::Project::does_exist($name, 1) or die "no such project: $name";
 121         my $proj = Girocco::Project->load($name);
 122         $proj or die "failed to load project $name";
 123
 124         my $user;
 125         if ($username && $username !~ /^%.*%$/) {
 126                 Girocco::User::does_exist($username, 1) or die "no such user: $username";
 127                 $user = Girocco::User->load($username);
 128                 $user or die "failed to load user $username";
 129         }
 130
 131         statmsg "ref-change $username $name ($ref: @{[substr($oldrev,0,$abbrev)]} -> @{[substr($newrev,0,$abbrev)]})";
 132         open STDIN, '<', '/dev/null';
 133         Girocco::Notify::ref_change($proj, $user, $ref, $oldrev, $newrev);
 134         return 0;
 135 }
 136
 137 #
 138 ## -------
 139 ## OStream
 140 ## -------
 141 #
 142
 143 package OStream;
 144
 145 use Carp 'croak';
 146 use Sys::Syslog qw(:DEFAULT :macros);
 147
 148 sub writeall {
 149         use POSIX qw();
 150         use Errno;
 151         my ($fd, $data) = @_;
 152         my $offset = 0;
 153         my $remaining = length($data);
 154         while ($remaining) {
 155                 my $bytes = POSIX::write(
 156                         $fd,
 157                         substr($data, $offset, $remaining),
 158                         $remaining);
 159                 next if !defined($bytes) && $!{EINTR};
 160                 croak "POSIX::write failed: $!" unless defined $bytes;
 161                 croak "POSIX::write wrote 0 bytes" unless $bytes;
 162                 $remaining -= $bytes;
 163                 $offset += $bytes;
 164         }
 165 }
 166
 167 sub dumpline {
 168         use POSIX qw(STDERR_FILENO);
 169         my ($self, $line) = @_;
 170         writeall(STDERR_FILENO, $line) if $self->{'stderr'};
 171         substr($line, -1, 1) = '' if substr($line, -1, 1) eq "\n";
 172         return unless length($line);
 173         syslog(LOG_NOTICE, "%s", $line) if $self->{'syslog'};
 174 }
 175
 176 sub TIEHANDLE {
 177         my $class = shift || 'OStream';
 178         my $mode = shift;
 179         my $syslogname = shift;
 180         my $syslogfacility = shift;
 181         defined($syslogfacility) or $syslogfacility = LOG_USER;
 182         my $self = {};
 183         $self->{'syslog'} = $mode > 0;
 184         $self->{'stderr'} = $mode <= 0 || $mode > 1;
 185         $self->{'lastline'} = '';
 186         if ($self->{'syslog'}) {
 187                 # Some Sys::Syslog have a stupid default setlogsock order
 188                 eval {Sys::Syslog::setlogsock("native"); 1;} or
 189                 eval {Sys::Syslog::setlogsock("unix");};
 190                 openlog($syslogname, "ndelay,pid", $syslogfacility)
 191                         or croak "Sys::Syslog::openlog failed: $!";
 192         }
 193         return bless $self, $class;
 194 }
 195
 196 sub BINMODE {return 1}
 197 sub FILENO {return undef}
 198 sub EOF {return 0}
 199 sub CLOSE {return 1}
 200
 201 sub PRINTF {
 202         my $self = shift;
 203         my $template = shift;
 204         return $self->PRINT(sprintf $template, @_);
 205 }
 206
 207 sub PRINT {
 208         my $self = shift;
 209         my $data = join('', $self->{'lastline'}, @_);
 210         my $pos = 0;
 211         while ((my $idx = index($data, "\n", $pos)) >= 0) {
 212                 ++$idx;
 213                 my $line = substr($data, $pos, $idx - $pos);
 214                 substr($data, $pos, $idx - $pos) = '';
 215                 $pos = $idx;
 216                 $self->dumpline($line);
 217         }
 218         $self->{'lastline'} = $data;
 219         return 1;
 220 }
 221
 222 sub DESTROY {
 223         my $self = shift;
 224         $self->dumpline($self->{'lastline'})
 225                 if length($self->{'lastline'});
 226         closelog;
 227 }
 228
 229 sub WRITE {
 230         my $self = shift;
 231         my ($scalar, $length, $offset) = @_;
 232         $scalar = '' if !defined($scalar);
 233         $length = length($scalar) if !defined($length);
 234         croak "OStream::WRITE invalid length $length"
 235                 if $length < 0;
 236         $offset = 0 if !defined($offset);
 237         $offset += length($scalar) if $offset < 0;
 238         croak "OStream::WRITE invalid write offset"
 239                 if $offset < 0 || $offset > $length;
 240         my $max = length($scalar) - $offset;
 241         $length = $max if $length > $max;
 242         $self->PRINT(substr($scalar, $offset, $length));
 243         return $length;
 244 }
 245
 246 #
 247 ## ----
 248 ## main
 249 ## ----
 250 #
 251
 252 package main;
 253
 254 my $sfac;
 255 Getopt::Long::Configure('bundling');
 256 my $parse_res = GetOptions(
 257         'help|?|h' => sub {pod2usage(-verbose => 2, -exitval => 0)},
 258         'quiet|q' => \$quiet,
 259         'no-quiet' => sub {$quiet = 0},
 260         'progress|P' => \$progress,
 261         'inetd|i' => sub {$inetd = 1; $syslog = 1; $quiet = 1;},
 262         'idle-timeout|t=i' => \$idle_timeout,
 263         'syslog|s:s' => \$sfac,
 264         'no-syslog' => sub {$syslog = 0; $sfac = undef;},
 265         'stderr' => \$stderr,
 266         'abbrev=i' => \$abbrev,
 267 ) || pod2usage(2);
 268 $syslog = 1 if defined($sfac);
 269 $progress = 1 unless $quiet;
 270 $abbrev = 128 unless $abbrev > 0;
 271 if (defined($idle_timeout)) {
 272         die "--idle-timeout must be a whole number" unless $idle_timeout =~ /^\d+$/;
 273         die "--idle-timeout may not be used without --inetd" unless $inetd;
 274 }
 275
 276 if ($syslog) {
 277         use Sys::Syslog qw();
 278         my $mode = 1;
 279         ++$mode if $stderr;
 280         $sfac = "user" unless defined($sfac) && $sfac ne "";
 281         my $ofac = $sfac;
 282         $sfac = uc($sfac);
 283         $sfac = 'LOG_'.$sfac unless $sfac =~ /^LOG_/;
 284         my $facility;
 285         my %badfac = map({("LOG_$_" => 1)}
 286                 (qw(PID CONS ODELAY NDELAY NOWAIT PERROR FACMASK NFACILITIES PRIMASK LFMT)));
 287         eval "\$facility = Sys::Syslog::$sfac; 1" or die "invalid syslog facility: $ofac";
 288         die "invalid syslog facility: $ofac"
 289                 if ($facility & ~0xf8) || ($facility >> 3) > 23 || $badfac{$sfac};
 290         tie *STDERR, 'OStream', $mode, $progname, $facility or die "tie failed";
 291 }
 292 if ($quiet) {
 293         open STDOUT, '>', '/dev/null';
 294 } elsif ($inetd) {
 295         *STDOUT = *STDERR;
 296 }
 297
 298 my $NAME;
 299
 300 if ($inetd) {
 301         open Server, '<&=0' or die "open: $!";
 302         my $sockname = getsockname Server;
 303         die "getsockname: $!" unless $sockname;
 304         die "socket already connected! must be 'wait' socket" if getpeername Server;
 305         die "getpeername: $!" unless $!{ENOTCONN};
 306         my $st = getsockopt Server, SOL_SOCKET, SO_TYPE;
 307         die "getsockopt(SOL_SOCKET, SO_TYPE): $!" unless $st;
 308         my $socktype = unpack('i', $st);
 309         die "stream socket required" unless defined $socktype && $socktype == SOCK_STREAM;
 310         die "AF_UNIX socket required" unless sockaddr_family($sockname) == AF_UNIX;
 311         $NAME = unpack_sockaddr_un $sockname;
 312         my $expected = $Girocco::Config::chroot.'/etc/taskd.socket';
 313         warn "listening on \"$NAME\" but expected \"$expected\"" unless $NAME eq $expected;
 314         my $mode = (stat($NAME))[2];
 315         die "stat: $!" unless $mode;
 316         $mode &= 07777;
 317         if (($mode & 0660) != 0660) {
 318                 chmod(($mode|0660), $NAME) == 1 or die "chmod ug+rw \"$NAME\": $!";
 319         }
 320 } else {
 321         $NAME = $Girocco::Config::chroot.'/etc/taskd.socket';
 322         my $uaddr = sockaddr_un($NAME);
 323
 324         socket(Server, PF_UNIX, SOCK_STREAM, 0) or die "socket: $!";
 325         unlink($NAME);
 326         bind(Server, $uaddr) or die "bind: $!";
 327         listen(Server, SOMAXCONN) or die "listen: $!";
 328         chmod 0666, $NAME or die "chmod: $!";
 329 }
 330
 331 my $fdset = '';
 332 vec($fdset, fileno(Server), 1) = 1;
 333 my $nextstatus = time + 60;
 334 statmsg "listening on $NAME";
 335 while (1) {
 336         my ($rout, $eout, $nfound);
 337         do {
 338                 my $wait;
 339                 $nextstatus += 60 while ($wait = $nextstatus - time) <= 0;
 340                 $nfound = select($rout=$fdset, undef, $eout=$fdset, $wait);
 341                 logmsg "select failed: $!" unless $nfound >= 0 || $!{EINTR};
 342                 my $now = time;
 343                 if ($idle_timeout && !$children && !$nfound && $now - $idlestart >= $idle_timeout) {
 344                         statmsg "idle timeout $idle_timeout exceeded now exiting";
 345                         exit 0;
 346                 }
 347                 if ($now >= $nextstatus) {
 348                         my $statmsg = "STATUS: $children active";
 349                         my $idlesecs;
 350                         $statmsg .= ", idle $idlesecs seconds"
 351                                 if !$children && ($idlesecs = $now - $idlestart) >= 2;
 352                         statmsg $statmsg;
 353                 }
 354         } while $nfound < 1;
 355         unless (accept(Client, Server)) {
 356                 logmsg "accept failed: $!" unless $!{EINTR};
 357                 next;
 358         }
 359         logmsg "connection on $NAME";
 360         spawn sub {
 361                 my $inp = <STDIN>;
 362                 chomp $inp if defined($inp);
 363                 $inp or exit 0; # ignore empty connects
 364                 my ($cmd, $arg) = $inp =~ /^([a-zA-Z0-9-]+)\s+(.*)$/;
 365                 if ($cmd eq 'clone') {
 366                         clone($arg);
 367                 } elsif ($cmd eq 'ref-change') {
 368                         ref_change($arg);
 369                 } else {
 370                         die "unknown command: $cmd";
 371                 }
 372         };
 373         close Client;
 374         sleep 1;
 375 }
 376
 377 #
 378 ## -------------
 379 ## Documentation
 380 ## -------------
 381 #
 382
 383 __END__
 384
 385 =head1 NAME
 386
 387 taskd.pl - Perform Girocco service tasks
 388
 389 =head1 SYNOPSIS
 390
 391 taskd.pl [options]
 392
 393  Options:
 394    -h | --help                           detailed instructions
 395    -q | --quiet                          run quietly
 396    --no-quiet                            do not run quietly
 397    -P | --progress                       show occasional status updates
 398    -i | --inetd                          run as inetd unix stream wait service
 399                                          implies --quiet --syslog
 400    -t SECONDS | --idle-timeout=SECONDS   how long to wait idle before exiting
 401                                          requires --inetd
 402    -s | --syslog[=facility]              send messages to syslog instead of
 403                                          stderr but see --stderr
 404                                          enabled by --inetd
 405    --no-syslog                           do not send message to syslog
 406    --stderr                              always send messages to stderr too
 407    --abbrev=n                            abbreviate hashes to n (default is 8)
 408
 409 =head1 OPTIONS
 410
 411 =over 8
 412
 413 =item B<--help>
 414
 415 Print the full description of taskd.pl's options.
 416
 417 =item B<--quiet>
 418
 419 Suppress non-error messages, e.g. for use when running this task as an inetd
 420 service.  Enabled by default by --inetd.
 421
 422 =item B<--no-quiet>
 423
 424 Enable non-error messages.  When running in --inetd mode these messages are
 425 sent to STDERR instead of STDOUT.
 426
 427 =item B<--progress>
 428
 429 Show information about the current status of the task operation occasionally.
 430 This is automatically enabled if --quiet is not given.
 431
 432 =item B<--inetd>
 433
 434 Run as an inetd wait service.  File descriptor 0 must be an unconnected unix
 435 stream socket ready to have accept called on it.  To be useful, the unix socket
 436 should be located at "$Girocco::Config::chroot/etc/taskd.socket".  A warning
 437 will be issued if the socket is not in the expected location.  Socket file
 438 permissions will be adjusted if necessary and if they cannot be taskd.pl will
 439 die.  The --inetd option also enables the --quiet and --syslog options but
 440 --no-quiet and --no-syslog may be used to alter that.
 441
 442 The correct specification for the inetd socket is a "unix" protocol "stream"
 443 socket in "wait" mode with user and group writable permissions (0660).  An
 444 attempt will be made to alter the socket's file mode if needed and if that
 445 cannot be accomplished taskd.pl will die.
 446
 447 Although most inetd stream services run in nowait mode, taskd.pl MUST be run
 448 in wait mode and will die if the passed in socket is already connected.
 449
 450 Note that while *BSD's inetd happily supports unix sockets (and so does
 451 Darwin's launchd), neither xinetd nor GNU's inetd supports unix sockets.
 452 However, systemd does seem to.
 453
 454 =item B<--idle-timeout=SECONDS>
 455
 456 Only permitted when running in --inetd mode.  After SECONDS of inactivity
 457 (i.e. all outstanding tasks have completed and no new requests have come in)
 458 exit normally.  The default is no timeout at all (a SECONDS value of 0).
 459 Note that it may actually take up to SECONDS+60 for the idle exit to occur.
 460
 461 =item B<--syslog[=facility]>
 462
 463 Normally error output is sent to STDERR.  With this option it's sent to
 464 syslog instead.  Note that when running in --inetd mode non-error output is
 465 also affected by this option as it's sent to STDERR in that case.  If
 466 not specified, the default for facility is LOG_USER.  Facility names are
 467 case-insensitive and the leading 'LOG_' is optional.  Messages are logged
 468 with the LOG_NOTICE priority.
 469
 470 =item B<--no-syslog>
 471
 472 Send error message output to STDERR but not syslog.
 473
 474 =item B<--stderr>
 475
 476 Always send error message output to STDERR.  If --syslog is in effect then
 477 a copy will also be sent to syslog.  In --inetd mode this applies to non-error
 478 messages as well.
 479
 480 =item B<--abbrev=n>
 481
 482 Abbreviate displayed hash values to only the first n hexadecimal characters.
 483 The default is 8 characters.  Set to 0 for no abbreviation at all.
 484
 485 =back
 486
 487 =head1 DESCRIPTION
 488
 489 taskd.pl is Girocco's service request servant; it listens for service requests
 490 such as new clone requests and ref update notifications and spawns a task to
 491 perform the requested action.
 492
 493 =cut