[Airflow integration - logs pipeline asset] Add support for new timestamp format (DataDog#22730)

* date parsing rules to supprot old and new formats

* add test with timestamp in new format

* keep samples as before

* test all formats

Author: aboitreaud

airflow/assets/logs/airflow.yaml

@@ -38,11 +38,15 @@ pipeline:
           Exception: Hello world fail!
       grok:
         supportRules: |
-          _date %{date("yyyy-MM-dd HH:mm:ss,SSS"):timestamp}
+          _date_legacy %{date("yyyy-MM-dd HH:mm:ss,SSS"):timestamp}
+          _date_iso_offset %{date("yyyy-MM-dd'T'HH:mm:ss.SSSZ"):timestamp}
+          _date_iso_zulu %{date("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'"):timestamp}
+          _date (%{_date_iso_offset}|%{_date_iso_zulu}|%{_date_legacy})
+          _date_any (%{date("yyyy-MM-dd'T'HH:mm:ss.SSSZ")}|%{date("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'")}|%{date("yyyy-MM-dd HH:mm:ss,SSS")})
           _file %{notSpace:filename}:%{integer:lineno}
           _base_prefix \[%{_date}\]\s+(\{\{%{_file}\}\}|\{%{_file}\})\s+%{word:level}\s+-
           # Some logs have duplicate prefixes, we only collect job_id and subtask name from first prefix if present.
-          _prefix (\[%{date("yyyy-MM-dd HH:mm:ss,SSS")}\]\s+\{\{%{notSpace}:%{integer}\}\}\s+%{word}\s+-\s+)?(Job %{integer:jobid}: Subtask %{word:subtask}\s+)?%{_base_prefix}
+          _prefix (\[%{_date_any}\]\s+(\{\{%{notSpace}:%{integer}\}\}|\{%{notSpace}:%{integer}\})\s+%{word}\s+-\s+)?(Job %{integer:jobid}: Subtask %{word:subtask}\s+)?%{_base_prefix}
 
         matchRules: |
           airflow_stacktrace %{_prefix}\s+%{data:message}\nTraceback \(most recent call last\):\n%{data:error.stack}

airflow/assets/logs/airflow_tests.yaml

@@ -214,6 +214,59 @@ tests:
     tags:
      - "source:LOGS_SOURCE"
     timestamp: 1577381689039
+ -
+  sample: "[2025-04-10T17:15:01.220Z] {logging_mixin.py:188} INFO - Running task"
+  result:
+    custom:
+      filename: "logging_mixin.py"
+      level: "INFO"
+      lineno: 188
+      timestamp: 1744305301220
+    message: "Running task"
+    status: "info"
+    tags:
+     - "source:LOGS_SOURCE"
+    timestamp: 1744305301220
+ -
+  sample: "[2025-04-10T17:15:01.220+0100] {base_task_runner.py:115} INFO - Job 12: Subtask my_task [2025-04-10T17:15:01.450+0100] {settings.py:211} INFO - Setting up DB connection pool (PID 42)"
+  result:
+    custom:
+      filename: "settings.py"
+      jobid: 12
+      level: "INFO"
+      lineno: 211
+      subtask: "my_task"
+      timestamp: 1744301701450
+    message: "Setting up DB connection pool (PID 42)"
+    status: "info"
+    tags:
+     - "source:LOGS_SOURCE"
+    timestamp: 1744301701450
+ -
+  sample: |
+    [2025-04-10T17:15:01.220+0000] {taskinstance.py:1058} ERROR - Task failed!
+    Traceback (most recent call last):
+      File "/usr/local/lib/python3.11/site-packages/airflow/operators/python.py", line 118, in execute_callable
+      return self.python_callable(*self.op_args, **self.op_kwargs)
+    Exception: Task failed!
+  result:
+    custom:
+      error:
+        kind: "Exception"
+        message: "Task failed!"
+        stack: |2-
+            File "/usr/local/lib/python3.11/site-packages/airflow/operators/python.py", line 118, in execute_callable
+            return self.python_callable(*self.op_args, **self.op_kwargs)
+          Exception: Task failed!
+      filename: "taskinstance.py"
+      level: "ERROR"
+      lineno: 1058
+      timestamp: 1744305301220
+    message: "Task failed!"
+    status: "error"
+    tags:
+     - "source:LOGS_SOURCE"
+    timestamp: 1744305301220
  -
   sample: "[2025-02-18 14:46:00,123] {{python_operator.py:105}} INFO - Task started"
   tags: