keyspace metrics

Signed-off-by: 童剑 <1045931706@qq.com>

Author: bufferflies

client/client.go

@@ -248,6 +248,7 @@ func createClientWithKeyspace(
 		callerComponent: adjustCallerComponent(callerComponent),
 		inner: &innerClient{
 			keyspaceID:              keyspaceID,
+			keyspaceName:            constants.NullKeyspaceName,
 			svrUrls:                 svrAddrs,
 			updateTokenConnectionCh: make(chan struct{}, 1),
 			ctx:                     clientCtx,
@@ -262,7 +263,21 @@ func createClientWithKeyspace(
 		opt(c.inner.option)
 	}
 
-	return c, c.inner.init(nil)
+	updateKeyspaceNameFunc := func() error {
+		metas, err := c.GetAllKeyspaces(clientCtx, keyspaceID, 1)
+		if err != nil {
+			return err
+		}
+		for _, m := range metas {
+			if m.GetId() == keyspaceID {
+				c.inner.keyspaceName = m.GetName()
+				break
+			}
+		}
+		return nil
+	}
+
+	return c, c.inner.init(updateKeyspaceNameFunc)
 }
 
 // APIVersion is the API version the server and the client is using.
@@ -368,6 +383,7 @@ func newClientWithKeyspaceName(
 			// Create a service discovery with null keyspace id, then query the real id with the keyspace name,
 			// finally update the keyspace id to the service discovery for the following interactions.
 			keyspaceID:              constants.NullKeyspaceID,
+			keyspaceName:            keyspaceName,
 			updateTokenConnectionCh: make(chan struct{}, 1),
 			ctx:                     clientCtx,
 			cancel:                  clientCancel,

client/clients/tso/client.go

@@ -88,13 +88,14 @@ type Cli struct {
 	// tsoReqPool is the pool to recycle `*tsoRequest`.
 	tsoReqPool *sync.Pool
 	// dispatcher is used to dispatch the TSO requests to the channel.
-	dispatcher atomic.Pointer[tsoDispatcher]
+	dispatcher   atomic.Pointer[tsoDispatcher]
+	keyspaceName string
 }
 
 // NewClient returns a new TSO client.
 func NewClient(
 	ctx context.Context, option *opt.Option,
-	svcDiscovery sd.ServiceDiscovery, factory tsoStreamBuilderFactory,
+	svcDiscovery sd.ServiceDiscovery, factory tsoStreamBuilderFactory, keyspaceName string,
 ) *Cli {
 	ctx, cancel := context.WithCancel(ctx)
 	c := &Cli{
@@ -114,6 +115,7 @@ func NewClient(
 				}
 			},
 		},
+		keyspaceName: keyspaceName,
 	}
 
 	c.svcDiscovery.ExecAndAddLeaderSwitchedCallback(c.updateTSOLeaderURL)
@@ -338,7 +340,7 @@ func (c *Cli) tryConnectToTSO(ctx context.Context) error {
 		}
 		if cc != nil {
 			cctx, cancel := context.WithCancel(ctx)
-			stream, err = c.tsoStreamBuilderFactory.makeBuilder(cc).build(cctx, cancel, c.option.Timeout)
+			stream, err = c.tsoStreamBuilderFactory.makeBuilder(cc, c.keyspaceName).build(cctx, cancel, c.option.Timeout)
 			failpoint.Inject("unreachableNetwork", func() {
 				stream = nil
 				err = status.New(codes.Unavailable, "unavailable").Err()
@@ -382,7 +384,7 @@ func (c *Cli) tryConnectToTSO(ctx context.Context) error {
 			// create the follower stream
 			cctx, cancel := context.WithCancel(ctx)
 			cctx = grpcutil.BuildForwardContext(cctx, forwardedHost)
-			stream, err = c.tsoStreamBuilderFactory.makeBuilder(backupClientConn).build(cctx, cancel, c.option.Timeout)
+			stream, err = c.tsoStreamBuilderFactory.makeBuilder(backupClientConn, c.keyspaceName).build(cctx, cancel, c.option.Timeout)
 			if err == nil {
 				forwardedHostTrim := tlsutil.TrimHTTPPrefix(forwardedHost)
 				addr := tlsutil.TrimHTTPPrefix(backupURL)
@@ -431,7 +433,7 @@ func (c *Cli) checkLeader(
 			if err == nil && resp.GetStatus() == healthpb.HealthCheckResponse_SERVING {
 				// create a stream of the original tso leader
 				cctx, cancel := context.WithCancel(ctx)
-				stream, err := c.tsoStreamBuilderFactory.makeBuilder(cc).build(cctx, cancel, c.option.Timeout)
+				stream, err := c.tsoStreamBuilderFactory.makeBuilder(cc, c.keyspaceName).build(cctx, cancel, c.option.Timeout)
 				if err == nil && stream != nil {
 					log.Info("[tso] recover the original tso stream since the network has become normal", zap.String("url", url))
 					c.conCtxMgr.CleanAllAndStore(cctx, cancel, url, stream)
@@ -518,7 +520,7 @@ func (c *Cli) getAllTSOStreamBuilders() map[string]tsoStreamBuilder {
 		resp, err := healthpb.NewHealthClient(cc).Check(healthCtx, &healthpb.HealthCheckRequest{Service: ""})
 		healthCancel()
 		if err == nil && resp.GetStatus() == healthpb.HealthCheckResponse_SERVING {
-			streamBuilders[addr] = c.makeBuilder(cc)
+			streamBuilders[addr] = c.makeBuilder(cc, c.keyspaceName)
 		}
 	}
 	return streamBuilders

client/clients/tso/dispatcher_test.go

@@ -30,6 +30,7 @@ import (
 	"github.com/pingcap/failpoint"
 	"github.com/pingcap/log"
 
+	"github.com/tikv/pd/client/constants"
 	"github.com/tikv/pd/client/opt"
 	cctx "github.com/tikv/pd/client/pkg/connectionctx"
 	"github.com/tikv/pd/client/pkg/utils/testutil"
@@ -73,7 +74,7 @@ func (m *mockTSOServiceProvider) updateConnectionCtxs(ctx context.Context) bool
 	cctx, cancel := context.WithCancel(ctx)
 	var stream *tsoStream
 	if m.createStream == nil {
-		stream = newTSOStream(cctx, mockStreamURL, newMockTSOStreamImpl(ctx, resultModeGenerated))
+		stream = newTSOStream(cctx, mockStreamURL, newMockTSOStreamImpl(ctx, resultModeGenerated), constants.DefaultKeyspaceName)
 	} else {
 		stream = m.createStream(ctx)
 	}
@@ -104,7 +105,7 @@ func (s *testTSODispatcherSuite) SetupTest() {
 	created := new(atomic.Bool)
 	createStream := func(ctx context.Context) *tsoStream {
 		s.streamInner = newMockTSOStreamImpl(ctx, resultModeGenerateOnSignal)
-		s.stream = newTSOStream(ctx, mockStreamURL, s.streamInner)
+		s.stream = newTSOStream(ctx, mockStreamURL, s.streamInner, constants.DefaultKeyspaceName)
 		created.Store(true)
 		return s.stream
 	}

client/clients/tso/stream.go

@@ -41,21 +41,21 @@ import (
 // TSO Stream Builder Factory
 
 type tsoStreamBuilderFactory interface {
-	makeBuilder(cc *grpc.ClientConn) tsoStreamBuilder
+	makeBuilder(cc *grpc.ClientConn, keyspaceName string) tsoStreamBuilder
 }
 
 // PDStreamBuilderFactory is a factory for building TSO streams to the PD cluster.
 type PDStreamBuilderFactory struct{}
 
-func (*PDStreamBuilderFactory) makeBuilder(cc *grpc.ClientConn) tsoStreamBuilder {
-	return &pdStreamBuilder{client: pdpb.NewPDClient(cc), serverURL: cc.Target()}
+func (*PDStreamBuilderFactory) makeBuilder(cc *grpc.ClientConn, keyspaceName string) tsoStreamBuilder {
+	return &pdStreamBuilder{client: pdpb.NewPDClient(cc), serverURL: cc.Target(), keyspaceName: keyspaceName}
 }
 
 // MSStreamBuilderFactory is a factory for building TSO streams to the microservice cluster.
 type MSStreamBuilderFactory struct{}
 
-func (*MSStreamBuilderFactory) makeBuilder(cc *grpc.ClientConn) tsoStreamBuilder {
-	return &msStreamBuilder{client: tsopb.NewTSOClient(cc), serverURL: cc.Target()}
+func (*MSStreamBuilderFactory) makeBuilder(cc *grpc.ClientConn, keyspaceName string) tsoStreamBuilder {
+	return &msStreamBuilder{client: tsopb.NewTSOClient(cc), serverURL: cc.Target(), keyspaceName: keyspaceName}
 }
 
 // TSO Stream Builder
@@ -65,8 +65,9 @@ type tsoStreamBuilder interface {
 }
 
 type pdStreamBuilder struct {
-	serverURL string
-	client    pdpb.PDClient
+	serverURL    string
+	keyspaceName string
+	client       pdpb.PDClient
 }
 
 func (b *pdStreamBuilder) build(ctx context.Context, cancel context.CancelFunc, timeout time.Duration) (*tsoStream, error) {
@@ -76,14 +77,15 @@ func (b *pdStreamBuilder) build(ctx context.Context, cancel context.CancelFunc,
 	stream, err := b.client.Tso(ctx)
 	done <- struct{}{}
 	if err == nil {
-		return newTSOStream(ctx, b.serverURL, pdTSOStreamAdapter{stream}), nil
+		return newTSOStream(ctx, b.serverURL, pdTSOStreamAdapter{stream}, b.keyspaceName), nil
 	}
 	return nil, err
 }
 
 type msStreamBuilder struct {
-	serverURL string
-	client    tsopb.TSOClient
+	serverURL    string
+	client       tsopb.TSOClient
+	keyspaceName string
 }
 
 func (b *msStreamBuilder) build(
@@ -95,7 +97,7 @@ func (b *msStreamBuilder) build(
 	stream, err := b.client.Tso(ctx)
 	done <- struct{}{}
 	if err == nil {
-		return newTSOStream(ctx, b.serverURL, tsoTSOStreamAdapter{stream}), nil
+		return newTSOStream(ctx, b.serverURL, tsoTSOStreamAdapter{stream}, b.keyspaceName), nil
 	}
 	return nil, err
 }
@@ -220,6 +222,8 @@ type tsoStream struct {
 
 	ongoingRequestCountGauge prometheus.Gauge
 	ongoingRequests          atomic.Int32
+
+	keyspaceName string
 }
 
 const (
@@ -235,7 +239,7 @@ const (
 	maxPendingRequestsInTSOStream = 64
 )
 
-func newTSOStream(ctx context.Context, serverURL string, stream grpcTSOStreamAdapter) *tsoStream {
+func newTSOStream(ctx context.Context, serverURL string, stream grpcTSOStreamAdapter, keyspaceName string) *tsoStream {
 	streamID := fmt.Sprintf("%s-%d", serverURL, streamIDAlloc.Add(1))
 	// To make error handling in `tsoDispatcher` work, the internal `cancel` and external `cancel` is better to be
 	// distinguished.
@@ -250,6 +254,7 @@ func newTSOStream(ctx context.Context, serverURL string, stream grpcTSOStreamAda
 		cancel: cancel,
 
 		ongoingRequestCountGauge: metrics.OngoingRequestCountGauge.WithLabelValues(streamID),
+		keyspaceName:             keyspaceName,
 	}
 	s.wg.Add(1)
 	go s.recvLoop(ctx)
@@ -390,6 +395,13 @@ func (s *tsoStream) recvLoop(ctx context.Context) {
 		// Update the metrics in seconds.
 		metrics.EstimateTSOLatencyGauge.WithLabelValues(s.streamID).Set(micros * 1e-6)
 	}
+	successObserver := sync.OnceValue(func() prometheus.Observer {
+		return metrics.RequestDuration.WithLabelValues("tso", s.keyspaceName)
+	})
+
+	failedObserver := sync.OnceValue(func() prometheus.Observer {
+		return metrics.RequestDuration.WithLabelValues("tso-failed", s.keyspaceName)
+	})
 
 recvLoop:
 	for {
@@ -401,7 +413,6 @@ recvLoop:
 		}
 
 		res, err := s.stream.Recv()
-
 		// Try to load the corresponding `batchedRequests`. If `Recv` is successful, there must be a request pending
 		// in the queue.
 		select {
@@ -419,7 +430,7 @@ recvLoop:
 			// Note that it's also possible that the stream is broken due to network without being requested. In this
 			// case, `Recv` may return an error while no request is pending.
 			if hasReq {
-				metrics.RequestFailedDurationTSO.Observe(latencySeconds)
+				successObserver().Observe(latencySeconds)
 			}
 			if err == io.EOF {
 				finishWithErr = errors.WithStack(errs.ErrClientTSOStreamClosed)
@@ -431,8 +442,7 @@ recvLoop:
 			finishWithErr = errors.New("tsoStream timing order broken")
 			break recvLoop
 		}
-
-		metrics.RequestDurationTSO.Observe(latencySeconds)
+		failedObserver().Observe(latencySeconds)
 		metrics.TSOBatchSize.Observe(float64(res.count))
 		updateEstimatedLatency(currentReq.startTime, latency)
 

client/clients/tso/stream_test.go

@@ -28,6 +28,7 @@ import (
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
 
+	"github.com/tikv/pd/client/constants"
 	"github.com/tikv/pd/client/errs"
 )
 
@@ -54,11 +55,12 @@ const (
 )
 
 type mockTSOStreamImpl struct {
-	ctx        context.Context
-	requestCh  chan requestMsg
-	resultCh   chan resultMsg
-	keyspaceID uint32
-	errorState error
+	ctx          context.Context
+	requestCh    chan requestMsg
+	resultCh     chan resultMsg
+	keyspaceID   uint32
+	keyspaceName string
+	errorState   error
 
 	resultMode resultMode
 	// Current progress of generating TSO results
@@ -67,10 +69,11 @@ type mockTSOStreamImpl struct {
 
 func newMockTSOStreamImpl(ctx context.Context, resultMode resultMode) *mockTSOStreamImpl {
 	return &mockTSOStreamImpl{
-		ctx:        ctx,
-		requestCh:  make(chan requestMsg, 64),
-		resultCh:   make(chan resultMsg, 64),
-		keyspaceID: 0,
+		ctx:          ctx,
+		requestCh:    make(chan requestMsg, 64),
+		resultCh:     make(chan resultMsg, 64),
+		keyspaceID:   constants.DefaultKeyspaceID,
+		keyspaceName: constants.DefaultKeyspaceName,
 
 		resultMode:     resultMode,
 		resGenPhysical: 10000,
@@ -271,7 +274,7 @@ type testTSOStreamSuite struct {
 func (s *testTSOStreamSuite) SetupTest() {
 	s.re = require.New(s.T())
 	s.inner = newMockTSOStreamImpl(context.Background(), resultModeManual)
-	s.stream = newTSOStream(context.Background(), mockStreamURL, s.inner)
+	s.stream = newTSOStream(context.Background(), mockStreamURL, s.inner, constants.DefaultKeyspaceName)
 }
 
 func (s *testTSOStreamSuite) TearDownTest() {
@@ -609,7 +612,7 @@ func BenchmarkTSOStreamSendRecv(b *testing.B) {
 	log.SetLevel(zapcore.FatalLevel)
 
 	streamInner := newMockTSOStreamImpl(context.Background(), resultModeGenerated)
-	stream := newTSOStream(context.Background(), mockStreamURL, streamInner)
+	stream := newTSOStream(context.Background(), mockStreamURL, streamInner, constants.DefaultKeyspaceName)
 	defer func() {
 		streamInner.stop()
 		stream.WaitForClosed()

client/constants/constants.go

@@ -26,6 +26,8 @@ const (
 	MaxKeyspaceID = uint32(0xFFFFFF)
 	// NullKeyspaceID is used for API v1 or legacy path where is keyspace agnostic.
 	NullKeyspaceID = uint32(0xFFFFFFFF)
+	// NullKeyspaceName is used for API v1 or legacy path where is keyspace agnostic.
+	NullKeyspaceName = ""
 	// DefaultKeyspaceGroupID is the default key space group id.
 	// We also reserved 0 for the keyspace group for the same purpose.
 	DefaultKeyspaceGroupID = uint32(0)

client/inner_client.go

@@ -42,6 +42,7 @@ const (
 
 type innerClient struct {
 	keyspaceID       uint32
+	keyspaceName     string
 	svrUrls          []string
 	serviceDiscovery sd.ServiceDiscovery
 	tokenDispatcher  *tokenDispatcher
@@ -62,7 +63,7 @@ type innerClient struct {
 func (c *innerClient) init(updateKeyspaceIDCb sd.UpdateKeyspaceIDFunc) error {
 	c.serviceDiscovery = sd.NewServiceDiscovery(
 		c.ctx, c.cancel, &c.wg, c.setServiceMode,
-		updateKeyspaceIDCb, c.keyspaceID, c.svrUrls, c.tlsCfg, c.option)
+		updateKeyspaceIDCb, c.keyspaceID, c.keyspaceName, c.svrUrls, c.tlsCfg, c.option)
 	if err := c.setup(); err != nil {
 		c.cancel()
 		if c.serviceDiscovery != nil {
@@ -181,21 +182,21 @@ func (c *innerClient) resetTSOClientLocked(mode pdpb.ServiceMode) {
 	switch mode {
 	case pdpb.ServiceMode_PD_SVC_MODE:
 		newTSOCli = tso.NewClient(c.ctx, c.option,
-			c.serviceDiscovery, &tso.PDStreamBuilderFactory{})
+			c.serviceDiscovery, &tso.PDStreamBuilderFactory{}, c.keyspaceName)
 	case pdpb.ServiceMode_API_SVC_MODE:
 		newTSOSvcDiscovery = sd.NewTSOServiceDiscovery(
 			c.ctx, c, c.serviceDiscovery,
 			c.keyspaceID, c.tlsCfg, c.option)
-		// At this point, the keyspace group isn't known yet. Starts from the default keyspace group,
-		// and will be updated later.
-		newTSOCli = tso.NewClient(c.ctx, c.option,
-			newTSOSvcDiscovery, &tso.MSStreamBuilderFactory{})
 		if err := newTSOSvcDiscovery.Init(); err != nil {
 			log.Error("[pd] failed to initialize tso service discovery",
 				zap.Strings("svr-urls", c.svrUrls),
 				zap.Error(err))
 			return
 		}
+		// At this point, the keyspace group isn't known yet. Starts from the default keyspace group,
+		// and will be updated later.
+		newTSOCli = tso.NewClient(c.ctx, c.option,
+			newTSOSvcDiscovery, &tso.MSStreamBuilderFactory{}, c.keyspaceName)
 	case pdpb.ServiceMode_UNKNOWN_SVC_MODE:
 		log.Warn("[pd] intend to switch to unknown service mode, just return")
 		return