Merge pull request #2 from fengys1996/feat/otlp_tracing_parse_body

feat(otlp): tracing parse body

src/frontend/src/instance/otlp.rs

@@ -24,6 +24,7 @@ use opentelemetry_proto::tonic::collector::trace::v1::{
 };
 use servers::error::{self, AuthSnafu, Result as ServerResult};
 use servers::otlp;
+use servers::otlp::plugin::TraceParserRef;
 use servers::query_handler::OpenTelemetryProtocolHandler;
 use session::context::QueryContextRef;
 use snafu::ResultExt;
@@ -70,7 +71,10 @@ impl OpenTelemetryProtocolHandler for Instance {
             .check_permission(ctx.current_user(), PermissionReq::Otlp)
             .context(AuthSnafu)?;
 
-        let (requests, rows) = otlp::traces::to_grpc_insert_requests(request)?;
+        let (requests, rows) = match self.plugins.get::<TraceParserRef>() {
+            Some(parser) => parser.parse(request)?,
+            None => otlp::trace::to_grpc_insert_requests(request)?,
+        };
 
         let _ = self
             .handle_row_inserts(requests, ctx)

src/servers/src/otlp.rs

@@ -13,4 +13,9 @@
 // limitations under the License.
 
 pub mod metrics;
-pub mod traces;
+pub mod plugin;
+pub mod trace;
+
+const GREPTIME_TIMESTAMP: &str = "greptime_timestamp";
+const GREPTIME_VALUE: &str = "greptime_value";
+const GREPTIME_COUNT: &str = "greptime_count";

src/servers/src/otlp/metrics.rs

@@ -18,12 +18,10 @@ use opentelemetry_proto::tonic::collector::metrics::v1::ExportMetricsServiceRequ
 use opentelemetry_proto::tonic::common::v1::{any_value, KeyValue};
 use opentelemetry_proto::tonic::metrics::v1::{metric, number_data_point, *};
 
+use super::{GREPTIME_COUNT, GREPTIME_TIMESTAMP, GREPTIME_VALUE};
 use crate::error::Result;
 use crate::row_writer::{self, MultiTableData, TableData};
 
-const GREPTIME_TIMESTAMP: &str = "greptime_timestamp";
-const GREPTIME_VALUE: &str = "greptime_value";
-const GREPTIME_COUNT: &str = "greptime_count";
 /// the default column count for table writer
 const APPROXIMATE_COLUMN_COUNT: usize = 8;
 

src/servers/src/otlp/traces.rs → src/servers/src/otlp/plugin.rs

@@ -12,21 +12,17 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::sync::Arc;
+
 use api::v1::RowInsertRequests;
 use opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest;
 
 use crate::error::Result;
 
-/// Convert OpenTelemetry traces to GreptimeDB row insert requests
-///
-/// See
-/// <https://github.com/open-telemetry/opentelemetry-proto/blob/main/opentelemetry/proto/trace/v1/trace.proto>
-/// for data structure of OTLP metrics.
-///
-/// Returns `InsertRequests` and total number of rows to ingest
-pub fn to_grpc_insert_requests(
-    _request: ExportTraceServiceRequest,
-) -> Result<(RowInsertRequests, usize)> {
-    // TODO:(fys, bobo)
-    todo!()
+/// Transformer helps to transform ExportTraceServiceRequest based on logic, like:
+///   - uplift some fields from Attributes (Map type) to column
+pub trait TraceParser: Send + Sync {
+    fn parse(&self, request: ExportTraceServiceRequest) -> Result<(RowInsertRequests, usize)>;
 }
+
+pub type TraceParserRef = Arc<dyn TraceParser>;