mirror of https://github.com/grpc/grpc-node.git
grpc-js-xds: Fix handling of resource validation errors
This commit is contained in:
Michael Lumish
parent
9dec365082
commit
856559cce1
|
|
@ -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<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) {
|
||||
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<ResponseType> implements XdsStreamState
|
|||
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) {
|
||||
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);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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));
|
||||
});
|
||||
});
|
||||
Loading…
Reference in New Issue