xds: Refactor to allow common code usage between the ext_proc client and server interceptors (#12883)

Moved some of the util classes for ext_proc to a new sub package io.grpc.xds.internal.extproc.

Author: kannanjgithub

xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java

@@ -29,22 +29,17 @@
 import io.envoyproxy.envoy.extensions.filters.http.ext_proc.v3.ExtProcOverrides;
 import io.envoyproxy.envoy.extensions.filters.http.ext_proc.v3.ExtProcPerRoute;
 import io.envoyproxy.envoy.extensions.filters.http.ext_proc.v3.ExternalProcessor;
-import io.envoyproxy.envoy.extensions.filters.http.ext_proc.v3.HeaderForwardingRules;
 import io.envoyproxy.envoy.extensions.filters.http.ext_proc.v3.ProcessingMode;
 import io.grpc.ClientInterceptor;
 import io.grpc.internal.GrpcUtil;
-import io.grpc.xds.Filter.FilterConfigParseContext;
-import io.grpc.xds.Filter.FilterContext;
-import io.grpc.xds.internal.MatcherParser;
-import io.grpc.xds.internal.Matchers;
+import io.grpc.xds.internal.HeaderForwardingRulesConfig;
 import io.grpc.xds.internal.grpcservice.CachedChannelManager;
 import io.grpc.xds.internal.grpcservice.GrpcServiceConfig;
 import io.grpc.xds.internal.grpcservice.GrpcServiceParseException;
 import io.grpc.xds.internal.headermutations.HeaderMutationRulesConfig;
 import io.grpc.xds.internal.headermutations.HeaderMutationRulesParseException;
 import io.grpc.xds.internal.headermutations.HeaderMutationRulesParser;
 import java.util.List;
-import java.util.Locale;
 import java.util.Optional;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
@@ -397,51 +392,5 @@ GrpcServiceConfig getGrpcServiceConfig() {
     }
   }
 
-  static final class HeaderForwardingRulesConfig {
-    private final ImmutableList<Matchers.StringMatcher> allowedHeaders;
-    private final ImmutableList<Matchers.StringMatcher> disallowedHeaders;
 
-    HeaderForwardingRulesConfig(
-        ImmutableList<Matchers.StringMatcher> allowedHeaders,
-        ImmutableList<Matchers.StringMatcher> disallowedHeaders) {
-      this.allowedHeaders = checkNotNull(allowedHeaders, "allowedHeaders");
-      this.disallowedHeaders = checkNotNull(disallowedHeaders, "disallowedHeaders");
-    }
-
-    static HeaderForwardingRulesConfig create(HeaderForwardingRules proto) {
-      ImmutableList<Matchers.StringMatcher> allowedHeaders = ImmutableList.of();
-      if (proto.hasAllowedHeaders()) {
-        allowedHeaders = MatcherParser.parseListStringMatcher(proto.getAllowedHeaders());
-      }
-      ImmutableList<Matchers.StringMatcher> disallowedHeaders = ImmutableList.of();
-      if (proto.hasDisallowedHeaders()) {
-        disallowedHeaders = MatcherParser.parseListStringMatcher(proto.getDisallowedHeaders());
-      }
-      return new HeaderForwardingRulesConfig(allowedHeaders, disallowedHeaders);
-    }
-
-    boolean isAllowed(String headerName) {
-      String lowerHeaderName = headerName.toLowerCase(Locale.ROOT);
-      if (!allowedHeaders.isEmpty()) {
-        boolean matched = false;
-        for (Matchers.StringMatcher matcher : allowedHeaders) {
-          if (matcher.matches(lowerHeaderName)) {
-            matched = true;
-            break;
-          }
-        }
-        if (!matched) {
-          return false;
-        }
-      }
-      if (!disallowedHeaders.isEmpty()) {
-        for (Matchers.StringMatcher matcher : disallowedHeaders) {
-          if (matcher.matches(lowerHeaderName)) {
-            return false;
-          }
-        }
-      }
-      return true;
-    }
-  }
 }

