Merge pull request #751 from netenglabs/fix-exceptions

Fix exception handling in the poller

Author: ddutt

suzieq/poller/sq_poller.py

@@ -52,7 +52,12 @@ async def start_controller(user_args: argparse.Namespace, config_data: Dict):
         if not log_stdout:
             print(f"ERROR: {error}")
         logger.error(error)
-        sys.exit(-1)
+        sys.exit(1)
+    except Exception as error:
+        if not log_stdout:
+            traceback.print_exc()
+        logger.critical(f'{error}\n{traceback.format_exc()}')
+        sys.exit(1)
 
 
 def controller_main():
@@ -198,8 +203,6 @@ def controller_main():
         asyncio.run(start_controller(args, cfg))
     except (KeyboardInterrupt, RuntimeError):
         pass
-    except Exception:  # pylint: disable=broad-except
-        traceback.print_exc()
 
 
 if __name__ == '__main__':

suzieq/poller/worker/nodes/node.py

@@ -146,8 +146,7 @@ async def initialize(self, **kwargs) -> TNode:
                                                           passphrase)
                 if not self.jump_host_key:
                     raise SqPollerConfError('Unable to read private key file'
-                                            f' at {pvtkey_file}'
-                                            )
+                                            f' at {pvtkey_file}')
         else:
             self.jump_host = None
             self.jump_host_key = None
@@ -463,6 +462,15 @@ async def _detect_node_type(self):
                 # that sets the device type
                 devtype = self.devtype
             except Exception:
+                self.logger.exception(f'{self.address}:{self.port}: Node '
+                                      'discovery failed due to exception')
+                # All the exceptions related to timeouts and authentication
+                # problems are already catched inside. If we get an
+                # exception here, this is unexpected and most likely something
+                # went wrong with the command output parsing.
+                # In this case there is not point in retrying discovery, it is
+                # likely a bug.
+                self._retry = 0
                 devtype = None
 
             if not devtype:
@@ -840,7 +848,6 @@ async def _exec_cmd(self, service_callback, cmd_list, cb_token,
         if only_one is True, commands are executed until the first one that
         succeeds, and the rest are ignored.
         '''
-
         if self.transport == "ssh":
             await self._ssh_gather(service_callback, cmd_list, cb_token,
                                    oformat, timeout, only_one)
@@ -1220,7 +1227,7 @@ async def _rest_gather(self, service_callback, cmd_list, cb_token,
                     result.append(self._create_error(cmd))
                 self.logger.error(
                     f"{self.transport}://{self.hostname}:{self.port}: Unable "
-                    "to communicate with node due to str(e)")
+                    f"to communicate with node due to {str(e)}")
 
         await service_callback(result, cb_token)
 
@@ -1331,8 +1338,8 @@ async def _init_rest(self):
             except Exception as e:
                 self.current_exception = e
                 self.logger.error(
-                    "{self.transport}://{self.hostname}:{self.port}: Unable to"
-                    " communicate with node due to {str(e)}")
+                    f"{self.transport}://{self.hostname}:{self.port}: Unable "
+                    f"to communicate with node due to {str(e)}")
 
     async def _rest_gather(self, service_callback, cmd_list, cb_token,
                            oformat='json', timeout=None):
@@ -1372,8 +1379,8 @@ async def _rest_gather(self, service_callback, cmd_list, cb_token,
                 self.current_exception = e
                 result.append(self._create_error(cmd_list))
                 self.logger.error(
-                    "{self.transport}://{self.hostname}:{self.port}: Unable "
-                    "to communicate with node due to {str(e)}")
+                    f"{self.transport}://{self.hostname}:{self.port}: Unable "
+                    f"to communicate with node due to {str(e)}")
 
         await service_callback(result, cb_token)
 
@@ -1700,10 +1707,10 @@ async def _ssh_gather(self, service_callback, cmd_list, cb_token, oformat,
                             await self._close_connection()
                             self.logger.debug("Closed conn successfully for "
                                               f"{self.hostname}")
-                        except Exception as e1:
+                        except Exception as close_exc:
                             self.logger.error(
                                 f"Caught an exception closing {self.hostname}"
-                                f" for {cmd}: {e1}")
+                                f" for {cmd}: {close_exc}")
                     else:
                         self.logger.error(
                             f"Unable to connect to {self.hostname} {cmd} "

suzieq/poller/worker/sq_worker.py

@@ -8,11 +8,10 @@
 from typing import Dict
 
 import uvloop
-
 from suzieq.poller.worker.worker import Worker
 from suzieq.poller.worker.writers.output_worker import OutputWorker
 from suzieq.shared.exceptions import InventorySourceError, SqPollerConfError
-from suzieq.shared.utils import poller_log_params, init_logger, load_sq_config
+from suzieq.shared.utils import init_logger, load_sq_config, poller_log_params
 
 
 async def start_worker(userargs: argparse.Namespace, cfg: Dict):
@@ -36,8 +35,14 @@ async def start_worker(userargs: argparse.Namespace, cfg: Dict):
         await worker.init_worker()
         await worker.run()
     except (SqPollerConfError, InventorySourceError) as error:
+        if not log_stdout:
+            print(error)
         logger.error(error)
-        print(error)
+        sys.exit(1)
+    except Exception as error:
+        if not log_stdout:
+            traceback.print_exc()
+        logger.critical(f'{error}\n{traceback.format_exc()}')
         sys.exit(1)
 
 
@@ -130,8 +135,6 @@ def worker_main():
         asyncio.run(start_worker(userargs, cfg))
     except (KeyboardInterrupt, RuntimeError):
         pass
-    except Exception:  # pylint: disable=broad-except
-        traceback.print_exc()
 
     sys.exit(0)
 

suzieq/poller/worker/worker.py

@@ -138,8 +138,11 @@ async def run(self):
             tasks = await self._pop_waiting_worker_tasks()
             while tasks:
                 try:
-                    _, pending = await asyncio.wait(
+                    done, pending = await asyncio.wait(
                         tasks, return_when=asyncio.FIRST_COMPLETED)
+                    for d in done:
+                        if d.exception():
+                            raise d.exception()
                     tasks = list(pending)
                     running_svcs = self.service_manager.running_services
                     if tasks and any(i._coro in running_svcs