diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 80be6b984d..010ae33852 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1608,18 +1608,30 @@ importers: '@crowd/database': specifier: workspace:* version: link:../../libs/database + '@crowd/logging': + specifier: workspace:* + version: link:../../libs/logging '@crowd/opensearch': specifier: workspace:* version: link:../../libs/opensearch '@crowd/queue': specifier: workspace:* version: link:../../libs/queue + '@crowd/telemetry': + specifier: workspace:* + version: link:../../libs/telemetry '@crowd/temporal': specifier: workspace:* version: link:../../libs/temporal + '@temporalio/activity': + specifier: ~1.11.1 + version: 1.11.1 '@temporalio/worker': specifier: ~1.11.1 version: 1.11.1 + '@temporalio/workflow': + specifier: ~1.11.1 + version: 1.11.1 devDependencies: '@types/config': specifier: ^3.3.1 diff --git a/services/archetypes/worker/package.json b/services/archetypes/worker/package.json index fa675430f9..0e7e63a5b4 100644 --- a/services/archetypes/worker/package.json +++ b/services/archetypes/worker/package.json @@ -14,7 +14,11 @@ "@crowd/opensearch": "workspace:*", "@crowd/queue": "workspace:*", "@crowd/temporal": "workspace:*", - "@temporalio/worker": "~1.11.1" + "@crowd/telemetry": "workspace:*", + "@crowd/logging": "workspace:*", + "@temporalio/worker": "~1.11.1", + "@temporalio/activity": "~1.11.1", + "@temporalio/workflow": "~1.11.1" }, "devDependencies": { "@types/config": "^3.3.1", diff --git a/services/archetypes/worker/src/activities/activityInterceptor.ts b/services/archetypes/worker/src/activities/activityInterceptor.ts new file mode 100644 index 0000000000..06e4e98844 --- /dev/null +++ b/services/archetypes/worker/src/activities/activityInterceptor.ts @@ -0,0 +1,40 @@ +import { Context } from '@temporalio/activity' +import { ActivityExecuteInput, ActivityInboundCallsInterceptor, Next } from '@temporalio/worker' + +import { getServiceChildLogger } from '@crowd/logging' +import telemetry from '@crowd/telemetry' + +export class ActivityMonitoringInterceptor implements ActivityInboundCallsInterceptor { + public constructor(private readonly ctx: Context) {} + async execute( + input: ActivityExecuteInput, + next: Next, + ): Promise { + const tags = { + workflow_type: this.ctx.info.workflowType, + task_queue: this.ctx.info.taskQueue, + activity_type: this.ctx.info.activityType, + } + + const timer = telemetry.timer('temporal.activity_execution_duration', tags) + + try { + const res = await next(input) + return res + } catch (err) { + const log = getServiceChildLogger('activity-interceptor', { + activityType: this.ctx.info.activityType, + activityId: this.ctx.info.activityId, + workflowType: this.ctx.info.workflowType, + taskQueue: this.ctx.info.taskQueue, + workflowId: this.ctx.info.workflowExecution.workflowId, + runId: this.ctx.info.workflowExecution.runId, + }) + + log.error(err, 'Error while processing an activity!') + throw err + } finally { + timer.stop() + } + } +} diff --git a/services/archetypes/worker/src/activities/index.ts b/services/archetypes/worker/src/activities/index.ts new file mode 100644 index 0000000000..bfe4530041 --- /dev/null +++ b/services/archetypes/worker/src/activities/index.ts @@ -0,0 +1,11 @@ +import telemetry from '@crowd/telemetry' + +async function telemetryDistribution( + name: string, + value: number, + tags?: Record, +) { + telemetry.distribution(name, value, tags) +} + +export { telemetryDistribution } diff --git a/services/archetypes/worker/src/index.ts b/services/archetypes/worker/src/index.ts index 6b647be16c..79b998c755 100644 --- a/services/archetypes/worker/src/index.ts +++ b/services/archetypes/worker/src/index.ts @@ -1,279 +1,2 @@ -import { NativeConnection, Worker as TemporalWorker, bundleWorkflowCode } from '@temporalio/worker' -import path from 'path' - -import { Config, Service } from '@crowd/archetype-standard' -import { IS_DEV_ENV, IS_STAGING_ENV, IS_TEST_ENV } from '@crowd/common' -import { DbStore, getDbConnection } from '@crowd/database' -import { OpenSearchService, getOpensearchClient } from '@crowd/opensearch' -import { IQueue, QueueFactory } from '@crowd/queue' -import { getDataConverter } from '@crowd/temporal' - -// List all required environment variables, grouped per "component". -// They are in addition to the ones required by the "standard" archetype. -const envvars = { - worker: [ - 'CROWD_TEMPORAL_SERVER_URL', - 'CROWD_TEMPORAL_NAMESPACE', - 'CROWD_TEMPORAL_TASKQUEUE', - 'CROWD_TEMPORAL_ENCRYPTION_KEY_ID', - 'CROWD_TEMPORAL_ENCRYPTION_KEY', - ], - postgres: [ - 'CROWD_DB_READ_HOST', - 'CROWD_DB_WRITE_HOST', - 'CROWD_DB_PORT', - 'CROWD_DB_USERNAME', - 'CROWD_DB_PASSWORD', - 'CROWD_DB_DATABASE', - ], - opensearch: - IS_DEV_ENV || IS_TEST_ENV || IS_STAGING_ENV - ? ['CROWD_OPENSEARCH_NODE'] - : ['CROWD_OPENSEARCH_USERNAME', 'CROWD_OPENSEARCH_PASSWORD', 'CROWD_OPENSEARCH_NODE'], - queue: ['CROWD_KAFKA_BROKERS'], -} - -/* -Options is used to configure the worker service. -*/ -export interface Options { - maxTaskQueueActivitiesPerSecond?: number - maxConcurrentActivityTaskExecutions?: number - postgres?: { - enabled: boolean - } - opensearch?: { - enabled: boolean - } - queue?: { - enabled: boolean - } -} - -/* -ServiceWorker holds all details and methods to run a worker services at crowd.dev. -*/ -export class ServiceWorker extends Service { - readonly options: Options - - protected _worker: TemporalWorker - - protected _postgresReader: DbStore - protected _postgresWriter: DbStore - - protected _opensearchService: OpenSearchService - - protected _queueClient: IQueue - - constructor(config: Config, opts: Options) { - super(config) - - this.options = opts - } - - get postgres(): { reader: DbStore; writer: DbStore } | null { - if (!this.options.postgres?.enabled) { - return null - } - - return { - reader: this._postgresReader, - writer: this._postgresWriter, - } - } - - get opensearch(): OpenSearchService { - if (!this.options.opensearch?.enabled) { - return null - } - - return this._opensearchService - } - - get queue(): IQueue { - if (!this.options.queue?.enabled) { - return null - } - - return this._queueClient - } - - // We first need to ensure a standard service can be initialized given the config - // and environment variables. - override async init(initWorker = true) { - try { - await super.init() - } catch (err) { - throw new Error(err) - } - - // We can now init tasks specific to a consumer service. Before actually - // starting the service, we need to ensure required environment variables - // are set. - const missing = [] - envvars.worker.forEach((envvar) => { - if (!process.env[envvar]) { - missing.push(envvar) - } - }) - - // Only validate PostgreSQL-related environment variables if enabled. - if (this.options.postgres?.enabled) { - envvars.postgres.forEach((envvar) => { - if (!process.env[envvar]) { - missing.push(envvar) - } - }) - } - - // Only validate OpenSearch-related environment variables if enabled. - if (this.options.opensearch?.enabled) { - envvars.opensearch.forEach((envvar) => { - if (!process.env[envvar]) { - missing.push(envvar) - } - }) - } - - // Only validate queue related environment variables if enabled - if (this.options.queue?.enabled) { - envvars.queue.forEach((envvar) => { - if (!process.env[envvar]) { - missing.push(envvar) - } - }) - } - - // There's no point in continuing if a variable is missing. - if (missing.length > 0) { - throw new Error(`Missing environment variables: ${missing.join(', ')}`) - } - - if (this.options.postgres?.enabled) { - try { - const dbConnection = await getDbConnection({ - host: process.env['CROWD_DB_READ_HOST'], - port: Number(process.env['CROWD_DB_PORT']), - user: process.env['CROWD_DB_USERNAME'], - password: process.env['CROWD_DB_PASSWORD'], - database: process.env['CROWD_DB_DATABASE'], - }) - - this._postgresReader = new DbStore(this.log, dbConnection) - } catch (err) { - throw new Error(err) - } - - try { - const dbConnection = await getDbConnection({ - host: process.env['CROWD_DB_WRITE_HOST'], - port: Number(process.env['CROWD_DB_PORT']), - user: process.env['CROWD_DB_USERNAME'], - password: process.env['CROWD_DB_PASSWORD'], - database: process.env['CROWD_DB_DATABASE'], - }) - - this._postgresWriter = new DbStore(this.log, dbConnection) - } catch (err) { - throw new Error(err) - } - } - - if (this.options.opensearch?.enabled) { - const osClient = await getOpensearchClient({ - username: process.env['CROWD_OPENSEARCH_USERNAME'], - password: process.env['CROWD_OPENSEARCH_PASSWORD'], - node: process.env['CROWD_OPENSEARCH_NODE'], - }) - try { - this._opensearchService = new OpenSearchService(this.log, osClient) - } catch (err) { - throw new Error(err) - } - } - - if (this.options.queue?.enabled) { - try { - this._queueClient = QueueFactory.createQueueService({ - brokers: process.env['CROWD_KAFKA_BROKERS'], - clientId: process.env['SERVICE'], //TODO:: make this configurable - }) - } catch (err) { - throw new Error(err) - } - } - - if (initWorker) { - try { - const certificate = process.env['CROWD_TEMPORAL_CERTIFICATE'] - const privateKey = process.env['CROWD_TEMPORAL_PRIVATE_KEY'] - - this.log.info( - { - address: process.env['CROWD_TEMPORAL_SERVER_URL'], - certificate: certificate ? 'yes' : 'no', - privateKey: privateKey ? 'yes' : 'no', - }, - 'Connecting to Temporal server as a worker!', - ) - - const connection = await NativeConnection.connect({ - address: process.env['CROWD_TEMPORAL_SERVER_URL'], - tls: - certificate && privateKey - ? { - clientCertPair: { - crt: Buffer.from(certificate, 'base64'), - key: Buffer.from(privateKey, 'base64'), - }, - } - : undefined, - }) - - const workflowBundle = await bundleWorkflowCode({ - workflowsPath: path.resolve('./src/workflows'), - }) - - this._worker = await TemporalWorker.create({ - connection: connection, - identity: this.name, - namespace: process.env['CROWD_TEMPORAL_NAMESPACE'], - taskQueue: process.env['CROWD_TEMPORAL_TASKQUEUE'], - enableSDKTracing: true, - showStackTraceSources: true, - workflowBundle: workflowBundle, - activities: require(path.resolve('./src/activities')), - dataConverter: await getDataConverter(), - maxTaskQueueActivitiesPerSecond: this.options.maxTaskQueueActivitiesPerSecond, - maxConcurrentActivityTaskExecutions: this.options.maxConcurrentActivityTaskExecutions, - }) - } catch (err) { - throw new Error(err) - } - } - } - - // Actually start the Temporal worker. - async start() { - try { - await this._worker.run() - } catch (err) { - throw new Error(err) - } - } - - // Stop allows to gracefully stop the service. Order for closing connections - // matters. We need to stop the Temporal worker before closing other connections. - protected override async stop() { - if (this.options.opensearch?.enabled) { - await this._opensearchService.client.close() - } - - if (this.options.postgres?.enabled) { - this._postgresWriter.dbInstance.end() - this._postgresReader.dbInstance.end() - } - - await super.stop() - } -} +export * from './worker' +export * from './interceptors' diff --git a/services/archetypes/worker/src/interceptors.ts b/services/archetypes/worker/src/interceptors.ts new file mode 100644 index 0000000000..d4e569b9df --- /dev/null +++ b/services/archetypes/worker/src/interceptors.ts @@ -0,0 +1,38 @@ +import { + Next, + WorkflowExecuteInput, + WorkflowInboundCallsInterceptor, + proxyActivities, + workflowInfo, +} from '@temporalio/workflow' + +import * as activities from './activities' + +const activity = proxyActivities({ + startToCloseTimeout: '10 seconds', +}) + +export class WorkflowMonitoringInterceptor implements WorkflowInboundCallsInterceptor { + async execute( + input: WorkflowExecuteInput, + next: Next, + ): Promise { + const info = workflowInfo() + + const tags = { + workflow_type: info.workflowType, + task_queue: info.taskQueue, + } + + const start = new Date() + + try { + const result = await next(input) + return result + } finally { + const end = new Date() + const duration = end.getTime() - start.getTime() + await activity.telemetryDistribution('temporal.workflow_execution_duration', duration, tags) + } + } +} diff --git a/services/archetypes/worker/src/logging.ts b/services/archetypes/worker/src/logging.ts new file mode 100644 index 0000000000..12d507acbf --- /dev/null +++ b/services/archetypes/worker/src/logging.ts @@ -0,0 +1,64 @@ +import { LogLevel as TemporalLogLevel, Logger as TemporalLogger } from '@temporalio/worker' + +import { LogLevel, Logger } from '@crowd/logging' + +export function createTemporalLogger(base: Logger): TemporalLogger { + return { + log(level: TemporalLogLevel, message: string, meta?: Record) { + const bunyanLevel = { + ['TRACE']: 'trace', + ['DEBUG']: 'debug', + ['INFO']: 'info', + ['WARN']: 'warn', + ['ERROR']: 'error', + }[level] as LogLevel + + if (meta) { + base[bunyanLevel](meta, message) + } else { + base[bunyanLevel](message) + } + }, + + // Map Temporal log levels to Bunyan levels + trace(message: string, meta?: Record) { + if (meta) { + base.trace(meta, message) + } else { + base.trace(message) + } + }, + + debug(message: string, meta?: Record) { + if (meta) { + base.debug(meta, message) + } else { + base.debug(message) + } + }, + + info(message: string, meta?: Record) { + if (meta) { + base.info(meta, message) + } else { + base.info(message) + } + }, + + warn(message: string, meta?: Record) { + if (meta) { + base.warn(meta, message) + } else { + base.warn(message) + } + }, + + error(message: string, meta?: Record) { + if (meta) { + base.error(meta, message) + } else { + base.error(message) + } + }, + } +} diff --git a/services/archetypes/worker/src/worker.ts b/services/archetypes/worker/src/worker.ts new file mode 100644 index 0000000000..73220f90b5 --- /dev/null +++ b/services/archetypes/worker/src/worker.ts @@ -0,0 +1,324 @@ +import { + NativeConnection, + Runtime, + Worker as TemporalWorker, + bundleWorkflowCode, + makeTelemetryFilterString, +} from '@temporalio/worker' +import fs from 'fs' +import path from 'path' + +import { Config, Service } from '@crowd/archetype-standard' +import { IS_DEV_ENV, IS_STAGING_ENV, IS_TEST_ENV } from '@crowd/common' +import { DbStore, getDbConnection } from '@crowd/database' +import { getServiceChildLogger } from '@crowd/logging' +import { OpenSearchService, getOpensearchClient } from '@crowd/opensearch' +import { IQueue, QueueFactory } from '@crowd/queue' +import { getDataConverter } from '@crowd/temporal' + +import * as metricActivities from './activities' +import { ActivityMonitoringInterceptor } from './activities/activityInterceptor' +import { createTemporalLogger } from './logging' + +// List all required environment variables, grouped per "component". +// They are in addition to the ones required by the "standard" archetype. +const envvars = { + worker: [ + 'CROWD_TEMPORAL_SERVER_URL', + 'CROWD_TEMPORAL_NAMESPACE', + 'CROWD_TEMPORAL_TASKQUEUE', + 'CROWD_TEMPORAL_ENCRYPTION_KEY_ID', + 'CROWD_TEMPORAL_ENCRYPTION_KEY', + ], + postgres: [ + 'CROWD_DB_READ_HOST', + 'CROWD_DB_WRITE_HOST', + 'CROWD_DB_PORT', + 'CROWD_DB_USERNAME', + 'CROWD_DB_PASSWORD', + 'CROWD_DB_DATABASE', + ], + opensearch: + IS_DEV_ENV || IS_TEST_ENV || IS_STAGING_ENV + ? ['CROWD_OPENSEARCH_NODE'] + : ['CROWD_OPENSEARCH_USERNAME', 'CROWD_OPENSEARCH_PASSWORD', 'CROWD_OPENSEARCH_NODE'], + queue: ['CROWD_KAFKA_BROKERS'], +} + +/* +Options is used to configure the worker service. +*/ +export interface Options { + maxTaskQueueActivitiesPerSecond?: number + maxConcurrentActivityTaskExecutions?: number + postgres?: { + enabled: boolean + } + opensearch?: { + enabled: boolean + } + queue?: { + enabled: boolean + } +} + +/* +ServiceWorker holds all details and methods to run a worker services at crowd.dev. +*/ +export class ServiceWorker extends Service { + readonly options: Options + + protected _worker: TemporalWorker + + protected _postgresReader: DbStore + protected _postgresWriter: DbStore + + protected _opensearchService: OpenSearchService + + protected _queueClient: IQueue + + constructor(config: Config, opts: Options) { + super(config) + + this.options = opts + } + + get postgres(): { reader: DbStore; writer: DbStore } | null { + if (!this.options.postgres?.enabled) { + return null + } + + return { + reader: this._postgresReader, + writer: this._postgresWriter, + } + } + + get opensearch(): OpenSearchService { + if (!this.options.opensearch?.enabled) { + return null + } + + return this._opensearchService + } + + get queue(): IQueue { + if (!this.options.queue?.enabled) { + return null + } + + return this._queueClient + } + + // We first need to ensure a standard service can be initialized given the config + // and environment variables. + override async init(initWorker = true) { + try { + await super.init() + } catch (err) { + throw new Error(err) + } + + // We can now init tasks specific to a consumer service. Before actually + // starting the service, we need to ensure required environment variables + // are set. + const missing = [] + envvars.worker.forEach((envvar) => { + if (!process.env[envvar]) { + missing.push(envvar) + } + }) + + // Only validate PostgreSQL-related environment variables if enabled. + if (this.options.postgres?.enabled) { + envvars.postgres.forEach((envvar) => { + if (!process.env[envvar]) { + missing.push(envvar) + } + }) + } + + // Only validate OpenSearch-related environment variables if enabled. + if (this.options.opensearch?.enabled) { + envvars.opensearch.forEach((envvar) => { + if (!process.env[envvar]) { + missing.push(envvar) + } + }) + } + + // Only validate queue related environment variables if enabled + if (this.options.queue?.enabled) { + envvars.queue.forEach((envvar) => { + if (!process.env[envvar]) { + missing.push(envvar) + } + }) + } + + // There's no point in continuing if a variable is missing. + if (missing.length > 0) { + throw new Error(`Missing environment variables: ${missing.join(', ')}`) + } + + if (this.options.postgres?.enabled) { + try { + const dbConnection = await getDbConnection({ + host: process.env['CROWD_DB_READ_HOST'], + port: Number(process.env['CROWD_DB_PORT']), + user: process.env['CROWD_DB_USERNAME'], + password: process.env['CROWD_DB_PASSWORD'], + database: process.env['CROWD_DB_DATABASE'], + }) + + this._postgresReader = new DbStore(this.log, dbConnection) + } catch (err) { + throw new Error(err) + } + + try { + const dbConnection = await getDbConnection({ + host: process.env['CROWD_DB_WRITE_HOST'], + port: Number(process.env['CROWD_DB_PORT']), + user: process.env['CROWD_DB_USERNAME'], + password: process.env['CROWD_DB_PASSWORD'], + database: process.env['CROWD_DB_DATABASE'], + }) + + this._postgresWriter = new DbStore(this.log, dbConnection) + } catch (err) { + throw new Error(err) + } + } + + if (this.options.opensearch?.enabled) { + const osClient = await getOpensearchClient({ + username: process.env['CROWD_OPENSEARCH_USERNAME'], + password: process.env['CROWD_OPENSEARCH_PASSWORD'], + node: process.env['CROWD_OPENSEARCH_NODE'], + }) + try { + this._opensearchService = new OpenSearchService(this.log, osClient) + } catch (err) { + throw new Error(err) + } + } + + if (this.options.queue?.enabled) { + try { + this._queueClient = QueueFactory.createQueueService({ + brokers: process.env['CROWD_KAFKA_BROKERS'], + clientId: process.env['SERVICE'], //TODO:: make this configurable + }) + } catch (err) { + throw new Error(err) + } + } + + if (initWorker) { + try { + Runtime.install({ + logger: createTemporalLogger(getServiceChildLogger('temporal-worker')), + telemetryOptions: { + logging: { + forward: {}, + filter: makeTelemetryFilterString({ core: 'INFO' }), + }, + }, + }) + + const certificate = process.env['CROWD_TEMPORAL_CERTIFICATE'] + const privateKey = process.env['CROWD_TEMPORAL_PRIVATE_KEY'] + + this.log.info( + { + address: process.env['CROWD_TEMPORAL_SERVER_URL'], + certificate: certificate ? 'yes' : 'no', + privateKey: privateKey ? 'yes' : 'no', + }, + 'Connecting to Temporal server as a worker!', + ) + + const connection = await NativeConnection.connect({ + address: process.env['CROWD_TEMPORAL_SERVER_URL'], + tls: + certificate && privateKey + ? { + clientCertPair: { + crt: Buffer.from(certificate, 'base64'), + key: Buffer.from(privateKey, 'base64'), + }, + } + : undefined, + }) + + const workflowInterceptorModules = [path.join(__dirname, 'workflowInterceptors')] + const serviceInterceptorsPath = path.resolve('./src/workflows/interceptors') + if (fs.existsSync(serviceInterceptorsPath)) { + workflowInterceptorModules.push(serviceInterceptorsPath) + } + + const workflowBundle = await bundleWorkflowCode({ + workflowsPath: path.resolve('./src/workflows'), + workflowInterceptorModules, + }) + + const dataConverter = await getDataConverter() + + this._worker = await TemporalWorker.create({ + connection: connection, + identity: this.name, + namespace: process.env['CROWD_TEMPORAL_NAMESPACE'], + taskQueue: process.env['CROWD_TEMPORAL_TASKQUEUE'], + enableSDKTracing: true, + showStackTraceSources: true, + workflowBundle, + activities: { + ...metricActivities, + ...require(path.resolve('./src/activities')), + }, + interceptors: { + activity: [ + (ctx) => { + return { + inbound: new ActivityMonitoringInterceptor(ctx), + } + }, + ], + }, + dataConverter: { + ...dataConverter, + }, + maxTaskQueueActivitiesPerSecond: this.options.maxTaskQueueActivitiesPerSecond, + maxConcurrentActivityTaskExecutions: this.options.maxConcurrentActivityTaskExecutions, + }) + } catch (err) { + throw new Error(err) + } + } + } + + // Actually start the Temporal worker. + async start() { + try { + await this._worker.run() + } catch (err) { + throw new Error(err) + } + } + + // Stop allows to gracefully stop the service. Order for closing connections + // matters. We need to stop the Temporal worker before closing other connections. + protected override async stop() { + if (this.options.opensearch?.enabled) { + await this._opensearchService.client.close() + } + + if (this.options.postgres?.enabled) { + this._postgresWriter.dbInstance.end() + this._postgresReader.dbInstance.end() + } + + await super.stop() + } +} diff --git a/services/archetypes/worker/src/workflowInterceptors/index.ts b/services/archetypes/worker/src/workflowInterceptors/index.ts new file mode 100644 index 0000000000..ef15a6a8b6 --- /dev/null +++ b/services/archetypes/worker/src/workflowInterceptors/index.ts @@ -0,0 +1,7 @@ +import { WorkflowInterceptorsFactory } from '@temporalio/workflow' + +import { WorkflowMonitoringInterceptor } from '../interceptors' + +export const interceptors: WorkflowInterceptorsFactory = () => ({ + inbound: [new WorkflowMonitoringInterceptor()], +}) diff --git a/services/libs/logging/src/types.ts b/services/libs/logging/src/types.ts index caf0a9ac34..4d6345fe44 100644 --- a/services/libs/logging/src/types.ts +++ b/services/libs/logging/src/types.ts @@ -1,3 +1,4 @@ import * as Bunyan from 'bunyan' export type Logger = Bunyan +export type LogLevel = Bunyan.LogLevel