From e52dc3fe02ea1322b9b9af78ea7011bcf6984dba Mon Sep 17 00:00:00 2001 From: AllisonWang Date: Mon, 19 Jun 2017 14:46:20 -0700 Subject: [PATCH] [AIRFLOW-1325] Use Elasticsearch as logging backend --- airflow/bin/cli.py | 3 +- airflow/config_templates/default_airflow.cfg | 4 + airflow/logging_backend.py | 64 +++++++ airflow/logging_backends/__init__.py | 13 ++ .../logging_backends/base_logging_backend.py | 61 +++++++ .../elasticsearch_logging_backend.py | 130 +++++++++++++ airflow/utils/imports.py | 32 ++++ airflow/www/templates/airflow/ti_logs.html | 70 +++++++ airflow/www/views.py | 46 ++++- setup.py | 5 + tests/logging_backends/__init__.py | 13 ++ tests/logging_backends/test_elasticsearch.sh | 69 +++++++ .../test_elasticsearch_logging_backend.py | 171 ++++++++++++++++++ tests/test_logging_backend.py | 32 ++++ tests/utils/test_imports.py | 35 ++++ 15 files changed, 739 insertions(+), 9 deletions(-) create mode 100644 airflow/logging_backend.py create mode 100644 airflow/logging_backends/__init__.py create mode 100644 airflow/logging_backends/base_logging_backend.py create mode 100644 airflow/logging_backends/elasticsearch_logging_backend.py create mode 100644 airflow/utils/imports.py create mode 100644 airflow/www/templates/airflow/ti_logs.html create mode 100644 tests/logging_backends/__init__.py create mode 100755 tests/logging_backends/test_elasticsearch.sh create mode 100644 tests/logging_backends/test_elasticsearch_logging_backend.py create mode 100644 tests/test_logging_backend.py create mode 100644 tests/utils/test_imports.py diff --git a/airflow/bin/cli.py b/airflow/bin/cli.py index 41f979fa51edb..1bac1cd305732 100755 --- a/airflow/bin/cli.py +++ b/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: diff --git a/airflow/config_templates/default_airflow.cfg b/airflow/config_templates/default_airflow.cfg index c6c1da2812b57..cc3f22223c446 100644 --- a/airflow/config_templates/default_airflow.cfg +++ b/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 diff --git a/airflow/logging_backend.py b/airflow/logging_backend.py new file mode 100644 index 0000000000000..7f8a0cbbd6957 --- /dev/null +++ b/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 diff --git a/airflow/logging_backends/__init__.py b/airflow/logging_backends/__init__.py new file mode 100644 index 0000000000000..9d7677a99b293 --- /dev/null +++ b/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. diff --git a/airflow/logging_backends/base_logging_backend.py b/airflow/logging_backends/base_logging_backend.py new file mode 100644 index 0000000000000..dd333ac67c2c0 --- /dev/null +++ b/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() diff --git a/airflow/logging_backends/elasticsearch_logging_backend.py b/airflow/logging_backends/elasticsearch_logging_backend.py new file mode 100644 index 0000000000000..5e47f69fcc38c --- /dev/null +++ b/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 diff --git a/airflow/utils/imports.py b/airflow/utils/imports.py new file mode 100644 index 0000000000000..2256c19dde289 --- /dev/null +++ b/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() + """ + 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 diff --git a/airflow/www/templates/airflow/ti_logs.html b/airflow/www/templates/airflow/ti_logs.html new file mode 100644 index 0000000000000..ee227c5813786 --- /dev/null +++ b/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() }} +

{{ title }}

+ 

+{% endblock %}
+{% block tail %}
+  {{ lib.form_js() }}
+  {{ super() }}
+  
+{% endblock %}
diff --git a/airflow/www/views.py b/airflow/www/views.py
index 541c3ff8afbb4..39d32136aa8b8 100644
--- a/airflow/www/views.py
+++ b/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
@@ -710,24 +712,52 @@ def rendered(self):
             form=form,
             title=title,)
 
+    @expose('/get_log')
+    @login_required
+    @wwwutils.action_logging
+    def get_log(self):
+        """ The 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)
+        start_ts = request.args.get('start_ts')
+
+        logs = logging_backend.get_logs(dag_id, task_id, execution_date,
+                                        gt_ts=start_ts)
+        next_start_ts = start_ts if not logs else logs[-1].timestamp
+        message = '\n'.join([log.message for log in logs])
+
+        return jsonify(message=message, next_start_ts=next_start_ts)
+
     @expose('/log')
     @login_required
     @wwwutils.action_logging
     def log(self):
