Merge branch 'master' into grpc-js_retry2

Author: murgatroid99

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,15 +329,33 @@ 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(() => {
+    console.log(`Accumulated stats: ${JSON.stringify(accumulatedStats, undefined, 2)}`);
+  }, 1000);
 }
 
 const callTypeEnumMap = {

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/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

@@ -52,6 +52,8 @@ export interface ChannelOptions {
    * 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;
@@ -82,6 +84,8 @@ export const recognizedOptions = {
   '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())

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

@@ -478,7 +478,7 @@ export class InternalChannel {
       parentCall: parentCall,
     };
 
-    const call = new ResolvingCall(this, method, finalOptions, this.filterStackFactory, this.credentials._getCallCredentials(), callNumber);
+    const call = new ResolvingCall(this, method, finalOptions, this.filterStackFactory.clone(), this.credentials._getCallCredentials(), callNumber);
 
     if (this.channelzEnabled) {
       this.callTracker.addCallStarted();