shunt/
telemetry.rs

1/// Fire-and-forget telemetry client.
2///
3/// Pushes request events and periodic heartbeats to a relay-server instance.
4/// All network failures are silently ignored — the relay is optional and must
5/// never block or degrade the proxy.
6use std::sync::Arc;
7use std::time::Duration;
8
9use serde_json::json;
10use tracing::debug;
11
12use crate::state::RequestLog;
13
14// ---------------------------------------------------------------------------
15// Client
16// ---------------------------------------------------------------------------
17
18#[derive(Clone)]
19pub struct TelemetryClient {
20    inner: Arc<Inner>,
21}
22
23struct Inner {
24    event_url:     String,
25    heartbeat_url: String,
26    token:         Option<String>,
27    instance:      String,
28    client:        reqwest::Client,
29}
30
31impl TelemetryClient {
32    pub fn new(base_url: &str, token: Option<String>, instance: String) -> Self {
33        let base = base_url.trim_end_matches('/');
34        let client = reqwest::Client::builder()
35            .timeout(Duration::from_secs(5))
36            .build()
37            .expect("reqwest client");
38
39        Self {
40            inner: Arc::new(Inner {
41                event_url:     format!("{base}/event"),
42                heartbeat_url: format!("{base}/heartbeat"),
43                token,
44                instance,
45                client,
46            }),
47        }
48    }
49
50    /// Push a completed request event — spawns a background task, never blocks.
51    pub fn push_event(&self, log: &RequestLog) {
52        let inner = Arc::clone(&self.inner);
53        let body = json!({
54            "instance":    inner.instance,
55            "ts_ms":       log.ts_ms,
56            "account":     log.account,
57            "model":       log.model,
58            "status":      log.status,
59            "duration_ms": log.duration_ms,
60        });
61
62        tokio::spawn(async move {
63            let mut req = inner.client.post(&inner.event_url).json(&body);
64            if let Some(ref t) = inner.token {
65                req = req.bearer_auth(t);
66            }
67            match req.send().await {
68                Ok(r) if !r.status().is_success() => {
69                    debug!(status = %r.status(), "relay rejected event");
70                }
71                Err(e) => debug!(err = %e, "relay event send failed"),
72                _ => {}
73            }
74        });
75    }
76
77    /// Push a full status snapshot as a heartbeat. Called periodically from a
78    /// background task — awaited by the caller so they can control the cadence.
79    pub async fn push_heartbeat(&self, status: serde_json::Value) {
80        let body = json!({
81            "instance": self.inner.instance,
82            "status":   status,
83        });
84        let mut req = self.inner.client.post(&self.inner.heartbeat_url).json(&body);
85        if let Some(ref t) = self.inner.token {
86            req = req.bearer_auth(t);
87        }
88        match req.send().await {
89            Ok(r) if !r.status().is_success() => {
90                debug!(status = %r.status(), "relay rejected heartbeat");
91            }
92            Err(e) => debug!(err = %e, "relay heartbeat send failed"),
93            _ => {}
94        }
95    }
96}
shunt/telemetry.rs

shunt/
telemetry.rs
