Remove OTEL tracing from our code

backend/src/api/index.ts

@@ -3,7 +3,6 @@ import { getUnleashClient } from '@crowd/feature-flags'
 import { getServiceLogger } from '@crowd/logging'
 import { getOpensearchClient } from '@crowd/opensearch'
 import { getRedisClient, getRedisPubSubPair, RedisPubSubReceiver } from '@crowd/redis'
-import { getServiceTracer } from '@crowd/tracing'
 import { ApiWebsocketMessage, Edition } from '@crowd/types'
 import bodyParser from 'body-parser'
 import bunyanMiddleware from 'bunyan-middleware'
@@ -37,7 +36,6 @@ import WebSockets from './websockets'
 import { databaseInit } from '@/database/databaseConnection'
 
 const serviceLogger = getServiceLogger()
-getServiceTracer()
 
 const app = express()
 

backend/src/bin/discord-ws.ts

@@ -4,7 +4,6 @@ import { processPaginated, timeout } from '@crowd/common'
 import { RedisCache, getRedisClient } from '@crowd/redis'
 import { getChildLogger, getServiceLogger } from '@crowd/logging'
 import { PlatformType } from '@crowd/types'
-import { SpanStatusCode, getServiceTracer } from '@crowd/tracing'
 import { DISCORD_CONFIG, REDIS_CONFIG } from '../conf'
 import SequelizeRepository from '../database/repositories/sequelizeRepository'
 import IntegrationRepository from '../database/repositories/integrationRepository'
@@ -15,7 +14,6 @@ import {
   getIntegrationStreamWorkerEmitter,
 } from '@/serverless/utils/serviceSQS'
 
