Skip to content

Streaming Logs from AWS CloudTrail and CloudWatch

This guide is for customers who want to send AWS CloudTrail events and CloudWatch Logs (such as application or API logs) into SenseOn. Both use the same mechanism: a small forwarder Lambda in your own AWS account, subscribed to whichever CloudWatch Log Groups you want ingested. CloudTrail rides this same path by enabling its CloudWatch Logs delivery, rather than a separate S3-based route.

How it works

Your log producers (CloudTrail, applications)
        -> CloudWatch Log Group (your account)
        -> Subscription filter
        -> Forwarder Lambda (your account)
        -> HTTPS POST -> SenseOn ingestion endpoint

You deploy the Lambda and point it at the ingestion endpoint SenseOn provides for your environment, attach a subscription filter per log group you want forwarded, and grant CloudWatch Logs permission to invoke it. Everything runs in your own account, nothing is pulled from it, and no cross-account IAM role or access into your account is required for this path. Step 3 covers both a manual (AWS Console/CLI) and a Terraform-based way to do this.

Prerequisites

  • Permissions in your AWS account to create a Lambda function, an IAM role/permission for it, and CloudWatch Logs subscription filters.
  • If you want CloudTrail included: an existing CloudTrail trail (or create a new one).
  • The ingestion endpoint URL for your environment, provided by your SenseOn contact.

Step 1: Enable CloudWatch Logs delivery on your CloudTrail trail

If you want CloudTrail events ingested, open your trail and enable CloudWatch Logs delivery, pointing at a dedicated log group.

This is additive: your trail keeps writing to its existing S3 destination exactly as before, this only adds a second, parallel delivery target. Nothing about your current CloudTrail setup needs to change.

If you're only sending CloudWatch application logs and not CloudTrail, skip this step.

Step 2: Identify your CloudWatch log groups

Note the name of every CloudWatch Log Group you want SenseOn to receive, your application/API logs, and (if applicable) the CloudTrail-destined log group from Step 1.

Step 3: Deploy the forwarder and subscribe your log groups

Choose whichever of these suits how you manage infrastructure. Both end up in the same place: a forwarder Lambda in your account, with a subscription filter per log group pointing at it.

Create a new Lambda function:

  • Name: SenseOn-Forwarder
  • Runtime: Python 3.12
  • Timeout: Under Basic settings, increase the execution timeout to 1 minute. The default of 3 seconds is too short for the code's own 10 second HTTP timeout to ever take effect, so ordinary network latency alone could get the function killed and trigger unnecessary CloudWatch Logs retries.

Paste the following directly into the console's default code editor (it already matches AWS's default filename and handler, lambda_function.py / lambda_handler, so no renaming or Handler configuration is needed):

import base64
import gzip
import json
import os
import urllib.request
from datetime import datetime, timezone

INGESTION_ENDPOINT_URL = os.environ["INGESTION_ENDPOINT_URL"]
DEBUG_LOG_ENVELOPES = os.environ.get("DEBUG_LOG_ENVELOPES", "") == "true"


def _to_iso8601(epoch_ms):
    return datetime.fromtimestamp(epoch_ms / 1000, tz=timezone.utc).strftime(
        "%Y-%m-%dT%H:%M:%S.%f"
    )[:-3] + "Z"


def _envelope(log_group, log_stream, event):
    return {
        "metadata": {
            "original_log_group": log_group,
            "resource_id": log_stream,
        },
        "log_event": {
            "eventId": event["id"],
            "logStreamName": log_stream,
            "timestamp_utc": _to_iso8601(event["timestamp"]),
            "message": event["message"],
        },
    }


def lambda_handler(event, _context):
    payload = json.loads(gzip.decompress(base64.b64decode(event["awslogs"]["data"])))
    log_group = payload["logGroup"]
    log_stream = payload["logStream"]

    envelopes = [_envelope(log_group, log_stream, e) for e in payload["logEvents"]]
    body = json.dumps(envelopes).encode("utf-8")

    if DEBUG_LOG_ENVELOPES:
        print(json.dumps(envelopes[:1]))

    headers = {"Content-Type": "application/json"}
    request = urllib.request.Request(
        INGESTION_ENDPOINT_URL, data=body, headers=headers, method="POST"
    )
    with urllib.request.urlopen(request, timeout=10) as response:
        response.read()

    return {"forwarded": len(envelopes)}

Environment variables

Variable Required Purpose
INGESTION_ENDPOINT_URL Yes The SenseOn ingestion endpoint provided for your environment.
DEBUG_LOG_ENVELOPES No Set to exactly true temporarily to have the function print the payload it sends, viewable in its own CloudWatch Logs group. Leave unset otherwise.

