feat: otlp tracing framework via http

Cargo.toml

@@ -87,7 +87,7 @@ meter-core = { git = "https://github.com/GreptimeTeam/greptime-meter.git", rev =
 metrics = "0.20"
 moka = "0.12"
 once_cell = "1.18"
-opentelemetry-proto = { version = "0.2", features = ["gen-tonic", "metrics"] }
+opentelemetry-proto = { version = "0.2", features = ["gen-tonic", "metrics", "traces"] }
 parquet = "43.0"
 paste = "1.0"
 prost = "0.11"

src/frontend/src/instance/otlp.rs

@@ -19,14 +19,17 @@ use metrics::counter;
 use opentelemetry_proto::tonic::collector::metrics::v1::{
     ExportMetricsServiceRequest, ExportMetricsServiceResponse,
 };
+use opentelemetry_proto::tonic::collector::trace::v1::{
+    ExportTraceServiceRequest, ExportTraceServiceResponse,
+};
 use servers::error::{self, AuthSnafu, Result as ServerResult};
 use servers::otlp;
 use servers::query_handler::OpenTelemetryProtocolHandler;
 use session::context::QueryContextRef;
 use snafu::ResultExt;
 
 use crate::instance::Instance;
-use crate::metrics::OTLP_METRICS_ROWS;
+use crate::metrics::{OTLP_METRICS_ROWS, OTLP_TRACES_ROWS};
 
 #[async_trait]
 impl OpenTelemetryProtocolHandler for Instance {
@@ -40,7 +43,7 @@ impl OpenTelemetryProtocolHandler for Instance {
             .as_ref()
             .check_permission(ctx.current_user(), PermissionReq::Otlp)
             .context(AuthSnafu)?;
-        let (requests, rows) = otlp::to_grpc_insert_requests(request)?;
+        let (requests, rows) = otlp::metrics::to_grpc_insert_requests(request)?;
         let _ = self
             .handle_row_inserts(requests, ctx)
             .await
@@ -55,4 +58,32 @@ impl OpenTelemetryProtocolHandler for Instance {
         };
         Ok(resp)
     }
+
+    async fn traces(
+        &self,
+        request: ExportTraceServiceRequest,
+        ctx: QueryContextRef,
+    ) -> ServerResult<ExportTraceServiceResponse> {
+        self.plugins
+            .get::<PermissionCheckerRef>()
+            .as_ref()
+            .check_permission(ctx.current_user(), PermissionReq::Otlp)
+            .context(AuthSnafu)?;
+
+        let (requests, rows) = otlp::traces::to_grpc_insert_requests(request)?;
+
+        let _ = self
+            .handle_row_inserts(requests, ctx)
+            .await
+            .map_err(BoxedError::new)
+            .context(error::ExecuteGrpcQuerySnafu)?;
+
+        counter!(OTLP_TRACES_ROWS, rows as u64);
+
+        let resp = ExportTraceServiceResponse {
+            // TODO(fys): add support for partial_success in future patch
+            partial_success: None,
+        };
+        Ok(resp)
+    }
 }

src/frontend/src/metrics.rs

@@ -22,3 +22,4 @@ pub(crate) const METRIC_RUN_SCRIPT_ELAPSED: &str = "frontend.run_script_elapsed"
 pub const PROM_STORE_REMOTE_WRITE_SAMPLES: &str = "frontend.prometheus.remote_write.samples";
 
 pub const OTLP_METRICS_ROWS: &str = "frontend.otlp.metrics.rows";
+pub const OTLP_TRACES_ROWS: &str = "frontend.otlp.traces.rows";

src/servers/src/http.rs

@@ -660,6 +660,7 @@ impl HttpServer {
     fn route_otlp<S>(&self, otlp_handler: OpenTelemetryProtocolHandlerRef) -> Router<S> {
         Router::new()
             .route("/v1/metrics", routing::post(otlp::metrics))
+            .route("/v1/traces", routing::post(otlp::traces))
             .with_state(otlp_handler)
     }
 

src/servers/src/http/otlp.rs

@@ -21,6 +21,9 @@ use hyper::Body;
 use opentelemetry_proto::tonic::collector::metrics::v1::{
     ExportMetricsServiceRequest, ExportMetricsServiceResponse,
 };
+use opentelemetry_proto::tonic::collector::trace::v1::{
+    ExportTraceServiceRequest, ExportTraceServiceResponse,
+};
 use prost::Message;
 use session::context::QueryContextRef;
 use snafu::prelude::*;
@@ -33,16 +36,19 @@ pub async fn metrics(
     State(handler): State<OpenTelemetryProtocolHandlerRef>,
     Extension(query_ctx): Extension<QueryContextRef>,
     RawBody(body): RawBody,
-) -> Result<OtlpResponse> {
+) -> Result<OtlpMetricsResponse> {
     let _timer = timer!(
-        crate::metrics::METRIC_HTTP_OPENTELEMETRY_ELAPSED,
+        crate::metrics::METRIC_HTTP_OPENTELEMETRY_METRICS_ELAPSED,
         &[(crate::metrics::METRIC_DB_LABEL, query_ctx.get_db_string())]
     );
-    let request = parse_body(body).await?;
-    handler.metrics(request, query_ctx).await.map(OtlpResponse)
+    let request = parse_metrics_body(body).await?;
+    handler
+        .metrics(request, query_ctx)
+        .await
+        .map(OtlpMetricsResponse)
 }
 
-async fn parse_body(body: Body) -> Result<ExportMetricsServiceRequest> {
+async fn parse_metrics_body(body: Body) -> Result<ExportMetricsServiceRequest> {
     hyper::body::to_bytes(body)
         .await
         .context(error::HyperSnafu)
@@ -51,9 +57,47 @@ async fn parse_body(body: Body) -> Result<ExportMetricsServiceRequest> {
         })
 }
 
-pub struct OtlpResponse(ExportMetricsServiceResponse);
+pub struct OtlpMetricsResponse(ExportMetricsServiceResponse);
+
+impl IntoResponse for OtlpMetricsResponse {
+    fn into_response(self) -> axum::response::Response {
+        (
+            [(header::CONTENT_TYPE, "application/x-protobuf")],
+            self.0.encode_to_vec(),
+        )
+            .into_response()
+    }
+}
+
+#[axum_macros::debug_handler]
+pub async fn traces(
+    State(handler): State<OpenTelemetryProtocolHandlerRef>,
+    Extension(query_ctx): Extension<QueryContextRef>,
+    RawBody(body): RawBody,
+) -> Result<OtlpTracesResponse> {
+    let _timer = timer!(
+        crate::metrics::METRIC_HTTP_OPENTELEMETRY_TRACES_ELAPSED,
+        &[(crate::metrics::METRIC_DB_LABEL, query_ctx.get_db_string())]
+    );
+    let request = parse_traces_body(body).await?;
+    handler
+        .traces(request, query_ctx)
+        .await
+        .map(OtlpTracesResponse)
+}
+
+async fn parse_traces_body(body: Body) -> Result<ExportTraceServiceRequest> {
+    hyper::body::to_bytes(body)
+        .await
+        .context(error::HyperSnafu)
+        .and_then(|buf| {
+            ExportTraceServiceRequest::decode(&buf[..]).context(error::DecodeOtlpRequestSnafu)
+        })
+}
+
+pub struct OtlpTracesResponse(ExportTraceServiceResponse);
 
-impl IntoResponse for OtlpResponse {
+impl IntoResponse for OtlpTracesResponse {
     fn into_response(self) -> axum::response::Response {
         (
             [(header::CONTENT_TYPE, "application/x-protobuf")],

src/servers/src/metrics.rs

@@ -37,7 +37,10 @@ pub(crate) const METRIC_HTTP_INFLUXDB_WRITE_ELAPSED: &str = "servers.http_influx
 pub(crate) const METRIC_HTTP_PROM_STORE_WRITE_ELAPSED: &str =
     "servers.http_prometheus_write_elapsed";
 pub(crate) const METRIC_HTTP_PROM_STORE_READ_ELAPSED: &str = "servers.http_prometheus_read_elapsed";
-pub(crate) const METRIC_HTTP_OPENTELEMETRY_ELAPSED: &str = "servers.http_otlp_elapsed";
+pub(crate) const METRIC_HTTP_OPENTELEMETRY_METRICS_ELAPSED: &str =
+    "servers.http_otlp_metrics_elapsed";
+pub(crate) const METRIC_HTTP_OPENTELEMETRY_TRACES_ELAPSED: &str =
+    "servers.http_otlp_traces_elapsed";
 pub(crate) const METRIC_TCP_OPENTSDB_LINE_WRITE_ELAPSED: &str =
     "servers.opentsdb_line_write_elapsed";
 pub(crate) const METRIC_HTTP_PROMQL_INSTANT_QUERY_ELAPSED: &str =