grahamc/hydra - Change 32KJOERMPFWZZZCIN6TGGVX72GMAJT6VCCIP7S65EHNF3KM42KWAC

Turn hydra-notify into a daemon

It now receives notifications about started/finished builds/steps via PostgreSQL. This gets rid of the (substantial) overhead of starting hydra-notify for every event. It also allows other programs (even on other machines) to listen to Hydra notifications.

Created by Eelco Dolstra on August 9, 2019

32KJOERMPFWZZZCIN6TGGVX72GMAJT6VCCIP7S65EHNF3KM42KWAC

Dependencies

In channels

main

upstream

Change contents

Insertion in src/hydra-queue-runner/builder.cc at line 101 [12.133]
[11.500]
[12.2164]
```
    auto conn(dbPool.get());
```

Replacement in src/hydra-queue-runner/builder.cc at line 127 [12.133]

B:BD[13.93] → [13.93:123]

B:BD[13.123] → [14.0:90]

              build = build2;
              enqueueNotificationItem({NotificationItem::Type::BuildStarted, build->id});

[13.93]

[11.501]

                build = build2;
                pqxx::work txn(*conn);
                notifyBuildStarted(txn, build->id);
                txn.commit();

Deletion in src/hydra-queue-runner/builder.cc at line 151 [12.133]
B:BD[12.3461] → [12.3461:3491]
```
    auto conn(dbPool.get());
```

Deletion in src/hydra-queue-runner/builder.cc at line 175 [12.133]

B:BD[2.596] → [2.596:597]

∅:D[2.597] → [15.33:320]

