diff --git a/airflow/config_templates/default_airflow.cfg b/airflow/config_templates/default_airflow.cfg index ddd1ba816e84b..38e13bc1c4b58 100644 --- a/airflow/config_templates/default_airflow.cfg +++ b/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 diff --git a/airflow/logging_backend.py b/airflow/logging_backend.py new file mode 100644 index 0000000000000..efd381945a109 --- /dev/null +++ b/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 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..10ea345c9868f --- /dev/null +++ b/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() diff --git a/airflow/logging_backends/elasticsearch_logging_backend.py b/airflow/logging_backends/elasticsearch_logging_backend.py new file mode 100644 index 0000000000000..8e6456b5b7572 --- /dev/null +++ b/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 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_streaming_log.html b/airflow/www/templates/airflow/ti_streaming_log.html new file mode 100644 index 0000000000000..8b759a1883490 --- /dev/null +++ b/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() }} +

{{ title }}

+ +
+ {% for log in logs %} +
+ 
{{ log }}
+
+ {% endfor %} +
+{% endblock %} +{% block tail %} + {{ lib.form_js() }} + {{ super() }} + +{% endblock %} diff --git a/airflow/www/views.py b/airflow/www/views.py index 046c2e1e21cfe..b2bf72399ec61 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 @@ -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) diff --git a/setup.py b/setup.py index dedcf767944fb..f5c56ae17dcf8 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', @@ -267,6 +271,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/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))