b1879132af
Yet again, manual testing is proving to be insufficient. I'm pretty sure I wrote this code but lost it in a rebase, or perhaps the switch to result classes. At any rate, this implements the actual "fetch a retry row and run it" for the hydra-notify daemon. Tested by hand.
139 lines
4.1 KiB
Perl
Executable File
139 lines
4.1 KiB
Perl
Executable File
#! /usr/bin/env perl
|
|
|
|
use strict;
|
|
use utf8;
|
|
use Getopt::Long;
|
|
use HTTP::Server::PSGI;
|
|
use Hydra::Event;
|
|
use Hydra::Event::BuildFinished;
|
|
use Hydra::Helper::AddBuilds;
|
|
use Hydra::Helper::Nix;
|
|
use Hydra::Plugin;
|
|
use Hydra::PostgresListener;
|
|
use Hydra::TaskDispatcher;
|
|
use Parallel::ForkManager;
|
|
use Prometheus::Tiny::Shared;
|
|
|
|
STDERR->autoflush(1);
|
|
STDOUT->autoflush(1);
|
|
binmode STDERR, ":encoding(utf8)";
|
|
|
|
my $config = getHydraConfig();
|
|
|
|
my $prom = Prometheus::Tiny::Shared->new;
|
|
# Note: It is very important to pre-declare any metrics before using them.
|
|
# Add a new declaration for any new metrics you create. Metrics which are
|
|
# not pre-declared disappear when their value is null. See:
|
|
# https://metacpan.org/pod/Prometheus::Tiny#declare
|
|
$prom->declare(
|
|
"event_loop_iterations",
|
|
type => "counter",
|
|
help => "Number of iterations through the event loop. Incremented at the start of the event loop."
|
|
);
|
|
$prom->declare(
|
|
"event_received",
|
|
type => "counter",
|
|
help => "Timestamp of the last time a new event was received."
|
|
);
|
|
$prom->declare(
|
|
"notify_event",
|
|
type => "counter",
|
|
help => "Number of events received on the given channel."
|
|
);
|
|
$prom->declare(
|
|
"notify_event_error",
|
|
type => "counter",
|
|
help => "Number of events received that were unprocessable by channel."
|
|
);
|
|
|
|
my $promCfg = Hydra::Helper::Nix::getHydraNotifyPrometheusConfig($config);
|
|
if (defined($promCfg)) {
|
|
print STDERR "Starting the Prometheus exporter, listening on http://${\$promCfg->{'listen_address'}}:${\$promCfg->{'port'}}/metrics.\n";
|
|
my $fork_manager = Parallel::ForkManager->new(1);
|
|
$fork_manager->start_child("metrics_exporter", sub {
|
|
my $server = HTTP::Server::PSGI->new(
|
|
host => $promCfg->{"listen_address"},
|
|
port => $promCfg->{"port"},
|
|
timeout => 1,
|
|
);
|
|
|
|
$server->run($prom->psgi);
|
|
});
|
|
} else {
|
|
print STDERR "Not starting the hydra-notify Prometheus exporter.\n";
|
|
}
|
|
|
|
my $queued_only;
|
|
|
|
GetOptions(
|
|
"queued-only" => \$queued_only
|
|
) or exit 1;
|
|
|
|
|
|
my $db = Hydra::Model::DB->new();
|
|
|
|
my @plugins = Hydra::Plugin->instantiate(db => $db, config => $config);
|
|
my $task_dispatcher = Hydra::TaskDispatcher->new(
|
|
$db,
|
|
$prom,
|
|
[@plugins],
|
|
sub {
|
|
my ($task) = @_;
|
|
$db->resultset("TaskRetries")->save_task($task);
|
|
}
|
|
);
|
|
|
|
my $dbh = $db->storage->dbh;
|
|
|
|
my $listener = Hydra::PostgresListener->new($dbh);
|
|
$listener->subscribe("build_started");
|
|
$listener->subscribe("build_finished");
|
|
$listener->subscribe("step_finished");
|
|
$listener->subscribe("hydra_notify_dump_metrics");
|
|
|
|
# Process builds that finished while hydra-notify wasn't running.
|
|
for my $build ($db->resultset('Builds')->search(
|
|
{ notificationpendingsince => { '!=', undef } }))
|
|
{
|
|
print STDERR "sending notifications for build ${\$build->id}...\n";
|
|
|
|
my $event = Hydra::Event->new_event("build_finished", $build->id);
|
|
$task_dispatcher->dispatch_event($event);
|
|
}
|
|
|
|
my $taskretries = $db->resultset('TaskRetries');
|
|
|
|
# Process incoming notifications.
|
|
while (!$queued_only) {
|
|
$prom->inc("event_loop_iterations");
|
|
my $messages = $listener->block_for_messages($taskretries->get_seconds_to_next_retry());
|
|
while (my $message = $messages->()) {
|
|
$prom->set("event_received", time());
|
|
my $channelName = $message->{"channel"};
|
|
my $pid = $message->{"pid"};
|
|
my $payload = $message->{"payload"};
|
|
|
|
$prom->inc("notify_event", { channel => $channelName });
|
|
|
|
if ($channelName eq "hydra_notify_dump_metrics") {
|
|
print STDERR "Dumping prometheus metrics:\n${\$prom->format}\n";
|
|
next;
|
|
}
|
|
|
|
eval {
|
|
my $event = Hydra::Event->new_event($channelName, $message->{"payload"});
|
|
$task_dispatcher->dispatch_event($event);
|
|
|
|
1;
|
|
} or do {
|
|
$prom->inc("notify_event_error", { channel => $channelName });
|
|
print STDERR "error processing message '$payload' on channel '$channelName': $@\n";
|
|
}
|
|
}
|
|
|
|
my $task = $taskretries->get_retryable_task();
|
|
if (defined($task)) {
|
|
$task_dispatcher->dispatch_task($task);
|
|
}
|
|
}
|