Feat/logs service

README.md

@@ -62,7 +62,6 @@ Options are configured via environment variables.
 | MESSAGES_PASSWORD              | No       | password              | Password to decrypt encrypted exit messages with. Needed only if you encrypt your exit messages                                                                                                                                                         |
 | MESSAGES_PASSWORD_FILE         | No       | password_inside.txt   | Path to a file with password inside to decrypt exit messages with. Needed only if you have encrypted exit messages. If used, MESSAGES_PASSWORD (not MESSAGES_PASSWORD_FILE) needs to be added to LOGGER_SECRETS in order to be sanitized                |
 | BLOCKS_PRELOAD                 | No       | 50000                 | Amount of blocks to load events from on start. Increase if daemon was not running for some time. Defaults to a week of blocks                                                                                                                           |
-| BLOCKS_LOOP                    | No       | 900                   | Amount of blocks to load events from on every poll. Defaults to 3 hours of blocks                                                                                                                                                                       |
 | JOB_INTERVAL                   | No       | 384000                | Time interval in milliseconds to run checks. Defaults to time of 1 epoch                                                                                                                                                                                |
 | HTTP_PORT                      | No       | 8989                  | Port to serve metrics and health check on                                                                                                                                                                                                               |
 | RUN_METRICS                    | No       | false                 | Enable metrics endpoint                                                                                                                                                                                                                                 |
@@ -129,7 +128,7 @@ Available metrics:
 - exit_messages: ['valid'] - Exit messages and their validity: JSON parseability, structure and signature
 - exit_actions: ['result'] - Statuses of initiated validator exits
 - event_security_verification: ['result'] - Statuses of exit event security verifications
-- polling_last_blocks_duration_seconds: ['eventsNumber'] - Duration of pooling last blocks in microseconds
+- polling_last_blocks_duration_seconds: ['eventsNumber'] - Duration of polling last blocks in microseconds
 - execution_request_duration_seconds: ['result', 'status', 'domain'] - Execution node request duration in microseconds
 - consensus_request_duration_seconds: ['result', 'status', 'domain'] - Consensus node request duration in microseconds
 - job_duration_seconds: ['name', 'interval', 'result'] - Duration of Ejector cycle cron job

package.json

@@ -17,7 +17,7 @@
     "@lodestar/utils": "1.2.2",
     "dotenv": "16.0.3",
     "ethers": "5.7.2",
-    "lido-nanolib": "1.4.0",
+    "lru-cache": "11.1.0",
     "node-fetch": "3.3.0",
     "prom-client": "14.1.0",
     "typescript": "4.9.3"
@@ -39,7 +39,7 @@
     "eslint": "8.29.0",
     "eslint-config-prettier": "8.5.0",
     "eslint-plugin-prettier": "4.2.1",
-    "nock": "^13.3.1",
+    "nock": "13.3.1",
     "prettier": "2.8.0",
     "ts-node": "10.9.1",
     "vite": "6.2.0",

src/app/module.ts

@@ -30,6 +30,7 @@ import { makeGsStore } from '../services/gs-store/service.js'
 import { makeApp } from './service.js'
 import { makeMessageReloader } from '../services/message-reloader/message-reloader.js'
 import { makeForkVersionResolver } from '../services/fork-version-resolver/service.js'
+import { makeExitLogsService } from '../services/exit-logs/service.js'
 
 dotenv.config()
 
