Added new metric: seconds since the last event has been synced (#28)

Author: pparshin

internal/bridge/replicator.go

@@ -33,8 +33,9 @@ type Bridge struct {
 	cancel context.CancelFunc
 	logger zerolog.Logger
 
-	dumping *atomic.Bool
-	running *atomic.Bool
+	dumping  *atomic.Bool
+	running  *atomic.Bool
+	syncedAt *atomic.Int64
 
 	syncCh    chan interface{}
 	closeOnce *sync.Once
@@ -45,6 +46,7 @@ func New(cfg *config.Config, logger zerolog.Logger) (*Bridge, error) {
 		logger:    logger,
 		dumping:   atomic.NewBool(false),
 		running:   atomic.NewBool(false),
+		syncedAt:  atomic.NewInt64(0),
 		syncCh:    make(chan interface{}, eventsBufSize),
 		closeOnce: &sync.Once{},
 	}
@@ -258,11 +260,7 @@ func (b *Bridge) newTarantoolClient(cfg *config.Config) {
 func (b *Bridge) Run() <-chan error {
 	defer b.setRunning(false)
 
-	go func() {
-		for range time.Tick(1 * time.Second) {
-			metrics.SetSecondsBehindMaster(b.Delay())
-		}
-	}()
+	go b.runBackgroundJobs()
 
 	maxErrs := 3
 	errCh := make(chan error, maxErrs)
@@ -328,6 +326,7 @@ func (b *Bridge) syncLoop() error {
 					return err
 				}
 			}
+			b.syncedAt.Store(time.Now().Unix())
 		case <-b.ctx.Done():
 			return nil
 		}
@@ -400,3 +399,21 @@ func (b *Bridge) setDumping(v bool) {
 func (b *Bridge) Dumping() bool {
 	return b.dumping.Load()
 }
+
+func (b *Bridge) runBackgroundJobs() {
+	go func() {
+		for range time.Tick(1 * time.Second) {
+			metrics.SetSecondsBehindMaster(b.Delay())
+		}
+	}()
+
+	go func() {
+		for range time.Tick(1 * time.Second) {
+			syncedAt := b.syncedAt.Load()
+			if syncedAt > 0 {
+				now := time.Now().Unix()
+				metrics.SetSyncedSecondsAgo(now - syncedAt)
+			}
+		}
+	}()
+}

internal/metrics/metrics.go

@@ -14,7 +14,13 @@ var (
 	secondsBehindMaster = prometheus.NewGauge(prometheus.GaugeOpts{
 		Namespace: "mysql2tarantool",
 		Name:      "seconds_behind",
-		Help:      "Current replication lag of the replicator",
+		Help:      "Current replication lag of the replicator. Calculates as diff between current timestamp and last event timestamp. The value updates only after receiving the event.",
+	})
+
+	syncedSecondsAgo = prometheus.NewGauge(prometheus.GaugeOpts{
+		Namespace: "mysql2tarantool",
+		Name:      "synced_seconds_ago",
+		Help:      "Seconds since the last event has been synced",
 	})
 
 	replState = prometheus.NewGauge(prometheus.GaugeOpts{
@@ -27,12 +33,17 @@ var (
 func Init() {
 	prometheus.MustRegister(secondsBehindMaster)
 	prometheus.MustRegister(replState)
+	prometheus.MustRegister(syncedSecondsAgo)
 }
 
 func SetSecondsBehindMaster(value uint32) {
 	secondsBehindMaster.Set(float64(value))
 }
 
+func SetSyncedSecondsAgo(sec int64) {
+	syncedSecondsAgo.Set(float64(sec))
+}
+
 func SetReplicationState(state ReplState) {
 	replState.Set(float64(state))
 }