openzeppelin_relayer/queues/pubsub/
monitoring.rs

1//! Cloud Monitoring backlog-depth read.
2//!
3//! ONE light, low-frequency, batched read of
4//! `subscription/num_undelivered_messages` across the project's subscriptions,
5//! mapped back to queue types. Feeds BOTH the `queue_depth` gauge and the
6//! health-endpoint depth. Any failure (or the emulator, which serves no Cloud
7//! Monitoring) yields "unavailable" — never a hardcoded 0. Requires ADC with
8//! `roles/monitoring.viewer`.
9
10use std::collections::HashMap;
11use std::sync::Arc;
12
13use gcloud_pubsub::client::google_cloud_auth::project::Config;
14use gcloud_pubsub::client::google_cloud_auth::token::DefaultTokenSourceProvider;
15use serde::Deserialize;
16use token_source::{TokenSource, TokenSourceProvider};
17
18use super::QueueType;
19
20const MONITORING_SCOPE: &str = "https://www.googleapis.com/auth/monitoring.read";
21const NUM_UNDELIVERED_METRIC: &str = "pubsub.googleapis.com/subscription/num_undelivered_messages";
22
23/// Builds an ADC token source scoped for Cloud Monitoring reads.
24///
25/// Uses the same `gcloud-auth` ADC resolution the Pub/Sub client uses
26/// (service-account file / `GOOGLE_APPLICATION_CREDENTIALS[_JSON]` / metadata
27/// server). The token is cached and refreshed internally by the provider.
28pub(crate) async fn monitoring_token_source() -> Result<Arc<dyn TokenSource>, String> {
29    let config = Config::default().with_scopes(&[MONITORING_SCOPE]);
30    let provider = DefaultTokenSourceProvider::new(config)
31        .await
32        .map_err(|e| format!("failed to init Cloud Monitoring ADC token source: {e}"))?;
33    Ok(provider.token_source())
34}
35
36/// Reads `num_undelivered_messages` for the project's subscriptions and maps
37/// them to queue types via `subscription_to_queue` (reverse of the backend's
38/// subscription-name map). One batched HTTP call.
39pub(crate) async fn read_backlog_depths(
40    http: &reqwest::Client,
41    token_source: &Arc<dyn TokenSource>,
42    project_id: &str,
43    subscription_to_queue: &HashMap<String, QueueType>,
44) -> Result<HashMap<QueueType, u64>, String> {
45    // `token()` returns the value already formatted as "Bearer <access_token>".
46    let token = token_source
47        .token()
48        .await
49        .map_err(|e| format!("Cloud Monitoring token fetch failed: {e}"))?;
50
51    let end = chrono::Utc::now();
52    let start = end - chrono::Duration::seconds(300);
53    let url = format!("https://monitoring.googleapis.com/v3/projects/{project_id}/timeSeries");
54
55    let resp = http
56        .get(&url)
57        .header("Authorization", token)
58        .query(&[
59            ("filter", build_filter(subscription_to_queue)),
60            ("interval.startTime", start.to_rfc3339()),
61            ("interval.endTime", end.to_rfc3339()),
62        ])
63        .send()
64        .await
65        .map_err(|e| format!("Cloud Monitoring request failed: {e}"))?;
66
67    if !resp.status().is_success() {
68        return Err(format!(
69            "Cloud Monitoring returned HTTP {}",
70            resp.status().as_u16()
71        ));
72    }
73
74    let body: TimeSeriesResponse = resp
75        .json()
76        .await
77        .map_err(|e| format!("Cloud Monitoring response parse failed: {e}"))?;
78
79    Ok(parse_depths(&body, subscription_to_queue))
80}
81
82/// Builds the Cloud Monitoring `filter` for the backlog read, scoped to our own
83/// subscription IDs via `one_of(...)`.
84///
85/// Filtering on `metric.type` alone would match every
86/// `num_undelivered_messages` series in the project; in a project with many
87/// subscriptions the response could paginate and our 8 series land beyond the
88/// first page, so their depths would read as unavailable. Constraining to the
89/// relayer's subscriptions keeps the response bounded and deterministic. Falls
90/// back to the metric-only filter if the subscription map is somehow empty (a
91/// `one_of()` with no arguments is rejected by the API).
92fn build_filter(subscription_to_queue: &HashMap<String, QueueType>) -> String {
93    let metric = format!("metric.type=\"{NUM_UNDELIVERED_METRIC}\"");
94    let mut ids: Vec<&str> = subscription_to_queue.keys().map(String::as_str).collect();
95    if ids.is_empty() {
96        return metric;
97    }
98    ids.sort_unstable(); // deterministic filter string
99    let subs = ids
100        .iter()
101        .map(|id| format!("\"{id}\""))
102        .collect::<Vec<_>>()
103        .join(",");
104    format!("{metric} AND resource.label.subscription_id=one_of({subs})")
105}
106
107/// Maps a parsed timeSeries response to per-queue depths using the latest point
108/// of each subscription's series.
109fn parse_depths(
110    body: &TimeSeriesResponse,
111    subscription_to_queue: &HashMap<String, QueueType>,
112) -> HashMap<QueueType, u64> {
113    let mut depths = HashMap::new();
114    for series in &body.time_series {
115        let Some(sub_id) = series.resource.labels.get("subscription_id") else {
116            continue;
117        };
118        let Some(&queue_type) = subscription_to_queue.get(sub_id) else {
119            continue; // not one of our subscriptions
120        };
121        // Points are newest-first; take the most recent int64 value.
122        if let Some(value) = series
123            .points
124            .first()
125            .and_then(|p| p.value.int64_value.as_deref())
126            .and_then(|v| v.parse::<u64>().ok())
127        {
128            depths.insert(queue_type, value);
129        }
130    }
131    depths
132}
133
134#[derive(Debug, Deserialize)]
135struct TimeSeriesResponse {
136    #[serde(rename = "timeSeries", default)]
137    time_series: Vec<TimeSeries>,
138}
139
140#[derive(Debug, Deserialize)]
141struct TimeSeries {
142    #[serde(default)]
143    resource: MonitoredResource,
144    #[serde(default)]
145    points: Vec<Point>,
146}
147
148#[derive(Debug, Default, Deserialize)]
149struct MonitoredResource {
150    #[serde(default)]
151    labels: HashMap<String, String>,
152}
153
154#[derive(Debug, Deserialize)]
155struct Point {
156    value: PointValue,
157}
158
159#[derive(Debug, Deserialize)]
160struct PointValue {
161    #[serde(rename = "int64Value", default)]
162    int64_value: Option<String>,
163}
164
165#[cfg(test)]
166mod tests {
167    use super::*;
168
169    fn sub_map() -> HashMap<String, QueueType> {
170        HashMap::from([
171            (
172                "relayer-status-check-evm-sub".to_string(),
173                QueueType::StatusCheckEvm,
174            ),
175            (
176                "relayer-transaction-request-sub".to_string(),
177                QueueType::TransactionRequest,
178            ),
179        ])
180    }
181
182    #[test]
183    fn test_parse_depths_maps_subscriptions_to_queues() {
184        // int64 values are JSON strings; newest point is first.
185        let json = r#"{
186          "timeSeries": [
187            {
188              "resource": { "labels": { "subscription_id": "relayer-status-check-evm-sub" } },
189              "points": [ { "value": { "int64Value": "42" } }, { "value": { "int64Value": "40" } } ]
190            },
191            {
192              "resource": { "labels": { "subscription_id": "relayer-transaction-request-sub" } },
193              "points": [ { "value": { "int64Value": "7" } } ]
194            },
195            {
196              "resource": { "labels": { "subscription_id": "some-other-sub" } },
197              "points": [ { "value": { "int64Value": "999" } } ]
198            }
199          ]
200        }"#;
201        let body: TimeSeriesResponse = serde_json::from_str(json).unwrap();
202        let depths = parse_depths(&body, &sub_map());
203
204        assert_eq!(depths.get(&QueueType::StatusCheckEvm), Some(&42)); // newest point
205        assert_eq!(depths.get(&QueueType::TransactionRequest), Some(&7));
206        assert_eq!(depths.len(), 2, "unknown subscriptions are ignored");
207    }
208
209    #[test]
210    fn test_build_filter_scopes_to_our_subscriptions() {
211        let filter = build_filter(&sub_map());
212        assert!(filter.starts_with(&format!("metric.type=\"{NUM_UNDELIVERED_METRIC}\"")));
213        assert!(filter.contains("resource.label.subscription_id=one_of("));
214        // Both subscription IDs are present, sorted for a deterministic string.
215        assert!(filter.contains(
216            "one_of(\"relayer-status-check-evm-sub\",\"relayer-transaction-request-sub\")"
217        ));
218    }
219
220    #[test]
221    fn test_build_filter_falls_back_when_empty() {
222        let filter = build_filter(&HashMap::new());
223        assert_eq!(filter, format!("metric.type=\"{NUM_UNDELIVERED_METRIC}\""));
224    }
225
226    #[test]
227    fn test_parse_depths_empty_response() {
228        let body: TimeSeriesResponse = serde_json::from_str(r#"{}"#).unwrap();
229        assert!(parse_depths(&body, &sub_map()).is_empty());
230    }
231
232    #[test]
233    fn test_parse_depths_skips_series_without_points() {
234        let json = r#"{"timeSeries":[{"resource":{"labels":{"subscription_id":"relayer-status-check-evm-sub"}},"points":[]}]}"#;
235        let body: TimeSeriesResponse = serde_json::from_str(json).unwrap();
236        assert!(parse_depths(&body, &sub_map()).is_empty());
237    }
238}
openzeppelin_relayer/queues/pubsub/monitoring.rs

openzeppelin_relayer/queues/pubsub/
monitoring.rs
