trying to isolate code metrics code (#6285)

* trying to isolate code from metrics

* added license

* minor refact

* Removing AI slop

* Making SourceActor's member private, and added a constructor

* Renaming ProcessorMailbox SourceSink

* Simplification of the dual sequencer/no sequencer logic

* AI slop

* CI

* fixing clippy

---------

Co-authored-by: Paul Masurel <paul@quickwit.io>

Author: fulmicoton-dd

.github/workflows/ci.yml

@@ -102,13 +102,17 @@ jobs:
         uses: taiki-e/install-action@aba36d755ec7ca22d38b12111787c26115943952
         with:
           tool: cargo-nextest
-      - name: cargo build
+      - name: cargo check without metrics
         if: always() && steps.modified.outputs.rust_src == 'true'
-        run: cargo build --features=postgres --tests --bin quickwit
+        run: cargo check --features=postgres --tests --bin quickwit
+        working-directory: ./quickwit
+      - name: cargo check with metrics
+        if: always() && steps.modified.outputs.rust_src == 'true'
+        run: cargo check --features=postgres,metrics --tests --bin quickwit
         working-directory: ./quickwit
       - name: cargo nextest
         if: always() && steps.modified.outputs.rust_src == 'true'
-        run: cargo nextest run --features=postgres --retries 1
+        run: cargo nextest run --features=postgres,metrics --retries 1
         working-directory: ./quickwit
       - name: Install python packages
         if: always() && steps.modified.outputs.rust_src == 'true'

quickwit/Cargo.lock

@@ -95,6 +95,9 @@ jemalloc-profiled = [
 ci-test = []
 pprof = ["quickwit-serve/pprof"]
 openssl-support = ["openssl-probe"]
+# metrics here refers to adding support for metrics ingestion within Quickwit.
+# (this is not about quickwit's metrics themselves)
+metrics = [ "quickwit-indexing/metrics" ]
 # Requires to enable tokio unstable via RUSTFLAGS="--cfg tokio_unstable"
 tokio-console = ["console-subscriber", "quickwit-common/named_tasks"]
 release-feature-set = [
@@ -116,6 +119,7 @@ release-feature-vendored-set = [
   "openssl-support",
   "pprof",
   "quickwit-indexing/kinesis",
+  "quickwit-indexing/metrics",
   "quickwit-indexing/pulsar",
   "quickwit-indexing/sqs",
   "quickwit-indexing/vrl",
@@ -129,6 +133,7 @@ release-macos-feature-vendored-set = [
   "jemalloc",
   "openssl-support",
   "quickwit-indexing/kinesis",
+  "quickwit-indexing/metrics",
   "quickwit-indexing/pulsar",
   "quickwit-indexing/sqs",
   "quickwit-indexing/vrl",

quickwit/quickwit-cli/Cargo.toml

@@ -37,7 +37,7 @@ use quickwit_config::{
     TransformConfig, VecSourceParams,
 };
 use quickwit_index_management::{IndexService, clear_cache_directory};
-use quickwit_indexing::IndexingPipeline;
+use quickwit_indexing::BoxedPipelineHandle;
 use quickwit_indexing::actors::{IndexingService, MergePipeline, MergeSchedulerService};
 use quickwit_indexing::models::{
     DetachIndexingPipeline, DetachMergePipeline, IndexingStatistics, SpawnPipeline,
@@ -751,7 +751,7 @@ async fn extract_split_cli(args: ExtractSplitArgs) -> anyhow::Result<()> {
 /// Starts a tokio task that displays the indexing statistics
 /// every once in awhile.
 pub async fn start_statistics_reporting_loop(
-    pipeline_handle: ActorHandle<IndexingPipeline>,
+    pipeline_handle: BoxedPipelineHandle,
     is_stdin: bool,
 ) -> anyhow::Result<IndexingStatistics> {
     let mut stdout_handle = stdout();

quickwit/quickwit-cli/src/tool.rs

@@ -49,6 +49,9 @@ pub mod tower;
 pub mod type_map;
 pub mod uri;
 
+mod metrics_specific;
+pub use metrics_specific::*;
+
 mod socket_addr_legacy_hash;
 
 use std::env;
@@ -215,14 +218,6 @@ macro_rules! assert_eventually {
     };
 }
 
-/// Returns whether the given index ID corresponds to a metrics index.
-///
-/// Metrics indexes use the Parquet/DataFusion pipeline instead of the Tantivy pipeline.
-/// An index is considered a metrics index if it starts with "otel-metrics" or "metrics-".
-pub fn is_metrics_index(index_id: &str) -> bool {
-    index_id.starts_with("otel-metrics") || index_id.starts_with("metrics-")
-}
-
 #[macro_export]
 macro_rules! ignore_error_kind {
     ($kind:path, $expr:expr) => {
@@ -423,27 +418,6 @@ mod tests {
         assert_eq!(div_ceil_u32(0, 3), 0);
     }
 
-    #[test]
-    fn test_is_metrics_index() {
-        // OpenTelemetry metrics indexes
-        assert!(is_metrics_index("otel-metrics-v0_7"));
-        assert!(is_metrics_index("otel-metrics"));
-        assert!(is_metrics_index("otel-metrics-custom"));
-
-        // Generic metrics indexes
-        assert!(is_metrics_index("metrics-default"));
-        assert!(is_metrics_index("metrics-"));
-        assert!(is_metrics_index("metrics-my-app"));
-
-        // Non-metrics indexes
-        assert!(!is_metrics_index("otel-logs-v0_7"));
-        assert!(!is_metrics_index("otel-traces-v0_7"));
-        assert!(!is_metrics_index("my-index"));
-        assert!(!is_metrics_index("logs-default"));
-        assert!(!is_metrics_index("metrics")); // No hyphen after "metrics"
-        assert!(!is_metrics_index("my-metrics-index")); // Not prefixed
-    }
-
     #[test]
     fn test_parse_bool_lenient() {
         assert_eq!(parse_bool_lenient("true"), Some(true));

quickwit/quickwit-common/src/lib.rs

@@ -0,0 +1,47 @@
+// Copyright 2021-Present Datadog, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+/// Returns whether the given index ID corresponds to a metrics index.
+///
+/// Metrics indexes use the Parquet/DataFusion pipeline instead of the Tantivy pipeline.
+/// An index is considered a metrics index if it starts with "otel-metrics" or "metrics-".
+pub fn is_metrics_index(index_id: &str) -> bool {
+    index_id.starts_with("otel-metrics") || index_id.starts_with("metrics-")
+}
+
+#[cfg(test)]
+mod tests {
+    use super::is_metrics_index;
+
+    #[test]
+    fn test_is_metrics_index() {
+        // OpenTelemetry metrics indexes
+        assert!(is_metrics_index("otel-metrics-v0_7"));
+        assert!(is_metrics_index("otel-metrics"));
+        assert!(is_metrics_index("otel-metrics-custom"));
+
+        // Generic metrics indexes
+        assert!(is_metrics_index("metrics-default"));
+        assert!(is_metrics_index("metrics-"));
+        assert!(is_metrics_index("metrics-my-app"));
+
+        // Non-metrics indexes
+        assert!(!is_metrics_index("otel-logs-v0_7"));
+        assert!(!is_metrics_index("otel-traces-v0_7"));
+        assert!(!is_metrics_index("my-index"));
+        assert!(!is_metrics_index("logs-default"));
+        assert!(!is_metrics_index("metrics")); // No hyphen after "metrics"
+        assert!(!is_metrics_index("my-metrics-index")); // Not prefixed
+    }
+}

quickwit/quickwit-common/src/metrics_specific.rs

@@ -49,7 +49,7 @@ utoipa = { workspace = true }
 vrl = { workspace = true, optional = true }
 warp = { workspace = true, optional = true }
 
-arrow = { workspace = true }
+arrow = { workspace = true, optional = true }
 
 quickwit-actors = { workspace = true }
 quickwit-aws = { workspace = true }
@@ -61,7 +61,7 @@ quickwit-doc-mapper = { workspace = true }
 quickwit-ingest = { workspace = true }
 quickwit-metastore = { workspace = true }
 quickwit-opentelemetry = { workspace = true }
-quickwit-parquet-engine = { workspace = true }
+quickwit-parquet-engine = { workspace = true, optional = true }
 quickwit-proto = { workspace = true }
 quickwit-storage = { workspace = true }
 
@@ -105,6 +105,7 @@ testsuite = [
   "quickwit-proto/testsuite",
   "quickwit-storage/testsuite"
 ]
+metrics = ["dep:arrow", "dep:quickwit-parquet-engine"]
 vrl = ["dep:vrl", "quickwit-config/vrl"]
 postgres = ["quickwit-metastore/postgres"]
 ci-test = []
@@ -113,7 +114,6 @@ ci-test = []
 bytes = { workspace = true }
 criterion = { workspace = true, features = ["async_tokio"] }
 mockall = { workspace = true }
-parquet = { workspace = true }
 proptest = { workspace = true }
 prost = { workspace = true }
 rand = { workspace = true }
@@ -122,8 +122,8 @@ sqlx = { workspace = true, features = ["runtime-tokio", "postgres"] }
 tempfile = { workspace = true }
 
 quickwit-actors = { workspace = true, features = ["testsuite"] }
-quickwit-parquet-engine = { workspace = true, features = ["testsuite"] }
 quickwit-cluster = { workspace = true, features = ["testsuite"] }
+quickwit-parquet-engine = { workspace = true, features = ["testsuite"] }
 quickwit-common = { workspace = true, features = ["testsuite"] }
 quickwit-config = { workspace = true, features = ["testsuite"] }
 quickwit-doc-mapper = { workspace = true, features = ["testsuite"] }

quickwit/quickwit-indexing/Cargo.toml

@@ -49,8 +49,8 @@ use tokio::sync::Semaphore;
 use tracing::{Span, info, info_span, warn};
 use ulid::Ulid;
 
-use crate::actors::IndexSerializer;
-use crate::actors::cooperative_indexing::{CooperativeIndexingCycle, CooperativeIndexingPeriod};
+use super::IndexSerializer;
+use super::cooperative_indexing::{CooperativeIndexingCycle, CooperativeIndexingPeriod};
 use crate::models::{
     CommitTrigger, EmptySplit, IndexedSplitBatchBuilder, IndexedSplitBuilder, NewPublishLock,
     NewPublishToken, ProcessedDoc, ProcessedDocBatch, PublishLock,
@@ -715,8 +715,7 @@ mod tests {
     use quickwit_proto::types::{IndexUid, NodeId, PipelineUid};
     use tantivy::{DateTime, doc};
 
-    use super::*;
-    use crate::actors::indexer::{IndexerCounters, record_timestamp};
+    use super::{IndexerCounters, record_timestamp, *};
 
     #[test]
     fn test_record_timestamp() {