xds/src/main/java/io/grpc/xds/ExternalProcessorFilter.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright 2026 The gRPC Authors
+ *
+ * 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,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.grpc.xds.internal;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.common.collect.ImmutableList;
+import io.envoyproxy.envoy.extensions.filters.http.ext_proc.v3.HeaderForwardingRules;
+import java.util.Locale;
+
+/**
+ * Configuration for header forwarding rules in external processing.
+ */
+public final class HeaderForwardingRulesConfig {
+  private final ImmutableList<Matchers.StringMatcher> allowedHeaders;
+  private final ImmutableList<Matchers.StringMatcher> disallowedHeaders;
+
+  public HeaderForwardingRulesConfig(
+      ImmutableList<Matchers.StringMatcher> allowedHeaders,
+      ImmutableList<Matchers.StringMatcher> disallowedHeaders) {
+    this.allowedHeaders = checkNotNull(allowedHeaders, "allowedHeaders");
+    this.disallowedHeaders = checkNotNull(disallowedHeaders, "disallowedHeaders");
+  }
+
+  public static HeaderForwardingRulesConfig create(HeaderForwardingRules proto) {
+    ImmutableList<Matchers.StringMatcher> allowedHeaders = ImmutableList.of();
+    if (proto.hasAllowedHeaders()) {
+      allowedHeaders = MatcherParser.parseListStringMatcher(proto.getAllowedHeaders());
+    }
+    ImmutableList<Matchers.StringMatcher> disallowedHeaders = ImmutableList.of();
+    if (proto.hasDisallowedHeaders()) {
+      disallowedHeaders = MatcherParser.parseListStringMatcher(proto.getDisallowedHeaders());
+    }
+    return new HeaderForwardingRulesConfig(allowedHeaders, disallowedHeaders);
+  }
+
+  public boolean isAllowed(String headerName) {
+    String lowerHeaderName = headerName.toLowerCase(Locale.ROOT);
+    if (!allowedHeaders.isEmpty()) {
+      boolean matched = false;
+      for (Matchers.StringMatcher matcher : allowedHeaders) {
+        if (matcher.matches(lowerHeaderName)) {
+          matched = true;
+          break;
+        }
+      }
+      if (!matched) {
+        return false;
+      }
+    }
+    if (!disallowedHeaders.isEmpty()) {
+      for (Matchers.StringMatcher matcher : disallowedHeaders) {
+        if (matcher.matches(lowerHeaderName)) {
+          return false;
+        }
+      }
+    }
+    return true;
+  }
+}

xds/src/main/java/io/grpc/xds/internal/HeaderForwardingRulesConfig.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright 2026 The gRPC Authors
+ *
+ * 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,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.grpc.xds.internal.extproc;
+
+/**
+ * States of the call inside the data plane.
+ */
+public enum DataPlaneCallState {
+  IDLE,
+  ACTIVE,
+  CLOSED
+}

xds/src/main/java/io/grpc/xds/internal/extproc/DataPlaneCallState.java

@@ -0,0 +1,28 @@
+/*
+ * Copyright 2026 The gRPC Authors
+ *
+ * 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,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.grpc.xds.internal.extproc;
+
+/**
+ * The event type representing the phase of external processing.
+ */
+public enum EventType {
+  REQUEST_HEADERS,
+  REQUEST_BODY,
+  RESPONSE_HEADERS,
+  RESPONSE_BODY,
+  RESPONSE_TRAILERS
+}

xds/src/main/java/io/grpc/xds/internal/extproc/EventType.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright 2026 The gRPC Authors
+ *
+ * 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,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.grpc.xds.internal.extproc;
+
+/**
+ * States of the stream with the external processor.
+ */
+public enum ExtProcStreamState {
+  ACTIVE,
+  DRAINING,
+  COMPLETED,
+  FAILED;
+
+  public boolean isCompleted() {
+    return this == COMPLETED || this == FAILED;
+  }
+
+  public boolean isFailed() {
+    return this == FAILED;
+  }
+
+  public boolean isDraining() {
+    return this == DRAINING;
+  }
+}

