-
Notifications
You must be signed in to change notification settings - Fork 2.1k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[release-17.0] Tablet throttler: empty list of probes on non-leader (#…
…13926) (#13952) Signed-off-by: Shlomi Noach <[email protected]> Co-authored-by: vitess-bot[bot] <108069721+vitess-bot[bot]@users.noreply.github.com> Co-authored-by: Shlomi Noach <[email protected]>
- Loading branch information
1 parent
0543850
commit 4f01ad8
Showing
2 changed files
with
218 additions
and
6 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,192 @@ | ||
| /* | ||
| Copyright 2017 GitHub Inc. | ||
| Licensed under MIT License. See https://github.com/github/freno/blob/master/LICENSE | ||
| */ | ||
|
|
||
| package throttle | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "sync/atomic" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/patrickmn/go-cache" | ||
| "github.com/stretchr/testify/assert" | ||
| "github.com/stretchr/testify/require" | ||
|
|
||
| "vitess.io/vitess/go/vt/topo" | ||
| "vitess.io/vitess/go/vt/vttablet/tabletserver/throttle/config" | ||
| "vitess.io/vitess/go/vt/vttablet/tabletserver/throttle/mysql" | ||
|
|
||
| topodatapb "vitess.io/vitess/go/vt/proto/topodata" | ||
| ) | ||
|
|
||
| const ( | ||
| waitForProbesTimeout = 30 * time.Second | ||
| ) | ||
|
|
||
| type FakeTopoServer struct { | ||
| } | ||
|
|
||
| func (ts *FakeTopoServer) GetTablet(ctx context.Context, alias *topodatapb.TabletAlias) (*topo.TabletInfo, error) { | ||
| tablet := &topo.TabletInfo{ | ||
| Tablet: &topodatapb.Tablet{ | ||
| Alias: alias, | ||
| Hostname: "127.0.0.1", | ||
| MysqlHostname: "127.0.0.1", | ||
| MysqlPort: 3306, | ||
| PortMap: map[string]int32{"vt": 5000}, | ||
| Type: topodatapb.TabletType_REPLICA, | ||
| }, | ||
| } | ||
| return tablet, nil | ||
| } | ||
|
|
||
| func (ts *FakeTopoServer) FindAllTabletAliasesInShard(ctx context.Context, keyspace, shard string) ([]*topodatapb.TabletAlias, error) { | ||
| aliases := []*topodatapb.TabletAlias{ | ||
| {Cell: "zone1", Uid: 100}, | ||
| {Cell: "zone2", Uid: 101}, | ||
| } | ||
| return aliases, nil | ||
| } | ||
|
|
||
| func (ts *FakeTopoServer) GetSrvKeyspace(ctx context.Context, cell, keyspace string) (*topodatapb.SrvKeyspace, error) { | ||
| ks := &topodatapb.SrvKeyspace{} | ||
| return ks, nil | ||
| } | ||
|
|
||
| type FakeHeartbeatWriter struct { | ||
| } | ||
|
|
||
| func (w FakeHeartbeatWriter) RequestHeartbeats() { | ||
| } | ||
|
|
||
| func TestIsAppThrottled(t *testing.T) { | ||
| throttler := Throttler{ | ||
| throttledApps: cache.New(cache.NoExpiration, 0), | ||
| heartbeatWriter: FakeHeartbeatWriter{}, | ||
| } | ||
| assert.False(t, throttler.IsAppThrottled("app1")) | ||
| assert.False(t, throttler.IsAppThrottled("app2")) | ||
| assert.False(t, throttler.IsAppThrottled("app3")) | ||
| assert.False(t, throttler.IsAppThrottled("app4")) | ||
| // | ||
| throttler.ThrottleApp("app1", time.Now().Add(time.Hour), DefaultThrottleRatio) | ||
| throttler.ThrottleApp("app2", time.Now(), DefaultThrottleRatio) | ||
| throttler.ThrottleApp("app3", time.Now().Add(time.Hour), DefaultThrottleRatio) | ||
| throttler.ThrottleApp("app4", time.Now().Add(time.Hour), 0) | ||
| assert.False(t, throttler.IsAppThrottled("app2")) // expired | ||
| assert.True(t, throttler.IsAppThrottled("app3")) | ||
| assert.False(t, throttler.IsAppThrottled("app4")) // ratio is zero | ||
| // | ||
| throttler.UnthrottleApp("app1") | ||
| throttler.UnthrottleApp("app2") | ||
| throttler.UnthrottleApp("app3") | ||
| throttler.UnthrottleApp("app4") | ||
| assert.False(t, throttler.IsAppThrottled("app1")) | ||
| assert.False(t, throttler.IsAppThrottled("app2")) | ||
| assert.False(t, throttler.IsAppThrottled("app3")) | ||
| assert.False(t, throttler.IsAppThrottled("app4")) | ||
| } | ||
|
|
||
| // TestRefreshMySQLInventory tests the behavior of the throttler's RefreshMySQLInventory() function, which | ||
| // is called periodically in actual throttler. For a given cluster name, it generates a list of probes | ||
| // the throttler will use to check metrics. | ||
| // On a "self" cluster, that list is expect to probe the tablet itself. | ||
| // On any other cluster, the list is expected to be empty if non-leader (only leader throttler, on a | ||
| // `PRIMARY` tablet, probes other tablets). On the leader, the list is expected to be non-empty. | ||
| func TestRefreshMySQLInventory(t *testing.T) { | ||
| metricsQuery := "select 1" | ||
| config.Settings().Stores.MySQL.Clusters = map[string]*config.MySQLClusterConfigurationSettings{ | ||
| selfStoreName: {}, | ||
| "ks1": {}, | ||
| "ks2": {}, | ||
| } | ||
| clusters := config.Settings().Stores.MySQL.Clusters | ||
| for _, s := range clusters { | ||
| s.MetricQuery = metricsQuery | ||
| s.ThrottleThreshold = &atomic.Uint64{} | ||
| s.ThrottleThreshold.Store(1) | ||
| } | ||
|
|
||
| throttler := &Throttler{ | ||
| mysqlClusterProbesChan: make(chan *mysql.ClusterProbes), | ||
| mysqlClusterThresholds: cache.New(cache.NoExpiration, 0), | ||
| ts: &FakeTopoServer{}, | ||
| mysqlInventory: mysql.NewInventory(), | ||
| } | ||
| throttler.metricsQuery.Store(metricsQuery) | ||
| throttler.initThrottleTabletTypes() | ||
|
|
||
| validateClusterProbes := func(t *testing.T, ctx context.Context) { | ||
| testName := fmt.Sprintf("leader=%v", atomic.LoadInt64(&throttler.isLeader)) | ||
| t.Run(testName, func(t *testing.T) { | ||
| // validateProbesCount expectes number of probes according to cluster name and throttler's leadership status | ||
| validateProbesCount := func(t *testing.T, clusterName string, probes *mysql.Probes) { | ||
| if clusterName == selfStoreName { | ||
| assert.Equal(t, 1, len(*probes)) | ||
| } else if atomic.LoadInt64(&throttler.isLeader) > 0 { | ||
| assert.NotZero(t, len(*probes)) | ||
| } else { | ||
| assert.Empty(t, *probes) | ||
| } | ||
| } | ||
| t.Run("waiting for probes", func(t *testing.T) { | ||
| ctx, cancel := context.WithTimeout(ctx, waitForProbesTimeout) | ||
| defer cancel() | ||
| numClusterProbesResults := 0 | ||
| for { | ||
| select { | ||
| case probes := <-throttler.mysqlClusterProbesChan: | ||
| // Worth noting that in this unit test, the throttler is _closed_. Its own Operate() function does | ||
| // not run, and therefore there is none but us to both populate `mysqlClusterProbesChan` as well as | ||
| // read from it. We do not compete here with any other goroutine. | ||
| assert.NotNil(t, probes) | ||
|
|
||
| throttler.updateMySQLClusterProbes(ctx, probes) | ||
|
|
||
| numClusterProbesResults++ | ||
| validateProbesCount(t, probes.ClusterName, probes.InstanceProbes) | ||
|
|
||
| if numClusterProbesResults == len(clusters) { | ||
| // Achieved our goal | ||
| return | ||
| } | ||
| case <-ctx.Done(): | ||
| assert.FailNowf(t, ctx.Err().Error(), "waiting for %d cluster probes", len(clusters)) | ||
| } | ||
| } | ||
| }) | ||
| t.Run("validating probes", func(t *testing.T) { | ||
| for clusterName := range clusters { | ||
| probes, ok := throttler.mysqlInventory.ClustersProbes[clusterName] | ||
| require.True(t, ok) | ||
| validateProbesCount(t, clusterName, probes) | ||
| } | ||
| }) | ||
| }) | ||
| } | ||
| // | ||
| ctx := context.Background() | ||
|
|
||
| t.Run("initial, not leader", func(t *testing.T) { | ||
| atomic.StoreInt64(&throttler.isLeader, 0) | ||
| throttler.refreshMySQLInventory(ctx) | ||
| validateClusterProbes(t, ctx) | ||
| }) | ||
|
|
||
| t.Run("promote", func(t *testing.T) { | ||
| atomic.StoreInt64(&throttler.isLeader, 1) | ||
| throttler.refreshMySQLInventory(ctx) | ||
| validateClusterProbes(t, ctx) | ||
| }) | ||
|
|
||
| t.Run("demote, expect cleanup", func(t *testing.T) { | ||
| atomic.StoreInt64(&throttler.isLeader, 0) | ||
| throttler.refreshMySQLInventory(ctx) | ||
| validateClusterProbes(t, ctx) | ||
| }) | ||
| } |