git.delta.rocks / fleet / refs/commits / 51d5ea52f7ad

--- /dev/null
+++ b/.cargo/config.toml
@@ -0,0 +1,2 @@
+[build]
+rustflags = ["--cfg", "tokio_unstable"]

--- /dev/null
+++ b/crates/goodlog-subscriber/Cargo.toml
@@ -0,0 +1,31 @@
+[package]
+name = "goodlog-subscriber"
+version.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+license.workspace = true
+
+[dependencies]
+tracing-subscriber.workspace = true
+
+# Tokio-console
+console-subscriber = { workspace = true, optional = true }
+
+# Indicatif
+human-repr = { workspace = true, optional = true }
+indicatif = { workspace = true, optional = true }
+tracing-indicatif = { workspace = true, optional = true }
+
+# OTEL
+opentelemetry.workspace = true
+opentelemetry-appender-tracing.workspace = true
+opentelemetry-exporter-env.workspace = true
+opentelemetry_sdk.workspace = true
+tracing-opentelemetry.workspace = true
+clap = { workspace = true, features = ["derive"] }
+anyhow.workspace = true
+tracing.workspace = true
+
+[features]
+indicatif = ["dep:tracing-indicatif", "dep:indicatif", "dep:human-repr"]
+tokio-console = ["dep:console-subscriber"]

--- /dev/null
+++ b/crates/goodlog-subscriber/src/lib.rs
@@ -0,0 +1,135 @@
+mod log_tree;
+
+use anyhow::Result;
+use clap::Parser;
+#[cfg(feature = "indicatif")]
+use human_repr::HumanCount;
+#[cfg(feature = "indicatif")]
+use indicatif::{ProgressState, ProgressStyle};
+use log_tree::TreeLayer;
+use opentelemetry::trace::TracerProvider;
+use opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge;
+use opentelemetry_exporter_env::{
+	OtlpBaseSettings, OtlpLogsSettings, OtlpTracesSettings, ResolvedOtlpSettings,
+};
+use opentelemetry_sdk::{logs::SdkLoggerProvider, trace::SdkTracerProvider};
+#[cfg(feature = "indicatif")]
+use tracing_indicatif::IndicatifLayer;
+use tracing_subscriber::layer::SubscriberExt as _;
+use tracing_subscriber::util::SubscriberInitExt as _;
+use tracing_subscriber::{EnvFilter, Layer as _};
+
+#[derive(Parser)]
+#[clap(next_help_heading = "Telemetry")]
+pub struct LogOpts {
+	/// Whatever opentelemetry logging should be enabled
+	#[clap(long, env = "OTEL_FLEET")]
+	otel: bool,
+	#[clap(flatten)]
+	otlp_base: OtlpBaseSettings,
+	#[clap(flatten)]
+	otel_logs: OtlpLogsSettings,
+	#[clap(flatten)]
+	otel_traces: OtlpTracesSettings,
+}
+
+pub fn setup_logging(opts: &LogOpts) -> 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(
+				"{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("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;
+					};
+					let pos = state.pos();
+					if pos > len {
+						let _ = write!(writer, "{}", pos.human_count_bare());
+					} else {
+						let _ = write!(writer, "{} / {}", pos.human_count_bare(), len.human_count_bare());
+					}
+				})
+				.with_key(
+					"color_start",
+					|state: &ProgressState, writer: &mut dyn 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 fmt::Write| {
+						if state.elapsed() > Duration::from_secs(30) {
+							let _ = write!(writer, "\x1b[0m");
+						}
+					},
+				),
+		)
+	};
+
+	// TODO: Default filter should be configurable
+	let filter =
+		EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("noq_udp=warn,info"));
+
+	let tree = {
+		#[cfg(feature = "indicatif")]
+		let writer = indicatif_layer.get_stderr_writer();
+		#[cfg(not(feature = "indicatif"))]
+		let writer = || std::io::stderr();
+		TreeLayer::new(writer)
+	};
+
+	let reg = tracing_subscriber::registry();
+
+	#[cfg(feature = "tokio-console")]
+	let reg = reg.with(console_subscriber::spawn());
+
+	let reg = reg.with(tree.with_filter(filter));
+
+	#[cfg(feature = "indicatif")]
+	let reg = reg.with(indicatif_layer);
+
+	if opts.otel {
+		let traces = ResolvedOtlpSettings::traces(&opts.otlp_base, &opts.otel_traces)?;
+		let span_exporter = traces.span_exporter()?;
+		let logs = ResolvedOtlpSettings::logs(&opts.otlp_base, &opts.otel_logs)?;
+		let log_exporter = logs.log_exporter()?;
+
+		let span_provider = SdkTracerProvider::builder()
+			.with_batch_exporter(span_exporter)
+			.build();
+		let log_provider = SdkLoggerProvider::builder()
+			.with_batch_exporter(log_exporter)
+			.build();
+
+		let logger = OpenTelemetryTracingBridge::new(&log_provider);
+		let tracer = span_provider.tracer("fleet");
+
+		reg.with(tracing_opentelemetry::layer().with_tracer(tracer))
+			.with(logger)
+			.init();
+	} else {
+		reg.init();
+	}
+
+	Ok(())
+}

--- /dev/null
+++ b/crates/goodlog-subscriber/src/log_tree.rs
@@ -0,0 +1,233 @@
+//! Streaming, indented tree formatter.
+
+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()
+	}
+}
+
+struct SpanData {
+	fields: String,
+	started: Instant,
+	opened: AtomicBool,
+}
+
+fn level_code(level: &Level) -> &'static str {
+	match *level {
+		// Red
+		Level::ERROR => "31m",
+		// Yellow
+		Level::WARN => "33m",
+		// Blue
+		Level::INFO => "34m",
+		// Default fg
+		Level::DEBUG => "39m",
+		// Gray
+		Level::TRACE => "90m",
+	}
+}
+
+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,
+	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![]),
+		}
+	}
+
+	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());
+	}
+}