Added a metric to track the number of running pipelines per index. (#5880)

Author: fulmicoton-dd

quickwit/quickwit-indexing/src/actors/indexing_pipeline.rs

@@ -23,6 +23,7 @@ use quickwit_actors::{
     QueueCapacity, Supervisable,
 };
 use quickwit_common::KillSwitch;
+use quickwit_common::metrics::OwnedGaugeGuard;
 use quickwit_common::pubsub::EventBroker;
 use quickwit_common::temp_dir::TempDirectory;
 use quickwit_config::{IndexingSettings, RetentionPolicy, SourceConfig};
@@ -120,6 +121,7 @@ pub struct IndexingPipeline {
     // requiring a respawn of the pipeline.
     // We keep the list of shards here however, to reassign them after a respawn.
     shard_ids: BTreeSet<ShardId>,
+    _indexing_pipelines_gauge_guard: OwnedGaugeGuard,
 }
 
 #[async_trait]
@@ -154,6 +156,10 @@ impl Actor for IndexingPipeline {
 
 impl IndexingPipeline {
     pub fn new(params: IndexingPipelineParams) -> Self {
+        let indexing_pipelines_gauge = crate::metrics::INDEXER_METRICS
+            .indexing_pipelines
+            .with_label_values([&params.pipeline_id.index_uid.index_id]);
+        let indexing_pipelines_gauge_guard = OwnedGaugeGuard::from_gauge(indexing_pipelines_gauge);
         let params_fingerprint = params.params_fingerprint;
         IndexingPipeline {
             params,
@@ -165,6 +171,7 @@ impl IndexingPipeline {
                 ..Default::default()
             },
             shard_ids: Default::default(),
+            _indexing_pipelines_gauge_guard: indexing_pipelines_gauge_guard,
         }
     }
 

quickwit/quickwit-indexing/src/metrics.rs

@@ -21,6 +21,7 @@ use quickwit_common::metrics::{
 pub struct IndexerMetrics {
     pub processed_docs_total: IntCounterVec<2>,
     pub processed_bytes: IntCounterVec<2>,
+    pub indexing_pipelines: IntGaugeVec<1>,
     pub backpressure_micros: IntCounterVec<1>,
     pub available_concurrent_upload_permits: IntGaugeVec<1>,
     pub split_builders: IntGauge,
@@ -51,6 +52,13 @@ impl Default for IndexerMetrics {
                 &[],
                 ["index", "docs_processed_status"],
             ),
+            indexing_pipelines: new_gauge_vec(
+                "indexing_pipelines",
+                "Number of running indexing pipelines",
+                "indexing",
+                &[],
+                ["index"],
+            ),
             backpressure_micros: new_counter_vec(
                 "backpressure_micros",
                 "Amount of time spent in backpressure (in micros). This time only includes the \