This is an automated cherry-pick of pingcap#39932

Signed-off-by: ti-chi-bot <ti-community-prow-bot@tidb.io>
ti-chi-bot · Dec 22, 2022 · 97354fb · 97354fb
1 parent 1164063
commit 97354fb
Show file tree

Hide file tree

Showing 17 changed files with 571 additions and 85 deletions.
diff --git a/kv/mpp.go b/kv/mpp.go
@@ -16,7 +16,6 @@ package kv
 
 import (
 	"context"
-	"sync"
 	"time"
 
 	"github.com/pingcap/kvproto/pkg/mpp"
@@ -81,7 +80,7 @@ type MPPDispatchRequest struct {
 type MPPClient interface {
 	// ConstructMPPTasks schedules task for a plan fragment.
 	// TODO:: This interface will be refined after we support more executors.
-	ConstructMPPTasks(context.Context, *MPPBuildTasksRequest, *sync.Map, time.Duration) ([]MPPTaskMeta, error)
+	ConstructMPPTasks(context.Context, *MPPBuildTasksRequest, time.Duration) ([]MPPTaskMeta, error)
 
 	// DispatchMPPTasks dispatches ALL mpp requests at once, and returns an iterator that transfers the data.
 	DispatchMPPTasks(ctx context.Context, vars interface{}, reqs []*MPPDispatchRequest, needTriggerFallback bool, startTs uint64) Response

diff --git a/metrics/metrics.go b/metrics/metrics.go
@@ -182,6 +182,7 @@ func RegisterMetrics() {
 	prometheus.MustRegister(TokenGauge)
 	prometheus.MustRegister(ConfigStatus)
 	prometheus.MustRegister(TiFlashQueryTotalCounter)
+	prometheus.MustRegister(TiFlashFailedMPPStoreState)
 	prometheus.MustRegister(SmallTxnWriteDuration)
 	prometheus.MustRegister(TxnWriteThroughput)
 	prometheus.MustRegister(LoadSysVarCacheCounter)
@@ -236,6 +237,7 @@ func ToggleSimplifiedMode(simplified bool) {
 		InfoCacheCounters,
 		ReadFromTableCacheCounter,
 		TiFlashQueryTotalCounter,
+		TiFlashFailedMPPStoreState,
 		CampaignOwnerCounter,
 		NonTransactionalDMLCount,
 		MemoryUsage,

diff --git a/metrics/server.go b/metrics/server.go
@@ -279,6 +279,14 @@ var (
 			Help:      "Counter of TiFlash queries.",
 		}, []string{LblType, LblResult})
 
+	TiFlashFailedMPPStoreState = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Namespace: "tidb",
+			Subsystem: "server",
+			Name:      "tiflash_failed_store",
+			Help:      "Statues of failed tiflash mpp store,-1 means detector heartbeat,0 means reachable,1 means abnormal.",
+		}, []string{LblAddress})
+
 	PDAPIExecutionHistogram = prometheus.NewHistogramVec(
 		prometheus.HistogramOpts{
 			Namespace: "tidb",

diff --git a/planner/core/fragment.go b/planner/core/fragment.go
@@ -378,7 +378,7 @@ func (e *mppTaskGenerator) constructMPPTasksImpl(ctx context.Context, ts *Physic
 		logutil.BgLogger().Warn("MPP store fail ttl is invalid", zap.Error(err))
 		ttl = 30 * time.Second
 	}
-	metas, err := e.ctx.GetMPPClient().ConstructMPPTasks(ctx, req, e.ctx.GetSessionVars().MPPStoreLastFailTime, ttl)
+	metas, err := e.ctx.GetMPPClient().ConstructMPPTasks(ctx, req, ttl)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}

diff --git a/sessionctx/sessionstates/session_states.go b/sessionctx/sessionstates/session_states.go
@@ -15,8 +15,6 @@
 package sessionstates
 
 import (
-	"time"
-
 	"github.com/pingcap/tidb/errno"
 	ptypes "github.com/pingcap/tidb/parser/types"
 	"github.com/pingcap/tidb/sessionctx/stmtctx"
@@ -79,7 +77,6 @@ type SessionStates struct {
 	FoundInPlanCache     bool                         `json:"in-plan-cache,omitempty"`
 	FoundInBinding       bool                         `json:"in-binding,omitempty"`
 	SequenceLatestValues map[int64]int64              `json:"seq-values,omitempty"`
-	MPPStoreLastFailTime map[string]time.Time         `json:"store-fail-time,omitempty"`
 	LastAffectedRows     int64                        `json:"affected-rows,omitempty"`
 	LastInsertID         uint64                       `json:"last-insert-id,omitempty"`
 	Warnings             []stmtctx.SQLWarn            `json:"warnings,omitempty"`

diff --git a/sessionctx/sessionstates/session_states_test.go b/sessionctx/sessionstates/session_states_test.go
@@ -20,9 +20,7 @@ import (
 	"fmt"
 	"strconv"
 	"strings"
-	"sync"
 	"testing"
-	"time"
 
 	"github.com/pingcap/errors"
 	"github.com/pingcap/tidb/config"
@@ -378,23 +376,6 @@ func TestSessionCtx(t *testing.T) {
 				tk.MustQuery("select nextval(test.s)").Check(testkit.Rows("2"))
 			},
 		},
-		{
-			// check MPPStoreLastFailTime
-			setFunc: func(tk *testkit.TestKit) any {
-				m := sync.Map{}
-				m.Store("store1", time.Now())
-				tk.Session().GetSessionVars().MPPStoreLastFailTime = &m
-				return tk.Session().GetSessionVars().MPPStoreLastFailTime
-			},
-			checkFunc: func(tk *testkit.TestKit, param any) {
-				failTime := tk.Session().GetSessionVars().MPPStoreLastFailTime
-				tm, ok := failTime.Load("store1")
-				require.True(t, ok)
-				v, ok := (param.(*sync.Map)).Load("store1")
-				require.True(t, ok)
-				require.True(t, tm.(time.Time).Equal(v.(time.Time)))
-			},
-		},
 		{
 			// check FoundInPlanCache
 			setFunc: func(tk *testkit.TestKit) any {

diff --git a/sessionctx/variable/session.go b/sessionctx/variable/session.go
@@ -1170,9 +1170,6 @@ type SessionVars struct {
 	// TemporaryTableData stores committed kv values for temporary table for current session.
 	TemporaryTableData TemporaryTableData
 
-	// MPPStoreLastFailTime records the lastest fail time that a TiFlash store failed. It maps store address(string) to fail time(time.Time).
-	MPPStoreLastFailTime *sync.Map
-
 	// MPPStoreFailTTL indicates the duration that protect TiDB from sending task to a new recovered TiFlash.
 	MPPStoreFailTTL string
 
@@ -1660,7 +1657,6 @@ func NewSessionVars(hctx HookContext) *SessionVars {
 		AllowFallbackToTiKV:           make(map[kv.StoreType]struct{}),
 		CTEMaxRecursionDepth:          DefCTEMaxRecursionDepth,
 		TMPTableSize:                  DefTiDBTmpTableMaxSize,
-		MPPStoreLastFailTime:          new(sync.Map),
 		MPPStoreFailTTL:               DefTiDBMPPStoreFailTTL,
 		Rng:                           mathutil.NewWithTime(),
 		StatsLoadSyncWait:             StatsLoadSyncWait.Load(),
@@ -2287,12 +2283,6 @@ func (s *SessionVars) EncodeSessionStates(ctx context.Context, sessionStates *se
 	}
 	sessionStates.LastFoundRows = s.LastFoundRows
 	sessionStates.SequenceLatestValues = s.SequenceState.GetAllStates()
-	sessionStates.MPPStoreLastFailTime = make(map[string]time.Time, 0)
-	s.MPPStoreLastFailTime.Range(
-		func(key, value interface{}) bool {
-			sessionStates.MPPStoreLastFailTime[key.(string)] = value.(time.Time)
-			return true
-		})
 	sessionStates.FoundInPlanCache = s.PrevFoundInPlanCache
 	sessionStates.FoundInBinding = s.PrevFoundInBinding
 
@@ -2328,9 +2318,6 @@ func (s *SessionVars) DecodeSessionStates(ctx context.Context, sessionStates *se
 	}
 	s.LastFoundRows = sessionStates.LastFoundRows
 	s.SequenceState.SetAllStates(sessionStates.SequenceLatestValues)
-	for k, v := range sessionStates.MPPStoreLastFailTime {
-		s.MPPStoreLastFailTime.Store(k, v)
-	}
 	s.FoundInPlanCache = sessionStates.FoundInPlanCache
 	s.FoundInBinding = sessionStates.FoundInBinding
 

diff --git a/store/copr/BUILD.bazel b/store/copr/BUILD.bazel
@@ -9,6 +9,7 @@ go_library(
         "coprocessor_cache.go",
         "key_ranges.go",
         "mpp.go",
+        "mpp_probe.go",
         "region_cache.go",
         "store.go",
     ],
@@ -65,6 +66,7 @@ go_test(
         "coprocessor_test.go",
         "key_ranges_test.go",
         "main_test.go",
+        "mpp_probe_test.go",
     ],
     embed = [":copr"],
     flaky = True,
@@ -74,11 +76,18 @@ go_test(
         "//store/driver/backoff",
         "//testkit/testsetup",
         "//util/paging",
+        "@com_github_pingcap_errors//:errors",
         "@com_github_pingcap_kvproto//pkg/coprocessor",
+<<<<<<< HEAD
+=======
+        "@com_github_pingcap_kvproto//pkg/mpp",
+        "@com_github_stathat_consistent//:consistent",
+>>>>>>> aeccf77637 (*: optimize mpp probe (#39932))
         "@com_github_stretchr_testify//require",
         "@com_github_tikv_client_go_v2//config",
         "@com_github_tikv_client_go_v2//testutils",
         "@com_github_tikv_client_go_v2//tikv",
+        "@com_github_tikv_client_go_v2//tikvrpc",
         "@org_uber_go_goleak//:goleak",
     ],
 )