diff --git a/Cargo.lock b/Cargo.lock index 1d0b7bd655..51e707f7a3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5855,6 +5855,7 @@ dependencies = [ "itertools 0.14.0", "jsonwebtoken", "log", + "macros", "mockito", "murmur3", "nix 0.31.2", diff --git a/Cargo.toml b/Cargo.toml index f9199653de..f5aa2f7bb0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -121,6 +121,7 @@ raft-proto = { version = "0.7.0", features = [ common = { path = "lib/common/common" } cancel = { path = "lib/common/cancel" } issues = { path = "lib/common/issues" } +macros = { path = "lib/macros" } segment = { path = "lib/segment", default-features = false } shard = { path = "lib/shard", default-features = false } collection = { path = "lib/collection" } diff --git a/docs/redoc/master/openapi.json b/docs/redoc/master/openapi.json index c4e8d2103e..fc3a48abd6 100644 --- a/docs/redoc/master/openapi.json +++ b/docs/redoc/master/openapi.json @@ -321,6 +321,15 @@ "minimum": 0 } }, + { + "name": "per_collection", + "in": "query", + "description": "If true, include per-collection request statistics in the response", + "required": false, + "schema": { + "type": "boolean" + } + }, { "name": "timeout", "in": "query", @@ -411,6 +420,15 @@ "type": "boolean" } }, + { + "name": "per_collection", + "in": "query", + "description": "If true, include per-collection request metrics with a collection label instead of global request metrics", + "required": false, + "schema": { + "type": "boolean" + } + }, { "name": "timeout", "in": "query", @@ -13877,6 +13895,18 @@ "$ref": "#/components/schemas/OperationDurationStatistics" } } + }, + "per_collection_responses": { + "type": "object", + "additionalProperties": { + "type": "object", + "additionalProperties": { + "type": "object", + "additionalProperties": { + "$ref": "#/components/schemas/OperationDurationStatistics" + } + } + } } } }, @@ -13894,6 +13924,18 @@ "$ref": "#/components/schemas/OperationDurationStatistics" } } + }, + "per_collection_responses": { + "type": "object", + "additionalProperties": { + "type": "object", + "additionalProperties": { + "type": "object", + "additionalProperties": { + "$ref": "#/components/schemas/OperationDurationStatistics" + } + } + } } } }, diff --git a/lib/common/common/src/types.rs b/lib/common/common/src/types.rs index 8b6182a66a..86d2347d3e 100644 --- a/lib/common/common/src/types.rs +++ b/lib/common/common/src/types.rs @@ -34,11 +34,16 @@ impl PartialOrd for ScoredPointOffset { pub struct TelemetryDetail { pub level: DetailsLevel, pub histograms: bool, + pub per_collection: bool, } impl TelemetryDetail { pub fn new(level: DetailsLevel, histograms: bool) -> Self { - Self { level, histograms } + Self { + level, + histograms, + per_collection: false, + } } } @@ -76,6 +81,7 @@ impl Default for TelemetryDetail { TelemetryDetail { level: DetailsLevel::Level0, histograms: false, + per_collection: false, } } } diff --git a/openapi/openapi-service.ytt.yaml b/openapi/openapi-service.ytt.yaml index 7e08b70d35..d881e5d709 100644 --- a/openapi/openapi-service.ytt.yaml +++ b/openapi/openapi-service.ytt.yaml @@ -39,6 +39,12 @@ paths: schema: type: integer minimum: 0 + - name: per_collection + in: query + description: "If true, include per-collection request statistics in the response" + required: false + schema: + type: boolean - name: timeout in: query description: "Timeout for this request" @@ -63,6 +69,12 @@ paths: required: false schema: type: boolean + - name: per_collection + in: query + description: "If true, include per-collection request metrics with a collection label instead of global request metrics" + required: false + schema: + type: boolean - name: timeout in: query description: "Timeout for this request" diff --git a/src/actix/actix_telemetry.rs b/src/actix/actix_telemetry.rs index 0d95be242e..c4dfab7e55 100644 --- a/src/actix/actix_telemetry.rs +++ b/src/actix/actix_telemetry.rs @@ -39,6 +39,7 @@ where let match_pattern = request .match_pattern() .unwrap_or_else(|| "unknown".to_owned()); + let request_key = format!("{} {}", request.method(), match_pattern); let future = self.service.call(request); let telemetry_data = self.telemetry_data.clone(); @@ -46,9 +47,16 @@ where let instant = std::time::Instant::now(); let response = future.await?; let status = response.response().status().as_u16(); + + let collection_name = response + .request() + .match_info() + .get("collection_name") + .map(|s| s.to_string()); + telemetry_data .lock() - .add_response(request_key, status, instant); + .add_response(request_key, status, instant, collection_name); Ok(response) }) } @@ -88,3 +96,54 @@ where })) } } + +#[cfg(test)] +mod tests { + use std::path::Path; + + use crate::actix::api::query_api::THIS_FILE; + + /// Recursively collect all `.rs` files under `dir`. + fn collect_rs_files(dir: &Path, out: &mut Vec) { + for entry in fs_err::read_dir(dir).unwrap() { + let entry = entry.unwrap(); + let path = entry.path(); + if path.is_dir() { + collect_rs_files(&path, out); + } else if path.extension().is_some_and(|ext| ext == "rs") { + out.push(path); + } + } + } + + #[test] + fn test_collection_routes_use_collection_name_param() { + let manifest_path = Path::new(env!("CARGO_MANIFEST_DIR")); + let file_path = manifest_path.join(THIS_FILE); + let actix_dir = file_path.parent().unwrap(); + + let mut rs_files = Vec::new(); + collect_rs_files(actix_dir, &mut rs_files); + + let mut bad_lines = Vec::new(); + for path in &rs_files { + let contents = fs_err::read_to_string(path).unwrap(); + for (line_no, line) in contents.lines().enumerate() { + let trimmed = line.trim(); + // Only check route attribute macros like #[get("/collections/{...")] + if trimmed.starts_with("#[") + && trimmed.contains("/collections/{") + && !trimmed.contains("{collection_name}") + { + bad_lines.push(format!("{}:{}: {}", path.display(), line_no + 1, trimmed)); + } + } + } + + assert!( + bad_lines.is_empty(), + "All collection routes must use {{collection_name}} as path parameter:\n{}", + bad_lines.join("\n"), + ); + } +} diff --git a/src/actix/api/collections_api.rs b/src/actix/api/collections_api.rs index 6ed885e768..cfea9f8d77 100644 --- a/src/actix/api/collections_api.rs +++ b/src/actix/api/collections_api.rs @@ -56,7 +56,7 @@ async fn get_aliases( helpers::time(do_list_aliases(dispatcher.toc(&auth, &pass), &auth)).await } -#[get("/collections/{name}")] +#[get("/collections/{collection_name}")] async fn get_collection( dispatcher: web::Data, collection: Path, @@ -68,13 +68,13 @@ async fn get_collection( helpers::time(do_get_collection( dispatcher.toc(&auth, &pass), &auth, - &collection.name, + &collection.collection_name, None, )) .await } -#[get("/collections/{name}/exists")] +#[get("/collections/{collection_name}/exists")] async fn get_collection_existence( dispatcher: web::Data, collection: Path, @@ -86,12 +86,12 @@ async fn get_collection_existence( helpers::time(do_collection_exists( dispatcher.toc(&auth, &pass), &auth, - &collection.name, + &collection.collection_name, )) .await } -#[get("/collections/{name}/aliases")] +#[get("/collections/{collection_name}/aliases")] async fn get_collection_aliases( dispatcher: web::Data, collection: Path, @@ -103,12 +103,12 @@ async fn get_collection_aliases( helpers::time(do_list_collection_aliases( dispatcher.toc(&auth, &pass), &auth, - &collection.name, + &collection.collection_name, )) .await } -#[put("/collections/{name}")] +#[put("/collections/{collection_name}")] async fn create_collection( dispatcher: web::Data, collection: Path, @@ -118,7 +118,7 @@ async fn create_collection( ) -> HttpResponse { let timing = Instant::now(); let create_collection_op = - CreateCollectionOperation::new(collection.name.clone(), operation.into_inner()); + CreateCollectionOperation::new(collection.collection_name.clone(), operation.into_inner()); let Ok(create_collection_op) = create_collection_op else { return process_response(create_collection_op, timing, None); @@ -134,7 +134,7 @@ async fn create_collection( process_response(response, timing, None) } -#[patch("/collections/{name}")] +#[patch("/collections/{collection_name}")] async fn update_collection( dispatcher: web::Data, collection: Path, @@ -143,7 +143,7 @@ async fn update_collection( ActixAuth(auth): ActixAuth, ) -> impl Responder { let timing = Instant::now(); - let name = collection.name.clone(); + let name = collection.collection_name.clone(); let response = dispatcher .submit_collection_meta_op( CollectionMetaOperations::UpdateCollection(UpdateCollectionOperation::new( @@ -157,7 +157,7 @@ async fn update_collection( process_response(response, timing, None) } -#[delete("/collections/{name}")] +#[delete("/collections/{collection_name}")] async fn delete_collection( dispatcher: web::Data, collection: Path, @@ -168,7 +168,7 @@ async fn delete_collection( let response = dispatcher .submit_collection_meta_op( CollectionMetaOperations::DeleteCollection(DeleteCollectionOperation( - collection.name.clone(), + collection.collection_name.clone(), )), auth, query.timeout(), @@ -195,7 +195,7 @@ async fn update_aliases( process_response(response, timing, None) } -#[get("/collections/{name}/cluster")] +#[get("/collections/{collection_name}/cluster")] async fn get_cluster_info( dispatcher: web::Data, collection: Path, @@ -207,12 +207,12 @@ async fn get_cluster_info( helpers::time(do_get_collection_cluster( dispatcher.toc(&auth, &pass), &auth, - &collection.name, + &collection.collection_name, )) .await } -#[post("/collections/{name}/cluster")] +#[post("/collections/{collection_name}/cluster")] async fn update_collection_cluster( dispatcher: web::Data, collection: Path, @@ -224,7 +224,7 @@ async fn update_collection_cluster( let wait_timeout = query.timeout(); let response = do_update_collection_cluster( &dispatcher.into_inner(), - collection.name.clone(), + collection.collection_name.clone(), operation.0, auth, wait_timeout, @@ -273,7 +273,7 @@ impl TryFrom<&OptimizationsParam> for OptimizationsRequestOptions { } } -#[get("/collections/{name}/optimizations")] +#[get("/collections/{collection_name}/optimizations")] fn get_optimizations( dispatcher: web::Data, collection: Path, @@ -284,7 +284,7 @@ fn get_optimizations( let options = OptimizationsRequestOptions::try_from(¶ms.into_inner())?; let pass = new_unchecked_verification_pass(); let collection_pass = auth.check_collection_access( - &collection.name, + &collection.collection_name, AccessRequirements::new(), "get_optimizations", )?; diff --git a/src/actix/api/count_api.rs b/src/actix/api/count_api.rs index 212346c6e5..e274e9c016 100644 --- a/src/actix/api/count_api.rs +++ b/src/actix/api/count_api.rs @@ -13,7 +13,7 @@ use crate::actix::helpers::{self, get_request_hardware_counter, process_response use crate::common::query::do_count_points; use crate::settings::ServiceConfig; -#[post("/collections/{name}/points/count")] +#[post("/collections/{collection_name}/points/count")] async fn count_points( dispatcher: web::Data, collection: Path, @@ -30,7 +30,7 @@ async fn count_points( let pass = match check_strict_mode( &count_request, params.timeout_as_secs(), - &collection.name, + &collection.collection_name, &dispatcher, &auth, ) @@ -47,7 +47,7 @@ async fn count_points( let request_hw_counter = get_request_hardware_counter( &dispatcher, - collection.name.clone(), + collection.collection_name.clone(), service_config.hardware_reporting(), None, ); @@ -56,7 +56,7 @@ async fn count_points( let result = do_count_points( dispatcher.toc(&auth, &pass), - &collection.name, + &collection.collection_name, count_request, params.consistency, params.timeout(), diff --git a/src/actix/api/debug_api.rs b/src/actix/api/debug_api.rs index 6924e96f9a..c74fcdffbd 100644 --- a/src/actix/api/debug_api.rs +++ b/src/actix/api/debug_api.rs @@ -41,7 +41,7 @@ mod staging { use super::*; use crate::actix::helpers; - #[get("/collections/{collection}/shards/{shard}/wal")] + #[get("/collections/{collection_name}/shards/{shard}/wal")] pub async fn get_shard_wal( dispatcher: web::Data, path: web::Path<(String, ShardId)>, @@ -95,7 +95,7 @@ mod staging { } } - #[get("/collections/{collection}/shards/{shard}/recovery_point")] + #[get("/collections/{collection_name}/shards/{shard}/recovery_point")] pub async fn get_shard_recovery_point( dispatcher: web::Data, path: web::Path<(String, ShardId)>, diff --git a/src/actix/api/discover_api.rs b/src/actix/api/discover_api.rs index 29f19722af..0ef23983bb 100644 --- a/src/actix/api/discover_api.rs +++ b/src/actix/api/discover_api.rs @@ -16,7 +16,7 @@ use crate::actix::helpers::{self, get_request_hardware_counter, process_response use crate::common::query::do_discover_batch_points; use crate::settings::ServiceConfig; -#[post("/collections/{name}/points/discover")] +#[post("/collections/{collection_name}/points/discover")] async fn discover_points( dispatcher: web::Data, collection: Path, @@ -33,7 +33,7 @@ async fn discover_points( let pass = match check_strict_mode( &discover_request, params.timeout_as_secs(), - &collection.name, + &collection.collection_name, &dispatcher, &auth, ) @@ -50,7 +50,7 @@ async fn discover_points( let request_hw_counter = get_request_hardware_counter( &dispatcher, - collection.name.clone(), + collection.collection_name.clone(), service_config.hardware_reporting(), None, ); @@ -60,7 +60,7 @@ async fn discover_points( let result = dispatcher .toc(&auth, &pass) .discover( - &collection.name, + &collection.collection_name, discover_request, params.consistency, shard_selection, @@ -79,7 +79,7 @@ async fn discover_points( helpers::process_response(result, timing, request_hw_counter.to_rest_api()) } -#[post("/collections/{name}/points/discover/batch")] +#[post("/collections/{collection_name}/points/discover/batch")] async fn discover_batch_points( dispatcher: web::Data, collection: Path, @@ -93,7 +93,7 @@ async fn discover_batch_points( let pass = match check_strict_mode_batch( request.searches.iter().map(|i| &i.discover_request), params.timeout_as_secs(), - &collection.name, + &collection.collection_name, &dispatcher, &auth, ) @@ -105,7 +105,7 @@ async fn discover_batch_points( let request_hw_counter = get_request_hardware_counter( &dispatcher, - collection.name.clone(), + collection.collection_name.clone(), service_config.hardware_reporting(), None, ); @@ -113,7 +113,7 @@ async fn discover_batch_points( let result = do_discover_batch_points( dispatcher.toc(&auth, &pass), - &collection.name, + &collection.collection_name, request, params.consistency, auth, diff --git a/src/actix/api/facet_api.rs b/src/actix/api/facet_api.rs index da083f13e7..abe5fdfa3a 100644 --- a/src/actix/api/facet_api.rs +++ b/src/actix/api/facet_api.rs @@ -14,7 +14,7 @@ use crate::actix::helpers::{ }; use crate::settings::ServiceConfig; -#[post("/collections/{name}/facet")] +#[post("/collections/{collection_name}/facet")] async fn facet( dispatcher: web::Data, collection: Path, @@ -33,7 +33,7 @@ async fn facet( let pass = match check_strict_mode( &facet_request, params.timeout_as_secs(), - &collection.name, + &collection.collection_name, &dispatcher, &auth, ) @@ -52,7 +52,7 @@ async fn facet( let request_hw_counter = get_request_hardware_counter( &dispatcher, - collection.name.clone(), + collection.collection_name.clone(), service_config.hardware_reporting(), None, ); @@ -60,7 +60,7 @@ async fn facet( let response = dispatcher .toc(&auth, &pass) .facet( - &collection.name, + &collection.collection_name, facet_params, shard_selection, params.consistency, diff --git a/src/actix/api/local_shard_api.rs b/src/actix/api/local_shard_api.rs index b1ecce298b..1fda9a6b1c 100644 --- a/src/actix/api/local_shard_api.rs +++ b/src/actix/api/local_shard_api.rs @@ -33,7 +33,7 @@ pub fn config_local_shard_api(cfg: &mut web::ServiceConfig) { .service(cleanup_shard); } -#[post("/collections/{collection}/shards/{shard}/points")] +#[post("/collections/{collection_name}/shards/{shard}/points")] async fn get_points( dispatcher: web::Data, ActixAuth(auth): ActixAuth, @@ -47,7 +47,7 @@ async fn get_points( let request_hw_counter = get_request_hardware_counter( &dispatcher, - path.collection.clone(), + path.collection_name.clone(), service_config.hardware_reporting(), None, ); @@ -55,7 +55,7 @@ async fn get_points( let records = query::do_get_points( dispatcher.toc(&auth, &pass), - &path.collection, + &path.collection_name, request.into_inner(), params.consistency, params.timeout(), @@ -74,7 +74,7 @@ async fn get_points( process_response(records, timing, request_hw_counter.to_rest_api()) } -#[post("/collections/{collection}/shards/{shard}/points/scroll")] +#[post("/collections/{collection_name}/shards/{shard}/points/scroll")] async fn scroll_points( dispatcher: web::Data, ActixAuth(auth): ActixAuth, @@ -93,7 +93,7 @@ async fn scroll_points( let pass = match check_strict_mode( &request, params.timeout_as_secs(), - &path.collection, + &path.collection_name, &dispatcher, &auth, ) @@ -105,7 +105,7 @@ async fn scroll_points( let request_hw_counter = get_request_hardware_counter( &dispatcher, - path.collection.clone(), + path.collection_name.clone(), service_config.hardware_reporting(), None, ); @@ -116,7 +116,7 @@ async fn scroll_points( get_hash_ring_filter( &dispatcher, &auth, - &path.collection.clone(), + &path.collection_name.clone(), AccessRequirements::new(), filter.expected_shard_id, &pass, @@ -132,7 +132,7 @@ async fn scroll_points( request.filter = merge_with_optional_filter(request.filter.take(), hash_ring_filter); dispatcher.toc(&auth, &pass).scroll( - &path.collection, + &path.collection_name, request, params.consistency, params.timeout(), @@ -150,7 +150,7 @@ async fn scroll_points( process_response(result, timing, request_hw_counter.to_rest_api()) } -#[post("/collections/{collection}/shards/{shard}/points/count")] +#[post("/collections/{collection_name}/shards/{shard}/points/count")] async fn count_points( dispatcher: web::Data, ActixAuth(auth): ActixAuth, @@ -167,7 +167,7 @@ async fn count_points( let pass = match check_strict_mode( &request, params.timeout_as_secs(), - &path.collection, + &path.collection_name, &dispatcher, &auth, ) @@ -179,7 +179,7 @@ async fn count_points( let request_hw_counter = get_request_hardware_counter( &dispatcher, - path.collection.clone(), + path.collection_name.clone(), service_config.hardware_reporting(), None, ); @@ -191,7 +191,7 @@ async fn count_points( Some(filter) => get_hash_ring_filter( &dispatcher, &auth, - &path.collection, + &path.collection_name, AccessRequirements::new(), filter.expected_shard_id, &pass, @@ -206,7 +206,7 @@ async fn count_points( query::do_count_points( dispatcher.toc(&auth, &pass), - &path.collection, + &path.collection_name, request, params.consistency, params.timeout(), @@ -230,7 +230,7 @@ pub struct CleanParams { pub timeout: Option, } -#[post("/collections/{collection}/shards/{shard}/cleanup")] +#[post("/collections/{collection_name}/shards/{shard}/cleanup")] async fn cleanup_shard( dispatcher: web::Data, ActixAuth(auth): ActixAuth, @@ -245,7 +245,13 @@ async fn cleanup_shard( let timeout = params.timeout.map(|sec| Duration::from_secs(sec.get())); dispatcher .toc(&auth, &pass) - .cleanup_local_shard(&path.collection, path.shard, auth, params.wait, timeout) + .cleanup_local_shard( + &path.collection_name, + path.shard, + auth, + params.wait, + timeout, + ) .await }) .await @@ -254,7 +260,7 @@ async fn cleanup_shard( #[derive(serde::Deserialize, validator::Validate)] struct CollectionShard { #[validate(length(min = 1, max = 255))] - collection: String, + collection_name: String, shard: ShardId, } diff --git a/src/actix/api/mod.rs b/src/actix/api/mod.rs index 2e64ad889a..cb1c6d44c6 100644 --- a/src/actix/api/mod.rs +++ b/src/actix/api/mod.rs @@ -32,7 +32,7 @@ struct StrictCollectionPath { length(min = 1, max = 255), custom(function = "validate_collection_name") )] - name: String, + collection_name: String, } /// A collection path with basic validation @@ -46,5 +46,5 @@ struct CollectionPath { length(min = 1, max = 255), custom(function = "validate_collection_name_legacy") )] - name: String, + collection_name: String, } diff --git a/src/actix/api/query_api.rs b/src/actix/api/query_api.rs index 538ba911db..b632ddb10f 100644 --- a/src/actix/api/query_api.rs +++ b/src/actix/api/query_api.rs @@ -24,7 +24,10 @@ use crate::common::inference::query_requests_rest::{ use crate::common::query::do_query_point_groups; use crate::settings::ServiceConfig; -#[post("/collections/{name}/points/query")] +#[cfg(test)] +pub const THIS_FILE: &str = file!(); + +#[post("/collections/{collection_name}/points/query")] #[allow(clippy::too_many_arguments)] async fn query_points( dispatcher: web::Data, @@ -42,7 +45,7 @@ async fn query_points( let request_hw_counter = get_request_hardware_counter( &dispatcher, - collection.name.clone(), + collection.collection_name.clone(), service_config.hardware_reporting(), None, ); @@ -66,7 +69,7 @@ async fn query_points( let pass = check_strict_mode( &request, params.timeout_as_secs(), - &collection.name, + &collection.collection_name, &dispatcher, &auth, ) @@ -75,7 +78,7 @@ async fn query_points( let points = dispatcher .toc(&auth, &pass) .query_batch( - &collection.name, + &collection.collection_name, vec![(request, shard_selection)], params.consistency, auth, @@ -104,7 +107,7 @@ async fn query_points( } #[allow(clippy::too_many_arguments)] -#[post("/collections/{name}/points/query/batch")] +#[post("/collections/{collection_name}/points/query/batch")] async fn query_points_batch( dispatcher: web::Data, collection: Path, @@ -118,7 +121,7 @@ async fn query_points_batch( let request_hw_counter = get_request_hardware_counter( &dispatcher, - collection.name.clone(), + collection.collection_name.clone(), service_config.hardware_reporting(), None, ); @@ -154,7 +157,7 @@ async fn query_points_batch( let pass = check_strict_mode_batch( batch.iter().map(|i| &i.0), params.timeout_as_secs(), - &collection.name, + &collection.collection_name, &dispatcher, &auth, ) @@ -163,7 +166,7 @@ async fn query_points_batch( let res = dispatcher .toc(&auth, &pass) .query_batch( - &collection.name, + &collection.collection_name, batch, params.consistency, auth, @@ -192,7 +195,7 @@ async fn query_points_batch( } #[allow(clippy::too_many_arguments)] -#[post("/collections/{name}/points/query/groups")] +#[post("/collections/{collection_name}/points/query/groups")] async fn query_points_groups( dispatcher: web::Data, collection: Path, @@ -209,7 +212,7 @@ async fn query_points_groups( let request_hw_counter = get_request_hardware_counter( &dispatcher, - collection.name.clone(), + collection.collection_name.clone(), service_config.hardware_reporting(), None, ); @@ -232,7 +235,7 @@ async fn query_points_groups( let pass = check_strict_mode( &request, params.timeout_as_secs(), - &collection.name, + &collection.collection_name, &dispatcher, &auth, ) @@ -240,7 +243,7 @@ async fn query_points_groups( let query_result = do_query_point_groups( dispatcher.toc(&auth, &pass), - &collection.name, + &collection.collection_name, request, params.consistency, shard_selection, diff --git a/src/actix/api/recommend_api.rs b/src/actix/api/recommend_api.rs index c9df77e661..e7557ac2ab 100644 --- a/src/actix/api/recommend_api.rs +++ b/src/actix/api/recommend_api.rs @@ -25,7 +25,7 @@ use crate::actix::auth::ActixAuth; use crate::actix::helpers::{self, get_request_hardware_counter, process_response_error}; use crate::settings::ServiceConfig; -#[post("/collections/{name}/points/recommend")] +#[post("/collections/{collection_name}/points/recommend")] async fn recommend_points( dispatcher: web::Data, collection: Path, @@ -42,7 +42,7 @@ async fn recommend_points( let pass = match check_strict_mode( &recommend_request, params.timeout_as_secs(), - &collection.name, + &collection.collection_name, &dispatcher, &auth, ) @@ -59,7 +59,7 @@ async fn recommend_points( let request_hw_counter = get_request_hardware_counter( &dispatcher, - collection.name.clone(), + collection.collection_name.clone(), service_config.hardware_reporting(), None, ); @@ -69,7 +69,7 @@ async fn recommend_points( let result = dispatcher .toc(&auth, &pass) .recommend( - &collection.name, + &collection.collection_name, recommend_request, params.consistency, shard_selection, @@ -121,7 +121,7 @@ async fn do_recommend_batch_points( .await } -#[post("/collections/{name}/points/recommend/batch")] +#[post("/collections/{collection_name}/points/recommend/batch")] async fn recommend_batch_points( dispatcher: web::Data, collection: Path, @@ -133,7 +133,7 @@ async fn recommend_batch_points( let pass = match check_strict_mode_batch( request.searches.iter().map(|i| &i.recommend_request), params.timeout_as_secs(), - &collection.name, + &collection.collection_name, &dispatcher, &auth, ) @@ -145,7 +145,7 @@ async fn recommend_batch_points( let request_hw_counter = get_request_hardware_counter( &dispatcher, - collection.name.clone(), + collection.collection_name.clone(), service_config.hardware_reporting(), None, ); @@ -153,7 +153,7 @@ async fn recommend_batch_points( let result = do_recommend_batch_points( dispatcher.toc(&auth, &pass), - &collection.name, + &collection.collection_name, request.into_inner(), params.consistency, auth, @@ -176,7 +176,7 @@ async fn recommend_batch_points( helpers::process_response(result, timing, request_hw_counter.to_rest_api()) } -#[post("/collections/{name}/points/recommend/groups")] +#[post("/collections/{collection_name}/points/recommend/groups")] async fn recommend_point_groups( dispatcher: web::Data, collection: Path, @@ -193,7 +193,7 @@ async fn recommend_point_groups( let pass = match check_strict_mode( &recommend_group_request, params.timeout_as_secs(), - &collection.name, + &collection.collection_name, &dispatcher, &auth, ) @@ -210,7 +210,7 @@ async fn recommend_point_groups( let request_hw_counter = get_request_hardware_counter( &dispatcher, - collection.name.clone(), + collection.collection_name.clone(), service_config.hardware_reporting(), None, ); @@ -218,7 +218,7 @@ async fn recommend_point_groups( let result = crate::common::query::do_recommend_point_groups( dispatcher.toc(&auth, &pass), - &collection.name, + &collection.collection_name, recommend_group_request, params.consistency, shard_selection, diff --git a/src/actix/api/retrieve_api.rs b/src/actix/api/retrieve_api.rs index adfe66a8af..41a1d96fdb 100644 --- a/src/actix/api/retrieve_api.rs +++ b/src/actix/api/retrieve_api.rs @@ -67,7 +67,7 @@ async fn do_get_point( .map(|points| points.into_iter().next()) } -#[get("/collections/{name}/points/{id}")] +#[get("/collections/{collection_name}/points/{id}")] async fn get_point( dispatcher: web::Data, collection: Path, @@ -78,7 +78,7 @@ async fn get_point( ) -> impl Responder { let pass = match check_strict_mode_timeout( params.timeout_as_secs(), - &collection.name, + &collection.collection_name, &dispatcher, &auth, ) @@ -97,7 +97,7 @@ async fn get_point( let request_hw_counter = get_request_hardware_counter( &dispatcher, - collection.name.clone(), + collection.collection_name.clone(), service_config.hardware_reporting(), None, ); @@ -105,7 +105,7 @@ async fn get_point( let res = do_get_point( dispatcher.toc(&auth, &pass), - &collection.name, + &collection.collection_name, point_id, params.consistency, params.timeout(), @@ -123,7 +123,7 @@ async fn get_point( process_response(res, timing, request_hw_counter.to_rest_api()) } -#[post("/collections/{name}/points")] +#[post("/collections/{collection_name}/points")] async fn get_points( dispatcher: web::Data, collection: Path, @@ -134,7 +134,7 @@ async fn get_points( ) -> impl Responder { let pass = match check_strict_mode_timeout( params.timeout_as_secs(), - &collection.name, + &collection.collection_name, &dispatcher, &auth, ) @@ -156,7 +156,7 @@ async fn get_points( let request_hw_counter = get_request_hardware_counter( &dispatcher, - collection.name.clone(), + collection.collection_name.clone(), service_config.hardware_reporting(), None, ); @@ -164,7 +164,7 @@ async fn get_points( let res = do_get_points( dispatcher.toc(&auth, &pass), - &collection.name, + &collection.collection_name, point_request, params.consistency, params.timeout(), @@ -183,7 +183,7 @@ async fn get_points( process_response(res, timing, request_hw_counter.to_rest_api()) } -#[post("/collections/{name}/points/scroll")] +#[post("/collections/{collection_name}/points/scroll")] async fn scroll_points( dispatcher: web::Data, collection: Path, @@ -200,7 +200,7 @@ async fn scroll_points( let pass = match check_strict_mode( &scroll_request, params.timeout_as_secs(), - &collection.name, + &collection.collection_name, &dispatcher, &auth, ) @@ -217,7 +217,7 @@ async fn scroll_points( let request_hw_counter = get_request_hardware_counter( &dispatcher, - collection.name.clone(), + collection.collection_name.clone(), service_config.hardware_reporting(), None, ); @@ -226,7 +226,7 @@ async fn scroll_points( let res = dispatcher .toc(&auth, &pass) .scroll( - &collection.name, + &collection.collection_name, scroll_request, params.consistency, params.timeout(), diff --git a/src/actix/api/search_api.rs b/src/actix/api/search_api.rs index 09310acf8f..1b011628c1 100644 --- a/src/actix/api/search_api.rs +++ b/src/actix/api/search_api.rs @@ -24,7 +24,7 @@ use crate::common::query::{ }; use crate::settings::ServiceConfig; -#[post("/collections/{name}/points/search")] +#[post("/collections/{collection_name}/points/search")] async fn search_points( dispatcher: web::Data, collection: Path, @@ -41,7 +41,7 @@ async fn search_points( let pass = match check_strict_mode( &search_request, params.timeout_as_secs(), - &collection.name, + &collection.collection_name, &dispatcher, &auth, ) @@ -58,7 +58,7 @@ async fn search_points( let request_hw_counter = get_request_hardware_counter( &dispatcher, - collection.name.clone(), + collection.collection_name.clone(), service_config.hardware_reporting(), None, ); @@ -67,7 +67,7 @@ async fn search_points( let result = do_core_search_points( dispatcher.toc(&auth, &pass), - &collection.name, + &collection.collection_name, search_request.into(), params.consistency, shard_selection, @@ -86,7 +86,7 @@ async fn search_points( process_response(result, timing, request_hw_counter.to_rest_api()) } -#[post("/collections/{name}/points/search/batch")] +#[post("/collections/{collection_name}/points/search/batch")] async fn batch_search_points( dispatcher: web::Data, collection: Path, @@ -117,7 +117,7 @@ async fn batch_search_points( let pass = match check_strict_mode_batch( requests.iter().map(|i| &i.0), params.timeout_as_secs(), - &collection.name, + &collection.collection_name, &dispatcher, &auth, ) @@ -129,7 +129,7 @@ async fn batch_search_points( let request_hw_counter = get_request_hardware_counter( &dispatcher, - collection.name.clone(), + collection.collection_name.clone(), service_config.hardware_reporting(), None, ); @@ -138,7 +138,7 @@ async fn batch_search_points( let result = do_search_batch_points( dispatcher.toc(&auth, &pass), - &collection.name, + &collection.collection_name, requests, params.consistency, auth, @@ -161,7 +161,7 @@ async fn batch_search_points( process_response(result, timing, request_hw_counter.to_rest_api()) } -#[post("/collections/{name}/points/search/groups")] +#[post("/collections/{collection_name}/points/search/groups")] async fn search_point_groups( dispatcher: web::Data, collection: Path, @@ -178,7 +178,7 @@ async fn search_point_groups( let pass = match check_strict_mode( &search_group_request, params.timeout_as_secs(), - &collection.name, + &collection.collection_name, &dispatcher, &auth, ) @@ -195,7 +195,7 @@ async fn search_point_groups( let request_hw_counter = get_request_hardware_counter( &dispatcher, - collection.name.clone(), + collection.collection_name.clone(), service_config.hardware_reporting(), None, ); @@ -203,7 +203,7 @@ async fn search_point_groups( let result = do_search_point_groups( dispatcher.toc(&auth, &pass), - &collection.name, + &collection.collection_name, search_group_request, params.consistency, shard_selection, @@ -216,7 +216,7 @@ async fn search_point_groups( process_response(result, timing, request_hw_counter.to_rest_api()) } -#[post("/collections/{name}/points/search/matrix/pairs")] +#[post("/collections/{collection_name}/points/search/matrix/pairs")] async fn search_points_matrix_pairs( dispatcher: web::Data, collection: Path, @@ -233,7 +233,7 @@ async fn search_points_matrix_pairs( let pass = match check_strict_mode( &search_request, params.timeout_as_secs(), - &collection.name, + &collection.collection_name, &dispatcher, &auth, ) @@ -250,7 +250,7 @@ async fn search_points_matrix_pairs( let request_hw_counter = get_request_hardware_counter( &dispatcher, - collection.name.clone(), + collection.collection_name.clone(), service_config.hardware_reporting(), None, ); @@ -258,7 +258,7 @@ async fn search_points_matrix_pairs( let response = do_search_points_matrix( dispatcher.toc(&auth, &pass), - &collection.name, + &collection.collection_name, CollectionSearchMatrixRequest::from(search_request), params.consistency, shard_selection, @@ -272,7 +272,7 @@ async fn search_points_matrix_pairs( process_response(response, timing, request_hw_counter.to_rest_api()) } -#[post("/collections/{name}/points/search/matrix/offsets")] +#[post("/collections/{collection_name}/points/search/matrix/offsets")] async fn search_points_matrix_offsets( dispatcher: web::Data, collection: Path, @@ -289,7 +289,7 @@ async fn search_points_matrix_offsets( let pass = match check_strict_mode( &search_request, params.timeout_as_secs(), - &collection.name, + &collection.collection_name, &dispatcher, &auth, ) @@ -306,7 +306,7 @@ async fn search_points_matrix_offsets( let request_hw_counter = get_request_hardware_counter( &dispatcher, - collection.name.clone(), + collection.collection_name.clone(), service_config.hardware_reporting(), None, ); @@ -314,7 +314,7 @@ async fn search_points_matrix_offsets( let response = do_search_points_matrix( dispatcher.toc(&auth, &pass), - &collection.name, + &collection.collection_name, CollectionSearchMatrixRequest::from(search_request), params.consistency, shard_selection, diff --git a/src/actix/api/service_api.rs b/src/actix/api/service_api.rs index 1937ec8732..4446818d06 100644 --- a/src/actix/api/service_api.rs +++ b/src/actix/api/service_api.rs @@ -33,6 +33,7 @@ use crate::tracing; pub struct TelemetryParam { pub anonymize: Option, pub details_level: Option, + pub per_collection: Option, #[validate(range(min = 1))] pub timeout: Option, } @@ -58,6 +59,7 @@ fn telemetry( let detail = TelemetryDetail { level: details_level, histograms: false, + per_collection: params.per_collection.unwrap_or(false), }; let telemetry_data = telemetry_collector .lock() @@ -76,6 +78,7 @@ fn telemetry( #[derive(Deserialize, Serialize, JsonSchema, Validate)] pub struct MetricsParam { pub anonymize: Option, + pub per_collection: Option, #[validate(range(min = 1))] pub timeout: Option, } @@ -101,6 +104,7 @@ async fn metrics( } let anonymize = params.anonymize.unwrap_or(false); + let per_collection = params.per_collection.unwrap_or(false); let telemetry_data = telemetry_collector .lock() .await @@ -109,6 +113,7 @@ async fn metrics( TelemetryDetail { level: DetailsLevel::Level4, histograms: true, + per_collection, }, None, params.timeout(), @@ -225,7 +230,7 @@ pub struct TruncateUnappliedWalParams { pub wait: Option, } -#[post("/collections/{name}/truncate_unapplied_wal")] +#[post("/collections/{collection_name}/truncate_unapplied_wal")] async fn truncate_unapplied_wal( dispatcher: web::Data, collection: Path, @@ -235,7 +240,7 @@ async fn truncate_unapplied_wal( let future = async move { let collection_pass = auth .check_global_access(AccessRequirements::new().manage(), "truncate_unapplied_wal")? - .issue_pass(&collection.name) + .issue_pass(&collection.collection_name) .into_static(); let pass = new_unchecked_verification_pass(); diff --git a/src/actix/api/shards_api.rs b/src/actix/api/shards_api.rs index 971e4ec0da..ab7b749bff 100644 --- a/src/actix/api/shards_api.rs +++ b/src/actix/api/shards_api.rs @@ -14,7 +14,7 @@ use crate::actix::auth::ActixAuth; use crate::actix::helpers::{self, process_response}; use crate::common::collections::{do_get_collection_shard_keys, do_update_collection_cluster}; -#[get("/collections/{name}/shards")] +#[get("/collections/{collection_name}/shards")] async fn list_shard_keys( dispatcher: web::Data, collection: Path, @@ -26,12 +26,12 @@ async fn list_shard_keys( helpers::time(do_get_collection_shard_keys( dispatcher.toc(&auth, &pass), &auth, - &collection.name, + &collection.collection_name, )) .await } -#[put("/collections/{name}/shards")] +#[put("/collections/{collection_name}/shards")] async fn create_shard_key( dispatcher: web::Data, collection: Path, @@ -51,7 +51,7 @@ async fn create_shard_key( let response = do_update_collection_cluster( &dispatcher, - collection.name.clone(), + collection.collection_name.clone(), operation, auth, wait_timeout, @@ -61,7 +61,7 @@ async fn create_shard_key( process_response(response, timing, None) } -#[post("/collections/{name}/shards/delete")] +#[post("/collections/{collection_name}/shards/delete")] async fn delete_shard_key( dispatcher: web::Data, collection: Path, @@ -81,7 +81,7 @@ async fn delete_shard_key( let response = do_update_collection_cluster( &dispatcher, - collection.name.clone(), + collection.collection_name.clone(), operation, auth, wait_timeout, diff --git a/src/actix/api/snapshot_api.rs b/src/actix/api/snapshot_api.rs index b186b3f6b9..142ca437ed 100644 --- a/src/actix/api/snapshot_api.rs +++ b/src/actix/api/snapshot_api.rs @@ -146,7 +146,7 @@ pub async fn do_get_snapshot( Ok(snapshot_stream) } -#[get("/collections/{name}/snapshots")] +#[get("/collections/{collection_name}/snapshots")] async fn list_snapshots( dispatcher: web::Data, path: web::Path, @@ -163,7 +163,7 @@ async fn list_snapshots( .await } -#[post("/collections/{name}/snapshots")] +#[post("/collections/{collection_name}/snapshots")] async fn create_snapshot( dispatcher: web::Data, path: web::Path, @@ -187,7 +187,7 @@ async fn create_snapshot( helpers::time_or_accept(future, params.wait.unwrap_or(true)).await } -#[post("/collections/{name}/snapshots/upload")] +#[post("/collections/{collection_name}/snapshots/upload")] async fn upload_snapshot( dispatcher: web::Data, http_client: web::Data, @@ -213,9 +213,12 @@ async fn upload_snapshot( } } - let snapshot_location = - do_save_uploaded_snapshot(dispatcher.toc(&auth, &pass), &collection.name, snapshot) - .await?; + let snapshot_location = do_save_uploaded_snapshot( + dispatcher.toc(&auth, &pass), + &collection.collection_name, + snapshot, + ) + .await?; // Snapshot is a local file, we do not need an API key for that let http_client = http_client.client(None)?; @@ -229,7 +232,7 @@ async fn upload_snapshot( do_recover_from_snapshot( dispatcher.get_ref(), - &collection.name, + &collection.collection_name, snapshot_recover, auth, http_client, @@ -240,7 +243,7 @@ async fn upload_snapshot( helpers::time_or_accept(future, wait.unwrap_or(true)).await } -#[put("/collections/{name}/snapshots/recover")] +#[put("/collections/{collection_name}/snapshots/recover")] async fn recover_from_snapshot( dispatcher: web::Data, http_client: web::Data, @@ -255,7 +258,7 @@ async fn recover_from_snapshot( do_recover_from_snapshot( dispatcher.get_ref(), - &collection.name, + &collection.collection_name, snapshot_recover, auth, http_client, @@ -266,7 +269,7 @@ async fn recover_from_snapshot( helpers::time_or_accept(future, params.wait.unwrap_or(true)).await } -#[get("/collections/{name}/snapshots/{snapshot_name}")] +#[get("/collections/{collection_name}/snapshots/{snapshot_name}")] async fn get_snapshot( dispatcher: web::Data, path: web::Path<(String, String)>, @@ -334,7 +337,7 @@ async fn delete_full_snapshot( helpers::time_or_accept(future, params.wait.unwrap_or(true)).await } -#[delete("/collections/{name}/snapshots/{snapshot_name}")] +#[delete("/collections/{collection_name}/snapshots/{snapshot_name}")] async fn delete_collection_snapshot( dispatcher: web::Data, path: web::Path<(String, String)>, @@ -351,7 +354,7 @@ async fn delete_collection_snapshot( helpers::time_or_accept(future, params.wait.unwrap_or(true)).await } -#[get("/collections/{collection}/shards/{shard}/snapshots")] +#[get("/collections/{collection_name}/shards/{shard}/snapshots")] async fn list_shard_snapshots( dispatcher: web::Data, path: web::Path<(String, ShardId)>, @@ -373,7 +376,7 @@ async fn list_shard_snapshots( helpers::time(future).await } -#[post("/collections/{collection}/shards/{shard}/snapshots")] +#[post("/collections/{collection_name}/shards/{shard}/snapshots")] async fn create_shard_snapshot( dispatcher: web::Data, path: web::Path<(String, ShardId)>, @@ -397,7 +400,7 @@ async fn create_shard_snapshot( helpers::time_or_accept(future, query.wait.unwrap_or(true)).await } -#[get("/collections/{collection}/shards/{shard}/snapshot")] +#[get("/collections/{collection_name}/shards/{shard}/snapshot")] async fn stream_shard_snapshot( dispatcher: web::Data, path: web::Path<(String, ShardId)>, @@ -418,7 +421,7 @@ async fn stream_shard_snapshot( } // TODO: `PUT` (same as `recover_from_snapshot`) or `POST`!? -#[put("/collections/{collection}/shards/{shard}/snapshots/recover")] +#[put("/collections/{collection_name}/shards/{shard}/snapshots/recover")] async fn recover_shard_snapshot( dispatcher: web::Data, http_client: web::Data, @@ -453,7 +456,7 @@ async fn recover_shard_snapshot( } // TODO: `POST` (same as `upload_snapshot`) or `PUT`!? -#[post("/collections/{collection}/shards/{shard}/snapshots/upload")] +#[post("/collections/{collection_name}/shards/{shard}/snapshots/upload")] async fn upload_shard_snapshot( dispatcher: web::Data, path: web::Path<(String, ShardId)>, @@ -521,7 +524,7 @@ async fn upload_shard_snapshot( helpers::time_or_accept(future, wait.unwrap_or(true)).await } -#[get("/collections/{collection}/shards/{shard}/snapshots/{snapshot}")] +#[get("/collections/{collection_name}/shards/{shard}/snapshots/{snapshot}")] async fn download_shard_snapshot( dispatcher: web::Data, path: web::Path<(String, ShardId, String)>, @@ -553,7 +556,7 @@ async fn download_shard_snapshot( Ok(snapshot_stream) } -#[delete("/collections/{collection}/shards/{shard}/snapshots/{snapshot}")] +#[delete("/collections/{collection_name}/shards/{shard}/snapshots/{snapshot}")] async fn delete_shard_snapshot( dispatcher: web::Data, path: web::Path<(String, ShardId, String)>, @@ -579,7 +582,7 @@ async fn delete_shard_snapshot( helpers::time_or_accept(future, query.wait.unwrap_or(true)).await } -#[post("/collections/{collection}/shards/{shard}/snapshot/partial/create")] +#[post("/collections/{collection_name}/shards/{shard}/snapshot/partial/create")] async fn create_partial_snapshot( dispatcher: web::Data, path: web::Path<(String, ShardId)>, @@ -604,7 +607,7 @@ async fn create_partial_snapshot( Ok(snapshot_stream) } -#[post("/collections/{collection}/shards/{shard}/snapshot/partial/recover")] +#[post("/collections/{collection_name}/shards/{shard}/snapshot/partial/recover")] async fn recover_partial_snapshot( dispatcher: web::Data, path: web::Path<(String, ShardId)>, @@ -696,7 +699,7 @@ pub struct PartialSnapshotRecoverFrom { api_key: Option, } -#[post("/collections/{collection}/shards/{shard}/snapshot/partial/recover_from")] +#[post("/collections/{collection_name}/shards/{shard}/snapshot/partial/recover_from")] async fn recover_partial_snapshot_from( dispatcher: web::Data, http_client: web::Data, @@ -853,7 +856,7 @@ async fn recover_partial_snapshot_from( helpers::time_or_accept(future, wait.unwrap_or(true)).await } -#[get("/collections/{collection}/shards/{shard}/snapshot/partial/manifest")] +#[get("/collections/{collection_name}/shards/{shard}/snapshot/partial/manifest")] async fn get_partial_snapshot_manifest( dispatcher: web::Data, path: web::Path<(String, ShardId)>, diff --git a/src/actix/api/update_api.rs b/src/actix/api/update_api.rs index ed01abf9c6..71ed1360aa 100644 --- a/src/actix/api/update_api.rs +++ b/src/actix/api/update_api.rs @@ -29,7 +29,7 @@ struct FieldPath { name: JsonPath, } -#[put("/collections/{name}/points")] +#[put("/collections/{collection_name}/points")] #[allow(clippy::too_many_arguments)] async fn upsert_points( dispatcher: web::Data, @@ -44,7 +44,7 @@ async fn upsert_points( let request_hw_counter = get_request_hardware_counter( &dispatcher, - collection.name.clone(), + collection.collection_name.clone(), service_config.hardware_reporting(), Some(params.wait), ); @@ -54,7 +54,7 @@ async fn upsert_points( let result_with_usage = do_upsert_points( StrictModeCheckedTocProvider::new(&dispatcher), - collection.into_inner().name, + collection.into_inner().collection_name, operation, InternalUpdateParams::default(), params.into_inner(), @@ -77,7 +77,7 @@ async fn upsert_points( ) } -#[post("/collections/{name}/points/delete")] +#[post("/collections/{collection_name}/points/delete")] async fn delete_points( dispatcher: web::Data, collection: Path, @@ -90,7 +90,7 @@ async fn delete_points( let request_hw_counter = get_request_hardware_counter( &dispatcher, - collection.name.clone(), + collection.collection_name.clone(), service_config.hardware_reporting(), Some(params.wait), ); @@ -98,7 +98,7 @@ async fn delete_points( let res = do_delete_points( StrictModeCheckedTocProvider::new(&dispatcher), - collection.into_inner().name, + collection.into_inner().collection_name, operation, InternalUpdateParams::default(), params.into_inner(), @@ -110,7 +110,7 @@ async fn delete_points( process_response(res, timing, request_hw_counter.to_rest_api()) } -#[put("/collections/{name}/points/vectors")] +#[put("/collections/{collection_name}/points/vectors")] #[allow(clippy::too_many_arguments)] async fn update_vectors( dispatcher: web::Data, @@ -125,7 +125,7 @@ async fn update_vectors( let request_hw_counter = get_request_hardware_counter( &dispatcher, - collection.name.clone(), + collection.collection_name.clone(), service_config.hardware_reporting(), Some(params.wait), ); @@ -135,7 +135,7 @@ async fn update_vectors( let res = do_update_vectors( StrictModeCheckedTocProvider::new(&dispatcher), - collection.into_inner().name, + collection.into_inner().collection_name, operation, InternalUpdateParams::default(), params.into_inner(), @@ -158,7 +158,7 @@ async fn update_vectors( ) } -#[post("/collections/{name}/points/vectors/delete")] +#[post("/collections/{collection_name}/points/vectors/delete")] async fn delete_vectors( dispatcher: web::Data, collection: Path, @@ -171,7 +171,7 @@ async fn delete_vectors( let request_hw_counter = get_request_hardware_counter( &dispatcher, - collection.name.clone(), + collection.collection_name.clone(), service_config.hardware_reporting(), Some(params.wait), ); @@ -179,7 +179,7 @@ async fn delete_vectors( let response = do_delete_vectors( StrictModeCheckedTocProvider::new(&dispatcher), - collection.into_inner().name, + collection.into_inner().collection_name, operation, InternalUpdateParams::default(), params.into_inner(), @@ -191,7 +191,7 @@ async fn delete_vectors( process_response(response, timing, request_hw_counter.to_rest_api()) } -#[post("/collections/{name}/points/payload")] +#[post("/collections/{collection_name}/points/payload")] async fn set_payload( dispatcher: web::Data, collection: Path, @@ -204,7 +204,7 @@ async fn set_payload( let request_hw_counter = get_request_hardware_counter( &dispatcher, - collection.name.clone(), + collection.collection_name.clone(), service_config.hardware_reporting(), Some(params.wait), ); @@ -212,7 +212,7 @@ async fn set_payload( let res = do_set_payload( StrictModeCheckedTocProvider::new(&dispatcher), - collection.into_inner().name, + collection.into_inner().collection_name, operation, InternalUpdateParams::default(), params.into_inner(), @@ -224,7 +224,7 @@ async fn set_payload( process_response(res, timing, request_hw_counter.to_rest_api()) } -#[put("/collections/{name}/points/payload")] +#[put("/collections/{collection_name}/points/payload")] async fn overwrite_payload( dispatcher: web::Data, collection: Path, @@ -237,7 +237,7 @@ async fn overwrite_payload( let request_hw_counter = get_request_hardware_counter( &dispatcher, - collection.name.clone(), + collection.collection_name.clone(), service_config.hardware_reporting(), Some(params.wait), ); @@ -245,7 +245,7 @@ async fn overwrite_payload( let res = do_overwrite_payload( StrictModeCheckedTocProvider::new(&dispatcher), - collection.into_inner().name, + collection.into_inner().collection_name, operation, InternalUpdateParams::default(), params.into_inner(), @@ -257,7 +257,7 @@ async fn overwrite_payload( process_response(res, timing, request_hw_counter.to_rest_api()) } -#[post("/collections/{name}/points/payload/delete")] +#[post("/collections/{collection_name}/points/payload/delete")] async fn delete_payload( dispatcher: web::Data, collection: Path, @@ -270,7 +270,7 @@ async fn delete_payload( let request_hw_counter = get_request_hardware_counter( &dispatcher, - collection.name.clone(), + collection.collection_name.clone(), service_config.hardware_reporting(), Some(params.wait), ); @@ -278,7 +278,7 @@ async fn delete_payload( let res = do_delete_payload( StrictModeCheckedTocProvider::new(&dispatcher), - collection.into_inner().name, + collection.into_inner().collection_name, operation, InternalUpdateParams::default(), params.into_inner(), @@ -290,7 +290,7 @@ async fn delete_payload( process_response(res, timing, request_hw_counter.to_rest_api()) } -#[post("/collections/{name}/points/payload/clear")] +#[post("/collections/{collection_name}/points/payload/clear")] async fn clear_payload( dispatcher: web::Data, collection: Path, @@ -303,7 +303,7 @@ async fn clear_payload( let request_hw_counter = get_request_hardware_counter( &dispatcher, - collection.name.clone(), + collection.collection_name.clone(), service_config.hardware_reporting(), Some(params.wait), ); @@ -311,7 +311,7 @@ async fn clear_payload( let res = do_clear_payload( StrictModeCheckedTocProvider::new(&dispatcher), - collection.into_inner().name, + collection.into_inner().collection_name, operation, InternalUpdateParams::default(), params.into_inner(), @@ -324,7 +324,7 @@ async fn clear_payload( } #[allow(clippy::too_many_arguments)] -#[post("/collections/{name}/points/batch")] +#[post("/collections/{collection_name}/points/batch")] async fn update_batch( dispatcher: web::Data, collection: Path, @@ -338,7 +338,7 @@ async fn update_batch( let request_hw_counter = get_request_hardware_counter( &dispatcher, - collection.name.clone(), + collection.collection_name.clone(), service_config.hardware_reporting(), Some(params.wait), ); @@ -348,7 +348,7 @@ async fn update_batch( let result_with_usage = do_batch_update_points( StrictModeCheckedTocProvider::new(&dispatcher), - collection.into_inner().name, + collection.into_inner().collection_name, operations.operations, InternalUpdateParams::default(), params.into_inner(), @@ -371,7 +371,7 @@ async fn update_batch( ) } -#[put("/collections/{name}/index")] +#[put("/collections/{collection_name}/index")] async fn create_field_index( dispatcher: web::Data, collection: Path, @@ -385,14 +385,14 @@ async fn create_field_index( let request_hw_counter = get_request_hardware_counter( &dispatcher, - collection.name.clone(), + collection.collection_name.clone(), service_config.hardware_reporting(), Some(params.wait), ); let response = do_create_index( dispatcher.into_inner(), - collection.into_inner().name, + collection.into_inner().collection_name, operation, InternalUpdateParams::default(), params.into_inner(), @@ -407,7 +407,7 @@ async fn create_field_index( ) } -#[delete("/collections/{name}/index/{field_name}")] +#[delete("/collections/{collection_name}/index/{field_name}")] async fn delete_field_index( dispatcher: web::Data, collection: Path, @@ -419,7 +419,7 @@ async fn delete_field_index( let response = do_delete_index( dispatcher.into_inner(), - collection.into_inner().name, + collection.into_inner().collection_name, field.name.clone(), InternalUpdateParams::default(), params.into_inner(), @@ -434,7 +434,7 @@ async fn delete_field_index( /// Accepts any staging operation and executes it on the collection. /// Only available when the `staging` feature is enabled. #[cfg(feature = "staging")] -#[post("/collections/{name}/debug")] +#[post("/collections/{collection_name}/debug")] async fn staging_operation( dispatcher: web::Data, collection: Path, @@ -447,7 +447,7 @@ async fn staging_operation( let timing = Instant::now(); let operation = operation.into_inner(); - let collection_name = collection.into_inner().name; + let collection_name = collection.into_inner().collection_name; let collection_operation = CollectionUpdateOperations::from(operation); diff --git a/src/common/metrics.rs b/src/common/metrics.rs index 8383b0dd0d..1efde1824d 100644 --- a/src/common/metrics.rs +++ b/src/common/metrics.rs @@ -27,31 +27,31 @@ use crate::common::telemetry_ops::requests_telemetry::{ /// /// This array *must* be sorted. const REST_ENDPOINT_WHITELIST: &[&str] = &[ - "/collections/{name}/index", - "/collections/{name}/points", - "/collections/{name}/points/batch", - "/collections/{name}/points/count", - "/collections/{name}/points/delete", - "/collections/{name}/points/discover", - "/collections/{name}/points/discover/batch", - "/collections/{name}/points/facet", - "/collections/{name}/points/payload", - "/collections/{name}/points/payload/clear", - "/collections/{name}/points/payload/delete", - "/collections/{name}/points/query", - "/collections/{name}/points/query/batch", - "/collections/{name}/points/query/groups", - "/collections/{name}/points/recommend", - "/collections/{name}/points/recommend/batch", - "/collections/{name}/points/recommend/groups", - "/collections/{name}/points/scroll", - "/collections/{name}/points/search", - "/collections/{name}/points/search/batch", - "/collections/{name}/points/search/groups", - "/collections/{name}/points/search/matrix/offsets", - "/collections/{name}/points/search/matrix/pairs", - "/collections/{name}/points/vectors", - "/collections/{name}/points/vectors/delete", + "/collections/{collection_name}/index", + "/collections/{collection_name}/points", + "/collections/{collection_name}/points/batch", + "/collections/{collection_name}/points/count", + "/collections/{collection_name}/points/delete", + "/collections/{collection_name}/points/discover", + "/collections/{collection_name}/points/discover/batch", + "/collections/{collection_name}/points/facet", + "/collections/{collection_name}/points/payload", + "/collections/{collection_name}/points/payload/clear", + "/collections/{collection_name}/points/payload/delete", + "/collections/{collection_name}/points/query", + "/collections/{collection_name}/points/query/batch", + "/collections/{collection_name}/points/query/groups", + "/collections/{collection_name}/points/recommend", + "/collections/{collection_name}/points/recommend/batch", + "/collections/{collection_name}/points/recommend/groups", + "/collections/{collection_name}/points/scroll", + "/collections/{collection_name}/points/search", + "/collections/{collection_name}/points/search/batch", + "/collections/{collection_name}/points/search/groups", + "/collections/{collection_name}/points/search/matrix/offsets", + "/collections/{collection_name}/points/search/matrix/pairs", + "/collections/{collection_name}/points/vectors", + "/collections/{collection_name}/points/vectors/delete", ]; /// Whitelist for GRPC endpoints in metrics output. @@ -662,54 +662,113 @@ impl MetricsProvider for RequestsTelemetry { impl MetricsProvider for WebApiTelemetry { fn add_metrics(&self, metrics: &mut MetricsData, prefix: Option<&str>) { - let mut builder = OperationDurationMetricsBuilder::default(); - for (endpoint, responses) in &self.responses { - let Some((method, endpoint)) = endpoint.split_once(' ') else { - continue; - }; - // Endpoint must be whitelisted - if REST_ENDPOINT_WHITELIST.binary_search(&endpoint).is_err() { - continue; + // Mode decision: when `per_collection_responses` is populated (i.e. per_collection + // was requested via query parameter), we render per-collection metrics with a + // `collection` label and skip global ones. + if self.per_collection_responses.is_empty() { + // Global mode: render global metrics as before + let mut builder = OperationDurationMetricsBuilder::default(); + for (endpoint, responses) in &self.responses { + let Some((method, endpoint)) = endpoint.split_once(' ') else { + continue; + }; + if REST_ENDPOINT_WHITELIST.binary_search(&endpoint).is_err() { + continue; + } + for (status, stats) in responses { + builder.add( + stats, + &[ + ("method", method), + ("endpoint", endpoint), + ("status", &status.to_string()), + ], + *status == REST_TIMINGS_FOR_STATUS, + ); + } } - for (status, stats) in responses { - builder.add( - stats, - &[ - ("method", method), - ("endpoint", endpoint), - ("status", &status.to_string()), - ], - *status == REST_TIMINGS_FOR_STATUS, - ); + builder.build(prefix, "rest", metrics); + } else { + // Per-collection mode: render per-collection metrics with `collection` label + let mut builder = OperationDurationMetricsBuilder::default(); + for (collection, methods) in &self.per_collection_responses { + for (endpoint, responses) in methods { + let Some((method, endpoint)) = endpoint.split_once(' ') else { + continue; + }; + if REST_ENDPOINT_WHITELIST.binary_search(&endpoint).is_err() { + continue; + } + for (status, stats) in responses { + builder.add( + stats, + &[ + ("method", method), + ("endpoint", endpoint), + ("status", &status.to_string()), + ("collection", collection), + ], + *status == REST_TIMINGS_FOR_STATUS, + ); + } + } } + builder.build(prefix, "rest", metrics); } - builder.build(prefix, "rest", metrics); } } impl MetricsProvider for GrpcTelemetry { fn add_metrics(&self, metrics: &mut MetricsData, prefix: Option<&str>) { - let mut builder = OperationDurationMetricsBuilder::default(); - for (endpoint, responses) in &self.responses { - // Endpoint must be whitelisted - if GRPC_ENDPOINT_WHITELIST - .binary_search(&endpoint.as_str()) - .is_err() - { - continue; + // Same mode-switching logic as WebApiTelemetry::add_metrics — see comment there. + if self.per_collection_responses.is_empty() { + // Global mode: render global metrics as before + let mut builder = OperationDurationMetricsBuilder::default(); + for (endpoint, responses) in &self.responses { + if GRPC_ENDPOINT_WHITELIST + .binary_search(&endpoint.as_str()) + .is_err() + { + continue; + } + for (status, stats) in responses { + builder.add( + stats, + &[ + ("endpoint", endpoint.as_str()), + ("status", &status.to_string()), + ], + true, + ); + } } - for (status, stats) in responses { - builder.add( - stats, - &[ - ("endpoint", endpoint.as_str()), - ("status", &status.to_string()), - ], - true, - ); + builder.build(prefix, "grpc", metrics); + } else { + // Per-collection mode: render per-collection metrics with `collection` label + let mut builder = OperationDurationMetricsBuilder::default(); + for (collection, methods) in &self.per_collection_responses { + for (endpoint, responses) in methods { + if GRPC_ENDPOINT_WHITELIST + .binary_search(&endpoint.as_str()) + .is_err() + { + continue; + } + for (status, stats) in responses { + builder.add( + stats, + &[ + ("endpoint", endpoint.as_str()), + ("status", &status.to_string()), + ("collection", collection), + ], + true, + ); + } + } } + builder.build(prefix, "grpc", metrics); } - builder.build(prefix, "grpc", metrics); } } @@ -1310,4 +1369,185 @@ mod tests { "GRPC_ENDPOINT_WHITELIST must be sorted in code to allow binary search", ); } + + #[test] + fn test_rest_whitelist_uses_collection_name_param() { + use super::REST_ENDPOINT_WHITELIST; + + for endpoint in REST_ENDPOINT_WHITELIST { + assert!( + endpoint.contains("{collection_name}"), + "REST_ENDPOINT_WHITELIST entry `{endpoint}` must use \ + `{{collection_name}}` as the collection path parameter", + ); + } + } + + #[test] + fn test_rest_metrics_global_mode() { + use std::collections::HashMap; + + use segment::common::operation_time_statistics::OperationDurationStatistics; + + use super::{MetricsData, MetricsProvider, WebApiTelemetry}; + + let mut responses = HashMap::new(); + let mut status_map = HashMap::new(); + status_map.insert( + 200u16, + OperationDurationStatistics { + count: 10, + ..Default::default() + }, + ); + responses.insert( + "POST /collections/{collection_name}/points/search".to_string(), + status_map, + ); + + let telemetry = WebApiTelemetry { + responses, + per_collection_responses: HashMap::new(), + }; + + let mut metrics = MetricsData::empty(); + telemetry.add_metrics(&mut metrics, None); + let output = metrics.format_metrics(); + + // Should contain global metrics without collection label + assert!(output.contains("rest_responses_total")); + assert!(!output.contains("collection=")); + } + + #[test] + fn test_rest_metrics_per_collection_mode() { + use std::collections::HashMap; + + use segment::common::operation_time_statistics::OperationDurationStatistics; + + use super::{MetricsData, MetricsProvider, WebApiTelemetry}; + + // Global responses present but per_collection too + let mut responses = HashMap::new(); + let mut status_map = HashMap::new(); + status_map.insert( + 200u16, + OperationDurationStatistics { + count: 10, + ..Default::default() + }, + ); + responses.insert( + "POST /collections/{collection_name}/points/search".to_string(), + status_map, + ); + + let mut per_collection = HashMap::new(); + let mut methods = HashMap::new(); + let mut col_status_map = HashMap::new(); + col_status_map.insert( + 200u16, + OperationDurationStatistics { + count: 5, + ..Default::default() + }, + ); + methods.insert( + "POST /collections/{collection_name}/points/search".to_string(), + col_status_map, + ); + per_collection.insert("my_collection".to_string(), methods); + + let telemetry = WebApiTelemetry { + responses, + per_collection_responses: per_collection, + }; + + let mut metrics = MetricsData::empty(); + telemetry.add_metrics(&mut metrics, None); + let output = metrics.format_metrics(); + + // Should contain collection label + assert!( + output.contains("collection=\"my_collection\""), + "Expected collection label in output:\n{output}" + ); + // Should still have rest_ prefix metrics + assert!(output.contains("rest_responses_total")); + } + + #[test] + fn test_grpc_metrics_per_collection_mode() { + use std::collections::HashMap; + + use segment::common::operation_time_statistics::OperationDurationStatistics; + + use super::{GrpcTelemetry, MetricsData, MetricsProvider}; + + let mut per_collection = HashMap::new(); + let mut methods = HashMap::new(); + let mut status_map = HashMap::new(); + status_map.insert( + 0i32, + OperationDurationStatistics { + count: 7, + ..Default::default() + }, + ); + methods.insert("/qdrant.Points/Search".to_string(), status_map); + per_collection.insert("test_col".to_string(), methods); + + let telemetry = GrpcTelemetry { + responses: HashMap::new(), + per_collection_responses: per_collection, + }; + + let mut metrics = MetricsData::empty(); + telemetry.add_metrics(&mut metrics, None); + let output = metrics.format_metrics(); + + assert!( + output.contains("collection=\"test_col\""), + "Expected collection label in output:\n{output}" + ); + assert!(output.contains("grpc_responses_total")); + } + + #[test] + fn test_per_collection_skips_non_whitelisted() { + use std::collections::HashMap; + + use segment::common::operation_time_statistics::OperationDurationStatistics; + + use super::{MetricsData, MetricsProvider, WebApiTelemetry}; + + let mut per_collection = HashMap::new(); + let mut methods = HashMap::new(); + let mut status_map = HashMap::new(); + status_map.insert( + 200u16, + OperationDurationStatistics { + count: 3, + ..Default::default() + }, + ); + // This endpoint is NOT in the whitelist + methods.insert("GET /collections".to_string(), status_map); + per_collection.insert("col".to_string(), methods); + + let telemetry = WebApiTelemetry { + responses: HashMap::new(), + per_collection_responses: per_collection, + }; + + let mut metrics = MetricsData::empty(); + telemetry.add_metrics(&mut metrics, None); + let output = metrics.format_metrics(); + + // Non-whitelisted endpoints should not appear + assert!( + !output.contains("collection=\"col\""), + "Non-whitelisted endpoint should not appear:\n{output}" + ); + } } diff --git a/src/common/telemetry_ops/requests_telemetry.rs b/src/common/telemetry_ops/requests_telemetry.rs index 9b2e520728..ded7b2a95c 100644 --- a/src/common/telemetry_ops/requests_telemetry.rs +++ b/src/common/telemetry_ops/requests_telemetry.rs @@ -11,18 +11,33 @@ use segment::common::operation_time_statistics::{ use serde::Serialize; use storage::rbac::{AccessRequirements, Auth}; +/// Wrapper for passing collection name through gRPC response extensions. +#[derive(Clone, Debug)] +pub struct CollectionName(pub String); + pub type HttpStatusCode = u16; pub type GrpcStatusCode = i32; +/// Per-collection key: collection_name -> (method -> status_code -> stats) +pub type PerCollectionResponses = + HashMap>>; + +type PerCollectionAggregators = + HashMap>>>>; + #[derive(Serialize, Clone, Default, Debug, JsonSchema)] pub struct WebApiTelemetry { pub responses: HashMap>, + #[serde(skip_serializing_if = "HashMap::is_empty", default)] + pub per_collection_responses: PerCollectionResponses, } #[derive(Serialize, Clone, Default, Debug, JsonSchema)] pub struct GrpcTelemetry { pub responses: HashMap>, + #[serde(skip_serializing_if = "HashMap::is_empty", default)] + pub per_collection_responses: PerCollectionResponses, } pub struct ActixTelemetryCollector { @@ -32,6 +47,8 @@ pub struct ActixTelemetryCollector { #[derive(Default)] pub struct ActixWorkerTelemetryCollector { methods: HashMap>>>, + /// collection_name -> method -> status_code -> aggregator + per_collection_methods: PerCollectionAggregators, } pub struct TonicTelemetryCollector { @@ -41,11 +58,13 @@ pub struct TonicTelemetryCollector { #[derive(Default)] pub struct TonicWorkerTelemetryCollector { methods: HashMap>>>, + /// collection_name -> method -> status_code -> aggregator + per_collection_methods: PerCollectionAggregators, } impl ActixTelemetryCollector { pub fn create_web_worker_telemetry(&mut self) -> Arc> { - let worker: Arc> = Default::default(); + let worker = Arc::new(Mutex::new(ActixWorkerTelemetryCollector::default())); self.workers.push(worker.clone()); worker } @@ -62,7 +81,7 @@ impl ActixTelemetryCollector { impl TonicTelemetryCollector { pub fn create_grpc_telemetry_collector(&mut self) -> Arc> { - let worker: Arc> = Default::default(); + let worker = Arc::new(Mutex::new(TonicWorkerTelemetryCollector::default())); self.workers.push(worker.clone()); worker } @@ -83,14 +102,27 @@ impl TonicWorkerTelemetryCollector { method: String, instant: std::time::Instant, status_code: GrpcStatusCode, + collection_name: Option, ) { let aggregator = self .methods - .entry(method) + .entry(method.clone()) .or_default() .entry(status_code) .or_insert_with(OperationDurationsAggregator::new); ScopeDurationMeasurer::new_with_instant(aggregator, instant); + + if let Some(collection) = collection_name { + let aggregator = self + .per_collection_methods + .entry(collection) + .or_default() + .entry(method) + .or_default() + .entry(status_code) + .or_insert_with(OperationDurationsAggregator::new); + ScopeDurationMeasurer::new_with_instant(aggregator, instant); + } } pub fn get_telemetry_data(&self, detail: TelemetryDetail) -> GrpcTelemetry { @@ -102,7 +134,30 @@ impl TonicWorkerTelemetryCollector { } responses.insert(method.clone(), status_codes_map); } - GrpcTelemetry { responses } + + let per_collection_responses = if detail.per_collection { + let mut per_collection_responses = HashMap::new(); + for (collection, methods) in &self.per_collection_methods { + let mut methods_map = HashMap::new(); + for (method, status_codes) in methods { + let mut status_codes_map = HashMap::new(); + for (status_code, aggregator) in status_codes { + status_codes_map + .insert(*status_code, aggregator.lock().get_statistics(detail)); + } + methods_map.insert(method.clone(), status_codes_map); + } + per_collection_responses.insert(collection.clone(), methods_map); + } + per_collection_responses + } else { + HashMap::new() + }; + + GrpcTelemetry { + responses, + per_collection_responses, + } } } @@ -112,14 +167,27 @@ impl ActixWorkerTelemetryCollector { method: String, status_code: HttpStatusCode, instant: std::time::Instant, + collection_name: Option, ) { let aggregator = self .methods - .entry(method) + .entry(method.clone()) .or_default() .entry(status_code) .or_insert_with(OperationDurationsAggregator::new); ScopeDurationMeasurer::new_with_instant(aggregator, instant); + + if let Some(collection) = collection_name { + let aggregator = self + .per_collection_methods + .entry(collection) + .or_default() + .entry(method) + .or_default() + .entry(status_code) + .or_insert_with(OperationDurationsAggregator::new); + ScopeDurationMeasurer::new_with_instant(aggregator, instant); + } } pub fn get_telemetry_data(&self, detail: TelemetryDetail) -> WebApiTelemetry { @@ -131,7 +199,30 @@ impl ActixWorkerTelemetryCollector { } responses.insert(method.clone(), status_codes_map); } - WebApiTelemetry { responses } + + let per_collection_responses = if detail.per_collection { + let mut per_collection_responses = HashMap::new(); + for (collection, methods) in &self.per_collection_methods { + let mut methods_map = HashMap::new(); + for (method, status_codes) in methods { + let mut status_codes_map = HashMap::new(); + for (status_code, aggregator) in status_codes { + status_codes_map + .insert(*status_code, aggregator.lock().get_statistics(detail)); + } + methods_map.insert(method.clone(), status_codes_map); + } + per_collection_responses.insert(collection.clone(), methods_map); + } + per_collection_responses + } else { + HashMap::new() + }; + + WebApiTelemetry { + responses, + per_collection_responses, + } } } @@ -144,6 +235,19 @@ impl GrpcTelemetry { *status_entry = status_entry.clone() + statistics.clone(); } } + for (collection, methods) in &other.per_collection_responses { + let col_entry = self + .per_collection_responses + .entry(collection.clone()) + .or_default(); + for (method, status_codes) in methods { + let method_entry = col_entry.entry(method.clone()).or_default(); + for (status_code, statistics) in status_codes { + let entry = method_entry.entry(*status_code).or_default(); + *entry = entry.clone() + statistics.clone(); + } + } + } } } @@ -156,6 +260,19 @@ impl WebApiTelemetry { *entry = entry.clone() + statistics.clone(); } } + for (collection, methods) in &other.per_collection_responses { + let col_entry = self + .per_collection_responses + .entry(collection.clone()) + .or_default(); + for (method, status_codes) in methods { + let method_entry = col_entry.entry(method.clone()).or_default(); + for (status_code, statistics) in status_codes { + let entry = method_entry.entry(*status_code).or_default(); + *entry = entry.clone() + statistics.clone(); + } + } + } } } @@ -194,7 +311,12 @@ impl Anonymize for WebApiTelemetry { .map(|(key, value)| (key.clone(), anonymize_collection_values(value))) .collect(); - WebApiTelemetry { responses } + // Anonymize per-collection: keep structure but anonymize the stats, not collection names + // (collection names are user data, so strip them on anonymize) + WebApiTelemetry { + responses, + per_collection_responses: HashMap::new(), + } } } @@ -206,6 +328,232 @@ impl Anonymize for GrpcTelemetry { .map(|(key, value)| (key.clone(), anonymize_collection_values(value))) .collect(); - GrpcTelemetry { responses } + GrpcTelemetry { + responses, + per_collection_responses: HashMap::new(), + } + } +} + +#[cfg(test)] +mod tests { + use common::types::{DetailsLevel, TelemetryDetail}; + use segment::common::anonymize::Anonymize; + + use super::*; + + fn detail_with_per_collection() -> TelemetryDetail { + TelemetryDetail { + level: DetailsLevel::Level0, + histograms: false, + per_collection: true, + } + } + + // Helper: actix add_response signature is (method, status_code, instant, collection) + fn actix_add( + worker: &mut ActixWorkerTelemetryCollector, + method: &str, + status: u16, + collection: Option<&str>, + ) { + worker.add_response( + method.into(), + status, + std::time::Instant::now(), + collection.map(String::from), + ); + } + + #[test] + fn test_actix_add_response_without_collection() { + let mut worker = ActixWorkerTelemetryCollector::default(); + actix_add(&mut worker, "GET /collections/{name}/points", 200, None); + + let telemetry = worker.get_telemetry_data(detail_with_per_collection()); + assert_eq!(telemetry.responses.len(), 1); + assert!(telemetry.per_collection_responses.is_empty()); + } + + #[test] + fn test_actix_add_response_with_collection() { + let mut worker = ActixWorkerTelemetryCollector::default(); + actix_add( + &mut worker, + "POST /collections/{name}/points/search", + 200, + Some("my_collection"), + ); + + let telemetry = worker.get_telemetry_data(detail_with_per_collection()); + + // Global should still be populated + assert_eq!(telemetry.responses.len(), 1); + assert!( + telemetry + .responses + .contains_key("POST /collections/{name}/points/search") + ); + + // Per-collection should also be populated + assert_eq!(telemetry.per_collection_responses.len(), 1); + let collection_methods = &telemetry.per_collection_responses["my_collection"]; + assert!(collection_methods.contains_key("POST /collections/{name}/points/search")); + } + + #[test] + fn test_actix_per_collection_hidden_by_default() { + let mut worker = ActixWorkerTelemetryCollector::default(); + actix_add( + &mut worker, + "POST /collections/{name}/points/search", + 200, + Some("my_collection"), + ); + + // Default detail has per_collection=false, so per_collection_responses should be empty + let telemetry = worker.get_telemetry_data(TelemetryDetail::default()); + assert_eq!(telemetry.responses.len(), 1); + assert!(telemetry.per_collection_responses.is_empty()); + } + + #[test] + fn test_actix_multiple_collections() { + let mut worker = ActixWorkerTelemetryCollector::default(); + let method = "POST /collections/{name}/points/search"; + + actix_add(&mut worker, method, 200, Some("col_a")); + actix_add(&mut worker, method, 200, Some("col_b")); + actix_add(&mut worker, method, 200, Some("col_a")); + + let telemetry = worker.get_telemetry_data(detail_with_per_collection()); + + // Global: 3 total requests + let global_stats = &telemetry.responses[method][&200]; + assert_eq!(global_stats.count, 3); + + // Per-collection: col_a=2, col_b=1 + assert_eq!(telemetry.per_collection_responses.len(), 2); + assert_eq!( + telemetry.per_collection_responses["col_a"][method][&200].count, + 2 + ); + assert_eq!( + telemetry.per_collection_responses["col_b"][method][&200].count, + 1 + ); + } + + #[test] + fn test_tonic_add_response_with_collection() { + let mut worker = TonicWorkerTelemetryCollector::default(); + worker.add_response( + "/qdrant.Points/Search".into(), + std::time::Instant::now(), + 0, + Some("my_collection".into()), + ); + + let telemetry = worker.get_telemetry_data(detail_with_per_collection()); + + assert_eq!(telemetry.responses.len(), 1); + assert_eq!(telemetry.per_collection_responses.len(), 1); + assert!( + telemetry + .per_collection_responses + .contains_key("my_collection") + ); + } + + #[test] + fn test_merge_per_collection_responses() { + let mut worker1 = ActixWorkerTelemetryCollector::default(); + let mut worker2 = ActixWorkerTelemetryCollector::default(); + let method = "POST /collections/{name}/points/search"; + + actix_add(&mut worker1, method, 200, Some("col_a")); + actix_add(&mut worker2, method, 200, Some("col_a")); + actix_add(&mut worker2, method, 200, Some("col_b")); + + let detail = detail_with_per_collection(); + let t1 = worker1.get_telemetry_data(detail); + let t2 = worker2.get_telemetry_data(detail); + + let mut merged = WebApiTelemetry::default(); + merged.merge(&t1); + merged.merge(&t2); + + // Global: 3 total + assert_eq!(merged.responses[method][&200].count, 3); + + // Per-collection: col_a=2 (merged from two workers), col_b=1 + assert_eq!( + merged.per_collection_responses["col_a"][method][&200].count, + 2 + ); + assert_eq!( + merged.per_collection_responses["col_b"][method][&200].count, + 1 + ); + } + + #[test] + fn test_collector_merges_workers() { + let mut collector = ActixTelemetryCollector { workers: vec![] }; + let w1 = collector.create_web_worker_telemetry(); + let w2 = collector.create_web_worker_telemetry(); + let method = "POST /collections/{name}/points/search"; + + w1.lock().add_response( + method.into(), + 200, + std::time::Instant::now(), + Some("col_a".into()), + ); + w2.lock().add_response( + method.into(), + 200, + std::time::Instant::now(), + Some("col_a".into()), + ); + + let telemetry = collector.get_telemetry_data(detail_with_per_collection()); + assert_eq!(telemetry.responses[method][&200].count, 2); + assert_eq!( + telemetry.per_collection_responses["col_a"][method][&200].count, + 2 + ); + } + + #[test] + fn test_anonymize_strips_per_collection() { + let mut worker = ActixWorkerTelemetryCollector::default(); + actix_add( + &mut worker, + "POST /collections/{name}/points/search", + 200, + Some("secret_collection"), + ); + + let telemetry = worker.get_telemetry_data(detail_with_per_collection()); + assert!(!telemetry.per_collection_responses.is_empty()); + + let anonymized = telemetry.anonymize(); + assert!(anonymized.per_collection_responses.is_empty()); + // Global responses should still be present + assert!(!anonymized.responses.is_empty()); + } + + #[test] + fn test_all_collections_tracked() { + let mut worker = ActixWorkerTelemetryCollector::default(); + let method = "POST /collections/{name}/points/search"; + + for i in 0..500 { + actix_add(&mut worker, method, 200, Some(&format!("col_{i}"))); + } + + let telemetry = worker.get_telemetry_data(detail_with_per_collection()); + assert_eq!(telemetry.per_collection_responses.len(), 500); } } diff --git a/src/common/telemetry_reporting.rs b/src/common/telemetry_reporting.rs index b15a4eac37..6299029fba 100644 --- a/src/common/telemetry_reporting.rs +++ b/src/common/telemetry_reporting.rs @@ -14,6 +14,7 @@ use super::telemetry::TelemetryCollector; const DETAIL: TelemetryDetail = TelemetryDetail { level: DetailsLevel::Level2, histograms: false, + per_collection: false, }; const REPORTING_INTERVAL: Duration = Duration::from_secs(60 * 60); // One hour diff --git a/src/tonic/api/mod.rs b/src/tonic/api/mod.rs index 7c47e759d0..a3f963cfe9 100644 --- a/src/tonic/api/mod.rs +++ b/src/tonic/api/mod.rs @@ -5,6 +5,7 @@ pub mod points_internal_api; pub mod qdrant_internal_api; pub mod raft_api; pub mod snapshots_api; +pub mod telemetry_wrapper; mod collections_common; mod query_common; diff --git a/src/tonic/api/points_api.rs b/src/tonic/api/points_api.rs index def8582ffe..4117b2dc1a 100644 --- a/src/tonic/api/points_api.rs +++ b/src/tonic/api/points_api.rs @@ -115,8 +115,8 @@ impl Points for PointsService { let auth = extract_auth(&mut request); let inner_request = request.into_inner(); - let hw_metrics = self - .get_request_collection_hw_usage_counter(inner_request.collection_name.clone(), None); + let collection_name = inner_request.collection_name.clone(); + let hw_metrics = self.get_request_collection_hw_usage_counter(collection_name, None); get( StrictModeCheckedTocProvider::new(&self.dispatcher), @@ -165,10 +165,8 @@ impl Points for PointsService { let auth = extract_auth(&mut request); - let hw_metrics = self.get_request_collection_hw_usage_counter( - request.get_ref().collection_name.clone(), - None, - ); + let collection_name = request.get_ref().collection_name.clone(); + let hw_metrics = self.get_request_collection_hw_usage_counter(collection_name, None); delete_vectors( StrictModeCheckedTocProvider::new(&self.dispatcher), @@ -435,9 +433,9 @@ impl Points for PointsService { let auth = extract_auth(&mut request); let inner_request = request.into_inner(); + let collection_name = inner_request.collection_name.clone(); - let hw_metrics = self - .get_request_collection_hw_usage_counter(inner_request.collection_name.clone(), None); + let hw_metrics = self.get_request_collection_hw_usage_counter(collection_name, None); scroll( StrictModeCheckedTocProvider::new(&self.dispatcher), @@ -678,10 +676,8 @@ impl Points for PointsService { ) -> Result, Status> { validate(request.get_ref())?; let auth = extract_auth(&mut request); - let hw_metrics = self.get_request_collection_hw_usage_counter( - request.get_ref().collection_name.clone(), - None, - ); + let collection_name = request.get_ref().collection_name.clone(); + let hw_metrics = self.get_request_collection_hw_usage_counter(collection_name, None); facet( StrictModeCheckedTocProvider::new(&self.dispatcher), request.into_inner(), diff --git a/src/tonic/api/points_internal_api.rs b/src/tonic/api/points_internal_api.rs index e622da1562..24e530fa77 100644 --- a/src/tonic/api/points_internal_api.rs +++ b/src/tonic/api/points_internal_api.rs @@ -43,6 +43,11 @@ fn full_internal_auth() -> Auth { } /// This API is intended for P2P communication within a distributed deployment. +/// +/// Note: unlike the public `PointsService`, this service does NOT attach +/// `CollectionName` to gRPC responses. Internal endpoints +/// (`/qdrant.PointsInternal/…`) are not in `GRPC_ENDPOINT_WHITELIST`, so +/// attaching a collection name would have no effect on `/metrics` output. pub struct PointsInternalService { toc: Arc, service_config: ServiceConfig, diff --git a/src/tonic/api/qdrant_internal_api.rs b/src/tonic/api/qdrant_internal_api.rs index 032e36b9c5..706e4ad8c5 100644 --- a/src/tonic/api/qdrant_internal_api.rs +++ b/src/tonic/api/qdrant_internal_api.rs @@ -89,6 +89,7 @@ impl QdrantInternal for QdrantInternalService { let detail = TelemetryDetail { level: details_level, histograms: false, + per_collection: false, }; let only_collections = diff --git a/src/tonic/api/telemetry_wrapper.rs b/src/tonic/api/telemetry_wrapper.rs new file mode 100644 index 0000000000..db3a5935d6 --- /dev/null +++ b/src/tonic/api/telemetry_wrapper.rs @@ -0,0 +1,640 @@ +//! Thin wrappers around gRPC service traits that extract `collection_name` +//! from every request and attach it as a [`CollectionName`] extension on the +//! response. The telemetry Tower layer ([`TonicTelemetryService`]) later reads +//! this extension to record per-collection metrics without the individual +//! handlers having to know about telemetry at all. + +use api::grpc::qdrant::points_server::Points; +use api::grpc::qdrant::shard_snapshots_server::ShardSnapshots; +use api::grpc::qdrant::snapshots_server::Snapshots; +use api::grpc::qdrant::{ + ClearPayloadPoints, CountPoints, CountResponse, CreateFieldIndexCollection, + CreateFullSnapshotRequest, CreateShardSnapshotRequest, CreateSnapshotRequest, + CreateSnapshotResponse, DeleteFieldIndexCollection, DeleteFullSnapshotRequest, + DeletePayloadPoints, DeletePointVectors, DeletePoints, DeleteShardSnapshotRequest, + DeleteSnapshotRequest, DeleteSnapshotResponse, DiscoverBatchPoints, DiscoverBatchResponse, + DiscoverPoints, DiscoverResponse, FacetCounts, FacetResponse, GetPoints, GetResponse, + ListFullSnapshotsRequest, ListShardSnapshotsRequest, ListSnapshotsRequest, + ListSnapshotsResponse, PointsOperationResponse, QueryBatchPoints, QueryBatchResponse, + QueryGroupsResponse, QueryPointGroups, QueryPoints, QueryResponse, RecommendBatchPoints, + RecommendBatchResponse, RecommendGroupsResponse, RecommendPointGroups, RecommendPoints, + RecommendResponse, RecoverShardSnapshotRequest, RecoverSnapshotResponse, ScrollPoints, + ScrollResponse, SearchBatchPoints, SearchBatchResponse, SearchGroupsResponse, + SearchMatrixOffsetsResponse, SearchMatrixPairsResponse, SearchMatrixPoints, SearchPointGroups, + SearchPoints, SearchResponse, SetPayloadPoints, UpdateBatchPoints, UpdateBatchResponse, + UpdatePointVectors, UpsertPoints, +}; +use tonic::{Request, Response, Status}; + +use crate::common::telemetry_ops::requests_telemetry::CollectionName; + +/// Wraps a [`Points`] service, attaching `collection_name` to every response. +pub struct PointsTelemetryWrapper { + inner: T, +} + +impl PointsTelemetryWrapper { + pub fn new(inner: T) -> Self { + Self { inner } + } +} + +#[tonic::async_trait] +impl Points for PointsTelemetryWrapper { + async fn upsert( + &self, + request: Request, + ) -> Result, Status> { + let cn = request.get_ref().collection_name.clone(); + let mut resp = self.inner.upsert(request).await?; + resp.extensions_mut().insert(CollectionName(cn)); + Ok(resp) + } + + async fn delete( + &self, + request: Request, + ) -> Result, Status> { + let cn = request.get_ref().collection_name.clone(); + let mut resp = self.inner.delete(request).await?; + resp.extensions_mut().insert(CollectionName(cn)); + Ok(resp) + } + + async fn get(&self, request: Request) -> Result, Status> { + let cn = request.get_ref().collection_name.clone(); + let mut resp = self.inner.get(request).await?; + resp.extensions_mut().insert(CollectionName(cn)); + Ok(resp) + } + + async fn update_vectors( + &self, + request: Request, + ) -> Result, Status> { + let cn = request.get_ref().collection_name.clone(); + let mut resp = self.inner.update_vectors(request).await?; + resp.extensions_mut().insert(CollectionName(cn)); + Ok(resp) + } + + async fn delete_vectors( + &self, + request: Request, + ) -> Result, Status> { + let cn = request.get_ref().collection_name.clone(); + let mut resp = self.inner.delete_vectors(request).await?; + resp.extensions_mut().insert(CollectionName(cn)); + Ok(resp) + } + + async fn set_payload( + &self, + request: Request, + ) -> Result, Status> { + let cn = request.get_ref().collection_name.clone(); + let mut resp = self.inner.set_payload(request).await?; + resp.extensions_mut().insert(CollectionName(cn)); + Ok(resp) + } + + async fn overwrite_payload( + &self, + request: Request, + ) -> Result, Status> { + let cn = request.get_ref().collection_name.clone(); + let mut resp = self.inner.overwrite_payload(request).await?; + resp.extensions_mut().insert(CollectionName(cn)); + Ok(resp) + } + + async fn delete_payload( + &self, + request: Request, + ) -> Result, Status> { + let cn = request.get_ref().collection_name.clone(); + let mut resp = self.inner.delete_payload(request).await?; + resp.extensions_mut().insert(CollectionName(cn)); + Ok(resp) + } + + async fn clear_payload( + &self, + request: Request, + ) -> Result, Status> { + let cn = request.get_ref().collection_name.clone(); + let mut resp = self.inner.clear_payload(request).await?; + resp.extensions_mut().insert(CollectionName(cn)); + Ok(resp) + } + + async fn update_batch( + &self, + request: Request, + ) -> Result, Status> { + let cn = request.get_ref().collection_name.clone(); + let mut resp = self.inner.update_batch(request).await?; + resp.extensions_mut().insert(CollectionName(cn)); + Ok(resp) + } + + async fn create_field_index( + &self, + request: Request, + ) -> Result, Status> { + let cn = request.get_ref().collection_name.clone(); + let mut resp = self.inner.create_field_index(request).await?; + resp.extensions_mut().insert(CollectionName(cn)); + Ok(resp) + } + + async fn delete_field_index( + &self, + request: Request, + ) -> Result, Status> { + let cn = request.get_ref().collection_name.clone(); + let mut resp = self.inner.delete_field_index(request).await?; + resp.extensions_mut().insert(CollectionName(cn)); + Ok(resp) + } + + async fn search( + &self, + request: Request, + ) -> Result, Status> { + let cn = request.get_ref().collection_name.clone(); + let mut resp = self.inner.search(request).await?; + resp.extensions_mut().insert(CollectionName(cn)); + Ok(resp) + } + + async fn search_batch( + &self, + request: Request, + ) -> Result, Status> { + let cn = request.get_ref().collection_name.clone(); + let mut resp = self.inner.search_batch(request).await?; + resp.extensions_mut().insert(CollectionName(cn)); + Ok(resp) + } + + async fn search_groups( + &self, + request: Request, + ) -> Result, Status> { + let cn = request.get_ref().collection_name.clone(); + let mut resp = self.inner.search_groups(request).await?; + resp.extensions_mut().insert(CollectionName(cn)); + Ok(resp) + } + + async fn scroll( + &self, + request: Request, + ) -> Result, Status> { + let cn = request.get_ref().collection_name.clone(); + let mut resp = self.inner.scroll(request).await?; + resp.extensions_mut().insert(CollectionName(cn)); + Ok(resp) + } + + async fn recommend( + &self, + request: Request, + ) -> Result, Status> { + let cn = request.get_ref().collection_name.clone(); + let mut resp = self.inner.recommend(request).await?; + resp.extensions_mut().insert(CollectionName(cn)); + Ok(resp) + } + + async fn recommend_batch( + &self, + request: Request, + ) -> Result, Status> { + let cn = request.get_ref().collection_name.clone(); + let mut resp = self.inner.recommend_batch(request).await?; + resp.extensions_mut().insert(CollectionName(cn)); + Ok(resp) + } + + async fn recommend_groups( + &self, + request: Request, + ) -> Result, Status> { + let cn = request.get_ref().collection_name.clone(); + let mut resp = self.inner.recommend_groups(request).await?; + resp.extensions_mut().insert(CollectionName(cn)); + Ok(resp) + } + + async fn discover( + &self, + request: Request, + ) -> Result, Status> { + let cn = request.get_ref().collection_name.clone(); + let mut resp = self.inner.discover(request).await?; + resp.extensions_mut().insert(CollectionName(cn)); + Ok(resp) + } + + async fn discover_batch( + &self, + request: Request, + ) -> Result, Status> { + let cn = request.get_ref().collection_name.clone(); + let mut resp = self.inner.discover_batch(request).await?; + resp.extensions_mut().insert(CollectionName(cn)); + Ok(resp) + } + + async fn count( + &self, + request: Request, + ) -> Result, Status> { + let cn = request.get_ref().collection_name.clone(); + let mut resp = self.inner.count(request).await?; + resp.extensions_mut().insert(CollectionName(cn)); + Ok(resp) + } + + async fn query( + &self, + request: Request, + ) -> Result, Status> { + let cn = request.get_ref().collection_name.clone(); + let mut resp = self.inner.query(request).await?; + resp.extensions_mut().insert(CollectionName(cn)); + Ok(resp) + } + + async fn query_batch( + &self, + request: Request, + ) -> Result, Status> { + let cn = request.get_ref().collection_name.clone(); + let mut resp = self.inner.query_batch(request).await?; + resp.extensions_mut().insert(CollectionName(cn)); + Ok(resp) + } + + async fn query_groups( + &self, + request: Request, + ) -> Result, Status> { + let cn = request.get_ref().collection_name.clone(); + let mut resp = self.inner.query_groups(request).await?; + resp.extensions_mut().insert(CollectionName(cn)); + Ok(resp) + } + + async fn facet( + &self, + request: Request, + ) -> Result, Status> { + let cn = request.get_ref().collection_name.clone(); + let mut resp = self.inner.facet(request).await?; + resp.extensions_mut().insert(CollectionName(cn)); + Ok(resp) + } + + async fn search_matrix_pairs( + &self, + request: Request, + ) -> Result, Status> { + let cn = request.get_ref().collection_name.clone(); + let mut resp = self.inner.search_matrix_pairs(request).await?; + resp.extensions_mut().insert(CollectionName(cn)); + Ok(resp) + } + + async fn search_matrix_offsets( + &self, + request: Request, + ) -> Result, Status> { + let cn = request.get_ref().collection_name.clone(); + let mut resp = self.inner.search_matrix_offsets(request).await?; + resp.extensions_mut().insert(CollectionName(cn)); + Ok(resp) + } +} + +/// Wraps a [`Snapshots`] service, attaching `collection_name` to every +/// collection-scoped response. Full-snapshot methods are passed through as-is. +pub struct SnapshotsTelemetryWrapper { + inner: T, +} + +impl SnapshotsTelemetryWrapper { + pub fn new(inner: T) -> Self { + Self { inner } + } +} + +#[tonic::async_trait] +impl Snapshots for SnapshotsTelemetryWrapper { + async fn create( + &self, + request: Request, + ) -> Result, Status> { + let cn = request.get_ref().collection_name.clone(); + let mut resp = self.inner.create(request).await?; + resp.extensions_mut().insert(CollectionName(cn)); + Ok(resp) + } + + async fn list( + &self, + request: Request, + ) -> Result, Status> { + let cn = request.get_ref().collection_name.clone(); + let mut resp = self.inner.list(request).await?; + resp.extensions_mut().insert(CollectionName(cn)); + Ok(resp) + } + + async fn delete( + &self, + request: Request, + ) -> Result, Status> { + let cn = request.get_ref().collection_name.clone(); + let mut resp = self.inner.delete(request).await?; + resp.extensions_mut().insert(CollectionName(cn)); + Ok(resp) + } + + async fn create_full( + &self, + request: Request, + ) -> Result, Status> { + self.inner.create_full(request).await + } + + async fn list_full( + &self, + request: Request, + ) -> Result, Status> { + self.inner.list_full(request).await + } + + async fn delete_full( + &self, + request: Request, + ) -> Result, Status> { + self.inner.delete_full(request).await + } +} + +/// Wraps a [`ShardSnapshots`] service, attaching `collection_name` to every response. +pub struct ShardSnapshotsTelemetryWrapper { + inner: T, +} + +impl ShardSnapshotsTelemetryWrapper { + pub fn new(inner: T) -> Self { + Self { inner } + } +} + +#[tonic::async_trait] +impl ShardSnapshots for ShardSnapshotsTelemetryWrapper { + async fn create( + &self, + request: Request, + ) -> Result, Status> { + let cn = request.get_ref().collection_name.clone(); + let mut resp = self.inner.create(request).await?; + resp.extensions_mut().insert(CollectionName(cn)); + Ok(resp) + } + + async fn list( + &self, + request: Request, + ) -> Result, Status> { + let cn = request.get_ref().collection_name.clone(); + let mut resp = self.inner.list(request).await?; + resp.extensions_mut().insert(CollectionName(cn)); + Ok(resp) + } + + async fn delete( + &self, + request: Request, + ) -> Result, Status> { + let cn = request.get_ref().collection_name.clone(); + let mut resp = self.inner.delete(request).await?; + resp.extensions_mut().insert(CollectionName(cn)); + Ok(resp) + } + + async fn recover( + &self, + request: Request, + ) -> Result, Status> { + let cn = request.get_ref().collection_name.clone(); + let mut resp = self.inner.recover(request).await?; + resp.extensions_mut().insert(CollectionName(cn)); + Ok(resp) + } +} + +#[cfg(test)] +mod tests { + use api::grpc::qdrant::points_server::Points; + use api::grpc::qdrant::shard_snapshots_server::ShardSnapshots; + use api::grpc::qdrant::snapshots_server::Snapshots; + use api::grpc::qdrant::*; + use tonic::{Request, Response, Status}; + + use super::*; + use crate::common::telemetry_ops::requests_telemetry::CollectionName; + + // Points + macro_rules! mock_and_test_points { + ($($method:ident($req:ident) -> $resp:ident),* $(,)?) => { + struct MockPoints; + + #[tonic::async_trait] + #[allow(unused_variables)] + impl Points for MockPoints { + $( + async fn $method(&self, r: Request<$req>) -> Result, Status> { + Ok(Response::new(Default::default())) + } + )* + } + + $( + #[tokio::test] + async fn $method() { + let w = PointsTelemetryWrapper::new(MockPoints); + let r = w + .$method(Request::new($req { + collection_name: stringify!($method).into(), + ..Default::default() + })) + .await + .unwrap(); + assert_eq!( + r.extensions().get::().unwrap().0, + stringify!($method), + ); + } + )* + }; + } + + mock_and_test_points! { + upsert(UpsertPoints) -> PointsOperationResponse, + delete(DeletePoints) -> PointsOperationResponse, + get(GetPoints) -> GetResponse, + update_vectors(UpdatePointVectors) -> PointsOperationResponse, + delete_vectors(DeletePointVectors) -> PointsOperationResponse, + set_payload(SetPayloadPoints) -> PointsOperationResponse, + overwrite_payload(SetPayloadPoints) -> PointsOperationResponse, + delete_payload(DeletePayloadPoints) -> PointsOperationResponse, + clear_payload(ClearPayloadPoints) -> PointsOperationResponse, + update_batch(UpdateBatchPoints) -> UpdateBatchResponse, + create_field_index(CreateFieldIndexCollection) -> PointsOperationResponse, + delete_field_index(DeleteFieldIndexCollection) -> PointsOperationResponse, + search(SearchPoints) -> SearchResponse, + search_batch(SearchBatchPoints) -> SearchBatchResponse, + search_groups(SearchPointGroups) -> SearchGroupsResponse, + scroll(ScrollPoints) -> ScrollResponse, + recommend(RecommendPoints) -> RecommendResponse, + recommend_batch(RecommendBatchPoints) -> RecommendBatchResponse, + recommend_groups(RecommendPointGroups) -> RecommendGroupsResponse, + discover(DiscoverPoints) -> DiscoverResponse, + discover_batch(DiscoverBatchPoints) -> DiscoverBatchResponse, + count(CountPoints) -> CountResponse, + query(QueryPoints) -> QueryResponse, + query_batch(QueryBatchPoints) -> QueryBatchResponse, + query_groups(QueryPointGroups) -> QueryGroupsResponse, + facet(FacetCounts) -> FacetResponse, + search_matrix_pairs(SearchMatrixPoints) -> SearchMatrixPairsResponse, + search_matrix_offsets(SearchMatrixPoints) -> SearchMatrixOffsetsResponse, + } + + // Snapshots + macro_rules! mock_and_test_snapshots { + ( + with_cn: { $($method:ident($req:ident) -> $resp:ident),* $(,)? } + passthrough: { $($pt_method:ident($pt_req:ident) -> $pt_resp:ident),* $(,)? } + ) => { + struct MockSnapshots; + + #[tonic::async_trait] + #[allow(unused_variables)] + impl Snapshots for MockSnapshots { + $( + async fn $method(&self, r: Request<$req>) -> Result, Status> { + Ok(Response::new(Default::default())) + } + )* + $( + async fn $pt_method(&self, r: Request<$pt_req>) -> Result, Status> { + Ok(Response::new(Default::default())) + } + )* + } + + mod snapshots_tests { + use super::*; + + $( + #[tokio::test] + #[allow(clippy::needless_update)] + async fn $method() { + let w = SnapshotsTelemetryWrapper::new(MockSnapshots); + let r = w + .$method(Request::new($req { + collection_name: stringify!($method).into(), + ..Default::default() + })) + .await + .unwrap(); + assert_eq!( + r.extensions().get::().unwrap().0, + stringify!($method), + ); + } + )* + + $( + #[tokio::test] + async fn $pt_method() { + let w = SnapshotsTelemetryWrapper::new(MockSnapshots); + let r = w + .$pt_method(Request::new($pt_req::default())) + .await + .unwrap(); + assert!( + r.extensions().get::().is_none(), + "passthrough method should not attach CollectionName", + ); + } + )* + } + }; + } + + mock_and_test_snapshots! { + with_cn: { + create(CreateSnapshotRequest) -> CreateSnapshotResponse, + list(ListSnapshotsRequest) -> ListSnapshotsResponse, + delete(DeleteSnapshotRequest) -> DeleteSnapshotResponse, + } + passthrough: { + create_full(CreateFullSnapshotRequest) -> CreateSnapshotResponse, + list_full(ListFullSnapshotsRequest) -> ListSnapshotsResponse, + delete_full(DeleteFullSnapshotRequest) -> DeleteSnapshotResponse, + } + } + + // ShardSnapshots + macro_rules! mock_and_test_shard_snapshots { + ($($method:ident($req:ident) -> $resp:ident),* $(,)?) => { + struct MockShardSnapshots; + + #[tonic::async_trait] + #[allow(unused_variables)] + impl ShardSnapshots for MockShardSnapshots { + $( + async fn $method(&self, r: Request<$req>) -> Result, Status> { + Ok(Response::new(Default::default())) + } + )* + } + + mod shard_snapshots_tests { + use super::*; + + $( + #[tokio::test] + async fn $method() { + let w = ShardSnapshotsTelemetryWrapper::new(MockShardSnapshots); + let r = w + .$method(Request::new($req { + collection_name: stringify!($method).into(), + ..Default::default() + })) + .await + .unwrap(); + assert_eq!( + r.extensions().get::().unwrap().0, + stringify!($method), + ); + } + )* + } + }; + } + + mock_and_test_shard_snapshots! { + create(CreateShardSnapshotRequest) -> CreateSnapshotResponse, + list(ListShardSnapshotsRequest) -> ListSnapshotsResponse, + delete(DeleteShardSnapshotRequest) -> DeleteSnapshotResponse, + recover(RecoverShardSnapshotRequest) -> RecoverSnapshotResponse, + } +} diff --git a/src/tonic/mod.rs b/src/tonic/mod.rs index 7c21f74277..b0d68fd270 100644 --- a/src/tonic/mod.rs +++ b/src/tonic/mod.rs @@ -48,6 +48,9 @@ use crate::tonic::api::points_api::PointsService; use crate::tonic::api::points_internal_api::PointsInternalService; use crate::tonic::api::qdrant_internal_api::QdrantInternalService; use crate::tonic::api::snapshots_api::{ShardSnapshotsService, SnapshotsService}; +use crate::tonic::api::telemetry_wrapper::{ + PointsTelemetryWrapper, ShardSnapshotsTelemetryWrapper, SnapshotsTelemetryWrapper, +}; #[derive(Default)] pub struct QdrantService {} @@ -177,13 +180,13 @@ pub fn init( .max_decoding_message_size(usize::MAX), ) .add_service( - PointsServer::new(points_service) + PointsServer::new(PointsTelemetryWrapper::new(points_service)) .send_compressed(CompressionEncoding::Gzip) .accept_compressed(CompressionEncoding::Gzip) .max_decoding_message_size(usize::MAX), ) .add_service( - SnapshotsServer::new(snapshot_service) + SnapshotsServer::new(SnapshotsTelemetryWrapper::new(snapshot_service)) .send_compressed(CompressionEncoding::Gzip) .accept_compressed(CompressionEncoding::Gzip) .max_decoding_message_size(usize::MAX), @@ -226,7 +229,6 @@ pub fn init_internal( runtime .block_on(async { let socket = SocketAddr::from((host.parse::().unwrap(), internal_grpc_port)); - let qdrant_service = QdrantService::default(); let points_internal_service = PointsInternalService::new(toc.clone(), settings.service.clone()); @@ -292,10 +294,12 @@ pub fn init_internal( .max_decoding_message_size(usize::MAX), ) .add_service( - ShardSnapshotsServer::new(shard_snapshots_service) - .send_compressed(CompressionEncoding::Gzip) - .accept_compressed(CompressionEncoding::Gzip) - .max_decoding_message_size(usize::MAX), + ShardSnapshotsServer::new(ShardSnapshotsTelemetryWrapper::new( + shard_snapshots_service, + )) + .send_compressed(CompressionEncoding::Gzip) + .accept_compressed(CompressionEncoding::Gzip) + .max_decoding_message_size(usize::MAX), ) .add_service( RaftServer::new(raft_service) diff --git a/src/tonic/tonic_telemetry.rs b/src/tonic/tonic_telemetry.rs index 23e6b4c416..bd37e2ae21 100644 --- a/src/tonic/tonic_telemetry.rs +++ b/src/tonic/tonic_telemetry.rs @@ -7,7 +7,7 @@ use tower::Service; use tower_layer::Layer; use crate::common::telemetry_ops::requests_telemetry::{ - TonicTelemetryCollector, TonicWorkerTelemetryCollector, + CollectionName, TonicTelemetryCollector, TonicWorkerTelemetryCollector, }; /// Based on https://grpc.io/docs/guides/status-codes/ @@ -70,9 +70,16 @@ where } }); + // Collection name is attached to response extensions by + // telemetry wrappers (see telemetry_wrapper.rs). + let collection_name = response + .extensions() + .get::() + .map(|cn| cn.0.clone()); + telemetry_data .lock() - .add_response(method_name, instant, status_code); + .add_response(method_name, instant, status_code, collection_name); Ok(response) }) } diff --git a/tests/openapi/test_service.py b/tests/openapi/test_service.py index 3e2fd1198f..ef887c02c9 100644 --- a/tests/openapi/test_service.py +++ b/tests/openapi/test_service.py @@ -26,6 +26,68 @@ def test_metrics(): assert 'collections_total ' in response.text +def test_metrics_default_no_per_collection(collection_name): + """By default (per_collection not requested), request metrics must NOT have a collection label.""" + # Make a request that hits a whitelisted endpoint so metrics are populated + request_with_validation( + api='/collections/{collection_name}/points/scroll', + method="POST", + path_params={'collection_name': collection_name}, + body={"limit": 1}, + ) + + response = request_with_validation( + api='/metrics', + method="GET", + ) + assert response.ok + + # REST request metrics should exist (global mode) + assert 'rest_responses_total' in response.text + + # Existing collection-level metrics should use "id" label, not "collection" + assert f'collection_points{{id="{collection_name}"}}' in response.text \ + or f'id="{collection_name}"' in response.text + + # Per-collection request metrics (collection= label on rest_/grpc_ metrics) should NOT appear + for line in response.text.splitlines(): + if line.startswith('#'): + continue + if 'rest_responses_' in line or 'grpc_responses_' in line: + assert 'collection=' not in line, ( + f"Per-collection label found in default mode: {line}" + ) + + +def test_metrics_with_per_collection(collection_name): + """When per_collection=true is requested, request metrics must have a collection label.""" + # Make a request that hits a whitelisted endpoint so metrics are populated + request_with_validation( + api='/collections/{collection_name}/points/scroll', + method="POST", + path_params={'collection_name': collection_name}, + body={"limit": 1}, + ) + + response = request_with_validation( + api='/metrics', + method="GET", + query_params={'per_collection': 'true'}, + ) + assert response.ok + + # Per-collection mode: collection label should appear + found_per_collection = False + for line in response.text.splitlines(): + if line.startswith('#'): + continue + if 'rest_responses_' in line and f'collection="{collection_name}"' in line: + found_per_collection = True + break + + assert found_per_collection, "Per-collection label not found when per_collection=true" + + def test_telemetry(): response = request_with_validation( api='/telemetry', @@ -38,11 +100,17 @@ def test_telemetry(): assert result['collections']['number_of_collections'] >= 1 - endpoint = result['requests']['rest']['responses']['PUT /collections/{name}/points'] + endpoint = result['requests']['rest']['responses']['PUT /collections/{collection_name}/points'] assert endpoint['200']['count'] > 0 assert 'avg_duration_micros' in endpoint['200'] + # By default, per_collection_responses should be absent (per_collection not requested) + assert 'per_collection_responses' not in result['requests']['rest'], \ + "per_collection_responses should not appear when per_collection is not requested" + assert 'per_collection_responses' not in result['requests']['grpc'], \ + "per_collection_responses should not appear when per_collection is not requested" + @pytest.mark.parametrize("level", [0, 1, 2, 3, 10]) def test_telemetry_detail(level: int): @@ -58,7 +126,7 @@ def test_telemetry_detail(level: int): assert result['collections']['number_of_collections'] >= 1 - endpoint = result['requests']['rest']['responses']['PUT /collections/{name}/points'] + endpoint = result['requests']['rest']['responses']['PUT /collections/{collection_name}/points'] assert endpoint['200']['count'] > 0 if level == 0: diff --git a/tests/per_collection_metrics_test.sh b/tests/per_collection_metrics_test.sh new file mode 100644 index 0000000000..4286ec70ad --- /dev/null +++ b/tests/per_collection_metrics_test.sh @@ -0,0 +1,81 @@ +#!/usr/bin/env bash +# This test checks that Qdrant exposes per-collection metrics when requested via query parameter. + +set -e + +QDRANT_HOST=${QDRANT_HOST:-'localhost:6333'} +COLLECTION_NAME="test_collection_metrics" + +echo "Using Qdrant host: $QDRANT_HOST" + +# 1. Cleanup & Create collection +echo "Creating collection $COLLECTION_NAME..." +# Attempt to delete collection if it exists, ignore output/errors +curl -X DELETE "http://$QDRANT_HOST/collections/$COLLECTION_NAME" -s > /dev/null || true + +curl -X PUT "http://$QDRANT_HOST/collections/$COLLECTION_NAME" \ + -H 'Content-Type: application/json' \ + --fail -s \ + --data-raw '{ + "vectors": { + "size": 4, + "distance": "Dot" + } + }' + + +# 2. Insert points +echo "Inserting points..." +curl -X PUT "http://$QDRANT_HOST/collections/$COLLECTION_NAME/points?wait=true" \ + -H 'Content-Type: application/json' \ + --fail -s \ + --data-raw '{ + "points": [ + {"id": 1, "vector": [0.05, 0.61, 0.76, 0.74]}, + {"id": 2, "vector": [0.19, 0.81, 0.75, 0.11]} + ] + }' + + +# 3. Search points (to generate read metrics) +echo "Searching points..." +curl -X POST "http://$QDRANT_HOST/collections/$COLLECTION_NAME/points/search" \ + -H 'Content-Type: application/json' \ + --fail -s \ + --data-raw '{ + "vector": [0.2,0.1,0.9,0.7], + "top": 3 + }' + + +# 4. Fetch metrics with per_collection=true and verify +echo "Fetching per-collection metrics..." +METRICS=$(curl -s --fail "http://$QDRANT_HOST/metrics?per_collection=true") + +echo "Verifying per-collection metrics..." + +# Check for existence of collection label in rest responses +if echo "$METRICS" | grep -q "rest_responses_total{.*collection=\"$COLLECTION_NAME\".*}"; then + echo "Found per-collection rest_responses_total" +else + echo "Failed to find per-collection rest_responses_total" + exit 1 +fi + +# 5. Fetch default metrics (no per_collection) and verify global mode +echo "Fetching default metrics..." +DEFAULT_METRICS=$(curl -s --fail "http://$QDRANT_HOST/metrics") + +# In default mode, rest_responses should NOT have collection label +if echo "$DEFAULT_METRICS" | grep -q 'rest_responses_total{.*collection='; then + echo "ERROR: Found per-collection label in default mode" + exit 1 +else + echo "Default mode correctly shows global metrics (no collection label)" +fi + +echo "All per-collection metrics verification passed!" + +# Cleanup +echo "Cleaning up..." +curl -X DELETE "http://$QDRANT_HOST/collections/$COLLECTION_NAME" -s