Moved some of the util classes for ext_proc to a new sub package io.grpc.xds.internal.extproc, with public access modifier:

1. Otel metric instruments out of the client interceptor
2. Moved `KnownLengthInputStream` used during observability mode to send message to the data plane out of the client interceptor.

Created a new class io.grpc.xds.ExternalProcessorUtil with the below. This class is kept here and not in the above sub-package because `toHeaderMap` operates on `HeaderMutationRulesConfig` which is package-private.
1. Method `outboundStreamFromByteString` used to construct the ProcessingRequest payload to Util as a static utility method.
2. Method `toHeaderMap` to Util.

Author: kannanjgithub

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

@@ -17,6 +17,12 @@
 package io.grpc.xds;
 
 import static com.google.common.base.Preconditions.checkNotNull;
+import static io.grpc.xds.ExternalProcessorUtil.outboundStreamToByteString;
+import static io.grpc.xds.ExternalProcessorUtil.toHeaderMap;
+import static io.grpc.xds.internal.extproc.ExternalProcessorMetricInstruments.clientHalfCloseDuration;
+import static io.grpc.xds.internal.extproc.ExternalProcessorMetricInstruments.clientHeadersDuration;
+import static io.grpc.xds.internal.extproc.ExternalProcessorMetricInstruments.serverHeadersDuration;
+import static io.grpc.xds.internal.extproc.ExternalProcessorMetricInstruments.serverTrailersDuration;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Joiner;
@@ -26,7 +32,6 @@
 import com.google.protobuf.ByteString;
 import com.google.protobuf.Struct;
 import com.google.protobuf.Value;
-import io.envoyproxy.envoy.config.core.v3.HeaderMap;
 import io.envoyproxy.envoy.extensions.filters.http.ext_proc.v3.ProcessingMode;
 import io.envoyproxy.envoy.service.ext_proc.v3.BodyMutation;
 import io.envoyproxy.envoy.service.ext_proc.v3.BodyResponse;
@@ -48,25 +53,24 @@
 import io.grpc.Context;
 import io.grpc.Deadline;
 import io.grpc.DoubleHistogramMetricInstrument;
-import io.grpc.Drainable;
 import io.grpc.ForwardingClientCall.SimpleForwardingClientCall;
 import io.grpc.ForwardingClientCallListener.SimpleForwardingClientCallListener;
-import io.grpc.KnownLength;
 import io.grpc.ManagedChannel;
 import io.grpc.Metadata;
 import io.grpc.MethodDescriptor;
-import io.grpc.MetricInstrumentRegistry;
 import io.grpc.MetricRecorder;
 import io.grpc.Status;
 import io.grpc.StatusRuntimeException;
 import io.grpc.internal.DelayedClientCall;
-import io.grpc.internal.GrpcUtil;
 import io.grpc.internal.SerializingExecutor;
 import io.grpc.stub.ClientCallStreamObserver;
 import io.grpc.stub.ClientResponseObserver;
+import io.grpc.xds.ExternalProcessorFilter.DataPlaneCallState;
+import io.grpc.xds.ExternalProcessorFilter.ExtProcStreamState;
 import io.grpc.xds.ExternalProcessorFilter.ExternalProcessorFilterConfig;
-import io.grpc.xds.ExternalProcessorFilter.HeaderForwardingRulesConfig;
 import io.grpc.xds.Filter.FilterContext;
+import io.grpc.xds.internal.extproc.ExternalProcessorMetricInstruments;
+import io.grpc.xds.internal.extproc.KnownLengthInputStream;
 import io.grpc.xds.internal.grpcservice.CachedChannelManager;
 import io.grpc.xds.internal.grpcservice.HeaderValue;
 import io.grpc.xds.internal.grpcservice.HeaderValueValidationUtils;
