From 65ced2cc7b0686dc8af5cdad499412fc8fd29d1d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Karol=20S=C3=B3jko?= Date: Mon, 9 Oct 2023 14:08:27 +0200 Subject: [PATCH] feat: add custom tracing on workers --- .github/workflows/common-deploy.yml | 4 -- ...ry-api-npm-1.6.0-58fdf34ce1-b8daefad2c.zip | Bin 410462 -> 410461 bytes packages/auth/bin/worker.ts | 6 ++ packages/auth/src/Bootstrap/Container.ts | 13 +++- .../SQSOpenTelemetryEventMessageHandler.ts | 56 ++++++++++++++++++ .../domain-events-infra/src/Infra/index.ts | 1 + packages/files/bin/worker.ts | 8 ++- packages/files/src/Bootstrap/Container.ts | 15 ++++- packages/revisions/bin/worker.ts | 6 ++ packages/revisions/src/Bootstrap/Container.ts | 13 +++- packages/syncing-server/bin/worker.ts | 6 ++ .../syncing-server/src/Bootstrap/Container.ts | 15 ++++- packages/websockets/bin/worker.ts | 4 ++ .../websockets/src/Bootstrap/Container.ts | 12 +++- 14 files changed, 143 insertions(+), 16 deletions(-) create mode 100644 packages/domain-events-infra/src/Infra/SQS/SQSOpenTelemetryEventMessageHandler.ts 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 77101aa631631d2df7a48b1fea2afd3b775a88f8..21a6a9bf5606056f3e084c336318847c1566af2c 100644 GIT binary patch delta 1150 zcmccDCV96_vY~~sg{g(Pg=GutK|j93(VpRJF5D`yWMW{DXPdn7jL!58mzX%F`~PEQ zn|{EOjir8Sh;R072Z6n>!+E|4MDMtCciEDKQxY;(dR~e`zN(|B2|@ShKm)=lqWl=NG$HR9-SIa^Gik^S9{vh0HP2(&{q;H{=zuTArFK7Tmjk za{sxbTs!)n>8wl(Wbd!1;;T>5z(|BB!X zs)AOHcYf{NbYWfKrG-@=(w#I<8gfkKSoU^7tJD6%f*_ultll&2mu?vgJgaYG<=iZ~ zAho4=Q;%%=b(2%I$~Ly`uKv5`z5QFnTm4>Ft336^!u*dN!N$(*Cta-EXLjm6&w8J= zV%{#xpZ~*REJ7bhMV#1C!zwBy{XXyai3`Vf@bgFf;QGJtN_I-y)q7^{@s*Re@@%%w z;w|EP!T-5}d*8k%Y=25u=}PL)V6E)Je!HTVb zF>3lnE4D%=wb1E_R&18j<-Rd4P2XU{7Rag)4wv6h~f{tY&_Hb(}14cXUC?+#F_~aRuBLR%h|J;Fx|<9h$L{caZcCD z0V?db2PzaTntsud&1Cul7B;r&yFez&l!6rMOfO($VxOK<%E~?clmpOdud`VNco?8y zN#o~os47;V`?~Fc?z66A6`20YfsKnvrw(Ej*i8;Q8QG>AI06mcQV$V6Fqu(!dO;J= z1r1I>OBIaRxDc8dCrrQK$d<_@(F9Tq3Lj3mS}mq8U{Q_94l`J$+aG1*nEt>CXz|}G8EZdg{vL1Rq{edDI%XD^2HmT|DmTYNE?t#-UTCrJ9-~W%5 z1EfrkDKKdIMME~D=?;fk+2O(>(=&ou1-KcYU`b>2^b1yOg-q(9aK!@Cf2y#FOrL4Z zCdKqQ3}mXs^taY*`b-{C5H$fdY^F?eW2Y+`15ICV3shUc$i_O|7sSf8VUuE9H=WUz zt%_-XB3y|t)BR+KW&=AmU#9=5(;4;Hbfzb;fP9+<^w?fIHZ3N$OtA1|haisWvi58y zOn0*(>IB|1vQBRSiHH?}jM16?A&rq^I{!vSw&^?VfPR%Ng$M^Yu-PzuDF<^Ve|X9| z-Jt?#c%MC+z;vrRRskM_zjf;%MlU$Y$~Rr#5opHNdbo2rr{^~T_0>DEu}zmZV&g*B zJh1`E6%QJKO8K0C?w4!=2^&pY%7`#Pi|K0<#3uWrKqudK0@^gK8KQl|X<%GAJF|(f KnYXY4ody7?ucZS3 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(