AWS CloudWatch Integration

CloudWatch Log Groups
        │
        ▼
  Subscription Filter
        │
        ▼
 Lambda Forwarder ──► POST /events/logs/batch ──► Operyn Ingestion
        │                                              │
        │                                         Normalise → Index → Enqueue
        │                                              │
        ▼                                              ▼
  CloudWatch Metrics                            Incident Engine
        │                                      (detect + diagnose)
        ▼
  Metric Stream → Firehose
        │
        ▼
 Lambda Forwarder ──► POST /events/metrics ──► Operyn Ingestion

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "logs:CreateLogGroup",
        "logs:CreateLogStream",
        "logs:PutLogEvents"
      ],
      "Resource": "arn:aws:logs:*:*:*"
    }
  ]
}

import { gunzipSync } from 'node:zlib';

const OPERYN_INGEST_URL = process.env.OPERYN_INGEST_URL;
const OPERYN_API_KEY = process.env.OPERYN_API_KEY;
const OPERYN_ORGANIZATION_ID = process.env.OPERYN_ORGANIZATION_ID;

const SERVICE_MAP = JSON.parse(process.env.SERVICE_MAP || '{}');

function deriveServiceName(logGroup) {
  if (SERVICE_MAP[logGroup]) return SERVICE_MAP[logGroup];

  const parts = logGroup.split('/');
  // /aws/lambda/billing-service → billing-service
  // /aws/ecs/api-gateway → api-gateway
  return parts[parts.length - 1] || 'unknown';
}

function extractLevel(message) {
  const upper = message.toUpperCase();
  if (upper.includes('FATAL')) return 'fatal';
  if (upper.includes('ERROR')) return 'error';
  if (upper.includes('WARN')) return 'warn';
  if (upper.includes('DEBUG')) return 'debug';
  return 'info';
}

export const handler = async (event) => {
  const compressed = Buffer.from(event.awslogs.data, 'base64');
  const payload = JSON.parse(gunzipSync(compressed).toString('utf-8'));

  const { logGroup, logStream, logEvents } = payload;
  const service = deriveServiceName(logGroup);

  const body = logEvents.map((le) => ({
    service,
    level: extractLevel(le.message),
    message: le.message,
    timestamp: new Date(le.timestamp).toISOString(),
    metadata: {
      source: 'cloudwatch',
      logGroup,
      logStream,
      awsRegion: process.env.AWS_REGION,
    },
  }));

  const response = await fetch(`${OPERYN_INGEST_URL}/events/logs/batch`, {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      'x-api-key': OPERYN_API_KEY,
      'x-organization-id': OPERYN_ORGANIZATION_ID,
    },
    body: JSON.stringify(body),
  });

  if (!response.ok) {
    const text = await response.text();
    throw new Error(`Ingestion failed (${response.status}): ${text}`);
  }

  const result = await response.json();
  console.log(`Forwarded ${result.accepted} events from ${logGroup}`);
  return result;
};

# Allow CloudWatch to invoke the Lambda
aws lambda add-permission \
  --function-name operyn-log-forwarder \
  --statement-id cloudwatch-invoke \
  --action lambda:InvokeFunction \
  --principal logs.amazonaws.com \
  --source-arn "arn:aws:logs:us-east-1:123456789012:log-group:/aws/ecs/billing-service:*"

# Create the subscription filter
aws logs put-subscription-filter \
  --log-group-name "/aws/ecs/billing-service" \
  --filter-name "operyn-forwarder" \
  --filter-pattern "" \
  --destination-arn "arn:aws:lambda:us-east-1:123456789012:function:operyn-log-forwarder"

resource "aws_lambda_permission" "cloudwatch" {
  for_each      = toset(var.log_groups)
  statement_id  = "cloudwatch-${md5(each.value)}"
  action        = "lambda:InvokeFunction"
  function_name = aws_lambda_function.operyn_forwarder.function_name
  principal     = "logs.amazonaws.com"
  source_arn    = "arn:aws:logs:${data.aws_region.current.name}:${data.aws_caller_identity.current.account_id}:log-group:${each.value}:*"
}

resource "aws_cloudwatch_log_subscription_filter" "operyn" {
  for_each        = toset(var.log_groups)
  name            = "operyn-forwarder"
  log_group_name  = each.value
  filter_pattern  = ""
  destination_arn = aws_lambda_function.operyn_forwarder.arn
}

# Manually send a test log
curl -X POST https://ingest.operyn.example.com/events/logs \
  -H "Content-Type: application/json" \
  -H "x-api-key: YOUR_API_KEY" \
  -H "x-organization-id: YOUR_ORG_ID" \
  -d '{
    "service": "billing-service",
    "level": "error",
    "message": "Test event from CloudWatch integration verification",
    "metadata": { "source": "cloudwatch", "env": "prod" }
  }'

export const handler = async (event) => {
  for (const record of event.records) {
    const data = JSON.parse(Buffer.from(record.data, 'base64').toString());
    const body = {
      service: `${data.namespace}/${data.metric_name}`.toLowerCase().replace(/\//g, '-'),
      name: data.metric_name,
      value: data.value.sum ?? data.value.average ?? data.value.max ?? 0,
      unit: data.unit,
      timestamp: new Date(data.timestamp).toISOString(),
      labels: data.dimensions || {},
    };

    await fetch(`${process.env.OPERYN_INGEST_URL}/events/metrics`, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'x-api-key': process.env.OPERYN_API_KEY,
        'x-organization-id': process.env.OPERYN_ORGANIZATION_ID,
      },
      body: JSON.stringify(body),
    });
  }
};