@@ -98,76 +102,6 @@
  */
 final class ExternalProcessorClientInterceptor implements ClientInterceptor {
 
-
-
-  @VisibleForTesting
-  static final DoubleHistogramMetricInstrument clientHeadersDuration;
-  @VisibleForTesting
-  static final DoubleHistogramMetricInstrument clientHalfCloseDuration;
-  @VisibleForTesting
-  static final DoubleHistogramMetricInstrument serverHeadersDuration;
-  @VisibleForTesting
-  static final 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 {
-    if (GrpcUtil.getFlag("GRPC_EXPERIMENTAL_XDS_EXT_PROC_ON_CLIENT", false)) {
-      MetricInstrumentRegistry registry = MetricInstrumentRegistry.getDefaultRegistry();
-
-      clientHeadersDuration = registry.registerDoubleHistogram(
-          "grpc.client_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("grpc.target"),
-          ImmutableList.of("grpc.lb.backend_service"),
-          true);
-
-      clientHalfCloseDuration = registry.registerDoubleHistogram(
-          "grpc.client_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("grpc.target"),
-          ImmutableList.of("grpc.lb.backend_service"),
-          true);
-
-      serverHeadersDuration = registry.registerDoubleHistogram(
-          "grpc.client_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("grpc.target"),
-          ImmutableList.of("grpc.lb.backend_service"),
-          true);
-
-      serverTrailersDuration = registry.registerDoubleHistogram(
-          "grpc.client_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("grpc.target"),
-          ImmutableList.of("grpc.lb.backend_service"),
-          true);
-    } else {
-      clientHeadersDuration = null;
-      clientHalfCloseDuration = null;
-      serverHeadersDuration = null;
-      serverTrailersDuration = null;
-    }
-  }
-
   private final ExternalProcessorFilterConfig filterConfig;
   private final ScheduledExecutorService scheduler;
   private final MetricRecorder metricsRecorder;
@@ -182,6 +116,7 @@ final class ExternalProcessorClientInterceptor implements ClientInterceptor {
     checkNotNull(cachedChannelManager, "cachedChannelManager");
     this.scheduler = checkNotNull(scheduler, "scheduler");
     this.metricsRecorder = checkNotNull(context.metricsRecorder(), "metricsRecorder");
+    ExternalProcessorMetricInstruments.initMetricInstruments();
     this.extProcChannel = cachedChannelManager.getChannel(filterConfig.getGrpcServiceConfig());
   }
 
@@ -259,48 +194,6 @@ public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
   }
 
   // --- SHARED UTILITY METHODS ---
