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

Adds step functions logs forwarding

Author: IvanTopolcic

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")