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 (ActivityType::Unknown, [])113 if s.starts_with("copying \"") && s.ends_with("\" to the store") =>114 {115 let tree = s116 .trim_start_matches("copying \"")117 .trim_end_matches("\" to the store");118 debug_span!(target: "nix::trees", "copying", tree)119 }120 (ActivityType::Unknown, [])121 if s.starts_with("copying '") && s.ends_with("' to the store") =>122 {123 let tree = s124 .trim_start_matches("copying '")125 .trim_end_matches("' to the store");126 debug_span!(target: "nix::trees", "copying", tree)127 }128 (ActivityType::Unknown, []) if s.starts_with("hashing '") && s.ends_with("'") => {129 let tree = s.trim_start_matches("hashing '").trim_end_matches("'");130 debug_span!(target: "nix::trees", "hashing", tree)131 }132 (ActivityType::Unknown, []) if s.starts_with("connecting to '") && s.ends_with("'") => {133 let host = s134 .trim_start_matches("connecting to '")135 .trim_end_matches("'");136 debug_span!(target: "nix::remote", "connecting", host)137 }138 (ActivityType::Unknown, [])139 if s.starts_with("copying outputs from '") && s.ends_with("'") =>140 {141 let host = s142 .trim_start_matches("copying outputs from '")143 .trim_end_matches("'");144 debug_span!(target: "nix::remote", "copying outputs", host)145 }146 (ActivityType::Unknown, [])147 if s.starts_with("copying dependencies to '") && s.ends_with("'") =>148 {149 let host = s150 .trim_start_matches("copying dependencies to '")151 .trim_end_matches("'");152 debug_span!(target: "nix::remote", "copying dependencies", host)153 }154 (ActivityType::Unknown, [])155 if s.starts_with("waiting for the upload lock to '") && s.ends_with("'") =>156 {157 let host = s158 .trim_start_matches("waiting for the upload lock to '")159 .trim_end_matches("'");160 debug_span!(target: "nix::remote", "waiting for upload lock", host)161 }162 (ActivityType::BuildWaiting, [])163 if s.starts_with("waiting for a machine to build '") && s.ends_with("'") =>164 {165 let drv = parse_drv(166 s.trim_start_matches("waiting for a machine to build '")167 .trim_end_matches("'"),168 );169 debug_span!(target: "nix::build-waiting", "waiting for available builder", drv)170 }171 (ActivityType::Unknown, []) if s == "querying info about missing paths" => {172 debug_span!(target: "nix::remote", "querying")173 }174 _ => into(format_args!("{}({values:?})", self.name())),175 }176 }177 fn from_int(v: u32) -> Self {178 match v {179 0 => Self::Unknown,180 100 => Self::CopyPath,181 101 => Self::FileTransfer,182 102 => Self::Realise,183 103 => Self::CopyPaths,184 104 => Self::Builds,185 105 => Self::Build,186 106 => Self::OptimiseStore,187 107 => Self::VerifyPaths,188 108 => Self::Substitute,189 109 => Self::QueryPathInfo,190 110 => Self::PostBuildHook,191 111 => Self::BuildWaiting,192 112 => Self::FetchTree,193 _ => {194 warn!("unknown nix action: {v}");195 Self::Unknown196 }197 }198 }199}200201#[derive(Debug)]202enum ResultType {203 FileLinked = 100,204 BuildLogLine = 101,205 UntrustedPath = 102,206 CorruptedPath = 103,207 SetPhase = 104,208 Progress = 105,209 SetExpected = 106,210 PostBuildLogLine = 107,211 FetchStatus = 108,212213 Unknown = 999,214}215impl ResultType {216 fn from_int(v: u32) -> Self {217 match v {218 100 => Self::FileLinked,219 101 => Self::BuildLogLine,220 102 => Self::UntrustedPath,221 103 => Self::CorruptedPath,222 104 => Self::SetPhase,223 105 => Self::Progress,224 106 => Self::SetExpected,225 107 => Self::PostBuildLogLine,226 108 => Self::FetchStatus,227228 _ => {229 warn!("unknown nix result: {v}");230 Self::Unknown231 }232 }233 }234}235#[derive(Clone, Copy)]236enum Verbosity {237 Error,238 Warn,239 Notice,240 Info,241 Talkative,242 Chatty,243 Debug,244 Vomit,245}246impl From<Verbosity> for tracing::Level {247 fn from(val: Verbosity) -> Self {248 match val {249 Verbosity::Error => Level::ERROR,250 Verbosity::Warn => Level::WARN,251 Verbosity::Notice => Level::WARN,252 Verbosity::Info => Level::INFO,253 Verbosity::Talkative => Level::DEBUG,254 Verbosity::Chatty => Level::DEBUG,255 Verbosity::Debug => Level::DEBUG,256 Verbosity::Vomit => Level::TRACE,257 }258 }259}260impl Verbosity {261 fn from_int(u: u32) -> Self {262 [263 Self::Error,264 Self::Warn,265 Self::Notice,266 Self::Info,267 Self::Talkative,268 Self::Chatty,269 Self::Debug,270 Self::Vomit,271 ]272 .get(u as usize)273 .cloned()274 .unwrap_or_else(|| {275 warn!("unknown log level: {u}");276 Verbosity::Vomit277 })278 }279}280281static NIX_SPAN_MAPPING: LazyLock<Mutex<HashMap<u64, Span>>> =282 LazyLock::new(|| Mutex::new(HashMap::new()));283284struct DrvGraphEntry {285 name: String,286 parent: Option<Utf8PathBuf>,287 span: Option<Span>,288 refcount: usize,289}290291static DRV_GRAPH: LazyLock<Mutex<HashMap<Utf8PathBuf, DrvGraphEntry>>> =292 LazyLock::new(|| Mutex::new(HashMap::new()));293294static ACTIVITY_TO_DRV: LazyLock<Mutex<HashMap<u64, Utf8PathBuf>>> =295 LazyLock::new(|| Mutex::new(HashMap::new()));296297pub struct BuildGraphGuard {298 paths: Vec<Utf8PathBuf>,299}300301impl Drop for BuildGraphGuard {302 fn drop(&mut self) {303 let mut drv_graph = DRV_GRAPH.lock().expect("not poisoned");304 for path in &self.paths {305 if let Some(entry) = drv_graph.get_mut(path) {306 entry.refcount -= 1;307 if entry.refcount == 0 {308 drv_graph.remove(path);309 }310 }311 }312 }313}314315pub fn register_build_graph(parent: &Span, graph: &crate::drv::DrvGraph) -> BuildGraphGuard {316 let mut drv_graph = DRV_GRAPH.lock().expect("not poisoned");317 let mut paths = Vec::new();318319 drv_graph320 .entry(graph.root.clone())321 .and_modify(|e| e.refcount += 1)322 .or_insert_with(|| DrvGraphEntry {323 name: graph.nodes[&graph.root].name.clone(),324 parent: None,325 span: Some(parent.clone()),326 refcount: 1,327 });328 paths.push(graph.root.clone());329330 let mut queue = VecDeque::new();331 queue.push_back(graph.root.clone());332333 let mut visited = std::collections::HashSet::new();334 visited.insert(graph.root.clone());335336 while let Some(path) = queue.pop_front() {337 let Some(node) = graph.nodes.get(&path) else {338 continue;339 };340 for dep_path in node.input_drvs.keys() {341 if !visited.insert(dep_path.clone()) {342 continue;343 }344 let Some(dep_node) = graph.nodes.get(dep_path) else {345 continue;346 };347 if let Some(entry) = drv_graph.get_mut(dep_path) {348 entry.refcount += 1;349 } else {350 drv_graph.insert(351 dep_path.clone(),352 DrvGraphEntry {353 name: dep_node.name.clone(),354 parent: Some(path.clone()),355 span: None,356 refcount: 1,357 },358 );359 }360 paths.push(dep_path.clone());361 queue.push_back(dep_path.clone());362 }363 }364365 BuildGraphGuard { paths }366}367368fn ensure_drv_span(drv_path: &Utf8Path) -> Option<Span> {369 let mut drv_graph = DRV_GRAPH.lock().expect("not poisoned");370371 if let Some(span) = drv_graph.get(drv_path).and_then(|e| e.span.clone()) {372 return Some(span);373 }374375 let mut chain = vec![];376 let mut current = drv_path.to_owned();377 loop {378 let Some(entry) = drv_graph.get(¤t) else {379 break;380 };381 if entry.span.is_some() {382 chain.push(current);383 break;384 }385 chain.push(current.clone());386 match &entry.parent {387 Some(p) => current = p.clone(),388 None => break,389 }390 }391392 if chain.is_empty() {393 return None;394 }395396 for i in (0..chain.len()).rev() {397 let path = &chain[i];398 if drv_graph.get(path).unwrap().span.is_some() {399 continue;400 }401 let parent_span = chain402 .get(i + 1)403 .and_then(|p| drv_graph.get(p))404 .and_then(|e| e.span.clone());405 let name = drv_graph.get(path).unwrap().name.clone();406 let span = {407 let _enter = parent_span.as_ref().map(|s| s.enter());408 info_span!(target: "nix::build", "building", drv = %name)409 };410 drv_graph.get_mut(path).unwrap().span = Some(span);411 }412413 drv_graph.get(drv_path).and_then(|e| e.span.clone())414}415416#[derive(Debug)]417enum FieldValue {418 Int(i32),419 Str(String),420}421422struct StartActivityBuilder {423 activity_id: u64,424 verbosity: Verbosity,425 typ: ActivityType,426 fields: Vec<FieldValue>,427}428impl StartActivityBuilder {429 fn add_int_field(&mut self, i: i32) {430 self.fields.push(FieldValue::Int(i));431 }432 fn add_string_field(&mut self, v: &[u8]) {433 let v = String::from_utf8_lossy(v);434 self.fields.push(FieldValue::Str(v.to_string()));435 }436 fn emit(&mut self, parent: u64, s: &str) {437 let graph_span = if matches!(self.typ, ActivityType::Build) {438 self.fields.first().and_then(|f| match f {439 FieldValue::Str(drv_path) => {440 let clean = parse_path(drv_path);441 let span = ensure_drv_span(&clean);442 if span.is_some() {443 ACTIVITY_TO_DRV444 .lock()445 .expect("not poisoned")446 .insert(self.activity_id, clean.to_owned());447 }448 span449 }450 _ => None,451 })452 } else {453 None454 };455456 let mut mapping = NIX_SPAN_MAPPING.lock().expect("not poisoned");457458 let span = if let Some(span) = graph_span {459 #[cfg(feature = "indicatif")]460 span.pb_start();461 span462 } else {463 let parent = mapping.get(&parent);464 let _in_parent = parent.map(|p| p.enter());465 let level: Level = self.verbosity.into();466 if level == Level::ERROR {467 self.typ468 .format(&self.fields, s, |v| error_span!("action", v))469 } else if level == Level::WARN {470 self.typ471 .format(&self.fields, s, |v| warn_span!("action", v))472 } else if level == Level::INFO {473 self.typ474 .format(&self.fields, s, |v| info_span!("action", v))475 } else if level == Level::DEBUG {476 self.typ477 .format(&self.fields, s, |v| debug_span!("action", v))478 } else {479 self.typ480 .format(&self.fields, s, |v| trace_span!("action", v))481 }482 };483 if !s.trim().is_empty() {484 let s = ansi_filter(s);485 #[cfg(feature = "indicatif")]486 {487 span.pb_set_message(&s);488 }489 let _e = span.enter();490 let level: Level = self.verbosity.into();491 if level == Level::ERROR {492 error!(target: "nix", "{}", s)493 } else if level == Level::WARN {494 warn!(target: "nix", "{}", s)495 } else if level == Level::INFO {496 info!(target: "nix", "{}", s)497 } else if level == Level::DEBUG {498 if s != "querying info about missing paths" {499 debug!(target: "nix", "{}", s)500 }501 } else {502 trace!(target: "nix", "{}", s)503 }504 } else {505 #[cfg(feature = "indicatif")]506 {507 span.pb_start();508 }509 }510 mapping.insert(self.activity_id, span);511 }512 fn emit_result(&mut self, ty: u32) {513 let mapping = NIX_SPAN_MAPPING.lock().expect("not poisoned");514515 let Some(parent) = mapping.get(&self.activity_id) else {516 panic!("unexpected result for dead parent");517 };518519 let _in_parent = parent.enter();520 let res = ResultType::from_int(ty);521522 use FieldValue::*;523 match (&res, self.fields.as_slice()) {524 525 (ResultType::BuildLogLine, [Str(s)]) => {526 let s = ansi_filter(s);527 info!("{s}");528 }529 530 531 532 (ResultType::SetExpected, [Int(act_ty), Int(_expected)]) => {533 let _act_ty = ActivityType::from_int(*act_ty as u32);534 }535 (ResultType::SetPhase, [Str(phase)]) => {536 537 debug!(target: "nix::phase", phase)538 }539 (ResultType::Progress, [Int(_done), Int(_expected), Int(_), Int(_)]) => {540 #[cfg(feature = "indicatif")]541 {542 parent.pb_set_length(*_expected as u64);543 parent.pb_set_position(*_done as u64);544 }545 }546 _ => warn!("unknown progress report: {:?}({:?})", &res, &self.fields),547 }548 }549}550fn new_start_activity(activity_id: u64, lvl: u32, typ: u32) -> Box<StartActivityBuilder> {551 Box::new(StartActivityBuilder {552 activity_id,553 verbosity: Verbosity::from_int(lvl),554 typ: ActivityType::from_int(typ),555 fields: vec![],556 })557}558559fn emit_warn(v: &str) {560 warn!(target: "nix::eval", "{v}")561}562fn emit_stop(v: u64) {563 {564 let mut mapping = NIX_SPAN_MAPPING.lock().expect("not poisoned");565 mapping.remove(&v);566 }567 if let Some(drv_path) = ACTIVITY_TO_DRV.lock().expect("not poisoned").remove(&v) {568 if let Some(entry) = DRV_GRAPH.lock().expect("not poisoned").get_mut(&drv_path) {569 entry.span = None;570 }571 }572}573fn emit_log(lvl: u32, v: &[u8]) {574 let verbosity = Verbosity::from_int(lvl);575 let level: Level = verbosity.into();576 let v = String::from_utf8_lossy(v);577 if level == Level::ERROR {578 error!(target: "nix", "{v}")579 } else if level == Level::WARN {580 warn!(target: "nix", "{v}")581 } else if level == Level::INFO {582 info!(target: "nix", "{v}")583 } else if level == Level::DEBUG {584 if v != "querying info about missing paths" {585 debug!(target: "nix", "{v}")586 }587 } else {588 trace!(target: "nix", "{v}")589 }590}591592struct AnsiFiltered {593 output: String,594}595impl vte::Perform for AnsiFiltered {596 fn print(&mut self, c: char) {597 self.output.push(c);598 }599600 fn execute(&mut self, byte: u8) {601 602 if byte == b'\n' {603 self.output.push('\n');604 } else if byte == b'\t' {605 606 self.output.push('\t');607 }608 }609610 fn osc_dispatch(&mut self, _params: &[&[u8]], _bell_terminated: bool) {}611 fn esc_dispatch(&mut self, _intermediates: &[u8], _ignore: bool, _byte: u8) {}612613 fn csi_dispatch(614 &mut self,615 params: &vte::Params,616 _intermediates: &[u8],617 _ignore: bool,618 action: char,619 ) {620 use std::fmt::Write;621 if action != 'm' {622 623 return;624 }625 self.output.push_str("\x1b[");626 for (i, par) in params.iter().enumerate() {627 if i != 0 {628 let _ = write!(self.output, ";");629 }630 for (i, sub) in par.iter().enumerate() {631 if i != 0 {632 let _ = write!(self.output, ":");633 }634 let _ = write!(self.output, "{sub}");635 }636 }637 self.output.push(action);638 }639}640fn ansi_filter(i: &str) -> String {641 let mut out = AnsiFiltered {642 output: String::new(),643 };644 let mut parser = Parser::new();645646 647 for chunk in i.as_bytes().chunks(50) {648 parser.advance(&mut out, chunk);649 }650651 out.output652}653654#[derive(Debug)]655pub struct StackFrame {656 pub msg: String,657 pub pos: String,658}659660#[derive(Debug)]661pub struct ErrorInfoBuilder {662 level: Level,663 msg: String,664 pub stack_frames: Vec<StackFrame>,665}666fn new_error_info(lvl: u32, v: &[u8]) -> Box<ErrorInfoBuilder> {667 let verbosity = Verbosity::from_int(lvl);668 let level: Level = verbosity.into();669 let v = String::from_utf8_lossy(v);670 Box::new(ErrorInfoBuilder {671 level,672 msg: v.to_string(),673 stack_frames: Vec::new(),674 })675}676impl ErrorInfoBuilder {677 fn push_stack_frame(&mut self, v: &[u8], pos: &[u8]) {678 let v = String::from_utf8_lossy(v);679 let pos = String::from_utf8_lossy(pos);680 self.stack_frames.push(StackFrame {681 msg: v.to_string(),682 pos: pos.to_string(),683 });684 }685 fn emit_error_info(&mut self) {686 error!("{}", self.msg);687 for frame in &self.stack_frames {688 error!(" {} at {}", frame.msg, frame.pos)689 }690 }691}692693#[cxx::bridge]694pub mod nix_logging_cxx {695 extern "Rust" {696 type StartActivityBuilder;697 fn new_start_activity(activity_id: u64, lvl: u32, typ: u32) -> Box<StartActivityBuilder>;698 fn add_int_field(&mut self, i: i32);699 fn add_string_field(&mut self, v: &[u8]);700 fn emit(&mut self, parent: u64, s: &str);701 fn emit_result(&mut self, ty: u32);702 }703 extern "Rust" {704 type ErrorInfoBuilder;705 fn new_error_info(lvl: u32, v: &[u8]) -> Box<ErrorInfoBuilder>;706 fn push_stack_frame(&mut self, v: &[u8], pos: &[u8]);707 fn emit_error_info(&mut self);708 }709 extern "Rust" {710 fn emit_warn(v: &str);711 fn emit_stop(id: u64);712 fn emit_log(lvl: u32, v: &[u8]);713 }714 unsafe extern "C++" {715 include!("nix-eval/src/logging.hh");716717 type nix_c_context = crate::nix_raw::c_context;718719 fn apply_tracing_logger();720 unsafe fn extract_error_info(ctx: *const nix_c_context) -> Box<ErrorInfoBuilder>;721 }722}723724unsafe impl ExternType for crate::nix_raw::c_context {725 type Id = cxx::type_id!("nix_c_context");726727 type Kind = cxx::kind::Opaque;728}
--- 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"))],
};