sandbox agent: minor fix to graceful shutdown

Author: iakkus

SandboxAgent/deployment.py

@@ -113,43 +113,47 @@ def get_all_children_pids(self):
         return children_pids
 
     def check_child_process(self):
+        log_filepath = None
         pid, status = os.waitpid(-1, os.WNOHANG|os.WUNTRACED|os.WCONTINUED)
-        failed_process_name = ""
+        if pid == 0:
+            return True, pid, "Unknown process", log_filepath
+
+        stopped_process_name = ""
         if os.WIFCONTINUED(status) or os.WIFSTOPPED(status):
             return False, _
         if os.WIFSIGNALED(status) or os.WIFEXITED(status):
             self._logger.error("Process with pid: " + str(pid) + " stopped.")
             if pid == self._fluentbit_actual_pid:
-                failed_process_name = "Fluent-bit"
+                stopped_process_name = "Fluent-bit"
                 log_filepath = "/opt/mfn/LoggingService/fluent-bit/fluent-bit.log"
             elif pid == self._queue_service_process.pid:
-                failed_process_name = "Queue service"
+                stopped_process_name = "Queue service"
                 log_filepath = None
             elif pid == self._frontend_process.pid:
-                failed_process_name = "Frontend"
+                stopped_process_name = "Frontend"
                 log_filepath = "/opt/mfn/logs/frontend.log"
             else:
                 for jrhp in self._javarequesthandler_process_list:
                     if pid == jrhp.pid:
-                        failed_process_name = "Java request handler"
+                        stopped_process_name = "Java request handler"
                         log_filepath = "/opt/mfn/logs/javaworker.log"
                         break
                 for state_name in self._functionworker_process_map:
                     process = self._functionworker_process_map[state_name]
                     if pid == process.pid:
-                        failed_process_name = "Function worker (" + state_name + ")"
+                        stopped_process_name = "Function worker (" + state_name + ")"
                         log_filepath = "/opt/mfn/logs/function_" + state_name + ".log"
                         del self._functionworker_process_map[state_name]
                         break
 
-            self._logger.error("Failed process name: " + failed_process_name)
+            self._logger.error("Stopped process name: " + stopped_process_name)
 
         if os.path.exists('/var/run/secrets/kubernetes.io'):
-            return True, pid, failed_process_name, log_filepath
+            return True, pid, stopped_process_name, log_filepath
         else:
             # TODO: try to relaunch some of the processes (FWs, fluentbit, frontend)
             self._logger.info(self._child_process_command_args_map[pid])
-            return True, pid, failed_process_name, log_filepath
+            return True, pid, stopped_process_name, log_filepath
 
     def shutdown(self):
         shutdown_message = {}

SandboxAgent/process_utils.py

@@ -73,7 +73,7 @@ def run_command(command, logger, custom_env=None, wait_output=False, process_log
 
 def run_command_return_output(cmd, logger):
     error = None
-    output = None
+    output = ""
     try:
         args = shlex.split(cmd)
         child = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE, close_fds=True)
@@ -82,7 +82,7 @@ def run_command_return_output(cmd, logger):
         if child.returncode != 0:
             error = child_stderr_bytes.decode().strip()
     except Exception as exc:
-        logger.error('[SandboxAgent] Could not execute command and return output: %s', str(exc))
+        logger.error('[SandboxAgent] Could not execute command and return output: %s, %s', cmd, str(exc))
         error = exc
 
     return output.strip(), error

SandboxAgent/sandboxagent.py

@@ -200,8 +200,6 @@ def _process_deployment_info(self):
                 else:
                     self._management_endpoints = json.loads(self._management_endpoints)
 
-
-
         if not has_error:
             self._logger.info("External endpoint: %s", self._external_endpoint)
             self._logger.info("Internal endpoint: %s", self._internal_endpoint)
@@ -224,30 +222,30 @@ def sigterm(self, signum, _):
         raise KeyboardInterrupt
 
     def sigchld(self, signum, _):
-        should_shutdown, pid, failed_process_name, log_filepath = self._deployment.check_child_process()
+        if self._shutting_down:
+            return
+
+        should_shutdown, pid, stopped_process_name, log_filepath = self._deployment.check_child_process()
 
         if should_shutdown:
-            self._update_deployment_status(True, "A sandbox process stopped unexpectedly: " + failed_process_name, log_filepath)
+            self._update_deployment_status(True, "A sandbox process stopped unexpectedly: " + stopped_process_name, log_filepath)
 
             if pid == self._queue_service_process.pid:
                 self._queue_service_process = None
             elif pid == self._frontend_process.pid:
                 self._frontend_process = None
-            elif failed_process_name == "Fluent-bit":
+            elif stopped_process_name == "Fluent-bit":
                 self._fluentbit_process = None
 
-            self.shutdown(reason="Process " + failed_process_name + " with pid: " + str(pid) + " stopped unexpectedly.")
+            self.shutdown(reason="Process " + stopped_process_name + " with pid: " + str(pid) + " stopped unexpectedly.")
 
     def shutdown(self, reason=None):
-        if self._shutting_down:
-            return
-
         self._shutting_down = True
         errmsg = ""
         if reason is not None:
             errmsg = "Shutting down sandboxagent due to reason: " + reason + "..."
             self._logger.info(errmsg)
-            # some process dies unexpectedly; need to stop as immediately as possible
+            # some process died unexpectedly; need to stop as immediately as possible
             if self._fluentbit_process is not None:
                 time.sleep(2) # flush interval of fluent-bit
             os._exit(1)
@@ -269,8 +267,8 @@ def shutdown(self, reason=None):
             self._local_queue_client.removeTopic(self._instructions_topic)
             self._local_queue_client.shutdown()
 
-            self._logger.info("Shutting down the queue service...")
-            process_utils.terminate_and_wait_child(self._queue_service_process, "queue service", 5, self._logger)
+            #self._logger.info("Shutting down the queue service...")
+            #process_utils.terminate_and_wait_child(self._queue_service_process, "queue service", 5, self._logger)
         else:
             self._logger.info("No queue service; most probably it was the reason of the shutdown.")
             self._logger.info("Force shutting down the function worker(s)...")