-        BASE_LOG_FOLDER = os.path.expanduser(
-            conf.get('core', 'BASE_LOG_FOLDER'))
         dag_id = request.args.get('dag_id')
         task_id = request.args.get('task_id')
         execution_date = request.args.get('execution_date')
         dag = dagbag.get_dag(dag_id)
-        log_relative = "{dag_id}/{task_id}/{execution_date}".format(
-            **locals())
-        loc = os.path.join(BASE_LOG_FOLDER, log_relative)
+        dttm = dateutil.parser.parse(execution_date)
+        form = DateTimeForm(data={'execution_date': dttm})
+
+        if conf.get('core', 'logging_backend_url'):
+            return self.render(
+                'airflow/ti_logs.html',
+                dag=dag, title="Log", dag_id=dag_id, task_id=task_id,
+                execution_date=execution_date, form=form)
+
+        base_log_folder = os.path.expanduser(conf.get('core', 'BASE_LOG_FOLDER'))
+        log_relative = "{dag_id}/{task_id}/{execution_date}".format(**locals())
+        loc = os.path.join(base_log_folder, log_relative)
         loc = loc.format(**locals())
         log = ""
         TI = models.TaskInstance
-        dttm = dateutil.parser.parse(execution_date)
-        form = DateTimeForm(data={'execution_date': dttm})
+
         session = Session()
         ti = session.query(TI).filter(
             TI.dag_id == dag_id, TI.task_id == task_id,
diff --git a/setup.py b/setup.py
index 80668dac7aff0..73d8ff8e012eb 100644
--- a/setup.py
+++ b/setup.py
@@ -136,6 +136,10 @@ def check_previous():
 ]
 docker = ['docker-py>=1.6.0']
 druid = ['pydruid>=0.2.1']