-  private static HeaderMap toHeaderMap(
-      Metadata metadata, Optional<HeaderForwardingRulesConfig> forwardRules) {
-    HeaderMap.Builder builder =
-        HeaderMap.newBuilder();
-
-    for (String key : metadata.keys()) {
-      if (forwardRules.isPresent() && !forwardRules.get().isAllowed(key)) {
-        continue;
-      }
-      // Map binary headers using raw_value
-      if (key.endsWith(Metadata.BINARY_HEADER_SUFFIX)) {
-        Metadata.Key<byte[]> binKey = Metadata.Key.of(key, Metadata.BINARY_BYTE_MARSHALLER);
-        Iterable<byte[]> values = metadata.getAll(binKey);
-        if (values != null) {
-          for (byte[] binValue : values) {
-            String base64Value = BaseEncoding.base64().encode(binValue);
-            io.envoyproxy.envoy.config.core.v3.HeaderValue headerValue =
-                io.envoyproxy.envoy.config.core.v3.HeaderValue.newBuilder()
-                    .setKey(key.toLowerCase(Locale.ROOT))
-                    .setRawValue(ByteString.copyFromUtf8(base64Value))
-                    .build();
-            builder.addHeaders(headerValue);
-          }
-        }
-      } else {
-        Metadata.Key<String> asciiKey = Metadata.Key.of(key, Metadata.ASCII_STRING_MARSHALLER);
-        Iterable<String> values = metadata.getAll(asciiKey);
-        if (values != null) {
-          for (String value : values) {
-            io.envoyproxy.envoy.config.core.v3.HeaderValue headerValue =
-                io.envoyproxy.envoy.config.core.v3.HeaderValue.newBuilder()
-                    .setKey(key.toLowerCase(Locale.ROOT))
-                    .setRawValue(ByteString.copyFromUtf8(value))
-                    .build();
-            builder.addHeaders(headerValue);
-          }
-        }
-      }
-    }
-    return builder.build();
-  }
-
   private static ImmutableMap<String, Struct> collectAttributes(
       ImmutableList<String> requestedAttributes,
       MethodDescriptor<?, ?> method,
@@ -407,31 +300,6 @@ private static class DataPlaneDelayedCall<ReqT, RespT> extends DelayedClientCall
    */
   private static class DataPlaneClientCall 
       extends SimpleForwardingClientCall<InputStream, InputStream> {
-    enum ExtProcStreamState {
-      ACTIVE,
-      DRAINING,
-      COMPLETED,
-      FAILED;
-
-      boolean isCompleted() {
-        return this == COMPLETED || this == FAILED;
-      }
-
-      boolean isFailed() {
-        return this == FAILED;
-      }
-
-      boolean isDraining() {
-        return this == DRAINING;
-      }
-    }
-
-    enum DataPlaneCallState {
-      IDLE,
-      ACTIVE,
-      CLOSED
-    }
-
     private enum EventType {
       REQUEST_HEADERS,
       REQUEST_BODY,
@@ -1395,7 +1263,7 @@ public void onMessage(InputStream message) {
     @Override
     public void onClose(Status status, Metadata trailers) {
       dataPlaneClientCall.setServerTrailersStartNanos(System.nanoTime());
-      DataPlaneClientCall.ExtProcStreamState extProcStreamState =
+      ExtProcStreamState extProcStreamState =
           dataPlaneClientCall.getExtProcStreamState().get();
       if (extProcStreamState.isFailed()
           && !dataPlaneClientCall.getConfig().getObservabilityMode()
@@ -1593,46 +1461,4 @@ private void sendResponseBodyToExtProc(
           .build());
     }
   }
-
-
-
-  private static ByteString outboundStreamToByteString(InputStream message) throws IOException {
-    if (message instanceof Drainable) {
-      int size = message.available();
-      ByteString.Output output =
-          size > 0 ? ByteString.newOutput(size) : ByteString.newOutput();
-      ((Drainable) message).drainTo(output);
-      return output.toByteString();
-    }
-    return ByteString.readFrom(message);
-  }
-
-  private static final class KnownLengthInputStream extends InputStream
-      implements KnownLength {
-    private final InputStream delegate;
-
-    KnownLengthInputStream(ByteString byteString) {
-      this.delegate = byteString.newInput();
-    }
-
-    @Override
-    public int read() throws IOException {
-      return delegate.read();
-    }
-
-    @Override
-    public int read(byte[] b, int off, int len) throws IOException {
-      return delegate.read(b, off, len);
-    }
-
-    @Override
-    public int available() throws IOException {
-      return delegate.available();
-    }
-
-    @Override
-    public void close() throws IOException {
-      delegate.close();
-    }
-  }
 }

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

@@ -33,10 +33,9 @@
 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.extproc.ExternalProcessorMetricInstruments;
 import io.grpc.xds.internal.grpcservice.CachedChannelManager;
 import io.grpc.xds.internal.grpcservice.GrpcServiceConfig;
 import io.grpc.xds.internal.grpcservice.GrpcServiceParseException;
@@ -57,6 +56,35 @@ public class ExternalProcessorFilter implements Filter {
   static final String TYPE_URL = 
       "type.googleapis.com/envoy.extensions.filters.http.ext_proc.v3.ExternalProcessor";
 
+  static {
+    ExternalProcessorMetricInstruments.initMetricInstruments();
+  }
+
+  enum ExtProcStreamState {
+    ACTIVE,
+    DRAINING,
+    COMPLETED,
+    FAILED;
+
+    boolean isCompleted() {
+      return this == COMPLETED || this == FAILED;
+    }
+
+    boolean isFailed() {
+      return this == FAILED;
+    }
+
+    boolean isDraining() {
+      return this == DRAINING;
+    }
+  }
+
+  enum DataPlaneCallState {
+    IDLE,
+    ACTIVE,
+    CLOSED
+  }
+
   private final CachedChannelManager cachedChannelManager;
   private final FilterContext context;
 

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

@@ -25,14 +25,14 @@
  * A registry for all supported {@link Filter}s. Filters can be queried from the registry
  * by any of the {@link Filter.Provider#typeUrls() type URLs}.
  */
-final class FilterRegistry {
+public final class FilterRegistry {
   private static FilterRegistry instance;
 
   private final Map<String, Filter.Provider> supportedFilters = new HashMap<>();
 
   private FilterRegistry() {}
 
-  static synchronized FilterRegistry getDefaultRegistry() {
+  public static synchronized FilterRegistry getDefaultRegistry() {
     if (instance == null) {
       instance = newRegistry().register(
               new FaultFilter.Provider(),
@@ -45,7 +45,7 @@ static synchronized FilterRegistry getDefaultRegistry() {
   }
 
   @VisibleForTesting
-  static synchronized void reset() {
+  public static synchronized void reset() {
     instance = null;
   }
 
@@ -65,7 +65,7 @@ FilterRegistry register(Filter.Provider... filters) {
   }
 
   @Nullable
-  Filter.Provider get(String typeUrl) {
+  public Filter.Provider get(String typeUrl) {
     return supportedFilters.get(typeUrl);
   }
 }