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

remove lock api (#7449)

Browse Source 
* remove lock api

* fix api count + remove tests

* remove another test

* remove more tests
This commit is contained in:
Andrey Vasnetsov
2025-10-29 17:36:54 +01:00
committed by timvisee
parent 783c128c1a
commit 2a423d1f11
16 changed files with 23 additions and 455 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -343,152 +343,6 @@
}
}
},
"/locks": {
"post": {
"summary": "Set lock options",
"description": "Set lock options. If write is locked, all write operations and collection creation are forbidden. Returns previous lock options",
"operationId": "post_locks",
"tags": [
"Service"
],
"deprecated": true,
"requestBody": {
"description": "Lock options and optional error message",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/LocksOption"
}
}
}
},
"responses": {
"default": {
"description": "error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ErrorResponse"
}
}
}
},
"4XX": {
"description": "error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ErrorResponse"
}
}
}
},
"200": {
"description": "successful operation",
"content": {
"application/json": {
"schema": {
"type": "object",
"properties": {
"usage": {
"default": null,
"anyOf": [
{
"$ref": "#/components/schemas/Usage"
},
{
"nullable": true
}
]
},
"time": {
"type": "number",
"format": "float",
"description": "Time spent to process this request",
"example": 0.002
},
"status": {
"type": "string",
"example": "ok"
},
"result": {
"$ref": "#/components/schemas/LocksOption"
}
}
}
}
}
}
}
},
"get": {
"summary": "Get lock options",
"description": "Get lock options. If write is locked, all write operations and collection creation are forbidden",
"operationId": "get_locks",
"tags": [
"Service"
],
"deprecated": true,
"responses": {
"default": {
"description": "error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ErrorResponse"
}
}
}
},
"4XX": {
"description": "error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ErrorResponse"
}
}
}
},
"200": {
"description": "successful operation",
"content": {
"application/json": {
"schema": {
"type": "object",
"properties": {
"usage": {
"default": null,
"anyOf": [
{
"$ref": "#/components/schemas/Usage"
},
{
"nullable": true
}
]
},
"time": {
"type": "number",
"format": "float",
"description": "Time spent to process this request",
"example": 0.002
},
"status": {
"type": "string",
"example": "ok"
},
"result": {
"$ref": "#/components/schemas/LocksOption"
}
}
}
}
}
}
}
}
},
"/healthz": {
"get": {
"summary": "Kubernetes healthz endpoint",
@@ -13824,21 +13678,6 @@
}
}
},
"LocksOption": {
"type": "object",
"required": [
"write"
],
"properties": {
"error_message": {
"type": "string",
"nullable": true
},
"write": {
"type": "boolean"
}
}
},
"SnapshotRecover": {
"type": "object",
"required": [

24
lib/shard/src/operations/mod.rs
View File

@@ -34,21 +34,6 @@ impl CollectionUpdateOperations {
)
}
pub fn is_write_operation(&self) -> bool {
match self {
CollectionUpdateOperations::PointOperation(operation) => operation.is_write_operation(),
CollectionUpdateOperations::VectorOperation(operation) => {
operation.is_write_operation()
}
CollectionUpdateOperations::PayloadOperation(operation) => {
operation.is_write_operation()
}
CollectionUpdateOperations::FieldIndexOperation(operation) => {
operation.is_write_operation()
}
}
}
pub fn point_ids(&self) -> Option<Vec<PointIdType>> {
match self {
Self::PointOperation(op) => op.point_ids(),
@@ -81,15 +66,6 @@ pub enum FieldIndexOperations {
DeleteIndex(JsonPath),
}
impl FieldIndexOperations {
pub fn is_write_operation(&self) -> bool {
match self {
FieldIndexOperations::CreateIndex(_) => true,
FieldIndexOperations::DeleteIndex(_) => false,
}
}
}
#[derive(Clone, Debug, PartialEq, Deserialize, Serialize, Hash)]
#[serde(rename_all = "snake_case")]
pub struct CreateIndex {

10
lib/shard/src/operations/payload_ops.rs
View File

@@ -27,16 +27,6 @@ pub enum PayloadOps {
}
impl PayloadOps {
pub fn is_write_operation(&self) -> bool {
match self {
PayloadOps::SetPayload(_) => true,
PayloadOps::DeletePayload(_) => false,
PayloadOps::ClearPayload { .. } => false,
PayloadOps::ClearPayloadByFilter(_) => false,
PayloadOps::OverwritePayload(_) => true,
}
}
pub fn point_ids(&self) -> Option<Vec<PointIdType>> {
match self {
Self::SetPayload(op) => op.points.clone(),

10
lib/shard/src/operations/point_ops.rs
View File

@@ -108,16 +108,6 @@ pub enum PointOperations {
}
impl PointOperations {
pub fn is_write_operation(&self) -> bool {
match self {
PointOperations::UpsertPoints(_) => true,
PointOperations::UpsertPointsConditional(_) => true,
PointOperations::DeletePoints { .. } => false,
PointOperations::DeletePointsByFilter(_) => false,
PointOperations::SyncPoints(_) => true,
}
}
pub fn point_ids(&self) -> Option<Vec<PointIdType>> {
match self {
Self::UpsertPoints(op) => Some(op.point_ids()),

8
lib/shard/src/operations/vector_ops.rs
View File

@@ -17,14 +17,6 @@ pub enum VectorOperations {
}
impl VectorOperations {
pub fn is_write_operation(&self) -> bool {
match self {
VectorOperations::UpdateVectors(_) => true,
VectorOperations::DeleteVectors(..) => false,
VectorOperations::DeleteVectorsByFilter(..) => false,
}
}
pub fn point_ids(&self) -> Option<Vec<PointIdType>> {
match self {
Self::UpdateVectors(op) => Some(op.points.iter().map(|point| point.id).collect()),

36
lib/storage/src/content_manager/toc/locks.rs
View File

@@ -1,36 +0,0 @@
use std::sync::atomic;
use super::TableOfContent;
use crate::content_manager::errors::StorageError;
pub const DEFAULT_WRITE_LOCK_ERROR_MESSAGE: &str = "Write operations are forbidden";
impl TableOfContent {
pub fn is_write_locked(&self) -> bool {
self.is_write_locked.load(atomic::Ordering::Relaxed)
}
pub fn get_lock_error_message(&self) -> Option<String> {
self.lock_error_message.lock().clone()
}
/// Returns an error if the write lock is set
pub fn check_write_lock(&self) -> Result<(), StorageError> {
if self.is_write_locked.load(atomic::Ordering::Relaxed) {
return Err(StorageError::Locked {
description: self
.lock_error_message
.lock()
.clone()
.unwrap_or_else(|| DEFAULT_WRITE_LOCK_ERROR_MESSAGE.to_string()),
});
}
Ok(())
}
pub fn set_locks(&self, is_write_locked: bool, error_message: Option<String>) {
self.is_write_locked
.store(is_write_locked, atomic::Ordering::Relaxed);
*self.lock_error_message.lock() = error_message;
}
}

6
lib/storage/src/content_manager/toc/mod.rs
View File

@@ -2,7 +2,6 @@ mod collection_container;
mod collection_meta_ops;
mod create_collection;
pub mod dispatcher;
mod locks;
mod point_ops;
mod point_ops_internal;
pub mod request_hw_counter;
@@ -16,7 +15,6 @@ use std::collections::{HashMap, HashSet};
use std::num::NonZeroU32;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use api::rest::models::HardwareUsage;
use collection::collection::{Collection, RequestShardTransfer};
@@ -73,8 +71,6 @@ pub struct TableOfContent {
consensus_proposal_sender: Option<OperationSender>,
/// Dispatcher for access to table of contents and consensus, if none - single node mode
toc_dispatcher: parking_lot::Mutex<Option<TocDispatcher>>,
is_write_locked: AtomicBool,
lock_error_message: parking_lot::Mutex<Option<String>>,
/// Prevent DDoS of too many concurrent updates in distributed mode.
/// One external update usually triggers multiple internal updates, which breaks internal
/// timings. For example, the health check timing and consensus timing.
@@ -199,8 +195,6 @@ impl TableOfContent {
channel_service,
consensus_proposal_sender,
toc_dispatcher: Default::default(),
is_write_locked: AtomicBool::new(false),
lock_error_message: parking_lot::Mutex::new(None),
update_rate_limiter: rate_limiter,
collection_create_lock: Default::default(),
collection_hw_metrics: DashMap::new(),

4
lib/storage/src/content_manager/toc/point_ops.rs
View File

@@ -534,10 +534,6 @@ impl TableOfContent {
None => None,
};
if operation.operation.is_write_operation() {
self.check_write_lock()?;
}
// TODO: `debug_assert(operation.clock_tag.is_none())` for `_update_shard_keys`/`update_from_client`!?
let res = match shard_selector {

5
lib/storage/src/dispatcher.rs
View File

@@ -83,7 +83,6 @@ impl Dispatcher {
let op = match operation {
CollectionMetaOperations::CreateCollection(mut op) => {
self.toc.check_write_lock()?;
if !op.is_distribution_set() {
match op.create_collection.sharding_method.unwrap_or_default() {
ShardingMethod::Auto => {
@@ -136,7 +135,6 @@ impl Dispatcher {
CollectionMetaOperations::CreateCollection(op)
}
CollectionMetaOperations::CreateShardKey(op) => {
self.toc.check_write_lock()?;
CollectionMetaOperations::CreateShardKey(op)
}
@@ -246,9 +244,6 @@ impl Dispatcher {
Ok(res)
} else {
if let CollectionMetaOperations::CreateCollection(_) = &operation {
self.toc.check_write_lock()?;
}
self.toc.perform_collection_meta_op(operation).await
}
}

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

@@ -75,31 +75,6 @@ paths:
"4XX":
description: error
/locks:
post:
summary: Set lock options
description: Set lock options. If write is locked, all write operations and collection creation are forbidden. Returns previous lock options
operationId: post_locks
tags:
- Service
deprecated: true #! Deprecated since Qdrant 1.15.0
requestBody:
description: Lock options and optional error message
content:
application/json:
schema:
$ref: "#/components/schemas/LocksOption"
responses: #@ response(reference("LocksOption"))
get:
summary: Get lock options
description: Get lock options. If write is locked, all write operations and collection creation are forbidden
operationId: get_locks
tags:
- Service
deprecated: true #! Deprecated since Qdrant 1.15.0
responses: #@ response(reference("LocksOption"))
/healthz:
get:
summary: Kubernetes healthz endpoint

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

@@ -6,21 +6,17 @@ use actix_web::http::header::ContentType;
use actix_web::rt::time::Instant;
use actix_web::web::Query;
use actix_web::{HttpResponse, Responder, get, post, web};
use actix_web_validator::Json;
use collection::operations::verification::new_unchecked_verification_pass;
use common::types::{DetailsLevel, TelemetryDetail};
use schemars::JsonSchema;
use segment::common::anonymize::Anonymize;
use serde::{Deserialize, Serialize};
use storage::content_manager::errors::StorageError;
use storage::dispatcher::Dispatcher;
use storage::rbac::AccessRequirements;
use tokio::sync::Mutex;
use crate::actix::auth::ActixAccess;
use crate::actix::helpers::{self, process_response_error};
use crate::common::health;
use crate::common::helpers::LocksOption;
use crate::common::metrics::MetricsData;
use crate::common::stacktrace::get_stack_trace;
use crate::common::telemetry::TelemetryCollector;
@@ -95,46 +91,6 @@ async fn metrics(
.body(MetricsData::from(telemetry_data).format_metrics())
}
#[post("/locks")]
fn put_locks(
dispatcher: web::Data<Dispatcher>,
locks_option: Json<LocksOption>,
ActixAccess(access): ActixAccess,
) -> impl Future<Output = HttpResponse> {
// Not a collection level request.
let pass = new_unchecked_verification_pass();
helpers::time(async move {
let toc = dispatcher.toc(&access, &pass);
access.check_global_access(AccessRequirements::new().manage())?;
let result = LocksOption {
write: toc.is_write_locked(),
error_message: toc.get_lock_error_message(),
};
toc.set_locks(locks_option.write, locks_option.error_message.clone());
Ok(result)
})
}
#[get("/locks")]
fn get_locks(
dispatcher: web::Data<Dispatcher>,
ActixAccess(access): ActixAccess,
) -> impl Future<Output = HttpResponse> {
// Not a collection level request.
let pass = new_unchecked_verification_pass();
helpers::time(async move {
access.check_global_access(AccessRequirements::new())?;
let toc = dispatcher.toc(&access, &pass);
let result = LocksOption {
write: toc.is_write_locked(),
error_message: toc.get_lock_error_message(),
};
Ok(result)
})
}
#[get("/stacktrace")]
fn get_stacktrace(ActixAccess(access): ActixAccess) -> impl Future<Output = HttpResponse> {
helpers::time(async move {
@@ -205,8 +161,6 @@ async fn update_logger_config(
pub fn config_service_api(cfg: &mut web::ServiceConfig) {
cfg.service(telemetry)
.service(metrics)
.service(put_locks)
.service(get_locks)
.service(get_stacktrace)
.service(healthz)
.service(livez)

9
src/common/helpers.rs
View File

@@ -3,21 +3,12 @@ use std::io;
use std::sync::atomic::{AtomicUsize, Ordering};
use fs_err as fs;
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use tokio::runtime;
use tokio::runtime::Runtime;
use tonic::transport::{Certificate, ClientTlsConfig, Identity, ServerTlsConfig};
use validator::Validate;
use crate::settings::{Settings, TlsConfig};
#[derive(Debug, Deserialize, Serialize, JsonSchema, Validate)]
pub struct LocksOption {
pub error_message: Option<String>,
pub write: bool,
}
pub fn create_search_runtime(max_search_threads: usize) -> io::Result<Runtime> {
let num_threads = common::defaults::search_thread_count(max_search_threads);
runtime::Builder::new_multi_thread()

2
src/schema_generator.rs
View File

@@ -30,7 +30,6 @@ use storage::content_manager::collection_meta_ops::{
};
use storage::types::ClusterStatus;
use crate::common::helpers::LocksOption;
use crate::common::telemetry::TelemetryData;
use crate::common::update::{CreateFieldIndex, UpdateOperations};
@@ -70,7 +69,6 @@ struct AllDefinitions {
ar: ClusterOperations,
at: SearchRequestBatch,
au: RecommendRequestBatch,
av: LocksOption,
aw: SnapshotRecover,
ax: CollectionsAliasesResponse,
ay: AliasDescription,

54
tests/consensus_tests/auth_tests/test_jwt_access.py
View File

@@ -591,8 +591,6 @@ ACTION_ACCESS = {
"livez": EndpointAccess(True, True, True, "GET /livez", "grpc.health.v1.Health/Check"),
"telemetry": EndpointAccess(True, True, True, "GET /telemetry"),
"metrics": EndpointAccess(True, False, True, "GET /metrics", coll_r=False),
"post_locks": EndpointAccess(False, False, True, "POST /locks"),
"get_locks": EndpointAccess(True, False, True, "GET /locks", coll_r=False),
"get_issues": EndpointAccess(True, True, True, "GET /issues"),
"clear_issues": EndpointAccess(False, False, True, "DELETE /issues"),
}
@@ -604,7 +602,7 @@ def test_all_actions_have_tests():
# a test_{action_name} exists in this file
test_name = f"test_{action_name}"
assert (
test_name in globals()
test_name in globals()
), f"An action is not tested: `{test_name}` was not found in this file"
@@ -624,7 +622,7 @@ def test_all_rest_endpoints_are_covered():
covered_endpoints = set(v.rest_endpoint for v in ACTION_ACCESS.values())
for endpoint in endpoint_paths:
assert (
endpoint in covered_endpoints
endpoint in covered_endpoints
), f"REST endpoint `{endpoint}` not found in any of the `ACTION_ACCESS` REST endpoints"
@@ -654,18 +652,18 @@ def test_all_grpc_endpoints_are_covered():
for method in service.method_names:
grpc_endpoint = f"{service_name}/{method}"
assert (
grpc_endpoint in covered_endpoints
grpc_endpoint in covered_endpoints
), f"gRPC endpoint `{grpc_endpoint}` not found in ACTION_ACCESS gRPC endpoints"
def check_rest_access(
method: str,
path: str,
body: Optional[Union[dict, Callable[[], dict]]],
should_succeed: bool,
token: str,
path_params: dict = {},
request_kwargs: dict = {},
method: str,
path: str,
body: Optional[Union[dict, Callable[[], dict]]],
should_succeed: bool,
token: str,
path_params: dict = {},
request_kwargs: dict = {},
):
if isfunction(body):
body = body()
@@ -687,11 +685,11 @@ def check_rest_access(
if should_succeed:
assert (
res.status_code < 500 and res.status_code != 403
res.status_code < 500 and res.status_code != 403
), f"{method} {path} failed with {res.status_code}: {res.text}"
else:
assert (
res.status_code == 403
res.status_code == 403
), f"{method} {path} should've gotten `403` status code, but got `{res.status_code}: {res.text}`"
except requests.exceptions.ConnectionError as e:
@@ -703,11 +701,11 @@ def check_rest_access(
def check_grpc_access(
client: grpc_requests.Client,
service: str,
method: str,
request: Optional[dict],
should_succeed: bool,
client: grpc_requests.Client,
service: str,
method: str,
request: Optional[dict],
should_succeed: bool,
):
if isfunction(request):
request = request()
@@ -720,7 +718,7 @@ def check_grpc_access(
pytest.fail(f"{service}/{method} failed with {e.code()}: {e.details()}")
else:
assert (
e.code() == grpc.StatusCode.PERMISSION_DENIED
e.code() == grpc.StatusCode.PERMISSION_DENIED
), f"{service}/{method} should've gotten `PERMISSION_DENIED` status code, but got `{e.code()}: {e.details()}`"
@@ -769,7 +767,7 @@ def get_auth_grpc_clients() -> GrpcClients:
def check_access(
action_name: str, rest_request=None, grpc_request=None, path_params={}, rest_req_kwargs={}
action_name: str, rest_request=None, grpc_request=None, path_params={}, rest_req_kwargs={}
):
action_access: EndpointAccess = ACTION_ACCESS[action_name]
@@ -1788,7 +1786,7 @@ def test_query_points():
"query": {
"nearest": {
"dense": {
"data": [0.1,0.2,0.3,0.4]
"data": [0.1, 0.2, 0.3, 0.4]
}
}
},
@@ -1802,9 +1800,9 @@ def test_query_batch_points():
rest_request={"searches": [{"query": [0.1, 0.2, 0.3, 0.4]}]},
path_params={"collection_name": COLL_NAME},
grpc_request={
"collection_name": COLL_NAME,
"collection_name": COLL_NAME,
"query_points": [
{
{
"query": {
"nearest": {
"dense": {
@@ -1900,14 +1898,6 @@ def test_metrics():
check_access("metrics")
def test_post_locks():
check_access("post_locks", rest_request={"write": False})
def test_get_locks():
check_access("get_locks")
def test_get_issues():
check_access("get_issues")

76
tests/openapi/test_db_lock.py
View File

@@ -1,76 +0,0 @@
import pytest
from .helpers.collection_setup import basic_collection_setup, drop_collection
from .helpers.helpers import request_with_validation
@pytest.fixture(autouse=True)
def setup(on_disk_vectors, collection_name):
basic_collection_setup(collection_name=collection_name, on_disk_vectors=on_disk_vectors)
yield
drop_collection(collection_name=collection_name)
def test_lock_db_for_writes(collection_name):
response = request_with_validation(
api='/locks',
method="POST",
path_params={},
body={
"write": True,
"error_message": "integration test"
}
)
assert response.ok
response = request_with_validation(
api='/locks',
method="GET",
)
assert response.ok
response = request_with_validation(
api='/collections/{collection_name}/points',
method="PUT",
path_params={'collection_name': collection_name},
query_params={'wait': 'true'},
body={
"points": [
{
"id": 10,
"vector": [0.05, -0.61, -0.76, 0.74],
"payload": {"city": "Gdansk"}
}
]
}
)
assert not response.ok
assert "integration test" in response.text
response = request_with_validation(
api='/locks',
method="POST",
path_params={},
body={
"write": False,
}
)
assert response.ok
response = request_with_validation(
api='/collections/{collection_name}/points',
method="PUT",
path_params={'collection_name': collection_name},
query_params={'wait': 'true'},
body={
"points": [
{
"id": 10,
"vector": [0.05, -0.61, -0.76, 0.74],
"payload": {"city": "Gdansk"}
}
]
}
)
assert response.ok

2
tests/openapi_consistency_check.sh
View File

@@ -37,7 +37,7 @@ rm -f ./docs/redoc/master/.diff.openapi.json
NUMBER_OF_APIS=$(cat ./docs/redoc/master/openapi.json | jq '[.paths[] | length] | add')
EXPECTED_NUMBER_OF_APIS=71
EXPECTED_NUMBER_OF_APIS=69
if [ "$NUMBER_OF_APIS" -ne "$EXPECTED_NUMBER_OF_APIS" ]; then
echo "ERROR: It looks like the total number of APIs has changed."