diff --git a/.github/workflows/common-deploy.yml b/.github/workflows/common-deploy.yml index 92380541b..c2cb65b20 100644 --- a/.github/workflows/common-deploy.yml +++ b/.github/workflows/common-deploy.yml @@ -38,10 +38,6 @@ jobs: run: | jq '(.containerDefinitions[] | select(.name=="${{ inputs.service_name }}-prod") | .environment[] | select(.name=="VERSION")).value = "${{ github.sha }}"' task-definition.json > tmp.json && mv tmp.json task-definition.json - - name: Fill in the new package version in the Amazon ECS task definition - run: | - jq '(.containerDefinitions[] | select(.name=="${{ inputs.service_name }}-prod") | .environment[] | select(.name=="npm_package_version")).value = "${{ github.sha }}"' task-definition.json > tmp.json && mv tmp.json task-definition.json - - name: Fill in the new image ID in the Amazon ECS task definition id: task-def-prod uses: aws-actions/amazon-ecs-render-task-definition@v1 diff --git a/.yarn/cache/@opentelemetry-api-npm-1.6.0-58fdf34ce1-b8daefad2c.zip b/.yarn/cache/@opentelemetry-api-npm-1.6.0-58fdf34ce1-b8daefad2c.zip index 77101aa63..21a6a9bf5 100644 Binary files a/.yarn/cache/@opentelemetry-api-npm-1.6.0-58fdf34ce1-b8daefad2c.zip and b/.yarn/cache/@opentelemetry-api-npm-1.6.0-58fdf34ce1-b8daefad2c.zip differ diff --git a/packages/auth/bin/worker.ts b/packages/auth/bin/worker.ts index d6d786228..36d618e1d 100644 --- a/packages/auth/bin/worker.ts +++ b/packages/auth/bin/worker.ts @@ -8,6 +8,7 @@ import { Env } from '../src/Bootstrap/Env' import { DomainEventSubscriberFactoryInterface } from '@standardnotes/domain-events' import * as dayjs from 'dayjs' import * as utc from 'dayjs/plugin/utc' +import { OpenTelemetrySDKInterface } from '@standardnotes/domain-events-infra' const container = new ContainerConfigLoader('worker') void container.load().then((container) => { @@ -20,6 +21,11 @@ void container.load().then((container) => { logger.info('Starting worker...') + if (!container.get(TYPES.Auth_IS_CONFIGURED_FOR_HOME_SERVER_OR_SELF_HOSTING)) { + const openTelemetrySDK = container.get(TYPES.Auth_OpenTelemetrySDK) + openTelemetrySDK.start() + } + const subscriberFactory: DomainEventSubscriberFactoryInterface = container.get( TYPES.Auth_DomainEventSubscriberFactory, ) diff --git a/packages/auth/src/Bootstrap/Container.ts b/packages/auth/src/Bootstrap/Container.ts index 6ed555f7c..df216df93 100644 --- a/packages/auth/src/Bootstrap/Container.ts +++ b/packages/auth/src/Bootstrap/Container.ts @@ -94,6 +94,7 @@ import { SNSDomainEventPublisher, SQSDomainEventSubscriberFactory, SQSEventMessageHandler, + SQSOpenTelemetryEventMessageHandler, } from '@standardnotes/domain-events-infra' import { GetUserSubscription } from '../Domain/UseCase/GetUserSubscription/GetUserSubscription' import { ChangeCredentials } from '../Domain/UseCase/ChangeCredentials/ChangeCredentials' @@ -333,7 +334,11 @@ export class ContainerConfigLoader { if (!isConfiguredForHomeServerOrSelfHosting) { container .bind(TYPES.Auth_OpenTelemetrySDK) - .toConstantValue(new OpenTelemetrySDK(ServiceIdentifier.NAMES.Auth)) + .toConstantValue( + new OpenTelemetrySDK( + this.mode === 'server' ? ServiceIdentifier.NAMES.Auth : ServiceIdentifier.NAMES.AuthWorker, + ), + ) } if (!isConfiguredForInMemoryCache) { @@ -1238,7 +1243,11 @@ export class ContainerConfigLoader { } else { container .bind(TYPES.Auth_DomainEventMessageHandler) - .toConstantValue(new SQSEventMessageHandler(eventHandlers, container.get(TYPES.Auth_Logger))) + .toConstantValue( + isConfiguredForHomeServerOrSelfHosting + ? new SQSEventMessageHandler(eventHandlers, container.get(TYPES.Auth_Logger)) + : new SQSOpenTelemetryEventMessageHandler(eventHandlers, container.get(TYPES.Auth_Logger)), + ) container .bind(TYPES.Auth_DomainEventSubscriberFactory) diff --git a/packages/domain-events-infra/src/Infra/SQS/SQSOpenTelemetryEventMessageHandler.ts b/packages/domain-events-infra/src/Infra/SQS/SQSOpenTelemetryEventMessageHandler.ts new file mode 100644 index 000000000..ae02fb3cd --- /dev/null +++ b/packages/domain-events-infra/src/Infra/SQS/SQSOpenTelemetryEventMessageHandler.ts @@ -0,0 +1,56 @@ +import { Logger } from 'winston' +import * as zlib from 'zlib' +import * as OpenTelemetryApi from '@opentelemetry/api' +import { + DomainEventHandlerInterface, + DomainEventInterface, + DomainEventMessageHandlerInterface, +} from '@standardnotes/domain-events' + +export class SQSOpenTelemetryEventMessageHandler implements DomainEventMessageHandlerInterface { + private currentSpan: OpenTelemetryApi.Span | undefined + + constructor( + private handlers: Map, + private logger: Logger, + ) {} + + async handleMessage(message: string): Promise { + const messageParsed = JSON.parse(message) + + const domainEventJson = zlib.unzipSync(Buffer.from(messageParsed.Message, 'base64')).toString() + + const domainEvent: DomainEventInterface = JSON.parse(domainEventJson) + + domainEvent.createdAt = new Date(domainEvent.createdAt) + + const handler = this.handlers.get(domainEvent.type) + if (!handler) { + this.logger.debug(`Event handler for event type ${domainEvent.type} does not exist`) + + return + } + + this.logger.debug(`Received event: ${domainEvent.type}`) + + const tracer = OpenTelemetryApi.trace.getTracer('sqs-handler') + + this.currentSpan = tracer.startSpan(domainEvent.type) + + await handler.handle(domainEvent) + + this.currentSpan.end() + + this.currentSpan = undefined + } + + async handleError(error: Error): Promise { + if (this.currentSpan) { + this.currentSpan.recordException(error) + this.currentSpan.end() + this.currentSpan = undefined + } + + this.logger.error('Error occured while handling SQS message: %O', error) + } +} diff --git a/packages/domain-events-infra/src/Infra/index.ts b/packages/domain-events-infra/src/Infra/index.ts index 29d4422f2..bf4e21cde 100644 --- a/packages/domain-events-infra/src/Infra/index.ts +++ b/packages/domain-events-infra/src/Infra/index.ts @@ -15,3 +15,4 @@ export * from './SQS/SQSNewRelicBounceNotificiationHandler' export * from './SQS/SQSDomainEventSubscriberFactory' export * from './SQS/SQSEventMessageHandler' export * from './SQS/SQSNewRelicEventMessageHandler' +export * from './SQS/SQSOpenTelemetryEventMessageHandler' diff --git a/packages/files/bin/worker.ts b/packages/files/bin/worker.ts index 5fcea6222..246566be0 100644 --- a/packages/files/bin/worker.ts +++ b/packages/files/bin/worker.ts @@ -8,8 +8,9 @@ import { Env } from '../src/Bootstrap/Env' import { DomainEventSubscriberFactoryInterface } from '@standardnotes/domain-events' import * as dayjs from 'dayjs' import * as utc from 'dayjs/plugin/utc' +import { OpenTelemetrySDKInterface } from '@standardnotes/domain-events-infra' -const container = new ContainerConfigLoader() +const container = new ContainerConfigLoader('worker') void container.load().then((container) => { dayjs.extend(utc) @@ -20,6 +21,11 @@ void container.load().then((container) => { logger.info('Starting worker...') + if (!container.get(TYPES.Files_IS_CONFIGURED_FOR_HOME_SERVER_OR_SELF_HOSTING)) { + const openTelemetrySDK = container.get(TYPES.Files_OpenTelemetrySDK) + openTelemetrySDK.start() + } + const subscriberFactory: DomainEventSubscriberFactoryInterface = container.get( TYPES.Files_DomainEventSubscriberFactory, ) diff --git a/packages/files/src/Bootstrap/Container.ts b/packages/files/src/Bootstrap/Container.ts index 13afa8aef..f6edbc4ae 100644 --- a/packages/files/src/Bootstrap/Container.ts +++ b/packages/files/src/Bootstrap/Container.ts @@ -21,6 +21,7 @@ import { SNSDomainEventPublisher, SQSDomainEventSubscriberFactory, SQSEventMessageHandler, + SQSOpenTelemetryEventMessageHandler, } from '@standardnotes/domain-events-infra' import { StreamDownloadFile } from '../Domain/UseCase/StreamDownloadFile/StreamDownloadFile' import { FileDownloaderInterface } from '../Domain/Services/FileDownloaderInterface' @@ -57,6 +58,8 @@ import { SharedVaultValetTokenAuthMiddleware } from '../Infra/InversifyExpress/M import { ServiceIdentifier } from '@standardnotes/domain-core' export class ContainerConfigLoader { + constructor(private mode: 'server' | 'worker' = 'server') {} + async load(configuration?: { directCallDomainEventPublisher?: DirectCallDomainEventPublisher logger?: Transform @@ -96,7 +99,11 @@ export class ContainerConfigLoader { if (!isConfiguredForHomeServerOrSelfHosting) { container .bind(TYPES.Files_OpenTelemetrySDK) - .toConstantValue(new OpenTelemetrySDK(ServiceIdentifier.NAMES.Files)) + .toConstantValue( + new OpenTelemetrySDK( + this.mode === 'server' ? ServiceIdentifier.NAMES.Files : ServiceIdentifier.NAMES.FilesWorker, + ), + ) } let logger: winston.Logger @@ -306,7 +313,11 @@ export class ContainerConfigLoader { } else { container .bind(TYPES.Files_DomainEventMessageHandler) - .toConstantValue(new SQSEventMessageHandler(eventHandlers, container.get(TYPES.Files_Logger))) + .toConstantValue( + isConfiguredForHomeServerOrSelfHosting + ? new SQSEventMessageHandler(eventHandlers, container.get(TYPES.Files_Logger)) + : new SQSOpenTelemetryEventMessageHandler(eventHandlers, container.get(TYPES.Files_Logger)), + ) container .bind(TYPES.Files_DomainEventSubscriberFactory) .toConstantValue( diff --git a/packages/revisions/bin/worker.ts b/packages/revisions/bin/worker.ts index a49d308ea..a581c8ac7 100644 --- a/packages/revisions/bin/worker.ts +++ b/packages/revisions/bin/worker.ts @@ -6,6 +6,7 @@ import TYPES from '../src/Bootstrap/Types' import { Env } from '../src/Bootstrap/Env' import { DomainEventSubscriberFactoryInterface } from '@standardnotes/domain-events' import { ContainerConfigLoader } from '../src/Bootstrap/Container' +import { OpenTelemetrySDKInterface } from '@standardnotes/domain-events-infra' const container = new ContainerConfigLoader('worker') void container.load().then((container) => { @@ -16,6 +17,11 @@ void container.load().then((container) => { logger.info('Starting worker...') + if (!container.get(TYPES.Revisions_IS_CONFIGURED_FOR_HOME_SERVER_OR_SELF_HOSTING)) { + const openTelemetrySDK = container.get(TYPES.Revisions_OpenTelemetrySDK) + openTelemetrySDK.start() + } + const subscriberFactory: DomainEventSubscriberFactoryInterface = container.get( TYPES.Revisions_DomainEventSubscriberFactory, ) diff --git a/packages/revisions/src/Bootstrap/Container.ts b/packages/revisions/src/Bootstrap/Container.ts index e3b69ad03..9121e29fb 100644 --- a/packages/revisions/src/Bootstrap/Container.ts +++ b/packages/revisions/src/Bootstrap/Container.ts @@ -42,6 +42,7 @@ import { SNSDomainEventPublisher, OpenTelemetrySDKInterface, OpenTelemetrySDK, + SQSOpenTelemetryEventMessageHandler, } from '@standardnotes/domain-events-infra' import { DumpRepositoryInterface } from '../Domain/Dump/DumpRepositoryInterface' import { AccountDeletionRequestedEventHandler } from '../Domain/Handler/AccountDeletionRequestedEventHandler' @@ -161,7 +162,11 @@ export class ContainerConfigLoader { if (!isConfiguredForHomeServerOrSelfHosting) { container .bind(TYPES.Revisions_OpenTelemetrySDK) - .toConstantValue(new OpenTelemetrySDK(ServiceIdentifier.NAMES.Revisions)) + .toConstantValue( + new OpenTelemetrySDK( + this.mode === 'server' ? ServiceIdentifier.NAMES.Revisions : ServiceIdentifier.NAMES.RevisionsWorker, + ), + ) } if (!isConfiguredForHomeServer) { @@ -525,7 +530,11 @@ export class ContainerConfigLoader { } else { container .bind(TYPES.Revisions_DomainEventMessageHandler) - .toConstantValue(new SQSEventMessageHandler(eventHandlers, container.get(TYPES.Revisions_Logger))) + .toConstantValue( + isConfiguredForHomeServerOrSelfHosting + ? new SQSEventMessageHandler(eventHandlers, container.get(TYPES.Revisions_Logger)) + : new SQSOpenTelemetryEventMessageHandler(eventHandlers, container.get(TYPES.Revisions_Logger)), + ) container .bind(TYPES.Revisions_DomainEventSubscriberFactory) diff --git a/packages/syncing-server/bin/worker.ts b/packages/syncing-server/bin/worker.ts index 3334f5736..baf811fa2 100644 --- a/packages/syncing-server/bin/worker.ts +++ b/packages/syncing-server/bin/worker.ts @@ -6,6 +6,7 @@ import TYPES from '../src/Bootstrap/Types' import { Env } from '../src/Bootstrap/Env' import { DomainEventSubscriberFactoryInterface } from '@standardnotes/domain-events' import { ContainerConfigLoader } from '../src/Bootstrap/Container' +import { OpenTelemetrySDKInterface } from '@standardnotes/domain-events-infra' const container = new ContainerConfigLoader('worker') void container.load().then((container) => { @@ -16,6 +17,11 @@ void container.load().then((container) => { logger.info('Starting worker...') + if (!container.get(TYPES.Sync_IS_CONFIGURED_FOR_HOME_SERVER_OR_SELF_HOSTING)) { + const openTelemetrySDK = container.get(TYPES.Sync_OpenTelemetrySDK) + openTelemetrySDK.start() + } + const subscriberFactory: DomainEventSubscriberFactoryInterface = container.get( TYPES.Sync_DomainEventSubscriberFactory, ) diff --git a/packages/syncing-server/src/Bootstrap/Container.ts b/packages/syncing-server/src/Bootstrap/Container.ts index c572647fd..57142353e 100644 --- a/packages/syncing-server/src/Bootstrap/Container.ts +++ b/packages/syncing-server/src/Bootstrap/Container.ts @@ -18,6 +18,7 @@ import { SNSDomainEventPublisher, SQSDomainEventSubscriberFactory, SQSEventMessageHandler, + SQSOpenTelemetryEventMessageHandler, } from '@standardnotes/domain-events-infra' import { DomainEventFactoryInterface } from '../Domain/Event/DomainEventFactoryInterface' import { DomainEventFactory } from '../Domain/Event/DomainEventFactory' @@ -244,7 +245,13 @@ export class ContainerConfigLoader { if (!isConfiguredForHomeServerOrSelfHosting) { container .bind(TYPES.Sync_OpenTelemetrySDK) - .toConstantValue(new OpenTelemetrySDK(ServiceIdentifier.NAMES.SyncingServer)) + .toConstantValue( + new OpenTelemetrySDK( + this.mode === 'server' + ? ServiceIdentifier.NAMES.SyncingServer + : ServiceIdentifier.NAMES.SyncingServerWorker, + ), + ) } if (!isConfiguredForInMemoryCache) { @@ -1168,7 +1175,11 @@ export class ContainerConfigLoader { } else { container .bind(TYPES.Sync_DomainEventMessageHandler) - .toConstantValue(new SQSEventMessageHandler(eventHandlers, container.get(TYPES.Sync_Logger))) + .toConstantValue( + isConfiguredForHomeServerOrSelfHosting + ? new SQSEventMessageHandler(eventHandlers, container.get(TYPES.Sync_Logger)) + : new SQSOpenTelemetryEventMessageHandler(eventHandlers, container.get(TYPES.Sync_Logger)), + ) } container diff --git a/packages/websockets/bin/worker.ts b/packages/websockets/bin/worker.ts index ac53b23e6..3f1ded84e 100644 --- a/packages/websockets/bin/worker.ts +++ b/packages/websockets/bin/worker.ts @@ -6,6 +6,7 @@ import { ContainerConfigLoader } from '../src/Bootstrap/Container' import TYPES from '../src/Bootstrap/Types' import { Env } from '../src/Bootstrap/Env' import { DomainEventSubscriberFactoryInterface } from '@standardnotes/domain-events' +import { OpenTelemetrySDKInterface } from '@standardnotes/domain-events-infra' const container = new ContainerConfigLoader() void container.load().then((container) => { @@ -16,6 +17,9 @@ void container.load().then((container) => { logger.info('Starting worker...') + const openTelemetrySDK = container.get(TYPES.WebSockets_OpenTelemetrySDK) + openTelemetrySDK.start() + const subscriberFactory: DomainEventSubscriberFactoryInterface = container.get(TYPES.DomainEventSubscriberFactory) subscriberFactory.create().start() }) diff --git a/packages/websockets/src/Bootstrap/Container.ts b/packages/websockets/src/Bootstrap/Container.ts index dc5e067cb..c093cc5c0 100644 --- a/packages/websockets/src/Bootstrap/Container.ts +++ b/packages/websockets/src/Bootstrap/Container.ts @@ -22,7 +22,7 @@ import { OpenTelemetrySDK, OpenTelemetrySDKInterface, SQSDomainEventSubscriberFactory, - SQSEventMessageHandler, + SQSOpenTelemetryEventMessageHandler, } from '@standardnotes/domain-events-infra' import { ApiGatewayAuthMiddleware } from '../Controller/ApiGatewayAuthMiddleware' @@ -42,6 +42,8 @@ import { WebSocketMessageRequestedEventHandler } from '../Domain/Handler/WebSock import { ServiceIdentifier } from '@standardnotes/domain-core' export class ContainerConfigLoader { + constructor(private mode: 'server' | 'worker' = 'server') {} + async load(): Promise { const env: Env = new Env() env.load() @@ -50,7 +52,11 @@ export class ContainerConfigLoader { container .bind(TYPES.WebSockets_OpenTelemetrySDK) - .toConstantValue(new OpenTelemetrySDK(ServiceIdentifier.NAMES.Websockets)) + .toConstantValue( + new OpenTelemetrySDK( + this.mode === 'server' ? ServiceIdentifier.NAMES.Websockets : ServiceIdentifier.NAMES.WebsocketsWorker, + ), + ) const redisUrl = env.get('REDIS_URL') const isRedisInClusterMode = redisUrl.indexOf(',') > 0 @@ -158,7 +164,7 @@ export class ContainerConfigLoader { container .bind(TYPES.DomainEventMessageHandler) - .toConstantValue(new SQSEventMessageHandler(eventHandlers, container.get(TYPES.Logger))) + .toConstantValue(new SQSOpenTelemetryEventMessageHandler(eventHandlers, container.get(TYPES.Logger))) container .bind(TYPES.DomainEventSubscriberFactory) .toConstantValue(