diff --git a/seaweed-worker/Cargo.lock b/seaweed-worker/Cargo.lock index aaf222bb4..4fad19e50 100644 --- a/seaweed-worker/Cargo.lock +++ b/seaweed-worker/Cargo.lock @@ -867,6 +867,8 @@ dependencies = [ "http 1.5.0", "http-body 1.1.0", "http-body-util", + "hyper", + "hyper-util", "itoa", "matchit", "memchr", @@ -875,10 +877,15 @@ dependencies = [ "pin-project-lite", "rustversion", "serde", + "serde_json", + "serde_path_to_error", + "serde_urlencoded", "sync_wrapper", + "tokio", "tower 0.5.3", "tower-layer", "tower-service", + "tracing", ] [[package]] @@ -899,6 +906,7 @@ dependencies = [ "sync_wrapper", "tower-layer", "tower-service", + "tracing", ] [[package]] @@ -2120,7 +2128,7 @@ version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "10d60334b3b2e7c9d91ef8150abfb6fa4c1c39ebbcf4a81c2e346aad939fee3e" dependencies = [ - "thiserror", + "thiserror 2.0.20", ] [[package]] @@ -3107,7 +3115,7 @@ dependencies = [ "jni-sys", "log", "simd_cesu8", - "thiserror", + "thiserror 2.0.20", "walkdir", "windows-link", ] @@ -4162,7 +4170,7 @@ dependencies = [ "serde", "serde_json", "serde_urlencoded", - "thiserror", + "thiserror 2.0.20", "tokio", "tracing", "url", @@ -4541,6 +4549,20 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "prometheus" +version = "0.13.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3d33c28a30771f7f96db69893f78b857f7450d7e0237e9c8fc6427a81bae7ed1" +dependencies = [ + "cfg-if", + "fnv", + "lazy_static", + "memchr", + "parking_lot", + "thiserror 1.0.69", +] + [[package]] name = "prost" version = "0.13.5" @@ -4690,7 +4712,7 @@ dependencies = [ "rustc-hash", "rustls", "socket2 0.6.5", - "thiserror", + "thiserror 2.0.20", "tokio", "tracing", "web-time", @@ -4713,7 +4735,7 @@ dependencies = [ "rustls", "rustls-pki-types", "slab", - "thiserror", + "thiserror 2.0.20", "tinyvec", "tracing", "web-time", @@ -4913,7 +4935,7 @@ checksum = "a4e608c6638b9c18977b00b475ac1f28d14e84b27d8d42f70e0bf1e3dec127ac" dependencies = [ "getrandom 0.2.17", "libredox", - "thiserror", + "thiserror 2.0.20", ] [[package]] @@ -5353,6 +5375,8 @@ version = "0.1.0" dependencies = [ "anyhow", "async-trait", + "axum", + "prometheus", "prost 0.13.5", "prost-types 0.13.5", "tokio", @@ -5440,6 +5464,17 @@ dependencies = [ "zmij", ] +[[package]] +name = "serde_path_to_error" +version = "0.1.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "10a9ff822e371bb5403e391ecd83e182e0e77ba7f6fe0160b795797109d1b457" +dependencies = [ + "itoa", + "serde", + "serde_core", +] + [[package]] name = "serde_repr" version = "0.1.21" @@ -5784,13 +5819,33 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "thiserror" +version = "1.0.69" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6aaf5339b578ea85b50e080feb250a3e8ae8cfcdff9a461c9ec2904bc923f52" +dependencies = [ + "thiserror-impl 1.0.69", +] + [[package]] name = "thiserror" version = "2.0.20" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ec86235f5fcc2a73650310756d2ac5b138a5780bbbdfae3eeccec992c435ba4f" dependencies = [ - "thiserror-impl", + "thiserror-impl 2.0.20", +] + +[[package]] +name = "thiserror-impl" +version = "1.0.69" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4fee6c4efc90059e10f81e6d42c60a18f76588c3d74cb83a0b242a2b6c7504c1" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", ] [[package]] @@ -6030,6 +6085,7 @@ dependencies = [ "tokio", "tower-layer", "tower-service", + "tracing", ] [[package]] @@ -6073,6 +6129,7 @@ version = "0.1.44" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100" dependencies = [ + "log", "pin-project-lite", "tracing-attributes", "tracing-core", @@ -6422,6 +6479,7 @@ dependencies = [ "lance", "lance-index", "lance-linalg", + "prometheus", "reqwest 0.12.28", "seaweed-worker-core", "serde", diff --git a/seaweed-worker/Cargo.toml b/seaweed-worker/Cargo.toml index ef33f34e7..b44e9883f 100644 --- a/seaweed-worker/Cargo.toml +++ b/seaweed-worker/Cargo.toml @@ -19,6 +19,9 @@ prost-types = "0.13" tokio = { version = "1", features = ["full"] } tokio-stream = "0.1" tonic = { version = "0.12", features = ["tls"] } +# Already in the tree via tonic; named here so the metrics server can use them. +axum = "0.7" +prometheus = { version = "0.13", default-features = false } tonic-build = "0.12" tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } diff --git a/seaweed-worker/README.md b/seaweed-worker/README.md index 7ef6fbe47..d2c00b564 100644 --- a/seaweed-worker/README.md +++ b/seaweed-worker/README.md @@ -20,6 +20,29 @@ The admin's *HTTP* address is what an operator has; the gRPC port is derived from it the way the Go side does. Dialling the HTTP port fails as "frame with invalid size", which reads like a protocol bug rather than a wrong port. +## Metrics + + cargo run -p weed-lance-worker -- --admin 127.0.0.1:23646 --metrics-port 9327 + +Serves `/health`, `/ready` and `/metrics` on that port, the same three the Go +worker serves under `weed worker -metricsPort`, so one scrape config covers +workers in either language. Off by default, and bound to loopback unless +`--metrics-ip` says otherwise, because the endpoint is unauthenticated. + +Names are `SeaweedFS_worker_*`, matching the Go side's convention. The pair +worth alerting on is `objects_seen_total` and `objects_skipped_total`: a sweep +that proposes nothing and a sweep that could read nothing look identical from +`proposals_total` alone. + + SeaweedFS_worker_connected 1 + SeaweedFS_worker_objects_seen_total{job_type="lance_compact"} 7 + SeaweedFS_worker_proposals_total{job_type="lance_compact"} 2 + SeaweedFS_worker_jobs_total{job_type="lance_compact",result="ok"} 2 + SeaweedFS_worker_lance_fragments_removed_total 25 + +`/ready` follows the control stream: a worker whose admin has gone away is +running but is not going to do anything. + ## Credentials The worker holds none. It asks the namespace to describe a table with diff --git a/seaweed-worker/crates/core/Cargo.toml b/seaweed-worker/crates/core/Cargo.toml index 8627e8a95..2af01bc3e 100644 --- a/seaweed-worker/crates/core/Cargo.toml +++ b/seaweed-worker/crates/core/Cargo.toml @@ -15,6 +15,8 @@ prost-types.workspace = true tokio.workspace = true tokio-stream.workspace = true tonic.workspace = true +axum.workspace = true +prometheus.workspace = true tracing.workspace = true [build-dependencies] diff --git a/seaweed-worker/crates/core/src/lib.rs b/seaweed-worker/crates/core/src/lib.rs index 315d87c59..7cea55daa 100644 --- a/seaweed-worker/crates/core/src/lib.rs +++ b/seaweed-worker/crates/core/src/lib.rs @@ -9,6 +9,7 @@ pub mod address; pub mod config; pub mod config_form; +pub mod metrics; pub mod registry; pub mod senders; pub mod stream; @@ -19,6 +20,7 @@ pub mod pb { } pub use config::{TlsOptions, WorkerOptions}; +pub use metrics::Metrics; pub use registry::{JobHandler, Preview, PreviewProvider, Registry}; pub use senders::{DetectionSender, ExecutionSender}; pub use stream::run; diff --git a/seaweed-worker/crates/core/src/metrics.rs b/seaweed-worker/crates/core/src/metrics.rs new file mode 100644 index 000000000..f3bb9fae7 --- /dev/null +++ b/seaweed-worker/crates/core/src/metrics.rs @@ -0,0 +1,395 @@ +//! What a worker will say about itself when nobody is watching the logs. +//! +//! The Go plugin worker serves `/health`, `/ready` and `/metrics` on an optional +//! port (`weed worker -metricsPort`); this is the same contract, so one scrape +//! config covers workers in either language. Names follow the Go side's +//! convention, `SeaweedFS__`, under a `worker` subsystem that +//! nothing else uses. +//! +//! Nothing here is Lance-specific. A worker for another format registers its own +//! collectors on the same registry and gets the same endpoint. + +use std::net::SocketAddr; +use std::sync::Arc; + +use anyhow::{Context, Result}; +use prometheus::{ + Encoder, HistogramOpts, HistogramVec, IntCounter, IntCounterVec, IntGauge, IntGaugeVec, Opts, + Registry, TextEncoder, +}; +use std::sync::atomic::{AtomicBool, Ordering}; +use tracing::{info, warn}; + +const NAMESPACE: &str = "SeaweedFS"; +const SUBSYSTEM: &str = "worker"; + +/// The metrics one worker process publishes, and the registry they live on. +/// +/// Cloneable because the stream loop, the job handlers and the HTTP server all +/// hold it; everything inside is already shared. +#[derive(Clone)] +pub struct Metrics { + registry: Registry, + connected: IntGauge, + connects: IntCounterVec, + slots_used: IntGaugeVec, + slots_total: IntGaugeVec, + detections: IntCounterVec, + detection_seconds: HistogramVec, + proposals: IntCounterVec, + objects_seen: IntCounterVec, + objects_skipped: IntCounterVec, + jobs: IntCounterVec, + job_seconds: HistogramVec, + previews: IntCounterVec, + ready: Arc, +} + +impl Metrics { + pub fn new(worker_id: &str, worker_version: &str) -> Result { + let registry = Registry::new(); + + // Build info as a constant 1, the way the Go side does it, so a scrape + // can tell which worker and which build answered. + let build = IntGaugeVec::new( + Opts::new("build_info", "Worker build information.") + .namespace(NAMESPACE) + .subsystem(SUBSYSTEM), + &["worker_id", "version"], + )?; + build.with_label_values(&[worker_id, worker_version]).set(1); + registry.register(Box::new(build))?; + + let connected = IntGauge::with_opts( + Opts::new("connected", "1 while the admin control stream is up.") + .namespace(NAMESPACE) + .subsystem(SUBSYSTEM), + )?; + let connects = IntCounterVec::new( + Opts::new( + "stream_events_total", + "Control stream lifecycle events by outcome (connected, closed, failed, shutdown).", + ) + .namespace(NAMESPACE) + .subsystem(SUBSYSTEM), + &["event"], + )?; + let slots_used = IntGaugeVec::new( + Opts::new("slots_used", "Slots currently held, by lane.") + .namespace(NAMESPACE) + .subsystem(SUBSYSTEM), + &["lane"], + )?; + let slots_total = IntGaugeVec::new( + Opts::new("slots_total", "Slots this worker advertises, by lane.") + .namespace(NAMESPACE) + .subsystem(SUBSYSTEM), + &["lane"], + )?; + let detections = IntCounterVec::new( + Opts::new( + "detections_total", + "Detection sweeps by job type and outcome.", + ) + .namespace(NAMESPACE) + .subsystem(SUBSYSTEM), + &["job_type", "result"], + )?; + let detection_seconds = HistogramVec::new( + HistogramOpts::new("detection_seconds", "How long a detection sweep took.") + .namespace(NAMESPACE) + .subsystem(SUBSYSTEM) + .buckets(prometheus::exponential_buckets(0.01, 2.0, 14)?), + &["job_type"], + )?; + let proposals = IntCounterVec::new( + Opts::new("proposals_total", "Jobs proposed to admin, by job type.") + .namespace(NAMESPACE) + .subsystem(SUBSYSTEM), + &["job_type"], + )?; + let objects_seen = IntCounterVec::new( + Opts::new( + "objects_seen_total", + "Objects a detection sweep read, by job type.", + ) + .namespace(NAMESPACE) + .subsystem(SUBSYSTEM), + &["job_type"], + )?; + let objects_skipped = IntCounterVec::new( + Opts::new( + "objects_skipped_total", + "Objects a sweep could not read, by job type and reason. A sweep that \ + proposes nothing looks the same as one that could read nothing; this is \ + what tells them apart.", + ) + .namespace(NAMESPACE) + .subsystem(SUBSYSTEM), + &["job_type", "reason"], + )?; + let jobs = IntCounterVec::new( + Opts::new("jobs_total", "Jobs executed by job type and outcome.") + .namespace(NAMESPACE) + .subsystem(SUBSYSTEM), + &["job_type", "result"], + )?; + let job_seconds = HistogramVec::new( + HistogramOpts::new("job_seconds", "How long a job took to run.") + .namespace(NAMESPACE) + .subsystem(SUBSYSTEM) + .buckets(prometheus::exponential_buckets(0.05, 2.0, 16)?), + &["job_type"], + )?; + let previews = IntCounterVec::new( + Opts::new( + "previews_total", + "Object previews served to admin, by outcome.", + ) + .namespace(NAMESPACE) + .subsystem(SUBSYSTEM), + &["result"], + )?; + + for collector in [ + Box::new(connected.clone()) as Box, + Box::new(connects.clone()), + Box::new(slots_used.clone()), + Box::new(slots_total.clone()), + Box::new(detections.clone()), + Box::new(detection_seconds.clone()), + Box::new(proposals.clone()), + Box::new(objects_seen.clone()), + Box::new(objects_skipped.clone()), + Box::new(jobs.clone()), + Box::new(job_seconds.clone()), + Box::new(previews.clone()), + ] { + registry.register(collector)?; + } + + Ok(Self { + registry, + connected, + connects, + slots_used, + slots_total, + detections, + detection_seconds, + proposals, + objects_seen, + objects_skipped, + jobs, + job_seconds, + previews, + ready: Arc::new(AtomicBool::new(false)), + }) + } + + /// The registry, so a worker can add collectors of its own. + pub fn registry(&self) -> &Registry { + &self.registry + } + + pub fn stream_connected(&self) { + self.connected.set(1); + self.ready.store(true, Ordering::Relaxed); + self.connects.with_label_values(&["connected"]).inc(); + } + + /// `event` is why the stream ended: closed, failed, or shutdown. + pub fn stream_ended(&self, event: &str) { + self.connected.set(0); + self.ready.store(false, Ordering::Relaxed); + self.connects.with_label_values(&[event]).inc(); + } + + pub fn set_slots(&self, lane: &str, used: i64, total: i64) { + self.slots_used.with_label_values(&[lane]).set(used); + self.slots_total.with_label_values(&[lane]).set(total); + } + + pub fn detection_finished(&self, job_type: &str, result: &str, seconds: f64, proposals: usize) { + self.detections + .with_label_values(&[job_type, result]) + .inc_by(1); + self.detection_seconds + .with_label_values(&[job_type]) + .observe(seconds); + self.proposals + .with_label_values(&[job_type]) + .inc_by(proposals as u64); + } + + pub fn object_seen(&self, job_type: &str) { + self.objects_seen.with_label_values(&[job_type]).inc(); + } + + pub fn object_skipped(&self, job_type: &str, reason: &str) { + self.objects_skipped + .with_label_values(&[job_type, reason]) + .inc(); + } + + pub fn job_finished(&self, job_type: &str, result: &str, seconds: f64) { + self.jobs.with_label_values(&[job_type, result]).inc(); + self.job_seconds + .with_label_values(&[job_type]) + .observe(seconds); + } + + pub fn preview_finished(&self, result: &str) { + self.previews.with_label_values(&[result]).inc(); + } + + /// A counter this worker's own jobs can raise, e.g. fragments removed. + /// Registered lazily so a format's numbers live beside the generic ones + /// without core having to know what they are. + pub fn counter(&self, name: &str, help: &str) -> Result { + let counter = IntCounter::with_opts( + Opts::new(name, help) + .namespace(NAMESPACE) + .subsystem(SUBSYSTEM), + )?; + self.registry.register(Box::new(counter.clone()))?; + Ok(counter) + } + + fn gather(&self) -> Result { + let mut buffer = Vec::new(); + TextEncoder::new().encode(&self.registry.gather(), &mut buffer)?; + Ok(String::from_utf8(buffer)?) + } + + fn is_ready(&self) -> bool { + self.ready.load(Ordering::Relaxed) + } +} + +/// Serves the metrics endpoints until the process stops. Failing to bind is +/// logged rather than fatal: a worker that cannot publish metrics should still +/// do its work. +pub async fn serve(metrics: Metrics, addr: SocketAddr) -> Result<()> { + use axum::extract::State; + use axum::http::StatusCode; + use axum::routing::get; + use axum::Router; + + let app = Router::new() + .route("/health", get(|| async { StatusCode::OK })) + .route( + "/ready", + get(|State(metrics): State| async move { + if metrics.is_ready() { + StatusCode::OK + } else { + StatusCode::SERVICE_UNAVAILABLE + } + }), + ) + .route( + "/metrics", + get(|State(metrics): State| async move { + match metrics.gather() { + Ok(body) => (StatusCode::OK, body), + Err(err) => (StatusCode::INTERNAL_SERVER_ERROR, format!("{err:#}")), + } + }), + ) + .with_state(metrics); + + let listener = tokio::net::TcpListener::bind(addr) + .await + .with_context(|| format!("bind the metrics server to {addr}"))?; + info!("serving worker metrics on http://{addr}/metrics"); + axum::serve(listener, app) + .await + .context("metrics server stopped") +} + +/// Starts the metrics server in the background, warning rather than failing when +/// it cannot start. +pub fn spawn(metrics: Metrics, addr: SocketAddr) { + tokio::spawn(async move { + if let Err(err) = serve(metrics, addr).await { + warn!("worker metrics server: {err:#}"); + } + }); +} + +#[cfg(test)] +mod tests { + use super::*; + + fn metrics() -> Metrics { + Metrics::new("worker-1", "0.1.0").expect("build metrics") + } + + // A scrape has to be able to tell these apart: a worker that swept and found + // nothing to do, and one that could not read anything it swept. + #[test] + fn a_sweep_reports_what_it_read_and_what_it_could_not() { + let metrics = metrics(); + metrics.object_seen("lance_compact"); + metrics.object_seen("lance_compact"); + metrics.object_skipped("lance_compact", "open"); + metrics.detection_finished("lance_compact", "ok", 0.5, 0); + + let text = metrics.gather().expect("gather"); + assert!(text.contains("SeaweedFS_worker_objects_seen_total{job_type=\"lance_compact\"} 2")); + assert!(text.contains( + "SeaweedFS_worker_objects_skipped_total{job_type=\"lance_compact\",reason=\"open\"} 1" + )); + assert!(text.contains( + "SeaweedFS_worker_detections_total{job_type=\"lance_compact\",result=\"ok\"} 1" + )); + assert!(text.contains("SeaweedFS_worker_proposals_total{job_type=\"lance_compact\"} 0")); + } + + // /ready follows the stream, because a worker with no admin behind it is + // running but not doing anything. + #[test] + fn readiness_follows_the_stream() { + let metrics = metrics(); + assert!(!metrics.is_ready(), "not ready before the stream is up"); + + metrics.stream_connected(); + assert!(metrics.is_ready()); + assert!(metrics + .gather() + .unwrap() + .contains("SeaweedFS_worker_connected 1")); + + metrics.stream_ended("closed"); + assert!(!metrics.is_ready()); + assert!(metrics + .gather() + .unwrap() + .contains("SeaweedFS_worker_connected 0")); + assert!(metrics + .gather() + .unwrap() + .contains("SeaweedFS_worker_stream_events_total{event=\"closed\"} 1")); + } + + #[test] + fn build_info_names_the_worker() { + let text = metrics().gather().expect("gather"); + assert!(text + .contains("SeaweedFS_worker_build_info{version=\"0.1.0\",worker_id=\"worker-1\"} 1")); + } + + // A format's own numbers land on the same registry, so one endpoint serves + // both and a second worker implementation needs no new plumbing. + #[test] + fn a_worker_can_add_counters_of_its_own() { + let metrics = metrics(); + let counter = metrics + .counter("lance_fragments_removed_total", "Fragments merged away.") + .expect("register"); + counter.inc_by(16); + + let text = metrics.gather().expect("gather"); + assert!(text.contains("SeaweedFS_worker_lance_fragments_removed_total 16")); + } +} diff --git a/seaweed-worker/crates/core/src/senders.rs b/seaweed-worker/crates/core/src/senders.rs index 70f067453..39c5bff7e 100644 --- a/seaweed-worker/crates/core/src/senders.rs +++ b/seaweed-worker/crates/core/src/senders.rs @@ -73,3 +73,131 @@ impl ExecutionSender for StreamSender { self.send(Body::JobCompleted(completed)) } } + +/// Wraps a sender so what passes through it is counted. The handlers report +/// their results to admin and nowhere else, so this is where a scrape can learn +/// what happened without every handler having to know about metrics. +pub struct MeteredSender<'a> { + inner: &'a StreamSender, + proposals: std::sync::atomic::AtomicUsize, + failed: std::sync::atomic::AtomicBool, +} + +impl<'a> MeteredSender<'a> { + pub fn new(inner: &'a StreamSender) -> Self { + Self { + inner, + proposals: std::sync::atomic::AtomicUsize::new(0), + failed: std::sync::atomic::AtomicBool::new(false), + } + } + + /// How many proposals went out, for the detection counter. + pub fn proposals(&self) -> usize { + self.proposals.load(std::sync::atomic::Ordering::Relaxed) + } + + /// Whether the handler reported a failure of its own. A handler that fails + /// by returning an error is counted by the caller; this catches the one that + /// reports failure and returns Ok. + pub fn reported_failure(&self) -> bool { + self.failed.load(std::sync::atomic::Ordering::Relaxed) + } +} + +impl DetectionSender for MeteredSender<'_> { + fn send_proposals(&self, proposals: DetectionProposals) -> Result<()> { + // Counted after the send, not before: a stream that closed mid-sweep + // would otherwise leave proposals_total claiming work admin never saw. + let count = proposals.proposals.len(); + self.inner.send_proposals(proposals)?; + self.proposals + .fetch_add(count, std::sync::atomic::Ordering::Relaxed); + Ok(()) + } + + fn send_complete(&self, complete: DetectionComplete) -> Result<()> { + if !complete.success { + self.failed + .store(true, std::sync::atomic::Ordering::Relaxed); + } + self.inner.send_complete(complete) + } + + fn send_activity(&self, activity: ActivityEvent) -> Result<()> { + self.inner.send_activity(activity) + } + + fn send_observations(&self, observations: WorkerObservations) -> Result<()> { + self.inner.send_observations(observations) + } +} + +impl ExecutionSender for MeteredSender<'_> { + fn send_progress(&self, progress: JobProgressUpdate) -> Result<()> { + self.inner.send_progress(progress) + } + + fn send_completed(&self, completed: JobCompleted) -> Result<()> { + if !completed.success { + self.failed + .store(true, std::sync::atomic::Ordering::Relaxed); + } + self.inner.send_completed(completed) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::pb::JobProposal; + + fn proposals(n: usize) -> DetectionProposals { + DetectionProposals { + proposals: (0..n).map(|_| JobProposal::default()).collect(), + ..Default::default() + } + } + + #[test] + fn proposals_are_counted_once_they_are_sent() { + let (tx, mut rx) = mpsc::unbounded_channel(); + let stream = StreamSender::new("worker-1".to_string(), tx); + let metered = MeteredSender::new(&stream); + + metered.send_proposals(proposals(3)).expect("send"); + assert_eq!(metered.proposals(), 3); + assert!(rx.try_recv().is_ok(), "the proposals reached the stream"); + } + + // Admin never saw these, so counting them would report work that was not + // handed over. + #[test] + fn proposals_are_not_counted_when_the_stream_is_gone() { + let (tx, rx) = mpsc::unbounded_channel(); + let stream = StreamSender::new("worker-1".to_string(), tx); + let metered = MeteredSender::new(&stream); + drop(rx); + + assert!(metered.send_proposals(proposals(3)).is_err()); + assert_eq!(metered.proposals(), 0); + } + + // A handler can report failure and still return Ok; the outcome has to come + // from what it said, not only from what it returned. + #[test] + fn a_reported_failure_is_remembered() { + let (tx, _rx) = mpsc::unbounded_channel(); + let stream = StreamSender::new("worker-1".to_string(), tx); + let metered = MeteredSender::new(&stream); + assert!(!metered.reported_failure()); + + metered + .send_complete(DetectionComplete { + success: false, + ..Default::default() + }) + .expect("send"); + assert!(metered.reported_failure()); + } +} diff --git a/seaweed-worker/crates/core/src/stream.rs b/seaweed-worker/crates/core/src/stream.rs index 054731890..9dc7770f2 100644 --- a/seaweed-worker/crates/core/src/stream.rs +++ b/seaweed-worker/crates/core/src/stream.rs @@ -1,5 +1,5 @@ use std::sync::Arc; -use std::time::Duration; +use std::time::{Duration, Instant}; use anyhow::{anyhow, Context, Result}; use tokio::sync::{mpsc, Semaphore}; @@ -8,6 +8,7 @@ use tonic::transport::{Certificate, Channel, ClientTlsConfig, Identity}; use tracing::{info, warn}; use crate::config::WorkerOptions; +use crate::metrics::Metrics; use crate::pb::{ admin_to_worker_message::Body as AdminBody, plugin_control_service_client::PluginControlServiceClient, @@ -16,7 +17,7 @@ use crate::pb::{ RunningWork, WorkerHeartbeat, WorkerHello, }; use crate::registry::Registry; -use crate::senders::StreamSender; +use crate::senders::{MeteredSender, StreamSender}; /// The protocol version this worker speaks, sent in WorkerHello. const PROTOCOL_VERSION: &str = "1"; @@ -25,6 +26,17 @@ const PROTOCOL_VERSION: &str = "1"; /// reconnecting on failure. The stream is the only channel: everything admin /// asks for and everything the worker reports flows through it. pub async fn run(options: WorkerOptions, registry: Registry) -> Result<()> { + let metrics = Metrics::new(&options.worker_id, &options.worker_version)?; + run_with_metrics(options, registry, metrics).await +} + +/// Runs with metrics a caller has already made, so it can serve them and add +/// collectors of its own before the stream starts. +pub async fn run_with_metrics( + options: WorkerOptions, + registry: Registry, + metrics: Metrics, +) -> Result<()> { if registry.is_empty() { return Err(anyhow!("no job handlers registered")); } @@ -36,20 +48,31 @@ pub async fn run(options: WorkerOptions, registry: Registry) -> Result<()> { )); } let slots = Slots::new(&options); + metrics.set_slots("detection", 0, slots.detection_total as i64); + metrics.set_slots("execution", 0, slots.execution_total as i64); loop { - match serve_once(&options, ®istry, &slots).await { - Err(err) => warn!("worker stream ended: {err:#}"), + match serve_once(&options, ®istry, &slots, &metrics).await { + Err(err) => { + metrics.stream_ended("failed"); + warn!("worker stream ended: {err:#}") + } // Admin asked this worker to stop, so stop. Reconnecting here would // make shutdown impossible: the worker would log back in. - Ok(Outcome::ShutdownRequested) => return Ok(()), + Ok(Outcome::ShutdownRequested) => { + metrics.stream_ended("shutdown"); + return Ok(()); + } // Admin closing a healthy stream is not an error, but reconnecting // in silence hides the reason - two workers sharing an id evict // each other and produce nothing but a login every few seconds. - Ok(Outcome::StreamClosed) => warn!( - "admin closed the stream; reconnecting in {:?}. If this repeats, check for \ - another worker using the id {}", - options.reconnect_delay, options.worker_id - ), + Ok(Outcome::StreamClosed) => { + metrics.stream_ended("closed"); + warn!( + "admin closed the stream; reconnecting in {:?}. If this repeats, check for \ + another worker using the id {}", + options.reconnect_delay, options.worker_id + ) + } } tokio::time::sleep(options.reconnect_delay).await; } @@ -131,6 +154,7 @@ async fn serve_once( options: &WorkerOptions, registry: &Registry, slots: &Slots, + metrics: &Metrics, ) -> Result { // Operators give the admin's HTTP address; the gRPC port is derived, the // same way the Go worker does it. @@ -157,7 +181,12 @@ async fn serve_once( .await? .into_inner(); - let heartbeat = spawn_heartbeat(sender.clone(), options.clone(), slots.clone()); + let heartbeat = spawn_heartbeat( + sender.clone(), + options.clone(), + slots.clone(), + metrics.clone(), + ); while let Some(message) = inbound.message().await? { let request_id = message.request_id.clone(); @@ -166,6 +195,7 @@ async fn serve_once( if !hello.accepted { return Err(anyhow!("admin rejected this worker: {}", hello.message)); } + metrics.stream_connected(); info!( "connected to admin at {} ({})", options.admin_address, grpc_address @@ -194,15 +224,28 @@ async fn serve_once( spawn_preview( registry.clone(), sender.clone(), + metrics.clone(), request_id.clone(), request, ); } Some(AdminBody::RunDetectionRequest(request)) => { - spawn_detection(registry.clone(), sender.clone(), slots.clone(), request); + spawn_detection( + registry.clone(), + sender.clone(), + slots.clone(), + metrics.clone(), + request, + ); } Some(AdminBody::ExecuteJobRequest(request)) => { - spawn_execution(registry.clone(), sender.clone(), slots.clone(), request); + spawn_execution( + registry.clone(), + sender.clone(), + slots.clone(), + metrics.clone(), + request, + ); } Some(AdminBody::CancelRequest(request)) => { // Cancellation needs a per-request handle to be honoured; until @@ -230,6 +273,7 @@ async fn serve_once( fn spawn_preview( registry: Registry, sender: StreamSender, + metrics: Metrics, request_id: String, request: RequestObjectPreview, ) { @@ -263,6 +307,7 @@ fn spawn_preview( }, }, }; + metrics.preview_finished(if response.success { "ok" } else { "failed" }); let _ = sender.send(WorkerBody::ObjectPreviewResponse(response)); }); } @@ -271,6 +316,7 @@ fn spawn_heartbeat( sender: StreamSender, options: WorkerOptions, slots: Slots, + metrics: Metrics, ) -> tokio::task::JoinHandle<()> { // The handle has to be the heartbeat's own, or aborting it aborts nothing // and every reconnect leaves another ticker running. @@ -278,6 +324,18 @@ fn spawn_heartbeat( let mut ticker = tokio::time::interval(options.heartbeat_interval); loop { ticker.tick().await; + // The heartbeat already computes this for admin; publish the same + // numbers so a scrape and the admin UI cannot disagree. + metrics.set_slots( + "detection", + slots.detection_used() as i64, + slots.detection_total as i64, + ); + metrics.set_slots( + "execution", + slots.execution_used() as i64, + slots.execution_total as i64, + ); let beat = WorkerHeartbeat { worker_id: options.worker_id.clone(), running_work: Vec::::new(), @@ -299,6 +357,7 @@ fn spawn_detection( registry: Registry, sender: StreamSender, slots: Slots, + metrics: Metrics, request: RunDetectionRequest, ) { tokio::spawn(async move { @@ -308,7 +367,10 @@ fn spawn_detection( // Held until the sweep finishes, so the worker keeps to the capacity it // advertised and the heartbeat reports the truth while it works. let _permit = slots.detection.acquire().await; - if let Err(err) = handler.detect(&request, &sender).await { + let metered = MeteredSender::new(&sender); + let started = Instant::now(); + let outcome = handler.detect(&request, &metered).await; + let result = if let Err(err) = &outcome { warn!("detection for {} failed: {err:#}", request.job_type); let _ = sender.send(WorkerBody::DetectionComplete( crate::pb::DetectionComplete { @@ -319,7 +381,18 @@ fn spawn_detection( total_proposals: 0, }, )); - } + "failed" + } else if metered.reported_failure() { + "failed" + } else { + "ok" + }; + metrics.detection_finished( + &request.job_type, + result, + started.elapsed().as_secs_f64(), + metered.proposals(), + ); }); } @@ -327,6 +400,7 @@ fn spawn_execution( registry: Registry, sender: StreamSender, slots: Slots, + metrics: Metrics, request: ExecuteJobRequest, ) { tokio::spawn(async move { @@ -344,17 +418,26 @@ fn spawn_execution( let Some(handler) = registry.get(&job_type) else { return; }; - if let Err(err) = handler.execute(&request, &sender).await { + let metered = MeteredSender::new(&sender); + let started = Instant::now(); + let outcome = handler.execute(&request, &metered).await; + let result = if let Err(err) = &outcome { warn!("job {job_id} failed: {err:#}"); let _ = sender.send(WorkerBody::JobCompleted(JobCompleted { request_id: request.request_id.clone(), job_id, - job_type, + job_type: job_type.clone(), success: false, error_message: format!("{err:#}"), ..Default::default() })); - } + "failed" + } else if metered.reported_failure() { + "failed" + } else { + "ok" + }; + metrics.job_finished(&job_type, result, started.elapsed().as_secs_f64()); }); } diff --git a/seaweed-worker/crates/lance/Cargo.toml b/seaweed-worker/crates/lance/Cargo.toml index 6b85eed76..ab59e1e6c 100644 --- a/seaweed-worker/crates/lance/Cargo.toml +++ b/seaweed-worker/crates/lance/Cargo.toml @@ -13,6 +13,7 @@ path = "src/main.rs" [dependencies] seaweed-worker-core = { path = "../core" } +prometheus.workspace = true # Only the S3 backend: the other object stores lance enables by default are # build time this worker never spends. lance = { version = "10", default-features = false, features = ["aws"] } diff --git a/seaweed-worker/crates/lance/src/jobs/cleanup.rs b/seaweed-worker/crates/lance/src/jobs/cleanup.rs index b24e4642a..84351e8b2 100644 --- a/seaweed-worker/crates/lance/src/jobs/cleanup.rs +++ b/seaweed-worker/crates/lance/src/jobs/cleanup.rs @@ -32,6 +32,7 @@ const MAX_MIN_VERSIONS: i64 = 1000; pub struct CleanupVersionsHandler { namespace_url: String, fallback: dataset::FallbackOptions, + metrics: Option, } impl CleanupVersionsHandler { @@ -39,9 +40,15 @@ impl CleanupVersionsHandler { Self { namespace_url, fallback: dataset::FallbackOptions::new(), + metrics: None, } } + pub fn with_metrics(mut self, metrics: Option) -> Self { + self.metrics = metrics; + self + } + pub fn with_fallback(mut self, fallback: dataset::FallbackOptions) -> Self { self.fallback = fallback; self @@ -147,13 +154,22 @@ impl JobHandler for CleanupVersionsHandler { let table = match dataset::open(&client, &id, &self.fallback).await { Ok(table) => table, Err(err) => { + if let Some(counters) = &self.metrics { + counters.worker.object_skipped(JOB_TYPE, "open"); + } warn!("skipping {encoded}: {err:#}"); continue; } }; + if let Some(counters) = &self.metrics { + counters.worker.object_seen(JOB_TYPE); + } let stats = match table.stats().await { Ok(stats) => stats, Err(err) => { + if let Some(counters) = &self.metrics { + counters.worker.object_skipped(JOB_TYPE, "stats"); + } warn!("skipping {encoded}: reading its stats failed: {err:#}"); continue; } @@ -281,6 +297,11 @@ impl JobHandler for CleanupVersionsHandler { .await .with_context(|| format!("clean up versions of {}", table.location))?; + if let Some(counters) = &self.metrics { + counters.versions_removed.inc_by(stats.old_versions as u64); + counters.bytes_reclaimed.inc_by(stats.bytes_removed as u64); + } + let mut output: HashMap = HashMap::new(); output.insert( "old_versions_removed".to_string(), diff --git a/seaweed-worker/crates/lance/src/jobs/compact.rs b/seaweed-worker/crates/lance/src/jobs/compact.rs index 5fc6333d3..e20bd3b6a 100644 --- a/seaweed-worker/crates/lance/src/jobs/compact.rs +++ b/seaweed-worker/crates/lance/src/jobs/compact.rs @@ -35,6 +35,7 @@ const MIN_FRAGMENTS_CEILING: i64 = 4096; pub struct CompactHandler { namespace_url: String, fallback: dataset::FallbackOptions, + metrics: Option, } impl CompactHandler { @@ -42,9 +43,15 @@ impl CompactHandler { Self { namespace_url, fallback: dataset::FallbackOptions::new(), + metrics: None, } } + pub fn with_metrics(mut self, metrics: Option) -> Self { + self.metrics = metrics; + self + } + /// Storage options to use where the namespace vends none. pub fn with_fallback(mut self, fallback: dataset::FallbackOptions) -> Self { self.fallback = fallback; @@ -143,6 +150,9 @@ impl JobHandler for CompactHandler { Err(err) => { // A table that cannot be opened is the next run's problem, // not a reason to abandon the whole sweep. + if let Some(counters) = &self.metrics { + counters.worker.object_skipped(JOB_TYPE, "open"); + } warn!("skipping {encoded}: {err:#}"); continue; } @@ -150,9 +160,15 @@ impl JobHandler for CompactHandler { // One unreadable table must not end the sweep: the tables already // read would lose their proposals, and admin would get no // completion for this request at all. + if let Some(counters) = &self.metrics { + counters.worker.object_seen(JOB_TYPE); + } let stats = match table.stats().await { Ok(stats) => stats, Err(err) => { + if let Some(counters) = &self.metrics { + counters.worker.object_skipped(JOB_TYPE, "stats"); + } warn!("skipping {encoded}: reading its stats failed: {err:#}"); continue; } @@ -264,6 +280,12 @@ impl JobHandler for CompactHandler { .await .with_context(|| format!("compact {}", table.location))?; + if let Some(counters) = &self.metrics { + counters + .fragments_removed + .inc_by(metrics.fragments_removed as u64); + } + let after = table.stats().await?; let mut output: HashMap = HashMap::new(); output.insert( diff --git a/seaweed-worker/crates/lance/src/jobs/indices.rs b/seaweed-worker/crates/lance/src/jobs/indices.rs index 5c8d70391..a3c23c61c 100644 --- a/seaweed-worker/crates/lance/src/jobs/indices.rs +++ b/seaweed-worker/crates/lance/src/jobs/indices.rs @@ -35,6 +35,7 @@ const MAX_UNINDEXED_CEILING: i64 = 100_000_000; pub struct OptimizeIndicesHandler { namespace_url: String, fallback: dataset::FallbackOptions, + metrics: Option, } impl OptimizeIndicesHandler { @@ -42,9 +43,15 @@ impl OptimizeIndicesHandler { Self { namespace_url, fallback: dataset::FallbackOptions::new(), + metrics: None, } } + pub fn with_metrics(mut self, metrics: Option) -> Self { + self.metrics = metrics; + self + } + pub fn with_fallback(mut self, fallback: dataset::FallbackOptions) -> Self { self.fallback = fallback; self @@ -149,14 +156,23 @@ impl JobHandler for OptimizeIndicesHandler { let table = match dataset::open(&client, &id, &self.fallback).await { Ok(table) => table, Err(err) => { + if let Some(counters) = &self.metrics { + counters.worker.object_skipped(JOB_TYPE, "open"); + } warn!("skipping {encoded}: {err:#}"); continue; } }; + if let Some(counters) = &self.metrics { + counters.worker.object_seen(JOB_TYPE); + } let unindexed = match unindexed_rows(&table).await { Ok(Some(unindexed)) => unindexed, Ok(None) => continue, Err(err) => { + if let Some(counters) = &self.metrics { + counters.worker.object_skipped(JOB_TYPE, "index_stats"); + } warn!("skipping {encoded}: reading its index stats failed: {err:#}"); continue; } @@ -232,6 +248,9 @@ impl JobHandler for OptimizeIndicesHandler { .with_context(|| format!("optimize indices of {}", table.location))?; let after = unindexed_rows(&table).await?.unwrap_or(0); + if let Some(counters) = &self.metrics { + counters.rows_indexed.inc_by(before.saturating_sub(after)); + } let mut output: HashMap = HashMap::new(); output.insert( "unindexed_rows_before".to_string(), diff --git a/seaweed-worker/crates/lance/src/jobs/mod.rs b/seaweed-worker/crates/lance/src/jobs/mod.rs index ae995f7e3..08cbdaa7e 100644 --- a/seaweed-worker/crates/lance/src/jobs/mod.rs +++ b/seaweed-worker/crates/lance/src/jobs/mod.rs @@ -41,16 +41,24 @@ pub(crate) fn table_id(parameters: &HashMap) -> Option, ) -> Vec> { vec![ Arc::new( - compact::CompactHandler::new(namespace_url.clone()).with_fallback(fallback.clone()), + compact::CompactHandler::new(namespace_url.clone()) + .with_fallback(fallback.clone()) + .with_metrics(metrics.clone()), ), Arc::new( indices::OptimizeIndicesHandler::new(namespace_url.clone()) - .with_fallback(fallback.clone()), + .with_fallback(fallback.clone()) + .with_metrics(metrics.clone()), + ), + Arc::new( + cleanup::CleanupVersionsHandler::new(namespace_url) + .with_fallback(fallback) + .with_metrics(metrics), ), - Arc::new(cleanup::CleanupVersionsHandler::new(namespace_url).with_fallback(fallback)), ] } diff --git a/seaweed-worker/crates/lance/src/lib.rs b/seaweed-worker/crates/lance/src/lib.rs index ee8b88f39..be5da0f7b 100644 --- a/seaweed-worker/crates/lance/src/lib.rs +++ b/seaweed-worker/crates/lance/src/lib.rs @@ -8,6 +8,7 @@ pub mod catalog; pub mod dataset; pub mod jobs; +pub mod metrics; pub mod preview; pub use jobs::handlers; diff --git a/seaweed-worker/crates/lance/src/main.rs b/seaweed-worker/crates/lance/src/main.rs index 6bc42a5bd..41bb1109b 100644 --- a/seaweed-worker/crates/lance/src/main.rs +++ b/seaweed-worker/crates/lance/src/main.rs @@ -1,11 +1,13 @@ use std::time::Duration; -use anyhow::Result; +use anyhow::{Context, Result}; use clap::Parser; +use std::net::{IpAddr, SocketAddr}; use std::sync::Arc; -use seaweed_worker_core::{Registry, TlsOptions, WorkerOptions}; +use seaweed_worker_core::{Metrics, Registry, TlsOptions, WorkerOptions}; use weed_lance_worker::handlers; +use weed_lance_worker::metrics::LanceMetrics; /// Mirrors `weed worker`'s flags, because this is the same contract from another /// language and an operator should not have to learn a second set of names. @@ -61,6 +63,16 @@ struct Args { /// this worker dials. #[arg(long)] tls_server_name: Option, + + /// Serve /health, /ready and /metrics on this port, the way + /// `weed worker -metricsPort` does. Zero, the default, serves nothing. + #[arg(long, default_value = "0", env = "WEED_METRICS_PORT")] + metrics_port: u16, + + /// Address the metrics server binds. Loopback by default, since the + /// endpoint is unauthenticated. + #[arg(long, default_value = "127.0.0.1", env = "WEED_METRICS_IP")] + metrics_ip: String, } impl Args { @@ -89,6 +101,19 @@ impl Args { } } +/// Builds the metrics bind address. Parsing the host separately is what makes an +/// IPv6 literal work: "::1" and 9327 joined with a colon is not an address, and +/// formatting them that way turns `--metrics-ip ::1` into a startup failure. +/// Brackets are accepted too, since that is how the same address is written in a +/// URL and an operator will reasonably try it. +fn metrics_address(ip: &str, port: u16) -> Result { + let host = ip.trim().trim_start_matches('[').trim_end_matches(']'); + let parsed: IpAddr = host + .parse() + .with_context(|| format!("parse the metrics address {ip}"))?; + Ok(SocketAddr::new(parsed, port)) +} + #[tokio::main] async fn main() -> Result<()> { tracing_subscriber::fmt() @@ -116,12 +141,54 @@ async fn main() -> Result<()> { fallback.insert("aws_secret_access_key".to_string(), secret); } + let metrics = Metrics::new(&options.worker_id, &options.worker_version)?; + let lance_metrics = LanceMetrics::new(&metrics)?; + if args.metrics_port > 0 { + seaweed_worker_core::metrics::spawn( + metrics.clone(), + metrics_address(&args.metrics_ip, args.metrics_port)?, + ); + } + let mut registry = Registry::new().with_preview(Arc::new( weed_lance_worker::preview::LancePreview::new(args.namespace.clone(), fallback.clone()), )); - for handler in handlers(args.namespace, fallback) { + for handler in handlers(args.namespace, fallback, Some(lance_metrics)) { registry = registry.register(handler); } - seaweed_worker_core::run(options, registry).await + seaweed_worker_core::stream::run_with_metrics(options, registry, metrics).await +} + +#[cfg(test)] +mod tests { + use super::metrics_address; + + #[test] + fn metrics_address_takes_ipv4_ipv6_and_brackets() { + assert_eq!( + metrics_address("127.0.0.1", 9327).unwrap().to_string(), + "127.0.0.1:9327" + ); + // Joining these with a colon gives "::1:9327", which is not an address. + assert_eq!( + metrics_address("::1", 9327).unwrap().to_string(), + "[::1]:9327" + ); + assert_eq!( + metrics_address("[::1]", 9327).unwrap().to_string(), + "[::1]:9327" + ); + assert_eq!( + metrics_address("0.0.0.0", 9327).unwrap().to_string(), + "0.0.0.0:9327" + ); + } + + #[test] + fn metrics_address_rejects_a_hostname() { + // Binding takes an address, not a name; saying so beats a confusing + // failure inside the server. + assert!(metrics_address("localhost", 9327).is_err()); + } } diff --git a/seaweed-worker/crates/lance/src/metrics.rs b/seaweed-worker/crates/lance/src/metrics.rs new file mode 100644 index 000000000..4eab963fb --- /dev/null +++ b/seaweed-worker/crates/lance/src/metrics.rs @@ -0,0 +1,45 @@ +//! What the Lance jobs reclaimed, in numbers a scrape can add up. +//! +//! The generic worker metrics say a job ran and how long it took. These say what +//! it did: a compaction sweep that runs every minute and removes nothing is a +//! different thing from one that never runs, and only these tell them apart. + +use anyhow::Result; +use prometheus::IntCounter; +use seaweed_worker_core::Metrics; + +/// Counters the Lance jobs raise, registered on the worker's own registry so +/// they are served from the same endpoint. +#[derive(Clone)] +pub struct LanceMetrics { + /// The worker's own metrics, so a job can also record what it read and what + /// it had to skip - the pair that tells "nothing needed doing" apart from + /// "nothing could be read". + pub worker: Metrics, + pub fragments_removed: IntCounter, + pub rows_indexed: IntCounter, + pub versions_removed: IntCounter, + pub bytes_reclaimed: IntCounter, +} + +impl LanceMetrics { + pub fn new(metrics: &Metrics) -> Result { + Ok(Self { + worker: metrics.clone(), + fragments_removed: metrics.counter( + "lance_fragments_removed_total", + "Fragments merged away by compaction.", + )?, + rows_indexed: metrics.counter( + "lance_rows_indexed_total", + "Rows brought under an index that did not cover them.", + )?, + versions_removed: metrics.counter( + "lance_versions_removed_total", + "Dataset versions removed by cleanup.", + )?, + bytes_reclaimed: metrics + .counter("lance_bytes_reclaimed_total", "Bytes freed by cleanup.")?, + }) + } +}