Tablet throttler: multi-metric support

.github/workflows/upgrade_downgrade_test_onlineddl_flow.yml

@@ -146,6 +146,7 @@ jobs:
       if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
       timeout-minutes: 10
       run: |
+        echo "building last release: $(git rev-parse HEAD)"
         source build.env
         make build
         mkdir -p /tmp/vitess-build-last/
@@ -168,6 +169,7 @@ jobs:
       if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
       timeout-minutes: 10
       run: |
+        echo "building next release: $(git rev-parse HEAD)"
         source build.env
         NOVTADMINBUILD=1 make build
         mkdir -p /tmp/vitess-build-next/
@@ -188,6 +190,7 @@ jobs:
       if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
       timeout-minutes: 10
       run: |
+        echo "building this SHA: $(git rev-parse HEAD)"
         source build.env
         make build
         mkdir -p /tmp/vitess-build-current/

go/cmd/vtctldclient/command/onlineddl.go

@@ -33,6 +33,7 @@ import (
 
 	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
 	vtctldatapb "vitess.io/vitess/go/vt/proto/vtctldata"
+	"vitess.io/vitess/go/vt/proto/vttime"
 )
 
 const (
@@ -307,7 +308,7 @@ func throttleCommandHelper(cmd *cobra.Command, throttleType bool) error {
 		rule.ExpiresAt = protoutil.TimeToProto(time.Now().Add(throttle.DefaultAppThrottleDuration))
 	} else {
 		rule.Ratio = 0
-		rule.ExpiresAt = protoutil.TimeToProto(time.Now())
+		rule.ExpiresAt = &vttime.Time{} // zero
 	}
 
 	if strings.ToLower(uuid) == AllMigrationsIndicator {

go/cmd/vtctldclient/command/throttler.go

@@ -27,7 +27,11 @@ import (
 
 	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
 	vtctldatapb "vitess.io/vitess/go/vt/proto/vtctldata"
+	"vitess.io/vitess/go/vt/proto/vttime"
+	"vitess.io/vitess/go/vt/topo/topoproto"
 	"vitess.io/vitess/go/vt/vttablet/tabletserver/throttle"
+	"vitess.io/vitess/go/vt/vttablet/tabletserver/throttle/base"
+	"vitess.io/vitess/go/vt/vttablet/tabletserver/throttle/throttlerapp"
 )
 
 var (
@@ -39,13 +43,32 @@ var (
 		Args:                  cobra.ExactArgs(1),
 		RunE:                  commandUpdateThrottlerConfig,
 	}
+	CheckThrottler = &cobra.Command{
+		Use:                   "CheckThrottler [--app-name <name>] <tablet alias>",
+		Short:                 "Issue a throttler check on the given tablet.",
+		Example:               "CheckThrottler --app-name online-ddl zone1-0000000101",
+		DisableFlagsInUseLine: true,
+		Args:                  cobra.ExactArgs(1),
+		RunE:                  commandCheckThrottler,
+	}
+
+	GetThrottlerStatus = &cobra.Command{
+		Use:                   "GetThrottlerStatus <tablet alias>",
+		Short:                 "Get the throttler status for the given tablet.",
+		Example:               "GetThrottlerStatus zone1-0000000101",
+		DisableFlagsInUseLine: true,
+		Args:                  cobra.ExactArgs(1),
+		RunE:                  commandGetThrottlerStatus,
+	}
 )
 
 var (
 	updateThrottlerConfigOptions vtctldatapb.UpdateThrottlerConfigRequest
 	throttledAppRule             topodatapb.ThrottledAppRule
 	unthrottledAppRule           topodatapb.ThrottledAppRule
 	throttledAppDuration         time.Duration
+
+	checkThrottlerOptions vtctldatapb.CheckThrottlerRequest
 )
 
 func commandUpdateThrottlerConfig(cmd *cobra.Command, args []string) error {
@@ -56,14 +79,24 @@ func commandUpdateThrottlerConfig(cmd *cobra.Command, args []string) error {
 		return fmt.Errorf("throttle-app and unthrottle-app are mutually exclusive")
 	}
 
+	if updateThrottlerConfigOptions.MetricName != "" && !cmd.Flags().Changed("threshold") {
+		return fmt.Errorf("--metric-name flag requires --threshold flag. Set threshold to 0 to disable the metric threshold configuration")
+	}
+	if cmd.Flags().Changed("app-name") != cmd.Flags().Changed("app-metrics") {
+		return fmt.Errorf("--app-name and --app-metrics must be set together")
+	}
+	if cmd.Flags().Changed("app-name") && updateThrottlerConfigOptions.AppName == "" {
+		return fmt.Errorf("--app-name must not be empty")
+	}
+
 	updateThrottlerConfigOptions.CustomQuerySet = cmd.Flags().Changed("custom-query")
 	updateThrottlerConfigOptions.Keyspace = keyspace
 
 	if throttledAppRule.Name != "" {
 		throttledAppRule.ExpiresAt = protoutil.TimeToProto(time.Now().Add(throttledAppDuration))
 		updateThrottlerConfigOptions.ThrottledApp = &throttledAppRule
 	} else if unthrottledAppRule.Name != "" {
-		unthrottledAppRule.ExpiresAt = protoutil.TimeToProto(time.Now())
+		unthrottledAppRule.ExpiresAt = &vttime.Time{} // zero
 		updateThrottlerConfigOptions.ThrottledApp = &unthrottledAppRule
 	}
 
@@ -74,9 +107,67 @@ func commandUpdateThrottlerConfig(cmd *cobra.Command, args []string) error {
 	return nil
 }
 
+func commandCheckThrottler(cmd *cobra.Command, args []string) error {
+	alias, err := topoproto.ParseTabletAlias(cmd.Flags().Arg(0))
+	if err != nil {
+		return err
+	}
+
+	cli.FinishedParsing(cmd)
+	if _, err := base.ScopeFromString(checkThrottlerOptions.Scope); err != nil {
+		return err
+	}
+	resp, err := client.CheckThrottler(commandCtx, &vtctldatapb.CheckThrottlerRequest{
+		TabletAlias:           alias,
+		AppName:               checkThrottlerOptions.AppName,
+		Scope:                 checkThrottlerOptions.Scope,
+		SkipRequestHeartbeats: checkThrottlerOptions.SkipRequestHeartbeats,
+		OkIfNotExists:         checkThrottlerOptions.OkIfNotExists,
+	})
+	if err != nil {
+		return err
+	}
+
+	data, err := cli.MarshalJSON(resp)
+	if err != nil {
+		return err
+	}
+
+	fmt.Printf("%s\n", data)
+
+	return nil
+}
+
+func commandGetThrottlerStatus(cmd *cobra.Command, args []string) error {
+	alias, err := topoproto.ParseTabletAlias(cmd.Flags().Arg(0))
+	if err != nil {
+		return err
+	}
+
+	cli.FinishedParsing(cmd)
+
+	resp, err := client.GetThrottlerStatus(commandCtx, &vtctldatapb.GetThrottlerStatusRequest{
+		TabletAlias: alias,
+	})
+	if err != nil {
+		return err
+	}
+
+	data, err := cli.MarshalJSON(resp)
+	if err != nil {
+		return err
+	}
+
+	fmt.Printf("%s\n", data)
+
+	return nil
+}
+
 func init() {
+	// UpdateThrottlerConfig
 	UpdateThrottlerConfig.Flags().BoolVar(&updateThrottlerConfigOptions.Enable, "enable", false, "Enable the throttler")
 	UpdateThrottlerConfig.Flags().BoolVar(&updateThrottlerConfigOptions.Disable, "disable", false, "Disable the throttler")
+	UpdateThrottlerConfig.Flags().StringVar(&updateThrottlerConfigOptions.MetricName, "metric-name", "", "name of the metric for which we apply a new threshold (requires ==threshold). Leave empty for v19 default metric/threshold")
 	UpdateThrottlerConfig.Flags().Float64Var(&updateThrottlerConfigOptions.Threshold, "threshold", 0, "threshold for the either default check (replication lag seconds) or custom check")
 	UpdateThrottlerConfig.Flags().StringVar(&updateThrottlerConfigOptions.CustomQuery, "custom-query", "", "custom throttler check query")
 	UpdateThrottlerConfig.Flags().BoolVar(&updateThrottlerConfigOptions.CheckAsCheckSelf, "check-as-check-self", false, "/throttler/check requests behave as is /throttler/check-self was called")
@@ -88,5 +179,17 @@ func init() {
 	UpdateThrottlerConfig.Flags().DurationVar(&throttledAppDuration, "throttle-app-duration", throttle.DefaultAppThrottleDuration, "duration after which throttled app rule expires (app specififed in --throttled-app)")
 	UpdateThrottlerConfig.Flags().BoolVar(&throttledAppRule.Exempt, "throttle-app-exempt", throttledAppRule.Exempt, "exempt this app from being at all throttled. WARNING: use with extreme care, as this is likely to push metrics beyond the throttler's threshold, and starve other apps")
 
+	UpdateThrottlerConfig.Flags().StringVar(&updateThrottlerConfigOptions.AppName, "app-name", "", "app name for which to assign metrics (requires --app-metrics)")
+	UpdateThrottlerConfig.Flags().StringSliceVar(&updateThrottlerConfigOptions.AppCheckedMetrics, "app-metrics", nil, "metrics to be used when checking the throttler for the app (requires --app-name). Empty to restore to default metrics")
+
 	Root.AddCommand(UpdateThrottlerConfig)
+	// Check Throttler
+	CheckThrottler.Flags().StringVar(&checkThrottlerOptions.AppName, "app-name", throttlerapp.VitessName.String(), "app to identify as")
+	CheckThrottler.Flags().StringVar(&checkThrottlerOptions.Scope, "scope", base.UndefinedScope.String(), "check scope ('shard', 'self' or leave empty for per-metric defaults)")
+	CheckThrottler.Flags().BoolVar(&checkThrottlerOptions.SkipRequestHeartbeats, "skip-heartbeats", false, "skip renewing heartbeat lease")
+	CheckThrottler.Flags().BoolVar(&checkThrottlerOptions.OkIfNotExists, "ok-if-not-exists", false, "return OK even if metric does not exist")
+	Root.AddCommand(CheckThrottler)
+
+	// GetThrottlerStatus
+	Root.AddCommand(GetThrottlerStatus)
 }

go/flags/endtoend/vtctldclient.txt

@@ -19,6 +19,7 @@ Available Commands:
   Backup                      Uses the BackupStorage service on the given tablet to create and store a new backup.
   BackupShard                 Finds the most up-to-date REPLICA, RDONLY, or SPARE tablet in the given shard and uses the BackupStorage service on that tablet to create and store a new backup.
   ChangeTabletType            Changes the db type for the specified tablet, if possible.
+  CheckThrottler              Issue a throttler check on the given tablet.
   CreateKeyspace              Creates the specified keyspace in the topology.
   CreateShard                 Creates the specified shard in the topology.
   DeleteCellInfo              Deletes the CellInfo for the provided cell.
@@ -55,6 +56,7 @@ Available Commands:
   GetTablet                   Outputs a JSON structure that contains information about the tablet.
   GetTabletVersion            Print the version of a tablet from its debug vars.
   GetTablets                  Looks up tablets according to filter criteria.
+  GetThrottlerStatus          Get the throttler status for the given tablet.
   GetTopologyPath             Gets the value associated with the particular path (key) in the topology server.
   GetVSchema                  Prints a JSON representation of a keyspace's topo record.
   GetWorkflows                Gets all vreplication workflows (Reshard, MoveTables, etc) in the given keyspace.

go/test/endtoend/cluster/vtctldclient_process.go

@@ -66,13 +66,13 @@ func (vtctldclient *VtctldClientProcess) ExecuteCommandWithOutput(args ...string
 		pArgs = append(pArgs, "--test.coverprofile="+getCoveragePath("vtctldclient-"+args[0]+".out"), "--test.v")
 	}
 	pArgs = append(pArgs, args...)
-	for i := 1; i <= retries; i++ {
+	for i := range retries {
 		tmpProcess := exec.Command(
 			vtctldclient.Binary,
 			filterDoubleDashArgs(pArgs, vtctldclient.VtctldClientMajorVersion)...,
 		)
 		msg := binlogplayer.LimitString(strings.Join(tmpProcess.Args, " "), 256) // limit log line length
-		log.Infof("Executing vtctldclient with command: %v (attempt %d of %d)", msg, i, retries)
+		log.Infof("Executing vtctldclient with command: %v (attempt %d of %d)", msg, i+1, retries)
 		resultByte, err = tmpProcess.CombinedOutput()
 		resultStr = string(resultByte)
 		if err == nil || !shouldRetry(resultStr) {

go/test/endtoend/onlineddl/flow/onlineddl_flow_test.go

@@ -222,7 +222,7 @@ func TestSchemaChange(t *testing.T) {
 	shards = clusterInstance.Keyspaces[0].Shards
 	require.Equal(t, 1, len(shards))
 
-	throttler.EnableLagThrottlerAndWaitForStatus(t, clusterInstance, time.Second)
+	throttler.EnableLagThrottlerAndWaitForStatus(t, clusterInstance)
 
 	t.Run("flow", func(t *testing.T) {
 		t.Run("create schema", func(t *testing.T) {
@@ -283,24 +283,18 @@ func TestSchemaChange(t *testing.T) {
 				onlineddl.CheckMigrationStatus(t, &vtParams, shards, uuid, schema.OnlineDDLStatusRunning)
 			})
 			t.Run("throttle online-ddl", func(t *testing.T) {
+				onlineddl.CheckThrottledApps(t, &vtParams, throttlerapp.OnlineDDLName, false)
 				onlineddl.ThrottleAllMigrations(t, &vtParams)
 				onlineddl.CheckThrottledApps(t, &vtParams, throttlerapp.OnlineDDLName, true)
-
-				for _, tab := range tablets {
-					body, err := throttleApp(tab.VttabletProcess, throttlerapp.OnlineDDLName)
-					assert.NoError(t, err)
-					assert.Contains(t, body, throttlerapp.OnlineDDLName)
-				}
 				waitForThrottleCheckStatus(t, throttlerapp.OnlineDDLName, primaryTablet, http.StatusExpectationFailed)
 			})
 			t.Run("unthrottle online-ddl", func(t *testing.T) {
 				onlineddl.UnthrottleAllMigrations(t, &vtParams)
-				onlineddl.CheckThrottledApps(t, &vtParams, throttlerapp.OnlineDDLName, false)
-
-				for _, tab := range tablets {
-					body, err := unthrottleApp(tab.VttabletProcess, throttlerapp.OnlineDDLName)
+				if !onlineddl.CheckThrottledApps(t, &vtParams, throttlerapp.OnlineDDLName, false) {
+					status, err := throttler.GetThrottlerStatus(&clusterInstance.VtctldClientProcess, primaryTablet)
 					assert.NoError(t, err)
-					assert.Contains(t, body, throttlerapp.OnlineDDLName)
+
+					t.Logf("Throttler status: %+v", status)
 				}
 				waitForThrottleCheckStatus(t, throttlerapp.OnlineDDLName, primaryTablet, http.StatusOK)
 			})

go/test/endtoend/onlineddl/revert/onlineddl_revert_test.go

@@ -21,6 +21,7 @@ import (
 	"flag"
 	"fmt"
 	"math/rand/v2"
+	"net/http"
 	"os"
 	"path"
 	"strings"
@@ -85,6 +86,7 @@ deletesAttempts=%d, deletesFailures=%d, deletesNoops=%d, deletes=%d,
 
 var (
 	clusterInstance *cluster.LocalProcessCluster
+	primaryTablet   *cluster.Vttablet
 	shards          []cluster.Shard
 	vtParams        mysql.ConnParams
 	mysqlVersion    string
@@ -188,7 +190,7 @@ func TestMain(m *testing.M) {
 			Host: clusterInstance.Hostname,
 			Port: clusterInstance.VtgateMySQLPort,
 		}
-
+		primaryTablet = clusterInstance.Keyspaces[0].Shards[0].PrimaryTablet()
 		return m.Run(), nil
 	}()
 	if err != nil {
@@ -205,7 +207,8 @@ func TestSchemaChange(t *testing.T) {
 	shards = clusterInstance.Keyspaces[0].Shards
 	require.Equal(t, 1, len(shards))
 
-	throttler.EnableLagThrottlerAndWaitForStatus(t, clusterInstance, time.Second)
+	throttler.EnableLagThrottlerAndWaitForStatus(t, clusterInstance)
+	throttler.WaitForCheckThrottlerResult(t, clusterInstance, primaryTablet, "test", nil, http.StatusOK, time.Minute)
 
 	t.Run("revertible", testRevertible)
 	t.Run("revert", testRevert)
@@ -410,7 +413,16 @@ func testRevertible(t *testing.T) {
 				// This is the migration we will test, and see whether it is revertible or not (and why not).
 				toStatement := fmt.Sprintf(createTableWrapper, testcase.toSchema)
 				uuid = testOnlineDDLStatement(t, toStatement, ddlStrategy, "vtgate", tableName, "")
-				onlineddl.CheckMigrationStatus(t, &vtParams, shards, uuid, schema.OnlineDDLStatusComplete)
+				if !onlineddl.CheckMigrationStatus(t, &vtParams, shards, uuid, schema.OnlineDDLStatusComplete) {
+					resp, err := throttler.CheckThrottler(clusterInstance, primaryTablet, "test", nil)
+					assert.NoError(t, err)
+					fmt.Println("Throttler check response: ", resp)
+
+					output, err := throttler.GetThrottlerStatusRaw(&clusterInstance.VtctldClientProcess, primaryTablet)
+					assert.NoError(t, err)
+					fmt.Println("Throttler status response: ", output)
+				}
+
 				checkTable(t, tableName, true)
 			})
 			t.Run("check migration", func(t *testing.T) {
@@ -557,7 +569,7 @@ func testRevert(t *testing.T) {
 		onlineddl.VtgateExecQuery(t, &vtParams, populatePartitionedTableStatement, "")
 	}
 
-	mysqlVersion = onlineddl.GetMySQLVersion(t, clusterInstance.Keyspaces[0].Shards[0].PrimaryTablet())
+	mysqlVersion = onlineddl.GetMySQLVersion(t, primaryTablet)
 	require.NotEmpty(t, mysqlVersion)
 
 	capableOf := mysql.ServerVersionCapableOf(mysqlVersion)