Merge pull request #908 from wavezhang/master

[healthcheck] Withdraw routes when exit

lib/exabgp/application/healthcheck.py

@@ -349,14 +349,6 @@ def setup_ips(ips, label, sudo=False):
             subprocess.check_call(
                 cmd, stdout=fnull, stderr=fnull)
 
-    # If we setup IPs we should also remove them on SIGTERM
-    def sigterm_handler(signum, frame):  # pylint: disable=W0612,W0613
-        remove_ips(ips, label, sudo)
-        sys.exit(0)
-
-    signal.signal(signal.SIGTERM, sigterm_handler)
-
-
 def remove_ips(ips, label, sudo=False):
     """Remove added IP on loopback interface"""
     existing = set(loopback_ips(label))
@@ -456,14 +448,15 @@ def loop(options):
         "FALLING",              # Checks are currently failing.
         "UP",                   # Service is considered as up.
         "DOWN",                 # Service is considered as down.
+        "EXIT",                 # Exit state
     )
 
     def exabgp(target):
         """Communicate new state to ExaBGP"""
-        if target not in (states.UP, states.DOWN, states.DISABLED):
+        if target not in (states.UP, states.DOWN, states.DISABLED, states.EXIT):
             return
         # dynamic ip management. When the service fail, remove the loopback
-        if target in (states.DOWN, states.DISABLED) and options.ip_dynamic:
+        if target in (states.DOWN, states.DISABLED, states.EXIT) and options.ip_dynamic:
             logger.info("service down, deleting loopback ips")
             remove_ips(options.ips, options.label, options.sudo)
         # if ips was deleted with dyn ip, re-setup them
@@ -472,7 +465,8 @@ def exabgp(target):
             setup_ips(options.ips, options.label, options.sudo)
 
         logger.info("send announces for %s state to ExaBGP", target)
-        metric = vars(options).get("{0}_metric".format(str(target).lower()))
+        default_metric = vars(options).get("{0}_metric".format(str(states.DISABLED).lower()))
+        metric = vars(options).get("{0}_metric".format(str(target).lower()), default_metric)
         for ip in options.ips:
             if options.withdraw_on_down:
                 command = "announce" if target is states.UP else "withdraw"
@@ -601,6 +595,13 @@ def one(checks, state):
 
     checks = 0
     state = states.INIT
+    # Do cleanups on SIGTERM
+    def sigterm_handler(signum, frame):  # pylint: disable=W0612,W0613
+        exabgp(states.EXIT)
+        sys.exit(0)
+
+    signal.signal(signal.SIGTERM, sigterm_handler)
+
     while True:
         checks, state = one(checks, state)
 

lib/exabgp/reactor/api/processes.py

@@ -29,6 +29,7 @@
 from exabgp.version import text as text_version
 
 from exabgp.configuration.environment import environment
+from threading import Thread
 
 
 # pylint: disable=no-self-argument,not-callable,unused-argument,invalid-name
@@ -73,6 +74,8 @@ def clean (self):
 		self._respawning = {}
 
 	def _handle_problem (self, process):
+		if process not in self._process:
+			return
 		if self.respawn_number:
 			self.logger.debug('issue with the process, restarting it','process')
 			self._terminate(process)
@@ -83,13 +86,18 @@ def _handle_problem (self, process):
 
 	def _terminate (self, process):
 		self.logger.debug('terminating process %s' % process,'process')
+		thread = Thread(target=self._terminate_run, args = (process,))
+		thread.start()
+		return thread
+
+	def _terminate_run (self, process):
 		try:
 			self._process[process].terminate()
-		except OSError:
+			self._process[process].wait()
+			del self._process[process]
+		except (OSError, KeyError):
 			# the process is most likely already dead
 			pass
-		self._process[process].wait()
-		del self._process[process]
 
 	def terminate (self):
 		for process in list(self._process):
@@ -104,7 +112,8 @@ def terminate (self):
 		time.sleep(0.1)
 		for process in list(self._process):
 			try:
-				self._terminate(process)
+				t = self._terminate(process)
+				t.join()
 			except OSError:
 				# we most likely received a SIGTERM signal and our child is already dead
 				self.logger.debug('child process %s was already dead' % process,'process')
@@ -236,6 +245,8 @@ def received (self):
 				except StopIteration:
 					if not consumed_data:
 						self._handle_problem(process)
+			except KeyError:
+				pass
 			except (subprocess.CalledProcessError,OSError,ValueError):
 				self._handle_problem(process)