[WIP] Tmp code to verify xDS HTTP Connect end-to-end

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

@@ -35,6 +35,8 @@
 import io.envoyproxy.envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager;
 import io.envoyproxy.envoy.extensions.load_balancing_policies.round_robin.v3.RoundRobin;
 import io.envoyproxy.envoy.extensions.load_balancing_policies.wrr_locality.v3.WrrLocality;
+import io.envoyproxy.envoy.extensions.transport_sockets.http_11_proxy.v3.Http11ProxyUpstreamTransport;
+import io.envoyproxy.envoy.extensions.transport_sockets.raw_buffer.v3.RawBuffer;
 import io.envoyproxy.envoy.extensions.transport_sockets.tls.v3.DownstreamTlsContext;
 import io.envoyproxy.envoy.extensions.transport_sockets.tls.v3.UpstreamTlsContext;
 import io.envoyproxy.envoy.service.discovery.v3.Resource;
@@ -72,8 +74,12 @@ private static JsonFormat.Printer newPrinter() {
               .add(ClusterConfig.getDescriptor())
               .add(ClusterLoadAssignment.getDescriptor())
               .add(WrrLocality.getDescriptor())
-              .add(TypedStruct.getDescriptor())
-              .add(RoundRobin.getDescriptor());
+              .add(RoundRobin.getDescriptor())
+              // Transports
+              .add(Http11ProxyUpstreamTransport.getDescriptor())
+              .add(RawBuffer.getDescriptor())
+              // Always last, guarding the semicolon.
+              .add(TypedStruct.getDescriptor());
       try {
         @SuppressWarnings("unchecked")
         Class<? extends Message> routeLookupClusterSpecifierClass =

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

@@ -40,6 +40,7 @@
 import io.envoyproxy.envoy.extensions.transport_sockets.http_11_proxy.v3.Http11ProxyUpstreamTransport;
 import io.envoyproxy.envoy.extensions.transport_sockets.tls.v3.CertificateValidationContext;
 import io.envoyproxy.envoy.extensions.transport_sockets.tls.v3.CommonTlsContext;
+import io.grpc.InternalLogId;
 import io.grpc.LoadBalancerRegistry;
 import io.grpc.NameResolver;
 import io.grpc.internal.GrpcUtil;
@@ -50,6 +51,8 @@
 import io.grpc.xds.XdsClusterResource.CdsUpdate;
 import io.grpc.xds.client.BackendMetricPropagation;
 import io.grpc.xds.client.XdsClient.ResourceUpdate;
+import io.grpc.xds.client.XdsLogger;
+import io.grpc.xds.client.XdsLogger.XdsLogLevel;
 import io.grpc.xds.client.XdsResourceType;
 import io.grpc.xds.internal.security.CommonTlsContextUtil;
 import java.util.List;
@@ -80,11 +83,24 @@ class XdsClusterResource extends XdsResourceType<CdsUpdate> {
       "type.googleapis.com/envoy.extensions.transport_sockets.tls.v3.UpstreamTlsContext";
   private static final String TYPE_URL_UPSTREAM_TLS_CONTEXT_V2 =
       "type.googleapis.com/envoy.api.v2.auth.UpstreamTlsContext";
-  static final String TRANSPORT_SOCKET_NAME_HTTP11_PROXY =
-      "type.googleapis.com/envoy.extensions.transport_sockets.http_11_proxy.v3"
-          + ".Http11ProxyUpstreamTransport";
+
+  // TODO(sergiitk): should this be in XdsResourceType next to TRANSPORT_SOCKET_NAME_TLS?
+  @VisibleForTesting
+  static final String TRANSPORT_SOCKET_NAME_HTTP11_PROXY = "envoy.transport_sockets.http_11_proxy";
+  @VisibleForTesting
+  static final String TRANSPORT_SOCKET_NAME_RAW_BUFFER = "envoy.transport_sockets.raw_buffer";
+
+  // TODO(sergiitk): tell Eric about the bug.
+  // static final String TRANSPORT_SOCKET_NAME_HTTP11_PROXY =
+  //     "type.googleapis.com/envoy.extensions.transport_sockets.http_11_proxy.v3"
+  //         + ".Http11ProxyUpstreamTransport";
+  // static final String TRANSPORT_SOCKET_NAME_RAW_BUFFER =
+  //     "type.googleapis.com/envoy.extensions.transport_sockets.raw_buffer.v3.RawBuffer";
+
   private final LoadBalancerRegistry loadBalancerRegistry
       = LoadBalancerRegistry.getDefaultRegistry();
+  private static final XdsLogger logger = XdsLogger.withLogId(
+        InternalLogId.allocate("xds-cluster-resource", "sergiitk"));
 
   private static final XdsClusterResource instance = new XdsClusterResource();
 
@@ -260,22 +276,44 @@ private static StructOrError<CdsUpdate.Builder> parseNonAggregateCluster(
     boolean hasTransportSocket = cluster.hasTransportSocket();
     TransportSocket transportSocket = cluster.getTransportSocket();
 
-    if (hasTransportSocket && !TRANSPORT_SOCKET_NAME_TLS.equals(transportSocket.getName())
-        && !(isEnabledXdsHttpConnect
-        && TRANSPORT_SOCKET_NAME_HTTP11_PROXY.equals(transportSocket.getName()))) {
-      return StructOrError.fromError(
-          "transport-socket with name " + transportSocket.getName() + " not supported.");
+
+    String transportSocketName = hasTransportSocket ? transportSocket.getName() : "<invalid>";
+    boolean socketIsTls = transportSocketName.equals(TRANSPORT_SOCKET_NAME_TLS);
+    boolean socketIsH1Proxy = transportSocketName.equals(TRANSPORT_SOCKET_NAME_HTTP11_PROXY);
+    boolean socketIsH1ProxyAndSupported = isEnabledXdsHttpConnect && socketIsH1Proxy;
+
+    if (hasTransportSocket && !(socketIsTls || socketIsH1ProxyAndSupported)) {
+      String errMsg = "transport-socket with name " + transportSocketName + " not supported";
+
+      @SuppressWarnings("AlreadyChecked") // todo remove
+      String extraErrorMsg = errMsg
+          + ", socketIsTls=" + socketIsTls
+          + ", socketIsH1Proxy=" + socketIsH1Proxy
+          + ", isEnabledXdsHttpConnect=" + isEnabledXdsHttpConnect
+          + ", socketIsH1ProxyAndSupported=" + socketIsH1ProxyAndSupported + ".";
+      logger.log(XdsLogLevel.FORCE_WARNING, extraErrorMsg);
+
+      return StructOrError.fromError(errMsg + ".");
     }
 
-    if (hasTransportSocket && isEnabledXdsHttpConnect
-        && TRANSPORT_SOCKET_NAME_HTTP11_PROXY.equals(transportSocket.getName())) {
+    if (hasTransportSocket && socketIsH1ProxyAndSupported) {
+      logger.log(XdsLogLevel.FORCE_INFO, "Enabling HTTP11 Proxy, trying to parse");
       isHttp11ProxyAvailable = true;
+
       try {
         Http11ProxyUpstreamTransport wrappedTransportSocket = transportSocket
-            .getTypedConfig().unpack(io.envoyproxy.envoy.extensions.transport_sockets
-                .http_11_proxy.v3.Http11ProxyUpstreamTransport.class);
+            .getTypedConfig().unpack(Http11ProxyUpstreamTransport.class);
+
         hasTransportSocket = wrappedTransportSocket.hasTransportSocket();
         transportSocket = wrappedTransportSocket.getTransportSocket();
+        transportSocketName = hasTransportSocket ? transportSocket.getName() : "<wrapped-invalid>";
+
+        logger.log(XdsLogLevel.FORCE_INFO,
+            "Enabled HTTP11 Proxy, wrapped socket: " + transportSocketName
+        );
+
+        // TODO(sergiitk): tell Eric this is missing the wrapped socket check.
+
       } catch (InvalidProtocolBufferException e) {
         return StructOrError.fromError(
             "Cluster " + clusterName + ": malformed Http11ProxyUpstreamTransport: " + e);

xds/src/test/java/io/grpc/xds/ClusterResolverLoadBalancerTest.java

@@ -416,8 +416,7 @@ public void endpointAddressRewritten_whenProxyMetadataIsInEndpointMetadata() {
     XdsClusterResource.isEnabledXdsHttpConnect = true;
     Cluster cluster = EDS_CLUSTER.toBuilder()
         .setTransportSocket(TransportSocket.newBuilder()
-            .setName(
-                "type.googleapis.com/" + Http11ProxyUpstreamTransport.getDescriptor().getFullName())
+            .setName("envoy.transport_sockets.http_11_proxy")
             .setTypedConfig(Any.pack(Http11ProxyUpstreamTransport.getDefaultInstance())))
         .build();
     // Proxy address in endpointMetadata, and no proxy in locality metadata
@@ -466,8 +465,7 @@ public void endpointAddressRewritten_whenProxyMetadataIsInLocalityMetadata() {
     XdsClusterResource.isEnabledXdsHttpConnect = true;
     Cluster cluster = EDS_CLUSTER.toBuilder()
         .setTransportSocket(TransportSocket.newBuilder()
-            .setName(
-                "type.googleapis.com/" + Http11ProxyUpstreamTransport.getDescriptor().getFullName())
+            .setName("envoy.transport_sockets.http_11_proxy")
             .setTypedConfig(Any.pack(Http11ProxyUpstreamTransport.getDefaultInstance())))
         .build();
     // No proxy address in endpointMetadata, and proxy in locality metadata

xds/src/test/java/io/grpc/xds/GrpcXdsClientImplDataTest.java

@@ -19,6 +19,7 @@
 import static com.google.common.truth.Truth.assertThat;
 import static io.envoyproxy.envoy.config.route.v3.RouteAction.ClusterSpecifierCase.CLUSTER_SPECIFIER_PLUGIN;
 import static io.grpc.xds.XdsClusterResource.TRANSPORT_SOCKET_NAME_HTTP11_PROXY;
+import static io.grpc.xds.XdsClusterResource.TRANSPORT_SOCKET_NAME_RAW_BUFFER;
 import static io.grpc.xds.XdsEndpointResource.GRPC_EXPERIMENTAL_XDS_DUALSTACK_ENDPOINTS;
 import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.fail;
@@ -97,6 +98,7 @@
 import io.envoyproxy.envoy.extensions.load_balancing_policies.client_side_weighted_round_robin.v3.ClientSideWeightedRoundRobin;
 import io.envoyproxy.envoy.extensions.load_balancing_policies.wrr_locality.v3.WrrLocality;
 import io.envoyproxy.envoy.extensions.transport_sockets.http_11_proxy.v3.Http11ProxyUpstreamTransport;
+import io.envoyproxy.envoy.extensions.transport_sockets.raw_buffer.v3.RawBuffer;
 import io.envoyproxy.envoy.extensions.transport_sockets.tls.v3.CertificateProviderPluginInstance;
 import io.envoyproxy.envoy.extensions.transport_sockets.tls.v3.CertificateValidationContext;
 import io.envoyproxy.envoy.extensions.transport_sockets.tls.v3.CommonTlsContext;
@@ -2639,6 +2641,50 @@ public void parseNonAggregateCluster_withHttp11ProxyTransportSocket() throws Exc
     assertThat(result.isHttp11ProxyAvailable()).isTrue();
   }
 
+  /**
+   * Http11ProxyUpstreamTransport with transport_socket=RawBuffer must behave the same as
+   * when transport_socket is unset.
+   */
+  @Test
+  public void parseNonAggregateCluster_withHttp11ProxyTransportSocket_rawBuffer() throws Exception {
+    XdsClusterResource.isEnabledXdsHttpConnect = true;
+
+    // Nested transport_socket (Http11ProxyUpstreamTransport.transport_socket).
+    TransportSocket transportSocketRawBuffer = TransportSocket.newBuilder()
+        .setName(TRANSPORT_SOCKET_NAME_RAW_BUFFER)
+        .setTypedConfig(Any.pack(RawBuffer.getDefaultInstance()))
+        .build();
+
+    Http11ProxyUpstreamTransport http11ProxyUpstreamTransport =
+        Http11ProxyUpstreamTransport.newBuilder()
+            .setTransportSocket(transportSocketRawBuffer)
+            .build();
+
+    TransportSocket transportSocket = TransportSocket.newBuilder()
+        .setName(TRANSPORT_SOCKET_NAME_HTTP11_PROXY)
+        .setTypedConfig(Any.pack(http11ProxyUpstreamTransport))
+        .build();
+
+    Cluster cluster = Cluster.newBuilder()
+        .setName("cluster-http11-proxy.googleapis.com")
+        .setType(DiscoveryType.EDS)
+        .setEdsClusterConfig(
+            EdsClusterConfig.newBuilder()
+                .setEdsConfig(
+                    ConfigSource.newBuilder().setAds(AggregatedConfigSource.getDefaultInstance()))
+                .setServiceName("service-http11-proxy.googleapis.com"))
+        .setLbPolicy(LbPolicy.ROUND_ROBIN)
+        .setTransportSocket(transportSocket)
+        .build();
+
+    CdsUpdate result =
+        XdsClusterResource.processCluster(cluster, null, LRS_SERVER_INFO,
+            LoadBalancerRegistry.getDefaultRegistry());
+
+    assertThat(result).isNotNull();
+    assertThat(result.isHttp11ProxyAvailable()).isTrue();
+  }
+
   @Test
   public void processCluster_parsesOrcaLrsPropagationMetrics() throws ResourceInvalidException {
     LoadStatsManager2.isEnabledOrcaLrsPropagation = true;

xds/third_party/envoy/import.sh

@@ -96,6 +96,7 @@ envoy/extensions/load_balancing_policies/ring_hash/v3/ring_hash.proto
 envoy/extensions/load_balancing_policies/round_robin/v3/round_robin.proto
 envoy/extensions/load_balancing_policies/wrr_locality/v3/wrr_locality.proto
 envoy/extensions/transport_sockets/http_11_proxy/v3/upstream_http_11_connect.proto
+envoy/extensions/transport_sockets/raw_buffer/v3/raw_buffer.proto
 envoy/extensions/transport_sockets/tls/v3/cert.proto
 envoy/extensions/transport_sockets/tls/v3/common.proto
 envoy/extensions/transport_sockets/tls/v3/secret.proto

xds/third_party/envoy/src/main/proto/envoy/extensions/transport_sockets/raw_buffer/v3/raw_buffer.proto

@@ -0,0 +1,21 @@
+syntax = "proto3";
+
+package envoy.extensions.transport_sockets.raw_buffer.v3;
+
+import "udpa/annotations/status.proto";
+import "udpa/annotations/versioning.proto";
+
+option java_package = "io.envoyproxy.envoy.extensions.transport_sockets.raw_buffer.v3";
+option java_outer_classname = "RawBufferProto";
+option java_multiple_files = true;
+option go_package = "github.com/envoyproxy/go-control-plane/envoy/extensions/transport_sockets/raw_buffer/v3;raw_bufferv3";
+option (udpa.annotations.file_status).package_version_status = ACTIVE;
+
+// [#protodoc-title: Raw Buffer]
+// [#extension: envoy.transport_sockets.raw_buffer]
+
+// Configuration for raw buffer transport socket.
+message RawBuffer {
+  option (udpa.annotations.versioning).previous_message_type =
+      "envoy.config.transport_socket.raw_buffer.v2.RawBuffer";
+}