feat: Propagate requestId for trace logging

Author: paveltiunov

packages/cubejs-api-gateway/SubscriptionServer.js

@@ -68,6 +68,7 @@ class SubscriptionServer {
         context,
         isSubscription,
         res: this.resultFn(connectionId, message.messageId),
+        requestId: message.requestId || `${connectionId}-${message.messageId}`,
         subscriptionState: async () => {
           const subscription = await this.subscriptionStore.getSubscription(connectionId, message.messageId);
           return subscription && subscription.state;

packages/cubejs-api-gateway/index.js

@@ -2,8 +2,9 @@ const jwt = require('jsonwebtoken');
 const R = require('ramda');
 const Joi = require('@hapi/joi');
 const moment = require('moment');
-const dateParser = require('./dateParser');
+const uuid = require('uuid/v4');
 
+const dateParser = require('./dateParser');
 const UserError = require('./UserError');
 const CubejsHandlerError = require('./CubejsHandlerError');
 const SubscriptionServer = require('./SubscriptionServer');
@@ -226,30 +227,34 @@ class ApiGateway {
       await this.load({
         query: req.query.query,
         context: this.contextByReq(req),
-        res: this.resToResultFn(res)
+        res: this.resToResultFn(res),
+        requestId: this.requestIdByReq(req)
       });
     }));
 
     app.get(`${this.basePath}/v1/subscribe`, this.checkAuthMiddleware, (async (req, res) => {
       await this.load({
         query: req.query.query,
         context: this.contextByReq(req),
-        res: this.resToResultFn(res)
+        res: this.resToResultFn(res),
+        requestId: this.requestIdByReq(req)
       });
     }));
 
     app.get(`${this.basePath}/v1/sql`, this.checkAuthMiddleware, (async (req, res) => {
       await this.sql({
         query: req.query.query,
         context: this.contextByReq(req),
-        res: this.resToResultFn(res)
+        res: this.resToResultFn(res),
+        requestId: this.requestIdByReq(req)
       });
     }));
 
     app.get(`${this.basePath}/v1/meta`, this.checkAuthMiddleware, (async (req, res) => {
       await this.meta({
         context: this.contextByReq(req),
-        res: this.resToResultFn(res)
+        res: this.resToResultFn(res),
+        requestId: this.requestIdByReq(req)
       });
     }));
   }
@@ -262,20 +267,22 @@ class ApiGateway {
     return requestStarted && (new Date().getTime() - requestStarted.getTime());
   }
 
