[AIRFLOW-1325][WIP]Airflow streaming log backed by ElasticSearch

airflow/config_templates/default_airflow.cfg

@@ -122,6 +122,10 @@ security =
 # values at runtime)
 unit_test_mode = False
 
+# Logging backend URL. This is used for Airflow webserver to fetch logs.
+# Example: elasticsearch://localhost:9200
+logging_backend_url =
+
 [cli]
 # In what way should the cli access the API. The LocalClient will use the
 # database directly, while the json_client will use the api running on the

airflow/logging_backend.py

@@ -0,0 +1,66 @@
+# -*- coding: utf-8 -*-
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from airflow import configuration, AirflowException
+from airflow.logging_backends.base_logging_backend import BaseLoggingBackend
+from airflow.utils.imports import import_class_by_name
+
+LOGGING_BACKEND = None
+
+# Backend alias to full class name mapping
+LOGGING_BACKEND_ALIAS = {
+    'elasticsearch': ('airflow.logging_backends.elasticsearch_logging_backend'
+                      '.ElasticsearchLoggingBackend'),
+}
+
+
+def cached_logging_backend():
+
+    if not configuration.get('core', 'logging_backend_url'):
+        return None
+
+    global LOGGING_BACKEND
+
+    if LOGGING_BACKEND:
+        return LOGGING_BACKEND
+
+    url = configuration.get('core', 'logging_backend_url')
+    LOGGING_BACKEND = _get_logging_backend(url)
+
+    return LOGGING_BACKEND
+
+
+def _get_logging_backend(url):
+    """
+    Get logging backend instance.
+    :param url: Url of the logging backend in form schema_alias://host
+    :raises ValueError: if url or schema invalid
+    :raises TypeError: if backend is not an instance of BaseLoggingBackend
+    """
+    parsed = url.split('://', 1)
+    if len(parsed) != 2:
+        raise ValueError("Logging backend url {} invalid. Please use correct "
+                         "format: schema_alias://host.".format(url))
+
+    schema, host = parsed
+    cls_name = LOGGING_BACKEND_ALIAS.get(schema)
+    if not cls_name:
+        raise ValueError("Logging backend alias {} not found".format(schema))
+
+    backend_cls = import_class_by_name(cls_name)
+    backend = backend_cls(host)
+    if not isinstance(backend, BaseLoggingBackend):
+        raise AirflowException("Backend is not an instance of BaseLoggingBackend")
+
+    return backend

airflow/logging_backends/__init__.py

@@ -0,0 +1,13 @@
+# -*- coding: utf-8 -*-
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.

airflow/logging_backends/base_logging_backend.py

@@ -0,0 +1,24 @@
+# -*- coding: utf-8 -*-
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+class BaseLoggingBackend(object):
+    def __init__(self, **kwargs):
+        self.schema_name = kwargs.get('schema_name')
+
+    def get_logs(self,**kwargs):
+        """
+        Return a list of airflow task instance logs.
+        """
+        raise NotImplementedError()

airflow/logging_backends/elasticsearch_logging_backend.py

