git.delta.rocks / jrsonnet / refs/commits / 4340a04aa508

difftreelog

feat add progress bar

Yaroslav Bolyukin2023-10-09parent: #03c441d.patch.diff
in: trunk

10 files changed

modifiedCargo.lockdiffbeforeafterboth
before · Cargo.lock
273 packageslockfile v3
deletedREADME.mddiffbeforeafterboth
--- a/README.md
+++ /dev/null
@@ -1,8 +0,0 @@
-# fleet
-
-Early prototype stage
-
-## Advantages over existing configuration systems (NixOps/Morph)
-
-- Modules can configure multiple hosts at once (I.e for wireguard/kubernetes installation)
-- Secrets can be securely stored in Git (No one except target hosts can decrypt them)
modifiedcmds/fleet/Cargo.tomldiffbeforeafterboth
--- a/cmds/fleet/Cargo.toml
+++ b/cmds/fleet/Cargo.toml
@@ -27,8 +27,10 @@
 	"unicode",
 ] }
 tokio = { version = "1.14.0", features = ["full"] }
-tracing = "0.1.29"
-tracing-subscriber = { version = "0.3.3", features = ["fmt", "env-filter"] }
+tracing = "0.1"
+tracing-subscriber = { version = "0.3", features = ["fmt", "env-filter"] }
 tokio-util = { version = "0.7.0", features = ["codec"] }
 async-trait = "0.1.52"
 futures = "0.3.17"
+tracing-indicatif = "0.3.5"
+indicatif = "0.17.7"
modifiedcmds/fleet/src/cmds/secrets/mod.rsdiffbeforeafterboth
--- a/cmds/fleet/src/cmds/secrets/mod.rs
+++ b/cmds/fleet/src/cmds/secrets/mod.rs
@@ -11,7 +11,7 @@
 	path::PathBuf,
 };
 use tokio::fs::read_to_string;
-use tracing::{info, warn, error};
+use tracing::{error, info, warn};
 
 #[derive(Parser)]
 pub enum Secrets {
@@ -281,8 +281,8 @@
 					secret.owners.first()
 				};
 				let Some(identity_holder) = identity_holder else {
-                    bail!("no available holder found");
-                };
+					bail!("no available holder found");
+				};
 				let target_recipients = futures::stream::iter(&target_machines)
 					.then(|m| async { config.key(m).await })
 					.collect::<Vec<_>>()
@@ -365,7 +365,9 @@
 							data.owners = expected_owners;
 							config.replace_shared(name.to_owned(), data);
 						} else if let Some(generator) = config
-							.shared_config_attr::<Option<String>>(&format!("sharedSecrets.\"{name}\".generator"))
+							.shared_config_attr::<Option<String>>(&format!(
+								"sharedSecrets.\"{name}\".generator"
+							))
 							.await?
 						{
 							todo!("regenerate secret {name} with {generator}");
modifiedcmds/fleet/src/command.rsdiffbeforeafterboth
--- a/cmds/fleet/src/command.rs
+++ b/cmds/fleet/src/command.rs
@@ -1,7 +1,6 @@
-use std::{ffi::OsStr, process::Stdio, task::Poll};
+use std::{collections::HashMap, ffi::OsStr, process::Stdio, task::Poll};
 
 use anyhow::{Context, Result};
-use async_trait::async_trait;
 use futures::StreamExt;
 use serde::{
 	de::{DeserializeOwned, Visitor},
@@ -9,7 +8,8 @@
 };
 use tokio::{io::AsyncRead, process::Command, select};
 use tokio_util::codec::{BytesCodec, FramedRead, LinesCodec};
-use tracing::{info, warn};
+use tracing::{info, info_span, warn, Span};
+use tracing_indicatif::span_ext::IndicatifSpanExt;
 
 fn escape_bash(input: &str, out: &mut String) {
 	const TO_ESCAPE: &str = "$ !\"#&'()*,;<>?[\\]^`{|}";
@@ -126,26 +126,15 @@
 
 	pub async fn run(self) -> Result<()> {
 		let str = self.clone().into_string();
-		info!("running {str}");
-		let mut cmd = self.into_command();
-		cmd.inherit_stdio();
-		let out = cmd.spawn()?.wait_with_output().await?;
-		if !out.status.success() {
-			anyhow::bail!("command '{}' failed with status {}", str, out.status);
-		}
+		let cmd = self.into_command();
+		run_nix_inner(str, cmd, &mut PlainHandler).await?;
 		Ok(())
 	}
 	pub async fn run_string(self) -> Result<String> {
 		let str = self.clone().into_string();
-		info!("running {str}");
-		let mut cmd = self.into_command();
-		cmd.inherit_stdio();
-		cmd.stdout(Stdio::piped());
-		let out = cmd.spawn()?.wait_with_output().await?;
-		if !out.status.success() {
-			anyhow::bail!("command '{}' failed with status {}", str, out.status);
-		}
-		Ok(String::from_utf8(out.stdout)?)
+		let cmd = self.into_command();
+		let v = run_nix_inner_stdout(str, cmd, &mut PlainHandler).await?;
+		Ok(v)
 	}
 	pub async fn run_nix_json<T: DeserializeOwned>(self) -> Result<T> {
 		let str = self.run_nix_string().await?;
@@ -154,17 +143,14 @@
 
 	pub async fn run_nix_string(self) -> Result<String> {
 		let str = self.clone().into_string();
-		let mut cmd = self.into_command();
-		cmd.stdout(Stdio::piped());
-		run_nix_inner(str, cmd).await.map(|v| v.unwrap())
+		let cmd = self.into_command();
+		run_nix_inner_stdout(str, cmd, &mut NixHandler::default()).await
 	}
 	pub async fn run_nix(self) -> Result<()> {
 		let str = self.clone().into_string();
 		let mut cmd = self.into_command();
 		cmd.stdout(Stdio::inherit());
-		run_nix_inner(str, cmd).await.map(|v| {
-			assert!(v.is_none());
-		})
+		run_nix_inner(str, cmd, &mut NixHandler::default()).await
 	}
 }
 
@@ -179,101 +165,290 @@
 	}
 }
 
