more logs

Author: SiyaoIsHiding

lib/control-connection.js

@@ -44,6 +44,9 @@ const schemaChangeTypes = {
 const supportedProductTypeKey = 'PRODUCT_TYPE';
 const supportedDbaas = 'DATASTAX_APOLLO';
 
+// Process-wide counter used to give each ControlConnection a stable id for log correlation
+let controlConnectionCounter = 0;
+
 /**
  * Represents a connection used by the driver to receive events and to check the status of the cluster.
  * <p>It uses an existing connection from the hosts' connection pool to maintain the driver metadata up-to-date.</p>
@@ -92,7 +95,8 @@ class ControlConnection extends events.EventEmitter {
 
     // Reference to the encoder of the last valid connection
     this._encoder = null;
-    this._debouncer = new EventDebouncer(options.refreshSchemaDelay, this.log.bind(this));
+    this._id = ++controlConnectionCounter;
+    this._debouncer = new EventDebouncer(options.refreshSchemaDelay, this.log.bind(this), this._id);
     this._profileManager = profileManager;
     this._triedHosts = null;
     this._resolvedContactPoints = new Map();
@@ -579,6 +583,7 @@ class ControlConnection extends events.EventEmitter {
     this.connection.on('nodeTopologyChange', this._nodeTopologyChangeHandler.bind(this));
     this.connection.on('nodeStatusChange', this._nodeStatusChangeHandler.bind(this));
     this.connection.on('nodeSchemaChange', this._nodeSchemaChangeHandler.bind(this));
+    console.log(`[cc ${this._id}] Listener count on ${this.connection.endpointFriendlyName}: ${this.connection.listenerCount('nodeSchemaChange')}`);
     const request = new requests.RegisterRequest(['TOPOLOGY_CHANGE', 'STATUS_CHANGE', 'SCHEMA_CHANGE']);
     await this.connection.send(request, null);
   }
@@ -633,7 +638,7 @@ class ControlConnection extends events.EventEmitter {
    * Handles a SCHEMA_CHANGE event
    */
   _nodeSchemaChangeHandler(event) {
-    this.log('info', 'Schema change', event);
+    this.log('info', `[cc ${this._id}] Schema change`, event);
     if (!this.options.isMetadataSyncEnabled) {
       return;
     }
@@ -659,10 +664,10 @@ class ControlConnection extends events.EventEmitter {
           delete self.metadata.keyspaces[event.keyspace];
         };
 
-        return this._scheduleObjectRefresh(handler, event.keyspace, null, processNow);
+        return this._scheduleObjectRefresh(handler, event.keyspace, null, processNow, event);
       }
 
-      return this._scheduleKeyspaceRefresh(event.keyspace, processNow);
+      return this._scheduleKeyspaceRefresh(event.keyspace, processNow, event);
     }
 
     const ksInfo = this.metadata.keyspaces[event.keyspace];
@@ -713,19 +718,20 @@ class ControlConnection extends events.EventEmitter {
    * @param {Boolean} processNow
    * @returns {Promise<void>}
    */
-  _scheduleObjectRefresh(handler, keyspace, cqlObject, processNow) {
-    return this._debouncer.eventReceived({ handler, keyspace, cqlObject }, processNow);
+  _scheduleObjectRefresh(handler, keyspace, cqlObject, processNow, event = null) {
+    return this._debouncer.eventReceived({ handler, keyspace, cqlObject, event}, processNow);
   }
 
   /**
    * @param {String} keyspace
    * @param {Boolean} processNow
    * @returns {Promise<void>}
    */
-  _scheduleKeyspaceRefresh(keyspace, processNow) {
+  _scheduleKeyspaceRefresh(keyspace, processNow, event = null) {
     return this._debouncer.eventReceived({
       handler: () => this.metadata.refreshKeyspace(keyspace),
-      keyspace
+      keyspace,
+      event
     }, processNow);
   }
 

lib/metadata/event-debouncer.js

@@ -35,11 +35,13 @@ class EventDebouncer {
    * @param {Number} delay
    * @param {Function} logger
    */
-  constructor(delay, logger) {
+  constructor(delay, logger, id) {
     this._delay = delay;
     this._logger = logger;
     this._queue = null;
     this._timeout = null;
+    // Identifier used to correlate debouncer log lines with the owning ControlConnection
+    this._id = id;
   }
 
   /**
@@ -92,7 +94,7 @@ class EventDebouncer {
       const q = self._queue;
       self._queue = null;
       self._timeout = null;
-      processQueue(q);
+      processQueue(q, self._id);
     }
     if (delay === 0) {
       // no delay, process immediately
@@ -129,7 +131,7 @@ class EventDebouncer {
  * @param {{callbacks: Array, keyspaces: Object, mainEvent: Object}} q
  * @private
  */
-function processQueue (q) {
+function processQueue (q, id) {
   if (q.mainEvent) {
     // refresh all by invoking 1 handler and invoke all pending callbacks
     return promiseUtils.toCallback(q.mainEvent.handler(), (err) => {
@@ -141,14 +143,16 @@ function processQueue (q) {
 
   utils.each(Object.keys(q.keyspaces), function eachKeyspace(name, next) {
     const keyspaceEvents = q.keyspaces[name];
+    console.log(`[deb ${id}] Current queue: ${JSON.stringify(q)}`);
     if (keyspaceEvents.mainEvent) {
       // refresh a keyspace
       return promiseUtils.toCallback(keyspaceEvents.mainEvent.handler(), function mainEventCallback(err) {
+        console.log(`[deb ${id}] Processed main event ${JSON.stringify(keyspaceEvents.mainEvent)}`);
         for (let i = 0; i < keyspaceEvents.events.length; i++) {
           // log if not main event so this event is getting dropped
           if (keyspaceEvents.events[i] !== keyspaceEvents.mainEvent) {
             // eslint-disable-next-line
-            console.log(util.format('Event for %s.%s is being dropped because a full refresh of the keyspace is scheduled', name, JSON.stringify(keyspaceEvents.events[i])));
+            console.log(util.format('[deb %s] Event for %s.%s is being dropped because a full refresh of the keyspace is scheduled', id, name, JSON.stringify(keyspaceEvents.events[i])));
           }
           keyspaceEvents.events[i].callback(err);
         }

test/integration/short/control-connection-tests.js

@@ -43,7 +43,11 @@ describe('ControlConnection', function () {
       options.refreshSchemaDelay = 100;
 
       const cc = newInstance(options);
-      const otherClient = new Client(helper.baseOptions);
+      const otherClient = new Client(utils.extend({}, helper.baseOptions, {
+        logEmitter: (event, level, className, message, furtherInfo) =>
+          // eslint-disable-next-line
+          console.log(`${new Date().toISOString()} [otherClient][${level}] ${className}: ${message}`, furtherInfo || '')
+      }));
 
       helper.afterThisTest(() => otherClient.shutdown());
       disposeAfter(cc);
@@ -160,7 +164,9 @@ describe('ControlConnection', function () {
     });
 
     it('should not break when refreshing concurrently', async () => {
-      const cc = newInstance();
+      const cc = newInstance({logEmitter: (event, level, className, message, furtherInfo) =>
+        // eslint-disable-next-line 
+        console.log(`${new Date().toISOString()} [${level}] ${className}: ${message}`, furtherInfo || '')});
       cc.options.policies.loadBalancing = new policies.loadBalancing.RoundRobinPolicy();
       disposeAfter(cc);