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(<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
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() }}
+  <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 %}
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))
