[AIRFLOW-1325] Use Elasticsearch as logging backend

airflow/bin/cli.py

@@ -481,7 +481,8 @@ def run(args, dag=None):
             DeprecationWarning)
         remote_base = conf.get('core', 'S3_LOG_FOLDER')
 
-    if os.path.exists(filename):
+    # Only move log to remote storage if logging backend is not setup.
+    if os.path.exists(filename) and not conf.get('core', 'logging_backend_url'):
         # read log and remove old logs to get just the latest additions
 
         with open(filename, 'r') as logfile:

airflow/config_templates/default_airflow.cfg

@@ -116,6 +116,10 @@ security =
 # values at runtime)
 unit_test_mode = False
 
+# Url for connecting logging backend
+# 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,64 @@
+# -*- 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
+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():
+    global LOGGING_BACKEND
+
+    if LOGGING_BACKEND or not configuration.get('core', 'logging_backend_url'):
+        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. Url must be in the form of schema_alias://host
+    Example url:
+        elasticsearch://localhost:9200
+    Thrown:
+        ValueError if url or schema invalid
+        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 TypeError("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,61 @@
+# -*- 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, dag_id, task_id, execution_date, **kwargs):
+        """
+        Get logs given dag_id, task_id and execution_date.
+        Return a list of Log obejct.
+        """
+        raise NotImplementedError()
+
+class Log(object):
+    """
+    An abstract class for single line of Airflow log.
+                       whole line is message
+                                v
+    [2017-06-09 13:14:24,965] {cli.py:389} INFO - Running on host
+                ^                   ^        ^           ^
+            [timestamp]    {filename:fileline} log_level - details
+    """
+    def __init__(self, log):
+        self.log = log
+
+    @property
+    def message(self):
+        raise NotImplementedError()
+
+    @property
+    def timestamp(self):
+        raise NotImplementedError()
+
+    @property
+    def filename(self):
+        raise NotImplementedError()
+
+    @property
+    def fileline(self):
+        raise NotImplementedError()
+
+    @property
+    def log_level(self):
+        raise NotImplementedError()
+
+    @property
+    def details(self):
+        raise NotImplementedError()

airflow/logging_backends/elasticsearch_logging_backend.py

@@ -0,0 +1,130 @@
+# -*- 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, Log
+from elasticsearch import Elasticsearch
+from elasticsearch_dsl import Search
+
+
+class ElasticsearchLoggingBackend(BaseLoggingBackend):
+    """
+    This search backend requires ES document to have the following schema:
+    {
+        "log_timestamp": "2017-06-19 15:08:17,077", // primary soring key
+        "log_filename": "cli.py",
+        "log_line_number": "389",
+        "log_level": "INFO",
+        "log_details": "Running on host",
+        "offset": "76",  // secondary sorting key when log_timestamp is the same
+        "dag_id": "example_skip_dag"
+        "task_id": "one_success",
+        "execution_date": "2017-06-13T00:00:00",
+        "message": "[2017-06-19 15:08:17,077] {cli.py:389} INFO - Running on host",
+        "file_id": "example_skip_dag-one_success-2017-06-13T00:00:00",
+    }
+    """
+
+    SCHEMA_NAME = 'elasticsearch'
+
+    def __init__(self, host=None, **kwargs):
+        super(ElasticsearchLoggingBackend, self).__init__(
+            schema_name=self.SCHEMA_NAME, **kwargs)
+
+        self.client = Elasticsearch([host])
+
+        # Disable elasticsearch logger unless CRITICAL level
+        logger = logging.getLogger('elasticsearch')
+        logger.setLevel(logging.CRITICAL)
+
+    def _search(self, **kwargs):
+        """
+        Return a list of documents given search conditions.
+        :arg dag_id: id of the dag
+        :arg task_id: id of the task
+        :arg execution_date: execution date of the dag run
+        :arg gt_ts: log has log_timestamp greater than gt_ts
+        :arg gte_ts: log has log_timestamp greater than or equal to gte_ts
+        :arg lt_ts: log has log_timestamp less than lt_ts
+        :arg lte_ts: log has log_timestamp less than or equal to lte_ts
+        :arg page: logs at given page
+        :arg max_page_size: maximum log lines returned in one page
+        """
+
+        dag_id = kwargs.get('dag_id')
+        task_id = kwargs.get('task_id')
+        execution_date = kwargs.get('execution_date')
+        gt_ts = kwargs.get('gt_ts')
+        gte_ts = kwargs.get('gte_ts')
+        lt_ts = kwargs.get('lt_ts')
+        lte_ts = kwargs.get('lte_ts')
+        page = kwargs.get('page') or 0
+        max_page_size = kwargs.get('max_page_size') or 10000
+
+        file_id = '-'.join([dag_id, task_id, execution_date])
+
+        s = Search(using=self.client) \
+            .query('match', file_id=file_id) \
+            .sort('log_timestamp', 'offset')
+
+        if lt_ts:
+            s = s.filter('range', log_timestamp={'lt': lt_ts})
+
+        if lte_ts:
+            s = s.filter('range', log_timestamp={'lte': lte_ts})
+
+        if gt_ts:
+            s = s.filter('range', log_timestamp={'gt': gt_ts})
+
+        if gte_ts:
+            s = s.filter('range', log_timestamp={'gte': gte_ts})
+
+        response = s[max_page_size * page:max_page_size].execute()
+
+        return response
+
+    def get_logs(self, dag_id, task_id, execution_date, **kwargs):
+        response = self._search(dag_id=dag_id, task_id=task_id,
+                                execution_date=execution_date, **kwargs)
+        logs = [ESLog(hit) for hit in response]
+        return logs
+
+class ESLog(Log):
+    def __init__(self, log):
+        super(ESLog, self).__init__(log)
+
+    @property
+    def message(self):
+        return self.log.message
+
+    @property
+    def timestamp(self):
+        return self.log.log_timestamp
+
+    @property
+    def filename(self):
+        return self.log.log_filename
+
+    @property
+    def fileline(self):
+        return self.log.log_line_number
+
+    @property
+    def log_level(self):
+        return self.log.log_level
+
+    @property
+    def details(self):
+        return self.log.log_details

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_logs.html

@@ -0,0 +1,70 @@
+{#
+  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>
+  <pre id="log"></pre>
+{% 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 = 50;
+    // Animation speed for auto tailing log display.
+    const ANIMATION_SPEED = 1000;
+
+    function recurse(delay=DELAY) {
+      return new Promise((resolve) => setTimeout(resolve, delay));
+    }
+
+    function autoTailingLog(startTimestamp=null) {
+      return Promise.resolve(
+        $.ajax({
+          url: "{{ url_for("airflow.get_log") }}",
+          data: {
+            dag_id: "{{ dag_id }}",
+            task_id: "{{ task_id }}",
+            execution_date: "{{ execution_date }}",
+            start_ts: startTimestamp,
+          },
+        })).then(res => {
+          console.log(res);
+          if (res && res.message) {
+            const docHeight = $(document).height();
+            $("#log").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(res.next_start_ts));
+        });
+    }
+
+    $(document).ready(function() {
+      autoTailingLog();
+    });
+
+  </script>
+{% endblock %}