git.delta.rocks / jrsonnet / refs/commits / 41ecf404fcd9

--- a/cmds/fleet/Cargo.toml
+++ b/cmds/fleet/Cargo.toml
@@ -7,19 +7,24 @@
 
 [dependencies]
 anyhow = "1.0"
-log = "0.4.14"
-env_logger = "0.9.0"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 time = { version = "0.3.2", features = ["serde"] }
 tempfile = "3.2"
 once_cell = "1.5"
 hostname = "0.3.1"
-age-core = "0.6.0"
+age-core = "0.7.0"
 peg = "0.7.0"
 nixlike = {path = "../../crates/nixlike"}
-age = { version = "0.6.0", features = ["ssh", "armor"] }
+age = { version = "0.7.0", features = ["ssh", "armor"] }
 base64 = "0.13.0"
 chrono = { version = "0.4.19", features = ["serde"] }
 z85 = "3.0.3"
+base58 = "*"
 structopt = "0.3.23"
+tokio = { version = "1.14.0", features = ["full"] }
+tracing = "0.1.29"
+tracing-subscriber = { version = "0.3.3", features = ["fmt", "env-filter"] }
+tokio-util = { version = "0.6.9", features = ["codec"] }
+async-trait = "0.1.52"
+futures = "0.3.17"

--- a/cmds/fleet/src/cmds/build_systems.rs
+++ b/cmds/fleet/src/cmds/build_systems.rs
@@ -1,11 +1,12 @@
-use std::{env::current_dir, process::Command};
+use std::{env::current_dir, time::Duration};
 
 use crate::{command::CommandExt, host::Config, nix::SYSTEMS_ATTRIBUTE};
 use anyhow::Result;
-use log::info;
 use structopt::StructOpt;
+use tokio::{process::Command, task::LocalSet, time::sleep};
+use tracing::{error, field, info, info_span, warn, Instrument};
 
