fix(root): add hyperdx and better logging

packages/common/src/id.ts

@@ -3,3 +3,7 @@ import { nanoid } from 'nanoid/non-secure';
 export function shortId() {
   return nanoid(4);
 }
+
+export function generateId() {
+  return nanoid(8);
+}

packages/db/package.json

@@ -19,6 +19,7 @@
     "@clickhouse/client": "^1.2.0",
     "@openpanel/common": "workspace:*",
     "@openpanel/constants": "workspace:*",
+    "@openpanel/logger": "workspace:*",
     "@openpanel/redis": "workspace:*",
     "@openpanel/validation": "workspace:*",
     "@prisma/client": "^5.1.1",

packages/db/src/buffers/buffer.ts

@@ -1,12 +1,6 @@
+import { createLogger } from '@openpanel/logger';
 import { getRedisCache } from '@openpanel/redis';
 
-const logger = {
-  debug: (...args: unknown[]) => console.log('[DEBUG]', ...args),
-  info: (...args: unknown[]) => console.log('[INFO]', ...args),
-  warn: (...args: unknown[]) => console.log('[WARN]', ...args),
-  error: (...args: unknown[]) => console.log('[ERROR]', ...args),
-};
-
 export const DELETE = '__DELETE__';
 
 export type QueueItem<T> = {
@@ -47,6 +41,7 @@ export abstract class RedisBuffer<T> {
   public prefix = 'op:buffer';
   public table: string;
   public batchSize?: number;
+  public logger: ReturnType<typeof createLogger>;
 
   // abstract methods
   public abstract onInsert?: OnInsert<T>;
@@ -58,6 +53,9 @@ export abstract class RedisBuffer<T> {
   constructor(options: { table: string; batchSize?: number }) {
     this.table = options.table;
     this.batchSize = options.batchSize;
+    this.logger = createLogger({ name: `buffer` }).child({
+      table: this.table,
+    });
   }
 
   public getKey(name?: string) {
@@ -73,12 +71,12 @@ export abstract class RedisBuffer<T> {
     await getRedisCache().rpush(this.getKey(), JSON.stringify(value));
 
     const length = await getRedisCache().llen(this.getKey());
-    logger.debug(
+    this.logger.debug(
       `Inserted item into buffer ${this.table}. Current length: ${length}`
     );
 
     if (this.batchSize && length >= this.batchSize) {
-      logger.info(
+      this.logger.info(
         `Buffer ${this.table} reached batch size (${this.batchSize}). Flushing...`
       );
       this.flush();
@@ -90,11 +88,13 @@ export abstract class RedisBuffer<T> {
       const queue = await this.getQueue(this.batchSize || -1);
 
       if (queue.length === 0) {
-        logger.debug(`Flush called on empty buffer ${this.table}`);
+        this.logger.debug(`Flush called on empty buffer ${this.table}`);
         return { count: 0, data: [] };
       }
 
-      logger.info(`Flushing ${queue.length} items from buffer ${this.table}`);
+      this.logger.info(
+        `Flushing ${queue.length} items from buffer ${this.table}`
+      );
 
       try {
         const indexes = await this.processQueue(queue);
@@ -105,18 +105,18 @@ export abstract class RedisBuffer<T> {
 
         if (this.onCompleted) {
           const res = await this.onCompleted(data);
-          logger.info(
+          this.logger.info(
             `Completed processing ${res.length} items from buffer ${this.table}`
           );
           return { count: res.length, data: res };
         }
 
-        logger.info(
+        this.logger.info(
           `Processed ${indexes.length} items from buffer ${this.table}`
         );
         return { count: indexes.length, data: indexes };
       } catch (e) {
-        logger.error(
+        this.logger.error(
           `Failed to process queue while flushing buffer ${this.table}:`,
           e
         );
@@ -126,12 +126,12 @@ export abstract class RedisBuffer<T> {
           data: JSON.stringify(queue.map((item) => item.event)),
           retries: 0,
         });
-        logger.warn(
+        this.logger.warn(
           `Stored ${queue.length} failed items in ${this.getKey(`failed:${timestamp}`)}`
         );
       }
     } catch (e) {
-      logger.error(
+      this.logger.error(
         `Failed to get queue while flushing buffer ${this.table}:`,
         e
       );
@@ -145,7 +145,9 @@ export abstract class RedisBuffer<T> {
     });
     multi.lrem(this.getKey(), 0, DELETE);
     await multi.exec();
-    logger.debug(`Deleted ${indexes.length} items from buffer ${this.table}`);
+    this.logger.debug(
+      `Deleted ${indexes.length} items from buffer ${this.table}`
+    );
   }
 
   public async getQueue(limit: number): Promise<QueueItem<T>[]> {
@@ -156,7 +158,9 @@ export abstract class RedisBuffer<T> {
         index,
       }))
       .filter((item): item is QueueItem<T> => item.event !== null);
-    logger.debug(`Retrieved ${result.length} items from buffer ${this.table}`);
+    this.logger.debug(
+      `Retrieved ${result.length} items from buffer ${this.table}`
+    );
     return result;
   }
 
@@ -164,7 +168,7 @@ export abstract class RedisBuffer<T> {
     try {
       return JSON.parse(item);
     } catch (e) {
-      logger.warn(`Failed to parse item in buffer ${this.table}:`, e);
+      this.logger.warn(`Failed to parse item in buffer ${this.table}:`, e);
       return null;
     }
   }

packages/db/src/buffers/event-buffer.ts

@@ -1,4 +1,4 @@
-import { groupBy, omit, pick } from 'ramda';
+import { groupBy, omit } from 'ramda';
 import SuperJSON from 'superjson';
 
 import { deepMergeObjects } from '@openpanel/common';

packages/db/src/clickhouse-client.ts

@@ -2,8 +2,11 @@ import type { ResponseJSON } from '@clickhouse/client';
 import { createClient } from '@clickhouse/client';
 import { escape } from 'sqlstring';
 
+import { createLogger } from '@openpanel/logger';
 import type { IInterval } from '@openpanel/validation';
 
+const logger = createLogger({ name: 'clickhouse' });
+
 export const TABLE_NAMES = {
   events: 'events_v2',
   profiles: 'profiles',
@@ -33,6 +36,10 @@ export const ch = new Proxy(originalCh, {
   get(target, property, receiver) {
     if (property === 'insert' || property === 'query') {
       return async (...args: any[]) => {
+        const childLogger = logger.child({
+          query: args[0].query,
+          property,
+        });
         try {
           // First attempt
           if (property in target) {
@@ -46,27 +53,26 @@ export const ch = new Proxy(originalCh, {
               error.message.includes('socket hang up') ||
               error.message.includes('Timeout error'))
           ) {
-            console.info(
-              `Caught ${error.message} error on ${property.toString()}, retrying once.`
-            );
+            childLogger.error(`Captured error`, {
+              error,
+            });
             await new Promise((resolve) => setTimeout(resolve, 500));
             try {
               // Retry once
+              childLogger.info(`Retrying query`);
               if (property in target) {
                 // @ts-expect-error
                 return await target[property](...args);
               }
             } catch (retryError) {
-              console.error(
-                `Retry failed for ${property.toString()}:`,
-                retryError
-              );
+              logger.error(`Retry failed`, retryError);
               throw retryError; // Rethrow or handle as needed
             }
           } else {
-            if (args[0].query) {
-              console.log('FAILED QUERY:', args[0].query);
-            }
+            logger.error('query failed', {
+              ...args[0],
+              error,
+            });
 
             // Handle other errors or rethrow them
             throw error;
@@ -103,10 +109,12 @@ export async function chQueryWithMeta<T extends Record<string, any>>(
     }),
   };
 
-  console.log(
-    `Query: (${Date.now() - start}ms, ${response.statistics?.elapsed}ms), Rows: ${json.rows}`,
-    query
-  );
+  logger.info('query info', {
+    query,
+    rows: json.rows,
+    stats: response.statistics,
+    elapsed: Date.now() - start,
+  });
 
   return response;
 }

packages/logger/index.ts

@@ -1,19 +1,45 @@
-import type { TransportTargetOptions } from 'pino';
-import pino from 'pino';
+import * as HyperDX from '@hyperdx/node-opentelemetry';
+import winston from 'winston';
 
-export function createLogger({ dataset }: { dataset: string }) {
-  const targets: TransportTargetOptions[] =
-    process.env.NODE_ENV === 'production' && process.env.BASELIME_API_KEY
-      ? []
-      : [
-          {
-            target: 'pino-pretty',
-          },
-        ];
+export { winston };
 
-  const transport = pino.transport({
-    targets,
+export type ILogger = winston.Logger & {
+  noop: (message: string) => (error: unknown) => void;
+};
+
+const logLevel = process.env.LOG_LEVEL ?? 'info';
+
+export function createLogger({ name }: { name: string }): ILogger {
+  const service = `${name}-${process.env.NODE_ENV ?? 'dev'}`;
+
+  const format = process.env.HYPERDX_API_KEY
+    ? winston.format.json()
+    : winston.format.prettyPrint();
+
+  const transports: winston.transport[] = [new winston.transports.Console()];
+  if (process.env.HYPERDX_API_KEY) {
+    transports.push(
+      HyperDX.getWinstonTransport(logLevel, {
+        detectResources: true,
+        service,
+      })
+    );
+  }
+
+  const logger = winston.createLogger({
+    defaultMeta: { service },
+    level: logLevel,
+    format,
+    transports,
+    // Add ISO levels of logging from PINO
+    levels: Object.assign(
+      { fatal: 0, warn: 4, trace: 7 },
+      winston.config.syslog.levels
+    ),
   });
 
-  return pino(transport);
+  return Object.assign(logger, {
+    noop: (message: string) => (error: unknown) =>
+      logger.error(`noop: ${message}`, error),
+  });
 }

packages/logger/package.json

@@ -8,8 +8,8 @@
     "typecheck": "tsc --noEmit"
   },
   "dependencies": {
-    "pino": "^8.17.2",
-    "pino-pretty": "^10.3.1"
+    "@hyperdx/node-opentelemetry": "^0.8.1",
+    "winston": "^3.14.2"
   },
   "devDependencies": {
     "@openpanel/eslint-config": "workspace:*",
@@ -28,4 +28,4 @@
     ]
   },
   "prettier": "@openpanel/prettier-config"
-}
+}