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}
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"))],
};