Merge branch 'main' into matthew.kim/metrics-wide-schema

Author: mattmkim

.github/workflows/ci.yml

@@ -143,7 +143,7 @@ jobs:
         if: always() && steps.modified.outputs.rust_src == 'true'
         run: |
           sudo apt-get update
-          sudo apt-get -y install protobuf-compiler
+          sudo apt-get -y install protobuf-compiler libcurl4-openssl-dev
       - name: Setup nightly Rust Toolchain (for rustfmt)
         if: steps.modified.outputs.rust_src == 'true'
         uses: dtolnay/rust-toolchain@efa25f7f19611383d5b0ccf2d1c8914531636bf9 # master

.github/workflows/coverage.yml

@@ -117,6 +117,7 @@ jobs:
           sudo apt update
           sudo apt install libsasl2-dev
           sudo apt install libsasl2-2
+          sudo apt install libcurl4-openssl-dev
 
       - uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405 # v.6.2.0
         with:

quickwit/quickwit-cluster/src/member.rs

@@ -86,7 +86,7 @@ impl NodeStateExt for NodeState {
     fn ingester_status(&self) -> IngesterStatus {
         self.get(INGESTER_STATUS_KEY)
             .and_then(IngesterStatus::from_json_str_name)
-            .unwrap_or_default()
+            .unwrap_or(IngesterStatus::Ready)
     }
 
     fn availability_zone(&self) -> Option<String> {
@@ -221,3 +221,17 @@ fn parse_enabled_services_str(
     }
     enabled_services
 }
+
+#[cfg(test)]
+mod tests {
+    use chitchat::NodeState;
+    use quickwit_proto::ingest::ingester::IngesterStatus;
+
+    use super::NodeStateExt;
+
+    #[test]
+    fn test_ingester_status_defaults_to_ready_when_key_absent() {
+        let node_state = NodeState::for_test();
+        assert_eq!(node_state.ingester_status(), IngesterStatus::Ready);
+    }
+}

quickwit/quickwit-proto/src/types/doc_mapping_uid.rs

@@ -99,6 +99,7 @@ impl prost::Message for DocMappingUid {
                 )?;
                 let ulid_bytes: [u8; ULID_SIZE] =
                     buffer.try_into().map_err(|buffer: Vec<u8>| {
+                        #[allow(deprecated)]
                         prost::DecodeError::new(format!(
                             "invalid length for field `doc_mapping_uid`, expected 16 bytes, got {}",
                             buffer.len()

quickwit/quickwit-proto/src/types/doc_uid.rs

@@ -98,6 +98,7 @@ impl prost::Message for DocUid {
                 )?;
                 let ulid_bytes: [u8; ULID_SIZE] =
                     buffer.try_into().map_err(|buffer: Vec<u8>| {
+                        #[allow(deprecated)]
                         prost::DecodeError::new(format!(
                             "invalid length for field `doc_uid`, expected 16 bytes, got {}",
                             buffer.len()

quickwit/quickwit-proto/src/types/index_uid.rs

@@ -137,6 +137,7 @@ impl prost::Message for IndexUid {
                 )?;
                 let ulid_bytes: [u8; ULID_SIZE] =
                     buffer.try_into().map_err(|buffer: Vec<u8>| {
+                        #[allow(deprecated)]
                         prost::DecodeError::new(format!(
                             "invalid length for field `incarnation_id`, expected 16 bytes, got {}",
                             buffer.len()

quickwit/quickwit-proto/src/types/pipeline_uid.rs

@@ -102,6 +102,7 @@ impl prost::Message for PipelineUid {
                 )?;
                 let ulid_bytes: [u8; ULID_SIZE] =
                     buffer.try_into().map_err(|buffer: Vec<u8>| {
+                        #[allow(deprecated)]
                         prost::DecodeError::new(format!(
                             "invalid length for field `pipeline_uid`, expected 16 bytes, got {}",
                             buffer.len()

quickwit/quickwit-proto/src/types/position.rs

@@ -273,6 +273,7 @@ impl prost::Message for Position {
                         error
                     },
                 )?;
+                #[allow(deprecated)]
                 let byte_string = ByteString::try_from(value)
                     .map_err(|_| DecodeError::new("position is not valid UTF-8"))?;
                 *self = Self::from(byte_string);

quickwit/quickwit-proto/src/types/shard_id.rs

@@ -104,6 +104,7 @@ impl prost::Message for ShardId {
                         error
                     },
                 )?;
+                #[allow(deprecated)]
                 let byte_string = ByteString::try_from(value)
                     .map_err(|_| DecodeError::new("shard_id is not valid UTF-8"))?;
                 *self = Self(byte_string);

quickwit/quickwit-serve/src/lib.rs

@@ -679,15 +679,13 @@ pub async fn serve_quickwit(
     .context("failed to start searcher service")?;
 
     // The control plane listens for local shards updates to learn about each shard's ingestion
-    // throughput. Ingesters (routers) do so to update their shard table.
-    let local_shards_update_listener_handle_opt = if node_config
-        .is_service_enabled(QuickwitService::ControlPlane)
-        || node_config.is_service_enabled(QuickwitService::Indexer)
-    {
-        Some(setup_local_shards_update_listener(cluster.clone(), event_broker.clone()).await)
-    } else {
-        None
-    };
+    // throughput.
+    let local_shards_update_listener_handle_opt =
+        if node_config.is_service_enabled(QuickwitService::ControlPlane) {
+            Some(setup_local_shards_update_listener(cluster.clone(), event_broker.clone()).await)
+        } else {
+            None
+        };
 
     let report_splits_subscription_handle_opt =
         // DISCLAIMER: This is quirky here: We base our decision to forward the split report depending