Add raft/health Prometheus metrics (rebased #105)

go/logic/orchestrator.go

@@ -90,7 +90,11 @@ func init() {
 	_ = metrics.Register("raft.is_leader", isRaftLeaderGauge)
 
 	ometrics.OnMetricsTick(func() {
-		discoveryQueueLengthGauge.Update(int64(discoveryQueue.QueueLen()))
+		if discoveryQueue != nil {
+			queueLen := discoveryQueue.QueueLen()
+			discoveryQueueLengthGauge.Update(int64(queueLen))
+			ometrics.DiscoveryQueueLength.Set(float64(queueLen))
+		}
 	})
 	ometrics.OnMetricsTick(func() {
 		if recentDiscoveryOperationKeys == nil {
@@ -116,13 +120,16 @@ func init() {
 	})
 }
 
+// IsLeader returns true if the current orchestrator instance is the cluster leader.
 func IsLeader() bool {
 	if orcraft.IsRaftEnabled() {
 		return orcraft.IsLeader()
 	}
 	return atomic.LoadInt64(&isElectedNode) == 1
 }
 
+// IsLeaderOrActive returns true if the current orchestrator instance is either the cluster leader
+// or an active member (when Raft is enabled, returns true if part of quorum).
 func IsLeaderOrActive() bool {
 	if orcraft.IsRaftEnabled() {
 		return orcraft.IsPartOfQuorum()
@@ -192,6 +199,7 @@ func handleDiscoveryRequests() {
 		_ = metrics.Register("discoveries.dead_instances_queue_length", deadInstancesDiscoveryQueueLengthGauge)
 		ometrics.OnMetricsTick(func() {
 			deadInstancesDiscoveryQueueLengthGauge.Update(int64(deadInstancesDiscoveryQueue.QueueLen()))
+			ometrics.DeadInstancesDiscoveryQueueLength.Set(float64(deadInstancesDiscoveryQueue.QueueLen()))
 		})
 
 		// create a pool of discovery workers
@@ -270,6 +278,7 @@ func DiscoverInstance(instanceKey inst.InstanceKey) {
 	}
 
 	discoveriesCounter.Inc(1)
+	ometrics.DiscoveriesTotal.Inc()
 
 	// First we've ever heard of this instance. Continue investigation:
 	var err error
@@ -290,6 +299,7 @@ func DiscoverInstance(instanceKey inst.InstanceKey) {
 
 	if instance == nil {
 		failedDiscoveriesCounter.Inc(1)
+		ometrics.DiscoveryErrorsTotal.Inc()
 		_ = discoveryMetrics.Append(&discovery.Metric{
 			Timestamp:       time.Now(),
 			InstanceKey:     instanceKey,
@@ -581,6 +591,7 @@ func ContinuousDiscovery() {
 	raftCaretakingTick := time.Tick(10 * time.Minute)
 	recoveryTick := time.Tick(time.Duration(config.RecoveryPollSeconds) * time.Second)
 	autoPseudoGTIDTick := time.Tick(time.Duration(config.PseudoGTIDIntervalSeconds) * time.Second)
+	dbMetricsTick := time.Tick(30 * time.Second)
 	var recoveryEntrance int64
 	var snapshotTopologiesTick <-chan time.Time
 	if config.Config.SnapshotTopologiesIntervalHours > 0 {
@@ -612,6 +623,35 @@ func ContinuousDiscovery() {
 	log.Infof("continuous discovery: starting")
 	for {
 		select {
+		case <-dbMetricsTick:
+			go func() {
+				if instances, err := inst.ReadDowntimedInstances(""); err == nil {
+					ometrics.DowntimedInstances.Set(float64(len(instances)))
+				}
+				if recoveries, err := ReadActiveRecoveries(); err == nil {
+					ometrics.ActiveRecoveries.Set(float64(len(recoveries)))
+				}
+				// Topology issue gauges: reuse discovery analysis. Cost is similar to a
+				// recovery tick; failures are ignored so metrics never block the loop.
+				if analysis, err := inst.GetReplicationAnalysis("", &inst.ReplicationAnalysisHints{}); err == nil {
+					issuesCount := make(map[string]int)
+					for _, a := range analysis {
+						issuesCount[string(a.Analysis)]++
+					}
+					ometrics.ActiveTopologyIssues.Reset()
+					for issueType, count := range issuesCount {
+						ometrics.ActiveTopologyIssues.WithLabelValues(issueType).Set(float64(count))
+					}
+				}
+				if clusters, err := inst.ReadClustersInfo(""); err == nil {
+					ometrics.ClustersTotal.Set(float64(len(clusters)))
+					var instancesTotal uint
+					for _, c := range clusters {
+						instancesTotal += c.CountInstances
+					}
+					ometrics.InstancesTotal.Set(float64(instancesTotal))
+				}
+			}()
 		case <-healthTick:
 			go func() {
 				onHealthTick()

go/logic/topology_recovery.go

@@ -1895,7 +1895,20 @@ func executeCheckAndRecoverFunction(analysisEntry inst.ReplicationAnalysis, cand
 	if isActionableRecovery || util.ClearToLog("executeCheckAndRecoverFunction: recovery", analysisEntry.AnalyzedInstanceKey.StringCode()) {
 		log.Infof("executeCheckAndRecoverFunction: proceeding with %+v recovery on %+v; isRecoverable?: %+v; skipProcesses: %+v", analysisEntry.Analysis, analysisEntry.AnalyzedInstanceKey, isActionableRecovery, skipProcesses)
 	}
+	startTime := time.Now()
 	recoveryAttempted, topologyRecovery, err = checkAndRecoverFunction(analysisEntry, candidateInstanceKey, forceInstanceRecovery, skipProcesses)
+
+	if recoveryAttempted && topologyRecovery != nil {
+		duration := time.Since(startTime).Seconds()
+		ometrics.RecoveryDurationSeconds.Observe(duration)
+
+		result := "success"
+		if !topologyRecovery.IsSuccessful {
+			result = "failed"
+		}
+		ometrics.RecoveriesTotal.WithLabelValues(string(analysisEntry.Analysis), result).Inc()
+	}
+
 	if !recoveryAttempted {
 		return recoveryAttempted, topologyRecovery, err
 	}

go/metrics/prometheus.go

@@ -17,9 +17,14 @@
 package metrics
 
 import (
+	"sync/atomic"
+	"time"
+
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/proxysql/golib/log"
 	"github.com/proxysql/orchestrator/go/config"
+	"github.com/proxysql/orchestrator/go/process"
+	orcraft "github.com/proxysql/orchestrator/go/raft"
 )
 
 // Prometheus metric variables, exported so other packages can update them.
@@ -54,6 +59,104 @@ var (
 		Help:    "Duration of recovery operations in seconds.",
 		Buckets: prometheus.ExponentialBuckets(0.5, 2, 12), // 0.5s to ~1024s
 	})
+
+	HealthIsReady = prometheus.NewGaugeFunc(prometheus.GaugeOpts{
+		Name: "orchestrator_health_is_ready",
+		Help: "1 if the node is ready, 0 otherwise.",
+	}, func() float64 {
+		if atomic.LoadInt64(&process.LastContinousCheckHealthy) == 1 {
+			return 1
+		}
+		return 0
+	})
+
+	HealthIsLive = prometheus.NewGaugeFunc(prometheus.GaugeOpts{
+		Name: "orchestrator_health_is_live",
+		Help: "1 if the process recently ran a health check tick (liveness), 0 otherwise.",
+	}, func() float64 {
+		// Liveness: health subsystem has run recently (even if the check failed).
+		// Readiness is LastContinousCheckHealthy (successful registration).
+		maxAge := time.Duration(config.HealthPollSeconds*3) * time.Second
+		if d := process.SinceLastHealthCheck(); d > 0 && d <= maxAge {
+			return 1
+		}
+		if d := process.SinceLastGoodHealthCheck(); d > 0 && d <= maxAge {
+			return 1
+		}
+		// ContinuousRegistration updates LastContinousCheckHealthy without touching
+		// SinceLastHealthCheck; treat a current healthy flag as live too.
+		if atomic.LoadInt64(&process.LastContinousCheckHealthy) == 1 {
+			return 1
+		}
+		return 0
+	})
+
+	RaftIsHealthy = prometheus.NewGaugeFunc(prometheus.GaugeOpts{
+		Name: "orchestrator_raft_is_healthy",
+		Help: "1 if raft is enabled and this node is healthy; 0 if unhealthy or raft disabled.",
+	}, func() float64 {
+		if orcraft.IsRaftEnabled() && orcraft.IsHealthy() {
+			return 1
+		}
+		return 0
+	})
+
+	RaftIsLeader = prometheus.NewGaugeFunc(prometheus.GaugeOpts{
+		Name: "orchestrator_raft_is_leader",
+		Help: "1 if the node is the raft leader, 0 otherwise.",
+	}, func() float64 {
+		if orcraft.IsRaftEnabled() && orcraft.IsLeader() {
+			return 1
+		}
+		return 0
+	})
+
+	DiscoveryQueueLength = prometheus.NewGauge(prometheus.GaugeOpts{
+		Name: "orchestrator_discovery_queue_length",
+		Help: "Length of the discovery queue.",
+	})
+
+	DeadInstancesDiscoveryQueueLength = prometheus.NewGauge(prometheus.GaugeOpts{
+		Name: "orchestrator_dead_instances_discovery_queue_length",
+		Help: "Length of the dead instances discovery queue.",
+	})
+
+	ActiveRecoveries = prometheus.NewGauge(prometheus.GaugeOpts{
+		Name: "orchestrator_active_recoveries",
+		Help: "Number of active (in-progress) recoveries.",
+	})
+
+	DowntimedInstances = prometheus.NewGauge(prometheus.GaugeOpts{
+		Name: "orchestrator_downtimed_instances_total",
+		Help: "Number of instances currently marked as downtimed.",
+	})
+
+	ActiveTopologyIssues = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "orchestrator_active_topology_issues",
+		Help: "Number of active topology issues, broken down by issue type.",
+	}, []string{"issue_type"})
+
+	RaftPeersTotal = prometheus.NewGaugeFunc(prometheus.GaugeOpts{
+		Name: "orchestrator_raft_peers_total",
+		Help: "Number of known raft peers.",
+	}, func() float64 {
+		if orcraft.IsRaftEnabled() {
+			if peers, err := orcraft.GetPeers(); err == nil {
+				return float64(len(peers))
+			}
+		}
+		return 0
+	})
+
+	RaftIsPartOfQuorum = prometheus.NewGaugeFunc(prometheus.GaugeOpts{
+		Name: "orchestrator_raft_is_part_of_quorum",
+		Help: "1 if the node is part of raft quorum, 0 otherwise.",
+	}, func() float64 {
+		if orcraft.IsRaftEnabled() && orcraft.IsPartOfQuorum() {
+			return 1
+		}
+		return 0
+	})
 )
 
 // InitPrometheus registers all Prometheus metrics. Should be called once at
@@ -70,5 +173,16 @@ func InitPrometheus() {
 		ClustersTotal,
 		RecoveriesTotal,
 		RecoveryDurationSeconds,
+		HealthIsReady,
+		HealthIsLive,
+		RaftIsHealthy,
+		RaftIsLeader,
+		DiscoveryQueueLength,
+		DeadInstancesDiscoveryQueueLength,
+		ActiveRecoveries,
+		DowntimedInstances,
+		ActiveTopologyIssues,
+		RaftPeersTotal,
+		RaftIsPartOfQuorum,
 	)
 }