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());
+ }
+}
1#![recursion_limit = "512"]23pub(crate) mod cmds;45use std::{process::ExitCode, sync::Arc};67use anyhow::{Context as _, Result, bail};8use camino::Utf8PathBuf;9use clap::{CommandFactory, Parser};10use cmds::{11 build_systems::{BuildSystems, Deploy},12 complete::Complete,13 info::Info,14 rollback::RollbackSingle,15 secrets::Secret,16 tf::Tf,17};18use fleet_base::{host::Config, opts::FleetOpts};19use futures::{TryStreamExt, stream::FuturesUnordered};20#[cfg(feature = "indicatif")]21use human_repr::HumanCount;22#[cfg(feature = "indicatif")]23use indicatif::{ProgressState, ProgressStyle};24use nix_eval::{25 eval_store, gc_register_my_thread, gc_unregister_my_thread, init_libraries, init_tokio_for_nix,26};27use opentelemetry::trace::TracerProvider;28use opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge;29use opentelemetry_exporter_env::{30 OtlpBaseSettings, OtlpLogsSettings, OtlpTracesSettings, ResolvedOtlpSettings,31};32use opentelemetry_sdk::{logs::SdkLoggerProvider, trace::SdkTracerProvider};33use tokio::task::spawn_blocking;34use tracing::{Instrument, error, info, info_span};35#[cfg(feature = "indicatif")]36use tracing_indicatif::IndicatifLayer;37use tracing_subscriber::{EnvFilter, prelude::*};3839#[derive(Parser)]40struct Prefetch {}41impl Prefetch {42 async fn run(&self, config: &Config) -> Result<()> {43 let mut prefetch_dir = config.directory.to_path_buf();44 prefetch_dir.push("prefetch");45 if !prefetch_dir.is_dir() {46 info!("nothing to prefetch: no prefetch directory");47 return Ok(());48 }49 let tasks = FuturesUnordered::new();50 for entry in std::fs::read_dir(&prefetch_dir)? {51 let entry = entry?;52 if !entry.metadata()?.is_file() {53 bail!("only files should exist in prefetch directory");54 }55 let name = entry.file_name().to_string_lossy().into_owned();56 let path =57 Utf8PathBuf::try_from(entry.path()).context("prefetch path should be utf8")?;58 let span = info_span!("prefetching", name = %name);59 tasks.push(async move {60 let store = eval_store();61 let added = spawn_blocking(move || store.add_file(&name, &path))62 .instrument(span.clone())63 .await??;64 let _g = span.enter();65 info!("{} -> {}", added.hash, added.store_path);66 anyhow::Ok(())67 });68 }69 tasks.try_collect::<Vec<()>>().await?;70 Ok(())71 }72}7374#[derive(Parser)]75enum Opts {76 77 BuildSystems(BuildSystems),78 79 Deploy(Deploy),80 81 RollbackSingle(RollbackSingle),82 83 #[clap(subcommand)]84 Secret(Secret),85 86 Prefetch(Prefetch),87 88 Info(Info),89 90 #[clap(hide(true))]91 Complete(Complete),92 93 Tf(Tf),94}9596#[derive(Parser)]97#[clap(version, author)]98struct RootOpts {99 #[clap(flatten)]100 fleet_opts: FleetOpts,101 #[clap(subcommand)]102 command: Opts,103 #[clap(long, next_help_heading = "Telemetry", env = "OTEL_FLEET")]104 otel: bool,105 #[clap(flatten)]106 otlp_base: OtlpBaseSettings,107 #[clap(flatten)]108 otel_logs: OtlpLogsSettings,109 #[clap(flatten)]110 otel_traces: OtlpTracesSettings,111}112113async fn run_command(config: &Config, opts: FleetOpts, command: Opts) -> Result<()> {114 match command {115 Opts::BuildSystems(c) => c.run(config, &opts).await?,116 Opts::Deploy(d) => d.run(config, &opts).await?,117 Opts::RollbackSingle(r) => r.run(config, &opts).await?,118 Opts::Secret(s) => s.run(config, &opts).await?,119 Opts::Info(i) => i.run(config).await?,120 Opts::Prefetch(p) => p.run(config).await?,121 Opts::Tf(t) => t.run(config).await?,122 123 Opts::Complete(c) => spawn_blocking(move || c.run(RootOpts::command())).await?,124 };125 Ok(())126}127128fn setup_logging(opts: &RootOpts) -> Result<()> {129 #[cfg(feature = "indicatif")]130 let indicatif_layer = {131 use std::time::Duration;132133 IndicatifLayer::new().with_max_progress_bars(10, Some(ProgressStyle::default_spinner()))134 .with_progress_style(135 ProgressStyle::with_template(136 "{color_start}{span_child_prefix} {span_name}{{{span_fields}}}{color_end} {wide_msg} {color_start}{download_progress} {elapsed}{color_end}",137 )138 .unwrap()139 .with_key("download_progress", |state: &ProgressState, writer: &mut dyn std::fmt::Write| {140 let Some(len) = state.len() else {141 return;142 };143 let pos = state.pos();144 if pos > len {145 let _ = write!(writer, "{}", pos.human_count_bare());146 } else {147 let _ = write!(writer, "{} / {}", pos.human_count_bare(), len.human_count_bare());148 }149 })150 .with_key(151 "color_start",152 |state: &ProgressState, writer: &mut dyn std::fmt::Write| {153 let elapsed = state.elapsed();154155 if elapsed > Duration::from_secs(60) {156 157 let _ = write!(writer, "\x1b[{}m", 1 + 30);158 } else if elapsed > Duration::from_secs(30) {159 160 let _ = write!(writer, "\x1b[{}m", 3 + 30);161 }162 },163 )164 .with_key(165 "color_end",166 |state: &ProgressState, writer: &mut dyn std::fmt::Write| {167 if state.elapsed() > Duration::from_secs(30) {168 let _ = write!(writer, "\x1b[0m");169 }170 },171 ),172 )173 };174175 let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));176177 let reg = tracing_subscriber::registry().with({178 let sub = tracing_subscriber::fmt::layer().without_time();179 #[cfg(feature = "indicatif")]180 let sub = sub.with_writer(indicatif_layer.get_stderr_writer());181 sub.with_filter(filter) 182 });183184 #[cfg(feature = "indicatif")]185 let reg = reg.with(indicatif_layer);186187 if opts.otel {188 let traces = ResolvedOtlpSettings::traces(&opts.otlp_base, &opts.otel_traces)?;189 let span_exporter = traces.span_exporter()?;190 let logs = ResolvedOtlpSettings::logs(&opts.otlp_base, &opts.otel_logs)?;191 let log_exporter = logs.log_exporter()?;192193 let span_provider = SdkTracerProvider::builder()194 .with_batch_exporter(span_exporter)195 .build();196 let log_provider = SdkLoggerProvider::builder()197 .with_batch_exporter(log_exporter)198 .build();199200 let logger = OpenTelemetryTracingBridge::new(&log_provider);201 let tracer = span_provider.tracer("fleet");202203 let reg = reg204 .with(tracing_opentelemetry::layer().with_tracer(tracer))205 .with(logger);206207 reg.init();208 } else {209 reg.init();210 };211212 Ok(())213}214215fn main() -> ExitCode {216 let opts = RootOpts::parse();217 if let Opts::Complete(c) = &opts.command {218 c.run(RootOpts::command());219 return ExitCode::SUCCESS;220 }221222 if let Err(e) = setup_logging(&opts) {223 eprintln!("{e:#}");224 return ExitCode::FAILURE;225 }226227 init_libraries();228229 let runtime = tokio::runtime::Builder::new_multi_thread()230 .enable_all()231 .on_thread_start(|| {232 gc_register_my_thread();233 })234 .on_thread_stop(|| {235 gc_unregister_my_thread();236 })237 .build()238 .expect("failed to build runtime");239 let runtime = Arc::new(runtime);240241 init_tokio_for_nix(runtime.clone());242243 runtime.block_on(async {244 tokio::task::spawn(async move {245 if let Err(e) = main_real(opts).await {246 error!("{e:#}");247 ExitCode::FAILURE248 } else {249 ExitCode::SUCCESS250 }251 })252 .await253 .expect("primary task panicked")254 })255 256}257258async fn main_real(opts: RootOpts) -> Result<()> {259 let config = opts.fleet_opts.build(matches!(260 opts.command,261 Opts::Deploy(_) | Opts::BuildSystems(_)262 ))?;263264 match run_command(&config, opts.fleet_opts, opts.command).await {265 Ok(()) => {266 config.save()?;267 Ok(())268 }269 Err(e) => {270 let _ = config.save();271 Err(e)272 }273 }274}275276#[cfg(test)]277mod tests {278 use super::*;279280 #[test]281 fn verify_command() {282 use clap::CommandFactory;283 RootOpts::command().debug_assert();284 }285}
1#![recursion_limit = "512"]23pub(crate) mod cmds;4mod log_tree;56use std::{process::ExitCode, sync::Arc};78use anyhow::{Context as _, Result, bail};9use camino::Utf8PathBuf;10use clap::{CommandFactory, Parser};11use cmds::{12 build_systems::{BuildSystems, Deploy},13 complete::Complete,14 info::Info,15 rollback::RollbackSingle,16 secrets::Secret,17 tf::Tf,18};19use fleet_base::{host::Config, opts::FleetOpts};20use futures::{TryStreamExt, stream::FuturesUnordered};21#[cfg(feature = "indicatif")]22use human_repr::HumanCount;23#[cfg(feature = "indicatif")]24use indicatif::{ProgressState, ProgressStyle};25use log_tree::TreeLayer;26use nix_eval::{27 eval_store, gc_register_my_thread, gc_unregister_my_thread, init_libraries, init_tokio_for_nix,28};29use opentelemetry::trace::TracerProvider;30use opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge;31use opentelemetry_exporter_env::{32 OtlpBaseSettings, OtlpLogsSettings, OtlpTracesSettings, ResolvedOtlpSettings,33};34use opentelemetry_sdk::{logs::SdkLoggerProvider, trace::SdkTracerProvider};35use tokio::task::spawn_blocking;36use tracing::{Instrument, error, info, info_span};37#[cfg(feature = "indicatif")]38use tracing_indicatif::IndicatifLayer;39use tracing_subscriber::{EnvFilter, prelude::*};4041#[derive(Parser)]42struct Prefetch {}43impl Prefetch {44 async fn run(&self, config: &Config) -> Result<()> {45 let mut prefetch_dir = config.directory.to_path_buf();46 prefetch_dir.push("prefetch");47 if !prefetch_dir.is_dir() {48 info!("nothing to prefetch: no prefetch directory");49 return Ok(());50 }51 let tasks = FuturesUnordered::new();52 for entry in std::fs::read_dir(&prefetch_dir)? {53 let entry = entry?;54 if !entry.metadata()?.is_file() {55 bail!("only files should exist in prefetch directory");56 }57 let name = entry.file_name().to_string_lossy().into_owned();58 let path =59 Utf8PathBuf::try_from(entry.path()).context("prefetch path should be utf8")?;60 let span = info_span!("prefetching", name = %name);61 tasks.push(async move {62 let store = eval_store();63 let added = spawn_blocking(move || store.add_file(&name, &path))64 .instrument(span.clone())65 .await??;66 let _g = span.enter();67 info!("{} -> {}", added.hash, added.store_path);68 anyhow::Ok(())69 });70 }71 tasks.try_collect::<Vec<()>>().await?;72 Ok(())73 }74}7576#[derive(Parser)]77enum Opts {78 79 BuildSystems(BuildSystems),80 81 Deploy(Deploy),82 83 RollbackSingle(RollbackSingle),84 85 #[clap(subcommand)]86 Secret(Secret),87 88 Prefetch(Prefetch),89 90 Info(Info),91 92 #[clap(hide(true))]93 Complete(Complete),94 95 Tf(Tf),96}9798#[derive(Parser)]99#[clap(version, author)]100struct RootOpts {101 #[clap(flatten)]102 fleet_opts: FleetOpts,103 #[clap(subcommand)]104 command: Opts,105 #[clap(long, next_help_heading = "Telemetry", env = "OTEL_FLEET")]106 otel: bool,107 #[clap(flatten)]108 otlp_base: OtlpBaseSettings,109 #[clap(flatten)]110 otel_logs: OtlpLogsSettings,111 #[clap(flatten)]112 otel_traces: OtlpTracesSettings,113}114115async fn run_command(config: &Config, opts: FleetOpts, command: Opts) -> Result<()> {116 match command {117 Opts::BuildSystems(c) => c.run(config, &opts).await?,118 Opts::Deploy(d) => d.run(config, &opts).await?,119 Opts::RollbackSingle(r) => r.run(config, &opts).await?,120 Opts::Secret(s) => s.run(config, &opts).await?,121 Opts::Info(i) => i.run(config).await?,122 Opts::Prefetch(p) => p.run(config).await?,123 Opts::Tf(t) => t.run(config).await?,124 125 Opts::Complete(c) => spawn_blocking(move || c.run(RootOpts::command())).await?,126 };127 Ok(())128}129130fn setup_logging(opts: &RootOpts) -> Result<()> {131 #[cfg(feature = "indicatif")]132 let indicatif_layer = {133 use std::fmt;134 use std::time::Duration;135136 IndicatifLayer::new().with_max_progress_bars(10, Some(ProgressStyle::default_spinner()))137 .with_span_child_prefix_indent(" ")138 .with_span_child_prefix_symbol("")139 .with_progress_style(140 ProgressStyle::with_template(141 "{span_child_prefix:.magenta}{chevron:.magenta}{span_name:<18.blue.bold} {wide_msg} {span_fields:.dim} {color_start}{download_progress} {elapsed}{color_end}",142 )143 .unwrap()144 .with_key("chevron", |_: &ProgressState, writer: &mut dyn fmt::Write| {145 let _ = write!(writer, "> ");146 })147 .with_key("download_progress", |state: &ProgressState, writer: &mut dyn fmt::Write| {148 let Some(len) = state.len() else {149 return;150 };151 let pos = state.pos();152 if pos > len {153 let _ = write!(writer, "{}", pos.human_count_bare());154 } else {155 let _ = write!(writer, "{} / {}", pos.human_count_bare(), len.human_count_bare());156 }157 })158 .with_key(159 "color_start",160 |state: &ProgressState, writer: &mut dyn fmt::Write| {161 let elapsed = state.elapsed();162163 if elapsed > Duration::from_secs(60) {164 165 let _ = write!(writer, "\x1b[{}m", 1 + 30);166 } else if elapsed > Duration::from_secs(30) {167 168 let _ = write!(writer, "\x1b[{}m", 3 + 30);169 }170 },171 )172 .with_key(173 "color_end",174 |state: &ProgressState, writer: &mut dyn fmt::Write| {175 if state.elapsed() > Duration::from_secs(30) {176 let _ = write!(writer, "\x1b[0m");177 }178 },179 ),180 )181 };182183 let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));184185 let tree = {186 #[cfg(feature = "indicatif")]187 let writer = indicatif_layer.get_stderr_writer();188 #[cfg(not(feature = "indicatif"))]189 let writer = || std::io::stderr();190 TreeLayer::new(writer)191 };192193 let reg = tracing_subscriber::registry().with(tree.with_filter(filter));194195 #[cfg(feature = "indicatif")]196 let reg = reg.with(indicatif_layer);197198 if opts.otel {199 let traces = ResolvedOtlpSettings::traces(&opts.otlp_base, &opts.otel_traces)?;200 let span_exporter = traces.span_exporter()?;201 let logs = ResolvedOtlpSettings::logs(&opts.otlp_base, &opts.otel_logs)?;202 let log_exporter = logs.log_exporter()?;203204 let span_provider = SdkTracerProvider::builder()205 .with_batch_exporter(span_exporter)206 .build();207 let log_provider = SdkLoggerProvider::builder()208 .with_batch_exporter(log_exporter)209 .build();210211 let logger = OpenTelemetryTracingBridge::new(&log_provider);212 let tracer = span_provider.tracer("fleet");213214 reg.with(tracing_opentelemetry::layer().with_tracer(tracer))215 .with(logger)216 .init();217 } else {218 reg.init();219 }220221 Ok(())222}223224fn main() -> ExitCode {225 let opts = RootOpts::parse();226 if let Opts::Complete(c) = &opts.command {227 c.run(RootOpts::command());228 return ExitCode::SUCCESS;229 }230231 if let Err(e) = setup_logging(&opts) {232 eprintln!("{e:#}");233 return ExitCode::FAILURE;234 }235236 init_libraries();237238 let runtime = tokio::runtime::Builder::new_multi_thread()239 .enable_all()240 .on_thread_start(|| {241 gc_register_my_thread();242 })243 .on_thread_stop(|| {244 gc_unregister_my_thread();245 })246 .build()247 .expect("failed to build runtime");248 let runtime = Arc::new(runtime);249250 init_tokio_for_nix(runtime.clone());251252 runtime.block_on(async {253 tokio::task::spawn(async move {254 if let Err(e) = main_real(opts).await {255 error!("{e:#}");256 ExitCode::FAILURE257 } else {258 ExitCode::SUCCESS259 }260 })261 .await262 .expect("primary task panicked")263 })264}265266async fn main_real(opts: RootOpts) -> Result<()> {267 let config = opts.fleet_opts.build(matches!(268 opts.command,269 Opts::Deploy(_) | Opts::BuildSystems(_)270 ))?;271272 match run_command(&config, opts.fleet_opts, opts.command).await {273 Ok(()) => {274 config.save()?;275 Ok(())276 }277 Err(e) => {278 let _ = config.save();279 Err(e)280 }281 }282}283284#[cfg(test)]285mod tests {286 use super::*;287288 #[test]289 fn verify_command() {290 use clap::CommandFactory;291 RootOpts::command().debug_assert();292 }293}
--- 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,
--- 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"))],
};