From 26efe297961b404656ce6a3c6bf39f05b680991a Mon Sep 17 00:00:00 2001 From: Pradeep Kintali Date: Thu, 15 Oct 2026 11:01:43 -0700 Subject: [PATCH 1/2] Add bounded direct ProxySQL capacity probes Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: dea961fa-5766-415f-a0f5-2562c55a0385 --- doc/mysql.md | 103 +++++++++- pkg/config/mysql_config.go | 18 +- pkg/config/proxysql_capacity_config.go | 101 ++++++++++ pkg/config/proxysql_capacity_config_test.go | 75 +++++++ pkg/mysql/mysql_throttle_metric.go | 17 +- pkg/mysql/probe.go | 19 +- pkg/mysql/probe_test.go | 15 ++ pkg/proxysql/capacity.go | 213 ++++++++++++++++++++ pkg/proxysql/capacity_integration_test.go | 76 +++++++ pkg/proxysql/capacity_test.go | 202 +++++++++++++++++++ pkg/proxysql/testdata/capacity.cnf | 23 +++ pkg/throttle/mysql.go | 5 +- pkg/throttle/mysql_test.go | 22 ++ pkg/throttle/throttler.go | 46 +++-- pkg/throttle/throttler_test.go | 63 ++++++ 15 files changed, 958 insertions(+), 40 deletions(-) create mode 100644 pkg/config/proxysql_capacity_config.go create mode 100644 pkg/config/proxysql_capacity_config_test.go create mode 100644 pkg/proxysql/capacity.go create mode 100644 pkg/proxysql/capacity_integration_test.go create mode 100644 pkg/proxysql/capacity_test.go create mode 100644 pkg/proxysql/testdata/capacity.cnf diff --git a/doc/mysql.md b/doc/mysql.md index b479b01..f6dba1d 100644 --- a/doc/mysql.md +++ b/doc/mysql.md @@ -160,7 +160,8 @@ A cluster can require additional cluster metrics to pass before `freno` permits "prod4": { "RequiredClusters": [ "prod4-primary", - "prod4-proxysql" + "prod4-proxysql-frontend", + "prod4-proxysql-writer-backend" ], "HAProxySettings": { "Host": "my.haproxy.mydomain.com", @@ -181,25 +182,54 @@ A cluster can require additional cluster metrics to pass before `freno` permits ] } }, - "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. - `FailOnNoHosts` changes the legacy no-host response from `HTTP 200` to `HTTP 500`. Enable it for primary and ProxySQL safety probes, where an empty roster cannot prove that work is safe. @@ -211,6 +241,63 @@ The ProxySQL query above is illustrative: deployments must provide a scalar quer Configuration loading rejects missing `RequiredClusters` references and dependency cycles. +### 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 c434aa5..533639a 100644 --- a/pkg/config/mysql_config.go +++ b/pkg/config/mysql_config.go @@ -29,10 +29,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. @@ -117,6 +118,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 ddf9759..5ac93c4 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() From b776839cab1c855945076b9fdd3770ad6166cd62 Mon Sep 17 00:00:00 2001 From: Pradeep Kintali Date: Thu, 15 Oct 2026 15:08:18 -0700 Subject: [PATCH 2/2] Apply shadow dependency evaluation to capacity probes Carry shadow graph evaluation, isolated HTTP and background counters, startup notices, and public rollout guidance into the capacity stack while preserving direct capacity sampling. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: fbc83df2-d1af-406f-b97b-51cf0aaa570d --- .gitignore | 2 +- README.md | 3 + cmd/freno/main.go | 34 ++++ cmd/freno/main_test.go | 111 +++++++++++++ doc/memcache.md | 3 + doc/mysql.md | 157 ++++++++++++++---- pkg/config/config_test.go | 43 ++++- pkg/config/mysql_config.go | 35 ++-- pkg/http/api.go | 4 +- pkg/http/api_test.go | 15 +- pkg/throttle/check.go | 56 +++---- pkg/throttle/check_result.go | 23 +++ pkg/throttle/shadow.go | 74 +++++++++ pkg/throttle/shadow_test.go | 309 +++++++++++++++++++++++++++++++++++ pkg/throttle/throttler.go | 12 ++ 15 files changed, 797 insertions(+), 84 deletions(-) create mode 100644 cmd/freno/main_test.go create mode 100644 pkg/throttle/shadow.go create mode 100644 pkg/throttle/shadow_test.go diff --git a/.gitignore b/.gitignore index 3b7d16e..f1f1f57 100644 --- a/.gitignore +++ b/.gitignore @@ -4,5 +4,5 @@ dist/ main conf/freno.local.conf.json .vendor/ -freno +/freno .idea/ diff --git a/README.md b/README.md index 9da2df5..4d3764a 100644 --- a/README.md +++ b/README.md @@ -35,6 +35,9 @@ Clients (application, scripts, jobs) are expected to consult with `freno`. `fren Read more about [freno and MySQL throttling](doc/mysql.md) +Before enabling required dependencies or recovery, read the +[MySQL safety behavior changes and rollout warnings](doc/mysql.md#behavior-changes). + ### Use cases `freno` is useful for bulk operations: massive loading/archiving tasks, schema migrations, mass updates. Such operations typically walk through thousands to millions of rows and may cause undesired effects such as MySQL replication lags. By breaking these tasks to small subtasks (e.g. `100` rows at a time), and by consulting `freno` before applying each such subtask, we are able to achieve the same result without ill effect to the database and to the application that uses it. diff --git a/cmd/freno/main.go b/cmd/freno/main.go index 3bf0a7f..459ada6 100644 --- a/cmd/freno/main.go +++ b/cmd/freno/main.go @@ -3,7 +3,10 @@ package main import ( "flag" "fmt" + "io" gohttp "net/http" + "os" + "sort" "strings" "github.com/github/freno/pkg/config" @@ -96,6 +99,11 @@ func main() { switch { case *http: + if !*quiet { + if err := logMySQLSafetyConfiguration(config.Settings().Stores.MySQL, os.Stderr); err != nil { + log.Fatalf("Error reporting MySQL safety configuration: %s", err) + } + } err := httpServe() log.Errore(err) default: @@ -116,6 +124,32 @@ func loadConfiguration(configFile string) { } } +func logMySQLSafetyConfiguration(settings config.MySQLConfigurationSettings, output io.Writer) error { + clusterNames := make([]string, 0, len(settings.Clusters)) + for name, cluster := range settings.Clusters { + if len(cluster.RequiredClusters) > 0 || len(cluster.ShadowRequiredClusters) > 0 || cluster.RecoveryDurationMillis > 0 || cluster.FailOnNoHosts { + clusterNames = append(clusterNames, name) + } + } + sort.Strings(clusterNames) + + for _, name := range clusterNames { + cluster := settings.Clusters[name] + recoveryThreshold := cluster.ThrottleThreshold + if cluster.RecoveryThreshold != nil { + recoveryThreshold = *cluster.RecoveryThreshold + } + // Safety notices must be visible at the default ERROR log level, without changing other log filtering. + if _, err := fmt.Fprintf(output, + "freno MySQL safety configuration: cluster=%q RequiredClusters=%q ShadowRequiredClusters=%q RecoveryDurationMillis=%d RecoveryThreshold=%g ThrottleThreshold=%g FailOnNoHosts=%t\n", + name, cluster.RequiredClusters, cluster.ShadowRequiredClusters, cluster.RecoveryDurationMillis, recoveryThreshold, cluster.ThrottleThreshold, cluster.FailOnNoHosts, + ); err != nil { + return fmt.Errorf("cannot log MySQL safety configuration for cluster %q: %w", name, err) + } + } + return nil +} + func httpServe() error { throttler := throttle.NewThrottler() log.Infof("Starting consensus service") diff --git a/cmd/freno/main_test.go b/cmd/freno/main_test.go new file mode 100644 index 0000000..b8fe135 --- /dev/null +++ b/cmd/freno/main_test.go @@ -0,0 +1,111 @@ +package main + +import ( + "bytes" + "errors" + "strings" + "testing" + + "github.com/github/freno/pkg/config" + "github.com/outbrain/golib/log" +) + +func TestLogMySQLSafetyConfiguration(t *testing.T) { + originalLogLevel := log.GetLevel() + log.SetLevel(log.ERROR) + defer log.SetLevel(originalLogLevel) + + recoveryThreshold := 3.0 + settings := config.MySQLConfigurationSettings{ + Clusters: map[string]*config.MySQLClusterConfigurationSettings{ + "legacy": {ThrottleThreshold: 5}, + "dependency": { + RequiredClusters: []string{"primary"}, + ThrottleThreshold: 5, + User: "do-not-log-user", + Password: "do-not-log-password", + MetricQuery: "do-not-log-query", + }, + "primary": {ThrottleThreshold: 50}, + "shadow-only": { + ShadowRequiredClusters: []string{"primary"}, + ThrottleThreshold: 5, + }, + "fail-closed": { + FailOnNoHosts: true, + ThrottleThreshold: 50, + }, + "recovery": { + RecoveryDurationMillis: 5000, + RecoveryThreshold: &recoveryThreshold, + ThrottleThreshold: 10, + }, + }, + } + var output bytes.Buffer + if err := logMySQLSafetyConfiguration(settings, &output); err != nil { + t.Fatal(err) + } + want := "freno MySQL safety configuration: cluster=\"dependency\" RequiredClusters=[\"primary\"] ShadowRequiredClusters=[] RecoveryDurationMillis=0 RecoveryThreshold=5 ThrottleThreshold=5 FailOnNoHosts=false\n" + + "freno MySQL safety configuration: cluster=\"fail-closed\" RequiredClusters=[] ShadowRequiredClusters=[] RecoveryDurationMillis=0 RecoveryThreshold=50 ThrottleThreshold=50 FailOnNoHosts=true\n" + + "freno MySQL safety configuration: cluster=\"recovery\" RequiredClusters=[] ShadowRequiredClusters=[] RecoveryDurationMillis=5000 RecoveryThreshold=3 ThrottleThreshold=10 FailOnNoHosts=false\n" + + "freno MySQL safety configuration: cluster=\"shadow-only\" RequiredClusters=[] ShadowRequiredClusters=[\"primary\"] RecoveryDurationMillis=0 RecoveryThreshold=5 ThrottleThreshold=5 FailOnNoHosts=false\n" + if got := output.String(); got != want { + t.Fatalf("startup safety configuration = %q, want %q", got, want) + } + if log.GetLevel() != log.ERROR { + t.Fatal("startup safety notices changed unrelated log filtering") + } +} + +func TestLogMySQLSafetyConfigurationWithoutSafetySettings(t *testing.T) { + settings := config.MySQLConfigurationSettings{ + Clusters: map[string]*config.MySQLClusterConfigurationSettings{ + "legacy": {ThrottleThreshold: 5}, + }, + } + var output bytes.Buffer + if err := logMySQLSafetyConfiguration(settings, &output); err != nil { + t.Fatal(err) + } + if output.Len() != 0 { + t.Fatalf("unexpected safety notice for legacy configuration: %q", output.String()) + } +} + +func TestLogMySQLSafetyConfigurationEscapesNames(t *testing.T) { + settings := config.MySQLConfigurationSettings{ + Clusters: map[string]*config.MySQLClusterConfigurationSettings{ + "line\nbreak": {RequiredClusters: []string{"probe\nname"}}, + }, + } + var output bytes.Buffer + if err := logMySQLSafetyConfiguration(settings, &output); err != nil { + t.Fatal(err) + } + got := output.String() + if strings.Count(got, "\n") != 1 || !strings.Contains(got, `cluster="line\nbreak" RequiredClusters=["probe\nname"]`) { + t.Fatalf("cluster names did not stay within one escaped log line: %q", got) + } +} + +type failingLogWriter struct { + err error +} + +func (writer failingLogWriter) Write([]byte) (int, error) { + return 0, writer.err +} + +func TestLogMySQLSafetyConfigurationWriteError(t *testing.T) { + settings := config.MySQLConfigurationSettings{ + Clusters: map[string]*config.MySQLClusterConfigurationSettings{ + "primary": {FailOnNoHosts: true}, + }, + } + writeErr := errors.New("log destination unavailable") + err := logMySQLSafetyConfiguration(settings, failingLogWriter{err: writeErr}) + if !errors.Is(err, writeErr) || !strings.Contains(err.Error(), `"primary"`) { + t.Fatalf("expected contextual startup logging error, got %v", err) + } +} diff --git a/doc/memcache.md b/doc/memcache.md index f269637..b2357d3 100644 --- a/doc/memcache.md +++ b/doc/memcache.md @@ -41,6 +41,9 @@ Optionally set `MemcachePath` (default is `"freno"`): required metric is unavailable, recovering, or above its configured threshold. Direct memcache clients therefore observe the same composite safety gate as HTTP clients. +- `ShadowRequiredClusters` never removes the requested cluster's key. Its + [background would-reject counters](mysql.md#shadow-counters-and-limits) describe + prospective dependency failures during leader aggregation, not client reads. ### Runtime access to memcache configuration diff --git a/doc/mysql.md b/doc/mysql.md index f081c03..33c0a64 100644 --- a/doc/mysql.md +++ b/doc/mysql.md @@ -153,12 +153,55 @@ Otherwise you may provide any query that returns a single row, single numeric co ### Composing replica, primary, and ProxySQL protection -A cluster can require additional cluster metrics to pass before `freno` permits work. This allows the normal replica-lag check to depend on separately sampled primary-load and ProxySQL-capacity checks: +#### Behavior changes + +**Read before enabling dependencies or recovery.** These settings are opt-in, but +they can reject application work as soon as the new configuration takes effect: + +- **Startup and leadership recovery:** a cluster with `RecoveryDurationMillis > 0` + must establish a full continuously healthy window after every process start or + leadership acquisition, even without `RequiredClusters`. Ordinary write checks + return `429` while a successfully collected metric is recovering; collection + errors can return `500` instead. Empty-host behavior is controlled separately by + `FailOnNoHosts`. +- **Configuration changes require a restart:** the running service does not + live-reload configuration. Applying changes, including rollback, requires + restarting the affected process. Metrics that retain a positive + `RecoveryDurationMillis` must establish a new healthy window. +- **Required dependencies fail closed:** missing runtime metrics or collection + failures, such as incorrect discovery or missing query privileges, return `500` + for checks that reach that dependency. An empty host list also fails with `500` + when the dependency has `FailOnNoHosts: true`. Overload or incomplete recovery + returns `429`; neither response permits writes. +- **Memcache decisions are affected:** when memcache publishing is configured, + aggregation deletes a cluster's published key while any required dependency is + missing, in error, recovering, or above its threshold. Previously cached client + reads are not synchronously invalidated. +- **Enforcement is not a dry run:** adding a probe and referencing it from + `RequiredClusters` in the same configuration change can immediately block work + after restart. First use `ShadowRequiredClusters` to measure prospective + dependency failures without changing the current decision. Standalone + collection without a shadow reference does not evaluate the composite graph. +- **Coordinate client wait budgets:** a budget shorter than the remaining + recovery window can expire before recovery finishes. Agree on safe retry/defer + behavior; do not bypass checks to avoid a timeout. + +At service startup, each cluster with required or shadow dependencies, a positive +recovery duration, or fail-on-no-hosts enabled produces one safety-configuration notice on +standard error. It includes the dependency names, effective recovery and throttle +thresholds, recovery duration, and no-host policy, without credentials or query +text. These notices are visible without `--verbose`; `--quiet` suppresses them. + +#### Configuration example + +A cluster can require additional cluster metrics to pass before `freno` permits +work. Start with `ShadowRequiredClusters` to evaluate separately sampled +primary-load and ProxySQL-capacity checks without enforcing them: ```json "Clusters": { "prod4": { - "RequiredClusters": [ + "ShadowRequiredClusters": [ "prod4-primary", "prod4-proxysql-frontend", "prod4-proxysql-writer-backend" @@ -232,6 +275,10 @@ 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 + the real check passes. Their failures do not change the HTTP response, priority + state, shared-domain health, or memcache publication for the requested cluster. + Checking a probe directly still applies that probe's ordinary admission rules. - `FailOnNoHosts` changes the legacy no-host response from `HTTP 200` to `HTTP 500`. Enable it for primary and ProxySQL safety probes, where an empty roster cannot prove that work is safe. - `RecoveryThreshold` is the lower threshold that begins recovery after a cluster has throttled. It must not exceed `ThrottleThreshold`. - `RecoveryDurationMillis` is the continuous time the metric must remain at or below `RecoveryThreshold` before checks return `HTTP 200` again. A new error or threshold violation resets the recovery window. @@ -239,7 +286,57 @@ or the legacy fail-open no-host behavior. - Metric sampling remains centralized in the `freno` leader and uses the existing metric cache. Application checks read the aggregated decision rather than querying the primary or ProxySQL for every batch. - Check responses include `MetricName`, identifying the replica, primary, or ProxySQL metric that blocked a composite decision. `recovery.mysql..active` reports whether the recovery latch is active. -Configuration loading rejects missing `RequiredClusters` references and dependency cycles. +Configuration loading rejects missing references in either dependency list, +overlap between the two lists on the same cluster, and cycles across their union. +To promote a dependency, remove it from `ShadowRequiredClusters` and add it to +`RequiredClusters` in the same reviewed configuration change. + +#### Shadow counters and limits + +Both metric endpoints (`/debug/metrics` and `/debug/vars`) export these counters +as integer values: + +| Counter | Meaning | +| --- | --- | +| `shadow.http.mysql..total` | Otherwise-allowed external HTTP checks with an applicable shadow graph | +| `shadow.http.mysql..would_reject` | Checks whose prospective metric dependencies would reject | +| `shadow.http.mysql..dependency..would_reject` | First blocking dependency, in configured traversal order | +| `shadow.http.mysql..status..would_reject` | Prospective dependency failure status (`429` or `500`) | +| `shadow.background.mysql..*` | Same suffixes for otherwise-healthy leader aggregation cycles | + +The HTTP denominator includes both write and read checks. Root-only read +threshold overrides do not weaken shadow dependencies. Internal self-checks are +excluded. Requests already rejected by the root or an enforced dependency do not +increment shadow counters. Unknown names resolved through `FallbackCluster` use +the effective configured cluster name, not the requested name. + +The observer includes shadows attached to enforced dependencies and recursively +follows both dependency lists on new targets. Each node is evaluated at most once +per observation, and each observation counts at most one rejection. Already +accepted nodes in the real required closure are not resampled. Dependency and +status breakdowns identify the first failure, not every failing probe. + +Shadow evaluation uses cached metric values, configured thresholds, recovery +results, and no-host policy; it does not issue SQL or simulate per-app throttling, +randomized policies, deprioritization, or shared-domain policy on new dependencies. +It is a metric-dependency dry run, not a full prediction of future admission. + +Background counters are sampling cycles, **not memcache client reads or +application demand**. They are collected even without memcache publishing. +Dependencies with any error (including legacy fail-open no-host results) are +unsafe for publication, matching the existing memcache path; HTTP shadow checks +retain the dependency's configured `FailOnNoHosts` behavior. Background metrics +are hidden on followers; HTTP counters remain process-local and visible. +The 25 ms aggregation loop can reuse 50 ms SQL samples and longer cached +results, so cycles are not independent samples. Do not combine HTTP and +background totals into one rejection rate. + +Compute rates from counter deltas over the same process/window, using +`would_reject / total`. A missing or non-increasing total means no observed +coverage, not a healthy candidate. The root rejection counter is registered at +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 @@ -301,48 +398,40 @@ docker stop freno-capacity-test ### Rolling out primary safety checks Deploy a binary containing these settings before deploying configuration that -uses them. The related #197 changes were merged into the #196 feature branch, -not independently into the default branch; a merged stacked PR is not proof of -a deployed server version. +uses them. Verify the capabilities of the deployed binary rather than assuming +that a configuration change installs new server functionality. 1. Deploy the binary with the existing configuration. Without required dependencies, fail-on-no-hosts, or a recovery duration, legacy cluster decisions retain their existing behavior. -2. Add separately named primary metrics in shadow mode, without adding them to - the application's normal cluster `RequiredClusters`. Validate writer - discovery, privileges, probe load, error rates, and recovery gauges. - Here, shadow mode means collection only: per-probe self-checks report errors, - but there is no `ShadowRequiredClusters` setting or composite "would reject" - counter. Do not add new probes and enforce them in the same deployment. +2. Add separately named primary metrics and reference them from the application's + normal cluster `ShadowRequiredClusters`, not `RequiredClusters`. Validate writer + discovery, privileges, probe load, error rates, recovery gauges, and both shadow + counter scopes. Do not add new probes and enforce them in the same deployment. 3. Review thresholds against recent traffic, known incidents, and per-host/shard - coverage. Five-minute Datadog averages are candidate baselines, not equivalent - to the instantaneous SQL samples used by the safety check. Missing data must - not be interpreted as zero or healthy. -4. In a separate reviewed configuration change, add a primary dependency to one - cluster whose probes have been validated. Exercise overload, probe + coverage. Averaged monitoring data is not equivalent to the instantaneous SQL + samples used by the safety check. Missing data must not be interpreted as zero + or healthy. +4. In a separate reviewed configuration change, move a primary dependency from + `ShadowRequiredClusters` to `RequiredClusters` on one cluster whose probes have + been validated. Exercise overload, probe failure, missing hosts, leadership changes, and the HTTP and memcache paths. Required probes fail closed; verify callers defer work rather than bypassing protection on an error. 5. Expand only after reviewing rejection rates, recovery duration, job backlog, application errors, and primary health against agreed success/rollback limits. -The fleet Puppet configuration uses a 30-second healthy recovery duration and -an empty enforcement allowlist. A leadership change starts a fresh recovery -window for configured metrics. Coordinate deployment/restart behavior with the -operators of active clients. - -Ruby clients default to a 10-second wait budget. A caller encountering the -beginning of a 30-second server recovery window can therefore receive -`WaitedTooLong` before the server recovers. Application owners must approve a -longer wait budget or implement safe retry/defer behavior. Neither a longer -timeout nor client-side consecutive checks guarantee recovery during overload. - -Rollback enforcement by removing the primary dependency from the normal -cluster, keeping its existing replica-lag configuration. In the Puppet rollout, -remove the cluster from `EnforcedClusters` with `EnforceAll` still false and -deploy the configuration. This is a configuration deployment, not an -instantaneous runtime toggle; account for service notifications and startup -recovery. Disable generation only after removing all references to its probes. +Choose recovery durations and client wait budgets for the workload, and +coordinate restarts and leadership transitions with application operators. +Neither a longer timeout nor client-side consecutive checks guarantee recovery +during overload. + +Rollback enforcement by moving the primary dependency from the normal cluster's +`RequiredClusters` back to `ShadowRequiredClusters`, keeping its existing replica-lag configuration. +Apply the configuration and restart the affected process; this is not an +instantaneous runtime toggle, and any remaining recovery-enabled metrics need +a new healthy window. Remove probe configurations only after removing every +reference to them. ### Scope boundaries diff --git a/pkg/config/config_test.go b/pkg/config/config_test.go index 20e1245..af83e79 100644 --- a/pkg/config/config_test.go +++ b/pkg/config/config_test.go @@ -139,7 +139,48 @@ func TestMySQLRequiredClusters(t *testing.T) { "replicas": {RequiredClusters: []string{"primary"}}, "primary": {RequiredClusters: []string{"replicas"}}, }, - wantErr: `Stores.MySQL.Clusters contains a RequiredClusters cycle involving "primary"`, + wantErr: `Stores.MySQL.Clusters contains a RequiredClusters/ShadowRequiredClusters cycle involving "primary"`, + }, + { + name: "unknown shadow dependency", + clusters: map[string]*MySQLClusterConfigurationSettings{ + "replicas": {ShadowRequiredClusters: []string{"primary"}}, + }, + wantErr: `Stores.MySQL.Clusters.replicas.ShadowRequiredClusters references unknown cluster "primary"`, + }, + { + name: "shadow cycle", + clusters: map[string]*MySQLClusterConfigurationSettings{ + "primary": {ShadowRequiredClusters: []string{"replicas"}}, + "replicas": {ShadowRequiredClusters: []string{"primary"}}, + }, + wantErr: `Stores.MySQL.Clusters contains a RequiredClusters/ShadowRequiredClusters cycle involving "primary"`, + }, + { + name: "mixed dependency cycle", + clusters: map[string]*MySQLClusterConfigurationSettings{ + "primary": {RequiredClusters: []string{"replicas"}}, + "replicas": {ShadowRequiredClusters: []string{"primary"}}, + }, + wantErr: `Stores.MySQL.Clusters contains a RequiredClusters/ShadowRequiredClusters cycle involving "primary"`, + }, + { + name: "enforced and shadow overlap", + clusters: map[string]*MySQLClusterConfigurationSettings{ + "primary": {}, + "replicas": {RequiredClusters: []string{"primary"}, ShadowRequiredClusters: []string{"primary"}}, + }, + wantErr: `Stores.MySQL.Clusters.replicas: cluster "primary" cannot appear in both RequiredClusters and ShadowRequiredClusters`, + }, + { + name: "nested and shared dependencies", + clusters: map[string]*MySQLClusterConfigurationSettings{ + "replicas": {RequiredClusters: []string{"required"}, ShadowRequiredClusters: []string{"primary", "primary"}}, + "required": {ShadowRequiredClusters: []string{"primary"}}, + "primary": {RequiredClusters: []string{"capacity"}, ShadowRequiredClusters: []string{"extra"}}, + "capacity": {}, + "extra": {RequiredClusters: []string{"capacity"}}, + }, }, { name: "null cluster", diff --git a/pkg/config/mysql_config.go b/pkg/config/mysql_config.go index 533639a..f2d5a19 100644 --- a/pkg/config/mysql_config.go +++ b/pkg/config/mysql_config.go @@ -22,6 +22,7 @@ type MySQLClusterConfigurationSettings struct { RecoveryDurationMillis int // continuous recovery duration required before writes resume FailOnNoHosts bool // reject checks when no eligible hosts remain instead of preserving legacy fail-open behavior RequiredClusters []string // additional MySQL cluster metrics that must pass for this cluster to pass + ShadowRequiredClusters []string // evaluate prospective dependencies without changing admission or memcache publication Port int // Specify if different than 3306 or if different than specified by MySQLConfigurationSettings IgnoreHostsCount int // Number of hosts that can be skipped/ignored even on error or on exceeding theesholds IgnoreHostsThreshold float64 // Threshold beyond which IgnoreHostsCount applies (default: 0) @@ -205,22 +206,36 @@ func (settings *MySQLConfigurationSettings) validateRequiredClusters() error { visit = func(clusterName string) error { switch states[clusterName] { case visiting: - return fmt.Errorf("Stores.MySQL.Clusters contains a RequiredClusters cycle involving %q", clusterName) + return fmt.Errorf("Stores.MySQL.Clusters contains a RequiredClusters/ShadowRequiredClusters cycle involving %q", clusterName) case visited: return nil } states[clusterName] = visiting - for _, requiredCluster := range settings.Clusters[clusterName].RequiredClusters { - if _, ok := settings.Clusters[requiredCluster]; !ok { - return fmt.Errorf( - "Stores.MySQL.Clusters.%s.RequiredClusters references unknown cluster %q", - clusterName, - requiredCluster, - ) + cluster := settings.Clusters[clusterName] + required := make(map[string]bool, len(cluster.RequiredClusters)) + for _, name := range cluster.RequiredClusters { + required[name] = true + } + for _, name := range cluster.ShadowRequiredClusters { + if required[name] { + return fmt.Errorf("Stores.MySQL.Clusters.%s: cluster %q cannot appear in both RequiredClusters and ShadowRequiredClusters", clusterName, name) } - if err := visit(requiredCluster); err != nil { - return err + } + for _, dependencies := range []struct { + field string + names []string + }{ + {"RequiredClusters", cluster.RequiredClusters}, + {"ShadowRequiredClusters", cluster.ShadowRequiredClusters}, + } { + for _, name := range dependencies.names { + if _, ok := settings.Clusters[name]; !ok { + return fmt.Errorf("Stores.MySQL.Clusters.%s.%s references unknown cluster %q", clusterName, dependencies.field, name) + } + if err := visit(name); err != nil { + return err + } } } states[clusterName] = visited diff --git a/pkg/http/api.go b/pkg/http/api.go index d0bd34d..5a605ef 100644 --- a/pkg/http/api.go +++ b/pkg/http/api.go @@ -366,7 +366,7 @@ type discardWriter struct{ io.Writer } func (discardWriter) Header() http.Header { return make(http.Header) } func (discardWriter) WriteHeader(int) {} -// Metrics exports process metrics, excluding aggregate and probe metrics on followers. +// Metrics exports process metrics, excluding leader-only metrics on followers. // Filtering happens here because published expvar values cannot be removed. func (api *APIImpl) Metrics(w http.ResponseWriter, r *http.Request, _ httprouter.Params) { metricsHandler.ServeHTTP(discardWriter{Writer: io.Discard}, r) @@ -375,7 +375,7 @@ func (api *APIImpl) Metrics(w http.ResponseWriter, r *http.Request, _ httprouter w.Header().Set("Content-Type", "application/json; charset=utf-8") metricValues := make(map[string]json.RawMessage) expvar.Do(func(metric expvar.KeyValue) { - if !isLeader && (strings.HasPrefix(metric.Key, "aggregated.") || strings.HasPrefix(metric.Key, "probes.")) { + if !isLeader && (strings.HasPrefix(metric.Key, "aggregated.") || strings.HasPrefix(metric.Key, "probes.") || strings.HasPrefix(metric.Key, "shadow.background.")) { return } metricValues[metric.Key] = json.RawMessage(metric.Value.String()) diff --git a/pkg/http/api_test.go b/pkg/http/api_test.go index 40eb7c6..1c548c1 100644 --- a/pkg/http/api_test.go +++ b/pkg/http/api_test.go @@ -120,9 +120,11 @@ func TestMetricsFiltersAggregateAndProbeMetricsFromFollowers(t *testing.T) { aggregatedMetricName = "aggregated.mysql.metrics-export-test" probeMetricName = "probes.total.metrics-export-test" processMetricName = "consensus.metrics-export-test" + backgroundMetricName = "shadow.background.mysql.metrics-export-test.would_reject" + httpMetricName = "shadow.http.mysql.metrics-export-test.total" ) - metricNames := []string{aggregatedMetricName, probeMetricName, processMetricName} + metricNames := []string{aggregatedMetricName, probeMetricName, processMetricName, backgroundMetricName, httpMetricName} for _, metricName := range metricNames { metrics.DefaultRegistry.Unregister(metricName) defer metrics.DefaultRegistry.Unregister(metricName) @@ -131,6 +133,8 @@ func TestMetricsFiltersAggregateAndProbeMetricsFromFollowers(t *testing.T) { metrics.GetOrRegisterGaugeFloat64(aggregatedMetricName, nil).Update(42) metrics.GetOrRegisterCounter(probeMetricName, nil).Inc(7) metrics.GetOrRegisterGauge(processMetricName, nil).Update(1) + metrics.GetOrRegisterCounter(backgroundMetricName, nil).Inc(2) + metrics.GetOrRegisterCounter(httpMetricName, nil).Inc(3) readMetrics := func(isLeader bool) map[string]interface{} { recorder := httptest.NewRecorder() @@ -149,6 +153,7 @@ func TestMetricsFiltersAggregateAndProbeMetricsFromFollowers(t *testing.T) { leaderMetrics := readMetrics(true) metrics.DefaultRegistry.Unregister(aggregatedMetricName) metrics.DefaultRegistry.Unregister(probeMetricName) + metrics.DefaultRegistry.Unregister(backgroundMetricName) if expvar.Get(aggregatedMetricName) == nil || expvar.Get(probeMetricName) == nil { t.Fatal("Expected aggregate and probe metrics to remain in expvar after unregister") } @@ -160,4 +165,12 @@ func TestMetricsFiltersAggregateAndProbeMetricsFromFollowers(t *testing.T) { if followerMetrics[aggregatedMetricName] != nil || followerMetrics[probeMetricName] != nil || followerMetrics[processMetricName] == nil { t.Errorf("Unexpected follower metrics: %v", followerMetrics) } + if leaderMetrics[backgroundMetricName] == nil || leaderMetrics[httpMetricName] == nil || + followerMetrics[backgroundMetricName] != nil || followerMetrics[httpMetricName] == nil { + t.Errorf("Unexpected shadow metrics: leader=%v follower=%v", leaderMetrics, followerMetrics) + } + if leaderMetrics[backgroundMetricName] != float64(2) || leaderMetrics[httpMetricName] != float64(3) || + followerMetrics[httpMetricName] != float64(3) { + t.Errorf("Expected integer shadow counter values, got leader=%v follower=%v", leaderMetrics, followerMetrics) + } } diff --git a/pkg/throttle/check.go b/pkg/throttle/check.go index ff0fb3b..34e367f 100644 --- a/pkg/throttle/check.go +++ b/pkg/throttle/check.go @@ -61,38 +61,19 @@ func (check *ThrottlerCheck) checkAppMetricResult(appName string, storeType stri return NewCheckResult(http.StatusExpectationFailed, value, threshold, fmt.Errorf("no app indicated")) } - statusCode := http.StatusInternalServerError // 500 - - if err == base.AppDeniedError { - // app specifically not allowed to get metrics - statusCode = http.StatusExpectationFailed // 417 - } else if err == base.NoSuchMetricError { - // not collected yet, or metric does not exist - statusCode = http.StatusNotFound // 404 - } else if errors.Is(err, base.NoHostsError) { - // Preserve legacy fail-open behavior unless the cluster explicitly requires - // an eligible host to prove that it is safe. - statusCode = http.StatusOK // 200 - if storeType == "mysql" { - if clusterSettings, ok := config.Settings().Stores.MySQL.Clusters[storeName]; ok && clusterSettings.FailOnNoHosts { - statusCode = http.StatusInternalServerError // 500 - } + failOnNoHosts := false + if storeType == "mysql" { + if clusterSettings, ok := config.Settings().Stores.MySQL.Clusters[storeName]; ok { + failOnNoHosts = clusterSettings.FailOnNoHosts } - } else if errors.Is(err, base.RecoveryNotCompleteError) { - statusCode = http.StatusTooManyRequests // 429 - } else if err != nil { - // any error - statusCode = http.StatusInternalServerError // 500 - } else if value > threshold { - // casual throttling - statusCode = http.StatusTooManyRequests // 429 - err = base.ThresholdExceededError - + } + checkResult = newMetricCheckResult(value, threshold, err, failOnNoHosts) + if err == nil && value > threshold { if !flags.LowPriority && !flags.ReadCheck && appName != frenoAppName { // low priority requests will henceforth be denied go check.throttler.nonLowPriorityAppRequestsThrottled.SetDefault(metricName, true) } - } else if appName != frenoAppName && check.throttler.getShareDomainSecondsSinceHealth(metricName) >= 1 { + } else if err == nil && checkResult.StatusCode == http.StatusOK && appName != frenoAppName && check.throttler.getShareDomainSecondsSinceHealth(metricName) >= 1 { // throttling based on shared domain metric. // we exclude the "freno" app itself, or else this could turn into a snowball: this service ("a") seeing // another service ("b") as unhealthy, itself becoming unhealthy, makind b's read into a's state as unheathly, @@ -100,13 +81,8 @@ func (check *ThrottlerCheck) checkAppMetricResult(appName string, storeType stri // The "freno" app is the one to generate those health metrics. It therefore must not participate the // shared-domain dependency check., - statusCode = http.StatusTooManyRequests // 429 - err = base.ThresholdExceededError - } else { - // all good! - statusCode = http.StatusOK // 200 + checkResult = NewCheckResult(http.StatusTooManyRequests, value, threshold, base.ThresholdExceededError) } - checkResult = NewCheckResult(statusCode, value, threshold, err) checkResult.MetricName = metricName return checkResult } @@ -154,6 +130,10 @@ func (check *ThrottlerCheck) checkMySQLCluster( // CheckAppStoreMetric func (check *ThrottlerCheck) Check(appName string, storeType string, storeName string, remoteAddr string, flags *CheckFlags) (checkResult *CheckResult) { + return check.check(appName, storeType, storeName, remoteAddr, flags, true) +} + +func (check *ThrottlerCheck) check(appName string, storeType string, storeName string, remoteAddr string, flags *CheckFlags, observeShadow bool) (checkResult *CheckResult) { requestedStoreName := storeName var metricResultFunc base.MetricResultFunc switch storeType { @@ -169,6 +149,12 @@ func (check *ThrottlerCheck) Check(appName string, storeType string, storeName s } if configured { checkResult = check.checkMySQLCluster(appName, storeName, flags, make(map[string]bool)) + if observeShadow && checkResult.StatusCode == http.StatusOK { + observeMySQLShadowDependencies("http", storeName, func(name string) *CheckResult { + metric, threshold := check.throttler.getMySQLClusterMetrics(name) + return mysqlDependencyCheckResult(name, metric, threshold, false) + }) + } } else { metricResultFunc = func() (metricResult base.MetricResult, threshold float64) { return base.NoSuchMetric, 0 @@ -230,8 +216,8 @@ func (check *ThrottlerCheck) localCheck(metricName string) (checkResult *CheckRe if err != nil { return NoSuchMetricCheckResult } - checkResult = check.Check(frenoAppName, storeType, storeName, "local", StandardCheckFlags) - go check.Check(frenoShareDmainAppName, storeType, storeName, "local", StandardCheckFlags) + checkResult = check.check(frenoAppName, storeType, storeName, "local", StandardCheckFlags, false) + go check.check(frenoShareDmainAppName, storeType, storeName, "local", StandardCheckFlags, false) if checkResult.StatusCode == http.StatusOK { check.throttler.markMetricHealthy(metricName) diff --git a/pkg/throttle/check_result.go b/pkg/throttle/check_result.go index 4e60bc1..a6730eb 100644 --- a/pkg/throttle/check_result.go +++ b/pkg/throttle/check_result.go @@ -1,11 +1,34 @@ package throttle import ( + "errors" "net/http" "github.com/github/freno/pkg/base" ) +func newMetricCheckResult(value, threshold float64, err error, failOnNoHosts bool) *CheckResult { + status := http.StatusOK + switch { + case err == base.AppDeniedError: + status = http.StatusExpectationFailed + case err == base.NoSuchMetricError: + status = http.StatusNotFound + case errors.Is(err, base.NoHostsError): + if failOnNoHosts { + status = http.StatusInternalServerError + } + case errors.Is(err, base.RecoveryNotCompleteError): + status = http.StatusTooManyRequests + case err != nil: + status = http.StatusInternalServerError + case value > threshold: + status = http.StatusTooManyRequests + err = base.ThresholdExceededError + } + return NewCheckResult(status, value, threshold, err) +} + // CheckResult is the result for an app inquiring on a metric. It also exports as JSON via the API type CheckResult struct { StatusCode int `json:"StatusCode"` diff --git a/pkg/throttle/shadow.go b/pkg/throttle/shadow.go new file mode 100644 index 0000000..8a2542e --- /dev/null +++ b/pkg/throttle/shadow.go @@ -0,0 +1,74 @@ +package throttle + +import ( + "fmt" + "net/http" + + "github.com/github/freno/pkg/base" + "github.com/github/freno/pkg/config" + metrics "github.com/rcrowley/go-metrics" +) + +func mysqlDependencyCheckResult(name string, metric base.MetricResult, threshold float64, background bool) *CheckResult { + settings := config.Settings().Stores.MySQL.Clusters[name] + value, err := metric.Get() + result := newMetricCheckResult(value, threshold, err, background || settings.FailOnNoHosts) + if result.StatusCode == http.StatusNotFound { + result = NewErrorCheckResult(http.StatusInternalServerError, fmt.Errorf("required MySQL cluster metric %q is unavailable", name)) + } + result.MetricName = "mysql/" + name + return result +} + +// Observe only new dependencies. The real required closure already passed and must +// not be resampled or attributed to a shadow failure. +func observeMySQLShadowDependencies(source, root string, evaluate func(string) *CheckResult) { + clusters := config.Settings().Stores.MySQL.Clusters + seen := make(map[string]bool) + var pending []string + var seed func(string) + seed = func(name string) { + if seen[name] { + return + } + seen[name] = true + pending = append(pending, clusters[name].ShadowRequiredClusters...) + for _, required := range clusters[name].RequiredClusters { + seed(required) + } + } + seed(root) + if len(pending) == 0 { + return + } + prefix := fmt.Sprintf("shadow.%s.mysql.%s", source, root) + metrics.GetOrRegisterCounter(prefix+".total", nil).Inc(1) + rejections := metrics.GetOrRegisterCounter(prefix+".would_reject", nil) + var visit func(string) *CheckResult + visit = func(name string) *CheckResult { + if seen[name] { + return nil + } + seen[name] = true + result := evaluate(name) + if result.StatusCode != http.StatusOK { + metrics.GetOrRegisterCounter(prefix+".dependency."+name+".would_reject", nil).Inc(1) + return result + } + for _, names := range [][]string{clusters[name].RequiredClusters, clusters[name].ShadowRequiredClusters} { + for _, dependency := range names { + if failure := visit(dependency); failure != nil { + return failure + } + } + } + return nil + } + for _, name := range pending { + if failure := visit(name); failure != nil { + rejections.Inc(1) + metrics.GetOrRegisterCounter(fmt.Sprintf("%s.status.%d.would_reject", prefix, failure.StatusCode), nil).Inc(1) + return + } + } +} diff --git a/pkg/throttle/shadow_test.go b/pkg/throttle/shadow_test.go new file mode 100644 index 0000000..c422ef0 --- /dev/null +++ b/pkg/throttle/shadow_test.go @@ -0,0 +1,309 @@ +package throttle + +import ( + "bufio" + "errors" + "fmt" + "io" + "net" + "net/http" + "strings" + "testing" + "time" + + "github.com/bradfitz/gomemcache/memcache" + "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 resetShadowCounters(t *testing.T) { + clear := func() { + metrics.DefaultRegistry.Each(func(name string, _ interface{}) { + if strings.HasPrefix(name, "shadow.") { + metrics.DefaultRegistry.Unregister(name) + } + }) + } + clear() + t.Cleanup(clear) +} + +func shadowCount(name string) int64 { + if counter := metrics.Get(name); counter != nil { + return counter.(metrics.Counter).Count() + } + return 0 +} + +func newShadowTestCheck(t *testing.T) (*ThrottlerCheck, *Throttler) { + original := config.Settings().Stores.MySQL + t.Cleanup(func() { config.Settings().Stores.MySQL = original }) + resetShadowCounters(t) + config.Settings().Stores.MySQL = config.MySQLConfigurationSettings{ + FallbackCluster: "root", + Clusters: map[string]*config.MySQLClusterConfigurationSettings{ + "root": {ThrottleThreshold: 5, RequiredClusters: []string{"required"}, ShadowRequiredClusters: []string{"shadow"}}, + "required": {ThrottleThreshold: 10}, + "shadow": {ThrottleThreshold: 20, FailOnNoHosts: true}, + }, + } + throttler := NewThrottler() + setTestMySQLClusterMetric(throttler, "root", 1, 5) + setTestMySQLClusterMetric(throttler, "required", 2, 10) + setTestMySQLClusterMetric(throttler, "shadow", 3, 20) + return NewThrottlerCheck(throttler), throttler +} + +func TestShadowHTTPDoesNotChangeAdmission(t *testing.T) { + for _, test := range []struct { + name string + metric base.MetricResult + failOpen bool + wantStatus int + }{ + {"healthy", base.NewSimpleMetricResult(20), false, 0}, + {"overloaded", base.NewSimpleMetricResult(21), false, 429}, + {"collection error", base.NewErrorMetricResult(0, errors.New("query denied")), false, 500}, + {"not collected", base.NoMetricResultYet, false, 500}, + {"missing metric", base.NoSuchMetric, false, 500}, + {"no hosts fail closed", base.NoHostsMetricResult, false, 500}, + {"no hosts legacy", base.NoHostsMetricResult, true, 0}, + {"recovering", base.NewErrorMetricResult(3, base.RecoveryNotCompleteError), false, 429}, + } { + t.Run(test.name, func(t *testing.T) { + check, throttler := newShadowTestCheck(t) + config.Settings().Stores.MySQL.Clusters["shadow"].FailOnNoHosts = !test.failOpen + throttler.aggregatedMetrics.Set("mysql/shadow", test.metric, cache.DefaultExpiration) + before := check.checkMySQLCluster("test-app", "root", &CheckFlags{}, make(map[string]bool)) + result := check.Check("test-app", "mysql", "root", "", &CheckFlags{}) + assert.Equal(t, before, result) + assert.Equal(t, http.StatusOK, result.StatusCode) + assert.Equal(t, int64(1), shadowCount("shadow.http.mysql.root.total")) + wantRejections := int64(0) + if test.wantStatus != 0 { + wantRejections = 1 + assert.Equal(t, int64(1), shadowCount(fmt.Sprintf("shadow.http.mysql.root.status.%d.would_reject", test.wantStatus))) + } + assert.Equal(t, wantRejections, shadowCount("shadow.http.mysql.root.would_reject")) + assert.Equal(t, wantRejections, shadowCount("shadow.http.mysql.root.dependency.shadow.would_reject")) + assert.Empty(t, throttler.nonLowPriorityAppRequestsThrottled.Items()) + assert.Empty(t, throttler.metricsHealth.Items()) + assert.Empty(t, throttler.mysqlRecoveryState) + assert.Equal(t, int64(0), shadowCount("shadow.background.mysql.root.total")) + }) + } +} + +func TestShadowHTTPOnlyOtherwiseAllowedRequests(t *testing.T) { + for _, rejected := range []string{"root", "required"} { + t.Run(rejected, func(t *testing.T) { + check, throttler := newShadowTestCheck(t) + throttler.aggregatedMetrics.Set("mysql/"+rejected, base.NewSimpleMetricResult(100), cache.DefaultExpiration) + result := check.Check("test-app", "mysql", "root", "", &CheckFlags{}) + assert.Equal(t, http.StatusTooManyRequests, result.StatusCode) + assert.Equal(t, "mysql/"+rejected, result.MetricName) + assert.Nil(t, metrics.Get("shadow.http.mysql.root.total")) + }) + } + t.Run("real app denial", func(t *testing.T) { + check, throttler := newShadowTestCheck(t) + throttler.ThrottleApp("test-app", time.Time{}, 1) + assert.Equal(t, http.StatusExpectationFailed, check.Check("test-app", "mysql", "root", "", &CheckFlags{}).StatusCode) + assert.Nil(t, metrics.Get("shadow.http.mysql.root.total")) + }) +} + +func TestShadowHTTPStrictThresholdsAndPromotion(t *testing.T) { + check, throttler := newShadowTestCheck(t) + throttler.aggregatedMetrics.Set("mysql/root", base.NewErrorMetricResult(6, base.RecoveryNotCompleteError), cache.DefaultExpiration) + throttler.aggregatedMetrics.Set("mysql/shadow", base.NewErrorMetricResult(3, base.RecoveryNotCompleteError), cache.DefaultExpiration) + flags := &CheckFlags{ReadCheck: true, OverrideThreshold: 100} + result := check.Check("test-app", "mysql", "unknown-client-name", "", flags) + assert.Equal(t, http.StatusOK, result.StatusCode) + assert.Equal(t, 100.0, result.Threshold) + assert.Equal(t, int64(1), shadowCount("shadow.http.mysql.root.status.429.would_reject")) + assert.Nil(t, metrics.Get("shadow.http.mysql.unknown-client-name.total")) + + root := config.Settings().Stores.MySQL.Clusters["root"] + root.RequiredClusters = append(root.RequiredClusters, root.ShadowRequiredClusters...) + root.ShadowRequiredClusters = nil + result = check.Check("test-app", "mysql", "root", "", flags) + assert.Equal(t, http.StatusTooManyRequests, result.StatusCode) + assert.Equal(t, "mysql/shadow", result.MetricName) + assert.Equal(t, 20.0, result.Threshold) + assert.Equal(t, int64(1), shadowCount("shadow.http.mysql.root.total")) +} + +func TestShadowSkipsAppPolicyAndSelfChecks(t *testing.T) { + check, throttler := newShadowTestCheck(t) + throttler.ThrottleApp("test-app/shadow", time.Time{}, 1) + throttler.shareDomainMetricHealth.SetDefault("mysql/shadow", &base.MetricHealth{SecondsSinceLastHealthy: 10}) + throttler.nonLowPriorityAppRequestsThrottled.SetDefault("mysql/shadow", true) + result := check.Check("test-app", "mysql", "root", "", &CheckFlags{LowPriority: true}) + assert.Equal(t, http.StatusOK, result.StatusCode) + assert.Equal(t, int64(0), shadowCount("shadow.http.mysql.root.would_reject")) + + throttler.aggregatedMetrics.Set("mysql/shadow", base.NewSimpleMetricResult(100), cache.DefaultExpiration) + result = check.localCheck("mysql/root") + deadline := time.Now().Add(time.Second) + for { + if _, finished := throttler.recentApps.Get(frenoShareDmainAppName + "/local"); finished { + break + } + if time.Now().After(deadline) { + t.Fatal("shared-domain self-check did not finish") + } + time.Sleep(time.Millisecond) + } + assert.Equal(t, http.StatusOK, result.StatusCode) + assert.Equal(t, int64(1), shadowCount("shadow.http.mysql.root.total")) + assert.Equal(t, int64(0), shadowCount("shadow.http.mysql.root.would_reject")) + _, markedHealthy := throttler.metricsHealth.Get("mysql/root") + assert.True(t, markedHealthy) +} + +func TestShadowGraphIncludesNestedDependenciesOnce(t *testing.T) { + newShadowTestCheck(t) + clusters := config.Settings().Stores.MySQL.Clusters + clusters["root"].ShadowRequiredClusters = nil + clusters["root"].RequiredClusters = append(clusters["root"].RequiredClusters, "shared") + clusters["required"].ShadowRequiredClusters = []string{"shadow", "shadow"} + clusters["shadow"].RequiredClusters = []string{"shared", "nested"} + clusters["shadow"].ShadowRequiredClusters = []string{"extra"} + clusters["shared"] = &config.MySQLClusterConfigurationSettings{} + clusters["nested"] = &config.MySQLClusterConfigurationSettings{} + clusters["extra"] = &config.MySQLClusterConfigurationSettings{RequiredClusters: []string{"nested"}} + calls := make(map[string]int) + evaluate := func(name string) *CheckResult { + calls[name]++ + return NewCheckResult(http.StatusOK, 0, 1, nil) + } + observeMySQLShadowDependencies("http", "root", evaluate) + assert.Equal(t, map[string]int{"shadow": 1, "nested": 1, "extra": 1}, calls) + assert.Equal(t, int64(1), shadowCount("shadow.http.mysql.root.total")) + assert.Equal(t, int64(0), shadowCount("shadow.http.mysql.root.would_reject")) + + calls = make(map[string]int) + observeMySQLShadowDependencies("http", "root", func(name string) *CheckResult { + if name == "nested" { + return NewErrorCheckResult(http.StatusInternalServerError, errors.New("missing")) + } + return evaluate(name) + }) + assert.Equal(t, int64(1), shadowCount("shadow.http.mysql.root.dependency.nested.would_reject")) + assert.Equal(t, 0, calls["extra"]) +} + +func addShadowInventoryMetric(throttler *Throttler, name string, metric base.MetricResult) { + key := mysql.InstanceKey{Hostname: name, Port: 3306} + probes := mysql.Probes{key: &mysql.Probe{Key: key}} + throttler.mysqlInventory.ClustersProbes[name] = &probes + throttler.mysqlInventory.InstanceKeyMetrics[mysql.GetClusterInstanceKey(name, &key)] = metric +} + +func TestShadowBackgroundSeparateFromHTTP(t *testing.T) { + for _, test := range []struct { + name string + root base.MetricResult + required base.MetricResult + shadow base.MetricResult + wantTotal int64 + wantReject int64 + }{ + {"healthy", base.NewSimpleMetricResult(1), base.NewSimpleMetricResult(2), base.NewSimpleMetricResult(3), 1, 0}, + {"overload", base.NewSimpleMetricResult(1), base.NewSimpleMetricResult(2), base.NewSimpleMetricResult(21), 1, 1}, + {"legacy no hosts still unsafe for memcache", base.NewSimpleMetricResult(1), base.NewSimpleMetricResult(2), base.NoHostsMetricResult, 1, 1}, + {"root denied", base.NewSimpleMetricResult(6), base.NewSimpleMetricResult(2), base.NewSimpleMetricResult(21), 0, 0}, + {"required denied", base.NewSimpleMetricResult(1), base.NewSimpleMetricResult(11), base.NewSimpleMetricResult(21), 0, 0}, + {"root error", base.NoMetricResultYet, base.NewSimpleMetricResult(2), base.NewSimpleMetricResult(21), 0, 0}, + } { + t.Run(test.name, func(t *testing.T) { + _, throttler := newShadowTestCheck(t) + config.Settings().Stores.MySQL.Clusters["shadow"].FailOnNoHosts = false + addShadowInventoryMetric(throttler, "root", test.root) + addShadowInventoryMetric(throttler, "required", test.required) + addShadowInventoryMetric(throttler, "shadow", test.shadow) + assert.NoError(t, throttler.aggregateMySQLMetrics()) + assert.Nil(t, metrics.Get("shadow.background.mysql.root.total")) + throttler.isLeader = true + assert.NoError(t, throttler.aggregateMySQLMetrics()) + assert.Equal(t, test.wantTotal, shadowCount("shadow.background.mysql.root.total")) + assert.Equal(t, test.wantReject, shadowCount("shadow.background.mysql.root.would_reject")) + assert.Nil(t, metrics.Get("shadow.http.mysql.root.total")) + }) + } +} + +// Read one real publication command, rather than inferring memcache behavior from +// the dependency predicate alone. +func TestShadowMemcachePublicationAndPromotion(t *testing.T) { + for _, enforce := range []bool{false, true} { + t.Run(fmt.Sprintf("enforced=%t", enforce), func(t *testing.T) { + _, throttler := newShadowTestCheck(t) + root := config.Settings().Stores.MySQL.Clusters["root"] + root.RequiredClusters = nil + if enforce { + root.RequiredClusters = root.ShadowRequiredClusters + root.ShadowRequiredClusters = nil + } + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer listener.Close() + type publication struct { + command string + err error + } + published := make(chan publication, 1) + go func() { + conn, err := listener.Accept() + if err != nil { + published <- publication{err: err} + return + } + defer conn.Close() + if err := conn.SetDeadline(time.Now().Add(2 * time.Second)); err != nil { + published <- publication{err: err} + return + } + reader := bufio.NewReader(conn) + line, err := reader.ReadString('\n') + if err == nil && strings.HasPrefix(line, "set ") { + _, err = reader.ReadString('\n') + } + if err == nil { + response := "STORED\r\n" + if strings.HasPrefix(line, "delete ") { + response = "DELETED\r\n" + } + _, err = io.WriteString(conn, response) + } + published <- publication{command: line, err: err} + }() + throttler.memcacheClient = memcache.New(listener.Addr().String()) + throttler.memcachePath = "shadow-test" + throttler.isLeader = true + addShadowInventoryMetric(throttler, "root", base.NewSimpleMetricResult(1)) + assert.NoError(t, throttler.aggregateMySQLMetrics()) + select { + case result := <-published: + assert.NoError(t, result.err) + want := "set shadow-test/mysql/root " + if enforce { + want = "delete shadow-test/mysql/root" + } + assert.True(t, strings.HasPrefix(result.command, want), result.command) + case <-time.After(3 * time.Second): + t.Fatal("timed out waiting for memcache publication") + } + if !enforce { + assert.Equal(t, int64(1), shadowCount("shadow.background.mysql.root.status.500.would_reject")) + } + }) + } +} diff --git a/pkg/throttle/throttler.go b/pkg/throttle/throttler.go index 5ac93c4..1a4d65d 100644 --- a/pkg/throttle/throttler.go +++ b/pkg/throttle/throttler.go @@ -472,6 +472,18 @@ func (throttler *Throttler) aggregateMySQLMetrics() error { for clusterName, aggregatedMetric := range aggregatedMetrics { metricName := fmt.Sprintf("mysql/%s", clusterName) go throttler.aggregatedMetrics.Set(metricName, aggregatedMetric, cache.DefaultExpiration) + value, err := aggregatedMetric.Get() + settings := config.Settings().Stores.MySQL.Clusters[clusterName] + if settings != nil && err == nil && value <= settings.ThrottleThreshold && + throttler.requiredMySQLClustersHealthy(clusterName, aggregatedMetrics, make(map[string]bool)) { + observeMySQLShadowDependencies("background", clusterName, func(name string) *CheckResult { + metric, found := aggregatedMetrics[name] + if !found { + metric = base.NoSuchMetric + } + return mysqlDependencyCheckResult(name, metric, config.Settings().Stores.MySQL.Clusters[name].ThrottleThreshold, true) + }) + } if throttler.memcacheClient != nil { requiredClustersHealthy := throttler.requiredMySQLClustersHealthy(clusterName, aggregatedMetrics, make(map[string]bool)) go func(metricName string, aggregatedMetric base.MetricResult, requiredClustersHealthy bool) {