Add grpc circuit breaker utility using interceptors

grpc-circuitbreaker-utils/build.gradle.kts

@@ -0,0 +1,28 @@
+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")
+
+  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,94 @@
+package org.hypertrace.circuitbreaker.grpcutils;
+
+import com.typesafe.config.Config;
+import java.time.Duration;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class CircuitBreakerConfigParser {
+
+  private static final Set<String> nonThresholdKeys =
+      Set.of("enabled", "defaultCircuitBreakerKey", "defaultThresholds");
+
+  // 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";
+
+  public static <T> CircuitBreakerConfiguration.CircuitBreakerConfigurationBuilder<T> parseConfig(
+      Config config) {
+    CircuitBreakerConfiguration.CircuitBreakerConfigurationBuilder<T> builder =
+        CircuitBreakerConfiguration.builder();
+    if (config.hasPath(ENABLED)) {
+      builder.enabled(config.getBoolean(ENABLED));
+    }
+
+    if (config.hasPath(DEFAULT_THRESHOLDS)) {
+      builder.defaultThresholds(
+          buildCircuitBreakerThresholds(config.getConfig(DEFAULT_THRESHOLDS)));
+    } else {
+      builder.defaultThresholds(buildCircuitBreakerDefaultThresholds());
+    }
+
+    Map<String, CircuitBreakerThresholds> circuitBreakerThresholdsMap =
+        config.root().keySet().stream()
+            .filter(key -> !nonThresholdKeys.contains(key)) // Filter out non-threshold keys
+            .collect(
+                Collectors.toMap(
+                    key -> key, // Circuit breaker key
+                    key -> buildCircuitBreakerThresholds(config.getConfig(key))));
+    builder.circuitBreakerThresholdsMap(circuitBreakerThresholdsMap);
+    log.debug("Loaded circuit breaker configs: {}", builder);
+    return builder;
+  }
+
+  private static CircuitBreakerThresholds buildCircuitBreakerThresholds(Config config) {
+    return CircuitBreakerThresholds.builder()
+        .failureRateThreshold((float) config.getDouble(FAILURE_RATE_THRESHOLD))
+        .slowCallRateThreshold((float) config.getDouble(SLOW_CALL_RATE_THRESHOLD))
+        .slowCallDurationThreshold(config.getDuration(SLOW_CALL_DURATION_THRESHOLD))
+        .slidingWindowType(getSlidingWindowType(config.getString(SLIDING_WINDOW_TYPE)))
+        .slidingWindowSize(config.getInt(SLIDING_WINDOW_SIZE))
+        .waitDurationInOpenState(config.getDuration(WAIT_DURATION_IN_OPEN_STATE))
+        .permittedNumberOfCallsInHalfOpenState(
+            config.getInt(PERMITTED_NUMBER_OF_CALLS_IN_HALF_OPEN_STATE))
+        .minimumNumberOfCalls(config.getInt(MINIMUM_NUMBER_OF_CALLS))
+        .build();
+  }
+
+  public static CircuitBreakerThresholds buildCircuitBreakerDefaultThresholds() {
+    return CircuitBreakerThresholds.builder()
+        .failureRateThreshold(50f)
+        .slowCallRateThreshold(50f)
+        .slowCallDurationThreshold(Duration.ofSeconds(2))
+        .slidingWindowType(CircuitBreakerThresholds.SlidingWindowType.TIME_BASED)
+        .slidingWindowSize(60)
+        .waitDurationInOpenState(Duration.ofSeconds(60))
+        .permittedNumberOfCallsInHalfOpenState(5)
+        .minimumNumberOfCalls(10)
+        .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,28 @@
+package org.hypertrace.circuitbreaker.grpcutils;
+
+import io.grpc.Status;
+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;
+  // Default value be "global" if not override.
+  String defaultCircuitBreakerKey = "global";
+  // 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, RuntimeException> 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,31 @@
+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
+  float failureRateThreshold;
+  // Percentage of slow calls to trigger OPEN state
+  float slowCallRateThreshold;
+  // Define what a "slow" call is
+  Duration slowCallDurationThreshold;
+  // Number of calls to consider in the sliding window
+  SlidingWindowType slidingWindowType;
+  int slidingWindowSize;
+  // Time before retrying after OPEN state
+  Duration waitDurationInOpenState;
+  // Minimum calls before evaluating failure rate
+  int minimumNumberOfCalls;
+  // Calls allowed in HALF_OPEN state before deciding to
+  // CLOSE or OPEN again
+  int permittedNumberOfCallsInHalfOpenState;
+
+  public enum SlidingWindowType {
+    COUNT_BASED,
+    TIME_BASED
+  }
+}

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

@@ -0,0 +1,41 @@
+package org.hypertrace.circuitbreaker.grpcutils.resilience;
+
+import io.github.resilience4j.circuitbreaker.CircuitBreakerConfig;
+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())));
+  }
+
+  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,25 @@
+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 resilicenceCircuitBreakerRegistry =
+        new ResilienceCircuitBreakerRegistryProvider(
+                circuitBreakerConfiguration.getDefaultThresholds())
+            .getCircuitBreakerRegistry();
+    ResilienceCircuitBreakerProvider resilienceCircuitBreakerProvider =
+        new ResilienceCircuitBreakerProvider(
+            resilicenceCircuitBreakerRegistry, resilienceCircuitBreakerConfigMap);
+    return new ResilienceCircuitBreakerInterceptor(
+        circuitBreakerConfiguration, clock, resilienceCircuitBreakerProvider);
+  }
+}