Throttler: return app name in check result, synthesize "why throttled" explanation from result

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

@@ -397,6 +397,8 @@ func TestSchemaChange(t *testing.T) {
 		assert.GreaterOrEqual(t, lastThrottledTimestamp, startedTimestamp)
 		component := row.AsString("component_throttled", "")
 		assert.Contains(t, []string{throttlerapp.VCopierName.String(), throttlerapp.VPlayerName.String()}, component)
+		reason := row.AsString("reason_throttled", "")
+		assert.Contains(t, reason, "is explicitly denied access")
 
 		// unthrottle
 		onlineddl.UnthrottleAllMigrations(t, &vtParams)

go/vt/binlog/binlogplayer/binlog_player.go

@@ -669,11 +669,11 @@ func GenerateUpdateHeartbeat(uid int32, timeUpdated int64) (string, error) {
 }
 
 // GenerateUpdateTimeThrottled returns a statement to record the latest throttle time in the _vt.vreplication table.
-func GenerateUpdateTimeThrottled(uid int32, timeThrottledUnix int64, componentThrottled string) (string, error) {
+func GenerateUpdateTimeThrottled(uid int32, timeThrottledUnix int64, componentThrottled string, reasonThrottled string) (string, error) {
 	if timeThrottledUnix == 0 {
 		return "", fmt.Errorf("timeUpdated cannot be zero")
 	}
-	return fmt.Sprintf("update _vt.vreplication set time_updated=%v, time_throttled=%v, component_throttled='%v' where id=%v", timeThrottledUnix, timeThrottledUnix, componentThrottled, uid), nil
+	return fmt.Sprintf("update _vt.vreplication set time_updated=%v, time_throttled=%v, component_throttled='%v', reason_throttled=%v where id=%v", timeThrottledUnix, timeThrottledUnix, componentThrottled, encodeString(MessageTruncate(reasonThrottled)), uid), nil
 }
 
 // StartVReplicationUntil returns a statement to start the replication with a stop position.

go/vt/binlog/binlogplayer/binlog_player_test.go

@@ -22,6 +22,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
+
 	"vitess.io/vitess/go/mysql/replication"
 	"vitess.io/vitess/go/mysql/sqlerror"
 	querypb "vitess.io/vitess/go/vt/proto/query"
@@ -454,3 +456,36 @@ func TestReadVReplicationStatus(t *testing.T) {
 		t.Errorf("ReadVReplicationStatus(482821) = %#v, want %#v", got, want)
 	}
 }
+
+func TestEncodeString(t *testing.T) {
+	tcases := []struct {
+		in, out string
+	}{
+		{
+			in:  "",
+			out: "''",
+		},
+		{
+			in:  "a",
+			out: "'a'",
+		},
+		{
+			in:  "here's",
+			out: "'here\\'s'",
+		},
+		{
+			in:  "online-ddl is denied access due to lag metric value 94.821447 exceeding threshold 5",
+			out: "'online-ddl is denied access due to lag metric value 94.821447 exceeding threshold 5'",
+		},
+		{
+			in:  "'a','b','c'",
+			out: "'\\'a\\',\\'b\\',\\'c\\''",
+		},
+	}
+	for _, tcase := range tcases {
+		t.Run(tcase.in, func(t *testing.T) {
+			out := encodeString(tcase.in)
+			assert.Equal(t, tcase.out, out)
+		})
+	}
+}