Fokko closed pull request #4005: [AIRFLOW-3160] Load latest_dagruns
asynchronously, speed up front page load time
URL: https://github.com/apache/incubator-airflow/pull/4005
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/airflow/www/templates/airflow/dags.html
b/airflow/www/templates/airflow/dags.html
index 0a7a6ec2a4..e3eda43cad 100644
--- a/airflow/www/templates/airflow/dags.html
+++ b/airflow/www/templates/airflow/dags.html
@@ -118,15 +118,11 @@ <h2>DAGs</h2>
<!-- Column 7: Last Run -->
<td class="text-nowrap latest_dag_run {{ dag.dag_id }}">
- {% if dag %}
- {% set last_run =
dag.get_last_dagrun(include_externally_triggered=True) %}
- {% if last_run and last_run.execution_date %}
- <a href="{{ url_for('airflow.graph', dag_id=dag.dag_id,
execution_date=last_run.execution_date) }}">
- {{ last_run.execution_date.strftime("%Y-%m-%d %H:%M")
}}
- </a>
- <span aria-hidden="true" id="statuses_info" title="Start
Date: {{ last_run.start_date.strftime("%Y-%m-%d %H:%M") }}" class="glyphicon
glyphicon-info-sign"></span>
- {% endif %}
- {% endif %}
+ <div height="10" width="10" id='last-run-{{ dag.safe_dag_id
}}' style="display: block;">
+ <a></a>
+ <img class="loading-last-run" width="15" src="{{
url_for("static", filename="loading.gif") }}">
+ <span aria-hidden="true" id="statuses_info" title=" "
class="glyphicon glyphicon-info-sign" style="display:none"></span>
+ </div>
</td>
<!-- Column 8: Dag Runs -->
@@ -318,6 +314,24 @@ <h2>DAGs</h2>
}
});
});
+ d3.json("{{ url_for('airflow.last_dagruns') }}", function(error, json) {
+ for(var safe_dag_id in json) {
+ dag_id = json[safe_dag_id].dag_id;
+ last_run = json[safe_dag_id].last_run;
+ g = d3.select('div#last-run-' + safe_dag_id)
+
+ g.selectAll('a')
+ .attr("href", "{{ url_for('airflow.graph') }}?dag_id=" + dag_id +
"&execution_date=" + last_run)
+ .text(last_run);
+
+ g.selectAll('span')
+ .attr("data-original-title", "Start Date: " + last_run)
+ .style('display', null);
+
+ g.selectAll(".loading-last-run").remove();
+ }
+ d3.selectAll(".loading-last-run").remove();
+ });
d3.json("{{ url_for('airflow.dag_stats') }}", function(error, json) {
for(var dag_id in json) {
states = json[dag_id];
diff --git a/airflow/www/views.py b/airflow/www/views.py
index 0aef2281e7..769a15772f 100644
--- a/airflow/www/views.py
+++ b/airflow/www/views.py
@@ -654,6 +654,26 @@ def task_stats(self, session=None):
payload[dag.safe_dag_id].append(d)
return wwwutils.json_response(payload)
+ @expose('/last_dagruns')
+ @login_required
+ @provide_session
+ def last_dagruns(self, session=None):
+ DagRun = models.DagRun
+
+ dags_to_latest_runs = dict(session.query(
+ DagRun.dag_id,
sqla.func.max(DagRun.execution_date).label('execution_date'))
+ .group_by(DagRun.dag_id).all())
+
+ payload = {}
+ for dag in dagbag.dags.values():
+ if dag.dag_id in dags_to_latest_runs and
dags_to_latest_runs[dag.dag_id]:
+ payload[dag.safe_dag_id] = {
+ 'dag_id': dag.dag_id,
+ 'last_run':
dags_to_latest_runs[dag.dag_id].strftime("%Y-%m-%d %H:%M")
+ }
+
+ return wwwutils.json_response(payload)
+
@expose('/code')
@login_required
def code(self):
diff --git a/airflow/www_rbac/templates/airflow/dags.html
b/airflow/www_rbac/templates/airflow/dags.html
index 5f492780d5..b071d0ae7b 100644
--- a/airflow/www_rbac/templates/airflow/dags.html
+++ b/airflow/www_rbac/templates/airflow/dags.html
@@ -119,15 +119,11 @@ <h2>DAGs</h2>
<!-- Column 7: Last Run -->
<td class="text-nowrap latest_dag_run {{ dag.dag_id }}">
- {% if dag %}
- {% set last_run =
dag.get_last_dagrun(include_externally_triggered=True) %}
- {% if last_run and last_run.execution_date %}
- <a href="{{ url_for('Airflow.graph', dag_id=dag.dag_id,
execution_date=last_run.execution_date) }}">
- {{ last_run.execution_date.strftime("%Y-%m-%d %H:%M")
}}
- </a>
- <span aria-hidden="true" id="statuses_info" title="Start
Date: {{ last_run.start_date.strftime("%Y-%m-%d %H:%M") }}" class="glyphicon
glyphicon-info-sign"></span>
- {% endif %}
- {% endif %}
+ <div height="10" width="10" id='last-run-{{ dag.safe_dag_id
}}' style="display: block;">
+ <a></a>
+ <img class="loading-last-run" width="15" src="{{
url_for("static", filename="loading.gif") }}">
+ <span aria-hidden="true" id="statuses_info" title=" "
class="glyphicon glyphicon-info-sign" style="display:none"></span>
+ </div>
</td>
<!-- Column 8: Dag Runs -->
@@ -318,6 +314,24 @@ <h2>DAGs</h2>
}
});
});
+ d3.json("{{ url_for('Airflow.last_dagruns') }}", function(error, json) {
+ for(var safe_dag_id in json) {
+ dag_id = json[safe_dag_id].dag_id;
+ last_run = json[safe_dag_id].last_run;
+ g = d3.select('div#last-run-' + safe_dag_id)
+
+ g.selectAll('a')
+ .attr("href", "{{ url_for('Airflow.graph') }}?dag_id=" + dag_id +
"&execution_date=" + last_run)
+ .text(last_run);
+
+ g.selectAll('span')
+ .attr("data-original-title", "Start Date: " + last_run)
+ .style('display', null);
+
+ g.selectAll(".loading-last-run").remove();
+ }
+ d3.selectAll(".loading-last-run").remove();
+ });
d3.json("{{ url_for('Airflow.dag_stats') }}", function(error, json) {
for(var dag_id in json) {
states = json[dag_id];
diff --git a/airflow/www_rbac/views.py b/airflow/www_rbac/views.py
index e6e505c41a..941b90d360 100644
--- a/airflow/www_rbac/views.py
+++ b/airflow/www_rbac/views.py
@@ -392,6 +392,33 @@ def task_stats(self, session=None):
payload[dag.safe_dag_id].append(d)
return wwwutils.json_response(payload)
+ @expose('/last_dagruns')
+ @has_access
+ @provide_session
+ def last_dagruns(self, session=None):
+ DagRun = models.DagRun
+
+ filter_dag_ids = appbuilder.sm.get_accessible_dag_ids()
+
+ if not filter_dag_ids:
+ return
+
+ dags_to_latest_runs = dict(session.query(
+ DagRun.dag_id,
sqla.func.max(DagRun.execution_date).label('execution_date'))
+ .group_by(DagRun.dag_id).all())
+
+ payload = {}
+ for dag in dagbag.dags.values():
+ dag_accessible = 'all_dags' in filter_dag_ids or dag.dag_id in
filter_dag_ids
+ if (dag_accessible and dag.dag_id in dags_to_latest_runs and
+ dags_to_latest_runs[dag.dag_id]):
+ payload[dag.safe_dag_id] = {
+ 'dag_id': dag.dag_id,
+ 'last_run':
dags_to_latest_runs[dag.dag_id].strftime("%Y-%m-%d %H:%M")
+ }
+
+ return wwwutils.json_response(payload)
+
@expose('/code')
@has_dag_access(can_dag_read=True)
@has_access
diff --git a/tests/core.py b/tests/core.py
index c37b1f9c8b..c8b823adb6 100644
--- a/tests/core.py
+++ b/tests/core.py
@@ -1830,17 +1830,6 @@ def test_index(self):
self.assertIn("DAGs", resp_html)
self.assertIn("example_bash_operator", resp_html)
- # The HTML should contain data for the last-run. A link to the
specific run,
- # and the text of the date.
- url = "/admin/airflow/graph?" + urlencode({
- "dag_id": self.dag_python.dag_id,
- "execution_date": self.dagrun_python.execution_date,
- }).replace("&", "&")
- self.assertIn(url, resp_html)
- self.assertIn(
- self.dagrun_python.execution_date.strftime("%Y-%m-%d %H:%M"),
- resp_html)
-
def test_query(self):
response = self.app.get('/admin/queryview/')
self.assertIn("Ad Hoc Query", response.data.decode('utf-8'))
@@ -1920,6 +1909,9 @@ def test_dag_views(self):
response = self.app.get(
'/admin/airflow/task_stats')
self.assertIn("example_bash_operator", response.data.decode('utf-8'))
+ response = self.app.get(
+ '/admin/airflow/last_dagruns')
+ self.assertIn("example_python_operator", response.data.decode('utf-8'))
url = (
"/admin/airflow/success?task_id=print_the_context&"
"dag_id=example_python_operator&upstream=false&downstream=false&"
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
