upstream-archive/qdrant
1
0
Fork 0
mirror of https://github.com/qdrant/qdrant.git synced 2026-07-23 11:11:00 -05:00
Code Issues Packages Projects Releases Wiki Activity

Implement Per collection metrics for Promethus (#8214)

Browse Source 
* Implement Per-Collection Prometheus Metrics

* Update config/config.yaml

Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>

* Update tests/per_collection_metrics_test.sh

Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>

* Update tests/per_collection_metrics_test.sh

Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>

* fix ci

* comment

* Update tests/per_collection_metrics_test.sh

Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>

* adress revew

* fix: linter

* refactor: cardinality limit, anonymize fix, cleanup

- Add max_per_collection_metrics config (default 256) to bound
  per-collection label cardinality
- Fix anonymize: strip per_collection_responses entirely instead of
  leaking hashed collection names
- Move CollectionName into requests_telemetry, remove
  telemetry_context.rs
- Add unit tests for cardinality limits
- Add integration test assertions for default mode
- Regenerate OpenAPI spec
- Document why internal gRPC doesn't attach CollectionName

* avoid cloning

* fix: address pr reviews

* fix: linter

* fix: linter

* feat: enforce {name} in actix api

* chore: remove empty line

* feat: add actix pre-commit hook

* feat: potential enforce tonic collection name

* feat: use proc_macro instead

* feat: use collection_name instead of name

* feat: add tonic telemetry tests

* fix: linter

* fix: linter

* chore: remove unneccessary clone

* fix: use collection_name everywhere

* fix: python tests and openapi

* chore: simplify collection_name tests

* actix collection_name enforcign: use relative path in test + avoid grep

* fix: remove macro

* chore: add some comment to telemetry_wrapper

* fix: clippy

---------

Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>
Co-authored-by: Daniel Boros <dancixx@gmail.com>
Co-authored-by: generall <andrey@vasnetsov.com>
This commit is contained in:
Excellencedev
2026-03-23 15:19:02 +01:00
committed by generall
parent b128e6042b
commit 77e2fe72f8
33 changed files with 1799 additions and 269 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
Cargo.lock generated
View File

@@ -5855,6 +5855,7 @@ dependencies = [
"itertools 0.14.0",
"jsonwebtoken",
"log",
"macros",
"mockito",
"murmur3",
"nix 0.31.2",

1
Cargo.toml
View File

@@ -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" }

42
docs/redoc/master/openapi.json
View File

@@ -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"
}
}
}
}
}
},

8
lib/common/common/src/types.rs
View File

@@ -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,
}
}
}

12
openapi/openapi-service.ytt.yaml
View File

@@ -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"

61
src/actix/actix_telemetry.rs
View File

@@ -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<std::path::PathBuf>) {
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"),
);
}
}

36
src/actix/api/collections_api.rs
View File

@@ -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<Dispatcher>,
collection: Path<CollectionPath>,
@@ -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<Dispatcher>,
collection: Path<CollectionPath>,
@@ -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<Dispatcher>,
collection: Path<CollectionPath>,
@@ -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<Dispatcher>,
collection: Path<StrictCollectionPath>,
@@ -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<Dispatcher>,
collection: Path<CollectionPath>,
@@ -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<Dispatcher>,
collection: Path<CollectionPath>,
@@ -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<Dispatcher>,
collection: Path<CollectionPath>,
@@ -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<Dispatcher>,
collection: Path<CollectionPath>,
@@ -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<Dispatcher>,
collection: Path<CollectionPath>,
@@ -284,7 +284,7 @@ fn get_optimizations(
let options = OptimizationsRequestOptions::try_from(&params.into_inner())?;
let pass = new_unchecked_verification_pass();
let collection_pass = auth.check_collection_access(
&collection.name,
&collection.collection_name,
AccessRequirements::new(),
"get_optimizations",
)?;

8
src/actix/api/count_api.rs
View File

@@ -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<Dispatcher>,
collection: Path<CollectionPath>,
@@ -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(),

4
src/actix/api/debug_api.rs
View File

@@ -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<Dispatcher>,
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<Dispatcher>,
path: web::Path<(String, ShardId)>,

16
src/actix/api/discover_api.rs
View File

@@ -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<Dispatcher>,
collection: Path<CollectionPath>,
@@ -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<Dispatcher>,
collection: Path<CollectionPath>,
@@ -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,

8
src/actix/api/facet_api.rs
View File

@@ -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<Dispatcher>,
collection: Path<CollectionPath>,
@@ -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,

38
src/actix/api/local_shard_api.rs
View File

@@ -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<Dispatcher>,
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<Dispatcher>,
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<Dispatcher>,
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<NonZeroU64>,
}
#[post("/collections/{collection}/shards/{shard}/cleanup")]
#[post("/collections/{collection_name}/shards/{shard}/cleanup")]
async fn cleanup_shard(
dispatcher: web::Data<Dispatcher>,
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,
}

