diff --git a/airflow/www/decorators.py b/airflow/www/decorators.py index 080fe682991c2..47a9847c2eecf 100644 --- a/airflow/www/decorators.py +++ b/airflow/www/decorators.py @@ -20,6 +20,7 @@ import gzip import logging from io import BytesIO as IO +from itertools import chain from typing import Callable, TypeVar, cast import pendulum @@ -48,13 +49,19 @@ def wrapper(*args, **kwargs): user = g.user.username fields_skip_logging = {'csrf_token', '_csrf_token'} + log_fields = { + k: v + for k, v in chain(request.values.items(), request.view_args.items()) + if k not in fields_skip_logging + } + log = Log( event=f.__name__, task_instance=None, owner=user, - extra=str([(k, v) for k, v in request.values.items() if k not in fields_skip_logging]), - task_id=request.values.get('task_id'), - dag_id=request.values.get('dag_id'), + extra=str([(k, log_fields[k]) for k in log_fields]), + task_id=log_fields.get('task_id'), + dag_id=log_fields.get('dag_id'), ) if 'execution_date' in request.values: diff --git a/airflow/www/templates/airflow/dag.html b/airflow/www/templates/airflow/dag.html index e57a1f1fc8b3b..0f4b967373da5 100644 --- a/airflow/www/templates/airflow/dag.html +++ b/airflow/www/templates/airflow/dag.html @@ -110,9 +110,9 @@

