chore: add schema urls to otlp logs (#4876)

* chore: add schema urls to otlp logs table

* chore: update meter-macros version to remove anymap warning

* chore: change span id and trace id to field

Cargo.toml

@@ -261,7 +261,7 @@ tokio-rustls = { git = "https://github.com/GreptimeTeam/tokio-rustls" }
 
 [workspace.dependencies.meter-macros]
 git = "https://github.com/GreptimeTeam/greptime-meter.git"
-rev = "80eb97c24c88af4dd9a86f8bbaf50e741d4eb8cd"
+rev = "a10facb353b41460eeb98578868ebf19c2084fac"
 
 [profile.release]
 debug = 1

src/frontend/src/instance/otlp.rs

@@ -113,6 +113,7 @@ impl OpenTelemetryProtocolHandler for Instance {
             .plugins
             .get::<OpenTelemetryProtocolInterceptorRef<servers::error::Error>>();
         interceptor_ref.pre_execute(ctx.clone())?;
+
         let (requests, rows) = otlp::logs::to_grpc_insert_requests(request, pipeline, table_name)?;
         self.handle_log_inserts(requests, ctx)
             .await

src/servers/src/http/otlp.rs

@@ -41,7 +41,7 @@ use snafu::prelude::*;
 
 use super::header::constants::GREPTIME_LOG_EXTRACT_KEYS_HEADER_NAME;
 use super::header::{write_cost_header_map, CONTENT_TYPE_PROTOBUF};
-use crate::error::{self, Result};
+use crate::error::{self, PipelineSnafu, Result};
 use crate::http::header::constants::{
     GREPTIME_LOG_PIPELINE_NAME_HEADER_NAME, GREPTIME_LOG_PIPELINE_VERSION_HEADER_NAME,
     GREPTIME_LOG_TABLE_NAME_HEADER_NAME,
@@ -227,15 +227,9 @@ pub async fn logs(
         .start_timer();
     let request = ExportLogsServiceRequest::decode(bytes).context(error::DecodeOtlpRequestSnafu)?;
 
-    let pipeline_way;
-    if let Some(pipeline_name) = &pipeline_info.pipeline_name {
+    let pipeline_way = if let Some(pipeline_name) = &pipeline_info.pipeline_name {
         let pipeline_version =
-            to_pipeline_version(pipeline_info.pipeline_version).map_err(|_| {
-                error::InvalidParameterSnafu {
-                    reason: GREPTIME_LOG_PIPELINE_VERSION_HEADER_NAME,
-                }
-                .build()
-            })?;
+            to_pipeline_version(pipeline_info.pipeline_version).context(PipelineSnafu)?;
         let pipeline = match handler
             .get_pipeline(pipeline_name, pipeline_version, query_ctx.clone())
             .await
@@ -245,10 +239,10 @@ pub async fn logs(
                 return Err(e);
             }
         };
-        pipeline_way = PipelineWay::Custom(pipeline);
+        PipelineWay::Custom(pipeline)
     } else {
-        pipeline_way = PipelineWay::OtlpLog(Box::new(select_info));
-    }
+        PipelineWay::OtlpLog(Box::new(select_info))
+    };
 
     handler
         .logs(request, pipeline_way, table_info.table_name, query_ctx)