Skip to content

Commit 8d297ee

Browse files
RusovDmitriyRusovDmitriy
authored
chore(query-orchestrator): Added metadata to cache and queue logs (#6751)
1 parent c609048 commit 8d297ee

File tree

4 files changed

+53
-17
lines changed

4 files changed

+53
-17
lines changed

‎packages/cubejs-base-driver/src/queue-driver.interface.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,7 @@ export interface AddToQueueOptions {
4242
stageQueryKey: string,
4343
requestId: string,
4444
orphanedTimeout?: number,
45+
queueId?: QueueId,
4546
}
4647

4748
export interface QueueDriverOptions {

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

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -109,6 +109,7 @@ export class LocalQueueDriverConnection {
109109
*/
110110
addToQueue(keyScore, queryKey, orphanedTime, queryHandler, query, priority, options) {
111111
const queryQueueObj = {
112+
queueId: options.queueId,
112113
queryHandler,
113114
query,
114115
queryKey,
@@ -142,8 +143,7 @@ export class LocalQueueDriverConnection {
142143

143144
return [
144145
added,
145-
// this driver doesnt support queue id
146-
null,
146+
queryQueueObj.queueId,
147147
Object.keys(this.toProcess).length,
148148
queryQueueObj.addedToQueueTime
149149
];
@@ -271,8 +271,7 @@ export class LocalQueueDriverConnection {
271271

272272
return [
273273
added,
274-
// this driver doesnt support queue id
275-
null,
274+
this.queryDef[key]?.queueId,
276275
this.queueArray(this.active),
277276
Object.keys(this.toProcess).length,
278277
this.queryDef[key],

‎packages/cubejs-query-orchestrator/src/orchestrator/QueryCache.ts‎

Lines changed: 20 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
1+
import crypto from 'crypto';
12
import csvWriter from 'csv-write-stream';
23
import LRUCache from 'lru-cache';
34
import { pipeline } from 'stream';
@@ -426,6 +427,7 @@ export class QueryCache {
426427
external,
427428
priority,
428429
requestId,
430+
spanId,
429431
inlineTables,
430432
useCsvQuery,
431433
lambdaTypes,
@@ -437,6 +439,7 @@ export class QueryCache {
437439
external: boolean,
438440
priority?: number,
439441
requestId?: string,
442+
spanId?: string,
440443
inlineTables?: InlineTables,
441444
useCsvQuery?: boolean,
442445
lambdaTypes?: TableStructure,
@@ -461,6 +464,7 @@ export class QueryCache {
461464
const opt = {
462465
stageQueryKey: cacheKey,
463466
requestId,
467+
spanId,
464468
};
465469

466470
if (!persistent) {
@@ -834,6 +838,7 @@ export class QueryCache {
834838
persistent?: boolean,
835839
}
836840
) {
841+
const spanId = crypto.randomBytes(16).toString('hex');
837842
options = options || { dataSource: 'default' };
838843
const { renewalThreshold } = options;
839844
const renewalKey = options.renewalKey && this.queryRedisKey(options.renewalKey);
@@ -844,6 +849,7 @@ export class QueryCache {
844849
priority: options.priority,
845850
external: options.external,
846851
requestId: options.requestId,
852+
spanId,
847853
persistent: options.persistent,
848854
dataSource: options.dataSource,
849855
useCsvQuery: options.useCsvQuery,
@@ -858,21 +864,23 @@ export class QueryCache {
858864
.cacheDriver
859865
.set(redisKey, result, expiration)
860866
.then(({ bytes }) => {
861-
this.logger('Renewed', { cacheKey, requestId: options.requestId });
867+
this.logger('Renewed', { cacheKey, requestId: options.requestId, spanId });
862868
this.logger('Outgoing network usage', {
863869
service: 'cache',
864870
requestId: options.requestId,
871+
spanId,
865872
bytes,
866873
cacheKey,
867874
});
868875
return res;
869876
});
870877
}).catch(e => {
871878
if (!(e instanceof ContinueWaitError)) {
872-
this.logger('Dropping Cache', { cacheKey, error: e.stack || e, requestId: options.requestId });
879+
this.logger('Dropping Cache', { cacheKey, error: e.stack || e, requestId: options.requestId, spanId });
873880
this.cacheDriver.remove(redisKey)
874881
.catch(err => this.logger('Error removing key', {
875882
cacheKey,
883+
spanId,
876884
error: err.stack || err,
877885
requestId: options.requestId
878886
}));
@@ -882,7 +890,7 @@ export class QueryCache {
882890
);
883891

884892
if (options.forceNoCache) {
885-
this.logger('Force no cache for', { cacheKey, requestId: options.requestId });
893+
this.logger('Force no cache for', { cacheKey, requestId: options.requestId, spanId });
886894
return fetchNew();
887895
}
888896

@@ -914,7 +922,8 @@ export class QueryCache {
914922
renewalKey: inMemoryValue.renewalKey,
915923
newRenewalKey: renewalKey,
916924
renewalThreshold,
917-
requestId: options.requestId
925+
requestId: options.requestId,
926+
spanId,
918927
});
919928
res = inMemoryValue;
920929
}
@@ -935,7 +944,8 @@ export class QueryCache {
935944
renewalKey: parsedResult.renewalKey,
936945
newRenewalKey: renewalKey,
937946
renewalThreshold,
938-
requestId: options.requestId
947+
requestId: options.requestId,
948+
spanId,
939949
});
940950
if (
941951
renewalKey && (
@@ -946,24 +956,24 @@ export class QueryCache {
946956
)
947957
) {
948958
if (options.waitForRenew) {
949-
this.logger('Waiting for renew', { cacheKey, renewalThreshold, requestId: options.requestId });
959+
this.logger('Waiting for renew', { cacheKey, renewalThreshold, requestId: options.requestId, spanId });
950960
return fetchNew();
951961
} else {
952-
this.logger('Renewing existing key', { cacheKey, renewalThreshold, requestId: options.requestId });
962+
this.logger('Renewing existing key', { cacheKey, renewalThreshold, requestId: options.requestId, spanId });
953963
fetchNew().catch(e => {
954964
if (!(e instanceof ContinueWaitError)) {
955-
this.logger('Error renewing', { cacheKey, error: e.stack || e, requestId: options.requestId });
965+
this.logger('Error renewing', { cacheKey, error: e.stack || e, requestId: options.requestId, spanId });
956966
}
957967
});
958968
}
959969
}
960-
this.logger('Using cache for', { cacheKey, requestId: options.requestId });
970+
this.logger('Using cache for', { cacheKey, requestId: options.requestId, spanId });
961971
if (options.useInMemory && renewedAgo + inMemoryCacheDisablePeriod <= renewalThreshold * 1000) {
962972
this.memoryCache.set(redisKey, parsedResult);
963973
}
964974
return parsedResult.result;
965975
} else {
966-
this.logger('Missing cache for', { cacheKey, requestId: options.requestId });
976+
this.logger('Missing cache for', { cacheKey, requestId: options.requestId, spanId });
967977
return fetchNew();
968978
}
969979
}

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

Lines changed: 29 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -167,6 +167,12 @@ export class QueryQueue {
167167
return stream;
168168
}
169169

170+
counter = 0;
171+
172+
generateQueueId() {
173+
return this.counter++;
174+
}
175+
170176
/**
171177
* Push query to the queue and call `QueryQueue.reconcileQueue()` method if
172178
* `options.skipQueue` is set to `false`, execute query skipping queue
@@ -189,6 +195,7 @@ export class QueryQueue {
189195
options,
190196
) {
191197
options = options || {};
198+
options.queueId = this.generateQueueId();
192199
if (this.skipQueue) {
193200
const queryDef = {
194201
queryHandler,
@@ -200,6 +207,8 @@ export class QueryQueue {
200207
addedToQueueTime: new Date().getTime(),
201208
};
202209
this.logger('Waiting for query', {
210+
queueId: options.queueId,
211+
spanId: options.spanId,
203212
queueSize: 0,
204213
queryKey: queryDef.queryKey,
205214
queuePrefix: this.redisQueuePrefix,
@@ -209,7 +218,7 @@ export class QueryQueue {
209218
if (queryHandler === 'stream') {
210219
throw new Error('Streaming queries to Cube Store aren\'t supported');
211220
}
212-
const result = await this.processQuerySkipQueue(queryDef);
221+
const result = await this.processQuerySkipQueue(queryDef, options.queueId);
213222
return this.parseResult(result);
214223
}
215224

@@ -251,6 +260,8 @@ export class QueryQueue {
251260

252261
if (added > 0) {
253262
this.logger('Added to queue', {
263+
queueId,
264+
spanId: options.spanId,
254265
priority,
255266
queueSize,
256267
queryKey,
@@ -273,6 +284,8 @@ export class QueryQueue {
273284

274285
if (queryDef) {
275286
this.logger('Waiting for query', {
287+
queueId,
288+
spanId: options.spanId,
276289
queueSize,
277290
queryKey: queryDef.queryKey,
278291
queuePrefix: this.redisQueuePrefix,
@@ -614,9 +627,10 @@ export class QueryQueue {
614627
* @param {*} query
615628
* @returns {Promise<{ result: undefined | Object, error: string | undefined }>}
616629
*/
617-
async processQuerySkipQueue(query) {
630+
async processQuerySkipQueue(query, queueId) {
618631
const startQueryTime = (new Date()).getTime();
619632
this.logger('Performing query', {
633+
queueId,
620634
queueSize: 0,
621635
queryKey: query.queryKey,
622636
queuePrefix: this.redisQueuePrefix,
@@ -639,6 +653,7 @@ export class QueryQueue {
639653
)
640654
};
641655
this.logger('Performing query completed', {
656+
queueId,
642657
queueSize: 0,
643658
duration: ((new Date()).getTime() - startQueryTime),
644659
queryKey: query.queryKey,
@@ -651,6 +666,7 @@ export class QueryQueue {
651666
error: (e.message || e).toString() // TODO error handling
652667
};
653668
this.logger('Error while querying', {
669+
queueId,
654670
queueSize: 0,
655671
duration: ((new Date()).getTime() - startQueryTime),
656672
queryKey: query.queryKey,
@@ -662,9 +678,10 @@ export class QueryQueue {
662678
if (e instanceof TimeoutError) {
663679
if (handler) {
664680
this.logger('Cancelling query due to timeout', {
681+
queueId,
665682
queryKey: query.queryKey,
666683
queuePrefix: this.redisQueuePrefix,
667-
requestId: query.requestId
684+
requestId: query.requestId,
668685
});
669686
await handler(query);
670687
}
@@ -708,6 +725,7 @@ export class QueryQueue {
708725
const startQueryTime = (new Date()).getTime();
709726
const timeInQueue = (new Date()).getTime() - query.addedToQueueTime;
710727
this.logger('Performing query', {
728+
queueId,
711729
processingId,
712730
queueSize,
713731
queryKey: query.queryKey,
@@ -755,6 +773,7 @@ export class QueryQueue {
755773
return queueConnection.optimisticQueryUpdate(queryKeyHashed, { cancelHandler }, processingId, queueId);
756774
} catch (e) {
757775
this.logger('Error while query update', {
776+
queueId,
758777
queryKey: query.queryKey,
759778
error: e.stack || e,
760779
queuePrefix: this.redisQueuePrefix,
@@ -775,6 +794,7 @@ export class QueryQueue {
775794
}
776795

777796
this.logger('Performing query completed', {
797+
queueId,
778798
processingId,
779799
queueSize,
780800
duration: ((new Date()).getTime() - startQueryTime),
@@ -793,6 +813,7 @@ export class QueryQueue {
793813
error: (e.message || e).toString() // TODO error handling
794814
};
795815
this.logger('Error while querying', {
816+
queueId,
796817
processingId,
797818
queueSize,
798819
duration: ((new Date()).getTime() - startQueryTime),
@@ -811,6 +832,7 @@ export class QueryQueue {
811832
const queryWithCancelHandle = await queueConnection.getQueryDef(queryKeyHashed);
812833
if (queryWithCancelHandle) {
813834
this.logger('Cancelling query due to timeout', {
835+
queueId,
814836
processingId,
815837
queryKey: queryWithCancelHandle.queryKey,
816838
queuePrefix: this.redisQueuePrefix,
@@ -830,6 +852,7 @@ export class QueryQueue {
830852

831853
if (!(await queueConnection.setResultAndRemoveQuery(queryKeyHashed, executionResult, processingId, queueId))) {
832854
this.logger('Orphaned execution result', {
855+
queueId,
833856
processingId,
834857
warn: 'Result for query was not set due to processing lock wasn\'t acquired',
835858
queryKey: query.queryKey,
@@ -855,6 +878,7 @@ export class QueryQueue {
855878
// }
856879

857880
this.logger('Skip processing', {
881+
queueId,
858882
processingId,
859883
queryKey: query && query.queryKey || queryKeyHashed,
860884
requestId: query && query.requestId,
@@ -869,6 +893,7 @@ export class QueryQueue {
869893
const currentProcessingId = await queueConnection.freeProcessingLock(queryKeyHashed, processingId, activated);
870894
if (currentProcessingId) {
871895
this.logger('Skipping free processing lock', {
896+
queueId,
872897
processingId,
873898
currentProcessingId,
874899
queryKey: query && query.queryKey || queryKeyHashed,
@@ -885,6 +910,7 @@ export class QueryQueue {
885910
}
886911
} catch (e) {
887912
this.logger('Queue storage error', {
913+
queueId,
888914
queryKey: query && query.queryKey || queryKeyHashed,
889915
requestId: query && query.requestId,
890916
error: (e.stack || e).toString(),

0 commit comments

Comments
 (0)