Airflow wrong log_id format-CodePudding

I'm using Airflow v2.2.3 and apache-airflow-providers-elasticsearch==2.1.0, running in Kubernetes.

Our logs are automatically shipped to Elasticsearch v7.6.2.

I set the following configurations in Airflow:

AIRFLOW__LOGGING__REMOTE_LOGGING=True
AIRFLOW__ELASTICSEARCH__HOST=<my-elasticsearch-host>:9200
AIRFLOW__ELASTICSEARCH__WRITE_STDOUT=True
AIRFLOW__ELASTICSEARCH__JSON_FORMAT=True
AIRFLOW__ELASTICSEARCH__LOG_ID_TEMPLATE={dag_id}-{task_id}-{execution_date}-{try_number}

I see the logs in stdout as json:

{
    "asctime": "2022-01-20 12:13:52,292",
    "filename": "taskinstance.py",
    "lineno": 1032,
    "levelname": "INFO",
    "message": "Dependencies all met for <TaskInstance: spark_jobs_8765280.check_rawevents_output scheduled__2022-01-19T07:00:00 00:00 [queued]>", 
    "offset": 1642680832292384000,
    "dag_id": "spark_jobs_8765280",
    "task_id": "check_raweven ts_output",
    "execution_date": "2022_01_19T07_00_00_000000",
    "try_number": "1",
    "log_id": "spark_jobs_8765280-check_rawevents_output-2022_01_19T07_00_00_000000-1"
}

and I do see those fields in Elasticsearch as well - so far so good. Now I'm expecting to see those logs via Airflow's web-server UI and nothing happened (airflow failed to retrieve the logs).

When I tried to manually update the log's document and change the format of "log_id" from:

"log_id": "spark_jobs_8765280-check_rawevents_output-2022_01_19T07_00_00_000000-1"

"log_id": "spark_jobs_8765280-check_rawevents_output-2022-01-19T07:00:00 00:00-1"

I DO see the logs in the UI!

Why does Airflow's UI try to get the log_id with the non escaped log_id rather than the correct one?

P.S - here is the log in Airflow web server:

[2022-01-20 14:08:21,170] {base.py:270} INFO - POST http://<my-elasticsearch-host>:9200/_count [status:200 request:0.008s]
10.1.19.65 - - [20/Jan/2022:14:08:21  0000] "GET /get_logs_with_metadata?dag_id=spark_jobs_8765280&task_id=check_rawevents_output&execution_date=2022-01-19T07:00:00+00:00&try_number=1&metadata={"end_of_log":false,"last_log_timestamp":"2022-01-20T14:05:55.439492+00:00","offset":"1691"} HTTP/1.1" 200 119 "https://<my-airflow-webserver-host>/log?dag_id=spark_jobs_8765280&task_id=check_rawevents_output&execution_date=2022-01-19T07:00:00+00:00" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/97.0.4692.71 Safari/537.36"

CodePudding user response：

I found the problem. I have a separate pod for airflow-scheduler and airflow-webserver.

I added the AIRFLOW__ELASTICSEARCH__JSON_FORMAT=True only to the airflow-scheduler the workers but not to the airflow-webserver.

I dig into the source code and I found that the webserver checks for the AIRFLOW__ELASTICSEARCH__JSON_FORMAT as well in order to transform the log_id in order to clean the dates to the right format:

if self.json_format:
    data_interval_start = self._clean_date(dag_run.data_interval_start)
    data_interval_end = self._clean_date(dag_run.data_interval_end)
    execution_date = self._clean_date(dag_run.execution_date)
else:
    data_interval_start = dag_run.data_interval_start.isoformat()
    data_interval_end = dag_run.data_interval_end.isoformat()
    execution_date = dag_run.execution_date.isoformat()