From 856559cce1d27771d50dc97d130f5dc423dd4723 Mon Sep 17 00:00:00 2001 From: Michael Lumish Date: Thu, 20 Apr 2023 14:34:06 -0700 Subject: [PATCH] grpc-js-xds: Fix handling of resource validation errors --- .../src/xds-stream-state/xds-stream-state.ts | 46 +++--- packages/grpc-js-xds/test/test-nack.ts | 156 ++++++++++++++++++ 2 files changed, 182 insertions(+), 20 deletions(-) create mode 100644 packages/grpc-js-xds/test/test-nack.ts diff --git a/packages/grpc-js-xds/src/xds-stream-state/xds-stream-state.ts b/packages/grpc-js-xds/src/xds-stream-state/xds-stream-state.ts index e20bc7e9..b2d7b227 100644 --- a/packages/grpc-js-xds/src/xds-stream-state/xds-stream-state.ts +++ b/packages/grpc-js-xds/src/xds-stream-state/xds-stream-state.ts @@ -15,7 +15,7 @@ * */ -import { experimental, logVerbosity, StatusObject } from "@grpc/grpc-js"; +import { experimental, logVerbosity, Metadata, status, StatusObject } from "@grpc/grpc-js"; import { Any__Output } from "../generated/google/protobuf/Any"; const TRACER_NAME = 'xds_client'; @@ -157,19 +157,32 @@ export abstract class BaseXdsStreamState implements XdsStreamState return Array.from(this.subscriptions.keys()); } handleResponses(responses: ResourcePair[]): HandleResponseResult { - const validResponses: ResponseType[] = []; let result: HandleResponseResult = { accepted: [], rejected: [], missing: [] } + const allResourceNames = new Set(); for (const {resource, raw} of responses) { const resourceName = this.getResourceName(resource); + allResourceNames.add(resourceName); + const subscriptionEntry = this.subscriptions.get(resourceName); if (this.validateResponse(resource)) { - validResponses.push(resource); result.accepted.push({ name: resourceName, raw: raw}); + if (subscriptionEntry) { + const watchers = subscriptionEntry.watchers; + for (const watcher of watchers) { + watcher.onValidUpdate(resource); + } + clearTimeout(subscriptionEntry.resourceTimer); + subscriptionEntry.cachedResponse = resource; + if (subscriptionEntry.deletionIgnored) { + experimental.log(logVerbosity.INFO, 'Received resource with previously ignored deletion: ' + resourceName); + subscriptionEntry.deletionIgnored = false; + } + } } else { this.trace('Validation failed for message ' + JSON.stringify(resource)); result.rejected.push({ @@ -177,23 +190,16 @@ export abstract class BaseXdsStreamState implements XdsStreamState raw: raw, error: `Validation failed for resource ${resourceName}` }); - } - } - const allResourceNames = new Set(); - for (const resource of validResponses) { - const resourceName = this.getResourceName(resource); - allResourceNames.add(resourceName); - const subscriptionEntry = this.subscriptions.get(resourceName); - if (subscriptionEntry) { - const watchers = subscriptionEntry.watchers; - for (const watcher of watchers) { - watcher.onValidUpdate(resource); - } - clearTimeout(subscriptionEntry.resourceTimer); - subscriptionEntry.cachedResponse = resource; - if (subscriptionEntry.deletionIgnored) { - experimental.log(logVerbosity.INFO, 'Received resource with previously ignored deletion: ' + resourceName); - subscriptionEntry.deletionIgnored = false; + if (subscriptionEntry) { + const watchers = subscriptionEntry.watchers; + for (const watcher of watchers) { + watcher.onTransientError({ + code: status.UNAVAILABLE, + details: `Validation failed for resource ${resourceName}`, + metadata: new Metadata() + }); + } + clearTimeout(subscriptionEntry.resourceTimer); } } } diff --git a/packages/grpc-js-xds/test/test-nack.ts b/packages/grpc-js-xds/test/test-nack.ts new file mode 100644 index 00000000..ad1aad44 --- /dev/null +++ b/packages/grpc-js-xds/test/test-nack.ts @@ -0,0 +1,156 @@ +/* + * Copyright 2023 gRPC authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +import * as assert from 'assert'; +import { register } from "../src"; +import { Backend } from "./backend"; +import { XdsTestClient } from "./client"; +import { FakeCluster, FakeRouteGroup } from "./framework"; +import { XdsServer } from "./xds-server"; + +register(); + +describe('Validation errors', () => { + let xdsServer: XdsServer; + let client: XdsTestClient; + beforeEach(done => { + xdsServer = new XdsServer(); + xdsServer.startServer(error => { + done(error); + }); + }); + afterEach(() => { + client?.close(); + xdsServer?.shutdownServer(); + }); + it('Should continue to use a valid resource after receiving an invalid EDS update', done => { + const cluster = new FakeCluster('cluster1', [{backends: [new Backend()], locality:{region: 'region1'}}]); + const routeGroup = new FakeRouteGroup('route1', [{cluster: cluster}]); + routeGroup.startAllBackends().then(() => { + xdsServer.setEdsResource(cluster.getEndpointConfig()); + xdsServer.setCdsResource(cluster.getClusterConfig()); + xdsServer.setRdsResource(routeGroup.getRouteConfiguration()); + xdsServer.setLdsResource(routeGroup.getListener()); + client = new XdsTestClient('route1', xdsServer); + client.startCalls(100); + routeGroup.waitForAllBackendsToReceiveTraffic().then(() => { + // After backends receive calls, set invalid EDS resource + xdsServer.setEdsResource({cluster_name: cluster.getEndpointConfig().cluster_name, endpoints: [{}]}); + let seenNack = false; + xdsServer.addResponseListener((typeUrl, responseState) => { + if (responseState.state === 'NACKED') { + if (seenNack) { + return; + } + seenNack = true; + routeGroup.waitForAllBackendsToReceiveTraffic().then(() => { + client.stopCalls(); + done(); + }); + } + }); + }, reason => done(reason)); + }, reason => done(reason)); + }); + it('Should continue to use a valid resource after receiving an invalid CDS update', done => { + const cluster = new FakeCluster('cluster1', [{backends: [new Backend()], locality:{region: 'region1'}}]); + const routeGroup = new FakeRouteGroup('route1', [{cluster: cluster}]); + routeGroup.startAllBackends().then(() => { + xdsServer.setEdsResource(cluster.getEndpointConfig()); + xdsServer.setCdsResource(cluster.getClusterConfig()); + xdsServer.setRdsResource(routeGroup.getRouteConfiguration()); + xdsServer.setLdsResource(routeGroup.getListener()); + client = new XdsTestClient('route1', xdsServer); + client.startCalls(100); + routeGroup.waitForAllBackendsToReceiveTraffic().then(() => { + // After backends receive calls, set invalid CDS resource + xdsServer.setCdsResource({name: cluster.getClusterConfig().name}); + let seenNack = false; + xdsServer.addResponseListener((typeUrl, responseState) => { + if (responseState.state === 'NACKED') { + if (seenNack) { + return; + } + seenNack = true; + routeGroup.waitForAllBackendsToReceiveTraffic().then(() => { + client.stopCalls(); + done(); + }); + } + }); + }, reason => done(reason)); + }, reason => done(reason)); + }); + it('Should continue to use a valid resource after receiving an invalid RDS update', done => { + const cluster = new FakeCluster('cluster1', [{backends: [new Backend()], locality:{region: 'region1'}}]); + const routeGroup = new FakeRouteGroup('route1', [{cluster: cluster}]); + routeGroup.startAllBackends().then(() => { + xdsServer.setEdsResource(cluster.getEndpointConfig()); + xdsServer.setCdsResource(cluster.getClusterConfig()); + xdsServer.setRdsResource(routeGroup.getRouteConfiguration()); + xdsServer.setLdsResource(routeGroup.getListener()); + client = new XdsTestClient('route1', xdsServer); + client.startCalls(100); + routeGroup.waitForAllBackendsToReceiveTraffic().then(() => { + // After backends receive calls, set invalid RDS resource + xdsServer.setRdsResource({name: routeGroup.getRouteConfiguration().name, virtual_hosts: [{domains: ['**']}]}); + let seenNack = false; + xdsServer.addResponseListener((typeUrl, responseState) => { + if (responseState.state === 'NACKED') { + if (seenNack) { + return; + } + seenNack = true; + routeGroup.waitForAllBackendsToReceiveTraffic().then(() => { + client.stopCalls(); + done(); + }); + } + }); + }, reason => done(reason)); + }, reason => done(reason)); + }); + it('Should continue to use a valid resource after receiving an invalid LDS update', done => { + const cluster = new FakeCluster('cluster1', [{backends: [new Backend()], locality:{region: 'region1'}}]); + const routeGroup = new FakeRouteGroup('route1', [{cluster: cluster}]); + routeGroup.startAllBackends().then(() => { + xdsServer.setEdsResource(cluster.getEndpointConfig()); + xdsServer.setCdsResource(cluster.getClusterConfig()); + xdsServer.setRdsResource(routeGroup.getRouteConfiguration()); + xdsServer.setLdsResource(routeGroup.getListener()); + client = new XdsTestClient('route1', xdsServer); + client.startCalls(100); + routeGroup.waitForAllBackendsToReceiveTraffic().then(() => { + // After backends receive calls, set invalid LDS resource + xdsServer.setLdsResource({name: routeGroup.getListener().name}); + let seenNack = false; + xdsServer.addResponseListener((typeUrl, responseState) => { + if (responseState.state === 'NACKED') { + if (seenNack) { + return; + } + seenNack = true; + routeGroup.waitForAllBackendsToReceiveTraffic().then(() => { + client.stopCalls(); + done(); + }); + } + }); + }, reason => done(reason)); + }, reason => done(reason)); + }); +}); \ No newline at end of file