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#[derive(Debug, Deserialize, Serialize, utoipa::IntoParams)]
21pub struct MetricsQuery {
22 pub name: Option<String>,
24 pub resource: Option<String>,
26 pub session_id: Option<String>,
28 pub conversation_id: Option<String>,
30 pub model: Option<String>,
32 pub attrs: Option<String>,
35 pub start_time: Option<i64>,
37 pub end_time: Option<i64>,
39 pub limit: Option<usize>,
41 pub offset: Option<usize>,
43}
44
45#[derive(Debug, Deserialize, Serialize, utoipa::IntoParams)]
47pub struct AggregateQuery {
48 pub name: String,
50 pub function: String,
52 pub bucket_size: Option<i64>,
54 pub start_time: Option<i64>,
56 pub end_time: Option<i64>,
58}
59
60#[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#[derive(Debug, Serialize, Deserialize, utoipa::ToSchema)]
72pub struct TimeBucket {
73 pub timestamp: i64,
74 pub value: f64,
75 pub count: usize,
76}
77
78#[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 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 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)); }
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 let mut metrics = state
183 .storage
184 .query_latest_metrics(¶ms)
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 if let Some(name_filter) = &query.name {
198 metrics.retain(|m| m.name.contains(name_filter));
199 }
200
201 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 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 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#[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#[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 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 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 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 let metrics = state.storage.query_metrics(¶ms).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 let metrics: Vec<_> = metrics
377 .into_iter()
378 .filter(|m| m.name == query.name)
379 .collect();
380
381 if metrics.is_empty() {
382 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 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 let result = if let Some(bucket_size) = query.bucket_size {
561 let bucket_size_ns = bucket_size * 1_000_000_000; 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 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#[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 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 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 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 let metrics = state.storage.query_metrics(¶ms).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 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 return Ok(Json(vec![]));
765 }
766
767 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 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 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#[derive(Debug, Deserialize, Serialize, utoipa::IntoParams, utoipa::ToSchema)]
835pub struct TimeseriesQuery {
836 pub start_time: Option<i64>,
838 pub end_time: Option<i64>,
840 pub step: Option<i64>,
842}
843
844#[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 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 let metrics = state.storage.query_metrics(¶ms).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 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 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}