Add channelz tests and fix some bugs

murgatroid99 · murgatroid99 · commit d60c4ea16f59 · 2021-09-27T10:46:39.000-07:00
diff --git a/packages/grpc-js/src/call-stream.ts b/packages/grpc-js/src/call-stream.ts
@@ -25,7 +25,7 @@ import { FilterStackFactory, FilterStack } from './filter-stack';
 import { Metadata } from './metadata';
 import { StreamDecoder } from './stream-decoder';
 import { ChannelImplementation } from './channel';
-import { Subchannel } from './subchannel';
+import { SubchannelCallStatsTracker, Subchannel } from './subchannel';
 import * as logging from './logging';
 import { LogVerbosity } from './constants';
 import { ServerSurfaceCall } from './server-call';
@@ -252,6 +252,8 @@ export class Http2CallStream implements Call {
   private statusWatchers: ((status: StatusObject) => void)[] = [];
   private streamEndWatchers: ((success: boolean) => void)[] = [];
 
+  private callStatsTracker: SubchannelCallStatsTracker | null = null;
+
   constructor(
     private readonly methodName: string,
     private readonly channel: ChannelImplementation,
@@ -468,10 +470,16 @@ export class Http2CallStream implements Call {
     this.endCall(status);
   }
 
+  private writeMessageToStream(message: Buffer, callback: WriteCallback) {
+    this.callStatsTracker?.addMessageSent();
+    this.http2Stream!.write(message, callback);
+  }
+
   attachHttp2Stream(
     stream: http2.ClientHttp2Stream,
     subchannel: Subchannel,
-    extraFilters: FilterFactory<Filter>[]
+    extraFilters: FilterFactory<Filter>[],
+    callStatsTracker: SubchannelCallStatsTracker
   ): void {
     this.filterStack.push(
       extraFilters.map((filterFactory) => filterFactory.createFilter(this))
@@ -484,6 +492,7 @@ export class Http2CallStream implements Call {
       );
       this.http2Stream = stream;
       this.subchannel = subchannel;
+      this.callStatsTracker = callStatsTracker;
       subchannel.addDisconnectListener(this.disconnectListener);
       subchannel.callRef();
       stream.on('response', (headers, flags) => {
@@ -549,6 +558,7 @@ export class Http2CallStream implements Call {
 
         for (const message of messages) {
           this.trace('parsed message of length ' + message.length);
+          this.callStatsTracker!.addMessageReceived();
           this.tryPush(message);
         }
       });
@@ -666,7 +676,7 @@ export class Http2CallStream implements Call {
             this.pendingWrite.length +
             ' (deferred)'
         );
-        stream.write(this.pendingWrite, this.pendingWriteCallback);
+        this.writeMessageToStream(this.pendingWrite, this.pendingWriteCallback);
       }
       this.maybeCloseWrites();
     }
@@ -802,7 +812,7 @@ export class Http2CallStream implements Call {
         this.pendingWriteCallback = cb;
       } else {
         this.trace('sending data chunk of length ' + message.message.length);
-        this.http2Stream.write(message.message, cb);
+        this.writeMessageToStream(message.message, cb);
         this.maybeCloseWrites();
       }
     }, this.handleFilterError.bind(this));
diff --git a/packages/grpc-js/src/channel.ts b/packages/grpc-js/src/channel.ts
@@ -103,6 +103,12 @@ export interface Channel {
     deadline: Date | number,
     callback: (error?: Error) => void
   ): void;
+  /**
+   * Get the channelz reference object for this channel. A request to the
+   * channelz service for the id in this object will provide information
+   * about this channel.
+   */
+  getChannelzRef(): ChannelRef;
   /**
    * Create a call object. Call is an opaque type that is used by the Client
    * class. This function is called by the gRPC library when starting a
@@ -243,7 +249,7 @@ export class ChannelImplementation implements Channel {
           Object.assign({}, this.options, subchannelArgs),
           this.credentials
         );
-        this.channelzTrace.addTrace('CT_INFO', 'Created or got existing subchannel', subchannel.getChannelzRef());
+        this.channelzTrace.addTrace('CT_INFO', 'Created subchannel or used existing subchannel', subchannel.getChannelzRef());
         return subchannel;
       },
       updateState: (connectivityState: ConnectivityState, picker: Picker) => {
@@ -677,6 +683,10 @@ export class ChannelImplementation implements Channel {
     this.connectivityStateWatchers.push(watcherObject);
   }
 
+  getChannelzRef() {
+    return this.channelzRef;
+  }
+
   createCall(
     method: string,
     deadline: Deadline,
diff --git a/packages/grpc-js/src/channelz.ts b/packages/grpc-js/src/channelz.ts
@@ -51,6 +51,7 @@ import { ChannelzDefinition, ChannelzHandlers } from "./generated/grpc/channelz/
 import { ProtoGrpcType as ChannelzProtoGrpcType } from "./generated/channelz";
 import type { loadSync } from '@grpc/proto-loader';
 import { registerAdminService } from "./admin";
+import { loadPackageDefinition } from "./make-client";
 
 export type TraceSeverity = 'CT_UNKNOWN' | 'CT_INFO' | 'CT_WARNING' | 'CT_ERROR';
 
@@ -455,9 +456,10 @@ function dateToProtoTimestamp(date?: Date | null): Timestamp | null {
   if (!date) {
     return null;
   }
+  const millisSinceEpoch = date.getTime();
   return {
-    seconds: date.getSeconds(),
-    nanos: date.getMilliseconds() * 1_000_000
+    seconds: (millisSinceEpoch / 1000) | 0,
+    nanos: (millisSinceEpoch % 1000) * 1_000_000
   }
 }
 
@@ -664,7 +666,10 @@ function GetServerSockets(call: ServerUnaryCall<GetServerSocketsRequest__Output,
   const startId = Number.parseInt(call.request.start_socket_id);
   const maxResults = Number.parseInt(call.request.max_results);
   const resolvedInfo = serverEntry.getInfo();
-  const allSockets = resolvedInfo.listenerChildren.sockets.concat(resolvedInfo.sessionChildren.sockets).sort((ref1, ref2) => ref1.id - ref2.id);
+  // If we wanted to include listener sockets in the result, this line would
+  // instead say
+  // const allSockets = resolvedInfo.listenerChildren.sockets.concat(resolvedInfo.sessionChildren.sockets).sort((ref1, ref2) => ref1.id - ref2.id);
+  const allSockets = resolvedInfo.sessionChildren.sockets.sort((ref1, ref2) => ref1.id - ref2.id);
   const resultList: SocketRefMessage[] = [];
   let i = 0;
   for (; i < allSockets.length; i++) {
@@ -709,10 +714,11 @@ export function getChannelzServiceDefinition(): ChannelzDefinition {
     defaults: true,
     oneofs: true,
     includeDirs: [
-      '../../proto'
+      `${__dirname}/../../proto`
     ]
-  }) as unknown as ChannelzProtoGrpcType;
-  loadedChannelzDefinition = loadedProto.grpc.channelz.v1.Channelz.service;
+  });
+  const channelzGrpcObject = loadPackageDefinition(loadedProto) as unknown as ChannelzProtoGrpcType;
+  loadedChannelzDefinition = channelzGrpcObject.grpc.channelz.v1.Channelz.service;
   return loadedChannelzDefinition;
 }
 
diff --git a/packages/grpc-js/src/server-call.ts b/packages/grpc-js/src/server-call.ts
@@ -570,7 +570,6 @@ export class Http2ServerCallStream<
       const response = this.serializeMessage(value!);
 
       this.write(response);
-      this.emit('sendMessage');
       this.sendStatus({ code: Status.OK, details: 'OK', metadata });
     } catch (err) {
       err.code = Status.INTERNAL;
diff --git a/packages/grpc-js/src/server.ts b/packages/grpc-js/src/server.ts
@@ -188,7 +188,7 @@ export class Server {
         const peerCertificate = tlsSocket.getPeerCertificate();
         tlsInfo = {
           cipherSuiteStandardName: cipherInfo.standardName ?? null,
-          cipherSuiteOtherName: cipherInfo.standardName ? cipherInfo.name: null,
+          cipherSuiteOtherName: cipherInfo.standardName ? null : cipherInfo.name,
           localCertificate: (certificate && 'raw' in certificate) ? certificate.raw : null,
           remoteCertificate: (peerCertificate && 'raw' in peerCertificate) ? peerCertificate.raw : null
         };
@@ -424,6 +424,7 @@ export class Server {
                   remoteFlowControlWindow: null
                 };
               });
+              this.listenerChildrenTracker.refChild(channelzRef);
               this.http2ServerList.push({server: http2Server, channelzRef: channelzRef});
               trace('Successfully bound ' + subchannelAddressToString(boundSubchannelAddress));
               resolve('port' in boundSubchannelAddress ? boundSubchannelAddress.port : portNum);
@@ -491,6 +492,7 @@ export class Server {
               remoteFlowControlWindow: null
             };
           });
+          this.listenerChildrenTracker.refChild(channelzRef);
           this.http2ServerList.push({server: http2Server, channelzRef: channelzRef});
           trace('Successfully bound ' + subchannelAddressToString(boundSubchannelAddress));
           resolve(
@@ -676,6 +678,10 @@ export class Server {
     throw new Error('Not yet implemented');
   }
 
+  getChannelzRef() {
+    return this.channelzRef;
+  }
+
   private _setupHandlers(
     http2Server: http2.Http2Server | http2.Http2SecureServer
   ): void {
diff --git a/packages/grpc-js/src/subchannel.ts b/packages/grpc-js/src/subchannel.ts
@@ -68,6 +68,11 @@ export type ConnectivityStateListener = (
   newState: ConnectivityState
 ) => void;
 
+export interface SubchannelCallStatsTracker {
+  addMessageSent(): void;
+  addMessageReceived(): void;
+}
+
 const {
   HTTP2_HEADER_AUTHORITY,
   HTTP2_HEADER_CONTENT_TYPE,
@@ -178,33 +183,6 @@ export class Subchannel {
   private messagesReceived = 0;
   private lastMessageSentTimestamp: Date | null = null;
   private lastMessageReceivedTimestamp: Date | null = null;
-  private MessageCountFilter = class extends BaseFilter implements Filter {
-    private session: http2.ClientHttp2Session;
-    constructor(private parent: Subchannel) {
-      super();
-      this.session = parent.session!;
-    }
-    sendMessage(message: Promise<WriteObject>): Promise<WriteObject> {
-      if (this.parent.session === this.session) {
-        this.parent.messagesSent += 1;
-        this.parent.lastMessageSentTimestamp = new Date();
-      }
-      return message;
-    }
-    receiveMessage(message: Promise<Buffer>): Promise<Buffer> {
-      if (this.parent.session === this.session) {
-        this.parent.messagesReceived += 1;
-        this.parent.lastMessageReceivedTimestamp = new Date();
-      }
-      return message;
-    }
-  };
-  private MessageCountFilterFactory = class implements FilterFactory<Filter> {
-    constructor(private parent: Subchannel) {}
-    createFilter(callStream: Call): Filter {
-      return new this.parent.MessageCountFilter(this.parent);
-    }
-  }
 
   /**
    * A class representing a connection to a single backend.
@@ -848,8 +826,15 @@ export class Subchannel {
         }
       }
     });
-    extraFilterFactories.push(new this.MessageCountFilterFactory(this));
-    callStream.attachHttp2Stream(http2Stream, this, extraFilterFactories);
+    callStream.attachHttp2Stream(http2Stream, this, extraFilterFactories, {
+      addMessageSent: () => {
+        this.messagesSent += 1;
+        this.lastMessageSentTimestamp = new Date();
+      },
+      addMessageReceived: () => {
+        this.messagesReceived += 1;
+      }
+    });
   }
 
   /**
diff --git a/packages/grpc-js/test/test-channelz.ts b/packages/grpc-js/test/test-channelz.ts
diff --git a/test/channelz/channelz_manual_test.js b/test/channelz/channelz_manual_test.js
diff --git a/test/interop/interop_server.js b/test/interop/interop_server.js
