grpc-js-xds: Implement EDS dualstack support

Author: murgatroid99

packages/grpc-js-xds/src/environment.ts

@@ -22,3 +22,4 @@ export const EXPERIMENTAL_FEDERATION = (process.env.GRPC_EXPERIMENTAL_XDS_FEDERA
 export const EXPERIMENTAL_CUSTOM_LB_CONFIG = (process.env.GRPC_EXPERIMENTAL_XDS_CUSTOM_LB_CONFIG ?? 'false') === 'true';
 export const EXPERIMENTAL_RING_HASH = (process.env.GRPC_XDS_EXPERIMENTAL_ENABLE_RING_HASH ?? 'false') === 'true';
 export const EXPERIMENTAL_PICK_FIRST = (process.env.GRPC_EXPERIMENTAL_PICKFIRST_LB_CONFIG ?? 'false') === 'true';
+export const EXPERIMENTAL_DUALSTACK_ENDPOINTS = (process.env.GRPC_EXPERIMENTAL_XDS_DUALSTACK_ENDPOINTS ?? 'false') === 'true';

packages/grpc-js-xds/src/load-balancer-xds-cluster-resolver.ts

@@ -17,7 +17,7 @@
 
 import { ChannelOptions, LoadBalancingConfig, Metadata, connectivityState, experimental, logVerbosity, status } from "@grpc/grpc-js";
 import { registerLoadBalancerType } from "@grpc/grpc-js/build/src/load-balancer";
-import { EXPERIMENTAL_OUTLIER_DETECTION } from "./environment";
+import { EXPERIMENTAL_DUALSTACK_ENDPOINTS, EXPERIMENTAL_OUTLIER_DETECTION } from "./environment";
 import { Locality__Output } from "./generated/envoy/config/core/v3/Locality";
 import { ClusterLoadAssignment__Output } from "./generated/envoy/config/endpoint/v3/ClusterLoadAssignment";
 import { LocalityEndpoint, PriorityChildRaw } from "./load-balancer-priority";
@@ -40,6 +40,7 @@ import parseLoadBalancingConfig = experimental.parseLoadBalancingConfig;
 import UnavailablePicker = experimental.UnavailablePicker;
 import { serverConfigEqual, validateXdsServerConfig, XdsServerConfig } from "./xds-bootstrap";
 import { EndpointResourceType } from "./xds-resource-type/endpoint-resource-type";
+import { SocketAddress__Output } from "./generated/envoy/config/core/v3/SocketAddress";
 
 const TRACER_NAME = 'xds_cluster_resolver';
 
@@ -175,13 +176,21 @@ function getEdsPriorities(edsUpdate: ClusterLoadAssignment__Output): PriorityEnt
       (lbEndpoint) => {
         /* The validator in the XdsClient class ensures that each endpoint has
          * a socket_address with an IP address and a port_value. */
-        const socketAddress = lbEndpoint.endpoint!.address!.socket_address!;
+        let socketAddresses: SocketAddress__Output[];
+        if (EXPERIMENTAL_DUALSTACK_ENDPOINTS) {
+          socketAddresses = [
+            lbEndpoint.endpoint!.address!.socket_address!,
+            ...lbEndpoint.endpoint!.additional_addresses.map(additionalAddress => additionalAddress.address!.socket_address!)
+          ];
+        } else {
+          socketAddresses = [lbEndpoint.endpoint!.address!.socket_address!];
+        }
         return {
           endpoint: {
-            addresses: [{
+            addresses: socketAddresses.map(socketAddress => ({
               host: socketAddress.address!,
-              port: socketAddress.port_value!,
-            }]
+              port: socketAddress.port_value!
+            }))
           },
           weight: lbEndpoint.load_balancing_weight?.value ?? 1
         };

packages/grpc-js-xds/src/xds-resource-type/endpoint-resource-type.ts

@@ -7,6 +7,7 @@ import { isIPv4, isIPv6 } from "net";
 import { Any__Output } from "../generated/google/protobuf/Any";
 import { EDS_TYPE_URL, decodeSingleResource } from "../resources";
 import { Watcher, XdsClient } from "../xds-client";
+import { EXPERIMENTAL_DUALSTACK_ENDPOINTS } from "../environment";
 
 const TRACER_NAME = 'xds_client';
 
@@ -39,6 +40,24 @@ export class EndpointResourceType extends XdsResourceType {
     return 'envoy.config.endpoint.v3.ClusterLoadAssignment';
   }
 
+  private validateAddress(socketAddress: SocketAddress__Output, seenAddresses: SocketAddress__Output[]): boolean {
+    if (socketAddress.port_specifier !== 'port_value') {
+      trace('EDS validation: socket_address.port_specifier !== "port_value"');
+      return false;
+    }
+    if (!(isIPv4(socketAddress.address) || isIPv6(socketAddress.address))) {
+      trace('EDS validation: address not a valid IPv4 or IPv6 address: ' + socketAddress.address);
+      return false;
+    }
+    for (const address of seenAddresses) {
+      if (addressesEqual(socketAddress, address)) {
+        trace('EDS validation: duplicate address seen: ' + address);
+        return false;
+      }
+    }
+    return true;
+  }
+
   private validateResource(message: ClusterLoadAssignment__Output): ClusterLoadAssignment__Output | null {
     const seenLocalities: {locality: Locality__Output, priority: number}[] = [];
     const seenAddresses: SocketAddress__Output[] = [];
@@ -61,21 +80,22 @@ export class EndpointResourceType extends XdsResourceType {
           trace('EDS validation: endpoint socket_address not set');
           return null;
         }
-        if (socketAddress.port_specifier !== 'port_value') {
-          trace('EDS validation: socket_address.port_specifier !== "port_value"');
-          return null;
-        }
-        if (!(isIPv4(socketAddress.address) || isIPv6(socketAddress.address))) {
-          trace('EDS validation: address not a valid IPv4 or IPv6 address: ' + socketAddress.address);
+        if (!this.validateAddress(socketAddress, seenAddresses)) {
           return null;
         }
-        for (const address of seenAddresses) {
-          if (addressesEqual(socketAddress, address)) {
-            trace('EDS validation: duplicate address seen: ' + address);
-            return null;
+        seenAddresses.push(socketAddress);
+        if (EXPERIMENTAL_DUALSTACK_ENDPOINTS && lb.endpoint?.additional_addresses) {
+          for (const additionalAddress of lb.endpoint.additional_addresses) {
+            if (!additionalAddress.address?.socket_address) {
+              trace('EDS validation: endpoint additional_addresses socket_address not set');
+              return null;
+            }
+            if (!this.validateAddress(additionalAddress.address.socket_address, seenAddresses)) {
+              return null;
+            }
+            seenAddresses.push(additionalAddress.address.socket_address);
           }
         }
-        seenAddresses.push(socketAddress);
       }
       priorityTotalWeights.set(endpoint.priority, (priorityTotalWeights.get(endpoint.priority) ?? 0) + (endpoint.load_balancing_weight?.value ?? 0));
     }