Add grpc circuit breaker utility using interceptors

build.gradle.kts

@@ -7,7 +7,7 @@ plugins {
   id("org.hypertrace.publish-plugin") version "1.0.5" apply false
   id("org.hypertrace.jacoco-report-plugin") version "0.2.1" apply false
   id("org.hypertrace.code-style-plugin") version "2.0.0" apply false
-  id("org.owasp.dependencycheck") version "10.0.3"
+  id("org.owasp.dependencycheck") version "12.1.0"
 }
 
 subprojects {

grpc-circuitbreaker-utils/build.gradle.kts

@@ -0,0 +1,29 @@
+plugins {
+  `java-library`
+  jacoco
+  id("org.hypertrace.publish-plugin")
+  id("org.hypertrace.jacoco-report-plugin")
+}
+
+dependencies {
+
+  api(platform("io.grpc:grpc-bom:1.68.3"))
+  api("io.grpc:grpc-api")
+  api(project(":grpc-context-utils"))
+
+  implementation("org.slf4j:slf4j-api:1.7.36")
+  implementation("io.github.resilience4j:resilience4j-circuitbreaker:1.7.1")
+  implementation("com.typesafe:config:1.4.2")
+  implementation("com.google.guava:guava:32.0.1-jre")
+
+  annotationProcessor("org.projectlombok:lombok:1.18.24")
+  compileOnly("org.projectlombok:lombok:1.18.24")
+
+  testImplementation("org.junit.jupiter:junit-jupiter:5.8.2")
+  testImplementation("org.mockito:mockito-core:5.8.0")
+  testImplementation("org.mockito:mockito-junit-jupiter:5.8.0")
+}
+
+tasks.test {
+  useJUnitPlatform()
+}

grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerConfigParser.java

@@ -0,0 +1,111 @@
+package org.hypertrace.circuitbreaker.grpcutils;
+
+import com.typesafe.config.Config;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class CircuitBreakerConfigParser {
+
+  // Percentage of failures to trigger OPEN state
+  private static final String FAILURE_RATE_THRESHOLD = "failureRateThreshold";
+  // Percentage of slow calls to trigger OPEN state
+  private static final String SLOW_CALL_RATE_THRESHOLD = "slowCallRateThreshold";
+  // Define what a "slow" call is
+  private static final String SLOW_CALL_DURATION_THRESHOLD = "slowCallDurationThreshold";
+  // Number of calls to consider in the sliding window
+  private static final String SLIDING_WINDOW_SIZE = "slidingWindowSize";
+  // Time before retrying after OPEN state
+  private static final String WAIT_DURATION_IN_OPEN_STATE = "waitDurationInOpenState";
+  // Minimum calls before evaluating failure rate
+  private static final String MINIMUM_NUMBER_OF_CALLS = "minimumNumberOfCalls";
+  // Calls allowed in HALF_OPEN state before deciding to
+  // CLOSE or OPEN again
+  private static final String PERMITTED_NUMBER_OF_CALLS_IN_HALF_OPEN_STATE =
+      "permittedNumberOfCallsInHalfOpenState";
+  private static final String SLIDING_WINDOW_TYPE = "slidingWindowType";
+  public static final String ENABLED = "enabled";
+  public static final String DEFAULT_THRESHOLDS = "defaultThresholds";
+  private static final Set<String> NON_THRESHOLD_KEYS = Set.of(ENABLED, DEFAULT_THRESHOLDS);
+
+  public static <T> CircuitBreakerConfiguration.CircuitBreakerConfigurationBuilder<T> parseConfig(
+      Config config) {
+    CircuitBreakerConfiguration.CircuitBreakerConfigurationBuilder<T> builder =
+        CircuitBreakerConfiguration.builder();
+    if (config.hasPath(ENABLED)) {
+      builder.enabled(config.getBoolean(ENABLED));
+    }
+
+    Map<String, CircuitBreakerThresholds> circuitBreakerThresholdsMap =
+        config.root().keySet().stream()
+            .filter(key -> !NON_THRESHOLD_KEYS.contains(key)) // Filter out non-threshold keys
+            .collect(
+                Collectors.toMap(
+                    key -> key, // Circuit breaker key
+                    key -> buildCircuitBreakerThresholds(config.getConfig(key))));
+
+    builder.defaultThresholds(
+        config.hasPath(DEFAULT_THRESHOLDS)
+            ? buildCircuitBreakerThresholds(config.getConfig(DEFAULT_THRESHOLDS))
+            : buildCircuitBreakerDefaultThresholds());
+
+    builder.circuitBreakerThresholdsMap(circuitBreakerThresholdsMap);
+    log.debug("Loaded circuit breaker configs: {}", builder);
+    return builder;
+  }
+
+  private static CircuitBreakerThresholds buildCircuitBreakerThresholds(Config config) {
+    CircuitBreakerThresholds.CircuitBreakerThresholdsBuilder builder =
+        CircuitBreakerThresholds.builder();
+
+    if (config.hasPath(FAILURE_RATE_THRESHOLD)) {
+      builder.failureRateThreshold((float) config.getDouble(FAILURE_RATE_THRESHOLD));
+    }
+
+    if (config.hasPath(SLOW_CALL_RATE_THRESHOLD)) {
+      builder.slowCallRateThreshold((float) config.getDouble(SLOW_CALL_RATE_THRESHOLD));
+    }
+
+    if (config.hasPath(SLOW_CALL_DURATION_THRESHOLD)) {
+      builder.slowCallDurationThreshold(config.getDuration(SLOW_CALL_DURATION_THRESHOLD));
+    }
+
+    if (config.hasPath(SLIDING_WINDOW_TYPE)) {
+      builder.slidingWindowType(getSlidingWindowType(config.getString(SLIDING_WINDOW_TYPE)));
+    }
+
+    if (config.hasPath(SLIDING_WINDOW_SIZE)) {
+      builder.slidingWindowSize(config.getInt(SLIDING_WINDOW_SIZE));
+    }
+
+    if (config.hasPath(WAIT_DURATION_IN_OPEN_STATE)) {
+      builder.waitDurationInOpenState(config.getDuration(WAIT_DURATION_IN_OPEN_STATE));
+    }
+
+    if (config.hasPath(PERMITTED_NUMBER_OF_CALLS_IN_HALF_OPEN_STATE)) {
+      builder.permittedNumberOfCallsInHalfOpenState(
+          config.getInt(PERMITTED_NUMBER_OF_CALLS_IN_HALF_OPEN_STATE));
+    }
+
+    if (config.hasPath(MINIMUM_NUMBER_OF_CALLS)) {
+      builder.minimumNumberOfCalls(config.getInt(MINIMUM_NUMBER_OF_CALLS));
+    }
+
+    if (config.hasPath(ENABLED)) {
+      builder.enabled(config.getBoolean(ENABLED));
+    }
+
+    return builder.build();
+  }
+
+  public static CircuitBreakerThresholds buildCircuitBreakerDefaultThresholds() {
+    return CircuitBreakerThresholds.builder().build();
+  }
+
+  private static CircuitBreakerThresholds.SlidingWindowType getSlidingWindowType(
+      String slidingWindowType) {
+    return CircuitBreakerThresholds.SlidingWindowType.valueOf(slidingWindowType);
+  }
+}

grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerConfiguration.java

@@ -0,0 +1,27 @@
+package org.hypertrace.circuitbreaker.grpcutils;
+
+import io.grpc.Status;
+import io.grpc.StatusRuntimeException;
+import java.util.Map;
+import java.util.function.BiFunction;
+import java.util.function.Function;
+import lombok.Builder;
+import lombok.Value;
+import org.hypertrace.core.grpcutils.context.RequestContext;
+
+@Value
+@Builder
+public class CircuitBreakerConfiguration<T> {
+  Class<T> requestClass;
+  BiFunction<RequestContext, T, String> keyFunction;
+  @Builder.Default boolean enabled = false;
+  // Standard/default thresholds
+  CircuitBreakerThresholds defaultThresholds;
+  // Custom overrides for specific cases (less common)
+  @Builder.Default Map<String, CircuitBreakerThresholds> circuitBreakerThresholdsMap = Map.of();
+
+  // New exception builder logic
+  @Builder.Default
+  Function<String, StatusRuntimeException> exceptionBuilder =
+      reason -> Status.RESOURCE_EXHAUSTED.withDescription(reason).asRuntimeException();
+}

grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerInterceptor.java

@@ -0,0 +1,23 @@
+package org.hypertrace.circuitbreaker.grpcutils;
+
+import io.grpc.CallOptions;
+import io.grpc.Channel;
+import io.grpc.ClientCall;
+import io.grpc.ClientInterceptor;
+import io.grpc.MethodDescriptor;
+
+public abstract class CircuitBreakerInterceptor implements ClientInterceptor {
+  @Override
+  public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
+      MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
+    if (!isCircuitBreakerEnabled()) {
+      return next.newCall(method, callOptions);
+    }
+    return createInterceptedCall(method, callOptions, next);
+  }
+
+  protected abstract boolean isCircuitBreakerEnabled();
+
+  protected abstract <ReqT, RespT> ClientCall<ReqT, RespT> createInterceptedCall(
+      MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next);
+}

grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerThresholds.java

@@ -0,0 +1,32 @@
+package org.hypertrace.circuitbreaker.grpcutils;
+
+import java.time.Duration;
+import lombok.Builder;
+import lombok.Value;
+
+@Value
+@Builder
+public class CircuitBreakerThresholds {
+  // Percentage of failures to trigger OPEN state
+  @Builder.Default float failureRateThreshold = 50f;
+  // Percentage of slow calls to trigger OPEN state
+  @Builder.Default float slowCallRateThreshold = 50f;
+  // Define what a "slow" call is
+  @Builder.Default Duration slowCallDurationThreshold = Duration.ofSeconds(2);
+  // Number of calls to consider in the sliding window
+  @Builder.Default SlidingWindowType slidingWindowType = SlidingWindowType.TIME_BASED;
+  @Builder.Default int slidingWindowSize = 60;
+  // Time before retrying after OPEN state
+  @Builder.Default Duration waitDurationInOpenState = Duration.ofSeconds(60);
+  // Minimum calls before evaluating failure rate
+  @Builder.Default int minimumNumberOfCalls = 10;
+  // Calls allowed in HALF_OPEN state before deciding to
+  // CLOSE or OPEN again
+  @Builder.Default int permittedNumberOfCallsInHalfOpenState = 5;
+  @Builder.Default boolean enabled = true;
+
+  public enum SlidingWindowType {
+    COUNT_BASED,
+    TIME_BASED
+  }
+}

grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/resilience/ResilienceCircuitBreakerConfigConverter.java

@@ -0,0 +1,50 @@
+package org.hypertrace.circuitbreaker.grpcutils.resilience;
+
+import io.github.resilience4j.circuitbreaker.CircuitBreakerConfig;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import org.hypertrace.circuitbreaker.grpcutils.CircuitBreakerThresholds;
+
+/** Utility class to parse CircuitBreakerConfiguration to Resilience4j CircuitBreakerConfig */
+class ResilienceCircuitBreakerConfigConverter {
+
+  public static Map<String, CircuitBreakerConfig> getCircuitBreakerConfigs(
+      Map<String, CircuitBreakerThresholds> configurationMap) {
+    return configurationMap.entrySet().stream()
+        .collect(Collectors.toMap(Map.Entry::getKey, entry -> convertConfig(entry.getValue())));
+  }
+
+  public static List<String> getDisabledKeys(
+      Map<String, CircuitBreakerThresholds> configurationMap) {
+    return configurationMap.entrySet().stream()
+        .filter(entry -> entry.getValue().isEnabled())
+        .map(Map.Entry::getKey)
+        .collect(Collectors.toList());
+  }
+
+  static CircuitBreakerConfig convertConfig(CircuitBreakerThresholds configuration) {
+    return CircuitBreakerConfig.custom()
+        .failureRateThreshold(configuration.getFailureRateThreshold())
+        .slowCallRateThreshold(configuration.getSlowCallRateThreshold())
+        .slowCallDurationThreshold(configuration.getSlowCallDurationThreshold())
+        .slidingWindowType(getSlidingWindowType(configuration.getSlidingWindowType()))
+        .slidingWindowSize(configuration.getSlidingWindowSize())
+        .waitDurationInOpenState(configuration.getWaitDurationInOpenState())
+        .permittedNumberOfCallsInHalfOpenState(
+            configuration.getPermittedNumberOfCallsInHalfOpenState())
+        .minimumNumberOfCalls(configuration.getMinimumNumberOfCalls())
+        .build();
+  }
+
+  private static CircuitBreakerConfig.SlidingWindowType getSlidingWindowType(
+      CircuitBreakerThresholds.SlidingWindowType slidingWindowType) {
+    switch (slidingWindowType) {
+      case COUNT_BASED:
+        return CircuitBreakerConfig.SlidingWindowType.COUNT_BASED;
+      case TIME_BASED:
+      default:
+        return CircuitBreakerConfig.SlidingWindowType.TIME_BASED;
+    }
+  }
+}

grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/resilience/ResilienceCircuitBreakerFactory.java

@@ -0,0 +1,29 @@
+package org.hypertrace.circuitbreaker.grpcutils.resilience;
+
+import io.github.resilience4j.circuitbreaker.CircuitBreakerConfig;
+import io.github.resilience4j.circuitbreaker.CircuitBreakerRegistry;
+import java.time.Clock;
+import java.util.Map;
+import org.hypertrace.circuitbreaker.grpcutils.CircuitBreakerConfiguration;
+
+public class ResilienceCircuitBreakerFactory {
+  public static ResilienceCircuitBreakerInterceptor getResilienceCircuitBreakerInterceptor(
+      CircuitBreakerConfiguration<?> circuitBreakerConfiguration, Clock clock) {
+    Map<String, CircuitBreakerConfig> resilienceCircuitBreakerConfigMap =
+        ResilienceCircuitBreakerConfigConverter.getCircuitBreakerConfigs(
+            circuitBreakerConfiguration.getCircuitBreakerThresholdsMap());
+    CircuitBreakerRegistry resilienceCircuitBreakerRegistry =
+        new ResilienceCircuitBreakerRegistryProvider(
+                circuitBreakerConfiguration.getDefaultThresholds())
+            .getCircuitBreakerRegistry();
+    ResilienceCircuitBreakerProvider resilienceCircuitBreakerProvider =
+        new ResilienceCircuitBreakerProvider(
+            resilienceCircuitBreakerRegistry,
+            resilienceCircuitBreakerConfigMap,
+            ResilienceCircuitBreakerConfigConverter.getDisabledKeys(
+                circuitBreakerConfiguration.getCircuitBreakerThresholdsMap()),
+            circuitBreakerConfiguration.getDefaultThresholds().isEnabled());
+    return new ResilienceCircuitBreakerInterceptor(
+        circuitBreakerConfiguration, clock, resilienceCircuitBreakerProvider);
+  }
+}