[CLOUDS-4012] Fix service tag setting in lambda forwarder

If the service tag in set in dd_tags use it, otherwise use the one coming from the logGroup
If none of the above, check if the tag is set on the lambda forwarder and use it, otherwise use the function name.
Finally, check if the service tag is set in the application level and override

aws/logs_monitoring/lambda_cache.py

@@ -78,5 +78,12 @@ def get(self, key):
             logger.debug("Local cache expired, fetching cache from S3")
             self._refresh()
 
+        if not self.should_fetch_tags():
+            logger.debug(
+                "Not fetching lambda function tags because the env variable DD_FETCH_LAMBDA_TAGS is "
+                "not set to true"
+            )
+            return []
         function_tags = self.tags_by_id.get(key, [])
+
         return function_tags

aws/logs_monitoring/lambda_function.py

@@ -257,14 +257,23 @@ def add_metadata_to_lambda_log(event):
     # Get custom tags of the Lambda function
     custom_lambda_tags = get_enriched_lambda_log_tags(event)
 
+    # If not set during parsing or set with a default value
     # Set the `service` tag and metadata field. If the Lambda function is
     # tagged with a `service` tag, use it, otherwise use the function name.
-    service_tag = next(
-        (tag for tag in custom_lambda_tags if tag.startswith("service:")),
-        f"service:{function_name}",
-    )
-    tags.append(service_tag)
-    event[DD_SERVICE] = service_tag.split(":")[1]
+    # Otherwise, remove the `service` tag from the Lambda function if it exists
+    if not event[DD_SERVICE] or event[DD_SERVICE] == event[DD_SOURCE]:
+        service_tag = next(
+            (tag for tag in custom_lambda_tags if tag.startswith("service:")),
+            f"service:{function_name}",
+        )
+        tags.append(service_tag)
+        event[DD_SERVICE] = service_tag.split(":")[1]
+    else:
+        # remove the service tag from the cusotm lambda tags if it exists
+        # as we don't want to add it again
+        custom_lambda_tags = [
+            tag for tag in custom_lambda_tags if not tag.startswith("service:")
+        ]
 
     # Check if one of the Lambda's custom tags is env
     # If an env tag exists, remove the env:none placeholder
@@ -319,7 +328,44 @@ def extract_ddtags_from_message(event):
                 if logger.isEnabledFor(logging.DEBUG):
                     logger.debug(f"Failed to extract ddtags from: {event}")
                 return
-        event[DD_CUSTOM_TAGS] = f"{event[DD_CUSTOM_TAGS]},{extracted_ddtags}"
+
+        event[DD_CUSTOM_TAGS] = _merge_custom_and_application_tags(event[DD_CUSTOM_TAGS], extracted_ddtags)
+
+        # Extract service tag from message.ddtags if exists
+        if extracted_ddtags.contains("service:"):
+            event[DD_SERVICE] = next(tag[8:] for tag in extracted_ddtags.split(",") if tag.startswith("service:"))
+
+
+def _merge_custom_and_application_tags(custom_tags, application_tags):
+    """Merge the custom tags added by the forwarder and the application.
+
+    The custom tags added by the forwarder are added to the top-level `ddtags`
+    field, while the custom tags added by the application are added to the
+    `message.ddtags` field.
+
+    Args:
+        custom_tags (str): the custom tags added by the forwarder
+        application_tags (str): the custom tags added by the application
+
+    Returns:
+        str: the merged custom tags
+    """
+    if not custom_tags:
+        return application_tags
+
+    if not application_tags:
+        return custom_tags
+
+    custom_tags_set = set(custom_tags.split(","))
+    application_tags_set = set(application_tags.split(","))
+    application_tags_keys = [tag.split(":")[0] for tag in application_tags_set]
+
+    for application_tag_key in application_tags_keys:
+        custom_tags_set = set(
+            [tag for tag in custom_tags_set if not tag.startswith(f"{application_tag_key}:")]
+        )
+
+    return ",".join(list(custom_tags_set.union(application_tags_set)))
 
 
 def extract_host_from_cloudtrails(event):

aws/logs_monitoring/parsing.py

@@ -186,7 +186,7 @@ def s3_handler(event, context, metadata):
         source = "transitgateway"
     metadata[DD_SOURCE] = source
 
-    metadata[DD_SERVICE] = get_service_from_tags(metadata)
+    metadata[DD_SERVICE] = get_service_from_tags_and_remove_duplicates(metadata)
 
     ##Get the ARN of the service and set it as the hostname
     hostname = parse_service_arn(source, key, bucket, context)
@@ -242,15 +242,21 @@ def s3_handler(event, context, metadata):
             yield structured_line
 
 
-def get_service_from_tags(metadata):
-    # Get service from dd_custom_tags if it exists
+def get_service_from_tags_and_remove_duplicates(metadata):
+    service = ""
     tagsplit = metadata[DD_CUSTOM_TAGS].split(",")
-    for tag in tagsplit:
+    for i, tag in enumerate(tagsplit):
         if tag.startswith("service:"):
-            return tag[8:]
+            if service:
+                # remove duplicate entry from the tags
+                del tagsplit[i]
+            else:
+                service = tag[8:]
+
+    metadata[DD_CUSTOM_TAGS] = ",".join(tagsplit)
 
     # Default service to source value
-    return metadata[DD_SOURCE]
+    return service if service else metadata[DD_SOURCE]
 
 
 def parse_event_source(event, key):
@@ -530,7 +536,8 @@ def awslogs_handler(event, context, metadata):
 
     # Set service from custom tags, which may include the tags set on the log group
     # Returns DD_SOURCE by default
-    metadata[DD_SERVICE] = get_service_from_tags(metadata)
+    metadata[DD_SERVICE] = get_service_from_tags_and_remove_duplicates(metadata)
+
 
     # Set host as log group where cloudwatch is source
     if metadata[DD_SOURCE] == "cloudwatch" or metadata.get(DD_HOST, None) == None:
@@ -640,7 +647,7 @@ def cwevent_handler(event, metadata):
     else:
         metadata[DD_SOURCE] = "cloudwatch"
 
-    metadata[DD_SERVICE] = get_service_from_tags(metadata)
+    metadata[DD_SERVICE] = get_service_from_tags_and_remove_duplicates(metadata)
 
     yield data