Poller: More minor fixes and debug messages

We were continuing to attempt to poll when we had authentication
failures and other such errors. When we retrieved netbox inventory, it
was hard to know how many entries we got and how many we ignored because
of an error. Finally, when we raised NotImplementedError, we didn't trap
it in an exception handling clause causing an entire worker to die and
we didn't print additional info about what was not implemented and on
what device IP.

This patch addresses all this.

Signed-off-by: Dinesh Dutt <[email protected]>

Author: ddutt

suzieq/poller/controller/source/netbox.py

@@ -173,6 +173,9 @@ async def get_inventory_list(self) -> List:
         except Exception as e:
             raise InventorySourceError(f'{self.name}: error while '
                                        f'getting devices: {e}')
+
+        logger.info(
+            f'Netbox: Retrieved inventory list of {len(devices)} devices')
         return devices
 
     async def _get_devices(self, url: str) -> Tuple[List, str]:
@@ -224,6 +227,7 @@ def get_field_value(entry: dict, fields_str: str):
             return cur_field
 
         inventory = {}
+        ignored_device_count = 0
 
         for device in inventory_list:
             ipv4 = get_field_value(device, 'primary_ip4.address')
@@ -245,6 +249,7 @@ def get_field_value(entry: dict, fields_str: str):
                 logger.warning(
                     f"Skipping {namespace}.{hostname}: doesn't have a "
                     "management IP")
+                ignored_device_count += 1
                 continue
 
             address = address.split("/")[0]
@@ -255,6 +260,10 @@ def get_field_value(entry: dict, fields_str: str):
                 'hostname': hostname,
             }
 
+        logger.info(
+            f'Netbox: Acting on inventory of {len(inventory)} devices, '
+            f'ignoring {ignored_device_count} devices')
+
         return inventory
 
     async def _execute(self):

suzieq/poller/worker/inventory/inventory.py

@@ -125,7 +125,12 @@ async def _init_nodes(self, inventory_list:
             )]
 
         for n in asyncio.as_completed(init_tasks):
-            newnode = await n
+            try:
+                newnode = await n
+            except Exception as e:  # pylint: disable=broad-except
+                logger.error(
+                    f'Encountered error {e} in initializing node')
+                continue
             if newnode.devtype == "unsupported":
                 logger.error(
                     f'Unsupported device type for '

suzieq/poller/worker/nodes/node.py

@@ -184,11 +184,11 @@ async def initialize(self, **kwargs) -> TNode:
 
         # Now we know the dev type, fetch the data we need about the
         # device to get cracking
-        if not self.devtype:
+        if not self.devtype and self._retry:
             self.backoff = min(600, self.backoff * 2) + \
                 (random.randint(0, 1000) / 1000)
             self.init_again_at = time.time() + self.backoff
-        elif self.devtype != 'unsupported':
+        elif self.devtype not in ['unsupported', None] and self._retry:
             # OK, we know the devtype, now initialize the info we need
             # to start proper operation
 
@@ -271,6 +271,9 @@ def _create_error(self, cmd) -> dict:
             status = HTTPStatus.REQUEST_TIMEOUT
         elif isinstance(self.current_exception, asyncssh.misc.ProtocolError):
             status = HTTPStatus.FORBIDDEN
+        elif isinstance(self.current_exception,
+                        asyncssh.misc.PermissionDenied):
+            status = HTTPStatus.FORBIDDEN
         elif hasattr(self.current_exception, 'code'):
             status = self.current_exception.code
         else:
@@ -631,14 +634,16 @@ async def _init_ssh(self, init_dev_data=True, use_lock=True) -> None:
                                       f'{self.address}:{self.port}, {e}')
                 self.current_exception = e
                 await self._close_connection()
+                self._conn = None
             finally:
                 if use_lock:
                     self.ssh_ready.release()
 
     @abstractmethod
     async def _init_rest(self):
         '''Check that connectivity exists and works'''
