David.kim/updated lambda forwarder

aws/logs_monitoring/lambda_function.py

@@ -148,6 +148,15 @@ def compileRegex(rule, pattern):
 ]
 
 
+# Used to store aws.dd_forwarder.* metrics and tags
+DD_FORWARDER_TELEMETRY = {
+    "logs_forwarded": 0,
+    "traces_forwarded": 0,
+    "metrics_forwarded": 0,
+    "tags": []
+}
+
+
 class RetriableException(Exception):
     pass
 
@@ -390,14 +399,24 @@ def datadog_forwarder(event, context):
 
     if DD_FORWARD_LOG:
         forward_logs(filter_logs(map(json.dumps, logs)))
+        DD_FORWARDER_TELEMETRY["logs_forwarded"]+=len(list(filter_logs(map(json.dumps, logs))))
 
     forward_metrics(metrics)
+    DD_FORWARDER_TELEMETRY["metrics_forwarded"]+=len(metrics)
 
     if len(trace_payloads) > 0:
+        DD_FORWARDER_TELEMETRY["traces_forwarded"]+=len(trace_payloads)
         forward_traces(trace_payloads)
 
     parse_and_submit_enhanced_metrics(logs)
 
+    send_dd_forwarder_telemetry()
+
+    # Reset DD_FORWARDER_TELEMETRY counters and tags after sending
+    DD_FORWARDER_TELEMETRY["logs_forwarded"] = 0
+    DD_FORWARDER_TELEMETRY["traces_forwarded"] = 0
+    DD_FORWARDER_TELEMETRY["metrics_forwarded"] = 0
+    DD_FORWARDER_TELEMETRY["tags"] = []
 
 lambda_handler = datadog_lambda_wrapper(datadog_forwarder)
 
@@ -515,6 +534,10 @@ def add_metadata_to_lambda_log(event):
     tags = list(set(tags))
     tags.sort()  # Keep order deterministic
 
+    # Set custom tags on DD_FORWARDER_TELEMETRY
+    DD_FORWARDER_TELEMETRY["tags"].extend(custom_lambda_tags)
+    DD_FORWARDER_TELEMETRY["tags"].extend(tags)
+
     event[DD_CUSTOM_TAGS] = ",".join([event[DD_CUSTOM_TAGS]] + tags)
 
 
@@ -532,6 +555,11 @@ def generate_metadata(context):
         "forwarder_memorysize": context.memory_limit_in_mb,
         "forwarder_version": DD_FORWARDER_VERSION,
     }
+
+    DD_FORWARDER_TELEMETRY["tags"].append("forwardername:"+dd_custom_tags_data["forwardername"])
+    DD_FORWARDER_TELEMETRY["tags"].append("forwarder_memorysize:"+dd_custom_tags_data["forwarder_memorysize"])
+    DD_FORWARDER_TELEMETRY["tags"].append("forwarder_version:"+DD_FORWARDER_VERSION)
+
     metadata[DD_CUSTOM_TAGS] = ",".join(
         filter(
             None,
@@ -989,3 +1017,30 @@ def parse_service_arn(source, key, bucket, context):
                     region, accountID, clustername
                 )
     return
+
+
+# Send forwarder metrics with tags
+def send_dd_forwarder_telemetry():
+    timestamp = time.time()
+    tags = list(set(DD_FORWARDER_TELEMETRY["tags"]))
+
+    lambda_stats.distribution(
+        "aws.dd_forwarder.logs_forwarded",
+        DD_FORWARDER_TELEMETRY["logs_forwarded"],
+        timestamp=timestamp,
+        tags=tags
+    )
+
+    lambda_stats.distribution(
+        "aws.dd_forwarder.traces_forwarded",
+        DD_FORWARDER_TELEMETRY["traces_forwarded"],
+        timestamp=timestamp,
+        tags=tags
+    )
+
+    lambda_stats.distribution(
+        "aws.dd_forwarder.metrics_forwarded",
+        DD_FORWARDER_TELEMETRY["metrics_forwarded"],
+        timestamp=timestamp,
+        tags=tags
+    )