From d4fb5b70264436e0c9055d0aa2b70a19d2288b13 Mon Sep 17 00:00:00 2001 From: Pradeep Kintali Date: Wed, 23 Sep 2026 15:58:18 -0700 Subject: [PATCH 1/9] Add composite MySQL safety checks Allow replica-lag checks to require cached primary and ProxySQL safety metrics, fail closed for configured probes, and hold recovery until a cooling period completes. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 0fbc396f-32f9-449a-8537-6264e8665c31 --- doc/mysql.md | 75 ++++++++++++++++ pkg/base/throttle_metric.go | 14 +++ pkg/config/config_test.go | 91 +++++++++++++++++++ pkg/config/mysql_config.go | 99 ++++++++++++++++++--- pkg/throttle/check.go | 60 +++++++++++-- pkg/throttle/throttler.go | 62 +++++++++++++ pkg/throttle/throttler_test.go | 158 +++++++++++++++++++++++++++++++++ 7 files changed, 540 insertions(+), 19 deletions(-) diff --git a/doc/mysql.md b/doc/mysql.md index 9815050..3aaebca 100644 --- a/doc/mysql.md +++ b/doc/mysql.md @@ -148,3 +148,78 @@ Noteworthy: `freno` explicitly recognizes `show global ...` statements and reads the result's numeric value. Otherwise you may provide any query that returns a single row, single numeric column. + +### 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: + +```json +"Clusters": { + "prod4": { + "RequiredClusters": [ + "prod4-primary", + "prod4-proxysql" + ], + "HAProxySettings": { + "Host": "my.haproxy.mydomain.com", + "Port": 1001, + "PoolName": "my_prod4_pool" + } + }, + "prod4-primary": { + "MetricQuery": "show global status like 'Threads_running'", + "CacheMillis": 500, + "ThrottleThreshold": 50, + "RecoveryThreshold": 35, + "RecoveryDurationMillis": 5000, + "FailOnNoHosts": true, + "StaticHostsSettings": { + "Hosts": [ + "my.prod4.primary.vip.example.com:3306" + ] + } + }, + "prod4-proxysql": { + "User": "${proxysql_stats_user}", + "Password": "${proxysql_stats_password}", + "MetricQuery": "select connection_pressure_percent from operational_metrics.proxysql_capacity limit 1", + "CacheMillis": 500, + "ThrottleThreshold": 80, + "RecoveryThreshold": 60, + "RecoveryDurationMillis": 5000, + "FailOnNoHosts": true, + "StaticHostsSettings": { + "Hosts": [ + "my.prod4.proxysql.example.com:6032" + ] + } + } +} +``` + +The ProxySQL query above is illustrative: deployments must provide a scalar query or operational view whose value increases as usable connection capacity is consumed. + +- `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. +- `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. +- 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. + +Configuration loading rejects missing `RequiredClusters` references and dependency cycles. + +### A1 repair mapping + +| A1 repair item | Freno implementation in this change | Remaining work | +| --- | --- | --- | +| Default transitions to one worker | Not a Freno responsibility. | Enforce in the transitions framework. | +| Audit custom worker configurations | Not a Freno responsibility. | Audit and constrain custom transitions in the application. | +| Identify Freno traffic as transitions | Existing Freno API already accepts the application name and emits per-app/per-store counters. | Complete caller rollout with `app=transitions` and monitor that identity. | +| Mark transition writes as low priority | Existing `p=low` behavior remains compatible with composite checks. | Complete application rollout. | +| Page on primary and ProxySQL connection failures | `FailOnNoHosts` and missing-required-metric handling prevent an unobservable probe failure from allowing work. | Add Datadog paging for probe and connectivity failures. | +| Page on sustained primary `Threads_running` | A cached primary metric can now be required by the replica-lag decision; recovery hysteresis prevents immediate release. | Configure cluster thresholds and add the sustained Datadog monitor. | +| Page before connection exhaustion | A cached ProxySQL capacity metric can now be required by the admission decision. | Define the production capacity query, reserve threshold, and Datadog monitor. | +| Detect rising concurrency with falling completions | A derived scalar metric can be configured as another required cluster check. | Define and validate the correlation query and alert thresholds. | +| Alert on sustained Freno rejection | Composite safety failures return normal Freno rejection/error status and existing counters remain available by app and store. | Add the rejection-ratio and minimum-volume Datadog monitor grouped by cluster and `app=transitions`. | +| Provide a cluster-wide emergency pause | Existing app/store throttling remains the emergency control and is evaluated for each required metric. | Operationalize a deterministic transitions-plus-cluster command and runbook. | + +These changes provide admission enforcement; they do not create Datadog monitors, change transition worker defaults, or bound replica-to-primary fallback inside the application. diff --git a/pkg/base/throttle_metric.go b/pkg/base/throttle_metric.go index 8c486fc..93362e3 100644 --- a/pkg/base/throttle_metric.go +++ b/pkg/base/throttle_metric.go @@ -18,6 +18,7 @@ type MetricResultFunc func() (metricResult MetricResult, threshold float64) var ThresholdExceededError = errors.New("Threshold exceeded") var NoHostsError = errors.New("No hosts found") +var RecoveryNotCompleteError = errors.New("Recovery period not complete") var noResultYetError = errors.New("Metric not collected yet") var NoSuchMetricError = errors.New("No such metric") @@ -63,3 +64,16 @@ func NewSimpleMetricResult(value float64) MetricResult { func (metricResult *simpleMetricResult) Get() (float64, error) { return metricResult.Value, nil } + +type errorMetricResult struct { + Value float64 + Err error +} + +func NewErrorMetricResult(value float64, err error) MetricResult { + return &errorMetricResult{Value: value, Err: err} +} + +func (metricResult *errorMetricResult) Get() (float64, error) { + return metricResult.Value, metricResult.Err +} diff --git a/pkg/config/config_test.go b/pkg/config/config_test.go index 21216e0..92ce27d 100644 --- a/pkg/config/config_test.go +++ b/pkg/config/config_test.go @@ -112,6 +112,97 @@ func TestMySQLFallbackCluster(t *testing.T) { } } +func TestMySQLRequiredClusters(t *testing.T) { + tests := []struct { + name string + clusters map[string]*MySQLClusterConfigurationSettings + wantErr string + }{ + { + name: "valid dependencies", + clusters: map[string]*MySQLClusterConfigurationSettings{ + "replicas": {RequiredClusters: []string{"primary", "proxysql"}}, + "primary": {}, + "proxysql": {}, + }, + }, + { + name: "unknown dependency", + clusters: map[string]*MySQLClusterConfigurationSettings{ + "replicas": {RequiredClusters: []string{"primary"}}, + }, + wantErr: `Stores.MySQL.Clusters.replicas.RequiredClusters references unknown cluster "primary"`, + }, + { + name: "dependency cycle", + clusters: map[string]*MySQLClusterConfigurationSettings{ + "replicas": {RequiredClusters: []string{"primary"}}, + "primary": {RequiredClusters: []string{"replicas"}}, + }, + wantErr: `Stores.MySQL.Clusters contains a RequiredClusters cycle involving "primary"`, + }, + { + name: "null cluster", + clusters: map[string]*MySQLClusterConfigurationSettings{ + "replicas": nil, + }, + wantErr: `Stores.MySQL.Clusters.replicas must not be null`, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + settings := MySQLConfigurationSettings{Clusters: test.clusters} + err := settings.postReadAdjustments() + if test.wantErr == "" && err != nil { + t.Fatalf("unexpected error: %v", err) + } + if test.wantErr != "" && (err == nil || err.Error() != test.wantErr) { + t.Fatalf("error = %v, want %q", err, test.wantErr) + } + }) + } +} + +func TestMySQLRecoveryConfiguration(t *testing.T) { + recoveryThreshold := 5.0 + tooHighRecoveryThreshold := 11.0 + tests := []struct { + name string + clusterSettings *MySQLClusterConfigurationSettings + wantErr string + }{ + {name: "disabled", clusterSettings: &MySQLClusterConfigurationSettings{ThrottleThreshold: 10}}, + {name: "valid", clusterSettings: &MySQLClusterConfigurationSettings{ThrottleThreshold: 10, RecoveryThreshold: &recoveryThreshold, RecoveryDurationMillis: 5000}}, + {name: "default recovery threshold", clusterSettings: &MySQLClusterConfigurationSettings{ThrottleThreshold: 10, RecoveryDurationMillis: 5000}}, + { + name: "negative duration", + clusterSettings: &MySQLClusterConfigurationSettings{ThrottleThreshold: 10, RecoveryDurationMillis: -1}, + wantErr: "Stores.MySQL.Clusters.primary.RecoveryDurationMillis must not be negative", + }, + { + name: "recovery threshold above throttle threshold", + clusterSettings: &MySQLClusterConfigurationSettings{ThrottleThreshold: 10, RecoveryThreshold: &tooHighRecoveryThreshold, RecoveryDurationMillis: 5000}, + wantErr: "Stores.MySQL.Clusters.primary.RecoveryThreshold must not exceed ThrottleThreshold", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + settings := MySQLConfigurationSettings{ + Clusters: map[string]*MySQLClusterConfigurationSettings{"primary": test.clusterSettings}, + } + err := settings.postReadAdjustments() + if test.wantErr == "" && err != nil { + t.Fatalf("unexpected error: %v", err) + } + if test.wantErr != "" && (err == nil || err.Error() != test.wantErr) { + t.Fatalf("error = %v, want %q", err, test.wantErr) + } + }) + } +} + func dump(path string, contents *ConfigurationSettings) error { json, _ := json.Marshal(contents) err := ioutil.WriteFile(path, json, 0644) diff --git a/pkg/config/mysql_config.go b/pkg/config/mysql_config.go index 8e5a7bf..3944a65 100644 --- a/pkg/config/mysql_config.go +++ b/pkg/config/mysql_config.go @@ -7,22 +7,27 @@ package config import ( "fmt" "os" + "sort" ) const DefaultMySQLPort = 3306 type MySQLClusterConfigurationSettings struct { - User string // override MySQLConfigurationSettings's, or leave empty to inherit those settings - Password string // override MySQLConfigurationSettings's, or leave empty to inherit those settings - MetricQuery string // override MySQLConfigurationSettings's, or leave empty to inherit those settings - CacheMillis int // override MySQLConfigurationSettings's, or leave empty to inherit those settings - ThrottleThreshold float64 // override MySQLConfigurationSettings's, or leave empty to inherit those settings - 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) - HttpCheckPort int // Specify if different than specified by MySQLConfigurationSettings. -1 to disable HTTP check - HttpCheckPath string // Specify if different than specified by MySQLConfigurationSettings - IgnoreHosts []string // override MySQLConfigurationSettings's, or leave empty to inherit those settings + User string // override MySQLConfigurationSettings's, or leave empty to inherit those settings + Password string // override MySQLConfigurationSettings's, or leave empty to inherit those settings + MetricQuery string // override MySQLConfigurationSettings's, or leave empty to inherit those settings + CacheMillis int // override MySQLConfigurationSettings's, or leave empty to inherit those settings + ThrottleThreshold float64 // override MySQLConfigurationSettings's, or leave empty to inherit those settings + RecoveryThreshold *float64 // optional threshold that must be met before a throttled cluster begins recovery + 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 + 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) + HttpCheckPort int // Specify if different than specified by MySQLConfigurationSettings. -1 to disable HTTP check + 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 @@ -75,6 +80,9 @@ func (settings *MySQLConfigurationSettings) postReadAdjustments() error { if settings.Port == 0 { settings.Port = DefaultMySQLPort } + if settings.Clusters == nil { + settings.Clusters = make(map[string]*MySQLClusterConfigurationSettings) + } if settings.FallbackCluster != "" { if _, ok := settings.Clusters[settings.FallbackCluster]; !ok { return fmt.Errorf("Stores.MySQL.FallbackCluster %q does not name a configured cluster", settings.FallbackCluster) @@ -97,7 +105,10 @@ func (settings *MySQLConfigurationSettings) postReadAdjustments() error { } } - for _, clusterSettings := range settings.Clusters { + for clusterName, clusterSettings := range settings.Clusters { + if clusterSettings == nil { + return fmt.Errorf("Stores.MySQL.Clusters.%s must not be null", clusterName) + } if err := clusterSettings.postReadAdjustments(); err != nil { return err } @@ -148,6 +159,70 @@ func (settings *MySQLConfigurationSettings) postReadAdjustments() error { if !clusterSettings.VitessSettings.IsEmpty() && len(clusterSettings.VitessSettings.Cells) < 1 { clusterSettings.VitessSettings.Cells = settings.VitessCells } + if clusterSettings.RecoveryDurationMillis < 0 { + return fmt.Errorf("Stores.MySQL.Clusters.%s.RecoveryDurationMillis must not be negative", clusterName) + } + if clusterSettings.RecoveryDurationMillis > 0 { + recoveryThreshold := clusterSettings.ThrottleThreshold + if clusterSettings.RecoveryThreshold != nil { + recoveryThreshold = *clusterSettings.RecoveryThreshold + } + if recoveryThreshold > clusterSettings.ThrottleThreshold { + return fmt.Errorf( + "Stores.MySQL.Clusters.%s.RecoveryThreshold must not exceed ThrottleThreshold", + clusterName, + ) + } + } + } + return settings.validateRequiredClusters() +} + +func (settings *MySQLConfigurationSettings) validateRequiredClusters() error { + const ( + unvisited = iota + visiting + visited + ) + states := make(map[string]int) + + var visit func(string) error + visit = func(clusterName string) error { + switch states[clusterName] { + case visiting: + return fmt.Errorf("Stores.MySQL.Clusters contains a RequiredClusters 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, + ) + } + if err := visit(requiredCluster); err != nil { + return err + } + } + states[clusterName] = visited + return nil + } + + clusterNames := make([]string, 0, len(settings.Clusters)) + for clusterName := range settings.Clusters { + clusterNames = append(clusterNames, clusterName) + } + sort.Strings(clusterNames) + for _, clusterName := range clusterNames { + if states[clusterName] == unvisited { + if err := visit(clusterName); err != nil { + return err + } + } } return nil } diff --git a/pkg/throttle/check.go b/pkg/throttle/check.go index 0c1cc7b..ea5cc88 100644 --- a/pkg/throttle/check.go +++ b/pkg/throttle/check.go @@ -67,8 +67,16 @@ func (check *ThrottlerCheck) checkAppMetricResult(appName string, storeType stri // not collected yet, or metric does not exist statusCode = http.StatusNotFound // 404 } else if errors.Is(err, base.NoHostsError) { - // If no hosts we report 0 for the metric. + // 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 + } + } + } else if errors.Is(err, base.RecoveryNotCompleteError) { + statusCode = http.StatusTooManyRequests // 429 } else if err != nil { // any error statusCode = http.StatusInternalServerError // 500 @@ -98,6 +106,44 @@ func (check *ThrottlerCheck) checkAppMetricResult(appName string, storeType stri return NewCheckResult(statusCode, value, threshold, err) } +func (check *ThrottlerCheck) checkMySQLCluster( + appName string, + clusterName string, + flags *CheckFlags, + checkedClusters map[string]bool, +) *CheckResult { + if checkedClusters[clusterName] { + return nil + } + checkedClusters[clusterName] = true + + metricResultFunc := func() (metricResult base.MetricResult, threshold float64) { + return check.throttler.getMySQLClusterMetrics(clusterName) + } + rootResult := check.checkAppMetricResult(appName, "mysql", clusterName, metricResultFunc, flags) + if rootResult.StatusCode != http.StatusOK { + return rootResult + } + + clusterSettings := config.Settings().Stores.MySQL.Clusters[clusterName] + for _, requiredCluster := range clusterSettings.RequiredClusters { + requiredFlags := *flags + requiredFlags.OverrideThreshold = 0 + requiredResult := check.checkMySQLCluster(appName, requiredCluster, &requiredFlags, checkedClusters) + if requiredResult == nil || requiredResult.StatusCode == http.StatusOK { + continue + } + if requiredResult.StatusCode == http.StatusNotFound { + return NewErrorCheckResult( + http.StatusInternalServerError, + fmt.Errorf("required MySQL cluster metric %q is unavailable", requiredCluster), + ) + } + return requiredResult + } + return rootResult +} + // CheckAppStoreMetric func (check *ThrottlerCheck) Check(appName string, storeType string, storeName string, remoteAddr string, flags *CheckFlags) (checkResult *CheckResult) { requestedStoreName := storeName @@ -114,9 +160,7 @@ func (check *ThrottlerCheck) Check(appName string, storeType string, storeName s } } if configured { - metricResultFunc = func() (metricResult base.MetricResult, threshold float64) { - return check.throttler.getMySQLClusterMetrics(storeName) - } + checkResult = check.checkMySQLCluster(appName, storeName, flags, make(map[string]bool)) } else { metricResultFunc = func() (metricResult base.MetricResult, threshold float64) { return base.NoSuchMetric, 0 @@ -125,11 +169,13 @@ func (check *ThrottlerCheck) Check(appName string, storeType string, storeName s } } if metricResultFunc == nil { - return NoSuchMetricCheckResult + if checkResult == nil { + return NoSuchMetricCheckResult + } + } else { + checkResult = check.checkAppMetricResult(appName, storeType, storeName, metricResultFunc, flags) } - checkResult = check.checkAppMetricResult(appName, storeType, storeName, metricResultFunc, flags) - go func(statusCode int) { metrics.GetOrRegisterCounter("check.any.total", nil).Inc(1) metrics.GetOrRegisterCounter(fmt.Sprintf("check.%s.total", appName), nil).Inc(1) diff --git a/pkg/throttle/throttler.go b/pkg/throttle/throttler.go index e0d103c..f432132 100644 --- a/pkg/throttle/throttler.go +++ b/pkg/throttle/throttler.go @@ -2,6 +2,7 @@ package throttle import ( "encoding/json" + "errors" "fmt" "io/ioutil" "math/rand" @@ -75,11 +76,18 @@ type Throttler struct { throttledAppsMutex sync.Mutex skippedHostsMutex sync.Mutex + mysqlRecoveryMutex sync.Mutex + mysqlRecoveryState map[string]*mysqlClusterRecoveryState nonLowPriorityAppRequestsThrottled *cache.Cache httpClient *http.Client } +type mysqlClusterRecoveryState struct { + throttled bool + healthySince time.Time +} + func NewThrottler() *Throttler { throttler := &Throttler{ isLeader: false, @@ -98,6 +106,7 @@ func NewThrottler() *Throttler { recentApps: cache.New(recentAppsExpiration, time.Minute), metricsHealth: cache.New(cache.NoExpiration, 0), shareDomainMetricHealth: cache.New(5*sharedDomainCollectInterval, sharedDomainCollectInterval), + mysqlRecoveryState: make(map[string]*mysqlClusterRecoveryState), nonLowPriorityAppRequestsThrottled: cache.New(nonDeprioritizedAppMapExpiration, nonDeprioritizedAppMapInterval), @@ -442,6 +451,7 @@ func (throttler *Throttler) aggregateMySQLMetrics() error { ignoreHostsCount := throttler.mysqlInventory.IgnoreHostsCount[clusterName] ignoreHostsThreshold := throttler.mysqlInventory.IgnoreHostsThreshold[clusterName] aggregatedMetric := aggregateMySQLProbes(probes, clusterName, throttler.mysqlInventory.InstanceKeyMetrics, throttler.mysqlInventory.ClusterInstanceHttpChecks, ignoreHostsCount, config.Settings().Stores.MySQL.IgnoreDialTcpErrors, ignoreHostsThreshold) + aggregatedMetric = throttler.stabilizeMySQLMetric(clusterName, aggregatedMetric, time.Now()) go throttler.aggregatedMetrics.Set(metricName, aggregatedMetric, cache.DefaultExpiration) if throttler.memcacheClient != nil { go func() { @@ -460,6 +470,58 @@ func (throttler *Throttler) aggregateMySQLMetrics() error { return nil } +func (throttler *Throttler) stabilizeMySQLMetric(clusterName string, metricResult base.MetricResult, now time.Time) base.MetricResult { + clusterSettings, ok := config.Settings().Stores.MySQL.Clusters[clusterName] + if !ok || clusterSettings.RecoveryDurationMillis <= 0 { + return metricResult + } + + value, err := metricResult.Get() + if errors.Is(err, base.NoHostsError) && !clusterSettings.FailOnNoHosts { + return metricResult + } + + throttler.mysqlRecoveryMutex.Lock() + defer throttler.mysqlRecoveryMutex.Unlock() + + state, ok := throttler.mysqlRecoveryState[clusterName] + if !ok { + state = &mysqlClusterRecoveryState{} + throttler.mysqlRecoveryState[clusterName] = state + } + + if err != nil || value > clusterSettings.ThrottleThreshold { + state.throttled = true + state.healthySince = time.Time{} + return metricResult + } + if !state.throttled { + return metricResult + } + + recoveryThreshold := clusterSettings.ThrottleThreshold + if clusterSettings.RecoveryThreshold != nil { + recoveryThreshold = *clusterSettings.RecoveryThreshold + } + if value > recoveryThreshold { + state.healthySince = time.Time{} + return base.NewErrorMetricResult(value, base.RecoveryNotCompleteError) + } + if state.healthySince.IsZero() { + state.healthySince = now + return base.NewErrorMetricResult(value, base.RecoveryNotCompleteError) + } + + recoveryDuration := time.Duration(clusterSettings.RecoveryDurationMillis) * time.Millisecond + if now.Sub(state.healthySince) < recoveryDuration { + return base.NewErrorMetricResult(value, base.RecoveryNotCompleteError) + } + + state.throttled = false + state.healthySince = time.Time{} + return metricResult +} + func (throttler *Throttler) pushStatusToExpVar() { metrics.DefaultRegistry.Each(func(metricName string, _ interface{}) { if strings.HasPrefix(metricName, "throttled_states.") { diff --git a/pkg/throttle/throttler_test.go b/pkg/throttle/throttler_test.go index 49826c8..3249323 100644 --- a/pkg/throttle/throttler_test.go +++ b/pkg/throttle/throttler_test.go @@ -43,6 +43,34 @@ func Test_checkAppMetricResult(t *testing.T) { assert.EqualError(t, result.Error, base.NoHostsError.Error()) assert.Equal(t, http.StatusOK, result.StatusCode) }) + t.Run("no hosts fail closed", func(t *testing.T) { + originalSettings := config.Settings().Stores.MySQL + defer func() { + config.Settings().Stores.MySQL = originalSettings + }() + config.Settings().Stores.MySQL = config.MySQLConfigurationSettings{ + Clusters: map[string]*config.MySQLClusterConfigurationSettings{ + "test-cluster": {FailOnNoHosts: true}, + }, + } + + check := NewThrottlerCheck(throttler) + metricResultFunc := func() (metricResult base.MetricResult, threshold float64) { + return base.NoHostsMetricResult, 10.0 + } + result := check.checkAppMetricResult("test-app", "mysql", "test-cluster", metricResultFunc, &CheckFlags{}) + assert.EqualError(t, result.Error, base.NoHostsError.Error()) + assert.Equal(t, http.StatusInternalServerError, result.StatusCode) + }) + t.Run("recovery period throttles", func(t *testing.T) { + check := NewThrottlerCheck(throttler) + metricResultFunc := func() (metricResult base.MetricResult, threshold float64) { + return base.NewErrorMetricResult(5.0, base.RecoveryNotCompleteError), 10.0 + } + result := check.checkAppMetricResult("test-app", "mysql", "test-cluster", metricResultFunc, &CheckFlags{}) + assert.EqualError(t, result.Error, base.RecoveryNotCompleteError.Error()) + assert.Equal(t, http.StatusTooManyRequests, result.StatusCode) + }) t.Run("low priority denied", func(t *testing.T) { check := NewThrottlerCheck(throttler) throttler.nonLowPriorityAppRequestsThrottled.SetDefault("mysql/test-cluster", true) @@ -179,3 +207,133 @@ func TestCheckMySQLFallbackUsesCanonicalState(t *testing.T) { }) } } + +func TestCheckMySQLRequiredClusters(t *testing.T) { + originalSettings := config.Settings().Stores.MySQL + defer func() { + config.Settings().Stores.MySQL = originalSettings + }() + + config.Settings().Stores.MySQL = config.MySQLConfigurationSettings{ + Clusters: map[string]*config.MySQLClusterConfigurationSettings{ + "replicas": { + RequiredClusters: []string{"primary", "proxysql"}, + }, + "primary": {}, + "proxysql": {}, + }, + } + + newCheck := func() (*ThrottlerCheck, *Throttler) { + throttler := NewThrottler() + setTestMySQLClusterMetric(throttler, "replicas", 1.0, 5.0) + setTestMySQLClusterMetric(throttler, "primary", 10.0, 50.0) + setTestMySQLClusterMetric(throttler, "proxysql", 20.0, 80.0) + return NewThrottlerCheck(throttler), throttler + } + + t.Run("all required clusters healthy", func(t *testing.T) { + check, _ := newCheck() + result := check.Check("transitions", "mysql", "replicas", "", &CheckFlags{}) + assert.Equal(t, http.StatusOK, result.StatusCode) + assert.Equal(t, 1.0, result.Value) + assert.Equal(t, 5.0, result.Threshold) + }) + + t.Run("primary cluster blocks admission", func(t *testing.T) { + check, throttler := newCheck() + setTestMySQLClusterMetric(throttler, "primary", 60.0, 50.0) + result := check.Check("transitions", "mysql", "replicas", "", &CheckFlags{}) + assert.Equal(t, http.StatusTooManyRequests, result.StatusCode) + assert.Equal(t, 60.0, result.Value) + assert.Equal(t, 50.0, result.Threshold) + }) + + t.Run("ProxySQL cluster blocks admission", func(t *testing.T) { + check, throttler := newCheck() + setTestMySQLClusterMetric(throttler, "proxysql", 90.0, 80.0) + result := check.Check("transitions", "mysql", "replicas", "", &CheckFlags{}) + assert.Equal(t, http.StatusTooManyRequests, result.StatusCode) + assert.Equal(t, 90.0, result.Value) + assert.Equal(t, 80.0, result.Threshold) + }) + + t.Run("read threshold override only applies to requested cluster", func(t *testing.T) { + check, throttler := newCheck() + setTestMySQLClusterMetric(throttler, "primary", 60.0, 50.0) + result := check.Check("transitions", "mysql", "replicas", "", &CheckFlags{ + ReadCheck: true, + OverrideThreshold: 100.0, + }) + assert.Equal(t, http.StatusTooManyRequests, result.StatusCode) + assert.Equal(t, 50.0, result.Threshold) + }) + + t.Run("missing required metric fails closed", func(t *testing.T) { + check, throttler := newCheck() + throttler.aggregatedMetrics.Delete("mysql/primary") + result := check.Check("transitions", "mysql", "replicas", "", &CheckFlags{OKIfNotExists: true}) + assert.Equal(t, http.StatusInternalServerError, result.StatusCode) + assert.EqualError(t, result.Error, `required MySQL cluster metric "primary" is unavailable`) + }) +} + +func TestStabilizeMySQLMetric(t *testing.T) { + originalSettings := config.Settings().Stores.MySQL + defer func() { + config.Settings().Stores.MySQL = originalSettings + }() + + recoveryThreshold := 40.0 + config.Settings().Stores.MySQL = config.MySQLConfigurationSettings{ + Clusters: map[string]*config.MySQLClusterConfigurationSettings{ + "primary": { + ThrottleThreshold: 50.0, + RecoveryThreshold: &recoveryThreshold, + RecoveryDurationMillis: 5000, + }, + }, + } + + throttler := NewThrottler() + start := time.Date(2026, time.September, 23, 12, 0, 0, 0, time.UTC) + + assertMetric := func(t *testing.T, result base.MetricResult, wantValue float64, wantErr error) { + t.Helper() + value, err := result.Get() + assert.Equal(t, wantValue, value) + assert.Equal(t, wantErr, err) + } + + assertMetric(t, throttler.stabilizeMySQLMetric("primary", base.NewSimpleMetricResult(60), start), 60, nil) + assertMetric(t, throttler.stabilizeMySQLMetric("primary", base.NewSimpleMetricResult(45), start.Add(time.Second)), 45, base.RecoveryNotCompleteError) + assertMetric(t, throttler.stabilizeMySQLMetric("primary", base.NewSimpleMetricResult(35), start.Add(2*time.Second)), 35, base.RecoveryNotCompleteError) + assertMetric(t, throttler.stabilizeMySQLMetric("primary", base.NewSimpleMetricResult(35), start.Add(6*time.Second)), 35, base.RecoveryNotCompleteError) + assertMetric(t, throttler.stabilizeMySQLMetric("primary", base.NewSimpleMetricResult(35), start.Add(7*time.Second)), 35, nil) +} + +func TestStabilizeMySQLMetricResetsRecoveryWindow(t *testing.T) { + originalSettings := config.Settings().Stores.MySQL + defer func() { + config.Settings().Stores.MySQL = originalSettings + }() + + config.Settings().Stores.MySQL = config.MySQLConfigurationSettings{ + Clusters: map[string]*config.MySQLClusterConfigurationSettings{ + "primary": { + ThrottleThreshold: 50.0, + RecoveryDurationMillis: 5000, + }, + }, + } + + throttler := NewThrottler() + start := time.Date(2026, time.September, 23, 12, 0, 0, 0, time.UTC) + + throttler.stabilizeMySQLMetric("primary", base.NewSimpleMetricResult(60), start) + throttler.stabilizeMySQLMetric("primary", base.NewSimpleMetricResult(40), start.Add(time.Second)) + throttler.stabilizeMySQLMetric("primary", base.NewSimpleMetricResult(55), start.Add(3*time.Second)) + result := throttler.stabilizeMySQLMetric("primary", base.NewSimpleMetricResult(40), start.Add(7*time.Second)) + _, err := result.Get() + assert.Equal(t, base.RecoveryNotCompleteError, err) +} From c4e653ecc12f7948ebac4bc4ee0ff30d14d60251 Mon Sep 17 00:00:00 2001 From: Pradeep Kintali Date: Wed, 23 Sep 2026 16:08:32 -0700 Subject: [PATCH 2/9] Harden primary protection recovery Require a full healthy window after process or leader startup, expose the blocking metric in check responses, and publish recovery latch state. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 0fbc396f-32f9-449a-8537-6264e8665c31 --- doc/mysql.md | 2 ++ pkg/throttle/check.go | 8 ++++++-- pkg/throttle/check_result.go | 1 + pkg/throttle/throttler.go | 11 ++++++++++- pkg/throttle/throttler_test.go | 31 +++++++++++++++++++++++++++++++ 5 files changed, 50 insertions(+), 3 deletions(-) diff --git a/doc/mysql.md b/doc/mysql.md index 3aaebca..ef335dc 100644 --- a/doc/mysql.md +++ b/doc/mysql.md @@ -203,7 +203,9 @@ The ProxySQL query above is illustrative: deployments must provide a scalar quer - `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. +- A new process or leader starts configured recovery metrics in the recovering state. It must observe a complete healthy window before admitting work. - 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. diff --git a/pkg/throttle/check.go b/pkg/throttle/check.go index ea5cc88..c47efb9 100644 --- a/pkg/throttle/check.go +++ b/pkg/throttle/check.go @@ -103,7 +103,9 @@ func (check *ThrottlerCheck) checkAppMetricResult(appName string, storeType stri // all good! statusCode = http.StatusOK // 200 } - return NewCheckResult(statusCode, value, threshold, err) + checkResult = NewCheckResult(statusCode, value, threshold, err) + checkResult.MetricName = metricName + return checkResult } func (check *ThrottlerCheck) checkMySQLCluster( @@ -134,10 +136,12 @@ func (check *ThrottlerCheck) checkMySQLCluster( continue } if requiredResult.StatusCode == http.StatusNotFound { - return NewErrorCheckResult( + checkResult := NewErrorCheckResult( http.StatusInternalServerError, fmt.Errorf("required MySQL cluster metric %q is unavailable", requiredCluster), ) + checkResult.MetricName = fmt.Sprintf("mysql/%s", requiredCluster) + return checkResult } return requiredResult } diff --git a/pkg/throttle/check_result.go b/pkg/throttle/check_result.go index 1a288fc..4e60bc1 100644 --- a/pkg/throttle/check_result.go +++ b/pkg/throttle/check_result.go @@ -11,6 +11,7 @@ type CheckResult struct { StatusCode int `json:"StatusCode"` Value float64 `json:"Value"` Threshold float64 `json:"Threshold"` + MetricName string `json:"MetricName,omitempty"` Error error `json:"-"` Message string `json:"Message"` } diff --git a/pkg/throttle/throttler.go b/pkg/throttle/throttler.go index f432132..7b51e97 100644 --- a/pkg/throttle/throttler.go +++ b/pkg/throttle/throttler.go @@ -486,16 +486,21 @@ func (throttler *Throttler) stabilizeMySQLMetric(clusterName string, metricResul state, ok := throttler.mysqlRecoveryState[clusterName] if !ok { - state = &mysqlClusterRecoveryState{} + // A new process or leader has no recovery history. Require a complete + // healthy window before admitting work rather than assuming the metric + // was healthy before this process began observing it. + state = &mysqlClusterRecoveryState{throttled: true} throttler.mysqlRecoveryState[clusterName] = state } if err != nil || value > clusterSettings.ThrottleThreshold { state.throttled = true state.healthySince = time.Time{} + metrics.GetOrRegisterGauge(fmt.Sprintf("recovery.mysql.%s.active", clusterName), nil).Update(1) return metricResult } if !state.throttled { + metrics.GetOrRegisterGauge(fmt.Sprintf("recovery.mysql.%s.active", clusterName), nil).Update(0) return metricResult } @@ -505,20 +510,24 @@ func (throttler *Throttler) stabilizeMySQLMetric(clusterName string, metricResul } if value > recoveryThreshold { state.healthySince = time.Time{} + metrics.GetOrRegisterGauge(fmt.Sprintf("recovery.mysql.%s.active", clusterName), nil).Update(1) return base.NewErrorMetricResult(value, base.RecoveryNotCompleteError) } if state.healthySince.IsZero() { state.healthySince = now + metrics.GetOrRegisterGauge(fmt.Sprintf("recovery.mysql.%s.active", clusterName), nil).Update(1) return base.NewErrorMetricResult(value, base.RecoveryNotCompleteError) } recoveryDuration := time.Duration(clusterSettings.RecoveryDurationMillis) * time.Millisecond if now.Sub(state.healthySince) < recoveryDuration { + metrics.GetOrRegisterGauge(fmt.Sprintf("recovery.mysql.%s.active", clusterName), nil).Update(1) return base.NewErrorMetricResult(value, base.RecoveryNotCompleteError) } state.throttled = false state.healthySince = time.Time{} + metrics.GetOrRegisterGauge(fmt.Sprintf("recovery.mysql.%s.active", clusterName), nil).Update(0) return metricResult } diff --git a/pkg/throttle/throttler_test.go b/pkg/throttle/throttler_test.go index 3249323..0695383 100644 --- a/pkg/throttle/throttler_test.go +++ b/pkg/throttle/throttler_test.go @@ -7,6 +7,7 @@ import ( "github.com/github/freno/pkg/base" "github.com/github/freno/pkg/config" + metrics "github.com/rcrowley/go-metrics" "github.com/stretchr/testify/assert" ) @@ -247,6 +248,7 @@ func TestCheckMySQLRequiredClusters(t *testing.T) { assert.Equal(t, http.StatusTooManyRequests, result.StatusCode) assert.Equal(t, 60.0, result.Value) assert.Equal(t, 50.0, result.Threshold) + assert.Equal(t, "mysql/primary", result.MetricName) }) t.Run("ProxySQL cluster blocks admission", func(t *testing.T) { @@ -275,6 +277,7 @@ func TestCheckMySQLRequiredClusters(t *testing.T) { result := check.Check("transitions", "mysql", "replicas", "", &CheckFlags{OKIfNotExists: true}) assert.Equal(t, http.StatusInternalServerError, result.StatusCode) assert.EqualError(t, result.Error, `required MySQL cluster metric "primary" is unavailable`) + assert.Equal(t, "mysql/primary", result.MetricName) }) } @@ -337,3 +340,31 @@ func TestStabilizeMySQLMetricResetsRecoveryWindow(t *testing.T) { _, err := result.Get() assert.Equal(t, base.RecoveryNotCompleteError, err) } + +func TestStabilizeMySQLMetricRequiresHealthyWindowAfterStartup(t *testing.T) { + originalSettings := config.Settings().Stores.MySQL + defer func() { + config.Settings().Stores.MySQL = originalSettings + }() + + config.Settings().Stores.MySQL = config.MySQLConfigurationSettings{ + Clusters: map[string]*config.MySQLClusterConfigurationSettings{ + "primary": { + ThrottleThreshold: 50.0, + RecoveryDurationMillis: 5000, + }, + }, + } + + throttler := NewThrottler() + start := time.Date(2026, time.September, 23, 12, 0, 0, 0, time.UTC) + + result := throttler.stabilizeMySQLMetric("primary", base.NewSimpleMetricResult(30), start) + _, err := result.Get() + assert.Equal(t, base.RecoveryNotCompleteError, err) + + result = throttler.stabilizeMySQLMetric("primary", base.NewSimpleMetricResult(30), start.Add(5*time.Second)) + _, err = result.Get() + assert.Nil(t, err) + assert.Equal(t, int64(0), metrics.Get("recovery.mysql.primary.active").(metrics.Gauge).Value()) +} From a6a32fc276ca777f57bd5d011fa75d28735bd4f8 Mon Sep 17 00:00:00 2001 From: Pradeep Kintali Date: Wed, 23 Sep 2026 17:32:15 -0700 Subject: [PATCH 3/9] Support primary Vitess tablet discovery Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 0fbc396f-32f9-449a-8537-6264e8665c31 --- doc/mysql.md | 4 +++- pkg/config/config_test.go | 44 +++++++++++++++++++++++++++++++++++ pkg/config/mysql_config.go | 5 ++++ pkg/config/vitess_config.go | 18 ++++++++++++++ pkg/vitess/api_client.go | 29 +++++++++++++++-------- pkg/vitess/api_client_test.go | 19 +++++++++++++++ 6 files changed, 108 insertions(+), 11 deletions(-) diff --git a/doc/mysql.md b/doc/mysql.md index ef335dc..0ac8325 100644 --- a/doc/mysql.md +++ b/doc/mysql.md @@ -96,7 +96,8 @@ Looking at clusters configuration: ], "VitessSettings": { "API": "https://vtctld.example.com/api/", - "Keyspace": "my_sharded_ks" + "Keyspace": "my_sharded_ks", + "TabletType": "REPLICA" } }, "local": { @@ -120,6 +121,7 @@ Noteworthy: - `prod4` chooses to (but doesn't have to) override the `ThrottleThreshold` to `0.8` seconds - `prod4` list of servers is dictated by `HAProxy`. `freno` will routinely and dynamically poll given HAProxy server for list of hosts. These will include any hosts not in `NOLB`. +- `sharded` discovers `REPLICA` tablets by default. Set `TabletType` to `MASTER` (or `PRIMARY`) for a separately configured primary safety metric. - `local` cluster chooses to override `User`, `Password` and `IgnoreHostsCount`. - `local` cluster defines a static list of hosts. diff --git a/pkg/config/config_test.go b/pkg/config/config_test.go index 92ce27d..41c1ceb 100644 --- a/pkg/config/config_test.go +++ b/pkg/config/config_test.go @@ -203,6 +203,50 @@ func TestMySQLRecoveryConfiguration(t *testing.T) { } } +func TestVitessTabletTypeConfiguration(t *testing.T) { + tests := []struct { + name string + tabletType string + want string + wantErr string + }{ + {name: "default replica", want: "REPLICA"}, + {name: "replica", tabletType: "replica", want: "REPLICA"}, + {name: "master", tabletType: "master", want: "MASTER"}, + {name: "primary alias", tabletType: "primary", want: "MASTER"}, + {name: "invalid", tabletType: "rdonly", wantErr: `Stores.MySQL.Clusters.vitess.VitessSettings: unsupported Vitess tablet type "rdonly"`}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + settings := MySQLConfigurationSettings{ + Clusters: map[string]*MySQLClusterConfigurationSettings{ + "vitess": { + VitessSettings: VitessConfigurationSettings{ + API: "https://vtctld.example.com/api", + Keyspace: "test", + TabletType: test.tabletType, + }, + }, + }, + } + err := settings.postReadAdjustments() + if test.wantErr == "" && err != nil { + t.Fatalf("unexpected error: %v", err) + } + if test.wantErr != "" { + if err == nil || err.Error() != test.wantErr { + t.Fatalf("error = %v, want %q", err, test.wantErr) + } + return + } + if got := settings.Clusters["vitess"].VitessSettings.TabletType; got != test.want { + t.Fatalf("TabletType = %q, want %q", got, test.want) + } + }) + } +} + func dump(path string, contents *ConfigurationSettings) error { json, _ := json.Marshal(contents) err := ioutil.WriteFile(path, json, 0644) diff --git a/pkg/config/mysql_config.go b/pkg/config/mysql_config.go index 3944a65..c434aa5 100644 --- a/pkg/config/mysql_config.go +++ b/pkg/config/mysql_config.go @@ -112,6 +112,11 @@ func (settings *MySQLConfigurationSettings) postReadAdjustments() error { if err := clusterSettings.postReadAdjustments(); err != nil { return err } + if !clusterSettings.VitessSettings.IsEmpty() { + if err := clusterSettings.VitessSettings.postReadAdjustments(); err != nil { + return fmt.Errorf("Stores.MySQL.Clusters.%s.VitessSettings: %w", clusterName, err) + } + } if clusterSettings.User == "" { clusterSettings.User = settings.User } diff --git a/pkg/config/vitess_config.go b/pkg/config/vitess_config.go index 79de29e..5468d1d 100644 --- a/pkg/config/vitess_config.go +++ b/pkg/config/vitess_config.go @@ -1,5 +1,10 @@ package config +import ( + "fmt" + "strings" +) + // // HAProxy-specific configuration // @@ -9,6 +14,7 @@ type VitessConfigurationSettings struct { Cells []string Keyspace string Shard string + TabletType string TimeoutSecs uint } @@ -21,3 +27,15 @@ func (settings *VitessConfigurationSettings) IsEmpty() bool { } return false } + +func (settings *VitessConfigurationSettings) postReadAdjustments() error { + switch strings.ToUpper(strings.TrimSpace(settings.TabletType)) { + case "", "REPLICA": + settings.TabletType = "REPLICA" + case "MASTER", "PRIMARY": + settings.TabletType = "MASTER" + default: + return fmt.Errorf("unsupported Vitess tablet type %q", settings.TabletType) + } + return nil +} diff --git a/pkg/vitess/api_client.go b/pkg/vitess/api_client.go index 87cc66c..294315e 100644 --- a/pkg/vitess/api_client.go +++ b/pkg/vitess/api_client.go @@ -61,9 +61,10 @@ func (t Tablet) IsServeable() bool { return true } -// IsValidReplica returns a bool reflecting if a tablet type is REPLICA -func (t Tablet) IsValidReplica() bool { - if t.Type != topodata.TabletType_REPLICA { +// IsValidType returns whether a tablet has the requested type and is eligible +// for direct metric collection. +func (t Tablet) IsValidType(tabletType topodata.TabletType) bool { + if t.Type != tabletType { return false } return t.IsServeable() @@ -101,19 +102,27 @@ func ParseCells(settings config.VitessConfigurationSettings) (cells []string) { return cells } -// filterReplicaTablets parses a list of tablets, returning replica tablets only -func filterReplicaTablets(settings config.VitessConfigurationSettings, tablets []Tablet) (replicas []Tablet) { +func configuredTabletType(settings config.VitessConfigurationSettings) topodata.TabletType { + if settings.TabletType == "MASTER" { + return topodata.TabletType_MASTER + } + return topodata.TabletType_REPLICA +} + +// filterTablets parses a list of tablets, returning only the configured type. +func filterTablets(settings config.VitessConfigurationSettings, tablets []Tablet) (filtered []Tablet) { validCells := ParseCells(settings) + tabletType := configuredTabletType(settings) for _, tablet := range tablets { - if tablet.HasValidCell(validCells) && tablet.IsValidReplica() { - replicas = append(replicas, tablet) + if tablet.HasValidCell(validCells) && tablet.IsValidType(tabletType) { + filtered = append(filtered, tablet) } } - return replicas + return filtered } // ParseTablets reads from vitess /api/keyspace//tablets/[shard] and returns a -// listing (mysql_hostname, mysql_port, type) of REPLICA tablets +// listing (mysql_hostname, mysql_port, type) of the configured tablet type. func ParseTablets(settings config.VitessConfigurationSettings) (tablets []Tablet, err error) { if settings.TimeoutSecs == 0 { httpClient.Timeout = defaultTimeout @@ -137,5 +146,5 @@ func ParseTablets(settings config.VitessConfigurationSettings) (tablets []Tablet return tablets, err } err = json.Unmarshal(body, &tablets) - return filterReplicaTablets(settings, tablets), err + return filterTablets(settings, tablets), err } diff --git a/pkg/vitess/api_client_test.go b/pkg/vitess/api_client_test.go index 97a753d..8a7bbef 100644 --- a/pkg/vitess/api_client_test.go +++ b/pkg/vitess/api_client_test.go @@ -141,6 +141,25 @@ func TestParseTablets(t *testing.T) { } }) + t.Run("primary", func(t *testing.T) { + tablets, err := ParseTablets(config.VitessConfigurationSettings{ + API: vitessApi.URL, + Keyspace: "test", + Shard: "00", + TabletType: "MASTER", + }) + if err != nil { + t.Fatalf("Expected no error, got %q", err) + } + + if len(tablets) != 1 { + t.Fatalf("Expected 1 tablet, got %d", len(tablets)) + } + if tablets[0].Type != topodata.TabletType_MASTER { + t.Fatalf("Expected master tablet, got %s", tablets[0].Type.String()) + } + }) + t.Run("not-found", func(t *testing.T) { tablets, err := ParseTablets(config.VitessConfigurationSettings{ API: vitessApi.URL, From e0392d8e21077f5cc00b9ef71b14fe0fe73f10da Mon Sep 17 00:00:00 2001 From: Pradeep Kintali Date: Mon, 28 Sep 2026 14:49:24 -0700 Subject: [PATCH 4/9] Address primary protection review feedback Preserve the replica validation API, apply read overrides consistently, reset recovery state on leadership regain, and enforce required clusters for direct memcache consumers. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 0fbc396f-32f9-449a-8537-6264e8665c31 --- doc/memcache.md | 4 +++ pkg/throttle/check.go | 4 +++ pkg/throttle/throttler.go | 54 +++++++++++++++++++++++++--- pkg/throttle/throttler_test.go | 66 ++++++++++++++++++++++++++++++++++ pkg/vitess/api_client.go | 5 +++ pkg/vitess/api_client_test.go | 12 +++++++ 6 files changed, 140 insertions(+), 5 deletions(-) diff --git a/doc/memcache.md b/doc/memcache.md index 7cbe91e..f269637 100644 --- a/doc/memcache.md +++ b/doc/memcache.md @@ -37,6 +37,10 @@ Optionally set `MemcachePath` (default is `"freno"`): - The value will be of the form `:`. - As example, it might be `1497418678836:0.54` where `1497418678836` is the unix epoch in milliseconds, and `0.54` is the aggregated value. - Embedding the epoch within the value allows the app to double-check the validity of the value, or go into more granular validation. +- For a MySQL cluster with `RequiredClusters`, the key is removed while any + required metric is unavailable, recovering, or above its configured + threshold. Direct memcache clients therefore observe the same composite + safety gate as HTTP clients. ### Runtime access to memcache configuration diff --git a/pkg/throttle/check.go b/pkg/throttle/check.go index c47efb9..ff0fb3b 100644 --- a/pkg/throttle/check.go +++ b/pkg/throttle/check.go @@ -54,6 +54,9 @@ func (check *ThrottlerCheck) checkAppMetricResult(appName string, storeType stri threshold = flags.OverrideThreshold } value, err := metricResult.Get() + if flags.ReadCheck && flags.OverrideThreshold > 0 && errors.Is(err, base.RecoveryNotCompleteError) && value <= threshold { + err = nil + } if appName == "" { return NewCheckResult(http.StatusExpectationFailed, value, threshold, fmt.Errorf("no app indicated")) } @@ -130,6 +133,7 @@ func (check *ThrottlerCheck) checkMySQLCluster( clusterSettings := config.Settings().Stores.MySQL.Clusters[clusterName] for _, requiredCluster := range clusterSettings.RequiredClusters { requiredFlags := *flags + requiredFlags.ReadCheck = false requiredFlags.OverrideThreshold = 0 requiredResult := check.checkMySQLCluster(appName, requiredCluster, &requiredFlags, checkedClusters) if requiredResult == nil || requiredResult.StatusCode == http.StatusOK { diff --git a/pkg/throttle/throttler.go b/pkg/throttle/throttler.go index 7b51e97..ddf9759 100644 --- a/pkg/throttle/throttler.go +++ b/pkg/throttle/throttler.go @@ -155,7 +155,7 @@ func (throttler *Throttler) Operate() { case <-leaderCheckTick: { // sparse - throttler.isLeader = throttler.isLeaderFunc() + throttler.updateLeadership(throttler.isLeaderFunc()) } case <-mysqlCollectTick: { @@ -446,16 +446,24 @@ func (throttler *Throttler) aggregateMySQLMetrics() error { if !throttler.isLeader { return nil } + aggregatedMetrics := make(map[string]base.MetricResult, len(throttler.mysqlInventory.ClustersProbes)) for clusterName, probes := range throttler.mysqlInventory.ClustersProbes { - metricName := fmt.Sprintf("mysql/%s", clusterName) ignoreHostsCount := throttler.mysqlInventory.IgnoreHostsCount[clusterName] ignoreHostsThreshold := throttler.mysqlInventory.IgnoreHostsThreshold[clusterName] aggregatedMetric := aggregateMySQLProbes(probes, clusterName, throttler.mysqlInventory.InstanceKeyMetrics, throttler.mysqlInventory.ClusterInstanceHttpChecks, ignoreHostsCount, config.Settings().Stores.MySQL.IgnoreDialTcpErrors, ignoreHostsThreshold) - aggregatedMetric = throttler.stabilizeMySQLMetric(clusterName, aggregatedMetric, time.Now()) + aggregatedMetrics[clusterName] = throttler.stabilizeMySQLMetric(clusterName, aggregatedMetric, time.Now()) + } + for clusterName, aggregatedMetric := range aggregatedMetrics { + metricName := fmt.Sprintf("mysql/%s", clusterName) go throttler.aggregatedMetrics.Set(metricName, aggregatedMetric, cache.DefaultExpiration) if throttler.memcacheClient != nil { - go func() { + requiredClustersHealthy := throttler.requiredMySQLClustersHealthy(clusterName, aggregatedMetrics, make(map[string]bool)) + go func(metricName string, aggregatedMetric base.MetricResult, requiredClustersHealthy bool) { memcacheKey := fmt.Sprintf("%s/%s", throttler.memcachePath, metricName) + if !requiredClustersHealthy { + throttler.memcacheClient.Delete(memcacheKey) + return + } value, err := aggregatedMetric.Get() if err != nil { throttler.memcacheClient.Delete(memcacheKey) @@ -464,12 +472,48 @@ func (throttler *Throttler) aggregateMySQLMetrics() error { entryVal := fmt.Sprintf("%d:%.6f", epochMillis, value) throttler.memcacheClient.Set(&memcache.Item{Key: memcacheKey, Value: []byte(entryVal), Expiration: 1}) } - }() + }(metricName, aggregatedMetric, requiredClustersHealthy) } } return nil } +func (throttler *Throttler) requiredMySQLClustersHealthy(clusterName string, aggregatedMetrics map[string]base.MetricResult, checkedClusters map[string]bool) bool { + if checkedClusters[clusterName] { + return true + } + checkedClusters[clusterName] = true + + clusterSettings, ok := config.Settings().Stores.MySQL.Clusters[clusterName] + if !ok { + return false + } + for _, requiredCluster := range clusterSettings.RequiredClusters { + requiredSettings, configured := config.Settings().Stores.MySQL.Clusters[requiredCluster] + requiredMetric, collected := aggregatedMetrics[requiredCluster] + if !configured || !collected { + return false + } + value, err := requiredMetric.Get() + if err != nil || value > requiredSettings.ThrottleThreshold { + return false + } + if !throttler.requiredMySQLClustersHealthy(requiredCluster, aggregatedMetrics, checkedClusters) { + return false + } + } + return true +} + +func (throttler *Throttler) updateLeadership(isLeader bool) { + if isLeader && !throttler.isLeader { + throttler.mysqlRecoveryMutex.Lock() + throttler.mysqlRecoveryState = make(map[string]*mysqlClusterRecoveryState) + throttler.mysqlRecoveryMutex.Unlock() + } + throttler.isLeader = isLeader +} + func (throttler *Throttler) stabilizeMySQLMetric(clusterName string, metricResult base.MetricResult, now time.Time) base.MetricResult { clusterSettings, ok := config.Settings().Stores.MySQL.Clusters[clusterName] if !ok || clusterSettings.RecoveryDurationMillis <= 0 { diff --git a/pkg/throttle/throttler_test.go b/pkg/throttle/throttler_test.go index 0695383..5c902d2 100644 --- a/pkg/throttle/throttler_test.go +++ b/pkg/throttle/throttler_test.go @@ -7,6 +7,7 @@ import ( "github.com/github/freno/pkg/base" "github.com/github/freno/pkg/config" + "github.com/patrickmn/go-cache" metrics "github.com/rcrowley/go-metrics" "github.com/stretchr/testify/assert" ) @@ -271,6 +272,32 @@ func TestCheckMySQLRequiredClusters(t *testing.T) { assert.Equal(t, 50.0, result.Threshold) }) + t.Run("read threshold override bypasses only the requested cluster recovery latch", func(t *testing.T) { + check, throttler := newCheck() + throttler.aggregatedMetrics.Set( + "mysql/replicas", + base.NewErrorMetricResult(45.0, base.RecoveryNotCompleteError), + cache.DefaultExpiration, + ) + result := check.Check("transitions", "mysql", "replicas", "", &CheckFlags{ + ReadCheck: true, + OverrideThreshold: 100.0, + }) + assert.Equal(t, http.StatusOK, result.StatusCode) + + throttler.aggregatedMetrics.Set( + "mysql/primary", + base.NewErrorMetricResult(40.0, base.RecoveryNotCompleteError), + cache.DefaultExpiration, + ) + result = check.Check("transitions", "mysql", "replicas", "", &CheckFlags{ + ReadCheck: true, + OverrideThreshold: 100.0, + }) + assert.Equal(t, http.StatusTooManyRequests, result.StatusCode) + assert.Equal(t, "mysql/primary", result.MetricName) + }) + t.Run("missing required metric fails closed", func(t *testing.T) { check, throttler := newCheck() throttler.aggregatedMetrics.Delete("mysql/primary") @@ -368,3 +395,42 @@ func TestStabilizeMySQLMetricRequiresHealthyWindowAfterStartup(t *testing.T) { assert.Nil(t, err) assert.Equal(t, int64(0), metrics.Get("recovery.mysql.primary.active").(metrics.Gauge).Value()) } + +func TestUpdateLeadershipResetsRecoveryStateOnRegain(t *testing.T) { + throttler := NewThrottler() + throttler.isLeader = true + throttler.mysqlRecoveryState["primary"] = &mysqlClusterRecoveryState{throttled: false} + + throttler.updateLeadership(false) + assert.Len(t, throttler.mysqlRecoveryState, 1) + + throttler.updateLeadership(true) + assert.Empty(t, throttler.mysqlRecoveryState) +} + +func TestRequiredMySQLClustersHealthyForMemcache(t *testing.T) { + originalSettings := config.Settings().Stores.MySQL + defer func() { + config.Settings().Stores.MySQL = originalSettings + }() + + config.Settings().Stores.MySQL = config.MySQLConfigurationSettings{ + Clusters: map[string]*config.MySQLClusterConfigurationSettings{ + "replicas": {RequiredClusters: []string{"primary"}}, + "primary": {ThrottleThreshold: 50.0}, + }, + } + throttler := NewThrottler() + aggregatedMetrics := map[string]base.MetricResult{ + "replicas": base.NewSimpleMetricResult(1.0), + "primary": base.NewSimpleMetricResult(40.0), + } + + assert.True(t, throttler.requiredMySQLClustersHealthy("replicas", aggregatedMetrics, make(map[string]bool))) + + aggregatedMetrics["primary"] = base.NewSimpleMetricResult(60.0) + assert.False(t, throttler.requiredMySQLClustersHealthy("replicas", aggregatedMetrics, make(map[string]bool))) + + delete(aggregatedMetrics, "primary") + assert.False(t, throttler.requiredMySQLClustersHealthy("replicas", aggregatedMetrics, make(map[string]bool))) +} diff --git a/pkg/vitess/api_client.go b/pkg/vitess/api_client.go index 294315e..fb9c945 100644 --- a/pkg/vitess/api_client.go +++ b/pkg/vitess/api_client.go @@ -70,6 +70,11 @@ func (t Tablet) IsValidType(tabletType topodata.TabletType) bool { return t.IsServeable() } +// IsValidReplica preserves the original replica-specific API for package consumers. +func (t Tablet) IsValidReplica() bool { + return t.IsValidType(topodata.TabletType_REPLICA) +} + var httpClient = http.Client{ Timeout: defaultTimeout, } diff --git a/pkg/vitess/api_client_test.go b/pkg/vitess/api_client_test.go index 8a7bbef..f9d7fb9 100644 --- a/pkg/vitess/api_client_test.go +++ b/pkg/vitess/api_client_test.go @@ -11,6 +11,18 @@ import ( "vitess.io/vitess/go/vt/proto/topodata" ) +func TestTabletIsValidReplica(t *testing.T) { + replica := Tablet{Type: topodata.TabletType_REPLICA} + primary := Tablet{Type: topodata.TabletType_MASTER} + + if !replica.IsValidReplica() { + t.Fatal("expected replica tablet to remain valid through compatibility method") + } + if primary.IsValidReplica() { + t.Fatal("expected primary tablet not to pass replica validation") + } +} + func TestParseTablets(t *testing.T) { vitessApi := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { switch r.URL.String() { From 308f5be46510be0c926d0d690f5467e2240c7eaf Mon Sep 17 00:00:00 2001 From: Pradeep Kintali Date: Thu, 1 Oct 2026 09:14:00 -0700 Subject: [PATCH 5/9] Generalize composite check documentation Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 0fbc396f-32f9-449a-8537-6264e8665c31 --- doc/mysql.md | 19 +++---------------- 1 file changed, 3 insertions(+), 16 deletions(-) diff --git a/doc/mysql.md b/doc/mysql.md index 0ac8325..5ce79d9 100644 --- a/doc/mysql.md +++ b/doc/mysql.md @@ -211,19 +211,6 @@ The ProxySQL query above is illustrative: deployments must provide a scalar quer Configuration loading rejects missing `RequiredClusters` references and dependency cycles. -### A1 repair mapping - -| A1 repair item | Freno implementation in this change | Remaining work | -| --- | --- | --- | -| Default transitions to one worker | Not a Freno responsibility. | Enforce in the transitions framework. | -| Audit custom worker configurations | Not a Freno responsibility. | Audit and constrain custom transitions in the application. | -| Identify Freno traffic as transitions | Existing Freno API already accepts the application name and emits per-app/per-store counters. | Complete caller rollout with `app=transitions` and monitor that identity. | -| Mark transition writes as low priority | Existing `p=low` behavior remains compatible with composite checks. | Complete application rollout. | -| Page on primary and ProxySQL connection failures | `FailOnNoHosts` and missing-required-metric handling prevent an unobservable probe failure from allowing work. | Add Datadog paging for probe and connectivity failures. | -| Page on sustained primary `Threads_running` | A cached primary metric can now be required by the replica-lag decision; recovery hysteresis prevents immediate release. | Configure cluster thresholds and add the sustained Datadog monitor. | -| Page before connection exhaustion | A cached ProxySQL capacity metric can now be required by the admission decision. | Define the production capacity query, reserve threshold, and Datadog monitor. | -| Detect rising concurrency with falling completions | A derived scalar metric can be configured as another required cluster check. | Define and validate the correlation query and alert thresholds. | -| Alert on sustained Freno rejection | Composite safety failures return normal Freno rejection/error status and existing counters remain available by app and store. | Add the rejection-ratio and minimum-volume Datadog monitor grouped by cluster and `app=transitions`. | -| Provide a cluster-wide emergency pause | Existing app/store throttling remains the emergency control and is evaluated for each required metric. | Operationalize a deterministic transitions-plus-cluster command and runbook. | - -These changes provide admission enforcement; they do not create Datadog monitors, change transition worker defaults, or bound replica-to-primary fallback inside the application. +### Scope boundaries + +These settings provide reusable admission checks and recovery behavior. Deployments remain responsible for choosing appropriate metrics and thresholds, configuring their host-discovery sources, monitoring probe health and rejection rates, and controlling application concurrency or fallback behavior. From 9a340aaa03073829293fbb9bfee355b952c39646 Mon Sep 17 00:00:00 2001 From: Pradeep Kintali Date: Thu, 15 Oct 2026 10:00:54 -0700 Subject: [PATCH 6/9] Document staged primary safety rollout and recovery budgets Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: dea961fa-5766-415f-a0f5-2562c55a0385 --- doc/mysql.md | 42 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 42 insertions(+) diff --git a/doc/mysql.md b/doc/mysql.md index 5ce79d9..b479b01 100644 --- a/doc/mysql.md +++ b/doc/mysql.md @@ -211,6 +211,48 @@ The ProxySQL query above is illustrative: deployments must provide a scalar quer Configuration loading rejects missing `RequiredClusters` references and dependency cycles. +### 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. + +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. +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. Add a primary dependency to one reviewed cluster. 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. + ### Scope boundaries These settings provide reusable admission checks and recovery behavior. Deployments remain responsible for choosing appropriate metrics and thresholds, configuring their host-discovery sources, monitoring probe health and rejection rates, and controlling application concurrency or fallback behavior. From ea47750263763777c646b95601a7698b005f1c15 Mon Sep 17 00:00:00 2001 From: Pradeep Kintali Date: Thu, 15 Oct 2026 12:38:38 -0700 Subject: [PATCH 7/9] Use canonical PRIMARY for Vitess discovery Preserve MASTER as a deprecated input alias and the existing numeric tablet mapping. Clarify that shadow collection precedes enforcement and is not composite dry-run counting. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: fbc83df2-d1af-406f-b97b-51cf0aaa570d --- doc/mysql.md | 8 +++++-- pkg/config/config_test.go | 6 ++++-- pkg/config/vitess_config.go | 4 ++-- pkg/vitess/api_client.go | 7 ++++-- pkg/vitess/api_client_test.go | 40 ++++++++++++++++++----------------- 5 files changed, 38 insertions(+), 27 deletions(-) diff --git a/doc/mysql.md b/doc/mysql.md index b479b01..74263d6 100644 --- a/doc/mysql.md +++ b/doc/mysql.md @@ -121,7 +121,7 @@ Noteworthy: - `prod4` chooses to (but doesn't have to) override the `ThrottleThreshold` to `0.8` seconds - `prod4` list of servers is dictated by `HAProxy`. `freno` will routinely and dynamically poll given HAProxy server for list of hosts. These will include any hosts not in `NOLB`. -- `sharded` discovers `REPLICA` tablets by default. Set `TabletType` to `MASTER` (or `PRIMARY`) for a separately configured primary safety metric. +- `sharded` discovers `REPLICA` tablets by default. Set `TabletType` to `PRIMARY` for a separately configured primary safety metric. `MASTER` is accepted as a deprecated alias and normalized to `PRIMARY`. - `local` cluster chooses to override `User`, `Password` and `IgnoreHostsCount`. - `local` cluster defines a static list of hosts. @@ -224,11 +224,15 @@ a deployed server version. 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. 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. Add a primary dependency to one reviewed cluster. Exercise overload, probe +4. In a separate reviewed configuration change, add a primary dependency to 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. diff --git a/pkg/config/config_test.go b/pkg/config/config_test.go index 41c1ceb..20e1245 100644 --- a/pkg/config/config_test.go +++ b/pkg/config/config_test.go @@ -212,8 +212,10 @@ func TestVitessTabletTypeConfiguration(t *testing.T) { }{ {name: "default replica", want: "REPLICA"}, {name: "replica", tabletType: "replica", want: "REPLICA"}, - {name: "master", tabletType: "master", want: "MASTER"}, - {name: "primary alias", tabletType: "primary", want: "MASTER"}, + {name: "primary", tabletType: "primary", want: "PRIMARY"}, + {name: "legacy master alias", tabletType: "master", want: "PRIMARY"}, + {name: "primary with whitespace", tabletType: " PRIMARY ", want: "PRIMARY"}, + {name: "legacy alias with whitespace", tabletType: " MASTER ", want: "PRIMARY"}, {name: "invalid", tabletType: "rdonly", wantErr: `Stores.MySQL.Clusters.vitess.VitessSettings: unsupported Vitess tablet type "rdonly"`}, } diff --git a/pkg/config/vitess_config.go b/pkg/config/vitess_config.go index 5468d1d..2b61d9b 100644 --- a/pkg/config/vitess_config.go +++ b/pkg/config/vitess_config.go @@ -32,8 +32,8 @@ func (settings *VitessConfigurationSettings) postReadAdjustments() error { switch strings.ToUpper(strings.TrimSpace(settings.TabletType)) { case "", "REPLICA": settings.TabletType = "REPLICA" - case "MASTER", "PRIMARY": - settings.TabletType = "MASTER" + case "PRIMARY", "MASTER": + settings.TabletType = "PRIMARY" default: return fmt.Errorf("unsupported Vitess tablet type %q", settings.TabletType) } diff --git a/pkg/vitess/api_client.go b/pkg/vitess/api_client.go index fb9c945..ec404a1 100644 --- a/pkg/vitess/api_client.go +++ b/pkg/vitess/api_client.go @@ -15,6 +15,9 @@ import ( const defaultTimeout = time.Duration(5) * time.Second +// The pinned Vitess enum names value 1 MASTER; current Vitess calls it PRIMARY. +const tabletTypePrimary = topodata.TabletType_MASTER + // TabletRealtimeStats represents realtime stats from a running instance of vttablet. type TabletRealtimeStats struct { HealthError string `json:"health_error,omitempty"` @@ -108,8 +111,8 @@ func ParseCells(settings config.VitessConfigurationSettings) (cells []string) { } func configuredTabletType(settings config.VitessConfigurationSettings) topodata.TabletType { - if settings.TabletType == "MASTER" { - return topodata.TabletType_MASTER + if settings.TabletType == "PRIMARY" || settings.TabletType == "MASTER" { + return tabletTypePrimary } return topodata.TabletType_REPLICA } diff --git a/pkg/vitess/api_client_test.go b/pkg/vitess/api_client_test.go index f9d7fb9..6e2b16f 100644 --- a/pkg/vitess/api_client_test.go +++ b/pkg/vitess/api_client_test.go @@ -13,7 +13,7 @@ import ( func TestTabletIsValidReplica(t *testing.T) { replica := Tablet{Type: topodata.TabletType_REPLICA} - primary := Tablet{Type: topodata.TabletType_MASTER} + primary := Tablet{Type: tabletTypePrimary} if !replica.IsValidReplica() { t.Fatal("expected replica tablet to remain valid through compatibility method") @@ -31,7 +31,7 @@ func TestParseTablets(t *testing.T) { { // primary (should be ignored) Alias: &topodata.TabletAlias{Cell: "cell1"}, - Type: topodata.TabletType_MASTER, + Type: tabletTypePrimary, }, { // replica without realtime tablet stats enabled (assumed to be healthy) @@ -153,24 +153,26 @@ func TestParseTablets(t *testing.T) { } }) - t.Run("primary", func(t *testing.T) { - tablets, err := ParseTablets(config.VitessConfigurationSettings{ - API: vitessApi.URL, - Keyspace: "test", - Shard: "00", - TabletType: "MASTER", + for _, tabletType := range []string{"PRIMARY", "MASTER"} { + t.Run(tabletType, func(t *testing.T) { + tablets, err := ParseTablets(config.VitessConfigurationSettings{ + API: vitessApi.URL, + Keyspace: "test", + Shard: "00", + TabletType: tabletType, + }) + if err != nil { + t.Fatalf("Expected no error, got %q", err) + } + + if len(tablets) != 1 { + t.Fatalf("Expected 1 tablet, got %d", len(tablets)) + } + if tablets[0].Type != tabletTypePrimary { + t.Fatalf("Expected primary tablet, got %s", tablets[0].Type.String()) + } }) - if err != nil { - t.Fatalf("Expected no error, got %q", err) - } - - if len(tablets) != 1 { - t.Fatalf("Expected 1 tablet, got %d", len(tablets)) - } - if tablets[0].Type != topodata.TabletType_MASTER { - t.Fatalf("Expected master tablet, got %s", tablets[0].Type.String()) - } - }) + } t.Run("not-found", func(t *testing.T) { tablets, err := ParseTablets(config.VitessConfigurationSettings{ From 55212337ffb8d4a1b3ea07d4328f2e6f63797a27 Mon Sep 17 00:00:00 2001 From: Pradeep Kintali Date: Thu, 15 Oct 2026 15:06:44 -0700 Subject: [PATCH 8/9] Add shadow MySQL dependency evaluation Evaluate prospective required metrics without changing admission or memcache publication. Report separate HTTP and background would-reject counters, validate dependency graphs, and document safe activation and recovery behavior. 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 74263d6..c75424b 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" ], @@ -202,6 +245,10 @@ A cluster can require additional cluster metrics to pass before `freno` permits The ProxySQL query above is illustrative: deployments must provide a scalar query or operational view whose value increases as usable connection capacity is consumed. - `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. @@ -209,53 +256,95 @@ The ProxySQL query above is illustrative: deployments must provide a scalar quer - 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. ### 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 c434aa5..53d7e23 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) @@ -195,22 +196,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 ddf9759..7f22eed 100644 --- a/pkg/throttle/throttler.go +++ b/pkg/throttle/throttler.go @@ -456,6 +456,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) { From a83dc722cf6554e522b4d8f00f4c07f867eb3887 Mon Sep 17 00:00:00 2001 From: Pradeep Kintali Date: Fri, 16 Oct 2026 09:49:10 -0700 Subject: [PATCH 9/9] Expose read-only MySQL policy diagnostics Explain cached source and dependency metrics with effective thresholds, recovery settings and enforcement modes without running checks or exposing connection details. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: fbc83df2-d1af-406f-b97b-51cf0aaa570d --- doc/http.md | 27 +++++++ pkg/http/api.go | 24 ++++++ pkg/http/api_test.go | 61 ++++++++++++++ pkg/throttle/explain.go | 140 ++++++++++++++++++++++++++++++++ pkg/throttle/throttler_test.go | 143 +++++++++++++++++++++++++++++++++ 5 files changed, 395 insertions(+) create mode 100644 pkg/throttle/explain.go diff --git a/doc/http.md b/doc/http.md index b2e06ac..018885e 100644 --- a/doc/http.md +++ b/doc/http.md @@ -113,6 +113,33 @@ Notes: - `/config/memcache`: show the [memcache](memcache.md) configuration used, so freno clients can use it to implement more efficient read strategies. +### Read-only MySQL diagnostics + +`GET /explain/mysql/` returns the source metric and all of its required +and shadow dependencies, even when the source is healthy. It requires the +current leader (`503` on a follower), uses exact configured names without +`FallbackCluster`, and returns `404` for an unknown cluster. + +Each entry in `Metrics` includes `MetricName`, the cached `Value`, `Threshold`, +effective `RecoveryThreshold`, `RecoveryDurationMillis`, `CacheMillis`, +`FailOnNoHosts`, and host-exclusion settings. `Mode` is `source`, `enforced`, or +`shadow`, relative to the requested source. Dependencies below a shadow edge +remain shadow-only unless also reachable through an entirely required path. +Dependency lists preserve the configured graph. + +`StatusCode` and `Message` describe each metric independently, not the source's +combined admission decision. `Value` is `null` when a usable measurement is +unavailable; recovery errors retain the observed value. Recovery may reject +even when the value is below the throttle threshold. Probe-error details are +redacted; credentials, SQL, connection settings, and host identities are not +exported. The response reflects loaded policy and cached aggregates, not +per-host values or sample timestamps. + +This endpoint does not run application policy, random throttling, priority or +shared-domain checks, issue SQL, advance recovery, or increment shadow/check +counters. A diagnostic HTTP `200` means the explanation was retrieved, **not** +permission to write. Continue to use the regular `/check` endpoint for admission. + # GET method `GET` and `HEAD` respond with same status codes. But `GET` requests compute and return additional data. Automated requests should not be interested in this data; the status code is what should guide the clients. However humans or manual requests may benefit from extra information supplied by the `GET` request. diff --git a/pkg/http/api.go b/pkg/http/api.go index 5a605ef..de264a6 100644 --- a/pkg/http/api.go +++ b/pkg/http/api.go @@ -2,6 +2,7 @@ package http import ( "encoding/json" + "errors" "expvar" "fmt" "io" @@ -12,6 +13,7 @@ import ( "strings" "time" + "github.com/github/freno/pkg/base" "github.com/github/freno/pkg/config" "github.com/github/freno/pkg/group" "github.com/github/freno/pkg/throttle" @@ -34,6 +36,7 @@ type API interface { ReadCheck(w http.ResponseWriter, r *http.Request, _ httprouter.Params) ReadCheckIfExists(w http.ResponseWriter, r *http.Request, _ httprouter.Params) AggregatedMetrics(w http.ResponseWriter, r *http.Request, _ httprouter.Params) + ExplainMySQL(w http.ResponseWriter, r *http.Request, _ httprouter.Params) MetricsHealth(w http.ResponseWriter, r *http.Request, _ httprouter.Params) ThrottleApp(w http.ResponseWriter, r *http.Request, _ httprouter.Params) UnthrottleApp(w http.ResponseWriter, r *http.Request, _ httprouter.Params) @@ -235,6 +238,26 @@ func (api *APIImpl) AggregatedMetrics(w http.ResponseWriter, r *http.Request, ps json.NewEncoder(w).Encode(responseMap) } +func (api *APIImpl) ExplainMySQL(w http.ResponseWriter, r *http.Request, ps httprouter.Params) { + w.Header().Set("Content-Type", "application/json") + if api.consensusService == nil || !api.consensusService.IsLeader() { + w.WriteHeader(http.StatusServiceUnavailable) + json.NewEncoder(w).Encode(NewGeneralResponse(http.StatusServiceUnavailable, "Diagnostics require the current leader")) + return + } + explanation, err := api.throttlerCheck.ExplainMySQLCluster(ps.ByName("storeName")) + if err != nil { + status, message := http.StatusInternalServerError, "Invalid MySQL dependency configuration" + if errors.Is(err, base.NoSuchMetricError) { + status, message = http.StatusNotFound, "No such MySQL cluster" + } + w.WriteHeader(status) + json.NewEncoder(w).Encode(NewGeneralResponse(status, message)) + return + } + json.NewEncoder(w).Encode(explanation) +} + // MetricsHealth returns the time since last "OK" check per-metric func (api *APIImpl) MetricsHealth(w http.ResponseWriter, r *http.Request, ps httprouter.Params) { w.Header().Set("Content-Type", "application/json") @@ -438,6 +461,7 @@ func ConfigureRoutes(api API) *httprouter.Router { register(router, "/check-read-if-exists/:app/:storeType/:storeName/:threshold", api.ReadCheckIfExists) register(router, "/aggregated-metrics", api.AggregatedMetrics) + register(router, "/explain/mysql/:storeName", api.ExplainMySQL) register(router, "/metrics-health", api.MetricsHealth) register(router, "/throttle-app/:app", api.ThrottleApp) diff --git a/pkg/http/api_test.go b/pkg/http/api_test.go index 1c548c1..18d5c7d 100644 --- a/pkg/http/api_test.go +++ b/pkg/http/api_test.go @@ -6,10 +6,12 @@ import ( "net/http" "net/http/httptest" "reflect" + "strings" "testing" "github.com/github/freno/pkg/config" "github.com/github/freno/pkg/group" + "github.com/github/freno/pkg/throttle" metrics "github.com/rcrowley/go-metrics" ) @@ -61,6 +63,65 @@ func TestRoutes(t *testing.T) { } } +func TestExplainMySQLRoute(t *testing.T) { + defer config.Reset() + config.Settings().Stores.MySQL = config.MySQLConfigurationSettings{ + User: "private-user", Password: "private-password", + ProxySQLPassword: "private-proxy-password", + MetricQuery: "private-query", + Clusters: map[string]*config.MySQLClusterConfigurationSettings{ + "source": { + ThrottleThreshold: 1, + ShadowRequiredClusters: []string{"source-primary"}, + }, + "source-primary": { + User: "private-cluster-user", Password: "private-cluster-password", + MetricQuery: "private-cluster-query", ThrottleThreshold: 75, + StaticHostsSettings: config.StaticHostsConfigurationSettings{Hosts: []string{"private-host"}}, + }, + "broken": {RequiredClusters: []string{"not-configured"}}, + }, + } + for _, test := range []struct { + name string + leader bool + cluster string + wantCode int + }{ + {"leader snapshot", true, "source", http.StatusOK}, + {"follower", false, "source", http.StatusServiceUnavailable}, + {"unknown cluster", true, "unknown", http.StatusNotFound}, + {"invalid dependencies", true, "broken", http.StatusInternalServerError}, + } { + t.Run(test.name, func(t *testing.T) { + check := throttle.NewThrottlerCheck(throttle.NewThrottler()) + api := NewAPIImpl(check, &metricsConsensusService{isLeader: test.leader}) + router := ConfigureRoutes(api) + recorder := httptest.NewRecorder() + router.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/explain/mysql/"+test.cluster, nil)) + if recorder.Code != test.wantCode { + t.Fatalf("status = %d, want %d: %s", recorder.Code, test.wantCode, recorder.Body.String()) + } + if strings.Contains(recorder.Body.String(), "private-") { + t.Fatal("diagnostic response exposed private configuration") + } + if test.wantCode == http.StatusOK { + var result throttle.MySQLExplanation + if err := json.Unmarshal(recorder.Body.Bytes(), &result); err != nil { + t.Fatal(err) + } + if len(result.Metrics) != 2 || result.Metrics[1].Threshold != 75 || + result.Metrics[1].Mode != "shadow" || result.Metrics[1].Value != nil { + t.Fatalf("unexpected diagnostic response: %s", recorder.Body.String()) + } + if !strings.Contains(recorder.Body.String(), `"Value":null`) { + t.Fatal("missing measurement must be JSON null") + } + } + }) + } +} + func TestMemcacheConfigWhenProvided(t *testing.T) { defer config.Reset() diff --git a/pkg/throttle/explain.go b/pkg/throttle/explain.go new file mode 100644 index 0000000..30780fa --- /dev/null +++ b/pkg/throttle/explain.go @@ -0,0 +1,140 @@ +package throttle + +import ( + "errors" + "fmt" + "math" + "net/http" + "sort" + + "github.com/github/freno/pkg/base" + "github.com/github/freno/pkg/config" +) + +type MySQLMetricExplanation struct { + MetricName string + Mode string + StatusCode int + Value *float64 + Threshold float64 + RecoveryThreshold float64 + RecoveryDurationMillis int + CacheMillis int + FailOnNoHosts bool + IgnoreHostsCount int + IgnoreHostsThreshold float64 + IgnoreDialTcpErrors bool + Message string + RequiredClusters []string `json:",omitempty"` + ShadowRequiredClusters []string `json:",omitempty"` +} + +type MySQLExplanation struct { + MetricName string + Metrics []MySQLMetricExplanation +} + +// ExplainMySQLCluster reads loaded policy and cached metrics without running +// admission checks, sampling probes, or changing recovery and shadow counters. +func (check *ThrottlerCheck) ExplainMySQLCluster(clusterName string) (*MySQLExplanation, error) { + settings := config.Settings().Stores.MySQL + if settings.Clusters[clusterName] == nil { + return nil, base.NoSuchMetricError + } + + modes := make(map[string]string) + var visit func(string, string) error + visit = func(name, mode string) error { + if previous, found := modes[name]; found && (previous != "shadow" || mode == "shadow") { + return nil + } + cluster := settings.Clusters[name] + if cluster == nil { + return fmt.Errorf("missing MySQL dependency configuration for %q", name) + } + modes[name] = mode + requiredMode := "enforced" + if mode == "shadow" { + requiredMode = "shadow" + } + for _, dependency := range cluster.RequiredClusters { + if err := visit(dependency, requiredMode); err != nil { + return err + } + } + for _, dependency := range cluster.ShadowRequiredClusters { + if err := visit(dependency, "shadow"); err != nil { + return err + } + } + return nil + } + if err := visit(clusterName, "source"); err != nil { + return nil, err + } + + names := make([]string, 0, len(modes)-1) + for name := range modes { + if name != clusterName { + names = append(names, name) + } + } + sort.Strings(names) + names = append([]string{clusterName}, names...) + + snapshot := check.AggregatedMetrics() + explanation := &MySQLExplanation{MetricName: "mysql/" + clusterName} + for _, name := range names { + cluster := settings.Clusters[name] + metricName := "mysql/" + name + metric, found := snapshot[metricName] + if !found { + metric = base.NoMetricResultYet + } + value, err := metric.Get() + result := newMetricCheckResult(value, cluster.ThrottleThreshold, err, cluster.FailOnNoHosts) + if err != nil { + switch { + case errors.Is(err, base.RecoveryNotCompleteError): + result.Message = base.RecoveryNotCompleteError.Error() + case errors.Is(err, base.NoHostsError): + result.Message = base.NoHostsError.Error() + case errors.Is(err, base.NoSuchMetricError): + result.Message = base.NoSuchMetricError.Error() + case metric == base.NoMetricResultYet: + result.Message = "Metric not collected yet" + default: + // Probe errors can contain connection details or query text. + result.Message = "Probe error; consult server logs" + } + } + var observedValue *float64 + if math.IsNaN(value) || math.IsInf(value, 0) { + result = NewErrorCheckResult(http.StatusInternalServerError, fmt.Errorf("invalid metric value")) + } else if err == nil || errors.Is(err, base.RecoveryNotCompleteError) { + observedValue = &value + } + recoveryThreshold := cluster.ThrottleThreshold + if cluster.RecoveryThreshold != nil { + recoveryThreshold = *cluster.RecoveryThreshold + } + explanation.Metrics = append(explanation.Metrics, MySQLMetricExplanation{ + MetricName: metricName, + Mode: modes[name], + StatusCode: result.StatusCode, + Value: observedValue, + Threshold: cluster.ThrottleThreshold, + RecoveryThreshold: recoveryThreshold, + RecoveryDurationMillis: cluster.RecoveryDurationMillis, + CacheMillis: cluster.CacheMillis, + FailOnNoHosts: cluster.FailOnNoHosts, + IgnoreHostsCount: cluster.IgnoreHostsCount, + IgnoreHostsThreshold: cluster.IgnoreHostsThreshold, + IgnoreDialTcpErrors: settings.IgnoreDialTcpErrors, + Message: result.Message, + RequiredClusters: cluster.RequiredClusters, + ShadowRequiredClusters: cluster.ShadowRequiredClusters, + }) + } + return explanation, nil +} diff --git a/pkg/throttle/throttler_test.go b/pkg/throttle/throttler_test.go index 2c8af65..9bcae24 100644 --- a/pkg/throttle/throttler_test.go +++ b/pkg/throttle/throttler_test.go @@ -1,6 +1,9 @@ package throttle import ( + "errors" + "fmt" + "math" "net/http" "testing" "time" @@ -13,6 +16,146 @@ import ( "github.com/stretchr/testify/assert" ) +func TestExplainMySQLCluster(t *testing.T) { + originalSettings := config.Settings().Stores.MySQL + defer func() { config.Settings().Stores.MySQL = originalSettings }() + recoveryThreshold := 100.0 + config.Settings().Stores.MySQL = config.MySQLConfigurationSettings{ + IgnoreDialTcpErrors: true, + FallbackCluster: "explain-source", + Clusters: map[string]*config.MySQLClusterConfigurationSettings{ + "explain-source": { + ThrottleThreshold: 1, + RequiredClusters: []string{"explain-required"}, + ShadowRequiredClusters: []string{"explain-candidate", "explain-missing"}, + }, + "explain-required": { + ThrottleThreshold: 10, + RecoveryDurationMillis: 30000, + RequiredClusters: []string{"explain-shared"}, + }, + "explain-candidate": { + ThrottleThreshold: 175, + RecoveryThreshold: &recoveryThreshold, + RecoveryDurationMillis: 30000, + CacheMillis: 5000, + IgnoreHostsCount: 1, + IgnoreHostsThreshold: 15, + FailOnNoHosts: true, + RequiredClusters: []string{"explain-shared", "explain-no-hosts"}, + }, + "explain-shared": {ThrottleThreshold: 20}, + "explain-missing": {ThrottleThreshold: 30}, + "explain-no-hosts": {ThrottleThreshold: 40, FailOnNoHosts: true}, + }, + } + throttler := NewThrottler() + setTestMySQLClusterMetric(throttler, "explain-source", 0.1, 1) + setTestMySQLClusterMetric(throttler, "explain-candidate", 200, 175) + throttler.aggregatedMetrics.SetDefault("mysql/explain-required", base.NewErrorMetricResult(3, fmt.Errorf("private-password: %w", base.RecoveryNotCompleteError))) + throttler.aggregatedMetrics.SetDefault("mysql/explain-shared", base.NewErrorMetricResult(0, errors.New("query failed with private-password"))) + throttler.aggregatedMetrics.SetDefault("mysql/explain-no-hosts", base.NoHostsMetricResult) + + counterName := "shadow.http.mysql.explain-source.total" + metrics.DefaultRegistry.Unregister(counterName) + defer metrics.DefaultRegistry.Unregister(counterName) + check := NewThrottlerCheck(throttler) + result, err := check.ExplainMySQLCluster("explain-source") + if !assert.NoError(t, err) { + return + } + assert.Equal(t, "mysql/explain-source", result.MetricName) + assert.Len(t, result.Metrics, 6) + assert.Equal(t, "mysql/explain-source", result.Metrics[0].MetricName) + byName := make(map[string]MySQLMetricExplanation) + for _, metric := range result.Metrics { + byName[metric.MetricName] = metric + } + source := byName["mysql/explain-source"] + assert.Equal(t, "source", source.Mode) + assert.Equal(t, http.StatusOK, source.StatusCode) + assert.Equal(t, 0.1, *source.Value) + assert.Equal(t, 1.0, source.RecoveryThreshold) + + candidate := byName["mysql/explain-candidate"] + assert.Equal(t, "shadow", candidate.Mode) + assert.Equal(t, http.StatusTooManyRequests, candidate.StatusCode) + assert.Equal(t, 200.0, *candidate.Value) + assert.Equal(t, 175.0, candidate.Threshold) + assert.Equal(t, 100.0, candidate.RecoveryThreshold) + assert.Equal(t, 30000, candidate.RecoveryDurationMillis) + assert.Equal(t, 5000, candidate.CacheMillis) + assert.True(t, candidate.FailOnNoHosts) + assert.Equal(t, 1, candidate.IgnoreHostsCount) + assert.Equal(t, 15.0, candidate.IgnoreHostsThreshold) + assert.True(t, candidate.IgnoreDialTcpErrors) + assert.Equal(t, []string{"explain-shared", "explain-no-hosts"}, candidate.RequiredClusters) + + required := byName["mysql/explain-required"] + assert.Equal(t, "enforced", required.Mode) + assert.Equal(t, http.StatusTooManyRequests, required.StatusCode) + assert.Equal(t, 3.0, *required.Value) + assert.Equal(t, base.RecoveryNotCompleteError.Error(), required.Message) + shared := byName["mysql/explain-shared"] + assert.Equal(t, "enforced", shared.Mode) + assert.Equal(t, http.StatusInternalServerError, shared.StatusCode) + assert.Nil(t, shared.Value) + assert.Equal(t, "Probe error; consult server logs", shared.Message) + assert.Equal(t, "shadow", byName["mysql/explain-no-hosts"].Mode) + assert.Equal(t, base.NoHostsError.Error(), byName["mysql/explain-no-hosts"].Message) + assert.Equal(t, http.StatusInternalServerError, byName["mysql/explain-no-hosts"].StatusCode) + assert.Nil(t, byName["mysql/explain-missing"].Value) + assert.Equal(t, http.StatusInternalServerError, byName["mysql/explain-missing"].StatusCode) + assert.Nil(t, metrics.DefaultRegistry.Get(counterName)) + assert.Empty(t, throttler.mysqlRecoveryState) + _, err = check.ExplainMySQLCluster("unknown") + assert.Equal(t, base.NoSuchMetricError, err) +} + +func TestExplainMySQLClusterPromotesSharedShadowDependency(t *testing.T) { + originalSettings := config.Settings().Stores.MySQL + defer func() { config.Settings().Stores.MySQL = originalSettings }() + config.Settings().Stores.MySQL = config.MySQLConfigurationSettings{ + Clusters: map[string]*config.MySQLClusterConfigurationSettings{ + "root": {RequiredClusters: []string{"first", "second"}}, + "first": { + ShadowRequiredClusters: []string{"shared"}, + }, + "second": {RequiredClusters: []string{"shared"}}, + "shared": {RequiredClusters: []string{"child"}}, + "child": {}, + }, + } + result, err := NewThrottlerCheck(NewThrottler()).ExplainMySQLCluster("root") + if !assert.NoError(t, err) { + return + } + for _, metric := range result.Metrics { + if metric.MetricName == "mysql/shared" || metric.MetricName == "mysql/child" { + assert.Equal(t, "enforced", metric.Mode) + } + } +} + +func TestExplainMySQLClusterInvalidValues(t *testing.T) { + originalSettings := config.Settings().Stores.MySQL + defer func() { config.Settings().Stores.MySQL = originalSettings }() + config.Settings().Stores.MySQL = config.MySQLConfigurationSettings{ + Clusters: map[string]*config.MySQLClusterConfigurationSettings{"root": {ThrottleThreshold: 1}}, + } + for _, value := range []float64{math.NaN(), math.Inf(1)} { + throttler := NewThrottler() + setTestMySQLClusterMetric(throttler, "root", value, 1) + result, err := NewThrottlerCheck(throttler).ExplainMySQLCluster("root") + if !assert.NoError(t, err) { + continue + } + assert.Equal(t, http.StatusInternalServerError, result.Metrics[0].StatusCode) + assert.Nil(t, result.Metrics[0].Value) + assert.Equal(t, "invalid metric value", result.Metrics[0].Message) + } +} + func TestRefreshMySQLInventoryStaticHostsIgnoreSettings(t *testing.T) { originalSettings := config.Settings().Stores.MySQL defer func() {