From 275ff4fa0115b9a47d5901f14e043f93f656e6bf Mon Sep 17 00:00:00 2001 From: Guillaume Jacquart Date: Thu, 15 Jan 2026 10:09:13 +0100 Subject: [PATCH] chore(core): Extract log streaming log streaming standalone handling of destinations (#24197) --- .../cli/src/controllers/e2e.controller.ts | 10 +- .../message-event-bus/message-event-bus.ts | 110 --------- ...rometheus-metrics.service.unmocked.test.ts | 2 +- .../message-event-bus-destination.ee.ts | 56 +---- .../log-streaming-destination.service.ts | 214 ++++++++++++++++-- .../log-streaming.controller.ts | 14 +- .../log-streaming.ee/log-streaming.module.ts | 1 + .../cli/test/integration/eventbus.ee.test.ts | 36 ++- .../integration/eventbus/syslog-tls.test.ts | 15 +- 9 files changed, 259 insertions(+), 199 deletions(-) diff --git a/packages/cli/src/controllers/e2e.controller.ts b/packages/cli/src/controllers/e2e.controller.ts index c0a30886d0c..601a0e55218 100644 --- a/packages/cli/src/controllers/e2e.controller.ts +++ b/packages/cli/src/controllers/e2e.controller.ts @@ -18,7 +18,6 @@ import { v4 as uuid } from 'uuid'; import { ActiveWorkflowManager } from '@/active-workflow-manager'; import { inE2ETests } from '@/constants'; -import { MessageEventBus } from '@/eventbus/message-event-bus/message-event-bus'; import type { FeatureReturnType } from '@/license'; import { License } from '@/license'; import { MfaService } from '@/mfa/mfa.service'; @@ -167,7 +166,6 @@ export class E2EController { private readonly cacheService: CacheService, private readonly push: Push, private readonly passwordUtility: PasswordUtility, - private readonly eventBus: MessageEventBus, private readonly userRepository: UserRepository, private readonly frontendService: FrontendService, private readonly executionsConfig: ExecutionsConfig, @@ -282,9 +280,11 @@ export class E2EController { } private async resetLogStreaming() { - for (const id in this.eventBus.destinations) { - await this.eventBus.removeDestination(id, false); - await this.logStreamingDestinationsService.deleteDestinationFromDb(id); + const destinations = await this.logStreamingDestinationsService.findDestination(); + for (const destination of destinations) { + if (destination.id) { + await this.logStreamingDestinationsService.removeDestination(destination.id, false); + } } } diff --git a/packages/cli/src/eventbus/message-event-bus/message-event-bus.ts b/packages/cli/src/eventbus/message-event-bus/message-event-bus.ts index 3ec6572a00c..f838e402f52 100644 --- a/packages/cli/src/eventbus/message-event-bus/message-event-bus.ts +++ b/packages/cli/src/eventbus/message-event-bus/message-event-bus.ts @@ -1,16 +1,11 @@ import { Logger } from '@n8n/backend-common'; import { GlobalConfig } from '@n8n/config'; import { ExecutionRepository, WorkflowRepository } from '@n8n/db'; -import { OnPubSubEvent } from '@n8n/decorators'; import { Service } from '@n8n/di'; // eslint-disable-next-line n8n-local-rules/misplaced-n8n-typeorm-import import { In, IsNull, Not } from '@n8n/typeorm'; import EventEmitter from 'events'; import uniqby from 'lodash/uniqBy'; -import type { MessageEventBusDestinationOptions } from 'n8n-workflow'; - -import { License } from '@/license'; -import { Publisher } from '@/scaling/pubsub/publisher.service'; import { ExecutionRecoveryService } from '../../executions/execution-recovery.service'; import type { EventMessageTypes } from '../event-message-classes/'; @@ -23,10 +18,6 @@ import { EventMessageAudit } from '../event-message-classes/event-message-audit' import type { EventMessageConfirmSource } from '../event-message-classes/event-message-confirm'; import type { EventMessageExecutionOptions } from '../event-message-classes/event-message-execution'; import { EventMessageExecution } from '../event-message-classes/event-message-execution'; -import { - EventMessageGeneric, - eventMessageGenericDestinationTestEvent, -} from '../event-message-classes/event-message-generic'; import type { EventMessageNodeOptions } from '../event-message-classes/event-message-node'; import { EventMessageNode } from '../event-message-classes/event-message-node'; import type { EventMessageQueueOptions } from '../event-message-classes/event-message-queue'; @@ -50,17 +41,6 @@ export interface MessageEventBusInitializeOptions { webhookProcessorId?: string; } -// TODO: remove startListening and potential close method, -// As it should be the responsibility of the event bus to listen and close destinations -export interface MessageEventBusDestinationType { - getId(): string; - hasSubscribedToEvent(msg: EventMessageTypes): boolean; - receiveFromEventBus(messageWithCallback: MessageWithCallback): Promise; - startListening(): void; - close(): Promise; - serialize(): MessageEventBusDestinationOptions; -} - @Service() // TODO: Convert to TypedEventEmitter // eslint-disable-next-line n8n-local-rules/no-type-unsafe-event-emitter @@ -69,19 +49,13 @@ export class MessageEventBus extends EventEmitter { logWriter: MessageEventBusLogWriter; - destinations: { - [key: string]: MessageEventBusDestinationType; - } = {}; - private pushIntervalTimer: NodeJS.Timeout; constructor( private readonly logger: Logger, private readonly executionRepository: ExecutionRepository, private readonly workflowRepository: WorkflowRepository, - private readonly publisher: Publisher, private readonly recoveryService: ExecutionRecoveryService, - private readonly license: License, private readonly globalConfig: GlobalConfig, ) { super(); @@ -226,36 +200,6 @@ export class MessageEventBus extends EventEmitter { this.isInitialized = true; } - async addDestination(destination: MessageEventBusDestinationType, notifyWorkers: boolean = true) { - await this.removeDestination(destination.getId(), false); - this.destinations[destination.getId()] = destination; - this.destinations[destination.getId()].startListening(); - if (notifyWorkers) { - void this.publisher.publishCommand({ command: 'restart-event-bus' }); - } - return destination; - } - - async findDestination(id?: string): Promise { - let result: MessageEventBusDestinationOptions[]; - if (id && Object.keys(this.destinations).includes(id)) { - result = [this.destinations[id].serialize()]; - } else { - result = Object.keys(this.destinations).map((e) => this.destinations[e].serialize()); - } - return result.sort((a, b) => (a.__type ?? '').localeCompare(b.__type ?? '')); - } - - async removeDestination(id: string, notifyWorkers: boolean = true) { - if (Object.keys(this.destinations).includes(id)) { - await this.destinations[id].close(); - delete this.destinations[id]; - } - if (notifyWorkers) { - void this.publisher.publishCommand({ command: 'restart-event-bus' }); - } - } - private async trySendingUnsent(msgs?: EventMessageTypes[]) { const unsentMessages = msgs ?? (await this.getEventsUnsent()); if (unsentMessages.length > 0) { @@ -270,76 +214,30 @@ export class MessageEventBus extends EventEmitter { async close() { this.logger.debug('Shutting down event writer...'); await this.logWriter?.close(); - for (const destinationName of Object.keys(this.destinations)) { - this.logger.debug( - `Shutting down event destination ${this.destinations[destinationName].getId()}...`, - ); - await this.destinations[destinationName].close(); - } this.isInitialized = false; this.logger.debug('EventBus shut down.'); } - @OnPubSubEvent('restart-event-bus') - async restart() { - await this.close(); - await this.initialize({ skipRecoveryPass: true }); - } - async send(msgs: EventMessageTypes | EventMessageTypes[]) { if (!Array.isArray(msgs)) { msgs = [msgs]; } for (const msg of msgs) { this.logWriter?.putMessage(msg); - // if there are no set up destinations, immediately mark the event as sent - if (!this.shouldSendMsg(msg)) { - this.confirmSent(msg, { id: '0', name: 'eventBus' }); - } await this.emitMessage(msg); } } - async testDestination(destinationId: string): Promise { - const msg = new EventMessageGeneric({ - eventName: eventMessageGenericDestinationTestEvent, - }); - const destination = await this.findDestination(destinationId); - if (destination.length > 0) { - const sendResult = await this.destinations[destinationId].receiveFromEventBus({ - msg, - confirmCallback: () => this.confirmSent(msg, { id: '0', name: 'eventBus' }), - }); - return sendResult; - } - return false; - } - confirmSent(msg: EventMessageTypes, source?: EventMessageConfirmSource) { this.logWriter?.confirmMessageSent(msg.id, source); } - private hasAnyDestinationSubscribedToEvent(msg: EventMessageTypes): boolean { - for (const destinationName of Object.keys(this.destinations)) { - if (this.destinations[destinationName].hasSubscribedToEvent(msg)) { - return true; - } - } - return false; - } - private async emitMessage(msg: EventMessageTypes) { this.emit('metrics.eventBus.event', msg); // generic emit for external modules to capture events // this is for internal use ONLY and not for use with custom destinations! this.emitMessageWithCallback('message', msg); - - if (this.shouldSendMsg(msg)) { - for (const destinationName of Object.keys(this.destinations)) { - this.emitMessageWithCallback(this.destinations[destinationName].getId(), msg); - } - } } private emitMessageWithCallback(eventName: string, msg: EventMessageTypes): boolean { @@ -348,14 +246,6 @@ export class MessageEventBus extends EventEmitter { return this.emit(eventName, msg, confirmCallback); } - shouldSendMsg(msg: EventMessageTypes): boolean { - return ( - this.license.isLogStreamingEnabled() && - Object.keys(this.destinations).length > 0 && - this.hasAnyDestinationSubscribedToEvent(msg) - ); - } - async getEventsAll(): Promise { const queryResult = await this.logWriter?.getMessagesAll(); const filtered = uniqby(queryResult, 'id'); diff --git a/packages/cli/src/metrics/__tests__/prometheus-metrics.service.unmocked.test.ts b/packages/cli/src/metrics/__tests__/prometheus-metrics.service.unmocked.test.ts index 04eb6a20393..bd8ad02f556 100644 --- a/packages/cli/src/metrics/__tests__/prometheus-metrics.service.unmocked.test.ts +++ b/packages/cli/src/metrics/__tests__/prometheus-metrics.service.unmocked.test.ts @@ -22,7 +22,7 @@ const eventService = mock(); const instanceSettings = mock({ instanceType: 'main' }); const workflowRepository = mock(); const app = mock(); -const eventBus = new MessageEventBus(mock(), mock(), mock(), mock(), mock(), mock(), mock()); +const eventBus = new MessageEventBus(mock(), mock(), mock(), mock(), mock()); describe('workflow_success_total', () => { test('support workflow id labels', async () => { diff --git a/packages/cli/src/modules/log-streaming.ee/destinations/message-event-bus-destination.ee.ts b/packages/cli/src/modules/log-streaming.ee/destinations/message-event-bus-destination.ee.ts index cbd237b62cc..f53c99dbc72 100644 --- a/packages/cli/src/modules/log-streaming.ee/destinations/message-event-bus-destination.ee.ts +++ b/packages/cli/src/modules/log-streaming.ee/destinations/message-event-bus-destination.ee.ts @@ -11,20 +11,15 @@ import type { INodeCredentials, MessageEventBusDestinationOptions } from 'n8n-wo import { MessageEventBusDestinationTypeNames } from 'n8n-workflow'; import { v4 as uuid } from 'uuid'; -import type { EventMessageTypes } from '@/eventbus'; import type { AbstractEventMessage } from '@/eventbus/event-message-classes/abstract-event-message'; -import type { EventMessageConfirmSource } from '@/eventbus/event-message-classes/event-message-confirm'; import type { MessageEventBus, - MessageEventBusDestinationType, MessageWithCallback, } from '@/eventbus/message-event-bus/message-event-bus'; import { License } from '@/license'; import { CircuitBreaker } from '@/utils/circuit-breaker'; -export abstract class MessageEventBusDestination - implements MessageEventBusDestinationOptions, MessageEventBusDestinationType -{ +export abstract class MessageEventBusDestination implements MessageEventBusDestinationOptions { // Since you can't have static abstract functions - this just serves as a reminder that you need to implement these. Please. // static abstract deserialize(): MessageEventBusDestination | null; readonly id: string; @@ -83,44 +78,6 @@ export abstract class MessageEventBusDestination this.logger.debug(`${this.__type}(${this.id}) event destination constructed`); } - startListening() { - if (this.enabled) { - this.eventBusInstance.on( - this.getId(), - async ( - msg: EventMessageTypes, - confirmCallback: (message: EventMessageTypes, src: EventMessageConfirmSource) => void, - ) => { - try { - await this.circuitBreakerInstance.execute(async () => { - await this.receiveFromEventBus({ msg, confirmCallback }); - }); - } catch (error) { - this.logger.error( - `${this.__type}(${this.id}) event destination ${this.label} failed to send message`, - { error }, - ); - } - }, - ); - this.logger.debug(`${this.id} listener started`); - } - } - - stopListening() { - this.eventBusInstance.removeAllListeners(this.getId()); - } - - enable() { - this.enabled = true; - this.startListening(); - } - - disable() { - this.enabled = false; - this.stopListening(); - } - getId() { return this.id; } @@ -148,11 +105,20 @@ export abstract class MessageEventBusDestination abstract receiveFromEventBus(emitterPayload: MessageWithCallback): Promise; + /** + * Send a message to this destination with circuit breaker protection + */ + async sendMessage(emitterPayload: MessageWithCallback): Promise { + return await this.circuitBreakerInstance.execute(async () => { + return await this.receiveFromEventBus(emitterPayload); + }); + } + toString() { return JSON.stringify(this.serialize()); } async close(): Promise { - this.stopListening(); + this.enabled = false; } } diff --git a/packages/cli/src/modules/log-streaming.ee/log-streaming-destination.service.ts b/packages/cli/src/modules/log-streaming.ee/log-streaming-destination.service.ts index e21174dcae4..189e1156a7c 100644 --- a/packages/cli/src/modules/log-streaming.ee/log-streaming-destination.service.ts +++ b/packages/cli/src/modules/log-streaming.ee/log-streaming-destination.service.ts @@ -1,27 +1,51 @@ -import { Logger } from '@n8n/backend-common'; +import { LicenseState, Logger } from '@n8n/backend-common'; +import { OnPubSubEvent } from '@n8n/decorators'; import { Service } from '@n8n/di'; import type { DeleteResult } from '@n8n/typeorm'; +import type { MessageEventBusDestinationOptions } from 'n8n-workflow'; -import type { MessageEventBusDestinationType } from '@/eventbus/message-event-bus/message-event-bus'; +import type { EventMessageTypes } from '@/eventbus'; +import type { EventMessageConfirmSource } from '@/eventbus/event-message-classes/event-message-confirm'; +import { + EventMessageGeneric, + eventMessageGenericDestinationTestEvent, +} from '@/eventbus/event-message-classes/event-message-generic'; import { MessageEventBus } from '@/eventbus/message-event-bus/message-event-bus'; +import { Publisher } from '@/scaling/pubsub/publisher.service'; import { EventDestinationsRepository } from './database/repositories/event-destination.repository'; import { messageEventBusDestinationFromDb } from './destinations/message-event-bus-destination-from-db'; +import { MessageEventBusDestination } from './destinations/message-event-bus-destination.ee'; /** - * Service to handle all database operations for log streaming destinations. - * This ensures the message event bus doesn't directly access the database. + * Service to handle all log streaming destination operations including: + * - Database persistence + * - Destination lifecycle management + * - Event forwarding from event bus to destinations */ @Service() export class LogStreamingDestinationService { + private destinations: { [key: string]: MessageEventBusDestination } = {}; + + private isListening = false; + + private readonly messageHandler: ( + msg: EventMessageTypes, + confirmCallback: (message: EventMessageTypes, src: EventMessageConfirmSource) => void, + ) => Promise; + constructor( private readonly logger: Logger, private readonly eventDestinationsRepository: EventDestinationsRepository, private readonly eventBus: MessageEventBus, - ) {} + private readonly licenseState: LicenseState, + private readonly publisher: Publisher, + ) { + this.messageHandler = this.handleMessage.bind(this); + } /** - * Load all destinations from database and add them to the event bus + * Load all destinations from database and add them to the local destinations map */ async loadDestinationsFromDb(): Promise { const savedEventDestinations = await this.eventDestinationsRepository.find({}); @@ -30,7 +54,8 @@ export class LogStreamingDestinationService { try { const destination = messageEventBusDestinationFromDb(this.eventBus, destinationData); if (destination) { - await this.eventBus.addDestination(destination, false); + this.destinations[destination.getId()] = destination; + this.logger.debug(`Loaded destination ${destination.getId()} from database`); } } catch (error) { this.logger.debug('Failed to load destination from database', { error }); @@ -42,7 +67,7 @@ export class LogStreamingDestinationService { /** * Save a destination to the database */ - async saveDestinationToDb(destination: MessageEventBusDestinationType) { + async saveDestinationToDb(destination: MessageEventBusDestination) { const data = { id: destination.getId(), destination: destination.serialize(), @@ -62,22 +87,179 @@ export class LogStreamingDestinationService { } /** - * Add a destination to the event bus and save to database + * Add a destination to the local map and save to database */ async addDestination( - destination: MessageEventBusDestinationType, + destination: MessageEventBusDestination, notifyWorkers: boolean = true, - ): Promise { - const addedDestination = await this.eventBus.addDestination(destination, notifyWorkers); - await this.saveDestinationToDb(addedDestination); - return addedDestination; + ): Promise { + // Remove any existing destination with the same ID + await this.destinations[destination.getId()]?.close(); + + // Add to local destinations map + this.destinations[destination.getId()] = destination; + this.logger.debug(`Added destination ${destination.getId()}`); + + // Save to database + await this.saveDestinationToDb(destination); + + // Notify workers to reload destinations + if (notifyWorkers) { + void this.publisher.publishCommand({ command: 'restart-event-bus' }); + } + + return destination; } /** - * Remove a destination from the event bus and delete from database + * Remove a destination from the local map and delete from database */ async removeDestination(id: string, notifyWorkers: boolean = true): Promise { - await this.eventBus.removeDestination(id, notifyWorkers); + // Close and remove from local map + if (this.destinations[id]) { + await this.destinations[id].close(); + delete this.destinations[id]; + this.logger.debug(`Removed destination ${id}`); + } + + // Notify workers to reload destinations + if (notifyWorkers) { + void this.publisher.publishCommand({ command: 'restart-event-bus' }); + } + + // Delete from database return await this.deleteDestinationFromDb(id); } + + /** + * Initialize the service by registering listener on event bus + */ + async initialize(): Promise { + if (this.isListening) { + return; + } + + this.logger.debug('Initializing log streaming destination service...'); + + // Register listener for all messages from event bus + this.eventBus.on('message', this.messageHandler); + this.isListening = true; + + this.logger.debug( + `Log streaming destination service initialized with ${Object.keys(this.destinations).length} destinations`, + ); + } + + /** + * Handle messages from event bus and forward to appropriate destinations + */ + private async handleMessage( + msg: EventMessageTypes, + confirmCallback: (message: EventMessageTypes, src: EventMessageConfirmSource) => void, + ) { + // If there are no destinations that should receive this message, mark it as sent immediately + if (!this.shouldSendMsg(msg)) { + this.eventBus.confirmSent(msg, { id: '0', name: 'eventBus' }); + return; + } + + for (const destination of Object.values(this.destinations)) { + if (destination.hasSubscribedToEvent(msg)) { + try { + // Call destination with circuit breaker protection + await destination.sendMessage({ msg, confirmCallback }); + } catch (error) { + this.logger.error(`Failed to send message to destination ${destination.getId()}`, { + error, + }); + } + } + } + } + + /** + * Find destination(s) by ID + */ + async findDestination(id?: string): Promise { + const result: MessageEventBusDestinationOptions[] = id + ? this.destinations[id] + ? [this.destinations[id].serialize()] + : [] + : Object.values(this.destinations).map((dest) => dest.serialize()); + return result.sort((a, b) => (a.__type ?? '').localeCompare(b.__type ?? '')); + } + + /** + * Test a destination by sending a test message + * Note: This bypasses the circuit breaker to allow testing even when circuit is open + */ + async testDestination(destinationId: string): Promise { + const msg = new EventMessageGeneric({ + eventName: eventMessageGenericDestinationTestEvent, + }); + const destination = await this.findDestination(destinationId); + if (destination.length > 0) { + const sendResult = await this.destinations[destinationId].receiveFromEventBus({ + msg, + confirmCallback: () => this.eventBus.confirmSent(msg, { id: '0', name: 'eventBus' }), + }); + return sendResult; + } + return false; + } + + /** + * Check if any destination is subscribed to the given event + */ + private hasAnyDestinationSubscribedToEvent(msg: EventMessageTypes): boolean { + return Object.values(this.destinations).some((destination) => + destination.hasSubscribedToEvent(msg), + ); + } + + /** + * Determine if a message should be sent to destinations + */ + shouldSendMsg(msg: EventMessageTypes): boolean { + return ( + this.licenseState.isLogStreamingLicensed() && + Object.keys(this.destinations).length > 0 && + this.hasAnyDestinationSubscribedToEvent(msg) + ); + } + + /** + * Close all destinations + */ + async close(): Promise { + this.logger.debug('Closing log streaming destination service...'); + + // Stop listening to event bus + if (this.isListening) { + this.eventBus.removeListener('message', this.messageHandler); + this.isListening = false; + } + + // Close all destinations + for (const destination of Object.values(this.destinations)) { + this.logger.debug(`Closing destination ${destination.getId()}...`); + await destination.close(); + } + + // Clear destinations map + this.destinations = {}; + + this.logger.debug('Log streaming destination service closed'); + } + + /** + * Handle restart event from pubsub + */ + @OnPubSubEvent('restart-event-bus') + async restart() { + this.logger.debug('Restarting log streaming destination service...'); + await this.close(); + await this.loadDestinationsFromDb(); + await this.initialize(); + } } diff --git a/packages/cli/src/modules/log-streaming.ee/log-streaming.controller.ts b/packages/cli/src/modules/log-streaming.ee/log-streaming.controller.ts index e50b409bddf..47202a72aeb 100644 --- a/packages/cli/src/modules/log-streaming.ee/log-streaming.controller.ts +++ b/packages/cli/src/modules/log-streaming.ee/log-streaming.controller.ts @@ -9,10 +9,7 @@ import { MessageEventBusDestinationTypeNames } from 'n8n-workflow'; import { BadRequestError } from '@/errors/response-errors/bad-request.error'; import { eventNamesAll } from '@/eventbus/event-message-classes'; -import { - MessageEventBus, - MessageEventBusDestinationType, -} from '@/eventbus/message-event-bus/message-event-bus'; +import { MessageEventBus } from '@/eventbus/message-event-bus/message-event-bus'; import { isMessageEventBusDestinationSentryOptions, @@ -23,6 +20,7 @@ import { MessageEventBusDestinationSyslog, } from './destinations/message-event-bus-destination-syslog.ee'; import { MessageEventBusDestinationWebhook } from './destinations/message-event-bus-destination-webhook.ee'; +import { MessageEventBusDestination } from './destinations/message-event-bus-destination.ee'; import { LogStreamingDestinationService } from './log-streaming-destination.service'; const isWithIdString = (candidate: unknown): candidate is { id: string } => { @@ -64,9 +62,9 @@ export class EventBusController { @GlobalScope('eventBusDestination:list') async getDestination(req: express.Request): Promise { if (isWithIdString(req.query)) { - return await this.eventBus.findDestination(req.query.id); + return await this.destinationService.findDestination(req.query.id); } else { - return await this.eventBus.findDestination(); + return await this.destinationService.findDestination(); } } @@ -74,7 +72,7 @@ export class EventBusController { @Post('/destination') @GlobalScope('eventBusDestination:create') async postDestination(req: AuthenticatedRequest): Promise { - let result: MessageEventBusDestinationType | undefined; + let result: MessageEventBusDestination | undefined; if (isMessageEventBusDestinationOptions(req.body)) { switch (req.body.__type) { case MessageEventBusDestinationTypeNames.sentry: @@ -119,7 +117,7 @@ export class EventBusController { @GlobalScope('eventBusDestination:test') async sendTestMessage(req: express.Request): Promise { if (isWithIdString(req.query)) { - return await this.eventBus.testDestination(req.query.id); + return await this.destinationService.testDestination(req.query.id); } return false; } diff --git a/packages/cli/src/modules/log-streaming.ee/log-streaming.module.ts b/packages/cli/src/modules/log-streaming.ee/log-streaming.module.ts index ddd13b38cb2..cf200d5d02d 100644 --- a/packages/cli/src/modules/log-streaming.ee/log-streaming.module.ts +++ b/packages/cli/src/modules/log-streaming.ee/log-streaming.module.ts @@ -22,6 +22,7 @@ export class LogStreamingModule implements ModuleInterface { const { LogStreamingDestinationService } = await import('./log-streaming-destination.service'); const destinationService = Container.get(LogStreamingDestinationService); await destinationService.loadDestinationsFromDb(); + await destinationService.initialize(); } async entities() { diff --git a/packages/cli/test/integration/eventbus.ee.test.ts b/packages/cli/test/integration/eventbus.ee.test.ts index fb84340c636..f388783507e 100644 --- a/packages/cli/test/integration/eventbus.ee.test.ts +++ b/packages/cli/test/integration/eventbus.ee.test.ts @@ -23,6 +23,7 @@ import { ExecutionRecoveryService } from '@/executions/execution-recovery.servic import type { MessageEventBusDestinationSentry } from '@/modules/log-streaming.ee/destinations/message-event-bus-destination-sentry.ee'; import type { MessageEventBusDestinationSyslog } from '@/modules/log-streaming.ee/destinations/message-event-bus-destination-syslog.ee'; import type { MessageEventBusDestinationWebhook } from '@/modules/log-streaming.ee/destinations/message-event-bus-destination-webhook.ee'; +import { LogStreamingDestinationService } from '@/modules/log-streaming.ee/log-streaming-destination.service'; import { Publisher } from '@/scaling/pubsub/publisher.service'; import { createUser } from './shared/db/users'; @@ -70,6 +71,7 @@ const testSentryDestination: MessageEventBusDestinationSentryOptions = { }; let eventBus: MessageEventBus; +let destinationService: LogStreamingDestinationService; async function confirmIdInAll(id: string) { const sent = await eventBus.getEventsAll(); @@ -96,6 +98,9 @@ beforeAll(async () => { eventBus = Container.get(MessageEventBus); await eventBus.initialize(); + + destinationService = Container.get(LogStreamingDestinationService); + await destinationService.initialize(); }); afterAll(async () => { @@ -140,7 +145,7 @@ describe('GET /eventbus/destination', () => { for (let index = 0; index < data.length; index++) { const destination = data[index]; - const foundDestinations = await eventBus.findDestination(destination.id); + const foundDestinations = await destinationService.findDestination(destination.id); expect(Array.isArray(foundDestinations)).toBeTruthy(); expect(foundDestinations.length).toBe(1); expect(foundDestinations[0].label).toBe(destination.label); @@ -177,11 +182,11 @@ test('should anonymize audit message to syslog ', async () => { id: uuid(), }); - const syslogDestination = eventBus.destinations[ + const syslogDestination = destinationService['destinations'][ testSyslogDestination.id! ] as MessageEventBusDestinationSyslog; - syslogDestination.enable(); + syslogDestination.enabled = true; const mockedSyslogClientLog = jest.spyOn(syslogDestination.client, 'log'); mockedSyslogClientLog.mockImplementation(async (m, _options, _cb) => { @@ -222,7 +227,7 @@ test('should anonymize audit message to syslog ', async () => { await eventBus.getEventsAll(); await confirmIdInAll(testAuditMessage.id); expect(mockedSyslogClientLog).toHaveBeenCalled(); - syslogDestination.disable(); + syslogDestination.enabled = false; eventBus.logWriter.worker?.removeListener('message', handler006); resolve(true); } @@ -237,11 +242,11 @@ test('should send message to webhook ', async () => { id: uuid(), }); - const webhookDestination = eventBus.destinations[ + const webhookDestination = destinationService['destinations'][ testWebhookDestination.id! ] as MessageEventBusDestinationWebhook; - webhookDestination.enable(); + webhookDestination.enabled = true; mockAxiosInstance.post.mockResolvedValue({ status: 200, data: { msg: 'OK' } }); mockAxiosInstance.request.mockResolvedValue({ status: 200, data: { msg: 'OK' } }); @@ -256,7 +261,7 @@ test('should send message to webhook ', async () => { } else if (msg.command === 'confirmMessageSent') { await confirmIdSent(testMessage.id); expect(mockAxiosInstance.request).toHaveBeenCalled(); - webhookDestination.disable(); + webhookDestination.enabled = false; eventBus.logWriter.worker?.removeListener('message', handler003); resolve(true); } @@ -271,11 +276,11 @@ test('should send message to sentry ', async () => { id: uuid(), }); - const sentryDestination = eventBus.destinations[ + const sentryDestination = destinationService['destinations'][ testSentryDestination.id! ] as MessageEventBusDestinationSentry; - sentryDestination.enable(); + sentryDestination.enabled = true; const mockedSentryCaptureMessage = jest.spyOn(sentryDestination.sentryClient!, 'captureMessage'); mockedSentryCaptureMessage.mockImplementation((_m, _level, _hint, _scope) => { @@ -296,7 +301,7 @@ test('should send message to sentry ', async () => { } else if (msg.command === 'confirmMessageSent') { await confirmIdSent(testMessage.id); expect(mockedSentryCaptureMessage).toHaveBeenCalled(); - sentryDestination.disable(); + sentryDestination.enabled = false; eventBus.logWriter.worker?.removeListener('message', handler004); resolve(true); } @@ -306,7 +311,13 @@ test('should send message to sentry ', async () => { }); test('DELETE /eventbus/destination delete all destinations by id', async () => { - const existingDestinationIds = [...Object.keys(eventBus.destinations)]; + const existingDestinations = await destinationService.findDestination(); + const existingDestinationIds = existingDestinations.reduce((acc, d) => { + if (d.id) { + acc.push(d.id); + } + return acc; + }, []); await Promise.all( existingDestinationIds.map(async (id) => { @@ -315,5 +326,6 @@ test('DELETE /eventbus/destination delete all destinations by id', async () => { }), ); - expect(Object.keys(eventBus.destinations).length).toBe(0); + const remainingDestinations = await destinationService.findDestination(); + expect(remainingDestinations.length).toBe(0); }); diff --git a/packages/cli/test/integration/eventbus/syslog-tls.test.ts b/packages/cli/test/integration/eventbus/syslog-tls.test.ts index 3ec7f371143..2d097f7b2ea 100644 --- a/packages/cli/test/integration/eventbus/syslog-tls.test.ts +++ b/packages/cli/test/integration/eventbus/syslog-tls.test.ts @@ -7,6 +7,7 @@ import type TestAgent from 'supertest/lib/agent'; import { MessageEventBus } from '@/eventbus/message-event-bus/message-event-bus'; import { EventMessageGeneric } from '@/eventbus/event-message-classes/event-message-generic'; +import { LogStreamingDestinationService } from '@/modules/log-streaming.ee/log-streaming-destination.service'; import { TlsSyslogServer } from './tls-server'; import { createUser } from '../shared/db/users'; @@ -17,12 +18,14 @@ jest.unmock('@/eventbus/message-event-bus/message-event-bus'); const tlsServer = new TlsSyslogServer(); let serverPort: number; let eventBus: MessageEventBus; +let destinationService: LogStreamingDestinationService; let authOwnerAgent: TestAgent; let logger: Logger; const testServer = utils.setupTestServer({ endpointGroups: ['eventBus'], enabledFeatures: ['feat:logStreaming'], + modules: ['log-streaming'], }); afterAll(async () => { @@ -41,6 +44,9 @@ beforeAll(async () => { eventBus = Container.get(MessageEventBus); logger = Container.get(Logger); await eventBus.initialize(); + + destinationService = Container.get(LogStreamingDestinationService); + await destinationService.initialize(); }); describe('TLS Syslog E2E', () => { @@ -51,8 +57,13 @@ describe('TLS Syslog E2E', () => { tlsServer.clearMessages(); }); - afterEach(() => { - eventBus.destinations = {}; + afterEach(async () => { + const destinations = await destinationService.findDestination(); + for (const destination of destinations) { + if (destination.id) { + await destinationService.removeDestination(destination.id, false); + } + } }); test('should send message over real TLS connection', async () => {