libstore: don't treat legacy ssh like a sandbox

treat it like ssh-ng instead, and have the build hook do the translation
of ssh stdout to the json log message steam the derivaiton goal can use.
since the ssh-ng path also handles all store urls that aren't legacy ssh
we now have a single logging system that handles every remote build kind
equally, without requiring fd passing into the build hook. this is later
required to rpc-ify the log stream emitted by build hooks to the daemon.

Change-Id: Ifb522eb8a9745029050f16b1b3b3601a6ddac748

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<void> startLogThread(int intoFD)
+    std::future<void>
+    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<ActivityId, Activity> 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<char>(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<void> Instance::build(BuildContext context)
 kj::Promise<void> 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<std::string_view>(context.getParams().getDescription()),
+                builder.storeUri),
+            store->printStorePath(drvPath)
+        );
         KJ_DEFER({
             // drop any existing ssh connection so the log thread can exit
             builder.sshStore = nullptr;

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<Result<Goal::WorkResult>> 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<int> getFd() override
-    {
-        return fd.get();
-    }
-};
+{};
 }
 
 kj::Promise<Result<HookResult>> 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<BuildHookLogger>(std::move(logPipe.writeSide)));
+    buildReq.setBuildLogger(kj::heap<BuildHookLogger>());
     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;
 

lix/libstore/build/derivation-goal.hh

@@ -352,7 +352,7 @@ public:
 
     kj::Promise<Result<WorkResult>> repairClosure() noexcept;
 
-    void started();
+    std::string buildDescription() const;
 
     WorkResult done(
         BuildResult::Status status,

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

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