chore(cubestore): Upgrade DF: Add compaction operation app_metrics

Author: srh

rust/cubestore/cubestore/src/app_metrics.rs

@@ -85,6 +85,30 @@ pub static CACHESTORE_ROCKSDB_CF_DEFAULT_SIZE: Gauge =
 pub static CACHESTORE_SCHEDULER_GC_QUEUE: Gauge =
     metrics::gauge("cs.cachestore.scheduler.gc_queue");
 
+// TODO: Maybe these should be a single metric that uses tags.
+pub static JOBS_PARTITION_COMPACTION: Counter =
+    metrics::counter("cs.jobs.partition_compaction.count");
+pub static JOBS_PARTITION_COMPACTION_COMPLETED: Counter =
+    metrics::counter("cs.jobs.partition_compaction.completed");
+pub static JOBS_PARTITION_COMPACTION_FAILURES: Counter =
+    metrics::counter("cs.jobs.partition_compaction.failures");
+pub static JOBS_MULTI_PARTITION_SPLIT: Counter =
+    metrics::counter("cs.jobs.multi_partition_split.count");
+pub static JOBS_MULTI_PARTITION_SPLIT_COMPLETED: Counter =
+    metrics::counter("cs.jobs.multi_partition_split.completed");
+pub static JOBS_MULTI_PARTITION_SPLIT_FAILURES: Counter =
+    metrics::counter("cs.jobs.multi_partition_split.failures");
+pub static JOBS_FINISH_MULTI_SPLIT: Counter = metrics::counter("cs.jobs.finish_multi_split.count");
+pub static JOBS_FINISH_MULTI_SPLIT_COMPLETED: Counter =
+    metrics::counter("cs.jobs.finish_multi_split.completed");
+pub static JOBS_FINISH_MULTI_SPLIT_FAILURES: Counter =
+    metrics::counter("cs.jobs.finish_multi_split.failures");
+pub static JOBS_REPARTITION_CHUNK: Counter = metrics::counter("cs.jobs.repartition_chunk.count");
+pub static JOBS_REPARTITION_CHUNK_COMPLETED: Counter =
+    metrics::counter("cs.jobs.repartition_chunk.completed");
+pub static JOBS_REPARTITION_CHUNK_FAILURES: Counter =
+    metrics::counter("cs.jobs.repartition_chunk.failures");
+
 /// RemoteFs metrics
 pub static REMOTE_FS_OPERATION_CORE: Counter = metrics::counter("cs.remote_fs.operations.core");
 pub static REMOTE_FS_FILES_TO_REMOVE: Gauge = metrics::gauge("cs.remote_fs.files_to_remove.count");

rust/cubestore/cubestore/src/cluster/ingestion/job_processor.rs

@@ -7,7 +7,7 @@ use crate::metastore::{MetaStore, RowKey, TableId};
 use crate::queryplanner::trace_data_loaded::DataLoadedSize;
 use crate::store::compaction::CompactionService;
 use crate::store::ChunkDataStore;
-use crate::CubeError;
+use crate::{app_metrics, CubeError};
 use async_trait::async_trait;
 use serde::{Deserialize, Serialize};
 use std::sync::Arc;
@@ -117,10 +117,16 @@ impl JobIsolatedProcessor {
                     let compaction_service = self.compaction_service.clone();
                     let partition_id = *partition_id;
                     let data_loaded_size = DataLoadedSize::new();
+                    app_metrics::JOBS_PARTITION_COMPACTION.add(1);
                     let r = compaction_service
                         .compact(partition_id, data_loaded_size.clone())
                         .await;
-                    r?;
+                    if let Err(e) = r {
+                        app_metrics::JOBS_PARTITION_COMPACTION_FAILURES.add(1);
+                        return Err(e);
+                    }
+                    app_metrics::JOBS_PARTITION_COMPACTION_COMPLETED.add(1);
+
                     Ok(JobProcessResult::new(data_loaded_size.get()))
                 } else {
                     Self::fail_job_row_key(job)
@@ -130,7 +136,13 @@ impl JobIsolatedProcessor {
                 if let RowKey::Table(TableId::MultiPartitions, id) = job.row_reference() {
                     let compaction_service = self.compaction_service.clone();
                     let id = *id;
-                    compaction_service.split_multi_partition(id).await?;
+                    app_metrics::JOBS_MULTI_PARTITION_SPLIT.add(1);
+                    let r = compaction_service.split_multi_partition(id).await;
+                    if let Err(e) = r {
+                        app_metrics::JOBS_MULTI_PARTITION_SPLIT_FAILURES.add(1);
+                        return Err(e);
+                    }
+                    app_metrics::JOBS_MULTI_PARTITION_SPLIT_COMPLETED.add(1);
                     Ok(JobProcessResult::default())
                 } else {
                     Self::fail_job_row_key(job)
@@ -143,9 +155,15 @@ impl JobIsolatedProcessor {
                     let compaction_service = self.compaction_service.clone();
                     let multi_part_id = *multi_part_id;
                     for p in meta_store.find_unsplit_partitions(multi_part_id).await? {
-                        compaction_service
+                        app_metrics::JOBS_FINISH_MULTI_SPLIT.add(1);
+                        let r = compaction_service
                             .finish_multi_split(multi_part_id, p)
-                            .await?
+                            .await;
+                        if let Err(e) = r {
+                            app_metrics::JOBS_FINISH_MULTI_SPLIT_FAILURES.add(1);
+                            return Err(e);
+                        }
+                        app_metrics::JOBS_FINISH_MULTI_SPLIT_COMPLETED.add(1);
                     }
 
                     Ok(JobProcessResult::default())
@@ -196,9 +214,16 @@ impl JobIsolatedProcessor {
                         ));
                     }
                     let data_loaded_size = DataLoadedSize::new();
-                    self.chunk_store
+                    app_metrics::JOBS_REPARTITION_CHUNK.add(1);
+                    let r = self
+                        .chunk_store
                         .repartition_chunk(chunk_id, data_loaded_size.clone())
-                        .await?;
+                        .await;
+                    if let Err(e) = r {
+                        app_metrics::JOBS_REPARTITION_CHUNK_FAILURES.add(1);
+                        return Err(e);
+                    }
+                    app_metrics::JOBS_REPARTITION_CHUNK_COMPLETED.add(1);
                     Ok(JobProcessResult::new(data_loaded_size.get()))
                 } else {
                     Self::fail_job_row_key(job)