SLS-2070: Adds step function forwarding to forwarder (#546)

Adds step functions logs forwarding

aws/logs_monitoring/parsing.py

@@ -276,6 +276,9 @@ def find_cloudwatch_source(log_group):
     ):
         return "apigateway"
 
+    if log_group.startswith("/aws/vendedlogs/states"):
+        return "stepfunction"
+
     # e.g. dms-tasks-test-instance
     if log_group.startswith("dms-tasks"):
         return "dms"
@@ -482,6 +485,19 @@ def awslogs_handler(event, context, metadata):
     if metadata[DD_SOURCE] == "cloudwatch" or metadata.get(DD_HOST, None) == None:
         metadata[DD_HOST] = aws_attributes["aws"]["awslogs"]["logGroup"]
 
+    if metadata[DD_SOURCE] == "stepfunction" and logs["logStream"].startswith(
+        "states/"
+    ):
+        try:
+            message = json.loads(logs["logEvents"][0]["message"])
+            if message.get("execution_arn") is not None:
+                execution_arn = message["execution_arn"]
+                arn_tokens = execution_arn.split(":")
+                arn_tokens[5] = "stateMachine"
+                metadata[DD_HOST] = ":".join(arn_tokens[:-1])
+        except Exception as e:
+            logger.debug("Unable to set stepfunction host: %s" % e)
+
     # When parsing rds logs, use the cloudwatch log group name to derive the
     # rds instance name, and add the log name of the stream ingested
     if metadata[DD_SOURCE] in ["rds", "mariadb", "mysql", "postgresql"]:

aws/logs_monitoring/tests/test_parsing.py

@@ -271,6 +271,14 @@ def test_carbon_black_event(self):
             "carbonblack",
         )
 
+    def test_step_function_event(self):
+        self.assertEqual(
+            parse_event_source(
+                {"awslogs": "logs"}, "/aws/vendedlogs/states/MyStateMachine-Logs"
+            ),
+            "stepfunction",
+        )
+
     def test_cloudwatch_source_if_none_found(self):
         self.assertEqual(parse_event_source({"awslogs": "logs"}, ""), "cloudwatch")