Utilize nix::Machine more fully

With https://github.com/NixOS/nix/pull/9839, the `storeUri` field is
much better structured, so we can use it while still opening the SSH
connection ourselves.
This commit is contained in:
John Ericson 2024-05-21 13:34:30 -04:00
parent 346badc66f
commit d55bea2a1e
5 changed files with 49 additions and 56 deletions

View File

@ -21,27 +21,22 @@
using namespace nix; using namespace nix;
namespace nix::build_remote { bool ::Machine::isLocalhost() const
static Strings extraStoreArgs(std::string & machine)
{ {
Strings result; return storeUri.params.empty() && std::visit(overloaded {
try { [](const StoreReference::Auto &) {
auto parsed = parseURL(machine); return true;
if (parsed.scheme != "ssh") { },
throw SysError("Currently, only (legacy-)ssh stores are supported!"); [](const StoreReference::Specified & s) {
} return
machine = parsed.authority.value_or(""); (s.scheme == "local" || s.scheme == "unix") ||
auto remoteStore = parsed.query.find("remote-store"); ((s.scheme == "ssh" || s.scheme == "ssh-ng") &&
if (remoteStore != parsed.query.end()) { s.authority == "localhost");
result = {"--store", shellEscape(remoteStore->second)}; },
} }, storeUri.variant);
} catch (BadURL &) {
// We just try to continue with `machine->sshName` here for backwards compat.
} }
return result; namespace nix::build_remote {
}
static std::unique_ptr<SSHMaster::Connection> openConnection( static std::unique_ptr<SSHMaster::Connection> openConnection(
::Machine::ptr machine, SSHMaster & master) ::Machine::ptr machine, SSHMaster & master)
@ -51,7 +46,11 @@ static std::unique_ptr<SSHMaster::Connection> openConnection(
command.push_back("--builders"); command.push_back("--builders");
command.push_back(""); command.push_back("");
} else { } else {
command.splice(command.end(), extraStoreArgs(machine->sshName)); auto remoteStore = machine->storeUri.params.find("remote-store");
if (remoteStore != machine->storeUri.params.end()) {
command.push_back("--store");
command.push_back(shellEscape(remoteStore->second));
}
} }
return master.startCommand(std::move(command), { return master.startCommand(std::move(command), {
@ -187,7 +186,7 @@ static BasicDerivation sendInputs(
MaintainCount<counter> mc2(nrStepsCopyingTo); MaintainCount<counter> mc2(nrStepsCopyingTo);
printMsg(lvlDebug, "sending closure of %s to %s", printMsg(lvlDebug, "sending closure of %s to %s",
localStore.printStorePath(step.drvPath), conn.machine->sshName); localStore.printStorePath(step.drvPath), conn.machine->storeUri.render());
auto now1 = std::chrono::steady_clock::now(); auto now1 = std::chrono::steady_clock::now();
@ -393,8 +392,13 @@ void State::buildRemote(ref<Store> destStore,
updateStep(ssConnecting); updateStep(ssConnecting);
auto * pSpecified = std::get_if<StoreReference::Specified>(&machine->storeUri.variant);
if (!pSpecified || pSpecified->scheme != "ssh") {
throw Error("Currently, only (legacy-)ssh stores are supported!");
}
SSHMaster master { SSHMaster master {
machine->sshName, pSpecified->authority,
machine->sshKey, machine->sshKey,
machine->sshPublicHostKey, machine->sshPublicHostKey,
false, // no SSH master yet false, // no SSH master yet
@ -445,11 +449,11 @@ void State::buildRemote(ref<Store> destStore,
conn.to, conn.to,
conn.from, conn.from,
our_version, our_version,
machine->sshName); machine->storeUri.render());
} catch (EndOfFile & e) { } catch (EndOfFile & e) {
child->sshPid.wait(); child->sshPid.wait();
std::string s = chomp(readFile(result.logFile)); std::string s = chomp(readFile(result.logFile));
throw Error("cannot connect to %1%: %2%", machine->sshName, s); throw Error("cannot connect to %1%: %2%", machine->storeUri.render(), s);
} }
{ {
@ -480,7 +484,7 @@ void State::buildRemote(ref<Store> destStore,
/* Do the build. */ /* Do the build. */
printMsg(lvlDebug, "building %s on %s", printMsg(lvlDebug, "building %s on %s",
localStore->printStorePath(step->drvPath), localStore->printStorePath(step->drvPath),
machine->sshName); machine->storeUri.render());
updateStep(ssBuilding); updateStep(ssBuilding);
@ -503,7 +507,7 @@ void State::buildRemote(ref<Store> destStore,
get a build log. */ get a build log. */
if (result.isCached) { if (result.isCached) {
printMsg(lvlInfo, "outputs of %s substituted or already valid on %s", printMsg(lvlInfo, "outputs of %s substituted or already valid on %s",
localStore->printStorePath(step->drvPath), machine->sshName); localStore->printStorePath(step->drvPath), machine->storeUri.render());
unlink(result.logFile.c_str()); unlink(result.logFile.c_str());
result.logFile = ""; result.logFile = "";
} }
@ -532,7 +536,7 @@ void State::buildRemote(ref<Store> destStore,
/* Copy each path. */ /* Copy each path. */
printMsg(lvlDebug, "copying outputs of %s from %s (%d bytes)", printMsg(lvlDebug, "copying outputs of %s from %s (%d bytes)",
localStore->printStorePath(step->drvPath), machine->sshName, totalNarSize); localStore->printStorePath(step->drvPath), machine->storeUri.render(), totalNarSize);
build_remote::copyPathsFromRemote(conn, narMembers, *localStore, *destStore, infos); build_remote::copyPathsFromRemote(conn, narMembers, *localStore, *destStore, infos);
auto now2 = std::chrono::steady_clock::now(); auto now2 = std::chrono::steady_clock::now();
@ -571,7 +575,7 @@ void State::buildRemote(ref<Store> destStore,
info->consecutiveFailures = std::min(info->consecutiveFailures + 1, (unsigned int) 4); info->consecutiveFailures = std::min(info->consecutiveFailures + 1, (unsigned int) 4);
info->lastFailure = now; info->lastFailure = now;
int delta = retryInterval * std::pow(retryBackoff, info->consecutiveFailures - 1) + (rand() % 30); int delta = retryInterval * std::pow(retryBackoff, info->consecutiveFailures - 1) + (rand() % 30);
printMsg(lvlInfo, "will disable machine %1% for %2%s", machine->sshName, delta); printMsg(lvlInfo, "will disable machine %1% for %2%s", machine->storeUri.render(), delta);
info->disabledUntil = now + std::chrono::seconds(delta); info->disabledUntil = now + std::chrono::seconds(delta);
} }
throw; throw;

View File

@ -41,7 +41,7 @@ void State::builder(MachineReservation::ptr reservation)
} catch (std::exception & e) { } catch (std::exception & e) {
printMsg(lvlError, "uncaught exception building %s on %s: %s", printMsg(lvlError, "uncaught exception building %s on %s: %s",
localStore->printStorePath(reservation->step->drvPath), localStore->printStorePath(reservation->step->drvPath),
reservation->machine->sshName, reservation->machine->storeUri.render(),
e.what()); e.what());
} }
} }
@ -150,7 +150,7 @@ State::StepResult State::doBuildStep(nix::ref<Store> destStore,
buildOptions.buildTimeout = build->buildTimeout; buildOptions.buildTimeout = build->buildTimeout;
printInfo("performing step %s %d times on %s (needed by build %d and %d others)", printInfo("performing step %s %d times on %s (needed by build %d and %d others)",
localStore->printStorePath(step->drvPath), buildOptions.nrRepeats + 1, machine->sshName, buildId, (dependents.size() - 1)); localStore->printStorePath(step->drvPath), buildOptions.nrRepeats + 1, machine->storeUri.render(), buildId, (dependents.size() - 1));
} }
if (!buildOneDone) if (!buildOneDone)
@ -196,7 +196,7 @@ State::StepResult State::doBuildStep(nix::ref<Store> destStore,
{ {
auto mc = startDbUpdate(); auto mc = startDbUpdate();
pqxx::work txn(*conn); pqxx::work txn(*conn);
stepNr = createBuildStep(txn, result.startTime, buildId, step, machine->sshName, bsBusy); stepNr = createBuildStep(txn, result.startTime, buildId, step, machine->storeUri.render(), bsBusy);
txn.commit(); txn.commit();
} }
@ -253,7 +253,7 @@ State::StepResult State::doBuildStep(nix::ref<Store> destStore,
/* Finish the step in the database. */ /* Finish the step in the database. */
if (stepNr) { if (stepNr) {
pqxx::work txn(*conn); pqxx::work txn(*conn);
finishBuildStep(txn, result, buildId, stepNr, machine->sshName); finishBuildStep(txn, result, buildId, stepNr, machine->storeUri.render());
txn.commit(); txn.commit();
} }
@ -261,7 +261,7 @@ State::StepResult State::doBuildStep(nix::ref<Store> destStore,
issue). Retry a number of times. */ issue). Retry a number of times. */
if (result.canRetry) { if (result.canRetry) {
printMsg(lvlError, "possibly transient failure building %s on %s: %s", printMsg(lvlError, "possibly transient failure building %s on %s: %s",
localStore->printStorePath(step->drvPath), machine->sshName, result.errorMsg); localStore->printStorePath(step->drvPath), machine->storeUri.render(), result.errorMsg);
assert(stepNr); assert(stepNr);
bool retry; bool retry;
{ {
@ -452,7 +452,7 @@ void State::failStep(
build->finishedInDB) build->finishedInDB)
continue; continue;
createBuildStep(txn, createBuildStep(txn,
0, build->id, step, machine ? machine->sshName : "", 0, build->id, step, machine ? machine->storeUri.render() : "",
result.stepStatus, result.errorMsg, buildId == build->id ? 0 : buildId); result.stepStatus, result.errorMsg, buildId == build->id ? 0 : buildId);
} }

View File

@ -255,7 +255,7 @@ system_time State::doDispatch()
/* Can this machine do this step? */ /* Can this machine do this step? */
if (!mi.machine->supportsStep(step)) { if (!mi.machine->supportsStep(step)) {
debug("machine '%s' does not support step '%s' (system type '%s')", debug("machine '%s' does not support step '%s' (system type '%s')",
mi.machine->sshName, localStore->printStorePath(step->drvPath), step->drv->platform); mi.machine->storeUri.render(), localStore->printStorePath(step->drvPath), step->drv->platform);
continue; continue;
} }

View File

@ -153,8 +153,8 @@ void State::parseMachines(const std::string & contents)
using MaxJobs = std::remove_const<decltype(nix::Machine::maxJobs)>::type; using MaxJobs = std::remove_const<decltype(nix::Machine::maxJobs)>::type;
auto machine = std::make_shared<::Machine>(nix::Machine { auto machine = std::make_shared<::Machine>(nix::Machine {
// `storeUri`, not yet used // `storeUri`
"", tokens[0],
// `systemTypes` // `systemTypes`
tokenizeString<StringSet>(tokens[1], ","), tokenizeString<StringSet>(tokens[1], ","),
// `sshKey` // `sshKey`
@ -175,25 +175,23 @@ void State::parseMachines(const std::string & contents)
: "", : "",
}); });
machine->sshName = tokens[0];
/* Re-use the State object of the previous machine with the /* Re-use the State object of the previous machine with the
same name. */ same name. */
auto i = oldMachines.find(machine->sshName); auto i = oldMachines.find(machine->storeUri.variant);
if (i == oldMachines.end()) if (i == oldMachines.end())
printMsg(lvlChatty, "adding new machine %1%", machine->sshName); printMsg(lvlChatty, "adding new machine %1%", machine->storeUri.render());
else else
printMsg(lvlChatty, "updating machine %1%", machine->sshName); printMsg(lvlChatty, "updating machine %1%", machine->storeUri.render());
machine->state = i == oldMachines.end() machine->state = i == oldMachines.end()
? std::make_shared<::Machine::State>() ? std::make_shared<::Machine::State>()
: i->second->state; : i->second->state;
newMachines[machine->sshName] = machine; newMachines[machine->storeUri.variant] = machine;
} }
for (auto & m : oldMachines) for (auto & m : oldMachines)
if (newMachines.find(m.first) == newMachines.end()) { if (newMachines.find(m.first) == newMachines.end()) {
if (m.second->enabled) if (m.second->enabled)
printInfo("removing machine %1%", m.first); printInfo("removing machine %1%", m.second->storeUri.render());
/* Add a disabled ::Machine object to make sure stats are /* Add a disabled ::Machine object to make sure stats are
maintained. */ maintained. */
auto machine = std::make_shared<::Machine>(*(m.second)); auto machine = std::make_shared<::Machine>(*(m.second));
@ -657,7 +655,7 @@ void State::dumpStatus(Connection & conn)
machine["avgStepTime"] = (float) s->totalStepTime / s->nrStepsDone; machine["avgStepTime"] = (float) s->totalStepTime / s->nrStepsDone;
machine["avgStepBuildTime"] = (float) s->totalStepBuildTime / s->nrStepsDone; machine["avgStepBuildTime"] = (float) s->totalStepBuildTime / s->nrStepsDone;
} }
statusJson["machines"][m->sshName] = machine; statusJson["machines"][m->storeUri.render()] = machine;
} }
} }

View File

@ -6,7 +6,6 @@
#include <map> #include <map>
#include <memory> #include <memory>
#include <queue> #include <queue>
#include <regex>
#include <prometheus/counter.h> #include <prometheus/counter.h>
#include <prometheus/gauge.h> #include <prometheus/gauge.h>
@ -240,10 +239,6 @@ struct Machine : nix::Machine
{ {
typedef std::shared_ptr<Machine> ptr; typedef std::shared_ptr<Machine> ptr;
/* TODO Get rid of: `nix::Machine::storeUri` is normalized in a way
we are not yet used to, but once we are, we don't need this. */
std::string sshName;
struct State { struct State {
typedef std::shared_ptr<State> ptr; typedef std::shared_ptr<State> ptr;
counter currentJobs{0}; counter currentJobs{0};
@ -293,11 +288,7 @@ struct Machine : nix::Machine
return true; return true;
} }
bool isLocalhost() bool isLocalhost() const;
{
std::regex r("^(ssh://|ssh-ng://)?localhost$");
return std::regex_search(sshName, r);
}
// A connection to a machine // A connection to a machine
struct Connection : nix::ServeProto::BasicClientConnection { struct Connection : nix::ServeProto::BasicClientConnection {
@ -357,7 +348,7 @@ private:
/* The build machines. */ /* The build machines. */
std::mutex machinesReadyLock; std::mutex machinesReadyLock;
typedef std::map<std::string, Machine::ptr> Machines; typedef std::map<nix::StoreReference::Variant, Machine::ptr> Machines;
nix::Sync<Machines> machines; // FIXME: use atomic_shared_ptr nix::Sync<Machines> machines; // FIXME: use atomic_shared_ptr
/* Various stats. */ /* Various stats. */