-  async meta({ context, res }) {
+  async meta({ context, res, requestId }) {
     const requestStarted = new Date();
     try {
       const metaConfig = await this.getCompilerApi(context).metaConfig();
       const cubes = metaConfig.map(c => c.config);
       res({ cubes });
     } catch (e) {
       this.handleError({
-        e, context, res, requestStarted
+        e, context, res, requestStarted, requestId
       });
     }
   }
 
-  async sql({ query, context, res }) {
+  async sql({
+    query, context, res, requestId
+  }) {
     const requestStarted = new Date();
     try {
       query = this.parseQueryParam(query);
@@ -289,18 +296,21 @@ class ApiGateway {
       });
     } catch (e) {
       this.handleError({
-        e, context, query, res, requestStarted
+        e, context, query, res, requestStarted, requestId
       });
     }
   }
 
-  async load({ query, context, res }) {
+  async load({
+    query, context, res, requestId
+  }) {
     const requestStarted = new Date();
     try {
       query = this.parseQueryParam(query);
       this.log(context, {
         type: 'Load Request',
-        query
+        query,
+        requestId
       });
       const normalizedQuery = await this.queryTransformer(normalizeQuery(query), context);
       const [compilerSqlResult, metaConfigResult] = await Promise.all([
@@ -320,15 +330,17 @@ class ApiGateway {
         query: sqlQuery.sql[0],
         values: sqlQuery.sql[1],
         continueWait: true,
-        renewQuery: normalizedQuery.renewQuery
+        renewQuery: normalizedQuery.renewQuery,
+        requestId
       };
       const response = await this.getAdapterApi({
         ...context, dataSource: sqlQuery.dataSource
       }).executeQuery(toExecute);
       this.log(context, {
         type: 'Load Request Success',
         query,
-        duration: this.duration(requestStarted)
+        duration: this.duration(requestStarted),
+        requestId
       });
       const flattenAnnotation = {
         ...annotation.measures,
@@ -347,13 +359,13 @@ class ApiGateway {
       });
     } catch (e) {
       this.handleError({
-        e, context, query, res, requestStarted
+        e, context, query, res, requestStarted, requestId
       });
     }
   }
 
   async subscribe({
-    query, context, res, subscribe, subscriptionState
+    query, context, res, subscribe, subscriptionState, requestId
   }) {
     const requestStarted = new Date();
     try {
@@ -373,6 +385,7 @@ class ApiGateway {
       await this.load({
         query,
         context,
+        requestId: `${requestId}-${uuid()}`,
         res: (message, opts) => {
           if (message.error) {
             error = { message, opts };
@@ -427,14 +440,19 @@ class ApiGateway {
     return { authInfo: req.authInfo };
   }
 
+  requestIdByReq(req) {
+    return req.headers['x-request-id'] || req.headers.traceparent || uuid();
+  }
+
   handleError({
-    e, context, query, res, requestStarted
+    e, context, query, res, requestStarted, requestId
   }) {
     if (e instanceof CubejsHandlerError) {
       this.log(context, {
         type: e.type,
         query,
         error: e.message,
+        requestId,
         duration: this.duration(requestStarted)
       });
       res({ error: e.message }, { status: e.status });
@@ -443,6 +461,7 @@ class ApiGateway {
         type: 'Continue wait',
         query,
         error: e.message,
+        requestId,
         duration: this.duration(requestStarted)
       });
       res(e, { status: 200 });
@@ -451,6 +470,7 @@ class ApiGateway {
         type: 'Orchestrator error',
         query,
         error: e.error,
+        requestId,
         duration: this.duration(requestStarted)
       });
       res(e, { status: 400 });
@@ -459,6 +479,7 @@ class ApiGateway {
         type: 'Internal Server Error',
         query,
         error: e.stack || e.toString(),
+        requestId,
         duration: this.duration(requestStarted)
       });
       res({ error: e.toString() }, { status: 500 });

packages/cubejs-api-gateway/package.json

@@ -21,7 +21,8 @@
     "jsonwebtoken": "^8.3.0",
     "moment": "^2.24.0",
     "moment-timezone": "^0.5.27",
-    "ramda": "^0.25.0"
+    "ramda": "^0.25.0",
+    "uuid": "^3.3.3"
   },
   "devDependencies": {
     "eslint": "^4.14.0",

packages/cubejs-api-gateway/yarn.lock

@@ -4498,7 +4498,7 @@ utils-merge@1.0.1:
   resolved "https://registry.yarnpkg.com/utils-merge/-/utils-merge-1.0.1.tgz#9f95710f50a267947b2ccc124741c1028427e713"
   integrity sha1-n5VxD1CiZ5R7LMwSR0HBAoQn5xM=
 
-uuid@^3.3.2:
+uuid@^3.3.2, uuid@^3.3.3:
   version "3.3.3"
   resolved "https://registry.yarnpkg.com/uuid/-/uuid-3.3.3.tgz#4568f0216e78760ee1dbf3a4d2cf53e224112866"
   integrity sha512-pW0No1RGHgzlpHJO1nsVrHKpOEIxkGg1xB+v0ZmdNH5OAeAwzAVrCnI2/6Mtx+Uys6iaylxa+D3g4j63IKKjSQ==

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

@@ -61,7 +61,7 @@ class LocalQueueDriverConnection {
 
   addToQueue(keyScore, queryKey, time, queryHandler, query, priority, options) {
     const queryQueueObj = {
-      queryHandler, query, queryKey, stageQueryKey: options.stageQueryKey, priority
+      queryHandler, query, queryKey, stageQueryKey: options.stageQueryKey, priority, requestId: options.requestId
     };
     const key = this.redisHash(queryKey);
     if (!this.queryDef[key]) {

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

@@ -165,6 +165,7 @@ class PreAggregationLoader {
     this.loadCache = loadCache;
     this.waitForRenew = options.waitForRenew;
     this.externalDriverFactory = preAggregations.externalDriverFactory;
+    this.requestId = options.requestId;
   }
 
   async loadPreAggregation() {
@@ -179,7 +180,11 @@ class PreAggregationLoader {
       if (versionEntryByStructureVersion) {
         this.loadPreAggregationWithKeys().catch(e => {
           if (!(e instanceof ContinueWaitError)) {
-            this.logger('Error loading pre-aggregation', { error: (e.stack || e), preAggregation: this.preAggregation });
+            this.logger('Error loading pre-aggregation', {
+              error: (e.stack || e),
+              preAggregation: this.preAggregation,
+              requestId: this.requestId
+            });
           }
         });
         return this.targetTableName(versionEntryByStructureVersion);
@@ -248,12 +253,18 @@ class PreAggregationLoader {
 
     if (versionEntry) {
       if (versionEntry.structure_version !== newVersionEntry.structure_version) {
-        this.logger('Invalidating pre-aggregation structure', { preAggregation: this.preAggregation });
+        this.logger('Invalidating pre-aggregation structure', {
+          preAggregation: this.preAggregation,
+          requestId: this.requestId
+        });
         await this.executeInQueue(invalidationKeys, 10, newVersionEntry);
         return mostRecentTargetTableName();
       } else if (versionEntry.content_version !== newVersionEntry.content_version) {
         if (this.waitForRenew) {
-          this.logger('Waiting for pre-aggregation renew', { preAggregation: this.preAggregation });
+          this.logger('Waiting for pre-aggregation renew', {
+            preAggregation: this.preAggregation,
+            requestId: this.requestId
+          });
           await this.executeInQueue(invalidationKeys, 0, newVersionEntry);
           return mostRecentTargetTableName();
         } else {
@@ -268,7 +279,10 @@ class PreAggregationLoader {
         }
       }
     } else {
-      this.logger('Creating pre-aggregation from scratch', { preAggregation: this.preAggregation });
+      this.logger('Creating pre-aggregation from scratch', {
+        preAggregation: this.preAggregation,
+        requestId: this.requestId
+      });
       await this.executeInQueue(invalidationKeys, 10, newVersionEntry);
       return mostRecentTargetTableName();
     }
@@ -283,7 +297,10 @@ class PreAggregationLoader {
   }
 
   scheduleRefresh(invalidationKeys, newVersionEntry) {
-    this.logger('Refreshing pre-aggregation content', { preAggregation: this.preAggregation });
+    this.logger('Refreshing pre-aggregation content', {
+      preAggregation: this.preAggregation,
+      requestId: this.requestId
+    });
     this.executeInQueue(invalidationKeys, 0, newVersionEntry)
       .then(() => {
         delete this.preAggregations.refreshErrors[newVersionEntry.table_name];
@@ -297,9 +314,11 @@ class PreAggregationLoader {
             this.preAggregations.refreshErrors[newVersionEntry.table_name][newVersionEntry.content_version].error = e;
             this.preAggregations.refreshErrors[newVersionEntry.table_name][newVersionEntry.content_version].counter += 1;
           }
-          this.logger('Error refreshing pre-aggregation', { error: (e.stack || e), preAggregation: this.preAggregation })
+          this.logger('Error refreshing pre-aggregation', {
+            error: (e.stack || e), preAggregation: this.preAggregation, requestId: this.requestId
+          });
         }
-      })
+      });
   }
 
   executeInQueue(invalidationKeys, priority, newVersionEntry) {
@@ -309,7 +328,8 @@ class PreAggregationLoader {
       {
         preAggregation: this.preAggregation,
         preAggregationsTablesToTempTables: this.preAggregationsTablesToTempTables,
-        newVersionEntry
+        newVersionEntry,
+        requestId: this.requestId
       },
       priority,
       { stageQueryKey: PreAggregations.preAggregationQueryCacheKey(this.preAggregation) }
@@ -360,14 +380,20 @@ class PreAggregationLoader {
       throw new Error(`Can't load external pre-aggregation: source driver doesn't support downloadTable()`);
     }
     const table = this.targetTableName(newVersionEntry);
-    this.logger('Downloading external pre-aggregation', { preAggregation: this.preAggregation });
+    this.logger('Downloading external pre-aggregation', {
+      preAggregation: this.preAggregation,
+      requestId: this.requestId
+    });
     const tableData = await client.downloadTable(table);
     const columns = await client.tableColumnTypes(table);
     const externalDriver = await this.externalDriverFactory();
     if (!externalDriver.uploadTable) {
       throw new Error(`Can't load external pre-aggregation: destination driver doesn't support uploadTable()`);
     }
-    this.logger('Uploading external pre-aggregation', { preAggregation: this.preAggregation });
+    this.logger('Uploading external pre-aggregation', {
+      preAggregation: this.preAggregation,
+      requestId: this.requestId
+    });
     await externalDriver.uploadTable(table, columns, tableData);
     await this.loadCache.reset(this.preAggregation);
     await this.dropOrphanedTables(externalDriver, table);
@@ -389,7 +415,10 @@ class PreAggregationLoader {
     const toDrop = actualTables
       .map(t => `${this.preAggregation.preAggregationsSchema}.${t.table_name || t.TABLE_NAME}`)
       .filter(t => tablesToSave.indexOf(t) === -1);
-    this.logger('Dropping orphaned tables', { tablesToDrop: JSON.stringify(toDrop) });
+    this.logger('Dropping orphaned tables', {
+      tablesToDrop: JSON.stringify(toDrop),
+      requestId: this.requestId
+    });
     await Promise.all(toDrop.map(table => client.dropTable(table)));
   }
 }
@@ -434,7 +463,7 @@ class PreAggregations {
         p,
         preAggregationsTablesToTempTables,
         loadCache,
-        { waitForRenew: queryBody.renewQuery }
+        { waitForRenew: queryBody.renewQuery, requestId: queryBody.requestId }
       );
       const preAggregationPromise = () => loader.loadPreAggregation().then(async targetTableName => {
         const usedPreAggregation = typeof targetTableName === 'string' ? { targetTableName } : targetTableName;
@@ -448,7 +477,9 @@ class PreAggregations {
   getQueue() {
     if (!this.queue) {
       this.queue = QueryCache.createQueue(`SQL_PRE_AGGREGATIONS_${this.redisPrefix}`, this.driverFactory, (client, q) => {
-        const { preAggregation, preAggregationsTablesToTempTables, newVersionEntry } = q;
+        const {
+          preAggregation, preAggregationsTablesToTempTables, newVersionEntry, requestId
+        } = q;
         const loader = new PreAggregationLoader(
           this.redisPrefix,
           this.driverFactory,
@@ -457,7 +488,8 @@ class PreAggregations {
           this,
           preAggregation,
           preAggregationsTablesToTempTables,
-          new PreAggregationLoadCache(this.redisPrefix, this.driverFactory, this.queryCache, this)
+          new PreAggregationLoadCache(this.redisPrefix, this.driverFactory, this.queryCache, this),
+          { requestId }
         );
         return loader.refresh(newVersionEntry)(client);
       }, {