diff --git a/payjoin-mailroom/CHANGELOG.md b/payjoin-mailroom/CHANGELOG.md index 378991892..e49ad5fd8 100644 --- a/payjoin-mailroom/CHANGELOG.md +++ b/payjoin-mailroom/CHANGELOG.md @@ -14,6 +14,7 @@ - Enable HTTP/2 multiplexing on relay-directory hop (#1655) - Bound OHTTP bootstrap tunnel resource usage and export its metrics (#1610) - Add per-request metrics middleware (#1674) +- Export only settled weekly windows with small-count suppression, quantization, and an attribute allowlist; precise metrics stay local ## 0.1.1 diff --git a/payjoin-mailroom/README.md b/payjoin-mailroom/README.md index 8f92053e7..c48fcf47c 100644 --- a/payjoin-mailroom/README.md +++ b/payjoin-mailroom/README.md @@ -47,8 +47,44 @@ systemctl enable --now payjoin-mailroom ## Telemetry payjoin-mailroom supports **optional** OpenTelemetry-based telemetry (metrics). -Build with `--features telemetry` and configure via the [`[telemetry]`](config.example.com) config section. +Build with `--features telemetry` and configure via the [`[telemetry]`](config.example.toml) config section. When no telemetry configuration is present, it falls back to local-only console tracing. +Metrics are local-only by default: nothing is exported unless a `[telemetry]` section is configured. + +### What leaves the operator boundary + +Precise metrics (per-request counters, live in-flight and tunnel gauges) never +leave the process. When export is configured, the only metrics pushed to the +OTLP endpoint are unlabelled coarse weekly gauges, designed to support +ecosystem traction measurement without providing per-request telemetry: + +- **Settled weekly windows.** Each exported count covers one completed + Monday-to-Monday UTC week (`EXPORT_WINDOW_DAYS` in `src/metrics.rs`). The + in-progress week is never exported, preventing live probing and daily + differencing of overlapping windows. +- **Small-count suppression.** Windows whose raw count is below + `suppression_threshold` (default 10) are dropped entirely, not rounded: a + small operator's weekly count of 1-2 could be tied to a known real-world + event. +- **Quantization.** Surviving counts are rounded to the nearest + `quantization_bin` (default 5), so exported values carry no small-integer + precision to subtract against. +- **Attribute minimization.** Exported metric points carry no attributes. The + resource carries only `service.name` and a Foundation-issued opaque + `reporter.id`; its operator mapping belongs outside Grafana. Tests fail if + any other metric or resource attribute appears. +- **Reliable coarse delivery.** The exporter pushes daily, but every push in a + reporting week carries the same frozen aggregate for the preceding completed + week. This provides retry opportunities without exposing daily traffic + volume; daily delivery time remains liveness metadata. + +The Foundation should restrict raw reporter-labelled series to its telemetry +operators, retain them briefly, and publish aggregate dashboards only. A future +collector can aggregate across reporters and remove `reporter.id` before data +reaches Grafana. + +`export_enabled = false` keeps the `[telemetry]` section's structured JSON +logging while disabling the metrics export entirely. ## Access Control diff --git a/payjoin-mailroom/config.example.toml b/payjoin-mailroom/config.example.toml index 11ffb52dc..4a42c69ff 100644 --- a/payjoin-mailroom/config.example.toml +++ b/payjoin-mailroom/config.example.toml @@ -2,7 +2,7 @@ # # Configuration can also be set via environment variables with the `PJ_` # prefix. Nested values use double underscores as separators, e.g. -# PJ_TELEMETRY__OPERATOR_DOMAIN="your-domain.example.com" +# PJ_TELEMETRY__REPORTER_ID="foundation-issued-opaque-id" # Address and port to listen on # listener = "[::]:8080" @@ -22,9 +22,23 @@ # Authentication token for the OTLP endpoint # auth_token = "" -# The domain you are running the payjoin-mailroom from. -# This serves as an identifier for metrics collection. -# operator_domain = "your-domain.example.com" +# Foundation-issued opaque reporting ID. Do not use a domain, hostname, or +# operator name. Its mapping to an operator is kept outside Grafana. +# reporter_id = "foundation-issued-opaque-id" + +# Master switch for the metrics export. When false, the [telemetry] section's +# structured JSON logging is kept but no metrics leave this machine. +# Omitting the whole [telemetry] section is also local-only. +# export_enabled = true + +# Exported counts cover one completed Monday-to-Monday UTC week and are coarsened +# before export so aggregates cannot deanonymize users or fingerprint this +# operator. Windows whose raw count is below this threshold are not exported +# at all. +# suppression_threshold = 10 + +# Exported counts are rounded to the nearest multiple of this bin. +# quantization_bin = 5 # --- Access-control (requires `access-control` feature) --- # [access_control] diff --git a/payjoin-mailroom/src/config.rs b/payjoin-mailroom/src/config.rs index 77debbf70..4fc2a034e 100644 --- a/payjoin-mailroom/src/config.rs +++ b/payjoin-mailroom/src/config.rs @@ -43,7 +43,42 @@ pub struct V1Config { pub struct TelemetryConfig { pub endpoint: String, pub auth_token: String, - pub operator_domain: String, + /// Foundation-issued opaque identifier for this reporting installation. + /// Never use a domain, hostname, or operator name here. + pub reporter_id: String, + /// Master switch for the metrics export. When `false` the mailroom keeps + /// the structured log format from this section but exports no metrics at + /// all; metrics stay local to the operator. Omitting the whole + /// `[telemetry]` section (the default) is also local-only. + #[serde(default = "default_export_enabled")] + pub export_enabled: bool, + /// Exported windows whose raw count is below this are dropped entirely. + /// See [`crate::metrics::ExportPolicy`]. + #[serde(default = "default_suppression_threshold")] + pub suppression_threshold: u64, + /// Exported counts are rounded to the nearest multiple of this bin. + /// See [`crate::metrics::ExportPolicy`]. + #[serde(default = "default_quantization_bin")] + pub quantization_bin: u64, +} + +#[cfg(feature = "telemetry")] +fn default_export_enabled() -> bool { true } + +#[cfg(feature = "telemetry")] +fn default_suppression_threshold() -> u64 { crate::metrics::DEFAULT_SUPPRESSION_THRESHOLD } + +#[cfg(feature = "telemetry")] +fn default_quantization_bin() -> u64 { crate::metrics::DEFAULT_QUANTIZATION_BIN } + +#[cfg(feature = "telemetry")] +impl TelemetryConfig { + pub fn export_policy(&self) -> crate::metrics::ExportPolicy { + crate::metrics::ExportPolicy { + suppression_threshold: self.suppression_threshold, + quantization_bin: self.quantization_bin, + } + } } #[cfg(feature = "acme")] diff --git a/payjoin-mailroom/src/lib.rs b/payjoin-mailroom/src/lib.rs index 65ec40260..34f58cf45 100644 --- a/payjoin-mailroom/src/lib.rs +++ b/payjoin-mailroom/src/lib.rs @@ -29,7 +29,7 @@ pub mod ohttp_relay; #[cfg(feature = "telemetry")] pub mod telemetry; -use crate::metrics::MetricsService; +use crate::metrics::{ExportPolicy, MetricsService}; use crate::middleware::{track_connections, track_metrics}; type DirectoryService = @@ -46,7 +46,7 @@ struct Services { pub async fn serve(config: Config, meter_provider: Option) -> anyhow::Result<()> { let sentinel_tag = generate_sentinel_tag(); - let metrics = MetricsService::new(meter_provider); + let metrics = build_metrics(&config, meter_provider); #[cfg(feature = "access-control")] let geoip = init_geoip(&config).await?; @@ -169,7 +169,7 @@ pub async fn serve_acme( .ok_or_else(|| anyhow::anyhow!("ACME configuration is required for serve_acme"))?; let sentinel_tag = generate_sentinel_tag(); - let metrics = MetricsService::new(meter_provider); + let metrics = build_metrics(&config, meter_provider); #[cfg(feature = "access-control")] let geoip = init_geoip(&config).await?; @@ -229,6 +229,29 @@ pub async fn serve_acme( /// at detecting self loops. fn generate_sentinel_tag() -> SentinelTag { SentinelTag::new(rand::thread_rng().gen()) } +/// Builds the metrics service for a server process. +/// +/// Precise instruments always stay in-process. When an export provider is +/// present, the only instruments registered on it are the coarse +/// settled-window gauges filtered through the operator's export policy: +/// nothing precise, live, or below-threshold leaves the operator boundary. +fn build_metrics(config: &Config, export_provider: Option) -> MetricsService { + match export_provider { + Some(provider) => MetricsService::with_export(&provider, export_policy(config)), + None => MetricsService::new(None), + } +} + +fn export_policy(config: &Config) -> ExportPolicy { + #[cfg(feature = "telemetry")] + if let Some(telemetry) = &config.telemetry { + return telemetry.export_policy(); + } + #[cfg(not(feature = "telemetry"))] + let _ = config; + ExportPolicy::default() +} + #[cfg(feature = "access-control")] impl Connected> for middleware::MaybePeerIp { fn connect_info(stream: IncomingStream<'_, Listener>) -> Self { diff --git a/payjoin-mailroom/src/main.rs b/payjoin-mailroom/src/main.rs index 60f314c53..2b8000ab7 100644 --- a/payjoin-mailroom/src/main.rs +++ b/payjoin-mailroom/src/main.rs @@ -12,7 +12,7 @@ async fn main() -> anyhow::Result<()> { #[cfg(feature = "telemetry")] let meter_provider = match &config.telemetry { - Some(telemetry) => Some(init_tracing_with_telemetry(telemetry)), + Some(telemetry) => init_tracing_with_telemetry(telemetry), None => init_tracing(), }; #[cfg(not(feature = "telemetry"))] @@ -35,19 +35,25 @@ fn init_tracing() -> Option { } #[cfg(feature = "telemetry")] -fn init_tracing_with_telemetry(telemetry: &config::TelemetryConfig) -> SdkMeterProvider { - let meter_provider = payjoin_mailroom::telemetry::build_otlp_meter_provider( - &telemetry.endpoint, - &telemetry.auth_token, - &telemetry.operator_domain, - ); +fn init_tracing_with_telemetry(telemetry: &config::TelemetryConfig) -> Option { + // export_enabled = false keeps this section's structured logging but + // builds no exporter at all: metrics stay local to the operator. + let meter_provider = telemetry.export_enabled.then(|| { + payjoin_mailroom::telemetry::build_otlp_meter_provider( + &telemetry.endpoint, + &telemetry.auth_token, + &telemetry.reporter_id, + ) + }); let env_filter = EnvFilter::builder().with_default_directive(LevelFilter::INFO.into()).from_env_lossy(); tracing_subscriber::fmt().json().with_target(true).with_env_filter(env_filter).init(); - opentelemetry::global::set_meter_provider(meter_provider.clone()); + if let Some(meter_provider) = &meter_provider { + opentelemetry::global::set_meter_provider(meter_provider.clone()); + } meter_provider } diff --git a/payjoin-mailroom/src/metrics.rs b/payjoin-mailroom/src/metrics.rs index 702678bc9..e54391ff7 100644 --- a/payjoin-mailroom/src/metrics.rs +++ b/payjoin-mailroom/src/metrics.rs @@ -18,10 +18,77 @@ pub(crate) const HTTP_REQUESTS_TOTAL: &str = "http_requests_total"; pub(crate) const DB_ENTRIES: &str = "db_entries_total"; pub(crate) const UNIQUE_SHORT_IDS: &str = "unique_short_ids"; +// Names of the coarse, settled-window gauges that make up the entire export +// surface. Everything above stays in-process; only these leave the operator +// boundary, and only after passing through an ExportPolicy. +pub(crate) const HTTP_REQUESTS_WEEKLY: &str = "http_requests_weekly"; +pub(crate) const HTTP_REQUESTS_STARTED_WEEKLY: &str = "http_requests_started_weekly"; +pub(crate) const DB_ENTRIES_WEEKLY: &str = "db_entries_weekly"; +pub(crate) const TUNNEL_SHEDS_WEEKLY: &str = "bootstrap_tunnel_sheds_weekly"; +pub(crate) const UNIQUE_SHORT_IDS_WEEKLY: &str = "unique_short_ids_weekly"; + const HLL_PRECISION: u8 = 14; const HOURLY_RETENTION_HOURS: u64 = 168; // 7 days const DAILY_RETENTION_DAYS: u64 = 90; +/// Number of UTC days in each exported reporting window. +/// +/// Counts that leave the operator boundary cover one completed, fixed UTC +/// week. The in-progress week is never exported, preventing live probing and +/// avoiding the daily differencing possible with a sliding window. +pub const EXPORT_WINDOW_DAYS: u64 = 7; + +/// Day since the UNIX epoch for Monday 1970-01-05, the reporting-week anchor. +const REPORTING_WEEK_ANCHOR_DAY: u64 = 4; + +/// Default for [`ExportPolicy::suppression_threshold`]. +pub const DEFAULT_SUPPRESSION_THRESHOLD: u64 = 10; + +/// Default for [`ExportPolicy::quantization_bin`]. +pub const DEFAULT_QUANTIZATION_BIN: u64 = 5; + +/// Coarsening applied to every count before it leaves the operator boundary. +/// +/// Exported aggregates are visible outside the operator, so small counts and +/// precise integers are a linkage risk: a weekly count of 1-2 at a small +/// operator can be tied to a known real-world event, and precise integers let +/// an attacker who sends known traffic infer the remainder by subtraction. +/// Below-threshold windows are dropped entirely (rounding them would still +/// reveal that a small count existed) and surviving counts are quantized. +#[derive(Debug, Clone, Copy)] +pub struct ExportPolicy { + /// Windows whose raw count is below this are not exported at all. + pub suppression_threshold: u64, + /// Surviving counts are rounded to the nearest multiple of this bin. + /// A value of 0 is treated as 1 (no rounding). + pub quantization_bin: u64, +} + +impl Default for ExportPolicy { + fn default() -> Self { + Self { + suppression_threshold: DEFAULT_SUPPRESSION_THRESHOLD, + quantization_bin: DEFAULT_QUANTIZATION_BIN, + } + } +} + +impl ExportPolicy { + /// Applies suppression, then quantization, to a raw windowed count. + /// + /// Suppression is evaluated on the raw count before any rounding, so a + /// count just under the threshold is dropped rather than rounded up into + /// visibility. Rounding is to the nearest multiple of the bin, with ties + /// rounding up. + pub fn apply(&self, raw: u64) -> Option { + if raw < self.suppression_threshold { + return None; + } + let bin = self.quantization_bin.max(1); + Some(raw.saturating_add(bin / 2) / bin * bin) + } +} + type HllSketch = HyperLogLogPlus<[u8; 8], RandomState>; const HOUR: Duration = Duration::from_secs(3600); @@ -37,6 +104,12 @@ trait SystemTimeExt { fn days_since_epoch(&self) -> u64 { self.intervals_since_epoch(DAY) } } +fn reporting_week(day: u64) -> u64 { + day.saturating_sub(REPORTING_WEEK_ANCHOR_DAY) / EXPORT_WINDOW_DAYS +} + +fn reporting_week_start(week: u64) -> u64 { week * EXPORT_WINDOW_DAYS + REPORTING_WEEK_ANCHOR_DAY } + impl SystemTimeExt for SystemTime { fn intervals_since_epoch(&self, interval: Duration) -> u64 { self.duration_since(UNIX_EPOCH).expect("system clock before UNIX epoch").as_secs() @@ -106,6 +179,15 @@ impl HllSketches { self.daily_union_count(today.saturating_sub(6)..=today) } + /// Unique-id estimate over the most recently completed UTC reporting + /// week. This is the only cardinality figure eligible for export. + fn settled_weekly_count(&mut self) -> u64 { + let current_week = reporting_week(SystemTime::now().days_since_epoch()); + let settled_week = current_week.saturating_sub(1); + let start = reporting_week_start(settled_week); + self.daily_union_count(start..=start + EXPORT_WINDOW_DAYS - 1) + } + fn monthly_count(&mut self) -> u64 { let today = SystemTime::now().days_since_epoch(); self.daily_union_count(today.saturating_sub(30)..=today) @@ -147,12 +229,61 @@ impl UniqueShortIdTracker { pub fn monthly_count(&self) -> u64 { self.inner.lock().expect("tracker lock poisoned").monthly_count() } + + pub(crate) fn settled_weekly_count(&self) -> u64 { + self.inner.lock().expect("tracker lock poisoned").settled_weekly_count() + } } impl Default for UniqueShortIdTracker { fn default() -> Self { Self::new() } } +/// Per-UTC-week counter buckets backing the settled-window export. +/// +/// Weeks are anchored on Monday UTC. Each `add` retains only the active and +/// immediately preceding week, the only two windows relevant to export. +#[derive(Default)] +struct WeeklyBuckets { + weeks: BTreeMap, +} + +impl WeeklyBuckets { + fn add(&mut self, day: u64) { + let week = reporting_week(day); + *self.weeks.entry(week).or_insert(0) += 1; + let cutoff = week.saturating_sub(1); + while let Some((&k, _)) = self.weeks.first_key_value() { + if k < cutoff { + self.weeks.pop_first(); + } else { + break; + } + } + } + + /// Count in the most recently completed fixed UTC week. + /// + /// The in-progress week is excluded, so traffic sent now is invisible + /// until the week completes. Releasing fixed weeks also prevents a viewer + /// from differencing adjacent rolling windows to recover daily traffic. + fn settled_window_count(&self, today: u64) -> u64 { + self.weeks.get(&reporting_week(today).saturating_sub(1)).copied().unwrap_or(0) + } +} + +/// In-process weekly accounting for every count eligible for export. +/// +/// Point-in-time gauges (requests in flight, open tunnels) are deliberately +/// absent: they are live values and never leave the process. +#[derive(Default)] +struct ExportWindows { + http_requests: WeeklyBuckets, + http_requests_started: WeeklyBuckets, + db_entries: WeeklyBuckets, + tunnel_sheds: WeeklyBuckets, +} + #[derive(Clone)] pub struct MetricsService { /// Total number of HTTP requests that ran to completion, by endpoint @@ -174,7 +305,12 @@ pub struct MetricsService { /// Total v1/v2 mailbox entries written, labelled by `version` db_entries_total: Counter, tracker: UniqueShortIdTracker, + /// Weekly buckets feeding the settled-window export gauges. + windows: Arc>, _unique_ids_gauge: Option>>, + _export_gauges: Vec>>, + /// Keeps the export pipeline alive for as long as the service exists. + _export_provider: Option, } impl fmt::Debug for MetricsService { @@ -184,7 +320,7 @@ impl fmt::Debug for MetricsService { } #[repr(u8)] -#[derive(Clone, Copy, PartialEq, Eq)] +#[derive(Clone, Copy, PartialEq, Eq, Hash)] pub enum PayjoinVersion { /// BIP 78 Payjoin One = 1, @@ -272,10 +408,117 @@ impl MetricsService { tunnel_sheds_total, db_entries_total, tracker, + windows: Arc::new(Mutex::new(ExportWindows::default())), _unique_ids_gauge: unique_ids_gauge, + _export_gauges: Vec::new(), + _export_provider: None, } } + /// Builds a service whose precise instruments stay in-process and whose + /// only exported instruments are the coarse settled-window gauges + /// registered on `export_provider`, filtered through `policy`. + /// + /// This is the constructor for anything that exports beyond the operator + /// boundary: no live counter, no in-progress window, and no count that + /// survives the policy's suppression threshold unquantized ever reaches + /// the export provider. + pub fn with_export(export_provider: &SdkMeterProvider, policy: ExportPolicy) -> Self { + let mut service = Self::new(None); + service.register_export_gauges(export_provider, policy); + service._export_provider = Some(export_provider.clone()); + service + } + + fn register_export_gauges(&mut self, provider: &SdkMeterProvider, policy: ExportPolicy) { + let meter = provider.meter("payjoin-mailroom"); + + let windows = self.windows.clone(); + let http_requests_weekly = meter + .u64_observable_gauge(HTTP_REQUESTS_WEEKLY) + .with_description( + "Completed HTTP requests in the last settled UTC reporting week, coarsened", + ) + .with_callback(move |observer| { + let today = SystemTime::now().days_since_epoch(); + let windows = windows.lock().expect("windows lock poisoned"); + if let Some(count) = policy.apply(windows.http_requests.settled_window_count(today)) + { + observer.observe(count, &[]); + } + }) + .build(); + + let windows = self.windows.clone(); + let http_requests_started_weekly = meter + .u64_observable_gauge(HTTP_REQUESTS_STARTED_WEEKLY) + .with_description( + "HTTP requests started in the last settled UTC reporting week, coarsened", + ) + .with_callback(move |observer| { + let today = SystemTime::now().days_since_epoch(); + let windows = windows.lock().expect("windows lock poisoned"); + if let Some(count) = + policy.apply(windows.http_requests_started.settled_window_count(today)) + { + observer.observe(count, &[]); + } + }) + .build(); + + let windows = self.windows.clone(); + let db_entries_weekly = meter + .u64_observable_gauge(DB_ENTRIES_WEEKLY) + .with_description( + "Mailbox entries stored in the last settled UTC reporting week, coarsened", + ) + .with_callback(move |observer| { + let today = SystemTime::now().days_since_epoch(); + let windows = windows.lock().expect("windows lock poisoned"); + if let Some(count) = policy.apply(windows.db_entries.settled_window_count(today)) { + observer.observe(count, &[]); + } + }) + .build(); + + let windows = self.windows.clone(); + let tunnel_sheds_weekly = meter + .u64_observable_gauge(TUNNEL_SHEDS_WEEKLY) + .with_description( + "OHTTP bootstrap tunnels shed in the last settled UTC reporting week, coarsened", + ) + .with_callback(move |observer| { + let today = SystemTime::now().days_since_epoch(); + let windows = windows.lock().expect("windows lock poisoned"); + if let Some(count) = policy.apply(windows.tunnel_sheds.settled_window_count(today)) + { + observer.observe(count, &[]); + } + }) + .build(); + + let tracker = self.tracker.clone(); + let unique_short_ids_weekly = meter + .u64_observable_gauge(UNIQUE_SHORT_IDS_WEEKLY) + .with_description( + "Estimated unique short IDs in the last settled UTC reporting week, coarsened", + ) + .with_callback(move |observer| { + if let Some(count) = policy.apply(tracker.settled_weekly_count()) { + observer.observe(count, &[]); + } + }) + .build(); + + self._export_gauges = vec![ + Arc::new(http_requests_weekly), + Arc::new(http_requests_started_weekly), + Arc::new(db_entries_weekly), + Arc::new(tunnel_sheds_weekly), + Arc::new(unique_short_ids_weekly), + ]; + } + pub fn record_http_request(&self, endpoint: &str, method: &str, status_code: u16) { self.http_requests_total.add( 1, @@ -285,6 +528,8 @@ impl MetricsService { KeyValue::new("status_code", status_code.to_string()), ], ); + let day = SystemTime::now().days_since_epoch(); + self.windows.lock().expect("windows lock poisoned").http_requests.add(day); } /// Records the start of an HTTP request and returns a guard that marks it @@ -299,6 +544,8 @@ impl MetricsService { pub(crate) fn track_request(&self) -> InFlightGuard { self.http_requests_started_total.add(1, &[]); self.http_requests_in_flight.add(1, &[]); + let day = SystemTime::now().days_since_epoch(); + self.windows.lock().expect("windows lock poisoned").http_requests_started.add(day); InFlightGuard { in_flight: self.http_requests_in_flight.clone() } } @@ -306,10 +553,16 @@ impl MetricsService { pub fn record_tunnel_close(&self) { self.active_tunnels.add(-1, &[]); } - pub fn record_tunnel_shed(&self) { self.tunnel_sheds_total.add(1, &[]); } + pub fn record_tunnel_shed(&self) { + self.tunnel_sheds_total.add(1, &[]); + let day = SystemTime::now().days_since_epoch(); + self.windows.lock().expect("windows lock poisoned").tunnel_sheds.add(day); + } pub fn record_db_entry(&self, version: PayjoinVersion) { self.db_entries_total.add(1, &[KeyValue::new("version", version.to_string())]); + let day = SystemTime::now().days_since_epoch(); + self.windows.lock().expect("windows lock poisoned").db_entries.add(day); } pub fn record_short_id(&self, id: &ShortId) { self.tracker.add_id(id); } @@ -429,4 +682,185 @@ mod tests { "the guard's Drop decremented in-flight while unwinding" ); } + + #[test] + fn weekly_buckets_exclude_the_in_progress_week() { + let mut buckets = WeeklyBuckets::default(); + let settled_day = reporting_week_start(100) + 1; + let active_day = reporting_week_start(101) + 1; + for _ in 0..3 { + buckets.add(settled_day); + buckets.add(active_day); + } + assert_eq!(buckets.settled_window_count(active_day), 3); + } + + #[test] + fn weekly_buckets_release_non_overlapping_windows_and_prune() { + let mut buckets = WeeklyBuckets::default(); + let first = reporting_week_start(100); + buckets.add(first); + buckets.add(first); + assert_eq!(buckets.settled_window_count(reporting_week_start(101)), 2); + assert_eq!(buckets.settled_window_count(reporting_week_start(102)), 0); + buckets.add(reporting_week_start(102)); + assert_eq!(buckets.weeks.len(), 1, "older reporting weeks are pruned on add"); + } + + /// Collects (metric name, attribute keys per data point) for everything + /// the exporter saw. Metrics with no data points still appear once with + /// an empty key list so name assertions can see them. + fn exported_points(exporter: &InMemoryMetricExporter) -> Vec<(String, Vec)> { + let mut points = Vec::new(); + for rm in exporter.get_finished_metrics().expect("metrics").iter() { + for sm in rm.scope_metrics() { + for m in sm.metrics() { + let name = m.name().to_string(); + match m.data() { + AggregatedMetrics::U64(MetricData::Gauge(gauge)) => + for dp in gauge.data_points() { + let keys = + dp.attributes().map(|kv| kv.key.as_str().to_string()).collect(); + points.push((name.clone(), keys)); + }, + _ => points.push((name.clone(), Vec::new())), + } + } + } + } + points + } + + fn in_memory_provider() -> (InMemoryMetricExporter, SdkMeterProvider) { + let exporter = InMemoryMetricExporter::default(); + let reader = PeriodicReader::builder(exporter.clone()).build(); + let provider = SdkMeterProvider::builder().with_reader(reader).build(); + (exporter, provider) + } + + /// The export surface must contain only the coarse settled-window gauges + /// with no metric attributes. Anything else on the export provider + /// is a leak: precise counters, live gauges, or an attribute that could + /// identify the operator or a client (hostname, IP, instance id, path). + #[test] + fn export_surface_is_windowed_gauges_with_allowed_attributes_only() { + const EXPORTED_METRICS: &[&str] = &[ + HTTP_REQUESTS_WEEKLY, + HTTP_REQUESTS_STARTED_WEEKLY, + DB_ENTRIES_WEEKLY, + TUNNEL_SHEDS_WEEKLY, + UNIQUE_SHORT_IDS_WEEKLY, + ]; + + let (exporter, provider) = in_memory_provider(); + // A permissive policy so every gauge observes a point (today's traffic + // is settled to 0, which threshold 0 still emits) and its attributes + // become visible to the audit. + let policy = ExportPolicy { suppression_threshold: 0, quantization_bin: 1 }; + let metrics = MetricsService::with_export(&provider, policy); + + metrics.record_http_request("/health", "GET", 200); + drop(metrics.track_request()); + metrics.record_db_entry(PayjoinVersion::Two); + metrics.record_tunnel_shed(); + metrics.record_short_id(&ShortId([0; 8])); + + provider.force_flush().expect("flush failed"); + + let points = exported_points(&exporter); + let names: std::collections::HashSet<&str> = + points.iter().map(|(name, _)| name.as_str()).collect(); + for expected in EXPORTED_METRICS { + assert!(names.contains(expected), "{expected} missing from the export surface"); + } + for (name, keys) in &points { + assert!( + EXPORTED_METRICS.contains(&name.as_str()), + "unexpected metric {name} on the export provider" + ); + assert!( + keys.is_empty(), + "disallowed attribute keys {keys:?} on exported metric {name}" + ); + } + } + + /// Under the default policy, traffic recorded today must be invisible in + /// the export: the in-progress window is never emitted, so there is no + /// live counter for an active prober to watch move. + #[test] + fn export_omits_in_progress_window() { + let (exporter, provider) = in_memory_provider(); + let metrics = MetricsService::with_export(&provider, ExportPolicy::default()); + + for _ in 0..100 { + metrics.record_http_request("/health", "GET", 200); + } + metrics.record_short_id(&ShortId([1; 8])); + + provider.force_flush().expect("flush failed"); + + // Today's traffic sits in the in-progress bucket, so every settled + // window is 0 and falls under the suppression threshold: the flush + // must carry no data points at all. + let points = exported_points(&exporter); + assert!( + points.is_empty(), + "data points left the process while their window was in progress: {points:?}" + ); + } + + /// Precise instruments registered via `new` are unaffected by the export + /// pipeline: an operator's own reader still sees exact counts. + #[test] + fn precise_local_metrics_remain_exact() { + let exporter = InMemoryMetricExporter::default(); + let reader = PeriodicReader::builder(exporter.clone()).build(); + let provider = SdkMeterProvider::builder().with_reader(reader).build(); + let metrics = MetricsService::new(Some(provider.clone())); + + for _ in 0..3 { + metrics.record_http_request("/health", "GET", 200); + } + provider.force_flush().expect("flush failed"); + assert_eq!(sum_u64(&exporter, HTTP_REQUESTS_TOTAL), 3); + } + + #[test] + fn export_policy_suppresses_below_threshold() { + let policy = ExportPolicy::default(); + assert_eq!(policy.apply(0), None, "zero is below the default threshold"); + assert_eq!(policy.apply(9), None, "counts under the threshold are dropped, not rounded"); + assert_eq!(policy.apply(10), Some(10), "the threshold itself is exported"); + } + + #[test] + fn export_policy_quantizes_to_nearest_bin() { + let policy = ExportPolicy { suppression_threshold: 0, quantization_bin: 5 }; + assert_eq!(policy.apply(11), Some(10), "11 rounds down to the nearest bin"); + assert_eq!(policy.apply(12), Some(10), "12 rounds down to the nearest bin"); + assert_eq!(policy.apply(13), Some(15), "13 rounds up to the nearest bin"); + assert_eq!(policy.apply(15), Some(15), "exact multiples are unchanged"); + assert_eq!(policy.apply(17), Some(15), "17 rounds down"); + assert_eq!(policy.apply(18), Some(20), "18 rounds up"); + } + + #[test] + fn export_policy_suppression_precedes_quantization() { + // 8 would quantize to 10, meeting the threshold, but suppression is + // decided on the raw count: small precise counts are the leak, so + // they must be dropped rather than rounded into visibility. + let policy = ExportPolicy { suppression_threshold: 10, quantization_bin: 5 }; + assert_eq!(policy.apply(8), None); + } + + #[test] + fn export_policy_degenerate_bins() { + let identity = ExportPolicy { suppression_threshold: 0, quantization_bin: 1 }; + assert_eq!(identity.apply(7), Some(7), "bin of 1 leaves counts unchanged"); + let zero_bin = ExportPolicy { suppression_threshold: 0, quantization_bin: 0 }; + assert_eq!(zero_bin.apply(7), Some(7), "bin of 0 is treated as 1"); + let large = ExportPolicy { suppression_threshold: 0, quantization_bin: 5 }; + assert_eq!(large.apply(u64::MAX), Some(u64::MAX / 5 * 5), "no overflow near u64::MAX"); + } } diff --git a/payjoin-mailroom/src/telemetry.rs b/payjoin-mailroom/src/telemetry.rs index 21912ea0b..3efe6cf21 100644 --- a/payjoin-mailroom/src/telemetry.rs +++ b/payjoin-mailroom/src/telemetry.rs @@ -10,9 +10,32 @@ use opentelemetry::KeyValue; use opentelemetry_http::hyper::HyperClient; use opentelemetry_http::{Bytes, HttpClient, HttpError, Request, Response}; use opentelemetry_otlp::{WithExportConfig, WithHttpConfig}; -use opentelemetry_sdk::metrics::SdkMeterProvider; +use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider}; use opentelemetry_sdk::Resource; +/// How often the exporter pushes to the collection endpoint. +/// +/// Exported values cover one completed UTC reporting week. Daily delivery +/// retries the same frozen value, avoiding a single weekly delivery attempt +/// without exposing daily traffic volume. +const EXPORT_INTERVAL: Duration = Duration::from_secs(24 * 60 * 60); + +/// Resource attributes attached to every exported metric. +/// +/// Built from an empty resource rather than the SDK default so nothing about +/// the host environment leaks into the export: the default builder includes +/// an environment detector (`OTEL_RESOURCE_ATTRIBUTES`) through which +/// hostnames or other identifying attributes could silently join the stream. +/// The export carries exactly the service name and a Foundation-issued opaque +/// reporter ID. The ID is necessary only until a collector aggregates reports; +/// its mapping to an operator belongs outside Grafana. +pub(crate) fn export_resource(reporter_id: &str) -> Resource { + Resource::builder_empty() + .with_service_name("payjoin-mailroom") + .with_attribute(KeyValue::new("reporter.id", reporter_id.to_string())) + .build() +} + /// Build an OTLP/HTTP `SdkMeterProvider` pinned to the mailroom's `ring` /// crypto provider. /// @@ -35,12 +58,9 @@ use opentelemetry_sdk::Resource; pub fn build_otlp_meter_provider( endpoint: &str, auth_token: &str, - operator_domain: &str, + reporter_id: &str, ) -> SdkMeterProvider { - let resource = Resource::builder() - .with_service_name("payjoin-mailroom") - .with_attribute(KeyValue::new("operator.domain", operator_domain.to_string())) - .build(); + let resource = export_resource(reporter_id); let headers: std::collections::HashMap = [("Authorization".to_string(), format!("Basic {auth_token}"))].into(); @@ -71,10 +91,9 @@ pub fn build_otlp_meter_provider( .build() .expect("Failed to build OTLP metric exporter"); - SdkMeterProvider::builder() - .with_periodic_exporter(metric_exporter) - .with_resource(resource) - .build() + let reader = PeriodicReader::builder(metric_exporter).with_interval(EXPORT_INTERVAL).build(); + + SdkMeterProvider::builder().with_reader(reader).with_resource(resource).build() } /// `HttpClient` adapter that runs the inner client on a captured Tokio handle. @@ -119,6 +138,23 @@ mod tests { use super::*; use crate::metrics::MetricsService; + #[test] + fn exporter_retries_frozen_weekly_values_daily() { + assert_eq!(EXPORT_INTERVAL, Duration::from_secs(24 * 60 * 60)); + } + + /// The exported resource must carry exactly the service name and the + /// Foundation-issued opaque reporting identifier. Anything else (hostname, + /// IP, instance id, domain, env-injected attributes) could identify the operator's + /// infrastructure, so this pins the exact key set rather than a subset. + #[test] + fn export_resource_carries_only_allowlisted_attributes() { + let resource = export_resource("reporter-opaque-test-id"); + let mut keys: Vec<&str> = resource.iter().map(|(key, _)| key.as_str()).collect(); + keys.sort_unstable(); + assert_eq!(keys, vec!["reporter.id", "service.name"]); + } + /// Regression test for the OTLP transport swap (opentelemetry 0.32). /// /// Replaces the manual `mock_otlp.py` + `curl` + 65s-wait smoke test.