Fix #965 - Make Flapping Ignore Commanded Stops

circus/arbiter.py

@@ -619,7 +619,7 @@ def reap_processes(self):
 
                     if pid in watchers_pids:
                         watcher = watchers_pids[pid]
-                        watcher.reap_process(pid, status)
+                        watcher.reap_process(pid, status, commanded_reap=False)
                 except OSError as e:
                     if e.errno == errno.EAGAIN:
                         sleep(0)

circus/plugins/flapping.py

@@ -54,12 +54,19 @@ def handle_stop(self):
 
     def handle_recv(self, data):
         watcher_name, action, msg = self.split_data(data)
-        if action == "reap":
-            timeline = self.timelines.get(watcher_name, [])
-            timeline.append(time.time())
-            self.timelines[watcher_name] = timeline
+        try:
+            message = self.load_message(msg)
+        except (ValueError, TypeError):
+            message = None
+            logger.error("Error while decoding json for message: %s", msg)
 
-            self.check(watcher_name)
+        if action == "reap":
+            commanded_reap = message and message.get('commanded_reap', False)
+            if not commanded_reap:
+                timeline = self.timelines.get(watcher_name, [])
+                timeline.append(time.time())
+                self.timelines[watcher_name] = timeline
+                self.check(watcher_name)
         elif action == "updated":
             self.update_conf(watcher_name)
 

circus/tests/test_plugin_flapping.py

@@ -1,4 +1,5 @@
 from mock import patch
+import json
 
 from circus.tests.support import TestCircus, EasyTestSuite
 from circus.plugins.flapping import Flapping
@@ -28,6 +29,17 @@ def test_reap_message_calls_check(self, check_mock):
 
         check_mock.assert_called_with('test')
 
+    @patch.object(Flapping, 'check')
+    def test_command_generated_reap_message_doesnt_call_check(self,
+                                                              check_mock):
+        plugin = self._flapping_plugin()
+        topic = 'watcher.test.reap'
+        message = json.dumps({'commanded_reap': True})
+
+        plugin.handle_recv([topic, message])
+
+        check_mock.assert_not_called()
+
     @patch.object(Flapping, 'cast')
     @patch('circus.plugins.flapping.Timer')
     def test_below_max_retry_triggers_restart(self, timer_mock, cast_mock):

circus/watcher.py

@@ -435,7 +435,7 @@ def notify_event(self, topic, msg):
             self.evpub_socket.send_multipart(multipart_msg)
 
     @util.debuglog
-    def reap_process(self, pid, status=None):
+    def reap_process(self, pid, status=None, commanded_reap=True):
         """ensure that the process is killed (and not a zombie)"""
         if pid not in self.processes:
             return
@@ -476,7 +476,8 @@ def reap_process(self, pid, status=None):
                     "reap",
                     {"process_pid": pid,
                      "time": time.time(),
-                     "exit_code": process.returncode()})
+                     "exit_code": process.returncode(),
+                     "commanded_reap": commanded_reap})
                 process.stop()
                 return
 
@@ -512,10 +513,11 @@ def reap_process(self, pid, status=None):
         self.notify_event("reap",
                           {"process_pid": pid,
                            "time": time.time(),
-                           "exit_code": exit_code})
+                           "exit_code": exit_code,
+                           "commanded_reap": commanded_reap})
 
     @util.debuglog
-    def reap_processes(self):
+    def reap_processes(self, commanded_reap=True):
         """Reap all the processes for this watcher.
         """
         if self.is_stopped():
@@ -524,7 +526,7 @@ def reap_processes(self):
 
         # reap_process changes our dict, look through the copy of keys
         for pid in list(self.processes.keys()):
-            self.reap_process(pid)
+            self.reap_process(pid, commanded_reap=commanded_reap)
 
     @gen.coroutine
     @util.debuglog