diff --git a/Cargo.lock b/Cargo.lock index 9248a104..9bb6057b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -780,6 +780,7 @@ dependencies = [ "tokio", "tokio-util", "toml", + "tower", "tracing", "tracing-subscriber", "trusted-proxies", @@ -838,6 +839,8 @@ dependencies = [ "futures", "http", "jacquard-common", + "opentelemetry", + "opentelemetry_sdk", "reqwest 0.13.1", "scc", "serde", diff --git a/Cargo.toml b/Cargo.toml index 54885355..016c9bad 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -208,9 +208,9 @@ tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt", "json"] metrics = "0.24" metrics-exporter-prometheus = { version = "0.18", default-features = false } -opentelemetry = { version = "0.33", default-features = false, features = ["logs"] } -opentelemetry_sdk = { version = "0.33", default-features = false, features = ["logs"] } -opentelemetry-otlp = { version = "0.33", default-features = false, features = ["http-proto", "reqwest-blocking-client", "logs"] } +opentelemetry = { version = "0.33", default-features = false, features = ["logs", "metrics"] } +opentelemetry_sdk = { version = "0.33", default-features = false, features = ["logs", "metrics"] } +opentelemetry-otlp = { version = "0.33", default-features = false, features = ["http-proto", "reqwest-blocking-client", "logs", "metrics"] } opentelemetry-appender-tracing = { version = "0.33", default-features = false, features = ["experimental_span_attributes"] } thiserror = "2" diff --git a/bobbin/crates/bobbin/Cargo.toml b/bobbin/crates/bobbin/Cargo.toml index adb6d200..4a23d0e6 100644 --- a/bobbin/crates/bobbin/Cargo.toml +++ b/bobbin/crates/bobbin/Cargo.toml @@ -34,6 +34,7 @@ tokio-util = { workspace = true } url = { workspace = true } tracing = { workspace = true } tracing-subscriber = { workspace = true } +opentelemetry = { workspace = true } opentelemetry_sdk = { workspace = true } opentelemetry-otlp = { workspace = true, features = ["internal-logs"] } opentelemetry-appender-tracing = { workspace = true } @@ -47,6 +48,6 @@ tikv-jemallocator = { version = "0.7", features = ["unprefixed_malloc_on_support tikv-jemalloc-ctl = { version = "0.7", features = ["stats", "profiling"] } [dev-dependencies] -opentelemetry = { workspace = true } opentelemetry_sdk = { workspace = true, features = ["testing"] } tokio = { workspace = true, features = ["test-util"] } +tower = { workspace = true } diff --git a/bobbin/crates/bobbin/src/config.rs b/bobbin/crates/bobbin/src/config.rs index 2538acae..416347a5 100644 --- a/bobbin/crates/bobbin/src/config.rs +++ b/bobbin/crates/bobbin/src/config.rs @@ -382,7 +382,7 @@ pub struct LogConfig { #[derive(Debug, Config)] pub struct OtlpConfig { - /// base url of an otlp/http collector for logs. + /// base url of an otlp/http collector for logs and metrics. /// unset disables opentelemetry export. #[config(env = "BOBBIN_OTLP_ENDPOINT")] pub endpoint: Option, diff --git a/bobbin/crates/bobbin/src/main.rs b/bobbin/crates/bobbin/src/main.rs index 6c62054b..cf29399b 100644 --- a/bobbin/crates/bobbin/src/main.rs +++ b/bobbin/crates/bobbin/src/main.rs @@ -39,6 +39,7 @@ use tracing_subscriber::{EnvFilter, layer::SubscriberExt as _, util::SubscriberI mod config; mod mem; +mod metrics; mod otlp; use config::{BobbinConfig, LogFormat}; @@ -95,7 +96,7 @@ async fn main() -> ExitCode { } }; - let export = match init_tracing(&cfg, Arc::new(SystemClock::new())) { + let export = match init_telemetry(&cfg, Arc::new(SystemClock::new())) { Ok(export) => export, Err(e) => { eprintln!("failed to install tracing subscriber: {e}"); @@ -133,10 +134,10 @@ impl tracing_subscriber::fmt::time::FormatTime for ClockTimer { } } -fn init_tracing( +fn init_telemetry( cfg: &BobbinConfig, clock: Arc, -) -> Result, String> { +) -> Result, String> { let combined = format!("{},{}", LevelFilter::INFO, cfg.log.filter); let filter = EnvFilter::try_new(&combined) .map_err(|e| format!("invalid log filter `{}`: {e}", cfg.log.filter))?; @@ -152,15 +153,18 @@ fn init_tracing( Some(tracing_subscriber::fmt::layer().json().with_timer(timer)), ), }; - let export = otlp::LogExport::from_config(&cfg.otlp)?; + let export = otlp::Export::from_config(&cfg.otlp)?; tracing_subscriber::registry() .with(filter) .with(text) .with(json) - .with(export.as_ref().map(otlp::LogExport::layer)) + .with(export.as_ref().map(otlp::Export::layer)) .try_init() .map_err(|e| e.to_string())?; + if let Some(export) = &export { + export.install_meters(); + } Ok(export) } @@ -245,6 +249,16 @@ async fn run(cfg: BobbinConfig) -> anyhow::Result<()> { let stream_health = Arc::new(StreamHealth::new()); let settlements = Arc::new(Settlements::new()); let warming_buffer = Arc::new(WarmingBuffer::new(hasher.clone())); + metrics::observe(metrics::Observed { + resolver: resolver.clone(), + identity: identity.clone(), + records: records.clone(), + coverage: coverage.clone(), + warming: warming_buffer.clone(), + stream: stream_health.clone(), + limiter: limiter.clone(), + budget, + }); let knot_registry = Arc::new(KnotRegistry::new()); let knots = Arc::new(KnotProxy::new( KnotProxyConfig { @@ -496,7 +510,7 @@ async fn run(cfg: BobbinConfig) -> anyhow::Result<()> { .with_settlements(settlements) .with_invites(invites) .with_stream_health(stream_health); - let app = router(state); + let app = router(state).layer(axum::middleware::from_fn(metrics::xrpc)); let _debug_server = match (debug_bind, mem_probe) { (Some(addr), Some(probe)) => { diff --git a/bobbin/crates/bobbin/src/mem/mod.rs b/bobbin/crates/bobbin/src/mem/mod.rs index e23ba39f..4e9cd38f 100644 --- a/bobbin/crates/bobbin/src/mem/mod.rs +++ b/bobbin/crates/bobbin/src/mem/mod.rs @@ -3,7 +3,7 @@ mod report; mod sizing; mod watcher; -pub use cgroup::{detect_budget, try_set_high}; +pub use cgroup::{detect_budget, read_current, try_set_high}; pub use report::{MemProbe, debug_router}; pub use sizing::{ingest_parallelism, lru_bytes, search_heap_bytes}; pub use watcher::{AdaptiveThresholds, spawn as spawn_adaptive_watcher}; diff --git a/bobbin/crates/bobbin/src/metrics.rs b/bobbin/crates/bobbin/src/metrics.rs new file mode 100644 index 00000000..18faac4b --- /dev/null +++ b/bobbin/crates/bobbin/src/metrics.rs @@ -0,0 +1,286 @@ +use std::sync::{Arc, LazyLock}; +use std::time::Instant; + +use axum::extract::{MatchedPath, Request}; +use axum::middleware::Next; +use axum::response::Response; +use bobbin_edge_index::{CoverageWatch, StreamHealth, StreamState}; +use bobbin_ingest::{RepoIdResolver, WarmingBuffer}; +use bobbin_record_lru::RecordStore; +use bobbin_resolver::IdentityResolver; +use bobbin_runtime::MemoryBudget; +use bobbin_xrpc::HeavyLimiter; +use opentelemetry::metrics::{Counter, Histogram, Meter, UpDownCounter}; +use opentelemetry::{KeyValue, global}; + +mod process; + +// the sdk default buckets assume milliseconds, so seconds get a 1, 2.5, 5 ladder from 100µs +// to 50s instead +const DURATION_BOUNDARIES: [f64; 18] = ladder(-4); + +// dividing by an exact power of ten rounds to the same float as the decimal literal, so the +// `le` labels come out clean +const fn ladder(lowest_exponent: i32) -> [f64; N] { + const STEPS: [f64; 3] = [1.0, 2.5, 5.0]; + let mut out = [0.0; N]; + let mut i = 0; + while i < N { + let exponent = lowest_exponent + (i / 3) as i32; + let mut scale = 1.0; + let mut k = 0; + while k < exponent.unsigned_abs() { + scale *= 10.0; + k += 1; + } + out[i] = if exponent < 0 { + STEPS[i % 3] / scale + } else { + STEPS[i % 3] * scale + }; + i += 1; + } + out +} + +struct Xrpc { + requests: Counter, + duration: Histogram, + in_flight: UpDownCounter, +} + +static XRPC: LazyLock = LazyLock::new(|| { + let meter = meter(); + Xrpc { + requests: meter + .u64_counter("bobbin_xrpc_requests") + .with_description("xrpc requests by route and status") + .build(), + duration: meter + .f64_histogram("bobbin_xrpc_duration") + .with_unit("s") + .with_description("xrpc request duration by route") + .with_boundaries(DURATION_BOUNDARIES.to_vec()) + .build(), + in_flight: meter + .i64_up_down_counter("bobbin_xrpc_in_flight") + .with_description("xrpc requests being handled right now") + .build(), + } +}); + +fn meter() -> Meter { + global::meter("bobbin") +} + +// decrements on drop so cancelled or timed out requests don't leak in-flight counts +struct InFlight(KeyValue); + +impl Drop for InFlight { + fn drop(&mut self) { + XRPC.in_flight.add(-1, std::slice::from_ref(&self.0)); + } +} + +pub async fn xrpc(request: Request, next: Next) -> Response { + let route = request + .extensions() + .get::() + .map_or("unmatched".to_owned(), |path| path.as_str().to_owned()); + let route = KeyValue::new("route", route); + XRPC.in_flight.add(1, std::slice::from_ref(&route)); + let _in_flight = InFlight(route.clone()); + let start = Instant::now(); + let response = next.run(request).await; + let status = KeyValue::new("status", i64::from(response.status().as_u16())); + XRPC.requests.add(1, &[route.clone(), status]); + XRPC.duration + .record(start.elapsed().as_secs_f64(), &[route]); + response +} + +pub struct Observed { + pub resolver: Arc, + pub identity: Arc, + pub records: Arc, + pub coverage: Arc, + pub warming: Arc, + pub stream: Arc, + pub limiter: Option>, + pub budget: Option, +} + +pub fn observe(observed: Observed) { + let meter = meter(); + let Observed { + resolver, + identity, + records, + coverage, + warming, + stream, + limiter, + budget, + } = observed; + process::observe(&meter, budget); + if let Some(limiter) = limiter { + observe_limiter(&meter, limiter); + } + + let latency = resolver.clone(); + meter + .u64_observable_counter("bobbin_resolver_lookups") + .with_description("repo did lookups by result") + .with_callback(move |o| { + let s = resolver.stats(); + for (result, value) in [ + ("hit", s.hits), + ("mapped", s.misses_mapped), + ("no_repo_did", s.misses_no_repo_did), + ("unresolvable", s.misses_unresolvable), + ("transient", s.misses_transient), + ("no_client", s.misses_no_client), + ] { + o.observe(value, &[KeyValue::new("result", result)]); + } + }) + .build(); + meter + .f64_observable_counter("bobbin_resolver_miss_latency") + .with_unit("s") + .with_description("time spent resolving repo did misses upstream") + .with_callback(move |o| { + o.observe(latency.stats().miss_latency_micros_sum as f64 / 1e6, &[]); + }) + .build(); + + let lookups = identity.clone(); + meter + .u64_observable_counter("bobbin_identity_lookups") + .with_description("identity cache lookups by result") + .with_callback(move |o| { + let s = lookups.stats(); + o.observe(s.hits, &[KeyValue::new("result", "hit")]); + o.observe(s.misses, &[KeyValue::new("result", "miss")]); + }) + .build(); + let warm = identity.clone(); + meter + .u64_observable_counter("bobbin_identity_warm") + .with_description("background identity warming by result") + .with_callback(move |o| { + let s = warm.stats(); + for (result, value) in [ + ("queued", s.warm_queued), + ("dropped", s.warm_dropped), + ("resolved", s.warm_resolved), + ("failed", s.warm_failed), + ] { + o.observe(value, &[KeyValue::new("result", result)]); + } + }) + .build(); + meter + .u64_observable_gauge("bobbin_identity_entries") + .with_description("identities held in the cache") + .with_callback(move |o| { + o.observe( + u64::try_from(identity.stats().entries).unwrap_or(u64::MAX), + &[], + ); + }) + .build(); + + let cache = records.clone(); + meter + .u64_observable_gauge("bobbin_record_cache_size") + .with_unit("By") + .with_description("bytes held by the record cache, next to its capacity") + .with_callback(move |o| { + if let Some(s) = cache.cache_stats() { + o.observe(s.weight, &[KeyValue::new("kind", "used")]); + o.observe(s.capacity, &[KeyValue::new("kind", "capacity")]); + } + }) + .build(); + meter + .u64_observable_gauge("bobbin_record_cache_entries") + .with_description("records held by the record cache") + .with_callback(move |o| { + if let Some(s) = records.cache_stats() { + o.observe(s.len, &[]); + } + }) + .build(); + + let ready = coverage.clone(); + meter + .u64_observable_gauge("bobbin_ingest_ready") + .with_description("1 once ingest has caught up with the firehose") + .with_callback(move |o| o.observe(u64::from(ready.snapshot().is_ready()), &[])) + .build(); + let cursor = coverage.clone(); + meter + .u64_observable_gauge("bobbin_ingest_cursor") + .with_description("last hydrant cursor ingest committed") + .with_callback(move |o| o.observe(cursor.snapshot().last_cursor().raw(), &[])) + .build(); + meter + .u64_observable_counter("bobbin_ingest_events") + .with_description("hydrant events ingest has committed") + .with_callback(move |o| o.observe(coverage.snapshot().events_processed(), &[])) + .build(); + meter + .u64_observable_gauge("bobbin_ingest_connected") + .with_description("1 while the hydrant stream is connected") + .with_callback(move |o| { + o.observe( + u64::from(stream.snapshot().state == StreamState::Connected), + &[], + ); + }) + .build(); + + let parked = warming.clone(); + meter + .u64_observable_gauge("bobbin_ingest_parked") + .with_description("upserts parked until their repo dids resolve") + .with_callback(move |o| o.observe(parked.snapshot().current_entries, &[])) + .build(); + meter + .u64_observable_counter("bobbin_ingest_parked_outcomes") + .with_description("what happened to parked upserts") + .with_callback(move |o| { + let s = warming.snapshot(); + for (outcome, value) in [ + ("parked", s.enqueued_total), + ("drained_observe", s.drained_observe_total), + ("drained_promote", s.drained_promote_total), + ("evicted", s.evicted_total), + ("rejected_after_seal", s.rejected_after_seal), + ] { + o.observe(value, &[KeyValue::new("outcome", outcome)]); + } + }) + .build(); +} + +fn observe_limiter(meter: &Meter, limiter: Arc) { + let usage = limiter.clone(); + meter + .u64_observable_gauge("bobbin_heavy_requests") + .with_description("heavy requests in flight, next to the limit memory pressure sets") + .with_callback(move |o| { + o.observe( + usage.in_flight() as u64, + &[KeyValue::new("kind", "in_flight")], + ); + o.observe(usage.limit() as u64, &[KeyValue::new("kind", "limit")]); + }) + .build(); + meter + .u64_observable_counter("bobbin_heavy_rejected") + .with_description("heavy requests shed as overloaded") + .with_callback(move |o| o.observe(limiter.rejected(), &[])) + .build(); +} diff --git a/bobbin/crates/bobbin/src/metrics/process.rs b/bobbin/crates/bobbin/src/metrics/process.rs new file mode 100644 index 00000000..5a43b19f --- /dev/null +++ b/bobbin/crates/bobbin/src/metrics/process.rs @@ -0,0 +1,140 @@ +use std::fs; +use std::time::{SystemTime, UNIX_EPOCH}; + +use bobbin_runtime::MemoryBudget; +use opentelemetry::KeyValue; +use opentelemetry::metrics::Meter; +use tikv_jemalloc_ctl::{epoch, stats}; + +use crate::mem; + +// standard linux user_hz for converting `utime` and `stime` into seconds +const TICKS_PER_SECOND: f64 = 100.0; + +pub(super) fn observe(meter: &Meter, budget: Option) { + let started = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_or(0.0, |since| since.as_secs_f64()); + meter + .f64_observable_gauge("process_start_time") + .with_unit("s") + .with_description("unix time the process started") + .with_callback(move |o| o.observe(started, &[])) + .build(); + meter + .f64_observable_counter("process_cpu") + .with_unit("s") + .with_description("user and system cpu time") + .with_callback(|o| { + if let Some(seconds) = cpu_seconds() { + o.observe(seconds, &[]); + } + }) + .build(); + meter + .u64_observable_gauge("process_resident_memory") + .with_unit("By") + .with_description("resident set size") + .with_callback(|o| { + if let Some(kib) = status_field("VmRSS:") { + o.observe(kib * 1024, &[]); + } + }) + .build(); + meter + .u64_observable_gauge("process_threads") + .with_description("os threads") + .with_callback(|o| { + if let Some(threads) = status_field("Threads:") { + o.observe(threads, &[]); + } + }) + .build(); + meter + .u64_observable_gauge("process_open_fds") + .with_description("open file descriptors") + .with_callback(|o| { + if let Ok(fds) = fs::read_dir("/proc/self/fd") { + o.observe(fds.count() as u64, &[]); + } + }) + .build(); + + meter + .u64_observable_gauge("bobbin_jemalloc") + .with_unit("By") + .with_description("jemalloc memory by kind") + .with_callback(|o| { + // jemalloc stats are cached until the epoch is advanced + if epoch::advance().is_err() { + return; + } + for (kind, bytes) in [ + ("allocated", stats::allocated::read()), + ("active", stats::active::read()), + ("resident", stats::resident::read()), + ("retained", stats::retained::read()), + ("mapped", stats::mapped::read()), + ] { + if let Ok(bytes) = bytes { + o.observe(bytes as u64, &[KeyValue::new("kind", kind)]); + } + } + }) + .build(); + meter + .u64_observable_gauge("bobbin_memory") + .with_unit("By") + .with_description("cgroup memory in use, next to the budget bobbin sized itself to") + .with_callback(move |o| { + if let Some(current) = mem::read_current() { + o.observe(current.bytes(), &[KeyValue::new("kind", "current")]); + } + if let Some(budget) = budget { + o.observe(budget.bytes(), &[KeyValue::new("kind", "budget")]); + } + }) + .build(); +} + +fn cpu_seconds() -> Option { + let ticks = cpu_ticks(&fs::read_to_string("/proc/self/stat").ok()?)?; + Some(ticks as f64 / TICKS_PER_SECOND) +} + +fn status_field(name: &str) -> Option { + status_value(&fs::read_to_string("/proc/self/status").ok()?, name) +} + +// comm can contain spaces and parentheses, so split past the last ')' to find `utime` and `stime` +fn cpu_ticks(stat: &str) -> Option { + let mut fields = stat.rsplit_once(')')?.1.split_whitespace().skip(11); + let user: u64 = fields.next()?.parse().ok()?; + let system: u64 = fields.next()?.parse().ok()?; + Some(user + system) +} + +fn status_value(status: &str, name: &str) -> Option { + let line = status.lines().find(|line| line.starts_with(name))?; + line[name.len()..].split_whitespace().next()?.parse().ok() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn cpu_ticks_sum_utime_and_stime_past_a_spaced_command() { + let stat = + "4242 (tokio worker 1) S 1 4242 4242 0 -1 4194560 812 0 0 0 150 25 0 0 20 0 9 0 12345"; + assert_eq!(cpu_ticks(stat), Some(175)); + } + + #[test] + fn status_values_are_read_by_field_name() { + let status = "Name:\tbobbin\nVmRSS:\t 20480 kB\nThreads:\t9\n"; + assert_eq!(status_value(status, "VmRSS:"), Some(20480)); + assert_eq!(status_value(status, "Threads:"), Some(9)); + assert_eq!(status_value(status, "VmSwap:"), None); + } +} diff --git a/bobbin/crates/bobbin/src/otlp.rs b/bobbin/crates/bobbin/src/otlp.rs index ef5ad65b..08a5dd77 100644 --- a/bobbin/crates/bobbin/src/otlp.rs +++ b/bobbin/crates/bobbin/src/otlp.rs @@ -1,11 +1,18 @@ use std::collections::HashMap; use std::fmt; +use std::time::Duration; use base64::Engine as _; use base64::engine::general_purpose::STANDARD as Base64; +use bobbin_runtime::{Entropy as _, OsEntropy}; +use opentelemetry::{KeyValue, global}; use opentelemetry_appender_tracing::layer::{OpenTelemetryTracingBridge, TracingSpanAttributes}; -use opentelemetry_otlp::{LogExporter, Protocol, WithExportConfig as _, WithHttpConfig as _}; -use opentelemetry_sdk::{Resource, logs::SdkLoggerProvider}; +use opentelemetry_otlp::{ + LogExporter, MetricExporter, Protocol, WithExportConfig as _, WithHttpConfig as _, +}; +use opentelemetry_sdk::Resource; +use opentelemetry_sdk::logs::SdkLoggerProvider; +use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider}; use serde::Deserialize; use tracing::Subscriber; use tracing_subscriber::Layer; @@ -18,6 +25,8 @@ use crate::config::OtlpConfig; // suppress http client and otlp logs so exporting a record cannot trigger another export loop const EXPORT_QUIET: [&str; 4] = ["h2", "hyper", "reqwest", "opentelemetry"]; +const METRICS_INTERVAL: Duration = Duration::from_secs(15); + #[derive(Deserialize)] #[serde(try_from = "String")] pub struct OtlpAuth(String); @@ -47,36 +56,56 @@ impl OtlpAuth { } } -pub struct LogExport { - provider: SdkLoggerProvider, +pub struct Export { + logs: SdkLoggerProvider, + metrics: SdkMeterProvider, } -impl LogExport { +impl Export { pub fn from_config(cfg: &OtlpConfig) -> Result, String> { let Some(base) = &cfg.endpoint else { return Ok(None); }; - let endpoint = signal_url(base, "logs"); - let exporter = LogExporter::builder() + let headers = cfg.auth.as_ref().map(OtlpAuth::headers).unwrap_or_default(); + let logs = LogExporter::builder() + .with_http() + .with_protocol(Protocol::HttpBinary) + .with_endpoint(signal_url(base, "logs").as_str()) + .with_headers(headers.clone()) + .build() + .map_err(|e| format!("otlp log exporter for `{base}`: {e}"))?; + let metrics = MetricExporter::builder() .with_http() .with_protocol(Protocol::HttpBinary) - .with_endpoint(endpoint.as_str()) - .with_headers(cfg.auth.as_ref().map(OtlpAuth::headers).unwrap_or_default()) + .with_endpoint(signal_url(base, "metrics").as_str()) + .with_headers(headers) .build() - .map_err(|e| format!("OTLP log exporter for `{endpoint}`: {e}"))?; - Ok(Some(Self::batched(exporter, &cfg.service_name))) + .map_err(|e| format!("otlp metric exporter for `{base}`: {e}"))?; + Ok(Some(Self::new(logs, metrics, &cfg.service_name))) } - fn batched( - exporter: impl opentelemetry_sdk::logs::LogExporter + 'static, + fn new( + logs: impl opentelemetry_sdk::logs::LogExporter + 'static, + metrics: impl opentelemetry_sdk::metrics::exporter::PushMetricExporter, service_name: &str, ) -> Self { let resource = Resource::builder() .with_service_name(service_name.to_owned()) + .with_attribute(KeyValue::new( + "service.instance.id", + format!("{:016x}", OsEntropy.next_u64()), + )) + .build(); + let reader = PeriodicReader::builder(metrics) + .with_interval(METRICS_INTERVAL) .build(); Self { - provider: SdkLoggerProvider::builder() - .with_batch_exporter(exporter) + logs: SdkLoggerProvider::builder() + .with_batch_exporter(logs) + .with_resource(resource.clone()) + .build(), + metrics: SdkMeterProvider::builder() + .with_reader(reader) .with_resource(resource) .build(), } @@ -86,7 +115,7 @@ impl LogExport { where S: Subscriber + for<'a> LookupSpan<'a>, { - OpenTelemetryTracingBridge::builder(&self.provider) + OpenTelemetryTracingBridge::builder(&self.logs) .with_tracing_span_attributes(TracingSpanAttributes::all()) .build() .with_filter( @@ -96,9 +125,16 @@ impl LogExport { ) } + pub fn install_meters(&self) { + global::set_meter_provider(self.metrics.clone()); + } + pub fn flush(self) { - if let Err(error) = self.provider.shutdown() { - eprintln!("could not flush OTLP log export: {error}"); + if let Err(error) = self.logs.shutdown() { + eprintln!("could not flush otlp logs: {error}"); + } + if let Err(error) = self.metrics.shutdown() { + eprintln!("could not flush otlp metrics: {error}"); } } } @@ -115,6 +151,7 @@ mod tests { use opentelemetry::logs::AnyValue; use opentelemetry::{Key, Value}; use opentelemetry_sdk::logs::InMemoryLogExporter; + use opentelemetry_sdk::metrics::InMemoryMetricExporter; use tracing_subscriber::layer::SubscriberExt as _; use super::*; @@ -126,7 +163,7 @@ mod tests { auth: None, service_name: "bobbin-test".to_owned(), }; - assert!(LogExport::from_config(&cfg).unwrap().is_none()); + assert!(Export::from_config(&cfg).unwrap().is_none()); } #[test] @@ -149,7 +186,11 @@ mod tests { #[test] fn export_ships_records_with_service_name_and_fields() { let exporter = InMemoryLogExporter::default(); - let export = LogExport::batched(exporter.clone(), "bobbin-test"); + let export = Export::new( + exporter.clone(), + InMemoryMetricExporter::default(), + "bobbin-test", + ); let subscriber = tracing_subscriber::registry().with(export.layer()); tracing::subscriber::with_default(subscriber, || { tracing::warn!(target: "bobbin_xrpc", did = "did:plc:dawn", "slow request"); @@ -159,7 +200,7 @@ mod tests { tracing::error!(target: "opentelemetry_sdk", "export failed"); tracing::warn!(target: "opentelemetry-otlp", "retries exhausted"); }); - export.provider.force_flush().unwrap(); + export.logs.force_flush().unwrap(); let logs = exporter.get_emitted_logs().unwrap(); let [log] = logs.as_slice() else { @@ -192,4 +233,56 @@ mod tests { assert!(OtlpAuth::try_from("no-colon".to_owned()).is_err()); assert!(OtlpAuth::try_from(":pass".to_owned()).is_err()); } + + #[tokio::test] + async fn installed_meters_export_xrpc_metrics_per_instance() { + use axum::routing::get; + use opentelemetry_sdk::metrics::data::{AggregatedMetrics, MetricData}; + use tower::ServiceExt as _; + + let metrics = InMemoryMetricExporter::default(); + let export = Export::new( + InMemoryLogExporter::default(), + metrics.clone(), + "bobbin-test", + ); + export.install_meters(); + let app = axum::Router::new() + .route("/xrpc/sh.tangled.test", get(|| async { "ok" })) + .layer(axum::middleware::from_fn(crate::metrics::xrpc)); + let request = axum::http::Request::get("/xrpc/sh.tangled.test") + .body(axum::body::Body::empty()) + .unwrap(); + assert!(app.oneshot(request).await.unwrap().status().is_success()); + export.metrics.force_flush().unwrap(); + + let exported = metrics.get_finished_metrics().unwrap(); + let resource = exported[0].resource(); + assert_eq!( + resource.get(&Key::from_static_str("service.name")), + Some(Value::from("bobbin-test")), + ); + assert!( + resource + .get(&Key::from_static_str("service.instance.id")) + .is_some() + ); + let requests = exported + .iter() + .flat_map(|rm| rm.scope_metrics()) + .flat_map(|sm| sm.metrics()) + .find(|m| m.name() == "bobbin_xrpc_requests") + .expect("request counter exported"); + let AggregatedMetrics::U64(MetricData::Sum(sum)) = requests.data() else { + panic!("requests is not a u64 sum"); + }; + let point = sum.data_points().next().expect("one request"); + assert_eq!(point.value(), 1); + let mut labels: Vec = point + .attributes() + .map(|kv| format!("{}={}", kv.key, kv.value)) + .collect(); + labels.sort(); + assert_eq!(labels, ["route=/xrpc/sh.tangled.test", "status=200"]); + } } diff --git a/bobbin/crates/ingest/Cargo.toml b/bobbin/crates/ingest/Cargo.toml index 255ae5b1..7a89925b 100644 --- a/bobbin/crates/ingest/Cargo.toml +++ b/bobbin/crates/ingest/Cargo.toml @@ -19,6 +19,7 @@ bytes = { workspace = true } chrono = { workspace = true } reqwest = { workspace = true } futures = { workspace = true } +opentelemetry = { workspace = true } scc = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } @@ -32,6 +33,7 @@ uuid = { workspace = true } [dev-dependencies] http = { workspace = true } +opentelemetry_sdk = { workspace = true, features = ["testing"] } tracing-subscriber = { workspace = true } tokio-util = { workspace = true } tokio = { workspace = true, features = ["test-util"] } diff --git a/bobbin/crates/ingest/src/analytics.rs b/bobbin/crates/ingest/src/analytics.rs index 6b5ff4d6..af523a3a 100644 --- a/bobbin/crates/ingest/src/analytics.rs +++ b/bobbin/crates/ingest/src/analytics.rs @@ -19,6 +19,8 @@ use tracing::{debug, instrument, warn}; use url::Url; use uuid::Uuid; +use crate::metrics::{self, AnalyticsResult}; + const QUEUE_CAPACITY: usize = 4096; const BATCH_MAX: usize = 100; const FLUSH_INTERVAL: Duration = Duration::from_secs(5); @@ -82,6 +84,7 @@ impl Analytics { }) { Ok(()) => {} Err(mpsc::error::TrySendError::Full(_)) => { + metrics::analytics_events(AnalyticsResult::QueueFull, 1); let dropped = self.dropped.fetch_add(1, Ordering::Relaxed) + 1; // log on powers of two so a stuck flusher does not flood the logs if dropped.is_power_of_two() { @@ -93,6 +96,7 @@ impl Analytics { } } Err(mpsc::error::TrySendError::Closed(_)) => { + metrics::analytics_events(AnalyticsResult::FlusherGone, 1); debug!(target: "bobbin_ingest::posthog", "flusher gone, event dropped"); } } @@ -457,12 +461,15 @@ async fn flush(client: &reqwest::Client, url: &Url, api_key: &str, batch: &mut V Ok(response) if response.status().is_success() => { // drain the body so hyper can reuse the connection let _ = response.bytes().await; + metrics::analytics_events(AnalyticsResult::Sent, count); debug!(target: "bobbin_ingest::posthog", count, "posthog batch accepted"); } Ok(response) => { + metrics::analytics_events(AnalyticsResult::Rejected, count); warn!(target: "bobbin_ingest::posthog", count, status = %response.status(), "posthog rejected batch, dropping it"); } Err(error) => { + metrics::analytics_events(AnalyticsResult::Unreachable, count); warn!(target: "bobbin_ingest::posthog", count, %error, "posthog unreachable, dropping batch"); } } diff --git a/bobbin/crates/ingest/src/lib.rs b/bobbin/crates/ingest/src/lib.rs index 42c3a5b1..1ce3d743 100644 --- a/bobbin/crates/ingest/src/lib.rs +++ b/bobbin/crates/ingest/src/lib.rs @@ -43,11 +43,13 @@ use url::Url; mod analytics; mod frame; +mod metrics; mod resolver; mod shadow; mod warming; use frame::HydrantStreamErrorFrame; pub use frame::{AccountFrame, FrameKind, HydrantFrame, IdentityFrame, RecordAction, RecordFrame}; +use metrics::{EdgeDrop, FrameDrop}; pub use resolver::{RepoIdResolver, Resolution}; pub use shadow::{WarmingShadowBuffer, WarmingShadowSnapshot}; pub use warming::{ParkedUpsert, WarmingBuffer, WarmingBufferSnapshot}; @@ -66,7 +68,6 @@ const CONTROL_CHANNEL_DEPTH: usize = 16; const READER_HOLD_LIMIT: usize = 64; const SEND_TIMEOUT: Duration = Duration::from_secs(10); const NORMAL_CLOSE: u16 = 1000; -const METRICS_DUMP_INTERVAL: Duration = Duration::from_secs(10); const WARMING_FLUSH_PARALLELISM: usize = 64; pub const DEFAULT_INGEST_PARALLELISM: NonZeroUsize = match NonZeroUsize::new(16) { Some(n) => n, @@ -159,6 +160,22 @@ pub enum DisconnectKind { } impl DisconnectKind { + pub const fn as_str(self) -> &'static str { + match self { + Self::Url => "url", + Self::UnknownScheme => "unknown_scheme", + Self::Network => "network", + Self::Decode => "decode", + Self::InvalidAtUri => "invalid_at_uri", + Self::Extract => "extract", + Self::ConnectTimeout => "connect_timeout", + Self::PongTimeout => "pong_timeout", + Self::SendTimeout => "send_timeout", + Self::ConsumerTooSlow => "consumer_too_slow", + Self::HydrantStream => "hydrant_stream", + } + } + pub fn from_error(err: &IngestError) -> Self { match err { IngestError::Url(_) => Self::Url, @@ -320,12 +337,8 @@ pub async fn run( runtime: IngestRuntime, ) -> Result<(), IngestError> { let frames = FrameCounter::default(); - let metrics_dumper = spawn_metrics_dumper(&runtime); let warming_flusher = spawn_warming_flusher(&runtime); let result = run_inner(config, &runtime, &frames).await; - if let Err(join) = metrics_dumper.await { - warn!(?join, "metrics dumper task panicked"); - } if let Some(handle) = warming_flusher && let Err(join) = handle.await { @@ -375,13 +388,17 @@ async fn run_inner( "hydrant stream errored" ), } - if let (Some(sink), Some(err)) = (runtime.disconnects.as_ref(), error.as_ref()) { - sink.record(DisconnectSnapshot { - kind: DisconnectKind::from_error(err), - message: err.to_string(), - at_unix_micros: runtime.clock.now_unix_micros(), - last_cursor: runtime.coverage.snapshot().last_cursor(), - }); + if let Some(err) = error.as_ref() { + let kind = DisconnectKind::from_error(err); + metrics::disconnected(kind); + if let Some(sink) = runtime.disconnects.as_ref() { + sink.record(DisconnectSnapshot { + kind, + message: err.to_string(), + at_unix_micros: runtime.clock.now_unix_micros(), + last_cursor: runtime.coverage.snapshot().last_cursor(), + }); + } } runtime.stream_health.disconnected( runtime.clock.now_unix_micros(), @@ -585,36 +602,6 @@ fn promote_at_head(runtime: &IngestRuntime, head: Hy ); } -fn spawn_metrics_dumper( - runtime: &IngestRuntime, -) -> tokio::task::JoinHandle<()> { - let rt = runtime.clone(); - tokio::spawn(async move { - loop { - tokio::select! { - biased; - _ = rt.cancel.cancelled() => break, - _ = rt.clock.sleep(METRICS_DUMP_INTERVAL) => { - let s = rt.resolver.stats(); - info!( - target: "bobbin_ingest::metrics", - resolver_hits = s.hits, - resolver_misses_mapped = s.misses_mapped, - resolver_misses_no_repo_did = s.misses_no_repo_did, - resolver_misses_unresolvable = s.misses_unresolvable, - resolver_misses_transient = s.misses_transient, - resolver_misses_no_client = s.misses_no_client, - resolver_miss_latency_micros_avg = s.miss_latency_micros_avg().unwrap_or(0), - resolver_miss_latency_micros_max = s.miss_latency_micros_max, - resolver_total = s.total(), - "resolver stats", - ); - } - } - } - }) -} - fn next_connect_cursor(snapshot: Coverage, start: HydrantCursor) -> HydrantCursor { if snapshot.events_processed() == 0 { start @@ -1209,6 +1196,9 @@ async fn commit_stage( } let settled = commit_pending(pending, &rt.pipeline_ctx()).await; let commit_end = rt.clock.now_instant(); + if let Some(settled) = &settled { + metrics::settled(settled); + } if let Some((watch, settled)) = rt.settlements.as_deref().zip(settled) { watch.settle(settled); } @@ -1267,6 +1257,7 @@ async fn prepare_record( ) -> PendingOp { let Some(record) = record else { debug!("record-typed frame missing payload, skipping"); + metrics::frame_dropped(FrameDrop::MissingPayload); return PendingOp::Noop; }; if !record.collection.as_ref().starts_with(TANGLED_PREFIX) @@ -1282,6 +1273,7 @@ async fn prepare_record( Ok(s) => s, Err(e) => { warn!(?e, "invalid frame source, dropping record"); + metrics::frame_dropped(FrameDrop::InvalidSource); return PendingOp::Noop; } }; @@ -1364,6 +1356,7 @@ async fn prepare_record( }, RecordAction::Other => { debug!(collection = %nsid, "ignoring unknown record action"); + metrics::frame_dropped(FrameDrop::UnknownAction); PendingOp::Noop } } @@ -1626,12 +1619,14 @@ async fn finalize_drained( ); log_unknown_state_variant(outcome, &source); index_search(ctx.search, ctx.resolver, &source, parsed).await; + let settled = Settlement { + source, + cid: settled_cid, + outcome: RecordOutcome::Indexed, + }; + metrics::settled(&settled); if let Some(watch) = ctx.settlements { - watch.settle(Settlement { - source, - cid: settled_cid, - outcome: RecordOutcome::Indexed, - }); + watch.settle(settled); } } } @@ -1872,6 +1867,7 @@ async fn normalize_subjects( source = edge.source.as_ref(), "dropping edge: target repo has no repoDid, no canonical DID subject available", ); + metrics::edge_dropped(EdgeDrop::NoRepoDid); None } Resolution::Unresolvable => { @@ -1883,6 +1879,7 @@ async fn normalize_subjects( source = edge.source.as_ref(), "dropping edge: repo unresolvable, rkey-form subject will not match bare-DID queries", ); + metrics::edge_dropped(EdgeDrop::RepoUnresolvable); None } } diff --git a/bobbin/crates/ingest/src/metrics.rs b/bobbin/crates/ingest/src/metrics.rs new file mode 100644 index 00000000..91e2c6e9 --- /dev/null +++ b/bobbin/crates/ingest/src/metrics.rs @@ -0,0 +1,216 @@ +use std::sync::LazyLock; + +use bobbin_edge_index::Settlement; +use bobbin_types::RECORD_COLLECTIONS; +use opentelemetry::metrics::{Counter, Meter}; +use opentelemetry::{KeyValue, global}; + +use crate::DisconnectKind; + +struct Counters { + records: Counter, + frames_dropped: Counter, + edges_dropped: Counter, + disconnects: Counter, + analytics_events: Counter, +} + +static COUNTERS: LazyLock = + LazyLock::new(|| Counters::new(&global::meter("bobbin_ingest"))); + +impl Counters { + fn new(meter: &Meter) -> Self { + let counter = |name, description| { + meter + .u64_counter(name) + .with_description(description) + .build() + }; + Self { + records: counter( + "bobbin_ingest_records", + "records off the firehose by collection, outcome and rejection reason", + ), + frames_dropped: counter( + "bobbin_ingest_frames_dropped", + "record frames dropped before they became a record", + ), + edges_dropped: counter( + "bobbin_ingest_edges_dropped", + "edges dropped because their repo subject has no canonical did", + ), + disconnects: counter( + "bobbin_ingest_disconnects", + "hydrant sessions that ended in an error", + ), + analytics_events: counter( + "bobbin_ingest_analytics_events", + "posthog events by whether they reached posthog", + ), + } + } + + fn settled(&self, settlement: &Settlement) { + // collapse unknown collections so arbitrary firehose strings cant blow up metric + // cardinality + let collection = settlement + .source + .collection() + .and_then(|c| RECORD_COLLECTIONS.binary_search(&c.as_ref()).ok()) + .map_or("other", |i| RECORD_COLLECTIONS[i]); + let reason = settlement + .outcome + .rejection() + .map_or("none", |r| r.as_str()); + self.records.add( + 1, + &[ + KeyValue::new("collection", collection), + KeyValue::new("outcome", settlement.outcome.as_str()), + KeyValue::new("reason", reason), + ], + ); + } +} + +#[derive(Clone, Copy)] +pub(crate) enum FrameDrop { + MissingPayload, + InvalidSource, + UnknownAction, +} + +#[derive(Clone, Copy)] +pub(crate) enum EdgeDrop { + NoRepoDid, + RepoUnresolvable, +} + +#[derive(Clone, Copy)] +pub(crate) enum AnalyticsResult { + Sent, + Rejected, + Unreachable, + QueueFull, + FlusherGone, +} + +pub(crate) fn settled(settlement: &Settlement) { + COUNTERS.settled(settlement); +} + +pub(crate) fn frame_dropped(reason: FrameDrop) { + let reason = match reason { + FrameDrop::MissingPayload => "missing_payload", + FrameDrop::InvalidSource => "invalid_source", + FrameDrop::UnknownAction => "unknown_action", + }; + COUNTERS + .frames_dropped + .add(1, &[KeyValue::new("reason", reason)]); +} + +pub(crate) fn edge_dropped(reason: EdgeDrop) { + let reason = match reason { + EdgeDrop::NoRepoDid => "no_repo_did", + EdgeDrop::RepoUnresolvable => "repo_unresolvable", + }; + COUNTERS + .edges_dropped + .add(1, &[KeyValue::new("reason", reason)]); +} + +pub(crate) fn disconnected(kind: DisconnectKind) { + COUNTERS + .disconnects + .add(1, &[KeyValue::new("kind", kind.as_str())]); +} + +pub(crate) fn analytics_events(result: AnalyticsResult, events: usize) { + let result = match result { + AnalyticsResult::Sent => "sent", + AnalyticsResult::Rejected => "rejected", + AnalyticsResult::Unreachable => "unreachable", + AnalyticsResult::QueueFull => "queue_full", + AnalyticsResult::FlusherGone => "flusher_gone", + }; + let events = u64::try_from(events).unwrap_or(u64::MAX); + COUNTERS + .analytics_events + .add(events, &[KeyValue::new("result", result)]); +} + +#[cfg(test)] +mod tests { + use bobbin_edge_index::{RecordOutcome, Rejection}; + use jacquard_common::types::string::AtUri; + use opentelemetry::metrics::MeterProvider as _; + use opentelemetry_sdk::metrics::data::{AggregatedMetrics, MetricData}; + use opentelemetry_sdk::metrics::{InMemoryMetricExporter, PeriodicReader, SdkMeterProvider}; + + use super::*; + + fn settlement(uri: &str, outcome: RecordOutcome) -> Settlement { + Settlement { + source: AtUri::new_owned(uri).unwrap(), + cid: None, + outcome, + } + } + + #[test] + fn records_are_labelled_by_known_collection_outcome_and_reason() { + let exporter = InMemoryMetricExporter::default(); + let provider = SdkMeterProvider::builder() + .with_reader(PeriodicReader::builder(exporter.clone()).build()) + .build(); + let counters = Counters::new(&provider.meter("test")); + counters.settled(&settlement( + "at://did:plc:nel/sh.tangled.feed.star/k1", + RecordOutcome::Indexed, + )); + counters.settled(&settlement( + "at://did:plc:nel/sh.tangled.made.up/k1", + RecordOutcome::rejected(Rejection::UnknownCollection, Some("sh.tangled.made.up")), + )); + provider.force_flush().unwrap(); + + let metrics = exporter.get_finished_metrics().unwrap(); + let records = metrics + .iter() + .flat_map(|rm| rm.scope_metrics()) + .flat_map(|sm| sm.metrics()) + .find(|m| m.name() == "bobbin_ingest_records") + .expect("records counter exported"); + let AggregatedMetrics::U64(MetricData::Sum(sum)) = records.data() else { + panic!("records is not a u64 sum"); + }; + let mut seen: Vec> = sum + .data_points() + .map(|point| { + let mut labels: Vec = point + .attributes() + .map(|kv| format!("{}={}", kv.key, kv.value)) + .collect(); + labels.sort(); + labels + }) + .collect(); + seen.sort(); + assert_eq!( + seen, + [ + [ + "collection=other", + "outcome=rejected", + "reason=unknownCollection" + ], + [ + "collection=sh.tangled.feed.star", + "outcome=indexed", + "reason=none" + ], + ], + ); + } +} diff --git a/bobbin/crates/xrpc/src/backpressure.rs b/bobbin/crates/xrpc/src/backpressure.rs index 3cd2d42b..f4e62d89 100644 --- a/bobbin/crates/xrpc/src/backpressure.rs +++ b/bobbin/crates/xrpc/src/backpressure.rs @@ -1,5 +1,5 @@ use std::sync::Arc; -use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; use bobbin_runtime::MemoryBudget; use tokio::sync::{OwnedSemaphorePermit, Semaphore}; @@ -61,6 +61,7 @@ pub enum PressureVerdict { struct LimiterCore { in_flight: AtomicUsize, + rejected: AtomicU64, limit: AtomicUsize, floor: usize, ceiling: usize, @@ -87,6 +88,7 @@ impl HeavyLimiter { Self { core: Arc::new(LimiterCore { in_flight: AtomicUsize::new(0), + rejected: AtomicU64::new(0), limit: AtomicUsize::new(ceiling), floor: 1, ceiling, @@ -103,6 +105,7 @@ impl HeavyLimiter { }) } else { self.core.in_flight.fetch_sub(1, Ordering::AcqRel); + self.core.rejected.fetch_add(1, Ordering::Relaxed); Err(XrpcError::overloaded()) } } @@ -120,6 +123,14 @@ impl HeavyLimiter { pub fn limit(&self) -> usize { self.core.limit.load(Ordering::Acquire) } + + pub fn in_flight(&self) -> usize { + self.core.in_flight.load(Ordering::Acquire) + } + + pub fn rejected(&self) -> u64 { + self.core.rejected.load(Ordering::Relaxed) + } } // a wait is idle, so this caps connection buildup rather than memory @@ -196,6 +207,7 @@ mod tests { let limiter = HeavyLimiter::new(MaxInFlight(1)); let held = limiter.try_enter().expect("first permit enters"); assert!(matches!(limiter.try_enter(), Err(XrpcError::Overloaded))); + assert_eq!((limiter.in_flight(), limiter.rejected()), (1, 1)); drop(held); assert!(limiter.try_enter().is_ok()); } diff --git a/bobbin/example.toml b/bobbin/example.toml index 0a043e38..4361e4ac 100644 --- a/bobbin/example.toml +++ b/bobbin/example.toml @@ -221,7 +221,7 @@ #filter = "info" [otlp] -# base url of an otlp/http collector for logs. +# base url of an otlp/http collector for logs and metrics. # unset disables opentelemetry export. # # Can also be specified via environment variable `BOBBIN_OTLP_ENDPOINT`.