-async fn run_nix_inner(str: String, mut cmd: Command) -> Result<Option<String>> {
+async fn run_nix_inner_stdout(
+	str: String,
+	cmd: Command,
+	handler: &mut dyn Handler,
+) -> Result<String> {
+	Ok(run_nix_inner_raw(str, cmd, true, handler)
+		.await?
+		.expect("has out"))
+}
+async fn run_nix_inner(str: String, cmd: Command, handler: &mut dyn Handler) -> Result<()> {
+	let v = run_nix_inner_raw(str, cmd, false, handler).await?;
+	assert!(v.is_none());
+	Ok(())
+}
+
+trait Handler {
+	fn handle_err(&mut self, e: &str);
+	fn handle_info(&mut self, e: &str);
+}
+
+struct PlainHandler;
+impl Handler for PlainHandler {
+	fn handle_err(&mut self, e: &str) {
+		info!(target: "log", "{e}");
+	}
+
+	fn handle_info(&mut self, e: &str) {
+		info!(target: "log", "{e}");
+	}
+}
+
+#[derive(Default)]
+struct NixHandler {
+	spans: HashMap<u64, Span>,
+}
+impl Handler for NixHandler {
+	fn handle_err(&mut self, e: &str) {
+		if let Some(e) = e.strip_prefix("@nix ") {
+			let log: NixLog = match serde_json::from_str(e) {
+				Ok(l) => l,
+				Err(err) => {
+					warn!("failed to parse nix log line {:?}: {}", e, err);
+					return;
+				}
+			};
+			match log {
+				NixLog::Msg { msg, raw_msg, .. } => {
+					if !(msg.starts_with("\u{1b}[35;1mwarning:\u{1b}[0m Git tree '") && msg.ends_with("' is dirty"))
+					&& !msg.starts_with("\u{1b}[35;1mwarning:\u{1b}[0m not writing modified lock file of flake")
+					&& msg != "\u{1b}[35;1mwarning:\u{1b}[0m \u{1b}[31;1merror:\u{1b}[0m SQLite database '\u{1b}[35;1m/nix/var/nix/db/db.sqlite\u{1b}[0m' is busy" {
+						if let Some(raw_msg) = raw_msg {
+							if !msg.is_empty() {
+								info!(target: "nix", "{}\n{}", raw_msg.trim_end(), msg.trim_end())
+							} else {
+								info!(target: "nix", "{}", raw_msg.trim_end())
+							}
+						} else {
+							info!(target: "nix", "{}", msg.trim_end())
+						}
+					}
+				}
+				NixLog::Start {
+					ref fields,
+					typ,
+					id,
+					..
+				} if typ == 105 && !fields.is_empty() => {
+					if let [LogField::String(drv), ..] = &fields[..] {
+						let mut drv = drv.as_str();
+						if let Some(pkg) = drv.strip_prefix("/nix/store/") {
+							let mut it = pkg.splitn(2, '-');
+							it.next();
+							if let Some(pkg) = it.next() {
+								drv = pkg;
+							}
+						}
+						info!(target: "nix","building {}", drv);
+						let span = info_span!("build", drv);
+						span.pb_start();
+						self.spans.insert(id, span);
+					} else {
+						warn!("bad build log: {:?}", log)
+					}
+				}
+				NixLog::Start {
+					ref fields,
+					typ,
+					id,
+					..
+				} if typ == 100 && fields.len() >= 3 => {
+					if let [LogField::String(drv), LogField::String(from), LogField::String(to), ..] =
+						&fields[..]
+					{
+						let mut drv = drv.as_str();
+
+						if let Some(pkg) = drv.strip_prefix("/nix/store/") {
+							let mut it = pkg.splitn(2, '-');
+							it.next();
+							if let Some(pkg) = it.next() {
+								drv = pkg;
+							}
+						}
+						info!(target: "nix","copying {} {} -> {}", drv, from, to);
+						let span = info_span!("copy", from, to, drv);
+						span.pb_start();
+						self.spans.insert(id, span);
+					} else {
+						warn!("bad copy log: {:?}", log)
+					}
+				}
+				NixLog::Start { text, typ, id, .. }
+					if typ == 0 || typ == 102 || typ == 103 || typ == 104 =>
+				{
+					if !text.is_empty()
+						&& text != "querying info about missing paths"
+						&& text != "copying 0 paths"
+					{
+						let span = info_span!("job");
+						span.pb_start();
+						span.pb_set_message(text.trim());
+						self.spans.insert(id, span);
+						info!(target: "nix", "{}", text);
+					}
+				}
+				NixLog::Start {
+					text,
+					level: 0,
+					typ: 108,
+					..
+				} if text.is_empty() => {
+					// Cache lookup? Coupled with copy log
+				}
+				NixLog::Start {
+					text,
+					level: 4,
+					typ: 109,
+					..
+				} if text.starts_with("querying info about ") => {
+					// Cache lookup
+				}
+				NixLog::Start {
+					text,
+					level: 4,
+					typ: 101,
+					..
+				} if text.starts_with("downloading ") => {
+					// NAR downloading, coupled with copy log
+				}
+				NixLog::Start {
+					text,
+					level: 1,
+					typ: 111,
+					..
+				} if text.starts_with("waiting for a machine to build ") => {
+					// Useless repeating notification about build
+				}
+				NixLog::Start {
+					text,
+					level: 3,
+					typ: 111,
+					..
+				} if text.starts_with("resolved derivation: ") => {
+					// CA resolved
+				}
+				NixLog::Start {
+					text,
+					level: 1,
+					typ: 111,
+					id,
+					..
+				} if text.starts_with("waiting for lock on ") => {
+					let mut drv = text.strip_prefix("waiting for lock on ").unwrap();
+					if let Some(txt) = drv.strip_prefix("\u{1b}[35;1m'") {
+						drv = txt;
+					}
+					if let Some(txt) = drv.strip_suffix("'\u{1b}[0m") {
+						drv = txt;
+					}
+					if let Some(txt) = drv.split("', '").next() {
+						drv = txt;
+					}
+					if let Some(pkg) = drv.strip_prefix("/nix/store/") {
+						let mut it = pkg.splitn(2, '-');
+						it.next();
+						if let Some(pkg) = it.next() {
+							drv = pkg;
+						}
+					}
+					let span = info_span!("waiting on drv", drv);
+					span.pb_start();
+					self.spans.insert(id, span);
+					// Concurrent build of the same message
+				}
+				NixLog::Stop { id, .. } => {
+					self.spans.remove(&id);
+				}
+				NixLog::Result { fields, id, typ } if typ == 101 && !fields.is_empty() => {
+					if let Some(span) = self.spans.get(&id) {
+						if let LogField::String(s) = &fields[0] {
+							span.pb_set_message(s.trim());
+						} else {
+							warn!("bad fields: {fields:?}");
+						}
+					} else {
+						warn!("unknown result id: {id} {typ} {fields:?}");
+					}
+					// dbg!(fields, id, typ);
+				}
+				NixLog::Result { fields, id, typ } if typ == 105 && fields.len() >= 4 => {
+					if let Some(span) = self.spans.get(&id) {
+						if let [LogField::Num(done), LogField::Num(expected), LogField::Num(_running), LogField::Num(_failed)] =
+							&fields[..4]
+						{
+							span.pb_set_length(*expected);
+							span.pb_set_position(*done);
+						} else {
+							warn!("bad fields: {fields:?}");
+						}
+					} else {
+						// warn!("unknown result id: {id} {typ} {fields:?}");
+						// Unaccounted progress.
+					}
+					// dbg!(fields, id, typ);
+				}
+				NixLog::Result { typ, .. } if typ == 104 || typ == 106 => {
+					// Set phase, expected
+				}
+				_ => warn!("unknown log: {:?}", log),
+			};
+		} else {
+			warn!(target = "nix", "unknown: {}", e.trim())
+		}
+	}
+	fn handle_info(&mut self, o: &str) {
+		self.handle_err(o)
+	}
+}
+
+async fn run_nix_inner_raw(
+	str: String,
+	mut cmd: Command,
+	want_stdout: bool,
+	handler: &mut dyn Handler,
+) -> Result<Option<String>> {
 	info!("running {str}");
 	cmd.arg("--log-format").arg("internal-json");
 	cmd.stderr(Stdio::piped());
+	cmd.stdout(Stdio::piped());
 	let mut child = cmd.spawn()?;
 	let mut stderr = child.stderr.take().unwrap();
-	let stdout = child.stdout.take();
-	let wants_stdout = stdout.is_some();
+	let stdout = child.stdout.take().unwrap();
 	let mut err = FramedRead::new(&mut stderr, LinesCodec::new());
-	let mut out: Box<dyn AsyncRead + Unpin> = stdout
-		.map(|s| Box::new(s) as Box<dyn AsyncRead + Unpin>)
+	let mut out: Option<Box<dyn AsyncRead + Unpin>> = Some(Box::new(stdout));
+	let mut ob = want_stdout
+		.then(|| out.take().unwrap())
+		.unwrap_or_else(|| Box::new(EmptyAsyncRead));
+	let mut ol = (!want_stdout)
+		.then(|| out.take().unwrap())
 		.unwrap_or_else(|| Box::new(EmptyAsyncRead));
-	let mut out = FramedRead::new(&mut out, BytesCodec::new());
+	let mut ob = FramedRead::new(&mut ob, BytesCodec::new());
+	let mut ol = FramedRead::new(&mut ol, LinesCodec::new());
 
 	// while let Some(line) = read.next().await? {}
 
-	let mut out_buf = if wants_stdout { Some(vec![]) } else { None };
+	let mut out_buf = if want_stdout { Some(vec![]) } else { None };
 	loop {
 		select! {
 			e = err.next() => {
 				if let Some(e) = e {
 					let e = e?;
-					if let Some(e) = e.strip_prefix("@nix ") {
-
-						let log: NixLog = match serde_json::from_str(e) {
-							Ok(l) => l,
-							Err(err) => {
-								warn!("failed to parse nix log line {:?}: {}", e, err);
-								continue;
-							},
-						};
-						match log {
-							NixLog::Msg { msg, raw_msg, .. } => {
-								if !(msg.starts_with("\u{1b}[35;1mwarning:\u{1b}[0m Git tree '") && msg.ends_with("' is dirty"))
-									&& !msg.starts_with("\u{1b}[35;1mwarning:\u{1b}[0m not writing modified lock file of flake")
-									&& msg != "\u{1b}[35;1mwarning:\u{1b}[0m \u{1b}[31;1merror:\u{1b}[0m SQLite database '\u{1b}[35;1m/nix/var/nix/db/db.sqlite\u{1b}[0m' is busy" {
-									if let Some(raw_msg) = raw_msg {
-										info!(target: "nix", "{raw_msg}\n{msg}")
-									}else {
-										info!(target: "nix", "{msg}")
-
-									}
-								}
-							},
-							NixLog::Start { ref fields, typ, .. } if typ == 105 && !fields.is_empty() => {
-								if let [LogField::String(drv), ..] = &fields[..] {
-									info!(target: "nix","building {}", drv)
-								} else {
-									warn!("bad build log: {:?}", log)
-								}
-							},
-							NixLog::Start { ref fields, typ, .. } if typ == 100 && fields.len() >= 3 => {
-								if let [LogField::String(drv), LogField::String(from), LogField::String(to), ..] = &fields[..] {
-									info!(target: "nix","copying {} {} -> {}", drv, from, to)
-								} else {
-									warn!("bad copy log: {:?}", log)
-								}
-							},
-							NixLog::Start { text, typ, .. } if typ == 0 || typ == 102 || typ == 103 || typ == 104 => {
-								if !text.is_empty() && text != "querying info about missing paths" && text != "copying 0 paths" {
-									info!(target: "nix", "{}", text)
-								}
-							},
-							NixLog::Start { text, level: 0, typ: 108, .. } if text.is_empty() => {
-								// Cache lookup? Coupled with copy log
-							},
-							NixLog::Start { text, level: 4, typ: 109, .. } if text.starts_with("querying info about ") => {
-								// Cache lookup
-							}
-							NixLog::Start { text, level: 4, typ: 101, .. } if text.starts_with("downloading ") => {
-								// NAR downloading, coupled with copy log
-							}
-							NixLog::Start { text, level: 1, typ: 111, .. } if text.starts_with("waiting for a machine to build ") => {
-								// Useless repeating notification about build
-							}
-							NixLog::Start { text, level: 3, typ: 111, .. } if text.starts_with("resolved derivation: ") => {
-								// CA resolved
-							}
-							NixLog::Start { text, level: 1, typ: 111, .. } if text.starts_with("waiting for lock on ") => {
-								// Concurrent build of the same message
-							}
-							NixLog::Stop { .. } => {},
-							NixLog::Result { .. } => {},
-							_ => warn!("unknown log: {:?}", log)
-						};
-					} else {
-						warn!(target="nix","unknown: {}", e)
-					}
+					handler.handle_err(&e);
 				}
 			},
-			o = out.next() => {
+			o = ob.next() => {
 				if let Some(o) = o {
 					out_buf.as_mut().expect("stdout == wants_stdout").extend_from_slice(&o?);
 				}
 			},
+			o = ol.next() => {
+				if let Some(o) = o {
+					let o = o?;
+					handler.handle_info(&o);
+				}
+			},
 			code = child.wait() => {
 				let code = code?;
 				if !code.success() {
@@ -285,17 +460,6 @@
 	}
 
 	Ok(out_buf.map(String::from_utf8).transpose()?)
-}
-
-#[async_trait]
-pub trait CommandExt {
-	// async fn run_nix(&mut self) -> Result<()>;
-	// async fn run_nix_json<T: DeserializeOwned>(&mut self) -> Result<T>;
-	// async fn run_nix_string(&mut self) -> Result<String>;
-	// async fn run(&mut self) -> Result<()>;
-	// async fn run_json<T: DeserializeOwned>(&mut self) -> Result<T>;
-	// async fn run_string(&mut self) -> Result<String>;
-	fn inherit_stdio(&mut self) -> &mut Self;
 }
 
 #[derive(Debug)]
@@ -361,14 +525,7 @@
 		id: u64,
 		#[serde(rename = "type")]
 		typ: u32,
+		#[serde(default)]
+		fields: Vec<LogField>,
 	},
-}
-
-#[async_trait]
-impl CommandExt for Command {
-	fn inherit_stdio(&mut self) -> &mut Self {
-		self.stderr(Stdio::inherit());
-		self.stdout(Stdio::inherit());
-		self
-	}
 }
modifiedcmds/fleet/src/host.rsdiffbeforeafterboth
--- a/cmds/fleet/src/host.rs
+++ b/cmds/fleet/src/host.rs
@@ -8,7 +8,7 @@
 	sync::Arc,
 };
 
-use anyhow::{Result, bail, Context};
+use anyhow::{bail, Context, Result};
 use clap::{ArgGroup, Parser};
 use serde::de::DeserializeOwned;
 use tempfile::NamedTempFile;
@@ -61,7 +61,12 @@
 		command.run().await
 	}
 	#[must_use]
-	pub async fn run_string_on(&self, host: &str, mut command: MyCommand, sudo: bool) -> Result<String> {
+	pub async fn run_string_on(
+		&self,
+		host: &str,
+		mut command: MyCommand,
+		sudo: bool,
+	) -> Result<String> {
 		if sudo {
 			command = command.sudo();
 		}
@@ -87,8 +92,7 @@
 			.arg(self.configuration_attr_name("configuredHosts"))
 			.args(["--apply", "builtins.attrNames", "--json", "--show-trace"])
 			.args(&self.nix_args);
-		cmd.run_nix_json()
-			.await
+		cmd.run_nix_json().await
 	}
 	pub async fn shared_config_attr<T: DeserializeOwned>(&self, attr: &str) -> Result<T> {
 		let mut cmd = MyCommand::new("nix");
@@ -96,8 +100,7 @@
 			.arg(self.configuration_attr_name(&format!("configUnchecked.{}", attr)))
 			.args(["--json", "--show-trace"])
 			.args(&self.nix_args);
-		cmd.run_nix_json()
-			.await
+		cmd.run_nix_json().await
 	}
 	pub async fn shared_config_attr_names(&self, attr: &str) -> Result<Vec<String>> {
 		let mut cmd = MyCommand::new("nix");
@@ -106,8 +109,7 @@
 			.args(["--apply", "builtins.attrNames"])
 			.args(["--json", "--show-trace"])
 			.args(&self.nix_args);
-		cmd.run_nix_json()
-			.await
+		cmd.run_nix_json().await
 	}
 	pub async fn config_attr<T: DeserializeOwned>(&self, host: &str, attr: &str) -> Result<T> {
 		let mut cmd = MyCommand::new("nix");
@@ -120,8 +122,7 @@
 			)
 			.args(["--json", "--show-trace"])
 			.args(&self.nix_args);
-		cmd.run_nix_json()
-			.await
+		cmd.run_nix_json().await
 	}
 
 	pub(super) fn data(&self) -> Ref<FleetData> {
@@ -151,14 +152,14 @@
 	pub fn list_secrets(&self, host: &str) -> Vec<String> {
 		let data = self.data();
 		let Some(host_secrets) = data.host_secrets.get(host) else {
-			return Vec::new(); 
+			return Vec::new();
 		};
 		host_secrets.keys().cloned().collect()
 	}
 	pub fn has_secret(&self, host: &str, secret: &str) -> bool {
 		let data = self.data();
 		let Some(host_secrets) = data.host_secrets.get(host) else {
-			return false; 
+			return false;
 		};
 		host_secrets.contains_key(secret)
 	}
@@ -168,23 +169,38 @@
 		host_secrets.insert(secret, value);
 	}
 
-	pub async fn decrypt_on_host(&self, host: &str, data: Vec<u8>) -> Result<Vec<u8>>{
+	pub async fn decrypt_on_host(&self, host: &str, data: Vec<u8>) -> Result<Vec<u8>> {
 		let data = z85::encode(&data);
 		let mut cmd = MyCommand::new("fleet-install-secrets");
 		cmd.arg("decrypt").eqarg("--secret", data);
 		cmd = cmd.sudo().ssh(host);
-		let encoded = cmd.run_string().await.context("failed to call remote host for decrypt")?.trim().to_owned();
+		let encoded = cmd
+			.run_string()
+			.await
+			.context("failed to call remote host for decrypt")?
+			.trim()
+			.to_owned();
 		Ok(z85::decode(encoded).context("bad encoded data? outdated host?")?)
 	}
-	pub async fn reencrypt_on_host(&self, host: &str, data: Vec<u8>, targets: Vec<String>) -> Result<Vec<u8>>{
+	pub async fn reencrypt_on_host(
+		&self,
+		host: &str,
+		data: Vec<u8>,
+		targets: Vec<String>,
+	) -> Result<Vec<u8>> {
 		let data = z85::encode(&data);
 		let mut recmd = MyCommand::new("fleet-install-secrets");
-		recmd.arg("reencrypt").eqarg("--secret",data);
+		recmd.arg("reencrypt").eqarg("--secret", data);
 		for target in targets {
 			recmd.eqarg("--targets", target);
 		}
 		recmd = recmd.sudo().ssh(host);
-		let encoded = recmd.run_string().await.context("failed to call remote host for decrypt")?.trim().to_owned();
+		let encoded = recmd
+			.run_string()
+			.await
+			.context("failed to call remote host for decrypt")?
+			.trim()
+			.to_owned();
 		Ok(z85::decode(encoded).context("bad encoded data? outdated host?")?)
 	}
 
@@ -192,11 +208,11 @@
 	pub fn host_secret(&self, host: &str, secret: &str) -> Result<FleetSecret> {
 		let data = self.data();
 		let Some(host_secrets) = data.host_secrets.get(host) else {
-            bail!("no secrets for machine {host}");
-        };
+			bail!("no secrets for machine {host}");
+		};
 		let Some(secret) = host_secrets.get(secret) else {
-            bail!("machine {host} has no secret {secret}");
-        };
+			bail!("machine {host} has no secret {secret}");
+		};
 		Ok(secret.clone())
 	}
 	#[must_use]
modifiedcmds/fleet/src/main.rsdiffbeforeafterboth
--- a/cmds/fleet/src/main.rs
+++ b/cmds/fleet/src/main.rs
@@ -7,16 +7,19 @@
 
 use std::ffi::OsString;
 use std::io;
+use std::time::Duration;
 
 use anyhow::{anyhow, bail, Result};
 use clap::Parser;
 
 use cmds::{build_systems::BuildSystems, info::Info, secrets::Secrets};
 use host::{Config, FleetOpts};
+use indicatif::{ProgressState, ProgressStyle};
 use tokio::fs;
 use tokio::process::Command;
 use tracing::{info, metadata::LevelFilter};
-use tracing_subscriber::EnvFilter;
+use tracing_indicatif::IndicatifLayer;
+use tracing_subscriber::{prelude::*, EnvFilter};
 
 #[derive(Parser)]
 struct Prefetch {}
@@ -77,21 +80,53 @@
 	};
 	Ok(())
 }
