feat: Introducing Logging Class (Disabled Usage of Logger)

postgres.Dockerfile

@@ -0,0 +1,4 @@
+FROM postgres:14
+RUN apt-get update && apt-get install -y postgresql-14-cron
+EXPOSE 5432
+CMD ["postgres", "-c", "shared_preload_libraries=pg_cron"]

docker-compose.yml

@@ -78,6 +78,8 @@ services:
       - "6379:6379"
 
   postgres:
+    build:
+      dockerfile: ./postgres.Dockerfile
     image: postgres:14
     restart: always
     volumes:
@@ -160,4 +162,4 @@ services:
 volumes:
   postgres:
   redis:
-  grafana:
+  grafana:

runner/src/hasura-client/hasura-client.test.ts

@@ -72,7 +72,7 @@ describe('HasuraClient', () => {
       });
     const client = new HasuraClient({ fetch: mockFetch as unknown as typeof fetch }, config);
 
-    await client.runMigrations('dbName', 'schemaName', 'CREATE TABLE blocks (height numeric)');
+    await client.runSql('dbName', 'schemaName', 'CREATE TABLE blocks (height numeric)');
 
     expect(mockFetch.mock.calls).toMatchSnapshot();
   });

runner/src/hasura-client/hasura-client.ts

@@ -155,11 +155,11 @@ export default class HasuraClient {
     });
   }
 