4
src/actix/api/mod.rs
View File

@@ -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,
}

27
src/actix/api/query_api.rs
View File

@@ -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<Dispatcher>,
@@ -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<Dispatcher>,
collection: Path<CollectionPath>,
@@ -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<Dispatcher>,
collection: Path<CollectionPath>,
@@ -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,

24
src/actix/api/recommend_api.rs
View File

@@ -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<Dispatcher>,
collection: Path<CollectionPath>,
@@ -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<Dispatcher>,
collection: Path<CollectionPath>,
@@ -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<Dispatcher>,
collection: Path<CollectionPath>,
@@ -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,

24
src/actix/api/retrieve_api.rs
View File

@@ -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<Dispatcher>,
collection: Path<CollectionPath>,
@@ -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<Dispatcher>,
collection: Path<CollectionPath>,
@@ -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<Dispatcher>,
collection: Path<CollectionPath>,
@@ -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(),

40
src/actix/api/search_api.rs
View File

@@ -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<Dispatcher>,
collection: Path<CollectionPath>,
@@ -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<Dispatcher>,
collection: Path<CollectionPath>,
@@ -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<Dispatcher>,
collection: Path<CollectionPath>,
@@ -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<Dispatcher>,
collection: Path<CollectionPath>,
@@ -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<Dispatcher>,
collection: Path<CollectionPath>,
@@ -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,

9
src/actix/api/service_api.rs
View File

@@ -33,6 +33,7 @@ use crate::tracing;
pub struct TelemetryParam {
pub anonymize: Option<bool>,
pub details_level: Option<usize>,
pub per_collection: Option<bool>,
#[validate(range(min = 1))]
pub timeout: Option<u64>,
}
@@ -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<bool>,
pub per_collection: Option<bool>,
#[validate(range(min = 1))]
pub timeout: Option<u64>,
}
@@ -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<bool>,
}
#[post("/collections/{name}/truncate_unapplied_wal")]
#[post("/collections/{collection_name}/truncate_unapplied_wal")]
async fn truncate_unapplied_wal(
dispatcher: web::Data<Dispatcher>,
collection: Path<CollectionPath>,
@@ -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();

12
src/actix/api/shards_api.rs
View File

@@ -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<Dispatcher>,
collection: Path<CollectionPath>,
@@ -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<Dispatcher>,
collection: Path<CollectionPath>,
@@ -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<Dispatcher>,
collection: Path<CollectionPath>,
@@ -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,

47
src/actix/api/snapshot_api.rs
View File

@@ -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<Dispatcher>,
path: web::Path<String>,
@@ -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<Dispatcher>,
path: web::Path<String>,
@@ -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<Dispatcher>,
http_client: web::Data<HttpClient>,
@@ -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<Dispatcher>,
http_client: web::Data<HttpClient>,
@@ -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<Dispatcher>,
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<Dispatcher>,
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<Dispatcher>,
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<Dispatcher>,
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<Dispatcher>,
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<Dispatcher>,
http_client: web::Data<HttpClient>,
@@ -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<Dispatcher>,
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<Dispatcher>,
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<Dispatcher>,
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<Dispatcher>,
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<Dispatcher>,
path: web::Path<(String, ShardId)>,
@@ -696,7 +699,7 @@ pub struct PartialSnapshotRecoverFrom {
api_key: Option<String>,
}
#[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<Dispatcher>,
http_client: web::Data<HttpClient>,
@@ -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<Dispatcher>,
path: web::Path<(String, ShardId)>,

68
src/actix/api/update_api.rs
View File

@@ -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<Dispatcher>,
@@ -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<Dispatcher>,
collection: Path<CollectionPath>,
@@ -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<Dispatcher>,
@@ -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<Dispatcher>,
collection: Path<CollectionPath>,
@@ -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<Dispatcher>,
collection: Path<CollectionPath>,
@@ -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<Dispatcher>,
collection: Path<CollectionPath>,
@@ -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<Dispatcher>,
collection: Path<CollectionPath>,
@@ -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<Dispatcher>,
collection: Path<CollectionPath>,
@@ -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<Dispatcher>,
collection: Path<CollectionPath>,
@@ -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<Dispatcher>,
collection: Path<CollectionPath>,
@@ -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<Dispatcher>,
collection: Path<CollectionPath>,
@@ -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<Dispatcher>,
collection: Path<CollectionPath>,
@@ -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);