-        raise NotImplementedError
+        raise NotImplementedError(
+            f'{self.address}: REST transport is not supported')
 
     # pylint: disable=unused-argument
     async def _local_gather(self, service_callback: Callable,
@@ -789,7 +794,7 @@ async def _exec_service(self, service_callback, svc_defn: dict,
         if not svc_defn:
             return result
 
-        if not self.devtype:
+        if not self.devtype and self._retry:
             if self.init_again_at < time.time():
                 await self._detect_node_type()
 
@@ -882,7 +887,8 @@ async def _fetch_init_dev_data(self):
         This is where the list of commands specific to the device for
         extracting the said info is specified.
         """
-        raise NotImplementedError
+        raise NotImplementedError(
+            f'{self.address}: initing base Node class')
 
     @abstractmethod
     async def _parse_init_dev_data(self, output: List,
@@ -897,7 +903,8 @@ async def _parse_init_dev_data(self, output: List,
             output: The list of outputs, one per command specified
             cb_token: The callback token we passed to the data fetcher function
         """
-        raise NotImplementedError
+        raise NotImplementedError(
+            f'{self.address}: parsing init base Node class')
 
     @abstractmethod
     def _extract_nos_version(self, data: str) -> str:
@@ -910,7 +917,8 @@ def _extract_nos_version(self, data: str) -> str:
         Returns:
             The version as a string
         """
-        raise NotImplementedError
+        raise NotImplementedError(
+            f'{self.address}: extracting NOS in init base Node class')
 
     @abstractmethod
     async def _rest_gather(self, service_callback: Callable,
@@ -932,7 +940,8 @@ async def _rest_gather(self, service_callback: Callable,
             only_one: Run till the first command in the list succeeds and
                 then return
         """
-        raise NotImplementedError
+        raise NotImplementedError(
+            f'{self.address}: REST transport is not supported')
 
     def post_commands(self, service_callback: Callable, svc_defn: Dict,
                       cb_token: RsltToken):
@@ -1148,7 +1157,7 @@ async def _parse_init_dev_data(self, output, cb_token) -> None:
                 self.logger.error(f'nodeinit: Error getting version for '
                                   f'{self.address}: {self.port}')
 
-        if output[1]["status"] == 0 or output[1]["status"] == 200:
+        if (len(output) > 1) and (output[1]["status"] in [0, 200]):
             if self.transport == 'ssh':
                 try:
                     data = json.loads(output[1]["data"])
@@ -1192,12 +1201,12 @@ async def _parse_init_dev_data(self, output, _) -> None:
             upsecs = output[0]["data"].split()[0]
             self.bootupTimestamp = int(int(time.time()*1000)
                                        - float(upsecs)*1000)
-        if output[1]["status"] == 0:
+        if (len(output) > 1) and (output[1]["status"] == 0):
             data = output[1].get("data", '')
             hostname = data.splitlines()[0].strip()
             self.hostname = hostname
 
-        if output[2]["status"] == 0:
+        if (len(output) > 2) and (output[2]["status"] == 0):
             data = output[2].get("data", '')
             self._extract_nos_version(data)
 
@@ -1291,11 +1300,13 @@ class IosXRNode(Node):
     '''IOSXR Node specific implementation'''
 
     async def _init_rest(self):
-        raise NotImplementedError
+        raise NotImplementedError(
+            f'{self.address}: REST transport is not supported')
 
     async def _rest_gather(self, service_callback, cmd_list, cb_token,
                            oformat='json', timeout=None):
-        raise NotImplementedError
+        raise NotImplementedError(
+            f'{self.address}: REST transport is not supported')
 
     async def _fetch_init_dev_data(self):
         """Fill in the boot time of the node by executing certain cmds"""
@@ -1363,7 +1374,7 @@ async def _parse_init_dev_data(self, output, cb_token) -> None:
 
             self._extract_nos_version(data)
 
-        if output[1]["status"] == 0:
+        if (len(output) > 1) and (output[1]["status"] == 0):
             data = output[1]['data']
             hostname = re.search(r'hostname (\S+)', data.strip())
             if hostname:
@@ -1385,11 +1396,13 @@ class IosXENode(Node):
     '''IOS-XE Node-sepcific telemetry gather implementation'''
 
     async def _init_rest(self):
-        raise NotImplementedError
+        raise NotImplementedError(
+            f'{self.address}: REST transport is not supported')
 
     async def _rest_gather(self, service_callback, cmd_list, cb_token,
                            oformat='json', timeout=None):
-        raise NotImplementedError
+        raise NotImplementedError(
+            f'{self.address}: REST transport is not supported')
 
     async def _fetch_init_dev_data(self):
         """Fill in the boot time of the node by executing certain cmds"""
@@ -1610,22 +1623,26 @@ class IOSNode(IosXENode):
     '''Classic IOS Node-specific implementation'''
 
     async def _init_rest(self):
-        raise NotImplementedError
+        raise NotImplementedError(
+            f'{self.address}: REST transport is not supported')
 
     async def _rest_gather(self, service_callback, cmd_list, cb_token,
                            oformat='json', timeout=None):
-        raise NotImplementedError
+        raise NotImplementedError(
+            f'{self.address}: REST transport is not supported')
 
 
 class JunosNode(Node):
     '''Juniper's Junos node-specific implementation'''
 
     async def _init_rest(self):
-        raise NotImplementedError
+        raise NotImplementedError(
+            f'{self.address}: REST transport is not supported')
 
     async def _rest_gather(self, service_callback, cmd_list, cb_token,
                            oformat='json', timeout=None):
-        raise NotImplementedError
+        raise NotImplementedError(
+            f'{self.address}: REST transport is not supported')
 
     async def _fetch_init_dev_data(self):
         """Fill in the boot time of the node by running requisite cmd"""
@@ -1653,7 +1670,7 @@ async def _parse_init_dev_data(self, output, cb_token) -> None:
             self.bootupTimestamp = (get_timestamp_from_junos_time(
                 timestr, output[0]['timestamp']/1000)/1000)
 
-        if output[1]["status"] == 0:
+        if (len(output) > 1) and (output[1]["status"] == 0):
             data = output[1]["data"]
             hmatch = re.search(r'\nHostname:\s+(\S+)\n', data)
             if hmatch:
@@ -1676,12 +1693,14 @@ class NxosNode(Node):
     '''Cisco's NXOS Node-specific implementation'''
 
     async def _init_rest(self):
-        raise NotImplementedError
+        raise NotImplementedError(
+            f'{self.address}: REST transport is not supported')
 
     async def _rest_gather(self, service_callback, cmd_list, cb_token,
                            oformat="json", timeout=None):
         '''Gather data for service via device REST API'''
-        raise NotImplementedError
+        raise NotImplementedError(
+            f'{self.address}: REST transport is not supported')
 
     async def _fetch_init_dev_data(self):
         """Fill in the boot time of the node by running requisite cmd"""
@@ -1731,12 +1750,14 @@ class SonicNode(Node):
     '''SONiC Node-specific implementtaion'''
 
     async def _init_rest(self):
-        raise NotImplementedError
+        raise NotImplementedError(
+            f'{self.address}: REST transport is not supported')
 
     async def _rest_gather(self, service_callback, cmd_list, cb_token,
                            oformat="json", timeout=None):
         '''Gather data for service via device REST API'''
-        raise NotImplementedError
+        raise NotImplementedError(
+            f'{self.address}: REST transport is not supported')
 
     async def _fetch_init_dev_data(self):
         """Fill in the boot time of the node by running requisite cmd"""
@@ -1751,9 +1772,9 @@ async def _parse_init_dev_data(self, output, cb_token) -> None:
             upsecs = output[0]["data"].split()[0]
             self.bootupTimestamp = int(int(time.time()*1000)
                                        - float(upsecs)*1000)
-        if output[1]["status"] == 0:
+        if (len(output) > 1) and (output[1]["status"] == 0):
             self.hostname = output[1]["data"].strip()
-        if output[2]["status"] == 0:
+        if (len(output) > 2) and (output[2]["status"] == 0):
             self._extract_nos_version(output[1]["data"])
 
     def _extract_nos_version(self, data: str) -> None: