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

Log slow operations during local shard WAL recovery (#9282)

Browse Source 
* feat: log slow operations during local shard WAL recovery

Warn when applying a single WAL operation during recovery of a local
shard takes longer than 30s, including the operation type (e.g.
PointOperation::UpsertPoints) so slow recoveries can be diagnosed.

Adds CollectionUpdateOperations::label() returning a human-readable
label including the inner variant.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* refactor: reuse audit operation names for slow WAL recovery logging

Move the canonical operation-name mapping next to the
CollectionUpdateOperations definition in the shard crate as an inherent
operation_name() method, and have the audit AuditableOperation impl
delegate to it. The slow WAL recovery warning now reuses these same
names (e.g. upsert_points) instead of a duplicated label mapping.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Andrey Vasnetsov
2026-06-03 10:44:32 +02:00
committed by timvisee
parent 05d87a6d50
commit 634d6c5bee
5 changed files with 65 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
lib/collection/src/shards/local_shard/mod.rs
View File

@@ -92,6 +92,9 @@ use crate::wal_delta::RecoverableWal;
/// If rendering WAL load progression in basic text form, report progression every 60 seconds.
const WAL_LOAD_REPORT_EVERY: Duration = Duration::from_secs(60);
/// Log a warning if applying a single WAL operation during recovery takes longer than this.
const WAL_SLOW_OP_REPORT_THRESHOLD: Duration = Duration::from_secs(30);
/// LocalShard
///
/// LocalShard is an entity that can be moved between peers and contains some part of one collections data.
@@ -788,6 +791,11 @@ impl LocalShard {
newest_clocks.advance_clock(clock_tag);
}
// Capture the operation type before `update.operation` is moved, so we can
// report it if applying this operation turns out to be slow.
let op_name = update.operation.operation_name();
let op_started = Instant::now();
// Propagate `CollectionError::ServiceError`, but skip other error types.
match &CollectionUpdater::update(
&self.segments,
@@ -822,6 +830,15 @@ impl LocalShard {
Ok(_) => (),
}
let op_elapsed = op_started.elapsed();
if op_elapsed >= WAL_SLOW_OP_REPORT_THRESHOLD {
log::warn!(
"Slow WAL operation during recovery: {op_name} took {op_elapsed:.2?}, \
collection: {collection_id}, \
op_num: {op_num}"
);
}
// Update progress bar or show text progress every WAL_LOAD_REPORT_EVERY
bar.inc(1);
if !show_progress_bar && last_progress_report.elapsed() >= WAL_LOAD_REPORT_EVERY {

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

@@ -1,3 +1,4 @@
pub mod operation_name;
pub mod optimization;
pub mod payload_ops;
pub mod point_ops;

44
lib/shard/src/operations/operation_name.rs Normal file
View File

@@ -0,0 +1,44 @@
use crate::operations::payload_ops::PayloadOps;
use crate::operations::point_ops::PointOperations;
use crate::operations::vector_name_ops::VectorNameOperations;
use crate::operations::vector_ops::VectorOperations;
use crate::operations::{CollectionUpdateOperations, FieldIndexOperations};
impl CollectionUpdateOperations {
/// A stable, human-readable name for the operation, including its inner variant.
///
/// Used for audit logging and for reporting slow operations during WAL recovery.
pub fn operation_name(&self) -> &'static str {
match self {
CollectionUpdateOperations::PointOperation(op) => match op {
PointOperations::UpsertPoints(_) => "upsert_points",
PointOperations::UpsertPointsConditional(_) => "upsert_points_conditional",
PointOperations::DeletePoints { .. } => "delete_points",
PointOperations::DeletePointsByFilter(_) => "delete_points_by_filter",
PointOperations::SyncPoints(_) => "sync_points",
},
CollectionUpdateOperations::VectorOperation(op) => match op {
VectorOperations::UpdateVectors(_) => "update_vectors",
VectorOperations::DeleteVectors(_, _) => "delete_vectors",
VectorOperations::DeleteVectorsByFilter(_, _) => "delete_vectors_by_filter",
},
CollectionUpdateOperations::PayloadOperation(op) => match op {
PayloadOps::SetPayload(_) => "set_payload",
PayloadOps::DeletePayload(_) => "delete_payload",
PayloadOps::ClearPayload { .. } => "clear_payload",
PayloadOps::ClearPayloadByFilter(_) => "clear_payload_by_filter",
PayloadOps::OverwritePayload(_) => "overwrite_payload",
},
CollectionUpdateOperations::FieldIndexOperation(op) => match op {
FieldIndexOperations::CreateIndex(_) => "create_field_index",
FieldIndexOperations::DeleteIndex(_) => "delete_field_index",
},
CollectionUpdateOperations::VectorNameOperation(op) => match op {
VectorNameOperations::CreateVectorName(_) => "create_vector_name",
VectorNameOperations::DeleteVectorName(_) => "delete_vector_name",
},
#[cfg(feature = "staging")]
CollectionUpdateOperations::StagingOperation(_) => "debug",
}
}
}

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

@@ -27,7 +27,6 @@ use shard::search::CoreSearchRequestBatch;
use super::TableOfContent;
use crate::content_manager::errors::{StorageError, StorageResult};
use crate::rbac::Auth;
use crate::rbac::auditable_operation::AuditableOperation;
impl TableOfContent {
/// Recommend points using positive and negative example from the request

38
lib/storage/src/rbac/auditable_operation.rs
View File

@@ -1,7 +1,4 @@
use shard::operations::payload_ops::PayloadOps;
use shard::operations::point_ops::PointOperations;
use shard::operations::vector_ops::VectorOperations;
use shard::operations::{CollectionUpdateOperations, FieldIndexOperations, VectorNameOperations};
use shard::operations::CollectionUpdateOperations;
use crate::content_manager::collection_meta_ops::CollectionMetaOperations;
@@ -11,37 +8,8 @@ pub trait AuditableOperation {
impl AuditableOperation for CollectionUpdateOperations {
fn operation_name(&self) -> &'static str {
match self {
CollectionUpdateOperations::PointOperation(op) => match op {
PointOperations::UpsertPoints(_) => "upsert_points",
PointOperations::UpsertPointsConditional(_) => "upsert_points_conditional",
PointOperations::DeletePoints { .. } => "delete_points",
PointOperations::DeletePointsByFilter(_) => "delete_points_by_filter",
PointOperations::SyncPoints(_) => "sync_points",
},
CollectionUpdateOperations::VectorOperation(op) => match op {
VectorOperations::UpdateVectors(_) => "update_vectors",
VectorOperations::DeleteVectors(_, _) => "delete_vectors",
VectorOperations::DeleteVectorsByFilter(_, _) => "delete_vectors_by_filter",
},
CollectionUpdateOperations::PayloadOperation(op) => match op {
PayloadOps::SetPayload(_) => "set_payload",
PayloadOps::DeletePayload(_) => "delete_payload",
PayloadOps::ClearPayload { .. } => "clear_payload",
PayloadOps::ClearPayloadByFilter(_) => "clear_payload_by_filter",
PayloadOps::OverwritePayload(_) => "overwrite_payload",
},
CollectionUpdateOperations::FieldIndexOperation(op) => match op {
FieldIndexOperations::CreateIndex(_) => "create_field_index",
FieldIndexOperations::DeleteIndex(_) => "delete_field_index",
},
CollectionUpdateOperations::VectorNameOperation(op) => match op {
VectorNameOperations::CreateVectorName(_) => "create_vector_name",
VectorNameOperations::DeleteVectorName(_) => "delete_vector_name",
},
#[cfg(feature = "staging")]
CollectionUpdateOperations::StagingOperation(_) => "debug",
}
// Delegate to the canonical mapping defined next to the enum in the `shard` crate.
CollectionUpdateOperations::operation_name(self)
}
}