From ee4cd768ead57770e117dca4cd8ca15da06cdf10 Mon Sep 17 00:00:00 2001 From: Egor Ivkov Date: Fri, 8 Apr 2022 14:42:38 +0300 Subject: [PATCH] Allows specifying collection meta operation commit timeout in API (#444) * Allows specifying collection meta operation commit timeout in API Also restructures a part of gRPC collection API * Review: link to dev docs * Review: Timeout error description Also committed generated protobuf code that somehow was left out * Review: seconds in API and minor proto fixes --- CONTRIBUTING.md | 2 +- docs/grpc/docs.md | 4 + docs/redoc/master/openapi.json | 34 ++++++ lib/api/src/grpc/proto/collections.proto | 4 + lib/api/src/grpc/qdrant.rs | 12 +++ .../src/content_manager/conversions.rs | 20 ++-- openapi/openapi-collections.ytt.yaml | 30 +++++- src/actix/api/collections_api.rs | 42 +++++++- src/tonic/api/collections_api.rs | 102 +++++++++--------- 9 files changed, 183 insertions(+), 67 deletions(-) diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 30441f08a0..ac5632b9b8 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -14,7 +14,7 @@ Pull requests are the best way to propose changes to the codebase (we use [Githu 1. Fork the repo and create your branch from `master`. 2. If you've added code that should be tested, add tests. -3. If you've changed APIs, update the documentation and API Schema definitions. +3. If you've changed APIs, update the documentation and API Schema definitions (see [development docs](https://github.com/qdrant/qdrant/blob/master/docs/DEVELOPMENT.md#api-changes)) 4. Ensure the test suite passes. 5. Make sure your code lints (with cargo). 6. Issue that pull request! diff --git a/docs/grpc/docs.md b/docs/grpc/docs.md index d6c0c414fa..14414c0d1f 100644 --- a/docs/grpc/docs.md +++ b/docs/grpc/docs.md @@ -133,6 +133,7 @@ | Field | Type | Label | Description | | ----- | ---- | ----- | ----------- | | actions | [AliasOperations](#qdrant-AliasOperations) | repeated | List of actions | +| timeout | [uint64](#uint64) | optional | Wait timeout for operation commit in seconds, if not specified - default value will be supplied | @@ -274,6 +275,7 @@ | wal_config | [WalConfigDiff](#qdrant-WalConfigDiff) | optional | Configuration of the Write-Ahead-Log | | optimizers_config | [OptimizersConfigDiff](#qdrant-OptimizersConfigDiff) | optional | Configuration of the optimizers | | shard_number | [uint32](#uint32) | optional | Number of shards in the collection, default = 1 | +| timeout | [uint64](#uint64) | optional | Wait timeout for operation commit in seconds, if not specified - default value will be supplied | @@ -304,6 +306,7 @@ | Field | Type | Label | Description | | ----- | ---- | ----- | ----------- | | collection_name | [string](#string) | | Name of the collection | +| timeout | [uint64](#uint64) | optional | Wait timeout for operation commit in seconds, if not specified - default value will be supplied | @@ -470,6 +473,7 @@ If indexation speed have more priority for your - make this parameter lower. If | ----- | ---- | ----- | ----------- | | collection_name | [string](#string) | | Name of the collection | | optimizers_config | [OptimizersConfigDiff](#qdrant-OptimizersConfigDiff) | optional | New configuration parameters for the collection | +| timeout | [uint64](#uint64) | optional | Wait timeout for operation commit in seconds, if not specified - default value will be supplied | diff --git a/docs/redoc/master/openapi.json b/docs/redoc/master/openapi.json index 9c43f4c135..aa0e85556a 100644 --- a/docs/redoc/master/openapi.json +++ b/docs/redoc/master/openapi.json @@ -2085,6 +2085,16 @@ "/collections/aliases": { "post": { "operationId": "update_aliases", + "parameters": [ + { + "description": "Wait for operation commit timeout in seconds. \nIf timeout is reached - request will return with service error.\n", + "in": "query", + "name": "timeout", + "schema": { + "type": "integer" + } + } + ], "requestBody": { "content": { "application/json": { @@ -2162,6 +2172,14 @@ "schema": { "type": "string" } + }, + { + "description": "Wait for operation commit timeout in seconds. \nIf timeout is reached - request will return with service error.\n", + "in": "query", + "name": "timeout", + "schema": { + "type": "integer" + } } ], "responses": { @@ -2296,6 +2314,14 @@ "schema": { "type": "string" } + }, + { + "description": "Wait for operation commit timeout in seconds. \nIf timeout is reached - request will return with service error.\n", + "in": "query", + "name": "timeout", + "schema": { + "type": "integer" + } } ], "requestBody": { @@ -2460,6 +2486,14 @@ "schema": { "type": "string" } + }, + { + "description": "Wait for operation commit timeout in seconds. \nIf timeout is reached - request will return with service error.\n", + "in": "query", + "name": "timeout", + "schema": { + "type": "integer" + } } ], "requestBody": { diff --git a/lib/api/src/grpc/proto/collections.proto b/lib/api/src/grpc/proto/collections.proto index 9b0ea154c7..d2fbc3eab5 100644 --- a/lib/api/src/grpc/proto/collections.proto +++ b/lib/api/src/grpc/proto/collections.proto @@ -137,15 +137,18 @@ message CreateCollection { optional WalConfigDiff wal_config = 5; // Configuration of the Write-Ahead-Log optional OptimizersConfigDiff optimizers_config = 6; // Configuration of the optimizers optional uint32 shard_number = 7; // Number of shards in the collection, default = 1 + optional uint64 timeout = 8; // Wait timeout for operation commit in seconds, if not specified - default value will be supplied } message UpdateCollection { string collection_name = 1; // Name of the collection optional OptimizersConfigDiff optimizers_config = 2; // New configuration parameters for the collection + optional uint64 timeout = 3; // Wait timeout for operation commit in seconds, if not specified - default value will be supplied } message DeleteCollection { string collection_name = 1; // Name of the collection + optional uint64 timeout = 2; // Wait timeout for operation commit in seconds, if not specified - default value will be supplied } message CollectionOperationResponse { @@ -183,6 +186,7 @@ message CollectionInfo { message ChangeAliases { repeated AliasOperations actions = 1; // List of actions + optional uint64 timeout = 2; // Wait timeout for operation commit in seconds, if not specified - default value will be supplied } message AliasOperations { diff --git a/lib/api/src/grpc/qdrant.rs b/lib/api/src/grpc/qdrant.rs index 19231e9892..fcff707f7c 100644 --- a/lib/api/src/grpc/qdrant.rs +++ b/lib/api/src/grpc/qdrant.rs @@ -138,6 +138,9 @@ pub struct CreateCollection { /// Number of shards in the collection, default = 1 #[prost(uint32, optional, tag="7")] pub shard_number: ::core::option::Option, + /// Wait timeout for operation commit in seconds, if not specified - default value will be supplied + #[prost(uint64, optional, tag="8")] + pub timeout: ::core::option::Option, } #[derive(Clone, PartialEq, ::prost::Message)] pub struct UpdateCollection { @@ -147,12 +150,18 @@ pub struct UpdateCollection { /// New configuration parameters for the collection #[prost(message, optional, tag="2")] pub optimizers_config: ::core::option::Option, + /// Wait timeout for operation commit in seconds, if not specified - default value will be supplied + #[prost(uint64, optional, tag="3")] + pub timeout: ::core::option::Option, } #[derive(Clone, PartialEq, ::prost::Message)] pub struct DeleteCollection { /// Name of the collection #[prost(string, tag="1")] pub collection_name: ::prost::alloc::string::String, + /// Wait timeout for operation commit in seconds, if not specified - default value will be supplied + #[prost(uint64, optional, tag="2")] + pub timeout: ::core::option::Option, } #[derive(Clone, PartialEq, ::prost::Message)] pub struct CollectionOperationResponse { @@ -228,6 +237,9 @@ pub struct ChangeAliases { /// List of actions #[prost(message, repeated, tag="1")] pub actions: ::prost::alloc::vec::Vec, + /// Wait timeout for operation commit in seconds, if not specified - default value will be supplied + #[prost(uint64, optional, tag="2")] + pub timeout: ::core::option::Option, } #[derive(Clone, PartialEq, ::prost::Message)] pub struct AliasOperations { diff --git a/lib/storage/src/content_manager/conversions.rs b/lib/storage/src/content_manager/conversions.rs index 20ec2aab2a..7146e533cb 100644 --- a/lib/storage/src/content_manager/conversions.rs +++ b/lib/storage/src/content_manager/conversions.rs @@ -43,20 +43,26 @@ impl TryFrom for CollectionMetaOperations { } } -impl From for CollectionMetaOperations { - fn from(value: api::grpc::qdrant::UpdateCollection) -> Self { - Self::UpdateCollection(UpdateCollectionOperation { +impl TryFrom for CollectionMetaOperations { + type Error = Status; + + fn try_from(value: api::grpc::qdrant::UpdateCollection) -> Result { + Ok(Self::UpdateCollection(UpdateCollectionOperation { collection_name: value.collection_name, update_collection: UpdateCollection { optimizers_config: value.optimizers_config.map(|v| v.into()), }, - }) + })) } } -impl From for CollectionMetaOperations { - fn from(value: api::grpc::qdrant::DeleteCollection) -> Self { - Self::DeleteCollection(DeleteCollectionOperation(value.collection_name)) +impl TryFrom for CollectionMetaOperations { + type Error = Status; + + fn try_from(value: api::grpc::qdrant::DeleteCollection) -> Result { + Ok(Self::DeleteCollection(DeleteCollectionOperation( + value.collection_name, + ))) } } diff --git a/openapi/openapi-collections.ytt.yaml b/openapi/openapi-collections.ytt.yaml index 89cff38974..f9c9d7c74b 100644 --- a/openapi/openapi-collections.ytt.yaml +++ b/openapi/openapi-collections.ytt.yaml @@ -61,6 +61,13 @@ paths: required: true schema: type: string + - name: timeout + in: query + description: | + Wait for operation commit timeout in seconds. + If timeout is reached - request will return with service error. + schema: + type: integer responses: #@ response(type("boolean")) patch: @@ -83,6 +90,13 @@ paths: required: true schema: type: string + - name: timeout + in: query + description: | + Wait for operation commit timeout in seconds. + If timeout is reached - request will return with service error. + schema: + type: integer responses: #@ response(type("boolean")) delete: @@ -98,6 +112,13 @@ paths: required: true schema: type: string + - name: timeout + in: query + description: | + Wait for operation commit timeout in seconds. + If timeout is reached - request will return with service error. + schema: + type: integer responses: #@ response(type("boolean")) /collections/aliases: @@ -112,7 +133,14 @@ paths: application/json: schema: $ref: "#/components/schemas/ChangeAliasesOperation" - + parameters: + - name: timeout + in: query + description: | + Wait for operation commit timeout in seconds. + If timeout is reached - request will return with service error. + schema: + type: integer responses: #@ response(type("boolean")) /collections/{collection_name}/index: diff --git a/src/actix/api/collections_api.rs b/src/actix/api/collections_api.rs index 9ad9f6a3e4..b8e8bfdae1 100644 --- a/src/actix/api/collections_api.rs +++ b/src/actix/api/collections_api.rs @@ -2,13 +2,26 @@ use crate::actix::helpers::process_response; use crate::common::collections::*; use actix_web::rt::time::Instant; use actix_web::{delete, get, patch, post, put, web, Responder}; +use serde::Deserialize; use std::sync::Arc; +use std::time::Duration; use storage::content_manager::collection_meta_ops::{ ChangeAliasesOperation, CollectionMetaOperations, CreateCollection, CreateCollectionOperation, DeleteCollectionOperation, UpdateCollection, UpdateCollectionOperation, }; use storage::content_manager::toc::TableOfContent; +#[derive(Debug, Deserialize)] +struct WaitTimeout { + timeout: Option, +} + +impl WaitTimeout { + pub fn timeout(&self) -> Option { + self.timeout.map(Duration::from_secs) + } +} + #[get("/collections")] async fn get_collections(toc: web::Data>) -> impl Responder { let timing = Instant::now(); @@ -43,6 +56,7 @@ async fn create_collection( toc: web::Data>, path: web::Path, operation: web::Json, + web::Query(query): web::Query, ) -> impl Responder { let timing = Instant::now(); let name = path.into_inner(); @@ -52,7 +66,7 @@ async fn create_collection( collection_name: name, create_collection: operation.0, }), - None, + query.timeout(), ) .await; process_response(response, timing) @@ -63,6 +77,7 @@ async fn update_collection( toc: web::Data>, path: web::Path, operation: web::Json, + web::Query(query): web::Query, ) -> impl Responder { let timing = Instant::now(); let name = path.into_inner(); @@ -72,7 +87,7 @@ async fn update_collection( collection_name: name, update_collection: operation.0, }), - None, + query.timeout(), ) .await; process_response(response, timing) @@ -82,13 +97,14 @@ async fn update_collection( async fn delete_collection( toc: web::Data>, path: web::Path, + web::Query(query): web::Query, ) -> impl Responder { let timing = Instant::now(); let name = path.into_inner(); let response = toc .submit_collection_operation( CollectionMetaOperations::DeleteCollection(DeleteCollectionOperation(name)), - None, + query.timeout(), ) .await; process_response(response, timing) @@ -98,10 +114,14 @@ async fn delete_collection( async fn update_aliases( toc: web::Data>, operation: web::Json, + web::Query(query): web::Query, ) -> impl Responder { let timing = Instant::now(); let response = toc - .submit_collection_operation(CollectionMetaOperations::ChangeAliases(operation.0), None) + .submit_collection_operation( + CollectionMetaOperations::ChangeAliases(operation.0), + query.timeout(), + ) .await; process_response(response, timing) } @@ -116,3 +136,17 @@ pub fn config_collections_api(cfg: &mut web::ServiceConfig) { .service(delete_collection) .service(update_aliases); } + +#[cfg(test)] +mod tests { + use super::WaitTimeout; + use actix_web::web::Query; + + #[test] + fn timeout_is_deserialized() { + let timeout: WaitTimeout = Query::from_query("").unwrap().0; + assert!(timeout.timeout.is_none()); + let timeout: WaitTimeout = Query::from_query("timeout=10").unwrap().0; + assert_eq!(timeout.timeout, Some(10)) + } +} diff --git a/src/tonic/api/collections_api.rs b/src/tonic/api/collections_api.rs index 577d7995ff..6e086fba4e 100644 --- a/src/tonic/api/collections_api.rs +++ b/src/tonic/api/collections_api.rs @@ -7,9 +7,8 @@ use api::grpc::qdrant::{ GetCollectionInfoRequest, GetCollectionInfoResponse, ListCollectionsRequest, ListCollectionsResponse, UpdateCollection, }; -use std::convert::TryFrom; use std::sync::Arc; -use std::time::Instant; +use std::time::{Duration, Instant}; use storage::content_manager::conversions::error_to_status; use storage::content_manager::toc::TableOfContent; @@ -22,6 +21,30 @@ impl CollectionsService { pub fn new(toc: Arc) -> Self { Self { toc } } + + async fn perform_operation( + &self, + request: Request, + ) -> Result, Status> + where + O: WithTimeout + + TryInto< + storage::content_manager::collection_meta_ops::CollectionMetaOperations, + Error = Status, + >, + { + let operation = request.into_inner(); + let wait_timeout = operation.wait_timeout(); + let timing = Instant::now(); + let result = self + .toc + .submit_collection_operation(operation.try_into()?, wait_timeout) + .await + .map_err(error_to_status)?; + + let response = CollectionOperationResponse::from((timing, result)); + Ok(Response::new(response)) + } } #[tonic::async_trait] @@ -58,75 +81,46 @@ impl Collections for CollectionsService { &self, request: Request, ) -> Result, Status> { - let operations = - storage::content_manager::collection_meta_ops::CollectionMetaOperations::try_from( - request.into_inner(), - )?; - let timing = Instant::now(); - let result = self - .toc - .submit_collection_operation(operations, None) - .await - .map_err(error_to_status)?; - - let response = CollectionOperationResponse::from((timing, result)); - Ok(Response::new(response)) + self.perform_operation(request).await } async fn update( &self, request: Request, ) -> Result, Status> { - let operations = - storage::content_manager::collection_meta_ops::CollectionMetaOperations::from( - request.into_inner(), - ); - let timing = Instant::now(); - let result = self - .toc - .submit_collection_operation(operations, None) - .await - .map_err(error_to_status)?; - - let response = CollectionOperationResponse::from((timing, result)); - Ok(Response::new(response)) + self.perform_operation(request).await } async fn delete( &self, request: Request, ) -> Result, Status> { - let operations = - storage::content_manager::collection_meta_ops::CollectionMetaOperations::from( - request.into_inner(), - ); - let timing = Instant::now(); - let result = self - .toc - .submit_collection_operation(operations, None) - .await - .map_err(error_to_status)?; - - let response = CollectionOperationResponse::from((timing, result)); - Ok(Response::new(response)) + self.perform_operation(request).await } async fn update_aliases( &self, request: Request, ) -> Result, Status> { - let operations = - storage::content_manager::collection_meta_ops::CollectionMetaOperations::try_from( - request.into_inner(), - )?; - let timing = Instant::now(); - let result = self - .toc - .submit_collection_operation(operations, None) - .await - .map_err(error_to_status)?; - - let response = CollectionOperationResponse::from((timing, result)); - Ok(Response::new(response)) + self.perform_operation(request).await } } + +trait WithTimeout { + fn wait_timeout(&self) -> Option; +} + +macro_rules! impl_with_timeout { + ($operation:ty) => { + impl WithTimeout for $operation { + fn wait_timeout(&self) -> Option { + self.timeout.map(Duration::from_secs) + } + } + }; +} + +impl_with_timeout!(CreateCollection); +impl_with_timeout!(UpdateCollection); +impl_with_timeout!(DeleteCollection); +impl_with_timeout!(ChangeAliases);