Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[AIRFLOW-3160] Load latest_dagruns asynchronously, speed up front page load time #4005

Merged
merged 1 commit into from
Nov 5, 2018
+96 −29
Merged

[AIRFLOW-3160] Load latest_dagruns asynchronously, speed up front page load time #4005

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 23 additions & 9 deletions airflow/www/templates/airflow/dags.html
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 -->
Expand Down Expand Up @@ -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];
Expand Down
20 changes: 20 additions & 0 deletions airflow/www/views.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down
32 changes: 23 additions & 9 deletions airflow/www_rbac/templates/airflow/dags.html
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 -->
Expand Down Expand Up @@ -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];
Expand Down
27 changes: 27 additions & 0 deletions airflow/www_rbac/views.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
14 changes: 3 additions & 11 deletions tests/core.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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("&", "&amp;")
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'))
Expand Down Expand Up @@ -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&"
Expand Down