[PP-14099]: Handle transforming ADOT logs

spec/fixtures/adot_fixtures.ts

@@ -0,0 +1,66 @@
+import { Fixture } from './general_fixtures'
+
+export const anADOTCloudWatchEvent: Fixture = {
+  input: {
+    deliveryStreamArn: 'someDeliveryStreamArn',
+    invocationId: 'someId',
+    region: 'eu-west-1',
+    records: [{
+      approximateArrivalTimestamp: 1234,
+      recordId: 'LogEvent-1',
+      data: Buffer.from(JSON.stringify({
+        owner: '223851549868',
+        logGroup: 'test-12_adot_frontend',
+        logStream: 'logStream',
+        subscriptionFilters: [],
+        messageType: 'DATA_MESSAGE',
+        logEvents: [
+          {
+            id: 'cloudwatch-log-message-id-1',
+            timestamp: 1234,
+            message: '2025/05/28 10:13:36 ADOT Collector version: v0.43.3'
+          },
+          {
+            id: 'cloudwatch-log-gmessage-id-2',
+            timestamp: 1235,
+            message: '2025-05-28T10:13:36.838Z warn awsemfexporter@v0.117.0/emf_exporter.go:92 the default value for DimensionRollupOption will be changing to NoDimensionRollupin a future release. See https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/23997 for moreinformation {"kind": "exporter", "data_type": "metrics", "name": "awsemf"}'
+          }
+        ]
+      })).toString('base64')
+    }]
+  },
+  expected: {
+    records: [{
+      result: 'Ok',
+      recordId: 'LogEvent-1',
+      data: Buffer.from([
+        {
+          host: 'logStream',
+          source: 'adot',
+          sourcetype: 'generic_single_line',
+          index: 'pay_devops',
+          event: '2025/05/28 10:13:36 ADOT Collector version: v0.43.3',
+          fields: {
+            account: 'test',
+            environment: 'test-12',
+            service: 'frontend'
+          },
+          time: 1234.000
+        },
+        {
+          host: 'logStream',
+          source: 'adot',
+          sourcetype: 'generic_single_line',
+          index: 'pay_devops',
+          event: '2025-05-28T10:13:36.838Z warn awsemfexporter@v0.117.0/emf_exporter.go:92 the default value for DimensionRollupOption will be changing to NoDimensionRollupin a future release. See https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/23997 for moreinformation {"kind": "exporter", "data_type": "metrics", "name": "awsemf"}',
+          fields: {
+            account: 'test',
+            environment: 'test-12',
+            service: 'frontend'
+          },
+          time: 1235.000
+        }
+      ].map(x => JSON.stringify(x)).join('\n')).toString('base64')
+    }]
+  }
+}

spec/index.test.ts

@@ -6,6 +6,10 @@ import {
   anInvalidApplicationLogFirehoseTransformationEventRecord
 } from './fixtures/application_fixtures'
 
+import {
+  anADOTCloudWatchEvent
+} from './fixtures/adot_fixtures'
+
 import {
   aBastionLogCloudWatchEvent
 } from './fixtures/bastion_fixtures'
@@ -81,6 +85,16 @@ describe('Processing CloudWatchLogEvents', () => {
       expect(Buffer.from(result.records[1].data as string, 'base64').toString()).toEqual(Buffer.from(expectedSecondRecord.data as string, 'base64').toString())
     })
   })
+  describe('From ADOT', () => {
+    test('should transform adot logs from CloudWatch', async () => {
+      const result = await handler(anADOTCloudWatchEvent.input, mockContext, mockCallback) as FirehoseTransformationResult
+
+      const expectedRecord = anADOTCloudWatchEvent.expected.records[0]
+      expect(result.records[0].result).toEqual(expectedRecord.result)
+      expect(result.records[0].recordId).toEqual(expectedRecord.recordId)
+      expect(Buffer.from(result.records[0].data as string, 'base64').toString()).toEqual(Buffer.from(expectedRecord.data as string, 'base64').toString())
+    })
+  })
   describe('From Bastion', () => {
     test('should transform bastion logs from CloudWatch', async () => {
       const result = await handler(aBastionLogCloudWatchEvent.input, mockContext, mockCallback) as FirehoseTransformationResult

src/extractTime.ts

@@ -153,6 +153,9 @@ export function parseTimeFromLog(log: string, logType: CloudWatchLogTypes): numb
   switch (logType) {
     case CloudWatchLogTypes.app:
       return extractAppLogTime(log)
+    case CloudWatchLogTypes.adot:
+      // Adot has many different time formats, so we'll rely on the cloudwatch timestamp instead
+      return undefined
     case CloudWatchLogTypes.bastion:
       return extractBastionLogTime(log)
     case CloudWatchLogTypes.squid:

src/transformData.ts

@@ -152,6 +152,9 @@ function sourceTypeFromLogGroup(logType: CloudWatchLogTypes, msg: string): strin
   switch (logType) {
     case CloudWatchLogTypes.app:
       return 'ST004:application_json'
+    case CloudWatchLogTypes.adot:
+      // ADOT has logs in many formats, so better to just set a geenric source type
+      return 'generic_single_line'
     case CloudWatchLogTypes.bastion:
       return 'linux_bastion'
     case CloudWatchLogTypes['nginx-forward-proxy']:
@@ -184,6 +187,8 @@ function indexFromLogType(logType: CloudWatchLogTypes): string {
   switch (logType) {
     case CloudWatchLogTypes.app:
       return 'pay_application'
+    case CloudWatchLogTypes.adot:
+      return 'pay_devops'
     case CloudWatchLogTypes.bastion:
       return 'pay_host'
     case CloudWatchLogTypes['nginx-forward-proxy']:

src/types.ts

@@ -37,6 +37,7 @@ export type EnvVars = {
 }
 
 export enum CloudWatchLogTypes {
+  'adot',
   'app',
   'apt',
   'audit',