[backend] send package status to a redis server if configured

srinidhibs · May 23, 2019 · d5217f0 · d5217f0
1 parent 503fbe9
commit d5217f0
Show file tree

Hide file tree

Showing 15 changed files with 710 additions and 51 deletions.
diff --git a/dist/obs-server.spec b/dist/obs-server.spec
@@ -674,6 +674,8 @@ fi
 /usr/lib/obs/server/bs_worker
 /usr/lib/obs/server/bs_signer
 /usr/lib/obs/server/bs_warden
+/usr/lib/obs/server/bs_redis
+/usr/lib/obs/server/bs_notifyforward
 /usr/lib/obs/server/worker
 /usr/lib/obs/server/worker-deltagen.spec
 %config(noreplace) /usr/lib/obs/server/BSConfig.pm

diff --git a/src/backend/BSConfiguration.pm b/src/backend/BSConfiguration.pm
@@ -92,4 +92,6 @@ $BSConfig::service_timeout          = $BSConfig::service_timeout          || 360
 
 $BSConfig::cloudupload_pubkey       = $BSConfig::cloudupload_pubkey       || '/etc/obs/cloudupload/_pubkey';
 
+$BSConfig::redisserver = undef unless $BSConfig::redisserver;
+
 1;
diff --git a/src/backend/BSRedisnotify.pm b/src/backend/BSRedisnotify.pm
@@ -0,0 +1,65 @@
+#
+# Copyright (c) 2019 SUSE LLC
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 2 as
+# published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program (see the file COPYING); if not, write to the
+# Free Software Foundation, Inc.,
+# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA
+#
+package BSRedisnotify;
+
+use BSConfiguration;
+use BSUtil;
+
+my $eventdir = "$BSConfig::bsdir/events";
+
+my $notifyforwarddir = "$eventdir/notifyforward";
+
+sub addforwardjob {
+  my (@job) = @_;
+  s/([\000-\037%|=\177-\237])/sprintf("%%%02X", ord($1))/ge for @job;
+  my $job = join('|', @job)."\n";
+  my $file;
+  mkdir_p($notifyforwarddir) unless -d $notifyforwarddir;
+  BSUtil::lockopen($file, '>>', "$notifyforwarddir/queue");
+  my $oldlen = -s $file;
+  (syswrite($file, $job) || 0) == length($job) || die("notifyforward/queue: $!\n");
+  close($file);
+  BSUtil::ping("$notifyforwarddir/.ping") unless $oldlen;
+}
+
+sub updateresult {
+  my ($prpa, $packstatus, $packerror, $jobs) = @_;
+  my @job = ('redis', 'updateresult', $prpa);
+  for my $packid (sort keys %$packstatus) {
+    my $code = $packstatus->{$packid};
+    my $details = $code eq 'scheduled' ? $jobs->{$packid} : $packerror->{$packid};
+    push @job, $packid, ($details ? "$code:$details" : $code);
+  }
+  addforwardjob(@job);
+}
+
+sub updateoneresult {
+  my ($prpa, $packid, $codedetails, $job) = @_;
+  my @job = ('redis', 'updateoneresult', $prpa, $packid, $codedetails);
+  push @job, "scheduled:$job" if $job;
+  addforwardjob(@job);
+}
+
+sub updatejobstatus {
+  my ($prpa, $job, $codedetails) = @_;
+  my @job = ('redis', 'updatejobstatus', $prpa, "scheduled:$job");
+  push @job, $codedetails if $codedetails;
+  addforwardjob(@job);
+}
+
+1;
diff --git a/src/backend/BSSched/BuildJob.pm b/src/backend/BSSched/BuildJob.pm
@@ -91,6 +91,7 @@ use BSSched::EventSource::Directory; 	# for sendevent
 use Build;
 use BSRPC;
 use BSCando;
+use BSRedisnotify;
 
 =head1 NAME
 
@@ -160,7 +161,7 @@ sub purgejob {
 =cut
 
 sub killjob {
-  my ($gctx, $job) = @_;
+  my ($gctx, $prp, $job) = @_;
 
   my $myjobsdir = $gctx->{'myjobsdir'};
   local *F;
@@ -192,6 +193,7 @@ sub killjob {
     };
     warn("kill $job: $@") if $@;
   }
+  BSRedisnotify::updatejobstatus("$prp/$gctx->{'arch'}", $job) if $BSConfig::redisserver;
   purgejob($gctx, $job);
   close(F);
 }
