feat: Enhanced trace logging

Author: paveltiunov

packages/cubejs-query-orchestrator/orchestrator/LocalQueueDriver.js

@@ -61,7 +61,13 @@ class LocalQueueDriverConnection {
 
   addToQueue(keyScore, queryKey, time, queryHandler, query, priority, options) {
     const queryQueueObj = {
-      queryHandler, query, queryKey, stageQueryKey: options.stageQueryKey, priority, requestId: options.requestId
+      queryHandler,
+      query,
+      queryKey,
+      stageQueryKey: options.stageQueryKey,
+      priority,
+      requestId: options.requestId,
+      addedToQueueTime: new Date().getTime()
     };
     const key = this.redisHash(queryKey);
     if (!this.queryDef[key]) {
@@ -124,8 +130,8 @@ class LocalQueueDriverConnection {
     return this.queueArray(this.active, new Date().getTime() - this.heartBeatTimeout * 1000);
   }
 
-  async getQueryStageState() {
-    return [this.queueArray(this.active), this.queueArray(this.toProcess), R.clone(this.queryDef)];
+  async getQueryStageState(onlyKeys) {
+    return [this.queueArray(this.active), this.queueArray(this.toProcess), onlyKeys ? {} : R.clone(this.queryDef)];
   }
 
   async getQueryDef(queryKey) {

packages/cubejs-query-orchestrator/orchestrator/QueryQueue.js

@@ -58,11 +58,25 @@ class QueryQueue {
           queueSize,
           queryKey,
           queuePrefix: this.redisQueuePrefix,
-          requestId: options.requestId
+          requestId: options.requestId,
         });
       }
 
       await this.reconcileQueue(redisClient);
+
+      const queryDef = await redisClient.getQueryDef(queryKey);
+      const [active, toProcess] = await redisClient.getQueryStageState(true);
+
+      this.logger('Waiting for query', {
+        queueSize,
+        queryKey: queryDef.queryKey,
+        queuePrefix: this.redisQueuePrefix,
+        requestId: options.requestId,
+        active: active.indexOf(redisClient.redisHash(queryKey)) !== -1,
+        queueIndex: toProcess.indexOf(redisClient.redisHash(queryKey)),
+        waitingForRequestId: queryDef.requestId
+      });
+
       result = await redisClient.getResultBlocking(queryKey);
       if (!result) {
         throw new ContinueWaitError();
@@ -178,11 +192,13 @@ class QueryQueue {
         if (query) {
           let executionResult;
           const startQueryTime = (new Date()).getTime();
+          const timeInQueue = (new Date()).getTime() - query.addedToQueueTime;
           this.logger('Performing query', {
             queueSize,
             queryKey: query.queryKey,
             queuePrefix: this.redisQueuePrefix,
-            requestId: query.requestId
+            requestId: query.requestId,
+            timeInQueue
           });
           await redisClient.optimisticQueryUpdate(queryKey, { startQueryTime });
 
@@ -216,7 +232,8 @@ class QueryQueue {
               duration: ((new Date()).getTime() - startQueryTime),
               queryKey: query.queryKey,
               queuePrefix: this.redisQueuePrefix,
-              requestId: query.requestId
+              requestId: query.requestId,
+              timeInQueue
             });
           } catch (e) {
             executionResult = {
@@ -255,6 +272,12 @@ class QueryQueue {
 
         await this.reconcileQueue(redisClient);
       }
+    } catch (e) {
+      this.logger('Queue storage error', {
+        queryKey,
+        error: (e.stack || e).toString(),
+        queuePrefix: this.redisQueuePrefix
+      });
     } finally {
       redisClient.release();
     }

packages/cubejs-query-orchestrator/orchestrator/RedisQueueDriver.js

@@ -34,7 +34,13 @@ class RedisQueueDriverConnection {
         this.queriesDefKey(),
         this.redisHash(queryKey),
         JSON.stringify({
-          queryHandler, query, queryKey, stageQueryKey: options.stageQueryKey, priority, requestId: options.requestId
+          queryHandler,
+          query,
+          queryKey,
+          stageQueryKey: options.stageQueryKey,
+          priority,
+          requestId: options.requestId,
+          addedToQueueTime: new Date().getTime()
         })
       ])
       .zcard(this.toProcessRedisKey())
@@ -91,12 +97,14 @@ class RedisQueueDriverConnection {
     );
   }
 
-  async getQueryStageState() {
-    const [active, toProcess, allQueryDefs] = await this.redisClient.multi()
+  async getQueryStageState(onlyKeys) {
+    let request = this.redisClient.multi()
       .zrange([this.activeRedisKey(), 0, -1])
-      .zrange([this.toProcessRedisKey(), 0, -1])
-      .hgetall(this.queriesDefKey())
-      .execAsync();
+      .zrange([this.toProcessRedisKey(), 0, -1]);
+    if (!onlyKeys) {
+      request = request.hgetall(this.queriesDefKey());
+    }
+    const [active, toProcess, allQueryDefs] = await request.execAsync();
     return [active, toProcess, R.map(q => JSON.parse(q), allQueryDefs || {})];
   }