Skip to content

Commit

Permalink
Merge pull request #2433 from murgatroid99/grpc-js-xds_watcher_valida…
Browse files Browse the repository at this point in the history
…tion_errors

grpc-js-xds: Fix handling of resource validation errors
  • Loading branch information
murgatroid99 committed Apr 24, 2023
2 parents 9271984 + edeeda6 commit 4fcd9c2
Show file tree
Hide file tree
Showing 3 changed files with 185 additions and 21 deletions.
2 changes: 1 addition & 1 deletion packages/grpc-js-xds/package.json
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
{
"name": "@grpc/grpc-js-xds",
"version": "1.8.1",
"version": "1.8.2",
"description": "Plugin for @grpc/grpc-js. Adds the xds:// URL scheme and associated features.",
"main": "build/src/index.js",
"scripts": {
Expand Down
44 changes: 24 additions & 20 deletions packages/grpc-js-xds/src/xds-stream-state/xds-stream-state.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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';
Expand Down Expand Up @@ -157,43 +157,47 @@ export abstract class BaseXdsStreamState<ResponseType> implements XdsStreamState
return Array.from(this.subscriptions.keys());
}
handleResponses(responses: ResourcePair<ResponseType>[]): HandleResponseResult {
const validResponses: ResponseType[] = [];
let result: HandleResponseResult = {
accepted: [],
rejected: [],
missing: []
}
const allResourceNames = new Set<string>();
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) {
for (const watcher of subscriptionEntry.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({
name: resourceName,
raw: raw,
error: `Validation failed for resource ${resourceName}`
});
}
}
const allResourceNames = new Set<string>();
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) {
for (const watcher of subscriptionEntry.watchers) {
watcher.onTransientError({
code: status.UNAVAILABLE,
details: `Validation failed for resource ${resourceName}`,
metadata: new Metadata()
});
}
clearTimeout(subscriptionEntry.resourceTimer);
}
}
}
Expand Down
160 changes: 160 additions & 0 deletions packages/grpc-js-xds/test/test-nack.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,160 @@
/*
* 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
const invalidEdsResource = {cluster_name: cluster.getEndpointConfig().cluster_name, endpoints: [{}]};
xdsServer.setEdsResource(invalidEdsResource);
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
const invalidCdsResource = {name: cluster.getClusterConfig().name};
xdsServer.setCdsResource(invalidCdsResource);
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
const invalidRdsResource = {name: routeGroup.getRouteConfiguration().name, virtual_hosts: [{domains: ['**']}]};
xdsServer.setRdsResource(invalidRdsResource);
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
const invalidLdsResource = {name: routeGroup.getListener().name};
xdsServer.setLdsResource(invalidLdsResource);
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));
});
});

0 comments on commit 4fcd9c2

Please sign in to comment.