Move log compression to a plugin

[?]

Mar 13, 2017, 4:18 PM

UVQJBDHNWHTGOAEL426F6RZPD67QABBRP73E6QB5ND5OV2S3267QC

Dependencies

[2] LVQXQIYA Kill active build steps when builds are cancelled
[3] X4KYZJBQ Use latest nixUnstable
[4] 24BMQDZA Start of single-process hydra-queue-runner
[5] HJOEIMLR Refactor
[6] MHVIT4JY Split hydra-queue-runner.cc more
[7] PLOZBRTR Add command ‘hydra-queue-runner --status’ to show current status
[8] FCTX433O Add buildStarted plugin hook
[9] K5G5GZY7 Guard against concurrent invocations of hydra-queue-runner
[10] RQUAATWB Add status dump facility
[11] C6HOMHZW Don't try to handle SIGINT
[12] BRAESISH Warn if PostgreSQL appears stalled
[13] BG6PEOB2 Make the output size limit configurable
[14] GS4BE6TB Asynchronously compress build logs
[15] NTEDD7T4 Provide a plugin hook for when build steps finish
[16] XV4AEKJC hydra-queue-runner: Handle status queries on the main thread
[17] XCDTFZUY hydra-queue-runner: Fix build
[18] NAYQT2GT hydra-queue-runner: Use cmdBuildDerivation
[19] IWB3F4Z6 Fail builds with previously failed steps early
[20] IK2UBDAU Revive jobset scheduling
[21] NKQOEVVP Get rid of "will retry" messages after "maybe cancelling..."
[22] IE2PRAQU hydra-queue-runner: Send build notifications
[23] OG3Z3QGC Namespace cleanup
[24] NJJ7H64S Very basic multi-threaded queue runner
[25] 7VQ4ALFY Update "make check" for the new queue runner
[26] B2L4T3X6 Sync with Nix
[*] 7LWB2J2Z Periodically clear orphaned build steps
[*] 5EQYVRWE Add a plugin mechanism

Change contents

edit in src/hydra-queue-runner/builder.cc at line 148

[2.1038]

[28.235]

}