@@ -49,18 +50,17 @@ export const makeAppModule = async () => {
 
   const metrics = makeMetrics({ PREFIX: config.PROM_PREFIX })
 
-  const executionApi = makeExecutionApi(
-    makeRequest([
-      retry(3),
-      loggerMiddleware(logger),
-      prom(metrics.executionRequestDurationSeconds),
-      notOkError(),
-      abort(30_000),
-    ]),
-    logger,
-    config,
-    metrics
-  )
+  const executionHttp = makeRequest([
+    retry(3),
+    loggerMiddleware(logger),
+    prom(metrics.executionRequestDurationSeconds),
+    notOkError(),
+    abort(30_000),
+  ])
+
+  const executionApi = makeExecutionApi(executionHttp, logger, config)
+
+  const exitLogs = makeExitLogsService(logger, executionApi, config, metrics)
 
   const consensusApi = makeConsensusApi(
     makeRequest([
@@ -110,6 +110,7 @@ export const makeAppModule = async () => {
     config,
     messageReloader,
     executionApi,
+    exitLogs,
     consensusApi,
     messagesProcessor,
     webhookProcessor,

src/app/service.ts

@@ -1,5 +1,6 @@
 import type { Dependencies } from './interface.js'
 import { MessageStorage } from '../services/job-processor/message-storage.js'
+import { getHeapStatistics } from 'v8'
 
 export const makeApp = ({
   config,
@@ -11,20 +12,22 @@ export const makeApp = ({
   consensusApi,
   appInfoReader,
 }: Dependencies) => {
-  const {
-    OPERATOR_ID,
-    BLOCKS_PRELOAD,
-    BLOCKS_LOOP,
-    JOB_INTERVAL,
-    OPERATOR_IDENTIFIERS,
-  } = config
+  const { OPERATOR_ID, BLOCKS_PRELOAD, JOB_INTERVAL, OPERATOR_IDENTIFIERS } =
+    config
 
   let ejectorCycleTimer: NodeJS.Timer | null = null
 
   const run = async () => {
     const version = await appInfoReader.getVersion()
     const mode = config.MESSAGES_LOCATION ? 'message' : 'webhook'
-    logger.info(`Validator Ejector v${version} started in ${mode} mode`, config)
+
+    const { heap_size_limit } = getHeapStatistics()
+    const heapLimit = Math.round(heap_size_limit / 1024 / 1024).toString()
+
+    logger.info(`Validator Ejector v${version} started in ${mode} mode`, {
+      ...config,
+      heapLimit,
+    })
 
     metrics.buildInfo
       .labels({
@@ -47,19 +50,19 @@ export const makeApp = ({
     )
 
     logger.info(`Loading initial events for ${BLOCKS_PRELOAD} last blocks`)
+    const fetchTimeStart = performance.now()
+
     await job.once({
-      eventsNumber: BLOCKS_PRELOAD,
       messageStorage: messageStorage,
     })
 
-    logger.info(
-      `Starting ${
-        JOB_INTERVAL / 1000
-      } seconds polling for ${BLOCKS_LOOP} last blocks`
-    )
+    const fetchTimeEnd = performance.now()
+    const fetchTime = Math.ceil(fetchTimeEnd - fetchTimeStart) / 1000
+    logger.info(`Initial events loaded in ${fetchTime} seconds`)
+
+    logger.info(`Starting ${JOB_INTERVAL / 1000} seconds polling`)
 
-    ejectorCycleTimer = job.pooling({
-      eventsNumber: BLOCKS_LOOP,
+    ejectorCycleTimer = job.loop({
       messageStorage: messageStorage,
     })
   }

src/lib/job/index.ts

@@ -42,12 +42,12 @@ export const makeJobRunner = <Initial>(
     }
   }
 
-  const pooling = (handlerValue: Initial) => {
+  const loop = (handlerValue: Initial) => {
     return setInterval(() => once(handlerValue), config.JOB_INTERVAL)
   }
 
   return {
     once,
-    pooling,
+    loop,
   }
 }

src/lib/logger/logger.spec.ts

@@ -31,9 +31,44 @@ describe('Logger', () => {
     expect(dateStr).toMatchInlineSnapshot(`"2022-01-01 13:58:22"`)
   })
 
-  test.todo('json output')
+  test('json output', () => {
+    const { restore, log } = mockConsole()
+    const logger = makeLogger({ format: 'json', level: 'info' })
+    const testMessage = 'test message'
+    const testData = { foo: 'bar', num: 42 }
 
-  test.todo('test simple output')
+    logger.info(testMessage, testData)
+
+    expect(log.info).toHaveBeenCalledTimes(1)
+    const loggedJson = JSON.parse(log.info.mock.calls[0][0])
+
+    expect(loggedJson).toHaveProperty('timestamp')
+    expect(loggedJson).toHaveProperty('level', 'info')
+    expect(loggedJson).toHaveProperty('message', testMessage)
+    expect(loggedJson).toHaveProperty('details')
+    expect(loggedJson.details).toEqual(testData)
+
+    restore()
+  })
+
+  test('test simple output', () => {
+    const { restore, log } = mockConsole()
+    const logger = makeLogger({ format: 'simple', level: 'warn' })
+    const testMessage = 'warning message'
+
+    logger.warn(testMessage)
+
+    expect(log.warn).toHaveBeenCalledTimes(1)
+    const loggedString = log.warn.mock.calls[0][0]
+
+    expect(typeof loggedString).toBe('string')
+    expect(loggedString).toContain(testMessage)
+    expect(loggedString).toContain('warn')
+    // Check for timestamp-like content (numbers and colons)
+    expect(loggedString).toMatch(/\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}/)
+
+    restore()
+  })
 
   describe('print level', () => {
     test('debug enabled: debug logs should be printing', () => {

src/lib/logger/printer.ts

@@ -2,7 +2,7 @@ import { stringify, sanitize } from './sanitizer.js'
 import { Sanitizer } from './types.js'
 
 const colorTable = {
-  debug: '\x1b[32m', // green
+  debug: '\x1b[34m', // blue
   info: '\x1b[32m', // green
   log: '\x1b[36m', // cyan
   warn: '\x1b[33m', // yellow

src/lib/request/index.ts

@@ -1,3 +1,4 @@
+import { LoggerService } from '../logger/index.js'
 import { HttpException } from './errors.js'
 import type {
   InternalConfig,
@@ -77,3 +78,36 @@ export const makeRequest = (initMiddlewares: Middleware[]) => {
     return chain(internalConfig, middleware)
   }
 }
+
+// Helper function to safely parse JSON responses
+export const safelyParseJsonResponse = async (
+  response: Response,
+  logger: LoggerService
+) => {
+  try {
+    const text = await response.text()
+    try {
+      return JSON.parse(text)
+    } catch (jsonError) {
+      // If it starts with < it's likely HTML/XML, otherwise it's some other non-JSON format
+      const isMarkup = text.trim().startsWith('<')
+      const content = `${text.substring(0, 200)}${
+        text.length > 200 ? '...' : ''
+      }`
+      const errorMessage = isMarkup
+        ? `Received markup (HTML/XML) response instead of JSON. Status:`
+        : `Invalid JSON response. Status:`
+
+      logger.error(
+        `${errorMessage} ${response.status} ${response.statusText}`,
+        { content }
+      )
+      throw new Error(errorMessage)
+    }
+  } catch (error) {
+    if (error instanceof Error) {
+      throw error
+    }
+    throw new Error(`Failed to process response: ${response.statusText}`)
+  }
+}

src/lib/request/types.ts

@@ -4,6 +4,7 @@ import type {
   RequestInfo as FetchRequestInfo,
 } from 'node-fetch'
 export interface RequestConfig extends RequestInit {
+  // To be used in the future in the balancer mechanism
   baseUrl?: FetchRequestInfo
   middlewares?: Middleware[]
 }

src/scripts/logs.ts

@@ -0,0 +1,63 @@
+import dotenv from 'dotenv'
+import {
+  makeLogger,
+  makeRequest,
+  retry,
+  abort,
+  notOkError,
+  prom,
+  logger as loggerMiddleware,
+} from '../lib/index.js'
+import { makeLoggerConfig, makeConfig } from '../services/config/service.js'
+import { makeExecutionApi } from '../services/execution-api/service.js'
+import { makeExitLogsService } from '../services/exit-logs/service.js'
+import { makeMetrics } from '../services/prom/service.js'
+
+dotenv.config()
+
+const run = async () => {
+  const loggerConfig = makeLoggerConfig({ env: process.env })
+
+  const logger = makeLogger({
+    level: loggerConfig.LOGGER_LEVEL,
+    format: loggerConfig.LOGGER_FORMAT,
+    sanitizer: {
+      secrets: loggerConfig.LOGGER_SECRETS,
+      replacer: '<secret>',
+    },
+  })
+
+  const config = makeConfig({ logger, env: process.env })
+
+  const operatorIds = config.OPERATOR_IDS
+
+  const metrics = makeMetrics({ PREFIX: config.PROM_PREFIX })
+
+  const executionHttp = makeRequest([
+    retry(3),
+    loggerMiddleware(logger),
+    prom(metrics.executionRequestDurationSeconds),
+    notOkError(),
+    abort(30_000),
+  ])
+
+  const executionApi = makeExecutionApi(executionHttp, logger, config)
+
+  const exitLogs = makeExitLogsService(logger, executionApi, config, metrics)
+
+  await executionApi.resolveExitBusAddress()
+  await executionApi.resolveConsensusAddress()
+  const fetchTimeStart = performance.now()
+  const lastBlockNumber = await executionApi.latestBlockNumber()
+
+  const logs = await exitLogs.getLogs(operatorIds, lastBlockNumber)
+
+  logger.info('logs fetched', { count: logs.length })
+
+  const fetchTimeEnd = performance.now()
+  const fetchTime = Math.ceil(fetchTimeEnd - fetchTimeStart) / 1000
+
+  logger.info(`Logs fetched in ${fetchTime} seconds`)
+}
+
+run().catch(console.error)