Add environment variable to support backend storage tag enrichment (#1006)

* feat(aws): AWSX-1566 Adding storage tag HTTP header

Signed-off-by: Vincent Boutour <vincent.boutour@datadoghq.com>

* feat(aws): AWSX-1566 Adding new variable in the cloudformation template

Signed-off-by: Vincent Boutour <vincent.boutour@datadoghq.com>

* feat: AWSX-1566 Restoring fetch in the integration test to see metrics

Signed-off-by: Vincent Boutour <vincent.boutour@datadoghq.com>

* fixup! feat: AWSX-1566 Restoring fetch in the integration test to see metrics

Signed-off-by: Vincent Boutour <vincent.boutour@datadoghq.com>

* docs(aws): AWSX-1566 Refine documentation around the new variable

Signed-off-by: Vincent Boutour <vincent.boutour@datadoghq.com>

---------

Signed-off-by: Vincent Boutour <vincent.boutour@datadoghq.com>

Author: ViBiOh

aws/logs_monitoring/README.md

@@ -278,7 +278,7 @@ You can run the Forwarder in a VPC private subnet and send data to Datadog over
 3. When installing the Forwarder with the CloudFormation template:
     1. Set `DdUseVPC` to `true`.
     2. Set `VPCSecurityGroupIds` and `VPCSubnetIds` based on your VPC settings.
-    3. Set `DdFetchLambdaTags`, `DdFetchStepFunctionsTags` and `DdFetchS3Tags` to `false`, because AWS Resource Groups Tagging API doesn't support PrivateLink.
+    3. Set `DdFetchLambdaTags`, `DdFetchStepFunctionsTags`, and `DdFetchS3Tags` to `false`, because AWS Resource Groups Tagging API doesn't support PrivateLink.
 
 ### AWS VPC and proxy support
 
@@ -287,7 +287,7 @@ If you must deploy the Forwarder to a VPC without direct public internet access,
 1. Unless the Forwarder is deployed to a public subnet, follow the [instructions][15] to add endpoints for Secrets Manager and S3 to the VPC, so that the Forwarder can access those services.
 2. Update your proxy with following configurations ([HAProxy][17] or [NGINX][18]). If you are using another proxy, or Web Proxy, allowlist the Datadog domain, for example: `.{{< region-param key="dd_site" code="true" >}}`.
 3. When installing the Forwarder with the CloudFormation template, set `DdUseVPC`, `VPCSecurityGroupIds`, and `VPCSubnetIds`.
-4. Ensure the `DdFetchLambdaTags`, `DdFetchStepFunctionsTags` and `DdFetchS3Tags` options are disabled, because AWS VPC does not yet offer an endpoint for the Resource Groups Tagging API.
+4. Ensure the `DdFetchLambdaTags`, `DdFetchStepFunctionsTags`, and `DdFetchS3Tags` options are disabled, because AWS VPC does not yet offer an endpoint for the Resource Groups Tagging API.
 5. If you are using HAProxy or NGINX:
 
 - Set `DdApiUrl` to `http://<proxy_host>:3834` or `https://<proxy_host>:3834`.
@@ -404,17 +404,23 @@ To test different patterns against your logs, turn on [debug logs](#troubleshoot
 
 ### Advanced (optional)
 
+`DdEnrichS3Tags`
+: Enabled by default. When enabled, instructs the Datadog backend to automatically enrich logs originating from S3 buckets with the tags associated with those buckets. This approach offers the same tag enrichment as `DdFetchS3Tags` but defers the operation after log ingestion, reducing Forwarder overhead. Requires [Resource Collection](https://docs.datadoghq.com/integrations/amazon-web-services/#resource-collection) to be enabled in your AWS integration.
+
+`DdEnrichCloudwatchTags`
+: Enabled by default. When enabled, instructs the Datadog backend to automatically enrich logs originating from CloudWatch LogGroups with the tags associated with those log groups. This approach offers the same tag enrichment as `DdFetchLogGroupTags` but defers the operation after log ingestion, reducing Forwarder overhead. Requires [Resource Collection](https://docs.datadoghq.com/integrations/amazon-web-services/#resource-collection) to be enabled in your AWS integration.
+
 `DdFetchLambdaTags`
 : Let the Forwarder fetch Lambda tags using GetResources API calls and apply them to logs, metrics, and traces. If set to true, permission `tag:GetResources` will be automatically added to the Lambda execution IAM role.
 
 `DdFetchLogGroupTags`
-: Let the forwarder fetch Log Group tags using ListTagsLogGroup and apply them to logs, metrics, and traces. If set to true, permission `logs:ListTagsForResource` will be automatically added to the Lambda execution IAM role.
+: **[DEPRECATED, use DdEnrichCloudwatchTags]** Let the forwarder fetch Log Group tags using ListTagsLogGroup and apply them to logs, metrics, and traces. If set to true, permission `logs:ListTagsForResource` will be automatically added to the Lambda execution IAM role.
 
 `DdFetchStepFunctionsTags`
 : Let the Forwarder fetch Step Functions tags using GetResources API calls and apply them to logs and traces (if Step Functions tracing is enabled). If set to true, permission `tag:GetResources` will be automatically added to the Lambda execution IAM role.
 
 `DdFetchS3Tags`
-: Let the Forwarder fetch S3 tags using GetResources API calls and apply them to logs and traces. If set to true, permission `tag:GetResources` will be automatically added to the Lambda execution IAM role.
+: **[DEPRECATED, use DdEnrichS3Tags]** Let the Forwarder fetch S3 tags using GetResources API calls and apply them to logs and traces. If set to true, permission `tag:GetResources` will be automatically added to the Lambda execution IAM role.
 
 `DdStepFunctionsTraceEnabled`
 : Set to true to enable tracing for all Step Functions.
@@ -567,11 +573,17 @@ To test different patterns against your logs, turn on [debug logs](#troubleshoot
 
 ### Advanced (optional)
 
+`DD_ENRICH_S3_TAGS`
+: Enabled by default. When enabled, instructs the Datadog backend to automatically enrich logs originating from S3 buckets with the tags associated with those buckets. This approach offers the same tag enrichment as `DD_FETCH_S3_TAGS` but defers the operation after log ingestion, reducing Forwarder overhead. Requires https://docs.datadoghq.com/integrations/amazon-web-services/#resource-collection to be enabled in your AWS integration.
+
+`DD_ENRICH_CLOUDWATCH_TAGS`
+: Enabled by default. When enabled, instructs the Datadog backend to automatically enrich logs originating from Cloudwatch LogGroup with the tags associated with those log groups. This approach offers the same tag enrichment as `DD_FETCH_LOG_GROUP_TAGS` but defers the operation after log ingestion, reducing Forwarder overhead. Requires https://docs.datadoghq.com/integrations/amazon-web-services/#resource-collection to be enabled in your AWS integration.
+
 `DD_FETCH_LAMBDA_TAGS`
 : Let the Forwarder fetch Lambda tags using GetResources API calls and apply them to logs, metrics, and traces. If set to true, permission `tag:GetResources` will be automatically added to the Lambda execution IAM role.
 
 `DD_FETCH_LOG_GROUP_TAGS`
-: Let the forwarder fetch Log Group tags using ListTagsLogGroup and apply them to logs, metrics, and traces. If set to true, permission `logs:ListTagsForResource` will be automatically added to the Lambda execution IAM role.
+: [DEPRECATED, use DD_ENRICH_CLOUDWATCH_TAGS] Let the forwarder fetch Log Group tags using ListTagsLogGroup and apply them to logs, metrics, and traces. If set to true, permission `logs:ListTagsForResource` will be automatically added to the Lambda execution IAM role.
 
 `DD_FETCH_STEP_FUNCTIONS_TAGS`
 : Let the Forwarder fetch Step Functions tags using GetResources API calls and apply them to logs and traces (if Step Functions tracing is enabled). If set to true, permission `tag:GetResources` will be automatically added to the Lambda execution IAM role.

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/settings.py

@@ -250,6 +250,23 @@ def __init__(self, name, pattern, placeholder, enabled=True):
 )
 
 
+DD_ENRICH_S3_TAGS = get_env_var("DD_ENRICH_S3_TAGS", default="true", boolean=True)
+
+DD_ENRICH_CLOUDWATCH_TAGS = get_env_var(
+    "DD_ENRICH_CLOUDWATCH_TAGS", default="true", boolean=True
+)
+
+if DD_FETCH_S3_TAGS and DD_ENRICH_S3_TAGS:
+    logger.warn(
+        "Enabling both DD_FETCH_S3_TAGS and DD_ENRICH_S3_TAGS might be unwanted"
+    )
+
+if DD_FETCH_LOG_GROUP_TAGS and DD_ENRICH_CLOUDWATCH_TAGS:
+    logger.warn(
+        "Enabling both DD_FETCH_LOG_GROUP_TAGS and DD_ENRICH_CLOUDWATCH_TAGS might be unwanted"
+    )
+
+
 def get_fetch_s3_tags():
     return DD_FETCH_S3_TAGS
 
@@ -266,6 +283,14 @@ def get_fetch_step_functions_tags():
     return DD_FETCH_STEP_FUNCTIONS_TAGS
 
 
+def get_enrich_s3_tags():
+    return DD_ENRICH_S3_TAGS
+
+
+def get_enrich_cloudwatch_tags():
+    return DD_ENRICH_CLOUDWATCH_TAGS
+
+
 DD_SOURCE = "ddsource"
 DD_CUSTOM_TAGS = "ddtags"
 DD_SERVICE = "service"

aws/logs_monitoring/template.yaml

@@ -75,6 +75,20 @@ Parameters:
     Type: String
     Default: ""
     Description: Add custom tags to forwarded logs, comma-delimited string, no trailing comma, e.g., env:prod,stack:classic
+  DdEnrichS3Tags:
+    Type: String
+    Default: true
+    AllowedValues:
+      - true
+      - false
+    Description: Instruct Datadog backend to enrich a log coming from a S3 bucket with the tag attached to this bucket. Datadog AWS Resource Collection needs to be enabled.
+  DdEnrichCloudwatchTags:
+    Type: String
+    Default: true
+    AllowedValues:
+      - true
+      - false
+    Description: Instruct Datadog backend to enrich a log coming from a Cloudwatch logGroup with the tag attached to this logGroup. Datadog AWS Resource Collection needs to be enabled.
   DdFetchLambdaTags:
     Type: String
     Default: true
@@ -88,7 +102,7 @@ Parameters:
     AllowedValues:
       - true
       - false
-    Description: Let the forwarder fetch Log Group tags using ListTagsLogGroup and apply them to logs, metrics and traces. If set to true, permission logs:ListTagsLogGroup will be automatically added to the Lambda execution IAM role. The tags are cached in memory and S3 so that they'll only be fetched when the function cold starts or when the TTL (1 hour) expires. The forwarder increments the aws.lambda.enhanced.list_tags_log_group_api_call metric for each API call made.
+    Description: (DEPRECATED in favor of DdEnrichCloudwatchTags) Let the forwarder fetch Log Group tags using ListTagsLogGroup and apply them to logs, metrics and traces. If set to true, permission logs:ListTagsLogGroup will be automatically added to the Lambda execution IAM role. The tags are cached in memory and S3 so that they'll only be fetched when the function cold starts or when the TTL (1 hour) expires. The forwarder increments the aws.lambda.enhanced.list_tags_log_group_api_call metric for each API call made.
   DdFetchStepFunctionsTags:
     Type: String
     Default: true
@@ -98,11 +112,11 @@ Parameters:
     Description: Let the forwarder fetch Step Functions tags using GetResources API calls and apply them to logs, metrics and traces. If set to true, permission tag:GetResources will be automatically added to the Lambda execution IAM role. The tags are cached in memory and S3 so that they'll only be fetched when the function cold starts or when the TTL (1 hour) expires. The forwarder increments the aws.lambda.enhanced.get_resources_api_calls metric for each API call made.
   DdFetchS3Tags:
     Type: String
-    Default: true
+    Default: false
     AllowedValues:
       - true
       - false
-    Description: Let the forwarder fetch S3 buckets tags using GetResources API calls and apply them to S3 based logs. If set to true, permission tag:GetResources will be automatically added to the Lambda execution IAM role. The tags are cached in memory and S3 so that they'll only be fetched when the function cold starts or when the TTL (1 hour) expires. The forwarder increments the aws.lambda.enhanced.get_resources_api_calls metric for each API call made.
+    Description: (DEPRECATED in favor of DdEnrichS3Tags) Let the forwarder fetch S3 buckets tags using GetResources API calls and apply them to S3 based logs. If set to true, permission tag:GetResources will be automatically added to the Lambda execution IAM role. The tags are cached in memory and S3 so that they'll only be fetched when the function cold starts or when the TTL (1 hour) expires. The forwarder increments the aws.lambda.enhanced.get_resources_api_calls metric for each API call made.
   DdNoSsl:
     Type: String
     Default: false
@@ -448,11 +462,13 @@ Resources:
             - !Ref DdTags
             - !Ref AWS::NoValue
           DD_TAGS_CACHE_TTL_SECONDS: !Ref TagsCacheTTLSeconds
+          DD_ENRICH_S3_TAGS: !Ref DdEnrichS3Tags
+          DD_ENRICH_CLOUDWATCH_TAGS: !Ref DdEnrichCloudwatchTags
+          DD_FETCH_S3_TAGS: !Ref DdFetchS3Tags
           DD_FETCH_LAMBDA_TAGS: !If
             - SetDdFetchLambdaTags
             - !Ref DdFetchLambdaTags
             - !Ref AWS::NoValue
-          DD_FETCH_S3_TAGS: !Ref DdFetchS3Tags
           DD_FETCH_LOG_GROUP_TAGS: !If
             - SetDdFetchLogGroupTags
             - !Ref DdFetchLogGroupTags
@@ -1018,6 +1034,8 @@ Metadata:
       - Label:
           default: Advanced (Optional)
         Parameters:
+          - DdEnrichS3Tags
+          - DdEnrichCloudwatchTags
           - DdFetchLambdaTags
           - DdFetchLogGroupTags
           - DdFetchStepFunctionsTags

aws/logs_monitoring/tools/build_bundle.sh

@@ -64,7 +64,7 @@ docker_build_zip() {
     # between different python runtimes.
     temp_dir=$(mktemp -d)
 
-    docker buildx build --platform  linux/arm64 --file "${DIR}/Dockerfile_bundle" -t "datadog-bundle:$1" .. --no-cache --build-arg "runtime=${PYTHON_VERSION}"
+    docker buildx build --platform linux/arm64 --file "${DIR}/Dockerfile_bundle" -t "datadog-bundle:$1" .. --no-cache --build-arg "runtime=${PYTHON_VERSION}"
 
     # Run the image by runtime tag, tar its generated `python` directory to sdout,
     # then extract it to a temp directory.

aws/logs_monitoring/tools/integration_tests/docker-compose.yml

@@ -40,9 +40,9 @@ services:
       DD_USE_COMPRESSION: "false"
       DD_ADDITIONAL_TARGET_LAMBDAS: "${EXTERNAL_LAMBDAS}"
       DD_S3_BUCKET_NAME: "${DD_S3_BUCKET_NAME}"
-      DD_FETCH_LAMBDA_TAGS: "true"
-      DD_FETCH_LOG_GROUP_TAGS: "true"
-      DD_FETCH_STEP_FUNCTIONS_TAGS: "false"  # intentionally set false to allow integration test for step function logs to run without hitting aws
+      DD_FETCH_LAMBDA_TAGS: "${DD_FETCH_LAMBDA_TAGS:-false}"
+      DD_FETCH_LOG_GROUP_TAGS: "${DD_FETCH_LOG_GROUP_TAGS:-false}"
+      DD_FETCH_STEP_FUNCTIONS_TAGS: "${DD_FETCH_STEP_FUNCTIONS_TAGS:-false}"
       DD_STORE_FAILED_EVENTS: "false"
       DD_TRACE_ENABLED: "true"
     expose:

aws/logs_monitoring/tools/integration_tests/integration_tests.sh

@@ -1,4 +1,4 @@
-#!/bin/bash
+#!/usr/bin/env bash
 
 # Unless explicitly stated otherwise all files in this repository are licensed
 # under the Apache License Version 2.0.
@@ -21,6 +21,7 @@ SNAPS=($SNAPSHOT_DIR)
 ADDITIONAL_LAMBDA=false
 CACHE_TEST=false
 DD_FETCH_LAMBDA_TAGS="true"
+DD_FETCH_LOG_GROUP_TAGS="true"
 DD_FETCH_STEP_FUNCTIONS_TAGS="true"
 
 script_start_time=$(date -u +"%Y-%m-%dT%H:%M:%SZ")
@@ -37,7 +38,6 @@ for arg in "$@"; do
                 shift
                 ;;
 
-
         # -u or --update
         # Update the snapshots to reflect this test run
         -u | --update)
@@ -152,6 +152,7 @@ LOG_LEVEL=${LOG_LEVEL} \
         AWS_ACCOUNT_ID=${AWS_ACCOUNT_ID} \
         SNAPSHOTS_DIR_NAME="./${SNAPSHOTS_DIR_NAME}" \
         DD_FETCH_LAMBDA_TAGS=${DD_FETCH_LAMBDA_TAGS} \
+        DD_FETCH_LOG_GROUP_TAGS=${DD_FETCH_LOG_GROUP_TAGS} \
         DD_FETCH_STEP_FUNCTIONS_TAGS=${DD_FETCH_STEP_FUNCTIONS_TAGS} \
         docker compose up --build --abort-on-container-exit
 

aws/logs_monitoring/tools/integration_tests/snapshots/cloudwatch_customized_log_group_lambda_invocation.json~snapshot

@@ -75,6 +75,7 @@
         "DD-API-KEY": "abcdefghijklmnopqrstuvwxyz012345",
         "DD-EVP-ORIGIN": "aws_forwarder",
         "DD-EVP-ORIGIN-VERSION": "<redacted from snapshot>",
+        "DD-STORAGE-TAG": "s3,cloudwatch",
         "Host": "recorder:8080",
         "User-Agent": "<redacted from snapshot>",
         "traceparent": "<redacted from snapshot>",

aws/logs_monitoring/tools/integration_tests/snapshots/cloudwatch_log.json~snapshot

@@ -48,6 +48,7 @@
         "DD-API-KEY": "abcdefghijklmnopqrstuvwxyz012345",
         "DD-EVP-ORIGIN": "aws_forwarder",
         "DD-EVP-ORIGIN-VERSION": "<redacted from snapshot>",
+        "DD-STORAGE-TAG": "s3,cloudwatch",
         "Host": "recorder:8080",
         "User-Agent": "<redacted from snapshot>",
         "traceparent": "<redacted from snapshot>",

aws/logs_monitoring/tools/integration_tests/snapshots/cloudwatch_log_cloudtrail.json~snapshot

@@ -84,6 +84,7 @@
         "DD-API-KEY": "abcdefghijklmnopqrstuvwxyz012345",
         "DD-EVP-ORIGIN": "aws_forwarder",
         "DD-EVP-ORIGIN-VERSION": "<redacted from snapshot>",
+        "DD-STORAGE-TAG": "s3,cloudwatch",
         "Host": "recorder:8080",
         "User-Agent": "<redacted from snapshot>",
         "traceparent": "<redacted from snapshot>",