Merge branch 'grpc:master' into master

Author: natiz

packages/grpc-js-xds/interop/xds-interop-client.ts

@@ -180,6 +180,27 @@ class CallStatsTracker {
   }
 }
 
+class RecentTimestampList {
+  private timeList: bigint[] = [];
+  private nextIndex = 0;
+
+  constructor(private readonly size: number) {}
+
+  isFull() {
+    return this.timeList.length === this.size;
+  }
+
+  insertTimestamp(timestamp: bigint) {
+    this.timeList[this.nextIndex] = timestamp;
+    this.nextIndex = (this.nextIndex + 1) % this.size;
+  }
+
+  getSpan(): bigint {
+    const lastIndex = (this.nextIndex + this.size - 1) % this.size;
+    return this.timeList[lastIndex] - this.timeList[this.nextIndex];
+  }
+}
+
 type CallType = 'EmptyCall' | 'UnaryCall';
 
 interface ClientConfiguration {
@@ -246,27 +267,34 @@ const callTimeHistogram: {[callType: string]: {[status: number]: number[]}} = {
   EmptyCall: {}
 }
 
-function makeSingleRequest(client: TestServiceClient, type: CallType, failOnFailedRpcs: boolean, callStatsTracker: CallStatsTracker) {
+/**
+ * Timestamps output by process.hrtime.bigint() are a bigint number of
+ * nanoseconds. This is the representation of 1 second in that context.
+ */
+const TIMESTAMP_ONE_SECOND = BigInt(1e9);
+
+function makeSingleRequest(client: TestServiceClient, type: CallType, failOnFailedRpcs: boolean, callStatsTracker: CallStatsTracker, callStartTimestamps: RecentTimestampList) {
   const callEnumName = callTypeEnumMapReverse[type];
   addAccumulatedCallStarted(callEnumName);
   const notifier = callStatsTracker.startCall();
   let gotMetadata: boolean = false;
   let hostname: string | null = null;
   let completed: boolean = false;
   let completedWithError: boolean = false;
-  const startTime = process.hrtime();
+  const startTime = process.hrtime.bigint();
   const deadline = new Date();
   deadline.setSeconds(deadline.getSeconds() + currentConfig.timeoutSec);
   const callback = (error: grpc.ServiceError | undefined, value: Empty__Output | undefined) => {
     const statusCode = error?.code ?? grpc.status.OK;
-    const duration = process.hrtime(startTime);
+    const duration = process.hrtime.bigint() - startTime;
+    const durationSeconds = Number(duration / TIMESTAMP_ONE_SECOND) | 0;
     if (!callTimeHistogram[type][statusCode]) {
       callTimeHistogram[type][statusCode] = [];
     }
-    if (callTimeHistogram[type][statusCode][duration[0]]) {
-      callTimeHistogram[type][statusCode][duration[0]] += 1;
+    if (callTimeHistogram[type][statusCode][durationSeconds]) {
+      callTimeHistogram[type][statusCode][durationSeconds] += 1;
     } else {
-      callTimeHistogram[type][statusCode][duration[0]] = 1;
+      callTimeHistogram[type][statusCode][durationSeconds] = 1;
     }
     addAccumulatedCallEnded(callEnumName, statusCode);
     if (error) {
@@ -301,13 +329,28 @@ function makeSingleRequest(client: TestServiceClient, type: CallType, failOnFail
       }
     }
   });
-
+  /* callStartTimestamps tracks the last N timestamps of started calls, where N
+   * is the target QPS. If the measured span of time between the first and last
+   * of those N calls is greater than 1 second, we make another call
+   * ~immediately to correct for that. */
+  callStartTimestamps.insertTimestamp(startTime);
+  if (callStartTimestamps.isFull()) {
+    if (callStartTimestamps.getSpan() > TIMESTAMP_ONE_SECOND) {
+      setImmediate(() => {
+        makeSingleRequest(client, type, failOnFailedRpcs, callStatsTracker, callStartTimestamps);
+      });
+    }
+  }
 }
 
 function sendConstantQps(client: TestServiceClient, qps: number, failOnFailedRpcs: boolean, callStatsTracker: CallStatsTracker) {
+  const callStartTimestampsTrackers: {[callType: string]: RecentTimestampList} = {};
+  for (const callType of ['EmptyCall', 'UnaryCall']) {
+    callStartTimestampsTrackers[callType] = new RecentTimestampList(qps);
+  }
   setInterval(() => {
     for (const callType of currentConfig.callTypes) {
-      makeSingleRequest(client, callType, failOnFailedRpcs, callStatsTracker);
+      makeSingleRequest(client, callType, failOnFailedRpcs, callStatsTracker, callStartTimestampsTrackers[callType]);
     }
   }, 1000/qps);
   setInterval(() => {

packages/grpc-js-xds/scripts/xds.sh

@@ -48,6 +48,8 @@ git clone -b master --single-branch --depth=1 https://github.com/grpc/grpc.git
 
 grpc/tools/run_tests/helper_scripts/prep_xds.sh
 
+mkdir -p "${KOKORO_ARTIFACTS_DIR}/github/grpc/reports"
+
 GRPC_NODE_TRACE=xds_client,xds_resolver,cds_balancer,eds_balancer,priority,weighted_target,round_robin,resolving_load_balancer,subchannel,keepalive,dns_resolver,fault_injection,http_filter,csds \
   GRPC_NODE_VERBOSITY=DEBUG \
   NODE_XDS_INTEROP_VERBOSITY=1 \
@@ -59,7 +61,7 @@ GRPC_NODE_TRACE=xds_client,xds_resolver,cds_balancer,eds_balancer,priority,weigh
     --gcp_suffix=$(date '+%s') \
     --verbose \
     ${XDS_V3_OPT-} \
-    --client_cmd="$(which node) --enable-source-maps grpc-node/packages/grpc-js-xds/build/interop/xds-interop-client \
+    --client_cmd="$(which node) --enable-source-maps --prof --logfile=${KOKORO_ARTIFACTS_DIR}/github/grpc/reports/prof.log grpc-node/packages/grpc-js-xds/build/interop/xds-interop-client \
       --server=xds:///{server_uri} \
       --stats_port={stats_port} \
       --qps={qps} \

packages/grpc-js-xds/src/xds-client.ts

@@ -18,7 +18,7 @@
 import * as protoLoader from '@grpc/proto-loader';
 // This is a non-public, unstable API, but it's very convenient
 import { loadProtosWithOptionsSync } from '@grpc/proto-loader/build/src/util';
-import { loadPackageDefinition, StatusObject, status, logVerbosity, Metadata, experimental, ChannelOptions, ClientDuplexStream, ServiceError, ChannelCredentials, Channel } from '@grpc/grpc-js';
+import { loadPackageDefinition, StatusObject, status, logVerbosity, Metadata, experimental, ChannelOptions, ClientDuplexStream, ServiceError, ChannelCredentials, Channel, connectivityState } from '@grpc/grpc-js';
 import * as adsTypes from './generated/ads';
 import * as lrsTypes from './generated/lrs';
 import { loadBootstrapInfo } from './xds-bootstrap';
@@ -255,6 +255,7 @@ export class XdsClient {
     DiscoveryRequest,
     DiscoveryResponse__Output
   > | null = null;
+  private receivedAdsResponseOnCurrentStream = false;
 
   private lrsNode: Node | null = null;
   private lrsClient: LoadReportingServiceClient | null = null;
@@ -373,6 +374,9 @@ export class XdsClient {
           {channelOverride: channel}
         );
         this.maybeStartAdsStream();
+        channel.watchConnectivityState(channel.getConnectivityState(false), Infinity, () => {
+          this.handleAdsConnectivityStateUpdate();
+        })
 
         this.lrsClient = new protoDefinitions.envoy.service.load_stats.v3.LoadReportingService(
           serverUri,
@@ -394,7 +398,29 @@ export class XdsClient {
     clearInterval(this.statsTimer);
   }
 
+  private handleAdsConnectivityStateUpdate() {
+    if (!this.adsClient) {
+      return;
+    }
+    const state = this.adsClient.getChannel().getConnectivityState(false);
+    if (state === connectivityState.READY && this.adsCall) {
+      this.reportAdsStreamStarted();
+    }
+    if (state === connectivityState.TRANSIENT_FAILURE) {
+      this.reportStreamError({
+        code: status.UNAVAILABLE,
+        details: 'No connection established to xDS server',
+        metadata: new Metadata()
+      });
+    }
+    this.adsClient.getChannel().watchConnectivityState(state, Infinity, () => {
+      this.handleAdsConnectivityStateUpdate();
+    });
+  }
+
   private handleAdsResponse(message: DiscoveryResponse__Output) {
+    this.receivedAdsResponseOnCurrentStream = true;
+    this.adsBackoff.reset();
     let handleResponseResult: {
       result: HandleResponseResult;
       serviceKind: AdsServiceKind;
@@ -466,7 +492,7 @@ export class XdsClient {
       'ADS stream ended. code=' + streamStatus.code + ' details= ' + streamStatus.details
     );
     this.adsCall = null;
-    if (streamStatus.code !== status.OK) {
+    if (streamStatus.code !== status.OK && !this.receivedAdsResponseOnCurrentStream) {
       this.reportStreamError(streamStatus);
     }
     /* If the backoff timer is no longer running, we do not need to wait any
@@ -496,7 +522,9 @@ export class XdsClient {
     if (this.adsCall !== null) {
       return;
     }
-    this.adsCall = this.adsClient.StreamAggregatedResources();
+    this.receivedAdsResponseOnCurrentStream = false;
+    const metadata = new Metadata({waitForReady: true});
+    this.adsCall = this.adsClient.StreamAggregatedResources(metadata);
     this.adsCall.on('data', (message: DiscoveryResponse__Output) => {
       this.handleAdsResponse(message);
     });
@@ -515,7 +543,9 @@ export class XdsClient {
         this.updateNames(service);
       }
     }
-    this.reportAdsStreamStarted();
+    if (this.adsClient.getChannel().getConnectivityState(false) === connectivityState.READY) {
+      this.reportAdsStreamStarted();
+    }
   }
 
   private maybeSendAdsMessage(typeUrl: string, resourceNames: string[], responseNonce: string, versionInfo: string, errorMessage?: string) {
@@ -547,10 +577,6 @@ export class XdsClient {
    * version info are updated so that it sends the post-update values.
    */
   ack(serviceKind: AdsServiceKind) {
-    /* An ack is the best indication of a successful interaction between the
-     * client and the server, so we can reset the backoff timer here. */
-    this.adsBackoff.reset();
-
     this.updateNames(serviceKind);
   }
 

packages/grpc-js-xds/src/xds-stream-state/xds-stream-state.ts

@@ -213,6 +213,9 @@ export abstract class BaseXdsStreamState<ResponseType> implements XdsStreamState
   }
 
   reportAdsStreamStart() {
+    if (this.isAdsStreamRunning) {
+      return;
+    }
     this.isAdsStreamRunning = true;
     for (const subscriptionEntry of this.subscriptions.values()) {
       if (subscriptionEntry.cachedResponse === null) {

packages/grpc-js/README.md

@@ -59,6 +59,9 @@ Many channel arguments supported in `grpc` are not supported in `@grpc/grpc-js`.
   - `grpc.default_compression_algorithm`
   - `grpc.enable_channelz`
   - `grpc.dns_min_time_between_resolutions_ms`
+  - `grpc.enable_retries`
+  - `grpc.per_rpc_retry_buffer_size`
+  - `grpc.retry_buffer_size`
   - `grpc-node.max_session_memory`
   - `channelOverride`
   - `channelFactoryOverride`

packages/grpc-js/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@grpc/grpc-js",
-  "version": "1.7.2",
+  "version": "1.7.3",
   "description": "gRPC Library for Node - pure JS implementation",
   "homepage": "https://grpc.io/",
   "repository": "https://github.com/grpc/grpc-node/tree/master/packages/grpc-js",
@@ -17,14 +17,14 @@
   "devDependencies": {
     "@types/gulp": "^4.0.6",
     "@types/gulp-mocha": "0.0.32",
-    "@types/lodash": "^4.14.108",
+    "@types/lodash": "^4.14.186",
     "@types/mocha": "^5.2.6",
     "@types/ncp": "^2.0.1",
     "@types/pify": "^3.0.2",
     "@types/semver": "^7.3.9",
     "clang-format": "^1.0.55",
     "execa": "^2.0.3",
-    "gts": "^2.0.0",
+    "gts": "^3.1.1",
     "gulp": "^4.0.2",
     "gulp-mocha": "^6.0.0",
     "lodash": "^4.17.4",
@@ -35,7 +35,7 @@
     "rimraf": "^3.0.2",
     "semver": "^7.3.5",
     "ts-node": "^8.3.0",
-    "typescript": "^3.7.2"
+    "typescript": "^4.8.4"
   },
   "contributors": [
     {

packages/grpc-js/src/call-credentials.ts

@@ -115,6 +115,10 @@ export abstract class CallCredentials {
                 reject(err);
                 return;
               }
+              if (!headers) {
+                reject(new Error('Headers not set by metadata plugin'));
+                return;
+              }
               resolve(headers);
             }
           );

packages/grpc-js/src/call-interface.ts

@@ -36,6 +36,10 @@ export interface StatusObject {
   metadata: Metadata;
 }
 
+export type PartialStatusObject = Pick<StatusObject, 'code' | 'details'> & {
+  metadata: Metadata | null;
+}
+
 export const enum WriteFlags {
   BufferHint = 1,
   NoCompress = 2,

packages/grpc-js/src/channel-options.ts

@@ -44,6 +44,16 @@ export interface ChannelOptions {
   'grpc.default_compression_algorithm'?: CompressionAlgorithms;
   'grpc.enable_channelz'?: number;
   'grpc.dns_min_time_between_resolutions_ms'?: number;
+  'grpc.enable_retries'?: number;
+  'grpc.per_rpc_retry_buffer_size'?: number;
+  /* This option is pattered like a core option, but the core does not have
+   * this option. It is closely related to the option
+   * grpc.per_rpc_retry_buffer_size, which is in the core. The core will likely
+   * implement this functionality using the ResourceQuota mechanism, so there
+   * will probably not be any collision or other inconsistency. */
+  'grpc.retry_buffer_size'?: number;
+  'grpc.max_connection_age_ms'?: number;
+  'grpc.max_connection_age_grace_ms'?: number;
   'grpc-node.max_session_memory'?: number;
   // eslint-disable-next-line @typescript-eslint/no-explicit-any
   [key: string]: any;
@@ -71,6 +81,11 @@ export const recognizedOptions = {
   'grpc.enable_http_proxy': true,
   'grpc.enable_channelz': true,
   'grpc.dns_min_time_between_resolutions_ms': true,
+  'grpc.enable_retries': true,
+  'grpc.per_rpc_retry_buffer_size': true,
+  'grpc.retry_buffer_size': true,
+  'grpc.max_connection_age_ms': true,
+  'grpc.max_connection_age_grace_ms': true,
   'grpc-node.max_session_memory': true,
 };
 

packages/grpc-js/src/client-interceptors.ts

@@ -34,6 +34,7 @@ import { Channel } from './channel';
 import { CallOptions } from './client';
 import { CallCredentials } from './call-credentials';
 import { ClientMethodDefinition } from './make-client';
+import { getErrorMessage } from './error';
 
 /**
  * Error class associated with passing both interceptors and interceptor
@@ -374,7 +375,7 @@ class BaseInterceptingCall implements InterceptingCallInterface {
     } catch (e) {
       this.call.cancelWithStatus(
         Status.INTERNAL,
-        `Request message serialization failure: ${e.message}`
+        `Request message serialization failure: ${getErrorMessage(e)}`
       );
       return;
     }
@@ -401,7 +402,7 @@ class BaseInterceptingCall implements InterceptingCallInterface {
         } catch (e) {
           readError = {
             code: Status.INTERNAL,
-            details: `Response message parsing error: ${e.message}`,
+            details: `Response message parsing error: ${getErrorMessage(e)}`,
             metadata: new Metadata(),
           };
           this.call.cancelWithStatus(readError.code, readError.details);

packages/grpc-js/src/error.ts

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2022 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.
+ *
+ */
+
+export function getErrorMessage(error: unknown): string {
+  if (error instanceof Error) {
+    return error.message;
+  } else {
+    return String(error);
+  }
+}
+
+export function getErrorCode(error: unknown): number | null {
+  if (
+    typeof error === 'object' &&
+    error !== null &&
+    'code' in error &&
+    typeof (error as Record<string, unknown>).code === 'number'
+  ) {
+    return (error as Record<string, number>).code;
+  } else {
+    return null;
+  }
+}

packages/grpc-js/src/filter-stack.ts

@@ -88,6 +88,10 @@ export class FilterStackFactory implements FilterFactory<FilterStack> {
     this.factories.unshift(...filterFactories);
   }
 
+  clone(): FilterStackFactory {
+    return new FilterStackFactory([...this.factories]);
+  }
+
   createFilter(): FilterStack {
     return new FilterStack(
       this.factories.map((factory) => factory.createFilter())