-const tracer = getServiceTracer()
 const log = getServiceLogger()
 
 async function executeIfNotExists(
@@ -58,51 +56,40 @@ async function spawnClient(
 
     logger.info({ payload }, 'Processing Discord WS Message!')
 
-    await tracer.startActiveSpan('ProcessDiscordWSMessage', async (span) => {
-      try {
-        const integration = (await IntegrationRepository.findByIdentifier(
-          guildId,
-          PlatformType.DISCORD,
-        )) as any
+    try {
+      const integration = (await IntegrationRepository.findByIdentifier(
+        guildId,
+        PlatformType.DISCORD,
+      )) as any
 
-        const result = await repo.create({
-          tenantId: integration.tenantId,
-          integrationId: integration.id,
-          type: WebhookType.DISCORD,
-          payload,
-        })
+      const result = await repo.create({
+        tenantId: integration.tenantId,
+        integrationId: integration.id,
+        type: WebhookType.DISCORD,
+        payload,
+      })
 
-        const streamEmitter = await getIntegrationStreamWorkerEmitter()
+      const streamEmitter = await getIntegrationStreamWorkerEmitter()
 
-        await streamEmitter.triggerWebhookProcessing(
-          integration.tenantId,
-          integration.platform,
-          result.id,
+      await streamEmitter.triggerWebhookProcessing(
+        integration.tenantId,
+        integration.platform,
+        result.id,
+      )
+    } catch (err) {
+      if (err.code === 404) {
+        logger.warn({ guildId }, 'No integration found for incoming Discord WS Message!')
+      } else {
+        logger.error(
+          err,
+          {
+            discordPayload: JSON.stringify(payload),
+            guildId,
+          },
+          'Error processing Discord WS Message!',
         )
-        span.setStatus({
-          code: SpanStatusCode.OK,
-        })
-      } catch (err) {
-        if (err.code === 404) {
-          logger.warn({ guildId }, 'No integration found for incoming Discord WS Message!')
-        } else {
-          span.setStatus({
-            code: SpanStatusCode.ERROR,
-            message: err,
-          })
-          logger.error(
-            err,
-            {
-              discordPayload: JSON.stringify(payload),
-              guildId,
-            },
-            'Error processing Discord WS Message!',
-          )
-        }
-      } finally {
-        span.end()
       }
-    })
+    }
   }
 
   const client = new Client({

backend/src/bin/job-generator.ts

@@ -1,32 +1,19 @@
 import { CronJob } from 'cron'
 import { getServiceLogger } from '@crowd/logging'
-import { SpanStatusCode, getServiceTracer } from '@crowd/tracing'
 import jobs from './jobs'
 
-const tracer = getServiceTracer()
 const log = getServiceLogger()
 
 for (const job of jobs) {
   const cronJob = new CronJob(
     job.cronTime,
     async () => {
-      await tracer.startActiveSpan(`ProcessingJob:${job.name}`, async (span) => {
-        log.info({ job: job.name }, 'Triggering job.')
-        try {
-          await job.onTrigger(log)
-          span.setStatus({
-            code: SpanStatusCode.OK,
-          })
-        } catch (err) {
-          span.setStatus({
-            code: SpanStatusCode.ERROR,
-            message: err,
-          })
-          log.error(err, { job: job.name }, 'Error while executing a job!')
-        } finally {
-          span.end()
-        }
-      })
+      log.info({ job: job.name }, 'Triggering job.')
+      try {
+        await job.onTrigger(log)
+      } catch (err) {
+        log.error(err, { job: job.name }, 'Error while executing a job!')
+      }
     },
     null,
     true,

backend/src/bin/nodejs-worker.ts

@@ -8,7 +8,6 @@ import {
   receiveMessage,
   sendMessage,
 } from '@crowd/sqs'
-import { SpanStatusCode, getServiceTracer } from '@crowd/tracing'
 import moment from 'moment'
 import { SQS_CONFIG } from '../conf'
 import { processDbOperationsMessage } from '../serverless/dbOperations/workDispatcher'
@@ -21,7 +20,6 @@ import { SQS_CLIENT } from '@/serverless/utils/serviceSQS'
 
 /* eslint-disable no-constant-condition */
 
-const tracer = getServiceTracer()
 const serviceLogger = getServiceLogger()
 
 let exiting = false
@@ -70,55 +68,38 @@ async function handleDelayedMessages() {
     const message = await receive(true)
 
     if (message) {
-      await tracer.startActiveSpan('ProcessDelayedMessage', async (span) => {
-        try {
-          const msg: NodeWorkerMessageBase = JSON.parse(message.Body)
-          const messageLogger = getChildLogger('messageHandler', serviceLogger, {
-            messageId: message.MessageId,
-            type: msg.type,
-          })
-
-          if (message.MessageAttributes && message.MessageAttributes.remainingDelaySeconds) {
-            // re-delay
-            const newDelay = parseInt(
-              message.MessageAttributes.remainingDelaySeconds.StringValue,
-              10,
-            )
-            const tenantId = message.MessageAttributes.tenantId.StringValue
-            messageLogger.debug({ newDelay, tenantId }, 'Re-delaying message!')
-            await sendNodeWorkerMessage(tenantId, msg, newDelay)
-          } else {
-            // just emit to the normal queue for processing
-            const tenantId = message.MessageAttributes.tenantId.StringValue
-
-            if (message.MessageAttributes.targetQueueUrl) {
-              const targetQueueUrl = message.MessageAttributes.targetQueueUrl.StringValue
-              messageLogger.debug({ tenantId, targetQueueUrl }, 'Successfully delayed a message!')
-              await sendMessage(SQS_CLIENT(), {
-                QueueUrl: targetQueueUrl,
-                MessageGroupId: tenantId,
-                MessageDeduplicationId: `${tenantId}-${moment().valueOf()}`,
-                MessageBody: JSON.stringify(msg),
-              })
-            } else {
-              messageLogger.debug({ tenantId }, 'Successfully delayed a message!')
-              await sendNodeWorkerMessage(tenantId, msg)
-            }
-          }
+      const msg: NodeWorkerMessageBase = JSON.parse(message.Body)
+      const messageLogger = getChildLogger('messageHandler', serviceLogger, {
+        messageId: message.MessageId,
+        type: msg.type,
+      })
 
-          await removeFromQueue(message.ReceiptHandle, true)
-          span.setStatus({
-            code: SpanStatusCode.OK,
-          })
-        } catch (err) {
-          span.setStatus({
-            code: SpanStatusCode.ERROR,
-            message: err,
+      if (message.MessageAttributes && message.MessageAttributes.remainingDelaySeconds) {
+        // re-delay
+        const newDelay = parseInt(message.MessageAttributes.remainingDelaySeconds.StringValue, 10)
+        const tenantId = message.MessageAttributes.tenantId.StringValue
+        messageLogger.debug({ newDelay, tenantId }, 'Re-delaying message!')
+        await sendNodeWorkerMessage(tenantId, msg, newDelay)
+      } else {
+        // just emit to the normal queue for processing
+        const tenantId = message.MessageAttributes.tenantId.StringValue
+
+        if (message.MessageAttributes.targetQueueUrl) {
+          const targetQueueUrl = message.MessageAttributes.targetQueueUrl.StringValue
+          messageLogger.debug({ tenantId, targetQueueUrl }, 'Successfully delayed a message!')
+          await sendMessage(SQS_CLIENT(), {
+            QueueUrl: targetQueueUrl,
+            MessageGroupId: tenantId,
+            MessageDeduplicationId: `${tenantId}-${moment().valueOf()}`,
+            MessageBody: JSON.stringify(msg),
           })
-        } finally {
-          span.end()
+        } else {
+          messageLogger.debug({ tenantId }, 'Successfully delayed a message!')
+          await sendNodeWorkerMessage(tenantId, msg)
         }
-      })
+      }
+
+      await removeFromQueue(message.ReceiptHandle, true)
     } else {
       delayedHandlerLogger.trace('No message received!')
     }
@@ -143,83 +124,71 @@ async function handleMessages() {
   handlerLogger.info('Listening for messages!')
 
   const processSingleMessage = async (message: SqsMessage): Promise<void> => {
-    await tracer.startActiveSpan('ProcessMessage', async (span) => {
-      const msg: NodeWorkerMessageBase = JSON.parse(message.Body)
-
-      const messageLogger = getChildLogger('messageHandler', serviceLogger, {
-        messageId: message.MessageId,
-        type: msg.type,
-      })
+    const msg: NodeWorkerMessageBase = JSON.parse(message.Body)
 
-      try {
-        if (
-          msg.type === NodeWorkerMessageType.NODE_MICROSERVICE &&
-          (msg as any).service === 'enrich_member_organizations'
-        ) {
-          messageLogger.warn(
-            'Skipping enrich_member_organizations message! Purging the queue because they are not needed anymore!',
-          )
-          await removeFromQueue(message.ReceiptHandle)
-          return
-        }
+    const messageLogger = getChildLogger('messageHandler', serviceLogger, {
+      messageId: message.MessageId,
+      type: msg.type,
+    })
 
-        messageLogger.info(
-          { messageType: msg.type, messagePayload: JSON.stringify(msg) },
-          'Received a new queue message!',
+    try {
+      if (
+        msg.type === NodeWorkerMessageType.NODE_MICROSERVICE &&
+        (msg as any).service === 'enrich_member_organizations'
+      ) {
+        messageLogger.warn(
+          'Skipping enrich_member_organizations message! Purging the queue because they are not needed anymore!',
         )
+        await removeFromQueue(message.ReceiptHandle)
+        return
+      }
 
-        let processFunction: (msg: NodeWorkerMessageBase, logger?: Logger) => Promise<void>
-
-        switch (msg.type) {
-          case NodeWorkerMessageType.INTEGRATION_PROCESS:
-            processFunction = processIntegration
-            break
-          case NodeWorkerMessageType.NODE_MICROSERVICE:
-            processFunction = processNodeMicroserviceMessage
-            break
-          case NodeWorkerMessageType.DB_OPERATIONS:
-            processFunction = processDbOperationsMessage
-            break
-          case NodeWorkerMessageType.PROCESS_WEBHOOK:
-            processFunction = processWebhook
-            break
-
-          default:
-            messageLogger.error('Error while parsing queue message! Invalid type.')
-        }
-
-        if (processFunction) {
-          await logExecutionTimeV2(
-            async () => {
-              // remove the message from the queue as it's about to be processed
-              await removeFromQueue(message.ReceiptHandle)
-              messagesInProgress.set(message.MessageId, msg)
-              try {
-                await processFunction(msg, messageLogger)
-              } catch (err) {
-                messageLogger.error(err, 'Error while processing queue message!')
-              } finally {
-                messagesInProgress.delete(message.MessageId)
-              }
-            },
-            messageLogger,
-            'Processing queue message!',
-          )
-        }
+      messageLogger.info(
+        { messageType: msg.type, messagePayload: JSON.stringify(msg) },
+        'Received a new queue message!',
+      )
+
+      let processFunction: (msg: NodeWorkerMessageBase, logger?: Logger) => Promise<void>
+
+      switch (msg.type) {
+        case NodeWorkerMessageType.INTEGRATION_PROCESS:
+          processFunction = processIntegration
+          break
+        case NodeWorkerMessageType.NODE_MICROSERVICE:
+          processFunction = processNodeMicroserviceMessage
+          break
+        case NodeWorkerMessageType.DB_OPERATIONS:
+          processFunction = processDbOperationsMessage
+          break
+        case NodeWorkerMessageType.PROCESS_WEBHOOK:
+          processFunction = processWebhook
+          break
+
+        default:
+          messageLogger.error('Error while parsing queue message! Invalid type.')
+      }
 
-        span.setStatus({
-          code: SpanStatusCode.OK,
-        })
-      } catch (err) {
-        span.setStatus({
-          code: SpanStatusCode.ERROR,
-          message: err,
-        })
-        messageLogger.error(err, { payload: msg }, 'Error while processing queue message!')
-      } finally {
-        span.end()
+      if (processFunction) {
+        await logExecutionTimeV2(
+          async () => {
+            // remove the message from the queue as it's about to be processed
+            await removeFromQueue(message.ReceiptHandle)
+            messagesInProgress.set(message.MessageId, msg)
+            try {
+              await processFunction(msg, messageLogger)
+            } catch (err) {
+              messageLogger.error(err, 'Error while processing queue message!')
+            } finally {
+              messagesInProgress.delete(message.MessageId)
+            }
+          },
+          messageLogger,
+          'Processing queue message!',
+        )
       }
-    })
+    } catch (err) {
+      messageLogger.error(err, { payload: msg }, 'Error while processing queue message!')
+    }
   }
 
   // noinspection InfiniteLoopJS