chore(pkg/replication): Add basic replication metrics

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

Author: Bartłomiej Święcki

pkg/replication/metrics.go

@@ -0,0 +1,91 @@
+/*
+Copyright 2022 Codenotary Inc. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package replication
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+// TODO: Metrics should be put behind abstract metrics interfaces to avoid direct dependency on prometheus SDK
+
+var (
+	_metricsTxWaitQueueHistogram = promauto.NewHistogramVec(prometheus.HistogramOpts{
+		Name:    "immudb_replication_tx_wait_queue",
+		Buckets: prometheus.ExponentialBucketsRange(0.001, 10.0, 16),
+		Help:    "histogram of time spent in the waiting queue before replicator picks up the transaction",
+	}, []string{"db"})
+
+	_metricsReplicationTimeHistogram = promauto.NewHistogramVec(prometheus.HistogramOpts{
+		Name:    "immudb_replication_commit_time",
+		Buckets: prometheus.ExponentialBucketsRange(0.001, 10.0, 16),
+		Help:    "histogram of time spent by replicators to replicate a single transaction",
+	}, []string{"db"})
+
+	_metricsReplicators = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "immudb_replication_replicators",
+		Help: "number of replicators available",
+	}, []string{"db"})
+
+	_metricsReplicatorsActive = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "immudb_replication_replicators_active",
+		Help: "number of replicators actively processing transactions",
+	}, []string{"db"})
+
+	_metricsReplicatorsInRetryDelay = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "immudb_replication_replicators_retry_delay",
+		Help: "number of replicators that are currently delaying the replication due to a replication error",
+	}, []string{"db"})
+
+	_metricsReplicationRetries = promauto.NewCounterVec(prometheus.CounterOpts{
+		Name: "immudb_replication_replicators_retries",
+		Help: "number of retries while replicating transactions caused by errors",
+	}, []string{"db"})
+)
+
+type metrics struct {
+	txWaitQueueHistogram     prometheus.Observer
+	replicationTimeHistogram prometheus.Observer
+	replicationRetries       prometheus.Counter
+	replicators              prometheus.Gauge
+	replicatorsActive        prometheus.Gauge
+	replicatorsInRetryDelay  prometheus.Gauge
+}
+
+// metricsForDb returns metrics object for particular database name
+func metricsForDb(dbName string) metrics {
+	return metrics{
+		txWaitQueueHistogram:     _metricsTxWaitQueueHistogram.WithLabelValues(dbName),
+		replicationTimeHistogram: _metricsReplicationTimeHistogram.WithLabelValues(dbName),
+		replicationRetries:       _metricsReplicationRetries.WithLabelValues(dbName),
+		replicators:              _metricsReplicators.WithLabelValues(dbName),
+		replicatorsActive:        _metricsReplicatorsActive.WithLabelValues(dbName),
+		replicatorsInRetryDelay:  _metricsReplicatorsInRetryDelay.WithLabelValues(dbName),
+	}
+}
+
+// reset ensures all necessary gauges are zeroed
+func (m *metrics) reset() {
+	m.replicators.Set(0)
+	m.replicatorsActive.Set(0)
+	m.replicatorsInRetryDelay.Set(0)
+}
+
+// replicationTimeHistogramTimer returns prometheus timer for replicationTimeHistogram
+func (m *metrics) replicationTimeHistogramTimer() *prometheus.Timer {
+	return prometheus.NewTimer(m.replicationTimeHistogram)
+}

pkg/replication/replicator.go

@@ -42,6 +42,11 @@ var ErrFollowerDivergedFromMaster = errors.New("follower diverged from master")
 var ErrNoSynchronousReplicationOnMaster = errors.New("master is not running with synchronous replication")
 var ErrInvalidReplicationMetadata = errors.New("invalid replication metadata retrieved")
 
+type prefetchTxEntry struct {
+	data    []byte
+	addedAt time.Time
+}
+
 type TxReplicator struct {
 	uuid xid.ID
 
@@ -61,7 +66,7 @@ type TxReplicator struct {
 
 	lastTx uint64
 
-	prefetchTxBuffer       chan []byte // buffered channel of exported txs
+	prefetchTxBuffer       chan prefetchTxEntry // buffered channel of exported txs
 	replicationConcurrency int
 
 	allowTxDiscarding bool
@@ -72,6 +77,8 @@ type TxReplicator struct {
 	running bool
 
 	mutex sync.Mutex
+
+	metrics metrics
 }
 
 func NewTxReplicator(uuid xid.ID, db database.DB, opts *Options, logger logger.Logger) (*TxReplicator, error) {
@@ -86,10 +93,11 @@ func NewTxReplicator(uuid xid.ID, db database.DB, opts *Options, logger logger.L
 		logger:                 logger,
 		_masterDB:              fullAddress(opts.masterDatabase, opts.masterAddress, opts.masterPort),
 		streamSrvFactory:       stream.NewStreamServiceFactory(opts.streamChunkSize),
-		prefetchTxBuffer:       make(chan []byte, opts.prefetchTxBufferSize),
+		prefetchTxBuffer:       make(chan prefetchTxEntry, opts.prefetchTxBufferSize),
 		replicationConcurrency: opts.replicationCommitConcurrency,
 		allowTxDiscarding:      opts.allowTxDiscarding,
 		delayer:                opts.delayer,
+		metrics:                metricsForDb(db.GetName()),
 	}, nil
 }
 
@@ -164,46 +172,77 @@ func (txr *TxReplicator) Start() error {
 		}
 	}()
 
+	txr.metrics.reset()
+
 	for i := 0; i < txr.replicationConcurrency; i++ {
 		go func() {
+			txr.metrics.replicators.Inc()
+			defer txr.metrics.replicators.Dec()
+
 			for etx := range txr.prefetchTxBuffer {
-				consecutiveFailures := 0
-
-				// replication must be retried as many times as necessary
-				for {
-					_, err := txr.db.ReplicateTx(etx)
-					if err == nil {
-						break // transaction successfully replicated
-					}
-					if errors.Is(err, ErrAlreadyStopped) {
-						return
-					}
-
-					if strings.Contains(err.Error(), "tx already committed") {
-						break // transaction successfully replicated
-					}
-
-					txr.logger.Infof("Failed to replicate transaction from '%s' to '%s'. Reason: %s", txr._masterDB, txr.db.GetName(), err.Error())
-
-					consecutiveFailures++
-
-					timer := time.NewTimer(txr.delayer.DelayAfter(consecutiveFailures))
-					select {
-					case <-txr.context.Done():
-						timer.Stop()
-						return
-					case <-timer.C:
-					}
+				txr.metrics.txWaitQueueHistogram.Observe(time.Since(etx.addedAt).Seconds())
+
+				if !txr.replicateSingleTx(etx.data) {
+					break
 				}
 			}
 		}()
 	}
 
-	txr.logger.Infof("Replication from '%s' to '%s' succesfully initialized", txr._masterDB, txr.db.GetName())
+	txr.logger.Infof("Replication from '%s' to '%s' successfully initialized", txr._masterDB, txr.db.GetName())
 
 	return nil
 }
 
+func (txr *TxReplicator) replicateSingleTx(data []byte) bool {
+	txr.metrics.replicatorsActive.Inc()
+	defer txr.metrics.replicatorsActive.Dec()
+	defer txr.metrics.replicationTimeHistogramTimer().ObserveDuration()
+
+	consecutiveFailures := 0
+
+	// replication must be retried as many times as necessary
+	for {
+		_, err := txr.db.ReplicateTx(data)
+		if err == nil {
+			break // transaction successfully replicated
+		}
+		if errors.Is(err, ErrAlreadyStopped) {
+			return false
+		}
+
+		if strings.Contains(err.Error(), "tx already committed") {
+			break // transaction successfully replicated
+		}
+
+		txr.logger.Infof("Failed to replicate transaction from '%s' to '%s'. Reason: %s", txr._masterDB, txr.db.GetName(), err.Error())
+
+		consecutiveFailures++
+
+		if !txr.replicationFailureDelay(consecutiveFailures) {
+			return false
+		}
+	}
+
+	return true
+}
+
+func (txr *TxReplicator) replicationFailureDelay(consecutiveFailures int) bool {
+	txr.metrics.replicationRetries.Inc()
+
+	txr.metrics.replicatorsInRetryDelay.Inc()
+	defer txr.metrics.replicatorsInRetryDelay.Dec()
+
+	timer := time.NewTimer(txr.delayer.DelayAfter(consecutiveFailures))
+	select {
+	case <-txr.context.Done():
+		timer.Stop()
+		return false
+	case <-timer.C:
+		return true
+	}
+}
+
 func fullAddress(db, address string, port int) string {
 	return fmt.Sprintf("%s@%s:%d", db, address, port)
 }
@@ -358,7 +397,10 @@ func (txr *TxReplicator) fetchNextTx() error {
 
 	if len(etx) > 0 {
 		// in some cases the transaction is not provided but only the master commit state
-		txr.prefetchTxBuffer <- etx
+		txr.prefetchTxBuffer <- prefetchTxEntry{
+			data:    etx,
+			addedAt: time.Now(),
+		}
 		txr.lastTx++
 	}