Fokko closed pull request #3930: [AIRFLOW-2548] Output plugin import errors to
web UI
URL: https://github.com/apache/incubator-airflow/pull/3930
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/airflow/plugins_manager.py b/airflow/plugins_manager.py
index c589e9b5a9..0eee5f8f93 100644
--- a/airflow/plugins_manager.py
+++ b/airflow/plugins_manager.py
@@ -34,6 +34,8 @@
log = LoggingMixin().log
+import_errors = {}
+
class AirflowPluginException(Exception):
pass
@@ -99,6 +101,7 @@ def validate(cls):
except Exception as e:
log.exception(e)
log.error('Failed to import plugin %s', filepath)
+ import_errors[filepath] = str(e)
def make_module(name, objects):
diff --git a/airflow/www/views.py b/airflow/www/views.py
index 8792191e21..0b7fde92f6 100644
--- a/airflow/www/views.py
+++ b/airflow/www/views.py
@@ -2126,6 +2126,14 @@ def get_int_arg(value, default=0):
"Broken DAG: [{ie.filename}] {ie.stacktrace}".format(ie=ie),
"error")
+ from airflow.plugins_manager import import_errors as
plugin_import_errors
+ for filename, stacktrace in plugin_import_errors.items():
+ flash(
+ "Broken plugin: [{filename}] {stacktrace}".format(
+ stacktrace=stacktrace,
+ filename=filename),
+ "error")
+
# get a list of all non-subdag dags visible to everyone
# optionally filter out "paused" dags
if hide_paused:
diff --git a/airflow/www_rbac/views.py b/airflow/www_rbac/views.py
index 49a9a734cc..1f69b7ac44 100644
--- a/airflow/www_rbac/views.py
+++ b/airflow/www_rbac/views.py
@@ -184,15 +184,20 @@ def get_int_arg(value, default=0):
if hide_paused:
sql_query = sql_query.filter(~DM.is_paused)
- # Get all the dag id the user could access
- filter_dag_ids = appbuilder.sm.get_accessible_dag_ids()
-
import_errors = session.query(models.ImportError).all()
for ie in import_errors:
flash(
"Broken DAG: [{ie.filename}] {ie.stacktrace}".format(ie=ie),
"error")
+ from airflow.plugins_manager import import_errors as
plugin_import_errors
+ for filename, stacktrace in plugin_import_errors.items():
+ flash(
+ "Broken plugin: [{filename}] {stacktrace}".format(
+ stacktrace=stacktrace,
+ filename=filename),
+ "error")
+
# get a list of all non-subdag dags visible to everyone
# optionally filter out "paused" dags
if hide_paused:
@@ -203,6 +208,9 @@ def get_int_arg(value, default=0):
unfiltered_webserver_dags = [dag for dag in dagbag.dags.values() if
not dag.parent_dag]
+ # Get all the dag id the user could access
+ filter_dag_ids = appbuilder.sm.get_accessible_dag_ids()
+
if 'all_dags' in filter_dag_ids:
orm_dags = {dag.dag_id: dag for dag
in sql_query
diff --git a/docs/plugins.rst b/docs/plugins.rst
index 4bd55cead9..59e505f0b1 100644
--- a/docs/plugins.rst
+++ b/docs/plugins.rst
@@ -119,6 +119,8 @@ For example,
* For ``Operator`` plugin, an ``execute`` method is compulsory.
* For ``Sensor`` plugin, a ``poke`` method returning a Boolean value is
compulsory.
+Make sure you restart the webserver and scheduler after making changes to
plugins so that they take effect.
+
Example
-------
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
