[close #372] add circuit breaker and smart rawkv client

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

@@ -92,6 +92,18 @@ public class ConfigUtils {
   public static final String TIKV_KEY_CERT_CHAIN = "tikv.key_cert_chain";
   public static final String TIKV_KEY_FILE = "tikv.key_file";
 
+  public static final String TiKV_CIRCUIT_BREAK_ENABLE = "tikv.circuit_break.enable";
+  public static final String TiKV_CIRCUIT_BREAK_AVAILABILITY_WINDOW_IN_SECONDS =
+      "tikv.circuit_break.trigger.availability.window_in_seconds";
+  public static final String TiKV_CIRCUIT_BREAK_AVAILABILITY_ERROR_THRESHOLD_PERCENTAGE =
+      "tikv.circuit_break.trigger.availability.error_threshold_percentage";
+  public static final String TiKV_CIRCUIT_BREAK_AVAILABILITY_REQUEST_VOLUMN_THRESHOLD =
+      "tikv.circuit_break.trigger.availability.request_volumn_threshold";
+  public static final String TiKV_CIRCUIT_BREAK_SLEEP_WINDOW_IN_SECONDS =
+      "tikv.circuit_break.trigger.sleep_window_in_seconds";
+  public static final String TiKV_CIRCUIT_BREAK_ATTEMPT_REQUEST_COUNT =
+      "tikv.circuit_break.trigger.attempt_request_count";
+
   public static final String TIFLASH_ENABLE = "tiflash.enable";
   public static final String DEF_PD_ADDRESSES = "127.0.0.1:2379";
   public static final String DEF_TIMEOUT = "200ms";
@@ -161,4 +173,11 @@ public class ConfigUtils {
   public static final int DEF_TIKV_GRPC_KEEPALIVE_TIMEOUT = 3;
   public static final boolean DEF_TIKV_TLS_ENABLE = false;
   public static final boolean DEF_TIFLASH_ENABLE = false;
+
+  public static final boolean DEF_TiKV_CIRCUIT_BREAK_ENABLE = false;
+  public static final int DEF_TiKV_CIRCUIT_BREAK_AVAILABILITY_WINDOW_IN_SECONDS = 60;
+  public static final int DEF_TiKV_CIRCUIT_BREAK_AVAILABILITY_ERROR_THRESHOLD_PERCENTAGE = 100;
+  public static final int DEF_TiKV_CIRCUIT_BREAK_AVAILABILITY_REQUST_VOLUMN_THRESHOLD = 10;
+  public static final int DEF_TiKV_CIRCUIT_BREAK_SLEEP_WINDOW_IN_SECONDS = 20;
+  public static final int DEF_TiKV_CIRCUIT_BREAK_ATTEMPT_REQUEST_COUNT = 10;
 }

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

@@ -130,6 +130,20 @@ private static void loadFromDefaultProperties() {
     setIfMissing(TIKV_RAWKV_CLEAN_TIMEOUT_IN_MS, DEF_TIKV_RAWKV_CLEAN_TIMEOUT_IN_MS);
     setIfMissing(TIKV_BO_REGION_MISS_BASE_IN_MS, DEF_TIKV_BO_REGION_MISS_BASE_IN_MS);
     setIfMissing(TIKV_RAWKV_SCAN_SLOWLOG_IN_MS, DEF_TIKV_RAWKV_SCAN_SLOWLOG_IN_MS);
+    setIfMissing(TiKV_CIRCUIT_BREAK_ENABLE, DEF_TiKV_CIRCUIT_BREAK_ENABLE);
+    setIfMissing(
+        TiKV_CIRCUIT_BREAK_AVAILABILITY_WINDOW_IN_SECONDS,
+        DEF_TiKV_CIRCUIT_BREAK_AVAILABILITY_WINDOW_IN_SECONDS);
+    setIfMissing(
+        TiKV_CIRCUIT_BREAK_AVAILABILITY_ERROR_THRESHOLD_PERCENTAGE,
+        DEF_TiKV_CIRCUIT_BREAK_AVAILABILITY_ERROR_THRESHOLD_PERCENTAGE);
+    setIfMissing(
+        TiKV_CIRCUIT_BREAK_AVAILABILITY_REQUEST_VOLUMN_THRESHOLD,
+        DEF_TiKV_CIRCUIT_BREAK_AVAILABILITY_REQUST_VOLUMN_THRESHOLD);
+    setIfMissing(
+        TiKV_CIRCUIT_BREAK_SLEEP_WINDOW_IN_SECONDS, DEF_TiKV_CIRCUIT_BREAK_SLEEP_WINDOW_IN_SECONDS);
+    setIfMissing(
+        TiKV_CIRCUIT_BREAK_ATTEMPT_REQUEST_COUNT, DEF_TiKV_CIRCUIT_BREAK_ATTEMPT_REQUEST_COUNT);
   }
 
   public static void listAll() {
@@ -360,6 +374,16 @@ private static ReplicaRead getReplicaRead(String key) {
   private int keepaliveTime = getInt(TIKV_GRPC_KEEPALIVE_TIME);
   private int keepaliveTimeout = getInt(TIKV_GRPC_KEEPALIVE_TIMEOUT);
 
+  private boolean circuitBreakEnable = getBoolean(TiKV_CIRCUIT_BREAK_ENABLE);
+  private int circuitBreakAvailabilityWindowInSeconds =
+      getInt(TiKV_CIRCUIT_BREAK_AVAILABILITY_WINDOW_IN_SECONDS);
+  private int circuitBreakAvailabilityErrorThresholdPercentage =
+      getInt(TiKV_CIRCUIT_BREAK_AVAILABILITY_ERROR_THRESHOLD_PERCENTAGE);
+  private int circuitBreakAvailabilityRequestVolumnThreshold =
+      getInt(TiKV_CIRCUIT_BREAK_AVAILABILITY_REQUEST_VOLUMN_THRESHOLD);
+  private int circuitBreakSleepWindowInSeconds = getInt(TiKV_CIRCUIT_BREAK_SLEEP_WINDOW_IN_SECONDS);
+  private int circuitBreakAttemptRequestCount = getInt(TiKV_CIRCUIT_BREAK_ATTEMPT_REQUEST_COUNT);
+
   public enum KVMode {
     TXN,
     RAW
@@ -881,4 +905,57 @@ public int getRawKVScanSlowLogInMS() {
   public void setRawKVScanSlowLogInMS(int rawKVScanSlowLogInMS) {
     this.rawKVScanSlowLogInMS = rawKVScanSlowLogInMS;
   }
+
+  public boolean isCircuitBreakEnable() {
+    return circuitBreakEnable;
+  }
+
+  public void setCircuitBreakEnable(boolean circuitBreakEnable) {
+    this.circuitBreakEnable = circuitBreakEnable;
+  }
+
+  public int getCircuitBreakAvailabilityWindowInSeconds() {
+    return circuitBreakAvailabilityWindowInSeconds;
+  }
+
+  public void setCircuitBreakAvailabilityWindowInSeconds(
+      int circuitBreakAvailabilityWindowInSeconds) {
+    this.circuitBreakAvailabilityWindowInSeconds = circuitBreakAvailabilityWindowInSeconds;
+  }
+
+  public int getCircuitBreakAvailabilityErrorThresholdPercentage() {
+    return circuitBreakAvailabilityErrorThresholdPercentage;
+  }
+
+  public void setCircuitBreakAvailabilityErrorThresholdPercentage(
+      int circuitBreakAvailabilityErrorThresholdPercentage) {
+    this.circuitBreakAvailabilityErrorThresholdPercentage =
+        circuitBreakAvailabilityErrorThresholdPercentage;
+  }
+
+  public int getCircuitBreakAvailabilityRequestVolumnThreshold() {
+    return circuitBreakAvailabilityRequestVolumnThreshold;
+  }
+
+  public void setCircuitBreakAvailabilityRequestVolumnThreshold(
+      int circuitBreakAvailabilityRequestVolumnThreshold) {
+    this.circuitBreakAvailabilityRequestVolumnThreshold =
+        circuitBreakAvailabilityRequestVolumnThreshold;
+  }
+
+  public int getCircuitBreakSleepWindowInSeconds() {
+    return circuitBreakSleepWindowInSeconds;
+  }
+
+  public void setCircuitBreakSleepWindowInSeconds(int circuitBreakSleepWindowInSeconds) {
+    this.circuitBreakSleepWindowInSeconds = circuitBreakSleepWindowInSeconds;
+  }
+
+  public int getCircuitBreakAttemptRequestCount() {
+    return circuitBreakAttemptRequestCount;
+  }
+
+  public void setCircuitBreakAttemptRequestCount(int circuitBreakAttemptRequestCount) {
+    this.circuitBreakAttemptRequestCount = circuitBreakAttemptRequestCount;
+  }
 }

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

@@ -43,6 +43,7 @@
 import org.tikv.kvproto.ImportSstpb;
 import org.tikv.kvproto.Metapb;
 import org.tikv.raw.RawKVClient;
+import org.tikv.raw.SmartRawKVClient;
 import org.tikv.txn.KVClient;
 import org.tikv.txn.TxnKVClient;
 
@@ -126,6 +127,11 @@ public RawKVClient createRawClient() {
     return new RawKVClient(this, this.getRegionStoreClientBuilder());
   }
 
+  public SmartRawKVClient createSmartRawClient() {
+    RawKVClient rawKVClient = createRawClient();
+    return new SmartRawKVClient(rawKVClient, getConf());
+  }
+
   public KVClient createKVClient() {
     checkIsClosed();
 

src/main/java/org/tikv/common/exception/CircuitBreakerOpenException.java

@@ -0,0 +1,22 @@
+/*
+ * Copyright 2018 PingCAP, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.tikv.common.exception;
+
+public class CircuitBreakerOpenException extends RuntimeException {
+  public CircuitBreakerOpenException() {
+    super("Circuit Breaker Opened");
+  }
+}