if (stepNr) {
/* 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});
edit in src/hydra-queue-runner/builder.cc at line 215

[4.4779]→[4.4779:4929](∅→∅),[4.4929]→[4.121:149](∅→∅),[4.149]→[2.1810:1884](∅→∅),[4.225]→[4.4984:5043](∅→∅),[2.1884]→[4.4984:5043](∅→∅),[4.4984]→[4.4984:5043](∅→∅)

/* Asynchronously compress the log. */
if (result.logFile != "") {
{
auto logCompressorQueue_(logCompressorQueue.lock());
assert(stepNr);
logCompressorQueue_->push({buildId, stepNr, result.logFile});
}
logCompressorWakeup.notify_one();
}
replacement in src/hydra-queue-runner/builder.cc at line 446

[4.14531]→[4.14531:14570](∅→∅)

if (quit) exit(0); // testing hack

[4.14531]

[4.14570]

if (quit) exit(0); // testing hack; FIXME: this won't run plugins
edit in src/hydra-queue-runner/hydra-queue-runner.cc at line 461

[4.3178]→[4.1361:1362](∅→∅),[4.1362]→[4.3178:3179](∅→∅),[4.3178]→[4.3178:3179](∅→∅),[4.3179]→[4.1363:1426](∅→∅),[4.1426]→[4.3179:3180](∅→∅),[4.3179]→[4.3179:3180](∅→∅),[4.3180]→[4.795:829](∅→∅),[4.829]→[4.1453:1656](∅→∅),[4.1453]→[4.1453:1656](∅→∅),[4.1656]→[4.830:883](∅→∅),[4.883]→[4.1712:1770](∅→∅),[4.1712]→[4.1712:1770](∅→∅)

void State::logCompressor()
{
while (true) {
try {

CompressionItem item;
{
auto logCompressorQueue_(logCompressorQueue.lock());
while (logCompressorQueue_->empty())
logCompressorQueue_.wait(logCompressorWakeup);
item = logCompressorQueue_->front();
logCompressorQueue_->pop();
}
edit in src/hydra-queue-runner/hydra-queue-runner.cc at line 462

[4.1771]→[4.884:937](∅→∅)

if (!pathExists(item.logPath)) continue;
edit in src/hydra-queue-runner/hydra-queue-runner.cc at line 463

[4.1820]→[4.938:1028](∅→∅),[4.1028]→[4.1905:1906](∅→∅),[4.1905]→[4.1905:1906](∅→∅),[4.1906]→[4.1029:1124](∅→∅),[4.1124]→[4.1955:2122](∅→∅),[4.1955]→[4.1955:2122](∅→∅),[4.2122]→[4.792:849](∅→∅),[4.849]→[4.2173:2245](∅→∅),[4.2173]→[4.2173:2245](∅→∅),[4.2245]→[4.1125:1204](∅→∅),[4.1204]→[4.1292:1346](∅→∅),[4.2319]→[4.1292:1346](∅→∅),[4.1346]→[4.2371:2388](∅→∅),[4.2371]→[4.2371:2388](∅→∅),[4.2388]→[3.149:183](∅→∅),[3.183]→[4.2426:2548](∅→∅),[4.2426]→[4.2426:2548](∅→∅),[4.2548]→[4.1205:1248](∅→∅),[4.1248]→[4.2586:2587](∅→∅),[4.2586]→[4.2586:2587](∅→∅),[4.2587]→[4.1249:1312](∅→∅),[4.1312]→[4.2661:2734](∅→∅),[4.2661]→[4.2661:2734](∅→∅),[4.2734]→[4.1313:1731](∅→∅),[4.1731]→[4.2853:3005](∅→∅),[4.2853]→[4.2853:3005](∅→∅),[4.3005]→[4.167:170](∅→∅),[4.170]→[4.1347:1348](∅→∅)

printMsg(lvlChatty, format("compressing log file ‘%1%’") % item.logPath);

Path dstPath = item.logPath + ".bz2";
Path tmpPath = dstPath + ".tmp";

AutoCloseFD fd = open(tmpPath.c_str(), O_CREAT | O_TRUNC | O_WRONLY, 0644);

// FIXME: use libbz2

Pid pid = startProcess([&]() {
if (dup2(fd.get(), STDOUT_FILENO) == -1)
throw SysError("cannot dup output pipe to stdout");
execlp("bzip2", "bzip2", "-c", item.logPath.c_str(), nullptr);
throw SysError("cannot start bzip2");
});

int res = pid.wait();

if (res != 0)
throw Error(format("bzip2 returned exit code %1% while compressing ‘%2%’")
% res % item.logPath);

if (rename(tmpPath.c_str(), dstPath.c_str()) != 0)
throw SysError(format("renaming ‘%1%’") % tmpPath);

if (unlink(item.logPath.c_str()) != 0)
throw SysError(format("unlinking ‘%1%’") % item.logPath);

/* Run plugins. We do this after log compression to ensure
that the log file doesn't change while the plugins may
be accessing it. */
enqueueNotificationItem({NotificationItem::Type::StepFinished, item.id, {}, item.stepNr, dstPath});

} catch (std::exception & e) {
printMsg(lvlError, format("log compressor: %1%") % e.what());
sleep(5);
}
}
}
edit in src/hydra-queue-runner/hydra-queue-runner.cc at line 806

[4.399]→[4.3010:3153](∅→∅)

/* Run a log compressor thread. If needed, we could start more
than one. */
std::thread(&State::logCompressor, this).detach();
edit in src/hydra-queue-runner/state.hh at line 324

[4.6085]→[4.6085:6122](∅→∅),[4.6122]→[4.2663:2842](∅→∅),[4.2842]→[4.2047:2096](∅→∅),[4.2047]→[4.2047:2096](∅→∅),[4.2096]→[4.6222:6223](∅→∅),[4.6222]→[4.6222:6223](∅→∅)

/* Log compressor work queue. */
struct CompressionItem
{
BuildID id;
unsigned int stepNr;
nix::Path logPath;
};
nix::Sync<std::queue<CompressionItem>> logCompressorQueue;
std::condition_variable logCompressorWakeup;
edit in src/hydra-queue-runner/state.hh at line 501

[4.8698]→[4.8698:8792](∅→∅)

/* Thread that asynchronously bzips logs of finished steps. */
void logCompressor();
file addition: CompressLog.pm (----------)

[29.1]

package Hydra::Plugin::CompressLog;

use strict;
use utf8;
use parent 'Hydra::Plugin';
use Hydra::Helper::CatalystUtils;

sub stepFinished {
my ($self, $step, $logPath) = @_;

my $doCompress = $self->{config}->{'compress_build_logs'} // "1";

if ($doCompress eq "1" && -e $logPath) {
print STDERR "compressing ‘$logPath’...\n";
system("bzip2", "--force", $logPath);
}
}

1;