Merge pull request #38 from monobilisim/feat/mysql-galera-monitoring

feat(mysqlHealth): add Galera Cluster Receive Queue and Flow Control monitoring

Author: ahmetsoguksu

README.md

@@ -153,7 +153,9 @@ These core tools will be available in every monokit installation.
 
 - mysqlHealth
   - Checks MySQL health, including read and write operations.
+  - Supports Galera Cluster monitoring (Receive Queue and Flow Control).
   - Sends alarm notifications to a Slack webhook.
+  - Opens Redmine issues for Galera Flow Control issues.
   - Config: `/etc/mono/db.yaml`
 
 - pgsqlHealth

common/db/main.go

@@ -4,10 +4,12 @@ type Mysql struct {
 	Process_limit int
 	Pmm_enabled   *bool `json:"pmm_enabled,omitempty"`
 	Cluster       struct {
-		Enabled          bool
-		Size             int
-		Check_table_day  string
-		Check_table_hour string
+		Enabled             bool
+		Size                int
+		Check_table_day     string
+		Check_table_hour    string
+		Receive_queue_limit int
+		Flow_control_limit  float64
 	}
 	Alarm struct {
 		Enabled bool

config/db.yml

@@ -9,10 +9,12 @@ postgres:
 
 mysql:
   process_limit: 50
-  cluster: 
+  cluster:
     enabled: false
     size: 3
     check_table_day: "Sun"
     check_table_hour: "05:00"
+    receive_queue_limit: 10
+    flow_control_limit: 0.20
   alarm:
     enabled: true

mysqlHealth/main.go

@@ -150,6 +150,12 @@ func Main(cmd *cobra.Command, args []string) {
 
 		// Check if cluster is synced
 		CheckClusterSynced()
+
+		// Check Galera receive queue
+		CheckReceiveQueue()
+
+		// Check Galera flow control
+		CheckFlowControl()
 	}
 
 	// check if time matches to configured time

mysqlHealth/mysql.go

@@ -8,13 +8,15 @@ import (
 	"fmt"
 	"os"
 	"os/exec"
+	"strconv"
 	"strings"
 	"time"
 
 	"github.com/go-ini/ini"
 	_ "github.com/go-sql-driver/mysql"      // Keep anonymous import for side effects
 	_mysql "github.com/go-sql-driver/mysql" // Import with alias
 	"github.com/monobilisim/monokit/common"
+	issues "github.com/monobilisim/monokit/common/redmine/issues"
 	"github.com/rs/zerolog/log"
 )
 
@@ -554,6 +556,87 @@ func CheckCertificationWaiting() {
 	}
 }
 
+func CheckReceiveQueue() {
+	limit := 10
+	if DbHealthConfig.Mysql.Cluster.Receive_queue_limit != 0 {
+		limit = DbHealthConfig.Mysql.Cluster.Receive_queue_limit
+	}
+
+	query := "SHOW GLOBAL STATUS WHERE Variable_name = 'wsrep_local_recv_queue'"
+	rows, err := Connection.Query(query)
+	if err != nil {
+		log.Error().Err(err).Msg("CheckReceiveQueue query failed")
+		return
+	}
+	defer rows.Close()
+
+	var name string
+	var value string
+	if rows.Next() {
+		if err := rows.Scan(&name, &value); err != nil {
+			log.Error().Err(err).Msg("Error scanning Receive Queue count")
+			return
+		}
+	}
+
+	count, _ := strconv.Atoi(value)
+
+	// Update health data
+	healthData.ClusterInfo.ReceiveQueue.Count = count
+	healthData.ClusterInfo.ReceiveQueue.Limit = limit
+	healthData.ClusterInfo.ReceiveQueue.Exceeded = count > limit
+
+	if count > limit {
+		msg := fmt.Sprintf("Galera Receive Queue değeri %d (Limit: %d)", count, limit)
+		// Only send alarm (webhook), no Redmine issue for Receive Queue as requested
+		common.AlarmCheckDown("receive queue", msg, false, "", "")
+	} else {
+		common.AlarmCheckUp("receive queue", fmt.Sprintf("Receive Queue OK: %d/%d", count, limit), false)
+	}
+}
+
+func CheckFlowControl() {
+	threshold := 0.2
+	if DbHealthConfig.Mysql.Cluster.Flow_control_limit != 0 {
+		threshold = DbHealthConfig.Mysql.Cluster.Flow_control_limit
+	}
+
+	query := "SHOW GLOBAL STATUS WHERE Variable_name = 'wsrep_flow_control_paused'"
+	rows, err := Connection.Query(query)
+	if err != nil {
+		log.Error().Err(err).Msg("CheckFlowControl query failed")
+		return
+	}
+	defer rows.Close()
+
+	var name string
+	var value string
+	if rows.Next() {
+		if err := rows.Scan(&name, &value); err != nil {
+			log.Error().Err(err).Msg("Error scanning Flow Control Paused value")
+			return
+		}
+	}
+
+	paused, _ := strconv.ParseFloat(value, 64)
+
+	// Update health data
+	healthData.ClusterInfo.FlowControlPaused = paused
+	healthData.ClusterInfo.FlowControlLimit = threshold
+
+	if paused > threshold {
+		msg := fmt.Sprintf("Galera Flow Control duraklama oranı %.4f (Limit: %.2f)", paused, threshold)
+		subject := fmt.Sprintf("%s için Galera Flow Control duraklama oranı %.2f üstüne çıktı", common.Config.Identifier, threshold)
+
+		common.AlarmCheckDown("flow control", msg, false, "", "")
+		issues.CheckDown("flow-control", subject, msg, false, 0)
+	} else {
+		msg := fmt.Sprintf("%s için Galera Flow Control duraklama oranı %.2f altına düştü", common.Config.Identifier, threshold)
+		common.AlarmCheckUp("flow control", "Flow Control OK", false)
+		issues.CheckUp("flow-control", msg)
+	}
+}
+
 func checkPMM() {
 	// Check if PMM monitoring is enabled in config (default: enabled)
 	if DbHealthConfig.Mysql.Pmm_enabled != nil && !*DbHealthConfig.Mysql.Pmm_enabled {

mysqlHealth/ui.go

@@ -40,6 +40,13 @@ type CertificationWaitingInfo struct {
 	Exceeded bool
 }
 
+// ReceiveQueueInfo contains information about Galera receive queue
+type ReceiveQueueInfo struct {
+	Count    int
+	Limit    int
+	Exceeded bool
+}
+
 // ClusterInfo contains information about MySQL cluster status
 type ClusterInfo struct {
 	Enabled           bool
@@ -48,6 +55,9 @@ type ClusterInfo struct {
 	Status            string
 	Nodes             []NodeInfo
 	Synced            bool
+	ReceiveQueue      ReceiveQueueInfo
+	FlowControlPaused float64
+	FlowControlLimit  float64
 }
 
 // NodeInfo represents a MySQL cluster node
@@ -69,6 +79,10 @@ func NewMySQLHealthData() *MySQLHealthData {
 	return &MySQLHealthData{
 		ClusterInfo: ClusterInfo{
 			Nodes: []NodeInfo{},
+			ReceiveQueue: ReceiveQueueInfo{
+				Limit: 10,
+			},
+			FlowControlLimit: 0.2,
 		},
 	}
 }
@@ -162,6 +176,36 @@ func (m *MySQLHealthData) RenderCompact() string {
 			"Cluster Nodes",
 			nodesAccessStatus,
 			isInaccessibleOK))
+		sb.WriteString("\n")
+
+		// Receive Queue status
+		isQueueOK := !m.ClusterInfo.ReceiveQueue.Exceeded
+		queueStatus := "within limit"
+		if !isQueueOK {
+			queueStatus = "exceeds limit"
+		}
+
+		sb.WriteString(common.StatusListItem(
+			"Receive Queue",
+			queueStatus,
+			fmt.Sprintf("%d", m.ClusterInfo.ReceiveQueue.Limit),
+			fmt.Sprintf("%d", m.ClusterInfo.ReceiveQueue.Count),
+			isQueueOK))
+		sb.WriteString("\n")
+
+		// Flow Control status
+		isFlowOK := m.ClusterInfo.FlowControlPaused <= m.ClusterInfo.FlowControlLimit
+		flowStatus := "within limit"
+		if !isFlowOK {
+			flowStatus = "exceeds limit"
+		}
+
+		sb.WriteString(common.StatusListItem(
+			"Flow Control",
+			flowStatus,
+			fmt.Sprintf("%.2f", m.ClusterInfo.FlowControlLimit),
+			fmt.Sprintf("%.2f", m.ClusterInfo.FlowControlPaused),
+			isFlowOK))
 
 		// Add cluster size display with custom color logic
 		clusterSize := m.ClusterInfo.ClusterSize