Tablet throttler: read and use MySQL host metrics

go/test/endtoend/tabletmanager/throttler_topo/throttler_test.go

@@ -215,7 +215,7 @@ func warmUpHeartbeat(t *testing.T) tabletmanagerdatapb.CheckThrottlerResponseCod
 }
 
 // waitForThrottleCheckStatus waits for the tablet to return the provided HTTP code in a throttle check
-func waitForThrottleCheckStatus(t *testing.T, tablet *cluster.Vttablet, wantCode tabletmanagerdatapb.CheckThrottlerResponseCode) bool {
+func waitForThrottleCheckStatus(t *testing.T, tablet *cluster.Vttablet, wantCode tabletmanagerdatapb.CheckThrottlerResponseCode) (*tabletmanagerdatapb.CheckThrottlerResponse, bool) {
 	_ = warmUpHeartbeat(t)
 	ctx, cancel := context.WithTimeout(context.Background(), onDemandHeartbeatDuration*4)
 	defer cancel()
@@ -229,11 +229,11 @@ func waitForThrottleCheckStatus(t *testing.T, tablet *cluster.Vttablet, wantCode
 		if wantCode == resp.Check.ResponseCode {
 			// Wait for any cached check values to be cleared and the new
 			// status value to be in effect everywhere before returning.
-			return true
+			return resp.Check, true
 		}
 		select {
 		case <-ctx.Done():
-			return assert.EqualValues(t, wantCode, resp.Check.StatusCode, "response: %+v", resp)
+			return resp.Check, false
 		case <-ticker.C:
 		}
 	}
@@ -779,16 +779,16 @@ func TestUpdateAppCheckedMetrics(t *testing.T) {
 		}
 		waitForThrottleCheckStatus(t, primaryTablet, tabletmanagerdatapb.CheckThrottlerResponseCode_THRESHOLD_EXCEEDED)
 	})