- + Zoom into Sub DAG
@@ -402,7 +402,7 @@
more_horiz diff --git a/airflow/www/utils.py b/airflow/www/utils.py index cea50114e12b1..8cc420caccc5b 100644 --- a/airflow/www/utils.py +++ b/airflow/www/utils.py @@ -356,8 +356,10 @@ def dag_link(attr): """Generates a URL to the Graph view for a Dag.""" dag_id = attr.get('dag_id') execution_date = attr.get('execution_date') + if not dag_id: + return Markup('None') url = url_for('Airflow.graph', dag_id=dag_id, execution_date=execution_date) - return Markup('{}').format(url, dag_id) if dag_id else Markup('None') + return Markup('{}').format(url, dag_id) def dag_run_link(attr): diff --git a/airflow/www/views.py b/airflow/www/views.py index b73530170bc87..b255ec67e8ff0 100644 --- a/airflow/www/views.py +++ b/airflow/www/views.py @@ -1054,15 +1054,24 @@ def last_dagruns(self, session=None): (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG_CODE), ] ) + def legacy_code(self): + """Redirect from url param.""" + return redirect(url_for('Airflow.code', **request.args)) + + @expose('/dags//code') + @auth.has_access( + [ + (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG), + (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG_CODE), + ] + ) @provide_session - def code(self, session=None): + def code(self, dag_id, session=None): """Dag Code.""" all_errors = "" dag_orm = None - dag_id = None try: - dag_id = request.args.get('dag_id') dag_orm = DagModel.get_dagmodel(dag_id, session=session) code = DagCode.get_code_by_fileloc(dag_orm.fileloc) html_code = Markup(highlight(code, lexers.PythonLexer(), HtmlFormatter(linenos=True))) @@ -1095,10 +1104,20 @@ def code(self, session=None): (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG_RUN), ] ) + def legacy_dag_details(self): + """Redirect from url param.""" + return redirect(url_for('Airflow.dag_details', **request.args)) + + @expose('/dags//details') + @auth.has_access( + [ + (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG), + (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG_RUN), + ] + ) @provide_session - def dag_details(self, session=None): + def dag_details(self, dag_id, session=None): """Get Dag details.""" - dag_id = request.args.get('dag_id') dag = current_app.dag_bag.get_dag(dag_id) dag_model = DagModel.get_dagmodel(dag_id) @@ -2301,6 +2320,34 @@ def success(self): State.SUCCESS, ) + @expose('/dags/') + @auth.has_access( + [ + (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG), + (permissions.ACTION_CAN_READ, permissions.RESOURCE_TASK_INSTANCE), + (permissions.ACTION_CAN_READ, permissions.RESOURCE_TASK_LOG), + ] + ) + @gzipped + @action_logging + def dag(self, dag_id): + """Redirect to default DAG view.""" + return redirect(url_for('Airflow.grid', dag_id=dag_id, **request.args)) + + @expose('/legacy_tree') + @auth.has_access( + [ + (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG), + (permissions.ACTION_CAN_READ, permissions.RESOURCE_TASK_INSTANCE), + (permissions.ACTION_CAN_READ, permissions.RESOURCE_TASK_LOG), + ] + ) + @gzipped + @action_logging + def legacy_tree(self): + """Redirect to the replacement - grid view.""" + return redirect(url_for('Airflow.grid', **request.args)) + @expose('/tree') @auth.has_access( [ @@ -2311,10 +2358,23 @@ def success(self): ) @gzipped @action_logging + def tree(self): + """Redirect to the replacement - grid view. Kept for backwards compatibility.""" + return redirect(url_for('Airflow.grid', **request.args)) + + @expose('/dags//grid') + @auth.has_access( + [ + (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG), + (permissions.ACTION_CAN_READ, permissions.RESOURCE_TASK_INSTANCE), + (permissions.ACTION_CAN_READ, permissions.RESOURCE_TASK_LOG), + ] + ) + @gzipped + @action_logging @provide_session - def tree(self, session=None): - """Get Dag as tree.""" - dag_id = request.args.get('dag_id') + def grid(self, dag_id, session=None): + """Get Dag's grid view.""" dag = current_app.dag_bag.get_dag(dag_id) dag_model = DagModel.get_dagmodel(dag_id) if not dag: @@ -2399,8 +2459,21 @@ def tree(self, session=None): ) @gzipped @action_logging + def legacy_calendar(self): + """Redirect from url param.""" + return redirect(url_for('Airflow.calendar', **request.args)) + + @expose('/dags//calendar') + @auth.has_access( + [ + (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG), + (permissions.ACTION_CAN_READ, permissions.RESOURCE_TASK_INSTANCE), + ] + ) + @gzipped + @action_logging @provide_session - def calendar(self, session=None): + def calendar(self, dag_id, session=None): """Get DAG runs as calendar""" def _convert_to_date(session, column): @@ -2410,7 +2483,6 @@ def _convert_to_date(session, column): else: return func.date(column) - dag_id = request.args.get('dag_id') dag = current_app.dag_bag.get_dag(dag_id) dag_model = DagModel.get_dagmodel(dag_id) if not dag: @@ -2476,10 +2548,23 @@ def _convert_to_date(session, column): ) @gzipped @action_logging + def legacy_graph(self): + """Redirect from url param.""" + return redirect(url_for('Airflow.graph', **request.args)) + + @expose('/dags//graph') + @auth.has_access( + [ + (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG), + (permissions.ACTION_CAN_READ, permissions.RESOURCE_TASK_INSTANCE), + (permissions.ACTION_CAN_READ, permissions.RESOURCE_TASK_LOG), + ] + ) + @gzipped + @action_logging @provide_session - def graph(self, session=None): + def graph(self, dag_id, session=None): """Get DAG as Graph.""" - dag_id = request.args.get('dag_id') dag = current_app.dag_bag.get_dag(dag_id) dag_model = DagModel.get_dagmodel(dag_id) if not dag: @@ -2569,11 +2654,22 @@ class GraphForm(DateTimeWithNumRunsWithDagRunsForm): ] ) @action_logging + def legacy_duration(self): + """Redirect from url param.""" + return redirect(url_for('Airflow.duration', **request.args)) + + @expose('/dags//duration') + @auth.has_access( + [ + (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG), + (permissions.ACTION_CAN_READ, permissions.RESOURCE_TASK_INSTANCE), + ] + ) + @action_logging @provide_session - def duration(self, session=None): + def duration(self, dag_id, session=None): """Get Dag as duration graph.""" default_dag_run = conf.getint('webserver', 'default_dag_run_display_number') - dag_id = request.args.get('dag_id') dag_model = DagModel.get_dagmodel(dag_id) dag: Optional[DAG] = current_app.dag_bag.get_dag(dag_id) @@ -2711,11 +2807,22 @@ def duration(self, session=None): ] ) @action_logging + def legacy_tries(self): + """Redirect from url param.""" + return redirect(url_for('Airflow.tries', **request.args)) + + @expose('/dags//tries') + @auth.has_access( + [ + (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG), + (permissions.ACTION_CAN_READ, permissions.RESOURCE_TASK_INSTANCE), + ] + ) + @action_logging @provide_session - def tries(self, session=None): + def tries(self, dag_id, session=None): """Shows all tries.""" default_dag_run = conf.getint('webserver', 'default_dag_run_display_number') - dag_id = request.args.get('dag_id') dag = current_app.dag_bag.get_dag(dag_id) dag_model = DagModel.get_dagmodel(dag_id) base_date = request.args.get('base_date') @@ -2788,11 +2895,22 @@ def tries(self, session=None): ] ) @action_logging + def legacy_landing_times(self): + """Redirect from url param.""" + return redirect(url_for('Airflow.landing_times', **request.args)) + + @expose('/dags//landing-times') + @auth.has_access( + [ + (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG), + (permissions.ACTION_CAN_READ, permissions.RESOURCE_TASK_INSTANCE), + ] + ) + @action_logging @provide_session - def landing_times(self, session=None): + def landing_times(self, dag_id, session=None): """Shows landing times.""" default_dag_run = conf.getint('webserver', 'default_dag_run_display_number') - dag_id = request.args.get('dag_id') dag: DAG = current_app.dag_bag.get_dag(dag_id) dag_model = DagModel.get_dagmodel(dag_id) base_date = request.args.get('base_date') @@ -2893,10 +3011,21 @@ def paused(self): ] ) @action_logging + def legacy_gantt(self): + """Redirect from url param.""" + return redirect(url_for('Airflow.gantt', **request.args)) + + @expose('/dags//gantt') + @auth.has_access( + [ + (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG), + (permissions.ACTION_CAN_READ, permissions.RESOURCE_TASK_INSTANCE), + ] + ) + @action_logging @provide_session - def gantt(self, session=None): + def gantt(self, dag_id, session=None): """Show GANTT chart.""" - dag_id = request.args.get('dag_id') dag = current_app.dag_bag.get_dag(dag_id) dag_model = DagModel.get_dagmodel(dag_id) diff --git a/tests/www/views/test_views_decorators.py b/tests/www/views/test_views_decorators.py index 9b5383f2d654e..c211806bd1b04 100644 --- a/tests/www/views/test_views_decorators.py +++ b/tests/www/views/test_views_decorators.py @@ -115,7 +115,7 @@ def _check_last_log(session, dag_id, event, execution_date): def test_action_logging_get(session, admin_client): url = ( - f'graph?dag_id=example_bash_operator&' + f'dags/example_bash_operator/graph?' f'execution_date={urllib.parse.quote_plus(str(EXAMPLE_DAG_DEFAULT_DATE))}' ) resp = admin_client.get(url, follow_redirects=True) @@ -131,6 +131,24 @@ def test_action_logging_get(session, admin_client): ) +def test_action_logging_get_legacy_view(session, admin_client): + url = ( + f'graph?dag_id=example_bash_operator&' + f'execution_date={urllib.parse.quote_plus(str(EXAMPLE_DAG_DEFAULT_DATE))}' + ) + resp = admin_client.get(url, follow_redirects=True) + check_content_in_response('runme_1', resp) + + # In mysql backend, this commit() is needed to write down the logs + session.commit() + _check_last_log( + session, + dag_id="example_bash_operator", + event="legacy_graph", + execution_date=EXAMPLE_DAG_DEFAULT_DATE, + ) + + def test_action_logging_post(session, admin_client): form = dict( task_id="runme_1", diff --git a/tests/www/views/test_views_tasks.py b/tests/www/views/test_views_tasks.py index d433542fd3264..bd0c7badd46f2 100644 --- a/tests/www/views/test_views_tasks.py +++ b/tests/www/views/test_views_tasks.py @@ -139,16 +139,26 @@ def client_ti_without_dag_edit(app): pytest.param( 'dag_details?dag_id=example_bash_operator', ['DAG Details'], - id="dag-details", + id="dag-details-url-param", ), pytest.param( 'dag_details?dag_id=example_subdag_operator.section-1', ['DAG Details'], + id="dag-details-subdag-url-param", + ), + pytest.param( + 'dags/example_subdag_operator.section-1/details', + ['DAG Details'], id="dag-details-subdag", ), pytest.param( 'graph?dag_id=example_bash_operator', ['runme_1'], + id='graph-url-param', + ), + pytest.param( + 'dags/example_bash_operator/graph', + ['runme_1'], id='graph', ), pytest.param( @@ -156,34 +166,69 @@ def client_ti_without_dag_edit(app): ['runme_1'], id='tree', ), + pytest.param( + 'dags/example_bash_operator/grid', + ['runme_1'], + id='grid', + ), pytest.param( 'tree?dag_id=example_subdag_operator.section-1', ['section-1-task-1'], - id="tree-subdag", + id="tree-subdag-url-param", + ), + pytest.param( + 'dags/example_subdag_operator.section-1/grid', + ['section-1-task-1'], + id="grid-subdag", ), pytest.param( 'duration?days=30&dag_id=example_bash_operator', ['example_bash_operator'], + id='duration-url-param', + ), + pytest.param( + 'dags/example_bash_operator/duration?days=30', + ['example_bash_operator'], id='duration', ), pytest.param( 'duration?days=30&dag_id=missing_dag', ['seems to be missing'], + id='duration-missing-url-param', + ), + pytest.param( + 'dags/missing_dag/duration?days=30', + ['seems to be missing'], id='duration-missing', ), pytest.param( 'tries?days=30&dag_id=example_bash_operator', ['example_bash_operator'], + id='tries-url-param', + ), + pytest.param( + 'dags/example_bash_operator/tries?days=30', + ['example_bash_operator'], id='tries', ), pytest.param( 'landing_times?days=30&dag_id=example_bash_operator', ['example_bash_operator'], + id='landing-times-url-param', + ), + pytest.param( + 'dags/example_bash_operator/landing-times?days=30', + ['example_bash_operator'], id='landing-times', ), pytest.param( 'gantt?dag_id=example_bash_operator', ['example_bash_operator'], + id="gantt-url-param", + ), + pytest.param( + 'dags/example_bash_operator/gantt', + ['example_bash_operator'], id="gantt", ), pytest.param( @@ -196,21 +241,41 @@ def client_ti_without_dag_edit(app): pytest.param( "graph?dag_id=example_bash_operator", ["example_bash_operator"], + id="existing-dagbag-graph-url-param", + ), + pytest.param( + "dags/example_bash_operator/graph", + ["example_bash_operator"], id="existing-dagbag-graph", ), pytest.param( "tree?dag_id=example_bash_operator", ["example_bash_operator"], - id="existing-dagbag-tree", + id="existing-dagbag-tree-url-param", + ), + pytest.param( + "dags/example_bash_operator/grid", + ["example_bash_operator"], + id="existing-dagbag-grid", ), pytest.param( "calendar?dag_id=example_bash_operator", ["example_bash_operator"], + id="existing-dagbag-calendar-url-param", + ), + pytest.param( + "dags/example_bash_operator/calendar", + ["example_bash_operator"], id="existing-dagbag-calendar", ), pytest.param( "dag_details?dag_id=example_bash_operator", ["example_bash_operator"], + id="existing-dagbag-dag-details-url-param", + ), + pytest.param( + "dags/example_bash_operator/details", + ["example_bash_operator"], id="existing-dagbag-dag-details", ), pytest.param( @@ -274,7 +339,7 @@ def test_tree_trigger_origin_tree_view(app, admin_client): url = 'tree?dag_id=test_tree_view' resp = admin_client.get(url, follow_redirects=True) - params = {'dag_id': 'test_tree_view', 'origin': '/tree?dag_id=test_tree_view'} + params = {'dag_id': 'test_tree_view', 'origin': '/dags/test_tree_view/grid'} href = f"/trigger?{html.escape(urllib.parse.urlencode(params))}" check_content_in_response(href, resp) @@ -288,9 +353,9 @@ def test_graph_trigger_origin_graph_view(app, admin_client): state=State.RUNNING, ) - url = 'graph?dag_id=test_tree_view' + url = '/dags/test_tree_view/graph' resp = admin_client.get(url, follow_redirects=True) - params = {'dag_id': 'test_tree_view', 'origin': '/graph?dag_id=test_tree_view'} + params = {'dag_id': 'test_tree_view', 'origin': '/dags/test_tree_view/graph'} href = f"/trigger?{html.escape(urllib.parse.urlencode(params))}" check_content_in_response(href, resp) @@ -304,9 +369,9 @@ def test_dag_details_trigger_origin_dag_details_view(app, admin_client): state=State.RUNNING, ) - url = 'dag_details?dag_id=test_graph_view' + url = '/dags/test_graph_view/details' resp = admin_client.get(url, follow_redirects=True) - params = {'dag_id': 'test_graph_view', 'origin': '/dag_details?dag_id=test_graph_view'} + params = {'dag_id': 'test_graph_view', 'origin': '/dags/test_graph_view/details'} href = f"/trigger?{html.escape(urllib.parse.urlencode(params))}" check_content_in_response(href, resp) @@ -348,7 +413,7 @@ def test_code_from_db(admin_client): dag = DagBag(include_examples=True).get_dag("example_bash_operator") DagCode(dag.fileloc, DagCode._get_code_from_file(dag.fileloc)).sync_to_db() url = 'code?dag_id=example_bash_operator' - resp = admin_client.get(url) + resp = admin_client.get(url, follow_redirects=True) check_content_not_in_response('Failed to load DAG file Code', resp) check_content_in_response('example_bash_operator', resp) @@ -358,7 +423,7 @@ def test_code_from_db_all_example_dags(admin_client): for dag in dagbag.dags.values(): DagCode(dag.fileloc, DagCode._get_code_from_file(dag.fileloc)).sync_to_db() url = 'code?dag_id=example_bash_operator' - resp = admin_client.get(url) + resp = admin_client.get(url, follow_redirects=True) check_content_not_in_response('Failed to load DAG file Code', resp) check_content_in_response('example_bash_operator', resp)