es_task_handler.py crashes accessing "View Logs in Elasticsearch"

[ejk43]

Apache Airflow version

2.3.2 (latest released)

What happened

Hi,

I have a local setup of airflow 2.3.2 running with airflow standalone. I am also running a docker-compose elasticsearch + kibana, and trying to exercise the elasticsearch backend integrations within Airflow...

I configured the frontend parameter in airflow.cfg, but when I click the link to "View Logs in Elasticsearch" I get the following failure in the webserver:

 webserver | [2022-06-17 16:10:39,448] {app.py:1891} ERROR - Exception on /redirect_to_external_log [GET]
 webserver | Traceback (most recent call last):
 webserver | File "/home/ejk/miniconda3/envs/local-airflow/lib/python3.8/site-packages/flask/app.py", line 2447, in wsgi_app
 webserver | response = self.full_dispatch_request()
 webserver | File "/home/ejk/miniconda3/envs/local-airflow/lib/python3.8/site-packages/flask/app.py", line 1952, in full_dispatch_request
 webserver | rv = self.handle_user_exception(e)
 webserver | File "/home/ejk/miniconda3/envs/local-airflow/lib/python3.8/site-packages/flask/app.py", line 1821, in handle_user_exception
 webserver | reraise(exc_type, exc_value, tb)
 webserver | File "/home/ejk/miniconda3/envs/local-airflow/lib/python3.8/site-packages/flask/_compat.py", line 39, in reraise
 webserver | raise value
 webserver | File "/home/ejk/miniconda3/envs/local-airflow/lib/python3.8/site-packages/flask/app.py", line 1950, in full_dispatch_request
 webserver | rv = self.dispatch_request()
 webserver | File "/home/ejk/miniconda3/envs/local-airflow/lib/python3.8/site-packages/flask/app.py", line 1936, in dispatch_request
 webserver | return self.view_functions[rule.endpoint](**req.view_args)
 webserver | File "/home/ejk/miniconda3/envs/local-airflow/lib/python3.8/site-packages/airflow/www/auth.py", line 43, in decorated
 webserver | return func(*args, **kwargs)
 webserver | File "/home/ejk/miniconda3/envs/local-airflow/lib/python3.8/site-packages/airflow/www/decorators.py", line 80, in wrapper
 webserver | return f(*args, **kwargs)
 webserver | File "/home/ejk/miniconda3/envs/local-airflow/lib/python3.8/site-packages/airflow/utils/session.py", line 71, in wrapper
 webserver | return func(*args, session=session, **kwargs)
 webserver | File "/home/ejk/miniconda3/envs/local-airflow/lib/python3.8/site-packages/airflow/www/views.py", line 1588, in redirect_to_external_log
 webserver | url = handler.get_external_log_url(ti, try_number)
 webserver | File "/home/ejk/miniconda3/envs/local-airflow/lib/python3.8/site-packages/airflow/providers/elasticsearch/log/es_task_handler.py", line 395, in get_external_log_url
 webserver | log_id = self._render_log_id(task_instance, try_number)
 webserver | File "/home/ejk/miniconda3/envs/local-airflow/lib/python3.8/site-packages/airflow/providers/elasticsearch/log/es_task_handler.py", line 128, in _render_log_id
 webserver | dag = ti.task.dag
 webserver | AttributeError: 'TaskInstance' object has no attribute 'task'

What you think should happen instead

This crash appears to trace back to this commit: 7be87d

If I edit es_task_handler.py to move this dag = ti.task.dag assignment BACK into the try/except statement (7be87d#diff-dd898ab2ed4bca853f1ce5cf52b6fbb37d5fc3545f28967c03dc499fabd3a746R107), I can successfully open Kibana from the link in the Airflow UI

Not sure what might be going wrong in the backend here, e.g. if ti.task is now supposed to be always valid? Or if something I'm doing got into a bad state... But the try/except was able to handle the failure gracefully, at least.

How to reproduce

1. Install airflow[elasticsearch] into a conda environment
2. Run elasticsearch using the docker-compose instructions: https://www.elastic.co/guide/en/elastic-stack-get-started/current/get-started-stack-docker.html#get-started-docker-tls
3. Start airflow using airflow standalone
4. Edit the generated airflow.cfg to: 1) Point to elasticsearch host URL, 2) json_format = True, 3) frontend = determined by the URL that opens kibana for me

Then, inside the Airflow UI, I turned on a few example DAGs. Specifically I triggered the Example Python Operator, then clicked the button to view in Kibana. I tried a few DAGs and tasks and found this crash all the time

Operating System

Ubuntu 20.04.4 LTS

Versions of Apache Airflow Providers

apache-airflow-providers-elasticsearch==4.0.0
apache-airflow-providers-ftp==2.1.2
apache-airflow-providers-http==2.1.2
apache-airflow-providers-imap==2.2.3
apache-airflow-providers-sqlite==2.1.3

Deployment

Virtualenv installation

Deployment details

No response

Anything else

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template!

[potiuk]

I think this has something to do with the #24511 that is likely going to be fixed in 2.3.3

[potiuk]

CC: @jedcunningham ?

[jedcunningham]

Nope, unfortunately #24511 won't fix this issue. I wonder why we don't see it with the file handler though.

I'm thinking it was unintentional. @uranusjr, do you remember?

@ejk43, want to open a PR pulling that (and the assert) back into the try?

[ejk43]

Sure I can definitely spin out a quick PR-- as long as moving the assignment & assert into the try statement is the extent of the changes you have in mind... I'm afraid I dont know enough of the context to understand what else might be going wrong.

I was guessing it was at least semi-intentional, since I also see the same pattern in file_task_handler.py in the same commit: 7be87d#diff-e7f34f73940eb52d92bb991abedc1c963431c5373c12dff739c8fb7d03e93d3aR87

[eladkal]

Does #24592 fixes the issue or we still have further tasks here?

[uranusjr]

I think #24592 is the fix to this.