mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-09-20 13:30:46 +02:00
seaweed-worker: serve health, readiness and metrics (#10848)
* seaweed-worker: serve health, readiness and metrics A Rust worker had no surface of its own. If it wedged, the only signals were its stdout and whatever admin could infer from a stream that had gone quiet; nothing could be scraped and nothing could be alerted on. --metrics-port serves /health, /ready and /metrics, the same three the Go worker serves under -metricsPort, so one scrape config covers workers in either language. Off by default, loopback unless --metrics-ip says otherwise, since the endpoint is unauthenticated. Names follow the Go convention, SeaweedFS_worker_*. The counters live in core and are raised where the stream already knows what happened - connect, close, detection, execution, preview - so a worker for another format gets them without writing any of this. Slots are published from the heartbeat that already computes them, so a scrape and the admin UI cannot disagree. The pair worth having is objects_seen_total and objects_skipped_total. A sweep that proposed nothing because there was nothing to do and a sweep that proposed nothing because it could not read anything are the same number of proposals; they are not the same event, and until now only a log line told them apart. The Lance jobs add what they reclaimed - fragments, rows brought under an index, versions, bytes - on the same registry, so one endpoint serves both. Claude-Session: https://claude.ai/code/session_01Rkp1Mw5E89Jp6dzJFYiMrm * seaweed-worker: fix the metrics address, the count, and a dead field Three from review. --metrics-ip ::1 failed at startup: the address was built by joining host and port with a colon, and "::1:9327" is not an address. It is parsed as a host and combined with SocketAddr::new now, so an IPv6 literal works, with or without the brackets an operator will reasonably type after seeing one in a URL. proposals_total counted before the send rather than after, so a stream that closed mid-sweep left the counter claiming proposals admin never received. And MeteredSender carried a Metrics clone and a job type it never read, kept alive by two statements that existed only to silence the warning about them. Everything is recorded by the caller, so both are gone. Claude-Session: https://claude.ai/code/session_01Rkp1Mw5E89Jp6dzJFYiMrm
This commit is contained in:
Chris Lu
Generated
+65
-7
@@ -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",
|
||||
|
||||
@@ -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"] }
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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]
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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_<subsystem>_<name>`, 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<AtomicBool>,
|
||||
}
|
||||
|
||||
impl Metrics {
|
||||
pub fn new(worker_id: &str, worker_version: &str) -> Result<Self> {
|
||||
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<dyn prometheus::core::Collector>,
|
||||
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<IntCounter> {
|
||||
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<String> {
|
||||
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<Metrics>| async move {
|
||||
if metrics.is_ready() {
|
||||
StatusCode::OK
|
||||
} else {
|
||||
StatusCode::SERVICE_UNAVAILABLE
|
||||
}
|
||||
}),
|
||||
)
|
||||
.route(
|
||||
"/metrics",
|
||||
get(|State(metrics): State<Metrics>| 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"));
|
||||
}
|
||||
}
|
||||
@@ -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());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<Outcome> {
|
||||
// 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::<RunningWork>::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());
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
@@ -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"] }
|
||||
|
||||
@@ -32,6 +32,7 @@ const MAX_MIN_VERSIONS: i64 = 1000;
|
||||
pub struct CleanupVersionsHandler {
|
||||
namespace_url: String,
|
||||
fallback: dataset::FallbackOptions,
|
||||
metrics: Option<crate::metrics::LanceMetrics>,
|
||||
}
|
||||
|
||||
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<crate::metrics::LanceMetrics>) -> 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<String, ConfigValue> = HashMap::new();
|
||||
output.insert(
|
||||
"old_versions_removed".to_string(),
|
||||
|
||||
@@ -35,6 +35,7 @@ const MIN_FRAGMENTS_CEILING: i64 = 4096;
|
||||
pub struct CompactHandler {
|
||||
namespace_url: String,
|
||||
fallback: dataset::FallbackOptions,
|
||||
metrics: Option<crate::metrics::LanceMetrics>,
|
||||
}
|
||||
|
||||
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<crate::metrics::LanceMetrics>) -> 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<String, ConfigValue> = HashMap::new();
|
||||
output.insert(
|
||||
|
||||
@@ -35,6 +35,7 @@ const MAX_UNINDEXED_CEILING: i64 = 100_000_000;
|
||||
pub struct OptimizeIndicesHandler {
|
||||
namespace_url: String,
|
||||
fallback: dataset::FallbackOptions,
|
||||
metrics: Option<crate::metrics::LanceMetrics>,
|
||||
}
|
||||
|
||||
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<crate::metrics::LanceMetrics>) -> 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<String, ConfigValue> = HashMap::new();
|
||||
output.insert(
|
||||
"unindexed_rows_before".to_string(),
|
||||
|
||||
@@ -41,16 +41,24 @@ pub(crate) fn table_id(parameters: &HashMap<String, ConfigValue>) -> Option<Vec<
|
||||
pub fn handlers(
|
||||
namespace_url: String,
|
||||
fallback: crate::dataset::FallbackOptions,
|
||||
metrics: Option<crate::metrics::LanceMetrics>,
|
||||
) -> Vec<Arc<dyn JobHandler>> {
|
||||
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)),
|
||||
]
|
||||
}
|
||||
|
||||
|
||||
@@ -8,6 +8,7 @@
|
||||
pub mod catalog;
|
||||
pub mod dataset;
|
||||
pub mod jobs;
|
||||
pub mod metrics;
|
||||
pub mod preview;
|
||||
|
||||
pub use jobs::handlers;
|
||||
|
||||
@@ -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<String>,
|
||||
|
||||
/// 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<SocketAddr> {
|
||||
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());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<Self> {
|
||||
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.")?,
|
||||
})
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user