diff --git a/doc/mysql.md b/doc/mysql.md index c75424b..33c0a64 100644 --- a/doc/mysql.md +++ b/doc/mysql.md @@ -203,7 +203,8 @@ primary-load and ProxySQL-capacity checks without enforcing them: "prod4": { "ShadowRequiredClusters": [ "prod4-primary", - "prod4-proxysql" + "prod4-proxysql-frontend", + "prod4-proxysql-writer-backend" ], "HAProxySettings": { "Host": "my.haproxy.mydomain.com", @@ -224,25 +225,54 @@ primary-load and ProxySQL-capacity checks without enforcing them: ] } }, - "prod4-proxysql": { + "prod4-proxysql-frontend": { "User": "${proxysql_stats_user}", "Password": "${proxysql_stats_password}", - "MetricQuery": "select connection_pressure_percent from operational_metrics.proxysql_capacity limit 1", - "CacheMillis": 500, + "Port": 6032, "ThrottleThreshold": 80, "RecoveryThreshold": 60, - "RecoveryDurationMillis": 5000, + "RecoveryDurationMillis": 30000, "FailOnNoHosts": true, + "ProxySQLCapacitySettings": { + "Metric": "frontend", + "FrontendMaxConnections": 200000, + "WriterHostgroupID": 30, + "BackendLimits": [ + {"Host": "/var/run/mysqld/mysqld.sock", "Port": 0, "MaxConnections": 1000} + ] + }, "StaticHostsSettings": { - "Hosts": [ - "my.prod4.proxysql.example.com:6032" + "Hosts": ["my.prod4.proxysql.example.com"] + } + }, + "prod4-proxysql-writer-backend": { + "User": "${proxysql_stats_user}", + "Password": "${proxysql_stats_password}", + "Port": 6032, + "ThrottleThreshold": 80, + "RecoveryThreshold": 60, + "RecoveryDurationMillis": 30000, + "FailOnNoHosts": true, + "ProxySQLCapacitySettings": { + "Metric": "writer-backend", + "FrontendMaxConnections": 200000, + "WriterHostgroupID": 30, + "BackendLimits": [ + {"Host": "/var/run/mysqld/mysqld.sock", "Port": 0, "MaxConnections": 1000} ] + }, + "StaticHostsSettings": { + "Hosts": ["my.prod4.proxysql.example.com"] } } } ``` -The ProxySQL query above is illustrative: deployments must provide a scalar query or operational view whose value increases as usable connection capacity is consumed. +The capacity limits and thresholds above are examples, not production-approved +values. Capacity probes execute built-in queries against ProxySQL's `stats` +database, not an application view or a MySQL backend. They require explicit +stats credentials and do not inherit MySQL credentials, queries, host ignores, +or the legacy fail-open no-host behavior. - `RequiredClusters` lists additional configured metrics that must return `HTTP 200`. Missing required runtime metrics fail closed. - `ShadowRequiredClusters` lists prospective dependencies, evaluated only after @@ -308,6 +338,63 @@ zero on its first observation; dependency/status counters appear only on failures. Counters reset on restart and do not replace probe health, discovery, or coverage validation. +### Direct ProxySQL capacity sampling + +`ProxySQLCapacitySettings.Metric` selects `frontend` or `writer-backend`. +Frontend utilization is `100 * Client_Connections_connected / +FrontendMaxConnections`. Writer utilization is the maximum of `100 * ConnUsed / +MaxConnections` across the expected backend endpoints in `WriterHostgroupID`. +`ConnFree` is not the configured capacity; `ConnUsed / (ConnUsed + ConnFree)` is +not used. Unknown ONLINE backends, duplicate endpoints, missing expected +backends, non-ONLINE expected backends, invalid counters, and query failures +reject writer capacity checks. Frontend and backend checks must both be attached +when enforcing full connection-capacity protection. + +The stats-only account cannot read runtime configuration tables. Operators must +verify every configured limit against runtime before enabling collection, and +revalidate limits whenever backend membership or ProxySQL configuration changes. +All proxies discovered by one configured metric must have matching verified +limits and backend endpoint mappings; otherwise split them into explicit probes. +There is no automatic runtime-limit drift detection. + +Only the leader samples. Both metrics share one reusable admin connection and +one five-second snapshot per proxy, credential set, and writer hostgroup. +A complete snapshot uses two small stats queries, at most 0.4 queries/second in +steady state. Failed polls replace the previous snapshot and are cached for the +same five seconds; application requests cannot trigger a hot retry loop. +The entire two-query batch has a one-second deadline, with one-second dial, +read, and write limits. Inactive endpoint pools are retired after ten minutes. + +An admin port must be explicit or default to 6032; common application ports +3306 and 6033 are rejected. For capacity probes, discovered port numbers are +replaced by the configured admin port, including Vitess tablet ports. HAProxy +and Vitess discovery are only appropriate after confirming that ProxySQL is +colocated with those database hosts. Detached deployments require an explicit +proxy roster. `ProxySQLSettings` is not suitable: it discovers MySQL backends. +Capacity probes do not inherit the replica HTTP check or honor host-skipping +settings that could hide a saturated proxy. + +These queries add no SQL or connections to the MySQL backend. They still consume +ProxySQL CPU/memory, including shared resources on a colocated primary host. +Measure that overhead during shadow rollout; do not claim zero host overhead. +No listener, permissions, deployment, or enforcement is enabled by adding code. + +The opt-in integration test uses an isolated ProxySQL fixture with monitoring +disabled and a listening backend sentinel. It verifies successful stats-only +sampling while the sentinel receives zero connections: + +```bash +docker run --rm --name freno-capacity-test \ + --mount type=bind,src="$PWD/pkg/proxysql/testdata/capacity.cnf",dst=/etc/proxysql.cnf,readonly \ + proxysql/proxysql:2.7.3 +# In another shell; no container ports need to be published: +docker run --rm --network container:freno-capacity-test \ + --mount type=bind,src="$PWD",dst=/source,readonly -w /source \ + -e FRENO_PROXYSQL_CAPACITY_INTEGRATION=1 golang:1.20 \ + go test -mod=vendor ./pkg/proxysql -run TestCapacityIntegrationNoBackendConnections -count=1 +docker stop freno-capacity-test +``` + ### Rolling out primary safety checks Deploy a binary containing these settings before deploying configuration that diff --git a/pkg/config/mysql_config.go b/pkg/config/mysql_config.go index 53d7e23..f2d5a19 100644 --- a/pkg/config/mysql_config.go +++ b/pkg/config/mysql_config.go @@ -30,10 +30,11 @@ type MySQLClusterConfigurationSettings struct { HttpCheckPath string // Specify if different than specified by MySQLConfigurationSettings IgnoreHosts []string // override MySQLConfigurationSettings's, or leave empty to inherit those settings - HAProxySettings HAProxyConfigurationSettings // If list of servers is to be acquired via HAProxy, provide this field - ProxySQLSettings ProxySQLConfigurationSettings // If list of servers is to be acquired via ProxySQL, provide this field - VitessSettings VitessConfigurationSettings // If list of servers is to be acquired via Vitess, provide this field - StaticHostsSettings StaticHostsConfigurationSettings + HAProxySettings HAProxyConfigurationSettings // If list of servers is to be acquired via HAProxy, provide this field + ProxySQLSettings ProxySQLConfigurationSettings // If list of servers is to be acquired via ProxySQL, provide this field + VitessSettings VitessConfigurationSettings // If list of servers is to be acquired via Vitess, provide this field + StaticHostsSettings StaticHostsConfigurationSettings + ProxySQLCapacitySettings *ProxySQLCapacityConfigurationSettings } // Hook to implement adjustments after reading each configuration file. @@ -118,6 +119,15 @@ func (settings *MySQLConfigurationSettings) postReadAdjustments() error { return fmt.Errorf("Stores.MySQL.Clusters.%s.VitessSettings: %w", clusterName, err) } } + if clusterSettings.ProxySQLCapacitySettings != nil { + if !clusterSettings.VitessSettings.IsEmpty() && len(clusterSettings.VitessSettings.Cells) == 0 { + clusterSettings.VitessSettings.Cells = settings.VitessCells + } + if err := clusterSettings.adjustProxySQLCapacity(); err != nil { + return fmt.Errorf("Stores.MySQL.Clusters.%s.ProxySQLCapacitySettings: %w", clusterName, err) + } + continue + } if clusterSettings.User == "" { clusterSettings.User = settings.User } diff --git a/pkg/config/proxysql_capacity_config.go b/pkg/config/proxysql_capacity_config.go new file mode 100644 index 0000000..3b41f87 --- /dev/null +++ b/pkg/config/proxysql_capacity_config.go @@ -0,0 +1,101 @@ +package config + +import ( + "fmt" + "math" +) + +const ProxySQLDefaultAdminPort = 6032 + +type ProxySQLBackendLimit struct { + Host string + Port int + MaxConnections int64 +} + +// ProxySQLCapacityConfigurationSettings uses explicitly verified limits because +// ProxySQL's stats-only account cannot read the runtime configuration tables. +type ProxySQLCapacityConfigurationSettings struct { + Metric string + FrontendMaxConnections int64 + WriterHostgroupID int + BackendLimits []ProxySQLBackendLimit +} + +func (settings *ProxySQLCapacityConfigurationSettings) Validate() error { + if settings.Metric != "frontend" && settings.Metric != "writer-backend" { + return fmt.Errorf("Metric must be frontend or writer-backend") + } + if settings.FrontendMaxConnections <= 0 || settings.WriterHostgroupID <= 0 { + return fmt.Errorf("FrontendMaxConnections and WriterHostgroupID must be positive") + } + if len(settings.BackendLimits) == 0 { + return fmt.Errorf("BackendLimits must identify the expected writer backends") + } + seen := make(map[string]bool) + for _, backend := range settings.BackendLimits { + if backend.Host == "" || backend.Port < 0 || backend.Port > 65535 || backend.MaxConnections <= 0 { + return fmt.Errorf("BackendLimits must contain a host, valid port (0 for a Unix socket), and positive MaxConnections") + } + key := fmt.Sprintf("%s:%d", backend.Host, backend.Port) + if seen[key] { + return fmt.Errorf("duplicate BackendLimits endpoint %q", key) + } + seen[key] = true + } + return nil +} + +func (settings *MySQLClusterConfigurationSettings) adjustProxySQLCapacity() error { + if err := settings.ProxySQLCapacitySettings.Validate(); err != nil { + return err + } + if settings.User == "" || settings.Password == "" { + return fmt.Errorf("explicit stats-only User and Password are required; MySQL credentials are not inherited") + } + if settings.MetricQuery != "" { + return fmt.Errorf("MetricQuery must be empty; capacity probes only execute built-in ProxySQL stats queries") + } + if settings.Port == 0 { + settings.Port = ProxySQLDefaultAdminPort + } + if settings.Port < 1 || settings.Port > 65535 || settings.Port == 3306 || settings.Port == 6033 { + return fmt.Errorf("Port must identify a ProxySQL admin listener, not a MySQL or ProxySQL application port") + } + if len(settings.ProxySQLSettings.Addresses) > 0 || settings.ProxySQLSettings.HostgroupID != 0 { + return fmt.Errorf("ProxySQLSettings discovers MySQL backends, not ProxySQL capacity endpoints") + } + discoverySources := 0 + for _, present := range []bool{ + !settings.HAProxySettings.IsEmpty(), + !settings.VitessSettings.IsEmpty(), + !settings.StaticHostsSettings.IsEmpty(), + } { + if present { + discoverySources++ + } + } + if discoverySources != 1 { + return fmt.Errorf("exactly one HAProxy, Vitess, or static proxy-host discovery source is required") + } + if math.IsNaN(settings.ThrottleThreshold) || math.IsInf(settings.ThrottleThreshold, 0) || + settings.ThrottleThreshold <= 0 || settings.ThrottleThreshold > 100 { + return fmt.Errorf("ThrottleThreshold must be a finite percentage in (0, 100]") + } + if settings.RecoveryThreshold == nil || math.IsNaN(*settings.RecoveryThreshold) || + math.IsInf(*settings.RecoveryThreshold, 0) || *settings.RecoveryThreshold < 0 || + *settings.RecoveryThreshold >= settings.ThrottleThreshold { + return fmt.Errorf("RecoveryThreshold must be a finite non-negative percentage below ThrottleThreshold") + } + if settings.RecoveryDurationMillis < 10000 { + return fmt.Errorf("RecoveryDurationMillis must be at least 10000 (two sampling intervals)") + } + if settings.IgnoreHostsCount != 0 || settings.IgnoreHostsThreshold != 0 || len(settings.IgnoreHosts) != 0 { + return fmt.Errorf("capacity probes must not ignore hosts or errors") + } + settings.FailOnNoHosts = true + settings.HttpCheckPort = -1 + settings.HttpCheckPath = "" + settings.CacheMillis = 0 // The sampler shares a bounded success/error cache across both capacity metrics. + return nil +} diff --git a/pkg/config/proxysql_capacity_config_test.go b/pkg/config/proxysql_capacity_config_test.go new file mode 100644 index 0000000..46de2a0 --- /dev/null +++ b/pkg/config/proxysql_capacity_config_test.go @@ -0,0 +1,75 @@ +package config + +import ( + "strings" + "testing" +) + +func capacityClusterSettings() *MySQLClusterConfigurationSettings { + recovery := 70.0 + return &MySQLClusterConfigurationSettings{ + User: "stats", Password: "test-only", ThrottleThreshold: 90, + RecoveryThreshold: &recovery, RecoveryDurationMillis: 30000, + StaticHostsSettings: StaticHostsConfigurationSettings{Hosts: []string{"proxy.example"}}, + ProxySQLCapacitySettings: &ProxySQLCapacityConfigurationSettings{ + Metric: "frontend", FrontendMaxConnections: 200000, WriterHostgroupID: 30, + BackendLimits: []ProxySQLBackendLimit{{Host: "/var/run/mysqld/mysqld.sock", Port: 0, MaxConnections: 1000}}, + }, + } +} + +func TestCapacityDoesNotInheritMySQLProbeSettings(t *testing.T) { + cluster := capacityClusterSettings() + settings := MySQLConfigurationSettings{ + User: "mysql-user", Password: "mysql-password", Port: 3306, + MetricQuery: "SELECT mysql_only", IgnoreHostsCount: 1, IgnoreHosts: []string{"proxy"}, + HttpCheckPort: 9210, CacheMillis: 20000, + Clusters: map[string]*MySQLClusterConfigurationSettings{"capacity": cluster}, + } + if err := settings.postReadAdjustments(); err != nil { + t.Fatal(err) + } + if cluster.Port != 6032 || cluster.MetricQuery != "" || cluster.IgnoreHostsCount != 0 || + len(cluster.IgnoreHosts) != 0 || cluster.HttpCheckPort != -1 || cluster.CacheMillis != 0 || !cluster.FailOnNoHosts { + t.Fatalf("capacity inherited unsafe MySQL settings: %+v", cluster) + } +} + +func TestCapacityConfigurationRejectsUnsafeInputs(t *testing.T) { + tests := []struct { + name string + change func(*MySQLClusterConfigurationSettings) + }{ + {"missing user", func(c *MySQLClusterConfigurationSettings) { c.User = "" }}, + {"missing password environment", func(c *MySQLClusterConfigurationSettings) { c.Password = "${FRENO_TEST_MISSING_STATS_PASSWORD}" }}, + {"MySQL port", func(c *MySQLClusterConfigurationSettings) { c.Port = 3306 }}, + {"application port", func(c *MySQLClusterConfigurationSettings) { c.Port = 6033 }}, + {"custom query", func(c *MySQLClusterConfigurationSettings) { c.MetricQuery = "SELECT mysql_only" }}, + {"missing discovery", func(c *MySQLClusterConfigurationSettings) { c.StaticHostsSettings.Hosts = nil }}, + {"backend discovery", func(c *MySQLClusterConfigurationSettings) { c.ProxySQLSettings.Addresses = []string{"proxy:6032"} }}, + {"ignored hosts", func(c *MySQLClusterConfigurationSettings) { c.IgnoreHostsCount = 1 }}, + {"no recovery", func(c *MySQLClusterConfigurationSettings) { c.RecoveryDurationMillis = 0 }}, + {"invalid throttle", func(c *MySQLClusterConfigurationSettings) { c.ThrottleThreshold = 101 }}, + {"missing recovery threshold", func(c *MySQLClusterConfigurationSettings) { c.RecoveryThreshold = nil }}, + {"invalid metric", func(c *MySQLClusterConfigurationSettings) { c.ProxySQLCapacitySettings.Metric = "unknown" }}, + {"zero frontend capacity", func(c *MySQLClusterConfigurationSettings) { c.ProxySQLCapacitySettings.FrontendMaxConnections = 0 }}, + {"zero backend capacity", func(c *MySQLClusterConfigurationSettings) { + c.ProxySQLCapacitySettings.BackendLimits[0].MaxConnections = 0 + }}, + {"missing backend limits", func(c *MySQLClusterConfigurationSettings) { c.ProxySQLCapacitySettings.BackendLimits = nil }}, + } + t.Setenv("FRENO_TEST_MISSING_STATS_PASSWORD", "") + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + cluster := capacityClusterSettings() + test.change(cluster) + settings := MySQLConfigurationSettings{ + User: "must-not-inherit", Password: "must-not-inherit", + Clusters: map[string]*MySQLClusterConfigurationSettings{"capacity": cluster}, + } + if err := settings.postReadAdjustments(); err == nil || !strings.Contains(err.Error(), "ProxySQLCapacitySettings") { + t.Fatalf("unsafe configuration accepted: %v", err) + } + }) + } +} diff --git a/pkg/mysql/mysql_throttle_metric.go b/pkg/mysql/mysql_throttle_metric.go index a67031e..8f396e9 100644 --- a/pkg/mysql/mysql_throttle_metric.go +++ b/pkg/mysql/mysql_throttle_metric.go @@ -10,12 +10,14 @@ import ( "strings" "time" + "github.com/github/freno/pkg/proxysql" "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) +var proxySQLCapacitySampler = proxysql.NewCapacitySampler() func getMySQLMetricCacheKey(probe *Probe) string { return fmt.Sprintf("%s:%s", probe.Key, probe.MetricQuery) @@ -68,9 +70,11 @@ func (metric *MySQLThrottleMetric) Get() (float64, error) { // ReadThrottleMetric returns replication lag for a given connection config; either by explicit query // or via 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 + if probe.ProxySQLCapacitySettings == nil { + if mySQLThrottleMetric := getCachedMySQLThrottleMetric(probe); mySQLThrottleMetric != nil { + return mySQLThrottleMetric + // On cached results we avoid taking latency metrics + } } started := time.Now() @@ -88,6 +92,13 @@ func ReadThrottleMetric(probe *Probe, clusterName string) (mySQLThrottleMetric * }() }(mySQLThrottleMetric, started) + if probe.ProxySQLCapacitySettings != nil { + mySQLThrottleMetric.Value, mySQLThrottleMetric.Err = proxySQLCapacitySampler.Read( + probe.Key.Hostname, probe.Key.Port, probe.User, probe.Password, *probe.ProxySQLCapacitySettings, + ) + return mySQLThrottleMetric + } + dbUri := probe.GetDBUri("information_schema") db, fromCache, err := sqlutils.GetDB(dbUri) diff --git a/pkg/mysql/probe.go b/pkg/mysql/probe.go index 4386068..d6e0015 100644 --- a/pkg/mysql/probe.go +++ b/pkg/mysql/probe.go @@ -18,15 +18,16 @@ const timeoutMillis = 1000 // Probe is the minimal configuration required to connect to a MySQL server type Probe struct { - Key InstanceKey - User string - Password string - MetricQuery string - CacheMillis int - QueryInProgress int64 - HttpCheckPort int - HttpCheckPath string - HttpCheckInProgress int64 + Key InstanceKey + User string + Password string + MetricQuery string + CacheMillis int + QueryInProgress int64 + HttpCheckPort int + HttpCheckPath string + HttpCheckInProgress int64 + ProxySQLCapacitySettings *config.ProxySQLCapacityConfigurationSettings } type Probes map[InstanceKey](*Probe) diff --git a/pkg/mysql/probe_test.go b/pkg/mysql/probe_test.go index 13c35c4..5a46109 100644 --- a/pkg/mysql/probe_test.go +++ b/pkg/mysql/probe_test.go @@ -66,3 +66,18 @@ func TestGetDBUri(t *testing.T) { dbUri = c.GetDBUri("test_database") test.S(t).ExpectEquals(dbUri, "gromit:penguin@tcp(myhost:3306)/test_database?interpolateParams=true&collation=utf8mb4_unicode_ci&timeout=1000ms") } + +func TestCapacityMetricDoesNotFallBackToMySQLQueries(t *testing.T) { + probe := &Probe{ + Key: InstanceKey{Hostname: "127.0.0.1", Port: 3306}, + MetricQuery: "SELECT 1", + ProxySQLCapacitySettings: &config.ProxySQLCapacityConfigurationSettings{Metric: "invalid"}, + } + metric := ReadThrottleMetric(probe, "capacity") + if metric.Err == nil || metric.Err.Error() != "Metric must be frontend or writer-backend" { + t.Fatalf("capacity probe fell through to ordinary SQL: %v", metric.Err) + } + if metric.ClusterName != "capacity" || metric.Key != probe.Key { + t.Fatalf("capacity metric lost its cluster/endpoint identity: %+v", metric) + } +} diff --git a/pkg/proxysql/capacity.go b/pkg/proxysql/capacity.go new file mode 100644 index 0000000..dda8bcf --- /dev/null +++ b/pkg/proxysql/capacity.go @@ -0,0 +1,213 @@ +package proxysql + +import ( + "context" + "crypto/sha256" + "database/sql" + "fmt" + "net" + "strconv" + "sync" + "time" + + "github.com/github/freno/pkg/config" + mysqldriver "github.com/go-sql-driver/mysql" + "github.com/outbrain/golib/log" +) + +const CapacityPollInterval = 5 * time.Second +const capacityQueryTimeout = time.Second +const capacityIdleTimeout = 10 * time.Minute + +const frontendCapacityQuery = `SELECT Variable_Value FROM stats_mysql_global WHERE Variable_Name = 'Client_Connections_connected'` +const backendCapacityQuery = `SELECT srv_host, srv_port, status, ConnUsed FROM stats_mysql_connection_pool WHERE hostgroup = ?` + +type capacityBackend struct { + host string + port int + status string + used int64 +} + +type capacitySnapshot struct { + frontend int64 + backends []capacityBackend +} + +type capacityEndpoint struct { + mu sync.Mutex + db *sql.DB + lastPoll time.Time + lastUsed time.Time // guarded by CapacitySampler.mu + snapshot capacitySnapshot + err error +} + +// CapacitySampler shares one connection and one success/error snapshot between +// the frontend and backend probes for each proxy, credential set, and hostgroup. +type CapacitySampler struct { + mu sync.Mutex + endpoints map[string]*capacityEndpoint + lastCleanup time.Time + now func() time.Time + openDB func(string, string, string) (*sql.DB, error) +} + +func NewCapacitySampler() *CapacitySampler { + return &CapacitySampler{ + endpoints: make(map[string]*capacityEndpoint), + now: time.Now, + openDB: openCapacityDB, + } +} + +func openCapacityDB(address, user, password string) (*sql.DB, error) { + settings := mysqldriver.NewConfig() + settings.Net = "tcp" + settings.Addr = address + settings.User = user + settings.Passwd = password + settings.DBName = config.ProxySQLDefaultDatabase + settings.Timeout = capacityQueryTimeout + settings.ReadTimeout = capacityQueryTimeout + settings.WriteTimeout = capacityQueryTimeout + settings.InterpolateParams = true + db, err := sql.Open("mysql", settings.FormatDSN()) + if err != nil { + return nil, err + } + db.SetMaxOpenConns(1) + db.SetMaxIdleConns(1) + db.SetConnMaxIdleTime(capacityIdleTimeout) + return db, nil +} + +func (sampler *CapacitySampler) endpoint(address, user, password string, hostgroup int) *capacityEndpoint { + key := fmt.Sprintf("%s:%d:%x", address, hostgroup, sha256.Sum256([]byte(user+"\x00"+password))) + now := sampler.now() + sampler.mu.Lock() + defer sampler.mu.Unlock() + if now.Sub(sampler.lastCleanup) >= capacityIdleTimeout { + for key, endpoint := range sampler.endpoints { + if now.Sub(endpoint.lastUsed) >= capacityIdleTimeout { + endpoint.mu.Lock() + if endpoint.db != nil { + if err := endpoint.db.Close(); err != nil { + log.Errorf("closing idle ProxySQL capacity pool: %v", err) + } + } + endpoint.mu.Unlock() + delete(sampler.endpoints, key) + } + } + sampler.lastCleanup = now + } + endpoint := sampler.endpoints[key] + if endpoint == nil { + endpoint = &capacityEndpoint{} + sampler.endpoints[key] = endpoint + } + endpoint.lastUsed = now + return endpoint +} + +func (sampler *CapacitySampler) Close() error { + sampler.mu.Lock() + defer sampler.mu.Unlock() + var firstErr error + for key, endpoint := range sampler.endpoints { + endpoint.mu.Lock() + if endpoint.db != nil { + if err := endpoint.db.Close(); err != nil && firstErr == nil { + firstErr = err + } + } + endpoint.mu.Unlock() + delete(sampler.endpoints, key) + } + return firstErr +} + +func (sampler *CapacitySampler) Read(host string, port int, user, password string, settings config.ProxySQLCapacityConfigurationSettings) (float64, error) { + if err := settings.Validate(); err != nil { + return 0, err + } + if host == "" || port < 1 || port > 65535 || port == 3306 || port == 6033 || user == "" || password == "" { + return 0, fmt.Errorf("ProxySQL capacity requires an explicit admin endpoint and stats credentials") + } + address := net.JoinHostPort(host, strconv.Itoa(port)) + endpoint := sampler.endpoint(address, user, password, settings.WriterHostgroupID) + endpoint.mu.Lock() + defer endpoint.mu.Unlock() + now := sampler.now() + if endpoint.lastPoll.IsZero() || now.Sub(endpoint.lastPoll) >= CapacityPollInterval { + endpoint.lastPoll = now + endpoint.snapshot = capacitySnapshot{} + endpoint.err = nil + if endpoint.db == nil { + endpoint.db, endpoint.err = sampler.openDB(address, user, password) + } + if endpoint.err == nil { + endpoint.snapshot, endpoint.err = readCapacitySnapshot(endpoint.db, settings.WriterHostgroupID) + } + } + if endpoint.err != nil { + return 0, fmt.Errorf("ProxySQL stats at %s: %w", address, endpoint.err) + } + if settings.Metric == "frontend" { + return 100 * float64(endpoint.snapshot.frontend) / float64(settings.FrontendMaxConnections), nil + } + limits := make(map[string]config.ProxySQLBackendLimit) + for _, backend := range settings.BackendLimits { + limits[net.JoinHostPort(backend.Host, strconv.Itoa(backend.Port))] = backend + } + seen := make(map[string]bool) + worst := float64(0) + for _, backend := range endpoint.snapshot.backends { + key := net.JoinHostPort(backend.host, strconv.Itoa(backend.port)) + limit, expected := limits[key] + if !expected && backend.status != "ONLINE" { + continue + } + if !expected || seen[key] || backend.status != "ONLINE" { + return 0, fmt.Errorf("writer backend %s is unknown, duplicated, or not ONLINE", key) + } + seen[key] = true + value := 100 * float64(backend.used) / float64(limit.MaxConnections) + if value > worst { + worst = value + } + } + if len(seen) != len(limits) { + return 0, fmt.Errorf("ProxySQL writer stats do not contain every configured backend") + } + return worst, nil +} + +func readCapacitySnapshot(db *sql.DB, hostgroup int) (capacitySnapshot, error) { + snapshot := capacitySnapshot{} + ctx, cancel := context.WithTimeout(context.Background(), capacityQueryTimeout) + defer cancel() + if err := db.QueryRowContext(ctx, frontendCapacityQuery).Scan(&snapshot.frontend); err != nil { + return snapshot, fmt.Errorf("frontend connections: %w", err) + } + if snapshot.frontend < 0 { + return snapshot, fmt.Errorf("frontend connections must not be negative") + } + rows, err := db.QueryContext(ctx, backendCapacityQuery, hostgroup) + if err != nil { + return snapshot, fmt.Errorf("writer connections: %w", err) + } + defer rows.Close() + for rows.Next() { + backend := capacityBackend{} + if err := rows.Scan(&backend.host, &backend.port, &backend.status, &backend.used); err != nil { + return snapshot, err + } + if backend.host == "" || backend.port < 0 || backend.port > 65535 || backend.used < 0 { + return snapshot, fmt.Errorf("invalid writer connection counters or endpoint") + } + snapshot.backends = append(snapshot.backends, backend) + } + return snapshot, rows.Err() +} diff --git a/pkg/proxysql/capacity_integration_test.go b/pkg/proxysql/capacity_integration_test.go new file mode 100644 index 0000000..eb9a10a --- /dev/null +++ b/pkg/proxysql/capacity_integration_test.go @@ -0,0 +1,76 @@ +package proxysql + +import ( + "errors" + "net" + "os" + "sync/atomic" + "testing" + "time" + + "github.com/github/freno/pkg/config" +) + +// Run only against testdata/capacity.cnf in an isolated container network. +func TestCapacityIntegrationNoBackendConnections(t *testing.T) { + if os.Getenv("FRENO_PROXYSQL_CAPACITY_INTEGRATION") != "1" { + t.Skip("requires the isolated ProxySQL capacity fixture") + } + backend, err := net.Listen("tcp", "127.0.0.1:3407") + if err != nil { + t.Fatal(err) + } + var connections int64 + done := make(chan struct{}) + go func() { + defer close(done) + for { + connection, err := backend.Accept() + if err != nil { + if !errors.Is(err, net.ErrClosed) { + t.Error(err) + } + return + } + atomic.AddInt64(&connections, 1) + connection.Close() + } + }() + defer func() { + backend.Close() + <-done + }() + deadline := time.Now().Add(15 * time.Second) + for { + connection, err := net.DialTimeout("tcp", "127.0.0.1:6032", time.Second) + if err == nil { + connection.Close() + break + } + if time.Now().After(deadline) { + t.Fatalf("ProxySQL fixture did not become ready: %v", err) + } + time.Sleep(100 * time.Millisecond) + } + + sampler := NewCapacitySampler() + defer sampler.Close() + now := time.Now() + sampler.now = func() time.Time { return now } + settings := config.ProxySQLCapacityConfigurationSettings{ + Metric: "frontend", FrontendMaxConnections: 200000, WriterHostgroupID: 30, + BackendLimits: []config.ProxySQLBackendLimit{{Host: "127.0.0.1", Port: 3407, MaxConnections: 1000}}, + } + for cycle := 0; cycle < 3; cycle++ { + for _, metric := range []string{"frontend", "writer-backend"} { + settings.Metric = metric + if value, err := sampler.Read("127.0.0.1", 6032, "stats", "capacity-test-only", settings); err != nil || value != 0 { + t.Fatalf("%s = %v, %v; want an idle fixture", metric, value, err) + } + } + now = now.Add(CapacityPollInterval) + } + if count := atomic.LoadInt64(&connections); count != 0 { + t.Fatalf("stats sampling opened %d backend connections; want zero", count) + } +} diff --git a/pkg/proxysql/capacity_test.go b/pkg/proxysql/capacity_test.go new file mode 100644 index 0000000..0c62999 --- /dev/null +++ b/pkg/proxysql/capacity_test.go @@ -0,0 +1,202 @@ +package proxysql + +import ( + "database/sql" + "errors" + "regexp" + "sync" + "testing" + "time" + + "github.com/DATA-DOG/go-sqlmock" + "github.com/github/freno/pkg/config" +) + +func capacityTestSettings() config.ProxySQLCapacityConfigurationSettings { + return config.ProxySQLCapacityConfigurationSettings{ + Metric: "frontend", FrontendMaxConnections: 200000, WriterHostgroupID: 30, + BackendLimits: []config.ProxySQLBackendLimit{ + {Host: "/var/run/mysqld/mysqld.sock", Port: 0, MaxConnections: 1000}, + }, + } +} + +func capacityTestSampler(t *testing.T) (*CapacitySampler, sqlmock.Sqlmock, *time.Time) { + t.Helper() + db, mock, err := sqlmock.New() + if err != nil { + t.Fatal(err) + } + db.SetMaxOpenConns(1) + db.SetMaxIdleConns(1) + now := time.Now() + sampler := NewCapacitySampler() + sampler.now = func() time.Time { return now } + sampler.openDB = func(address, user, password string) (*sql.DB, error) { + if address != "proxy.example:6032" || user != "stats" || password != "test-only" { + t.Errorf("unexpected capacity endpoint or credentials") + } + return db, nil + } + t.Cleanup(func() { + if err := mock.ExpectationsWereMet(); err != nil { + t.Error(err) + } + mock.ExpectClose() + if err := sampler.Close(); err != nil { + t.Error(err) + } + }) + return sampler, mock, &now +} + +func expectCapacitySnapshot(mock sqlmock.Sqlmock, connected interface{}, backends *sqlmock.Rows) { + mock.ExpectQuery(regexp.QuoteMeta(frontendCapacityQuery)). + WillReturnRows(sqlmock.NewRows([]string{"Variable_Value"}).AddRow(connected)) + mock.ExpectQuery(regexp.QuoteMeta(backendCapacityQuery)).WithArgs(30).WillReturnRows(backends) +} + +func writerRows() *sqlmock.Rows { + return sqlmock.NewRows([]string{"srv_host", "srv_port", "status", "ConnUsed"}) +} + +func TestCapacitySharedSnapshotAndPollingBound(t *testing.T) { + sampler, mock, now := capacityTestSampler(t) + expectCapacitySnapshot(mock, "150000", writerRows().AddRow("/var/run/mysqld/mysqld.sock", 0, "ONLINE", 900)) + settings := capacityTestSettings() + for i := 0; i < 100; i++ { + value, err := sampler.Read("proxy.example", 6032, "stats", "test-only", settings) + if err != nil || value != 75 { + t.Fatalf("frontend = %v, %v; want 75", value, err) + } + settings.Metric = "writer-backend" + value, err = sampler.Read("proxy.example", 6032, "stats", "test-only", settings) + if err != nil || value != 90 { + t.Fatalf("writer = %v, %v; want 90", value, err) + } + settings.Metric = "frontend" + } + *now = now.Add(CapacityPollInterval - time.Nanosecond) + if _, err := sampler.Read("proxy.example", 6032, "stats", "test-only", settings); err != nil { + t.Fatal(err) + } + *now = now.Add(time.Nanosecond) + expectCapacitySnapshot(mock, 100000, writerRows().AddRow("/var/run/mysqld/mysqld.sock", 0, "ONLINE", 500)) + value, err := sampler.Read("proxy.example", 6032, "stats", "test-only", settings) + if err != nil || value != 50 { + t.Fatalf("new sample = %v, %v; want 50", value, err) + } +} + +func TestCapacityFailureIsCachedAndReplacesHealthySnapshot(t *testing.T) { + sampler, mock, now := capacityTestSampler(t) + settings := capacityTestSettings() + expectCapacitySnapshot(mock, 100000, writerRows().AddRow("/var/run/mysqld/mysqld.sock", 0, "ONLINE", 500)) + if _, err := sampler.Read("proxy.example", 6032, "stats", "test-only", settings); err != nil { + t.Fatal(err) + } + *now = now.Add(CapacityPollInterval) + mock.ExpectQuery(regexp.QuoteMeta(frontendCapacityQuery)).WillReturnError(errors.New("stats unavailable")) + for i := 0; i < 100; i++ { + for _, metric := range []string{"frontend", "writer-backend"} { + settings.Metric = metric + if _, err := sampler.Read("proxy.example", 6032, "stats", "test-only", settings); err == nil { + t.Fatal("failed poll must not return a healthy cached sample") + } + } + } + *now = now.Add(CapacityPollInterval) + expectCapacitySnapshot(mock, 80000, writerRows().AddRow("/var/run/mysqld/mysqld.sock", 0, "ONLINE", 400)) + if value, err := sampler.Read("proxy.example", 6032, "stats", "test-only", settings); err != nil || value != 40 { + t.Fatalf("recovery sample = %v, %v", value, err) + } +} + +func TestCapacityWriterCoverageAndWorstBackend(t *testing.T) { + tests := []struct { + name string + rows *sqlmock.Rows + want float64 + wantErr bool + }{ + {"worst, not average", writerRows().AddRow("db1", 3306, "ONLINE", 100).AddRow("db2", 3306, "ONLINE", 950), 95, false}, + {"missing writer", writerRows().AddRow("db1", 3306, "ONLINE", 100), 0, true}, + {"empty pool", writerRows(), 0, true}, + {"shunned writer", writerRows().AddRow("db1", 3306, "ONLINE", 100).AddRow("db2", 3306, "SHUNNED", 0), 0, true}, + {"unknown online writer", writerRows().AddRow("db3", 3306, "ONLINE", 100), 0, true}, + {"duplicate writer", writerRows().AddRow("db1", 3306, "ONLINE", 1).AddRow("db1", 3306, "ONLINE", 2), 0, true}, + {"negative count", writerRows().AddRow("db1", 3306, "ONLINE", -1), 0, true}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + sampler, mock, _ := capacityTestSampler(t) + settings := capacityTestSettings() + settings.Metric = "writer-backend" + settings.BackendLimits = []config.ProxySQLBackendLimit{ + {Host: "db1", Port: 3306, MaxConnections: 1000}, + {Host: "db2", Port: 3306, MaxConnections: 1000}, + } + expectCapacitySnapshot(mock, 100000, test.rows) + value, err := sampler.Read("proxy.example", 6032, "stats", "test-only", settings) + if (err != nil) != test.wantErr || value != test.want { + t.Fatalf("got %v, %v; want %v, error=%v", value, err, test.want, test.wantErr) + } + }) + } +} + +func TestCapacityInvalidFrontendStats(t *testing.T) { + for _, counter := range []interface{}{nil, "not-a-number", int64(-1)} { + t.Run("invalid counter", func(t *testing.T) { + sampler, mock, _ := capacityTestSampler(t) + mock.ExpectQuery(regexp.QuoteMeta(frontendCapacityQuery)). + WillReturnRows(sqlmock.NewRows([]string{"Variable_Value"}).AddRow(counter)) + if _, err := sampler.Read("proxy.example", 6032, "stats", "test-only", capacityTestSettings()); err == nil { + t.Fatal("invalid frontend stats must fail") + } + }) + } +} + +func TestCapacityConcurrentReadsShareOnePoll(t *testing.T) { + sampler, mock, _ := capacityTestSampler(t) + expectCapacitySnapshot(mock, 100000, writerRows().AddRow("/var/run/mysqld/mysqld.sock", 0, "ONLINE", 500)) + var workers sync.WaitGroup + for i := 0; i < 20; i++ { + workers.Add(1) + go func() { + defer workers.Done() + if _, err := sampler.Read("proxy.example", 6032, "stats", "test-only", capacityTestSettings()); err != nil { + t.Error(err) + } + }() + } + workers.Wait() +} + +func TestCapacityQueryDeadline(t *testing.T) { + sampler, mock, _ := capacityTestSampler(t) + mock.ExpectQuery(regexp.QuoteMeta(frontendCapacityQuery)). + WillDelayFor(2 * capacityQueryTimeout). + WillReturnRows(sqlmock.NewRows([]string{"Variable_Value"}).AddRow(100)) + start := time.Now() + if _, err := sampler.Read("proxy.example", 6032, "stats", "test-only", capacityTestSettings()); err == nil { + t.Fatal("query deadline must fail the probe") + } + if elapsed := time.Since(start); elapsed > 1500*time.Millisecond { + t.Fatalf("query deadline took %v, want approximately one second", elapsed) + } +} + +func TestCapacityRejectsApplicationEndpointsBeforeConnecting(t *testing.T) { + sampler := NewCapacitySampler() + sampler.openDB = func(_, _, _ string) (*sql.DB, error) { + t.Fatal("must not connect to an application endpoint") + return nil, nil + } + for _, port := range []int{3306, 6033} { + if _, err := sampler.Read("primary.example", port, "stats", "test-only", capacityTestSettings()); err == nil { + t.Fatal("application port must be rejected") + } + } +} diff --git a/pkg/proxysql/testdata/capacity.cnf b/pkg/proxysql/testdata/capacity.cnf new file mode 100644 index 0000000..20cba7e --- /dev/null +++ b/pkg/proxysql/testdata/capacity.cnf @@ -0,0 +1,23 @@ +datadir="/tmp/proxysql-capacity-test" +admin_variables= +{ + admin_credentials="admin:admin" + stats_credentials="stats:capacity-test-only" + mysql_ifaces="0.0.0.0:6032" +} +mysql_variables= +{ + threads=1 + max_connections=200000 + interfaces="127.0.0.1:6033" + monitor_enabled=false +} +mysql_servers= +( + { + address="127.0.0.1" + port=3407 + hostgroup=30 + max_connections=1000 + } +) diff --git a/pkg/throttle/mysql.go b/pkg/throttle/mysql.go index 2489b34..3bffb1c 100644 --- a/pkg/throttle/mysql.go +++ b/pkg/throttle/mysql.go @@ -21,7 +21,10 @@ func aggregateMySQLProbes( // so it's safe to iterate it probeValues := []float64{} for _, probe := range *probes { - if clusterInstanceHttpChecksMap[mysql.MySQLHttpCheckHashKey(clusterName, &probe.Key)] == http.StatusNotFound { + if probe.ProxySQLCapacitySettings != nil { + ignoreHostsCount = 0 + ignoreDialTcpErrors = false + } else if clusterInstanceHttpChecksMap[mysql.MySQLHttpCheckHashKey(clusterName, &probe.Key)] == http.StatusNotFound { continue } instanceMetricResult, ok := instanceResultsMap[mysql.GetClusterInstanceKey(clusterName, &probe.Key)] diff --git a/pkg/throttle/mysql_test.go b/pkg/throttle/mysql_test.go index 2b13f1a..8a6382f 100644 --- a/pkg/throttle/mysql_test.go +++ b/pkg/throttle/mysql_test.go @@ -10,6 +10,7 @@ import ( "testing" "github.com/github/freno/pkg/base" + "github.com/github/freno/pkg/config" "github.com/github/freno/pkg/mysql" "github.com/outbrain/golib/log" @@ -28,6 +29,27 @@ func init() { log.SetLevel(log.ERROR) } +func TestCapacityAggregationDoesNotIgnoreFailedOrSaturatedHosts(t *testing.T) { + cluster := "capacity" + capacity := &config.ProxySQLCapacityConfigurationSettings{Metric: "frontend"} + probes := mysql.Probes{ + key1: {Key: key1, ProxySQLCapacitySettings: capacity}, + key2: {Key: key2, ProxySQLCapacitySettings: capacity}, + } + results := mysql.InstanceMetricResultMap{ + mysql.GetClusterInstanceKey(cluster, &key1): base.NewSimpleMetricResult(10), + mysql.GetClusterInstanceKey(cluster, &key2): base.NewSimpleMetricResult(99), + } + value, err := aggregateMySQLProbes(&probes, cluster, results, nil, 1, true, 0).Get() + if err != nil || value != 99 { + t.Fatalf("capacity aggregation ignored saturation: %v, %v", value, err) + } + results[mysql.GetClusterInstanceKey(cluster, &key2)] = base.NewErrorMetricResult(0, base.NoHostsError) + if _, err := aggregateMySQLProbes(&probes, cluster, results, nil, 1, true, 0).Get(); err == nil { + t.Fatal("capacity aggregation ignored a failed host") + } +} + func TestAggregateMySQLProbesNoErrors(t *testing.T) { clusterName := "c0" key1cluster := mysql.GetClusterInstanceKey(clusterName, &key1) diff --git a/pkg/throttle/throttler.go b/pkg/throttle/throttler.go index 7f22eed..1a4d65d 100644 --- a/pkg/throttle/throttler.go +++ b/pkg/throttle/throttler.go @@ -286,16 +286,20 @@ func (throttler *Throttler) refreshMySQLInventory() error { log.Debugf("refreshing MySQL inventory") addInstanceKey := func(key *mysql.InstanceKey, clusterName string, clusterSettings *config.MySQLClusterConfigurationSettings, probes *mysql.Probes) { - for _, ignore := range clusterSettings.IgnoreHosts { - if strings.Contains(key.StringCode(), ignore) { - log.Debugf("instance key ignored: %+v", key) + if clusterSettings.ProxySQLCapacitySettings != nil { + key = &mysql.InstanceKey{Hostname: key.Hostname, Port: clusterSettings.Port} + } else { + for _, ignore := range clusterSettings.IgnoreHosts { + if strings.Contains(key.StringCode(), ignore) { + log.Debugf("instance key ignored: %+v", key) + return + } + } + if _, skipped := throttler.skippedHosts.Get(key.Hostname); skipped { + log.Debugf("host skipped: %+v", key.Hostname) return } } - if _, skipped := throttler.skippedHosts.Get(key.Hostname); skipped { - log.Debugf("host skipped: %+v", key.Hostname) - return - } if !key.IsValid() { log.Debugf("read invalid instance key: [%+v] for cluster %+v", key, clusterName) return @@ -303,13 +307,14 @@ func (throttler *Throttler) refreshMySQLInventory() error { log.Debugf("read instance key: %+v", key) probe := &mysql.Probe{ - Key: *key, - User: clusterSettings.User, - Password: clusterSettings.Password, - MetricQuery: clusterSettings.MetricQuery, - CacheMillis: clusterSettings.CacheMillis, - HttpCheckPath: clusterSettings.HttpCheckPath, - HttpCheckPort: clusterSettings.HttpCheckPort, + Key: *key, + User: clusterSettings.User, + Password: clusterSettings.Password, + MetricQuery: clusterSettings.MetricQuery, + CacheMillis: clusterSettings.CacheMillis, + HttpCheckPath: clusterSettings.HttpCheckPath, + HttpCheckPort: clusterSettings.HttpCheckPort, + ProxySQLCapacitySettings: clusterSettings.ProxySQLCapacitySettings, } (*probes)[*key] = probe } @@ -319,7 +324,15 @@ func (throttler *Throttler) refreshMySQLInventory() error { clusterSettings := clusterSettings // config may dynamically change, but internal structure (config.Settings().Stores.MySQL.Clusters in our case) // is immutable and can only be _replaced_. Hence, it's safe to read in a goroutine: - go func() error { + go func() (err error) { + defer func() { + if err != nil && clusterSettings.ProxySQLCapacitySettings != nil { + // A failed refresh must not leave an old proxy roster looking healthy. + throttler.mysqlClusterProbesChan <- &mysql.ClusterProbes{ + ClusterName: clusterName, InstanceProbes: mysql.NewProbes(), + } + } + }() throttler.mysqlClusterThresholds.Set(clusterName, clusterSettings.ThrottleThreshold, cache.DefaultExpiration) if !clusterSettings.HAProxySettings.IsEmpty() { poolName := clusterSettings.HAProxySettings.PoolName @@ -336,6 +349,9 @@ func (throttler *Throttler) refreshMySQLInventory() error { totalHosts = append(totalHosts, hosts...) log.Debugf("Read %+v hosts from haproxy %s/#%s", len(hosts), u.String(), poolName) } else { + if clusterSettings.ProxySQLCapacitySettings != nil { + return log.Errorf("Unable to discover ProxySQL capacity hosts from %s/#%s: %+v", u.String(), poolName, err) + } log.Errorf("Unable to get HAproxy hosts from %s/#%s: %+v", u.String(), poolName, err) } } diff --git a/pkg/throttle/throttler_test.go b/pkg/throttle/throttler_test.go index 5c902d2..e0476d6 100644 --- a/pkg/throttle/throttler_test.go +++ b/pkg/throttle/throttler_test.go @@ -2,16 +2,79 @@ package throttle import ( "net/http" + "net/http/httptest" "testing" "time" "github.com/github/freno/pkg/base" "github.com/github/freno/pkg/config" + "github.com/github/freno/pkg/mysql" "github.com/patrickmn/go-cache" metrics "github.com/rcrowley/go-metrics" "github.com/stretchr/testify/assert" ) +func TestCapacityDiscoveryUsesAdminPortWithoutMySQLHostSkipping(t *testing.T) { + original := config.Settings().Stores.MySQL + defer func() { config.Settings().Stores.MySQL = original }() + capacity := &config.ProxySQLCapacityConfigurationSettings{Metric: "frontend"} + config.Settings().Stores.MySQL = config.MySQLConfigurationSettings{ + Clusters: map[string]*config.MySQLClusterConfigurationSettings{ + "capacity": { + Port: 6032, User: "stats", Password: "test-only", HttpCheckPort: -1, + IgnoreHosts: []string{"proxy"}, + StaticHostsSettings: config.StaticHostsConfigurationSettings{Hosts: []string{"proxy.example:3306"}}, + ProxySQLCapacitySettings: capacity, + }, + }, + } + throttler := NewThrottler() + throttler.isLeader = true + throttler.skippedHosts.SetDefault("proxy.example", true) + if err := throttler.refreshMySQLInventory(); err != nil { + t.Fatal(err) + } + select { + case cluster := <-throttler.mysqlClusterProbesChan: + probe := (*cluster.InstanceProbes)[mysql.InstanceKey{Hostname: "proxy.example", Port: 6032}] + if len(*cluster.InstanceProbes) != 1 || probe == nil || probe.ProxySQLCapacitySettings != capacity { + t.Fatalf("discovery did not retain the admin probe: %+v", cluster) + } + case <-time.After(time.Second): + t.Fatal("capacity discovery did not publish probes") + } +} + +func TestCapacityFailedDiscoveryClearsThePreviousRoster(t *testing.T) { + original := config.Settings().Stores.MySQL + defer func() { config.Settings().Stores.MySQL = original }() + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Write([]byte("invalid HAProxy CSV")) + })) + defer server.Close() + config.Settings().Stores.MySQL = config.MySQLConfigurationSettings{ + Clusters: map[string]*config.MySQLClusterConfigurationSettings{ + "capacity": { + HAProxySettings: config.HAProxyConfigurationSettings{Addresses: server.URL, PoolName: "writer"}, + ProxySQLCapacitySettings: &config.ProxySQLCapacityConfigurationSettings{Metric: "frontend"}, + }, + }, + } + throttler := NewThrottler() + throttler.isLeader = true + if err := throttler.refreshMySQLInventory(); err != nil { + t.Fatal(err) + } + select { + case cluster := <-throttler.mysqlClusterProbesChan: + if cluster.ClusterName != "capacity" || len(*cluster.InstanceProbes) != 0 { + t.Fatalf("failed discovery retained old hosts: %+v", cluster) + } + case <-time.After(time.Second): + t.Fatal("failed discovery must explicitly clear capacity probes") + } +} + func Test_checkAppMetricResult(t *testing.T) { throttler := NewThrottler()