Grant CloudWatch Logs permission to invoke the Lambda

CloudWatch Logs needs explicit permission to call this Lambda for a given log group, it isn't automatic.

Console: if you create the subscription filter below through the CloudWatch console and choose this Lambda as the destination, the required permission is added for you automatically. Nothing else to do.

AWS CLI: if you use aws logs put-subscription-filter directly, add the permission first, or the command fails with an access error:

aws lambda add-permission \
  --function-name SenseOn-Forwarder \
  --statement-id AllowCloudWatchLogsInvoke-<log-group-short-name> \
  --action lambda:InvokeFunction \
  --principal logs.<region>.amazonaws.com \
  --source-arn arn:aws:logs:<region>:<account-id>:log-group:<log-group-name>:* \
  --source-account <account-id>

Give each log group's permission a unique --statement-id if you're forwarding more than one (CloudTrail's and an application log group both need their own).

Create the subscription filter

One per log group you want ingested. Name: SenseOn-Forwarder-SF, this name only needs to be unique within its own log group, so it can be reused across every log group you subscribe.

Point it at the Lambda above. A blank filter pattern forwards every event from that log group.

⚠ Think before excluding events by type. If you're sending CloudTrail for security monitoring, avoid filtering out read-only calls (Describe*/List*/Get*). These are exactly how AWS account enumeration and reconnaissance activity shows up, filtering them out for volume reasons can blind detection to the activity it's meant to catch. If volume genuinely needs managing, filtering by identity type (for example excluding calls made by AWS service principals) is a safer lever than filtering by verb.

Troubleshooting

[ERROR] Runtime.HandlerNotFound: Handler 'lambda_handler' missing on module 'lambda_function' This shouldn't come up given the code above already matches AWS's default filename and handler, but if the source was renamed or pasted into a differently-named file, go to the Code tab's Runtime settings panel and confirm Handler reads lambda_function.lambda_handler, and that the source file is named lambda_function.py.

Task timed out after 3.00 seconds The function's execution timeout wasn't raised above. Go to Basic settings and confirm Timeout is set to at least 1 minute, not the AWS default of 3 seconds.

This stack only ever reads the log groups you list (a data source, never a resource), so it can't create, modify, or delete them, and its own teardown is safe to run at any time -- it only removes what it created.

Download the Terraform module (zip) to get every file below already in place, rather than creating each one by hand. Unzip it, review the contents against the code shown below, then skip ahead to terraform init.

Prerequisites beyond the ones above: Terraform >= 1.5, the aws CLI, and jq -- the last two are what a guardrail script below shells out to.

The files, for reference (identical to what's in the zip):

versions.tf:

terraform {
  required_version = ">= 1.5"

  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "~> 5.0"
    }
    archive = {
      source  = "hashicorp/archive"
      version = "~> 2.4"
    }
    external = {
      source  = "hashicorp/external"
      version = "~> 2.3"
    }
  }
}

provider "aws" {
  region  = var.aws_region
  profile = var.aws_profile
}

variables.tf:

variable "aws_region" {
  description = "Region the log groups being subscribed to (and this stack's own resources) live in."
  type        = string
  default     = "eu-west-1"
}

variable "aws_profile" {
  description = "Named AWS CLI profile to authenticate with (leave null to use the default credential chain)."
  type        = string
  default     = null
}

variable "name_prefix" {
  description = "Prefix applied to every resource this stack creates, so it's identifiable in the account and can be torn down cleanly on its own."
  type        = string
  default     = "senseon-forwarder"
}

variable "log_group_names" {
  description = <<-EOT
    CloudWatch Log Group names to forward to SenseOn. Every name must
    already exist (CloudTrail delivery, an application log group, etc.) --
    this stack only reads each one via a data source and subscribes to it,
    it never creates, modifies, or deletes the log group itself. Add or
    remove a name here and re-apply to subscribe/unsubscribe that one
    source; nothing else is affected.
  EOT
  type        = list(string)

  validation {
    condition     = length(var.log_group_names) > 0
    error_message = "log_group_names must list at least one existing CloudWatch Log Group name."
  }

  validation {
    condition     = alltrue([for n in var.log_group_names : trimspace(n) == n && n != ""])
    error_message = "log_group_names entries must be non-empty and have no leading/trailing whitespace."
  }

  validation {
    condition     = length(var.log_group_names) == length(distinct(var.log_group_names))
    error_message = "log_group_names must not contain duplicate entries."
  }
}

