Feat[STATS]: Report partition offsets (#922)

waldgange · web-flow · commit e8e59ec8a97e · 2025-09-16T16:56:57.000+03:00
Signed-off-by: Anton Pryakhin &lt;apryakhin1@bloomberg.net&gt;
diff --git a/src/groups/mqb/mqbs/mqbs_filestore.cpp b/src/groups/mqb/mqbs/mqbs_filestore.cpp
@@ -3973,10 +3973,11 @@ int FileStore::issueSyncPointInternal(SyncPointType::Enum type,
                     immediateFlush);
 
     // Report cluster's partition stats
-    d_clusterStats_p->setPartitionOutstandingBytes(
-        d_config.partitionId(),
-        fs->d_outstandingBytesData,
-        fs->d_outstandingBytesJournal);
+    d_clusterStats_p->setPartitionBytes(d_config.partitionId(),
+                                        fs->d_outstandingBytesData,
+                                        fs->d_outstandingBytesJournal,
+                                        fs->d_dataFilePosition,
+                                        fs->d_journalFilePosition);
 
     return rc_SUCCESS;
 }
@@ -4324,10 +4325,10 @@ int FileStore::writeQueueCreationRecord(
     // Create in-memory record.
     DataStoreRecordHandle handle;
     DataStoreRecord       record(RecordType::e_QUEUE_OP,
-                                 recordOffset,
-                                 queueRecLength);
+                           recordOffset,
+                           queueRecLength);
     DataStoreRecordKey    key(recHeader.sequenceNumber(),
-                              recHeader.primaryLeaseId());
+                           recHeader.primaryLeaseId());
 
     insertDataStoreRecord(&handle, key, record);
 
@@ -5207,10 +5208,12 @@ int FileStore::open(const QueueKeyInfoMap& queueKeyInfoMap)
     BSLS_ASSERT_SAFE(d_isOpen);
 
     // Report cluster's partition stats
-    d_clusterStats_p->setPartitionOutstandingBytes(
-        d_config.partitionId(),
-        d_fileSets[0].get()->d_outstandingBytesData,
-        d_fileSets[0].get()->d_outstandingBytesJournal);
+    const FileSet* fs = d_fileSets[0].get();
+    d_clusterStats_p->setPartitionBytes(d_config.partitionId(),
+                                        fs->d_outstandingBytesData,
+                                        fs->d_outstandingBytesJournal,
+                                        fs->d_dataFilePosition,
+                                        fs->d_journalFilePosition);
 
     return rc_SUCCESS;
 }
diff --git a/src/groups/mqb/mqbstat/mqbstat_clusterstats.cpp b/src/groups/mqb/mqbstat/mqbstat_clusterstats.cpp
@@ -91,6 +91,12 @@ struct ClusterStatsIndex {
         ,
         e_PARTITION_JOURNAL_BYTES
         // Value: Outstanding bytes in the journal file of the partition.
+        ,
+        e_PARTITION_DATA_OFFSET_BYTES
+        // Value: Offset bytes in the data file of the partition.
+        ,
+        e_PARTITION_JOURNAL_OFFSET_BYTES
+        // Value: Offset bytes in the journal file of the partition.
     };
 };
 
@@ -196,6 +202,12 @@ bsls::Types::Int64 ClusterStats::getValue(const bmqst::StatContext& context,
         return value == bsl::numeric_limits<bsls::Types::Int64>::min() ? 0
                                                                        : value;
     }
