tikv · ti-chi-bot · Jul 30, 2026 · Jul 24, 2026
diff --git a/pkg/errs/errno.go b/pkg/errs/errno.go
@@ -86,9 +86,6 @@ var (
 	ErrNotStarted                 = status.Error(codes.Unavailable, "server not started")
 	ErrEtcdNotStarted             = status.Error(codes.Unavailable, "server is started, but etcd not started")
 	ErrFollowerHandlingNotAllowed = status.Error(codes.Unavailable, "not leader and follower handling not allowed")
-
-	// Unimplemented indicates operation is not implemented or not supported.
-	ErrRecvNotSupported = status.Error(codes.Unimplemented, "recv is not supported by this stream")
 )
 
 // common error in multiple packages

diff --git a/pkg/mcs/metastorage/server/grpc_service.go b/pkg/mcs/metastorage/server/grpc_service.go
@@ -82,7 +82,6 @@ func (s *Service) checkServing() error {
 
 // Watch watches the key with a given prefix and revision.
 func (s *Service) Watch(req *meta_storagepb.WatchRequest, server meta_storagepb.MetaStorage_WatchServer) error {
-	server = newWatchMetricsStream(server)
 	if err := s.checkServing(); err != nil {
 		return err
 	}

diff --git a/pkg/mcs/metastorage/server/metrics.go b/pkg/mcs/metastorage/server/metrics.go
diff --git a/pkg/mcs/resourcemanager/server/grpc_service.go b/pkg/mcs/resourcemanager/server/grpc_service.go
@@ -188,7 +188,6 @@ func (s *Service) ModifyResourceGroup(_ context.Context, req *rmpb.PutResourceGr
 
 // AcquireTokenBuckets implements ResourceManagerServer.AcquireTokenBuckets.
 func (s *Service) AcquireTokenBuckets(stream rmpb.ResourceManager_AcquireTokenBucketsServer) error {
-	stream = newAcquireTokenBucketsMetricsStream(stream)
 	for {
 		select {
 		case <-s.ctx.Done():

diff --git a/pkg/mcs/resourcemanager/server/metrics.go b/pkg/mcs/resourcemanager/server/metrics.go
@@ -23,8 +23,6 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 
 	rmpb "github.com/pingcap/kvproto/pkg/resource_manager"
-
-	"github.com/tikv/pd/pkg/utils/grpcutil"
 )
 
 const (
@@ -60,8 +58,6 @@ const (
 )
 
 var (
-	grpcStreamSendDuration = grpcutil.NewGRPCStreamSendDuration(namespace, serverSubsystem)
-
 	// RU cost metrics.
 	// `sum` is added to the name to maintain compatibility with the previous use of histogram.
 	readRequestUnitCost = prometheus.NewCounterVec(
@@ -313,7 +309,6 @@ type requestMetricsKey struct {
 }
 
 func init() {
-	prometheus.MustRegister(grpcStreamSendDuration)
 	prometheus.MustRegister(readRequestUnitCost)
 	prometheus.MustRegister(writeRequestUnitCost)
 	prometheus.MustRegister(activeRequestUnitCost)
@@ -1052,7 +1047,3 @@ func (t *maxPerSecCostTracker) flushMetrics() {
 		t.maxPerSecWRU = 0
 	}
 }
-
-func newAcquireTokenBucketsMetricsStream(stream rmpb.ResourceManager_AcquireTokenBucketsServer) rmpb.ResourceManager_AcquireTokenBucketsServer {
-	return grpcutil.NewMetricsStream(stream, stream.Send, stream.Recv, grpcStreamSendDuration, "acquire-token-buckets")
-}
diff --git a/pkg/mcs/router/server/grpc_service.go b/pkg/mcs/router/server/grpc_service.go
@@ -129,7 +129,6 @@ func (s *Service) GetRegionByID(_ctx context.Context, request *pdpb.GetRegionByI
 
 // QueryRegion implements the QueryRegion RPC method.
 func (s *Service) QueryRegion(stream routerpb.Router_QueryRegionServer) error {
-	stream = newQueryRegionMetricsStream(stream)
 	for {
 		request, err := stream.Recv()
 		if err == io.EOF {

diff --git a/pkg/mcs/router/server/metrics.go b/pkg/mcs/router/server/metrics.go
@@ -16,10 +16,6 @@ package server
 
 import (
 	"github.com/prometheus/client_golang/prometheus"
-
-	"github.com/pingcap/kvproto/pkg/routerpb"
-
-	"github.com/tikv/pd/pkg/utils/grpcutil"
 )
 
 const (
@@ -28,8 +24,6 @@ const (
 )
 
 var (
-	grpcStreamSendDuration = grpcutil.NewGRPCStreamSendDuration(namespace, serverSubsystem)
-
 	queryRegionDuration = prometheus.NewHistogram(
 		prometheus.HistogramOpts{
 			Namespace: namespace,
@@ -57,12 +51,7 @@ var (
 )
 
 func init() {
-	prometheus.MustRegister(grpcStreamSendDuration)
 	prometheus.MustRegister(regionRequestCounter)
 	prometheus.MustRegister(queryRegionDuration)
 	prometheus.MustRegister(regionSyncerStatus)
 }
-
-func newQueryRegionMetricsStream(stream routerpb.Router_QueryRegionServer) routerpb.Router_QueryRegionServer {
-	return grpcutil.NewMetricsStream(stream, stream.Send, stream.Recv, grpcStreamSendDuration, "query-region")
-}
diff --git a/pkg/mcs/scheduling/server/grpc_service.go b/pkg/mcs/scheduling/server/grpc_service.go
@@ -116,9 +116,8 @@ func (s *heartbeatServer) recv() (*schedulingpb.RegionHeartbeatRequest, error) {
 
 // RegionHeartbeat implements gRPC SchedulingServer.
 func (s *Service) RegionHeartbeat(stream schedulingpb.Scheduling_RegionHeartbeatServer) error {
-	wrappedStream := newRegionHeartbeatMetricsStream(stream)
 	var (
-		server   = &heartbeatServer{stream: wrappedStream}
+		server   = &heartbeatServer{stream: stream}
 		cancel   context.CancelFunc
 		lastBind time.Time
 	)
@@ -183,7 +182,6 @@ func (s *Service) RegionHeartbeat(stream schedulingpb.Scheduling_RegionHeartbeat
 
 // RegionBuckets implements gRPC SchedulingServer.
 func (s *Service) RegionBuckets(stream schedulingpb.Scheduling_RegionBucketsServer) error {
-	stream = newRegionBucketsMetricsStream(stream)
 	var cancel context.CancelFunc
 	defer func() {
 		// cancel the forward stream

diff --git a/pkg/mcs/scheduling/server/metrics.go b/pkg/mcs/scheduling/server/metrics.go
@@ -14,22 +14,14 @@
 
 package server
 
-import (
-	"github.com/prometheus/client_golang/prometheus"
-
-	"github.com/pingcap/kvproto/pkg/schedulingpb"
-
-	"github.com/tikv/pd/pkg/utils/grpcutil"
-)
+import "github.com/prometheus/client_golang/prometheus"
 
 const (
 	namespace       = "scheduling"
 	serverSubsystem = "server"
 )
 
 var (
-	grpcStreamSendDuration = grpcutil.NewGRPCStreamSendDuration(namespace, serverSubsystem)
-
 	// Store heartbeat metrics
 	storeHeartbeatHandleDuration = prometheus.NewHistogramVec(
 		prometheus.HistogramOpts{
@@ -94,7 +86,6 @@ var (
 )
 
 func init() {
-	prometheus.MustRegister(grpcStreamSendDuration)
 	prometheus.MustRegister(storeHeartbeatHandleDuration)
 	prometheus.MustRegister(storeHeartbeatCounter)
 	prometheus.MustRegister(regionHeartbeatHandleDuration)
@@ -103,11 +94,3 @@ func init() {
 	prometheus.MustRegister(regionBucketsCounter)
 	prometheus.MustRegister(regionBucketsReportInterval)
 }
-
-func newRegionHeartbeatMetricsStream(stream schedulingpb.Scheduling_RegionHeartbeatServer) schedulingpb.Scheduling_RegionHeartbeatServer {
-	return grpcutil.NewMetricsStream(stream, stream.Send, stream.Recv, grpcStreamSendDuration, "region-heartbeat")
-}
-
-func newRegionBucketsMetricsStream(stream schedulingpb.Scheduling_RegionBucketsServer) schedulingpb.Scheduling_RegionBucketsServer {
-	return grpcutil.NewMetricsStream(stream, stream.Send, stream.Recv, grpcStreamSendDuration, "region-buckets")
-}
diff --git a/pkg/mcs/tso/server/grpc_service.go b/pkg/mcs/tso/server/grpc_service.go
@@ -77,7 +77,6 @@ func (s *Service) RegisterRESTHandler(userDefineHandlers map[string]http.Handler
 
 // Tso returns a stream of timestamps
 func (s *Service) Tso(stream tsopb.TSO_TsoServer) error {
-	stream = newTsoMetricsStream(stream)
 	ctx, cancel := context.WithCancel(stream.Context())
 	defer cancel()
 	for {

diff --git a/pkg/mcs/tso/server/metrics.go b/pkg/mcs/tso/server/metrics.go
@@ -14,19 +14,11 @@
 
 package server
 
-import (
-	"github.com/prometheus/client_golang/prometheus"
-
-	"github.com/pingcap/kvproto/pkg/tsopb"
-
-	"github.com/tikv/pd/pkg/utils/grpcutil"
-)
+import "github.com/prometheus/client_golang/prometheus"
 
 const namespace = "tso"
 
 var (
-	grpcStreamSendDuration = grpcutil.NewGRPCStreamSendDuration(namespace, "server")
-
 	timeJumpBackCounter = prometheus.NewCounter(
 		prometheus.CounterOpts{
 			Namespace: namespace,
@@ -54,12 +46,7 @@ var (
 )
 
 func init() {
-	prometheus.MustRegister(grpcStreamSendDuration)
 	prometheus.MustRegister(timeJumpBackCounter)
 	prometheus.MustRegister(metaDataGauge)
 	prometheus.MustRegister(tsoHandleDuration)
 }
-
-func newTsoMetricsStream(stream tsopb.TSO_TsoServer) tsopb.TSO_TsoServer {
-	return grpcutil.NewMetricsStream(stream, stream.Send, stream.Recv, grpcStreamSendDuration, "tso")
-}
diff --git a/pkg/utils/grpcutil/stream.go b/pkg/utils/grpcutil/stream.go