From 31b6eb27861217cb8829d1e1ced392bc494eeaa4 Mon Sep 17 00:00:00 2001 From: eldritch horrors Date: Sat, 18 Oct 2025 19:43:44 +0200 Subject: [PATCH] libstore: convert build-hook logging to rpc Change-Id: I0c20f89de113dce6032c93a32e9fcd43b4478f55 --- lix/legacy/build-remote.cc | 271 +++++++++++--------- lix/libstore/build/derivation-goal.cc | 134 ++++------ lix/libstore/build/derivation-goal.hh | 1 - lix/libstore/build/hook-instance.capnp | 10 +- lix/libstore/build/hook-instance.cc | 62 ++++- lix/libstore/build/hook-instance.hh | 30 ++- lix/libstore/build/local-derivation-goal.cc | 4 - lix/libstore/build/local-derivation-goal.hh | 2 +- 8 files changed, 268 insertions(+), 246 deletions(-) diff --git a/lix/legacy/build-remote.cc b/lix/legacy/build-remote.cc index 2af4542ff..27521446f 100644 --- a/lix/legacy/build-remote.cc +++ b/lix/legacy/build-remote.cc @@ -3,16 +3,15 @@ #include "lix/libutil/c-calls.hh" #include "lix/libutil/error.hh" #include "lix/libutil/file-descriptor.hh" +#include "lix/libutil/logging-rpc.hh" #include "lix/libutil/logging.hh" #include "lix/libutil/rpc.hh" -#include "lix/libutil/types-rpc.hh" +#include "lix/libutil/types-rpc.hh" // IWYU pragma: keep #include "lix/libutil/types.hh" #include #include -#include #include #include -#include #include #include #include @@ -47,10 +46,11 @@ namespace { struct Instance final : rpc::build_remote::HookInstance::Server { unsigned int maxBuildJobs; - bool initialized = false; + bool initialized = false, used = false; kj::Promise init(InitContext context) override; + kj::Promise buildImpl(BuildContext context); kj::Promise build(BuildContext context) override; }; } @@ -188,11 +188,10 @@ struct BuilderConnection // start the thread that reads ssh stderr and turns it into log items. // this future *must* outlive sshStore, otherwise it will never finish - std::future - startLogThread(const std::string & buildDescription, const std::string & drvPath) - { + kj::Promise> startLogThread(std::string buildDescription, std::string drvPath) + try { if (!logPipe.readSide) { - return {}; + co_return result::success(); } logPipe.writeSide.close(); @@ -200,54 +199,51 @@ struct BuilderConnection // NOTE this is very similar to handleBuilderOutput in DerivationGoal, but unlike // the derivation goal we do not need to handle EIO from a pty here. we also have // no timeouts or limits to keep track of, which makes deduplication less useful. - return std::async( - std::launch::async, - [this, buildDescription, drvPath](int from) { - AsyncIoRoot aio; - - auto act = logger->startActivity( - lvlInfo, actBuild, buildDescription, Logger::Fields{drvPath, storeUri, 1, 1} - ); - - std::map activities; - - auto reader = AIO().lowLevelProvider.wrapInputFd(from); - - LogLineSplitter splitter; - - auto flushLine = [&](const std::string & line) { - if (const auto state = - handleJSONLogMessage(line, act, activities, "the derivation builder")) - { - if (state == Logger::BufferState::NeedsFlush) { - aio.blockOn(act.getLogger().flush()); - } - } else { - ACTIVITY_RESULT_SYNC(aio, act, resBuildLogLine, line); - } - }; - - auto buf = kj::heapArray(4096); - while (true) { - const auto got = aio.blockOn(reader->tryRead(buf.begin(), 1, buf.size())); - if (got == 0) { - break; - } - - std::string_view data{buf.begin(), got}; - while (!data.empty()) { - if (auto line = splitter.feed(data)) { - flushLine(*line); - } - } - } - - if (auto left = splitter.finish(); !left.empty()) { - flushLine(left); - } - }, - logPipe.readSide.get() + auto act = logger->startActivity( + lvlInfo, actBuild, buildDescription, Logger::Fields{drvPath, storeUri, 1, 1} ); + + std::map activities; + + auto reader = AIO().lowLevelProvider.wrapInputFd(logPipe.readSide.get()); + + LogLineSplitter splitter; + + auto flushLine = [&](const std::string & line) { + if (const auto state = + handleJSONLogMessage(line, act, activities, "the derivation builder")) + { + return *state; + } else { + return act.result(resBuildLogLine, line); + } + }; + + auto buf = kj::heapArray(4096); + while (true) { + const auto got = co_await reader->tryRead(buf.begin(), 1, buf.size()); + if (got == 0) { + break; + } + + std::string_view data{buf.begin(), got}; + while (!data.empty()) { + if (auto line = splitter.feed(data)) { + if (flushLine(*line) == Logger::BufferState::NeedsFlush) { + TRY_AWAIT(act.getLogger().flush()); + } + } + } + } + + if (auto line = splitter.finish(); !line.empty()) { + (void) flushLine(line); + TRY_AWAIT(act.getLogger().flush()); + } + + co_return result::success(); + } catch (...) { + co_return result::current_exception(); } }; @@ -256,21 +252,16 @@ struct AcceptedBuild final : rpc::build_remote::HookInstance::AcceptedBuild::Ser ref store; StorePath drvPath; BuilderConnection builder; - rpc::build_remote::HookInstance::BuildLogger::Client buildLogger; + bool used = false; - AcceptedBuild( - ref store, - StorePath drvPath, - BuilderConnection builder, - rpc::build_remote::HookInstance::BuildLogger::Client buildLogger - ) + AcceptedBuild(ref store, StorePath drvPath, BuilderConnection builder) : store(store) , drvPath(drvPath) , builder(std::move(builder)) - , buildLogger(std::move(buildLogger)) { } + kj::Promise runImpl(RunContext context); kj::Promise run(RunContext context) override; }; @@ -365,8 +356,6 @@ try { static int main_build_remote(AsyncIoRoot & aio, std::string programName, Strings argv) { { - logger = makeJSONLogger(*logger); - /* Ensure we don't get any SSH passphrase or host key popups. */ unsetenv("DISPLAY"); unsetenv("SSH_ASKPASS"); @@ -393,6 +382,8 @@ kj::Promise Instance::init(InitContext context) throw Error("build hook can only be initialized once"); } + logger = rpc::log::makeRpcLoggerClient(context.getParams().getLogger()); + /* Read the parent's settings. */ for (const auto & [name, value] : rpc::to(context.getParams().getSettings())) { settings.set(name, value); @@ -413,85 +404,109 @@ kj::Promise Instance::init(InitContext context) return kj::READY_NOW; } -kj::Promise Instance::build(BuildContext context) +kj::Promise Instance::buildImpl(BuildContext context) { - try { - if (!initialized) { - throw Error("build hook not fully initialized"); - } + if (!initialized) { + throw Error("build hook not fully initialized"); + } - // FIXME this does not open a daemon connection for historical reasons. - // we may create a lot of build hook instances, and having each of them - // also create a daemon instance is inefficient and wasteful. in future - // versions of the build hook (where we don't need one hook process per - // build) we should change this to using a daemon connection, ideally a - // daemon connection provided by the parent via file descriptor passing - auto store = TRY_AWAIT(openStore(settings.storeUri, {}, AllowDaemon::Disallow)); + // FIXME this does not open a daemon connection for historical reasons. + // we may create a lot of build hook instances, and having each of them + // also create a daemon instance is inefficient and wasteful. in future + // versions of the build hook (where we don't need one hook process per + // build) we should change this to using a daemon connection, ideally a + // daemon connection provided by the parent via file descriptor passing + auto store = TRY_AWAIT(openStore(settings.storeUri, {}, AllowDaemon::Disallow)); - /* It would be more appropriate to use $XDG_RUNTIME_DIR, since - that gets cleared on reboot, but it wouldn't work on macOS. */ - auto currentLoadName = "/current-load"; - if (auto localStore = store.try_cast_shared()) - currentLoad = std::string { localStore->config().stateDir } + currentLoadName; - else - currentLoad = settings.nixStateDir + currentLoadName; + /* It would be more appropriate to use $XDG_RUNTIME_DIR, since + that gets cleared on reboot, but it wouldn't work on macOS. */ + auto currentLoadName = "/current-load"; + if (auto localStore = store.try_cast_shared()) { + currentLoad = std::string{localStore->config().stateDir} + currentLoadName; + } else { + currentLoad = settings.nixStateDir + currentLoadName; + } - auto machines = getMachines(); - debug("got %d remote builders", machines.size()); + auto machines = getMachines(); + debug("got %d remote builders", machines.size()); - if (machines.empty()) { - context.getResults().initResult().initGood().setDeclinePermanently(); + if (machines.empty()) { + context.getResults().initResult().initGood().setDeclinePermanently(); + co_return; + } + + auto amWilling = context.getParams().getAmWilling(); + auto neededSystem = rpc::to(context.getParams().getNeededSystem()); + auto drvPath = from(context.getParams().getDrvPath(), *store); + auto requiredFeatures = + rpc::to>(context.getParams().getRequiredFeatures()); + + auto result = TRY_AWAIT(connectToBuilder( + store, drvPath, machines, maxBuildJobs, amWilling, neededSystem, requiredFeatures + )); + + if (auto immediateResponse = std::get_if(&result)) { + switch (*immediateResponse) { + case BuildRejected::Temporarily: + context.getResults().initResult().initGood().setPostpone(); + co_return; + case BuildRejected::Permanently: + context.getResults().initResult().initGood().setDecline(); co_return; } - - auto amWilling = context.getParams().getAmWilling(); - auto neededSystem = rpc::to(context.getParams().getNeededSystem()); - auto drvPath = from(context.getParams().getDrvPath(), *store); - auto requiredFeatures = - rpc::to>(context.getParams().getRequiredFeatures()); - auto buildLogger = context.getParams().getBuildLogger(); - - auto result = TRY_AWAIT(connectToBuilder( - store, drvPath, machines, maxBuildJobs, amWilling, neededSystem, requiredFeatures - )); - - if (auto immediateResponse = std::get_if(&result)) { - switch (*immediateResponse) { - case BuildRejected::Temporarily: - context.getResults().initResult().initGood().setPostpone(); - co_return; - case BuildRejected::Permanently: - context.getResults().initResult().initGood().setDecline(); - co_return; - } - } - - auto builder = std::get_if(&result); - assert(builder); - - auto ac = context.getResults().initResult().initGood().initAccept(); - ac.setMachine(kj::heap(store, drvPath, std::move(*builder), buildLogger)); - } catch (...) { - RPC_FILL(context.getResults(), initResult, std::current_exception()); } + + auto builder = std::get_if(&result); + assert(builder); + + auto ac = context.getResults().initResult().initGood().initAccept(); + ac.setMachine(kj::heap(store, drvPath, std::move(*builder))); +} + +kj::Promise Instance::build(BuildContext context) +try { + if (used) { + throw Error("build hooks can only accept a single job"); + } + used = true; // lock out other rpc calls during processing + co_await buildImpl(context); + TRY_AWAIT(logger->flush()); + used = context.getResults().getResult().getGood().isAccept(); +} catch (...) { + RPC_FILL(context.getResults(), getResult, std::current_exception()); } kj::Promise AcceptedBuild::run(RunContext context) { try { - auto logThread = builder.startLogThread( + auto oldLogger = logger; + logger = rpc::log::makeRpcLoggerClient(context.getParams().getLogger()); + TRY_AWAIT(oldLogger->flush()); + KJ_DEFER({ + delete logger; + logger = oldLogger; + }); + + if (used) { + throw Error("build hooks builds are single-use items"); + } + used = true; + co_await runImpl(context); + TRY_AWAIT(logger->flush()); + } catch (...) { + RPC_FILL(context.getResults(), getResult, std::current_exception()); + } +} + +kj::Promise AcceptedBuild::runImpl(RunContext context) +{ + try { + auto logHandler = builder.startLogThread( fmt("%s on '%s'", rpc::to(context.getParams().getDescription()), builder.storeUri), store->printStorePath(drvPath) ); - KJ_DEFER({ - // drop any existing ssh connection so the log thread can exit - builder.sshStore = nullptr; - if (logThread.valid()) { - logThread.get(); - } - }); auto & sshStore = builder.sshStore; auto & storeUri = builder.storeUri; @@ -602,6 +617,10 @@ kj::Promise AcceptedBuild::run(RunContext context) ); } + // drop store connection, let log handler process any remaining input + builder.sshStore = nullptr; + TRY_AWAIT(logHandler); + context.getResults().initResult().setGood(); } catch (...) { RPC_FILL(context.getResults(), initResult, std::current_exception()); diff --git a/lix/libstore/build/derivation-goal.cc b/lix/libstore/build/derivation-goal.cc index a46afa5ec..799363154 100644 --- a/lix/libstore/build/derivation-goal.cc +++ b/lix/libstore/build/derivation-goal.cc @@ -24,12 +24,14 @@ #include #include #include +#include #include #include #include #include #include #include +#include #include #include #include @@ -798,10 +800,7 @@ int DerivationGoal::getChildStatus() return hook->kill(); } -void DerivationGoal::closeReadPipes() -{ - hook->fromHook.reset(); -} +void DerivationGoal::closeReadPipes() {} void DerivationGoal::cleanupHookFinally() { @@ -1035,8 +1034,22 @@ try { } namespace { -struct BuildHookLogger final : rpc::build_remote::HookInstance::BuildLogger::Server -{}; +struct ActivityTrackingHookLogger final : HookInstance::HookLogger +{ + kj::TimePoint & tracker; + + ActivityTrackingHookLogger(const Activity & act, FinishSink * logSink, kj::TimePoint & tracker) + : HookLogger(act, logSink) + , tracker(tracker) + { + } + + kj::Promise push(PushContext context) override + { + tracker = AIO().provider.getTimer().now(); + return HookLogger::push(context); + } +}; } kj::Promise> DerivationGoal::tryBuildHook() @@ -1052,18 +1065,16 @@ try { hook = std::move(worker.hook.instances.front()); worker.hook.instances.pop_front(); } else { - hook = TRY_AWAIT(HookInstance::create()); + hook = TRY_AWAIT(HookInstance::create(worker.act)); } KJ_DEFER(hook = nullptr); - auto output = wrapChildHandler(handleRawChildStream()); auto buildReq = hook->rpc->buildRequest(); RPC_FILL(buildReq, setAmWilling, slotToken.valid()); RPC_FILL(buildReq, setNeededSystem, drv->platform); RPC_FILL(buildReq, initDrvPath, drvPath, worker.store); RPC_FILL(buildReq, initRequiredFeatures, parsedDrv->getRequiredSystemFeatures()); - buildReq.setBuildLogger(kj::heap()); auto buildRespPromise = buildReq.send(); auto buildResp = TRY_AWAIT_RPC(buildRespPromise); @@ -1085,9 +1096,15 @@ try { // the build was accepted by the hook, we can free the slot for another build now hookSlot = {}; + /* Create the log file and pipe. */ + openLogFile(); + auto runReq = buildResp.getAccept().getMachine().runRequest(); /* Tell the hook all the inputs that have to be copied to the remote system. */ + runReq.setLogger( + kj::heap(worker.act, logSink.get(), lastChildActivity) + ); RPC_FILL(runReq, initInputs, inputPaths, worker.store); /* Tell the hooks the missing outputs that have to be copied back @@ -1103,9 +1120,6 @@ try { RPC_FILL(runReq, setDescription, buildDescription()); } - /* Create the log file and pipe. */ - openLogFile(); - auto runPromise = runReq.send(); // build via hook is now properly running. wait for it to finish @@ -1113,21 +1127,30 @@ try { buildResult.startTime = time(0); // inexact mcRunningBuilds = worker.runningBuilds.addTemporarily(1); - auto result = co_await runPromise; + auto result = TRY_AWAIT(wrapChildHandler( + runPromise + .then( + // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines) + [&](auto result) -> kj::Promise>> { + try { + std::shared_ptr remoteError; + if (result.getResult().isBad()) { + remoteError = + std::make_shared(from(result.getResult().getBad())); + logErrorInfo(remoteError->info().level, remoteError->info()); + } + // close the rpc connection to have the hook exit + hook->rpc = nullptr; + hook->wait(); + co_return TRY_AWAIT(buildDone(remoteError)); + } catch (...) { + co_return result::current_exception(); + } + } + ) + )); - // close the rpc connection to have the hook exit - hook->rpc = nullptr; - - if (auto error = TRY_AWAIT(output)) { - co_return HookResult::Accept{std::move(*error)}; - } - - std::shared_ptr remoteError; - if (result.getResult().isBad()) { - remoteError = std::make_shared(from(result.getResult().getBad())); - logErrorInfo(remoteError->info().level, remoteError->info()); - } - co_return HookResult::Accept{TRY_AWAIT(buildDone(remoteError))}; + co_return HookResult::Accept{std::move(*result)}; } catch (...) { co_return result::current_exception(); } @@ -1225,65 +1248,6 @@ Goal::WorkResult DerivationGoal::tooMuchLogs() getName(), settings.maxLogSize)); } -kj::Promise>> DerivationGoal::handleRawChildStream() noexcept -try { - assert(hook); - - std::string currentHookLine; - - AsyncFdIoStream in(AsyncFdIoStream::shared_fd{}, hook->fromHook.get()); - - auto buf = kj::heapArray(4096); - while (true) { - const auto got = TRY_AWAIT(in.read(buf.begin(), buf.size())); - if (!got) { - co_return std::nullopt; - } - - std::string_view data = {buf.begin(), *got}; - lastChildActivity = AIO().provider.getTimer().now(); - - for (auto c : data) - if (c == '\n') { - auto json = parseJSONMessage(currentHookLine, "the derivation builder"); - if (json) { - auto s = handleJSONLogMessage( - *json, worker.act, hook->activities, "the derivation builder" - ); - // ensure that logs from a builder using `ssh-ng://` as protocol - // are also available to `nix log`. - if (s && logSink) { - const auto type = (*json)["type"]; - const auto fields = (*json)["fields"]; - if (type == resBuildLogLine) { - const std::string logLine = - (fields.size() > 0 ? fields[0].get() : "") + "\n"; - (*logSink)(logLine); - } else if (type == resSetPhase && ! fields.is_null()) { - const auto phase = fields[0]; - if (! phase.is_null()) { - // nixpkgs' stdenv produces lines in the log to signal - // phase changes. - // We want to get the same lines in case of remote builds. - // The format is: - // @nix { "action": "setPhase", "phase": "$curPhase" } - const auto logLine = JSON::object({ - {"action", "setPhase"}, - {"phase", phase} - }); - (*logSink)("@nix " + logLine.dump(-1, ' ', false, JSON::error_handler_t::replace) + "\n"); - } - } - } - } - currentHookLine.clear(); - } else - currentHookLine += c; - } -} catch (...) { - co_return result::current_exception(); -} - kj::Promise>> DerivationGoal::wrapChildHandler(kj::Promise>> handler) noexcept { diff --git a/lix/libstore/build/derivation-goal.hh b/lix/libstore/build/derivation-goal.hh index 9e95161e5..59f548ef2 100644 --- a/lix/libstore/build/derivation-goal.hh +++ b/lix/libstore/build/derivation-goal.hh @@ -311,7 +311,6 @@ protected: kj::Promise>> wrapChildHandler(kj::Promise>> handler) noexcept; - virtual kj::Promise>> handleRawChildStream() noexcept; kj::Promise>> monitorForSilence() noexcept; WorkResult tooMuchLogs(); diff --git a/lix/libstore/build/hook-instance.capnp b/lix/libstore/build/hook-instance.capnp index cada46b39..40e60c870 100644 --- a/lix/libstore/build/hook-instance.capnp +++ b/lix/libstore/build/hook-instance.capnp @@ -5,15 +5,13 @@ $Cxx.namespace("nix::rpc::build_remote"); $Cxx.allowCancellation; using Types = import "/lix/libutil/types.capnp"; +using Log = import "/lix/libutil/logging.capnp"; using StoreTypes = import "/lix/libstore/types.capnp"; interface HookInstance { - interface BuildLogger { - # will be used later - } - interface AcceptedBuild { run @0 ( + logger :Log.LogStream, inputs :List(StoreTypes.StorePath), # actual a set wantedOutputs :List(Data), # actually StringSet description :Text, # root activity description for this build @@ -32,13 +30,13 @@ interface HookInstance { } init @0 ( - settings :Types.Settings + logger :Log.LogStream, + settings :Types.Settings, ) -> (result :Types.ResultV); build @1 ( amWilling :Bool, neededSystem :Data, drvPath :StoreTypes.StorePath, requiredFeatures :List(Data), - buildLogger :BuildLogger, ) -> (result :Types.Result(BuildResponse)); } diff --git a/lix/libstore/build/hook-instance.cc b/lix/libstore/build/hook-instance.cc index 7e050d37e..b9074b847 100644 --- a/lix/libstore/build/hook-instance.cc +++ b/lix/libstore/build/hook-instance.cc @@ -4,15 +4,63 @@ #include "lix/libutil/file-system.hh" #include "lix/libstore/globals.hh" #include "lix/libstore/build/hook-instance.hh" +#include "lix/libutil/json.hh" +#include "lix/libutil/logging.hh" +#include "lix/libutil/result.hh" #include "lix/libutil/rpc.hh" +#include "lix/libutil/serialise.hh" #include "lix/libutil/strings.hh" +#include "lix/libutil/logging-rpc.hh" // IWYU pragma: keep #include "lix/libutil/types-rpc.hh" // IWYU pragma: keep #include #include +#include namespace nix { -kj::Promise>> HookInstance::create() +void HookInstance::HookLogger::emitLog(rpc::log::Event::Result::Reader r) +{ + auto type = rpc::log::from(r.getType()); + auto fields = r.getFields(); + if (!type) { + return; + } + + // ensure that logs from a builder using `ssh-ng://` as protocol + // are also available to `nix log`. + if (type == resBuildLogLine) { + if (fields.size() > 0 && fields[0].isS()) { + (*logSink)(fmt("%s\n", rpc::to(fields[0].getS()))); + } else { + (*logSink)("\n"); + } + } else if (type == resSetPhase && fields.size() > 0 && fields[0].isS()) { + // nixpkgs' stdenv produces lines in the log to signal phase changes. + // We want to get the same lines in case of remote builds. + // The format is: + // @nix { "action": "setPhase", "phase": "$curPhase" } + const auto phase = rpc::to(fields[0].getS()); + const auto logLine = JSON::object({{"action", "setPhase"}, {"phase", phase}}); + (*logSink)("@nix " + logLine.dump(-1, ' ', false, JSON::error_handler_t::replace) + "\n"); + } +} + +kj::Promise HookInstance::HookLogger::push(PushContext context) +{ + try { + auto e = context.getParams().getE(); + if (logSink && e.isResult()) { + emitLog(e.getResult()); + } + } catch (std::exception & e) { // NOLINT(lix-foreign-exceptions) + printError("error in log processor: %s", e.what()); + throw; // NOLINT(lix-foreign-exceptions) + } + + return RpcLoggerServer::push(context); +} + +kj::Promise>> HookInstance::create(const Activity & act) try { debug("starting build hook '%s'", concatStringsSep(" ", settings.buildHook.get())); @@ -32,10 +80,6 @@ try { args.push_back(std::to_string(verbosity)); - /* Create a pipe to get the output of the child. */ - Pipe fromHook_; - fromHook_.create(); - /* Create the communication pipes. */ auto [selfRPC, hookRPC] = SocketPair::stream(); @@ -43,9 +87,6 @@ try { /* Fork the hook. */ auto pid = startProcess([&]() { - if (dup2(fromHook_.writeSide.get(), STDERR_FILENO) == -1) - throw SysError("cannot pipe standard error into log file"); - commonExecveingChildInit(); if (chdir("/") == -1) throw SysError("changing into /"); @@ -71,14 +112,13 @@ try { { auto initReq = rpc.initRequest(); + initReq.setLogger(kj::heap(act, nullptr)); RPC_FILL(initReq, initSettings, settings); TRY_AWAIT_RPC(initReq.send()); } co_return std::make_unique( - std::move(fromHook_.readSide), - kj::heap(std::move(rpc)).attach(std::move(conn), std::move(client)), - std::move(pid) + kj::heap(std::move(rpc)).attach(std::move(conn), std::move(client)), std::move(pid) ); } catch (...) { co_return result::current_exception(); diff --git a/lix/libstore/build/hook-instance.hh b/lix/libstore/build/hook-instance.hh index 5bddf3ba7..26f8413e1 100644 --- a/lix/libstore/build/hook-instance.hh +++ b/lix/libstore/build/hook-instance.hh @@ -2,9 +2,11 @@ ///@file #include "hook-instance.capnp.h" +#include "lix/libutil/logging-rpc.hh" #include "lix/libutil/logging.hh" #include "lix/libutil/processes.hh" #include "lix/libutil/serialise.hh" +#include "logging.capnp.h" #include #include #include @@ -14,22 +16,26 @@ namespace nix { struct HookInstance { - /** - * Pipe for the hook's standard output/error. - */ - AutoCloseFD fromHook; + struct HookLogger : rpc::log::RpcLoggerServer + { + FinishSink * logSink; + + HookLogger(const Activity & act, FinishSink * logSink) + : rpc::log::RpcLoggerServer(act) + , logSink(logSink) + { + } + + void emitLog(rpc::log::Event::Result::Reader r); + kj::Promise push(PushContext context) override; + }; kj::Own rpc; - std::map activities; + static kj::Promise>> create(const Activity & act); - static kj::Promise>> create(); - - HookInstance( - AutoCloseFD fromHook, kj::Own rpc, Pid pid - ) - : fromHook(std::move(fromHook)) - , rpc(std::move(rpc)) + HookInstance(kj::Own rpc, Pid pid) + : rpc(std::move(rpc)) , pidOrStatus(std::move(pid)) { } diff --git a/lix/libstore/build/local-derivation-goal.cc b/lix/libstore/build/local-derivation-goal.cc index 2a2fedde0..67ce34b4b 100644 --- a/lix/libstore/build/local-derivation-goal.cc +++ b/lix/libstore/build/local-derivation-goal.cc @@ -2687,10 +2687,6 @@ StorePath LocalDerivationGoal::makeFallbackPath(const StorePath & path) kj::Promise>> LocalDerivationGoal::handleRawChildStream() noexcept try { - if (hook) { - co_return TRY_AWAIT(DerivationGoal::handleRawChildStream()); - } - AsyncFdIoStream in(AsyncFdIoStream::shared_fd{}, builderOutPTY.get()); std::map builderActivities; diff --git a/lix/libstore/build/local-derivation-goal.hh b/lix/libstore/build/local-derivation-goal.hh index 284a3cb0b..b705f7227 100644 --- a/lix/libstore/build/local-derivation-goal.hh +++ b/lix/libstore/build/local-derivation-goal.hh @@ -324,7 +324,7 @@ protected: */ virtual Pid startChild(std::function openSlave); - kj::Promise>> handleRawChildStream() noexcept override; + kj::Promise>> handleRawChildStream() noexcept; /** * Set up the system call filtering required for the sandbox.