grpc-js-xds: interop: add custom_lb test, reformat test list

packages/grpc-js-xds/interop/xds-interop-client.ts

@@ -92,7 +92,7 @@ class RpcBehaviorLoadBalancer implements LoadBalancer {
     const childChannelControlHelper = createChildChannelControlHelper(channelControlHelper, {
       updateState: (connectivityState, picker) => {
         if (connectivityState === grpc.connectivityState.READY && this.latestConfig) {
-          picker = new RpcBehaviorPicker(picker, this.latestConfig.getLoadBalancerName());
+          picker = new RpcBehaviorPicker(picker, this.latestConfig.getRpcBehavior());
         }
         channelControlHelper.updateState(connectivityState, picker);
       }

packages/grpc-js-xds/scripts/xds_k8s_lb.sh

@@ -165,7 +165,16 @@ main() {
   # Run tests
   cd "${TEST_DRIVER_FULL_DIR}"
   local failed_tests=0
-  test_suites=("baseline_test" "api_listener_test" "change_backend_service_test" "failover_test" "remove_neg_test" "round_robin_test" "outlier_detection_test")
+  test_suites=(
+    "api_listener_test"
+    "baseline_test"
+    "change_backend_service_test"
+    "custom_lb_test"
+    "failover_test"
+    "outlier_detection_test"
+    "remove_neg_test"
+    "round_robin_test"
+  )
   for test in "${test_suites[@]}"; do
     run_test $test || (( ++failed_tests ))
   done

packages/grpc-js-xds/test/backend.ts

@@ -48,6 +48,18 @@ export class Backend {
   Echo(call: ServerUnaryCall<EchoRequest__Output, EchoResponse>, callback: sendUnaryData<EchoResponse>) {
     // call.request.params is currently ignored
     this.addCall();
+    for (const behaviorEntry of call.metadata.get('rpc-behavior')) {
+      if (typeof behaviorEntry !== 'string') {
+        continue;
+      }
+      for (const behavior of behaviorEntry.split(',')) {
+        if (behavior.startsWith('error-code-')) {
+          const errorCode = Number(behavior.substring('error-code-'.length));
+          callback({code: errorCode, details: 'rpc-behavior error code'});
+          return;
+        }
+      }
+    }
     callback(null, {message: call.request.message});
   }
 
@@ -87,7 +99,7 @@ export class Backend {
       });
     });
   }
-  
+
   getPort(): number {
     if (this.port === null) {
       throw new Error('Port not set. Backend not yet started.');
@@ -125,4 +137,4 @@ export class Backend {
       });
     });
   }
-}
+}

packages/grpc-js-xds/test/test-custom-lb-policies.ts

@@ -24,6 +24,98 @@ import { XdsServer } from "./xds-server";
 import * as assert from 'assert';
 import { WrrLocality } from "../src/generated/envoy/extensions/load_balancing_policies/wrr_locality/v3/WrrLocality";
 import { TypedStruct } from "../src/generated/xds/type/v3/TypedStruct";
