cherry-pick #558

Signed-off-by: iosmanthus <myosmanthustree@gmail.com>

Author: iosmanthus

metrics/grafana/client_java_summary.json

@@ -65,7 +65,9 @@ public void close() {}
    * @return a ByteString value if key exists, ByteString.EMPTY if key does not exist
    */
   public ByteString get(ByteString key, long version) throws GrpcException {
-    BackOffer backOffer = ConcreteBackOffer.newGetBackOff();
+    BackOffer backOffer =
+        ConcreteBackOffer.newGetBackOff(
+            clientBuilder.getRegionManager().getPDClient().getClusterId());
     while (true) {
       RegionStoreClient client = clientBuilder.build(key);
       try {

src/main/java/org/tikv/common/KVClient.java

@@ -113,6 +113,7 @@ public class PDClient extends AbstractGRPCClient<PDBlockingStub, PDFutureStub>
       HistogramUtils.buildDuration()
           .name("client_java_pd_get_region_by_requests_latency")
           .help("pd getRegionByKey request latency.")
+          .labelNames("cluster")
           .register();
 
   private PDClient(TiConfiguration conf, ChannelFactory channelFactory) {
@@ -203,7 +204,7 @@ private GetOperatorResponse getOperator(long regionId) {
         () -> GetOperatorRequest.newBuilder().setHeader(header).setRegionId(regionId).build();
     // get operator no need to handle error and no need back offer.
     return callWithRetry(
-        ConcreteBackOffer.newCustomBackOff(0),
+        ConcreteBackOffer.newCustomBackOff(0, getClusterId()),
         PDGrpc.getGetOperatorMethod(),
         request,
         new NoopHandler<>());
@@ -231,7 +232,8 @@ private boolean isScatterRegionFinish(GetOperatorResponse resp) {
 
   @Override
   public Pair<Metapb.Region, Metapb.Peer> getRegionByKey(BackOffer backOffer, ByteString key) {
-    Histogram.Timer requestTimer = PD_GET_REGION_BY_KEY_REQUEST_LATENCY.startTimer();
+    Histogram.Timer requestTimer =
+        PD_GET_REGION_BY_KEY_REQUEST_LATENCY.labels(getClusterId().toString()).startTimer();
     try {
       if (conf.getKvMode() == KVMode.TXN) {
         CodecDataOutput cdo = new CodecDataOutput();
@@ -745,7 +747,8 @@ private Metapb.Region decodeRegion(Metapb.Region region) {
     return builder.build();
   }
 
-  public long getClusterId() {
+  @Override
+  public Long getClusterId() {
     return header.getClusterId();
   }
 

src/main/java/org/tikv/common/PDClient.java

@@ -65,4 +65,6 @@ List<Pdpb.Region> scanRegions(
   List<Store> getAllStores(BackOffer backOffer);
 
   TiConfiguration.ReplicaRead getReplicaRead();
+
+  Long getClusterId();
 }

src/main/java/org/tikv/common/ReadOnlyPDClient.java

@@ -78,7 +78,9 @@ public List<org.tikv.common.BytePairWrapper> batchGet(int backOffer, List<byte[]
     try (KVClient client = new KVClient(session, session.getRegionStoreClientBuilder())) {
       List<KvPair> kvPairList =
           client.batchGet(
-              ConcreteBackOffer.newCustomBackOff(backOffer), list, timestamp.getVersion());
+              ConcreteBackOffer.newCustomBackOff(backOffer, session.getPDClient().getClusterId()),
+              list,
+              timestamp.getVersion());
       return kvPairList
           .stream()
           .map(

src/main/java/org/tikv/common/Snapshot.java

@@ -60,7 +60,8 @@ public static int compareTo(String v0, String v1) {
   public static boolean minTiKVVersion(String version, PDClient pdClient) {
     StoreVersion storeVersion = new StoreVersion(version);
 
-    BackOffer bo = ConcreteBackOffer.newCustomBackOff(BackOffer.PD_INFO_BACKOFF);
+    BackOffer bo =
+        ConcreteBackOffer.newCustomBackOff(BackOffer.PD_INFO_BACKOFF, pdClient.getClusterId());
     List<Metapb.Store> storeList =
         pdClient
             .getAllStores(bo)

src/main/java/org/tikv/common/StoreVersion.java

@@ -91,13 +91,13 @@ public TiSession(TiConfiguration conf) {
     if (conf.isWarmUpEnable() && conf.isRawKVMode()) {
       warmUp();
     }
-    this.circuitBreaker = new CircuitBreakerImpl(conf);
+    this.circuitBreaker = new CircuitBreakerImpl(conf, getPDClient().getClusterId());
     logger.info("TiSession initialized in " + conf.getKvMode() + " mode");
   }
 
   private synchronized void warmUp() {
     long warmUpStartTime = System.nanoTime();
-    BackOffer backOffer = ConcreteBackOffer.newRawKVBackOff();
+    BackOffer backOffer = ConcreteBackOffer.newRawKVBackOff(getPDClient().getClusterId());
 
     try {
       // let JVM ClassLoader load gRPC error related classes
@@ -128,7 +128,9 @@ private synchronized void warmUp() {
         }
         for (Pdpb.Region region : regions) {
           regionManager.insertRegionToCache(
-              regionManager.createRegion(region.getRegion(), ConcreteBackOffer.newGetBackOff()));
+              regionManager.createRegion(
+                  region.getRegion(),
+                  ConcreteBackOffer.newGetBackOff(getPDClient().getClusterId())));
         }
         startKey = regions.get(regions.size() - 1).getRegion().getEndKey();
       } while (!startKey.isEmpty());
@@ -226,7 +228,8 @@ public TiConfiguration getConf() {
   public TiTimestamp getTimestamp() {
     checkIsClosed();
 
-    return getPDClient().getTimestamp(ConcreteBackOffer.newTsoBackOff());
+    return getPDClient()
+        .getTimestamp(ConcreteBackOffer.newTsoBackOff(getPDClient().getClusterId()));
   }
 
   public Snapshot createSnapshot() {
@@ -459,7 +462,7 @@ public void splitRegionAndScatter(
                 .stream()
                 .map(k -> Key.toRawKey(k).next().toByteString())
                 .collect(Collectors.toList()),
-            ConcreteBackOffer.newCustomBackOff(splitRegionBackoffMS));
+            ConcreteBackOffer.newCustomBackOff(splitRegionBackoffMS, getPDClient().getClusterId()));
 
     // scatter region
     for (Metapb.Region newRegion : newRegions) {
@@ -482,7 +485,9 @@ public void splitRegionAndScatter(
           return;
         }
         getPDClient()
-            .waitScatterRegionFinish(newRegion, ConcreteBackOffer.newCustomBackOff((int) remainMS));
+            .waitScatterRegionFinish(
+                newRegion,
+                ConcreteBackOffer.newCustomBackOff((int) remainMS, getPDClient().getClusterId()));
       }
     } else {
       logger.info("skip to wait scatter region finish");

src/main/java/org/tikv/common/TiSession.java

@@ -74,7 +74,9 @@ TiRegion loadCurrentRegionToCache() throws GrpcException {
     try (RegionStoreClient client = builder.build(startKey)) {
       client.setTimeout(conf.getScanTimeout());
       region = client.getRegion();
-      BackOffer backOffer = ConcreteBackOffer.newScannerNextMaxBackOff();
+      BackOffer backOffer =
+          ConcreteBackOffer.newScannerNextMaxBackOff(
+              builder.getRegionManager().getPDClient().getClusterId());
       currentCache = client.scan(backOffer, startKey, version);
       return region;
     }
@@ -86,7 +88,8 @@ private ByteString resolveCurrentLock(Kvrpcpb.KvPair current) {
         builder.getRegionManager().getRegionStorePairByKey(current.getKey());
     TiRegion region = pair.first;
     TiStore store = pair.second;
-    BackOffer backOffer = ConcreteBackOffer.newGetBackOff();
+    BackOffer backOffer =
+        ConcreteBackOffer.newGetBackOff(builder.getRegionManager().getPDClient().getClusterId());
     try (RegionStoreClient client = builder.build(region, store)) {
       return client.get(backOffer, current.getKey(), version);
     } catch (Exception e) {

src/main/java/org/tikv/common/operation/iterator/ConcreteScanIterator.java

@@ -33,19 +33,19 @@ public abstract class RetryPolicy<RespT> {
       HistogramUtils.buildDuration()
           .name("client_java_grpc_single_requests_latency")
           .help("grpc request latency.")
-          .labelNames("type")
+          .labelNames("type", "cluster")
           .register();
   public static final Histogram CALL_WITH_RETRY_DURATION =
       HistogramUtils.buildDuration()
           .name("client_java_call_with_retry_duration")
           .help("callWithRetry duration.")
-          .labelNames("type")
+          .labelNames("type", "cluster")
           .register();
   public static final Counter GRPC_REQUEST_RETRY_NUM =
       Counter.build()
           .name("client_java_grpc_requests_retry_num")
           .help("grpc request retry num.")
-          .labelNames("type")
+          .labelNames("type", "cluster")
           .register();
 
   // handles PD and TiKV's error.
@@ -70,16 +70,16 @@ private void rethrowNotRecoverableException(Exception e) {
   }
 
   public RespT callWithRetry(Callable<RespT> proc, String methodName, BackOffer backOffer) {
-    Histogram.Timer callWithRetryTimer = CALL_WITH_RETRY_DURATION.labels(methodName).startTimer();
+    String[] labels = new String[] {methodName, backOffer.getClusterId().toString()};
+    Histogram.Timer callWithRetryTimer = CALL_WITH_RETRY_DURATION.labels(labels).startTimer();
     SlowLogSpan callWithRetrySlowLogSpan =
         backOffer.getSlowLog().start("callWithRetry " + methodName);
     try {
       while (true) {
         RespT result = null;
         try {
           // add single request duration histogram
-          Histogram.Timer requestTimer =
-              GRPC_SINGLE_REQUEST_LATENCY.labels(methodName).startTimer();
+          Histogram.Timer requestTimer = GRPC_SINGLE_REQUEST_LATENCY.labels(labels).startTimer();
           SlowLogSpan slowLogSpan = backOffer.getSlowLog().start("gRPC " + methodName);
           try {
             result = proc.call();
@@ -93,7 +93,7 @@ public RespT callWithRetry(Callable<RespT> proc, String methodName, BackOffer ba
           backOffer.checkTimeout();
           boolean retry = handler.handleRequestError(backOffer, e);
           if (retry) {
-            GRPC_REQUEST_RETRY_NUM.labels(methodName).inc();
+            GRPC_REQUEST_RETRY_NUM.labels(labels).inc();
             continue;
           } else {
             return result;
@@ -104,7 +104,7 @@ public RespT callWithRetry(Callable<RespT> proc, String methodName, BackOffer ba
         if (handler != null) {
           boolean retry = handler.handleResponseError(backOffer, result);
           if (retry) {
-            GRPC_REQUEST_RETRY_NUM.labels(methodName).inc();
+            GRPC_REQUEST_RETRY_NUM.labels(labels).inc();
             continue;
           }
         }

src/main/java/org/tikv/common/policy/RetryPolicy.java

@@ -52,12 +52,14 @@ public abstract class AbstractRegionStoreClient
       HistogramUtils.buildDuration()
           .name("client_java_seek_leader_store_duration")
           .help("seek leader store duration.")
+          .labelNames("cluster")
           .register();
 
   public static final Histogram SEEK_PROXY_STORE_DURATION =
       HistogramUtils.buildDuration()
           .name("client_java_seek_proxy_store_duration")
           .help("seek proxy store duration.")
+          .labelNames("cluster")
           .register();
 
   protected final RegionManager regionManager;
@@ -181,7 +183,10 @@ private void updateClientStub() {
   }
 
   private Boolean seekLeaderStore(BackOffer backOffer) {
-    Histogram.Timer switchLeaderDurationTimer = SEEK_LEADER_STORE_DURATION.startTimer();
+    Histogram.Timer switchLeaderDurationTimer =
+        SEEK_LEADER_STORE_DURATION
+            .labels(regionManager.getPDClient().getClusterId().toString())
+            .startTimer();
     SlowLogSpan slowLogSpan = backOffer.getSlowLog().start("seekLeaderStore");
     try {
       List<Metapb.Peer> peers = region.getFollowerList();
@@ -229,7 +234,10 @@ private Boolean seekLeaderStore(BackOffer backOffer) {
 
   private boolean seekProxyStore(BackOffer backOffer) {
     SlowLogSpan slowLogSpan = backOffer.getSlowLog().start("seekProxyStore");
-    Histogram.Timer grpcForwardDurationTimer = SEEK_PROXY_STORE_DURATION.startTimer();
+    Histogram.Timer grpcForwardDurationTimer =
+        SEEK_PROXY_STORE_DURATION
+            .labels(regionManager.getPDClient().getClusterId().toString())
+            .startTimer();
     try {
       logger.info(String.format("try grpc forward: region[%d]", region.getId()));
       // when current leader cannot be reached