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, drv::DrvGraph, 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::BifConfig;21use remowt_ui_prompt::auto::AutoPrompter;22use remowt_ui_prompt::bifrost::PromptEndpoints;23use remowt_ui_prompt::{PrependSourcePrompter, Source};24use tempfile::NamedTempFile;25use tokio::{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 pub host_config: Option<Value>,108 pub nixos_config: OnceLock<Value>,109 pub nixos_unchecked_config: OnceLock<Value>,110 pub pkgs_override: Option<Value>,111112 113 pub local: bool,114 pub remowt: OnceCell<Remowt>,115 nix_store: OnceCell<Arc<Store>>,116 nix_plugin: OnceCell<()>,117}118119const NIX_PLUGIN_ID: u16 = 2;120121fn agents_dir() -> Result<PathBuf> {122 std::env::var_os("REMOWT_AGENTS_DIR")123 .map(PathBuf::from)124 .or_else(|| option_env!("REMOWT_AGENTS_DIR").map(PathBuf::from))125 .ok_or_else(|| {126 anyhow!("no remowt-agents bundle; set REMOWT_AGENTS_DIR to a remowt-agents output")127 })128}129130fn agent_bundle() -> Result<AgentBundle> {131 AgentBundle::from_dir(agents_dir()?)132}133134enum RemoteDerivationMode {135 136 CopySigned,137 138 CopyPrivileged,139 140 Attic { name: String },141}142143impl ConfigHost {144 pub fn set_session_destination(&self, dest: String) {145 self.session_destination146 .set(dest)147 .expect("session destination is already set")148 }149 pub fn set_deploy_kind(&self, kind: DeployKind) {150 self.deploy_kind151 .set(kind)152 .expect("deploy kind is already set");153 }154 pub fn set_legacy_ssh_store(&self, legacy: bool) {155 self.legacy_ssh_store156 .set(legacy)157 .expect("legacy ssh store is already set")158 }159 pub async fn deploy_kind(&self) -> Result<DeployKind> {160 self.deploy_kind.get_or_try_init(|| async {161 let remowt = self.remowt().await?;162 let fs = remowt.endpoints::<FsClient<_>>();163 let is_fleet_managed = match fs.file_exists(Utf8PathBuf::from("/etc/FLEET_HOST")).await {164 Ok(v) => v,165 Err(e) => {166 bail!("failed to query remote system kind: {e}");167 }168 };169 if !is_fleet_managed {170 bail!(171 "{}",172 indoc::indoc! {"173 host is not marked as managed by fleet174 if you're not trying to lustrate/install system from scratch,175 you should either176 1. manually create /etc/FLEET_HOST file on the target host,177 2. use ?deploy_kind=fleet host argument if you're upgrading from older version of fleet178 3. use ?deploy_kind=upgrade_to_fleet if you're upgrading from plain nixos to fleet-managed nixos179 for installation use ?deploy_kind=nixos_install / ?deploy_kind=nixos_lustrate 180 "}181 );182 }183 Ok(DeployKind::Fleet)184 }).await.copied()185 }186 async fn new_remowt(&self) -> Result<Remowt> {187 let bundle = agent_bundle()?;188 let conn = if self.local {189 Remowt::connect_local(&bundle, "remowt-fleet".to_owned())190 .await191 .context("starting local remowt agent")?192 } else {193 let dest = self194 .session_destination195 .get()196 .cloned()197 .unwrap_or_else(|| self.name.clone());198 Remowt::connect(&dest, &bundle, "remowt-fleet".to_owned())199 .await200 .map_err(|e| anyhow!("remowt error while connecting to {}: {e:#?}", self.name))?201 };202 PromptEndpoints(PrependSourcePrompter {203 prompter: AutoPrompter::new().await,204 source: if self.local {205 vec![]206 } else {207 vec![Source(std::borrow::Cow::Owned(format!(208 "ssh host: {}",209 self.name210 )))]211 },212 description: "".to_owned(),213 })214 .register_endpoints(&mut conn.rpc());215 Ok(conn)216 }217218 219 pub async fn remowt(&self) -> Result<Remowt> {220 self.remowt221 .get_or_try_init(|| self.new_remowt())222 .await223 .cloned()224 }225226 pub async fn nix_client(&self) -> Result<NixClient<BifConfig>> {227 let remowt = self.remowt().await?;228 let plugin_id = self.ensure_nix_plugin().await?;229 Ok(remowt.plugin_endpoints(plugin_id))230 }231 pub fn ensure_nix_plugin(&self) -> Pin<Box<dyn Future<Output = Result<u16>> + Send + '_>> {232 Box::pin(async {233 self.nix_plugin234 .get_or_try_init(|| async {235 let pkgs = self.pkgs()?;236 let name = "remowt-plugin-fleet";237 let plugin = nix_go!(pkgs[{ name }]);238 let built = plugin239 .build("out")240 .context("failed to build the fleet nix plugin")?;241 let copied = self242 .remote_derivation(&built)243 .await244 .context("failed to copy the fleet nix plugin to the host store")?;245 let bin = copied.join("bin/remowt-plugin-fleet");246 self.remowt()247 .await?248 .run0_load_plugin_path(NIX_PLUGIN_ID, bin.as_str())249 .await250 .context("failed to load the fleet nix plugin")?;251 anyhow::Ok(())252 })253 .await?;254 Ok(NIX_PLUGIN_ID)255 })256 }257258 async fn new_nix_store(&self) -> Result<Store> {259 let conn = self.remowt().await?;260 let socket = match self.deploy_kind().await? {261 DeployKind::NixosInstall => {262 remowt_fleet::nix_store_socket(conn, "/mnt?require-sigs=false").await?263 }264 _ => remowt_fleet::nix_store_socket(conn, "auto").await?,265 };266 267 let uri = format!("unix://{}", socket.display());268 Store::open(&uri)269 }270 async fn nix_store(&self) -> Result<Arc<Store>> {271 self.nix_store272 .get_or_try_init(async || Ok(Arc::new(self.new_nix_store().await?)))273 .await274 .cloned()275 }276277 pub async fn decrypt(&self, data: SecretData) -> Result<Vec<u8>> {278 ensure!(data.encrypted, "secret is not encrypted");279 let remowt = self.remowt().await?;280 let mut cmd = remowt.cmd("fleet-install-secrets");281 cmd.arg("decrypt").eqarg("--secret", data.to_string());282 let encoded = cmd283 .sudo()284 .run_string()285 .await286 .context("failed to call remote host for decrypt")?;287 let data: SecretData = encoded.parse().map_err(|e| anyhow!("{e}"))?;288 ensure!(!data.encrypted, "secret came out encrypted");289 Ok(data.data)290 }291 pub async fn reencrypt_distribution(292 &self,293 data: &FleetSecretDistribution,294 targets: BTreeSet<SecretOwner>,295 now: DateTime<Utc>,296 ) -> Result<FleetSecretDistribution> {297 let mut parts = BTreeMap::new();298 for (part_name, part) in &data.secret.parts {299 parts.insert(300 part_name.clone(),301 if part.raw.encrypted {302 FleetSecretPart {303 raw: self.reencrypt(part.raw.clone(), targets.clone()).await?,304 }305 } else {306 part.clone()307 },308 );309 }310 let secret = FleetSecretData {311 created_at: data.secret.created_at,312 expires_at: data.secret.expires_at,313 generation_data: data.secret.generation_data.clone(),314 parts,315 };316 Ok(FleetSecretDistribution::new(targets, secret, now))317 }318 pub async fn reencrypt(319 &self,320 data: SecretData,321 targets: BTreeSet<SecretOwner>,322 ) -> Result<SecretData> {323 let remowt = self.remowt().await?;324 ensure!(data.encrypted, "secret is not encrypted");325 let mut cmd = remowt.cmd("fleet-install-secrets");326 cmd.arg("reencrypt").eqarg("--secret", data.to_string());327 for target in targets {328 let key = self.config.key(&target).await?;329 cmd.eqarg("--targets", key);330 }331 let encoded = cmd332 .sudo()333 .run_string()334 .await335 .context("failed to call remote host for decrypt")?;336 let data: SecretData = encoded.parse().map_err(|e| anyhow!("{e}"))?;337 ensure!(data.encrypted, "secret came out not encrypted");338 Ok(data)339 }340 341 #[instrument(skip(self))]342 pub async fn remote_derivation(&self, path: &Utf8Path) -> Result<Utf8PathBuf> {343 let path = path.to_owned();344 if self.local {345 346 return Ok(path);347 }348 let sign: Pin<Box<dyn Future<Output = Result<()>> + Send>> = {349 let path = path.clone();350 let graph = DrvGraph::resolve(&eval_store(), &path)351 .context("failed to resolve graph to be uploaded")?;352 info!("signing {} paths", graph.nodes.len());353354 Box::pin(async move {355 let local = self.config.local_host();356 let plugin_id = local.ensure_nix_plugin().await?;357 let nix = local358 .remowt()359 .await?360 .plugin_endpoints::<remowt_fleet::NixClient<_>>(plugin_id);361 nix.sign_closure(path, Utf8PathBuf::from("/etc/nix/private-key"))362 .await363 .map_err(|e| anyhow!("{e:?}"))?364 .map_err(|e| anyhow!("{e}"))?;365 Ok(())366 })367 };368 if let Err(e) = sign.await {369 warn!("failed to sign store paths: {e}");370 }371 let store = self.nix_store().await?;372 let eval_store = eval_store();373 {374 let path = path.clone();375 spawn_blocking(move || eval_store.copy_to(&store, path.as_ref()))376 .await377 .expect("copy_to panicked")378 .context("copying closure to remote store")?;379 }380 Ok(path)381 }382}383384struct HostSecretDefinition(Value);385386impl ConfigHost {387 388 389 pub fn tags(&self) -> Result<Vec<String>> {390 if let Some(v) = self.groups.get() {391 return Ok(v.clone());392 }393 let Some(host_config) = &self.host_config else {394 return Ok(vec![]);395 };396 let tags: Vec<String> = nix_go_json!(host_config.tags);397398 let _ = self.groups.set(tags.clone());399400 Ok(tags)401 }402 pub fn nixos_config(&self) -> Result<Value> {403 if let Some(v) = self.nixos_config.get() {404 return Ok(v.clone());405 }406 let Some(host_config) = &self.host_config else {407 bail!("local host has no nixos_config");408 };409 let nixos_config = nix_go!(host_config.nixos.config);410 assert_warn("nixos config evaluation", &nixos_config)?;411412 let _ = self.nixos_config.set(nixos_config.clone());413414 Ok(nixos_config)415 }416 pub fn nixos_unchecked_config(&self) -> Result<Value> {417 if let Some(v) = self.nixos_unchecked_config.get() {418 return Ok(v.clone());419 }420 let Some(host_config) = &self.host_config else {421 bail!("local host has no nixos_config");422 };423 let nixos_config = nix_go!(host_config.nixos_unchecked.config);424425 let _ = self.nixos_unchecked_config.set(nixos_config.clone());426427 Ok(nixos_config)428 }429430 pub fn list_defined_secrets(&self) -> Result<Vec<String>> {431 let nixos = self.nixos_unchecked_config()?;432 let secrets = nix_go!(nixos.secrets);433 secrets.list_fields()434 }435436 437 pub fn pkgs(&self) -> Result<Value> {438 if let Some(value) = &self.pkgs_override {439 return Ok(value.clone());440 }441 let Some(host_config) = &self.host_config else {442 bail!("local host has no host_config");443 };444 445 Ok(nix_go!(host_config.nixos.options._module.args.value.pkgs))446 }447}448449#[derive(Clone)]450pub struct SharedSecretDefinition(Value);451impl SharedSecretDefinition {452 pub fn expected_owners(&self) -> Result<BTreeSet<SecretOwner>> {453 let secret = &self.0;454 Ok(nix_go_json!(secret.expectedOwners))455 }456 pub fn allow_different(&self) -> Result<bool> {457 let secret = &self.0;458 Ok(nix_go_json!(secret.allowDifferent))459 }460 pub fn regenerate_on_owner_added(&self) -> Result<bool> {461 let secret = &self.0;462 Ok(nix_go_json!(secret.regenerateOnOwnerAdded))463 }464 pub fn regenerate_on_owner_removed(&self) -> Result<bool> {465 let secret = &self.0;466 Ok(nix_go_json!(secret.regenerateOnOwnerRemoved))467 }468 pub fn generator(&self) -> Result<Value> {469 let secret = &self.0;470 Ok(nix_go!(secret.generator))471 }472}473474impl Config {475 pub fn tagged_hostnames(&self, tag: &str) -> Result<Vec<String>> {476 let config = &self.config_field;477 let tagged: Vec<String> = nix_go_json!(config.taggedWith[{ tag }]);478 Ok(tagged)479 }480 pub fn expand_owner_set(&self, owners: Vec<String>) -> Result<BTreeSet<String>> {481 let mut out = BTreeSet::new();482 for owner in owners {483 if let Some(tag) = owner.strip_prefix('@') {484 let hosts = self.tagged_hostnames(tag)?;485 out.extend(hosts);486 } else {487 out.insert(owner);488 }489 }490 Ok(out)491 }492 pub fn local_host(&self) -> Arc<ConfigHost> {493 self.local_host494 .get_or_init(|| {495 Arc::new(ConfigHost {496 config: self.clone(),497 name: "<virtual localhost>".to_owned(),498 host_config: None,499 nixos_config: OnceLock::new(),500 nixos_unchecked_config: OnceLock::new(),501 groups: {502 let cell = OnceLock::new();503 let _ = cell.set(vec![]);504 cell505 },506 pkgs_override: Some(self.default_pkgs.clone()),507508 local: true,509 remowt: OnceCell::new(),510 nix_store: OnceCell::new(),511 nix_plugin: OnceCell::new(),512 deploy_kind: OnceCell::new(),513 session_destination: OnceLock::new(),514 legacy_ssh_store: OnceLock::new(),515 })516 })517 .clone()518 }519520 pub fn preferred_hosts(521 &self,522 filter: impl Fn(&str) -> bool,523 ) -> Result<impl Iterator<Item = Result<ConfigHost>>> {524 let prefer = self525 .prefer_identities526 .iter()527 .filter_map(|v| v.as_host())528 .collect::<HashSet<_>>();529 let config = &self.config_field;530 let mut names = nix_go!(config.hosts).list_fields()?;531 names.retain(|s| filter(s));532 names.sort_by_key(|h| prefer.contains(h.as_str()));533534 Ok(names.into_iter().map(|h| self.host(&h)))535 }536537 pub fn host(&self, name: &str) -> Result<ConfigHost> {538 let config = &self.config_field;539 let host_config = nix_go!(config.hosts[{ name }]);540541 Ok(ConfigHost {542 config: self.clone(),543 name: name.to_owned(),544 host_config: Some(host_config),545 nixos_config: OnceLock::new(),546 nixos_unchecked_config: OnceLock::new(),547 groups: OnceLock::new(),548 pkgs_override: None,549550 551 local: self.localhost == name,552 remowt: OnceCell::new(),553 nix_store: OnceCell::new(),554 nix_plugin: OnceCell::new(),555 deploy_kind: OnceCell::new(),556 session_destination: OnceLock::new(),557 legacy_ssh_store: OnceLock::new(),558 })559 }560 pub fn list_hosts(&self) -> Result<Vec<ConfigHost>> {561 let config = &self.config_field;562 let names = nix_go!(config.hosts).list_fields()?;563 let mut out = vec![];564 for name in names {565 out.push(self.host(&name)?);566 }567 Ok(out)568 }569 570 pub fn system_config(&self, host: &str) -> Result<Value> {571 let fleet_field = &self.config_field;572 Ok(nix_go!(fleet_field.hosts[{ host }].nixos.config))573 }574575 pub fn secret_definition(&self, secret: &str) -> Result<Option<SharedSecretDefinition>> {576 let config = &self.config_field;577 let shared_secrets = nix_go!(config.secrets);578 if !shared_secrets.has_field(secret)? {579 return Ok(None);580 }581 Ok(Some(SharedSecretDefinition(nix_go!(582 shared_secrets[secret]583 ))))584 }585586 pub fn save(&self) -> Result<()> {587 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.")?;588 let data = nixlike::serialize(&*self.data)?;589 tempfile.write_all(590 format!(591 "# This file contains fleet state and shouldn't be edited by hand\n\n{data}\n\n# vim: ts=2 et nowrap\n"592 )593 .as_bytes(),594 )?;595 let mut fleet_data_path = self.directory.clone();596 fleet_data_path.push("fleet.nix");597 tempfile.persist(fleet_data_path)?;598 Ok(())599 }600}
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"))],
};