@@ -0,0 +1,74 @@
+# -*- coding: utf-8 -*-
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import logging
+
+from airflow.logging_backends.base_logging_backend import BaseLoggingBackend
+from elasticsearch import Elasticsearch
+from elasticsearch_dsl import Search
+
+
+class ElasticsearchLoggingBackend(BaseLoggingBackend):
+
+    SCHEMA_NAME = 'elasticsearch'
+    # Maximum number of logs to return per search call.
+    MAX_PER_PAGE = 1000
+
+    def __init__(self, host=None, **kwargs):
+        super(ElasticsearchLoggingBackend, self).__init__(
+            schema_name=self.SCHEMA_NAME, **kwargs)
+
+        self.client = Elasticsearch([host])
+
+        # Prevent adding ElasticSearch logs into task instance logs.
+        # TODO: we should find a place to place ElasticSearch logs.
+        logger = logging.getLogger('elasticsearch')
+        logger.setLevel(logging.CRITICAL)
+
+    def _search(self, **kwargs):
+        """
+        Return a list of documents given search conditions.
+        :param dag_id: id of the dag
+        :param task_id: id of the task
+        :param execution_date: execution date of the task instance
+        :param try_number: try_number of the task instance
+        :param offset: filter log with offset strictly greater than offset
+        :param page: logs at given page. Default value is 0.
+        """
+
+        dag_id = kwargs.get('dag_id')
+        task_id = kwargs.get('task_id')
+        execution_date = kwargs.get('execution_date')
+        try_number = kwargs.get('try_number')
+        offset = kwargs.get('offset')
+        page = kwargs.get('page') or 0
+
+        log_id = '-'.join([dag_id, task_id, execution_date, try_number])
+
+        s = Search(using=self.client) \
+            .query('match', log_id=log_id) \
+            .sort('offset')
+
+        # Offset is the unique key for sorting logs given log_id.
+        if offset:
+            s = s.filter('range', offset={'gt': offset})
+
+        response = s[self.MAX_PER_PAGE * page:self.MAX_PER_PAGE].execute()
+
+        return response
+
+    def get_logs(self, **kwargs):
+        response = self._search(**kwargs)
+        logs = [hit for hit in response]
+        return logs

airflow/utils/imports.py

@@ -0,0 +1,32 @@
+# -*- coding: utf-8 -*-
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import importlib
+
+
+def import_class_by_name(name):
+    """
+    Return a reference to a class given its full class name.
+    Example:
+        name = module.submodule.MyClass
+        will return a reference to MyClass, which can be instantiate
+        using my_class = MyClass(<params>)
+    """
+    parsed = name.rsplit('.', 1)
+    if len(parsed) != 2:
+        raise ValueError("Invalid class name {} to import.".format(name))
+    module_name, cls_name = parsed
+    module = importlib.import_module(module_name)
+    cls = getattr(module, cls_name)
+    return cls

airflow/www/templates/airflow/ti_streaming_log.html

@@ -0,0 +1,110 @@
+{#
+  Licensed to the Apache Software Foundation (ASF) under one or more
+  contributor license agreements.  See the NOTICE file distributed with
+  this work for additional information regarding copyright ownership.
+  The ASF licenses this file to You under the Apache License, Version 2.0
+  (the "License"); you may not use this file except in compliance with
+  the License.  You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+
+#}
+{% extends "airflow/task_instance.html" %}
+{% block title %}Airflow - DAGs{% endblock %}
+
+{% block body %}
+  {{ super() }}
+  <h4>{{ title }}</h4>
+  <ul class="nav nav-pills" role="tablist">
+    {% for log in logs %}
+      <li role="presentation" class="{{ 'active' if loop.last else '' }}">
+        <a href="#{{ loop.index }}" aria-controls="{{ loop.index }}" role="tab" data-toggle="tab">
+          {{ loop.index }}
+        </a>
+      </li>
+    {% endfor %}
+  </ul>
+  <div class="tab-content">
+    {% for log in logs %}
+      <div role="tabpanel" class="tab-pane {{ 'active' if loop.last else '' }}" id="{{ loop.index }}">
+        <pre><code id="try-{{ loop.index }}">{{ log }}</code></pre>
+      </div>
+    {% endfor %}
+  </div>
+{% endblock %}
+{% block tail %}
+  {{ lib.form_js() }}
+  {{ super() }}
+  <script>
+    // Time interval to wait before next log fetching. Default 2s.
+    const DELAY = 2e3;
+    // Distance away from page bottom to enable auto tailing.
+    const AUTO_TAILING_OFFSET = 30;
+    // Animation speed for auto tailing log display.
+    const ANIMATION_SPEED = 1000;
+    // Total number of tabs to show.
+    const TOTAL_ATTEMPTS = "{{ logs|length }}";
+
+    // Recursively fetch logs from flask endpoint.
+    function recurse(delay=DELAY) {
+      return new Promise((resolve) => setTimeout(resolve, delay));
+    }
+
+    // Fetch all logs for given try_number
+    function staticLog(try_number) {
+      $.ajax({
+        url: "{{ url_for("airflow.get_log_js") }}",
+        data: {
+          dag_id: "{{ dag_id }}",
+          task_id: "{{ task_id }}",
+          execution_date: "{{ execution_date }}",
+          try_number: try_number,
+          offset: null,
+        },
+      }).then(res => {
+        if (res && res.message) {
+          $(`#try-${try_number}`).append(res.message + '\n');
+        }
+      });
+    }
+
+    // Streaming log with auto-tailing.
+    function autoTailingLog(try_number, offset=null) {
+      return Promise.resolve(
+        $.ajax({
+          url: "{{ url_for("airflow.get_log_js") }}",
+          data: {
+            dag_id: "{{ dag_id }}",
+            task_id: "{{ task_id }}",
+            execution_date: "{{ execution_date }}",
+            try_number: try_number,
+            offset: offset,
+          },
+        })).then(res => {
+          if (res && res.message) {
+            const docHeight = $(document).height();
+            $(`#try-${try_number}`).append(res.message + '\n');
+            // Auto scroll window to the end if current window location is near the end
+            if($(window).scrollTop() + $(window).height() > docHeight - AUTO_TAILING_OFFSET) {
+              $("html, body").animate({ scrollTop: $(document).height() }, ANIMATION_SPEED);
+            }
+          }
+          return recurse().then(() => autoTailingLog(try_number, res.next_offset));
+        });
+    }
+    $(document).ready(function() {
+      // Lazily load past task instance logs.
+      for(let i = 1; i < TOTAL_ATTEMPTS; i++) {
+        staticLog(i);
+      }
+      // Get latest attempt logs with auto-tailing feature.
+      autoTailingLog(Math.max(TOTAL_ATTEMPTS, 1));
+    });
+  </script>
+{% endblock %}

airflow/www/views.py

@@ -35,7 +35,8 @@
 from sqlalchemy import or_, desc, and_, union_all
 
 from flask import (
-    redirect, url_for, request, Markup, Response, current_app, render_template, make_response)
+    redirect, url_for, request, Markup, Response, current_app,
+    render_template, make_response, jsonify)
 from flask_admin import BaseView, expose, AdminIndexView
 from flask_admin.contrib.sqla import ModelView
 from flask_admin.actions import action
