upstream-archive/qdrant
1
0
Fork 0
mirror of https://github.com/qdrant/qdrant.git synced 2026-08-04 09:01:00 -05:00
Code Issues Packages Projects Releases Wiki Activity

Allows specifying collection meta operation commit timeout in API (#444)

Browse Source 
* 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
This commit is contained in:
Egor Ivkov
2022-04-08 14:42:38 +03:00
committed by GitHub
parent 4d320b8585
commit ee4cd768ea
9 changed files with 183 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
CONTRIBUTING.md
View File

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

4
docs/grpc/docs.md
View File

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

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

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

4
lib/api/src/grpc/proto/collections.proto
View File

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

12
lib/api/src/grpc/qdrant.rs
View File

@@ -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<u32>,
/// 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<u64>,
}
#[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<OptimizersConfigDiff>,
/// 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<u64>,
}
#[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<u64>,
}
#[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<AliasOperations>,
/// 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<u64>,
}
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct AliasOperations {

20
lib/storage/src/content_manager/conversions.rs
View File

@@ -43,20 +43,26 @@ impl TryFrom<api::grpc::qdrant::CreateCollection> for CollectionMetaOperations {
}
}
impl From<api::grpc::qdrant::UpdateCollection> for CollectionMetaOperations {
fn from(value: api::grpc::qdrant::UpdateCollection) -> Self {
Self::UpdateCollection(UpdateCollectionOperation {
impl TryFrom<api::grpc::qdrant::UpdateCollection> for CollectionMetaOperations {
type Error = Status;
fn try_from(value: api::grpc::qdrant::UpdateCollection) -> Result<Self, Self::Error> {
Ok(Self::UpdateCollection(UpdateCollectionOperation {
collection_name: value.collection_name,
update_collection: UpdateCollection {
optimizers_config: value.optimizers_config.map(|v| v.into()),
},
})
}))
}
}
impl From<api::grpc::qdrant::DeleteCollection> for CollectionMetaOperations {
fn from(value: api::grpc::qdrant::DeleteCollection) -> Self {
Self::DeleteCollection(DeleteCollectionOperation(value.collection_name))
impl TryFrom<api::grpc::qdrant::DeleteCollection> for CollectionMetaOperations {
type Error = Status;
fn try_from(value: api::grpc::qdrant::DeleteCollection) -> Result<Self, Self::Error> {
Ok(Self::DeleteCollection(DeleteCollectionOperation(
value.collection_name,
)))
}
}

30
openapi/openapi-collections.ytt.yaml
View File

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

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

@@ -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<u64>,
}
impl WaitTimeout {
pub fn timeout(&self) -> Option<Duration> {
self.timeout.map(Duration::from_secs)
}
}
#[get("/collections")]
async fn get_collections(toc: web::Data<Arc<TableOfContent>>) -> impl Responder {
let timing = Instant::now();
@@ -43,6 +56,7 @@ async fn create_collection(
toc: web::Data<Arc<TableOfContent>>,
path: web::Path<String>,
operation: web::Json<CreateCollection>,
web::Query(query): web::Query<WaitTimeout>,
) -> 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<Arc<TableOfContent>>,
path: web::Path<String>,
operation: web::Json<UpdateCollection>,
web::Query(query): web::Query<WaitTimeout>,
) -> 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<Arc<TableOfContent>>,
path: web::Path<String>,
web::Query(query): web::Query<WaitTimeout>,
) -> 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<Arc<TableOfContent>>,
operation: web::Json<ChangeAliasesOperation>,
web::Query(query): web::Query<WaitTimeout>,
) -> 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))
}
}

102
src/tonic/api/collections_api.rs
View File

@@ -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<TableOfContent>) -> Self {
Self { toc }
}
async fn perform_operation<O>(
&self,
request: Request<O>,
) -> Result<Response<CollectionOperationResponse>, 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<CreateCollection>,
) -> Result<Response<CollectionOperationResponse>, 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<UpdateCollection>,
) -> Result<Response<CollectionOperationResponse>, 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<DeleteCollection>,
) -> Result<Response<CollectionOperationResponse>, 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<ChangeAliases>,
) -> Result<Response<CollectionOperationResponse>, 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<Duration>;
}
macro_rules! impl_with_timeout {
($operation:ty) => {
impl WithTimeout for $operation {
fn wait_timeout(&self) -> Option<Duration> {
self.timeout.map(Duration::from_secs)
}
}
};
}
impl_with_timeout!(CreateCollection);
impl_with_timeout!(UpdateCollection);
impl_with_timeout!(DeleteCollection);
impl_with_timeout!(ChangeAliases);