grpc-js-xds: Complete federation implementation

Author: murgatroid99

packages/grpc-js-xds/package.json

@@ -45,7 +45,8 @@
   "dependencies": {
     "@grpc/proto-loader": "^0.6.0",
     "google-auth-library": "^7.0.2",
-    "re2-wasm": "^1.0.1"
+    "re2-wasm": "^1.0.1",
+    "vscode-uri": "^3.0.7"
   },
   "peerDependencies": {
     "@grpc/grpc-js": "~1.8.0"

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

@@ -15,19 +15,18 @@
  *
  */
 
-import { Node } from "./generated/envoy/config/core/v3/Node";
 import { ClientConfig, _envoy_service_status_v3_ClientConfig_GenericXdsConfig as GenericXdsConfig } from "./generated/envoy/service/status/v3/ClientConfig";
 import { ClientStatusDiscoveryServiceHandlers } from "./generated/envoy/service/status/v3/ClientStatusDiscoveryService";
 import { ClientStatusRequest__Output } from "./generated/envoy/service/status/v3/ClientStatusRequest";
 import { ClientStatusResponse } from "./generated/envoy/service/status/v3/ClientStatusResponse";
 import { Timestamp } from "./generated/google/protobuf/Timestamp";
-import { AdsTypeUrl, CDS_TYPE_URL, EDS_TYPE_URL, LDS_TYPE_URL, RDS_TYPE_URL } from "./resources";
-import { HandleResponseResult } from "./xds-stream-state/xds-stream-state";
+import { xdsResourceNameToString } from "./resources";
 import { sendUnaryData, ServerDuplexStream, ServerUnaryCall, status, experimental, loadPackageDefinition, logVerbosity } from '@grpc/grpc-js';
 import { loadSync } from "@grpc/proto-loader";
 import { ProtoGrpcType as CsdsProtoGrpcType } from "./generated/csds";
 
 import registerAdminService = experimental.registerAdminService;
+import { XdsClient } from "./xds-client";
 
 const TRACER_NAME = 'csds';
 
@@ -47,115 +46,46 @@ function dateToProtoTimestamp(date?: Date | null): Timestamp | null {
   }
 }
 
-let clientNode: Node | null = null;
+const registeredClients: XdsClient[] = [];
 