-#[derive(StructOpt)]
+#[derive(StructOpt, Clone)]
 pub struct BuildSystems {
 	/// --builders arg for nix
 	#[structopt(long)]
@@ -78,117 +79,158 @@
 }
 
 impl BuildSystems {
-	pub fn run(self, config: &Config) -> Result<()> {
-		let hosts = config.list_hosts()?;
-
+	pub async fn run(self, config: &Config) -> Result<()> {
+		let hosts = config.list_hosts().await?;
+		let set = LocalSet::new();
+		let this = &self;
 		for host in hosts.iter() {
 			if config.should_skip(host) {
 				continue;
 			}
-			info!("Building host {}", host);
-			let built = {
-				let dir = tempfile::tempdir()?;
-				dir.path().to_owned()
-			};
+			let config = config.clone();
+			let host = host.clone();
+			let this = this.clone();
+			let span = info_span!("deployment", host = field::display(&host));
+			set.spawn_local(
+				(async move {
+					let res: Result<()> = try {
+						info!("building");
+						let built = {
+							let dir = tempfile::tempdir()?;
+							dir.path().to_owned()
+						};
 
-			let mut nix_build = if self.privileged_build {
-				let mut out = Command::new("sudo");
-				out.arg("nix");
-				out
-			} else {
-				Command::new("nix")
-			};
-			nix_build
-				.args(&["build", "--impure", "--no-link", "--out-link"])
-				.arg(&built)
-				.arg(format!(
-					"{}.{}.config.system.build.toplevel",
-					SYSTEMS_ATTRIBUTE, host,
-				));
+						let mut nix_build = if this.privileged_build {
+							let mut out = Command::new("sudo");
+							out.arg("nix");
+							out
+						} else {
+							Command::new("nix")
+						};
+						nix_build
+							.args(&[
+								"build",
+								"--impure",
+								"--json",
+								// "--show-trace",
+								"--no-link",
+								"--out-link",
+							])
+							.arg(&built)
+							.arg(format!(
+								"{}.{}.config.system.build.toplevel",
+								SYSTEMS_ATTRIBUTE, host,
+							));
+
+						if let Some(builders) = &this.builders {
+							nix_build.arg("--builders").arg(builders);
+						}
+						if let Some(jobs) = &this.jobs {
+							nix_build.arg("--max-jobs");
+							nix_build.arg(format!("{}", jobs));
+						}
+						if !this.fail_fast {
+							nix_build.arg("--keep-going");
+						}
 
-			if let Some(builders) = &self.builders {
-				nix_build.arg("--builders").arg(builders);
-			}
-			if let Some(jobs) = &self.jobs {
-				nix_build.arg("--max-jobs");
-				nix_build.arg(format!("{}", jobs));
-			}
-			if !self.fail_fast {
-				nix_build.arg("--keep-going");
-			}
+						nix_build.run_nix().await?;
+						let built = std::fs::canonicalize(built)?;
 
-			nix_build.inherit_stdio().run()?;
-			let built = std::fs::canonicalize(built)?;
-			info!("Built closure: {:?}", built);
+						let action = Action::from(this.subcommand.clone());
 
-			let action = Action::from(self.subcommand.clone());
+						match action {
+							Action::Upload(action) => {
+								if !config.is_local(&host) {
+									info!("uploading system closure");
+									let mut tries = 0;
+									loop {
+										match Command::new("nix")
+											.args(&["copy", "--to"])
+											.arg(format!("ssh://root@{}", host))
+											.arg(&built)
+											.inherit_stdio()
+											.run_nix()
+											.await
+										{
+											Ok(()) => break,
+											Err(e) if tries < 3 => {
+												tries += 1;
+												warn!("Copy failure ({}/3): {}", tries, e);
+												sleep(Duration::from_millis(5000)).await;
+											}
+											Err(e) => return Err(e),
+										}
+									}
+								}
+								if let Some(action) = action {
+									if action.should_switch_profile() {
+										info!("switching generation");
+										config
+											.command_on(&host, "nix-env", true)
+											.args(&["-p", "/nix/var/nix/profiles/system", "--set"])
+											.arg(&built)
+											.inherit_stdio()
+											.run()
+											.await?;
+									}
+									info!("executing activation script");
+									let mut switch_script = built.clone();
+									switch_script.push("bin");
+									switch_script.push("switch-to-configuration");
+									config
+										.command_on(&host, switch_script, true)
+										.arg(action.name())
+										.inherit_stdio()
+										.run()
+										.await?;
+								}
+							}
+							Action::Package(PackageAction::SdImage) => {
+								let mut out = current_dir()?;
+								out.push(format!("sd-image-{}", host));
 
-			match action {
-				Action::Upload(action) => {
-					if !config.is_local(host) {
-						info!("Uploading system closure");
-						Command::new("nix")
-							.args(&["copy", "--to"])
-							.arg(format!("ssh://root@{}", host))
-							.arg(&built)
-							.inherit_stdio()
-							.run()?;
-					}
-					if let Some(action) = action {
-						if action.should_switch_profile() {
-							info!("Switching generation");
-							config
-								.command_on(host, "nix-env", true)
-								.args(&["-p", "/nix/var/nix/profiles/system", "--set"])
-								.arg(&built)
-								.inherit_stdio()
-								.run()?;
-						}
-						info!("Executing activation script");
-						let mut switch_script = built.clone();
-						switch_script.push("bin");
-						switch_script.push("switch-to-configuration");
-						config
-							.command_on(host, switch_script, true)
-							.arg(action.name())
-							.inherit_stdio()
-							.run()?;
-					}
-				}
-				Action::Package(PackageAction::SdImage) => {
-					let mut out = current_dir()?;
-					out.push(format!("sd-image-{}", host));
+								info!("building sd image to {:?}", out);
+								let mut nix_build = if this.privileged_build {
+									let mut out = Command::new("sudo");
+									out.arg("nix");
+									out
+								} else {
+									Command::new("nix")
+								};
+								nix_build
+									.args(&["build", "--impure", "--no-link", "--out-link"])
+									.arg(&out)
+									.arg(format!(
+										"{}.{}.config.system.build.sdImage",
+										SYSTEMS_ATTRIBUTE, host,
+									));
+								if let Some(builders) = &this.builders {
+									nix_build.arg("--builders").arg(builders);
+								}
+								if let Some(jobs) = &this.jobs {
+									nix_build.arg("--max-jobs");
+									nix_build.arg(format!("{}", jobs));
+								}
+								if !this.fail_fast {
+									nix_build.arg("--keep-going");
+								}
 
-					info!("Building sd image to {:?}", out);
-					let mut nix_build = if self.privileged_build {
-						let mut out = Command::new("sudo");
-						out.arg("nix");
-						out
-					} else {
-						Command::new("nix")
+								nix_build.inherit_stdio().run_nix().await?;
+							}
+						};
 					};
-					nix_build
-						.args(&["build", "--impure", "--no-link", "--out-link"])
-						.arg(&out)
-						.arg(format!(
-							"{}.{}.config.system.build.sdImage",
-							SYSTEMS_ATTRIBUTE, host,
-						));
-					if let Some(builders) = &self.builders {
-						nix_build.arg("--builders").arg(builders);
+					match res {
+						Ok(_) => {}
+						Err(e) => {
+							error!("failed to deploy host: {}", e)
+						}
 					}
-					if let Some(jobs) = &self.jobs {
-						nix_build.arg("--max-jobs");
-						nix_build.arg(format!("{}", jobs));
-					}
-					if !self.fail_fast {
-						nix_build.arg("--keep-going");
-					}
-
-					nix_build.inherit_stdio().run()?;
-				}
-			};
+					Ok(())
+				})
+				.instrument(span),
+			);
 		}
+		set.await;
 		Ok(())
 	}
 }

--- a/cmds/fleet/src/cmds/info.rs
+++ b/cmds/fleet/src/cmds/info.rs
@@ -30,13 +30,13 @@
 }
 
 impl Info {
-	pub fn run(self, config: &Config) -> Result<()> {
+	pub async fn run(self, config: &Config) -> Result<()> {
 		let mut data = Vec::new();
 		match self.cmd {
 			InfoCmd::ListHosts { ref tagged } => {
-				'host: for host in config.list_hosts()? {
+				'host: for host in config.list_hosts().await? {
 					if !tagged.is_empty() {
-						let tags: Vec<String> = config.config_attr(&host, "tags")?;
+						let tags: Vec<String> = config.config_attr(&host, "tags").await?;
 						for tag in tagged {
 							if !tags.contains(&tag) {
 								continue 'host;
@@ -57,10 +57,18 @@
 				);
 				let mut out = <BTreeSet<String>>::new();
 				if external {
-					out.extend(config.config_attr::<Vec<String>>(&host, "network.externalIps")?);
+					out.extend(
+						config
+							.config_attr::<Vec<String>>(&host, "network.externalIps")
+							.await?,
+					);
 				}
 				if internal {
-					out.extend(config.config_attr::<Vec<String>>(&host, "network.internalIps")?);
+					out.extend(
+						config
+							.config_attr::<Vec<String>>(&host, "network.internalIps")
+							.await?,
+					);
 				}
 				for ip in out {
 					data.push(ip);

--- a/cmds/fleet/src/cmds/secrets/mod.rs
+++ b/cmds/fleet/src/cmds/secrets/mod.rs
@@ -3,6 +3,7 @@
 	host::Config,
 };
 use anyhow::{bail, Result};
+use futures::{StreamExt, TryStreamExt};
 use std::{
 	io::{self, Cursor, Read},
 	path::PathBuf,
@@ -44,14 +45,14 @@
 }
 
 impl Secrets {
-	pub fn run(self, config: &Config) -> Result<()> {
+	pub async fn run(self, config: &Config) -> Result<()> {
 		match self {
 			Secrets::ForceKeys => {
-				for host in config.list_hosts()? {
+				for host in config.list_hosts().await? {
 					if config.should_skip(&host) {
 						continue;
 					}
-					config.key(&host)?;
+					config.key(&host).await?;
 				}
 			}
 			Secrets::AddShared {
@@ -61,10 +62,10 @@
 				public,
 				public_file,
 			} => {
-				let recipients = machines
-					.iter()
-					.map(|m| config.recipient(m))
-					.collect::<Result<Vec<_>>>()?;
+				let recipients = futures::stream::iter(machines.iter())
+					.then(|m| config.recipient(m))
+					.try_collect::<Vec<_>>()
+					.await?;
 
 				let secret = {
 					let mut input = vec![];
@@ -117,7 +118,7 @@
 				public,
 				public_file,
 			} => {
-				let recipient = config.recipient(&machine)?;
+				let recipient = config.recipient(&machine).await?;
 
 				let secret = {
 					let mut input = vec![];

--- a/cmds/fleet/src/command.rs
+++ b/cmds/fleet/src/command.rs
@@ -1,42 +1,204 @@
-use std::{
-	ffi::OsStr,
-	process::{Command, Stdio},
-};
+use std::{ffi::OsStr, process::Stdio};
 
 use anyhow::{Context, Result};
-use serde::de::DeserializeOwned;
+use async_trait::async_trait;
+use futures::StreamExt;
+use serde::{
+	de::{DeserializeOwned, Visitor},
+	Deserialize, 
+};
+use tokio::{process::Command, select};
+use tokio_util::codec::{BytesCodec, FramedRead, LinesCodec};
+use tracing::{info, warn};
 
+#[async_trait]
 pub trait CommandExt {
-	fn run(&mut self) -> Result<()>;
-	fn run_json<T: DeserializeOwned>(&mut self) -> Result<T>;
-	fn run_string(&mut self) -> Result<String>;
+	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;
 	fn ssh_on(host: impl AsRef<OsStr>, command: impl AsRef<OsStr>) -> Self;
 }
 
+#[derive(Debug)]
+enum LogField {
+	String(String),
+	Num(u64),
+}
+
+impl<'de> Deserialize<'de> for LogField {
+	fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+	where
+		D: serde::Deserializer<'de>,
+	{
+		struct StringOrNum;
+		impl<'de> Visitor<'de> for StringOrNum {
+			type Value = LogField;
+
+			fn expecting(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
+				write!(f, "string or unsigned")
+			}
+
+			fn visit_str<E>(self, v: &str) -> Result<Self::Value, E>
+			where
+				E: serde::de::Error,
+			{
+				Ok(LogField::String(v.to_owned()))
+			}
+
+			fn visit_u64<E>(self, v: u64) -> Result<Self::Value, E>
+			where
+				E: serde::de::Error,
+			{
+				Ok(LogField::Num(v))
+			}
+		}
+
+		deserializer.deserialize_any(StringOrNum)
+	}
+}
+
+#[derive(Deserialize, Debug)]
+#[serde(rename_all = "camelCase", tag = "action")]
+enum NixLog {
+	Msg {
+		level: u32,
+		msg: String,
+		raw_msg: Option<String>,
+	},
+	Start {
+		id: u64,
+		level: u32,
+		#[serde(default)]
+		fields: Vec<LogField>,
+		text: String,
+		#[serde(rename = "type")]
+		typ: u32,
+	},
+	Stop {
+		id: u64,
+	},
+	Result {
+		id: u64,
+		#[serde(rename = "type")]
+		typ: u32,
+	},
+}
+
+#[async_trait]
 impl CommandExt for Command {
+	async fn run_nix(&mut self) -> Result<()> {
+		self.run_nix_string().await.map(|_| ())
+	}
+	async fn run_nix_json<T: DeserializeOwned>(&mut self) -> Result<T> {
+		let str = self.run_nix_string().await?;
+		serde_json::from_str(&str).with_context(|| format!("{:?}", str))
+	}
+
+	async fn run_nix_string(&mut self) -> Result<String> {
+		self.arg("--log-format").arg("internal-json");
+		self.stderr(Stdio::piped());
+		self.stdout(Stdio::piped());
+		let mut child = self.spawn()?;
+		let mut stderr = child.stderr.take().unwrap();
+		let mut stdout = child.stdout.take().unwrap();
+		let mut err = FramedRead::new(&mut stderr, LinesCodec::new());
+		let mut out = FramedRead::new(&mut stdout, BytesCodec::new());
+
+		// while let Some(line) = read.next().await? {}
+
+		let mut out_buf = vec![];
+		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.ends_with(" is dirty") && msg.contains("warning:") && msg.contains(" Git tree ")) {
+										info!(target: "nix", "{}", raw_msg.unwrap_or(msg))
+									}
+								},
+								NixLog::Start { ref fields, typ, .. } if typ == 105 && fields.len() >= 1 => {
+									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::Stop { .. } => {},
+								NixLog::Result { .. } => {},
+								_ => warn!("unknown log: {:?}", log)
+							};
+						} else {
+							warn!(target="nix","unknown: {}", e)
+						}
+					}
+				},
+				o = out.next() => {
+					if let Some(o) = o {
+						out_buf.extend_from_slice(&o?);
+					}
+				},
+				code = child.wait() => {
+					let code = code?;
+					if !code.success() {
+						anyhow::bail!("command ({:?}) failed with status {}", self, code);
+					}
+					break;
+				}
+			}
+		}
+
+		Ok(String::from_utf8(out_buf)?)
+	}
+
 	fn inherit_stdio(&mut self) -> &mut Self {
 		self.stderr(Stdio::inherit());
 		self
 	}
 
-	fn run(&mut self) -> Result<()> {
+	async fn run(&mut self) -> Result<()> {
 		self.inherit_stdio();
-		let out = self.output()?;
+		let out = self.output().await?;
 		if !out.status.success() {
 			anyhow::bail!("command ({:?}) failed with status {}", self, out.status);
 		}
 		Ok(())
 	}
 
-	fn run_json<T: DeserializeOwned>(&mut self) -> Result<T> {
-		let str = self.run_string()?;
+	async fn run_json<T: DeserializeOwned>(&mut self) -> Result<T> {
+		let str = self.run_string().await?;
 		serde_json::from_str(&str).with_context(|| format!("{:?}", str))
 	}
 
-	fn run_string(&mut self) -> Result<String> {
+	async fn run_string(&mut self) -> Result<String> {
 		self.inherit_stdio();
-		let out = self.output()?;
+		let out = self.output().await?;
 		if !out.status.success() {
 			anyhow::bail!("command ({:?}) failed with status {}", self, out.status);
 		}

--- a/cmds/fleet/src/host.rs
+++ b/cmds/fleet/src/host.rs
@@ -4,7 +4,6 @@
 	ffi::{OsStr, OsString},
 	ops::Deref,
 	path::PathBuf,
-	process::Command,
 	sync::Arc,
 };
 
@@ -12,6 +11,7 @@
 use serde::de::DeserializeOwned;
 use structopt::clap::ArgGroup;
 use structopt::StructOpt;
+use tokio::process::Command;
 
 use crate::{command::CommandExt, fleetdata::FleetData};
 
@@ -73,15 +73,15 @@
 		str
 	}
 
-	pub fn list_hosts(&self) -> Result<Vec<String>> {
+	pub async fn list_hosts(&self) -> Result<Vec<String>> {
 		Command::new("nix")
 			.arg("eval")
 			.arg(self.full_attr_name("fleetConfigurations.default.configuredHosts"))
 			.args(&["--apply", "builtins.attrNames", "--json", "--show-trace"])
-			.inherit_stdio()
-			.run_json()
+			.run_nix_json()
+			.await
 	}
-	pub fn config_attr<T: DeserializeOwned>(&self, host: &str, attr: &str) -> Result<T> {
+	pub async fn config_attr<T: DeserializeOwned>(&self, host: &str, attr: &str) -> Result<T> {
 		Command::new("nix")
 			.arg("eval")
 			.arg(self.full_attr_name(&format!(
@@ -89,8 +89,8 @@
 				host, attr
 			)))
 			.args(&["--json", "--show-trace"])
-			.inherit_stdio()
-			.run_json()
+			.run_nix_json()
+			.await
 	}
 
 	pub fn data(&self) -> Ref<FleetData> {

--- a/cmds/fleet/src/keys.rs
+++ b/cmds/fleet/src/keys.rs
@@ -2,7 +2,7 @@
 
 use crate::{command::CommandExt, host::Config};
 use anyhow::{anyhow, Result};
-use log::warn;
+use tracing::warn;
 
 impl Config {
 	pub fn cached_key(&self, host: &str) -> Option<String> {
@@ -21,7 +21,7 @@
 		host.encryption_key = key.trim().to_string();
 	}
 
-	pub fn key(&self, host: &str) -> anyhow::Result<String> {
+	pub async fn key(&self, host: &str) -> anyhow::Result<String> {
 		if let Some(key) = self.cached_key(host) {
 			Ok(key)
 		} else {
@@ -29,19 +29,20 @@
 			let key = self
 				.command_on(host, "cat", false)
 				.arg("/etc/ssh/ssh_host_ed25519_key.pub")
-				.run_string()?;
+				.run_string()
+				.await?;
 			self.update_key(host, key.clone());
 			Ok(key)
 		}
 	}
-	pub fn recipient(&self, host: &str) -> anyhow::Result<age::ssh::Recipient> {
-		let key = self.key(host)?;
+	pub async fn recipient(&self, host: &str) -> anyhow::Result<age::ssh::Recipient> {
+		let key = self.key(host).await?;
 		age::ssh::Recipient::from_str(&key).map_err(|e| anyhow!("parse recipient error: {:?}", e))
 	}
 
-	pub fn orphaned_data(&self) -> Result<Vec<String>> {
+	pub async fn orphaned_data(&self) -> Result<Vec<String>> {
 		let mut out = Vec::new();
-		let host_names = self.list_hosts()?;
+		let host_names = self.list_hosts().await?;
 		for hostname in self
 			.data()
 			.hosts

--- a/cmds/fleet/src/main.rs
+++ b/cmds/fleet/src/main.rs
@@ -1,3 +1,5 @@
+#![feature(try_blocks)]
+
 pub mod command;
 pub mod host;
 pub mod keys;
@@ -7,12 +9,16 @@
 
 mod fleetdata;
 
-use anyhow::Result;
+use std::io;
+
+use anyhow::{anyhow, Result};
 use structopt::clap::AppSettings::*;
 use structopt::StructOpt;
 
 use cmds::{build_systems::BuildSystems, info::Info, secrets::Secrets};
 use host::{Config, FleetOpts};
+use tracing::{info, metadata::LevelFilter};
+use tracing_subscriber::EnvFilter;
 
 #[derive(StructOpt)]
 enum Opts {
@@ -38,23 +44,34 @@
 	command: Opts,
 }
 
-fn run_command(config: &Config, command: Opts) -> Result<()> {
+async fn run_command(config: &Config, command: Opts) -> Result<()> {
 	match command {
-		Opts::BuildSystems(c) => c.run(config)?,
-		Opts::Secrets(s) => s.run(config)?,
-		Opts::Info(i) => i.run(config)?,
+		Opts::BuildSystems(c) => c.run(config).await?,
+		Opts::Secrets(s) => s.run(config).await?,
+		Opts::Info(i) => i.run(config).await?,
 	};
 	Ok(())
 }
 
-fn main() -> Result<()> {
-	env_logger::Builder::new()
-		.filter_level(log::LevelFilter::Info)
-		.init();
+#[tokio::main]
+async fn main() -> Result<()> {
+	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))?;
+
+	info!("Starting");
 	let opts = RootOpts::from_args();
 	let config = opts.fleet_opts.build()?;
 
-	match run_command(&config, opts.command) {
+	match run_command(&config, opts.command).await {
 		Ok(()) => {
 			config.save()?;
 			Ok(())

--- a/modules/fleet/meta.nix
+++ b/modules/fleet/meta.nix
@@ -45,6 +45,6 @@
     hosts = fleet.hostsToAttrs (host: {
       modules = config.globalModules;
     });
-    globalModules = import ../nixos/_modules.nix;
+    globalModules = import ../../nixos/modules/module-list.nix;
   };
 }