-  async runMigrations (source: string, schemaName: string, migration: string): Promise<any> {
+  async runSql (source: string, schemaName: string, sqlScript: string): Promise<any> {
     return await this.executeSql(
       `
       set schema '${schemaName}';
-      ${migration}
+      ${sqlScript}
       `,
       { source, readOnly: false }
     );
@@ -172,7 +172,6 @@ export default class HasuraClient {
         source,
       }
     );
-
     return tablesInSource
       .filter(({ schema }: { schema: string }) => schema === schemaName)
       .map(({ name }: { name: string }) => name);

runner/src/indexer-logger/indexer-logger.test.ts

@@ -0,0 +1,116 @@
+import pgFormat from 'pg-format';
+import IndexerLogger from './indexer-logger';
+import type PgClient from '../pg-client';
+import { LogLevel } from '../stream-handler/stream-handler';
+import { LogType, type LogEntry } from './indexer-logger';
+
+describe('IndexerLogger', () => {
+  let pgClient: PgClient;
+  let query: jest.Mock;
+
+  beforeEach(() => {
+    query = jest.fn().mockReturnValue({ rows: [] });
+    pgClient = {
+      query,
+      format: pgFormat
+    } as unknown as PgClient;
+  });
+
+  const mockDatabaseConnectionParameters = {
+    username: 'test_user',
+    password: 'test_password',
+    host: 'test_host',
+    port: 5432,
+    database: 'test_database'
+  };
+  const functionName = 'testFunction';
+
+  describe('writeLog', () => {
+    it('should insert a single log entry into the database', async () => {
+      const indexerLogger = new IndexerLogger(functionName, 5, mockDatabaseConnectionParameters, pgClient);
+      const logEntry: LogEntry = {
+        blockHeight: 123,
+        logTimestamp: new Date(),
+        logType: LogType.SYSTEM,
+        logLevel: LogLevel.INFO,
+        message: 'Test log message'
+      };
+
+      await indexerLogger.writeLogs(logEntry);
+
+      const expectedQueryStructure = `INSERT INTO "${functionName}".__logs (block_height, date, timestamp, type, level, message) VALUES`;
+      expect(query.mock.calls[0][0]).toContain(expectedQueryStructure);
+    });
+
+    it('should handle errors when inserting a single log entry', async () => {
+      query.mockRejectedValueOnce(new Error('Failed to insert log'));
+
+      const indexerLogger = new IndexerLogger(functionName, 5, mockDatabaseConnectionParameters, pgClient);
+      const logEntry: LogEntry = {
+        blockHeight: 123,
+        logTimestamp: new Date(),
+        logType: LogType.SYSTEM,
+        logLevel: LogLevel.INFO,
+        message: 'Test log message'
+      };
+
+      await expect(indexerLogger.writeLogs(logEntry)).rejects.toThrow('Failed to insert log');
+    });
+
+    it('should insert a batch of log entries into the database', async () => {
+      const indexerLogger = new IndexerLogger(functionName, 5, mockDatabaseConnectionParameters, pgClient);
+      const logEntries: LogEntry[] = [
+        {
+          blockHeight: 123,
+          logTimestamp: new Date(),
+          logType: LogType.SYSTEM,
+          logLevel: LogLevel.INFO,
+          message: 'Test log message 1'
+        },
+        {
+          blockHeight: 124,
+          logTimestamp: new Date(),
+          logType: LogType.SYSTEM,
+          logLevel: LogLevel.INFO,
+          message: 'Test log message 2'
+        }
+      ];
+
+      await indexerLogger.writeLogs(logEntries);
+
+      const expectedQuery = `INSERT INTO "${functionName}".__logs (block_height, date, timestamp, type, level, message) VALUES`;
+      expect(query.mock.calls[0][0]).toContain(expectedQuery);
+    });
+
+    it('should handle errors when inserting a batch of log entries', async () => {
+      query.mockRejectedValueOnce(new Error('Failed to insert batch of logs'));
+
+      const indexerLogger = new IndexerLogger(functionName, 5, mockDatabaseConnectionParameters, pgClient);
+      const logEntries: LogEntry[] = [
+        {
+          blockHeight: 123,
+          logTimestamp: new Date(),
+          logType: LogType.SYSTEM,
+          logLevel: LogLevel.INFO,
+          message: 'Test log message 1'
+        },
+        {
+          blockHeight: 124,
+          logTimestamp: new Date(),
+          logType: LogType.SYSTEM,
+          logLevel: LogLevel.INFO,
+          message: 'Test log message 2'
+        }
+      ];
+
+      await expect(indexerLogger.writeLogs(logEntries)).rejects.toThrow('Failed to insert batch of logs');
+    });
+
+    it('should handle empty log entry', async () => {
+      const indexerLogger = new IndexerLogger(functionName, 5, mockDatabaseConnectionParameters, pgClient);
+      const logEntries: LogEntry[] = [];
+      await indexerLogger.writeLogs(logEntries);
+      expect(query).not.toHaveBeenCalled();
+    });
+  });
+});

runner/src/indexer-logger/indexer-logger.ts

@@ -0,0 +1,77 @@
+import format from 'pg-format';
+import { wrapError } from '../utility';
+import PgClient from '../pg-client';
+import { type DatabaseConnectionParameters } from '../provisioner/provisioner';
+import { LogLevel } from '../stream-handler/stream-handler';
+import { trace } from '@opentelemetry/api';
+
+export interface LogEntry {
+  blockHeight: number
+  logTimestamp: Date
+  logType: string
+  logLevel: LogLevel
+  message: string
+}
+
+export enum LogType {
+  SYSTEM = 'system',
+  USER = 'user',
+}
+export default class IndexerLogger {
+  tracer = trace.getTracer('queryapi-runner-indexer-logger');
+
+  private readonly pgClient: PgClient;
+  private readonly schemaName: string;
+  private readonly logInsertQueryTemplate: string = 'INSERT INTO %I.__logs (block_height, date, timestamp, type, level, message) VALUES %L';
+  private readonly loggingLevel: number;
+
+  constructor (
+    functionName: string,
+    loggingLevel: number,
+    databaseConnectionParameters: DatabaseConnectionParameters,
+    pgClientInstance: PgClient | undefined = undefined
+  ) {
+    const pgClient = pgClientInstance ?? new PgClient({
+      user: databaseConnectionParameters.username,
+      password: databaseConnectionParameters.password,
+      host: process.env.PGHOST,
+      port: Number(databaseConnectionParameters.port),
+      database: databaseConnectionParameters.database,
+    });
+
+    this.pgClient = pgClient;
+    this.schemaName = functionName.replace(/[^a-zA-Z0-9]/g, '_');
+    this.loggingLevel = loggingLevel;
+  }
+
+  private shouldLog (logLevel: LogLevel): boolean {
+    return logLevel >= this.loggingLevel;
+  }
+
+  async writeLogs (
+    logEntries: LogEntry | LogEntry[],
+  ): Promise<void> {
+    const entriesArray = (Array.isArray(logEntries) ? logEntries : [logEntries]).filter(entry => this.shouldLog(entry.logLevel)); ;
+    if (entriesArray.length === 0) return;
+
+    const spanMessage = `call writeLog function of IndexerLogger for ${entriesArray.length === 1 ? 'single entry' : `batch of ${entriesArray.length}`}`;
+    const writeLogSpan = this.tracer.startSpan(spanMessage);
+
+    await wrapError(async () => {
+      const values = entriesArray.map(entry => [
+        entry.blockHeight,
+        entry.logTimestamp,
+        entry.logTimestamp,
+        entry.logType,
+        LogLevel[entry.logLevel],
+        entry.message
+      ]);
+
+      const query = format(this.logInsertQueryTemplate, this.schemaName, values);
+      await this.pgClient.query(query);
+    }, `Failed to insert ${entriesArray.length > 1 ? 'logs' : 'log'} into the ${this.schemaName}.__logs table`)
+      .finally(() => {
+        writeLogSpan.end();
+      });
+  }
+}

runner/src/indexer/__snapshots__/indexer.test.ts.snap

@@ -534,4 +534,4 @@ exports[`Indexer unit tests Indexer.runFunctions() supplies the required role to
     },
   ],
 ]
-`;
+`;