chore(pkg/replication): Add TX gap metrics

Added metrics:

* immudb_replication_primary_committed_tx_id - this is the latest
  known committed transaction ID of the prinary
* immudb_replication_allow_commit_up_to_tx_id - tx ID the replica
  is allowed to commit up to

Signed-off-by: Bartłomiej Święcki <[email protected]>

Author: Bartłomiej Święcki

pkg/replication/metrics.go

@@ -55,6 +55,16 @@ var (
 		Name: "immudb_replication_replicators_retries",
 		Help: "number of retries while replicating transactions caused by errors",
 	}, []string{"db"})
+
+	_metricsReplicationPrimaryCommittedTxID = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "immudb_replication_primary_committed_tx_id",
+		Help: "the latest know transaction ID committed on the primary node",
+	}, []string{"db"})
+
+	_metricsAllowCommitUpToTxID = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "immudb_replication_allow_commit_up_to_tx_id",
+		Help: "most recently received confirmation up to which commit id the replica is allowed to durably commit",
+	}, []string{"db"})
 )
 
 type metrics struct {
@@ -64,6 +74,8 @@ type metrics struct {
 	replicators              prometheus.Gauge
 	replicatorsActive        prometheus.Gauge
 	replicatorsInRetryDelay  prometheus.Gauge
+	primaryCommittedTxID     prometheus.Gauge
+	allowCommitUpToTxID      prometheus.Gauge
 }
 
 // metricsForDb returns metrics object for particular database name
@@ -75,6 +87,8 @@ func metricsForDb(dbName string) metrics {
 		replicators:              _metricsReplicators.WithLabelValues(dbName),
 		replicatorsActive:        _metricsReplicatorsActive.WithLabelValues(dbName),
 		replicatorsInRetryDelay:  _metricsReplicatorsInRetryDelay.WithLabelValues(dbName),
+		primaryCommittedTxID:     _metricsReplicationPrimaryCommittedTxID.WithLabelValues(dbName),
+		allowCommitUpToTxID:      _metricsAllowCommitUpToTxID.WithLabelValues(dbName),
 	}
 }
 
@@ -83,6 +97,8 @@ func (m *metrics) reset() {
 	m.replicators.Set(0)
 	m.replicatorsActive.Set(0)
 	m.replicatorsInRetryDelay.Set(0)
+	m.primaryCommittedTxID.Set(0)
+	m.allowCommitUpToTxID.Set(0)
 }
 
 // replicationTimeHistogramTimer returns prometheus timer for replicationTimeHistogram

pkg/replication/replicator.go

@@ -369,19 +369,27 @@ func (txr *TxReplicator) fetchNextTx() error {
 	if syncReplicationEnabled {
 		md := exportTxStream.Trailer()
 
-		if len(md.Get("may-commit-up-to-txid-bin")) == 0 || len(md.Get("may-commit-up-to-alh-bin")) == 0 {
+		if len(md.Get("may-commit-up-to-txid-bin")) == 0 ||
+			len(md.Get("may-commit-up-to-alh-bin")) == 0 ||
+			len(md.Get("committed-txid-bin")) == 0 {
 			return ErrNoSynchronousReplicationOnMaster
 		}
 
-		if len(md.Get("may-commit-up-to-txid-bin")[0]) != 8 || len(md.Get("may-commit-up-to-alh-bin")[0]) != sha256.Size {
+		if len(md.Get("may-commit-up-to-txid-bin")[0]) != 8 ||
+			len(md.Get("may-commit-up-to-alh-bin")[0]) != sha256.Size ||
+			len(md.Get("committed-txid-bin")[0]) != 8 {
 			return ErrInvalidReplicationMetadata
 		}
 
 		mayCommitUpToTxID := binary.BigEndian.Uint64([]byte(md.Get("may-commit-up-to-txid-bin")[0]))
+		committedTxID := binary.BigEndian.Uint64([]byte(md.Get("committed-txid-bin")[0]))
 
 		var mayCommitUpToAlh [sha256.Size]byte
 		copy(mayCommitUpToAlh[:], []byte(md.Get("may-commit-up-to-alh-bin")[0]))
 
+		txr.metrics.primaryCommittedTxID.Set(float64(committedTxID))
+		txr.metrics.allowCommitUpToTxID.Set(float64(mayCommitUpToTxID))
+
 		if mayCommitUpToTxID > commitState.TxId {
 			err = txr.db.AllowCommitUpto(mayCommitUpToTxID, mayCommitUpToAlh)
 			if err != nil {

pkg/server/stream_replication.go

@@ -38,12 +38,19 @@ func (s *ImmuServer) ExportTx(req *schema.ExportTxRequest, txsServer schema.Immu
 
 	defer func() {
 		if req.FollowerState != nil {
-			var bTxID [8]byte
-			binary.BigEndian.PutUint64(bTxID[:], mayCommitUpToTxID)
+			var bMayCommitUpToTxID [8]byte
+			binary.BigEndian.PutUint64(bMayCommitUpToTxID[:], mayCommitUpToTxID)
+
+			var bCommittedTxID [8]byte
+			state, err := db.CurrentState()
+			if err == nil {
+				binary.BigEndian.PutUint64(bCommittedTxID[:], state.TxId)
+			}
 
 			md := metadata.Pairs(
-				"may-commit-up-to-txid-bin", string(bTxID[:]),
+				"may-commit-up-to-txid-bin", string(bMayCommitUpToTxID[:]),
 				"may-commit-up-to-alh-bin", string(mayCommitUpToAlh[:]),
+				"committed-txid-bin", string(bCommittedTxID[:]),
 			)
 
 			txsServer.SetTrailer(md)