fix: Report more accurate time to APM on heavy used deployments (#9667)

The idea behind this change: 

When an async function is called in an async function with await, the real execution can be delayed to microtasking in Node.js. It results in incorrect timestamps, which I see in APM.

Author: ovr

packages/cubejs-backend-shared/src/track.ts

@@ -5,14 +5,17 @@ import { internalExceptions } from './errors';
 
 export type BaseEvent = {
   event: string,
+  // It's possible to fill timestamp at the place of logging, otherwise, it will be filled in automatically
+  timestamp?: string,
   [key: string]: any,
 };
 
-export type Event = BaseEvent & {
+export type Event = {
   id: string,
   clientTimestamp: string,
   anonymousId: string,
   platform: string,
+  arch: string,
   nodeVersion: string,
   sentFrom: 'backend';
 };
@@ -79,8 +82,8 @@ export async function track(opts: BaseEvent) {
 
   trackEvents.push({
     ...opts,
+    clientTimestamp: opts.timestamp || new Date().toJSON(),
     id: crypto.randomBytes(16).toString('hex'),
-    clientTimestamp: new Date().toJSON(),
     platform: process.platform,
     arch: process.arch,
     nodeVersion: process.version,

packages/cubejs-server-core/src/core/DevServer.ts

@@ -11,7 +11,7 @@ import jwt from 'jsonwebtoken';
 import isDocker from 'is-docker';
 import type { Application as ExpressApplication, Request, Response } from 'express';
 import type { ChildProcess } from 'child_process';
-import { executeCommand, getEnv, keyByDataSource, packageExists } from '@cubejs-backend/shared';
+import { executeCommand, getAnonymousId, getEnv, keyByDataSource, packageExists } from '@cubejs-backend/shared';
 import crypto from 'crypto';
 
 import type { BaseDriver } from '@cubejs-backend/query-orchestrator';
@@ -115,7 +115,7 @@ export class DevServer {
       res.json({
         cubejsToken,
         basePath: options.basePath,
-        anonymousId: this.cubejsServer.anonymousId,
+        anonymousId: getAnonymousId(),
         coreServerVersion: this.cubejsServer.coreServerVersion,
         dockerVersion: this.options.dockerVersion || null,
         projectFingerprint: this.cubejsServer.projectFingerprint,

packages/cubejs-server-core/src/core/agentCollect.ts

@@ -8,6 +8,7 @@ import crypto from 'crypto';
 import WebSocket from 'ws';
 import zlib from 'zlib';
 import { promisify } from 'util';
+import { LoggerFnParams, LoggerFn } from './types';
 
 const deflate = promisify(zlib.deflate);
 interface AgentTransport {
@@ -175,11 +176,11 @@ const clearTransport = () => {
   agentInterval = null;
 };
 
-export default async (event: Record<string, any>, endpointUrl: string, logger: any) => {
+export const agentCollect = async (event: LoggerFnParams, endpointUrl: string, logger: LoggerFn) => {
   trackEvents.push({
+    timestamp: new Date().toJSON(),
     ...event,
     id: crypto.randomBytes(16).toString('hex'),
-    timestamp: new Date().toJSON(),
     instanceId: getEnv('instanceId'),
   });
   lastEvent = new Date();
@@ -227,3 +228,5 @@ export default async (event: Record<string, any>, endpointUrl: string, logger: a
     }, getEnv('agentFlushInterval'));
   }
 };
+
+export default agentCollect;

packages/cubejs-server-core/src/core/server.ts

@@ -14,7 +14,6 @@ import {
   CancelableInterval,
   createCancelableInterval,
   formatDuration,
-  getAnonymousId,
   getEnv,
   assertDataSource,
   getRealType,
@@ -31,7 +30,7 @@ import { RefreshScheduler, ScheduledRefreshOptions } from './RefreshScheduler';
 import { OrchestratorApi, OrchestratorApiOptions } from './OrchestratorApi';
 import { CompilerApi } from './CompilerApi';
 import { DevServer } from './DevServer';
-import agentCollect from './agentCollect';
+import { agentCollect } from './agentCollect';
 import { OrchestratorStorage } from './OrchestratorStorage';
 import { prodLogger, devLogger } from './logger';
 import { OptsHandler } from './OptsHandler';
@@ -60,6 +59,7 @@ import type {
   DriverConfig,
   ScheduledRefreshTimeZonesFn,
   ContextToCubeStoreRouterIdFn,
+  LoggerFnParams,
 } from './types';
 import {
   ContextToOrchestratorIdFn,
@@ -171,8 +171,6 @@ export class CubejsServerCore {
 
   public projectFingerprint: string | null = null;
 
-  public anonymousId: string | null = null;
-
   public coreServerVersion: string | null = null;
 
   protected contextAcceptor: ContextAcceptor;
@@ -233,7 +231,7 @@ export class CubejsServerCore {
 
     this.startScheduledRefreshTimer();
 
-    this.event = async (name, props) => {
+    this.event = async (event, props: LoggerFnParams) => {
       if (!this.options.telemetry) {
         return;
       }
@@ -248,15 +246,12 @@ export class CubejsServerCore {
         }
       }
 
-      if (!this.anonymousId) {
-        this.anonymousId = getAnonymousId();
-      }
-
       const internalExceptionsEnv = getEnv('internalExceptions');
 
       try {
         await track({
-          event: name,
+          timestamp: new Date().toJSON(),
+          event,
           projectFingerprint: this.projectFingerprint,
           coreServerVersion: this.coreServerVersion,
           dockerVersion: getEnv('dockerImageVersion'),
@@ -410,7 +405,12 @@ export class CubejsServerCore {
     if (agentEndpointUrl) {
       const oldLogger = this.logger;
       this.preAgentLogger = oldLogger;
+
       this.logger = (msg, params) => {
+        // Filling timestamp as much as earlier as we can, otherwise it can be incorrect. Because next code is async
+        // with await points which can be delayed with Node.js micro-tasking.
+        params.timestamp = params.timestamp || new Date().toJSON();
+
         oldLogger(msg, params);
         agentCollect(
           {

packages/cubejs-server-core/src/core/types.ts

@@ -167,7 +167,12 @@ export type ExternalDbTypeFn = (context: RequestContext) => DatabaseType;
 export type ExternalDriverFactoryFn = (context: RequestContext) => Promise<BaseDriver> | BaseDriver;
 export type ExternalDialectFactoryFn = (context: RequestContext) => BaseQuery;
 
-export type LoggerFn = (msg: string, params: Record<string, any>) => void;
+export type LoggerFnParams = {
+  // It's possible to fill timestamp at the place of logging, otherwise, it will be filled in automatically
+  timestamp?: string,
+  [key: string]: any,
+};
+export type LoggerFn = (msg: string, params: LoggerFnParams) => void;
 
 export type BiToolSyncConfig = {
   type: string;

rust/cubesql/cubesql/src/sql/compiler_cache.rs

@@ -191,7 +191,8 @@ impl CompilerCache for CompilerCacheImpl {
                 .get(&(compiler_id, protocol.clone()))
                 .cloned()
         };
-        // Double checked locking
+
+        // Double-checked locking
         let cache_entry = if let Some(cache_entry) = cache_entry {
             cache_entry
         } else {