mysql_throttle_metric.go

/*
   Copyright 2017 GitHub Inc.
	 See https://github.com/github/freno/blob/master/LICENSE
*/

package mysql

import (
	"errors"
	"fmt"
	"strings"
	"time"

	"github.com/go-sql-driver/mysql"
	"github.com/outbrain/golib/sqlutils"
	"github.com/patrickmn/go-cache"
	metrics "github.com/rcrowley/go-metrics"
)

var mysqlMetricCache = cache.New(cache.NoExpiration, 10*time.Millisecond)

func getMySQLMetricCacheKey(probe *Probe) string {
	return fmt.Sprintf("%s:%s", probe.Key, probe.MetricQuery)
}

func cacheMySQLThrottleMetric(probe *Probe, mySQLThrottleMetric *MySQLThrottleMetric) *MySQLThrottleMetric {
	if mySQLThrottleMetric.Err != nil {
		return mySQLThrottleMetric
	}
	if probe.CacheMillis > 0 {
		mysqlMetricCache.Set(getMySQLMetricCacheKey(probe), mySQLThrottleMetric, time.Duration(probe.CacheMillis)*time.Millisecond)
	}
	return mySQLThrottleMetric
}

func getCachedMySQLThrottleMetric(probe *Probe) *MySQLThrottleMetric {
	if probe.CacheMillis == 0 {
		return nil
	}
	if metric, found := mysqlMetricCache.Get(getMySQLMetricCacheKey(probe)); found {
		mySQLThrottleMetric, _ := metric.(*MySQLThrottleMetric)
		return mySQLThrottleMetric
	}
	return nil
}

type MySQLThrottleMetric struct {
	ClusterName string
	Key         InstanceKey
	Value       float64
	Err         error
}

func NewMySQLThrottleMetric() *MySQLThrottleMetric {
	return &MySQLThrottleMetric{Value: 0}
}

func (metric *MySQLThrottleMetric) GetClusterInstanceKey() ClusterInstanceKey {
	return GetClusterInstanceKey(metric.ClusterName, &metric.Key)
}

func (metric *MySQLThrottleMetric) HashCode() string {
	return metric.GetClusterInstanceKey().HashCode()
}

func (metric *MySQLThrottleMetric) Get() (float64, error) {
	return metric.Value, metric.Err
}

// ReadThrottleMetric returns replication lag for a given connection config; either by explicit query
// or via SHOW REPLICA STATUS / SHOW SLAVE STATUS
func ReadThrottleMetric(probe *Probe, clusterName string) (mySQLThrottleMetric *MySQLThrottleMetric) {
	if mySQLThrottleMetric := getCachedMySQLThrottleMetric(probe); mySQLThrottleMetric != nil {
		return mySQLThrottleMetric
		// On cached results we avoid taking latency metrics
	}

	started := time.Now()
	mySQLThrottleMetric = NewMySQLThrottleMetric()
	mySQLThrottleMetric.ClusterName = clusterName
	mySQLThrottleMetric.Key = probe.Key

	defer func(metric *MySQLThrottleMetric, started time.Time) {
		go func() {
			metrics.GetOrRegisterTimer("probes.latency", nil).Update(time.Since(started))
			metrics.GetOrRegisterCounter("probes.total", nil).Inc(1)
			if metric.Err != nil {
				metrics.GetOrRegisterCounter("probes.error", nil).Inc(1)
			}
		}()
	}(mySQLThrottleMetric, started)

	dbUri := probe.GetDBUri("information_schema")
	db, fromCache, err := sqlutils.GetDB(dbUri)

	if err != nil {
		mySQLThrottleMetric.Err = err
		return mySQLThrottleMetric
	}
	if !fromCache {
		db.SetMaxOpenConns(maxPoolConnections)
		db.SetMaxIdleConns(maxIdleConnections)
	}
	if strings.HasPrefix(strings.ToLower(probe.MetricQuery), "select") {
		mySQLThrottleMetric.Err = db.QueryRow(probe.MetricQuery).Scan(&mySQLThrottleMetric.Value)
		return cacheMySQLThrottleMetric(probe, mySQLThrottleMetric)
	}

	if strings.HasPrefix(strings.ToLower(probe.MetricQuery), "show global") {
		var variableName string // just a placeholder
		mySQLThrottleMetric.Err = db.QueryRow(probe.MetricQuery).Scan(&variableName, &mySQLThrottleMetric.Value)
		return cacheMySQLThrottleMetric(probe, mySQLThrottleMetric)
	}

	if probe.MetricQuery != "" {
		mySQLThrottleMetric.Err = fmt.Errorf("Unsupported metrics query type: %s", probe.MetricQuery)
		return mySQLThrottleMetric
	}

	// No metric query? By default we look at replication lag.
	// Try SHOW REPLICA STATUS first (MySQL 8.0.22+, required in 8.4+), fall back to
	// SHOW SLAVE STATUS for older MySQL versions that don't recognise the new syntax.

	mySQLThrottleMetric.Err = sqlutils.QueryRowsMap(db, `show replica status`, func(m sqlutils.RowMap) error {
		replicaIORunning := m.GetString("Replica_IO_Running")
		replicaSQLRunning := m.GetString("Replica_SQL_Running")
		secondsBehindSource := m.GetNullInt64("Seconds_Behind_Source")
		if !secondsBehindSource.Valid {
			return fmt.Errorf("replication not running; Replica_IO_Running=%+v, Replica_SQL_Running=%+v", replicaIORunning, replicaSQLRunning)
		}
		mySQLThrottleMetric.Value = float64(secondsBehindSource.Int64)
		return nil
	})

	// MySQL error 1064 means syntax error — the server doesn't understand SHOW REPLICA STATUS
	// (MySQL < 8.0.22). Fall back to the legacy SHOW SLAVE STATUS command.
	if mySQLThrottleMetric.Err != nil {
		var mysqlErr *mysql.MySQLError
		if errors.As(mySQLThrottleMetric.Err, &mysqlErr) && mysqlErr.Number == 1064 {
			originalErr := mySQLThrottleMetric.Err
			fallbackErr := sqlutils.QueryRowsMap(db, `show slave status`, func(m sqlutils.RowMap) error {
				slaveIORunning := m.GetString("Slave_IO_Running")
				slaveSQLRunning := m.GetString("Slave_SQL_Running")
				secondsBehindMaster := m.GetNullInt64("Seconds_Behind_Master")
				if !secondsBehindMaster.Valid {
					return fmt.Errorf("replication not running; Slave_IO_Running=%+v, Slave_SQL_Running=%+v", slaveIORunning, slaveSQLRunning)
				}
				mySQLThrottleMetric.Value = float64(secondsBehindMaster.Int64)
				return nil
			})
			if fallbackErr == nil {
				mySQLThrottleMetric.Err = nil
			} else {
				// Both commands failed; surface the original error as it's more informative.
				mySQLThrottleMetric.Err = originalErr
			}
		}
		// Non-syntax errors (permissions, connectivity, replication issues) are kept as-is.
	}

	return cacheMySQLThrottleMetric(probe, mySQLThrottleMetric)
}