Add log_id field to log lines on ES handler (#10411)

* Add `log_id` field to log lines on ES handler

* Add `offset` field to log lines on ES handler

it will be set to the epoch timestamp in nanoseconds (this will just be
used for ordering log lines when displayed in the webserver UI).

* Update UPDATING.md

With information regarding log_id and offset fields in JSON log lines written to stdout

Author: eyalzek

UPDATING.md

@@ -518,6 +518,20 @@ The following configurations have been moved from `[core]` to the new `[logging]
 * `dag_processor_manager_log_location`
 * `task_log_reader`
 
+#### Changes to Elasticsearch logging provider
+
+When JSON output to stdout is enabled, log lines will now contain the `log_id` & `offset` fields, this should make reading task logs from elasticsearch on the webserver work out of the box. Example configuration:
+```ini
+[logging]
+remote_logging = True
+[elasticsearch]
+host = http://es-host:9200
+write_stdout = True
+json_format = True
+```
+
+Note that the webserver expects the log line data itself to be present in the `message` field of the document.
+
 #### Remove gcp_service_account_keys option in airflow.cfg file
 
 This option has been removed because it is no longer supported by the Google Kubernetes Engine. The new

airflow/providers/elasticsearch/log/es_task_handler.py

@@ -19,6 +19,7 @@
 import logging
 import sys
 from datetime import datetime
+from time import time
 from typing import Optional, Tuple
 from urllib.parse import quote
 
@@ -224,6 +225,8 @@ def set_context(self, ti: TaskInstance) -> None:
                     'task_id': str(ti.task_id),
                     'execution_date': self._clean_execution_date(ti.execution_date),
                     'try_number': str(ti.try_number),
+                    'log_id': self._render_log_id(ti, ti.try_number),
+                    'offset': int(time() * (10 ** 9)),
                 },
             )