Throttler: Expose Tablet's Config & Leverage to Deflake Tests

examples/common/scripts/vttablet-up.sh

@@ -46,7 +46,6 @@ vttablet \
  --init_shard $shard \
  --init_tablet_type $tablet_type \
  --health_check_interval 5s \
- --enable_replication_reporter \
  --backup_storage_implementation file \
  --file_backup_storage_root $VTDATAROOT/backups \
  --restore_from_backup \
@@ -56,6 +55,7 @@ vttablet \
  --pid_file $VTDATAROOT/$tablet_dir/vttablet.pid \
  --vtctld_addr http://$hostname:$vtctld_web_port/ \
  --disable_active_reparents \
+ --throttler-config-via-topo --heartbeat_enable --heartbeat_interval=250ms --heartbeat_on_demand_duration=5s \
  > $VTDATAROOT/$tablet_dir/vttablet.out 2>&1 &
 
 # Block waiting for the tablet to be listening

go/flags/endtoend/vttablet.txt

@@ -304,7 +304,7 @@ Usage of vttablet:
       --tablet_protocol string                                           Protocol to use to make queryservice RPCs to vttablets. (default "grpc")
       --throttle_check_as_check_self                                     Should throttler/check return a throttler/check-self result (changes throttler behavior for writes)
       --throttle_metrics_query SELECT                                    Override default heartbeat/lag metric. Use either SELECT (must return single row, single value) or `SHOW GLOBAL ... LIKE ...` queries. Set -throttle_metrics_threshold respectively.
-      --throttle_metrics_threshold float                                 Override default throttle threshold, respective to -throttle_metrics_query (default 1.7976931348623157e+308)
+      --throttle_metrics_threshold float                                 Override default throttle threshold, respective to --throttle_metrics_query (default 1.7976931348623157e+308)
       --throttle_tablet_types string                                     Comma separated VTTablet types to be considered by the throttler. default: 'replica'. example: 'replica,rdonly'. 'replica' aways implicitly included (default "replica")
       --throttle_threshold duration                                      Replication lag threshold for default lag throttling (default 1s)
       --throttler-config-via-topo                                        When 'true', read config from topo service and ignore throttle_threshold, throttle_metrics_threshold, throttle_metrics_query, throttle_check_as_check_self

go/test/endtoend/onlineddl/vrepl/onlineddl_vrepl_test.go

@@ -30,6 +30,7 @@ import (
 	"vitess.io/vitess/go/mysql"
 	"vitess.io/vitess/go/test/endtoend/cluster"
 	"vitess.io/vitess/go/test/endtoend/onlineddl"
+	"vitess.io/vitess/go/test/endtoend/throttler"
 	"vitess.io/vitess/go/vt/schema"
 	"vitess.io/vitess/go/vt/vttablet/tabletmanager/vreplication"
 	throttlebase "vitess.io/vitess/go/vt/vttablet/tabletserver/throttle/base"
@@ -259,13 +260,13 @@ func TestSchemaChange(t *testing.T) {
 	err := clusterInstance.WaitForTabletsToHealthyInVtgate()
 	require.NoError(t, err)
 
-	_, err = onlineddl.UpdateThrottlerTopoConfig(clusterInstance, true, false, 0, "", false)
+	_, err = throttler.UpdateThrottlerTopoConfig(clusterInstance, true, false, 0, "", false)
 	require.NoError(t, err)
 
 	for _, ks := range clusterInstance.Keyspaces {
 		for _, shard := range ks.Shards {
 			for _, tablet := range shard.Vttablets {
-				onlineddl.WaitForThrottlerStatusEnabled(t, tablet, extendedMigrationWait)
+				throttler.WaitForThrottlerStatusEnabled(t, tablet, true, nil, extendedMigrationWait)
 			}
 		}
 	}

go/test/endtoend/onlineddl/vtctlutil.go

@@ -17,8 +17,6 @@ limitations under the License.
 package onlineddl
 
 import (
-	"context"
-	"fmt"
 	"testing"
 	"time"
 
@@ -36,51 +34,3 @@ func CheckCancelAllMigrationsViaVtctl(t *testing.T, vtctlclient *cluster.VtctlCl
 	_, err := vtctlclient.ApplySchemaWithOutput(keyspace, cancelQuery, cluster.VtctlClientParams{})
 	assert.NoError(t, err)
 }
-
-// UpdateThrottlerTopoConfig runs vtctlclient UpdateThrottlerConfig.
-// This retries the command until it succeeds or times out as the
-// SrvKeyspace record may not yet exist for a newly created
-// Keyspace that is still initializing before it becomes serving.
-func UpdateThrottlerTopoConfig(clusterInstance *cluster.LocalProcessCluster, enable bool, disable bool, threshold float64, metricsQuery string, viaVtctldClient bool) (result string, err error) {
-	args := []string{}
-	clientfunc := clusterInstance.VtctldClientProcess.ExecuteCommandWithOutput
-	if !viaVtctldClient {
-		args = append(args, "--")
-		clientfunc = clusterInstance.VtctlclientProcess.ExecuteCommandWithOutput
-	}
-	args = append(args, "UpdateThrottlerConfig")
-	if enable {
-		args = append(args, "--enable")
-	}
-	if disable {
-		args = append(args, "--disable")
-	}
-	if threshold > 0 {
-		args = append(args, "--threshold", fmt.Sprintf("%f", threshold))
-	}
-	if metricsQuery != "" {
-		args = append(args, "--custom-query", metricsQuery)
-		args = append(args, "--check-as-check-self")
-	} else {
-		args = append(args, "--check-as-check-shard")
-	}
-	args = append(args, clusterInstance.Keyspaces[0].Name)
-
-	ctx, cancel := context.WithTimeout(context.Background(), throttlerConfigTimeout)
-	defer cancel()
-
-	ticker := time.NewTicker(time.Second)
-	defer ticker.Stop()
-
-	for {
-		result, err = clientfunc(args...)
-		if err == nil {
-			return result, nil
-		}
-		select {
-		case <-ctx.Done():
-			return "", fmt.Errorf("timed out waiting for UpdateThrottlerConfig to succeed after %v. Last seen value: %+v, error: %v", throttlerConfigTimeout, result, err)
-		case <-ticker.C:
-		}
-	}
-}