Skip to main content

otelite_api/api/
metrics.rs

1use super::validate_time_range;
2use crate::server::{AppState, QueryCache};
3use axum::{
4    extract::{Query, State},
5    http::StatusCode,
6    response::{IntoResponse, Json},
7};
8use otelite_core::api::{
9    ErrorResponse, HistogramBucket, HistogramValue, MetricResponse, MetricValue, Quantile,
10    Resource, SummaryValue,
11};
12use otelite_core::query::{Operator, QueryPredicate, QueryValue};
13use otelite_core::storage::QueryParams;
14use otelite_core::telemetry::metric::MetricType;
15use otelite_core::telemetry::Metric;
16use serde::{Deserialize, Serialize};
17use std::collections::HashMap;
18
19/// Query parameters for listing metrics
20#[derive(Debug, Deserialize, Serialize, utoipa::IntoParams)]
21pub struct MetricsQuery {
22    /// Filter by metric name
23    pub name: Option<String>,
24    /// Filter by resource attribute (format: key=value)
25    pub resource: Option<String>,
26    /// Filter by session ID (session.id attribute)
27    pub session_id: Option<String>,
28    /// Filter by gen_ai.conversation.id attribute
29    pub conversation_id: Option<String>,
30    /// Filter by model (matches gen_ai.request.model)
31    pub model: Option<String>,
32    /// JSON-encoded list of attribute filters: [{"key":"x","op":"=","value":"y"}, ...]
33    /// Supported ops: "=" (Equal), "!=" (NotEqual)
34    pub attrs: Option<String>,
35    /// Start time (nanoseconds since Unix epoch)
36    pub start_time: Option<i64>,
37    /// End time (nanoseconds since Unix epoch)
38    pub end_time: Option<i64>,
39    /// Maximum number of metrics to return (capped at 1000)
40    pub limit: Option<usize>,
41    /// Offset for pagination
42    pub offset: Option<usize>,
43}
44
45/// Query parameters for aggregating metrics
46#[derive(Debug, Deserialize, Serialize, utoipa::IntoParams)]
47pub struct AggregateQuery {
48    /// Metric name to aggregate
49    pub name: String,
50    /// Aggregation function (sum, avg, min, max)
51    pub function: String,
52    /// Time bucket size in seconds (for time-series aggregation)
53    pub bucket_size: Option<i64>,
54    /// Start time (nanoseconds since Unix epoch)
55    pub start_time: Option<i64>,
56    /// End time (nanoseconds since Unix epoch)
57    pub end_time: Option<i64>,
58}
59
60/// Response structure for aggregated metrics
61#[derive(Debug, Serialize, Deserialize, utoipa::ToSchema)]
62pub struct AggregateResponse {
63    pub name: String,
64    pub function: String,
65    pub result: f64,
66    pub count: usize,
67    pub buckets: Option<Vec<TimeBucket>>,
68}
69
70/// Time bucket for time-series aggregation
71#[derive(Debug, Serialize, Deserialize, utoipa::ToSchema)]
72pub struct TimeBucket {
73    pub timestamp: i64,
74    pub value: f64,
75    pub count: usize,
76}
77
78/// List metrics with optional filtering
79#[utoipa::path(
80    get,
81    path = "/api/metrics",
82    params(MetricsQuery),
83    responses(
84        (status = 200, description = "List of metrics", body = Vec<MetricResponse>),
85        (status = 400, description = "Invalid query parameters or time range", body = ErrorResponse),
86        (status = 500, description = "Internal server error", body = ErrorResponse)
87    ),
88    tag = "metrics"
89)]
90pub async fn list_metrics(
91    State(state): State<AppState>,
92    Query(query): Query<MetricsQuery>,
93) -> Result<Json<Vec<MetricResponse>>, (StatusCode, Json<ErrorResponse>)> {
94    validate_time_range(query.start_time, query.end_time)?;
95
96    // Check cache first
97    let cache_key = QueryCache::make_key(&query);
98    if let Some(cached) = state.cache.metrics.get(&cache_key) {
99        if let Ok(response) = serde_json::from_str(&cached) {
100            return Ok(Json(response));
101        }
102    }
103
104    // Build query parameters
105    let mut params = QueryParams::default();
106
107    if let Some(start) = query.start_time {
108        params.start_time = Some(start);
109    }
110    if let Some(end) = query.end_time {
111        params.end_time = Some(end);
112    }
113    if let Some(limit) = query.limit {
114        params.limit = Some(limit.min(1000)); // cap to prevent runaway queries
115    }
116
117    if let Some(ref sid) = query.session_id {
118        if !sid.is_empty() {
119            params.predicates.push(QueryPredicate {
120                field: "session.id".to_string(),
121                operator: Operator::Equal,
122                value: QueryValue::String(sid.clone()),
123            });
124        }
125    }
126
127    if let Some(ref cid) = query.conversation_id {
128        if !cid.is_empty() {
129            params.predicates.push(QueryPredicate {
130                field: "gen_ai.conversation.id".to_string(),
131                operator: Operator::Equal,
132                value: QueryValue::String(cid.clone()),
133            });
134        }
135    }
136
137    if let Some(ref model) = query.model {
138        if !model.is_empty() {
139            params.predicates.push(QueryPredicate {
140                field: "gen_ai.request.model".to_string(),
141                operator: Operator::Equal,
142                value: QueryValue::String(model.clone()),
143            });
144        }
145    }
146
147    if let Some(ref attrs_json) = query.attrs {
148        if !attrs_json.is_empty() {
149            #[derive(serde::Deserialize)]
150            struct AttrFilter {
151                key: String,
152                op: String,
153                value: Option<String>,
154            }
155            let filters = serde_json::from_str::<Vec<AttrFilter>>(attrs_json).map_err(|e| {
156                (
157                    StatusCode::BAD_REQUEST,
158                    Json(ErrorResponse::bad_request(format!(
159                        "attrs must be valid JSON attribute filters: {e}"
160                    ))),
161                )
162            })?;
163            for f in filters {
164                let op = match f.op.as_str() {
165                    "=" => Operator::Equal,
166                    "!=" => Operator::NotEqual,
167                    _ => continue,
168                };
169                if let Some(v) = f.value {
170                    params.predicates.push(QueryPredicate {
171                        field: f.key,
172                        operator: op,
173                        value: QueryValue::String(v),
174                    });
175                }
176            }
177        }
178    }
179
180    // Query metrics from storage: use latest-per-name so that high-frequency
181    // counters don't push gauges/histograms out of the result window.
182    let mut metrics = state
183        .storage
184        .query_latest_metrics(&params)
185        .await
186        .map_err(|e| {
187            (
188                StatusCode::INTERNAL_SERVER_ERROR,
189                Json(ErrorResponse::storage_error(format!(
190                    "query metrics: {}",
191                    e
192                ))),
193            )
194        })?;
195
196    // Filter by name if specified
197    if let Some(name_filter) = &query.name {
198        metrics.retain(|m| m.name.contains(name_filter));
199    }
200
201    // Filter by resource if specified
202    if let Some(resource_filter) = &query.resource {
203        if let Some((key, value)) = resource_filter.split_once('=') {
204            metrics.retain(|m| {
205                m.resource
206                    .as_ref()
207                    .and_then(|r| r.attributes.get(key))
208                    .map(|v| v == value)
209                    .unwrap_or(false)
210            });
211        }
212    }
213
214    // Apply pagination manually (limit capped at 1000, matching traces endpoint)
215    let offset = query.offset.unwrap_or(0);
216    let limit = query.limit.unwrap_or(100).min(1000);
217
218    let metrics: Vec<_> = metrics.into_iter().skip(offset).take(limit).collect();
219
220    // Convert to response format
221    let response: Vec<MetricResponse> = metrics
222        .into_iter()
223        .map(|metric| {
224            let (metric_type_str, value) = match &metric.metric_type {
225                MetricType::Gauge(v) => ("gauge", MetricValue::Gauge(*v)),
226                MetricType::Counter(v) => ("counter", MetricValue::Counter(*v as i64)),
227                MetricType::CounterDouble(v) => ("counter", MetricValue::CounterDouble(*v)),
228                MetricType::Histogram {
229                    count,
230                    sum,
231                    buckets,
232                } => (
233                    "histogram",
234                    MetricValue::Histogram(HistogramValue {
235                        count: *count,
236                        sum: *sum,
237                        buckets: buckets
238                            .iter()
239                            .map(|b| HistogramBucket {
240                                upper_bound: b.upper_bound,
241                                count: b.count,
242                            })
243                            .collect(),
244                    }),
245                ),
246                MetricType::Summary {
247                    count,
248                    sum,
249                    quantiles,
250                } => (
251                    "summary",
252                    MetricValue::Summary(SummaryValue {
253                        count: *count,
254                        sum: *sum,
255                        quantiles: quantiles
256                            .iter()
257                            .map(|q| Quantile {
258                                quantile: q.quantile,
259                                value: q.value,
260                            })
261                            .collect(),
262                    }),
263                ),
264            };
265
266            MetricResponse {
267                name: metric.name,
268                description: metric.description,
269                unit: metric.unit,
270                metric_type: metric_type_str.to_string(),
271                value,
272                timestamp: metric.timestamp,
273                attributes: metric.attributes,
274                resource: metric.resource.map(|r| Resource {
275                    attributes: r.attributes,
276                }),
277            }
278        })
279        .collect();
280
281    Ok(Json(response))
282}
283
284/// Get list of unique metric names
285#[utoipa::path(
286    get,
287    path = "/api/metrics/names",
288    responses(
289        (status = 200, description = "List of unique metric names", body = Vec<String>),
290        (status = 500, description = "Internal server error", body = ErrorResponse)
291    ),
292    tag = "metrics"
293)]
294pub async fn list_metric_names(
295    State(state): State<AppState>,
296) -> Result<Json<Vec<String>>, (StatusCode, Json<ErrorResponse>)> {
297    let names = state
298        .storage
299        .query_distinct_metric_names()
300        .await
301        .map_err(|e| {
302            (
303                StatusCode::INTERNAL_SERVER_ERROR,
304                Json(ErrorResponse::storage_error(format!(
305                    "list metric names: {}",
306                    e
307                ))),
308            )
309        })?;
310
311    Ok(Json(names))
312}
313
314/// Aggregate metrics by function
315#[utoipa::path(
316    get,
317    path = "/api/metrics/aggregate",
318    params(AggregateQuery),
319    responses(
320        (status = 200, description = "Aggregated metric result (count=0 if metric exists but no data in the requested window)", body = AggregateResponse),
321        (status = 400, description = "Invalid aggregation function or time range", body = ErrorResponse),
322        (status = 404, description = "Metric name has never been ingested", body = ErrorResponse),
323        (status = 500, description = "Internal server error", body = ErrorResponse)
324    ),
325    tag = "metrics"
326)]
327pub async fn aggregate_metrics(
328    State(state): State<AppState>,
329    Query(query): Query<AggregateQuery>,
330) -> Result<Json<AggregateResponse>, (StatusCode, Json<ErrorResponse>)> {
331    validate_time_range(query.start_time, query.end_time)?;
332
333    // A zero or negative bucket size would divide by zero when grouping.
334    if let Some(bucket_size) = query.bucket_size {
335        if bucket_size < 1 {
336            return Err((
337                StatusCode::BAD_REQUEST,
338                Json(ErrorResponse::bad_request(format!(
339                    "Invalid bucket_size '{}': must be a positive number of seconds",
340                    bucket_size
341                ))),
342            ));
343        }
344    }
345
346    // Check cache first
347    let cache_key = QueryCache::make_key(&query);
348    if let Some(cached) = state.cache.metrics.get(&cache_key) {
349        if let Ok(response) = serde_json::from_str(&cached) {
350            return Ok(Json(response));
351        }
352    }
353
354    // Build query parameters
355    let mut params = QueryParams::default();
356
357    if let Some(start) = query.start_time {
358        params.start_time = Some(start);
359    }
360    if let Some(end) = query.end_time {
361        params.end_time = Some(end);
362    }
363
364    // Query metrics from storage
365    let metrics = state.storage.query_metrics(&params).await.map_err(|e| {
366        (
367            StatusCode::INTERNAL_SERVER_ERROR,
368            Json(ErrorResponse::storage_error(format!(
369                "aggregate metrics: {}",
370                e
371            ))),
372        )
373    })?;
374
375    // Filter by name
376    let metrics: Vec<_> = metrics
377        .into_iter()
378        .filter(|m| m.name == query.name)
379        .collect();
380
381    if metrics.is_empty() {
382        // Check whether the metric name has ever been ingested (outside the time window).
383        // 404 = never seen; 200 + zero aggregate = exists but no data in the requested window.
384        let all_params = QueryParams::default();
385        let exists = state
386            .storage
387            .query_metrics(&all_params)
388            .await
389            .map_err(|e| {
390                (
391                    StatusCode::INTERNAL_SERVER_ERROR,
392                    Json(ErrorResponse::storage_error(format!(
393                        "check metric existence: {}",
394                        e
395                    ))),
396                )
397            })?
398            .iter()
399            .any(|m| m.name == query.name);
400        if !exists {
401            return Err((
402                StatusCode::NOT_FOUND,
403                Json(ErrorResponse::not_found(format!("Metric '{}'", query.name))),
404            ));
405        }
406        return Ok(Json(AggregateResponse {
407            name: query.name,
408            function: query.function,
409            result: 0.0,
410            count: 0,
411            buckets: None,
412        }));
413    }
414
415    // Perform aggregation based on function
416    let result = match query.function.as_str() {
417        "sum" => {
418            let mut sum = 0.0;
419            let mut count = 0;
420            for metric in &metrics {
421                match &metric.metric_type {
422                    MetricType::Gauge(v) => {
423                        sum += v;
424                        count += 1;
425                    },
426                    MetricType::Counter(v) => {
427                        sum += *v as f64;
428                        count += 1;
429                    },
430                    MetricType::CounterDouble(v) => {
431                        sum += *v;
432                        count += 1;
433                    },
434                    MetricType::Histogram { sum: s, .. } => {
435                        sum += s;
436                        count += 1;
437                    },
438                    MetricType::Summary { sum: s, .. } => {
439                        sum += s;
440                        count += 1;
441                    },
442                }
443            }
444            AggregateResponse {
445                name: query.name.clone(),
446                function: "sum".to_string(),
447                result: sum,
448                count,
449                buckets: None,
450            }
451        },
452        "avg" => {
453            let mut sum = 0.0;
454            let mut count = 0;
455            for metric in &metrics {
456                match &metric.metric_type {
457                    MetricType::Gauge(v) => {
458                        sum += v;
459                        count += 1;
460                    },
461                    MetricType::Counter(v) => {
462                        sum += *v as f64;
463                        count += 1;
464                    },
465                    MetricType::CounterDouble(v) => {
466                        sum += *v;
467                        count += 1;
468                    },
469                    MetricType::Histogram {
470                        sum: s, count: c, ..
471                    } => {
472                        sum += s;
473                        count += *c as usize;
474                    },
475                    MetricType::Summary {
476                        sum: s, count: c, ..
477                    } => {
478                        sum += s;
479                        count += *c as usize;
480                    },
481                }
482            }
483            let avg = if count > 0 { sum / count as f64 } else { 0.0 };
484            AggregateResponse {
485                name: query.name.clone(),
486                function: "avg".to_string(),
487                result: avg,
488                count,
489                buckets: None,
490            }
491        },
492        "min" => {
493            let mut min = f64::MAX;
494            let mut count = 0;
495            for metric in &metrics {
496                match &metric.metric_type {
497                    MetricType::Gauge(v) => {
498                        min = min.min(*v);
499                        count += 1;
500                    },
501                    MetricType::Counter(v) => {
502                        min = min.min(*v as f64);
503                        count += 1;
504                    },
505                    MetricType::CounterDouble(v) => {
506                        min = min.min(*v);
507                        count += 1;
508                    },
509                    _ => {},
510                }
511            }
512            AggregateResponse {
513                name: query.name.clone(),
514                function: "min".to_string(),
515                result: if count > 0 { min } else { 0.0 },
516                count,
517                buckets: None,
518            }
519        },
520        "max" => {
521            let mut max = f64::MIN;
522            let mut count = 0;
523            for metric in &metrics {
524                match &metric.metric_type {
525                    MetricType::Gauge(v) => {
526                        max = max.max(*v);
527                        count += 1;
528                    },
529                    MetricType::Counter(v) => {
530                        max = max.max(*v as f64);
531                        count += 1;
532                    },
533                    MetricType::CounterDouble(v) => {
534                        max = max.max(*v);
535                        count += 1;
536                    },
537                    _ => {},
538                }
539            }
540            AggregateResponse {
541                name: query.name.clone(),
542                function: "max".to_string(),
543                result: if count > 0 { max } else { 0.0 },
544                count,
545                buckets: None,
546            }
547        },
548        _ => {
549            return Err((
550                StatusCode::BAD_REQUEST,
551                Json(ErrorResponse::bad_request(format!(
552                    "Invalid aggregation function '{}'. Use: sum, avg, min, max",
553                    query.function
554                ))),
555            ))
556        },
557    };
558
559    // If bucket_size is specified, perform time-series aggregation
560    let result = if let Some(bucket_size) = query.bucket_size {
561        let bucket_size_ns = bucket_size * 1_000_000_000; // Convert seconds to nanoseconds
562
563        // Group metrics by time bucket
564        let mut buckets: HashMap<i64, Vec<&Metric>> = HashMap::new();
565        for metric in &metrics {
566            let bucket_timestamp = (metric.timestamp / bucket_size_ns) * bucket_size_ns;
567            buckets.entry(bucket_timestamp).or_default().push(metric);
568        }
569
570        // Aggregate each bucket
571        let mut time_buckets: Vec<TimeBucket> = buckets
572            .into_iter()
573            .map(|(timestamp, bucket_metrics)| {
574                let (sum, count) = match query.function.as_str() {
575                    "sum" => {
576                        let mut sum = 0.0;
577                        let mut count = 0;
578                        for metric in bucket_metrics {
579                            match &metric.metric_type {
580                                MetricType::Gauge(v) => {
581                                    sum += v;
582                                    count += 1;
583                                },
584                                MetricType::Counter(v) => {
585                                    sum += *v as f64;
586                                    count += 1;
587                                },
588                                MetricType::CounterDouble(v) => {
589                                    sum += *v;
590                                    count += 1;
591                                },
592                                MetricType::Histogram { sum: s, .. } => {
593                                    sum += s;
594                                    count += 1;
595                                },
596                                MetricType::Summary { sum: s, .. } => {
597                                    sum += s;
598                                    count += 1;
599                                },
600                            }
601                        }
602                        (sum, count)
603                    },
604                    "avg" => {
605                        let mut sum = 0.0;
606                        let mut count = 0;
607                        for metric in bucket_metrics {
608                            match &metric.metric_type {
609                                MetricType::Gauge(v) => {
610                                    sum += v;
611                                    count += 1;
612                                },
613                                MetricType::Counter(v) => {
614                                    sum += *v as f64;
615                                    count += 1;
616                                },
617                                MetricType::CounterDouble(v) => {
618                                    sum += *v;
619                                    count += 1;
620                                },
621                                MetricType::Histogram {
622                                    sum: s, count: c, ..
623                                } => {
624                                    sum += s;
625                                    count += *c as usize;
626                                },
627                                MetricType::Summary {
628                                    sum: s, count: c, ..
629                                } => {
630                                    sum += s;
631                                    count += *c as usize;
632                                },
633                            }
634                        }
635                        let avg = if count > 0 { sum / count as f64 } else { 0.0 };
636                        (avg, count)
637                    },
638                    _ => (0.0, 0),
639                };
640
641                TimeBucket {
642                    timestamp,
643                    value: sum,
644                    count,
645                }
646            })
647            .collect();
648
649        time_buckets.sort_by_key(|b| b.timestamp);
650
651        AggregateResponse {
652            buckets: Some(time_buckets),
653            ..result
654        }
655    } else {
656        result
657    };
658
659    Ok(Json(result))
660}
661
662/// Get time-series data for a specific metric
663#[utoipa::path(
664    get,
665    path = "/api/metrics/{name}/timeseries",
666    params(
667        ("name" = String, Path, description = "Metric name"),
668        ("start_time" = Option<i64>, Query, description = "Start time (nanoseconds since Unix epoch)"),
669        ("end_time" = Option<i64>, Query, description = "End time (nanoseconds since Unix epoch)"),
670        ("step" = Option<i64>, Query, description = "Time step in seconds (default: 60)")
671    ),
672    responses(
673        (status = 200, description = "Time-series data points", body = Vec<TimeBucket>),
674        (status = 404, description = "Metric not found", body = ErrorResponse),
675        (status = 500, description = "Internal server error", body = ErrorResponse)
676    ),
677    tag = "metrics"
678)]
679pub async fn get_metric_timeseries(
680    State(state): State<AppState>,
681    axum::extract::Path(name): axum::extract::Path<String>,
682    Query(query): Query<TimeseriesQuery>,
683) -> Result<Json<Vec<TimeBucket>>, (StatusCode, Json<ErrorResponse>)> {
684    // Reject a non-positive step up front: a zero or negative bucket size
685    // would divide by zero (or invert bucketing) below.
686    if let Some(step) = query.step {
687        if step < 1 {
688            return Err((
689                StatusCode::BAD_REQUEST,
690                Json(ErrorResponse::bad_request(format!(
691                    "Invalid step '{}': must be a positive number of seconds",
692                    step
693                ))),
694            ));
695        }
696    }
697
698    // Check cache first
699    let cache_key = QueryCache::make_key(&(&name, &query));
700    if let Some(cached) = state.cache.metrics.get(&cache_key) {
701        if let Ok(response) = serde_json::from_str(&cached) {
702            return Ok(Json(response));
703        }
704    }
705
706    // Build query parameters. The name predicate keeps the filter in SQL
707    // (an idx_metrics_name_ts seek) instead of pulling every metric in the
708    // window into memory and filtering here.
709    let name_predicate = QueryPredicate {
710        field: "name".to_string(),
711        operator: Operator::Equal,
712        value: QueryValue::String(name.clone()),
713    };
714    let mut params = QueryParams {
715        predicates: vec![name_predicate.clone()],
716        ..Default::default()
717    };
718    if let Some(start) = query.start_time {
719        params.start_time = Some(start);
720    }
721    if let Some(end) = query.end_time {
722        params.end_time = Some(end);
723    }
724
725    // Query metrics from storage
726    let metrics = state.storage.query_metrics(&params).await.map_err(|e| {
727        (
728            StatusCode::INTERNAL_SERVER_ERROR,
729            Json(ErrorResponse::storage_error(format!(
730                "query metric timeseries: {}",
731                e
732            ))),
733        )
734    })?;
735
736    // If time-filtered query is empty, check if metric exists at all (without time filter).
737    // A single-row existence probe, not a full-table load.
738    if metrics.is_empty() {
739        let exists_params = QueryParams {
740            predicates: vec![name_predicate],
741            limit: Some(1),
742            ..Default::default()
743        };
744        let all_metrics = state
745            .storage
746            .query_metrics(&exists_params)
747            .await
748            .map_err(|e| {
749                (
750                    StatusCode::INTERNAL_SERVER_ERROR,
751                    Json(ErrorResponse::storage_error(format!(
752                        "check metric existence: {}",
753                        e
754                    ))),
755                )
756            })?;
757        if all_metrics.is_empty() {
758            return Err((
759                StatusCode::NOT_FOUND,
760                Json(ErrorResponse::not_found(format!("Metric '{}'", name))),
761            ));
762        }
763        // Metric exists but no data in the requested time window — return empty
764        return Ok(Json(vec![]));
765    }
766
767    // Group by time buckets
768    let step_seconds = query.step.unwrap_or(60);
769    let bucket_size_ns = step_seconds * 1_000_000_000;
770
771    let mut buckets: HashMap<i64, Vec<&Metric>> = HashMap::new();
772    for metric in &metrics {
773        let bucket_timestamp = (metric.timestamp / bucket_size_ns) * bucket_size_ns;
774        buckets.entry(bucket_timestamp).or_default().push(metric);
775    }
776
777    // Calculate average value per bucket
778    let mut time_buckets: Vec<TimeBucket> = buckets
779        .into_iter()
780        .map(|(timestamp, bucket_metrics)| {
781            let mut sum = 0.0;
782            let mut count = 0;
783
784            for metric in bucket_metrics {
785                match &metric.metric_type {
786                    MetricType::Gauge(v) => {
787                        sum += v;
788                        count += 1;
789                    },
790                    MetricType::Counter(v) => {
791                        sum += *v as f64;
792                        count += 1;
793                    },
794                    MetricType::CounterDouble(v) => {
795                        sum += *v;
796                        count += 1;
797                    },
798                    MetricType::Histogram {
799                        sum: s, count: c, ..
800                    } => {
801                        sum += s;
802                        count += *c as usize;
803                    },
804                    MetricType::Summary {
805                        sum: s, count: c, ..
806                    } => {
807                        sum += s;
808                        count += *c as usize;
809                    },
810                }
811            }
812
813            let value = if count > 0 { sum / count as f64 } else { 0.0 };
814
815            TimeBucket {
816                timestamp,
817                value,
818                count,
819            }
820        })
821        .collect();
822
823    time_buckets.sort_by_key(|b| b.timestamp);
824
825    // Cache the result
826    if let Ok(json) = serde_json::to_string(&time_buckets) {
827        state.cache.metrics.insert(cache_key, json);
828    }
829
830    Ok(Json(time_buckets))
831}
832
833/// Query parameters for time-series data
834#[derive(Debug, Deserialize, Serialize, utoipa::IntoParams, utoipa::ToSchema)]
835pub struct TimeseriesQuery {
836    /// Start time (nanoseconds since Unix epoch)
837    pub start_time: Option<i64>,
838    /// End time (nanoseconds since Unix epoch)
839    pub end_time: Option<i64>,
840    /// Time step in seconds (default: 60)
841    pub step: Option<i64>,
842}
843
844/// Export metrics as JSON
845#[utoipa::path(
846    get,
847    path = "/api/metrics/export",
848    params(MetricsQuery),
849    responses(
850        (status = 200, description = "Exported metrics in JSON format"),
851        (status = 400, description = "Invalid time range", body = ErrorResponse),
852        (status = 500, description = "Internal server error", body = ErrorResponse)
853    ),
854    tag = "metrics"
855)]
856pub async fn export_metrics(
857    State(state): State<AppState>,
858    Query(query): Query<MetricsQuery>,
859) -> Result<impl IntoResponse, (StatusCode, Json<ErrorResponse>)> {
860    validate_time_range(query.start_time, query.end_time)?;
861
862    // Build query parameters
863    let mut params = QueryParams::default();
864
865    if let Some(start) = query.start_time {
866        params.start_time = Some(start);
867    }
868    if let Some(end) = query.end_time {
869        params.end_time = Some(end);
870    }
871
872    // Query metrics from storage
873    let metrics = state.storage.query_metrics(&params).await.map_err(|e| {
874        (
875            StatusCode::INTERNAL_SERVER_ERROR,
876            Json(ErrorResponse::storage_error(format!(
877                "export metrics: {}",
878                e
879            ))),
880        )
881    })?;
882
883    // Filter by name if specified
884    let metrics: Vec<_> = if let Some(name_filter) = &query.name {
885        metrics
886            .into_iter()
887            .filter(|m| m.name.contains(name_filter))
888            .collect()
889    } else {
890        metrics
891    };
892
893    // Convert to JSON
894    let json = serde_json::to_string_pretty(&metrics).map_err(|e| {
895        (
896            StatusCode::INTERNAL_SERVER_ERROR,
897            Json(ErrorResponse::internal_error(format!(
898                "Failed to serialize metrics: {}",
899                e
900            ))),
901        )
902    })?;
903
904    Ok((StatusCode::OK, [("Content-Type", "application/json")], json))
905}