From 57fb4f534d2ae73cffb5673c32164bc7a93e6d47 Mon Sep 17 00:00:00 2001 From: eldritch horrors Date: Thu, 16 Jul 2026 21:40:43 +0200 Subject: [PATCH] libstore: handle ssh{,-ng}:// logs in their store class build-remote should not be the only thing that ever reads logs from remote store connections. ssh-ng is better off here because it does not transport its logs on the ssh stderr channel, but ssh *does* do this. neither have any ssh logs shown after connection setup except when using an ssh:// url as a remote builder, which is not markedly helpful for debugging anything at all. let's have both kinds of ssh store classes handle logs themselves to alleviate all of that. this also improves the utility of remote builder checks in `nix doctor`. Change-Id: I23947a9666bc07561eac341578ed45993f1b9839 --- lix/legacy/build-remote.cc | 83 +----------- lix/libstore/build/derivation-goal.cc | 1 - lix/libstore/build/hook-instance.capnp | 1 - lix/libstore/legacy-ssh-store.cc | 166 ++++++++++++++++-------- lix/libstore/machines.cc | 8 +- lix/libstore/machines.hh | 2 +- lix/libstore/remote-store-connection.hh | 8 ++ lix/libstore/remote-store.cc | 11 +- lix/libstore/ssh-store.cc | 85 ++++++++---- lix/libstore/ssh.cc | 17 ++- lix/libstore/ssh.hh | 6 +- lix/nix/doctor.cc | 2 +- 12 files changed, 211 insertions(+), 179 deletions(-) diff --git a/lix/legacy/build-remote.cc b/lix/legacy/build-remote.cc index 47ddba0ad..2311e6d11 100644 --- a/lix/legacy/build-remote.cc +++ b/lix/legacy/build-remote.cc @@ -185,67 +185,6 @@ struct BuilderConnection AutoCloseFD slotLock; std::shared_ptr sshStore; std::string storeUri; - Pipe logPipe; - - // start the thread that reads ssh stderr and turns it into log items. - // this future *must* outlive sshStore, otherwise it will never finish - kj::Promise> startLogThread(std::string buildDescription, std::string drvPath) - try { - if (!logPipe.readSide) { - co_return result::success(); - } - - 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. - 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(); - } }; struct AcceptedBuild final : rpc::build_remote::HookInstance::AcceptedBuild::Server @@ -327,22 +266,16 @@ try { lock.reset(); std::shared_ptr sshStore; - Pipe logPipe; try { auto act = logger->startActivity(lvlTalkative, actUnknown, fmt("connecting to '%s'", bestMachine->name)); - std::tie(sshStore, logPipe) = TRY_AWAIT(bestMachine->openStore()); + sshStore = TRY_AWAIT(bestMachine->openStore()); TRY_AWAIT(sshStore->connect()); - co_return BuilderConnection{ - std::move(bestSlotLock), sshStore, bestMachine->storeUri, std::move(logPipe) - }; + co_return BuilderConnection{std::move(bestSlotLock), sshStore, bestMachine->storeUri}; } catch (std::exception & e) { // NOLINT(lix-foreign-exceptions) - std::string msg = logPipe.readSide ? chomp(drainFD(logPipe.readSide.get(), false)) : ""; - printError( - "cannot build on '%s': %s%s", bestMachine->name, e.what(), msg.empty() ? "" : ": " + msg - ); + printError("cannot build on '%s': %s", bestMachine->name, e.what()); bestMachine->enabled = false; } } @@ -502,13 +435,6 @@ kj::Promise AcceptedBuild::run(RunContext context) 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) - ); - auto & sshStore = builder.sshStore; auto & storeUri = builder.storeUri; @@ -618,9 +544,6 @@ kj::Promise> AcceptedBuild::runImpl(RunContext context) ); } - // drop store connection, let log handler process any remaining input - builder.sshStore = nullptr; - TRY_AWAIT(logHandler); co_return result::success(); } catch (...) { co_return result::current_exception(); diff --git a/lix/libstore/build/derivation-goal.cc b/lix/libstore/build/derivation-goal.cc index f0f5e312f..2ee96af66 100644 --- a/lix/libstore/build/derivation-goal.cc +++ b/lix/libstore/build/derivation-goal.cc @@ -1136,7 +1136,6 @@ try { missingOutputs.insert(outputName); } RPC_FILL(runReq, initWantedOutputs, missingOutputs); - RPC_FILL(runReq, setDescription, buildDescription()); } auto runPromise = LIX_WRAP_RPC_PROMISE_V1(runReq.send()); diff --git a/lix/libstore/build/hook-instance.capnp b/lix/libstore/build/hook-instance.capnp index f19cfde31..6cf4ac4a4 100644 --- a/lix/libstore/build/hook-instance.capnp +++ b/lix/libstore/build/hook-instance.capnp @@ -14,7 +14,6 @@ interface HookInstance $Types.throws(Types.v1Errors) { logger :Log.LogStream, inputs :List(StoreTypes.StorePath), # actual a set wantedOutputs :List(Data), # actually StringSet - description :Text, # root activity description for this build ); } diff --git a/lix/libstore/legacy-ssh-store.cc b/lix/libstore/legacy-ssh-store.cc index 7ea1a69b3..ea4240332 100644 --- a/lix/libstore/legacy-ssh-store.cc +++ b/lix/libstore/legacy-ssh-store.cc @@ -1,4 +1,6 @@ #include "lix/libstore/legacy-ssh-store.hh" +#include "libutil/error.hh" +#include "libutil/logging.hh" #include "lix/libutil/archive.hh" #include "lix/libutil/async-io.hh" #include "lix/libutil/async.hh" @@ -20,6 +22,8 @@ #include "path-info.hh" #include "path.hh" #include +#include +#include #include namespace nix { @@ -95,25 +99,25 @@ struct LegacySSHStoreConfig : CommonSSHStoreConfig } }; -struct LegacySSHStoreConfigWithLog : LegacySSHStoreConfig -{ - using LegacySSHStoreConfig::LegacySSHStoreConfig; - - // Hack for getting remote build log output. - // Intentionally not in `LegacySSHStoreConfig` so that it doesn't appear in - // the documentation - const Setting logFD{this, -1, "log-fd", "file descriptor to which SSH's stderr is connected"}; -}; - struct LegacySSHStore final : public Store { - LegacySSHStoreConfigWithLog config_; + LegacySSHStoreConfig config_; - LegacySSHStoreConfigWithLog & config() override { return config_; } - const LegacySSHStoreConfigWithLog & config() const override { return config_; } + LegacySSHStoreConfig & config() override + { + return config_; + } + const LegacySSHStoreConfig & config() const override + { + return config_; + } struct Connection { + // make sure this is destroyed last so the sshConn that feeds it dies first + kj::Promise logHandlerPromise{nullptr}; + std::list act; + ref fromBuf{make_ref()}; std::unique_ptr sshConn; ServeProto::Version remoteVersion; @@ -136,6 +140,54 @@ struct LegacySSHStore final : public Store }; } + kj::Promise logHandler(std::string storeUri) + try { + // 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. + + std::map activities; + + auto stderrPipe = std::move(sshConn->stderrPipe); + auto reader = AIO().lowLevelProvider.wrapInputFd(stderrPipe.get()); + + LogLineSplitter splitter; + + auto flushLine = [&](const std::string & line) { + if (const auto state = handleJSONLogMessage(line, act.back(), activities, storeUri)) { + return *state; + } else { + return act.back().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.back().getLogger().flush()); + } + } + } + } + + if (auto line = splitter.finish(); !line.empty()) { + (void) flushLine(line); + TRY_AWAIT(act.back().getLogger().flush()); + } + } catch (std::exception & e) { // NOLINT(lix-foreign-exceptions) + logException("remote store error", e); + } catch (...) { + std::terminate(); + } + template kj::Promise> sendArg(AsyncOutputStream & stream, StringSink & buffer, Arg && arg) @@ -210,24 +262,18 @@ struct LegacySSHStore final : public Store static std::set uriSchemes() { return {"ssh"}; } - LegacySSHStore( - const std::string & scheme, const std::string & host, LegacySSHStoreConfigWithLog config - ) + LegacySSHStore(const std::string & scheme, const std::string & host, LegacySSHStoreConfig config) : Store(config) , config_(std::move(config)) , host(host) - , connections(make_ref>( - std::max(1, (int) config_.maxConnections), - [this]() { return openConnection(); }, - [](const ref & r) { return r->good; } - )) - , ssh( - host, - config_.port, - config_.sshKey, - config_.sshPublicHostKey, - config_.compress, - config_.logFD) + , connections( + make_ref>( + std::max(1, (int) config_.maxConnections), + [this]() { return openConnection(); }, + [](const ref & r) { return r->good; } + ) + ) + , ssh(host, config_.port, config_.sshKey, config_.sshPublicHostKey, config_.compress) { } @@ -236,37 +282,43 @@ struct LegacySSHStore final : public Store auto conn = make_ref(); conn->sshConn = ssh.startCommand( fmt("%s --serve --write", config_.remoteProgram) - + (config_.remoteStore.get() == "" - ? "" - : " --store " + shellEscape(config_.remoteStore.get())) + + (config_.remoteStore.get() == "" ? "" : " --store " + shellEscape(config_.remoteStore.get())) ); - FdSink to(conn->sshConn->socket.get()); - FdSource from(conn->sshConn->socket.get(), conn->fromBuf); - conn->store = this; try { - to << SERVE_MAGIC_1 << SERVE_PROTOCOL_VERSION; - to.flush(); + FdSink to(conn->sshConn->socket.get()); + FdSource from(conn->sshConn->socket.get(), conn->fromBuf); + conn->store = this; - uint64_t magic = readNum(from); - if (magic != SERVE_MAGIC_2) - throw Error("'nix-store --serve' protocol mismatch from '%s'", host); - conn->remoteVersion = readNum(from); - if (GET_PROTOCOL_MAJOR(conn->remoteVersion) != 0x200) - throw Error("unsupported 'nix-store --serve' protocol version on '%s'", host); + try { + to << SERVE_MAGIC_1 << SERVE_PROTOCOL_VERSION; + to.flush(); - /* No longer support protocols this old*/ - if (GET_PROTOCOL_MINOR(conn->remoteVersion) < 4) { - throw Error( - "remote '%s' is too old (protocol version %x)", host, conn->remoteVersion - ); + uint64_t magic = readNum(from); + if (magic != SERVE_MAGIC_2) { + throw Error("'nix-store --serve' protocol mismatch from '%s'", host); + } + conn->remoteVersion = readNum(from); + if (GET_PROTOCOL_MAJOR(conn->remoteVersion) != 0x200) { + throw Error("unsupported 'nix-store --serve' protocol version on '%s'", host); + } + + /* No longer support protocols this old*/ + if (GET_PROTOCOL_MINOR(conn->remoteVersion) < 4) { + throw Error("remote '%s' is too old (protocol version %x)", host, conn->remoteVersion); + } + + } catch (EndOfFile & e) { + throw Error("cannot connect to '%1%'", host); } - } catch (EndOfFile & e) { - throw Error("cannot connect to '%1%'", host); + conn->act.emplace_back(logger->startActivity(lvlDebug, actUnknown, "remote store " + getUri())); + conn->logHandlerPromise = conn->logHandler(host); + return {conn}; + } catch (Error & e) { + std::string msg = chomp(drainFD(conn->sshConn->stderrPipe.get(), false)); + throw Error("cannot connect to %s: %s (%s)", getUri(), e.msg(), msg); } - - return {conn}; } catch (...) { return {result::current_exception()}; }; @@ -426,6 +478,18 @@ public: try { auto conn(TRY_AWAIT(connections->get())); + // this is a duplicate of DerivationGoal::buildDescription because ugh. + // getting that information into here where needed is nigh *impossible* + auto description = fmt(buildMode == bmRepair ? "repairing outputs of '%s'" + : buildMode == bmCheck ? "checking outputs of '%s'" + : "building '%s'", + printStorePath(drvPath)) + + "on " + getUri(); + conn->act.emplace_back(logger->startActivity( + lvlInfo, actBuild, description, Logger::Fields{printStorePath(drvPath), getUri(), 1, 1} + )); + KJ_DEFER(conn->act.pop_back()); + co_return TRY_AWAIT(conn->sendCommand( ServeProto::Command::BuildDerivation, printStorePath(drvPath), diff --git a/lix/libstore/machines.cc b/lix/libstore/machines.cc index adcb4e9da..cb04b6c34 100644 --- a/lix/libstore/machines.cc +++ b/lix/libstore/machines.cc @@ -34,10 +34,8 @@ bool Machine::mandatoryMet(const std::set & features) const }); } -kj::Promise, Pipe>>> Machine::openStore() const +kj::Promise>> Machine::openStore() const try { - Pipe pipe; - StoreConfig::Params storeParams; if (storeUri.starts_with("ssh://")) { // Remote builds become flakey, when having more than one ssh connection @@ -45,8 +43,6 @@ try { } if (storeUri.starts_with("ssh://") || storeUri.starts_with("ssh-ng://")) { - pipe.create(); - storeParams["log-fd"] = std::to_string(pipe.writeSide.get()); if (sshKey != "") storeParams["ssh-key"] = sshKey; if (sshPublicHostKey != "") @@ -65,7 +61,7 @@ try { append(mandatoryFeatures); } - co_return {TRY_AWAIT(nix::openStore(storeUri, storeParams)), std::move(pipe)}; + co_return TRY_AWAIT(nix::openStore(storeUri, storeParams)); } catch (...) { co_return result::current_exception(); } diff --git a/lix/libstore/machines.hh b/lix/libstore/machines.hh index cb0cca90f..21884ebfb 100644 --- a/lix/libstore/machines.hh +++ b/lix/libstore/machines.hh @@ -42,7 +42,7 @@ struct Machine { */ bool mandatoryMet(const std::set & features) const; - kj::Promise, Pipe>>> openStore() const; + kj::Promise>> openStore() const; }; typedef std::vector Machines; diff --git a/lix/libstore/remote-store-connection.hh b/lix/libstore/remote-store-connection.hh index 6ec14eda5..6f9a4a218 100644 --- a/lix/libstore/remote-store-connection.hh +++ b/lix/libstore/remote-store-connection.hh @@ -102,6 +102,14 @@ struct RemoteStore::Connection }; kj::Promise> processStderr(AsyncFdIoStream & stream); + + /** + * hook to provide additional error information for a failed connection. + */ + virtual std::string connectErrorInfo() + { + return ""; + } }; /** diff --git a/lix/libstore/remote-store.cc b/lix/libstore/remote-store.cc index 300a00e0f..c9d73511e 100644 --- a/lix/libstore/remote-store.cc +++ b/lix/libstore/remote-store.cc @@ -1,3 +1,4 @@ +#include "libutil/fmt.hh" #include "lix/libutil/async-collect.hh" #include "lix/libutil/async-io.hh" #include "lix/libutil/async.hh" @@ -114,7 +115,15 @@ try { } } catch (Error & e) { - throw Error("cannot open connection to remote store '%s': %s", getUri(), e.what()); + auto info = conn.connectErrorInfo(); + throw Error( + "cannot open connection to remote store '%s': %s%s%s%s", + getUri(), + e.what(), + Uncolored(info.empty() ? "" : " ("), + info, + Uncolored(info.empty() ? "" : ")") + ); } TRY_AWAIT(setOptions(conn)); diff --git a/lix/libstore/ssh-store.cc b/lix/libstore/ssh-store.cc index 147da1e7b..4a996b306 100644 --- a/lix/libstore/ssh-store.cc +++ b/lix/libstore/ssh-store.cc @@ -28,48 +28,25 @@ struct SSHStoreConfig : virtual RemoteStoreConfig, virtual CommonSSHStoreConfig } }; -struct SSHStoreConfigWithLog : SSHStoreConfig -{ - SSHStoreConfigWithLog(const Params & params) - : StoreConfig(params) - , RemoteStoreConfig(params) - , CommonSSHStoreConfig(params) - , SSHStoreConfig(params) - { - } - - // Hack for getting ssh errors into build-remote. - // Intentionally not in `SSHStoreConfig` so that it doesn't appear in - // the documentation - const Setting logFD{ - this, -1, "log-fd", "file descriptor to which SSH's stderr is connected" - }; -}; - class SSHStore final : public RemoteStore { - SSHStoreConfigWithLog config_; + SSHStoreConfig config_; public: - SSHStore(const std::string & scheme, const std::string & host, SSHStoreConfigWithLog config) + SSHStore(const std::string & scheme, const std::string & host, SSHStoreConfig config) : Store(config) , RemoteStore(config) , config_(std::move(config)) , host(host) - , ssh(host, - config_.port, - config_.sshKey, - config_.sshPublicHostKey, - config_.compress, - config_.logFD) + , ssh(host, config_.port, config_.sshKey, config_.sshPublicHostKey, config_.compress) { } - SSHStoreConfigWithLog & config() override + SSHStoreConfig & config() override { return config_; } - const SSHStoreConfigWithLog & config() const override + const SSHStoreConfig & config() const override { return config_; } @@ -93,12 +70,54 @@ protected: struct Connection : RemoteStore::Connection { + // make sure this is destroyed last so the sshConn that feeds it dies first + kj::Promise logHandlerPromise{nullptr}; std::unique_ptr sshConn; + ~Connection() noexcept = default; + int getFD() const override { return sshConn->socket.get(); } + + std::string connectErrorInfo() override + { + return chomp(drainFD(sshConn->stderrPipe.get(), false)); + } + + kj::Promise logHandler(std::string storeUri) + try { + auto stderrPipe = std::move(sshConn->stderrPipe); + auto reader = AIO().lowLevelProvider.wrapInputFd(stderrPipe.get()); + + LogLineSplitter splitter; + + auto flushLine = [&](const std::string & line) { debug("ssh(%s): %s", storeUri, 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)) { + flushLine(*line); + } + } + } + + if (auto line = splitter.finish(); !line.empty()) { + flushLine(line); + } + } catch (std::exception & e) { // NOLINT(lix-foreign-exceptions) + logException("remote store error", e); + } catch (...) { + std::terminate(); + } }; kj::Promise>> openConnection() override; @@ -117,6 +136,16 @@ protected: */ return {result::success()}; }; + + kj::Promise> initConnection(RemoteStore::Connection & conn) override + try { + TRY_AWAIT(RemoteStore::initConnection(conn)); + auto & sshConn = static_cast(conn); + sshConn.logHandlerPromise = sshConn.logHandler(getUri()); + co_return result::success(); + } catch (...) { + co_return result::current_exception(); + } }; kj::Promise>> SSHStore::openConnection() diff --git a/lix/libstore/ssh.cc b/lix/libstore/ssh.cc index cb9970b3c..e4607d13b 100644 --- a/lix/libstore/ssh.cc +++ b/lix/libstore/ssh.cc @@ -14,14 +14,19 @@ namespace nix { -SSH::SSH(const std::string & host, const std::optional port, const std::string & keyFile, const std::string & sshPublicHostKey, bool compress, int logFD) +SSH::SSH( + const std::string & host, + const std::optional port, + const std::string & keyFile, + const std::string & sshPublicHostKey, + bool compress +) : host(host) , port(port) , fakeSSH(host == "localhost") , keyFile(keyFile) , sshPublicHostKey(sshPublicHostKey) , compress(compress) - , logFD(logFD) { if (host == "" || host.starts_with("-")) throw Error("invalid SSH host name '%s'", host); @@ -55,6 +60,9 @@ std::unique_ptr SSH::startCommand(const std::string & command) { auto [parent, child] = SocketPair::stream(); auto conn = std::make_unique(); + Pipe stderrPipe; + + stderrPipe.create(); std::optional>> resumeLoggerDefer; if (!fakeSSH) { @@ -79,9 +87,7 @@ std::unique_ptr SSH::startCommand(const std::string & command) options.redirections.push_back({.dup = STDIN_FILENO, .from = child.get()}); options.redirections.push_back({.dup = STDOUT_FILENO, .from = child.get()}); - if (logFD != -1) { - options.redirections.push_back({.dup = STDERR_FILENO, .from = logFD}); - } + options.redirections.push_back({.dup = STDERR_FILENO, .from = stderrPipe.writeSide.get()}); auto [pid, _stdout] = runProgram2(options).release(); conn->sshPid = std::move(pid); @@ -89,6 +95,7 @@ std::unique_ptr SSH::startCommand(const std::string & command) child.close(); conn->socket = std::move(parent); + conn->stderrPipe = std::move(stderrPipe.readSide); return conn; } diff --git a/lix/libstore/ssh.hh b/lix/libstore/ssh.hh index b4db256e5..1682074c5 100644 --- a/lix/libstore/ssh.hh +++ b/lix/libstore/ssh.hh @@ -18,7 +18,6 @@ private: const std::string keyFile; const std::string sshPublicHostKey; const bool compress; - const int logFD; struct State { @@ -34,13 +33,12 @@ public: const std::optional port, const std::string & keyFile, const std::string & sshPublicHostKey, - bool compress, - int logFD); + bool compress); struct Connection { Pid sshPid; - AutoCloseFD socket; + AutoCloseFD socket, stderrPipe; }; std::unique_ptr startCommand(const std::string & command); diff --git a/lix/nix/doctor.cc b/lix/nix/doctor.cc index 4322cfee6..0318d91bb 100644 --- a/lix/nix/doctor.cc +++ b/lix/nix/doctor.cc @@ -313,7 +313,7 @@ struct CmdDoctor : StoreCommand checkInfo(fmt("attempting connection to %s", m.name)); try { auto store = aio().blockOn(m.openStore()); - success &= runPerStore(store.first); + success &= runPerStore(store); } catch (nix::Error & e) { success &= checkFail(fmt("connection failed: %s", e.what())); }