@@ -204,7 +206,7 @@ sub killjob {
 =cut
 
 sub killscheduled {
-  my ($gctx, $job) = @_;
+  my ($gctx, $prp, $job) = @_;
 
   my $myjobsdir = $gctx->{'myjobsdir'};
   return if -e "$myjobsdir/$job:status";
@@ -240,7 +242,7 @@ sub killbuilding {
   }
   for my $job (@jobs) {
     print "        killing obsolete job $job\n";
-    killjob($gctx, $job);
+    killjob($gctx, $prp, $job);
   }
 }
 
@@ -269,11 +271,11 @@ sub killunwantedjobs {
       }
       if ($status eq 'disabled' || $status eq 'excluded' || $status eq 'locked') {
 	print "        killing old job $job, now in disabled/excluded/locked state\n";
-	killjob($gctx, $job);
+	killjob($gctx, $prp, $job);
       } elsif ($status eq 'blocked' || $status eq 'unresolvable' || $status eq 'broken') {
 	# blocked jobs get removed, if they are currently not building. building jobs
 	# stay since they may become valid again
-	killscheduled($gctx, $job);
+	killscheduled($gctx, $prp, $job);
       }
     }
   }
@@ -381,23 +383,22 @@ sub update_buildavg {
 =cut
 
 sub jobfinished {
-  my ($ectx, $job, $js) = @_;
+  my ($ectx, $job, $info, $js) = @_;
 
   my $gctx = $ectx->{'gctx'};
   my $myjobsdir = $gctx->{'myjobsdir'};
-  my $info = readxml("$myjobsdir/$job", $BSXML::buildinfo, 1);
   my $jobdatadir = "$myjobsdir/$job:dir";
-  if (!$info || ! -d $jobdatadir) {
-    print "  - $job is bad\n";
+  if (! -d $jobdatadir) {
+    print "  - $job has no data dir\n";
     return;
   }
   # dispatch to specialized versions for aggregates and deltas
   if ($info->{'file'} eq '_aggregate') {
-    BSSched::BuildJob::Aggregate::jobfinished($ectx, $job, $js);
+    BSSched::BuildJob::Aggregate::jobfinished($ectx, $job, $info, $js);
     return ;
   }
   if ($info->{'file'} eq '_delta') {
-    BSSched::BuildJob::DeltaRpm::jobfinished($ectx, $job, $js);
+    BSSched::BuildJob::DeltaRpm::jobfinished($ectx, $job, $info, $js);
     return ;
   }
 
@@ -410,10 +411,6 @@ sub jobfinished {
   my $prp = "$projid/$repoid";
 
   my $now = time(); # ensure that we use the same time in all logs
-  if ($info->{'arch'} ne $myarch) {
-    print "  - $job has bad arch\n";
-    return;
-  }
   my $projpacks = $gctx->{'projpacks'};
   if (!$projpacks->{$projid}) {
     print "  - $job belongs to an unknown project ($projid/$packid)\n";
@@ -467,7 +464,8 @@ sub jobfinished {
   }
 
   # update packstatus so that it doesn't fall back to scheduled
-  patchpackstatus($gctx, $prp, $packid, $code);
+  patchpackstatus($gctx, $prp, $packid, $code, $job);
+  $info->{'packstatus_patched'} = 1;
 
   my $meta = $all{'meta'} ? "$jobdatadir/meta" : undef;
   if ($code eq 'unchanged') {
@@ -678,7 +676,7 @@ sub fakejobfinished_nouseforbuild {
 =cut
 
 sub patchpackstatus {
-  my ($gctx, $prp, $packid, $code) = @_;
+  my ($gctx, $prp, $packid, $code, $job) = @_;
 
   my $reporoot = $gctx->{'reporoot'};
   my $myarch = $gctx->{'arch'};
@@ -687,6 +685,7 @@ sub patchpackstatus {
   BSUtil::appendstr("$gdst/:packstatus.finished", "$code $packid\n");
   # touch mtime to make watchers see a change
   utime(time, time, "$gdst/:packstatus");
+  BSRedisnotify::updateoneresult("$prp/$myarch", $packid, "finished:$code", $job) if $BSConfig::redisserver;
 }
 
 
@@ -989,7 +988,7 @@ sub create {
     my @otherjobs = find_otherjobs($ctx, $jobprefix);
     for my $otherjob (@otherjobs) {
       print "        killing old job $otherjob\n" if $ctx->{'verbose'};
-      killjob($gctx, $otherjob);
+      killjob($gctx, $prp, $otherjob);
     }
   }
 

diff --git a/src/backend/BSSched/BuildJob/Aggregate.pm b/src/backend/BSSched/BuildJob/Aggregate.pm
@@ -476,21 +476,16 @@ sub build {
 =cut
 
 sub jobfinished {
-  my ($ectx, $job, $js) = @_;
+  my ($ectx, $job, $info, $js) = @_;
 
   my $gctx = $ectx->{'gctx'};
 
   my $changed = $gctx->{'changed_med'};
   my $myjobsdir = $gctx->{'myjobsdir'};
   my $myarch = $gctx->{'arch'};
-  my $info = readxml("$myjobsdir/$job", $BSXML::buildinfo, 1);
   my $jobdatadir = "$myjobsdir/$job:dir";
-  if (!$info || ! -d $jobdatadir) {
-    print "  - $job is bad\n";
-    return;
-  }
-  if ($info->{'arch'} ne $myarch) {
-    print "  - $job has bad arch\n";
+  if (! -d $jobdatadir) {
+    print "  - $job has no data dir\n";
     return;
   }
   my $projid = $info->{'project'};
@@ -532,7 +527,8 @@ sub jobfinished {
   # update meta
   mkdir_p("$gdst/:meta");
   rename("$jobdatadir/meta", "$gdst/:meta/$packid") || die("rename $jobdatadir/meta $gdst/:meta/$packid: $!\n");
-  BSSched::BuildJob::patchpackstatus($gctx, $prp, $packid, 'succeeded');
+  BSSched::BuildJob::patchpackstatus($gctx, $prp, $packid, 'succeeded', $job);
+  $info->{'packstatus_patched'} = 1;
 }
 
 sub readcontainerinfo {

diff --git a/src/backend/BSSched/BuildJob/DeltaRpm.pm b/src/backend/BSSched/BuildJob/DeltaRpm.pm
@@ -150,21 +150,16 @@ sub build {
 =cut
 
 sub jobfinished {
-  my ($ectx, $job, $js) = @_;
+  my ($ectx, $job, $info, $js) = @_;
 
   my $gctx = $ectx->{'gctx'};
 
   my $changed = $gctx->{'changed_med'};
   my $myjobsdir = $gctx->{'myjobsdir'};
   my $myarch = $gctx->{'arch'};
-  my $info = readxml("$myjobsdir/$job", $BSXML::buildinfo, 1);
   my $jobdatadir = "$myjobsdir/$job:dir";
-  if (!$info || ! -d $jobdatadir) {
-    print "  - $job is bad\n";
-    return;
-  }
-  if ($info->{'arch'} ne $myarch) {
-    print "  - $job has bad arch\n";
+  if (! -d $jobdatadir) {
+    print "  - $job has no data dir\n";
     return;
   }
   my $projid = $info->{'project'};

diff --git a/src/backend/BSSched/Checker.pm b/src/backend/BSSched/Checker.pm
@@ -24,6 +24,7 @@ use Digest::MD5 ();
 use BSUtil;
 use BSSolv;
 use BSNotify;
+use BSRedisnotify;
 
 use BSSched::ProjPacks;
 use BSSched::BuildRepo;
@@ -101,6 +102,23 @@ sub generate_random_id {
   return Digest::MD5::md5_hex($random);
 }
 
+=head2 notify - send repo changed notification
+
+=cut
+
+sub notify {
+  my ($ctx, $type, $buildid) = @_;
+
+  my $myarch = $ctx->{'gctx'}->{'arch'};
+  if ($BSConfig::redisserver) {
+    # use the redis forwarder to send the notification
+    BSRedisnotify::addforwardjob($type, "project=$ctx->{'project'}", "repo=$ctx->{'repository'}", "arch=$myarch", "buildid=$buildid");
+    return;
+  }
+  my $body = { project => $ctx->{'project'}, 'repo' => $ctx->{'repository'}, 'arch' => $myarch, 'buildid' => $buildid };
+  BSNotify::notify($type, $body);
+}
+
 =head2 set_repo_state - update the :schedulerstate file of a prp
 
 =cut
@@ -132,7 +150,7 @@ sub set_repo_state {
     if (!$newstate->{'oldbuildid'}) {
       my $id = generate_random_id($oldstate) . '-inprogress';
       $newstate->{'oldbuildid'} = $id;
-      BSNotify::notify('REPO_BUILD_STARTED', { project => $ctx->{'project'}, 'repo' => $ctx->{'repository'}, 'arch' => $myarch, 'buildid' => $id} );
+      $ctx->notify('REPO_BUILD_STARTED', $id);
     }
   } elsif ($state eq 'finished') {
     # we're done (for now). generate repostateid
@@ -149,7 +167,7 @@ sub set_repo_state {
 	# but the repo changed, send synthetic event
 	print "sending synthetic REPO_BUILD_STARTED event\n";
 	my $id = generate_random_id($oldstate) . '-inprogress';
-	BSNotify::notify('REPO_BUILD_STARTED', { project => $ctx->{'project'}, 'repo' => $ctx->{'repository'}, 'arch' => $myarch, 'buildid' => $id} );
+	$ctx->notify('REPO_BUILD_STARTED', $id);
 	delete $newstate->{'buildid'};
 	$newstate->{'oldbuildid'} = $id;
       }
@@ -158,7 +176,7 @@ sub set_repo_state {
       my $id = delete $newstate->{'oldbuildid'};
       $id =~ s/-inprogress$//;
       $newstate->{'buildid'} = $id;
-      BSNotify::notify('REPO_BUILD_FINISHED', { project => $ctx->{'project'}, 'repo' => $ctx->{'repository'}, 'arch' => $myarch, 'buildid' => $id} );
+      $ctx->notify('REPO_BUILD_FINISHED', $id);
     }
     $newstate->{'repostateid'} = $repostateid;
   }
@@ -1021,6 +1039,7 @@ sub checkpkgs {
   } else {
     unlink("$gdst/:packstatus.finished");
   }
+  BSRedisnotify::updateresult("$prp/$myarch", \%packstatus, \%packerror, \%building) if $BSConfig::redisserver;
 
   my $schedulerstate;
   if (keys %building) {

diff --git a/src/backend/BSSched/EventHandler.pm b/src/backend/BSSched/EventHandler.pm
@@ -27,6 +27,7 @@ use BSUtil;
 use BSXML;
 use BSConfiguration;
 use BSSolv;
+use BSRedisnotify;
 use BSSched::Checker;
 use BSSched::BuildResult;
 use BSSched::BuildRepo;
@@ -94,6 +95,7 @@ sub event_built {
   my ($ectx, $ev) = @_;
 
   my $gctx = $ectx->{'gctx'};
+  my $myarch = $gctx->{'arch'};
   my $job = $ev->{'job'};
   local *F;
   my $myjobsdir = $gctx->{'myjobsdir'};
@@ -108,12 +110,21 @@ sub event_built {
     close F;
     return;
   }
-  if ($ev->{'type'} eq 'built') {
-    BSSched::BuildJob::jobfinished($ectx, $job, $js);
-  } elsif ($ev->{'type'} eq 'uploadbuild') {
-    BSSched::BuildJob::Upload::jobfinished($ectx, $job, $js);
-  } elsif ($ev->{'type'} eq 'import') {
-    BSSched::BuildJob::Import::jobfinished($ectx, $job, $js);
+  my $info = readxml("$myjobsdir/$job", $BSXML::buildinfo, 1);
+  if (!$info) {
+    print "  - $job has bad info\n";
+  } elsif ($info->{'arch'} ne $myarch) {
+    print "  - $job has bad arch\n";
+  } else {
+    if ($ev->{'type'} eq 'built') {
+      BSSched::BuildJob::jobfinished($ectx, $job, $info, $js);
+    } elsif ($ev->{'type'} eq 'uploadbuild') {
+      BSSched::BuildJob::Upload::jobfinished($ectx, $job, $info, $js);
+    } elsif ($ev->{'type'} eq 'import') {
+      BSSched::BuildJob::Import::jobfinished($ectx, $job, $info, $js);
+    }
+    my $prp = "$info->{'project'}/$info->{'repository'}";
+    BSRedisnotify::updatejobstatus("$prp/$myarch", $job) if $BSConfig::redisserver && !$info->{'packstatus_patched'};
   }
   BSSched::BuildJob::purgejob($gctx, $job);
   close F;