+    case Stat::e_PARTITION_DATA_OFFSET: {
+        return STAT_SINGLE(value, e_PARTITION_DATA_OFFSET_BYTES);
+    }
+    case Stat::e_PARTITION_JOURNAL_OFFSET: {
+        return STAT_SINGLE(value, e_PARTITION_JOURNAL_OFFSET_BYTES);
+    }
     case Stat::e_PARTITION_DATA_UTILIZATION_MAX: {
         const bsls::Types::Int64 value = STAT_RANGE(rangeMax,
                                                     e_PARTITION_DATA_BYTES);
@@ -395,9 +407,11 @@ ClusterStats& ClusterStats::setNodeRoleForPartition(int partitionId,
 }
 
 ClusterStats&
-ClusterStats::setPartitionOutstandingBytes(int                partitionId,
-                                           bsls::Types::Int64 dataBytes,
-                                           bsls::Types::Int64 journalBytes)
+ClusterStats::setPartitionBytes(int                partitionId,
+                                bsls::Types::Int64 outstandingDataBytes,
+                                bsls::Types::Int64 outstandingJournalBytes,
+                                bsls::Types::Int64 offsetDataBytes,
+                                bsls::Types::Int64 offsetJournalBytes)
 {
     // PRECONDITIONS
     BSLS_ASSERT_SAFE(partitionId >= 0 &&
@@ -408,10 +422,16 @@ ClusterStats::setPartitionOutstandingBytes(int                partitionId,
 
     d_partitionsStatContexts[partitionId]->reportValue(
         ClusterStatsIndex::e_PARTITION_DATA_BYTES,
-        dataBytes);
+        outstandingDataBytes);
     d_partitionsStatContexts[partitionId]->reportValue(
         ClusterStatsIndex::e_PARTITION_JOURNAL_BYTES,
-        journalBytes);
+        outstandingJournalBytes);
+    d_partitionsStatContexts[partitionId]->setValue(
+        ClusterStatsIndex::e_PARTITION_DATA_OFFSET_BYTES,
+        offsetDataBytes);
+    d_partitionsStatContexts[partitionId]->setValue(
+        ClusterStatsIndex::e_PARTITION_JOURNAL_OFFSET_BYTES,
+        offsetJournalBytes);
     return *this;
 }
 
@@ -571,7 +591,9 @@ ClusterStatsUtil::initializeStatContextCluster(int               historySize,
         .value("partition_status")
         .value("partition.rollover_time", bmqst::StatValue::e_DISCRETE)
         .value("partition.data_bytes", bmqst::StatValue::e_DISCRETE)
-        .value("partition.journal_bytes", bmqst::StatValue::e_DISCRETE);
+        .value("partition.journal_bytes", bmqst::StatValue::e_DISCRETE)
+        .value("partition.data_offset_bytes")
+        .value("partition.journal_offset_bytes");
 
     // NOTE: For the clusters, the stat context will have two levels of
     //       children, first level is per cluster, and second level is per
diff --git a/src/groups/mqb/mqbstat/mqbstat_clusterstats.h b/src/groups/mqb/mqbstat/mqbstat_clusterstats.h
@@ -106,6 +106,11 @@ class ClusterStats {
             /// Maximum observed outstanding bytes in the journal file of the
             /// partition.
             e_PARTITION_JOURNAL_CONTENT,
+            /// Latest observed offset bytes in the data file of the partition.
+            e_PARTITION_DATA_OFFSET,
+            /// Latest observed offset bytes in the journal file of the
+            /// partition.
+            e_PARTITION_JOURNAL_OFFSET,
             /// Maximum observed utilization of the data file of the partition.
             e_PARTITION_DATA_UTILIZATION_MAX,
             /// Maximum observed utilization of the journal file of the
@@ -258,11 +263,13 @@ class ClusterStats {
 
     /// Set the partition outstanding bytes of the specified data and
     /// journal files for the specified `partitionId` to the corresponding
-    /// specified `dataBytes` and `journalBytes` values.
-    ClusterStats&
-    setPartitionOutstandingBytes(int                partitionId,
-                                 bsls::Types::Int64 dataBytes,
-                                 bsls::Types::Int64 journalBytes);
+    /// specified `outstandingDataBytes`, `outstandingJournalBytes`,
+    /// `offsetDataBytes` and `offsetJournalBytes` values.
+    ClusterStats& setPartitionBytes(int                partitionId,
+                                    bsls::Types::Int64 outstandingDataBytes,
+                                    bsls::Types::Int64 outstandingJournalBytes,
+                                    bsls::Types::Int64 offsetDataBytes,
+                                    bsls::Types::Int64 offsetJournalBytes);
 
     /// Return a pointer to the statcontext.
     bmqst::StatContext* statContext();
diff --git a/src/plugins/bmqprometheus/bmqprometheus_prometheusstatconsumer.cpp b/src/plugins/bmqprometheus/bmqprometheus_prometheusstatconsumer.cpp
@@ -736,11 +736,14 @@ void PrometheusStatConsumer::captureClusterPartitionsStats()
             // Generate the metric name from the partition name (e.g.,
             // 'cluster_partition1_rollover_time')
             const bsl::string prefix = "cluster_" + partitionIt->name() + "_";
-            const bsl::string rollover_time = prefix + "rollover_time";
+            const bsl::string rollover_time        = prefix + "rollover_time";
+            const bsl::string journal_offset_bytes = prefix +
+                                                     "journal_offset_bytes";
             const bsl::string journal_outstanding_bytes =
                 prefix + "journal_outstanding_bytes";
             const bsl::string journal_utilization = prefix +
                                                     "journal_utilization_max";
+            const bsl::string data_offset_bytes = prefix + "data_offset_bytes";
             const bsl::string data_outstanding_bytes =
                 prefix + "data_outstanding_bytes";
             const bsl::string data_utilization = prefix +
@@ -750,13 +753,19 @@ void PrometheusStatConsumer::captureClusterPartitionsStats()
                 {rollover_time.c_str(),
                  mqbstat::ClusterStats::Stat::e_PARTITION_ROLLOVER_TIME,
                  false},
+                {journal_offset_bytes.c_str(),
+                 mqbstat::ClusterStats::Stat::e_PARTITION_JOURNAL_OFFSET,
+                 false},
                 {journal_outstanding_bytes.c_str(),
                  mqbstat::ClusterStats::Stat::e_PARTITION_JOURNAL_CONTENT,
                  false},
                 {journal_utilization.c_str(),
                  mqbstat::ClusterStats::Stat::
                      e_PARTITION_JOURNAL_UTILIZATION_MAX,
                  false},
+                {data_offset_bytes.c_str(),
+                 mqbstat::ClusterStats::Stat::e_PARTITION_DATA_OFFSET,
+                 false},
                 {data_outstanding_bytes.c_str(),
                  mqbstat::ClusterStats::Stat::e_PARTITION_DATA_CONTENT,
                  false},
