Skip to content

Commit 8ddb32f

Browse files
GnaneshGnaniGnaneshGnani
authored
Otel Ingest Driver (#179)
1 parent b5da069 commit 8ddb32f

9 files changed

Lines changed: 811 additions & 37 deletions

File tree

‎Cargo.lock‎

Lines changed: 231 additions & 33 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎asap-query-engine/Cargo.toml‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,9 @@ datafusion = "43"
4747
arrow = "53.4.1"
4848
futures = "0.3"
4949
prost = "0.13"
50+
opentelemetry-proto = { version = "0.28", features = ["gen-tonic", "gen-tonic-messages", "metrics"] }
51+
tonic = "0.12"
52+
tokio-stream = "0.1"
5053
snap = "1"
5154
regex = "1"
5255
prometheus = "0.13"

‎asap-query-engine/src/drivers/ingest/mod.rs‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,10 @@
11
pub mod kafka;
2+
pub mod otel;
23
pub mod prometheus_remote_write;
34
pub mod victoriametrics_remote_write;
45

56
pub use kafka::{KafkaConsumer, KafkaConsumerConfig};
7+
pub use otel::{OtlpReceiver, OtlpReceiverConfig};
68
// pub use prometheus_remote_write::{PrometheusRemoteWriteConfig, PrometheusRemoteWriteServer};
79
// pub use victoriametrics_remote_write::{
810
// VictoriaMetricsRemoteWriteConfig, VictoriaMetricsRemoteWriteServer,

‎asap-query-engine/src/drivers/ingest/otel.rs‎

Lines changed: 344 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,344 @@
1+
//! OTLP ingest driver.
2+
//!
3+
//! Accepts OTLP metrics via gRPC (4317) and HTTP (4318, POST /v1/metrics),
4+
//! parses ExportMetricsServiceRequest, logs counts at DEBUG, and leaves
5+
//! handoff to precompute engine as TODO.
6+
7+
use std::collections::HashMap;
8+
9+
use axum::{body::Bytes, extract::State, routing::post, Json, Router};
10+
use opentelemetry_proto::tonic::collector::metrics::v1::{
11+
metrics_service_server::MetricsService, ExportMetricsServiceRequest,
12+
ExportMetricsServiceResponse,
13+
};
14+
use opentelemetry_proto::tonic::metrics::v1::number_data_point::Value as NumberValue;
15+
use prost::Message;
16+
use tonic::{Request, Response, Status};
17+
use tracing::{debug, error, info};
18+
19+
/// Configuration for the OTLP receiver.
20+
#[derive(Debug, Clone)]
21+
pub struct OtlpReceiverConfig {
22+
pub grpc_port: u16,
23+
pub http_port: u16,
24+
}
25+
26+
/// OTLP receiver that accepts metrics via gRPC and HTTP.
27+
pub struct OtlpReceiver {
28+
config: OtlpReceiverConfig,
29+
}
30+
31+
impl OtlpReceiver {
32+
pub fn new(config: OtlpReceiverConfig) -> Self {
33+
Self { config }
34+
}
35+
36+
pub async fn run(&self) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
37+
let grpc_addr = std::net::SocketAddr::from(([0, 0, 0, 0], self.config.grpc_port));
38+
let http_addr = std::net::SocketAddr::from(([0, 0, 0, 0], self.config.http_port));
39+
40+
let grpc_svc = MetricsServiceImpl;
41+
let grpc_svc =
42+
opentelemetry_proto::tonic::collector::metrics::v1::metrics_service_server::MetricsServiceServer::new(
43+
grpc_svc,
44+
);
45+
46+
let app = Router::new()
47+
.route("/v1/metrics", post(handle_otlp_http))
48+
.with_state(());
49+
50+
let grpc_listener = tokio::net::TcpListener::bind(grpc_addr).await?;
51+
let http_listener = tokio::net::TcpListener::bind(http_addr).await?;
52+
53+
info!("OTLP gRPC listening on {}", grpc_addr);
54+
info!("OTLP HTTP listening on {} (POST /v1/metrics)", http_addr);
55+
56+
tokio::select! {
57+
r = tonic::transport::Server::builder()
58+
.add_service(grpc_svc)
59+
.serve_with_incoming(tokio_stream::wrappers::TcpListenerStream::new(grpc_listener)) => {
60+
if let Err(e) = r {
61+
error!("OTLP gRPC server error: {}", e);
62+
}
63+
}
64+
r = axum::serve(http_listener, app) => {
65+
if let Err(e) = r {
66+
error!("OTLP HTTP server error: {}", e);
67+
}
68+
}
69+
}
70+
71+
Ok(())
72+
}
73+
}
74+
75+
struct MetricsServiceImpl;
76+
77+
#[tonic::async_trait]
78+
impl MetricsService for MetricsServiceImpl {
79+
async fn export(
80+
&self,
81+
request: Request<ExportMetricsServiceRequest>,
82+
) -> Result<Response<ExportMetricsServiceResponse>, Status> {
83+
let req = request.into_inner();
84+
process_otlp_request(&req);
85+
Ok(Response::new(ExportMetricsServiceResponse {
86+
partial_success: None,
87+
}))
88+
}
89+
}
90+
91+
async fn handle_otlp_http(
92+
State(_state): State<()>,
93+
body: Bytes,
94+
) -> Result<Json<serde_json::Value>, (axum::http::StatusCode, String)> {
95+
let req = ExportMetricsServiceRequest::decode(body.as_ref()).map_err(|e| {
96+
(
97+
axum::http::StatusCode::BAD_REQUEST,
98+
format!("Protobuf decode error: {}", e),
99+
)
100+
})?;
101+
process_otlp_request(&req);
102+
Ok(Json(serde_json::json!({"rejected": 0})))
103+
}
104+
105+
/// A parsed metric data point: name, labels, timestamp (nanos), and numeric value.
106+
#[derive(Debug)]
107+
pub struct MetricPoint {
108+
pub name: String,
109+
pub labels: HashMap<String, String>,
110+
pub timestamp_nanos: u64,
111+
pub value: f64,
112+
}
113+
114+
fn process_otlp_request(request: &ExportMetricsServiceRequest) {
115+
let resource_count = request.resource_metrics.len();
116+
let total_points = otlp_to_record_count(request);
117+
debug!(
118+
"OTLP ingest: received {} resource metrics, {} total data points",
119+
resource_count, total_points
120+
);
121+
122+
let points = otlp_to_metric_points(request);
123+
if let Some(first) = points.first() {
124+
debug!(
125+
"OTLP parse example: {} {:?} @{}ns = {}",
126+
first.name, first.labels, first.timestamp_nanos, first.value
127+
);
128+
}
129+
// TODO: Pass metrics to precompute engine.
130+
}
131+
132+
/// Count total data points by traversing resource_metrics -> scope_metrics -> metrics.
133+
/// Reuses the same conversion traversal as asap-otel-ingest (Gauge, Sum, Histogram, etc.).
134+
fn otlp_to_record_count(request: &ExportMetricsServiceRequest) -> usize {
135+
let mut count = 0;
136+
for resource_metrics in &request.resource_metrics {
137+
for scope_metrics in &resource_metrics.scope_metrics {
138+
for metric in &scope_metrics.metrics {
139+
if metric.name.is_empty() {
140+
continue;
141+
}
142+
143+
use opentelemetry_proto::tonic::metrics::v1::metric::Data;
144+
match &metric.data {
145+
Some(Data::Gauge(g)) => count += g.data_points.len(),
146+
Some(Data::Sum(s)) => count += s.data_points.len(),
147+
Some(Data::Histogram(hist)) => {
148+
for dp in &hist.data_points {
149+
if dp.sum.is_some() {
150+
count += 1;
151+
}
152+
count += 1; // _count
153+
count += dp.bucket_counts.len(); // _bucket per le
154+
}
155+
}
156+
Some(Data::ExponentialHistogram(eh)) => {
157+
for dp in &eh.data_points {
158+
if dp.sum.is_some() {
159+
count += 1;
160+
}
161+
count += 1; // _count
162+
count += 1; // _scale
163+
}
164+
}
165+
Some(Data::Summary(summary)) => {
166+
for dp in &summary.data_points {
167+
count += 1; // _sum
168+
count += 1; // _count
169+
count += dp.quantile_values.len();
170+
}
171+
}
172+
None => {}
173+
}
174+
}
175+
}
176+
}
177+
count
178+
}
179+
180+
/// Parse OTLP request and convert to metric data points (name, labels, timestamp, value).
181+
fn otlp_to_metric_points(request: &ExportMetricsServiceRequest) -> Vec<MetricPoint> {
182+
let mut points = Vec::new();
183+
for resource_metrics in &request.resource_metrics {
184+
let resource_attrs = resource_metrics
185+
.resource
186+
.as_ref()
187+
.map(|r| attributes_to_map(&r.attributes))
188+
.unwrap_or_default();
189+
190+
for scope_metrics in &resource_metrics.scope_metrics {
191+
let scope_attrs = scope_metrics
192+
.scope
193+
.as_ref()
194+
.map(|s| attributes_to_map(&s.attributes))
195+
.unwrap_or_default();
196+
197+
for metric in &scope_metrics.metrics {
198+
if metric.name.is_empty() {
199+
continue;
200+
}
201+
202+
let base_labels: HashMap<String, String> = scope_attrs
203+
.iter()
204+
.chain(resource_attrs.iter())
205+
.map(|(k, v)| (k.clone(), v.clone()))
206+
.collect();
207+
208+
use opentelemetry_proto::tonic::metrics::v1::metric::Data;
209+
match &metric.data {
210+
Some(Data::Gauge(g)) => {
211+
for dp in &g.data_points {
212+
let labels = merge_point_attributes(&base_labels, &dp.attributes);
213+
let value = number_value_to_f64(&dp.value);
214+
points.push(MetricPoint {
215+
name: metric.name.clone(),
216+
labels,
217+
timestamp_nanos: dp.time_unix_nano,
218+
value,
219+
});
220+
}
221+
}
222+
Some(Data::Sum(s)) => {
223+
for dp in &s.data_points {
224+
let labels = merge_point_attributes(&base_labels, &dp.attributes);
225+
let value = number_value_to_f64(&dp.value);
226+
points.push(MetricPoint {
227+
name: metric.name.clone(),
228+
labels,
229+
timestamp_nanos: dp.time_unix_nano,
230+
value,
231+
});
232+
}
233+
}
234+
Some(Data::Histogram(hist)) => {
235+
for dp in &hist.data_points {
236+
let labels = merge_point_attributes(&base_labels, &dp.attributes);
237+
if let Some(sum) = dp.sum {
238+
points.push(MetricPoint {
239+
name: format!("{}_sum", metric.name),
240+
labels: labels.clone(),
241+
timestamp_nanos: dp.time_unix_nano,
242+
value: sum,
243+
});
244+
}
245+
points.push(MetricPoint {
246+
name: format!("{}_count", metric.name),
247+
labels: labels.clone(),
248+
timestamp_nanos: dp.time_unix_nano,
249+
value: dp.count as f64,
250+
});
251+
}
252+
}
253+
Some(Data::ExponentialHistogram(eh)) => {
254+
for dp in &eh.data_points {
255+
let labels = merge_point_attributes(&base_labels, &dp.attributes);
256+
if let Some(sum) = dp.sum {
257+
points.push(MetricPoint {
258+
name: format!("{}_sum", metric.name),
259+
labels: labels.clone(),
260+
timestamp_nanos: dp.time_unix_nano,
261+
value: sum,
262+
});
263+
}
264+
points.push(MetricPoint {
265+
name: format!("{}_count", metric.name),
266+
labels: labels.clone(),
267+
timestamp_nanos: dp.time_unix_nano,
268+
value: dp.count as f64,
269+
});
270+
}
271+
}
272+
Some(Data::Summary(sm)) => {
273+
for dp in &sm.data_points {
274+
let labels = merge_point_attributes(&base_labels, &dp.attributes);
275+
points.push(MetricPoint {
276+
name: format!("{}_sum", metric.name),
277+
labels: labels.clone(),
278+
timestamp_nanos: dp.time_unix_nano,
279+
value: dp.sum,
280+
});
281+
points.push(MetricPoint {
282+
name: format!("{}_count", metric.name),
283+
labels: labels.clone(),
284+
timestamp_nanos: dp.time_unix_nano,
285+
value: dp.count as f64,
286+
});
287+
}
288+
}
289+
None => {}
290+
}
291+
}
292+
}
293+
}
294+
points
295+
}
296+
297+
fn merge_point_attributes(
298+
base: &HashMap<String, String>,
299+
attrs: &[opentelemetry_proto::tonic::common::v1::KeyValue],
300+
) -> HashMap<String, String> {
301+
let mut m = base.clone();
302+
for (k, v) in attributes_to_map(attrs) {
303+
m.insert(k, v);
304+
}
305+
m
306+
}
307+
308+
fn number_value_to_f64(v: &Option<NumberValue>) -> f64 {
309+
match v {
310+
Some(NumberValue::AsDouble(x)) => *x,
311+
Some(NumberValue::AsInt(x)) => *x as f64,
312+
None => 0.0,
313+
}
314+
}
315+
316+
fn any_value_to_string(v: &opentelemetry_proto::tonic::common::v1::AnyValue) -> String {
317+
use opentelemetry_proto::tonic::common::v1::any_value::Value as AnyValueVariant;
318+
match &v.value {
319+
Some(AnyValueVariant::StringValue(s)) => s.clone(),
320+
Some(AnyValueVariant::IntValue(i)) => i.to_string(),
321+
Some(AnyValueVariant::DoubleValue(d)) => d.to_string(),
322+
Some(AnyValueVariant::BoolValue(b)) => b.to_string(),
323+
Some(AnyValueVariant::BytesValue(bytes)) => format!("{:?}", bytes),
324+
_ => String::new(),
325+
}
326+
}
327+
328+
fn attributes_to_map(
329+
attrs: &[opentelemetry_proto::tonic::common::v1::KeyValue],
330+
) -> HashMap<String, String> {
331+
let mut m = HashMap::new();
332+
for kv in attrs {
333+
let key = kv.key.clone();
334+
let value = kv
335+
.value
336+
.as_ref()
337+
.map(any_value_to_string)
338+
.unwrap_or_default();
339+
if !key.is_empty() {
340+
m.insert(key, value);
341+
}
342+
}
343+
m
344+
}

‎asap-query-engine/src/drivers/mod.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,5 +2,5 @@ pub mod ingest;
22
pub mod query;
33

44
// Re-export commonly used types for convenience
5-
pub use ingest::{KafkaConsumer, KafkaConsumerConfig};
5+
pub use ingest::{KafkaConsumer, KafkaConsumerConfig, OtlpReceiver, OtlpReceiverConfig};
66
pub use query::{AdapterConfig, HttpServer, HttpServerConfig};

‎asap-query-engine/src/lib.rs‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,10 @@ pub use stores::{SimpleMapStore, Store, StoreResult};
2424

2525
pub use engines::{InstantVector, QueryResult, SimpleEngine};
2626

27-
pub use drivers::{HttpServer, HttpServerConfig, KafkaConsumer, KafkaConsumerConfig};
27+
pub use drivers::{
28+
HttpServer, HttpServerConfig, KafkaConsumer, KafkaConsumerConfig, OtlpReceiver,
29+
OtlpReceiverConfig,
30+
};
2831

2932
pub use utils::{normalize_spatial_filter, read_inference_config, read_streaming_config};
3033

0 commit comments

Comments
 (0)