variable "filter_pattern" {
  description = <<-EOT
    CloudWatch Logs subscription filter pattern applied to every log group
    listed above. Defaults to "" (no filtering): for CloudTrail sources in
    particular, Describe*/List*/Get* calls are how AWS enumeration/discovery
    (MITRE ATT&CK TA0007) actually shows up, so excluding them by verb would
    blind detection rather than cut noise.
  EOT
  type        = string
  default     = ""
}

variable "ingestion_endpoint_url" {
  description = "SenseOn's HTTP log ingestion endpoint the forwarder Lambda POSTs to. No authentication scheme exists for this endpoint yet -- TLS is the only protection in transit."
  type        = string
}

variable "reserved_concurrent_executions" {
  description = "Caps how many forwarder invocations can run at once, so a burst from a noisy log group can't consume the account's shared Lambda concurrency pool and starve other functions. -1 removes the cap."
  type        = number
  default     = 10
}

variable "debug_log_envelopes" {
  description = "Print the exact envelope sent to SenseOn into the forwarder's own logs. A troubleshooting/sample-gathering aid, not for routine use -- puts real log content into CloudWatch. Turn back off afterwards."
  type        = bool
  default     = false
}

variable "log_retention_days" {
  description = "Retention for the forwarder Lambda's own execution log group. Never applied to your own log groups -- those are untouched by this stack."
  type        = number
  default     = 14
}

variable "permissions_boundary_arn" {
  description = "Optional IAM permissions boundary ARN to attach to the forwarder's role. Leave null (default) unless your account requires one -- some organisations' SCPs reject any CreateRole call that omits it."
  type        = string
  default     = null
}

variable "tags" {
  description = "Additional tags applied to every resource this stack creates."
  type        = map(string)
  default     = {}
}

forwarder.tf:

locals {
  tags = merge(
    { ManagedBy = "senseon-log-forwarder-terraform" },
    var.tags,
  )

  # AWS hard limit, not adjustable via Service Quotas.
  subscription_filter_limit = 5
  filter_name               = "${var.name_prefix}-to-forwarder"
}

data "aws_caller_identity" "current" {}

# Data source, never a resource: your log group must not be something
# this stack's state can create, modify, or delete.
data "aws_cloudwatch_log_group" "source" {
  for_each = toset(var.log_group_names)
  name     = each.value
}

# No `aws` provider data source lists a log group's existing subscription
# filters, so the only way to check before adding ours is a real API call.
data "external" "existing_filters" {
  for_each = data.aws_cloudwatch_log_group.source

  program = ["bash", "${path.module}/scripts/check_subscription_filters.sh"]
  query = {
    log_group_name = each.value.name
    aws_region     = var.aws_region
    aws_profile    = coalesce(var.aws_profile, "")
  }
}

locals {
  existing_filter_names = {
    for k, v in data.external.existing_filters : k => jsondecode(v.result.names)
  }
  existing_filter_counts = {
    for k, v in data.external.existing_filters : k => tonumber(v.result.count)
  }
}

data "archive_file" "forwarder" {
  type        = "zip"
  source_dir  = "${path.module}/lambda_src"
  output_path = "${path.module}/.build/forwarder.zip"
}

data "aws_iam_policy_document" "lambda_trust" {
  statement {
    effect  = "Allow"
    actions = ["sts:AssumeRole"]
    principals {
      type        = "Service"
      identifiers = ["lambda.amazonaws.com"]
    }
  }
}

resource "aws_iam_role" "forwarder" {
  name                 = "${var.name_prefix}-forwarder"
  assume_role_policy   = data.aws_iam_policy_document.lambda_trust.json
  permissions_boundary = var.permissions_boundary_arn
  tags                 = local.tags
}

resource "aws_cloudwatch_log_group" "forwarder" {
  name              = "/aws/lambda/${var.name_prefix}-forwarder"
  retention_in_days = var.log_retention_days
  tags              = local.tags
}

# Scoped to this Lambda's own log group only, not the account-wide logs:*
# granted by AWSLambdaBasicExecutionRole -- least privilege for a stack
# landing in your account.
data "aws_iam_policy_document" "forwarder_logging" {
  statement {
    effect = "Allow"
    actions = [
      "logs:CreateLogStream",
      "logs:PutLogEvents",
    ]
    resources = ["${aws_cloudwatch_log_group.forwarder.arn}:*"]
  }
}