+elasticsearch = [
+    'elasticsearch>=5.0.0,<6.0.0',
+    'elasticsearch-dsl>=5.0.0,<6.0.0'
+]
 emr = ['boto3>=1.0.0']
 gcp_api = [
     'httplib2',
@@ -264,6 +268,7 @@ def do_setup():
             'doc': doc,
             'docker': docker,
             'druid': druid,
+            'elasticsearch': elasticsearch,
             'emr': emr,
             'gcp_api': gcp_api,
             'github_enterprise': github_enterprise,
diff --git a/tests/logging_backends/__init__.py b/tests/logging_backends/__init__.py
new file mode 100644
index 0000000000000..9d7677a99b293
--- /dev/null
+++ b/tests/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.
diff --git a/tests/logging_backends/test_elasticsearch.sh b/tests/logging_backends/test_elasticsearch.sh
new file mode 100755
index 0000000000000..0a84d977c4fda
--- /dev/null
+++ b/tests/logging_backends/test_elasticsearch.sh
@@ -0,0 +1,69 @@
+#!/usr/bin/env bash
+#
+# 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.
+
+# This is a script for testing Elasticsearch logging backend using Docker.
+# Check if user has docker installed
+docker -v 1>/dev/null
+if [ $? -ne 0 ]; then
+  echo "Please install Docker to run the unit test. Abort"
+  exit
+fi
+# Get elasticsearch running in docker
+echo "Starting Elasticsearch Docker image..."
+CONTAINER_ID=$(docker run -d \
+               -p 9200:9200 \
+               -e "cluster.name=airflow" \
+               -e "http.host=0.0.0.0" \
+               -e "transport.host=127.0.0.1" \
+               -e "xpack.security.enabled=false" \
+               -e "xpack.monitoring.enabled=false" \
+               -e "xpack.ml.enabled=false" \
+               -e "xpack.graph.enabled=false" \
+               -e "xpack.watcher.enabled=false" \
+               docker.elastic.co/elasticsearch/elasticsearch:5.4.1)
+if [ $? -ne 0 ]; then
+  echo "Abort."
+  exit
+fi
+echo "Container ${CONTAINER_ID} started."
+# Wait for Elasticsearch to start (~20s)
+echo "Waiting for Elasticsearch to initialize..."
+echo -ne '##                        (18%)\r'
+sleep 3
+echo -ne '#####                     (25%)\r'
+sleep 3
+echo -ne '########                  (42%)\r'
+sleep 3
+echo -ne '#############             (66%)\r'
+sleep 3
+echo -ne '################          (78%)\r'
+sleep 3
+echo -ne '###################       (89%)\r'
+sleep 3
+echo -ne '#####################     (95%)\r'
+sleep 3
+echo -ne '#######################   (100%)\r'
+echo -ne '\n'
+# Run integration test
+echo "Running Elasticsearch logging backend test..."
+export AIRFLOW_RUNALL_TESTS=1
+python test_elasticsearch_logging_backend.py
+export AIRFLOW_RUNALL_TESTS=0
+# Stop container
+docker stop $CONTAINER_ID 1>/dev/null
+if [ $? -ne 0 ]; then
+  echo "Failed to stop container ${CONTAINER_ID}. Please clean up manually."
+else
+  echo "Container ${CONTAINER_ID} stopped."
+fi
diff --git a/tests/logging_backends/test_elasticsearch_logging_backend.py b/tests/logging_backends/test_elasticsearch_logging_backend.py
new file mode 100644
index 0000000000000..14c64f19f34c9
--- /dev/null
+++ b/tests/logging_backends/test_elasticsearch_logging_backend.py
@@ -0,0 +1,171 @@
+# -*- 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 json
+import os
+import time
+import unittest
+
+if 'AIRFLOW_RUNALL_TESTS' in os.environ:
+
+    from airflow.logging_backends.elasticsearch_logging_backend import ElasticsearchLoggingBackend
+    from elasticsearch import Elasticsearch
+
+
+    class TestElasticsearchLoggingBackend(unittest.TestCase):
+
+        INDEX_NAME = "airflow"
+        TEMPLATE_NAME = "airflow"
+        TEMPLATE_FILENAME = os.path.join(os.path.dirname(__file__), 'test_elasticsearch_template.json')
+        DOC_TYPE = "log"
+
+        def setUp(self):
+            self.host = "localhost:9200"
+            self.backend = ElasticsearchLoggingBackend(host=self.host)
+            self.es = self.backend.client
+            self.doc1 = {
+                "log_timestamp": "2017-06-19 15:08:17,077",
+                "log_filename": "cli.py",
+                "log_details": "Running on host",
+                "offset": "76",
+                "execution_date": "2017-06-13T00:00:00",
+                "input_type": "log",
+                "log_level": "INFO",
+                "task_id": "one_success",
+                "source": "airflow/logs/example_skip_dag/one_success/2017-06-13T00:00:00",
+                "message": "[2017-06-19 15:08:17,077] {cli.py:389} INFO - Running on host",
+                "type": "log",
+                "file_id": "example_skip_dag-one_success-2017-06-13T00:00:00",
+                "log_line_number": "389",
+                "dag_id": "example_skip_dag"
+            }
+
+            self.doc2 = {
+                "log_timestamp": "2017-06-19 15:08:17,077",
+                "log_filename": "models.py",
+                "log_details": "Filling up the DagBag from airflow/airflow/example_dags/example_skip_dag.py",
+                "offset": "221",
+                "execution_date": "2017-06-13T00:00:00",
+                "input_type": "log",
+                "log_level": "INFO",
+                "task_id": "one_success",
+                "source": "airflow/logs/example_skip_dag/one_success/2017-06-13T00:00:00",
+                "message": "[2017-06-19 15:08:17,077] {models.py:176} INFO - Filling up the DagBag from airflow/airflow/example_dags/example_skip_dag.py",
+                "type": "log",
+                "file_id": "example_skip_dag-one_success-2017-06-13T00:00:00",
+                "log_line_number": "176",
+                "dag_id": "example_skip_dag"
+            }
+
+            self.doc3 = {
+                "log_timestamp": "2017-06-19 15:08:18,077",
+                "log_filename": "base_task_runner.py",
+                "log_details": "Running: ['bash', '-c', u'airflow run example_skip_dag one_success 2017-06-13T00:00:00 --job_id 101 --raw -sd DAGS_FOLDER/example_dags/example_skip_dag.py']",
+                "offset": "437",
+                "execution_date": "2017-06-13T00:00:00",
+                "input_type": "log",
+                "log_level": "INFO",
+                "task_id": "one_success",
+                "source": "airflow/logs/example_skip_dag/one_success/2017-06-13T00:00:00",
+                "message": "[2017-06-19 15:08:18,077] {base_task_runner.py:112} INFO - Running: ['bash', '-c', u'airflow run example_skip_dag one_success 2017-06-13T00:00:00 --job_id 101 --raw -sd DAGS_FOLDER/example_dags/example_skip_dag.py']",
+                "type": "log",
+                "file_id": "example_skip_dag-one_success-2017-06-13T00:00:00",
+                "log_line_number": "112",
+                "dag_id": "example_skip_dag"
+            }
+
+            # Load Elasticsearch 5 test template
+            self.template = {
+                "template" : "*",
+                "mappings": {
+                    "_default_": {
+                        "_all": { "enabled": "false" },
+                        "dynamic_date_formats": [
+                            "strict_date_optional_time",
+                            "yyyy/MM/dd HH:mm:ss Z||yyyy/MM/dd Z",
+                            "yyyy-MM-dd HH:mm:ss,SSS"
+                        ],
+                        "numeric_detection": "true",
+                        "dynamic_templates": [
+                            {
+                                "timestamp_template" : {
+                                    "path_match": "*",
+                                    "match_mapping_type": "date",
+                                    "mapping": {
+                                        "type": "date",
+                                        "index": "not_analyzed",
+                                        "doc_values": "true",
+                                        "ignore_malformed": "true"
+                                    }
+                                }
+                            },
+                            {
+                                "integer_template": {
+                                    "path_match": "*",
+                                    "match_mapping_type": "long",
+                                    "mapping": {
+                                        "type": "integer",
+                                        "index": "not_analyzed",
+                                        "doc_values": "true"
+                                    }
+                                }
+                            },
+                            {
+                                "default_template" : {
+                                    "path_match": "*",
+                                    "match_mapping_type": "string",
+                                    "mapping": {
+                                        "type": "keyword",
+                                        "index": "not_analyzed",
+                                        "doc_values": "true"
+                                    }
+                                }
+                            }
+                         ],
+                         "properties" : {
+                            "message" : { "type" : "text" }
+                        }
+                    }
+                }
+            }
+
+            # Make sure Elasticsearch has no old index, template and pipeline.
+            self.es.indices.delete(index="*", ignore=[404, 400])
+            self.es.indices.delete_template(name="*")
+            self.es.ingest.delete_pipeline(id="*")
+            # Add test template
+            self.es.indices.put_template(name=self.TEMPLATE_NAME, body=self.template)
+            # Index documents
+            self.es.index(index=self.INDEX_NAME, doc_type=self.DOC_TYPE, id=1, body=self.doc1)
+            self.es.index(index=self.INDEX_NAME, doc_type=self.DOC_TYPE, id=2, body=self.doc2)
+            self.es.index(index=self.INDEX_NAME, doc_type=self.DOC_TYPE, id=3, body=self.doc3)
+            # Give Elasticsearch sometime to index document
+            time.sleep(3)
+
+        def test_get_logs(self):
+            dag_id = "example_skip_dag"
+            task_id = "one_success"
+            execution_date = "2017-06-13T00:00:00"
+
+            logs = self.backend.get_logs(dag_id, task_id, execution_date)
+
+            self.assertEqual(len(logs), 3)
+            # Check logs are sorted
+            self.assertEqual(logs[0].message, self.doc1['message'])
+            self.assertEqual(logs[1].message, self.doc2['message'])
+            self.assertEqual(logs[2].message, self.doc3['message'])
+
+    if __name__ == "__main__":
+        unittest.main()
diff --git a/tests/test_logging_backend.py b/tests/test_logging_backend.py
new file mode 100644
index 0000000000000..06df444047c3e
--- /dev/null
+++ b/tests/test_logging_backend.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 unittest
+
+from airflow import configuration
+from airflow.logging_backend import cached_logging_backend, _get_logging_backend
+from airflow.logging_backends.base_logging_backend import BaseLoggingBackend
+
+class LoggingBackendTest(unittest.TestCase):
+
+    def test_invalid_logging_backend(self):
+        invalid_schema = "invalid_schema://localhost:1234"
+        invalid_url_format = "invalid_url_format"
+        self.assertRaises(ValueError, _get_logging_backend, invalid_schema)
+        self.assertRaises(ValueError, _get_logging_backend, invalid_url_format)
+
+    def test_default_test_configuration(self):
+        configuration.load_test_config()
+        logging_backend = cached_logging_backend()
+        self.assertIsNone(logging_backend)
diff --git a/tests/utils/test_imports.py b/tests/utils/test_imports.py
new file mode 100644
index 0000000000000..c68b3eb84d66b
--- /dev/null
+++ b/tests/utils/test_imports.py
@@ -0,0 +1,35 @@
+# -*- 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 types
+import unittest
+
+from airflow.utils.imports import import_class_by_name
+
+class TestImports(unittest.TestCase):
+
+    def test_import_class_by_name(self):
+        cls_name = 'airflow.operators.bash_operator.BashOperator'
+        invalid_cls_name_1 = 'airflow.operators.bash_operator.BashOpera'
+        invalid_cls_name_2 = 'airflow.operators.bash_operator'
+
+        cls_reference = import_class_by_name(cls_name)
+        self.assertEqual(cls_reference.__name__, "BashOperator")
+        self.assertTrue(callable(cls_reference))
+
+        self.assertRaises(AttributeError, import_class_by_name, invalid_cls_name_1)
+
+        invalid_cls_reference = import_class_by_name(invalid_cls_name_2)
+        self.assertEqual(type(invalid_cls_reference), types.ModuleType)
+        self.assertFalse(callable(invalid_cls_reference))