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/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/doc/memcache.md b/doc/memcache.md index 7cbe91e..b2357d3 100644 --- a/doc/memcache.md +++ b/doc/memcache.md @@ -37,6 +37,13 @@ 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. +- `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 9815050..c75424b 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 `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. @@ -148,3 +150,202 @@ 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 + +#### 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": { + "ShadowRequiredClusters": [ + "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. +- `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. +- 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 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. 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 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. 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. + +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 + +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. 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..af83e79 100644 --- a/pkg/config/config_test.go +++ b/pkg/config/config_test.go @@ -112,6 +112,184 @@ 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/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", + 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 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: "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"`}, + } + + 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 8e5a7bf..53d7e23 100644 --- a/pkg/config/mysql_config.go +++ b/pkg/config/mysql_config.go @@ -7,22 +7,28 @@ 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 + 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) + 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 +81,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,10 +106,18 @@ 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 } + 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 } @@ -148,6 +165,84 @@ 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/ShadowRequiredClusters cycle involving %q", clusterName) + case visited: + return nil + } + + states[clusterName] = visiting + 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) + } + } + 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 + 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/config/vitess_config.go b/pkg/config/vitess_config.go index 79de29e..2b61d9b 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 "PRIMARY", "MASTER": + settings.TabletType = "PRIMARY" + default: + return fmt.Errorf("unsupported Vitess tablet type %q", settings.TabletType) + } + return nil +} diff --git a/pkg/http/api.go b/pkg/http/api.go index d0bd34d..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") @@ -366,7 +389,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 +398,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()) @@ -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 40eb7c6..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() @@ -120,9 +181,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 +194,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 +214,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 +226,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 0c1cc7b..34e367f 100644 --- a/pkg/throttle/check.go +++ b/pkg/throttle/check.go @@ -54,34 +54,26 @@ 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")) } - 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) { - // If no hosts we report 0 for the metric. - statusCode = http.StatusOK // 200 - } else if err != nil { - // any error - statusCode = http.StatusInternalServerError // 500 - } else if value > threshold { - // casual throttling - statusCode = http.StatusTooManyRequests // 429 - err = base.ThresholdExceededError - + failOnNoHosts := false + if storeType == "mysql" { + if clusterSettings, ok := config.Settings().Stores.MySQL.Clusters[storeName]; ok { + failOnNoHosts = clusterSettings.FailOnNoHosts + } + } + 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, @@ -89,17 +81,59 @@ 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) } - return NewCheckResult(statusCode, value, threshold, err) + checkResult.MetricName = metricName + return checkResult +} + +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.ReadCheck = false + requiredFlags.OverrideThreshold = 0 + requiredResult := check.checkMySQLCluster(appName, requiredCluster, &requiredFlags, checkedClusters) + if requiredResult == nil || requiredResult.StatusCode == http.StatusOK { + continue + } + if requiredResult.StatusCode == http.StatusNotFound { + 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 + } + return rootResult } // 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 { @@ -114,8 +148,12 @@ 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)) + 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) { @@ -125,11 +163,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) @@ -176,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 1a288fc..a6730eb 100644 --- a/pkg/throttle/check_result.go +++ b/pkg/throttle/check_result.go @@ -1,16 +1,40 @@ 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"` 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/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/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 618d91f..f48c9a4 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), @@ -146,7 +155,7 @@ func (throttler *Throttler) Operate() { case <-leaderCheckTick: { // sparse - throttler.isLeader = throttler.isLeaderFunc() + throttler.updateLeadership(throttler.isLeaderFunc()) } case <-mysqlCollectTick: { @@ -439,15 +448,36 @@ 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) + 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) + 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 { - 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) @@ -456,12 +486,109 @@ 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 { + 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 { + // 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 + } + + recoveryThreshold := clusterSettings.ThrottleThreshold + if clusterSettings.RecoveryThreshold != nil { + recoveryThreshold = *clusterSettings.RecoveryThreshold + } + 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 +} + 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 9a4fa61..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" @@ -8,9 +11,151 @@ import ( "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 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() { @@ -125,6 +270,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) @@ -261,3 +434,228 @@ 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) + assert.Equal(t, "mysql/primary", result.MetricName) + }) + + 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("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") + 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) + }) +} + +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) +} + +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()) +} + +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 87cc66c..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"` @@ -61,14 +64,20 @@ 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() } +// 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, } @@ -101,19 +110,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 == "PRIMARY" || settings.TabletType == "MASTER" { + return tabletTypePrimary + } + 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 +154,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..6e2b16f 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: tabletTypePrimary} + + 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() { @@ -19,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) @@ -141,6 +153,27 @@ func TestParseTablets(t *testing.T) { } }) + 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()) + } + }) + } + t.Run("not-found", func(t *testing.T) { tablets, err := ParseTablets(config.VitessConfigurationSettings{ API: vitessApi.URL,