xds/src/main/java/io/grpc/xds/internal/extproc/ExtProcStreamState.java

@@ -0,0 +1,100 @@
+/*
+ * Copyright 2026 The gRPC Authors
+ *
+ * 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,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.grpc.xds.internal.extproc;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ImmutableList;
+import io.grpc.DoubleHistogramMetricInstrument;
+import io.grpc.MetricInstrumentRegistry;
+import io.grpc.internal.GrpcUtil;
+import java.util.List;
+
+/**
+ * Holds metric instrument definitions for server-side external processing.
+ */
+public final class ExternalProcessorServerInterceptorMetricInstruments {
+  @VisibleForTesting
+  public static DoubleHistogramMetricInstrument clientHeadersDuration;
+  @VisibleForTesting
+  public static DoubleHistogramMetricInstrument clientHalfCloseDuration;
+  @VisibleForTesting
+  public static DoubleHistogramMetricInstrument serverHeadersDuration;
+  @VisibleForTesting
+  public static DoubleHistogramMetricInstrument serverTrailersDuration;
+
+  // Copied from io.grpc.opentelemetry.internal.OpenTelemetryConstants.LATENCY_BUCKETS
+  private static final List<Double> LATENCY_BUCKETS = ImmutableList.of(
+      0d,     0.00001d, 0.00005d, 0.0001d, 0.0003d, 0.0006d, 0.0008d, 0.001d, 0.002d,
+      0.003d, 0.004d,   0.005d,   0.006d,  0.008d,  0.01d,   0.013d,  0.016d, 0.02d,
+      0.025d, 0.03d,    0.04d,    0.05d,   0.065d,  0.08d,   0.1d,    0.13d,  0.16d,
+      0.2d,   0.25d,    0.3d,     0.4d,    0.5d,    0.65d,   0.8d,    1d,     2d,
+      5d,     10d,      20d,      50d,     100d);
+
+  static {
+    initMetricInstruments();
+  }
+
+  public static synchronized void initMetricInstruments() {
+    if (GrpcUtil.getFlag("GRPC_EXPERIMENTAL_XDS_EXT_PROC_ON_SERVER", false)) {
+      if (clientHeadersDuration == null) {
+        MetricInstrumentRegistry registry = MetricInstrumentRegistry.getDefaultRegistry();
+
+        clientHeadersDuration = registry.registerDoubleHistogram(
+            "grpc.server_ext_proc.client_headers_duration",
+            "Time between when the ext_proc filter sees the client's headers and when "
+                + "it allows those headers to continue on to the next filter",
+            "s",
+            LATENCY_BUCKETS,
+            ImmutableList.of(),
+            ImmutableList.of(),
+            true);
+
+        clientHalfCloseDuration = registry.registerDoubleHistogram(
+            "grpc.server_ext_proc.client_half_close_duration",
+            "Time between when the ext_proc filter sees the client's half-close and when "
+                + "it allows that half-close to continue on to the next filter",
+            "s",
+            LATENCY_BUCKETS,
+            ImmutableList.of(),
+            ImmutableList.of(),
+            true);
+
+        serverHeadersDuration = registry.registerDoubleHistogram(
+            "grpc.server_ext_proc.server_headers_duration",
+            "Time between when the ext_proc filter sees the server's headers and when "
+                + "it allows those headers to continue on to the next filter",
+            "s",
+            LATENCY_BUCKETS,
+            ImmutableList.of(),
+            ImmutableList.of(),
+            true);
+
+        serverTrailersDuration = registry.registerDoubleHistogram(
+            "grpc.server_ext_proc.server_trailers_duration",
+            "Time between when the ext_proc filter sees the server's trailers and when "
+                + "it allows those trailers to continue on to the next filter",
+            "s",
+            LATENCY_BUCKETS,
+            ImmutableList.of(),
+            ImmutableList.of(),
+            true);
+      }
+    }
+  }
+
+  private ExternalProcessorServerInterceptorMetricInstruments() {}
+}