guiwf/lix
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

libstore: handle ssh{,-ng}:// logs in their store class

Browse Source 
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
This commit is contained in:
eldritch horrors
2026-07-17 18:00:14 +00:00
parent 553ae61ffd
commit 57fb4f534d
12 changed files with 211 additions and 179 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lix/legacy/build-remote.cc
+3 -80
View File
@@ -185,67 +185,6 @@ struct BuilderConnection
AutoCloseFD slotLock;
std::shared_ptr<Store> 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<Result<void>> 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<ActivityId, Activity> 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<char>(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<Store> 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<void> AcceptedBuild::run(RunContext context)
kj::Promise<Result<void>> AcceptedBuild::runImpl(RunContext context)
{
try {
auto logHandler = builder.startLogThread(
fmt("%s on '%s'",
rpc::to<std::string_view>(context.getParams().getDescription()),
builder.storeUri),
store->printStorePath(drvPath)
);
auto & sshStore = builder.sshStore;
auto & storeUri = builder.storeUri;
@@ -618,9 +544,6 @@ kj::Promise<Result<void>> 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();
lix/libstore/build/derivation-goal.cc
-1
View File
@@ -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());
lix/libstore/build/hook-instance.capnp
-1
View File
@@ -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
);
}
lix/libstore/legacy-ssh-store.cc
+115 -51
View File
@@ -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 <cstdint>
#include <exception>
#include <kj/async.h>
#include <optional>
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<int> 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<void> logHandlerPromise{nullptr};
std::list<Activity> act;
ref<IoBuffer> fromBuf{make_ref<IoBuffer>()};
std::unique_ptr<SSH::Connection> sshConn;
ServeProto::Version remoteVersion;
@@ -136,6 +140,54 @@ struct LegacySSHStore final : public Store
};
}
kj::Promise<void> 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<ActivityId, Activity> 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<char>(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<typename Arg>
kj::Promise<Result<void>>
sendArg(AsyncOutputStream & stream, StringSink & buffer, Arg && arg)
@@ -210,24 +262,18 @@ struct LegacySSHStore final : public Store
static std::set<std::string> 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<Pool<Connection>>(
std::max(1, (int) config_.maxConnections),
[this]() { return openConnection(); },
[](const ref<Connection> & r) { return r->good; }
))
, ssh(
host,
config_.port,
config_.sshKey,
config_.sshPublicHostKey,
config_.compress,
config_.logFD)
, connections(
make_ref<Pool<Connection>>(
std::max(1, (int) config_.maxConnections),
[this]() { return openConnection(); },
[](const ref<Connection> & 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<Connection>();
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<uint64_t>(from);
if (magic != SERVE_MAGIC_2)
throw Error("'nix-store --serve' protocol mismatch from '%s'", host);
conn->remoteVersion = readNum<unsigned>(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<uint64_t>(from);
if (magic != SERVE_MAGIC_2) {
throw Error("'nix-store --serve' protocol mismatch from '%s'", host);
}
conn->remoteVersion = readNum<unsigned>(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<BuildResult>(
ServeProto::Command::BuildDerivation,
printStorePath(drvPath),
lix/libstore/machines.cc
+2 -6
View File
@@ -34,10 +34,8 @@ bool Machine::mandatoryMet(const std::set<std::string> & features) const
});
}
kj::Promise<Result<std::pair<ref<Store>, Pipe>>> Machine::openStore() const
kj::Promise<Result<ref<Store>>> 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();
}
lix/libstore/machines.hh
+1 -1
View File
@@ -42,7 +42,7 @@ struct Machine {
*/
bool mandatoryMet(const std::set<std::string> & features) const;
kj::Promise<Result<std::pair<ref<Store>, Pipe>>> openStore() const;
kj::Promise<Result<ref<Store>>> openStore() const;
};
typedef std::vector<Machine> Machines;
lix/libstore/remote-store-connection.hh
+8
View File
@@ -102,6 +102,14 @@ struct RemoteStore::Connection
};
kj::Promise<Result<RemoteError>> processStderr(AsyncFdIoStream & stream);
/**
* hook to provide additional error information for a failed connection.
*/
virtual std::string connectErrorInfo()
{
return "";
}
};
/**
lix/libstore/remote-store.cc
+10 -1
View File
@@ -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));
lix/libstore/ssh-store.cc
+57 -28
View File
@@ -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<int> 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<void> logHandlerPromise{nullptr};
std::unique_ptr<SSH::Connection> 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<void> 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<char>(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<Result<ref<RemoteStore::Connection>>> openConnection() override;
@@ -117,6 +136,16 @@ protected:
*/
return {result::success()};
};
kj::Promise<Result<void>> initConnection(RemoteStore::Connection & conn) override
try {
TRY_AWAIT(RemoteStore::initConnection(conn));
auto & sshConn = static_cast<Connection &>(conn);
sshConn.logHandlerPromise = sshConn.logHandler(getUri());
co_return result::success();
} catch (...) {
co_return result::current_exception();
}
};
kj::Promise<Result<ref<RemoteStore::Connection>>> SSHStore::openConnection()
lix/libstore/ssh.cc
+12 -5
View File
@@ -14,14 +14,19 @@
namespace nix {
SSH::SSH(const std::string & host, const std::optional<uint16_t> port, const std::string & keyFile, const std::string & sshPublicHostKey, bool compress, int logFD)
SSH::SSH(
const std::string & host,
const std::optional<uint16_t> 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::Connection> SSH::startCommand(const std::string & command)
{
auto [parent, child] = SocketPair::stream();
auto conn = std::make_unique<Connection>();
Pipe stderrPipe;
stderrPipe.create();
std::optional<Finally<std::function<void()>>> resumeLoggerDefer;
if (!fakeSSH) {
@@ -79,9 +87,7 @@ std::unique_ptr<SSH::Connection> 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::Connection> SSH::startCommand(const std::string & command)
child.close();
conn->socket = std::move(parent);
conn->stderrPipe = std::move(stderrPipe.readSide);
return conn;
}
lix/libstore/ssh.hh
+2 -4
View File
@@ -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<uint16_t> 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<Connection> startCommand(const std::string & command);
lix/nix/doctor.cc
+1 -1
View File
@@ -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()));
}