grpc
/
grpc-node
mirror of https://github.com/grpc/grpc-node.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' into grpc-js-xds_federation

This commit is contained in:
Michael Lumish 2023-04-25 09:43:54 -07:00
parent 546696c366 eeaa6c0e6e
commit 2da3f868f2
7 changed files with 358 additions and 58 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
packages/grpc-js-xds/package.json
View File

@ -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": {

56
packages/grpc-js-xds/src/xds-stream-state/xds-stream-state.ts
View File

@ -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,35 @@ 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) {
/* Use process.nextTick to prevent errors from the watcher from
* bubbling up through here. */
process.nextTick(() => {
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,27 +193,19 @@ 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) {
/* Use process.nextTick to prevent errors from the watcher from
* bubbling up through here. */
process.nextTick(() => {
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) {
/* Use process.nextTick to prevent errors from the watcher from
* bubbling up through here. */
process.nextTick(() => {
watcher.onTransientError({
code: status.UNAVAILABLE,
details: `Validation failed for resource ${resourceName}`,
metadata: new Metadata()
});
});
}
clearTimeout(subscriptionEntry.resourceTimer);
}
}
}

161
packages/grpc-js-xds/test/test-nack.ts Normal file
View File

@ -0,0 +1,161 @@
/*
* 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 { Cluster } from '../src/generated/envoy/config/cluster/v3/Cluster';
import { Backend } from "./backend";
import { XdsTestClient } from "./client";
import { FakeEdsCluster, 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 FakeEdsCluster('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 FakeEdsCluster('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: Cluster = {name: cluster.getClusterConfig().name, type: 'EDS'};
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 FakeEdsCluster('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 FakeEdsCluster('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));
});
});

2
packages/grpc-js/package.json
View File

@ -1,6 +1,6 @@
{
"name": "@grpc/grpc-js",
"version": "1.8.13",
"version": "1.8.14",
"description": "gRPC Library for Node - pure JS implementation",
"homepage": "https://grpc.io/",
"repository": "https://github.com/grpc/grpc-node/tree/master/packages/grpc-js",

2
packages/grpc-js/src/load-balancer-pick-first.ts
View File

@ -322,12 +322,12 @@ export class PickFirstLoadBalancer implements LoadBalancer {
);
}
this.currentPick = subchannel;
this.updateState(ConnectivityState.READY, new PickFirstPicker(subchannel));
subchannel.addConnectivityStateListener(this.pickedSubchannelStateListener);
subchannel.ref();
this.channelControlHelper.addChannelzChild(subchannel.getChannelzRef());
this.resetSubchannelList();
clearTimeout(this.connectionDelayTimeout);
this.updateState(ConnectivityState.READY, new PickFirstPicker(subchannel));
}
private updateState(newState: ConnectivityState, picker: Picker) {

63
packages/grpc-js/src/subchannel.ts
View File

@ -60,7 +60,7 @@ export class Subchannel {
* state changes. Will be modified by `addConnectivityStateListener` and
* `removeConnectivityStateListener`
*/
private stateListeners: ConnectivityStateListener[] = [];
private stateListeners: Set<ConnectivityStateListener> = new Set();
private backoffTimeout: BackoffTimeout;
@ -261,9 +261,7 @@ export class Subchannel {
default:
throw new Error(`Invalid state: unknown ConnectivityState ${newState}`);
}
/* We use a shallow copy of the stateListeners array in case a listener
* is removed during this iteration */
for (const listener of [...this.stateListeners]) {
for (const listener of this.stateListeners) {
listener(this, previousState, newState, this.keepaliveTime);
}
return true;
@ -291,13 +289,15 @@ export class Subchannel {
if (this.channelzEnabled) {
this.channelzTrace.addTrace('CT_INFO', 'Shutting down');
}
this.transitionToState(
[ConnectivityState.CONNECTING, ConnectivityState.READY],
ConnectivityState.IDLE
);
if (this.channelzEnabled) {
unregisterChannelzRef(this.channelzRef);
}
process.nextTick(() => {
this.transitionToState(
[ConnectivityState.CONNECTING, ConnectivityState.READY],
ConnectivityState.IDLE
);
});
}
}
@ -339,20 +339,22 @@ export class Subchannel {
* Otherwise, do nothing.
*/
startConnecting() {
/* First, try to transition from IDLE to connecting. If that doesn't happen
* because the state is not currently IDLE, check if it is
* TRANSIENT_FAILURE, and if so indicate that it should go back to
* connecting after the backoff timer ends. Otherwise do nothing */
if (
!this.transitionToState(
[ConnectivityState.IDLE],
ConnectivityState.CONNECTING
)
) {
if (this.connectivityState === ConnectivityState.TRANSIENT_FAILURE) {
this.continueConnecting = true;
process.nextTick(() => {
/* First, try to transition from IDLE to connecting. If that doesn't happen
* because the state is not currently IDLE, check if it is
* TRANSIENT_FAILURE, and if so indicate that it should go back to
* connecting after the backoff timer ends. Otherwise do nothing */
if (
!this.transitionToState(
[ConnectivityState.IDLE],
ConnectivityState.CONNECTING
)
) {
if (this.connectivityState === ConnectivityState.TRANSIENT_FAILURE) {
this.continueConnecting = true;
}
}
}
});
}
/**
@ -368,7 +370,7 @@ export class Subchannel {
* @param listener
*/
addConnectivityStateListener(listener: ConnectivityStateListener) {
this.stateListeners.push(listener);
this.stateListeners.add(listener);
}
/**
@ -377,21 +379,20 @@ export class Subchannel {
* `addConnectivityStateListener`
*/
removeConnectivityStateListener(listener: ConnectivityStateListener) {
const listenerIndex = this.stateListeners.indexOf(listener);
if (listenerIndex > -1) {
this.stateListeners.splice(listenerIndex, 1);
}
this.stateListeners.delete(listener);
}
/**
* Reset the backoff timeout, and immediately start connecting if in backoff.
*/
resetBackoff() {
this.backoffTimeout.reset();
this.transitionToState(
[ConnectivityState.TRANSIENT_FAILURE],
ConnectivityState.CONNECTING
);
process.nextTick(() => {
this.backoffTimeout.reset();
this.transitionToState(
[ConnectivityState.TRANSIENT_FAILURE],
ConnectivityState.CONNECTING
);
});
}
getAddress(): string {

130
packages/grpc-js/test/test-global-subchannel-pool.ts Normal file
View File

@ -0,0 +1,130 @@
/*
* 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 * as path from 'path';
import * as grpc from '../src';
import {sendUnaryData, Server, ServerCredentials, ServerUnaryCall, ServiceClientConstructor, ServiceError} from '../src';
import {loadProtoFile} from './common';
const protoFile = path.join(__dirname, 'fixtures', 'echo_service.proto');
const echoService =
loadProtoFile(protoFile).EchoService as ServiceClientConstructor;
describe('Global subchannel pool', () => {
let server: Server;
let serverPort: number;
let client1: InstanceType<grpc.ServiceClientConstructor>;
let client2: InstanceType<grpc.ServiceClientConstructor>;
let promises: Promise<any>[];
before(done => {
server = new Server();
server.addService(echoService.service, {
echo(call: ServerUnaryCall<any, any>, callback: sendUnaryData<any>) {
callback(null, call.request);
},
});
server.bindAsync(
'localhost:0', ServerCredentials.createInsecure(), (err, port) => {
assert.ifError(err);
serverPort = port;
server.start();
done();
});
});
beforeEach(() => {
promises = [];
})
after(done => {
server.tryShutdown(done);
});
function callService(client: InstanceType<grpc.ServiceClientConstructor>) {
return new Promise<void>((resolve) => {
const request = {value: 'test value', value2: 3};
client.echo(request, (error: ServiceError, response: any) => {
assert.ifError(error);
assert.deepStrictEqual(response, request);
resolve();
});
})
}
function connect() {
const grpcOptions = {
'grpc.use_local_subchannel_pool': 0,
}
client1 = new echoService(
`127.0.0.1:${serverPort}`, grpc.credentials.createInsecure(),
grpcOptions);
client2 = new echoService(
`127.0.0.1:${serverPort}`, grpc.credentials.createInsecure(),
grpcOptions);
}
/* This is a regression test for a bug where client1.close in the
* waitForReady callback would cause the subchannel to transition to IDLE
* even though client2 is also using it. */
it('Should handle client.close calls in waitForReady',
done => {
connect();
promises.push(new Promise<void>((resolve) => {
client1.waitForReady(Date.now() + 50, (error) => {
assert.ifError(error);
client1.close();
resolve();
});
}))
promises.push(new Promise<void>((resolve) => {
client2.waitForReady(Date.now() + 50, (error) => {
assert.ifError(error);
resolve();
});
}))
Promise.all(promises).then(() => {done()});
})
it('Call the service', done => {
promises.push(callService(client2));
Promise.all(promises).then(() => {
done();
});
})
it('Should complete the client lifecycle without error', done => {
setTimeout(() => {
client1.close();
client2.close();
done()
}, 500);
});
});