Add a prefix to tags cache file names (#751)

* Add a prefix for tags cache files

Use lambda function name as a prefix when creating tags cache files accessed by the forwarder.
This will allow to use the same S3 bucket to store cache files from different forwarderst push

* Update readme file to include a section about troubleshooting s3 triggers creation

* Update aws/logs_monitoring/README.md

Co-authored-by: Michel Daviot <michel@daviot.info>

* Add cache prefix on init

* Updat readme

* update comment

---------

Co-authored-by: Michel Daviot <michel@daviot.info>

Author: ge0Aja

aws/logs_monitoring/README.md

@@ -138,6 +138,10 @@ If you can't install the Forwarder using the provided CloudFormation template, y
 
 If you encounter issues upgrading to the latest version, check the Troubleshooting section.
 
+### Upgrade an older version to +3.106.0
+Starting version 3.106.0 Lambda function has been updated to add a prefix to cache filenames stored in the S3 bucket configured in `DD_S3_BUCKET_NAME`.  
+This allows to use the same bucket to store cache files from several functions. 
+
 ### Upgrade an older version to +3.99.0
 
 Since version 3.99.0 the Lambda function has been updated to require **Python 3.11**. If upgrading an older forwarder installation to +3.99.0 or above, ensure the AWS Lambda function is configured to use Python 3.11
@@ -204,6 +208,14 @@ If you still couldn't figure out, please create a ticket for [Datadog Support][1
 ### JSON-formatted logs are not appearing in Datadog
 If your logs contain an attribute that Datadog parses as a timestamp, you need to make sure that the timestamp is both current and in the correct format. See [Log Date Remapper][24] to learn about which attributes are parsed as timestamps and how to make sure that the timestamp is valid.
 
+### Issue creating S3 triggers
+In case you encounter the following error when creating S3 triggers, we recommend considering following a fanout architecture proposed by AWS [in this article](https://aws.amazon.com/blogs/compute/fanout-s3-event-notifications-to-multiple-endpoints/)
+
+```
+An error occurred when creating the trigger: Configuration is ambiguously defined. Cannot have overlapping suffixes in two rules if the prefixes are overlapping for the same event type.
+```
+
+
 ## Contributing
 
 We love pull requests. Here's a quick guide.

aws/logs_monitoring/caching/base_tags_cache.py

@@ -14,41 +14,55 @@
 
 JITTER_MIN = 1
 JITTER_MAX = 100
-
 DD_TAGS_CACHE_TTL_SECONDS = DD_TAGS_CACHE_TTL_SECONDS + randint(JITTER_MIN, JITTER_MAX)
-s3_client = boto3.resource("s3")
-logger = logging.getLogger()
-logger.setLevel(logging.getLevelName(os.environ.get("DD_LOG_LEVEL", "INFO").upper()))
 
 
 class BaseTagsCache(object):
-    CACHE_FILENAME = None
-    CACHE_LOCK_FILENAME = None
-
-    def get_resources_paginator(self):
-        return self.resource_tagging_client.get_paginator("get_resources")
-
-    def __init__(self, tags_ttl_seconds=DD_TAGS_CACHE_TTL_SECONDS):
+    def __init__(
+        self,
+        prefix,
+        cache_filename,
+        cache_lock_filename,
+        tags_ttl_seconds=DD_TAGS_CACHE_TTL_SECONDS,
+    ):
         self.tags_ttl_seconds = tags_ttl_seconds
         self.tags_by_id = {}
         self.last_tags_fetch_time = 0
-        self.logger = logger
+        self.cache_prefix = prefix
+        self.cache_filename = cache_filename
+        self.cache_lock_filename = cache_lock_filename
+        self.logger = logging.getLogger()
+        self.logger.setLevel(
+            logging.getLevelName(os.environ.get("DD_LOG_LEVEL", "INFO").upper())
+        )
         self.resource_tagging_client = boto3.client("resourcegroupstaggingapi")
+        self.s3_client = boto3.resource("s3")
+
+    def get_resources_paginator(self):
+        return self.resource_tagging_client.get_paginator("get_resources")
+
+    def get_cache_name_with_prefix(self):
+        return f"{self.cache_prefix}_{self.cache_filename}"
+
+    def get_cache_lock_with_prefix(self):
+        return f"{self.cache_prefix}_{self.cache_lock_filename}"
 
     def write_cache_to_s3(self, data):
         """Writes tags cache to s3"""
         try:
             self.logger.debug("Trying to write data to s3: {}".format(data))
-            s3_object = s3_client.Object(DD_S3_BUCKET_NAME, self.CACHE_FILENAME)
+            s3_object = self.s3_client.Object(
+                DD_S3_BUCKET_NAME, self.get_cache_name_with_prefix()
+            )
             s3_object.put(Body=(bytes(json.dumps(data).encode("UTF-8"))))
         except ClientError:
             send_forwarder_internal_metrics("s3_cache_write_failure")
             self.logger.debug("Unable to write new cache to S3", exc_info=True)
 
     def acquire_s3_cache_lock(self):
         """Acquire cache lock"""
-        cache_lock_object = s3_client.Object(
-            DD_S3_BUCKET_NAME, self.CACHE_LOCK_FILENAME
+        cache_lock_object = self.s3_client.Object(
+            DD_S3_BUCKET_NAME, self.get_cache_lock_with_prefix()
         )
         try:
             file_content = cache_lock_object.get()
@@ -74,8 +88,8 @@ def acquire_s3_cache_lock(self):
     def release_s3_cache_lock(self):
         """Release cache lock"""
         try:
-            cache_lock_object = s3_client.Object(
-                DD_S3_BUCKET_NAME, self.CACHE_LOCK_FILENAME
+            cache_lock_object = self.s3_client.Object(
+                DD_S3_BUCKET_NAME, self.get_cache_lock_with_prefix()
             )
             cache_lock_object.delete()
             send_forwarder_internal_metrics("s3_cache_lock_released")
@@ -87,7 +101,9 @@ def release_s3_cache_lock(self):
     def get_cache_from_s3(self):
         """Retrieves tags cache from s3 and returns the body along with
         the last modified datetime for the cache"""
-        cache_object = s3_client.Object(DD_S3_BUCKET_NAME, self.CACHE_FILENAME)
+        cache_object = self.s3_client.Object(
+            DD_S3_BUCKET_NAME, self.get_cache_name_with_prefix()
+        )
         try:
             file_content = cache_object.get()
             tags_cache = json.loads(file_content["Body"].read().decode("utf-8"))
@@ -109,7 +125,7 @@ def _refresh(self):
         if not self.should_fetch_tags():
             self.logger.debug(
                 "Not fetching custom tags because the env variable for the cache {} is not set to true".format(
-                    self.CACHE_FILENAME
+                    self.cache_filename
                 )
             )
             return

aws/logs_monitoring/caching/cache_layer.py

@@ -0,0 +1,24 @@
+from caching.cloudwatch_log_group_cache import CloudwatchLogGroupTagsCache
+from caching.step_functions_cache import StepFunctionsTagsCache
+from caching.s3_tags_cache import S3TagsCache
+from caching.lambda_cache import LambdaTagsCache
+
+
+class CacheLayer:
+    def __init__(self, prefix):
+        self._cloudwatch_log_group_cache = CloudwatchLogGroupTagsCache(prefix)
+        self._s3_tags_cache = S3TagsCache(prefix)
+        self._step_functions_cache = StepFunctionsTagsCache(prefix)
+        self._lambda_cache = LambdaTagsCache(prefix)
+
+    def get_cloudwatch_log_group_tags_cache(self):
+        return self._cloudwatch_log_group_cache
+
+    def get_s3_tags_cache(self):
+        return self._s3_tags_cache
+
+    def get_step_functions_tags_cache(self):
+        return self._step_functions_cache
+
+    def get_lambda_tags_cache(self):
+        return self._lambda_cache

aws/logs_monitoring/caching/cloudwatch_log_group_cache.py

@@ -9,8 +9,11 @@
 
 
 class CloudwatchLogGroupTagsCache(BaseTagsCache):
-    CACHE_FILENAME = DD_S3_LOG_GROUP_CACHE_FILENAME
-    CACHE_LOCK_FILENAME = DD_S3_LOG_GROUP_CACHE_LOCK_FILENAME
+    def __init__(self, prefix):
+        super().__init__(
+            prefix, DD_S3_LOG_GROUP_CACHE_FILENAME, DD_S3_LOG_GROUP_CACHE_LOCK_FILENAME
+        )
+        self.cloudwatch_logs_client = boto3.client("logs")
 
     def should_fetch_tags(self):
         return os.environ.get("DD_FETCH_LOG_GROUP_TAGS", "false").lower() == "true"
@@ -69,11 +72,10 @@ def get(self, log_group):
         return log_group_tags
 
     def _get_log_group_tags(self, log_group):
-        cloudwatch_logs_client = boto3.client("logs")
         response = None
         try:
             send_forwarder_internal_metrics("list_tags_log_group_api_call")
-            response = cloudwatch_logs_client.list_tags_log_group(
+            response = self.cloudwatch_logs_client.list_tags_log_group(
                 logGroupName=log_group
             )
         except Exception as e:

aws/logs_monitoring/caching/lambda_cache.py

@@ -1,6 +1,5 @@
 import os
 from botocore.exceptions import ClientError
-
 from caching.base_tags_cache import BaseTagsCache
 from caching.common import (
     send_forwarder_internal_metrics,
@@ -14,8 +13,8 @@
 
 
 class LambdaTagsCache(BaseTagsCache):
-    CACHE_FILENAME = DD_S3_CACHE_FILENAME
-    CACHE_LOCK_FILENAME = DD_S3_CACHE_LOCK_FILENAME
+    def __init__(self, prefix):
+        super().__init__(prefix, DD_S3_CACHE_FILENAME, DD_S3_CACHE_LOCK_FILENAME)
 
     def should_fetch_tags(self):
         return os.environ.get("DD_FETCH_LAMBDA_TAGS", "false").lower() == "true"

aws/logs_monitoring/caching/s3_tags_cache.py

@@ -12,8 +12,10 @@
 
 
 class S3TagsCache(BaseTagsCache):
-    CACHE_FILENAME = DD_S3_TAGS_CACHE_FILENAME
-    CACHE_LOCK_FILENAME = DD_S3_TAGS_CACHE_LOCK_FILENAME
+    def __init__(self, prefix):
+        super().__init__(
+            prefix, DD_S3_TAGS_CACHE_FILENAME, DD_S3_TAGS_CACHE_LOCK_FILENAME
+        )
 
     def should_fetch_tags(self):
         return True

aws/logs_monitoring/caching/step_functions_cache.py

@@ -1,6 +1,5 @@
 import os
 from botocore.exceptions import ClientError
-
 from caching.base_tags_cache import BaseTagsCache
 from caching.common import (
     sanitize_aws_tag_string,
@@ -15,8 +14,12 @@
 
 
 class StepFunctionsTagsCache(BaseTagsCache):
-    CACHE_FILENAME = DD_S3_STEP_FUNCTIONS_CACHE_FILENAME
-    CACHE_LOCK_FILENAME = DD_S3_STEP_FUNCTIONS_CACHE_LOCK_FILENAME
+    def __init__(self, prefix):
+        super().__init__(
+            prefix,
+            DD_S3_STEP_FUNCTIONS_CACHE_FILENAME,
+            DD_S3_STEP_FUNCTIONS_CACHE_LOCK_FILENAME,
+        )
 
     def should_fetch_tags(self):
         return os.environ.get("DD_FETCH_STEP_FUNCTIONS_TAGS", "false").lower() == "true"

aws/logs_monitoring/enhanced_lambda_metrics.py

@@ -7,7 +7,6 @@
 import re
 import datetime
 from time import time
-from caching.lambda_cache import LambdaTagsCache
 
 ENHANCED_METRICS_NAMESPACE_PREFIX = "aws.lambda.enhanced"
 
@@ -82,10 +81,6 @@
     )
     DD_SUBMIT_ENHANCED_METRICS = False
 
-# Store the cache in the global scope so that it will be reused as long as
-# the log forwarder Lambda container is running
-account_lambda_custom_tags_cache = LambdaTagsCache()
-
 
 class DatadogMetricPoint(object):
     """Holds a datapoint's data so that it can be prepared for submission to DD
@@ -141,7 +136,7 @@ def get_last_modified_time(s3_file):
     return last_modified_unix_time
 
 
-def parse_and_submit_enhanced_metrics(logs):
+def parse_and_submit_enhanced_metrics(logs, cache_layer):
     """Parses enhanced metrics from logs and submits them to DD with tags
 
     Args:
@@ -155,7 +150,7 @@ def parse_and_submit_enhanced_metrics(logs):
     for log in logs:
         try:
             enhanced_metrics = generate_enhanced_lambda_metrics(
-                log, account_lambda_custom_tags_cache
+                log, cache_layer.get_lambda_tags_cache()
             )
             for enhanced_metric in enhanced_metrics:
                 enhanced_metric.submit_to_dd()
@@ -334,25 +329,6 @@ def calculate_estimated_cost(billed_duration_ms, memory_allocated):
     return BASE_LAMBDA_INVOCATION_PRICE + gb_seconds * LAMBDA_PRICE_PER_GB_SECOND
 
 
-def get_enriched_lambda_log_tags(log_event):
-    """Retrieves extra tags from lambda, either read from the function arn, or by fetching lambda tags from the function itself.
-
-    Args:
-        log (dict<str, str | dict | int>): a log parsed from the event in the split method
-    """
-    # Note that this arn attribute has been lowercased already
-    log_function_arn = log_event.get("lambda", {}).get("arn")
-
-    if not log_function_arn:
-        return []
-    tags_from_arn = parse_lambda_tags_from_arn(log_function_arn)
-    lambda_custom_tags = account_lambda_custom_tags_cache.get(log_function_arn)
-
-    # Combine and dedup tags
-    tags = list(set(tags_from_arn + lambda_custom_tags))
-    return tags
-
-
 def create_timeout_enhanced_metric(log_line):
     """Parses and returns a value of 1 if a timeout occurred for the function
 

aws/logs_monitoring/lambda_function.py

@@ -8,13 +8,15 @@
 import boto3
 import logging
 import requests
+from hashlib import sha1
 from datadog_lambda.wrapper import datadog_lambda_wrapper
 from datadog import api
 from enhanced_lambda_metrics import parse_and_submit_enhanced_metrics
 from steps.parsing import parse
 from steps.enrichment import enrich
 from steps.transformation import transform
 from steps.splitting import split
+from caching.cache_layer import CacheLayer
 from forwarders import (
     forward_metrics,
     forward_traces,
@@ -29,6 +31,7 @@
     DD_ADDITIONAL_TARGET_LAMBDAS,
 )
 
+
 logger = logging.getLogger()
 logger.setLevel(logging.getLevelName(os.environ.get("DD_LOG_LEVEL", "INFO").upper()))
 
@@ -56,6 +59,8 @@
 api._api_host = DD_API_URL
 api._cacert = not DD_SKIP_SSL_VALIDATION
 
+cache_layer = None
+
 
 def datadog_forwarder(event, context):
     """The actual lambda function entry point"""
@@ -66,8 +71,10 @@ def datadog_forwarder(event, context):
     if DD_ADDITIONAL_TARGET_LAMBDAS:
         invoke_additional_target_lambdas(event)
 
-    parsed = parse(event, context)
-    enriched = enrich(parsed)
+    init_cache_layer(context)
+
+    parsed = parse(event, context, cache_layer)
+    enriched = enrich(parsed, cache_layer)
     transformed = transform(enriched)
     metrics, logs, trace_payloads = split(transformed)
 
@@ -79,7 +86,21 @@ def datadog_forwarder(event, context):
     if len(trace_payloads) > 0:
         forward_traces(trace_payloads)
 
-    parse_and_submit_enhanced_metrics(logs)
+    parse_and_submit_enhanced_metrics(logs, cache_layer)
+
+
+def init_cache_layer(context):
+    global cache_layer
+    if cache_layer is None:
+        # set the prefix for cache layer
+        try:
+            if not cache_layer:
+                function_arn = context.invoked_function_arn.lower()
+                prefix = sha1(function_arn.encode("UTF-8")).hexdigest()
+                cache_layer = CacheLayer(prefix)
+        except Exception as e:
+            logger.exception(f"Failed to create cache layer due to {e}")
+            raise
 
 
 def invoke_additional_target_lambdas(event):