From fd0c87c787d0626b3546fa571541c9c809711821 Mon Sep 17 00:00:00 2001 From: DanConwayDev Date: Thu, 4 Dec 2025 15:17:04 +0000 Subject: add prometheus metrics --- src/metrics/bandwidth.rs | 301 +++++++++++++++++++++++++++++ src/metrics/connection.rs | 337 +++++++++++++++++++++++++++++++++ src/metrics/mod.rs | 469 ++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 1107 insertions(+) create mode 100644 src/metrics/bandwidth.rs create mode 100644 src/metrics/connection.rs create mode 100644 src/metrics/mod.rs (limited to 'src/metrics') diff --git a/src/metrics/bandwidth.rs b/src/metrics/bandwidth.rs new file mode 100644 index 0000000..d2c53e8 --- /dev/null +++ b/src/metrics/bandwidth.rs @@ -0,0 +1,301 @@ +//! Repository bandwidth tracking with cardinality control. +//! +//! This module tracks bandwidth per repository but only exposes the top N +//! repositories to Prometheus to prevent cardinality explosion with many repos. +//! +//! # Cardinality Control +//! +//! - All per-repo bandwidth is tracked internally in a `DashMap` +//! - Every 60 seconds, the top 10 are calculated and exposed to Prometheus +//! - Previous repo labels are cleared before setting new ones +//! - Prometheus only ever sees ~10 label values, keeping cardinality low + +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::{Duration, Instant}; + +use dashmap::DashMap; +use prometheus::{GaugeVec, Opts, Registry}; + +/// Default number of top repositories to expose in metrics +const DEFAULT_TOP_N: usize = 10; + +/// Default refresh interval for top-N calculation (60 seconds) +const DEFAULT_REFRESH_INTERVAL: Duration = Duration::from_secs(60); + +/// Tracks bandwidth per repository with top-N exposure to Prometheus. +/// +/// # Design +/// +/// All repositories are tracked internally for accurate total bandwidth, +/// but only the top N by bytes transferred are exposed to Prometheus. +/// This prevents cardinality explosion when hosting thousands of repositories. +/// +/// # Thread Safety +/// +/// Uses `DashMap` for lock-free concurrent access and atomics for +/// the refresh timestamp. +pub struct BandwidthTracker { + /// Internal: tracks ALL repos (memory only, not exposed) + all_repos: DashMap, + + /// Exposed to Prometheus: only top N repos + top_repos_gauge: GaugeVec, + + /// Last refresh timestamp (stored as nanos since some epoch) + last_refresh_nanos: AtomicU64, + + /// Instant when the tracker was created (for relative timing) + start_instant: Instant, + + /// Number of top repos to expose + top_n: usize, + + /// Refresh interval + refresh_interval: Duration, +} + +impl BandwidthTracker { + /// Creates a new BandwidthTracker and registers metrics with Prometheus. + /// + /// Uses default settings: + /// - Top 10 repositories exposed + /// - 60 second refresh interval + pub fn new(registry: &Registry) -> Self { + Self::with_config(registry, DEFAULT_TOP_N, DEFAULT_REFRESH_INTERVAL) + } + + /// Creates a new BandwidthTracker with custom configuration. + /// + /// # Arguments + /// + /// * `registry` - Prometheus registry to register metrics with + /// * `top_n` - Number of top repositories to expose in metrics + /// * `refresh_interval` - How often to recalculate the top-N list + pub fn with_config(registry: &Registry, top_n: usize, refresh_interval: Duration) -> Self { + let top_repos_gauge = GaugeVec::new( + Opts::new( + "ngit_git_top_repos_bytes", + "Top repositories by bandwidth (refreshed periodically)", + ), + &["repo"], + ) + .unwrap(); + registry.register(Box::new(top_repos_gauge.clone())).unwrap(); + + Self { + all_repos: DashMap::new(), + top_repos_gauge, + last_refresh_nanos: AtomicU64::new(0), + start_instant: Instant::now(), + top_n, + refresh_interval, + } + } + + /// Records bytes transferred for a repository. + /// + /// # Arguments + /// + /// * `repo_id` - Repository identifier (e.g., npub or repo name) + /// * `bytes` - Number of bytes transferred + pub fn record_transfer(&self, repo_id: &str, bytes: u64) { + self.all_repos + .entry(repo_id.to_string()) + .and_modify(|v| *v = v.saturating_add(bytes)) + .or_insert(bytes); + } + + /// Conditionally refreshes the top-N list if the refresh interval has elapsed. + /// + /// This method is designed to be called frequently (e.g., on every + /// `/metrics` request) without performance impact - it only does work + /// when the refresh interval has elapsed. + pub fn maybe_refresh_top_n(&self) { + let elapsed_nanos = self.start_instant.elapsed().as_nanos() as u64; + let last_refresh = self.last_refresh_nanos.load(Ordering::Relaxed); + let interval_nanos = self.refresh_interval.as_nanos() as u64; + + // Check if enough time has passed since last refresh + if elapsed_nanos.saturating_sub(last_refresh) >= interval_nanos { + // Try to update the timestamp atomically to prevent concurrent refreshes + if self + .last_refresh_nanos + .compare_exchange(last_refresh, elapsed_nanos, Ordering::SeqCst, Ordering::Relaxed) + .is_ok() + { + self.refresh_top_n(); + } + } + } + + /// Forces a refresh of the top-N list. + /// + /// This recalculates which repositories are in the top N by bandwidth + /// and updates the Prometheus gauges accordingly. + pub fn refresh_top_n(&self) { + // Collect all repo data + let mut sorted: Vec<_> = self + .all_repos + .iter() + .map(|r| (r.key().clone(), *r.value())) + .collect(); + + // Sort by bytes descending + sorted.sort_by(|a, b| b.1.cmp(&a.1)); + + // Clear old labels and set new top N + self.top_repos_gauge.reset(); + for (repo, bytes) in sorted.into_iter().take(self.top_n) { + self.top_repos_gauge + .with_label_values(&[&repo]) + .set(bytes as f64); + } + } + + /// Returns the total bytes transferred for a specific repository. + /// + /// Returns `None` if the repository has not been seen. + pub fn get_repo_bytes(&self, repo_id: &str) -> Option { + self.all_repos.get(repo_id).map(|v| *v) + } + + /// Returns the total bytes transferred across all repositories. + pub fn total_bytes(&self) -> u64 { + self.all_repos.iter().map(|r| *r.value()).sum() + } + + /// Returns the number of repositories being tracked. + pub fn repo_count(&self) -> usize { + self.all_repos.len() + } + + /// Returns the top N repositories by bandwidth. + /// + /// This is a snapshot and may not match the Prometheus gauges if + /// a refresh hasn't occurred recently. + pub fn get_top_repos(&self) -> Vec<(String, u64)> { + let mut sorted: Vec<_> = self + .all_repos + .iter() + .map(|r| (r.key().clone(), *r.value())) + .collect(); + + sorted.sort_by(|a, b| b.1.cmp(&a.1)); + sorted.truncate(self.top_n); + sorted + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn test_registry() -> Registry { + Registry::new() + } + + #[test] + fn test_bandwidth_tracking() { + let registry = test_registry(); + let tracker = BandwidthTracker::new(®istry); + + // Record transfers + tracker.record_transfer("repo-a", 1000); + tracker.record_transfer("repo-b", 2000); + tracker.record_transfer("repo-a", 500); // Additional transfer to repo-a + + assert_eq!(tracker.get_repo_bytes("repo-a"), Some(1500)); + assert_eq!(tracker.get_repo_bytes("repo-b"), Some(2000)); + assert_eq!(tracker.get_repo_bytes("repo-c"), None); + assert_eq!(tracker.total_bytes(), 3500); + assert_eq!(tracker.repo_count(), 2); + } + + #[test] + fn test_top_n_repos() { + let registry = test_registry(); + let tracker = BandwidthTracker::with_config(®istry, 3, Duration::from_secs(60)); + + // Create 5 repos with different bandwidth + tracker.record_transfer("repo-1", 100); + tracker.record_transfer("repo-2", 500); + tracker.record_transfer("repo-3", 200); + tracker.record_transfer("repo-4", 800); + tracker.record_transfer("repo-5", 300); + + let top = tracker.get_top_repos(); + assert_eq!(top.len(), 3); + assert_eq!(top[0], ("repo-4".to_string(), 800)); + assert_eq!(top[1], ("repo-2".to_string(), 500)); + assert_eq!(top[2], ("repo-5".to_string(), 300)); + } + + #[test] + fn test_refresh_updates_gauge() { + let registry = test_registry(); + let tracker = BandwidthTracker::new(®istry); + + tracker.record_transfer("high-bandwidth-repo", 10_000_000); + tracker.record_transfer("low-bandwidth-repo", 1000); + + // Force a refresh + tracker.refresh_top_n(); + + // Verify the gauge values (we can't easily access them directly, + // but we can verify the tracker state is correct) + assert_eq!(tracker.repo_count(), 2); + assert_eq!(tracker.total_bytes(), 10_001_000); + } + + #[test] + fn test_saturating_add() { + let registry = test_registry(); + let tracker = BandwidthTracker::new(®istry); + + // Test that we don't overflow + tracker.record_transfer("huge-repo", u64::MAX - 100); + tracker.record_transfer("huge-repo", 200); + + // Should saturate to MAX, not overflow + assert_eq!(tracker.get_repo_bytes("huge-repo"), Some(u64::MAX)); + } + + #[test] + fn test_maybe_refresh_respects_interval() { + let registry = test_registry(); + // Use a very short interval for testing + let tracker = BandwidthTracker::with_config(®istry, 10, Duration::from_millis(10)); + + tracker.record_transfer("repo-a", 1000); + + // First call should trigger refresh (no previous refresh) + tracker.maybe_refresh_top_n(); + + // Add more data + tracker.record_transfer("repo-b", 2000); + + // Immediate second call should NOT trigger refresh + let count_before = tracker.repo_count(); + tracker.maybe_refresh_top_n(); + assert_eq!(tracker.repo_count(), count_before); + + // Wait for interval to pass + std::thread::sleep(Duration::from_millis(15)); + + // Now it should refresh + tracker.maybe_refresh_top_n(); + } + + #[test] + fn test_empty_tracker() { + let registry = test_registry(); + let tracker = BandwidthTracker::new(®istry); + + assert_eq!(tracker.total_bytes(), 0); + assert_eq!(tracker.repo_count(), 0); + assert!(tracker.get_top_repos().is_empty()); + + // Refresh should not panic on empty data + tracker.refresh_top_n(); + } +} \ No newline at end of file diff --git a/src/metrics/connection.rs b/src/metrics/connection.rs new file mode 100644 index 0000000..6a7f406 --- /dev/null +++ b/src/metrics/connection.rs @@ -0,0 +1,337 @@ +//! Connection tracking with privacy-preserving abuse detection. +//! +//! This module tracks WebSocket connections per IP address internally for abuse +//! detection, but NEVER exposes IP addresses in Prometheus metrics. Only aggregate +//! counts are exposed. +//! +//! # Privacy Model +//! +//! | Data | Location | Exposed? | +//! |------|----------|----------| +//! | Total connections | Prometheus | ✅ Yes | +//! | Unique IP count | Prometheus | ✅ Yes | +//! | Flagged abuser count | Prometheus | ✅ Yes | +//! | Actual IP addresses | Internal HashMap | ❌ No | +//! | IP + abuse flag | Logs (when flagged) | ⚠️ Logs only | + +use std::net::IpAddr; +use std::time::Instant; + +use dashmap::DashMap; +use prometheus::{IntGauge, Opts, Registry}; +use tracing::warn; + +/// Information about connections from a specific IP address. +struct ConnectionInfo { + /// Number of active connections from this IP + count: u32, + /// When the first connection from this IP was established + first_seen: Instant, + /// Whether this IP has been flagged as potentially abusive + flagged_as_abuse: bool, +} + +/// Tracks WebSocket connections per IP with abuse detection. +/// +/// # Thread Safety +/// +/// Uses `DashMap` for lock-free concurrent access, as connection tracking +/// happens across multiple tokio tasks. +/// +/// # Privacy +/// +/// IP addresses are stored internally only for abuse detection and are +/// NEVER exposed in Prometheus metrics. Only aggregate counts are exposed: +/// - Total active connections +/// - Number of unique IPs +/// - Number of IPs flagged as potential abusers +pub struct ConnectionTracker { + /// Active connections per IP (INTERNAL ONLY - never exposed to metrics) + connections: DashMap, + + /// Threshold for abuse flagging (connections per IP) + abuse_threshold: u32, + + /// Prometheus gauge: total active connections + active_connections: IntGauge, + + /// Prometheus gauge: number of unique IPs connected + unique_ips: IntGauge, + + /// Prometheus gauge: number of IPs flagged as potential abusers + flagged_abusers: IntGauge, +} + +impl ConnectionTracker { + /// Creates a new ConnectionTracker and registers metrics with Prometheus. + /// + /// # Arguments + /// + /// * `abuse_threshold` - Number of connections from a single IP before flagging + /// * `registry` - Prometheus registry to register metrics with + pub fn new(abuse_threshold: u32, registry: &Registry) -> Self { + let active_connections = IntGauge::with_opts( + Opts::new( + "ngit_websocket_connections_active", + "Current active WebSocket connections", + ) + ).unwrap(); + registry.register(Box::new(active_connections.clone())).unwrap(); + + let unique_ips = IntGauge::with_opts( + Opts::new( + "ngit_websocket_unique_ips", + "Number of unique IP addresses connected (NOT the IPs themselves)", + ) + ).unwrap(); + registry.register(Box::new(unique_ips.clone())).unwrap(); + + let flagged_abusers = IntGauge::with_opts( + Opts::new( + "ngit_websocket_flagged_abusers", + "Number of IPs exceeding connection threshold", + ) + ).unwrap(); + registry.register(Box::new(flagged_abusers.clone())).unwrap(); + + Self { + connections: DashMap::new(), + abuse_threshold, + active_connections, + unique_ips, + flagged_abusers, + } + } + + /// Called when a new WebSocket connection is established. + /// + /// This method: + /// 1. Increments the connection count for this IP + /// 2. Checks if the IP has exceeded the abuse threshold + /// 3. Logs a warning if abuse is detected (IP is logged here only) + /// 4. Updates Prometheus metrics (aggregate counts only) + /// + /// # Privacy + /// + /// The IP address is logged only when abuse is detected. It is NEVER + /// exposed in Prometheus metrics. + pub fn on_connect(&self, ip: IpAddr) { + let mut is_new_ip = false; + let mut newly_flagged = false; + + self.connections + .entry(ip) + .and_modify(|info| { + info.count += 1; + // Check if this connection pushes us over the threshold + if !info.flagged_as_abuse && info.count >= self.abuse_threshold { + info.flagged_as_abuse = true; + newly_flagged = true; + } + }) + .or_insert_with(|| { + is_new_ip = true; + ConnectionInfo { + count: 1, + first_seen: Instant::now(), + flagged_as_abuse: false, + } + }); + + // Update Prometheus metrics (aggregate counts only) + self.active_connections.inc(); + + if is_new_ip { + self.unique_ips.inc(); + } + + if newly_flagged { + self.flagged_abusers.inc(); + // Log the abuse detection - IP is only exposed in logs, not metrics + warn!( + ip = %ip, + threshold = self.abuse_threshold, + "Potential abuse detected: IP exceeded connection threshold" + ); + } + } + + /// Called when a WebSocket connection is closed. + /// + /// This method: + /// 1. Decrements the connection count for this IP + /// 2. Removes the IP from tracking if count reaches 0 + /// 3. Updates the abuse flag count if the IP was flagged + /// 4. Updates Prometheus metrics (aggregate counts only) + pub fn on_disconnect(&self, ip: IpAddr) { + let mut remove_entry = false; + let mut was_flagged = false; + let mut had_connection = false; + + if let Some(mut entry) = self.connections.get_mut(&ip) { + had_connection = true; + entry.count = entry.count.saturating_sub(1); + if entry.count == 0 { + remove_entry = true; + was_flagged = entry.flagged_as_abuse; + } + } + + // Remove the entry if count is 0 + if remove_entry { + self.connections.remove(&ip); + self.unique_ips.dec(); + if was_flagged { + self.flagged_abusers.dec(); + } + } + + // Update total connections only if this IP had a tracked connection + if had_connection { + self.active_connections.dec(); + } + } + + /// Returns the current number of active connections. + pub fn active_connections(&self) -> u64 { + self.active_connections.get() as u64 + } + + /// Returns the current number of unique IPs. + pub fn unique_ip_count(&self) -> u64 { + self.unique_ips.get() as u64 + } + + /// Returns the current number of flagged abusers. + pub fn flagged_abuser_count(&self) -> u64 { + self.flagged_abusers.get() as u64 + } + + /// Returns the connection count for a specific IP (for internal use only). + /// + /// # Privacy + /// + /// This is an internal method. The returned data should NEVER be exposed + /// in metrics or logs without privacy consideration. + #[cfg(test)] + pub(crate) fn connection_count(&self, ip: &IpAddr) -> Option { + self.connections.get(ip).map(|info| info.count) + } + + /// Returns whether an IP is flagged as abusive (for internal use only). + #[cfg(test)] + pub(crate) fn is_flagged(&self, ip: &IpAddr) -> bool { + self.connections + .get(ip) + .map(|info| info.flagged_as_abuse) + .unwrap_or(false) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::net::{Ipv4Addr, Ipv6Addr}; + + fn test_registry() -> Registry { + Registry::new() + } + + #[test] + fn test_connection_tracking() { + let registry = test_registry(); + let tracker = ConnectionTracker::new(5, ®istry); + let ip = IpAddr::V4(Ipv4Addr::new(192, 168, 1, 1)); + + // Connect + tracker.on_connect(ip); + assert_eq!(tracker.active_connections(), 1); + assert_eq!(tracker.unique_ip_count(), 1); + assert_eq!(tracker.connection_count(&ip), Some(1)); + + // Connect again from same IP + tracker.on_connect(ip); + assert_eq!(tracker.active_connections(), 2); + assert_eq!(tracker.unique_ip_count(), 1); // Still 1 unique IP + assert_eq!(tracker.connection_count(&ip), Some(2)); + + // Disconnect one + tracker.on_disconnect(ip); + assert_eq!(tracker.active_connections(), 1); + assert_eq!(tracker.unique_ip_count(), 1); + assert_eq!(tracker.connection_count(&ip), Some(1)); + + // Disconnect last + tracker.on_disconnect(ip); + assert_eq!(tracker.active_connections(), 0); + assert_eq!(tracker.unique_ip_count(), 0); + assert_eq!(tracker.connection_count(&ip), None); + } + + #[test] + fn test_multiple_ips() { + let registry = test_registry(); + let tracker = ConnectionTracker::new(5, ®istry); + let ip1 = IpAddr::V4(Ipv4Addr::new(192, 168, 1, 1)); + let ip2 = IpAddr::V4(Ipv4Addr::new(192, 168, 1, 2)); + let ip3 = IpAddr::V6(Ipv6Addr::new(0x2001, 0xdb8, 0, 0, 0, 0, 0, 1)); + + tracker.on_connect(ip1); + tracker.on_connect(ip2); + tracker.on_connect(ip3); + + assert_eq!(tracker.active_connections(), 3); + assert_eq!(tracker.unique_ip_count(), 3); + + tracker.on_disconnect(ip2); + assert_eq!(tracker.active_connections(), 2); + assert_eq!(tracker.unique_ip_count(), 2); + } + + #[test] + fn test_abuse_detection() { + let registry = test_registry(); + let threshold = 3; + let tracker = ConnectionTracker::new(threshold, ®istry); + let abuser_ip = IpAddr::V4(Ipv4Addr::new(10, 0, 0, 1)); + let normal_ip = IpAddr::V4(Ipv4Addr::new(10, 0, 0, 2)); + + // Normal user with 1 connection + tracker.on_connect(normal_ip); + assert!(!tracker.is_flagged(&normal_ip)); + assert_eq!(tracker.flagged_abuser_count(), 0); + + // Abuser approaching threshold + tracker.on_connect(abuser_ip); + tracker.on_connect(abuser_ip); + assert!(!tracker.is_flagged(&abuser_ip)); + assert_eq!(tracker.flagged_abuser_count(), 0); + + // Abuser hits threshold + tracker.on_connect(abuser_ip); + assert!(tracker.is_flagged(&abuser_ip)); + assert_eq!(tracker.flagged_abuser_count(), 1); + + // Normal user still not flagged + assert!(!tracker.is_flagged(&normal_ip)); + + // Abuser disconnects all - should be removed from flagged count + tracker.on_disconnect(abuser_ip); + tracker.on_disconnect(abuser_ip); + tracker.on_disconnect(abuser_ip); + assert_eq!(tracker.flagged_abuser_count(), 0); + assert_eq!(tracker.active_connections(), 1); // Only normal user remains + } + + #[test] + fn test_disconnect_without_connect() { + let registry = test_registry(); + let tracker = ConnectionTracker::new(5, ®istry); + let ip = IpAddr::V4(Ipv4Addr::new(192, 168, 1, 1)); + + // Disconnect without connect should not panic or go negative + tracker.on_disconnect(ip); + assert_eq!(tracker.active_connections(), 0); + assert_eq!(tracker.unique_ip_count(), 0); + } +} \ No newline at end of file diff --git a/src/metrics/mod.rs b/src/metrics/mod.rs new file mode 100644 index 0000000..4a4fe57 --- /dev/null +++ b/src/metrics/mod.rs @@ -0,0 +1,469 @@ +//! Prometheus metrics for ngit-grasp relay. +//! +//! This module provides comprehensive monitoring metrics including: +//! - WebSocket connection tracking (with privacy-preserving IP aggregation) +//! - Git operation metrics (clone, fetch, push) +//! - Repository bandwidth tracking (top-N only for cardinality control) +//! - Nostr event metrics +//! +//! # Privacy +//! IP addresses are NEVER exposed in metrics. The `ConnectionTracker` maintains +//! per-IP counts internally only for abuse detection. Only aggregate counts +//! are exposed to Prometheus. + +pub mod bandwidth; +pub mod connection; + +use std::sync::Arc; +use std::time::Instant; + +use lazy_static::lazy_static; +use prometheus::{ + Counter, CounterVec, Encoder, Gauge, GaugeVec, Histogram, HistogramOpts, HistogramVec, Opts, + Registry, TextEncoder, +}; + +use bandwidth::BandwidthTracker; +use connection::ConnectionTracker; + +lazy_static! { + /// Global Prometheus registry for ngit-grasp metrics + pub static ref REGISTRY: Registry = Registry::new(); +} + +/// Central metrics collection for ngit-grasp relay. +/// +/// Thread-safe and designed for concurrent access from multiple tokio tasks. +#[derive(Clone)] +pub struct Metrics { + inner: Arc, +} + +struct MetricsInner { + /// Connection tracking with abuse detection + pub connection_tracker: ConnectionTracker, + + /// Repository bandwidth tracking (top-N only) + pub bandwidth_tracker: BandwidthTracker, + + // === WebSocket Metrics === + /// Total WebSocket connections since startup + pub websocket_connections_total: Counter, + /// Connection duration histogram + pub websocket_connection_duration: Histogram, + /// Messages received by type (REQ, EVENT, CLOSE) + pub websocket_messages_received: CounterVec, + /// Messages sent by type (EVENT, EOSE, OK, NOTICE) + pub websocket_messages_sent: CounterVec, + + // === Git Operation Metrics === + /// Git operations by type and status + pub git_operations_total: CounterVec, + /// Git operation duration histogram + pub git_operation_duration: HistogramVec, + /// Total bytes transferred + pub git_bytes_total: CounterVec, + /// Push authorization results + pub git_push_authorization: CounterVec, + + // === Nostr Event Metrics === + /// Events received by kind + pub events_received_total: CounterVec, + /// Events successfully stored by kind + pub events_stored_total: CounterVec, + /// Events rejected by kind and reason + pub events_rejected_total: CounterVec, + + // === Repository Metrics === + /// Total repositories hosted + pub repositories_total: Gauge, + + // === System Health Metrics === + /// Server start time for uptime calculation + pub start_time: Instant, + /// Build information gauge + pub build_info: GaugeVec, +} + +impl Metrics { + /// Creates a new Metrics instance and registers all metrics with Prometheus. + /// + /// # Arguments + /// * `abuse_threshold` - Number of connections from a single IP before flagging as abuse + pub fn new(abuse_threshold: u32) -> Self { + let inner = MetricsInner::new(abuse_threshold); + Self { + inner: Arc::new(inner), + } + } + + /// Returns the connection tracker for WebSocket connection management. + pub fn connection_tracker(&self) -> &ConnectionTracker { + &self.inner.connection_tracker + } + + /// Returns the bandwidth tracker for repository bandwidth tracking. + pub fn bandwidth_tracker(&self) -> &BandwidthTracker { + &self.inner.bandwidth_tracker + } + + // === WebSocket Recording Methods === + + /// Record a new WebSocket connection + pub fn record_websocket_connection(&self) { + self.inner.websocket_connections_total.inc(); + } + + /// Start timing a WebSocket connection, returns timer that records on drop + pub fn start_connection_timer(&self) -> HistogramTimer { + HistogramTimer::new(self.inner.websocket_connection_duration.clone()) + } + + /// Record a received WebSocket message + pub fn record_message_received(&self, msg_type: &str) { + self.inner + .websocket_messages_received + .with_label_values(&[msg_type]) + .inc(); + } + + /// Record a sent WebSocket message + pub fn record_message_sent(&self, msg_type: &str) { + self.inner + .websocket_messages_sent + .with_label_values(&[msg_type]) + .inc(); + } + + // === Git Operation Recording Methods === + + /// Record a git operation completion + pub fn record_git_operation(&self, operation: &str, status: &str) { + self.inner + .git_operations_total + .with_label_values(&[operation, status]) + .inc(); + } + + /// Start timing a git operation, returns a timer + pub fn start_git_operation_timer(&self, operation: &str) -> GitOperationTimer { + GitOperationTimer::new(self.inner.git_operation_duration.clone(), operation.to_string()) + } + + /// Record bytes transferred for a git operation + pub fn record_git_bytes(&self, direction: &str, bytes: u64) { + self.inner + .git_bytes_total + .with_label_values(&[direction]) + .inc_by(bytes as f64); + } + + /// Record a push authorization result + pub fn record_push_authorization(&self, result: &str) { + self.inner + .git_push_authorization + .with_label_values(&[result]) + .inc(); + } + + // === Nostr Event Recording Methods === + + /// Record a received Nostr event + pub fn record_event_received(&self, kind: u64) { + self.inner + .events_received_total + .with_label_values(&[&kind.to_string()]) + .inc(); + } + + /// Record a stored Nostr event + pub fn record_event_stored(&self, kind: u64) { + self.inner + .events_stored_total + .with_label_values(&[&kind.to_string()]) + .inc(); + } + + /// Record a rejected Nostr event + pub fn record_event_rejected(&self, kind: u64, reason: &str) { + self.inner + .events_rejected_total + .with_label_values(&[&kind.to_string(), reason]) + .inc(); + } + + // === Repository Metrics === + + /// Set the total number of repositories + pub fn set_repositories_total(&self, count: u64) { + self.inner.repositories_total.set(count as f64); + } + + /// Increment the repository count + pub fn inc_repositories_total(&self) { + self.inner.repositories_total.inc(); + } + + // === Rendering === + + /// Render all metrics in Prometheus text format. + /// + /// This method: + /// 1. Refreshes the top-N bandwidth metrics if needed + /// 2. Updates uptime + /// 3. Gathers all metrics from the registry + /// 4. Encodes them in Prometheus text format + pub fn render(&self) -> String { + // Refresh top-N bandwidth repos if needed + self.inner.bandwidth_tracker.maybe_refresh_top_n(); + + // Gather and encode metrics + let encoder = TextEncoder::new(); + let metric_families = REGISTRY.gather(); + let mut buffer = Vec::new(); + encoder.encode(&metric_families, &mut buffer).unwrap(); + + // Add uptime as a comment (it's derived, not a registered metric) + let uptime = self.inner.start_time.elapsed().as_secs(); + let mut output = String::from_utf8(buffer).unwrap(); + output.push_str(&format!( + "\n# HELP ngit_uptime_seconds Seconds since server startup\n# TYPE ngit_uptime_seconds counter\nngit_uptime_seconds {}\n", + uptime + )); + + output + } + + /// Check if the system is under high load (for sync scheduling) + pub fn is_high_load(&self, threshold: u64) -> bool { + self.inner.connection_tracker.active_connections() > threshold + } +} + +impl MetricsInner { + fn new(abuse_threshold: u32) -> Self { + // Create connection tracker + let connection_tracker = ConnectionTracker::new(abuse_threshold, ®ISTRY); + + // Create bandwidth tracker + let bandwidth_tracker = BandwidthTracker::new(®ISTRY); + + // WebSocket metrics + let websocket_connections_total = Counter::with_opts( + Opts::new( + "ngit_websocket_connections_total", + "Total WebSocket connections since startup", + ) + ).unwrap(); + REGISTRY.register(Box::new(websocket_connections_total.clone())).unwrap(); + + let websocket_connection_duration = Histogram::with_opts( + HistogramOpts::new( + "ngit_websocket_connection_duration_seconds", + "Duration of WebSocket connections", + ) + .buckets(vec![1.0, 5.0, 15.0, 30.0, 60.0, 300.0, 900.0, 3600.0]), + ).unwrap(); + REGISTRY.register(Box::new(websocket_connection_duration.clone())).unwrap(); + + let websocket_messages_received = CounterVec::new( + Opts::new( + "ngit_websocket_messages_received_total", + "WebSocket messages received by type", + ), + &["type"], + ).unwrap(); + REGISTRY.register(Box::new(websocket_messages_received.clone())).unwrap(); + + let websocket_messages_sent = CounterVec::new( + Opts::new( + "ngit_websocket_messages_sent_total", + "WebSocket messages sent by type", + ), + &["type"], + ).unwrap(); + REGISTRY.register(Box::new(websocket_messages_sent.clone())).unwrap(); + + // Git operation metrics + let git_operations_total = CounterVec::new( + Opts::new( + "ngit_git_operations_total", + "Git operations by type and status", + ), + &["operation", "status"], + ).unwrap(); + REGISTRY.register(Box::new(git_operations_total.clone())).unwrap(); + + let git_operation_duration = HistogramVec::new( + HistogramOpts::new( + "ngit_git_operation_duration_seconds", + "Duration of git operations", + ) + .buckets(vec![0.1, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0]), + &["operation"], + ).unwrap(); + REGISTRY.register(Box::new(git_operation_duration.clone())).unwrap(); + + let git_bytes_total = CounterVec::new( + Opts::new( + "ngit_git_bytes_total", + "Total bytes transferred for git operations", + ), + &["direction"], + ).unwrap(); + REGISTRY.register(Box::new(git_bytes_total.clone())).unwrap(); + + let git_push_authorization = CounterVec::new( + Opts::new( + "ngit_git_push_authorization_total", + "Push authorization results", + ), + &["result"], + ).unwrap(); + REGISTRY.register(Box::new(git_push_authorization.clone())).unwrap(); + + // Nostr event metrics + let events_received_total = CounterVec::new( + Opts::new( + "ngit_events_received_total", + "Nostr events received by kind", + ), + &["kind"], + ).unwrap(); + REGISTRY.register(Box::new(events_received_total.clone())).unwrap(); + + let events_stored_total = CounterVec::new( + Opts::new( + "ngit_events_stored_total", + "Nostr events successfully stored by kind", + ), + &["kind"], + ).unwrap(); + REGISTRY.register(Box::new(events_stored_total.clone())).unwrap(); + + let events_rejected_total = CounterVec::new( + Opts::new( + "ngit_events_rejected_total", + "Nostr events rejected by kind and reason", + ), + &["kind", "reason"], + ).unwrap(); + REGISTRY.register(Box::new(events_rejected_total.clone())).unwrap(); + + // Repository metrics + let repositories_total = Gauge::with_opts( + Opts::new( + "ngit_repositories_total", + "Total repositories hosted", + ) + ).unwrap(); + REGISTRY.register(Box::new(repositories_total.clone())).unwrap(); + + // Build info + let build_info = GaugeVec::new( + Opts::new( + "ngit_build_info", + "Build information", + ), + &["version", "commit"], + ).unwrap(); + REGISTRY.register(Box::new(build_info.clone())).unwrap(); + + // Set build info gauge to 1 (it's just for labels) + build_info + .with_label_values(&[env!("CARGO_PKG_VERSION"), option_env!("GIT_HASH").unwrap_or("unknown")]) + .set(1.0); + + Self { + connection_tracker, + bandwidth_tracker, + websocket_connections_total, + websocket_connection_duration, + websocket_messages_received, + websocket_messages_sent, + git_operations_total, + git_operation_duration, + git_bytes_total, + git_push_authorization, + events_received_total, + events_stored_total, + events_rejected_total, + repositories_total, + start_time: Instant::now(), + build_info, + } + } +} + +/// Timer for tracking WebSocket connection duration. +/// Records the elapsed time when dropped. +pub struct HistogramTimer { + histogram: Histogram, + start: Instant, +} + +impl HistogramTimer { + fn new(histogram: Histogram) -> Self { + Self { + histogram, + start: Instant::now(), + } + } +} + +impl Drop for HistogramTimer { + fn drop(&mut self) { + let elapsed = self.start.elapsed().as_secs_f64(); + self.histogram.observe(elapsed); + } +} + +/// Timer for tracking Git operation duration. +/// Records the elapsed time when dropped. +pub struct GitOperationTimer { + histogram_vec: HistogramVec, + operation: String, + start: Instant, +} + +impl GitOperationTimer { + fn new(histogram_vec: HistogramVec, operation: String) -> Self { + Self { + histogram_vec, + operation, + start: Instant::now(), + } + } +} + +impl Drop for GitOperationTimer { + fn drop(&mut self) { + let elapsed = self.start.elapsed().as_secs_f64(); + self.histogram_vec + .with_label_values(&[&self.operation]) + .observe(elapsed); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_metrics_creation() { + // Note: This test may fail if run with other tests due to global registry + // In production, consider using a test-specific registry + let metrics = Metrics::new(10); + + // Test that we can record metrics without panicking + metrics.record_websocket_connection(); + metrics.record_message_received("REQ"); + metrics.record_message_sent("EVENT"); + metrics.record_git_operation("clone", "success"); + metrics.record_git_bytes("in", 1024); + metrics.record_event_received(1); + metrics.record_event_stored(1); + metrics.record_event_rejected(1, "invalid_signature"); + metrics.set_repositories_total(5); + } +} \ No newline at end of file -- cgit v1.2.3