diff --git a/src/datanode/src/datanode.rs b/src/datanode/src/datanode.rs index 03fcef49261..4ae69a513ec 100644 --- a/src/datanode/src/datanode.rs +++ b/src/datanode/src/datanode.rs @@ -82,6 +82,7 @@ pub struct Datanode { heartbeat_task: Option, region_event_receiver: Option, region_server: RegionServer, + object_store_manager: ObjectStoreManagerRef, greptimedb_telemetry_task: Arc, leases_notifier: Option>, plugins: Plugins, @@ -151,6 +152,10 @@ impl Datanode { self.region_server.clone() } + pub fn object_store_manager(&self) -> &ObjectStoreManagerRef { + &self.object_store_manager + } + pub fn plugins(&self) -> Plugins { self.plugins.clone() } @@ -293,8 +298,10 @@ impl DatanodeBuilder { Some(node_id), gc_enabled, )); + let object_store_manager = Self::build_object_store_manager(&self.opts.storage).await?; let region_server = self .new_region_server( + object_store_manager.clone(), schema_metadata_manager, region_event_listener, file_ref_manager, @@ -359,6 +366,7 @@ impl DatanodeBuilder { services: ServerHandlers::default(), heartbeat_task, region_server, + object_store_manager, greptimedb_telemetry_task, region_event_receiver, leases_notifier, @@ -434,6 +442,7 @@ impl DatanodeBuilder { async fn new_region_server( &mut self, + object_store_manager: ObjectStoreManagerRef, schema_metadata_manager: SchemaMetadataManagerRef, event_listener: RegionServerEventListenerRef, file_ref_manager: FileReferenceManagerRef, @@ -470,7 +479,6 @@ impl DatanodeBuilder { ); region_server.install_remote_dyn_filter_receiver_injector(&self.plugins); - let object_store_manager = Self::build_object_store_manager(&opts.storage).await?; let engines = self .build_store_engines( object_store_manager, diff --git a/src/frontend/AGENTS.md b/src/frontend/AGENTS.md index b5c0871fc11..dd868b80097 100644 --- a/src/frontend/AGENTS.md +++ b/src/frontend/AGENTS.md @@ -47,6 +47,7 @@ remote datanodes via `operator`/`client`. ## Public surface - `Instance` (`instance.rs`) — the business-logic container. +- `OtlpMetricIngestor` (`instance/otlp.rs`, re-exported by `instance`) — optional converted-metric storage hook, after OTLP permission checks and translation. - `Frontend` (`frontend.rs`) — lifecycle wrapper around `Instance` + servers + heartbeat. - Created from `cmd`: `src/cmd/src/frontend.rs` (distributed) and `src/cmd/src/standalone.rs` (standalone, with embedded datanode). diff --git a/src/frontend/src/instance.rs b/src/frontend/src/instance.rs index dd89a1f3f95..9e6d57c9f3e 100644 --- a/src/frontend/src/instance.rs +++ b/src/frontend/src/instance.rs @@ -22,6 +22,7 @@ mod log_handler; mod logs; mod opentsdb; mod otlp; +pub use otlp::{OtlpMetricIngestor, OtlpMetricIngestorRef}; pub mod prom_store; mod promql; mod region_query; diff --git a/src/frontend/src/instance/otlp.rs b/src/frontend/src/instance/otlp.rs index 4c80f1d17f2..c350ecee7a0 100644 --- a/src/frontend/src/instance/otlp.rs +++ b/src/frontend/src/instance/otlp.rs @@ -18,6 +18,7 @@ pub mod trace_types; use std::sync::Arc; +use api::v1::RowInsertRequests; use async_trait::async_trait; use auth::{ OTLP_WRITE, PermissionChecker, PermissionCheckerRef, PermissionReq, PermissionTableTarget, @@ -52,6 +53,20 @@ use self::trace_ingest::trace_conventions; use crate::instance::Instance; use crate::metrics::{OTLP_LOGS_ROWS, OTLP_METRICS_ROWS, OTLP_RESOURCE_INFO_WRITE_ERRORS}; +/// Optional storage path for converted OTLP metrics, after protocol and table permissions. +#[async_trait] +pub trait OtlpMetricIngestor: Send + Sync { + async fn check_write(&self) -> ServerResult<()>; + + async fn ingest( + &self, + requests: RowInsertRequests, + ctx: QueryContextRef, + ) -> ServerResult; +} + +pub type OtlpMetricIngestorRef = Arc; + fn trace_permission_targets( table_name: &str, groups: &[TraceSpanGroup], @@ -101,6 +116,10 @@ impl OpenTelemetryProtocolHandler for Instance { .plugins .get::>(); interceptor_ref.pre_execute(ctx.clone())?; + let metric_ingestor = self.plugins.get::(); + if let Some(ingestor) = &metric_ingestor { + ingestor.check_write().await?; + } let ctx = Arc::new(ctx.fork()); let input_names = request @@ -121,6 +140,9 @@ impl OpenTelemetryProtocolHandler for Instance { metric_ctx.is_legacy = is_legacy; metric_ctx.resource_info = self.otlp_resource_info; + let conversion_timer = servers::metrics::METRIC_OTLP_METRICS_STAGE_ELAPSED + .with_label_values(&["convert"]) + .start_timer(); let otlp::metrics::MetricsConversion { requests, rows, @@ -128,6 +150,7 @@ impl OpenTelemetryProtocolHandler for Instance { resource_info, mut outcome, } = otlp::metrics::to_grpc_insert_requests(request, &mut metric_ctx)?; + drop(conversion_timer); if outcome.rejected_data_points > 0 { warn!( "Rejected {} OTLP metrics data points: {}", @@ -161,7 +184,9 @@ impl OpenTelemetryProtocolHandler for Instance { }; // OTLP tables have one sample field in both the legacy and physical paths. - let output = if metric_ctx.is_legacy || !metric_ctx.with_metric_engine { + let output = if let Some(ingestor) = metric_ingestor { + ingestor.ingest(requests, ctx.clone()).await + } else if metric_ctx.is_legacy || !metric_ctx.with_metric_engine { self.handle_row_inserts(requests, ctx.clone(), false, true) .await .map_err(BoxedError::new) diff --git a/src/servers/src/http/otlp.rs b/src/servers/src/http/otlp.rs index 2d5385c5e33..3ef217de7a9 100644 --- a/src/servers/src/http/otlp.rs +++ b/src/servers/src/http/otlp.rs @@ -100,11 +100,15 @@ pub async fn metrics( let _timer = crate::metrics::METRIC_HTTP_OPENTELEMETRY_METRICS_ELAPSED .with_label_values(&[db.as_str()]) .start_timer(); + let decode_timer = crate::metrics::METRIC_OTLP_METRICS_STAGE_ELAPSED + .with_label_values(&["http_decode"]) + .start_timer(); let request = ExportMetricsServiceRequest::decode(bytes).with_context(|_| { error::DecodeOtlpRequestSnafu { content_type: content_type_to_string(content_type.as_ref()), } })?; + drop(decode_timer); let OtlpState { with_metric_engine, diff --git a/src/servers/src/metrics.rs b/src/servers/src/metrics.rs index 0d70540be31..f8e9d930616 100644 --- a/src/servers/src/metrics.rs +++ b/src/servers/src/metrics.rs @@ -51,6 +51,13 @@ pub(crate) const METRIC_FAILURE_VALUE: &str = "failure"; lazy_static! { + pub static ref METRIC_OTLP_METRICS_STAGE_ELAPSED: HistogramVec = register_histogram_vec!( + "greptime_servers_otlp_metrics_stage_seconds", + "OTLP metrics decoding and conversion duration in seconds", + &["stage"], + prometheus::exponential_buckets(0.00001, 4.0, 12).unwrap() + ).unwrap(); + pub static ref HTTP_REQUEST_COUNTER: IntCounterVec = register_int_counter_vec!( "greptime_servers_http_request_counter", "servers http request counter",