22 files changed
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -5084,6 +5084,7 @@
"remowt-ui-prompt",
"tokio",
"tracing",
+ "tracing-journald",
"tracing-subscriber",
]
@@ -6576,6 +6577,17 @@
]
[[package]]
+name = "tracing-journald"
+version = "0.3.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2d3a81ed245bfb62592b1e2bc153e77656d94ee6a0497683a65a12ccaf2438d0"
+dependencies = [
+ "libc",
+ "tracing-core",
+ "tracing-subscriber",
+]
+
+[[package]]
name = "tracing-log"
version = "0.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -75,6 +75,7 @@
tokio = { version = "1.45.1", features = ["fs", "macros", "rt", "rt-multi-thread", "sync", "time"] }
tracing = "0.1"
tracing-indicatif = "0.3.13"
+tracing-journald = "0.3.2"
tracing-opentelemetry = "0.33.0"
uuid = { version = "1", features = ["v4"] }
# For fixed coloring needs to be updated to https://github.com/tokio-rs/tracing/pull/3484
--- a/cmds/fleet/src/cmds/build_systems.rs
+++ b/cmds/fleet/src/cmds/build_systems.rs
@@ -14,6 +14,8 @@
use tokio::task::spawn_blocking;
use tracing::{Instrument, error, field, info, info_span, warn};
+const TARGET: &str = "build-systems";
+
#[derive(Parser)]
pub struct Deploy {
/// Disable automatic rollback
@@ -32,7 +34,7 @@
}
async fn build_task(config: Config, hostname: String, build_attr: &str) -> Result<Utf8PathBuf> {
- info!("building");
+ info!(target: TARGET, "building");
let host = config.host(&hostname)?;
// let action = Action::from(self.subcommand.clone());
let nixos = host.nixos_config()?;
@@ -43,7 +45,7 @@
// We already have system profiles for backups.
if !host.local {
- info!("adding gc root");
+ info!(target: TARGET, "adding gc root");
let local = config.local_host();
let plugin_id = local.ensure_nix_plugin().await?;
let nix = local
@@ -70,7 +72,7 @@
let build_attr = self.build_attr.clone();
for host in hosts {
let config = config.clone();
- let span = info_span!("build", host = field::display(&host.name));
+ let span = info_span!(target: TARGET, "build", host = field::display(&host.name));
let hostname = host.name;
let build_attr = build_attr.clone();
tasks.push(
@@ -78,7 +80,7 @@
let built = match build_task(config, hostname.clone(), &build_attr).await {
Ok(path) => path,
Err(e) => {
- error!("failed to deploy host: {:?}", e);
+ error!(target: TARGET, "failed to deploy host: {:?}", e);
return;
}
};
@@ -86,9 +88,9 @@
let mut out = current_dir().expect("cwd exists");
out.push(format!("built-{hostname}"));
- info!("linking iso image to {:?}", out);
+ info!(target: TARGET, "linking iso image to {:?}", out);
if let Err(e) = symlink(built, out) {
- error!("failed to symlink: {e}")
+ error!(target: TARGET, "failed to symlink: {e}")
}
})
.instrument(span),
@@ -105,7 +107,7 @@
let tasks = FuturesUnordered::new();
for host in hosts.into_iter() {
let config = config.clone();
- let span = info_span!("deploy", host = field::display(&host.name));
+ let span = info_span!(target: TARGET, "deploy", host = field::display(&host.name));
let hostname = host.name.clone();
let opts = opts.clone();
if let Some(deploy_kind) = opts.action_attr::<DeployKind>(&host, "deploy_kind")? {
@@ -125,7 +127,7 @@
{
Ok(path) => path,
Err(e) => {
- error!("failed to build host system closure: {:?}", e);
+ error!(target: TARGET, "failed to build host system closure: {:?}", e);
return;
}
};
@@ -133,7 +135,7 @@
let deploy_kind = match host.deploy_kind().await {
Ok(v) => v,
Err(e) => {
- error!("failed to query target deploy kind: {e}");
+ error!(target: TARGET, "failed to query target deploy kind: {e}");
return;
}
};
@@ -141,7 +143,7 @@
// TODO: Make disable_rollback a host attribute instead
let mut disable_rollback = self.disable_rollback;
if !disable_rollback && deploy_kind != DeployKind::Fleet {
- warn!("disabling rollback, as not supported by non-fleet deployment kinds");
+ warn!(target: TARGET, "disabling rollback, as not supported by non-fleet deployment kinds");
disable_rollback = true;
}
@@ -149,7 +151,7 @@
match upload_task(&host, GenerationStorage::Deployer, built).await {
Ok(v) => v,
Err(e) => {
- error!("upload failed: {e}");
+ error!(target: TARGET, "upload failed: {e}");
return;
}
};
@@ -169,7 +171,7 @@
)
.await
{
- error!("activation failed: {e}");
+ error!(target: TARGET, "activation failed: {e}");
}
})
.instrument(span),
--- /dev/null
+++ b/cmds/fleet/src/log_tree.rs
@@ -0,0 +1,243 @@
+//! Streaming, indented tree formatter.
+//!
+//! Unlike `tracing-tree` it keeps no per-thread depth state, so interleaved
+//! parallel output stays correct; unlike `tracing-forest` it doesn't buffer
+//! whole trees, so logs stream live. Visual style mirrors `@meteor-it/logger`'s
+//! node receiver: a fixed-width, right-aligned, level-colored target column,
+//! then a magenta `>`/`<` gutter for span enter/exit (with duration on exit),
+//! and 2-space indentation per depth for events.
+
+use std::{
+ fmt::Write as _,
+ io::{IsTerminal as _, Write as _},
+ sync::{
+ Mutex,
+ atomic::{AtomicBool, Ordering},
+ },
+ time::{Duration, Instant},
+};
+
+use tracing::{Event, Level, Subscriber, field::Field, field::Visit, span::Attributes, span::Id};
+use tracing_subscriber::{
+ fmt::MakeWriter,
+ layer::{Context, Layer},
+ registry::{LookupSpan, Scope},
+};
+
+/// Width of the right-aligned target column.
+const NAME_LIMIT: usize = 18;
+
+#[derive(Default)]
+struct Fields {
+ message: String,
+ rest: String,
+}
+
+impl Visit for Fields {
+ fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) {
+ if field.name() == "message" {
+ let _ = write!(self.message, "{value:?}");
+ } else {
+ let _ = write!(self.rest, " {}={:?}", field.name(), value);
+ }
+ }
+}
+
+impl Fields {
+ fn render(&self) -> String {
+ format!("{}{}", self.message, self.rest).trim().to_owned()
+ }
+}
+
+/// Per-span state stashed in the registry's extensions.
+struct SpanData {
+ fields: String,
+ started: Instant,
+ // Whether the span's enter line was written, so we only emit an exit line
+ // for spans that actually appeared.
+ opened: AtomicBool,
+}
+
+/// SGR code → escape for the level-colored target column.
+fn level_code(level: &Level) -> &'static str {
+ match *level {
+ Level::ERROR => "31m", // red
+ Level::WARN => "33m", // yellow
+ Level::INFO => "34m", // blue
+ Level::DEBUG => "39m", // default fg
+ Level::TRACE => "90m", // gray
+ }
+}
+
+/// 2-space indent per depth, then a gutter symbol.
+fn gutter(depth: usize, sym: char) -> String {
+ let mut s = " ".repeat(depth);
+ s.push(sym);
+ s
+}
+
+fn fmt_dur(d: Duration) -> String {
+ if d.as_secs_f64() >= 1.0 {
+ format!("{:.2}s", d.as_secs_f64())
+ } else if d.as_millis() >= 1 {
+ format!("{}ms", d.as_millis())
+ } else {
+ format!("{}µs", d.as_micros())
+ }
+}
+
+pub struct TreeLayer<W> {
+ writer: W,
+ color: bool,
+ // Serializes the whole compute-and-write so enter/exit decisions, output
+ // ordering, and the writes stay consistent under parallel emission.
+ last: Mutex<Vec<Id>>,
+}
+
+impl<W> TreeLayer<W> {
+ pub fn new(writer: W) -> Self {
+ Self {
+ writer,
+ color: std::io::stderr().is_terminal() && std::env::var_os("NO_COLOR").is_none(),
+ last: Mutex::new(vec![]),
+ }
+ }
+
+ /// Raw SGR escape, or empty when color is disabled.
+ fn sgr(&self, code: &str) -> String {
+ if self.color {
+ format!("\x1b[{code}")
+ } else {
+ String::new()
+ }
+ }
+
+ fn name_col(&self, mut target: &str, code: &str) -> String {
+ let w = NAME_LIMIT;
+ if target.len() > w {
+ target = &target[..w];
+ }
+ if self.color {
+ format!("\x1b[{code}\x1b[1m{target:>w$}\x1b[0m")
+ } else {
+ format!("{target:>w$}")
+ }
+ }
+
+ fn print_ancestors<S>(&self, scope: Option<Scope<'_, S>>, out: &mut String) -> usize
+ where
+ S: Subscriber + for<'a> LookupSpan<'a>,
+ {
+ let scope: Vec<_> = scope.map(|s| s.from_root().collect()).unwrap_or_default();
+
+ let mut _last = self.last.lock().expect("not poisoned");
+
+ let skip = scope
+ .iter()
+ .enumerate()
+ .take_while(|(i, s)| _last.get(*i) == Some(&s.id()))
+ .count();
+
+ for (depth, span) in scope.iter().enumerate().skip(skip) {
+ let ext = span.extensions();
+ let Some(data) = ext.get::<SpanData>() else {
+ continue;
+ };
+ let newly_open = !data.opened.swap(true, Ordering::Relaxed);
+ let _ = writeln!(
+ out,
+ "{} {}{}{} {}{}{}",
+ self.name_col(span.metadata().target(), "44m"),
+ self.sgr("35m"),
+ gutter(depth, if newly_open { '>' } else { '.' }),
+ self.sgr("1m"),
+ span.name(),
+ data.fields,
+ self.sgr("0m"),
+ );
+ }
+
+ *_last = scope.iter().map(|v| v.id()).collect();
+
+ scope.len()
+ }
+}
+
+impl<S, W> Layer<S> for TreeLayer<W>
+where
+ S: Subscriber + for<'a> LookupSpan<'a>,
+ W: for<'a> MakeWriter<'a> + 'static,
+{
+ fn on_new_span(&self, attrs: &Attributes<'_>, id: &Id, ctx: Context<'_, S>) {
+ let span = ctx.span(id).expect("new span exists");
+ let mut fields = Fields::default();
+ attrs.record(&mut fields);
+ span.extensions_mut().insert(SpanData {
+ fields: fields.rest,
+ started: Instant::now(),
+ opened: AtomicBool::new(false),
+ });
+ }
+
+ fn on_event(&self, event: &Event<'_>, ctx: Context<'_, S>) {
+ let mut out = String::new();
+
+ let scope = ctx.event_scope(event);
+ let pad = self.print_ancestors(scope, &mut out);
+
+ let mut fields = Fields::default();
+ event.record(&mut fields);
+ let message = fields.render();
+ let meta = event.metadata();
+ for (i, message) in message.split('\n').enumerate() {
+ // Event line: black bg + colored target column, then indent + message.
+ let _ = writeln!(
+ out,
+ "{}{}{}{}{}",
+ self.sgr("40m"),
+ self.name_col(
+ if i == 0 { meta.target() } else { "|" },
+ level_code(meta.level())
+ ),
+ self.sgr("0m"),
+ gutter(pad, ' '),
+ message,
+ );
+ }
+
+ let _ = self.writer.make_writer().write_all(out.as_bytes());
+ }
+
+ fn on_close(&self, id: Id, ctx: Context<'_, S>) {
+ let Some(span) = ctx.span(&id) else { return };
+ let (opened, dur) = span
+ .extensions()
+ .get::<SpanData>()
+ .map_or((false, Duration::ZERO), |d| {
+ (d.opened.load(Ordering::Relaxed), d.started.elapsed())
+ });
+ if !opened {
+ return;
+ }
+
+ let mut out = String::new();
+ let _pad = self.print_ancestors(ctx.span_scope(&id), &mut out);
+
+ let depth = span.scope().skip(1).count(); // ancestors == depth
+ let _ = writeln!(
+ out,
+ "{} {}{}{} {}{} (Done in {}){}",
+ self.name_col(span.metadata().target(), "44m"),
+ self.sgr("35m"),
+ gutter(depth, '<'),
+ self.sgr("1m"),
+ span.name(),
+ self.sgr("22m"),
+ fmt_dur(dur),
+ self.sgr("0m"),
+ );
+
+ let _guard = self.last.lock().expect("not poisoned");
+ let _ = self.writer.make_writer().write_all(out.as_bytes());
+ }
+}
--- a/cmds/fleet/src/main.rs
+++ b/cmds/fleet/src/main.rs
@@ -1,6 +1,7 @@
#![recursion_limit = "512"]
pub(crate) mod cmds;
+mod log_tree;
use std::{process::ExitCode, sync::Arc};
@@ -21,6 +22,7 @@
use human_repr::HumanCount;
#[cfg(feature = "indicatif")]
use indicatif::{ProgressState, ProgressStyle};
+use log_tree::TreeLayer;
use nix_eval::{
eval_store, gc_register_my_thread, gc_unregister_my_thread, init_libraries, init_tokio_for_nix,
};
@@ -128,15 +130,21 @@
fn setup_logging(opts: &RootOpts) -> Result<()> {
#[cfg(feature = "indicatif")]
let indicatif_layer = {
+ use std::fmt;
use std::time::Duration;
IndicatifLayer::new().with_max_progress_bars(10, Some(ProgressStyle::default_spinner()))
+ .with_span_child_prefix_indent(" ")
+ .with_span_child_prefix_symbol("")
.with_progress_style(
ProgressStyle::with_template(
- "{color_start}{span_child_prefix} {span_name}{{{span_fields}}}{color_end} {wide_msg} {color_start}{download_progress} {elapsed}{color_end}",
+ "{span_child_prefix:.magenta}{chevron:.magenta}{span_name:<18.blue.bold} {wide_msg} {span_fields:.dim} {color_start}{download_progress} {elapsed}{color_end}",
)
.unwrap()
- .with_key("download_progress", |state: &ProgressState, writer: &mut dyn std::fmt::Write| {
+ .with_key("chevron", |_: &ProgressState, writer: &mut dyn fmt::Write| {
+ let _ = write!(writer, "> ");
+ })
+ .with_key("download_progress", |state: &ProgressState, writer: &mut dyn fmt::Write| {
let Some(len) = state.len() else {
return;
};
@@ -149,7 +157,7 @@
})
.with_key(
"color_start",
- |state: &ProgressState, writer: &mut dyn std::fmt::Write| {
+ |state: &ProgressState, writer: &mut dyn fmt::Write| {
let elapsed = state.elapsed();
if elapsed > Duration::from_secs(60) {
@@ -163,7 +171,7 @@
)
.with_key(
"color_end",
- |state: &ProgressState, writer: &mut dyn std::fmt::Write| {
+ |state: &ProgressState, writer: &mut dyn fmt::Write| {
if state.elapsed() > Duration::from_secs(30) {
let _ = write!(writer, "\x1b[0m");
}
@@ -174,12 +182,15 @@
let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
- let reg = tracing_subscriber::registry().with({
- let sub = tracing_subscriber::fmt::layer().without_time();
+ let tree = {
#[cfg(feature = "indicatif")]
- let sub = sub.with_writer(indicatif_layer.get_stderr_writer());
- sub.with_filter(filter) // .without,
- });
+ let writer = indicatif_layer.get_stderr_writer();
+ #[cfg(not(feature = "indicatif"))]
+ let writer = || std::io::stderr();
+ TreeLayer::new(writer)
+ };
+
+ let reg = tracing_subscriber::registry().with(tree.with_filter(filter));
#[cfg(feature = "indicatif")]
let reg = reg.with(indicatif_layer);
@@ -200,14 +211,12 @@
let logger = OpenTelemetryTracingBridge::new(&log_provider);
let tracer = span_provider.tracer("fleet");
- let reg = reg
- .with(tracing_opentelemetry::layer().with_tracer(tracer))
- .with(logger);
-
- reg.init();
+ reg.with(tracing_opentelemetry::layer().with_tracer(tracer))
+ .with(logger)
+ .init();
} else {
reg.init();
- };
+ }
Ok(())
}
@@ -252,7 +261,6 @@
.await
.expect("primary task panicked")
})
- // async_main(opts)
}
async fn main_real(opts: RootOpts) -> Result<()> {
--- a/crates/fleet-base/src/deploy.rs
+++ b/crates/fleet-base/src/deploy.rs
@@ -290,8 +290,10 @@
if !host.local {
info!("uploading system closure");
let mut tries = 0;
+ // TODO: Use gc prefix?..
loop {
- match host.remote_derivation(&generation).await {
+ let name = format!("{}-system", host.gc_root_prefix());
+ match host.remote_derivation(&generation, Some(&name)).await {
Ok(remote) => {
assert!(remote == generation, "CA derivations aren't implemented");
return Ok(remote);
--- a/crates/fleet-base/src/host.rs
+++ b/crates/fleet-base/src/host.rs
@@ -13,16 +13,16 @@
use camino::{Utf8Path, Utf8PathBuf};
use chrono::{DateTime, Utc};
use fleet_shared::SecretData;
-use nix_eval::{Store, Value, drv::DrvGraph, eval_store, nix_go, nix_go_json, util::assert_warn};
+use nix_eval::{Store, Value, eval_store, nix_go, nix_go_json, util::assert_warn};
use remowt_client::{AgentBundle, Remowt};
use remowt_endpoints::fs::FsClient;
use remowt_fleet::NixClient;
-use remowt_link_shared::BifConfig;
+use remowt_link_shared::{Address, BifConfig};
use remowt_ui_prompt::auto::AutoPrompter;
use remowt_ui_prompt::bifrost::PromptEndpoints;
use remowt_ui_prompt::{PrependSourcePrompter, Source};
use tempfile::NamedTempFile;
-use tokio::{sync::OnceCell, task::spawn_blocking};
+use tokio::{process::Command, sync::OnceCell, task::spawn_blocking};
use tracing::{info, instrument, warn};
use crate::fleetdata::{
@@ -104,11 +104,14 @@
session_destination: OnceLock<String>,
legacy_ssh_store: OnceLock<bool>,
+ /// fleetConfiguration.hosts.host, None for local (as local is only used for local tool overrides)
pub host_config: Option<Value>,
pub nixos_config: OnceLock<Value>,
pub nixos_unchecked_config: OnceLock<Value>,
pub pkgs_override: Option<Value>,
+ binary_cache: OnceCell<Option<BinaryCache>>,
+
// TODO: Move command helpers away with connectivity refactor
pub local: bool,
pub remowt: OnceCell<Remowt>,
@@ -131,13 +134,35 @@
AgentBundle::from_dir(agents_dir()?)
}
-enum RemoteDerivationMode {
- /// Closure is (pre)signed, remote host is expected to trust our signatures.
- CopySigned,
- /// Closure is not signed, using escalated nix daemon to bypass signature verification.
- CopyPrivileged,
- /// Closure is uploaded to attic, remote host is expected to trust its signature.
- Attic { name: String },
+#[derive(Clone)]
+pub enum BinaryCache {
+ Attic { name: String, pin: bool },
+}
+impl BinaryCache {
+ async fn upload(&self, local: &Remowt, path: &Utf8Path, pin_name: Option<&str>) -> Result<()> {
+ match self {
+ BinaryCache::Attic { name, pin } => {
+ let mut cmd = local.cmd("attic");
+ cmd.arg("push").arg(name).arg(path.as_str());
+ cmd.run()
+ .await
+ .context("failed to push path to attic (is attic-cli on PATH?)")?;
+
+ if *pin && let Some(pin_name) = pin_name {
+ let mut cmd = local.cmd("attic");
+ cmd.arg("pin")
+ .arg("create")
+ .arg(name)
+ .arg(pin_name)
+ .arg(path.as_str());
+ cmd.run()
+ .await
+ .context("failed to pin path in attic (is your attic-cli compatible?)")?;
+ }
+ Ok(())
+ }
+ }
+ }
}
impl ConfigHost {
@@ -156,6 +181,32 @@
.set(legacy)
.expect("legacy ssh store is already set")
}
+ pub fn binary_cache(&self) -> Result<Option<BinaryCache>> {
+ if let Some(v) = self.binary_cache.get() {
+ return Ok(v.clone());
+ }
+ let cache = if let Some(host_config) = &self.host_config {
+ let cache = nix_go!(host_config.cache);
+ let enable: bool = nix_go_json!(cache.enable);
+ if enable {
+ let kind: String = nix_go_json!(cache.kind);
+ let name: String = nix_go_json!(cache.name);
+ match kind.as_str() {
+ "attic" => {
+ let pin: bool = nix_go_json!(cache.pin);
+ Some(BinaryCache::Attic { name, pin })
+ }
+ v => bail!("unknown binary cache kind: {v}"),
+ }
+ } else {
+ None
+ }
+ } else {
+ None
+ };
+ let _ = self.binary_cache.set(cache.clone());
+ Ok(cache)
+ }
pub async fn deploy_kind(&self) -> Result<DeployKind> {
self.deploy_kind.get_or_try_init(|| async {
let remowt = self.remowt().await?;
@@ -238,8 +289,10 @@
let built = plugin
.build("out")
.context("failed to build the fleet nix plugin")?;
+ let pin = format!("{}-remowt-plugin-fleet", self.gc_root_prefix());
let copied = self
- .remote_derivation(&built)
+ // TODO: Pin? But this pin will be host-specific, and heavy because of nix dependency...
+ .remote_derivation(&built, Some(&pin))
.await
.context("failed to copy the fleet nix plugin to the host store")?;
let bin = copied.join("bin/remowt-plugin-fleet");
@@ -248,6 +301,13 @@
.run0_load_plugin_path(NIX_PLUGIN_ID, bin.as_str())
.await
.context("failed to load the fleet nix plugin")?;
+ // Should not fail, as plugin loading waits for connection on remote, and we only wait for forward... On the other hand, we should monitor connection to the remote itself somewhere too.
+ self.remowt()
+ .await?
+ .rpc()
+ .wait_for_connection_to(Address::Plugin(NIX_PLUGIN_ID))
+ .await
+ .map_err(|_| anyhow!("failed to wait"))?;
anyhow::Ok(())
})
.await?;
@@ -339,17 +399,31 @@
}
/// Returns path for futureproofing, as path might change i.e on conversion to CA
#[instrument(skip(self))]
- pub async fn remote_derivation(&self, path: &Utf8Path) -> Result<Utf8PathBuf> {
+ pub async fn remote_derivation(
+ &self,
+ path: &Utf8Path,
+ pin_name: Option<&str>,
+ ) -> Result<Utf8PathBuf> {
let path = path.to_owned();
if self.local {
// Path is located locally, thus already trusted.
return Ok(path);
}
+ let cache = self.binary_cache()?;
+ let mut cache = if let Some(cache) = cache {
+ let local = self.config.local_host().remowt().await?;
+ let path = path.to_owned();
+ let pin_name = pin_name.map(str::to_owned);
+ Some(tokio::spawn(async move {
+ if let Err(e) = cache.upload(&local, &path, pin_name.as_deref()).await {
+ warn!("failed to upload closure {path} to cache: {e:?}");
+ }
+ }))
+ } else {
+ None
+ };
let sign: Pin<Box<dyn Future<Output = Result<()>> + Send>> = {
let path = path.clone();
- let graph = DrvGraph::resolve(&eval_store(), &path)
- .context("failed to resolve graph to be uploaded")?;
- info!("signing {} paths", graph.nodes.len());
Box::pin(async move {
let local = self.config.local_host();
@@ -377,13 +451,62 @@
.expect("copy_to panicked")
.context("copying closure to remote store")?;
}
+
+ if let Some(cache) = cache.take() {
+ let _ = cache.await;
+ }
+
Ok(path)
}
+
+ /// Like [`Self::remote_derivation`], but instead of copying the closure over
+ /// the nix daemon tunnel, uploads it to an attic binary cache and has the
+ /// remote host substitute it from there. The remote is expected to already
+ /// trust the cache's signing key and have it configured as a substituter.
+ #[instrument(skip(self))]
+ pub async fn remote_derivation_attic(
+ &self,
+ path: &Utf8Path,
+ cache_name: &str,
+ ) -> Result<Utf8PathBuf> {
+ let path = path.to_owned();
+ if self.local {
+ // Path is located locally, thus already trusted.
+ return Ok(path);
+ }
+
+ info!("uploading {path} closure to attic cache {cache_name}");
+ let status = Command::new("attic")
+ .arg("push")
+ .arg(cache_name)
+ .arg(path.as_str())
+ .status()
+ .await
+ .context("failed to spawn attic-cli, is it on PATH?")?;
+ ensure!(status.success(), "attic push failed: {status}");
+
+ info!("substituting {path} on the remote host");
+ let nix = self.nix_client().await?;
+ let substituted = nix
+ .substitute(vec![path.clone()])
+ .await
+ .map_err(|e| anyhow!("{e:?}"))?
+ .map_err(|e| anyhow!("{e}"))?;
+ ensure!(
+ substituted.contains(&path),
+ "remote host failed to substitute {path} from attic cache {cache_name}"
+ );
+ Ok(path)
+ }
}
struct HostSecretDefinition(Value);
impl ConfigHost {
+ pub fn gc_root_prefix(&self) -> String {
+ format!("{}-{}", self.config.gc_root_prefix(), self.name)
+ }
+
// TOCTOU is possible here in case if config is changed, but this case is not handled anywhere anyway,
// assuming getting tags always returns the same value.
pub fn tags(&self) -> Result<Vec<String>> {
@@ -504,6 +627,7 @@
cell
},
pkgs_override: Some(self.default_pkgs.clone()),
+ binary_cache: OnceCell::new(),
local: true,
remowt: OnceCell::new(),
@@ -546,6 +670,7 @@
nixos_unchecked_config: OnceLock::new(),
groups: OnceLock::new(),
pkgs_override: None,
+ binary_cache: OnceCell::new(),
// TODO: Remove with connectivit refactor
local: self.localhost == name,
--- a/crates/fleet-base/src/keys.rs
+++ b/crates/fleet-base/src/keys.rs
@@ -9,6 +9,9 @@
use crate::{fleetdata::SecretOwner, host::Config};
impl Config {
+ pub fn gc_root_prefix(&self) -> String {
+ self.data.gc_root_prefix.clone()
+ }
fn cached_host_key(&self, host: &str) -> Option<String> {
let hosts = self.data.hosts.read().expect("no poisoning");
let key = hosts.get(host).map(|h| &h.encryption_key);
--- a/crates/fleet-base/src/primops.rs
+++ b/crates/fleet-base/src/primops.rs
@@ -119,7 +119,7 @@
.context("failed to build generator for target host")?;
let generator = host_on
- .remote_derivation(&generator)
+ .remote_derivation(&generator, None)
.await
.context("failed to copy generator to target host")?;
--- a/crates/nix-eval/src/lib.rs
+++ b/crates/nix-eval/src/lib.rs
@@ -39,7 +39,7 @@
make_attrs, make_bindings_builder, make_list, make_list_builder, realised_string,
realised_string_free, realised_string_get_buffer_size, realised_string_get_buffer_start,
realised_string_get_store_path, realised_string_get_store_path_count, register_primop,
- set_err_msg, setting_set, state_free, store_copy_closure, store_free, store_open,
+ set_err_msg, setting_get, setting_set, state_free, store_copy_closure, store_free, store_open,
store_parse_path, store_path_free, store_path_name, string_realise, value, value_call,
value_decref, value_force, value_incref,
};
@@ -390,6 +390,14 @@
with_default_context(|c, _| unsafe { setting_set(c, s.as_ptr(), v.as_ptr()) }).map(|_| ())
}
+pub fn get_setting(s: &CStr) -> Result<String> {
+ let mut out = String::new();
+ with_default_context(|c, _| unsafe {
+ setting_get(c, s.as_ptr(), Some(copy_nix_str), (&raw mut out).cast())
+ })?;
+ Ok(out)
+}
+
#[derive(Debug)]
pub struct AddedFile {
pub store_path: Utf8PathBuf,
1use std::collections::{HashMap, VecDeque};2use std::fmt::Arguments;3use std::sync::{LazyLock, Mutex};45use camino::{Utf8Path, Utf8PathBuf};6use cxx::ExternType;7use tracing::{8 Level, Span, debug, debug_span, error, error_span, info, info_span, trace, trace_span, warn,9 warn_span,10};11#[cfg(feature = "indicatif")]12use tracing_indicatif::span_ext::IndicatifSpanExt as _;13use vte::Parser;1415use crate::drv::extract_drv_name;1617#[derive(Debug)]18enum ActivityType {19 Unknown = 0,20 CopyPath = 100,21 FileTransfer = 101,22 Realise = 102,23 CopyPaths = 103,24 Builds = 104,25 Build = 105,26 OptimiseStore = 106,27 VerifyPaths = 107,28 Substitute = 108,29 QueryPathInfo = 109,30 PostBuildHook = 110,31 BuildWaiting = 111,32 FetchTree = 112,33}3435fn strip_prefix_suffix<'s, 'p>(a: &'s str, pref: &'p str, suff: &'p str) -> Option<&'s str> {36 a.strip_prefix(pref)?.strip_suffix(suff)37}3839fn parse_path(path: &str) -> Utf8PathBuf {40 Utf8PathBuf::from(strip_prefix_suffix(path, "\x1b[35;1m", "\x1b[0m").unwrap_or(path))41}4243fn parse_drv(drv: &str) -> String {44 let drv = parse_path(drv);45 extract_drv_name(&drv)46}47fn parse_host(host: &str) -> &str {48 if host.is_empty() || host == "local" {49 return "local";50 }51 52 host.strip_prefix("https://").unwrap_or(host)53}5455impl ActivityType {56 fn name(&self) -> &'static str {57 match self {58 ActivityType::Unknown => "nix",59 ActivityType::CopyPath => "nix::copy-path",60 ActivityType::FileTransfer => "nix::file-transfer",61 ActivityType::Realise => "nix::realise",62 ActivityType::CopyPaths => "nix::copy-paths",63 ActivityType::Builds => "nix::builds",64 ActivityType::Build => "nix::build",65 ActivityType::OptimiseStore => "nix::optimise-store",66 ActivityType::VerifyPaths => "nix::verify-paths",67 ActivityType::Substitute => "nix::substitute",68 ActivityType::QueryPathInfo => "nix::query-path-info",69 ActivityType::PostBuildHook => "nix::post-build-hook",70 ActivityType::BuildWaiting => "nix::build-waiting",71 ActivityType::FetchTree => "nix::fetch-tree",72 }73 }74 fn format(75 &self,76 values: &[FieldValue],77 s: &str,78 into: impl FnOnce(Arguments<'_>) -> Span,79 ) -> Span {80 use FieldValue::*;81 match (self, values) {82 (ActivityType::QueryPathInfo, [Str(drv), Str(host)]) => {83 let drv = parse_drv(drv);84 let host = parse_host(host);85 debug_span!(target: "nix::query-path-info", "querying", drv, host)86 }87 (ActivityType::Substitute, [Str(drv), Str(host)]) => {88 let drv = parse_drv(drv);89 let host = parse_host(host);90 debug_span!(target: "nix::substitute", "substituting", drv, host)91 }92 (ActivityType::CopyPath, [Str(drv), Str(from), Str(to)]) => {93 let drv = parse_drv(drv);94 let from = parse_host(from);95 let to = parse_host(to);96 debug_span!(target: "nix::copy-path", "copying", drv, from, to)97 }98 (ActivityType::Build, [Str(drv), Str(host), Int(_), Int(_)]) => {99 let drv = parse_drv(drv);100 let host = parse_host(host);101 info_span!(target: "nix::build", "building", drv, host)102 }103 (ActivityType::FileTransfer, [Str(file)]) => {104 info_span!(target: "nix::file-transfer", "downloading", file)105 }106 (ActivityType::Realise, []) => {107 debug_span!(target: "nix::realise", "realising")108 }109 (ActivityType::CopyPaths, []) => {110 debug_span!(target: "nix::copy-paths", "copying paths")111 }112 113 114 115 (ActivityType::Builds, []) => {116 debug_span!(target: "nix::builds", "building")117 }118 (ActivityType::Unknown, [])119 if s.starts_with("copying \"") && s.ends_with("\" to the store") =>120 {121 let tree = s122 .trim_start_matches("copying \"")123 .trim_end_matches("\" to the store");124 debug_span!(target: "nix::trees", "copying", tree)125 }126 (ActivityType::Unknown, [])127 if s.starts_with("copying '") && s.ends_with("' to the store") =>128 {129 let tree = s130 .trim_start_matches("copying '")131 .trim_end_matches("' to the store");132 debug_span!(target: "nix::trees", "copying", tree)133 }134 (ActivityType::Unknown, []) if s.starts_with("hashing '") && s.ends_with("'") => {135 let tree = s.trim_start_matches("hashing '").trim_end_matches("'");136 debug_span!(target: "nix::trees", "hashing", tree)137 }138 (ActivityType::Unknown, []) if s.starts_with("connecting to '") && s.ends_with("'") => {139 let host = s140 .trim_start_matches("connecting to '")141 .trim_end_matches("'");142 debug_span!(target: "nix::remote", "connecting", host)143 }144 (ActivityType::Unknown, [])145 if s.starts_with("copying outputs from '") && s.ends_with("'") =>146 {147 let host = s148 .trim_start_matches("copying outputs from '")149 .trim_end_matches("'");150 debug_span!(target: "nix::remote", "copying outputs", host)151 }152 (ActivityType::Unknown, [])153 if s.starts_with("copying dependencies to '") && s.ends_with("'") =>154 {155 let host = s156 .trim_start_matches("copying dependencies to '")157 .trim_end_matches("'");158 debug_span!(target: "nix::remote", "copying dependencies", host)159 }160 (ActivityType::Unknown, [])161 if s.starts_with("waiting for the upload lock to '") && s.ends_with("'") =>162 {163 let host = s164 .trim_start_matches("waiting for the upload lock to '")165 .trim_end_matches("'");166 debug_span!(target: "nix::remote", "waiting for upload lock", host)167 }168 (ActivityType::BuildWaiting, [])169 if s.starts_with("waiting for a machine to build '") && s.ends_with("'") =>170 {171 let drv = parse_drv(172 s.trim_start_matches("waiting for a machine to build '")173 .trim_end_matches("'"),174 );175 debug_span!(target: "nix::build-waiting", "waiting for available builder", drv)176 }177 (ActivityType::Unknown, []) if s == "querying info about missing paths" => {178 debug_span!(target: "nix::remote", "querying")179 }180 _ => into(format_args!("{}({values:?})", self.name())),181 }182 }183 fn from_int(v: u32) -> Self {184 match v {185 0 => Self::Unknown,186 100 => Self::CopyPath,187 101 => Self::FileTransfer,188 102 => Self::Realise,189 103 => Self::CopyPaths,190 104 => Self::Builds,191 105 => Self::Build,192 106 => Self::OptimiseStore,193 107 => Self::VerifyPaths,194 108 => Self::Substitute,195 109 => Self::QueryPathInfo,196 110 => Self::PostBuildHook,197 111 => Self::BuildWaiting,198 112 => Self::FetchTree,199 _ => {200 warn!("unknown nix action: {v}");201 Self::Unknown202 }203 }204 }205}206207#[derive(Debug)]208enum ResultType {209 FileLinked = 100,210 BuildLogLine = 101,211 UntrustedPath = 102,212 CorruptedPath = 103,213 SetPhase = 104,214 Progress = 105,215 SetExpected = 106,216 PostBuildLogLine = 107,217 FetchStatus = 108,218219 Unknown = 999,220}221impl ResultType {222 fn from_int(v: u32) -> Self {223 match v {224 100 => Self::FileLinked,225 101 => Self::BuildLogLine,226 102 => Self::UntrustedPath,227 103 => Self::CorruptedPath,228 104 => Self::SetPhase,229 105 => Self::Progress,230 106 => Self::SetExpected,231 107 => Self::PostBuildLogLine,232 108 => Self::FetchStatus,233234 _ => {235 warn!("unknown nix result: {v}");236 Self::Unknown237 }238 }239 }240}241#[derive(Clone, Copy)]242enum Verbosity {243 Error,244 Warn,245 Notice,246 Info,247 Talkative,248 Chatty,249 Debug,250 Vomit,251}252impl From<Verbosity> for tracing::Level {253 fn from(val: Verbosity) -> Self {254 match val {255 Verbosity::Error => Level::ERROR,256 Verbosity::Warn => Level::WARN,257 Verbosity::Notice => Level::WARN,258 Verbosity::Info => Level::INFO,259 Verbosity::Talkative => Level::DEBUG,260 Verbosity::Chatty => Level::DEBUG,261 Verbosity::Debug => Level::DEBUG,262 Verbosity::Vomit => Level::TRACE,263 }264 }265}266impl Verbosity {267 fn from_int(u: u32) -> Self {268 [269 Self::Error,270 Self::Warn,271 Self::Notice,272 Self::Info,273 Self::Talkative,274 Self::Chatty,275 Self::Debug,276 Self::Vomit,277 ]278 .get(u as usize)279 .cloned()280 .unwrap_or_else(|| {281 warn!("unknown log level: {u}");282 Verbosity::Vomit283 })284 }285}286287static NIX_SPAN_MAPPING: LazyLock<Mutex<HashMap<u64, Span>>> =288 LazyLock::new(|| Mutex::new(HashMap::new()));289290struct DrvGraphEntry {291 name: String,292 parent: Option<Utf8PathBuf>,293 span: Option<Span>,294 refcount: usize,295}296297static DRV_GRAPH: LazyLock<Mutex<HashMap<Utf8PathBuf, DrvGraphEntry>>> =298 LazyLock::new(|| Mutex::new(HashMap::new()));299300static ACTIVITY_TO_DRV: LazyLock<Mutex<HashMap<u64, Utf8PathBuf>>> =301 LazyLock::new(|| Mutex::new(HashMap::new()));302303pub struct BuildGraphGuard {304 paths: Vec<Utf8PathBuf>,305}306307impl Drop for BuildGraphGuard {308 fn drop(&mut self) {309 let mut drv_graph = DRV_GRAPH.lock().expect("not poisoned");310 for path in &self.paths {311 if let Some(entry) = drv_graph.get_mut(path) {312 entry.refcount -= 1;313 if entry.refcount == 0 {314 drv_graph.remove(path);315 }316 }317 }318 }319}320321pub fn register_build_graph(parent: &Span, graph: &crate::drv::DrvGraph) -> BuildGraphGuard {322 let mut drv_graph = DRV_GRAPH.lock().expect("not poisoned");323 let mut paths = Vec::new();324325 drv_graph326 .entry(graph.root.clone())327 .and_modify(|e| e.refcount += 1)328 .or_insert_with(|| DrvGraphEntry {329 name: graph.nodes[&graph.root].name.clone(),330 parent: None,331 span: Some(parent.clone()),332 refcount: 1,333 });334 paths.push(graph.root.clone());335336 let mut queue = VecDeque::new();337 queue.push_back(graph.root.clone());338339 let mut visited = std::collections::HashSet::new();340 visited.insert(graph.root.clone());341342 while let Some(path) = queue.pop_front() {343 let Some(node) = graph.nodes.get(&path) else {344 continue;345 };346 for dep_path in node.input_drvs.keys() {347 if !visited.insert(dep_path.clone()) {348 continue;349 }350 let Some(dep_node) = graph.nodes.get(dep_path) else {351 continue;352 };353 if let Some(entry) = drv_graph.get_mut(dep_path) {354 entry.refcount += 1;355 } else {356 drv_graph.insert(357 dep_path.clone(),358 DrvGraphEntry {359 name: dep_node.name.clone(),360 parent: Some(path.clone()),361 span: None,362 refcount: 1,363 },364 );365 }366 paths.push(dep_path.clone());367 queue.push_back(dep_path.clone());368 }369 }370371 BuildGraphGuard { paths }372}373374375376377378pub fn remove_drv_span(drv_path: &Utf8Path) {379 if let Some(entry) = DRV_GRAPH.lock().expect("not poisoned").get_mut(drv_path) {380 entry.span = None;381 }382}383384fn ensure_drv_span(drv_path: &Utf8Path) -> Option<Span> {385 let mut drv_graph = DRV_GRAPH.lock().expect("not poisoned");386387 if let Some(span) = drv_graph.get(drv_path).and_then(|e| e.span.clone()) {388 return Some(span);389 }390391 let mut chain = vec![];392 let mut current = drv_path.to_owned();393 loop {394 let Some(entry) = drv_graph.get(¤t) else {395 break;396 };397 if entry.span.is_some() {398 chain.push(current);399 break;400 }401 chain.push(current.clone());402 match &entry.parent {403 Some(p) => current = p.clone(),404 None => break,405 }406 }407408 if chain.is_empty() {409 return None;410 }411412 for i in (0..chain.len()).rev() {413 let path = &chain[i];414 if drv_graph.get(path).unwrap().span.is_some() {415 continue;416 }417 let parent_span = chain418 .get(i + 1)419 .and_then(|p| drv_graph.get(p))420 .and_then(|e| e.span.clone());421 let name = drv_graph.get(path).unwrap().name.clone();422 let span = {423 let _enter = parent_span.as_ref().map(|s| s.enter());424 info_span!(target: "nix::build", "building", drv = %name)425 };426 drv_graph.get_mut(path).unwrap().span = Some(span);427 }428429 drv_graph.get(drv_path).and_then(|e| e.span.clone())430}431432#[derive(Debug)]433enum FieldValue {434 Int(i32),435 Str(String),436}437438struct StartActivityBuilder {439 activity_id: u64,440 verbosity: Verbosity,441 typ: ActivityType,442 fields: Vec<FieldValue>,443}444impl StartActivityBuilder {445 fn add_int_field(&mut self, i: i32) {446 self.fields.push(FieldValue::Int(i));447 }448 fn add_string_field(&mut self, v: &[u8]) {449 let v = String::from_utf8_lossy(v);450 self.fields.push(FieldValue::Str(v.to_string()));451 }452 fn emit(&mut self, parent: u64, s: &str) {453 let graph_span = if matches!(self.typ, ActivityType::Build) {454 self.fields.first().and_then(|f| match f {455 FieldValue::Str(drv_path) => {456 let clean = parse_path(drv_path);457 let span = ensure_drv_span(&clean);458 if span.is_some() {459 ACTIVITY_TO_DRV460 .lock()461 .expect("not poisoned")462 .insert(self.activity_id, clean.to_owned());463 }464 span465 }466 _ => None,467 })468 } else {469 None470 };471472 let mut mapping = NIX_SPAN_MAPPING.lock().expect("not poisoned");473474 let span = if let Some(span) = graph_span {475 #[cfg(feature = "indicatif")]476 span.pb_start();477 span478 } else {479 let parent = mapping.get(&parent);480 let _in_parent = parent.map(|p| p.enter());481 let level: Level = self.verbosity.into();482 if level == Level::ERROR {483 self.typ484 .format(&self.fields, s, |v| error_span!("action", v))485 } else if level == Level::WARN {486 self.typ487 .format(&self.fields, s, |v| warn_span!("action", v))488 } else if level == Level::INFO {489 self.typ490 .format(&self.fields, s, |v| info_span!("action", v))491 } else if level == Level::DEBUG {492 self.typ493 .format(&self.fields, s, |v| debug_span!("action", v))494 } else {495 self.typ496 .format(&self.fields, s, |v| trace_span!("action", v))497 }498 };499 if !s.trim().is_empty() {500 let s = ansi_filter(s);501 #[cfg(feature = "indicatif")]502 {503 span.pb_set_message(&s);504 }505 let _e = span.enter();506 let level: Level = self.verbosity.into();507 if level == Level::ERROR {508 error!(target: "nix", "{}", s)509 } else if level == Level::WARN {510 warn!(target: "nix", "{}", s)511 } else if level == Level::INFO {512 info!(target: "nix", "{}", s)513 } else if level == Level::DEBUG {514 if s != "querying info about missing paths" {515 debug!(target: "nix", "{}", s)516 }517 } else {518 trace!(target: "nix", "{}", s)519 }520 } else {521 #[cfg(feature = "indicatif")]522 {523 span.pb_start();524 }525 }526 mapping.insert(self.activity_id, span);527 }528 fn emit_result(&mut self, ty: u32) {529 let mapping = NIX_SPAN_MAPPING.lock().expect("not poisoned");530531 let Some(parent) = mapping.get(&self.activity_id) else {532 panic!("unexpected result for dead parent");533 };534535 let _in_parent = parent.enter();536 let res = ResultType::from_int(ty);537538 use FieldValue::*;539 match (&res, self.fields.as_slice()) {540 541 (ResultType::BuildLogLine, [Str(s)]) => {542 let s = ansi_filter(s);543 info!(target: "nix::build", "{s}");544 }545 546 547 548 (ResultType::SetExpected, [Int(act_ty), Int(_expected)]) => {549 let _act_ty = ActivityType::from_int(*act_ty as u32);550 }551 (ResultType::SetPhase, [Str(phase)]) => {552 553 debug!(target: "nix::phase", phase)554 }555 (ResultType::Progress, [Int(_done), Int(_expected), Int(_), Int(_)]) => {556 #[cfg(feature = "indicatif")]557 {558 parent.pb_set_length(*_expected as u64);559 parent.pb_set_position(*_done as u64);560 }561 }562 _ => warn!("unknown progress report: {:?}({:?})", &res, &self.fields),563 }564 }565}566fn new_start_activity(activity_id: u64, lvl: u32, typ: u32) -> Box<StartActivityBuilder> {567 Box::new(StartActivityBuilder {568 activity_id,569 verbosity: Verbosity::from_int(lvl),570 typ: ActivityType::from_int(typ),571 fields: vec![],572 })573}574575fn emit_warn(v: &str) {576 warn!(target: "nix::eval", "{v}")577}578fn emit_stop(v: u64) {579 {580 let mut mapping = NIX_SPAN_MAPPING.lock().expect("not poisoned");581 mapping.remove(&v);582 }583 if let Some(drv_path) = ACTIVITY_TO_DRV.lock().expect("not poisoned").remove(&v) {584 if let Some(entry) = DRV_GRAPH.lock().expect("not poisoned").get_mut(&drv_path) {585 entry.span = None;586 }587 }588}589fn emit_log(lvl: u32, v: &[u8]) {590 let verbosity = Verbosity::from_int(lvl);591 let level: Level = verbosity.into();592 let v = String::from_utf8_lossy(v);593 if level == Level::ERROR {594 error!(target: "nix", "{v}")595 } else if level == Level::WARN {596 warn!(target: "nix", "{v}")597 } else if level == Level::INFO {598 info!(target: "nix", "{v}")599 } else if level == Level::DEBUG {600 if v != "querying info about missing paths" {601 debug!(target: "nix", "{v}")602 }603 } else {604 trace!(target: "nix", "{v}")605 }606}607608struct AnsiFiltered {609 output: String,610}611impl vte::Perform for AnsiFiltered {612 fn print(&mut self, c: char) {613 self.output.push(c);614 }615616 fn execute(&mut self, byte: u8) {617 618 if byte == b'\n' {619 self.output.push('\n');620 } else if byte == b'\t' {621 622 self.output.push('\t');623 }624 }625626 fn osc_dispatch(&mut self, _params: &[&[u8]], _bell_terminated: bool) {}627 fn esc_dispatch(&mut self, _intermediates: &[u8], _ignore: bool, _byte: u8) {}628629 fn csi_dispatch(630 &mut self,631 params: &vte::Params,632 _intermediates: &[u8],633 _ignore: bool,634 action: char,635 ) {636 use std::fmt::Write;637 if action != 'm' {638 639 return;640 }641 self.output.push_str("\x1b[");642 for (i, par) in params.iter().enumerate() {643 if i != 0 {644 let _ = write!(self.output, ";");645 }646 for (i, sub) in par.iter().enumerate() {647 if i != 0 {648 let _ = write!(self.output, ":");649 }650 let _ = write!(self.output, "{sub}");651 }652 }653 self.output.push(action);654 }655}656fn ansi_filter(i: &str) -> String {657 let mut out = AnsiFiltered {658 output: String::new(),659 };660 let mut parser = Parser::new();661662 663 for chunk in i.as_bytes().chunks(50) {664 parser.advance(&mut out, chunk);665 }666667 out.output668}669670#[derive(Debug)]671pub struct StackFrame {672 pub msg: String,673 pub pos: String,674}675676#[derive(Debug)]677pub struct ErrorInfoBuilder {678 level: Level,679 msg: String,680 pub stack_frames: Vec<StackFrame>,681}682fn new_error_info(lvl: u32, v: &[u8]) -> Box<ErrorInfoBuilder> {683 let verbosity = Verbosity::from_int(lvl);684 let level: Level = verbosity.into();685 let v = String::from_utf8_lossy(v);686 Box::new(ErrorInfoBuilder {687 level,688 msg: v.to_string(),689 stack_frames: Vec::new(),690 })691}692impl ErrorInfoBuilder {693 fn push_stack_frame(&mut self, v: &[u8], pos: &[u8]) {694 let v = String::from_utf8_lossy(v);695 let pos = String::from_utf8_lossy(pos);696 self.stack_frames.push(StackFrame {697 msg: v.to_string(),698 pos: pos.to_string(),699 });700 }701 fn emit_error_info(&mut self) {702 error!("{}", self.msg);703 for frame in &self.stack_frames {704 error!(" {} at {}", frame.msg, frame.pos)705 }706 }707}708709#[cxx::bridge]710pub mod nix_logging_cxx {711 extern "Rust" {712 type StartActivityBuilder;713 fn new_start_activity(activity_id: u64, lvl: u32, typ: u32) -> Box<StartActivityBuilder>;714 fn add_int_field(&mut self, i: i32);715 fn add_string_field(&mut self, v: &[u8]);716 fn emit(&mut self, parent: u64, s: &str);717 fn emit_result(&mut self, ty: u32);718 }719 extern "Rust" {720 type ErrorInfoBuilder;721 fn new_error_info(lvl: u32, v: &[u8]) -> Box<ErrorInfoBuilder>;722 fn push_stack_frame(&mut self, v: &[u8], pos: &[u8]);723 fn emit_error_info(&mut self);724 }725 extern "Rust" {726 fn emit_warn(v: &str);727 fn emit_stop(id: u64);728 fn emit_log(lvl: u32, v: &[u8]);729 }730 unsafe extern "C++" {731 include!("nix-eval/src/logging.hh");732733 type nix_c_context = crate::nix_raw::c_context;734735 fn apply_tracing_logger();736 unsafe fn extract_error_info(ctx: *const nix_c_context) -> Box<ErrorInfoBuilder>;737 }738}739740unsafe impl ExternType for crate::nix_raw::c_context {741 type Id = cxx::type_id!("nix_c_context");742743 type Kind = cxx::kind::Opaque;744}
--- a/crates/nix-eval/src/scheduler.rs
+++ b/crates/nix-eval/src/scheduler.rs
@@ -45,9 +45,36 @@
},
}
+// Derivations whose names contain any of these are "heavy": memory-hungry builds that already
+// saturate all cores internally (e.g. composable-kernels), so running anything alongside them
+// just thrashes RAM. A heavy build grabs every semaphore permit and runs exclusively.
+const DEFAULT_HEAVY_DRVS: &[&str] = &[
+ "composable-kernel",
+ "composable_kernel",
+ "ghc",
+ "gfortran",
+ "llvm",
+ "clang",
+];
+
+fn heavy_patterns_from_env() -> Vec<String> {
+ let mut pats: Vec<String> = DEFAULT_HEAVY_DRVS.iter().map(|s| (*s).to_owned()).collect();
+ if let Ok(extra) = std::env::var("FLEET_HEAVY_DRVS") {
+ pats.extend(
+ extra
+ .split(',')
+ .map(str::trim)
+ .filter(|s| !s.is_empty())
+ .map(str::to_owned),
+ );
+ }
+ pats
+}
+
pub struct Scheduler {
store: Arc<Store>,
parallelism: usize,
+ heavy_patterns: Vec<String>,
events: broadcast::Sender<BuildEvent>,
}
@@ -58,10 +85,20 @@
Self {
store: eval_store(),
parallelism,
+ heavy_patterns: heavy_patterns_from_env(),
events,
}
}
+ // Permits a drv must hold to run. Heavy drvs take all of them, pausing everything else.
+ fn permits_for(&self, name: &str) -> u32 {
+ if self.heavy_patterns.iter().any(|p| name.contains(p)) {
+ self.parallelism as u32
+ } else {
+ 1
+ }
+ }
+
pub fn subscribe(&self) -> broadcast::Receiver<BuildEvent> {
self.events.subscribe()
}
@@ -139,6 +176,7 @@
.get(&path)
.map(|n| n.name.clone())
.unwrap_or_default();
+ crate::logging::remove_drv_span(&path);
let _ = self.events.send(BuildEvent::DrvCancelled {
drv_path: path.clone(),
name,
@@ -153,8 +191,21 @@
let graph = graph.clone();
let wanted_here = wanted.get(&path).cloned().unwrap_or_default();
let store = self.store.clone();
+ let permits = graph
+ .nodes
+ .get(&path)
+ .map(|n| self.permits_for(&n.name))
+ .unwrap_or(1);
in_flight.push(tokio::spawn(async move {
- let _permit = sem.acquire_owned().await.expect("semaphore not closed");
+ if permits > 1 {
+ debug!(permits, "heavy drv: waiting for exclusive build slot");
+ }
+ // Heavy drvs acquire every permit, so they only start once all in-flight
+ // builds drain and nothing new can slip in (the semaphore is FIFO-fair).
+ let _permit = sem
+ .acquire_many_owned(permits)
+ .await
+ .expect("semaphore not closed");
let node = graph
.nodes
.get(&path)
@@ -224,6 +275,7 @@
propagate_done(&dependents, &mut indeg, &mut ready, &finished);
}
Err(e) => {
+ crate::logging::remove_drv_span(&finished);
failed.insert(finished.clone(), format!("{e:#}"));
mark_tainted(&dependents, &finished, &mut tainted);
propagate_done(&dependents, &mut indeg, &mut ready, &finished);
@@ -368,11 +420,23 @@
// TODO: Parallelism as a metric works poorly with multiple machines, but I haven't thought about bringing
// hercy here yet. In case of remote machines - they will handle parallelism on their own, and this one
// will work as a hard cap.
+/// Number of derivations to build concurrently, taken from Nix's `max-jobs`
+/// setting. `auto` resolves to the CPU count, matching Nix itself.
+fn max_jobs() -> usize {
+ let cores = || {
+ std::thread::available_parallelism()
+ .map(|p| p.get())
+ .unwrap_or(4)
+ };
+ match crate::get_setting(c"max-jobs") {
+ Ok(v) if v.trim() == "auto" => cores(),
+ Ok(v) => v.trim().parse().unwrap_or_else(|_| cores()),
+ Err(_) => cores(),
+ }
+}
+
pub fn build_graph_sync(graph: Arc<DrvGraph>, root_outputs: Vec<String>) -> Result<()> {
- let parallelism = std::thread::available_parallelism()
- .map(|p| p.get())
- .unwrap_or(4);
- let scheduler = Scheduler::new(parallelism);
+ let scheduler = Scheduler::new(max_jobs());
crate::await_in_nix(async move { scheduler.run(graph, root_outputs).await })
.context("scheduler run")
}
--- a/crates/remowt-fleet/src/lib.rs
+++ b/crates/remowt-fleet/src/lib.rs
@@ -28,6 +28,8 @@
Sign(String),
#[error("listing generations failed: {0}")]
ListGenerations(String),
+ #[error("substitution failed: {0}")]
+ Substitute(String),
}
#[endpoints(ns = 91)]
@@ -60,6 +62,20 @@
.map_err(|e| NixError::Sign(e.to_string()))
}
+ /// Download the given paths (and their closures) into the local store using
+ /// the configured substituters. Used to fetch closures that were uploaded to
+ /// a binary cache (e.g. attic) instead of copied over the nix daemon tunnel.
+ #[endpoints(id = 6)]
+ async fn substitute(&self, paths: Vec<Utf8PathBuf>) -> Result<Vec<Utf8PathBuf>, NixError> {
+ spawn_blocking(move || {
+ let store = eval_store();
+ store.substitute_paths(&paths)
+ })
+ .await
+ .expect("substitution panicked")
+ .map_err(|e| NixError::Substitute(e.to_string()))
+ }
+
#[endpoints(id = 5)]
async fn list_generations(
&self,
--- /dev/null
+++ b/modules/cache.nix
@@ -0,0 +1,61 @@
+{
+ lib,
+ fleetLib,
+ config,
+ ...
+}:
+let
+ inherit (lib.options) mkOption mkEnableOption literalExpression;
+ inherit (lib.types)
+ enum
+ str
+ bool
+ submodule
+ ;
+ inherit (fleetLib.options) mkHostsOption;
+
+ _file = ./cache.nix;
+
+ cacheOptions = {
+ enable = mkEnableOption "fleet binary cache";
+ kind = mkOption {
+ description = ''
+ Kind of the binary cache to use.
+ '';
+ type = enum [ "attic" ];
+ default = "attic";
+ };
+ name = mkOption {
+ description = ''
+ Name of the binary cache.
+ '';
+ type = str;
+ };
+ pin = mkOption {
+ description = ''
+ Whether to pin pushed paths in the attic cache, preventing them from
+ being garbage collected.
+
+ Requires compatible fork of attic server+client: https://github.com/zhaofengli/attic/pull/227#issuecomment-4759154796
+ '';
+ type = bool;
+ default = false;
+ };
+ };
+in
+{
+ options = {
+ cache = cacheOptions;
+ hosts = mkHostsOption {
+ inherit _file;
+ options.cache = mkOption {
+ description = ''
+ Binary cache configuration for this host.
+ '';
+ type = submodule { options = cacheOptions; };
+ default = config.cache;
+ defaultText = literalExpression "fleetConfiguration.cache";
+ };
+ };
+ };
+}
--- a/modules/module-list.nix
+++ b/modules/module-list.nix
@@ -1,5 +1,6 @@
[
./assertions.nix
+ ./cache.nix
./fleetLib.nix
./hosts.nix
./nixos.nix
--- a/pkgs/default.nix
+++ b/pkgs/default.nix
@@ -12,8 +12,6 @@
fleet-generator-helper = callPackage ./fleet-generator-helper.nix { inherit craneLib; };
inherit remowt-agents-bundle;
- remowt-plugin-fleet = callPackage ./remowt-plugin-fleet.nix {
- inherit craneLib inputs remowt-agents-bundle;
- };
+ remowt-plugin-fleet = callPackage ./remowt-plugin-fleet.nix { inherit craneLib inputs; };
remowt-ssh = callPackage ./remowt-ssh.nix { inherit craneLib remowt-agents-bundle; };
}
--- a/remowt/cmds/remowt-agent/src/main.rs
+++ b/remowt/cmds/remowt-agent/src/main.rs
@@ -273,10 +273,11 @@
let helper = SocketHelper {
fallback: SuidHelper,
};
- register_auth_agent(&system_conn, Agent::new(helper, RofiPrompter)).await?;
+ let rofi = RofiPrompter::default();
+ register_auth_agent(&system_conn, Agent::new(helper, rofi.clone())).await?;
let mut rpc = Rpc::<BifConfig>::new(Address::User);
- serve_prompts(&mut rpc, RofiPrompter);
+ serve_prompts(&mut rpc, rofi);
gateway::serve(rpc.clone(), &gateway::local_socket()?).await?;
--- a/remowt/cmds/remowt-ssh/Cargo.toml
+++ b/remowt/cmds/remowt-ssh/Cargo.toml
@@ -8,6 +8,7 @@
[dependencies]
clap = { workspace = true, features = ["derive"] }
tracing-subscriber.workspace = true
+tracing-journald.workspace = true
remowt-link-shared.workspace = true
remowt-client.workspace = true
tokio = { workspace = true, features = [
--- a/remowt/cmds/remowt-ssh/src/main.rs
+++ b/remowt/cmds/remowt-ssh/src/main.rs
@@ -20,6 +20,8 @@
use tokio::io::{AsyncRead, ReadBuf};
use tokio::signal::unix::{signal, SignalKind};
use tracing::debug;
+use tracing_subscriber::prelude::*;
+use tracing_subscriber::EnvFilter;
#[derive(Parser)]
enum Opts {
@@ -45,10 +47,16 @@
#[tokio::main(flavor = "current_thread")]
async fn main() -> anyhow::Result<()> {
- tracing_subscriber::fmt()
- .with_writer(std::io::stderr)
- .without_time()
- .init();
+ // Log to the journal, not stderr: this is an interactive client that puts
+ // the terminal in raw mode, so anything on stderr corrupts the session
+ // output. If the journal socket is unavailable, drop logs rather than fall
+ // back to stderr.
+ if let Ok(journald) = tracing_journald::layer() {
+ tracing_subscriber::registry()
+ .with(EnvFilter::from_default_env())
+ .with(journald.with_syslog_identifier("remowt-ssh".to_owned()))
+ .init();
+ }
let opts = Opts::parse();
let bundle = AgentBundle::from_dir(agents_dir()?)?;
--- a/remowt/crates/remowt-ui-prompt/Cargo.toml
+++ b/remowt/crates/remowt-ui-prompt/Cargo.toml
@@ -12,5 +12,5 @@
remowt-link-shared.workspace = true
serde.workspace = true
thiserror.workspace = true
-tokio = { workspace = true, features = ["io-util", "macros", "process", "rt"] }
+tokio = { workspace = true, features = ["io-util", "macros", "process", "rt", "sync"] }
tracing.workspace = true
--- a/remowt/crates/remowt-ui-prompt/src/auto.rs
+++ b/remowt/crates/remowt-ui-prompt/src/auto.rs
@@ -24,7 +24,7 @@
};
Self {
remote,
- fallback: RofiPrompter,
+ fallback: RofiPrompter::default(),
}
}
--- a/remowt/crates/remowt-ui-prompt/src/rofi.rs
+++ b/remowt/crates/remowt-ui-prompt/src/rofi.rs
@@ -1,13 +1,18 @@
use std::process::Stdio;
+use std::sync::Arc;
use tokio::io::AsyncWriteExt;
use tokio::process::Command;
+use tokio::sync::Mutex;
use tracing::trace;
use crate::{Error, Prompter, Result, Source};
-#[derive(Clone)]
-pub struct RofiPrompter;
+#[derive(Clone, Default)]
+pub struct RofiPrompter {
+ // Rofi can't run concurrently; serialize invocations.
+ lock: Arc<Mutex<()>>,
+}
fn fixup_prompt(prompt: &str) -> &str {
// Rofi always appends such suffix
@@ -27,6 +32,7 @@
source: &[Source],
) -> Result<u32> {
trace!("rofi radio");
+ let _guard = self.lock.lock().await;
let mut cmd = rofi_command();
let mesg = if source.is_empty() {
description.to_owned()
@@ -113,6 +119,7 @@
source: &[Source],
) -> Result<String> {
trace!("rofi text");
+ let _guard = self.lock.lock().await;
let mut cmd = rofi_command();
let mesg = if source.is_empty() {
description.to_owned()
@@ -162,6 +169,7 @@
async fn display_text(&self, error: bool, description: &str, source: &[Source]) -> Result<()> {
trace!("rofi display");
+ let _guard = self.lock.lock().await;
let mut cmd = rofi_command();
let mut mesg = if source.is_empty() {
description.to_owned()
@@ -206,7 +214,7 @@
#[ignore = "interactive"]
async fn test() {
let prompter = PrependSourcePrompter {
- prompter: RofiPrompter,
+ prompter: RofiPrompter::default(),
description: "test".to_owned(),
source: vec![Source(Cow::Borrowed("ssh"))],
};