Merge pull request #668 from NixOS/notifications

Turn hydra-notify into a daemon

hydra-module.nix

@@ -379,6 +379,23 @@ in
           };
       };
 
+    systemd.services.hydra-notify =
+      { wantedBy = [ "multi-user.target" ];
+        requires = [ "hydra-init.service" ];
+        after = [ "hydra-init.service" ];
+        restartTriggers = [ hydraConf ];
+        environment = env // {
+          PGPASSFILE = "${baseDir}/pgpass-queue-runner"; # grrr
+        };
+        serviceConfig =
+          { ExecStart = "@${cfg.package}/bin/hydra-notify hydra-notify";
+            # FIXME: run this under a less privileged user?
+            User = "hydra-queue-runner";
+            Restart = "always";
+            RestartSec = 5;
+          };
+      };
+
     # If there is less than a certain amount of free disk space, stop
     # the queue/evaluator to prevent builds from failing or aborting.
     systemd.services.hydra-check-space =

src/hydra-queue-runner/builder.cc

@@ -99,6 +99,8 @@ State::StepResult State::doBuildStep(nix::ref<Store> destStore,
     unsigned int maxSilentTime, buildTimeout;
     unsigned int repeats = step->isDeterministic ? 1 : 0;
 
+    auto conn(dbPool.get());
+
     {
         std::set<Build::ptr> dependents;
         std::set<Step::ptr> steps;
@@ -122,8 +124,10 @@ State::StepResult State::doBuildStep(nix::ref<Store> destStore,
 
         for (auto build2 : dependents) {
             if (build2->drvPath == step->drvPath) {
-              build = build2;
+                build = build2;
-              enqueueNotificationItem({NotificationItem::Type::BuildStarted, build->id});
+                pqxx::work txn(*conn);
+                notifyBuildStarted(txn, build->id);
+                txn.commit();
             }
             {
                 auto i = jobsetRepeats.find(std::make_pair(build2->projectName, build2->jobsetName));
@@ -144,8 +148,6 @@ State::StepResult State::doBuildStep(nix::ref<Store> destStore,
 
     bool quit = buildId == buildOne && step->drvPath == buildDrvPath;
 
-    auto conn(dbPool.get());
-
     RemoteResult result;
     BuildOutput res;
     unsigned int stepNr = 0;
@@ -170,11 +172,6 @@ State::StepResult State::doBuildStep(nix::ref<Store> destStore,
             } catch (...) {
                 ignoreException();
             }
-
-            /* Asynchronously run plugins. FIXME: if we're killed,
-               plugin actions might not be run. Need to ensure
-               at-least-once semantics. */
-            enqueueNotificationItem({NotificationItem::Type::StepFinished, buildId, {}, stepNr, result.logFile});
         }
     });
 
@@ -231,6 +228,11 @@ State::StepResult State::doBuildStep(nix::ref<Store> destStore,
     time_t stepStopTime = time(0);
     if (!result.stopTime) result.stopTime = stepStopTime;
 
+    /* For standard failures, we don't care about the error
+       message. */
+    if (result.stepStatus != bsAborted)
+        result.errorMsg = "";
+
     /* Account the time we spent building this step by dividing it
        among the jobsets that depend on it. */
     {
@@ -243,6 +245,13 @@ State::StepResult State::doBuildStep(nix::ref<Store> destStore,
         }
     }
 
+    /* Finish the step in the database. */
+    if (stepNr) {
+        pqxx::work txn(*conn);
+        finishBuildStep(txn, result, buildId, stepNr, machine->sshName);
+        txn.commit();
+    }
+
     /* The step had a hopefully temporary failure (e.g. network
        issue). Retry a number of times. */
     if (result.canRetry) {
@@ -256,11 +265,6 @@ State::StepResult State::doBuildStep(nix::ref<Store> destStore,
         }
         if (retry) {
             auto mc = startDbUpdate();
-            {
-            pqxx::work txn(*conn);
-            finishBuildStep(txn, result, buildId, stepNr, machine->sshName);
-            txn.commit();
-            }
             stepFinished = true;
             if (quit) exit(1);
             return sRetry;
@@ -315,8 +319,6 @@ State::StepResult State::doBuildStep(nix::ref<Store> destStore,
 
                 pqxx::work txn(*conn);
 
-                finishBuildStep(txn, result, buildId, stepNr, machine->sshName);
-
                 for (auto & b : direct) {
                     printMsg(lvlInfo, format("marking build %1% as succeeded") % b->id);
                     markSucceededBuild(txn, b, res, buildId != b->id || result.isCached,
@@ -342,8 +344,12 @@ State::StepResult State::doBuildStep(nix::ref<Store> destStore,
 
         /* Send notification about the builds that have this step as
            the top-level. */
-        for (auto id : buildIDs)
+        {
-            enqueueNotificationItem({NotificationItem::Type::BuildFinished, id});
+            pqxx::work txn(*conn);
+            for (auto id : buildIDs)
+                notifyBuildFinished(txn, id, {});
+            txn.commit();
+        }
 
         /* Wake up any dependent steps that have no other
            dependencies. */
@@ -369,11 +375,6 @@ State::StepResult State::doBuildStep(nix::ref<Store> destStore,
 
     } else {
 
-        /* For standard failures, we don't care about the error
-           message. */
-        if (result.stepStatus != bsAborted)
-            result.errorMsg = "";
-
         /* Register failure in the database for all Build objects that
            directly or indirectly depend on this step. */
 
@@ -419,11 +420,6 @@ State::StepResult State::doBuildStep(nix::ref<Store> destStore,
                         result.stepStatus, result.errorMsg, buildId == build2->id ? 0 : buildId);
                 }
 
-                if (result.stepStatus != bsCachedFailure && !stepFinished) {
-                    assert(stepNr);
-                    finishBuildStep(txn, result, buildId, stepNr, machine->sshName);
-                }
-
                 /* Mark all builds that depend on this derivation as failed. */
                 for (auto & build2 : indirect) {
                     if (build2->finishedInDB) continue;
@@ -462,11 +458,10 @@ State::StepResult State::doBuildStep(nix::ref<Store> destStore,
 
         /* Send notification about this build and its dependents. */
         {
-            auto notificationSenderQueue_(notificationSenderQueue.lock());
+            pqxx::work txn(*conn);
-            notificationSenderQueue_->push(NotificationItem{NotificationItem::Type::BuildFinished, buildId, dependentIDs});
+            notifyBuildFinished(txn, buildId, dependentIDs);
+            txn.commit();
         }
-        notificationSenderWakeup.notify_one();
-
     }
 
     // FIXME: keep stats about aborted steps?

src/hydra-queue-runner/hydra-queue-runner.cc

@@ -268,6 +268,9 @@ unsigned int State::createBuildStep(pqxx::work & txn, time_t startTime, BuildID
             ("insert into BuildStepOutputs (build, stepnr, name, path) values ($1, $2, $3, $4)")
             (buildId)(stepNr)(output.first)(output.second.path).exec();
 
+    if (status == bsBusy)
+        txn.exec(fmt("notify step_started, '%d\t%d'", buildId, stepNr));
+
     return stepNr;
 }
 
@@ -299,6 +302,9 @@ void State::finishBuildStep(pqxx::work & txn, const RemoteResult & result,
         (result.timesBuilt, result.timesBuilt > 0)
         (result.isNonDeterministic, result.timesBuilt > 1)
         .exec();
+    assert(result.logFile.find('\t') == std::string::npos);
+    txn.exec(fmt("notify step_finished, '%d\t%d\t%s'",
+            buildId, stepNr, result.logFile));
 }
 
 
@@ -450,74 +456,20 @@ bool State::checkCachedFailure(Step::ptr step, Connection & conn)
 }
 
 
-void State::notificationSender()
+void State::notifyBuildStarted(pqxx::work & txn, BuildID buildId)
 {
-    while (true) {
+    txn.exec(fmt("notify build_started, '%s'", buildId));
-        try {
+}
 
-            NotificationItem item;
-            {
-                auto notificationSenderQueue_(notificationSenderQueue.lock());
-                while (notificationSenderQueue_->empty())
-                    notificationSenderQueue_.wait(notificationSenderWakeup);
-                item = notificationSenderQueue_->front();
-                notificationSenderQueue_->pop();
-            }
 
-            MaintainCount<counter> mc(nrNotificationsInProgress);
+void State::notifyBuildFinished(pqxx::work & txn, BuildID buildId,
-
+    const std::vector<BuildID> & dependentIds)
-            printMsg(lvlChatty, format("sending notification about build %1%") % item.id);
+{
-
+    auto payload = fmt("%d ", buildId);
-            auto now1 = std::chrono::steady_clock::now();
+    for (auto & d : dependentIds)
-
+        payload += fmt("%d ", d);
-            Pid pid = startProcess([&]() {
+    // FIXME: apparently parameterized() doesn't support NOTIFY.
-                Strings argv;
+    txn.exec(fmt("notify build_finished, '%s'", payload));
-                switch (item.type) {
-                    case NotificationItem::Type::BuildStarted:
-                        argv = {"hydra-notify", "build-started", std::to_string(item.id)};
-                        for (auto id : item.dependentIds)
-                            argv.push_back(std::to_string(id));
-                        break;
-                    case NotificationItem::Type::BuildFinished:
-                        argv = {"hydra-notify", "build-finished", std::to_string(item.id)};
-                        for (auto id : item.dependentIds)
-                            argv.push_back(std::to_string(id));
-                        break;
-                    case NotificationItem::Type::StepFinished:
-                        argv = {"hydra-notify", "step-finished", std::to_string(item.id), std::to_string(item.stepNr), item.logPath};
-                        break;
-                };
-                printMsg(lvlChatty, "Executing hydra-notify " + concatStringsSep(" ", argv));
-                execvp("hydra-notify", (char * *) stringsToCharPtrs(argv).data()); // FIXME: remove cast
-                throw SysError("cannot start hydra-notify");
-            });
-
-            int res = pid.wait();
-
-            if (!statusOk(res))
-                throw Error("notification about build %d failed: %s", item.id, statusToString(res));
-
-            auto now2 = std::chrono::steady_clock::now();
-
-            if (item.type == NotificationItem::Type::BuildFinished) {
-                auto conn(dbPool.get());
-                pqxx::work txn(*conn);
-                txn.parameterized
-                    ("update Builds set notificationPendingSince = null where id = $1")
-                    (item.id)
-                    .exec();
-                txn.commit();
-            }
-
-            nrNotificationTimeMs += std::chrono::duration_cast<std::chrono::milliseconds>(now2 - now1).count();
-            nrNotificationsDone++;
-
-        } catch (std::exception & e) {
-            nrNotificationsFailed++;
-            printMsg(lvlError, format("notification sender: %1%") % e.what());
-            sleep(5);
-        }
-    }
 }
 
 
@@ -589,13 +541,6 @@ void State::dumpStatus(Connection & conn, bool log)
         root.attr("nrDbConnections", dbPool.count());
         root.attr("nrActiveDbUpdates", nrActiveDbUpdates);
         root.attr("memoryTokensInUse", memoryTokens.currentUse());
-        root.attr("nrNotificationsDone", nrNotificationsDone);
-        root.attr("nrNotificationsFailed", nrNotificationsFailed);
-        root.attr("nrNotificationsInProgress", nrNotificationsInProgress);
-        root.attr("nrNotificationsPending", notificationSenderQueue.lock()->size());
-        root.attr("nrNotificationTimeMs", nrNotificationTimeMs);
-        uint64_t nrNotificationsTotal = nrNotificationsDone + nrNotificationsFailed;
-        root.attr("nrNotificationTimeAvgMs", nrNotificationsTotal == 0 ? 0.0 : (float) nrNotificationTimeMs / nrNotificationsTotal);
 
         {
             auto nested = root.object("machines");
@@ -843,24 +788,6 @@ void State::run(BuildID buildOne)
 
     std::thread(&State::dispatcher, this).detach();
 
-    /* Idem for notification sending. */
-    auto maxConcurrentNotifications = config->getIntOption("max-concurrent-notifications", 2);
-    for (uint64_t i = 0; i < maxConcurrentNotifications; ++i)
-        std::thread(&State::notificationSender, this).detach();
-
-    /* Enqueue notification items for builds that were finished
-       previously, but for which we didn't manage to send
-       notifications. */
-    {
-        auto conn(dbPool.get());
-        pqxx::work txn(*conn);
-        auto res = txn.parameterized("select id from Builds where notificationPendingSince > 0").exec();
-        for (auto const & row : res) {
-            auto id = row["id"].as<BuildID>();
-            enqueueNotificationItem({NotificationItem::Type::BuildFinished, id});
-        }
-    }
-
     /* Periodically clean up orphaned busy steps in the database. */
     std::thread([&]() {
         while (true) {

src/hydra-queue-runner/queue-monitor.cc

@@ -193,13 +193,12 @@ bool State::getQueuedBuilds(Connection & conn,
                     (build->id)
                     ((int) (ex.step->drvPath == build->drvPath ? bsFailed : bsDepFailed))
                     (time(0)).exec();
+                notifyBuildFinished(txn, build->id, {});
                 txn.commit();
                 build->finishedInDB = true;
                 nrBuildsDone++;
             }
 
-            enqueueNotificationItem({NotificationItem::Type::BuildFinished, build->id});
-
             return;
         }
 
@@ -230,13 +229,12 @@ bool State::getQueuedBuilds(Connection & conn,
             time_t now = time(0);
             printMsg(lvlInfo, format("marking build %1% as succeeded (cached)") % build->id);
             markSucceededBuild(txn, build, res, true, now, now);
+            notifyBuildFinished(txn, build->id, {});
             txn.commit();
             }
 
             build->finishedInDB = true;
 
-            enqueueNotificationItem({NotificationItem::Type::BuildFinished, build->id});
-
             return;
         }
 

src/hydra-queue-runner/state.hh

@@ -347,39 +347,6 @@ private:
     counter bytesSent{0};
     counter bytesReceived{0};
     counter nrActiveDbUpdates{0};
-    counter nrNotificationsDone{0};
-    counter nrNotificationsFailed{0};
-    counter nrNotificationsInProgress{0};
-    counter nrNotificationTimeMs{0};
-
-    /* Notification sender work queue. FIXME: if hydra-queue-runner is
-       killed before it has finished sending notifications about a
-       build, then the notifications may be lost. It would be better
-       to mark builds with pending notification in the database. */
-    struct NotificationItem
-    {
-        enum class Type : char {
-           BuildStarted,
-           BuildFinished,
-           StepFinished,
-        };
-        Type type;
-        BuildID id;
-        std::vector<BuildID> dependentIds;
-        unsigned int stepNr;
-        nix::Path logPath;
-    };
-    nix::Sync<std::queue<NotificationItem>> notificationSenderQueue;
-    std::condition_variable notificationSenderWakeup;
-
-    void enqueueNotificationItem(const NotificationItem && item)
-    {
-        {
-            auto notificationSenderQueue_(notificationSenderQueue.lock());
-            notificationSenderQueue_->emplace(item);
-        }
-        notificationSenderWakeup.notify_one();
-    }
 
     /* Specific build to do for --build-one (testing only). */
     BuildID buildOne;
@@ -540,9 +507,10 @@ private:
 
     bool checkCachedFailure(Step::ptr step, Connection & conn);
 
-    /* Thread that asynchronously invokes hydra-notify to send build
+    void notifyBuildStarted(pqxx::work & txn, BuildID buildId);
-       notifications. */
+
-    void notificationSender();
+    void notifyBuildFinished(pqxx::work & txn, BuildID buildId,
+        const std::vector<BuildID> & dependentIds);
 
     /* Acquire the global queue runner lock, or null if somebody else
        has it. */

src/lib/Hydra/Controller/API.pm

@@ -76,7 +76,7 @@ sub latestbuilds : Chained('api') PathPart('latestbuilds') Args(0) {
 sub jobsetToHash {
     my ($jobset) = @_;
     return {
-        project => $jobset->project->name,
+        project => $jobset->get_column('project'),
         name => $jobset->name,
         nrscheduled => $jobset->get_column("nrscheduled"),
         nrsucceeded => $jobset->get_column("nrsucceeded"),
@@ -206,12 +206,12 @@ sub scmdiff : Path('/api/scmdiff') Args(0) {
 
 sub triggerJobset {
     my ($self, $c, $jobset, $force) = @_;
-    print STDERR "triggering jobset ", $jobset->project->name . ":" . $jobset->name, "\n";
+    print STDERR "triggering jobset ", $jobset->get_column('project') . ":" . $jobset->name, "\n";
     txn_do($c->model('DB')->schema, sub {
         $jobset->update({ triggertime => time });
         $jobset->update({ forceeval => 1 }) if $force;
     });
-    push @{$c->{stash}->{json}->{jobsetsTriggered}}, $jobset->project->name . ":" . $jobset->name;
+    push @{$c->{stash}->{json}->{jobsetsTriggered}}, $jobset->get_column('project') . ":" . $jobset->name;
 }
 
 

src/lib/Hydra/Controller/JobsetEval.pm

@@ -142,7 +142,7 @@ sub release : Chained('evalChain') PathPart('release') Args(0) {
     $releaseName ||= $_->releasename foreach @builds;
 
     # If no release name has been defined by any of the builds, compose one of the project name and evaluation id
-    $releaseName = $eval->project->name."-".$eval->id unless defined $releaseName;
+    $releaseName = $eval->get_column('project') . "-" . $eval->id unless defined $releaseName;
 
     my $release;
 

src/lib/Hydra/Helper/CatalystUtils.pm

@@ -60,9 +60,9 @@ sub getNextBuild {
     (my $nextBuild) = $c->model('DB::Builds')->search(
       { finished => 1
       , system => $build->system
-      , project => $build->project->name
+      , project => $build->get_column('project')
-      , jobset => $build->jobset->name
+      , jobset => $build->get_column('jobset')
-      , job => $build->job->name
+      , job => $build->get_column('job')
       , 'me.id' =>  { '>' => $build->id }
       }, {rows => 1, order_by => "me.id ASC"});
 
@@ -77,9 +77,9 @@ sub getPreviousSuccessfulBuild {
     (my $prevBuild) = $c->model('DB::Builds')->search(
       { finished => 1
       , system => $build->system
-      , project => $build->project->name
+      , project => $build->get_column('project')
-      , jobset => $build->jobset->name
+      , jobset => $build->get_column('jobset')
-      , job => $build->job->name
+      , job => $build->get_column('job')
       , buildstatus => 0
       , 'me.id' =>  { '<' => $build->id }
       }, {rows => 1, order_by => "me.id DESC"});
@@ -289,7 +289,7 @@ sub parseJobsetName {
 
 sub showJobName {
     my ($build) = @_;
-    return $build->project->name . ":" . $build->jobset->name . ":" . $build->job->name;
+    return $build->get_column('project') . ":" . $build->get_column('jobset') . ":" . $build->get_column('job');
 }
 
 

src/lib/Hydra/Plugin.pm

@@ -12,11 +12,15 @@ sub new {
     return $self;
 }
 
+sub isEnabled {
+    return 1;
+}
+
 sub instantiate {
     my ($class, %args) = @_;
     my $plugins = [];
     $args{plugins} = $plugins;
-    push @$plugins, $class->plugins(%args);
+    push @$plugins, grep { $_->isEnabled } $class->plugins(%args);
     return @$plugins;
 }
 

src/lib/Hydra/Plugin/BitBucketStatus.pm

@@ -7,6 +7,11 @@ use JSON;
 use LWP::UserAgent;
 use Hydra::Helper::CatalystUtils;
 
+sub isEnabled {
+    my ($self) = @_;
+    return $self->{config}->{enable_bitbucket_status} == 1;
+}
+
 sub toBitBucketState {
     my ($buildStatus) = @_;
     if ($buildStatus == 0) {

src/lib/Hydra/Plugin/CircleCINotification.pm

@@ -7,6 +7,11 @@ use LWP::UserAgent;
 use Hydra::Helper::CatalystUtils;
 use JSON;
 
+sub isEnabled {
+    my ($self) = @_;
+    return defined $self->{config}->{circleci};
+}
+
 sub buildFinished {
     my ($self, $build, $dependents) = @_;
     my $cfg = $self->{config}->{circleci};

src/lib/Hydra/Plugin/CoverityScan.pm

@@ -6,6 +6,11 @@ use File::Basename;
 use LWP::UserAgent;
 use Hydra::Helper::CatalystUtils;
 
+sub isEnabled {
+    my ($self) = @_;
+    return defined $self->{config}->{coverityscan};
+}
+
 sub buildFinished {
     my ($self, $b, $dependents) = @_;
 

src/lib/Hydra/Plugin/EmailNotification.pm

@@ -9,6 +9,10 @@ use Hydra::Helper::Nix;
 use Hydra::Helper::CatalystUtils;
 use Hydra::Helper::Email;
 
+sub isEnabled {
+    my ($self) = @_;
+    return $self->{config}->{email_notification} == 1;
+}
 
 my $template = <<EOF;
 Hi,
@@ -44,8 +48,6 @@ EOF
 sub buildFinished {
     my ($self, $build, $dependents) = @_;
 
-    return unless $self->{config}->{email_notification} // 0;
-
     die unless $build->finished;
 
     # Figure out to whom to send notification for each build.  For
@@ -98,7 +100,7 @@ sub buildFinished {
             , dependents => [grep { $_->id != $build->id } @builds]
             , baseurl => getBaseUrl($self->{config})
             , showJobName => \&showJobName, showStatus => \&showStatus
-            , showSystem => index($build->job->name, $build->system) == -1
+            , showSystem => index($build->get_column('job'), $build->system) == -1
             , nrCommits => $nrCommits
             , authorList => $authorList
             };
@@ -117,9 +119,9 @@ sub buildFinished {
 
         sendEmail(
             $self->{config}, $to, $subject, $body,
-            [ 'X-Hydra-Project'  => $build->project->name,
+            [ 'X-Hydra-Project'  => $build->get_column('project'),
-            , 'X-Hydra-Jobset'   => $build->jobset->name,
+            , 'X-Hydra-Jobset'   => $build->get_column('jobset'),
-            , 'X-Hydra-Job'      => $build->job->name,
+            , 'X-Hydra-Job'      => $build->get_column('job'),
             , 'X-Hydra-System'   => $build->system
             ]);
     }

src/lib/Hydra/Plugin/GithubStatus.pm

@@ -8,6 +8,11 @@ use LWP::UserAgent;
 use Hydra::Helper::CatalystUtils;
 use List::Util qw(max);
 
+sub isEnabled {
+    my ($self) = @_;
+    return defined $self->{config}->{githubstatus};
+}
+
 sub toGithubState {
     my ($buildStatus) = @_;
     if ($buildStatus == 0) {

src/lib/Hydra/Plugin/GitlabStatus.pm

@@ -16,6 +16,11 @@ use List::Util qw(max);
 #   - gitlab_project_id => ID of the project in Gitlab, i.e. in the above
 #     case the ID in gitlab of "nixexprs"
 
+sub isEnabled {
+    my ($self) = @_;
+    return defined $self->{config}->{gitlab_authorization};
+}
+
 sub toGitlabState {
     my ($status, $buildStatus) = @_;
     if ($status == 0) {
@@ -50,7 +55,7 @@ sub common {
                 state => $state,
                 target_url => "$baseurl/build/" . $b->id,
                 description => "Hydra build #" . $b->id . " of $jobName",
-                name => "Hydra " . $b->job->name,
+                name => "Hydra " . $b->get_column('job'),
             });
         while (my $eval = $evals->next) {
             my $gitlabstatusInput = $eval->jobsetevalinputs->find({ name => "gitlab_status_repo" });

src/lib/Hydra/Plugin/HipChatNotification.pm

@@ -5,6 +5,11 @@ use parent 'Hydra::Plugin';
 use LWP::UserAgent;
 use Hydra::Helper::CatalystUtils;
 
+sub isEnabled {
+    my ($self) = @_;
+    return defined $self->{config}->{hipchat};
+}
+
 sub buildFinished {
     my ($self, $build, $dependents) = @_;
 
@@ -54,7 +59,7 @@ sub buildFinished {
 
         my $msg = "";
         $msg .= "<img src='$img'/> ";
-        $msg .= "Job <a href='$baseurl/job/${\$build->project->name}/${\$build->jobset->name}/${\$build->job->name}'>${\showJobName($build)}</a>";
+        $msg .= "Job <a href='$baseurl/job/${\$build->get_column('project')}/${\$build->get_column('jobset')}/${\$build->get_column('job')}'>${\showJobName($build)}</a>";
         $msg .= " (and ${\scalar @deps} others)" if scalar @deps > 0;
         $msg .= ": <a href='$baseurl/build/${\$build->id}'>" . showStatus($build) . "</a>";
 

src/lib/Hydra/Plugin/InfluxDBNotification.pm

@@ -7,6 +7,11 @@ use HTTP::Request;
 use LWP::UserAgent;
 # use Hydra::Helper::CatalystUtils;
 
+sub isEnabled {
+    my ($self) = @_;
+    return defined $self->{config}->{influxdb};
+}
+
 sub toBuildStatusDetailed {
     my ($buildStatus) = @_;
     if ($buildStatus == 0) {
@@ -99,10 +104,10 @@ sub buildFinished {
         my $tagSet = {
             status  => toBuildStatusClass($b->buildstatus),
             result  => toBuildStatusDetailed($b->buildstatus),
-            project => $b->project->name,
+            project => $b->get_column('project'),
-            jobset  => $b->jobset->name,
+            jobset  => $b->get_column('jobset'),
-            repo    => ($b->jobset->name =~ /^(.*)\.pr-/) ? $1 : $b->jobset->name,
+            repo    => ($b->get_column('jobset') =~ /^(.*)\.pr-/) ? $1 : $b->get_column('jobset'),
-            job     => $b->job->name,
+            job     => $b->get_column('job'),
             system  => $b->system,
             cached  => $b->iscachedbuild ? "true" : "false",
         };

src/lib/Hydra/Plugin/RunCommand.pm

@@ -5,6 +5,11 @@ use parent 'Hydra::Plugin';
 use experimental 'smartmatch';
 use JSON;
 
+sub isEnabled {
+    my ($self) = @_;
+    return defined $self->{config}->{runcommand};
+}
+
 sub configSectionMatches {
     my ($name, $project, $jobset, $job) = @_;
 

src/lib/Hydra/Plugin/S3Backup.pm

@@ -14,6 +14,11 @@ use Nix::Store;
 use Hydra::Model::DB;
 use Hydra::Helper::CatalystUtils;
 
+sub isEnabled {
+    my ($self) = @_;
+    return defined $self->{config}->{s3backup};
+}
+
 my $client;
 my %compressors = (
     xz => "| $Nix::Config::xz",

src/lib/Hydra/Plugin/SlackNotification.pm

@@ -7,6 +7,11 @@ use LWP::UserAgent;
 use Hydra::Helper::CatalystUtils;
 use JSON;
 
+sub isEnabled {
+    my ($self) = @_;
+    return defined $self->{config}->{slack};
+}
+
 sub renderDuration {
     my ($build) = @_;
     my $duration = $build->stoptime - $build->starttime;
@@ -76,7 +81,7 @@ sub buildFinished {
             "danger";
 
         my $text = "";
-        $text .= "Job <$baseurl/job/${\$build->project->name}/${\$build->jobset->name}/${\$build->job->name}|${\showJobName($build)}>";
+        $text .= "Job <$baseurl/job/${\$build->get_column('project')}/${\$build->get_column('jobset')}/${\$build->get_column('job')}|${\showJobName($build)}>";
         $text .= " (and ${\scalar @deps} others)" if scalar @deps > 0;
         $text .= ": <$baseurl/build/${\$build->id}|" . showStatus($build) . ">". " in " . renderDuration($build);
 

src/script/hydra-eval-jobset

@@ -438,7 +438,7 @@ sub checkBuild {
                 # semantically unnecessary (because they're implied by
                 # the eval), but they give a factor 1000 speedup on
                 # the Nixpkgs jobset with PostgreSQL.
-                { project => $jobset->project->name, jobset => $jobset->name, job => $jobName,
+                { project => $jobset->get_column('project'), jobset => $jobset->name, job => $jobName,
                   name => $firstOutputName, path => $firstOutputPath },
                 { rows => 1, columns => ['id'], join => ['buildoutputs'] });
             if (defined $prevBuild) {
@@ -489,7 +489,7 @@ sub checkBuild {
         $buildMap->{$build->id} = { id => $build->id, jobName => $jobName, new => 1, drvPath => $drvPath };
         $$jobOutPathMap{$jobName . "\t" . $firstOutputPath} = $build->id;
 
-        print STDERR "added build ${\$build->id} (${\$jobset->project->name}:${\$jobset->name}:$jobName)\n";
+        print STDERR "added build ${\$build->id} (${\$jobset->get_column('project')}:${\$jobset->name}:$jobName)\n";
     });
 
     return $build;
@@ -531,7 +531,7 @@ sub sendJobsetErrorNotification() {
     return if $jobset->project->owner->emailonerror == 0;
     return if $errorMsg eq "";
 
-    my $projectName = $jobset->project->name;
+    my $projectName = $jobset->get_column('project');
     my $jobsetName = $jobset->name;
     my $body = "Hi,\n"
         . "\n"
@@ -568,7 +568,7 @@ sub permute {
 
 
 sub checkJobsetWrapped {
-    my ($jobset) = @_;
+    my ($jobset, $tmpId) = @_;
     my $project = $jobset->project;
     my $jobsetsJobset = length($project->declfile) && $jobset->name eq ".jobsets";
     my $inputInfo = {};
@@ -607,6 +607,7 @@ sub checkJobsetWrapped {
         print STDERR $fetchError;
         txn_do($db, sub {
             $jobset->update({ lastcheckedtime => time, fetcherrormsg => $fetchError }) if !$dryRun;
+            $db->storage->dbh->do("notify eval_failed, ?", undef, join('\t', $tmpId));
         });
         return;
     }
@@ -622,6 +623,7 @@ sub checkJobsetWrapped {
         Net::Statsd::increment("hydra.evaluator.unchanged_checkouts");
         txn_do($db, sub {
             $jobset->update({ lastcheckedtime => time, fetcherrormsg => undef });
+            $db->storage->dbh->do("notify eval_cached, ?", undef, join('\t', $tmpId));
         });
         return;
     }
@@ -690,6 +692,9 @@ sub checkJobsetWrapped {
             , nrbuilds => $jobsetChanged ? scalar(keys %buildMap) : undef
             });
 
+        $db->storage->dbh->do("notify eval_added, ?", undef,
+                              join('\t', $tmpId, $ev->id));
+
         if ($jobsetChanged) {
             # Create JobsetEvalMembers mappings.
             while (my ($id, $x) = each %buildMap) {
@@ -767,10 +772,6 @@ sub checkJobsetWrapped {
     Net::Statsd::increment("hydra.evaluator.evals");
     Net::Statsd::increment("hydra.evaluator.cached_evals") unless $jobsetChanged;
 
-    #while (my ($id, $x) = each %buildMap) {
-    #    system("hydra-notify build-queued $id") if $x->{new};
-    #}
-
     # Store the error messages for jobs that failed to evaluate.
     my $msg = "";
     foreach my $job (values %{$jobs}) {
@@ -788,8 +789,15 @@ sub checkJobset {
 
     my $startTime = clock_gettime(CLOCK_MONOTONIC);
 
+    # Add an ID to eval_* notifications so receivers can correlate
+    # them.
+    my $tmpId = "${startTime}.$$";
+
+    $db->storage->dbh->do("notify eval_started, ?", undef,
+                          join('\t', $tmpId, $jobset->get_column('project'), $jobset->name));
+
     eval {
-        checkJobsetWrapped($jobset);
+        checkJobsetWrapped($jobset, $tmpId);
     };
     my $checkError = $@;
 
@@ -802,6 +810,7 @@ sub checkJobset {
         txn_do($db, sub {
             $jobset->update({lastcheckedtime => time});
             setJobsetError($jobset, $checkError);
+            $db->storage->dbh->do("notify eval_failed, ?", undef, join('\t', $tmpId));
         }) if !$dryRun;
         $failed = 1;
     }

src/script/hydra-notify

@@ -5,6 +5,7 @@ use utf8;
 use Hydra::Plugin;
 use Hydra::Helper::Nix;
 use Hydra::Helper::AddBuilds;
+use IO::Select;
 
 STDERR->autoflush(1);
 binmode STDERR, ":encoding(utf8)";
@@ -15,20 +16,37 @@ my $db = Hydra::Model::DB->new();
 
 my @plugins = Hydra::Plugin->instantiate(db => $db, config => $config);
 
-my $cmd = shift @ARGV or die "Syntax: hydra-notify build-started BUILD | build-finished BUILD-ID [BUILD-IDs...] | step-finished BUILD-ID STEP-NR LOG-PATH\n";
+my $dbh = $db->storage->dbh;
 
-my $buildId = shift @ARGV or die;
+$dbh->do("listen build_started");
-my $build = $db->resultset('Builds')->find($buildId)
+$dbh->do("listen build_finished");
-    or die "build $buildId does not exist\n";
+$dbh->do("listen step_finished");
+
+sub buildStarted {
+    my ($buildId) = @_;
+
+    my $build = $db->resultset('Builds')->find($buildId)
+        or die "build $buildId does not exist\n";
+
+    foreach my $plugin (@plugins) {
+        eval { $plugin->buildStarted($build); };
+        if ($@) {
+            print STDERR "$plugin->buildStarted: $@\n";
+        }
+    }
+}
+
+sub buildFinished {
+    my ($build, @deps) = @_;
 
-if ($cmd eq "build-finished") {
     my $project = $build->project;
-    my $jobset = $build->jobset;
+    my $jobsetName = $build->get_column('jobset');
-    if (length($project->declfile) && $jobset->name eq ".jobsets" && $build->iscurrent) {
+    if (length($project->declfile) && $jobsetName eq ".jobsets" && $build->iscurrent) {
         handleDeclarativeJobsetBuild($db, $project, $build);
     }
+
     my @dependents;
-    foreach my $id (@ARGV) {
+    foreach my $id (@deps) {
         my $dep = $db->resultset('Builds')->find($id)
             or die "build $id does not exist\n";
         push @dependents, $dep;
@@ -40,33 +58,20 @@ if ($cmd eq "build-finished") {
             print STDERR "$plugin->buildFinished: $@\n";
         }
     }
+
+    $build->update({ notificationpendingsince => undef });
 }
 
-elsif ($cmd eq "build-queued") {
+sub stepFinished {
-    foreach my $plugin (@plugins) {
+    my ($buildId, $stepNr, $logPath) = @_;
-        eval { $plugin->buildQueued($build); };
-        if ($@) {
-            print STDERR "$plugin->buildQueued: $@\n";
-        }
-    }
-}
 
-elsif ($cmd eq "build-started") {
+    my $build = $db->resultset('Builds')->find($buildId)
-    foreach my $plugin (@plugins) {
+        or die "build $buildId does not exist\n";
-        eval { $plugin->buildStarted($build); };
-        if ($@) {
-            print STDERR "$plugin->buildStarted: $@\n";
-        }
-    }
-}
 
-elsif ($cmd eq "step-finished") {
-    die if scalar @ARGV < 2;
-    my $stepNr = shift @ARGV;
     my $step = $build->buildsteps->find({stepnr => $stepNr})
         or die "step $stepNr does not exist\n";
-    my $logPath = shift @ARGV;
+
-    $logPath = undef if $logPath eq "";
+    $logPath = undef if $logPath eq "-";
 
     foreach my $plugin (@plugins) {
         eval { $plugin->stepFinished($step, $logPath); };
@@ -76,6 +81,43 @@ elsif ($cmd eq "step-finished") {
     }
 }
 
-else {
+# Process builds that finished while hydra-notify wasn't running.
-    die "unknown action ‘$cmd’";
+for my $build ($db->resultset('Builds')->search(
+                   { notificationpendingsince => { '!=', undef } }))
+{
+    my $buildId = $build->id;
+    print STDERR "sending notifications for build ${\$buildId}...\n";
+    buildFinished($build);
+}
+
+# Process incoming notifications.
+my $fd = $dbh->func("getfd");
+my $sel = IO::Select->new($fd);
+
+while (1) {
+    $sel->can_read;
+
+    while (my $notify = $dbh->func("pg_notifies")) {
+
+        my ($channelName, $pid, $payload) = @$notify;
+        #print STDERR "got '$channelName' from $pid: $payload\n";
+
+        my @payload = split /\t/, $payload;
+
+        eval {
+            if ($channelName eq "build_started") {
+                buildStarted(int($payload[0]));
+            } elsif ($channelName eq "build_finished") {
+                my $buildId = int($payload[0]);
+                my $build = $db->resultset('Builds')->find($buildId)
+                    or die "build $buildId does not exist\n";
+                buildFinished($build, @payload[1..$#payload]);
+            } elsif ($channelName eq "step_finished") {
+                stepFinished(int($payload[0]), int($payload[1]));
+            }
+        };
+        if ($@) {
+            print STDERR "error processing message '$payload' on channel '$channelName': $@\n";
+        }
+    }
 }