364
src/common/metrics.rs
View File

@@ -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}"
);
}
}

364
src/common/telemetry_ops/requests_telemetry.rs
View File

@@ -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<StatusCode> =
HashMap<String, HashMap<String, HashMap<StatusCode, OperationDurationStatistics>>>;
type PerCollectionAggregators<StatusCode> =
HashMap<String, HashMap<String, HashMap<StatusCode, Arc<Mutex<OperationDurationsAggregator>>>>>;
#[derive(Serialize, Clone, Default, Debug, JsonSchema)]
pub struct WebApiTelemetry {
pub responses: HashMap<String, HashMap<HttpStatusCode, OperationDurationStatistics>>,
#[serde(skip_serializing_if = "HashMap::is_empty", default)]
pub per_collection_responses: PerCollectionResponses<HttpStatusCode>,
}
#[derive(Serialize, Clone, Default, Debug, JsonSchema)]
pub struct GrpcTelemetry {
pub responses: HashMap<String, HashMap<GrpcStatusCode, OperationDurationStatistics>>,
#[serde(skip_serializing_if = "HashMap::is_empty", default)]
pub per_collection_responses: PerCollectionResponses<GrpcStatusCode>,
}
pub struct ActixTelemetryCollector {
@@ -32,6 +47,8 @@ pub struct ActixTelemetryCollector {
#[derive(Default)]
pub struct ActixWorkerTelemetryCollector {
methods: HashMap<String, HashMap<HttpStatusCode, Arc<Mutex<OperationDurationsAggregator>>>>,
/// collection_name -> method -> status_code -> aggregator
per_collection_methods: PerCollectionAggregators<HttpStatusCode>,
}
pub struct TonicTelemetryCollector {
@@ -41,11 +58,13 @@ pub struct TonicTelemetryCollector {
#[derive(Default)]
pub struct TonicWorkerTelemetryCollector {
methods: HashMap<String, HashMap<GrpcStatusCode, Arc<Mutex<OperationDurationsAggregator>>>>,
/// collection_name -> method -> status_code -> aggregator
per_collection_methods: PerCollectionAggregators<GrpcStatusCode>,
}
impl ActixTelemetryCollector {
pub fn create_web_worker_telemetry(&mut self) -> Arc<Mutex<ActixWorkerTelemetryCollector>> {
let worker: Arc<Mutex<_>> = 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<Mutex<TonicWorkerTelemetryCollector>> {
let worker: Arc<Mutex<_>> = 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<String>,
) {
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<String>,
) {
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);
}
}

1
src/common/telemetry_reporting.rs
View File

@@ -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

1
src/tonic/api/mod.rs
View File

@@ -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;

20
src/tonic/api/points_api.rs
View File

@@ -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<Response<FacetResponse>, 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(),

5
src/tonic/api/points_internal_api.rs
View File

@@ -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<TableOfContent>,
service_config: ServiceConfig,

1
src/tonic/api/qdrant_internal_api.rs
View File

@@ -89,6 +89,7 @@ impl QdrantInternal for QdrantInternalService {
let detail = TelemetryDetail {
level: details_level,
histograms: false,
per_collection: false,
};
let only_collections =

640
src/tonic/api/telemetry_wrapper.rs Normal file
View File

@@ -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<T> {
inner: T,
}
impl<T> PointsTelemetryWrapper<T> {
pub fn new(inner: T) -> Self {
Self { inner }
}
}
#[tonic::async_trait]
impl<T: Points> Points for PointsTelemetryWrapper<T> {
async fn upsert(
&self,
request: Request<UpsertPoints>,
) -> Result<Response<PointsOperationResponse>, 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<DeletePoints>,
) -> Result<Response<PointsOperationResponse>, 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<GetPoints>) -> Result<Response<GetResponse>, 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<UpdatePointVectors>,
) -> Result<Response<PointsOperationResponse>, 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<DeletePointVectors>,
) -> Result<Response<PointsOperationResponse>, 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<SetPayloadPoints>,
) -> Result<Response<PointsOperationResponse>, 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<SetPayloadPoints>,
) -> Result<Response<PointsOperationResponse>, 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<DeletePayloadPoints>,
) -> Result<Response<PointsOperationResponse>, 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<ClearPayloadPoints>,
) -> Result<Response<PointsOperationResponse>, 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<UpdateBatchPoints>,
) -> Result<Response<UpdateBatchResponse>, 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<CreateFieldIndexCollection>,
) -> Result<Response<PointsOperationResponse>, 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<DeleteFieldIndexCollection>,
) -> Result<Response<PointsOperationResponse>, 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<SearchPoints>,
) -> Result<Response<SearchResponse>, 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<SearchBatchPoints>,
) -> Result<Response<SearchBatchResponse>, 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<SearchPointGroups>,
) -> Result<Response<SearchGroupsResponse>, 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<ScrollPoints>,
) -> Result<Response<ScrollResponse>, 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<RecommendPoints>,
) -> Result<Response<RecommendResponse>, 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<RecommendBatchPoints>,
) -> Result<Response<RecommendBatchResponse>, 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<RecommendPointGroups>,
) -> Result<Response<RecommendGroupsResponse>, 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<DiscoverPoints>,
) -> Result<Response<DiscoverResponse>, 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<DiscoverBatchPoints>,
) -> Result<Response<DiscoverBatchResponse>, 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<CountPoints>,
) -> Result<Response<CountResponse>, 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<QueryPoints>,
) -> Result<Response<QueryResponse>, 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<QueryBatchPoints>,
) -> Result<Response<QueryBatchResponse>, 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<QueryPointGroups>,
) -> Result<Response<QueryGroupsResponse>, 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<FacetCounts>,
) -> Result<Response<FacetResponse>, 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<SearchMatrixPoints>,
) -> Result<Response<SearchMatrixPairsResponse>, 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<SearchMatrixPoints>,
) -> Result<Response<SearchMatrixOffsetsResponse>, 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<T> {
inner: T,
}
impl<T> SnapshotsTelemetryWrapper<T> {
pub fn new(inner: T) -> Self {
Self { inner }
}
}
#[tonic::async_trait]
impl<T: Snapshots> Snapshots for SnapshotsTelemetryWrapper<T> {
async fn create(
&self,
request: Request<CreateSnapshotRequest>,
) -> Result<Response<CreateSnapshotResponse>, 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<ListSnapshotsRequest>,
) -> Result<Response<ListSnapshotsResponse>, 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<DeleteSnapshotRequest>,
) -> Result<Response<DeleteSnapshotResponse>, 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<CreateFullSnapshotRequest>,
) -> Result<Response<CreateSnapshotResponse>, Status> {
self.inner.create_full(request).await
}
async fn list_full(
&self,
request: Request<ListFullSnapshotsRequest>,
) -> Result<Response<ListSnapshotsResponse>, Status> {
self.inner.list_full(request).await
}
async fn delete_full(
&self,
request: Request<DeleteFullSnapshotRequest>,
) -> Result<Response<DeleteSnapshotResponse>, Status> {
self.inner.delete_full(request).await
}
}
/// Wraps a [`ShardSnapshots`] service, attaching `collection_name` to every response.
pub struct ShardSnapshotsTelemetryWrapper<T> {
inner: T,
}
impl<T> ShardSnapshotsTelemetryWrapper<T> {
pub fn new(inner: T) -> Self {
Self { inner }
}
}
#[tonic::async_trait]
impl<T: ShardSnapshots> ShardSnapshots for ShardSnapshotsTelemetryWrapper<T> {
async fn create(
&self,
request: Request<CreateShardSnapshotRequest>,
) -> Result<Response<CreateSnapshotResponse>, 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<ListShardSnapshotsRequest>,
) -> Result<Response<ListSnapshotsResponse>, 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<DeleteShardSnapshotRequest>,
) -> Result<Response<DeleteSnapshotResponse>, 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<RecoverShardSnapshotRequest>,
) -> Result<Response<RecoverSnapshotResponse>, 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<Response<$resp>, 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::<CollectionName>().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<Response<$resp>, Status> {
Ok(Response::new(Default::default()))
}
)*
$(
async fn $pt_method(&self, r: Request<$pt_req>) -> Result<Response<$pt_resp>, 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::<CollectionName>().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::<CollectionName>().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<Response<$resp>, 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::<CollectionName>().unwrap().0,
stringify!($method),
);
}
)*
}
};
}
mock_and_test_shard_snapshots! {
create(CreateShardSnapshotRequest) -> CreateSnapshotResponse,
list(ListShardSnapshotsRequest) -> ListSnapshotsResponse,
delete(DeleteShardSnapshotRequest) -> DeleteSnapshotResponse,
recover(RecoverShardSnapshotRequest) -> RecoverSnapshotResponse,
}
}

18
src/tonic/mod.rs
View File

@@ -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::<IpAddr>().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)

11
src/tonic/tonic_telemetry.rs
View File

@@ -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::<CollectionName>()
.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)
})
}

72
tests/openapi/test_service.py
View File

@@ -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:

81
tests/per_collection_metrics_test.sh Normal file
View File

@@ -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