diff --git a/lix/legacy/build-remote.cc b/lix/legacy/build-remote.cc index 8593f7ba3..989be86d8 100644 --- a/lix/legacy/build-remote.cc +++ b/lix/legacy/build-remote.cc @@ -184,7 +184,8 @@ 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(int intoFD) + std::future + startLogThread(const std::string & buildDescription, const std::string & drvPath) { if (!logPipe.readSide) { return {}; @@ -192,18 +193,56 @@ struct BuilderConnection logPipe.writeSide.close(); + // 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, - [](int from, int to) { + [this, buildDescription, drvPath](int from) { AsyncIoRoot aio; - auto reader = AIO().lowLevelProvider.wrapInputFd(from); - auto writer = AIO().lowLevelProvider.wrapOutputFd(to); + auto act = logger->startActivity( + lvlInfo, actBuild, buildDescription, Logger::Fields{drvPath, storeUri, 1, 1} + ); - reader->pumpTo(*writer).wait(aio.kj.waitScope); + 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(), - intoFD + logPipe.readSide.get() ); } }; @@ -420,12 +459,12 @@ kj::Promise Instance::build(BuildContext context) kj::Promise AcceptedBuild::run(RunContext context) { try { - const int logFD = (co_await buildLogger.getFd()).orDefault(-1); - if (logFD < 0) { - throw Error("build-hook needs a logFD from the builder to build"); - } - - auto logThread = builder.startLogThread(logFD); + auto logThread = 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; diff --git a/lix/libstore/build/derivation-goal.cc b/lix/libstore/build/derivation-goal.cc index 94d5ae2cb..d364e1bf9 100644 --- a/lix/libstore/build/derivation-goal.cc +++ b/lix/libstore/build/derivation-goal.cc @@ -608,21 +608,14 @@ try { co_return result::current_exception(); } -void DerivationGoal::started() +std::string DerivationGoal::buildDescription() const { - auto msg = fmt( - buildMode == bmRepair ? "repairing outputs of '%s'" : - buildMode == bmCheck ? "checking outputs of '%s'" : - "building '%s'", worker.store.printStorePath(drvPath)); - fmt("building '%s'", worker.store.printStorePath(drvPath)); - if (hook) msg += fmt(" on '%s'", machineName); - act = logger->startActivity( - lvlInfo, - actBuild, - msg, - Logger::Fields{worker.store.printStorePath(drvPath), hook ? machineName : "", 1, 1} + return fmt( + buildMode == bmRepair ? "repairing outputs of '%s'" + : buildMode == bmCheck ? "checking outputs of '%s'" + : "building '%s'", + worker.store.printStorePath(drvPath) ); - mcRunningBuilds = worker.runningBuilds.addTemporarily(1); } kj::Promise> DerivationGoal::tryToBuild() noexcept @@ -1047,16 +1040,7 @@ try { namespace { struct BuildHookLogger final : rpc::build_remote::HookInstance::BuildLogger::Server -{ - AutoCloseFD fd; - - BuildHookLogger(AutoCloseFD fd) : fd(std::move(fd)) {} - - kj::Maybe getFd() override - { - return fd.get(); - } -}; +{}; } kj::Promise> DerivationGoal::tryBuildHook() @@ -1075,12 +1059,6 @@ try { hook = TRY_AWAIT(HookInstance::create()); } - // open a pipe to receive logs directly from the hook - Pipe logPipe; - logPipe.create(); - builderOutFD = &logPipe.readSide; - KJ_DEFER(builderOutFD = nullptr); - KJ_DEFER(hook = nullptr); auto output = handleChildOutput(); @@ -1089,7 +1067,7 @@ try { RPC_FILL(buildReq, setNeededSystem, drv->platform); RPC_FILL(buildReq, initDrvPath, drvPath, worker.store); RPC_FILL(buildReq, initRequiredFeatures, parsedDrv->getRequiredSystemFeatures()); - buildReq.setBuildLogger(kj::heap(std::move(logPipe.writeSide))); + buildReq.setBuildLogger(kj::heap()); auto buildRespPromise = buildReq.send(); auto buildResp = TRY_AWAIT_RPC(buildRespPromise); @@ -1128,6 +1106,7 @@ try { missingOutputs.insert(outputName); } RPC_FILL(runReq, initWantedOutputs, missingOutputs); + RPC_FILL(runReq, setDescription, buildDescription()); } /* Create the log file and pipe. */ @@ -1138,7 +1117,7 @@ try { // build via hook is now properly running. wait for it to finish actLock.reset(); buildResult.startTime = time(0); // inexact - started(); + mcRunningBuilds = worker.runningBuilds.addTemporarily(1); auto result = co_await runPromise; diff --git a/lix/libstore/build/derivation-goal.hh b/lix/libstore/build/derivation-goal.hh index 0627e0d40..3becd9d3d 100644 --- a/lix/libstore/build/derivation-goal.hh +++ b/lix/libstore/build/derivation-goal.hh @@ -352,7 +352,7 @@ public: kj::Promise> repairClosure() noexcept; - void started(); + std::string buildDescription() const; WorkResult done( BuildResult::Status status, diff --git a/lix/libstore/build/hook-instance.capnp b/lix/libstore/build/hook-instance.capnp index 05840e8d5..810a068a7 100644 --- a/lix/libstore/build/hook-instance.capnp +++ b/lix/libstore/build/hook-instance.capnp @@ -9,13 +9,14 @@ using StoreTypes = import "/lix/libstore/types.capnp"; interface HookInstance { interface BuildLogger { - # only used for fd passing + # will be used later } interface AcceptedBuild { run @0 ( inputs :List(StoreTypes.StorePath), # actual a set wantedOutputs :List(Data), # actually StringSet + description :Text, # root activity description for this build ) -> (result :Types.ResultV); } diff --git a/lix/libstore/build/local-derivation-goal.cc b/lix/libstore/build/local-derivation-goal.cc index 6fa1c33d1..852c70333 100644 --- a/lix/libstore/build/local-derivation-goal.cc +++ b/lix/libstore/build/local-derivation-goal.cc @@ -279,7 +279,13 @@ retry: /* Okay, we have to build. */ TRY_AWAIT(startBuilder()); - started(); + act = logger->startActivity( + lvlInfo, + actBuild, + buildDescription(), + Logger::Fields{worker.store.printStorePath(drvPath), "", 1, 1} + ); + mcRunningBuilds = worker.runningBuilds.addTemporarily(1); if (auto error = TRY_AWAIT(handleChildOutput())) { co_return std::move(*error); }