From 1b2f0fd800b3b02a230d3613273f796fbc469c11 Mon Sep 17 00:00:00 2001
From: Kevin Wang <kevin@kevinwang.com>
Date: Tue, 17 Apr 2018 11:55:06 -0700
Subject: [PATCH] Add configuration to enable/disable the pause button (#2399)

---
 doc/configuration.rst                       |  4 ++++
 luigi/scheduler.py                          | 12 ++++++++++--
 luigi/static/visualiser/js/luigi.js         |  6 ++++++
 luigi/static/visualiser/js/visualiserApp.js |  6 +++++-
 test/scheduler_test.py                      |  8 ++++++++
 5 files changed, 33 insertions(+), 3 deletions(-)

diff --git a/doc/configuration.rst b/doc/configuration.rst
index 265755dbc5..06bed2f648 100644
--- a/doc/configuration.rst
+++ b/doc/configuration.rst
@@ -722,6 +722,10 @@ worker_disconnect_delay
   scheduler before removing it and marking all of its running tasks as
   failed. Defaults to 60.
 
+pause_enabled
+  If false, disables pause/unpause operations and hides the pause toggle from
+  the visualiser.
+
 
 [sendgrid]
 ----------
diff --git a/luigi/scheduler.py b/luigi/scheduler.py
index f83893e3c0..405507028b 100644
--- a/luigi/scheduler.py
+++ b/luigi/scheduler.py
@@ -147,6 +147,8 @@ class scheduler(Config):
 
     prune_on_get_work = parameter.BoolParameter(default=False)
 
+    pause_enabled = parameter.BoolParameter(default=True)
+
     def _get_retry_policy(self):
         return RetryPolicy(self.retry_count, self.disable_hard_timeout, self.disable_window)
 
@@ -930,17 +932,23 @@ def disable_worker(self, worker):
     def set_worker_processes(self, worker, n):
         self._state.get_worker(worker).add_rpc_message('set_worker_processes', n=n)
 
+    @rpc_method()
+    def is_pause_enabled(self):
+        return {'enabled': self._config.pause_enabled}
+
     @rpc_method()
     def is_paused(self):
         return {'paused': self._paused}
 
     @rpc_method()
     def pause(self):
-        self._paused = True
+        if self._config.pause_enabled:
+            self._paused = True
 
     @rpc_method()
     def unpause(self):
-        self._paused = False
+        if self._config.pause_enabled:
+            self._paused = False
 
     @rpc_method()
     def update_resources(self, **resources):
diff --git a/luigi/static/visualiser/js/luigi.js b/luigi/static/visualiser/js/luigi.js
index c44b776350..d6eec8a337 100644
--- a/luigi/static/visualiser/js/luigi.js
+++ b/luigi/static/visualiser/js/luigi.js
@@ -158,6 +158,12 @@ var LuigiAPI = (function() {
         });
     };
 
+    LuigiAPI.prototype.isPauseEnabled = function(callback) {
+        jsonRPC(this.urlRoot + '/is_pause_enabled', {}, function(response) {
+            callback(response.response.enabled);
+        });
+    };
+
     LuigiAPI.prototype.pause = function() {
         jsonRPC(this.urlRoot + '/pause');
     };
diff --git a/luigi/static/visualiser/js/visualiserApp.js b/luigi/static/visualiser/js/visualiserApp.js
index 4605a27047..3a6715a16b 100644
--- a/luigi/static/visualiser/js/visualiserApp.js
+++ b/luigi/static/visualiser/js/visualiserApp.js
@@ -1011,7 +1011,11 @@ function visualiserApp(luigi) {
     $(document).ready(function() {
         loadTemplates();
 
-        luigi.isPaused(createPauseToggle);
+        luigi.isPauseEnabled(function(enabled) {
+            if (enabled) {
+                luigi.isPaused(createPauseToggle);
+            }
+        });
 
         luigi.getWorkerList(function(workers) {
             $("#workerList").append(renderWorkers(workers));
diff --git a/test/scheduler_test.py b/test/scheduler_test.py
index b4f4f6f548..21199c96a5 100644
--- a/test/scheduler_test.py
+++ b/test/scheduler_test.py
@@ -229,6 +229,14 @@ def test_per_task_retry_policy(self):
         task_9.add_failure()
         self.assertTrue(task_9.has_excessive_failures())
 
+    @with_config({'scheduler': {'pause_enabled': 'false'}})
+    def test_pause_disabled(self):
+        s = luigi.scheduler.Scheduler()
+        self.assertFalse(s.is_pause_enabled()['enabled'])
+        self.assertFalse(s.is_paused()['paused'])
+        s.pause()
+        self.assertFalse(s.is_paused()['paused'])
+
 
 class SchedulerWorkerTest(unittest.TestCase):
     def get_pending_ids(self, worker, state):