From 33a72d75d348bc0b65d26593049eeab6909c6da3 Mon Sep 17 00:00:00 2001 From: Fabien Vinas Date: Wed, 3 May 2017 23:39:14 +0200 Subject: [PATCH 1/6] Added progress percentage support aside status message --- luigi/scheduler.py | 23 +++++++++++- luigi/static/visualiser/index.html | 5 +++ luigi/static/visualiser/js/luigi.js | 6 +++ luigi/static/visualiser/js/visualiserApp.js | 9 ++++- luigi/task.py | 3 +- luigi/worker.py | 5 +++ test/scheduler_api_test.py | 6 +++ test/task_progress_percentage_test.py | 41 +++++++++++++++++++++ 8 files changed, 94 insertions(+), 4 deletions(-) create mode 100644 test/task_progress_percentage_test.py diff --git a/luigi/scheduler.py b/luigi/scheduler.py index d2deda3131..0060bea9f3 100644 --- a/luigi/scheduler.py +++ b/luigi/scheduler.py @@ -275,7 +275,7 @@ def __eq__(self, other): class Task(object): def __init__(self, task_id, status, deps, resources=None, priority=0, family='', module=None, - params=None, tracking_url=None, status_message=None, retry_policy='notoptional'): + params=None, tracking_url=None, status_message=None, progress_percentage=None, retry_policy='notoptional'): self.id = task_id self.stakeholders = set() # workers ids that are somehow related to this task (i.e. don't prune while any of these workers are still active) self.workers = OrderedSet() # workers ids that can perform task - task is 'BROKEN' if none of these workers are active @@ -301,6 +301,7 @@ def __init__(self, task_id, status, deps, resources=None, priority=0, family='', self.failures = Failures(self.retry_policy.disable_window) self.tracking_url = tracking_url self.status_message = status_message + self.progress_percentage = progress_percentage self.scheduler_disable_time = None self.runnable = False self.batchable = False @@ -1195,7 +1196,8 @@ def _serialize_task(self, task_id, include_deps=True, deps=None): 'priority': task.priority, 'resources': task.resources, 'tracking_url': getattr(task, "tracking_url", None), - 'status_message': getattr(task, "status_message", None) + 'status_message': getattr(task, "status_message", None), + 'progress_percentage': getattr(task, "progress_percentage", None) } if task.status == DISABLED: ret['re_enable_able'] = task.scheduler_disable_time is not None @@ -1454,6 +1456,23 @@ def get_task_status_message(self, task_id): else: return {"taskId": task_id, "statusMessage": ""} + @rpc_method() + def set_task_progress_percentage(self, task_id, progress_percentage): + if self._state.has_task(task_id): + task = self._state.get_task(task_id) + task.progress_percentage = progress_percentage + if task.status == RUNNING and task.batch_id is not None: + for batch_task in self._state.get_batch_running_tasks(task.batch_id): + batch_task.progress_percentage = progress_percentage + + @rpc_method() + def get_task_progress_percentage(self, task_id): + if self._state.has_task(task_id): + task = self._state.get_task(task_id) + return {"taskId": task_id, "progressPercentage": task.progress_percentage} + else: + return {"taskId": task_id, "progressPercentage": 0} + def _update_task_history(self, task, status, host=None): try: if status == DONE or status == FAILED: diff --git a/luigi/static/visualiser/index.html b/luigi/static/visualiser/index.html index d90efdc808..bd83444a56 100644 --- a/luigi/static/visualiser/index.html +++ b/luigi/static/visualiser/index.html @@ -272,6 +272,11 @@ 
{{statusMessage}}
+
+
+ {{progressPercentage}}% +
+
diff --git a/luigi/static/visualiser/js/luigi.js b/luigi/static/visualiser/js/luigi.js index 1399068b12..d9607dcfda 100644 --- a/luigi/static/visualiser/js/luigi.js +++ b/luigi/static/visualiser/js/luigi.js @@ -99,6 +99,12 @@ var LuigiAPI = (function() { }); }; + LuigiAPI.prototype.getTaskProgressPercentage = function(taskId, callback) { + return jsonRPC(this.urlRoot + "/get_task_progress_percentage", {task_id: taskId}, function(response) { + callback(response.response); + }); + }; + LuigiAPI.prototype.getRunningTaskList = function(callback) { return jsonRPC(this.urlRoot + "/task_list", {status: "RUNNING", upstream_status: "", search: searchTerm()}, function(response) { callback(flatten(response.response)); diff --git a/luigi/static/visualiser/js/visualiserApp.js b/luigi/static/visualiser/js/visualiserApp.js index 977457366a..055a70720d 100644 --- a/luigi/static/visualiser/js/visualiserApp.js +++ b/luigi/static/visualiser/js/visualiserApp.js @@ -71,7 +71,8 @@ function visualiserApp(luigi) { graph: (task.status == "PENDING" || task.status == "RUNNING" || task.status == "DONE"), error: task.status == "FAILED", re_enable: task.status == "DISABLED" && task.re_enable_able, - statusMessage: task.status_message + statusMessage: task.status_message, + progressPercentage: task.progress_percentage }; } @@ -288,6 +289,12 @@ function visualiserApp(luigi) { luigi.getTaskStatusMessage(data.taskId, function(data) { $("#statusMessageModal pre").html(data.statusMessage); }); + luigi.getTaskProgressPercentage(data.taskId, function(data) { + $("#statusMessageModal .progress-bar") + .attr('aria-valuenow', data.progressPercentage) + .text(data.progressPercentage + '%') + .css({'width': data.progressPercentage + '%'}); + }); }).trigger('click'); $("#statusMessageModal").modal({}); } diff --git a/luigi/task.py b/luigi/task.py index 8709bc8b65..b2b75f8886 100644 --- a/luigi/task.py +++ b/luigi/task.py @@ -437,6 +437,7 @@ def __init__(self, *args, **kwargs): self.set_tracking_url = None self.set_status_message = None + self.set_progress_percentage = None def initialized(self): """ @@ -675,7 +676,7 @@ def _dump(self): pickle.dumps(self) """ - unpicklable_properties = ('set_tracking_url', 'set_status_message') + unpicklable_properties = ('set_tracking_url', 'set_status_message', 'set_progress_percentage') reserved_properties = {} for property_name in unpicklable_properties: if hasattr(self, property_name): diff --git a/luigi/worker.py b/luigi/worker.py index 4677914e1e..c01ca153bf 100644 --- a/luigi/worker.py +++ b/luigi/worker.py @@ -125,11 +125,13 @@ def __init__(self, task, worker_id, result_queue, status_reporter, def _run_get_new_deps(self): self.task.set_tracking_url = self.status_reporter.update_tracking_url self.task.set_status_message = self.status_reporter.update_status + self.task.set_progress_percentage = self.status_reporter.update_progress_percentage task_gen = self.task.run() self.task.set_tracking_url = None self.task.set_status_message = None + self.task.set_progress_percentage = None if not isinstance(task_gen, types.GeneratorType): return None @@ -268,6 +270,9 @@ def update_tracking_url(self, tracking_url): def update_status(self, message): self._scheduler.set_task_status_message(self._task_id, message) + def update_progress_percentage(self, percentage): + self._scheduler.set_task_progress_percentage(self._task_id, percentage) + class SingleProcessPool(object): """ diff --git a/test/scheduler_api_test.py b/test/scheduler_api_test.py index 0d88d046b1..adbc328711 100644 --- a/test/scheduler_api_test.py +++ b/test/scheduler_api_test.py @@ -482,6 +482,12 @@ def test_batch_update_status(self): for task_id in ('A_1', 'A_2', 'A_1_2'): self.assertEqual('test message', self.sch.get_task_status_message(task_id)['statusMessage']) + def test_batch_update_progress(self): + self._start_simple_batch() + self.sch.set_task_progress_percentage('A_1_2', 30) + for task_id in ('A_1', 'A_2', 'A_1_2'): + self.assertEqual(30, self.sch.get_task_progress_percentage(task_id)['progressPercentage']) + def test_batch_tracking_url(self): self._start_simple_batch() self.sch.add_task(worker=WORKER, task_id='A_1_2', tracking_url='http://test.tracking.url/') diff --git a/test/task_progress_percentage_test.py b/test/task_progress_percentage_test.py new file mode 100644 index 0000000000..8d30eef9dd --- /dev/null +++ b/test/task_progress_percentage_test.py @@ -0,0 +1,41 @@ +# -*- coding: utf-8 -*- +# +# Copyright 2012-2015 Spotify AB +# +# 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 helpers import LuigiTestCase + +import luigi +import luigi.scheduler +import luigi.worker + +luigi.notifications.DEBUG = True + + +class TaskProgressPercentageTest(LuigiTestCase): + + def test_run(self): + percentage = 30 + sch = luigi.scheduler.Scheduler() + with luigi.worker.Worker(scheduler=sch) as w: + class MyTask(luigi.Task): + def run(self): + self.set_progress_percentage(percentage) + + task = MyTask() + w.add(task) + w.run() + + self.assertEqual(sch.get_task_progress_percentage(task.task_id)["progressPercentage"], percentage) From 66c8223bfc369fa1c25645cf9fcb73d03ac6a51e Mon Sep 17 00:00:00 2001 From: Fabien Vinas Date: Thu, 4 May 2017 00:14:59 +0200 Subject: [PATCH 2/6] Added auto-refresh of messages & progress when status message modal is open --- luigi/static/visualiser/js/visualiserApp.js | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/luigi/static/visualiser/js/visualiserApp.js b/luigi/static/visualiser/js/visualiserApp.js index 055a70720d..85e5d14816 100644 --- a/luigi/static/visualiser/js/visualiserApp.js +++ b/luigi/static/visualiser/js/visualiserApp.js @@ -297,6 +297,14 @@ function visualiserApp(luigi) { }); }).trigger('click'); $("#statusMessageModal").modal({}); + var refreshInterval = setInterval(function() { + if ($("#statusMessageModal").is(":hidden")) + clearInterval(refreshInterval) + else + $("#statusMessageModal .refresh").trigger('click'); + }, + 500 + ); } function preProcessGraph(dependencyGraph) { From 73b10a607e34471c8f844ffdc490f46c993f3cc4 Mon Sep 17 00:00:00 2001 From: Fabien Vinas Date: Thu, 4 May 2017 23:34:08 +0200 Subject: [PATCH 3/6] doc + handling cases when there is no status message but a progress bar --- doc/tasks.rst | 6 ++++-- luigi/scheduler.py | 4 ++-- luigi/static/visualiser/index.html | 4 ++++ luigi/static/visualiser/js/visualiserApp.js | 19 ++++++++++++++----- 4 files changed, 24 insertions(+), 9 deletions(-) diff --git a/doc/tasks.rst b/doc/tasks.rst index 4304d99993..754550b51e 100644 --- a/doc/tasks.rst +++ b/doc/tasks.rst @@ -184,8 +184,8 @@ Task status tracking For long-running or remote tasks it is convenient to see extended status information not only on the command line or in your logs but also in the GUI of the central scheduler. Luigi implements -dynamic status messages and tracking urls which may point to an external monitoring system. You -can set this information using callbacks within Task.run_: +dynamic status messages, progress bar and tracking urls which may point to an external monitoring system. +You can set this information using callbacks within Task.run_: .. code:: python @@ -199,6 +199,8 @@ can set this information using callbacks within Task.run_: # do some hard work here if i % 10 == 0: self.set_status_message("Progress: %d / 100" % i) + # displays a progress bar in the scheduler UI + self.set_progress_percentage(i) .. _Events: diff --git a/luigi/scheduler.py b/luigi/scheduler.py index 0060bea9f3..42abd89676 100644 --- a/luigi/scheduler.py +++ b/luigi/scheduler.py @@ -1197,7 +1197,7 @@ def _serialize_task(self, task_id, include_deps=True, deps=None): 'resources': task.resources, 'tracking_url': getattr(task, "tracking_url", None), 'status_message': getattr(task, "status_message", None), - 'progress_percentage': getattr(task, "progress_percentage", None) + 'progress_percentage': getattr(task, "progress_percentage", -1) } if task.status == DISABLED: ret['re_enable_able'] = task.scheduler_disable_time is not None @@ -1471,7 +1471,7 @@ def get_task_progress_percentage(self, task_id): task = self._state.get_task(task_id) return {"taskId": task_id, "progressPercentage": task.progress_percentage} else: - return {"taskId": task_id, "progressPercentage": 0} + return {"taskId": task_id, "progressPercentage": None} def _update_task_history(self, task, status, host=None): try: diff --git a/luigi/static/visualiser/index.html b/luigi/static/visualiser/index.html index bd83444a56..172eeab59d 100644 --- a/luigi/static/visualiser/index.html +++ b/luigi/static/visualiser/index.html @@ -245,6 +245,10 @@ {{#re_enable}}Re-enable{{/re_enable}} {{#trackingUrl}}{{/trackingUrl}} {{#statusMessage}}{{/statusMessage}} + {{^statusMessage}} + {{#progressPercentage}} + {{/progressPercentage}} + {{/statusMessage}}