B:BD[15.33] → [15.33:320]


            /* 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});

Replacement in src/hydra-queue-runner/builder.cc at line 342 [12.133]

B:BD[12.8217] → [14.226:341]

        for (auto id : buildIDs)
            enqueueNotificationItem({NotificationItem::Type::BuildFinished, id});

[12.8217]

[12.8514]

        {
            pqxx::work txn(*conn);
            for (auto id : buildIDs)
                notifyBuildFinished(txn, id, {});
            txn.commit();
        }

Replacement in src/hydra-queue-runner/builder.cc at line 466 [12.133]

B:BD[12.13945] → [12.13945:14020]

B:BD[12.14020] → [16.2506:2630]

            auto notificationSenderQueue_(notificationSenderQueue.lock());
            notificationSenderQueue_->push(NotificationItem{NotificationItem::Type::BuildFinished, buildId, dependentIDs});

[12.13945]

[12.14107]

            pqxx::work txn(*conn);
            notifyBuildFinished(txn, buildId, dependentIDs);
            txn.commit();

Deletion in src/hydra-queue-runner/builder.cc at line 470 [12.133]
B:BD[12.14117] → [12.14117:14165]
```
        notificationSenderWakeup.notify_one();
```

Insertion in src/hydra-queue-runner/hydra-queue-runner.cc at line 302 [17.4840]

[11.2024]

[17.13421]

    assert(result.logFile.find('\'') == std::string::npos);
    txn.exec(fmt("notify step_finished, '%d %d %s'", buildId, stepNr,
            result.logFile.empty() ? "-" : result.logFile));

Replacement in src/hydra-queue-runner/hydra-queue-runner.cc at line 456 [17.4840]
B:BD[18.1348] → [18.1348:1381]
```
void State::notificationSender()
```
[19.1820]
[18.1381]
```
void State::notifyBuildStarted(pqxx::work & txn, BuildID buildId)
```

Replacement in src/hydra-queue-runner/hydra-queue-runner.cc at line 458 [17.4840]

B:BD[18.1383] → [18.1383:1416]

∅:D[18.1416] → [20.170:171]

B:BD[20.170] → [20.170:171]

B:BD[20.171] → [18.1417:1801]

B:BD[18.1801] → [21.0:1]

B:BD[21.1] → [8.512:578]

∅:D[21.58] → [18.1801:1802]

∅:D[8.578] → [18.1801:1802]

B:BD[18.1801] → [18.1801:1802]

B:BD[18.1802] → [13.653:744]

∅:D[13.744] → [18.1896:1897]

B:BD[18.1896] → [18.1896:1897]

B:BD[18.1897] → [21.59:118]

∅:D[21.118] → [18.1897:1940]

B:BD[18.1897] → [18.1897:1940]

B:BD[18.1940] → [14.1732:2662]

B:BD[14.2662] → [10.0:94]

∅:D[10.94] → [18.2117:2300]

∅:D[22.212] → [18.2117:2300]

∅:D[14.2662] → [18.2117:2300]

B:BD[18.2117] → [18.2117:2300]

B:BD[18.2300] → [23.184:218]

∅:D[23.218] → [18.2338:2339]

B:BD[18.2338] → [18.2338:2339]

B:BD[18.2339] → [5.0:133]

∅:D[5.133] → [18.2505:2506]

∅:D[13.990] → [18.2505:2506]

B:BD[18.2505] → [18.2505:2506]

B:BD[18.2506] → [21.119:178]

B:BD[21.178] → [7.396:771]

    while (true) {
        try {
            NotificationItem item;
            {
                auto notificationSenderQueue_(notificationSenderQueue.lock());
                while (notificationSenderQueue_->empty())
                    notificationSenderQueue_.wait(notificationSenderWakeup);
                item = notificationSenderQueue_->front();
                notificationSenderQueue_->pop();
            }
            MaintainCount<counter> mc(nrNotificationsInProgress);
            printMsg(lvlChatty, format("sending notification about build %1%") % item.id);
            auto now1 = std::chrono::steady_clock::now();
            Pid pid = startProcess([&]() {
                Strings argv;
                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();
            }

[18.1383]

[7.771]

    txn.exec(fmt("notify build_started, '%s'", buildId));
}

Deletion in src/hydra-queue-runner/hydra-queue-runner.cc at line 461 [17.4840]

∅:D[7.772] → [21.178:325]

B:BD[21.178] → [21.178:325]

            nrNotificationTimeMs += std::chrono::duration_cast<std::chrono::milliseconds>(now2 - now1).count();
            nrNotificationsDone++;

Replacement in src/hydra-queue-runner/hydra-queue-runner.cc at line 462 [17.4840]

∅:D[21.326] → [18.2506:2545]

B:BD[18.2506] → [18.2506:2545]

B:BD[18.2545] → [21.327:364]

∅:D[21.364] → [18.2545:2662]

B:BD[18.2545] → [18.2545:2662]

        } catch (std::exception & e) {
            nrNotificationsFailed++;
            printMsg(lvlError, format("notification sender: %1%") % e.what());
            sleep(5);
        }
    }

[21.326]

[18.2662]

void State::notifyBuildFinished(pqxx::work & txn, BuildID buildId,
    const std::vector<BuildID> & dependentIds)
{
    auto payload = fmt("%d ", buildId);
    for (auto & d : dependentIds)
        payload += fmt("%d ", d);
    // FIXME: apparently parameterized() doesn't support NOTIFY.
    txn.exec(fmt("notify build_finished, '%s'", payload));

Deletion in src/hydra-queue-runner/hydra-queue-runner.cc at line 541 [17.4840]

B:BD[24.1487] → [21.365:938]

        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);

Deletion in src/hydra-queue-runner/hydra-queue-runner.cc at line 787 [17.4840]

B:BD[25.92] → [26.398:399]

B:BD[19.3153] → [18.2722:2763]

B:BD[18.2763] → [6.0:221]


    /* 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();

Deletion in src/hydra-queue-runner/hydra-queue-runner.cc at line 788 [17.4840]

B:BD[18.2824] → [7.773:1280]

    /* 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});
        }
    }

Insertion in src/hydra-queue-runner/queue-monitor.cc at line 196 [12.20067]
[27.2304]
[27.2304]
```
                notifyBuildFinished(txn, build->id, {});
```

Deletion in src/hydra-queue-runner/queue-monitor.cc at line 201 [12.20067]

B:BD[27.2424] → [3.0:90]


            enqueueNotificationItem({NotificationItem::Type::BuildFinished, build->id});

Insertion in src/hydra-queue-runner/queue-monitor.cc at line 232 [12.20067]
[12.25116]
[12.25116]
```
            notifyBuildFinished(txn, build->id, {});
```
Deletion in src/hydra-queue-runner/queue-monitor.cc at line 238 [12.20067]
B:BD[3.92] → [3.92:181]
∅:D[3.181] → [12.25183:25184]
B:BD[12.25183] → [12.25183:25184]
```
            enqueueNotificationItem({NotificationItem::Type::BuildFinished, build->id});
```

Deletion in src/hydra-queue-runner/state.hh at line 350 [29.1128]

B:BD[28.910] → [21.939:1092]

∅:D[28.910] → [29.6084:6085]

∅:D[21.1092] → [29.6084:6085]

B:BD[29.6084] → [29.6084:6085]

B:BD[29.6223] → [29.6223:6498]

B:BD[29.6498] → [13.991:1058]

B:BD[13.1058] → [14.2843:2919]

∅:D[14.2919] → [13.1098:1191]

B:BD[13.1098] → [13.1098:1191]

B:BD[13.1191] → [14.2920:2976]

∅:D[14.2976] → [13.1191:1198]

B:BD[13.1191] → [13.1191:1198]

∅:D[13.1198] → [30.2097:2220]

B:BD[29.6569] → [30.2097:2220]

∅:D[30.2220] → [29.6691:6692]

B:BD[29.6691] → [29.6691:6692]

B:BD[29.6692] → [14.2977:3249]

    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();
    }

Replacement in src/hydra-queue-runner/state.hh at line 510 [29.1128]

B:BD[29.8792] → [29.8792:8917]

    /* Thread that asynchronously invokes hydra-notify to send build
       notifications. */
    void notificationSender();