+import { connectivityState, experimental, logVerbosity } from "@grpc/grpc-js";
+
+import TypedLoadBalancingConfig = experimental.TypedLoadBalancingConfig;
+import LoadBalancer = experimental.LoadBalancer;
+import ChannelControlHelper = experimental.ChannelControlHelper;
+import ChildLoadBalancerHandler = experimental.ChildLoadBalancerHandler;
+import SubchannelAddress = experimental.SubchannelAddress;
+import Picker = experimental.Picker;
+import PickArgs = experimental.PickArgs;
+import PickResult = experimental.PickResult;
+import PickResultType = experimental.PickResultType;
+import createChildChannelControlHelper = experimental.createChildChannelControlHelper;
+import parseLoadBalancingConfig = experimental.parseLoadBalancingConfig;
+import registerLoadBalancerType = experimental.registerLoadBalancerType;
+
+const LB_POLICY_NAME = 'test.RpcBehaviorLoadBalancer';
+
+class RpcBehaviorLoadBalancingConfig implements TypedLoadBalancingConfig {
+  constructor(private rpcBehavior: string) {}
+  getLoadBalancerName(): string {
+    return LB_POLICY_NAME;
+  }
+  toJsonObject(): object {
+    return {
+      [LB_POLICY_NAME]: {
+        'rpcBehavior': this.rpcBehavior
+      }
+    };
+  }
+  getRpcBehavior() {
+    return this.rpcBehavior;
+  }
+  static createFromJson(obj: any): RpcBehaviorLoadBalancingConfig {
+    if (!('rpcBehavior' in obj && typeof obj.rpcBehavior === 'string')) {
+      throw new Error(`${LB_POLICY_NAME} parsing error: expected string field rpcBehavior`);
+    }
+    return new RpcBehaviorLoadBalancingConfig(obj.rpcBehavior);
+  }
+}
+
+class RpcBehaviorPicker implements Picker {
+  constructor(private wrappedPicker: Picker, private rpcBehavior: string) {}
+  pick(pickArgs: PickArgs): PickResult {
+    const wrappedPick = this.wrappedPicker.pick(pickArgs);
+    if (wrappedPick.pickResultType === PickResultType.COMPLETE) {
+      pickArgs.metadata.add('rpc-behavior', this.rpcBehavior);
+    }
+    return wrappedPick;
+  }
+}
+
+const RPC_BEHAVIOR_CHILD_CONFIG = parseLoadBalancingConfig({round_robin: {}});
+
+/**
+ * Load balancer implementation for Custom LB policy test
+ */
+class RpcBehaviorLoadBalancer implements LoadBalancer {
+  private child: ChildLoadBalancerHandler;
+  private latestConfig: RpcBehaviorLoadBalancingConfig | null = null;
+  constructor(channelControlHelper: ChannelControlHelper) {
+    const childChannelControlHelper = createChildChannelControlHelper(channelControlHelper, {
+      updateState: (state, picker) => {
+        if (state === connectivityState.READY && this.latestConfig) {
+          picker = new RpcBehaviorPicker(picker, this.latestConfig.getRpcBehavior());
+        }
+        channelControlHelper.updateState(state, picker);
+      }
+    });
+    this.child = new ChildLoadBalancerHandler(childChannelControlHelper);
+  }
+  updateAddressList(addressList: SubchannelAddress[], lbConfig: TypedLoadBalancingConfig, attributes: { [key: string]: unknown; }): void {
+    if (!(lbConfig instanceof RpcBehaviorLoadBalancingConfig)) {
+      return;
+    }
+    this.latestConfig = lbConfig;
+    this.child.updateAddressList(addressList, RPC_BEHAVIOR_CHILD_CONFIG, attributes);
+  }
+  exitIdle(): void {
+    this.child.exitIdle();
+  }
+  resetBackoff(): void {
+    this.child.resetBackoff();
+  }
+  destroy(): void {
+    this.child.destroy();
+  }
+  getTypeName(): string {
+    return LB_POLICY_NAME;
+  }
+}
+
+registerLoadBalancerType(LB_POLICY_NAME, RpcBehaviorLoadBalancer, RpcBehaviorLoadBalancingConfig);
 
 describe('Custom LB policies', () => {
   let xdsServer: XdsServer;
@@ -163,4 +255,47 @@ describe('Custom LB policies', () => {
       client.sendOneCall(done);
     }, reason => done(reason));
   });
+  it('Should handle a custom LB policy', done => {
+    const childPolicy: TypedStruct & AnyExtension = {
+      '@type': 'type.googleapis.com/xds.type.v3.TypedStruct',
+      type_url: 'test.RpcBehaviorLoadBalancer',
+      value: {
+        fields: {
+          rpcBehavior: {stringValue: 'error-code-15'}
+        }
+      }
+    };
+    const lbPolicy: WrrLocality & AnyExtension = {
+      '@type': 'type.googleapis.com/envoy.extensions.load_balancing_policies.wrr_locality.v3.WrrLocality',
+      endpoint_picking_policy: {
+        policies: [
+          {
+            typed_extension_config: {
+              name: 'child',
+              typed_config: childPolicy
+            }
+          }
+        ]
+      }
+    };
+    const cluster = new FakeEdsCluster('cluster1', 'endpoint1', [{backends: [new Backend()], locality:{region: 'region1'}}], lbPolicy);
+    const routeGroup = new FakeRouteGroup('listener1', 'route1', [{cluster: cluster}]);
+    routeGroup.startAllBackends().then(() => {
+      xdsServer.setEdsResource(cluster.getEndpointConfig());
+      xdsServer.setCdsResource(cluster.getClusterConfig());
+      xdsServer.setRdsResource(routeGroup.getRouteConfiguration());
+      xdsServer.setLdsResource(routeGroup.getListener());
+      xdsServer.addResponseListener((typeUrl, responseState) => {
+        if (responseState.state === 'NACKED') {
+          client.stopCalls();
+          assert.fail(`Client NACKED ${typeUrl} resource with message ${responseState.errorMessage}`);
+        }
+      })
+      client = XdsTestClient.createFromServer('listener1', xdsServer);
+      client.sendOneCall(error => {
+        assert.strictEqual(error?.code, 15);
+        done();
+      });
+    }, reason => done(reason));
+  })
 });