-	t.Run("assigning 'loadavg' metrics to 'test' app", func(t *testing.T) {
+	t.Run("assigning 'threads_running' metrics to 'test' app", func(t *testing.T) {
 		{
-			req := &vtctldatapb.UpdateThrottlerConfigRequest{MetricName: "loadavg", Threshold: 7777}
+			req := &vtctldatapb.UpdateThrottlerConfigRequest{MetricName: base.ThreadsRunningMetricName.String(), Threshold: 7777}
 			_, err := throttler.UpdateThrottlerTopoConfig(clusterInstance, req, nil, nil)
 			assert.NoError(t, err)
 		}
 		{
 			req := &vtctldatapb.UpdateThrottlerConfigRequest{}
 			appCheckedMetrics := map[string]*topodatapb.ThrottlerConfig_MetricNames{
-				testAppName.String(): {Names: []string{"loadavg"}},
+				testAppName.String(): {Names: []string{base.ThreadsRunningMetricName.String()}},
 			}
 			_, err := throttler.UpdateThrottlerTopoConfig(clusterInstance, req, nil, appCheckedMetrics)
 			assert.NoError(t, err)
@@ -802,18 +802,18 @@ func TestUpdateAppCheckedMetrics(t *testing.T) {
 		for _, tablet := range []cluster.Vttablet{*primaryTablet, *replicaTablet} {
 			throttler.WaitForThrottlerStatusEnabled(t, &clusterInstance.VtctldClientProcess, &tablet, true, &throttler.Config{Query: throttler.DefaultQuery, Threshold: unreasonablyLowThreshold.Seconds()}, throttlerEnabledTimeout)
 		}
-		t.Run("validating OK response from throttler since it's checking loadavg", func(t *testing.T) {
-			if !waitForThrottleCheckStatus(t, primaryTablet, tabletmanagerdatapb.CheckThrottlerResponseCode_OK) {
+		t.Run("validating OK response from throttler since it's checking threads_running", func(t *testing.T) {
+			if _, ok := waitForThrottleCheckStatus(t, primaryTablet, tabletmanagerdatapb.CheckThrottlerResponseCode_OK); !ok {
 				t.Logf("throttler primary status: %+v", throttleStatus(t, primaryTablet))
 				t.Logf("throttler replica status: %+v", throttleStatus(t, replicaTablet))
 			}
 		})
 	})
-	t.Run("assigning 'loadavg,lag' metrics to 'test' app", func(t *testing.T) {
+	t.Run("assigning 'threads_running,lag' metrics to 'test' app", func(t *testing.T) {
 		{
 			req := &vtctldatapb.UpdateThrottlerConfigRequest{}
 			appCheckedMetrics := map[string]*topodatapb.ThrottlerConfig_MetricNames{
-				testAppName.String(): {Names: []string{"loadavg,lag"}},
+				testAppName.String(): {Names: []string{base.ThreadsRunningMetricName.String(), base.LagMetricName.String()}},
 			}
 			_, err := throttler.UpdateThrottlerTopoConfig(clusterInstance, req, nil, appCheckedMetrics)
 			assert.NoError(t, err)
@@ -831,9 +831,51 @@ func TestUpdateAppCheckedMetrics(t *testing.T) {
 			waitForThrottleCheckStatus(t, primaryTablet, tabletmanagerdatapb.CheckThrottlerResponseCode_THRESHOLD_EXCEEDED)
 		})
 	})
+	t.Run("assigning 'mysqld-loadavg,mysqld-datadir-used-ratio' metrics to 'test' app", func(t *testing.T) {
+		{
+			req := &vtctldatapb.UpdateThrottlerConfigRequest{MetricName: base.MysqldDatadirUsedRatioMetricName.String(), Threshold: 0.9999}
+			_, err := throttler.UpdateThrottlerTopoConfig(clusterInstance, req, nil, nil)
+			assert.NoError(t, err)
+		}
+		{
+			req := &vtctldatapb.UpdateThrottlerConfigRequest{MetricName: base.MysqldLoadAvgMetricName.String(), Threshold: 5555}
+			_, err := throttler.UpdateThrottlerTopoConfig(clusterInstance, req, nil, nil)
+			assert.NoError(t, err)
+		}
+		{
+			req := &vtctldatapb.UpdateThrottlerConfigRequest{}
+			appCheckedMetrics := map[string]*topodatapb.ThrottlerConfig_MetricNames{
+				testAppName.String(): {Names: []string{base.MysqldDatadirUsedRatioMetricName.String(), base.MysqldLoadAvgMetricName.String()}},
+			}
+			_, err := throttler.UpdateThrottlerTopoConfig(clusterInstance, req, nil, appCheckedMetrics)
+			assert.NoError(t, err)
+		}
+		{
+			req := &vtctldatapb.UpdateThrottlerConfigRequest{Threshold: extremelyHighThreshold.Seconds()}
+			_, err := throttler.UpdateThrottlerTopoConfig(clusterInstance, req, nil, nil)
+			assert.NoError(t, err)
+		}
+		// Wait for the throttler to be enabled everywhere with new config.
+		for _, tablet := range []cluster.Vttablet{*primaryTablet, *replicaTablet} {
+			throttler.WaitForThrottlerStatusEnabled(t, &clusterInstance.VtctldClientProcess, &tablet, true, &throttler.Config{Query: throttler.DefaultQuery, Threshold: extremelyHighThreshold.Seconds()}, throttlerEnabledTimeout)
+		}
+		t.Run("validating OK response from throttler since it's checking mysqld-loadavg,mysqld-datadir-used-ratio", func(t *testing.T) {
+			resp, ok := waitForThrottleCheckStatus(t, primaryTablet, tabletmanagerdatapb.CheckThrottlerResponseCode_OK)
+			if !ok {
+				t.Logf("response: %+v", resp)
+				t.Logf("throttler primary status: %+v", throttleStatus(t, primaryTablet))
+				t.Logf("throttler replica status: %+v", throttleStatus(t, replicaTablet))
+			}
+			require.Contains(t, resp.Metrics, base.MysqldDatadirUsedRatioMetricName.String())
+			require.Contains(t, resp.Metrics, base.MysqldLoadAvgMetricName.String())
+			assert.NotContains(t, resp.Metrics, base.ThreadsRunningMetricName.String())
+
+			assert.NotZero(t, resp.Metrics[base.MysqldDatadirUsedRatioMetricName.String()].Value)
+		})
+	})
 	t.Run("removing assignment from 'test' app and restoring defaults", func(t *testing.T) {
 		{
-			req := &vtctldatapb.UpdateThrottlerConfigRequest{MetricName: "loadavg", Threshold: 0}
+			req := &vtctldatapb.UpdateThrottlerConfigRequest{MetricName: base.ThreadsRunningMetricName.String(), Threshold: 0}
 			_, err := throttler.UpdateThrottlerTopoConfig(clusterInstance, req, nil, nil)
 			assert.NoError(t, err)
 		}

go/test/endtoend/throttler/util.go

@@ -529,7 +529,7 @@ func WaitForValidData(t *testing.T, tablet *cluster.Vttablet, timeout time.Durat
 	selfCheckURL := fmt.Sprintf("http://localhost:%d/throttler/check-self", tablet.HTTPPort)
 	ctx, cancel := context.WithTimeout(context.Background(), timeout)
 	defer cancel()
-	ticker := time.NewTicker(500 * time.Millisecond)
+	ticker := time.NewTicker(time.Second)
 	defer ticker.Stop()
 
 	for {
@@ -548,8 +548,10 @@ func WaitForValidData(t *testing.T, tablet *cluster.Vttablet, timeout time.Durat
 		}
 		select {
 		case <-ctx.Done():
-			t.Errorf("timed out waiting for %s tablet's throttler to return a valid result after %v; last seen value: %+v",
-				tablet.Alias, timeout, checkResp)
+			respByte, _ := io.ReadAll(checkResp.Body)
+			body := string(respByte)
+			require.Failf(t, "time out", "waiting for %s tablet's throttler to return a valid result after %v; last seen result: %+v",
+				tablet.Alias, timeout, body)
 			return
 		case <-ticker.C:
 		}

go/textutil/strings.go

@@ -90,6 +90,19 @@ func SingleWordCamel(w string) string {
 	return strings.ToUpper(w[0:1]) + strings.ToLower(w[1:])
 }
 
+var multiWordSplitterRegexp = regexp.MustCompile(`[-_.\s]+`)
+
+// SingleWordCamel takes a single word and returns is in Camel case; basically
+// just capitalizing the first letter and making sure the rest are lower case.
+func MultiWordCamel(w string) string {
+	var b strings.Builder
+	words := multiWordSplitterRegexp.Split(w, -1)
+	for _, word := range words {
+		b.WriteString(SingleWordCamel(word))
+	}
+	return b.String()
+}
+
 // ValueIsSimulatedNull returns true if the slice value represents
 // a NULL or unknown/unspecified value. This is used to distinguish
 // between a zero value empty slice and a user provided value of an

go/textutil/strings_test.go

@@ -122,6 +122,64 @@ func TestSingleWordCamel(t *testing.T) {
 	}
 }
 
+func TestMultiWordCamel(t *testing.T) {
+	tt := []struct {
+		word   string
+		expect string
+	}{
+		{
+			word:   "",
+			expect: "",
+		},
+		{
+			word:   "_",
+			expect: "",
+		},
+		{
+			word:   "_a",
+			expect: "A",
+		},
+		{
+			word:   "A",
+			expect: "A",
+		},
+		{
+			word:   "mysql",
+			expect: "Mysql",
+		},
+		{
+			word:   "mySQL",
+			expect: "Mysql",
+		},
+		{
+			word:   "foo-bar",
+			expect: "FooBar",
+		},
+		{
+			word:   "mysql-server",
+			expect: "MysqlServer",
+		},
+		{
+			word:   "io_util",
+			expect: "IoUtil",
+		},
+		{
+			word:   "there and back again",
+			expect: "ThereAndBackAgain",
+		},
+		{
+			word:   "combine_all_OF the\tabove",
+			expect: "CombineAllOfTheAbove",
+		},
+	}
+	for _, tc := range tt {
+		t.Run(tc.word, func(t *testing.T) {
+			camel := MultiWordCamel(tc.word)
+			assert.Equal(t, tc.expect, camel)
+		})
+	}
+}
+
 func TestValueIsSimulatedNull(t *testing.T) {
 	tt := []struct {
 		name   string

go/vt/mysqlctl/mysqld.go

@@ -1333,7 +1333,7 @@ func hostMetrics(ctx context.Context, cnf *Mycnf) (*mysqlctlpb.HostMetricsRespon
 	_ = func() error {
 		metric := newMetric("loadavg")
 		if runtime.GOOS != "linux" {
-			return withError(metric, fmt.Errorf("loadavg metric is only available on Linux"))
+			return nil
 		}
 		content, err := os.ReadFile("/proc/loadavg")
 		if err != nil {

go/vt/vttablet/tabletserver/throttle/base/metric_cache.go

@@ -49,6 +49,7 @@ import (
 	"github.com/patrickmn/go-cache"
 
 	"vitess.io/vitess/go/stats"
+	"vitess.io/vitess/go/vt/vttablet/tmclient"
 )
 
 // MetricsQueryType indicates the type of metrics query on MySQL backend. See following.
@@ -142,13 +143,13 @@ func (metric *ThrottleMetric) WithError(err error) *ThrottleMetric {
 
 // ReadThrottleMetrics returns a metric for the given probe. Either by explicit query
 // or via SHOW REPLICA STATUS
-func ReadThrottleMetrics(ctx context.Context, probe *Probe, metricsFunc func(context.Context) ThrottleMetrics) ThrottleMetrics {
+func ReadThrottleMetrics(ctx context.Context, probe *Probe, tmClient tmclient.TabletManagerClient, metricsFunc func(context.Context, tmclient.TabletManagerClient) ThrottleMetrics) ThrottleMetrics {
 	if metrics := getCachedThrottleMetrics(probe); metrics != nil {
 		return metrics
 	}
 
 	started := time.Now()
-	throttleMetrics := metricsFunc(ctx)
+	throttleMetrics := metricsFunc(ctx, tmClient)
 
 	go func(metrics ThrottleMetrics, started time.Time) {
 		stats.GetOrNewGauge("ThrottlerProbesLatency", "probes latency").Set(time.Since(started).Nanoseconds())

go/vt/vttablet/tabletserver/throttle/base/metric_name.go

@@ -60,11 +60,13 @@ func (names MetricNames) Unique() MetricNames {
 }
 
 const (
-	DefaultMetricName        MetricName = "default"
-	LagMetricName            MetricName = "lag"
-	ThreadsRunningMetricName MetricName = "threads_running"
-	CustomMetricName         MetricName = "custom"
-	LoadAvgMetricName        MetricName = "loadavg"
+	DefaultMetricName                MetricName = "default"
+	LagMetricName                    MetricName = "lag"
+	ThreadsRunningMetricName         MetricName = "threads_running"
+	CustomMetricName                 MetricName = "custom"
+	LoadAvgMetricName                MetricName = "loadavg"
+	MysqldLoadAvgMetricName          MetricName = "mysqld-loadavg"
+	MysqldDatadirUsedRatioMetricName MetricName = "mysqld-datadir-used-ratio"
 )
 
 func (metric MetricName) DefaultScope() Scope {

go/vt/vttablet/tabletserver/throttle/base/metric_name_test.go

@@ -21,6 +21,9 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"vitess.io/vitess/go/textutil"
 )
 
 func TestAggregateName(t *testing.T) {
@@ -238,4 +241,25 @@ func TestKnownMetricNames(t *testing.T) {
 	assert.Contains(t, KnownMetricNames, LoadAvgMetricName)
 	assert.Contains(t, KnownMetricNames, CustomMetricName)
 	assert.Contains(t, KnownMetricNames, DefaultMetricName)
+	assert.Contains(t, KnownMetricNames, MysqldLoadAvgMetricName)
+	assert.Contains(t, KnownMetricNames, MysqldDatadirUsedRatioMetricName)
+}
+
+func TestSingleWordCamelKnownMetricNames(t *testing.T) {
+	expectCases := map[MetricName]string{
+		LagMetricName:                    "Lag",
+		ThreadsRunningMetricName:         "ThreadsRunning",
+		LoadAvgMetricName:                "Loadavg",
+		CustomMetricName:                 "Custom",
+		DefaultMetricName:                "Default",
+		MysqldLoadAvgMetricName:          "MysqldLoadavg",
+		MysqldDatadirUsedRatioMetricName: "MysqldDatadirUsedRatio",
+	}
+	for _, metricName := range KnownMetricNames {
+		t.Run(metricName.String(), func(t *testing.T) {
+			expect, ok := expectCases[metricName]
+			require.True(t, ok)
+			assert.Equal(t, expect, textutil.MultiWordCamel(metricName.String()))
+		})
+	}
 }

go/vt/vttablet/tabletserver/throttle/base/self_metric.go

@@ -21,15 +21,24 @@ import (
 	"fmt"
 	"strconv"
 
+	"vitess.io/vitess/go/vt/topo"
 	"vitess.io/vitess/go/vt/vttablet/tabletserver/connpool"
+	"vitess.io/vitess/go/vt/vttablet/tmclient"
 )
 
+type SelfMetricReadParams struct {
+	Throttler  metricsPublisher
+	Conn       *connpool.Conn
+	TmClient   tmclient.TabletManagerClient
+	TabletInfo *topo.TabletInfo
+}
+
 type SelfMetric interface {
 	Name() MetricName
 	DefaultScope() Scope
 	DefaultThreshold() float64
 	RequiresConn() bool
-	Read(ctx context.Context, throttler ThrottlerMetricsPublisher, conn *connpool.Conn) *ThrottleMetric
+	Read(ctx context.Context, params *SelfMetricReadParams) *ThrottleMetric
 }
 
 var (

go/vt/vttablet/tabletserver/throttle/base/self_metric_custom_query.go

@@ -18,8 +18,6 @@ package base
 
 import (
 	"context"
-
-	"vitess.io/vitess/go/vt/vttablet/tabletserver/connpool"
 )
 
 var _ SelfMetric = registerSelfMetric(&CustomQuerySelfMetric{})
@@ -43,6 +41,6 @@ func (m *CustomQuerySelfMetric) RequiresConn() bool {
 	return true
 }
 
-func (m *CustomQuerySelfMetric) Read(ctx context.Context, throttler ThrottlerMetricsPublisher, conn *connpool.Conn) *ThrottleMetric {
-	return ReadSelfMySQLThrottleMetric(ctx, conn, throttler.GetCustomMetricsQuery())
+func (m *CustomQuerySelfMetric) Read(ctx context.Context, params *SelfMetricReadParams) *ThrottleMetric {
+	return ReadSelfMySQLThrottleMetric(ctx, params.Conn, params.Throttler.GetCustomMetricsQuery())
 }

go/vt/vttablet/tabletserver/throttle/base/self_metric_default.go

@@ -19,8 +19,6 @@ package base
 import (
 	"context"
 	"fmt"
-
-	"vitess.io/vitess/go/vt/vttablet/tabletserver/connpool"
 )
 
 var _ SelfMetric = registerSelfMetric(&DefaultSelfMetric{})
@@ -44,7 +42,7 @@ func (m *DefaultSelfMetric) RequiresConn() bool {
 	return false
 }
 
-func (m *DefaultSelfMetric) Read(ctx context.Context, throttler ThrottlerMetricsPublisher, conn *connpool.Conn) *ThrottleMetric {
+func (m *DefaultSelfMetric) Read(ctx context.Context, params *SelfMetricReadParams) *ThrottleMetric {
 	return &ThrottleMetric{
 		Err: fmt.Errorf("unexpected direct call to DefaultSelfMetric.Read"),
 	}

go/vt/vttablet/tabletserver/throttle/base/self_metric_lag.go

@@ -23,7 +23,6 @@ import (
 
 	"vitess.io/vitess/go/constants/sidecar"
 	"vitess.io/vitess/go/vt/sqlparser"
-	"vitess.io/vitess/go/vt/vttablet/tabletserver/connpool"
 )
 
 var (
@@ -65,6 +64,6 @@ func (m *LagSelfMetric) RequiresConn() bool {
 	return true
 }
 
-func (m *LagSelfMetric) Read(ctx context.Context, throttler ThrottlerMetricsPublisher, conn *connpool.Conn) *ThrottleMetric {
-	return ReadSelfMySQLThrottleMetric(ctx, conn, m.GetQuery())
+func (m *LagSelfMetric) Read(ctx context.Context, params *SelfMetricReadParams) *ThrottleMetric {
+	return ReadSelfMySQLThrottleMetric(ctx, params.Conn, m.GetQuery())
 }