[29.8698]

[29.8917]

    void notifyBuildStarted(pqxx::work & txn, BuildID buildId);
    void notifyBuildFinished(pqxx::work & txn, BuildID buildId,
        const std::vector<BuildID> & dependentIds);

Insertion in src/script/hydra-notify at line 8 [18.2994]
[31.5666]
[18.3128]
```
use IO::Select;
```
Insertion in src/script/hydra-notify at line 18 [18.2994]
[18.3326]
[18.3326]
```
my $dbh = $db->storage->dbh;
```

Replacement in src/script/hydra-notify at line 21 [18.2994]

B:BD[18.3327] → [14.3662:3820]

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";

[18.3327]

[18.3412]

$dbh->do("listen build_started");
$dbh->do("listen build_finished");
$dbh->do("listen step_finished");
sub buildStarted {
    my ($buildId) = @_;

Replacement in src/script/hydra-notify at line 28 [18.2994]

B:BD[18.3413] → [13.1658:1791]

my $buildId = shift @ARGV or die;
my $build = $db->resultset('Builds')->find($buildId)
    or die "build $buildId does not exist\n";

[18.3413]

[14.3821]

    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";
        }
    }
}

Replacement in src/script/hydra-notify at line 39 [18.2994]
∅:D[14.3822] → [13.1791:1823]
B:BD[13.1791] → [13.1791:1823]
```
if ($cmd eq "build-finished") {
```
[14.3822]
[31.5667]
```
sub buildFinished {
    my ($build, @deps) = @_;
```
Insertion in src/script/hydra-notify at line 47 [18.2994]
[31.5892]
[18.3581]
Replacement in src/script/hydra-notify at line 49 [18.2994]
B:BD[18.3601] → [18.3601:3630]
```
    foreach my $id (@ARGV) {
```
[18.3601]
[18.3630]
```
    foreach my $id (@deps) {
```
Insertion in src/script/hydra-notify at line 61 [18.2994]
[14.4016]
[14.4016]
```
    $build->update({ notificationpendingsince => undef });
```

Replacement in src/script/hydra-notify at line 65 [18.2994]

B:BD[14.4019] → [4.265:473]

elsif ($cmd eq "build-queued") {
    foreach my $plugin (@plugins) {
        eval { $plugin->buildQueued($build); };
        if ($@) {
            print STDERR "$plugin->buildQueued: $@\n";
        }
    }
}

[14.4019]

[4.473]

