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);
1use std::{2 collections::{BTreeMap, BTreeSet, HashSet},3 future::Future,4 io::Write,5 ops::Deref,6 path::PathBuf,7 pin::Pin,8 str::FromStr,9 sync::{Arc, OnceLock},10};1112use anyhow::{Context, Result, anyhow, bail, ensure};13use camino::{Utf8Path, Utf8PathBuf};14use chrono::{DateTime, Utc};15use fleet_shared::SecretData;16use nix_eval::{Store, Value, eval_store, nix_go, nix_go_json, util::assert_warn};17use remowt_client::{AgentBundle, Remowt};18use remowt_endpoints::fs::FsClient;19use remowt_fleet::NixClient;20use remowt_link_shared::{Address, BifConfig};21use remowt_ui_prompt::auto::AutoPrompter;22use remowt_ui_prompt::bifrost::PromptEndpoints;23use remowt_ui_prompt::{PrependSourcePrompter, Source};24use tempfile::NamedTempFile;25use tokio::{process::Command, sync::OnceCell, task::spawn_blocking};26use tracing::{info, instrument, warn};2728use crate::fleetdata::{29 FleetData, FleetSecretData, FleetSecretDistribution, FleetSecretPart, SecretOwner,30};3132pub struct FleetConfigInternals {33 pub prefer_identities: BTreeSet<SecretOwner>,34 pub now: DateTime<Utc>,3536 37 pub directory: PathBuf,38 39 pub local_system: String,40 pub data: Arc<FleetData>,41 42 pub config_field: Value,43 44 pub flake_outputs: Value,45 46 pub localhost: String,4748 49 pub default_pkgs: Value,50 51 pub nixpkgs: Value,5253 pub local_host: OnceLock<Arc<ConfigHost>>,54}555657#[derive(Clone)]58pub struct Config(pub Arc<FleetConfigInternals>);5960impl Deref for Config {61 type Target = FleetConfigInternals;6263 fn deref(&self) -> &Self::Target {64 &self.065 }66}6768#[derive(Clone, PartialEq, Copy, Debug)]69pub enum DeployKind {70 71 UpgradeToFleet,72 73 Fleet,74 75 76 NixosInstall,77 78 79 80 NixosLustrate,81}8283impl FromStr for DeployKind {84 type Err = anyhow::Error;85 fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {86 match s {87 "upgrade-to-fleet" => Ok(Self::UpgradeToFleet),88 "fleet" => Ok(Self::Fleet),89 "nixos-install" => Ok(Self::NixosInstall),90 "nixos-lustrate" => Ok(Self::NixosLustrate),91 v => bail!(92 "unknown deploy_kind: {v}; expected on of \"upgrade-to-fleet\", \"fleet\", \"nixos-install\", \"nixos-lustrate\""93 ),94 }95 }96}97pub struct ConfigHost {98 config: Config,99 pub name: String,100 groups: OnceLock<Vec<String>>,101102 103 deploy_kind: OnceCell<DeployKind>,104 session_destination: OnceLock<String>,105 legacy_ssh_store: OnceLock<bool>,106107 108 pub host_config: Option<Value>,109 pub nixos_config: OnceLock<Value>,110 pub nixos_unchecked_config: OnceLock<Value>,111 pub pkgs_override: Option<Value>,112113 binary_cache: OnceCell<Option<BinaryCache>>,114115 116 pub local: bool,117 pub remowt: OnceCell<Remowt>,118 nix_store: OnceCell<Arc<Store>>,119 nix_plugin: OnceCell<()>,120}121122const NIX_PLUGIN_ID: u16 = 2;123124fn agents_dir() -> Result<PathBuf> {125 std::env::var_os("REMOWT_AGENTS_DIR")126 .map(PathBuf::from)127 .or_else(|| option_env!("REMOWT_AGENTS_DIR").map(PathBuf::from))128 .ok_or_else(|| {129 anyhow!("no remowt-agents bundle; set REMOWT_AGENTS_DIR to a remowt-agents output")130 })131}132133fn agent_bundle() -> Result<AgentBundle> {134 AgentBundle::from_dir(agents_dir()?)135}136137#[derive(Clone)]138pub enum BinaryCache {139 Attic { name: String, pin: bool },140}141impl BinaryCache {142 async fn upload(&self, local: &Remowt, path: &Utf8Path, pin_name: Option<&str>) -> Result<()> {143 match self {144 BinaryCache::Attic { name, pin } => {145 let mut cmd = local.cmd("attic");146 cmd.arg("push").arg(name).arg(path.as_str());147 cmd.run()148 .await149 .context("failed to push path to attic (is attic-cli on PATH?)")?;150151 if *pin && let Some(pin_name) = pin_name {152 let mut cmd = local.cmd("attic");153 cmd.arg("pin")154 .arg("create")155 .arg(name)156 .arg(pin_name)157 .arg(path.as_str());158 cmd.run()159 .await160 .context("failed to pin path in attic (is your attic-cli compatible?)")?;161 }162 Ok(())163 }164 }165 }166}167168impl ConfigHost {169 pub fn set_session_destination(&self, dest: String) {170 self.session_destination171 .set(dest)172 .expect("session destination is already set")173 }174 pub fn set_deploy_kind(&self, kind: DeployKind) {175 self.deploy_kind176 .set(kind)177 .expect("deploy kind is already set");178 }179 pub fn set_legacy_ssh_store(&self, legacy: bool) {180 self.legacy_ssh_store181 .set(legacy)182 .expect("legacy ssh store is already set")183 }184 pub fn binary_cache(&self) -> Result<Option<BinaryCache>> {185 if let Some(v) = self.binary_cache.get() {186 return Ok(v.clone());187 }188 let cache = if let Some(host_config) = &self.host_config {189 let cache = nix_go!(host_config.cache);190 let enable: bool = nix_go_json!(cache.enable);191 if enable {192 let kind: String = nix_go_json!(cache.kind);193 let name: String = nix_go_json!(cache.name);194 match kind.as_str() {195 "attic" => {196 let pin: bool = nix_go_json!(cache.pin);197 Some(BinaryCache::Attic { name, pin })198 }199 v => bail!("unknown binary cache kind: {v}"),200 }201 } else {202 None203 }204 } else {205 None206 };207 let _ = self.binary_cache.set(cache.clone());208 Ok(cache)209 }210 pub async fn deploy_kind(&self) -> Result<DeployKind> {211 self.deploy_kind.get_or_try_init(|| async {212 let remowt = self.remowt().await?;213 let fs = remowt.endpoints::<FsClient<_>>();214 let is_fleet_managed = match fs.file_exists(Utf8PathBuf::from("/etc/FLEET_HOST")).await {215 Ok(v) => v,216 Err(e) => {217 bail!("failed to query remote system kind: {e}");218 }219 };220 if !is_fleet_managed {221 bail!(222 "{}",223 indoc::indoc! {"224 host is not marked as managed by fleet225 if you're not trying to lustrate/install system from scratch,226 you should either227 1. manually create /etc/FLEET_HOST file on the target host,228 2. use ?deploy_kind=fleet host argument if you're upgrading from older version of fleet229 3. use ?deploy_kind=upgrade_to_fleet if you're upgrading from plain nixos to fleet-managed nixos230 for installation use ?deploy_kind=nixos_install / ?deploy_kind=nixos_lustrate 231 "}232 );233 }234 Ok(DeployKind::Fleet)235 }).await.copied()236 }237 async fn new_remowt(&self) -> Result<Remowt> {238 let bundle = agent_bundle()?;239 let conn = if self.local {240 Remowt::connect_local(&bundle, "remowt-fleet".to_owned())241 .await242 .context("starting local remowt agent")?243 } else {244 let dest = self245 .session_destination246 .get()247 .cloned()248 .unwrap_or_else(|| self.name.clone());249 Remowt::connect(&dest, &bundle, "remowt-fleet".to_owned())250 .await251 .map_err(|e| anyhow!("remowt error while connecting to {}: {e:#?}", self.name))?252 };253 PromptEndpoints(PrependSourcePrompter {254 prompter: AutoPrompter::new().await,255 source: if self.local {256 vec![]257 } else {258 vec![Source(std::borrow::Cow::Owned(format!(259 "ssh host: {}",260 self.name261 )))]262 },263 description: "".to_owned(),264 })265 .register_endpoints(&mut conn.rpc());266 Ok(conn)267 }268269 270 pub async fn remowt(&self) -> Result<Remowt> {271 self.remowt272 .get_or_try_init(|| self.new_remowt())273 .await274 .cloned()275 }276277 pub async fn nix_client(&self) -> Result<NixClient<BifConfig>> {278 let remowt = self.remowt().await?;279 let plugin_id = self.ensure_nix_plugin().await?;280 Ok(remowt.plugin_endpoints(plugin_id))281 }282 pub fn ensure_nix_plugin(&self) -> Pin<Box<dyn Future<Output = Result<u16>> + Send + '_>> {283 Box::pin(async {284 self.nix_plugin285 .get_or_try_init(|| async {286 let pkgs = self.pkgs()?;287 let name = "remowt-plugin-fleet";288 let plugin = nix_go!(pkgs[{ name }]);289 let built = plugin290 .build("out")291 .context("failed to build the fleet nix plugin")?;292 let pin = format!("{}-remowt-plugin-fleet", self.gc_root_prefix());293 let copied = self294 295 .remote_derivation(&built, Some(&pin))296 .await297 .context("failed to copy the fleet nix plugin to the host store")?;298 let bin = copied.join("bin/remowt-plugin-fleet");299 self.remowt()300 .await?301 .run0_load_plugin_path(NIX_PLUGIN_ID, bin.as_str())302 .await303 .context("failed to load the fleet nix plugin")?;304 305 self.remowt()306 .await?307 .rpc()308 .wait_for_connection_to(Address::Plugin(NIX_PLUGIN_ID))309 .await310 .map_err(|_| anyhow!("failed to wait"))?;311 anyhow::Ok(())312 })313 .await?;314 Ok(NIX_PLUGIN_ID)315 })316 }317318 async fn new_nix_store(&self) -> Result<Store> {319 let conn = self.remowt().await?;320 let socket = match self.deploy_kind().await? {321 DeployKind::NixosInstall => {322 remowt_fleet::nix_store_socket(conn, "/mnt?require-sigs=false").await?323 }324 _ => remowt_fleet::nix_store_socket(conn, "auto").await?,325 };326 327 let uri = format!("unix://{}", socket.display());328 Store::open(&uri)329 }330 async fn nix_store(&self) -> Result<Arc<Store>> {331 self.nix_store332 .get_or_try_init(async || Ok(Arc::new(self.new_nix_store().await?)))333 .await334 .cloned()335 }336337 pub async fn decrypt(&self, data: SecretData) -> Result<Vec<u8>> {338 ensure!(data.encrypted, "secret is not encrypted");339 let remowt = self.remowt().await?;340 let mut cmd = remowt.cmd("fleet-install-secrets");341 cmd.arg("decrypt").eqarg("--secret", data.to_string());342 let encoded = cmd343 .sudo()344 .run_string()345 .await346 .context("failed to call remote host for decrypt")?;347 let data: SecretData = encoded.parse().map_err(|e| anyhow!("{e}"))?;348 ensure!(!data.encrypted, "secret came out encrypted");349 Ok(data.data)350 }351 pub async fn reencrypt_distribution(352 &self,353 data: &FleetSecretDistribution,354 targets: BTreeSet<SecretOwner>,355 now: DateTime<Utc>,356 ) -> Result<FleetSecretDistribution> {357 let mut parts = BTreeMap::new();358 for (part_name, part) in &data.secret.parts {359 parts.insert(360 part_name.clone(),361 if part.raw.encrypted {362 FleetSecretPart {363 raw: self.reencrypt(part.raw.clone(), targets.clone()).await?,364 }365 } else {366 part.clone()367 },368 );369 }370 let secret = FleetSecretData {371 created_at: data.secret.created_at,372 expires_at: data.secret.expires_at,373 generation_data: data.secret.generation_data.clone(),374 parts,375 };376 Ok(FleetSecretDistribution::new(targets, secret, now))377 }378 pub async fn reencrypt(379 &self,380 data: SecretData,381 targets: BTreeSet<SecretOwner>,382 ) -> Result<SecretData> {383 let remowt = self.remowt().await?;384 ensure!(data.encrypted, "secret is not encrypted");385 let mut cmd = remowt.cmd("fleet-install-secrets");386 cmd.arg("reencrypt").eqarg("--secret", data.to_string());387 for target in targets {388 let key = self.config.key(&target).await?;389 cmd.eqarg("--targets", key);390 }391 let encoded = cmd392 .sudo()393 .run_string()394 .await395 .context("failed to call remote host for decrypt")?;396 let data: SecretData = encoded.parse().map_err(|e| anyhow!("{e}"))?;397 ensure!(data.encrypted, "secret came out not encrypted");398 Ok(data)399 }400 401 #[instrument(skip(self))]402 pub async fn remote_derivation(403 &self,404 path: &Utf8Path,405 pin_name: Option<&str>,406 ) -> Result<Utf8PathBuf> {407 let path = path.to_owned();408 if self.local {409 410 return Ok(path);411 }412 let cache = self.binary_cache()?;413 let mut cache = if let Some(cache) = cache {414 let local = self.config.local_host().remowt().await?;415 let path = path.to_owned();416 let pin_name = pin_name.map(str::to_owned);417 Some(tokio::spawn(async move {418 if let Err(e) = cache.upload(&local, &path, pin_name.as_deref()).await {419 warn!("failed to upload closure {path} to cache: {e:?}");420 }421 }))422 } else {423 None424 };425 let sign: Pin<Box<dyn Future<Output = Result<()>> + Send>> = {426 let path = path.clone();427428 Box::pin(async move {429 let local = self.config.local_host();430 let plugin_id = local.ensure_nix_plugin().await?;431 let nix = local432 .remowt()433 .await?434 .plugin_endpoints::<remowt_fleet::NixClient<_>>(plugin_id);435 nix.sign_closure(path, Utf8PathBuf::from("/etc/nix/private-key"))436 .await437 .map_err(|e| anyhow!("{e:?}"))?438 .map_err(|e| anyhow!("{e}"))?;439 Ok(())440 })441 };442 if let Err(e) = sign.await {443 warn!("failed to sign store paths: {e}");444 }445 let store = self.nix_store().await?;446 let eval_store = eval_store();447 {448 let path = path.clone();449 spawn_blocking(move || eval_store.copy_to(&store, path.as_ref()))450 .await451 .expect("copy_to panicked")452 .context("copying closure to remote store")?;453 }454455 if let Some(cache) = cache.take() {456 let _ = cache.await;457 }458459 Ok(path)460 }461462 463 464 465 466 #[instrument(skip(self))]467 pub async fn remote_derivation_attic(468 &self,469 path: &Utf8Path,470 cache_name: &str,471 ) -> Result<Utf8PathBuf> {472 let path = path.to_owned();473 if self.local {474 475 return Ok(path);476 }477478 info!("uploading {path} closure to attic cache {cache_name}");479 let status = Command::new("attic")480 .arg("push")481 .arg(cache_name)482 .arg(path.as_str())483 .status()484 .await485 .context("failed to spawn attic-cli, is it on PATH?")?;486 ensure!(status.success(), "attic push failed: {status}");487488 info!("substituting {path} on the remote host");489 let nix = self.nix_client().await?;490 let substituted = nix491 .substitute(vec![path.clone()])492 .await493 .map_err(|e| anyhow!("{e:?}"))?494 .map_err(|e| anyhow!("{e}"))?;495 ensure!(496 substituted.contains(&path),497 "remote host failed to substitute {path} from attic cache {cache_name}"498 );499 Ok(path)500 }501}502503struct HostSecretDefinition(Value);504505impl ConfigHost {506 pub fn gc_root_prefix(&self) -> String {507 format!("{}-{}", self.config.gc_root_prefix(), self.name)508 }509510 511 512 pub fn tags(&self) -> Result<Vec<String>> {513 if let Some(v) = self.groups.get() {514 return Ok(v.clone());515 }516 let Some(host_config) = &self.host_config else {517 return Ok(vec![]);518 };519 let tags: Vec<String> = nix_go_json!(host_config.tags);520521 let _ = self.groups.set(tags.clone());522523 Ok(tags)524 }525 pub fn nixos_config(&self) -> Result<Value> {526 if let Some(v) = self.nixos_config.get() {527 return Ok(v.clone());528 }529 let Some(host_config) = &self.host_config else {530 bail!("local host has no nixos_config");531 };532 let nixos_config = nix_go!(host_config.nixos.config);533 assert_warn("nixos config evaluation", &nixos_config)?;534535 let _ = self.nixos_config.set(nixos_config.clone());536537 Ok(nixos_config)538 }539 pub fn nixos_unchecked_config(&self) -> Result<Value> {540 if let Some(v) = self.nixos_unchecked_config.get() {541 return Ok(v.clone());542 }543 let Some(host_config) = &self.host_config else {544 bail!("local host has no nixos_config");545 };546 let nixos_config = nix_go!(host_config.nixos_unchecked.config);547548 let _ = self.nixos_unchecked_config.set(nixos_config.clone());549550 Ok(nixos_config)551 }552553 pub fn list_defined_secrets(&self) -> Result<Vec<String>> {554 let nixos = self.nixos_unchecked_config()?;555 let secrets = nix_go!(nixos.secrets);556 secrets.list_fields()557 }558559 560 pub fn pkgs(&self) -> Result<Value> {561 if let Some(value) = &self.pkgs_override {562 return Ok(value.clone());563 }564 let Some(host_config) = &self.host_config else {565 bail!("local host has no host_config");566 };567 568 Ok(nix_go!(host_config.nixos.options._module.args.value.pkgs))569 }570}571572#[derive(Clone)]573pub struct SharedSecretDefinition(Value);574impl SharedSecretDefinition {575 pub fn expected_owners(&self) -> Result<BTreeSet<SecretOwner>> {576 let secret = &self.0;577 Ok(nix_go_json!(secret.expectedOwners))578 }579 pub fn allow_different(&self) -> Result<bool> {580 let secret = &self.0;581 Ok(nix_go_json!(secret.allowDifferent))582 }583 pub fn regenerate_on_owner_added(&self) -> Result<bool> {584 let secret = &self.0;585 Ok(nix_go_json!(secret.regenerateOnOwnerAdded))586 }587 pub fn regenerate_on_owner_removed(&self) -> Result<bool> {588 let secret = &self.0;589 Ok(nix_go_json!(secret.regenerateOnOwnerRemoved))590 }591 pub fn generator(&self) -> Result<Value> {592 let secret = &self.0;593 Ok(nix_go!(secret.generator))594 }595}596597impl Config {598 pub fn tagged_hostnames(&self, tag: &str) -> Result<Vec<String>> {599 let config = &self.config_field;600 let tagged: Vec<String> = nix_go_json!(config.taggedWith[{ tag }]);601 Ok(tagged)602 }603 pub fn expand_owner_set(&self, owners: Vec<String>) -> Result<BTreeSet<String>> {604 let mut out = BTreeSet::new();605 for owner in owners {606 if let Some(tag) = owner.strip_prefix('@') {607 let hosts = self.tagged_hostnames(tag)?;608 out.extend(hosts);609 } else {610 out.insert(owner);611 }612 }613 Ok(out)614 }615 pub fn local_host(&self) -> Arc<ConfigHost> {616 self.local_host617 .get_or_init(|| {618 Arc::new(ConfigHost {619 config: self.clone(),620 name: "<virtual localhost>".to_owned(),621 host_config: None,622 nixos_config: OnceLock::new(),623 nixos_unchecked_config: OnceLock::new(),624 groups: {625 let cell = OnceLock::new();626 let _ = cell.set(vec![]);627 cell628 },629 pkgs_override: Some(self.default_pkgs.clone()),630 binary_cache: OnceCell::new(),631632 local: true,633 remowt: OnceCell::new(),634 nix_store: OnceCell::new(),635 nix_plugin: OnceCell::new(),636 deploy_kind: OnceCell::new(),637 session_destination: OnceLock::new(),638 legacy_ssh_store: OnceLock::new(),639 })640 })641 .clone()642 }643644 pub fn preferred_hosts(645 &self,646 filter: impl Fn(&str) -> bool,647 ) -> Result<impl Iterator<Item = Result<ConfigHost>>> {648 let prefer = self649 .prefer_identities650 .iter()651 .filter_map(|v| v.as_host())652 .collect::<HashSet<_>>();653 let config = &self.config_field;654 let mut names = nix_go!(config.hosts).list_fields()?;655 names.retain(|s| filter(s));656 names.sort_by_key(|h| prefer.contains(h.as_str()));657658 Ok(names.into_iter().map(|h| self.host(&h)))659 }660661 pub fn host(&self, name: &str) -> Result<ConfigHost> {662 let config = &self.config_field;663 let host_config = nix_go!(config.hosts[{ name }]);664665 Ok(ConfigHost {666 config: self.clone(),667 name: name.to_owned(),668 host_config: Some(host_config),669 nixos_config: OnceLock::new(),670 nixos_unchecked_config: OnceLock::new(),671 groups: OnceLock::new(),672 pkgs_override: None,673 binary_cache: OnceCell::new(),674675 676 local: self.localhost == name,677 remowt: OnceCell::new(),678 nix_store: OnceCell::new(),679 nix_plugin: OnceCell::new(),680 deploy_kind: OnceCell::new(),681 session_destination: OnceLock::new(),682 legacy_ssh_store: OnceLock::new(),683 })684 }685 pub fn list_hosts(&self) -> Result<Vec<ConfigHost>> {686 let config = &self.config_field;687 let names = nix_go!(config.hosts).list_fields()?;688 let mut out = vec![];689 for name in names {690 out.push(self.host(&name)?);691 }692 Ok(out)693 }694 695 pub fn system_config(&self, host: &str) -> Result<Value> {696 let fleet_field = &self.config_field;697 Ok(nix_go!(fleet_field.hosts[{ host }].nixos.config))698 }699700 pub fn secret_definition(&self, secret: &str) -> Result<Option<SharedSecretDefinition>> {701 let config = &self.config_field;702 let shared_secrets = nix_go!(config.secrets);703 if !shared_secrets.has_field(secret)? {704 return Ok(None);705 }706 Ok(Some(SharedSecretDefinition(nix_go!(707 shared_secrets[secret]708 ))))709 }710711 pub fn save(&self) -> Result<()> {712 let mut tempfile = NamedTempFile::new_in(self.directory.clone()).context("failed to create updated version of fleet.nix in the same directory as original.\nDo you have write access to it? Access only to the fleet.nix won't be enough, the directory is used for atomic overwrite operation.\nIt is not recommended to use fleet by root anyway, move fleet project to your home directory.")?;713 let data = nixlike::serialize(&*self.data)?;714 tempfile.write_all(715 format!(716 "# This file contains fleet state and shouldn't be edited by hand\n\n{data}\n\n# vim: ts=2 et nowrap\n"717 )718 .as_bytes(),719 )?;720 let mut fleet_data_path = self.directory.clone();721 fleet_data_path.push("fleet.nix");722 tempfile.persist(fleet_data_path)?;723 Ok(())724 }725}
--- 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,
--- a/crates/nix-eval/src/logging.rs
+++ b/crates/nix-eval/src/logging.rs
@@ -109,6 +109,12 @@
(ActivityType::CopyPaths, []) => {
debug_span!(target: "nix::copy-paths", "copying paths")
}
+ // Umbrella progress activity. Nix emits it at lvlError ("always show"),
+ // which would otherwise become a noisy ERROR `action` span — give it an
+ // explicit debug level like the other container activities.
+ (ActivityType::Builds, []) => {
+ debug_span!(target: "nix::builds", "building")
+ }
(ActivityType::Unknown, [])
if s.starts_with("copying \"") && s.ends_with("\" to the store") =>
{
@@ -365,6 +371,16 @@
BuildGraphGuard { paths }
}
+/// Drop the lazily-created `building` span for a derivation. Used when a build
+/// fails: its dependents will never build, so their speculatively-created spans
+/// (made when a dependency started building) would otherwise linger until the
+/// whole build graph is torn down.
+pub fn remove_drv_span(drv_path: &Utf8Path) {
+ if let Some(entry) = DRV_GRAPH.lock().expect("not poisoned").get_mut(drv_path) {
+ entry.span = None;
+ }
+}
+
fn ensure_drv_span(drv_path: &Utf8Path) -> Option<Span> {
let mut drv_graph = DRV_GRAPH.lock().expect("not poisoned");
@@ -524,7 +540,7 @@
// ResultType::FileLinked => todo!(),
(ResultType::BuildLogLine, [Str(s)]) => {
let s = ansi_filter(s);
- info!("{s}");
+ info!(target: "nix::build", "{s}");
}
// ResultType::UntrustedPath => todo!(),
// ResultType::CorruptedPath => todo!(),
--- 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"))],
};