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

grpc-js: Use an object to represent subchannel addresses

This commit is contained in:
murgatroid99 2020-01-29 09:42:13 -08:00
parent 6994f1b1df
commit 57c18382d8
11 changed files with 136 additions and 98 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
packages/grpc-js/src/channel.ts
View File

@ -38,6 +38,7 @@ import { getDefaultAuthority } from './resolver';
import { LoadBalancingConfig } from './load-balancing-config';
import { ServiceConfig, validateServiceConfig } from './service-config';
import { trace } from './logging';
import { SubchannelAddress } from './subchannel';
export enum ConnectivityState {
CONNECTING,
@ -145,7 +146,7 @@ export class ChannelImplementation implements Channel {
this.subchannelPool = getSubchannelPool((options['grpc.use_local_subchannel_pool'] ?? 0) === 0);
const channelControlHelper: ChannelControlHelper = {
createSubchannel: (
subchannelAddress: string,
subchannelAddress: SubchannelAddress,
subchannelArgs: ChannelOptions
) => {
return this.subchannelPool.getOrCreateSubchannel(

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

@ -30,7 +30,7 @@ import {
UnavailablePicker,
} from './picker';
import { LoadBalancingConfig } from './load-balancing-config';
import { Subchannel, ConnectivityStateListener } from './subchannel';
import { Subchannel, ConnectivityStateListener, SubchannelAddress } from './subchannel';
import * as logging from './logging';
import { LogVerbosity } from './constants';
@ -76,7 +76,7 @@ export class PickFirstLoadBalancer implements LoadBalancer {
/**
* The list of backend addresses most recently passed to `updateAddressList`.
*/
private latestAddressList: string[] = [];
private latestAddressList: SubchannelAddress[] = [];
/**
* The list of subchannels this load balancer is currently attempting to
* connect to.
@ -369,7 +369,7 @@ export class PickFirstLoadBalancer implements LoadBalancer {
}
updateAddressList(
addressList: string[],
addressList: SubchannelAddress[],
lbConfig: LoadBalancingConfig | null
): void {
// lbConfig has no useful information for pick first load balancing

4
packages/grpc-js/src/load-balancer-round-robin.ts
View File

@ -30,7 +30,7 @@ import {
UnavailablePicker,
} from './picker';
import { LoadBalancingConfig } from './load-balancing-config';
import { Subchannel, ConnectivityStateListener } from './subchannel';
import { Subchannel, ConnectivityStateListener, SubchannelAddress } from './subchannel';
const TYPE_NAME = 'round_robin';
@ -168,7 +168,7 @@ export class RoundRobinLoadBalancer implements LoadBalancer {
}
updateAddressList(
addressList: string[],
addressList: SubchannelAddress[],
lbConfig: LoadBalancingConfig | null
): void {
this.resetSubchannelList();

6
packages/grpc-js/src/load-balancer.ts
View File

@ -16,7 +16,7 @@
*/
import { ChannelOptions } from './channel-options';
import { Subchannel } from './subchannel';
import { Subchannel, SubchannelAddress } from './subchannel';
import { ConnectivityState } from './channel';
import { Picker } from './picker';
import { LoadBalancingConfig } from './load-balancing-config';
@ -34,7 +34,7 @@ export interface ChannelControlHelper {
* @param subchannelArgs Extra channel arguments specified by the load balancer
*/
createSubchannel(
subchannelAddress: string,
subchannelAddress: SubchannelAddress,
subchannelArgs: ChannelOptions
): Subchannel;
/**
@ -66,7 +66,7 @@ export interface LoadBalancer {
* if one was provided
*/
updateAddressList(
addressList: string[],
addressList: SubchannelAddress[],
lbConfig: LoadBalancingConfig | null
): void;
/**

27
packages/grpc-js/src/resolver-dns.ts
View File

@ -30,6 +30,7 @@ import { StatusObject } from './call-stream';
import { Metadata } from './metadata';
import * as logging from './logging';
import { LogVerbosity } from './constants';
import { SubchannelAddress } from './subchannel';
const TRACER_NAME = 'dns_resolver';
@ -112,7 +113,7 @@ const dnsLookupPromise = util.promisify(dns.lookup);
* @param target
* @return An "IP:port" string in an array if parsing was successful, `null` otherwise
*/
function parseIP(target: string): string[] | null {
function parseIP(target: string): SubchannelAddress[] | null {
/* These three regular expressions are all mutually exclusive, so we just
* want the first one that matches the target string, if any do. */
const ipv4Match = IPV4_REGEX.exec(target);
@ -123,14 +124,14 @@ function parseIP(target: string): string[] | null {
}
// ipv6 addresses should be bracketed
const addr = ipv4Match ? match[1] : `[${match[1]}]`;
const addr = match[1];
let port: string;
if (match[2]) {
port = match[2];
} else {
port = DEFAULT_PORT;
}
return [`${addr}:${port}`];
return [{host: addr, port: +port}];
}
/**
@ -161,7 +162,7 @@ function mergeArrays<T>(...arrays: T[][]): T[] {
* Resolver implementation that handles DNS names and IP addresses.
*/
class DnsResolver implements Resolver {
private readonly ipResult: string[] | null;
private readonly ipResult: SubchannelAddress[] | null;
private readonly dnsHostname: string | null;
private readonly port: string | null;
/* The promise results here contain, in order, the A record, the AAAA record,
@ -222,23 +223,21 @@ class DnsResolver implements Resolver {
this.pendingResultPromise.then(
([addressList, txtRecord]) => {
this.pendingResultPromise = null;
const ip4Addresses: string[] = addressList
.filter(addr => addr.family === 4)
.map(addr => `${addr.address}:${this.port}`);
let ip6Addresses: string[];
const ip4Addresses: dns.LookupAddress[] = addressList
.filter(addr => addr.family === 4);
let ip6Addresses: dns.LookupAddress[];
if (semver.satisfies(process.version, IPV6_SUPPORT_RANGE)) {
ip6Addresses = addressList
.filter(addr => addr.family === 6)
.map(addr => `[${addr.address}]:${this.port}`);
ip6Addresses = addressList.filter(addr => addr.family === 6);
} else {
ip6Addresses = [];
}
const allAddresses: string[] = mergeArrays(
const allAddresses: SubchannelAddress[] = mergeArrays(
ip4Addresses,
ip6Addresses
);
).map(addr => {return {host: addr.address, port: +this.port!};});
const allAddressesString: string = '[' + allAddresses.map(addr => addr.host + ':' + addr.port).join(',') + ']';
trace(
'Resolved addresses for target ' + this.target + ': ' + allAddresses
'Resolved addresses for target ' + this.target + ': ' + allAddressesString
);
if (allAddresses.length === 0) {
this.listener.onError(this.defaultResolutionError);

5
packages/grpc-js/src/resolver-uds.ts
View File

@ -20,6 +20,7 @@ import {
registerResolver,
registerDefaultResolver,
} from './resolver';
import { SubchannelAddress } from './subchannel';
function getUdsName(target: string): string {
/* Due to how this resolver is registered, it should only be constructed
@ -34,9 +35,9 @@ function getUdsName(target: string): string {
}
class UdsResolver implements Resolver {
private addresses: string[] = [];
private addresses: SubchannelAddress[] = [];
constructor(target: string, private listener: ResolverListener) {
this.addresses = [getUdsName(target)];
this.addresses = [{path: getUdsName(target)}];
}
updateResolution(): void {
process.nextTick(

3
packages/grpc-js/src/resolver.ts
View File

@ -20,6 +20,7 @@ import { ServiceConfig } from './service-config';
import * as resolver_dns from './resolver-dns';
import * as resolver_uds from './resolver-uds';
import { StatusObject } from './call-stream';
import { SubchannelAddress } from './subchannel';
/**
* A listener object passed to the resolver's constructor that provides name
@ -36,7 +37,7 @@ export interface ResolverListener {
* service configuration was invalid
*/
onSuccessfulResolution(
addressList: string[],
addressList: SubchannelAddress[],
serviceConfig: ServiceConfig | null,
serviceConfigError: StatusObject | null
): void;

9
packages/grpc-js/src/resolving-load-balancer.ts
View File

@ -34,6 +34,7 @@ import { StatusObject } from './call-stream';
import { Metadata } from './metadata';
import * as logging from './logging';
import { LogVerbosity } from './constants';
import { SubchannelAddress } from './subchannel';
const TRACER_NAME = 'resolving_load_balancer';
@ -132,7 +133,7 @@ export class ResolvingLoadBalancer implements LoadBalancer {
this.updateState(ConnectivityState.IDLE, new QueuePicker(this));
this.innerResolver = createResolver(target, {
onSuccessfulResolution: (
addressList: string[],
addressList: SubchannelAddress[],
serviceConfig: ServiceConfig | null,
serviceConfigError: ServiceError | null
) => {
@ -243,7 +244,7 @@ export class ResolvingLoadBalancer implements LoadBalancer {
this.innerChannelControlHelper = {
createSubchannel: (
subchannelAddress: string,
subchannelAddress: SubchannelAddress,
subchannelArgs: ChannelOptions
) => {
return this.channelControlHelper.createSubchannel(
@ -289,7 +290,7 @@ export class ResolvingLoadBalancer implements LoadBalancer {
this.replacementChannelControlHelper = {
createSubchannel: (
subchannelAddress: string,
subchannelAddress: SubchannelAddress,
subchannelArgs: ChannelOptions
) => {
return this.channelControlHelper.createSubchannel(
@ -409,7 +410,7 @@ export class ResolvingLoadBalancer implements LoadBalancer {
}
updateAddressList(
addressList: string[],
addressList: SubchannelAddress[],
lbConfig: LoadBalancingConfig | null
) {
throw new Error('updateAddressList not supported on ResolvingLoadBalancer');

73
packages/grpc-js/src/subchannel-pool.ts
View File

@ -16,7 +16,7 @@
*/
import { ChannelOptions, channelOptionsEqual } from './channel-options';
import { Subchannel } from './subchannel';
import { Subchannel, SubchannelAddress, subchannelAddressEqual } from './subchannel';
import { ChannelCredentials } from './channel-credentials';
// 10 seconds in milliseconds. This value is arbitrary.
@ -28,13 +28,12 @@ const REF_CHECK_INTERVAL = 10_000;
export class SubchannelPool {
private pool: {
[channelTarget: string]: {
[subchannelTarget: string]: Array<{
channelArguments: ChannelOptions;
channelCredentials: ChannelCredentials;
subchannel: Subchannel;
}>;
};
[channelTarget: string]: Array<{
subchannelAddress: SubchannelAddress;
channelArguments: ChannelOptions;
channelCredentials: ChannelCredentials;
subchannel: Subchannel;
}>;
} = Object.create(null);
/**
@ -62,23 +61,20 @@ export class SubchannelPool {
* do not need to be filtered */
// tslint:disable-next-line:forin
for (const channelTarget in this.pool) {
// tslint:disable-next-line:forin
for (const subchannelTarget in this.pool[channelTarget]) {
const subchannelObjArray = this.pool[channelTarget][subchannelTarget];
const subchannelObjArray = this.pool[channelTarget];
const refedSubchannels = subchannelObjArray.filter(
value => !value.subchannel.unrefIfOneRef()
);
const refedSubchannels = subchannelObjArray.filter(
value => !value.subchannel.unrefIfOneRef()
);
if (refedSubchannels.length > 0) {
allSubchannelsUnrefed = false;
}
/* For each subchannel in the pool, try to unref it if it has
* exactly one ref (which is the ref from the pool itself). If that
* does happen, remove the subchannel from the pool */
this.pool[channelTarget][subchannelTarget] = refedSubchannels;
if (refedSubchannels.length > 0) {
allSubchannelsUnrefed = false;
}
/* For each subchannel in the pool, try to unref it if it has
* exactly one ref (which is the ref from the pool itself). If that
* does happen, remove the subchannel from the pool */
this.pool[channelTarget] = refedSubchannels;
}
/* Currently we do not delete keys with empty values. If that results
* in significant memory usage we should change it. */
@ -114,25 +110,24 @@ export class SubchannelPool {
*/
getOrCreateSubchannel(
channelTarget: string,
subchannelTarget: string,
subchannelTarget: SubchannelAddress,
channelArguments: ChannelOptions,
channelCredentials: ChannelCredentials
): Subchannel {
this.ensureCleanupTask();
if (channelTarget in this.pool) {
if (subchannelTarget in this.pool[channelTarget]) {
const subchannelObjArray = this.pool[channelTarget][subchannelTarget];
for (const subchannelObj of subchannelObjArray) {
if (
channelOptionsEqual(
channelArguments,
subchannelObj.channelArguments
) &&
channelCredentials._equals(subchannelObj.channelCredentials)
) {
return subchannelObj.subchannel;
}
const subchannelObjArray = this.pool[channelTarget];
for (const subchannelObj of subchannelObjArray) {
if (
subchannelAddressEqual(subchannelTarget, subchannelObj.subchannelAddress) &&
channelOptionsEqual(
channelArguments,
subchannelObj.channelArguments
) &&
channelCredentials._equals(subchannelObj.channelCredentials)
) {
return subchannelObj.subchannel;
}
}
}
@ -144,12 +139,10 @@ export class SubchannelPool {
channelCredentials
);
if (!(channelTarget in this.pool)) {
this.pool[channelTarget] = Object.create(null);
this.pool[channelTarget] = [];
}
if (!(subchannelTarget in this.pool[channelTarget])) {
this.pool[channelTarget][subchannelTarget] = [];
}
this.pool[channelTarget][subchannelTarget].push({
this.pool[channelTarget].push({
subchannelAddress: subchannelTarget,
channelArguments,
channelCredentials,
subchannel,

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

@ -20,12 +20,13 @@ import { ChannelCredentials } from './channel-credentials';
import { Metadata } from './metadata';
import { Http2CallStream } from './call-stream';
import { ChannelOptions } from './channel-options';
import { PeerCertificate, checkServerIdentity } from 'tls';
import { PeerCertificate, checkServerIdentity, TLSSocket } from 'tls';
import { ConnectivityState } from './channel';
import { BackoffTimeout, BackoffOptions } from './backoff-timeout';
import { getDefaultAuthority } from './resolver';
import * as logging from './logging';
import { LogVerbosity } from './constants';
import { SocketConnectOpts } from 'net';
const { version: clientVersion } = require('../../package.json');
@ -73,6 +74,22 @@ function uniformRandom(min: number, max: number) {
const tooManyPingsData: Buffer = Buffer.from('too_many_pings', 'ascii');
/**
* This represents a single backend address to connect to. This interface is a
* subset of net.SocketConnectOpts, i.e. the options described at
* https://nodejs.org/api/net.html#net_socket_connect_options_connectlistener.
* Those are in turn a subset of the options that can be passed to http2.connect.
*/
export interface SubchannelAddress {
port?: number;
host?: string;
path?: string;
}
export function subchannelAddressEqual(address1: SubchannelAddress, address2: SubchannelAddress) : boolean {
return address1.port === address2.port && address1.host === address2.host && address1.path === address2.path;
}
export class Subchannel {
/**
* The subchannel's current connectivity state. Invariant: `session` === `null`
@ -135,6 +152,11 @@ export class Subchannel {
*/
private refcount = 0;
/**
* A string representation of the subchannel address, for logging/tracing
*/
private subchannelAddressString: string;
/**
* A class representing a connection to a single backend.
* @param channelTarget The target string for the channel as a whole
@ -147,7 +169,7 @@ export class Subchannel {
*/
constructor(
private channelTarget: string,
private subchannelAddress: string,
private subchannelAddress: SubchannelAddress,
private options: ChannelOptions,
private credentials: ChannelCredentials
) {
@ -187,6 +209,11 @@ export class Subchannel {
);
}
}, backoffOptions);
if (subchannelAddress.host || subchannelAddress.port) {
this.subchannelAddressString = `${subchannelAddress.host}:${subchannelAddress.port}`;
} else {
this.subchannelAddressString = `${subchannelAddress.path}`;
}
}
/**
@ -225,10 +252,11 @@ export class Subchannel {
}
private startConnectingInternal() {
const connectionOptions: http2.SecureClientSessionOptions =
let connectionOptions: http2.SecureClientSessionOptions =
this.credentials._getConnectionOptions() || {};
let addressScheme = 'http://';
if ('secureContext' in connectionOptions) {
connectionOptions.protocol = 'https:';
addressScheme = 'https://';
// If provided, the value of grpc.ssl_target_name_override should be used
// to override the target hostname when checking server identity.
@ -248,8 +276,21 @@ export class Subchannel {
connectionOptions.servername = getDefaultAuthority(this.channelTarget);
}
}
connectionOptions = Object.assign(connectionOptions, this.subchannelAddress);
/* http2.connect uses the options here:
* https://github.com/nodejs/node/blob/70c32a6d190e2b5d7b9ff9d5b6a459d14e8b7d59/lib/internal/http2/core.js#L3028-L3036
* The spread operator overides earlier values with later ones, so any port
* or host values in the options will be used rather than any values extracted
* from the first argument. In addition, the path overrides the host and port,
* as documented for plaintext connections here:
* https://nodejs.org/api/net.html#net_socket_connect_options_connectlistener
* and for TLS connections here:
* https://nodejs.org/api/tls.html#tls_tls_connect_options_callback.
* The first argument just needs to be parseable as a URL and the scheme
* determines whether the connection will be established over TLS or not.
*/
const session = http2.connect(
addressScheme + this.subchannelAddress,
addressScheme + getDefaultAuthority(this.channelTarget),
connectionOptions
);
this.session = session;
@ -328,7 +369,7 @@ export class Subchannel {
return false;
}
trace(
this.subchannelAddress +
this.subchannelAddressString +
' ' +
ConnectivityState[this.connectivityState] +
' -> ' +
@ -400,7 +441,7 @@ export class Subchannel {
callRef() {
trace(
this.subchannelAddress +
this.subchannelAddressString +
' callRefcount ' +
this.callRefcount +
' -> ' +
@ -417,7 +458,7 @@ export class Subchannel {
callUnref() {
trace(
this.subchannelAddress +
this.subchannelAddressString +
' callRefcount ' +
this.callRefcount +
' -> ' +
@ -435,7 +476,7 @@ export class Subchannel {
ref() {
trace(
this.subchannelAddress +
this.subchannelAddressString +
' callRefcount ' +
this.refcount +
' -> ' +
@ -446,7 +487,7 @@ export class Subchannel {
unref() {
trace(
this.subchannelAddress +
this.subchannelAddressString +
' callRefcount ' +
this.refcount +
' -> ' +
@ -557,6 +598,6 @@ export class Subchannel {
}
getAddress(): string {
return this.subchannelAddress;
return this.subchannelAddressString;
}
}

37
packages/grpc-js/test/test-resolver.ts
View File

@ -21,6 +21,7 @@ import * as assert from 'assert';
import * as resolverManager from '../src/resolver';
import { ServiceConfig } from '../src/service-config';
import { StatusObject } from '../src/call-stream';
import { SubchannelAddress } from '../src/subchannel';
describe('Name Resolver', () => {
describe('DNS Names', function() {
@ -33,11 +34,11 @@ describe('Name Resolver', () => {
const target = 'localhost:50051';
const listener: resolverManager.ResolverListener = {
onSuccessfulResolution: (
addressList: string[],
addressList: SubchannelAddress[],
serviceConfig: ServiceConfig | null,
serviceConfigError: StatusObject | null
) => {
assert(addressList.includes('127.0.0.1:50051'));
assert(addressList.some(addr => addr.host === '127.0.0.1' && addr.port === 50051));
// We would check for the IPv6 address but it needs to be omitted on some Node versions
done();
},
@ -52,11 +53,11 @@ describe('Name Resolver', () => {
const target = 'localhost';
const listener: resolverManager.ResolverListener = {
onSuccessfulResolution: (
addressList: string[],
addressList: SubchannelAddress[],
serviceConfig: ServiceConfig | null,
serviceConfigError: StatusObject | null
) => {
assert(addressList.includes('127.0.0.1:443'));
assert(addressList.some(addr => addr.host === '127.0.0.1' && addr.port === 443));
// We would check for the IPv6 address but it needs to be omitted on some Node versions
done();
},
@ -71,11 +72,11 @@ describe('Name Resolver', () => {
const target = '1.2.3.4';
const listener: resolverManager.ResolverListener = {
onSuccessfulResolution: (
addressList: string[],
addressList: SubchannelAddress[],
serviceConfig: ServiceConfig | null,
serviceConfigError: StatusObject | null
) => {
assert(addressList.includes('1.2.3.4:443'));
assert(addressList.some(addr => addr.host === '1.2.3.4' && addr.port === 443));
// We would check for the IPv6 address but it needs to be omitted on some Node versions
done();
},
@ -90,11 +91,11 @@ describe('Name Resolver', () => {
const target = '::1';
const listener: resolverManager.ResolverListener = {
onSuccessfulResolution: (
addressList: string[],
addressList: SubchannelAddress[],
serviceConfig: ServiceConfig | null,
serviceConfigError: StatusObject | null
) => {
assert(addressList.includes('[::1]:443'));
assert(addressList.some(addr => addr.host === '::1' && addr.port === 443));
// We would check for the IPv6 address but it needs to be omitted on some Node versions
done();
},
@ -109,11 +110,11 @@ describe('Name Resolver', () => {
const target = '[::1]:50051';
const listener: resolverManager.ResolverListener = {
onSuccessfulResolution: (
addressList: string[],
addressList: SubchannelAddress[],
serviceConfig: ServiceConfig | null,
serviceConfigError: StatusObject | null
) => {
assert(addressList.includes('[::1]:50051'));
assert(addressList.some(addr => addr.host === '::1' && addr.port === 50051));
// We would check for the IPv6 address but it needs to be omitted on some Node versions
done();
},
@ -128,7 +129,7 @@ describe('Name Resolver', () => {
const target = 'example.com';
const listener: resolverManager.ResolverListener = {
onSuccessfulResolution: (
addressList: string[],
addressList: SubchannelAddress[],
serviceConfig: ServiceConfig | null,
serviceConfigError: StatusObject | null
) => {
@ -146,7 +147,7 @@ describe('Name Resolver', () => {
const target = 'loopback4.unittest.grpc.io';
const listener: resolverManager.ResolverListener = {
onSuccessfulResolution: (
addressList: string[],
addressList: SubchannelAddress[],
serviceConfig: ServiceConfig | null,
serviceConfigError: StatusObject | null
) => {
@ -166,7 +167,7 @@ describe('Name Resolver', () => {
const target = 'network-tools.com';
const listener: resolverManager.ResolverListener = {
onSuccessfulResolution: (
addressList: string[],
addressList: SubchannelAddress[],
serviceConfig: ServiceConfig | null,
serviceConfigError: StatusObject | null
) => {
@ -196,7 +197,7 @@ describe('Name Resolver', () => {
const target2 = 'grpc-test4.sandbox.googleapis.com';
const listener: resolverManager.ResolverListener = {
onSuccessfulResolution: (
addressList: string[],
addressList: SubchannelAddress[],
serviceConfig: ServiceConfig | null,
serviceConfigError: StatusObject | null
) => {
@ -218,11 +219,11 @@ describe('Name Resolver', () => {
const target = 'unix:socket';
const listener: resolverManager.ResolverListener = {
onSuccessfulResolution: (
addressList: string[],
addressList: SubchannelAddress[],
serviceConfig: ServiceConfig | null,
serviceConfigError: StatusObject | null
) => {
assert(addressList.includes('socket'));
assert(addressList.some(addr => addr.path = 'socket'));
done();
},
onError: (error: StatusObject) => {
@ -236,11 +237,11 @@ describe('Name Resolver', () => {
const target = 'unix:///tmp/socket';
const listener: resolverManager.ResolverListener = {
onSuccessfulResolution: (
addressList: string[],
addressList: SubchannelAddress[],
serviceConfig: ServiceConfig | null,
serviceConfigError: StatusObject | null
) => {
assert(addressList.includes('/tmp/socket'));
assert(addressList.some(addr => addr.path = '/tmp/socket'));
done();
},
onError: (error: StatusObject) => {