AWS Forwarder v5

aws/logs_monitoring/CHANGELOG.md

@@ -0,0 +1,120 @@
+# Datadog Lambda Forwarder Changelog
+
+## v5.0.0 - BREAKING CHANGES
+
+### Overview
+
+Version 5.0.0 of the Datadog Lambda Forwarder introduces several breaking changes that remove deprecated features and improve log filtering behavior. This release introduces a new way to enrich your logs with tags that will reduce AWS Lambda related cost (S3, KMS and Lambda).
+
+### New Features
+
+#### 1. Backend Storage Tag Enrichment
+
+**Added:**
+
+- New `DD_ENRICH_S3_TAGS` / `DdEnrichS3Tags` parameter (default: `true`)
+- New `DD_ENRICH_CLOUDWATCH_TAGS` / `DdEnrichCloudwatchTags` parameter (default: `true`)
+- These instruct the Datadog backend to automatically enrich logs with resource tags **after ingestion**
+- New cloudwatch tags can appear on logs, check your Datadog log index configuration to ensure smooth transition.
+
+**Benefits:**
+
+- **Reduces forwarder cost** and execution time
+- Provides the same tag enrichment as `DdFetchS3Tags` and `DdFetchLogGroupTags`
+- Requires [Resource Collection](https://docs.datadoghq.com/integrations/amazon-web-services/#resource-collection) enabled in your AWS integration
+
+**Deprecation Notice:**
+
+- `DdFetchS3Tags` is now marked as **DEPRECATED** in favor of `DdEnrichS3Tags`
+- `DdFetchLogGroupTags` is now marked as **DEPRECATED** in favor of `DdEnrichCloudwatchTags`
+- `DD_FETCH_S3_TAGS` now defaults to `false` (previously `true`)
+
+---
+
+### Breaking Changes
+
+#### 1. Changed Regex Matching Behavior for Log Filtering
+
+**What Changed:**
+
+- `IncludeAtMatch` / `INCLUDE_AT_MATCH` and `ExcludeAtMatch` / `EXCLUDE_AT_MATCH` regex patterns now match **only against the log message** itself
+- Previously, these patterns matched against the **entire JSON-formatted log**
+
+**Migration Required:**
+
+- **Review and update your filtering regex patterns**
+- If your patterns relied on matching against JSON structure or metadata fields, they will need to be rewritten
+- Example changes needed:
+    - **Before (v4)**: `\"awsRegion\":\"us-east-1\"` (matched JSON with escaped quotes)
+    - **After (v5)**: `"awsRegion":"us-east-1"` (matches the message content directly)
+- Patterns that matched the `message` field content should continue to work with minimal changes
+
+---
+
+#### 2. Removed TCP Transport Support
+
+**What Changed:**
+
+- Removed the `DD_USE_TCP` / `DdUseTcp` environment variable and parameter
+- Deleted the TCP client implementation
+- All logs now **must** be sent via HTTP/HTTPS
+
+**Migration Required:**
+
+- Remove any configuration setting `DD_USE_TCP=true` or `DdUseTcp=true`
+- The forwarder will now exclusively use HTTP transport
+- If you were using TCP with custom ports (10516), these configurations will be ignored
+- The default HTTP endpoint is now `http-intake.logs.<DD_SITE>` on port 443
+
+---
+
+#### 3. Removed Deprecated PrivateLink Environment Variable
+
+**What Changed:**
+
+- Removed the `DD_USE_PRIVATE_LINK` / `DdUsePrivateLink` environment variable and parameter
+
+**Migration Required:**
+
+- Remove any configuration setting `DD_USE_PRIVATE_LINK=true`
+- **AWS PrivateLink is still fully supported**, but you must follow [PrivateLink documentation](https://docs.datadoghq.com/agent/guide/private-link/):
+    1. Set up VPC endpoints for `api`, `http-logs.intake`, and `trace.agent` as documented
+    2. Configure the forwarder with `DdUseVPC=true`
+    3. Set `VPCSecurityGroupIds` and `VPCSubnetIds`
+
+**Why This Changed:**
+
+- The variable was previously deprecated but not removed from past versions.
+
+---
+
+### Upgrade Instructions
+
+Follow the usual [documentation](https://docs.datadoghq.com/logs/guide/forwarder/?tab=cloudformation#upgrade-to-a-new-version) about upgrading your Lambda Forwarder.
+
+#### Pre-Upgrade Checklist
+
+1. **Verify you're not using TCP transport:**
+
+    ```bash
+    aws lambda get-function-configuration --function-name "<YOUR_FORWARDER>" --query 'Environment.Variables.DD_USE_TCP'
+    ```
+
+2. **Verify you're not using the deprecated PrivateLink variable:**
+
+    ```bash
+    aws lambda get-function-configuration --function-name "<YOUR_FORWARDER>" --query 'Environment.Variables.DD_USE_PRIVATE_LINK'
+    ```
+
+3. **Review your log filtering patterns:**
+    - If using `IncludeAtMatch` or `ExcludeAtMatch`, test your patterns against log messages only
+    - Remove any JSON escaping (e.g., `\"` → `"`)
+
+#### Testing
+
+After upgrading:
+
+1. Verify logs are being forwarded to Datadog
+2. Check that filtering rules still work as expected
+3. Confirm tag enrichment is working (check logs in Datadog Explorer)
+4. Monitor forwarder execution duration and errors in CloudWatch

aws/logs_monitoring/forwarder.py

@@ -4,34 +4,33 @@
 # Copyright 2021 Datadog, Inc.
 
 
-import logging
 import json
+import logging
 import os
 
-from telemetry import send_event_metric, send_log_metric
-from trace_forwarder.connection import TraceConnection
-from logs.datadog_http_client import DatadogHTTPClient
 from logs.datadog_batcher import DatadogBatcher
 from logs.datadog_client import DatadogClient
-from logs.datadog_tcp_client import DatadogTCPClient
+from logs.datadog_http_client import DatadogHTTPClient
+from logs.datadog_matcher import DatadogMatcher
 from logs.datadog_scrubber import DatadogScrubber
-from logs.helpers import filter_logs, add_retry_tag
-from retry.storage import Storage
+from logs.helpers import add_retry_tag
 from retry.enums import RetryPrefix
+from retry.storage import Storage
 from settings import (
     DD_API_KEY,
-    DD_USE_TCP,
+    DD_FORWARD_LOG,
     DD_NO_SSL,
-    DD_SKIP_SSL_VALIDATION,
-    DD_URL,
     DD_PORT,
-    DD_TRACE_INTAKE_URL,
-    DD_FORWARD_LOG,
+    DD_SKIP_SSL_VALIDATION,
     DD_STORE_FAILED_EVENTS,
-    SCRUBBING_RULE_CONFIGS,
-    INCLUDE_AT_MATCH,
+    DD_TRACE_INTAKE_URL,
+    DD_URL,
     EXCLUDE_AT_MATCH,
+    INCLUDE_AT_MATCH,
+    SCRUBBING_RULE_CONFIGS,
 )
+from telemetry import send_event_metric, send_log_metric
+from trace_forwarder.connection import TraceConnection
 
 logger = logging.getLogger()
 logger.setLevel(logging.getLevelName(os.environ.get("DD_LOG_LEVEL", "INFO").upper()))
@@ -83,35 +82,35 @@ def _forward_logs(self, logs, key=None):
             logger.debug(f"Forwarding {len(logs)} logs")
 
         scrubber = DatadogScrubber(SCRUBBING_RULE_CONFIGS)
+        matcher = DatadogMatcher(
+            include_pattern=INCLUDE_AT_MATCH, exclude_pattern=EXCLUDE_AT_MATCH
+        )
+
         logs_to_forward = []
         for log in logs:
             if key:
                 log = add_retry_tag(log)
 
-            # apply scrubbing rules to inner log message if exists
+            evaluated_log = log
+
+            # apply scrubbing rules to inner log message
             if isinstance(log, dict) and log.get("message"):
                 try:
                     log["message"] = scrubber.scrub(log["message"])
+                    evaluated_log = log["message"]
                 except Exception as e:
                     logger.exception(
                         f"Exception while scrubbing log message {log['message']}: {e}"
                     )
 
-            logs_to_forward.append(json.dumps(log, ensure_ascii=False))
+            if matcher.match(evaluated_log):
+                logs_to_forward.append(json.dumps(log, ensure_ascii=False))
 
-        logs_to_forward = filter_logs(
-            logs_to_forward, INCLUDE_AT_MATCH, EXCLUDE_AT_MATCH
+        batcher = DatadogBatcher(512 * 1000, 4 * 1000 * 1000, 400)
+        cli = DatadogHTTPClient(
+            DD_URL, DD_PORT, DD_NO_SSL, DD_SKIP_SSL_VALIDATION, DD_API_KEY, scrubber
         )
 
-        if DD_USE_TCP:
-            batcher = DatadogBatcher(256 * 1000, 256 * 1000, 1)
-            cli = DatadogTCPClient(DD_URL, DD_PORT, DD_NO_SSL, DD_API_KEY, scrubber)
-        else:
-            batcher = DatadogBatcher(512 * 1000, 4 * 1000 * 1000, 400)
-            cli = DatadogHTTPClient(
-                DD_URL, DD_PORT, DD_NO_SSL, DD_SKIP_SSL_VALIDATION, DD_API_KEY, scrubber
-            )
-
         failed_logs = []
         with DatadogClient(cli) as client:
             for batch in batcher.batch(logs_to_forward):

aws/logs_monitoring/logs/datadog_http_client.py

@@ -4,25 +4,41 @@
 # Copyright 2021 Datadog, Inc.
 
 
-import os
 import logging
-
+import os
 from concurrent.futures import as_completed
+
 from requests_futures.sessions import FuturesSession
-from logs.helpers import compress_logs
-from logs.exceptions import ScrubbingException
 
+from logs.exceptions import ScrubbingException
+from logs.helpers import compress_logs
 from settings import (
-    DD_USE_COMPRESSION,
     DD_COMPRESSION_LEVEL,
-    DD_MAX_WORKERS,
     DD_FORWARDER_VERSION,
+    DD_MAX_WORKERS,
+    DD_USE_COMPRESSION,
+    get_enrich_cloudwatch_tags,
+    get_enrich_s3_tags,
 )
 
 logger = logging.getLogger()
 logger.setLevel(logging.getLevelName(os.environ.get("DD_LOG_LEVEL", "INFO").upper()))
 
 
+def get_dd_storage_tag_header():
+    storage_tag = ""
+
+    if get_enrich_s3_tags():
+        storage_tag += "s3"
+
+    if get_enrich_cloudwatch_tags():
+        if storage_tag != "":
+            storage_tag += ","
+        storage_tag += "cloudwatch"
+
+    return storage_tag
+
+
 class DatadogHTTPClient(object):
     """
     Client that sends a batch of logs over HTTP.
@@ -37,6 +53,10 @@ class DatadogHTTPClient(object):
     _HEADERS["DD-EVP-ORIGIN"] = "aws_forwarder"
     _HEADERS["DD-EVP-ORIGIN-VERSION"] = DD_FORWARDER_VERSION
 
+    storage_tag = get_dd_storage_tag_header()
+    if storage_tag != "":
+        _HEADERS["DD-STORAGE-TAG"] = storage_tag
+
     def __init__(
         self, host, port, no_ssl, skip_ssl_validation, api_key, scrubber, timeout=10
     ):

aws/logs_monitoring/logs/datadog_matcher.py

@@ -0,0 +1,48 @@
+# Unless explicitly stated otherwise all files in this repository are licensed
+# under the Apache License Version 2.0.
+# This product includes software developed at Datadog (https://www.datadoghq.com/).
+# Copyright 2021 Datadog, Inc.
+
+
+import logging
+import os
+import re
+
+from logs.exceptions import ScrubbingException
+from logs.helpers import compileRegex
+
+logger = logging.getLogger()
+logger.setLevel(logging.getLevelName(os.environ.get("DD_LOG_LEVEL", "INFO").upper()))
+
+
+class DatadogMatcher(object):
+    def __init__(self, include_pattern=None, exclude_pattern=None):
+        self._include_regex = None
+        self._exclude_regex = None
+
+        if include_pattern is not None:
+            logger.debug(f"Applying include pattern: {include_pattern}")
+            self._include_regex = compileRegex("INCLUDE_AT_MATCH", include_pattern)
+
+        if exclude_pattern is not None:
+            logger.debug(f"Applying exclude pattern: {exclude_pattern}")
+            self._exclude_regex = compileRegex("EXCLUDE_AT_MATCH", exclude_pattern)
+
+    def match(self, log):
+        try:
+            if self._exclude_regex is not None and re.search(
+                self._exclude_regex, str(log)
+            ):
+                logger.debug("Exclude pattern matched, excluding log event")
+                return False
+
+            if self._include_regex is not None and not re.search(
+                self._include_regex, str(log)
+            ):
+                logger.debug("Include pattern did not match, excluding log event")
+                return False
+
+            return True
+
+        except ScrubbingException:
+            raise Exception("could not filter the payload")