resource "aws_iam_role_policy" "forwarder_logging" {
  name   = "${var.name_prefix}-forwarder-logging"
  role   = aws_iam_role.forwarder.id
  policy = data.aws_iam_policy_document.forwarder_logging.json
}

resource "aws_lambda_function" "forwarder" {
  function_name                  = "${var.name_prefix}-forwarder"
  role                           = aws_iam_role.forwarder.arn
  handler                        = "forwarder.handler"
  runtime                        = "python3.12"
  timeout                        = 60
  reserved_concurrent_executions = var.reserved_concurrent_executions
  filename                       = data.archive_file.forwarder.output_path
  source_code_hash               = data.archive_file.forwarder.output_base64sha256

  environment {
    variables = {
      INGESTION_ENDPOINT_URL = var.ingestion_endpoint_url
      DEBUG_LOG_ENVELOPES    = var.debug_log_envelopes ? "true" : "false"
    }
  }

  depends_on = [aws_iam_role_policy.forwarder_logging]

  tags = local.tags
}

# Keyed by log group name so edits to log_group_names map to exactly one
# filter+permission pair, leaving every other source untouched.
resource "aws_lambda_permission" "invoke" {
  for_each = data.aws_cloudwatch_log_group.source

  statement_id   = "AllowInvoke-${substr(md5(each.key), 0, 16)}"
  action         = "lambda:InvokeFunction"
  function_name  = aws_lambda_function.forwarder.function_name
  principal      = "logs.${var.aws_region}.amazonaws.com"
  source_arn     = "${each.value.arn}:*"
  source_account = data.aws_caller_identity.current.account_id
}

resource "aws_cloudwatch_log_subscription_filter" "to_forwarder" {
  for_each = data.aws_cloudwatch_log_group.source

  name            = local.filter_name
  log_group_name  = each.value.name
  filter_pattern  = var.filter_pattern
  destination_arn = aws_lambda_function.forwarder.arn

  lifecycle {
    precondition {
      condition = (
        contains(local.existing_filter_names[each.key], local.filter_name) ||
        local.existing_filter_counts[each.key] < local.subscription_filter_limit
      )
      error_message = "Log group ${each.key} already has ${local.existing_filter_counts[each.key]} subscription filter(s); AWS allows at most ${local.subscription_filter_limit} per log group, a hard limit that can't be raised. Remove or consolidate an existing filter on this log group before adding SenseOn's, or drop it from log_group_names."
    }
  }

  depends_on = [aws_lambda_permission.invoke]
}

outputs.tf:

output "forwarder_function_name" {
  value = aws_lambda_function.forwarder.function_name
}

output "forwarder_log_group_name" {
  value = aws_cloudwatch_log_group.forwarder.name
}

output "subscribed_log_groups" {
  description = "Every log group currently being forwarded, and the subscription filter name attached to it."
  value = {
    for name, filter in aws_cloudwatch_log_subscription_filter.to_forwarder :
    name => filter.name
  }
}

lambda_src/forwarder.py:

import base64
import gzip
import json
import os
import urllib.request
from datetime import datetime, timezone

INGESTION_ENDPOINT_URL = os.environ["INGESTION_ENDPOINT_URL"]
DEBUG_LOG_ENVELOPES = os.environ.get("DEBUG_LOG_ENVELOPES", "") == "true"


def _to_iso8601(epoch_ms):
    return datetime.fromtimestamp(epoch_ms / 1000, tz=timezone.utc).strftime(
        "%Y-%m-%dT%H:%M:%S.%f"
    )[:-3] + "Z"


def _envelope(log_group, log_stream, event):
    return {
        "metadata": {
            "original_log_group": log_group,
            "resource_id": log_stream,
        },
        "log_event": {
            "eventId": event["id"],
            "logStreamName": log_stream,
            "timestamp_utc": _to_iso8601(event["timestamp"]),
            "message": event["message"],
        },
    }


