David.kim/updated lambda forwarder

aws/logs_monitoring/enhanced_lambda_metrics.py

@@ -297,6 +297,17 @@ def parse_get_resources_response_for_tags_by_arn(get_resources_page):
     return tags_by_arn
 
 
+def get_forwarder_telemetry_prefix_and_tags():
+    """Retrieves prefix and tags used when submitting telemetry metrics
+    Used to overcome circular import"""
+    from lambda_function import (
+        DD_FORWARDER_TELEMETRY_NAMESPACE_PREFIX,
+        DD_FORWARDER_TELEMETRY_TAGS,
+    )
+
+    return DD_FORWARDER_TELEMETRY_NAMESPACE_PREFIX, DD_FORWARDER_TELEMETRY_TAGS
+
+
 def build_tags_by_arn_cache():
     """Makes API calls to GetResources to get the live tags of the account's Lambda functions
 
@@ -307,14 +318,14 @@ def build_tags_by_arn_cache():
     """
     tags_by_arn_cache = {}
     get_resources_paginator = resource_tagging_client.get_paginator("get_resources")
+    prefix, tags = get_forwarder_telemetry_prefix_and_tags()
 
     try:
         for page in get_resources_paginator.paginate(
             ResourceTypeFilters=[GET_RESOURCES_LAMBDA_FILTER], ResourcesPerPage=100
         ):
             lambda_stats.distribution(
-                "{}.get_resources_api_calls".format(ENHANCED_METRICS_NAMESPACE_PREFIX),
-                1,
+                "{}.get_resources_api_calls".format(prefix), 1, tags=tags,
             )
             page_tags_by_arn = parse_get_resources_response_for_tags_by_arn(page)
             tags_by_arn_cache.update(page_tags_by_arn)

aws/logs_monitoring/lambda_function.py

@@ -148,6 +148,11 @@ def compileRegex(rule, pattern):
 ]
 
 
+# Used to build and pass aws.dd_forwarder.* telemetry tags
+DD_FORWARDER_TELEMETRY_TAGS = []
+DD_FORWARDER_TELEMETRY_NAMESPACE_PREFIX = "aws.dd_forwarder"
+
+
 class RetriableException(Exception):
     pass
 
@@ -389,7 +394,7 @@ def datadog_forwarder(event, context):
     metrics, logs, trace_payloads = split(enrich(parse(event, context)))
 
     if DD_FORWARD_LOG:
-        forward_logs(filter_logs(map(json.dumps, logs)))
+        forward_logs(logs)
 
     forward_metrics(metrics)
 
@@ -404,6 +409,7 @@ def datadog_forwarder(event, context):
 
 def forward_logs(logs):
     """Forward logs to Datadog"""
+    logs_to_forward = filter_logs(list(map(json.dumps, logs)))
     scrubber = DatadogScrubber(SCRUBBING_RULE_CONFIGS)
     if DD_USE_TCP:
         batcher = DatadogBatcher(256 * 1000, 256 * 1000, 1)
@@ -415,7 +421,7 @@ def forward_logs(logs):
         )
 
     with DatadogClient(cli) as client:
-        for batch in batcher.batch(logs):
+        for batch in batcher.batch(logs_to_forward):
             try:
                 client.send(batch)
             except Exception:
@@ -424,10 +430,17 @@ def forward_logs(logs):
                 if logger.isEnabledFor(logging.DEBUG):
                     logger.debug(f"Forwarded log batch: {json.dumps(batch)}")
 
+    lambda_stats.distribution(
+        "{}.logs_forwarded".format(DD_FORWARDER_TELEMETRY_NAMESPACE_PREFIX),
+        len(logs_to_forward),
+        tags=DD_FORWARDER_TELEMETRY_TAGS,
+    )
+
 
 def parse(event, context):
     """Parse Lambda input to normalized events"""
     metadata = generate_metadata(context)
+    event_type = "unknown"
     try:
         # Route to the corresponding parser
         event_type = parse_event_type(event)