+fn elapsed_subsec(state: &ProgressState, writer: &mut dyn std::fmt::Write) {
+	let _ = writer.write_str(&format!("{:?}", state.elapsed()));
+}
 
 #[tokio::main]
 async fn main() -> Result<()> {
+	let indicatif_layer = IndicatifLayer::new().with_progress_style(
+		ProgressStyle::with_template(
+			"{color_start}{span_child_prefix} {span_name}{{{span_fields}}}{color_end} {wide_msg} {color_start}{pos:>7}/{len:7}{elapsed}{color_end}",
+		)
+		.unwrap()
+		.with_key(
+			"color_start",
+			|state: &ProgressState, writer: &mut dyn std::fmt::Write| {
+				let elapsed = state.elapsed();
+
+				if elapsed > Duration::from_secs(60) {
+					// Red
+					let _ = write!(writer, "\x1b[{}m", 1 + 30);
+				} else if elapsed > Duration::from_secs(30) {
+					// Yellow
+					let _ = write!(writer, "\x1b[{}m", 3 + 30);
+				}
+			},
+		)
+		.with_key(
+			"color_end",
+			|state: &ProgressState, writer: &mut dyn std::fmt::Write| {
+				if state.elapsed() > Duration::from_secs(30) {
+					let _ = write!(writer, "\x1b[0m");
+				}
+			},
+		),
+	);
+
 	let filter = EnvFilter::from_default_env().add_directive(LevelFilter::INFO.into());
-	tracing_subscriber::FmtSubscriber::builder()
-		.with_env_filter(filter)
-		.without_time()
-		.with_target(false)
-		.with_writer(|| {
-			// eprintln!("Line");
-			io::stderr()
-		})
-		.try_init()
-		.map_err(|e| anyhow!("Failed to initialize logger: {}", e))?;
 
+	tracing_subscriber::registry()
+		.with(
+			tracing_subscriber::fmt::layer()
+				.without_time()
+				.with_target(false)
+				.with_writer(indicatif_layer.get_stderr_writer())
+				.with_filter(filter), // .withou,
+		)
+		.with(indicatif_layer)
+		.init();
 	info!("Starting");
 	let mut os_args = std::env::args_os();
 	let opts = RootOpts::parse_from((&mut os_args).take_while(|v| v != "--"));
modifiedflake.lockdiffbeforeafterboth
--- a/flake.lock
+++ b/flake.lock
@@ -1,12 +1,15 @@
 {
   "nodes": {
     "flake-utils": {
+      "inputs": {
+        "systems": "systems"
+      },
       "locked": {
-        "lastModified": 1667395993,
-        "narHash": "sha256-nuEHfE/LcWyuSWnS8t12N1wc105Qtau+/OdUAjtQ0rA=",
+        "lastModified": 1694529238,
+        "narHash": "sha256-zsNZZGTGnMOf9YpHKJqMSsa0dXbfmxeoJ7xHlrt+xmY=",
         "owner": "numtide",
         "repo": "flake-utils",
-        "rev": "5aed5285a952e0b949eb3ba02c12fa4fcfef535f",
+        "rev": "ff7b65b44d01cf9ba6a71320833626af21126384",
         "type": "github"
       },
       "original": {
@@ -16,12 +19,15 @@
       }
     },
     "flake-utils_2": {
+      "inputs": {
+        "systems": "systems_2"
+      },
       "locked": {
-        "lastModified": 1659877975,
-        "narHash": "sha256-zllb8aq3YO3h8B/U0/J1WBgAL8EX5yWf5pMj3G0NAmc=",
+        "lastModified": 1681202837,
+        "narHash": "sha256-H+Rh19JDwRtpVPAWp64F+rlEtxUWBAQW28eAi3SRSzg=",
         "owner": "numtide",
         "repo": "flake-utils",
-        "rev": "c0e246b9b83f637f4681389ecabcb2681b4f3af0",
+        "rev": "cfacdce06f30d2b68473a46042957675eebb3401",
         "type": "github"
       },
       "original": {
@@ -32,11 +38,11 @@
     },
     "nixpkgs": {
       "locked": {
-        "lastModified": 1670700221,
-        "narHash": "sha256-+Fy/Wu8qeAppA14R4gLSlxmD0jGNVWYrgAJUaL23qkI=",
+        "lastModified": 1696884899,
+        "narHash": "sha256-SZILkoh8KZxjvFHO3yzOUw7n1Mf9WqMdUqoxf8eKPM4=",
         "owner": "nixos",
         "repo": "nixpkgs",
-        "rev": "ccf0f09e2e6744dcd721860a44c633e8708fde2b",
+        "rev": "ba10489eae3b2b2f665947b516e7043594a235c8",
         "type": "github"
       },
       "original": {
@@ -61,11 +67,11 @@
         ]
       },
       "locked": {
-        "lastModified": 1670639101,
-        "narHash": "sha256-UvPSgbtaOk9WcgVqywnvQXOEEHx6OXdG+QXIwnbyvCw=",
+        "lastModified": 1696817516,
+        "narHash": "sha256-Xt9OY4Wnk9/vuUfA0OHFtmSlaen5GyiS9msgwOz3okI=",
         "owner": "oxalica",
         "repo": "rust-overlay",
-        "rev": "d00c488cb455c21fea731167bf8c1b8da605aac3",
+        "rev": "c0df7f2a856b5ff27a3ce314f6d7aacf5fda546f",
         "type": "github"
       },
       "original": {
@@ -73,6 +79,36 @@
         "repo": "rust-overlay",
         "type": "github"
       }
+    },
+    "systems": {
+      "locked": {
+        "lastModified": 1681028828,
+        "narHash": "sha256-Vy1rq5AaRuLzOxct8nz4T6wlgyUR7zLU309k9mBC768=",
+        "owner": "nix-systems",
+        "repo": "default",
+        "rev": "da67096a3b9bf56a91d16901293e51ba5b49a27e",
+        "type": "github"
+      },
+      "original": {
+        "owner": "nix-systems",
+        "repo": "default",
+        "type": "github"
+      }
+    },
+    "systems_2": {
+      "locked": {
+        "lastModified": 1681028828,
+        "narHash": "sha256-Vy1rq5AaRuLzOxct8nz4T6wlgyUR7zLU309k9mBC768=",
+        "owner": "nix-systems",
+        "repo": "default",
+        "rev": "da67096a3b9bf56a91d16901293e51ba5b49a27e",
+        "type": "github"
+      },
+      "original": {
+        "owner": "nix-systems",
+        "repo": "default",
+        "type": "github"
+      }
     }
   },
   "root": "root",
modifiedflake.nixdiffbeforeafterboth
--- a/flake.nix
+++ b/flake.nix
@@ -15,7 +15,7 @@
           inherit system; overlays = [ (import rust-overlay) ];
         };
       llvmPkgs = pkgs.buildPackages.llvmPackages_11;
-      rust = (pkgs.rustChannelOf { date = "2022-12-02"; channel = "nightly"; }).default.override { extensions = [ "rust-src" "rust-analyzer" ]; };
+      rust = (pkgs.rustChannelOf { date = "2023-10-05"; channel = "nightly"; }).default.override { extensions = [ "rust-src" "rust-analyzer" ]; };
       rustPlatform = pkgs.makeRustPlatform { cargo = rust; rustc = rust; };
     in
     {
@@ -27,7 +27,7 @@
           cargo-udeps
           cargo-fuzz
 
-          pkgconfig
+          pkg-config
           openssl
         ];
       };
modifiedpkgs/fleet-install-secrets.nixdiffbeforeafterboth
--- a/pkgs/fleet-install-secrets.nix
+++ b/pkgs/fleet-install-secrets.nix
@@ -10,7 +10,7 @@
   cargoLock = {
     lockFile = ../Cargo.lock;
     outputHashes = {
-      "alejandra-3.0.0" = "sha256-YSdHsJ73G7TEFzbmpZ2peuMefIa9/vNB2g+xdiyma3U=";
+      "alejandra-3.0.0" = "sha256-lStDIPizbJipd1JpNKX1olBKzyIosyC2U/mVFwJPcZE=";
     };
   };
 }