sub stepFinished {
    my ($buildId, $stepNr, $logPath) = @_;

Replacement in src/script/hydra-notify at line 68 [18.2994]

∅:D[4.474] → [14.4019:4230]

B:BD[14.4019] → [14.4019:4230]

elsif ($cmd eq "build-started") {
    foreach my $plugin (@plugins) {
        eval { $plugin->buildStarted($build); };
        if ($@) {
            print STDERR "$plugin->buildStarted: $@\n";
        }
    }
}

[4.474]

[14.4230]

    my $build = $db->resultset('Builds')->find($buildId)
        or die "build $buildId does not exist\n";

Deletion in src/script/hydra-notify at line 71 [18.2994]
B:BD[14.4231] → [14.4231:4265]
B:BD[14.4265] → [9.0:59]
```
elsif ($cmd eq "step-finished") {
    die if scalar @ARGV < 2;
    my $stepNr = shift @ARGV;
```

Replacement in src/script/hydra-notify at line 73 [18.2994]

B:BD[14.4411] → [9.60:131]

    my $logPath = shift @ARGV;
    $logPath = undef if $logPath eq "";

[14.4411]

[14.4449]


    $logPath = undef if $logPath eq "-";

Replacement in src/script/hydra-notify at line 84 [18.2994]

B:BD[18.3833] → [18.3833:3877]

else {
    die "unknown action ‘$cmd’";

[18.3833]

[18.3877]

# Process builds that finished while hydra-notify wasn't running.
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;
    my $notify = $dbh->func("pg_notifies");
    next if !$notify;
    my ($channelName, $pid, $payload) = @$notify;
    #print STDERR "got '$channelName' from $pid: $payload\n";
    my @payload = split / /, $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";
    }

Turn hydra-notify into a daemon

Dependencies

In channels

Change contents

Insertion in src/hydra-queue-runner/builder.cc at line 101 [12.133]

Replacement in src/hydra-queue-runner/builder.cc at line 127 [12.133]

Deletion in src/hydra-queue-runner/builder.cc at line 151 [12.133]

Deletion in src/hydra-queue-runner/builder.cc at line 175 [12.133]

Replacement in src/hydra-queue-runner/builder.cc at line 342 [12.133]

Replacement in src/hydra-queue-runner/builder.cc at line 466 [12.133]

Deletion in src/hydra-queue-runner/builder.cc at line 470 [12.133]

Insertion in src/hydra-queue-runner/hydra-queue-runner.cc at line 302 [17.4840]

Replacement in src/hydra-queue-runner/hydra-queue-runner.cc at line 456 [17.4840]

Replacement in src/hydra-queue-runner/hydra-queue-runner.cc at line 458 [17.4840]

Deletion in src/hydra-queue-runner/hydra-queue-runner.cc at line 461 [17.4840]

Replacement in src/hydra-queue-runner/hydra-queue-runner.cc at line 462 [17.4840]

Deletion in src/hydra-queue-runner/hydra-queue-runner.cc at line 541 [17.4840]

Deletion in src/hydra-queue-runner/hydra-queue-runner.cc at line 787 [17.4840]

Deletion in src/hydra-queue-runner/hydra-queue-runner.cc at line 788 [17.4840]

Insertion in src/hydra-queue-runner/queue-monitor.cc at line 196 [12.20067]

Deletion in src/hydra-queue-runner/queue-monitor.cc at line 201 [12.20067]

Insertion in src/hydra-queue-runner/queue-monitor.cc at line 232 [12.20067]

Deletion in src/hydra-queue-runner/queue-monitor.cc at line 238 [12.20067]

Deletion in src/hydra-queue-runner/state.hh at line 350 [29.1128]

Replacement in src/hydra-queue-runner/state.hh at line 510 [29.1128]

Insertion in src/script/hydra-notify at line 8 [18.2994]

Insertion in src/script/hydra-notify at line 18 [18.2994]

Replacement in src/script/hydra-notify at line 21 [18.2994]

Replacement in src/script/hydra-notify at line 28 [18.2994]

Replacement in src/script/hydra-notify at line 39 [18.2994]

Insertion in src/script/hydra-notify at line 47 [18.2994]

Replacement in src/script/hydra-notify at line 49 [18.2994]

Insertion in src/script/hydra-notify at line 61 [18.2994]

Replacement in src/script/hydra-notify at line 65 [18.2994]

Replacement in src/script/hydra-notify at line 68 [18.2994]

Deletion in src/script/hydra-notify at line 71 [18.2994]

Replacement in src/script/hydra-notify at line 73 [18.2994]

Replacement in src/script/hydra-notify at line 84 [18.2994]