fix(meta): remove last_committed_barrier_time series when database has no streaming job (#26231)

src/meta/src/barrier/checkpoint/control.rs

@@ -24,7 +24,7 @@ use itertools::Itertools;
 use risingwave_common::catalog::{DatabaseId, TableId};
 use risingwave_common::id::JobId;
 use risingwave_common::metrics::{LabelGuardedHistogram, LabelGuardedIntGauge};
-use risingwave_common::util::epoch::EpochPair;
+use risingwave_common::util::epoch::{Epoch, EpochPair};
 use risingwave_common::util::stream_graph_visitor::visit_stream_node_cont;
 use risingwave_meta_model::WorkerId;
 use risingwave_pb::common::WorkerNode;
@@ -662,11 +662,14 @@ pub(in crate::barrier) struct DatabaseCheckpointControl {
     finishing_jobs_collector:
         BarrierItemCollector<JobId, (Vec<BarrierCompleteResponse>, TrackingJob), ()>,
     /// The barrier that are completing.
-    /// Some(`prev_epoch`)
-    completing_barrier: Option<u64>,
+    completing_barrier: Option<EpochPair>,
 
     committed_epoch: Option<u64>,
 
+    /// `None` while the database has no streaming job, so that a frozen timestamp does not
+    /// render as an ever-growing barrier pending time.
+    last_committed_barrier_time: Option<LabelGuardedIntGauge>,
+
     pub(super) database_info: InflightDatabaseInfo,
     pub independent_checkpoint_job_controls: HashMap<JobId, IndependentCheckpointJobControl>,
 }
@@ -680,6 +683,7 @@ impl DatabaseCheckpointControl {
             finishing_jobs_collector: BarrierItemCollector::new(false),
             completing_barrier: None,
             committed_epoch: None,
+            last_committed_barrier_time: None,
             database_info: InflightDatabaseInfo::empty(database_id, shared_actor_infos),
             independent_checkpoint_job_controls: Default::default(),
         }
@@ -699,6 +703,7 @@ impl DatabaseCheckpointControl {
             finishing_jobs_collector: BarrierItemCollector::new(false),
             completing_barrier: None,
             committed_epoch: Some(committed_epoch),
+            last_committed_barrier_time: None,
             database_info,
             independent_checkpoint_job_controls,
         }
@@ -979,7 +984,7 @@ impl DatabaseCheckpointControl {
                     resps_to_commit,
                     self.collect_backfill_pinned_upstream_log_epoch(),
                 );
-                self.completing_barrier = Some(info.barrier_info.prev_epoch());
+                self.completing_barrier = Some(info.barrier_info.epoch());
                 task.finished_jobs.extend(staging_commit_info.finished_jobs);
                 task.finished_cdc_table_backfill
                     .extend(staging_commit_info.finished_cdc_table_backfill);
@@ -1019,10 +1024,17 @@ impl DatabaseCheckpointControl {
         independent_job_epochs: Vec<(JobId, u64)>,
     ) {
         {
-            if let Some(prev_epoch) = self.completing_barrier.take() {
-                assert_eq!(command_prev_epoch, Some(prev_epoch));
-                self.committed_epoch = Some(prev_epoch);
-                partial_graph_manager.ack_completed(self.partial_graph_id, prev_epoch);
+            if let Some(epoch) = self.completing_barrier.take() {
+                assert_eq!(command_prev_epoch, Some(epoch.prev));
+                self.committed_epoch = Some(epoch.prev);
+                partial_graph_manager.ack_completed(self.partial_graph_id, epoch.prev);
+                self.last_committed_barrier_time
+                    .get_or_insert_with(|| {
+                        GLOBAL_META_METRICS
+                            .last_committed_barrier_time
+                            .with_guarded_label_values(&[&self.database_id.to_string()])
+                    })
+                    .set(Epoch(epoch.curr).as_unix_secs() as i64);
             } else {
                 assert_eq!(command_prev_epoch, None);
             };
@@ -1199,6 +1211,8 @@ impl DatabaseCheckpointControl {
                 self.independent_checkpoint_job_controls.is_empty(),
                 "should not have snapshot backfill job when there is no normal job in database"
             );
+            // Drop the guard to remove the metric series of this database.
+            self.last_committed_barrier_time = None;
             // skip the command when there is nothing to do with the barrier
             for mut notifier in notifiers {
                 notifier.notify_started();

src/meta/src/barrier/complete_task.rs

@@ -148,10 +148,6 @@ impl CompleteBarrierTask {
                         &info.barrier_info,
                         command_name,
                     );
-                    GLOBAL_META_METRICS
-                        .last_committed_barrier_time
-                        .with_label_values(&[database_id.to_string().as_str()])
-                        .set(info.barrier_info.curr_epoch.value().as_unix_secs() as i64);
                 }
             }
 

src/meta/src/rpc/metrics.rs

@@ -95,7 +95,7 @@ pub struct MetaMetrics {
     /// The number of in-flight barriers
     pub in_flight_barrier_nums: LabelGuardedIntGaugeVec,
     /// The timestamp (UNIX epoch seconds) of the last committed barrier's epoch time.
-    pub last_committed_barrier_time: IntGaugeVec,
+    pub last_committed_barrier_time: LabelGuardedIntGaugeVec,
     /// The barrier interval of each database
     pub barrier_interval_by_database: GaugeVec,
 
@@ -314,7 +314,7 @@ impl MetaMetrics {
             registry
         )
         .unwrap();
-        let last_committed_barrier_time = register_int_gauge_vec_with_registry!(
+        let last_committed_barrier_time = register_guarded_int_gauge_vec_with_registry!(
             "last_committed_barrier_time",
             "The timestamp (UNIX epoch seconds) of the last committed barrier's epoch time.",
             &["database_id"],