@@ -448,9 +461,25 @@ def parse(event, context):
         )
         events = [err_message]
 
+    set_forwarder_telemetry_tags(context, event_type)
+
     return normalize_events(events, metadata)
 
 
+def set_forwarder_telemetry_tags(context, event_type):
+    """Helper function to set tags on telemetry metrics
+    Do not submit telemetry metrics before this helper function is invoked
+    """
+    global DD_FORWARDER_TELEMETRY_TAGS
+
+    DD_FORWARDER_TELEMETRY_TAGS = [
+        f"forwardername:{context.function_name.lower()}",
+        f"forwarder_memorysize:{context.memory_limit_in_mb}",
+        f"forwarder_version:{DD_FORWARDER_VERSION}",
+        f"event_type:{event_type}",
+    ]
+
+
 def enrich(events):
     """Adds event-specific tags and attributes to each event
 
@@ -572,6 +601,7 @@ def generate_metadata(context):
         "forwarder_memorysize": context.memory_limit_in_mb,
         "forwarder_version": DD_FORWARDER_VERSION,
     }
+
     metadata[DD_CUSTOM_TAGS] = ",".join(
         filter(
             None,
@@ -675,6 +705,12 @@ def forward_metrics(metrics):
             if logger.isEnabledFor(logging.DEBUG):
                 logger.debug(f"Forwarded metric: {json.dumps(metric)}")
 
+    lambda_stats.distribution(
+        "{}.metrics_forwarded".format(DD_FORWARDER_TELEMETRY_NAMESPACE_PREFIX),
+        len(metrics),
+        tags=DD_FORWARDER_TELEMETRY_TAGS,
+    )
+
 
 def forward_traces(trace_payloads):
     try:
@@ -687,20 +723,37 @@ def forward_traces(trace_payloads):
         if logger.isEnabledFor(logging.DEBUG):
             logger.debug(f"Forwarded traces: {json.dumps(trace_payloads)}")
 
+    lambda_stats.distribution(
+        "{}.traces_forwarded".format(DD_FORWARDER_TELEMETRY_NAMESPACE_PREFIX),
+        len(trace_payloads),
+        tags=DD_FORWARDER_TELEMETRY_TAGS,
+    )
+
 
 # Utility functions
 
 
 def normalize_events(events, metadata):
     normalized = []
+    events_counter = 0
+
     for event in events:
+        events_counter += 1
         if isinstance(event, dict):
             normalized.append(merge_dicts(event, metadata))
         elif isinstance(event, str):
             normalized.append(merge_dicts({"message": event}, metadata))
         else:
             # drop this log
             continue
+
+    """Submit count of total events"""
+    lambda_stats.distribution(
+        "{}.incoming_events".format(DD_FORWARDER_TELEMETRY_NAMESPACE_PREFIX),
+        events_counter,
+        tags=DD_FORWARDER_TELEMETRY_TAGS,
+    )
+
     return normalized
 
 

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

@@ -44,6 +44,57 @@
           "host": "testLogGroup"
         }
       ]
+    },
+    {
+      "path": "/api/v1/distribution_points?api_key=abcdefghijklmnopqrstuvwxyz012345",
+      "verb": "POST",
+      "content-type": "application/json",
+      "data": {
+        "series": [
+          {
+            "metric": "aws.dd_forwarder.incoming_events",
+            "points": "<redacted from snapshot>",
+            "type": "distribution",
+            "host": null,
+            "device": null,
+            "tags": [
+              "forwardername:test",
+              "forwarder_memorysize:1536",
+              "forwarder_version:<redacted from snapshot>",
+              "event_type:awslogs"
+            ],
+            "interval": 10
+          },
+          {
+            "metric": "aws.dd_forwarder.logs_forwarded",
+            "points": "<redacted from snapshot>",
+            "type": "distribution",
+            "host": null,
+            "device": null,
+            "tags": [
+              "forwardername:test",
+              "forwarder_memorysize:1536",
+              "forwarder_version:<redacted from snapshot>",
+              "event_type:awslogs"
+            ],
+            "interval": 10
+          },
+          {
+            "metric": "aws.dd_forwarder.metrics_forwarded",
+            "points": "<redacted from snapshot>",
+            "type": "distribution",
+            "host": null,
+            "device": null,
+            "tags": [
+              "forwardername:test",
+              "forwarder_memorysize:1536",
+              "forwarder_version:<redacted from snapshot>",
+              "event_type:awslogs"
+            ],
+            "interval": 10
+          }
+        ]
+      }
     }
   ]
 }

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

@@ -72,6 +72,57 @@
           "host": "arn:aws:lambda:us-east-1:0:function:storms-cloudwatch-event"
         }
       ]
+    },
+    {
+      "path": "/api/v1/distribution_points?api_key=abcdefghijklmnopqrstuvwxyz012345",
+      "verb": "POST",
+      "content-type": "application/json",
+      "data": {
+        "series": [
+          {
+            "metric": "aws.dd_forwarder.incoming_events",
+            "points": "<redacted from snapshot>",
+            "type": "distribution",
+            "host": null,
+            "device": null,
+            "tags": [
+              "forwardername:test",
+              "forwarder_memorysize:1536",
+              "forwarder_version:<redacted from snapshot>",
+              "event_type:awslogs"
+            ],
+            "interval": 10
+          },
+          {
+            "metric": "aws.dd_forwarder.logs_forwarded",
+            "points": "<redacted from snapshot>",
+            "type": "distribution",
+            "host": null,
+            "device": null,
+            "tags": [
+              "forwardername:test",
+              "forwarder_memorysize:1536",
+              "forwarder_version:<redacted from snapshot>",
+              "event_type:awslogs"
+            ],
+            "interval": 10
+          },
+          {
+            "metric": "aws.dd_forwarder.metrics_forwarded",
+            "points": "<redacted from snapshot>",
+            "type": "distribution",
+            "host": null,
+            "device": null,
+            "tags": [
+              "forwardername:test",
+              "forwarder_memorysize:1536",
+              "forwarder_version:<redacted from snapshot>",
+              "event_type:awslogs"
+            ],
+            "interval": 10
+          }
+        ]
+      }
     }
   ]
 }

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

@@ -25,6 +25,57 @@
           "host": "testLogGroup"
         }
       ]
+    },
+    {
+      "path": "/api/v1/distribution_points?api_key=abcdefghijklmnopqrstuvwxyz012345",
+      "verb": "POST",
+      "content-type": "application/json",
+      "data": {
+        "series": [
+          {
+            "metric": "aws.dd_forwarder.incoming_events",
+            "points": "<redacted from snapshot>",
+            "type": "distribution",
+            "host": null,
+            "device": null,
+            "tags": [
+              "forwardername:test",
+              "forwarder_memorysize:1536",
+              "forwarder_version:<redacted from snapshot>",
+              "event_type:awslogs"
+            ],
+            "interval": 10
+          },
+          {
+            "metric": "aws.dd_forwarder.logs_forwarded",
+            "points": "<redacted from snapshot>",
+            "type": "distribution",
+            "host": null,
+            "device": null,
+            "tags": [
+              "forwardername:test",
+              "forwarder_memorysize:1536",
+              "forwarder_version:<redacted from snapshot>",
+              "event_type:awslogs"
+            ],
+            "interval": 10
+          },
+          {
+            "metric": "aws.dd_forwarder.metrics_forwarded",
+            "points": "<redacted from snapshot>",
+            "type": "distribution",
+            "host": null,
+            "device": null,
+            "tags": [
+              "forwardername:test",
+              "forwarder_memorysize:1536",
+              "forwarder_version:<redacted from snapshot>",
+              "event_type:awslogs"
+            ],
+            "interval": 10
+          }
+        ]
+      }
     }
   ]
 }