Adding more logs to diagnostic shard truncation problems. (#4635)

quickwit/Cargo.toml

@@ -154,7 +154,7 @@ matches = "0.1.9"
 md5 = "0.7"
 mime_guess = "2.0.4"
 mockall = "0.11"
-mrecordlog = { git = "https://github.com/quickwit-oss/mrecordlog", rev = "2914cad" }
+mrecordlog = { git = "https://github.com/quickwit-oss/mrecordlog", rev = "bc6a998" }
 new_string_template = "1.4.0"
 nom = "7.1.3"
 num_cpus = "1"

quickwit/quickwit-indexing/src/models/shard_positions.rs

@@ -29,7 +29,7 @@ use quickwit_cluster::{Cluster, ListenerHandle};
 use quickwit_common::pubsub::{Event, EventBroker};
 use quickwit_proto::indexing::ShardPositionsUpdate;
 use quickwit_proto::types::{Position, ShardId, SourceUid};
-use tracing::{error, info, warn};
+use tracing::{debug, error, info, warn};
 
 /// Prefix used in chitchat to publish the shard positions.
 const SHARD_POSITIONS_PREFIX: &str = "indexer.shard_positions:";
@@ -223,6 +223,7 @@ impl Handler<ClusterShardPositionsUpdate> for ShardPositionsService {
             shard_positions,
         } = update;
         let updated_shard_positions = self.apply_update(&source_uid, shard_positions);
+        debug!(updated_shard_positions=?updated_shard_positions, "cluster position update");
         if !updated_shard_positions.is_empty() {
             self.publish_shard_updates_to_event_broker(source_uid, updated_shard_positions);
         }
@@ -276,6 +277,7 @@ impl ShardPositionsService {
         source_uid: SourceUid,
         shard_positions: Vec<(ShardId, Position)>,
     ) {
+        debug!(shard_positions=?shard_positions, "shard positions updates");
         self.event_broker.publish(ShardPositionsUpdate {
             source_uid,
             updated_shard_positions: shard_positions,

quickwit/quickwit-ingest/src/ingest_v2/ingester.rs

@@ -1098,22 +1098,25 @@ impl IngesterService for Ingester {
 impl EventSubscriber<ShardPositionsUpdate> for WeakIngesterState {
     async fn handle_event(&mut self, shard_positions_update: ShardPositionsUpdate) {
         let Some(state) = self.upgrade() else {
+            warn!("ingester state update failed");
             return;
         };
         let Ok(mut state_guard) =
             with_lock_metrics!(state.lock_fully().await, "gc_shards", "write")
         else {
+            error!("failed to lock the ingester state");
             return;
         };
         let index_uid = shard_positions_update.source_uid.index_uid;
         let source_id = shard_positions_update.source_uid.source_id;
 
         for (shard_id, shard_position) in shard_positions_update.updated_shard_positions {
             let queue_id = queue_id(&index_uid, &source_id, &shard_id);
-
             if shard_position.is_eof() {
+                info!(shard = queue_id, "deleting shard");
                 state_guard.delete_shard(&queue_id).await;
             } else {
+                info!(shard=queue_id, shard_position=%shard_position, "truncating shard");
                 state_guard.truncate_shard(&queue_id, &shard_position).await;
             }
         }