-const configStatus = {
-  [EDS_TYPE_URL]: new Map<string, GenericXdsConfig>(),
-  [CDS_TYPE_URL]: new Map<string, GenericXdsConfig>(),
-  [RDS_TYPE_URL]: new Map<string, GenericXdsConfig>(),
-  [LDS_TYPE_URL]: new Map<string, GenericXdsConfig>()
-};
-
-/**
- * This function only accepts a v3 Node message, because we are only supporting
- * v3 CSDS and it only handles v3 Nodes. If the client is actually using v2 xDS
- * APIs, it should just provide the equivalent v3 Node message.
- * @param node The Node message for the client that is requesting resources
- */
-export function setCsdsClientNode(node: Node) {
-  clientNode = node;
-}
-
-/**
- * Update the config status maps from the list of names of requested resources
- * for a specific type URL. These lists are the source of truth for determining
- * what resources will be listed in the CSDS response. Any resource that is not
- * in this list will never actually be applied anywhere.
- * @param typeUrl The resource type URL
- * @param names The list of resource names that are being requested
- */
-export function updateCsdsRequestedNameList(typeUrl: AdsTypeUrl, names: string[]) {
-  trace('Update type URL ' + typeUrl + ' with names [' + names + ']');
-  const currentTime = dateToProtoTimestamp(new Date());
-  const configMap = configStatus[typeUrl];
-  for (const name of names) {
-    if (!configMap.has(name)) {
-      configMap.set(name, {
-        type_url: typeUrl,
-        name: name,
-        last_updated: currentTime,
-        client_status: 'REQUESTED'
-      });
-    }
-  }
-  for (const name of configMap.keys()) {
-    if (!names.includes(name)) {
-      configMap.delete(name);
-    }
-  }
+export function registerXdsClientWithCsds(client: XdsClient) {
+  registeredClients.push(client);
 }
 
-/**
- * Update the config status maps from the result of parsing a single ADS
- * response. All resources that validated are considered "ACKED", and all
- * resources that failed validation are considered "NACKED".
- * @param typeUrl The type URL of resources in this response
- * @param versionInfo The version info field from this response
- * @param updates The lists of resources that passed and failed validation
- */
-export function updateCsdsResourceResponse(typeUrl: AdsTypeUrl, versionInfo: string, updates: HandleResponseResult) {
-  const currentTime = dateToProtoTimestamp(new Date());
-  const configMap = configStatus[typeUrl];
-  for (const {name, raw} of updates.accepted) {
-    const mapEntry = configMap.get(name);
-    if (mapEntry) {
-      trace('Updated ' + typeUrl + ' resource ' + name + ' to state ACKED');
-      mapEntry.client_status = 'ACKED';
-      mapEntry.version_info = versionInfo;
-      mapEntry.xds_config = raw;
-      mapEntry.error_state = null;
-      mapEntry.last_updated = currentTime;
-    }
-  }
-  for (const {name, error, raw} of updates.rejected) {
-    const mapEntry = configMap.get(name);
-    if (mapEntry) {
-      trace('Updated ' + typeUrl + ' resource ' + name + ' to state NACKED');
-      mapEntry.client_status = 'NACKED';
-      mapEntry.error_state = {
-        failed_configuration: raw,
-        last_update_attempt: currentTime,
-        details: error,
-        version_info: versionInfo
-      };
-    }
-  }
-  for (const name of updates.missing) {
-    const mapEntry = configMap.get(name);
-    if (mapEntry) {
-      trace('Updated ' + typeUrl + ' resource ' + name + ' to state DOES_NOT_EXIST');
-      mapEntry.client_status = 'DOES_NOT_EXIST';
-      mapEntry.version_info = versionInfo;
-      mapEntry.xds_config = null;
-      mapEntry.error_state = null;
-      mapEntry.last_updated = currentTime;
+function getCurrentConfigList(): ClientConfig[] {
+  const result: ClientConfig[] = [];
+  for (const client of registeredClients) {
+    if (!client.adsNode) {
+      continue;
     }
-  }
-}
-
-function getCurrentConfig(): ClientConfig {
-  const genericConfigList: GenericXdsConfig[] = [];
-  for (const configMap of Object.values(configStatus)) {
-    for (const configValue of configMap.values()) {
-      genericConfigList.push(configValue);
+    const genericConfigList: GenericXdsConfig[] = [];
+    for (const [authority, authorityState] of client.authorityStateMap) {
+      for (const [type, typeMap] of authorityState.resourceMap) {
+        for (const [key, resourceState] of typeMap) {
+          const typeUrl = type.getFullTypeUrl();
+          const meta = resourceState.meta;
+          genericConfigList.push({
+            name: xdsResourceNameToString({authority, key}, typeUrl),
+            client_status: meta.clientStatus,
+            version_info: meta.version,
+            xds_config: meta.clientStatus === 'ACKED' ? meta.rawResource : undefined,
+            last_updated: meta.updateTime ? dateToProtoTimestamp(meta.updateTime) : undefined,
+            error_state: meta.clientStatus === 'NACKED' ? {
+              details: meta.failedDetails,
+              failed_configuration: meta.rawResource,
+              last_update_attempt: meta.failedUpdateTime ? dateToProtoTimestamp(meta.failedUpdateTime) : undefined,
+              version_info: meta.failedVersion
+            } : undefined
+          });
+        }
+      }
     }
+    result.push({
+      node: client.adsNode,
+      generic_xds_configs: genericConfigList
+    });
   }
-  const config = {
-    node: clientNode,
-    generic_xds_configs: genericConfigList
-  };
-  trace('Sending current config ' + JSON.stringify(config, undefined, 2));
-  return config;
+  return result;
 }
 
 const csdsImplementation: ClientStatusDiscoveryServiceHandlers = {
@@ -169,7 +99,7 @@ const csdsImplementation: ClientStatusDiscoveryServiceHandlers = {
       return;
     }
     callback(null, {
-      config: [getCurrentConfig()]
+      config: getCurrentConfigList()
     });
   },
   StreamClientStatus(call: ServerDuplexStream<ClientStatusRequest__Output, ClientStatusResponse>) {
@@ -182,7 +112,7 @@ const csdsImplementation: ClientStatusDiscoveryServiceHandlers = {
         return;
       }
       call.write({
-        config: [getCurrentConfig()]
+        config: getCurrentConfigList()
       });
     });
     call.on('end', () => {

packages/grpc-js-xds/src/load-balancer-cds.ts

@@ -16,7 +16,7 @@
  */
 
 import { connectivityState, status, Metadata, logVerbosity, experimental } from '@grpc/grpc-js';
-import { getSingletonXdsClient, XdsClient } from './xds-client';
+import { getSingletonXdsClient, Watcher, XdsClient } from './xds-client';
 import { Cluster__Output } from './generated/envoy/config/cluster/v3/Cluster';
 import SubchannelAddress = experimental.SubchannelAddress;
 import UnavailablePicker = experimental.UnavailablePicker;
@@ -29,13 +29,12 @@ import OutlierDetectionLoadBalancingConfig = experimental.OutlierDetectionLoadBa
 import SuccessRateEjectionConfig = experimental.SuccessRateEjectionConfig;
 import FailurePercentageEjectionConfig = experimental.FailurePercentageEjectionConfig;
 import QueuePicker = experimental.QueuePicker;
-import { Watcher } from './xds-stream-state/xds-stream-state';
 import { OutlierDetection__Output } from './generated/envoy/config/cluster/v3/OutlierDetection';
 import { Duration__Output } from './generated/google/protobuf/Duration';
 import { EXPERIMENTAL_OUTLIER_DETECTION } from './environment';
 import { DiscoveryMechanism, XdsClusterResolverChildPolicyHandler, XdsClusterResolverLoadBalancingConfig } from './load-balancer-xds-cluster-resolver';
 import { CLUSTER_CONFIG_TYPE_URL, decodeSingleResource } from './resources';
-import { CdsUpdate, OutlierDetectionUpdate } from './xds-stream-state/cds-state';
+import { CdsUpdate, ClusterResourceType, OutlierDetectionUpdate } from './xds-resource-type/cluster-resource-type';
 
 const TRACER_NAME = 'cds_balancer';
 
@@ -195,8 +194,8 @@ export class CdsLoadBalancer implements LoadBalancer {
       return;
     }
     trace('Adding watcher for cluster ' + cluster);
-    const watcher: Watcher<CdsUpdate> = {
-      onValidUpdate: (update) => {
+    const watcher: Watcher<CdsUpdate> = new Watcher<CdsUpdate>({
+      onResourceChanged: (update) => {
         this.clusterTree[cluster].latestUpdate = update;
         if (update.type === 'AGGREGATE') {
           const children = update.aggregateChildren
@@ -227,15 +226,17 @@ export class CdsLoadBalancer implements LoadBalancer {
         }
       },
       onResourceDoesNotExist: () => {
+        trace('Received onResourceDoesNotExist update for cluster ' + cluster);
         if (cluster in this.clusterTree) {
           this.clusterTree[cluster].latestUpdate = undefined;
           this.clusterTree[cluster].children = [];
         }
         this.channelControlHelper.updateState(connectivityState.TRANSIENT_FAILURE, new UnavailablePicker({code: status.UNAVAILABLE, details: `CDS resource ${cluster} does not exist`, metadata: new Metadata()}));
         this.childBalancer.destroy();
       },
-      onTransientError: (statusObj) => {
+      onError: (statusObj) => {
         if (!this.updatedChild) {
+          trace('Transitioning to transient failure due to onError update for cluster' + cluster);
           this.channelControlHelper.updateState(
             connectivityState.TRANSIENT_FAILURE,
             new UnavailablePicker({
@@ -246,19 +247,23 @@ export class CdsLoadBalancer implements LoadBalancer {
           );
         }
       }
-    };
+    });
     this.clusterTree[cluster] = {
       watcher: watcher,
       children: []
     };
-    this.xdsClient?.addClusterWatcher(cluster, watcher);
+    if (this.xdsClient) {
+      ClusterResourceType.startWatch(this.xdsClient, cluster, watcher);
+    }
   }
 
   private removeCluster(cluster: string) {
     if (!(cluster in this.clusterTree)) {
       return;
     }
-    this.xdsClient?.removeClusterWatcher(cluster, this.clusterTree[cluster].watcher);
+    if (this.xdsClient) {
+      ClusterResourceType.cancelWatch(this.xdsClient, cluster, this.clusterTree[cluster].watcher);
+    }
     delete this.clusterTree[cluster];
   }
 

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

@@ -23,9 +23,8 @@ import { ClusterLoadAssignment__Output } from "./generated/envoy/config/endpoint
 import { LrsLoadBalancingConfig } from "./load-balancer-lrs";
 import { LocalitySubchannelAddress, PriorityChild, PriorityLoadBalancingConfig } from "./load-balancer-priority";
 import { WeightedTarget, WeightedTargetLoadBalancingConfig } from "./load-balancer-weighted-target";
-import { getSingletonXdsClient, XdsClient } from "./xds-client";
+import { getSingletonXdsClient, Watcher, XdsClient } from "./xds-client";
 import { DropCategory, XdsClusterImplLoadBalancingConfig } from "./load-balancer-xds-cluster-impl";
-import { Watcher } from "./xds-stream-state/xds-stream-state";
 
 import LoadBalancingConfig = experimental.LoadBalancingConfig;
 import validateLoadBalancingConfig = experimental.validateLoadBalancingConfig;
@@ -38,6 +37,7 @@ import ChannelControlHelper = experimental.ChannelControlHelper;
 import OutlierDetectionLoadBalancingConfig = experimental.OutlierDetectionLoadBalancingConfig;
 import subchannelAddressToString = experimental.subchannelAddressToString;
 import { serverConfigEqual, validateXdsServerConfig, XdsServerConfig } from "./xds-bootstrap";
+import { EndpointResourceType } from "./xds-resource-type/endpoint-resource-type";
 
 const TRACER_NAME = 'xds_cluster_resolver';
 
@@ -377,24 +377,26 @@ export class XdsClusterResolver implements LoadBalancer {
         };
         if (mechanism.type === 'EDS') {
           const edsServiceName = mechanism.eds_service_name ?? mechanism.cluster;
-          const watcher: Watcher<ClusterLoadAssignment__Output> = {
-            onValidUpdate: update => {
+          const watcher: Watcher<ClusterLoadAssignment__Output> = new Watcher<ClusterLoadAssignment__Output>({
+            onResourceChanged: update => {
               mechanismEntry.latestUpdate = getEdsPriorities(update);
               this.maybeUpdateChild();
             },
             onResourceDoesNotExist: () => {
               trace('Resource does not exist: ' + edsServiceName);
               mechanismEntry.latestUpdate = [{localities: [], dropCategories: []}];
             },
-            onTransientError: error => {
+            onError: error => {
               if (!mechanismEntry.latestUpdate) {
                 trace('xDS request failed with error ' + error);
                 mechanismEntry.latestUpdate = [{localities: [], dropCategories: []}];
               }
             }
-          };
+          });
           mechanismEntry.watcher = watcher;
-          this.xdsClient?.addEndpointWatcher(edsServiceName, watcher);
+          if (this.xdsClient) {
+            EndpointResourceType.startWatch(this.xdsClient, edsServiceName, watcher);
+          }
         } else {
           const resolver = createResolver({scheme: 'dns', path: mechanism.dns_hostname!}, {
             onSuccessfulResolution: addressList => {
@@ -434,7 +436,9 @@ export class XdsClusterResolver implements LoadBalancer {
     for (const mechanismEntry of this.discoveryMechanismList) {
       if (mechanismEntry.watcher) {
         const edsServiceName = mechanismEntry.discoveryMechanism.eds_service_name ?? mechanismEntry.discoveryMechanism.cluster;
-        this.xdsClient?.removeEndpointWatcher(edsServiceName, mechanismEntry.watcher);
+        if (this.xdsClient) {
+          EndpointResourceType.cancelWatch(this.xdsClient, edsServiceName, mechanismEntry.watcher);
+        }
       }
       mechanismEntry.resolver?.destroy();
     }