hdds_c/
telemetry.rs

1// SPDX-License-Identifier: Apache-2.0 OR MIT
2// Copyright (c) 2025-2026 naskel.com
3
4//! Telemetry and metrics collection for HDDS C FFI
5
6use std::ffi::CStr;
7use std::os::raw::c_char;
8use std::ptr;
9use std::sync::Arc;
10
11use super::HddsError;
12
13/// Opaque handle to a MetricsCollector
14#[repr(C)]
15pub struct HddsMetrics {
16    _private: [u8; 0],
17}
18
19/// Opaque handle to a telemetry Exporter
20#[repr(C)]
21pub struct HddsTelemetryExporter {
22    _private: [u8; 0],
23}
24
25/// Telemetry metrics snapshot (C-compatible)
26#[repr(C)]
27#[derive(Debug, Clone, Copy, Default)]
28pub struct HddsMetricsSnapshot {
29    /// Timestamp in nanoseconds since epoch
30    pub timestamp_ns: u64,
31    /// Total messages sent
32    pub messages_sent: u64,
33    /// Total messages received
34    pub messages_received: u64,
35    /// Total messages dropped
36    pub messages_dropped: u64,
37    /// Total bytes sent
38    pub bytes_sent: u64,
39    /// Latency p50 in nanoseconds
40    pub latency_p50_ns: u64,
41    /// Latency p99 in nanoseconds
42    pub latency_p99_ns: u64,
43    /// Latency p999 in nanoseconds
44    pub latency_p999_ns: u64,
45    /// Merge full count (backpressure events)
46    pub merge_full_count: u64,
47    /// Would-block count (send buffer full)
48    pub would_block_count: u64,
49}
50
51// =============================================================================
52// Global Metrics
53// =============================================================================
54
55/// Initialize the global metrics collector
56///
57/// Creates a thread-safe metrics collector for the entire HDDS instance.
58/// Safe to call multiple times - subsequent calls return the same instance.
59///
60/// # Safety
61/// The returned handle must be released with `hdds_telemetry_release`.
62///
63/// # Returns
64/// Handle to the metrics collector, or NULL on error
65#[no_mangle]
66pub unsafe extern "C" fn hdds_telemetry_init() -> *mut HddsMetrics {
67    let metrics = hdds::telemetry::init_metrics();
68    Arc::into_raw(metrics) as *mut HddsMetrics
69}
70
71/// Get the global metrics collector (if initialized)
72///
73/// # Safety
74/// The returned handle must be released with `hdds_telemetry_release`.
75///
76/// # Returns
77/// Handle to metrics collector, or NULL if not initialized
78#[no_mangle]
79pub unsafe extern "C" fn hdds_telemetry_get() -> *mut HddsMetrics {
80    match hdds::telemetry::get_metrics_opt() {
81        Some(metrics) => Arc::into_raw(metrics) as *mut HddsMetrics,
82        None => ptr::null_mut(),
83    }
84}
85
86/// Release a metrics handle
87///
88/// # Safety
89/// - `metrics` must be a valid pointer from `hdds_telemetry_init` or `hdds_telemetry_get`
90#[no_mangle]
91pub unsafe extern "C" fn hdds_telemetry_release(metrics: *mut HddsMetrics) {
92    if !metrics.is_null() {
93        let _ = Arc::from_raw(metrics.cast::<hdds::telemetry::MetricsCollector>());
94    }
95}
96
97/// Take a snapshot of current metrics
98///
99/// # Safety
100/// - `metrics` must be a valid handle
101/// - `out` must be a valid pointer to `HddsMetricsSnapshot`
102///
103/// # Returns
104/// `HddsError::HddsOk` on success
105#[no_mangle]
106pub unsafe extern "C" fn hdds_telemetry_snapshot(
107    metrics: *mut HddsMetrics,
108    out: *mut HddsMetricsSnapshot,
109) -> HddsError {
110    if metrics.is_null() || out.is_null() {
111        return HddsError::HddsInvalidArgument;
112    }
113
114    // Clone the Arc to avoid consuming it
115    let arc = Arc::from_raw(metrics.cast::<hdds::telemetry::MetricsCollector>());
116    let clone = arc.clone();
117    let _ = Arc::into_raw(arc); // Put it back
118
119    let frame = clone.snapshot();
120
121    // Extract fields from frame
122    let mut snapshot = HddsMetricsSnapshot {
123        timestamp_ns: frame.ts_ns,
124        ..Default::default()
125    };
126
127    use hdds::telemetry::metrics::*;
128    for field in &frame.fields {
129        match field.tag {
130            TAG_MESSAGES_SENT => snapshot.messages_sent = field.value_u64,
131            TAG_MESSAGES_RECEIVED => snapshot.messages_received = field.value_u64,
132            TAG_MESSAGES_DROPPED => snapshot.messages_dropped = field.value_u64,
133            TAG_BYTES_SENT => snapshot.bytes_sent = field.value_u64,
134            TAG_LATENCY_P50 => snapshot.latency_p50_ns = field.value_u64,
135            TAG_LATENCY_P99 => snapshot.latency_p99_ns = field.value_u64,
136            TAG_LATENCY_P999 => snapshot.latency_p999_ns = field.value_u64,
137            TAG_MERGE_FULL_COUNT => snapshot.merge_full_count = field.value_u64,
138            TAG_WOULD_BLOCK_COUNT => snapshot.would_block_count = field.value_u64,
139            _ => {}
140        }
141    }
142
143    *out = snapshot;
144    HddsError::HddsOk
145}
146
147/// Record a latency sample
148///
149/// # Safety
150/// - `metrics` must be a valid handle
151///
152/// # Arguments
153/// * `start_ns` - Start timestamp in nanoseconds
154/// * `end_ns` - End timestamp in nanoseconds
155#[no_mangle]
156pub unsafe extern "C" fn hdds_telemetry_record_latency(
157    metrics: *mut HddsMetrics,
158    start_ns: u64,
159    end_ns: u64,
160) {
161    if metrics.is_null() {
162        return;
163    }
164
165    let arc = Arc::from_raw(metrics.cast::<hdds::telemetry::MetricsCollector>());
166    arc.add_latency_sample(start_ns, end_ns);
167    let _ = Arc::into_raw(arc);
168}
169
170// =============================================================================
171// Telemetry Exporter (TCP streaming server)
172// =============================================================================
173
174/// Start the telemetry export server
175///
176/// Creates a TCP server that streams metrics to connected clients (e.g., HDDS Viewer).
177///
178/// # Safety
179/// - `bind_addr` must be a valid null-terminated C string.
180/// - The returned handle must be released with `hdds_telemetry_stop_exporter`.
181///
182/// # Arguments
183/// * `bind_addr` - IP address to bind (e.g., "127.0.0.1" or "0.0.0.0")
184/// * `port` - Port number (default: 4242)
185///
186/// # Returns
187/// Handle to exporter, or NULL on error
188#[no_mangle]
189pub unsafe extern "C" fn hdds_telemetry_start_exporter(
190    bind_addr: *const c_char,
191    port: u16,
192) -> *mut HddsTelemetryExporter {
193    if bind_addr.is_null() {
194        return ptr::null_mut();
195    }
196
197    let Ok(addr_str) = CStr::from_ptr(bind_addr).to_str() else {
198        return ptr::null_mut();
199    };
200
201    match hdds::telemetry::init_exporter(addr_str, port) {
202        Ok(exporter) => Arc::into_raw(exporter) as *mut HddsTelemetryExporter,
203        Err(e) => {
204            log::error!("Failed to start telemetry exporter: {}", e);
205            ptr::null_mut()
206        }
207    }
208}
209
210/// Stop and release the telemetry exporter
211///
212/// # Safety
213/// - `exporter` must be a valid pointer from `hdds_telemetry_start_exporter`
214#[no_mangle]
215pub unsafe extern "C" fn hdds_telemetry_stop_exporter(exporter: *mut HddsTelemetryExporter) {
216    if !exporter.is_null() {
217        let _ = Arc::from_raw(exporter.cast::<hdds::telemetry::Exporter>());
218    }
219}
hdds_c/telemetry.rs

hdds_c/
telemetry.rs