def handler(event, _context):
    payload = json.loads(gzip.decompress(base64.b64decode(event["awslogs"]["data"])))
    log_group = payload["logGroup"]
    log_stream = payload["logStream"]

    envelopes = [_envelope(log_group, log_stream, e) for e in payload["logEvents"]]
    body = json.dumps(envelopes).encode("utf-8")

    if DEBUG_LOG_ENVELOPES:
        print(json.dumps(envelopes[:1]))

    request = urllib.request.Request(
        INGESTION_ENDPOINT_URL,
        data=body,
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    with urllib.request.urlopen(request, timeout=10) as response:
        response.read()

    return {"forwarded": len(envelopes)}

scripts/check_subscription_filters.sh (used by forwarder.tf to check the AWS limit below before adding a filter -- make it executable with chmod +x):

#!/usr/bin/env bash
set -euo pipefail

query="$(cat)"
log_group_name="$(jq -r '.log_group_name' <<<"${query}")"
aws_region="$(jq -r '.aws_region' <<<"${query}")"
aws_profile="$(jq -r '.aws_profile' <<<"${query}")"

profile_args=()
if [[ -n "${aws_profile}" && "${aws_profile}" != "null" ]]; then
  profile_args=(--profile "${aws_profile}")
fi

names_json="$(aws logs describe-subscription-filters \
  "${profile_args[@]}" --region "${aws_region}" \
  --log-group-name "${log_group_name}" \
  --query 'subscriptionFilters[].filterName' --output json)"

jq -n --argjson names "${names_json}" \
  '{count: ($names | length | tostring), names: ($names | tostring)}'

example.tfvars is included in the zip with the same placeholders shown below. Copy it and fill in your own values -- keep the real terraform.tfvars out of version control, since it's specific to your account:

aws_region  = "eu-west-1"
aws_profile = "your-aws-cli-profile" # named AWS CLI profile for your account
name_prefix = "senseon-forwarder"

# Every name here must already exist -- this stack only reads and
# subscribes to them, it never creates or modifies the log group itself.
log_group_names = [
  "/aws/cloudtrail/your-cloudtrail-log-group",
  # "/ecs/your-app",
]

ingestion_endpoint_url = "https://your-ingestion-endpoint.senseon.io"

cp example.tfvars terraform.tfvars
# edit terraform.tfvars with your own values

Then apply:

terraform init
terraform plan  -var-file=terraform.tfvars
terraform apply -var-file=terraform.tfvars

Guardrails built into this stack

  • Subscription filter limit. AWS caps every log group at 5 subscription filters, a hard limit that can't be raised. The precondition above checks the real count via the guardrail script before adding SenseOn's filter, and fails plan/apply up front with the log group name and its current count if adding this one would exceed it -- instead of a raw LimitExceededException mid-apply.
  • Least-privilege IAM. The forwarder's role can only write to its own log group, and its aws_lambda_permission is scoped by both source_arn and source_account (the standard guard against a same-ARN-pattern resource in a different account invoking this function).
  • Concurrency cap. reserved_concurrent_executions (default 10) stops a burst from a noisy log group consuming your account's shared Lambda concurrency pool.

One thing this can't guard against: if a log group you list already has an unrelated subscription filter that happens to share this stack's exact filter name (${name_prefix}-to-forwarder), applying will silently take it over -- CloudWatch subscription filters carry no ownership tag, so there's no way to tell "ours" from "someone else's" by name alone. Pick a name_prefix unlikely to collide with anything already in your account.

If apply fails

executable file not found in $PATH -- aws or jq isn't installed on the machine running Terraform; both are required by the guardrail script above.

Resource precondition failed citing the subscription filter limit -- the named log group is already at AWS's 5-filter cap. Remove or consolidate an existing filter on it before adding SenseOn's, or drop it from log_group_names.

Adding or removing a source

Edit log_group_names and re-apply. terraform plan shows exactly one filter and one permission being added or removed, per log group changed -- every other source is untouched.

Decommissioning

terraform destroy -var-file=terraform.tfvars

This removes the forwarder Lambda, its role, its own log group, and every subscription filter and permission this stack created. The log groups being forwarded, and any data already delivered to SenseOn, are unaffected.

Step 4: Verify it's working

Trigger an event in the source log group (any AWS API call generates a CloudTrail event; a test log line works for application log groups), then check the forwarder Lambda's own CloudWatch Logs group (/aws/lambda/SenseOn-Forwarder for the Console path, or terraform output forwarder_log_group_name for the Terraform path). A clean run with no exceptions means it reached the ingestion endpoint successfully.

Good practice

  • Additive, not disruptive: enabling CloudWatch Logs delivery on an existing CloudTrail trail never changes or replaces its existing S3 delivery.
  • No cross-account access required: this is a push model from your account, SenseOn never needs a role or credentials into your AWS account for this path.
  • Filter deliberately, not by default: see the warning in Step 3's Console tab, unfiltered is the safer starting point for a source used for detection.
  • Turn off DEBUG_LOG_ENVELOPES when done: it's a diagnostic aid, not something to leave on, since it prints real log content into CloudWatch.
  • The stack only ever removes what it created: your log groups are read, never owned, by the stack in Step 3's Terraform tab, so applying it and later tearing it back down is safe against a real log group at any time.