[cluster telemetry] Internal telemetry endpoint (#7672)

coszio · web-flow · commit 8ab86b67360d · 2025-12-23T11:32:44.000-06:00
* move QdrantInternalService to its own file

* implement internal telemetry endpoint

cleaner diff on src/main.rs

upd response in proto

* use try_from conversion

* rename request/response in proto
diff --git a/lib/api/src/grpc/proto/qdrant_internal_service.proto b/lib/api/src/grpc/proto/qdrant_internal_service.proto
@@ -14,9 +14,8 @@ service QdrantInternal {
   rpc WaitOnConsensusCommit(WaitOnConsensusCommitRequest)
       returns (WaitOnConsensusCommitResponse) {}
 
-  // Get telemetry from a peer
-  rpc GetPeerTelemetry(GetPeerTelemetryRequest)
-      returns (GetPeerTelemetryResponse) {}
+  // Get telemetry
+  rpc GetTelemetry(GetTelemetryRequest) returns (GetTelemetryResponse) {}
 }
 
 message GetConsensusCommitRequest {}
diff --git a/lib/api/src/grpc/proto/telemetry_internal.proto b/lib/api/src/grpc/proto/telemetry_internal.proto
@@ -5,14 +5,14 @@ import "collections.proto";
 package qdrant;
 option csharp_namespace = "Qdrant.Client.Grpc";
 
-message GetPeerTelemetryRequest {
-  // The peer id to ask for telemetry
-  uint64 peer_id = 1;
+message GetTelemetryRequest {
   // The level of detail needed
-  uint32 details_level = 2;
+  uint32 details_level = 1;
+  // Timeout in secs for the request
+  uint64 timeout = 2;
 }
 
-message GetPeerTelemetryResponse {
+message GetTelemetryResponse {
   PeerTelemetry result = 1;
   double time = 2;
 }
diff --git a/lib/api/src/grpc/qdrant.rs b/lib/api/src/grpc/qdrant.rs
@@ -12449,18 +12449,18 @@ pub mod points_internal_server {
 #[derive(serde::Serialize)]
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
-pub struct GetPeerTelemetryRequest {
-    /// The peer id to ask for telemetry
-    #[prost(uint64, tag = "1")]
-    pub peer_id: u64,
+pub struct GetTelemetryRequest {
     /// The level of detail needed
-    #[prost(uint32, tag = "2")]
+    #[prost(uint32, tag = "1")]
     pub details_level: u32,
+    /// Timeout in secs for the request
+    #[prost(uint64, tag = "2")]
+    pub timeout: u64,
 }
 #[derive(serde::Serialize)]
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
-pub struct GetPeerTelemetryResponse {
+pub struct GetTelemetryResponse {
     #[prost(message, optional, tag = "1")]
     pub result: ::core::option::Option<PeerTelemetry>,
     #[prost(double, tag = "2")]
@@ -12913,12 +12913,12 @@ pub mod qdrant_internal_client {
                 );
             self.inner.unary(req, path, codec).await
         }
-        /// Get telemetry from a peer
-        pub async fn get_peer_telemetry(
+        /// Get telemetry
+        pub async fn get_telemetry(
             &mut self,
-            request: impl tonic::IntoRequest<super::GetPeerTelemetryRequest>,
+            request: impl tonic::IntoRequest<super::GetTelemetryRequest>,
         ) -> std::result::Result<
-            tonic::Response<super::GetPeerTelemetryResponse>,
+            tonic::Response<super::GetTelemetryResponse>,
             tonic::Status,
         > {
             self.inner
@@ -12932,11 +12932,11 @@ pub mod qdrant_internal_client {
                 })?;
             let codec = tonic::codec::ProstCodec::default();
             let path = http::uri::PathAndQuery::from_static(
-                "/qdrant.QdrantInternal/GetPeerTelemetry",
+                "/qdrant.QdrantInternal/GetTelemetry",
             );
             let mut req = request.into_request();
             req.extensions_mut()
-                .insert(GrpcMethod::new("qdrant.QdrantInternal", "GetPeerTelemetry"));
+                .insert(GrpcMethod::new("qdrant.QdrantInternal", "GetTelemetry"));
             self.inner.unary(req, path, codec).await
         }
     }
@@ -12964,12 +12964,12 @@ pub mod qdrant_internal_server {
             tonic::Response<super::WaitOnConsensusCommitResponse>,
             tonic::Status,
         >;
-        /// Get telemetry from a peer
-        async fn get_peer_telemetry(
+        /// Get telemetry
+        async fn get_telemetry(
             &self,
-            request: tonic::Request<super::GetPeerTelemetryRequest>,
+            request: tonic::Request<super::GetTelemetryRequest>,
         ) -> std::result::Result<
-            tonic::Response<super::GetPeerTelemetryResponse>,
+            tonic::Response<super::GetTelemetryResponse>,
             tonic::Status,
         >;
     }
@@ -13149,26 +13149,25 @@ pub mod qdrant_internal_server {
                     };
                     Box::pin(fut)
                 }
-                "/qdrant.QdrantInternal/GetPeerTelemetry" => {
+                "/qdrant.QdrantInternal/GetTelemetry" => {
                     #[allow(non_camel_case_types)]
-                    struct GetPeerTelemetrySvc<T: QdrantInternal>(pub Arc<T>);
+                    struct GetTelemetrySvc<T: QdrantInternal>(pub Arc<T>);
                     impl<
                         T: QdrantInternal,
-                    > tonic::server::UnaryService<super::GetPeerTelemetryRequest>
-                    for GetPeerTelemetrySvc<T> {
-                        type Response = super::GetPeerTelemetryResponse;
+                    > tonic::server::UnaryService<super::GetTelemetryRequest>
+                    for GetTelemetrySvc<T> {
+                        type Response = super::GetTelemetryResponse;
                         type Future = BoxFuture<
                             tonic::Response<Self::Response>,
                             tonic::Status,
                         >;
                         fn call(
                             &mut self,
-                            request: tonic::Request<super::GetPeerTelemetryRequest>,
+                            request: tonic::Request<super::GetTelemetryRequest>,
                         ) -> Self::Future {
                             let inner = Arc::clone(&self.0);
                             let fut = async move {
-                                <T as QdrantInternal>::get_peer_telemetry(&inner, request)
-                                    .await
+                                <T as QdrantInternal>::get_telemetry(&inner, request).await
                             };
                             Box::pin(fut)
                         }
@@ -13180,7 +13179,7 @@ pub mod qdrant_internal_server {
                     let inner = self.inner.clone();
                     let fut = async move {
                         let inner = inner.0;
-                        let method = GetPeerTelemetrySvc(inner);
+                        let method = GetTelemetrySvc(inner);
                         let codec = tonic::codec::ProstCodec::default();
                         let mut grpc = tonic::server::Grpc::new(codec)
                             .apply_compression_config(
diff --git a/src/common/telemetry.rs b/src/common/telemetry.rs
@@ -62,10 +62,6 @@ pub struct TelemetryData {
 }
 
 impl TelemetryCollector {
-    pub fn reporting_id(&self) -> String {
-        self.process_id.to_string()
-    }
-
     pub fn generate_id() -> Uuid {
         Uuid::new_v4()
     }
diff --git a/src/consensus.rs b/src/consensus.rs
@@ -27,6 +27,7 @@ use tokio::time::sleep;
 use tonic::transport::{ClientTlsConfig, Uri};
 
 use crate::common::helpers;
+use crate::common::telemetry::TelemetryCollector;
 use crate::common::telemetry_ops::requests_telemetry::TonicTelemetryCollector;
 use crate::settings::{ConsensusConfig, Settings};
 use crate::tonic::init_internal;
@@ -68,7 +69,8 @@ impl Consensus {
         settings: Settings,
         channel_service: ChannelService,
         propose_receiver: mpsc::Receiver<ConsensusOperations>,
-        telemetry_collector: Arc<parking_lot::Mutex<TonicTelemetryCollector>>,
+        telemetry_collector: Arc<tokio::sync::Mutex<TelemetryCollector>>,
+        tonic_telemetry_collector: Arc<parking_lot::Mutex<TonicTelemetryCollector>>,
         toc: Arc<TableOfContent>,
         runtime: Handle,
         reinit: bool,
@@ -156,6 +158,7 @@ impl Consensus {
                     toc,
                     state_ref,
                     telemetry_collector,
+                    tonic_telemetry_collector,
                     settings,
                     p2p_host,
                     p2p_port,
diff --git a/src/main.rs b/src/main.rs
@@ -401,7 +401,8 @@ fn main() -> anyhow::Result<()> {
     // It decides if query should go directly to the ToC or through the consensus.
     let mut dispatcher = Dispatcher::new(toc_arc.clone());
 
-    let (telemetry_collector, dispatcher_arc, health_checker) = if is_distributed_deployment {
+    let (telemetry_collector, tonic_telemetry_collector, dispatcher_arc, health_checker);
+    if is_distributed_deployment {
         let consensus_state: ConsensusStateRef = ConsensusManager::new(
             persistent_consensus_state,
             toc_arc.clone(),
@@ -418,26 +419,28 @@ fn main() -> anyhow::Result<()> {
         let toc_dispatcher = TocDispatcher::new(Arc::downgrade(&toc_arc), consensus_state.clone());
         toc_arc.with_toc_dispatcher(toc_dispatcher);
 
-        let dispatcher_arc = Arc::new(dispatcher);
+        dispatcher_arc = Arc::new(dispatcher);
 
         // Monitoring and telemetry.
-        let telemetry_collector =
+        let telemetry =
             TelemetryCollector::new(settings.clone(), dispatcher_arc.clone(), reporting_id);
-        let tonic_telemetry_collector = telemetry_collector.tonic_telemetry_collector.clone();
+        tonic_telemetry_collector = telemetry.tonic_telemetry_collector.clone();
+
+        telemetry_collector = Arc::new(tokio::sync::Mutex::new(telemetry));
 
         // `raft` crate uses `slog` crate so it is needed to use `slog_stdlog::StdLog` to forward
         // logs from it to `log` crate
         let slog_logger = slog::Logger::root(slog_stdlog::StdLog.fuse(), slog::o!());
 
         // Runs raft consensus in a separate thread.
         // Create a pipe `message_sender` to communicate with the consensus
-        let health_checker = Arc::new(common::health::HealthChecker::spawn(
+        health_checker = Some(Arc::new(common::health::HealthChecker::spawn(
             toc_arc.clone(),
             consensus_state.clone(),
             &runtime_handle,
             // NOTE: `wait_for_bootstrap` should be calculated *before* starting `Consensus` thread
             consensus_state.is_new_deployment() && bootstrap.is_some(),
-        ));
+        )));
 
         let handle = Consensus::run(
             &slog_logger,
@@ -447,7 +450,8 @@ fn main() -> anyhow::Result<()> {
             settings.clone(),
             channel_service,
             propose_receiver,
-            tonic_telemetry_collector,
+            telemetry_collector.clone(),
+            tonic_telemetry_collector.clone(),
             toc_arc.clone(),
             runtime_handle.clone(),
             args.reinit,
@@ -499,27 +503,23 @@ fn main() -> anyhow::Result<()> {
                 collections_to_recover_in_consensus,
             ));
         }
-
-        (telemetry_collector, dispatcher_arc, Some(health_checker))
     } else {
         log::info!("Distributed mode disabled");
-        let dispatcher_arc = Arc::new(dispatcher);
+        dispatcher_arc = Arc::new(dispatcher);
 
         // Monitoring and telemetry.
-        let telemetry_collector =
+        let telemetry =
             TelemetryCollector::new(settings.clone(), dispatcher_arc.clone(), reporting_id);
-        (telemetry_collector, dispatcher_arc, None)
-    };
 
-    let tonic_telemetry_collector = telemetry_collector.tonic_telemetry_collector.clone();
+        tonic_telemetry_collector = telemetry.tonic_telemetry_collector.clone();
+        telemetry_collector = Arc::new(tokio::sync::Mutex::new(telemetry));
+        health_checker = None;
+    };
 
     //
     // Telemetry reporting
     //
 
-    let reporting_id = telemetry_collector.reporting_id();
-    let telemetry_collector = Arc::new(tokio::sync::Mutex::new(telemetry_collector));
-
     if reporting_enabled {
         log::info!("Telemetry reporting enabled, id: {reporting_id}");
 
diff --git a/src/tonic/api/mod.rs b/src/tonic/api/mod.rs
@@ -2,6 +2,7 @@ pub mod collections_api;
 pub mod collections_internal_api;
 pub mod points_api;
 pub mod points_internal_api;
+pub mod qdrant_internal_api;
 pub mod raft_api;
 pub mod snapshots_api;
 
diff --git a/src/tonic/api/qdrant_internal_api.rs b/src/tonic/api/qdrant_internal_api.rs
@@ -0,0 +1,107 @@
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+use api::grpc::qdrant_internal_server::QdrantInternal;
+use api::grpc::{
+    GetConsensusCommitRequest, GetConsensusCommitResponse, GetTelemetryRequest,
+    GetTelemetryResponse, PeerTelemetry, WaitOnConsensusCommitRequest,
+    WaitOnConsensusCommitResponse,
+};
+use common::types::{DetailsLevel, TelemetryDetail};
+use storage::content_manager::consensus_manager::ConsensusStateRef;
+use storage::rbac::Access;
+use tokio::sync::Mutex;
+use tonic::{Request, Response, Status};
+
+use crate::common::telemetry::TelemetryCollector;
+use crate::settings::Settings;
+
+pub struct QdrantInternalService {
+    /// Telemetry collector
+    telemetry_collector: Arc<Mutex<TelemetryCollector>>,
+    /// Qdrant settings
+    settings: Settings,
+    /// Consensus state
+    consensus_state: ConsensusStateRef,
+}
+
+impl QdrantInternalService {
+    pub fn new(
+        telemetry_collector: Arc<Mutex<TelemetryCollector>>,
+        settings: Settings,
+        consensus_state: ConsensusStateRef,
+    ) -> Self {
+        Self {
+            telemetry_collector,
+            settings,
+            consensus_state,
+        }
+    }
+}
+
+#[tonic::async_trait]
+impl QdrantInternal for QdrantInternalService {
+    async fn get_consensus_commit(
+        &self,
+        _: tonic::Request<GetConsensusCommitRequest>,
+    ) -> Result<Response<GetConsensusCommitResponse>, Status> {
+        let persistent = self.consensus_state.persistent.read();
+        let commit = persistent.state.hard_state.commit as _;
+        let term = persistent.state.hard_state.term as _;
+        Ok(Response::new(GetConsensusCommitResponse { commit, term }))
+    }
+
+    async fn wait_on_consensus_commit(
+        &self,
+        request: Request<WaitOnConsensusCommitRequest>,
+    ) -> Result<Response<WaitOnConsensusCommitResponse>, Status> {
+        let request = request.into_inner();
+        let commit = request.commit as u64;
+        let term = request.term as u64;
+        let timeout = Duration::from_secs(request.timeout as u64);
+        let consensus_tick = Duration::from_millis(self.settings.cluster.consensus.tick_period_ms);
+        let ok = self
+            .consensus_state
+            .wait_for_consensus_commit(commit, term, consensus_tick, timeout)
+            .await
+            .is_ok();
+        Ok(Response::new(WaitOnConsensusCommitResponse { ok }))
+    }
+
+    async fn get_telemetry(
+        &self,
+        request: Request<GetTelemetryRequest>,
+    ) -> Result<Response<GetTelemetryResponse>, Status> {
+        let request = request.into_inner();
+
+        if request.details_level < 2 {
+            return Err(Status::invalid_argument(
+                "details_level for internal service must be >= 2",
+            ));
+        }
+
+        let details_level = DetailsLevel::from(request.details_level.max(2) as usize);
+
+        let detail = TelemetryDetail {
+            level: details_level,
+            histograms: false,
+        };
+
+        let timing = Instant::now();
+        let timeout = Duration::from_secs(request.timeout);
+
+        let access = Access::full("internal service");
+
+        let telemetry_collector = self.telemetry_collector.lock().await;
+        let telemetry_data = telemetry_collector
+            .prepare_data(&access, detail, Some(timeout))
+            .await?;
+
+        let response = GetTelemetryResponse {
+            result: Some(PeerTelemetry::try_from(telemetry_data)?),
+            time: timing.elapsed().as_secs_f64(),
+        };
+
+        Ok(Response::new(response))
+    }
+}
diff --git a/src/tonic/mod.rs b/src/tonic/mod.rs

Original file line number	Diff line number	Diff line change
`@@ -62,10 +62,6 @@ pub struct TelemetryData {`
`62`	`62`	`}`
`63`	`63`
`64`	`64`	`impl TelemetryCollector {`
`65`		`- pub fn reporting_id(&self) -> String {`
`66`		`- self.process_id.to_string()`
`67`		`- }`
`68`		`-`
`69`	`65`	`pub fn generate_id() -> Uuid {`
`70`	`66`	`Uuid::new_v4()`
`71`	`67`	`}`