xds: xDS-based HTTP CONNECT configuration (#11861)

Author: shivaspeaks

xds/BUILD.bazel

@@ -85,6 +85,7 @@ java_proto_library(
         "@envoy_api//envoy/extensions/load_balancing_policies/ring_hash/v3:pkg",
         "@envoy_api//envoy/extensions/load_balancing_policies/round_robin/v3:pkg",
         "@envoy_api//envoy/extensions/load_balancing_policies/wrr_locality/v3:pkg",
+        "@envoy_api//envoy/extensions/transport_sockets/http_11_proxy/v3:pkg",
         "@envoy_api//envoy/extensions/transport_sockets/tls/v3:pkg",
         "@envoy_api//envoy/service/discovery/v3:pkg",
         "@envoy_api//envoy/service/load_stats/v3:pkg",

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

@@ -243,7 +243,9 @@ private void handleClusterDiscovered() {
       }
 
       ClusterResolverConfig config = new ClusterResolverConfig(
-          Collections.unmodifiableList(instances), configOrError.getConfig());
+          Collections.unmodifiableList(instances),
+          configOrError.getConfig(),
+          root.result.isHttp11ProxyAvailable());
       if (childLb == null) {
         childLb = lbRegistry.getProvider(CLUSTER_RESOLVER_POLICY_NAME).newLoadBalancer(helper);
       }

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

@@ -25,6 +25,7 @@
 import com.google.protobuf.Struct;
 import io.grpc.Attributes;
 import io.grpc.EquivalentAddressGroup;
+import io.grpc.HttpConnectProxiedSocketAddress;
 import io.grpc.InternalLogId;
 import io.grpc.LoadBalancer;
 import io.grpc.LoadBalancerProvider;
@@ -59,6 +60,8 @@
 import io.grpc.xds.client.XdsClient.ResourceWatcher;
 import io.grpc.xds.client.XdsLogger;
 import io.grpc.xds.client.XdsLogger.XdsLogLevel;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.util.ArrayList;
@@ -430,8 +433,18 @@ public void run() {
                           .set(XdsAttributes.ATTR_SERVER_WEIGHT, weight)
                           .set(XdsAttributes.ATTR_ADDRESS_NAME, endpoint.hostname())
                           .build();
-                  EquivalentAddressGroup eag = new EquivalentAddressGroup(
-                      endpoint.eag().getAddresses(), attr);
+
+                  EquivalentAddressGroup eag;
+                  if (config.isHttp11ProxyAvailable()) {
+                    List<SocketAddress> rewrittenAddresses = new ArrayList<>();
+                    for (SocketAddress addr : endpoint.eag().getAddresses()) {
+                      rewrittenAddresses.add(rewriteAddress(
+                          addr, endpoint.endpointMetadata(), localityLbInfo.localityMetadata()));
+                    }
+                    eag = new EquivalentAddressGroup(rewrittenAddresses, attr);
+                  } else {
+                    eag = new EquivalentAddressGroup(endpoint.eag().getAddresses(), attr);
+                  }
                   eag = AddressFilter.setPathFilter(eag, Arrays.asList(priorityName, localityName));
                   addresses.add(eag);
                 }
@@ -469,6 +482,35 @@ public void run() {
         new EndpointsUpdated().run();
       }
 
+      private SocketAddress rewriteAddress(SocketAddress addr,
+          ImmutableMap<String, Object> endpointMetadata,
+          ImmutableMap<String, Object> localityMetadata) {
+        if (!(addr instanceof InetSocketAddress)) {
+          return addr;
+        }
+
+        SocketAddress proxyAddress;
+        try {
+          proxyAddress = (SocketAddress) endpointMetadata.get(
+              "envoy.http11_proxy_transport_socket.proxy_address");
+          if (proxyAddress == null) {
+            proxyAddress = (SocketAddress) localityMetadata.get(
+                "envoy.http11_proxy_transport_socket.proxy_address");
+          }
+        } catch (ClassCastException e) {
+          return addr;
+        }
+
+        if (proxyAddress == null) {
+          return addr;
+        }
+
+        return HttpConnectProxiedSocketAddress.newBuilder()
+            .setTargetAddress((InetSocketAddress) addr)
+            .setProxyAddress(proxyAddress)
+            .build();
+      }
+
       private List<String> generatePriorityNames(String name,
           Map<Locality, LocalityLbEndpoints> localityLbEndpoints) {
         TreeMap<Integer, List<Locality>> todo = new TreeMap<>();

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

@@ -74,10 +74,17 @@ static final class ClusterResolverConfig {
     final List<DiscoveryMechanism> discoveryMechanisms;
     // GracefulSwitch configuration
     final Object lbConfig;
+    private final boolean isHttp11ProxyAvailable;
 
-    ClusterResolverConfig(List<DiscoveryMechanism> discoveryMechanisms, Object lbConfig) {
+    ClusterResolverConfig(List<DiscoveryMechanism> discoveryMechanisms, Object lbConfig,
+        boolean isHttp11ProxyAvailable) {
       this.discoveryMechanisms = checkNotNull(discoveryMechanisms, "discoveryMechanisms");
       this.lbConfig = checkNotNull(lbConfig, "lbConfig");
+      this.isHttp11ProxyAvailable = isHttp11ProxyAvailable;
+    }
+
+    boolean isHttp11ProxyAvailable() {
+      return isHttp11ProxyAvailable;
     }
 
     @Override

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

@@ -21,6 +21,7 @@
 import com.google.auto.value.AutoValue;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
 import io.grpc.EquivalentAddressGroup;
 import java.net.InetSocketAddress;
 import java.util.List;
@@ -41,11 +42,13 @@ abstract static class LocalityLbEndpoints {
     // Locality's priority level.
     abstract int priority();
 
+    abstract ImmutableMap<String, Object> localityMetadata();
+
     static LocalityLbEndpoints create(List<LbEndpoint> endpoints, int localityWeight,
-        int priority) {
+        int priority, ImmutableMap<String, Object> localityMetadata) {
       checkArgument(localityWeight > 0, "localityWeight must be greater than 0");
       return new AutoValue_Endpoints_LocalityLbEndpoints(
-          ImmutableList.copyOf(endpoints), localityWeight, priority);
+          ImmutableList.copyOf(endpoints), localityWeight, priority, localityMetadata);
     }
   }
 
@@ -63,17 +66,20 @@ abstract static class LbEndpoint {
 
     abstract String hostname();
 
+    abstract ImmutableMap<String, Object> endpointMetadata();
+
     static LbEndpoint create(EquivalentAddressGroup eag, int loadBalancingWeight,
-        boolean isHealthy, String hostname) {
-      return new AutoValue_Endpoints_LbEndpoint(eag, loadBalancingWeight, isHealthy, hostname);
+        boolean isHealthy, String hostname, ImmutableMap<String, Object> endpointMetadata) {
+      return new AutoValue_Endpoints_LbEndpoint(
+          eag, loadBalancingWeight, isHealthy, hostname, endpointMetadata);
     }
 
     // Only for testing.
     @VisibleForTesting
-    static LbEndpoint create(
-        String address, int port, int loadBalancingWeight, boolean isHealthy, String hostname) {
+    static LbEndpoint create(String address, int port, int loadBalancingWeight, boolean isHealthy,
+        String hostname, ImmutableMap<String, Object> endpointMetadata) {
       return LbEndpoint.create(new EquivalentAddressGroup(new InetSocketAddress(address, port)),
-          loadBalancingWeight, isHealthy, hostname);
+          loadBalancingWeight, isHealthy, hostname, endpointMetadata);
     }
   }
 

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

@@ -36,6 +36,7 @@
 import io.grpc.Status;
 import io.grpc.auth.MoreCallCredentials;
 import io.grpc.xds.MetadataRegistry.MetadataValueParser;
+import io.grpc.xds.client.XdsResourceType.ResourceInvalidException;
 import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.concurrent.ScheduledExecutorService;
@@ -240,11 +241,16 @@ public String getTypeUrl() {
     }
 
     @Override
-    public String parse(Any any) throws InvalidProtocolBufferException {
-      Audience audience = any.unpack(Audience.class);
+    public String parse(Any any) throws ResourceInvalidException {
+      Audience audience;
+      try {
+        audience = any.unpack(Audience.class);
+      } catch (InvalidProtocolBufferException ex) {
+        throw new ResourceInvalidException("Invalid Resource in address proto", ex);
+      }
       String url = audience.getUrl();
       if (url.isEmpty()) {
-        throw new InvalidProtocolBufferException(
+        throw new ResourceInvalidException(
             "Audience URL is empty. Metadata value must contain a valid URL.");
       }
       return url;

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

@@ -17,9 +17,14 @@
 package io.grpc.xds;
 
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ImmutableMap;
 import com.google.protobuf.Any;
-import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.Struct;
+import io.envoyproxy.envoy.config.core.v3.Metadata;
 import io.grpc.xds.GcpAuthenticationFilter.AudienceMetadataParser;
+import io.grpc.xds.XdsEndpointResource.AddressMetadataParser;
+import io.grpc.xds.client.XdsResourceType.ResourceInvalidException;
+import io.grpc.xds.internal.ProtobufJsonConverter;
 import java.util.HashMap;
 import java.util.Map;
 
@@ -36,6 +41,7 @@ final class MetadataRegistry {
 
   private MetadataRegistry() {
     registerParser(new AudienceMetadataParser());
+    registerParser(new AddressMetadataParser());
   }
 
   static MetadataRegistry getInstance() {
@@ -55,6 +61,54 @@ void removeParser(MetadataValueParser parser) {
     supportedParsers.remove(parser.getTypeUrl());
   }
 
+  /**
+   * Parses cluster metadata into a structured map.
+   *
+   * <p>Values in {@code typed_filter_metadata} take precedence over
+   * {@code filter_metadata} when keys overlap, following Envoy API behavior. See
+   * <a href="https://github.com/envoyproxy/envoy/blob/main/api/envoy/config/core/v3/base.proto#L217-L259">
+   *   Envoy metadata documentation </a> for details.
+   *
+   * @param metadata the {@link Metadata} containing the fields to parse.
+   * @return an immutable map of parsed metadata.
+   * @throws ResourceInvalidException if parsing {@code typed_filter_metadata} fails.
+   */
+  public ImmutableMap<String, Object> parseMetadata(Metadata metadata)
+      throws ResourceInvalidException {
+    ImmutableMap.Builder<String, Object> parsedMetadata = ImmutableMap.builder();
+
+    // Process typed_filter_metadata
+    for (Map.Entry<String, Any> entry : metadata.getTypedFilterMetadataMap().entrySet()) {
+      String key = entry.getKey();
+      Any value = entry.getValue();
+      MetadataValueParser parser = findParser(value.getTypeUrl());
+      if (parser != null) {
+        try {
+          Object parsedValue = parser.parse(value);
+          parsedMetadata.put(key, parsedValue);
+        } catch (ResourceInvalidException e) {
+          throw new ResourceInvalidException(
+              String.format("Failed to parse metadata key: %s, type: %s. Error: %s",
+                  key, value.getTypeUrl(), e.getMessage()), e);
+        }
+      }
+    }
+    // building once to reuse in the next loop
+    ImmutableMap<String, Object> intermediateParsedMetadata = parsedMetadata.build();
+
+    // Process filter_metadata for remaining keys
+    for (Map.Entry<String, Struct> entry : metadata.getFilterMetadataMap().entrySet()) {
+      String key = entry.getKey();
+      if (!intermediateParsedMetadata.containsKey(key)) {
+        Struct structValue = entry.getValue();
+        Object jsonValue = ProtobufJsonConverter.convertToJson(structValue);
+        parsedMetadata.put(key, jsonValue);
+      }
+    }
+
+    return parsedMetadata.build();
+  }
+
   interface MetadataValueParser {
 
     String getTypeUrl();
@@ -64,8 +118,8 @@ interface MetadataValueParser {
      *
      * @param any the {@link Any} object to parse.
      * @return the parsed metadata value.
-     * @throws InvalidProtocolBufferException if the parsing fails.
+     * @throws ResourceInvalidException if the parsing fails.
      */
-    Object parse(Any any) throws InvalidProtocolBufferException;
+    Object parse(Any any) throws ResourceInvalidException;
   }
 }