@@ -59,6 +60,7 @@
 from airflow import models
 from airflow import settings
 from airflow.api.common.experimental.mark_tasks import set_dag_run_state
+from airflow.logging_backend import cached_logging_backend
 from airflow.exceptions import AirflowException
 from airflow.settings import Session
 from airflow.models import XCom, DagRun
@@ -771,6 +773,29 @@ def _get_log(self, ti, log_filename):
 
         return log
 
+    @expose('/get_log_js')
+    @login_required
+    @wwwutils.action_logging
+    def get_log_js(self):
+        """ JavaScript endpoint for fetching logs. """
+        logging_backend = cached_logging_backend()
+        if not logging_backend:
+            raise AirflowException("Logging backend not found.")
+
+        dag_id = request.args.get('dag_id')
+        task_id = request.args.get('task_id')
+        execution_date = request.args.get('execution_date')
+        dttm = dateutil.parser.parse(execution_date)
+        try_number = request.args.get('try_number')
+        offset = request.args.get('offset')
+
+        logs = logging_backend.get_logs(dag_id=dag_id, task_id=task_id,
+            execution_date=execution_date, try_number=try_number, offset=offset)
+        next_offset = offset if not logs else logs[-1].offset
+        message = '\n'.join([log.message for log in logs])
+
+        return jsonify(message=message, next_offset=next_offset)
+
     @expose('/log')
     @login_required
     @wwwutils.action_logging
@@ -792,6 +817,13 @@ def log(self):
             logs = ["*** Task instance did not exist in the DB\n"]
         else:
             logs = [''] * ti.try_number
+            if conf.get('core', 'logging_backend_url'):
+                return self.render(
+                    'airflow/ti_streaming_log.html',
+                    logs=logs, dag=dag, title="Log by attempts",
+                    dag_id=dag.dag_id, task_id=task_id,
+                    execution_date=execution_date, form=form)
+
             for try_number in range(ti.try_number):
                 log_filename = get_log_filename(
                     dag_id, task_id, execution_date, try_number)