spectra_backend_tensorbase/
metrics.rs

1//! TensorBase metrics storage adapter.
2
3use async_trait::async_trait;
4use chrono::{DateTime, Utc};
5use serde_json::Value;
6use spectra_backend_remote_common::RemoteMetricsBackend;
7use spectra_core::{
8    MetricPoint, MetricWriteRow, MetricsQueryRange, MetricsStorageBackend, Result,
9    StorageEngineType,
10};
11
12/// Scale-out TensorBase metrics storage over its ClickHouse-compatible protocol.
13///
14/// Use [`connect`](Self::connect) with a full protocol URL or
15/// [`connect_host`](Self::connect_host) for the default native port (`9528`). Pair this type
16/// with `TensorBaseEventsBackend` when building the runtime.
17///
18/// # Examples
19///
20/// Facade wiring through `Spectra::builder()` (requires the `spectra` crate with the
21/// `tensorbase` feature):
22///
23/// ```ignore
24/// use std::sync::Arc;
25/// use spectra::{Spectra, TensorBaseEventsBackend, TensorBaseMetricsBackend};
26///
27/// # async fn start() -> spectra::Result<()> {
28/// let url = "tcp://127.0.0.1:9528";
29/// let spectra = Spectra::builder()
30///     .metrics_backend(Arc::new(TensorBaseMetricsBackend::connect(url).await?))
31///     .events_backend(Arc::new(TensorBaseEventsBackend::connect(url).await?))
32///     .build()?;
33/// # let _ = spectra;
34/// # Ok(())
35/// # }
36/// ```
37pub struct TensorBaseMetricsBackend(RemoteMetricsBackend);
38
39impl TensorBaseMetricsBackend {
40    /// Connect with a full TensorBase protocol URL and ensure metrics tables exist.
41    ///
42    /// Accepts `tcp://host:9528` (native) or an HTTP ClickHouse-compatible endpoint. The call
43    /// is async and executes DDL.
44    ///
45    /// # Examples
46    ///
47    /// ```ignore
48    /// # async fn example() -> spectra_core::Result<()> {
49    /// use spectra_backend_tensorbase::TensorBaseMetricsBackend;
50    ///
51    /// let backend = TensorBaseMetricsBackend::connect("tcp://127.0.0.1:9528").await?;
52    /// # let _ = backend;
53    /// # Ok(())
54    /// # }
55    /// ```
56    pub async fn connect(url: &str) -> Result<Self> {
57        Ok(Self(
58            RemoteMetricsBackend::connect(
59                url,
60                StorageEngineType::TensorBase,
61                &crate::ddl::metrics_ddl(),
62            )
63            .await?,
64        ))
65    }
66
67    /// Connect using a host name and the default native port (`9528`).
68    ///
69    /// Equivalent to [`connect`](Self::connect) with `tcp://{host}:9528`.
70    ///
71    /// # Examples
72    ///
73    /// ```ignore
74    /// # async fn example() -> spectra_core::Result<()> {
75    /// use spectra_backend_tensorbase::TensorBaseMetricsBackend;
76    ///
77    /// let backend = TensorBaseMetricsBackend::connect_host("127.0.0.1").await?;
78    /// # let _ = backend;
79    /// # Ok(())
80    /// # }
81    /// ```
82    pub async fn connect_host(host: &str) -> Result<Self> {
83        Self::connect(&crate::ddl::default_url(host)).await
84    }
85
86    /// In-memory stub that preserves the TensorBase engine type for storage-contract tests.
87    ///
88    /// # Examples
89    ///
90    /// ```
91    /// use spectra_backend_tensorbase::TensorBaseMetricsBackend;
92    /// use spectra_core::{MetricsStorageBackend, StorageEngineType};
93    ///
94    /// let backend = TensorBaseMetricsBackend::in_memory_stub();
95    /// assert_eq!(backend.engine_type(), StorageEngineType::TensorBase);
96    /// ```
97    pub fn in_memory_stub() -> Self {
98        Self(RemoteMetricsBackend::in_memory_for_test(
99            StorageEngineType::TensorBase,
100        ))
101    }
102
103    /// In-memory stub for unit tests in this crate.
104    #[cfg(test)]
105    pub fn in_memory_for_test() -> Self {
106        Self::in_memory_stub()
107    }
108}
109
110#[async_trait]
111impl MetricsStorageBackend for TensorBaseMetricsBackend {
112    fn engine_type(&self) -> StorageEngineType {
113        self.0.engine_type()
114    }
115
116    async fn record_counter(
117        &self,
118        name: &str,
119        labels: &Value,
120        delta: i64,
121        ts: DateTime<Utc>,
122    ) -> Result<()> {
123        self.0.record_counter(name, labels, delta, ts).await
124    }
125
126    async fn record_gauge(
127        &self,
128        name: &str,
129        labels: &Value,
130        value: f64,
131        ts: DateTime<Utc>,
132    ) -> Result<()> {
133        self.0.record_gauge(name, labels, value, ts).await
134    }
135
136    async fn record_metrics_batch(&self, rows: &[MetricWriteRow]) -> Result<()> {
137        self.0.record_metrics_batch(rows).await
138    }
139
140    async fn query_range(&self, query: MetricsQueryRange) -> Result<Vec<MetricPoint>> {
141        self.0.query_range(query).await
142    }
143}
144
145#[cfg(test)]
146mod tests {
147    use super::*;
148    use chrono::Duration;
149    use serde_json::json;
150
151    #[tokio::test]
152    async fn tensorbase_metrics_roundtrip_in_memory() {
153        let backend = TensorBaseMetricsBackend::in_memory_for_test();
154        let ts = Utc::now();
155        backend
156            .record_counter("hits", &json!({}), 2, ts)
157            .await
158            .expect("write");
159        let points = backend
160            .query_range(MetricsQueryRange {
161                metric_name: "hits".into(),
162                start: ts - Duration::seconds(1),
163                end: ts + Duration::seconds(1),
164                label_matchers: vec![],
165            })
166            .await
167            .expect("query");
168        assert_eq!(points.len(), 1);
169    }
170
171    #[tokio::test]
172    #[ignore = "requires SPECTRA_TENSORBASE_URL"]
173    async fn tensorbase_metrics_integration() {
174        let url = std::env::var("SPECTRA_TENSORBASE_URL").expect("SPECTRA_TENSORBASE_URL");
175        let backend = TensorBaseMetricsBackend::connect(&url)
176            .await
177            .expect("connect");
178        let ts = Utc::now();
179        backend
180            .record_counter("integration_hits", &json!({}), 1, ts)
181            .await
182            .expect("write");
183        let points = backend
184            .query_range(MetricsQueryRange {
185                metric_name: "integration_hits".into(),
186                start: ts - Duration::seconds(5),
187                end: ts + Duration::seconds(5),
188                label_matchers: vec![],
189            })
190            .await
191            .expect("query");
192        assert!(!points.is_empty());
193    }
194}
spectra_backend_tensorbase/metrics.rs

spectra_backend_tensorbase/
metrics.rs
