fix ingester issue

Author: abbi4code

switchmap/poller/snmp/async_snmp_info.py

@@ -1,7 +1,6 @@
 """Async module to aggregate query results."""
 
 import time
-import sys
 from collections import defaultdict
 from switchmap.core import log
 import asyncio
@@ -57,10 +56,10 @@ async def everything(self):
         keys = ["misc", "system", "layer1", "layer2", "layer3"]
         for key, result in zip(keys, results):
             if isinstance(result, Exception):
-                log.log2warning(f"{key} failed: {result}")
+                log.log2warning(1004, f"{key} failed: {result}")
             elif result:
                 data[key] = result
-        print(f"final data: {data}")
+
         # Return
         return data
 

switchmap/poller/snmp/mib/generic/mib_if.py

@@ -5,6 +5,8 @@
 from switchmap.poller.snmp.base_query import Query
 from switchmap.core import general, log
 import asyncio
+from switchmap.core import general, log
+import asyncio
 
 
 def get_query():

switchmap/server/db/ingest/ingest.py

@@ -78,48 +78,39 @@ def process(self):
         with tempfile.TemporaryDirectory(
             dir=self._config.ingest_directory()
         ) as tmpdir:
-            # Only run the ingest if there is no poller lock file
-            # This helps to prevent ingesting files while polling is
-            # still running. This is only effective when the poller
-            # and ingester are running on the same machine
-            if os.path.isfile(poller_lock_file) is False:
-                # Copy files from cache to ingest
-                files.move_yaml_files(cache_directory, tmpdir)
-
-                # Parallel process the files
-                setup_success = setup(tmpdir, self._config)
-
-                if bool(setup_success) is True:
-                    # Populate the arguments
-                    arguments = [
-                        [
-                            IngestArgument(
-                                idx_zone=item.idx_zone,
-                                data=item.data,
-                                filepath=item.filepath,
-                                config=item.config,
-                                dns=self._dns,
-                            )
-                        ]
-                        for item in setup_success.zones
+            # ASYNC ARCHITECTURE: Allow ingester to run while poller is active
+            # Copy files from cache to ingest
+            files.move_yaml_files(cache_directory, tmpdir)
+
+            # Parallel process the files
+            setup_success = setup(tmpdir, self._config)
+
+            if bool(setup_success) is True:
+                # Populate the arguments
+                arguments = [
+                    [
+                        IngestArgument(
+                            idx_zone=item.idx_zone,
+                            data=item.data,
+                            filepath=item.filepath,
+                            config=item.config,
+                            dns=self._dns,
+                        )
                     ]
+                    for item in setup_success.zones
+                ]
 
-                    # Process the device independent zone data in the
-                    # database first
-                    if bool(arguments) is True:
-                        pairmacips = self.zone(arguments)
+                # Process the device independent zone data in the
+                # database first
+                if bool(arguments) is True:
+                    pairmacips = self.zone(arguments)
 
-                    # Process the device dependent in the database second
-                    if bool(pairmacips):
-                        self.device(arguments)
+                # Process the device dependent in the database second
+                if bool(pairmacips):
+                    self.device(arguments)
 
-                    # Cleanup
-                    self.cleanup(setup_success.event)
-            else:
-                log_message = f"""\
-Poller lock file {poller_lock_file} exists. Skipping processing of cache \
-files. Is the poller running or did it crash unexpectedly?"""
-                log.log2info(1077, log_message)
+                # Cleanup
+                self.cleanup(setup_success.event)
 
     def zone(self, arguments):
         """Ingest the files' zone data.

switchmap/server/db/ingest/update/device.py

@@ -34,6 +34,34 @@
 )
 
 
+def _decode_mac_address(encoded_mac):
+    """Decode double-encoded MAC addresses from async poller.
+    
+    Args:
+        encoded_mac: MAC address that may be double hex-encoded
+        
+    Returns:
+        str: Properly formatted MAC address or original if already valid
+        
+    """
+    import binascii
+    
+    try:
+        # Try to decode hex-encoded string to ASCII
+        if isinstance(encoded_mac, str) and len(encoded_mac) > 12:
+            decoded = binascii.unhexlify(encoded_mac).decode('ascii')
+            # Check if it starts with '0x' (hex prefix)
+            if decoded.startswith('0x'):
+                return decoded[2:]  # Return MAC without '0x' prefix
+        
+        # If decoding fails or doesn't match pattern, return original
+        return encoded_mac
+        
+    except Exception:
+        # If any decoding fails, return original
+        return encoded_mac
+
+
 def process(data, idx_zone, dns=True):
     """Process data received from a device.
 

switchmap/server/db/ingest/update/zone.py

@@ -17,6 +17,34 @@
 )
 
 
+def _decode_mac_address(encoded_mac):
+    """Decode double-encoded MAC addresses from async poller.
+    
+    Args:
+        encoded_mac: MAC address that may be double hex-encoded
+        
+    Returns:
+        str: Properly formatted MAC address or original if already valid
+        
+    """
+    import binascii
+    
+    try:
+        # Try to decode hex-encoded string to ASCII
+        if isinstance(encoded_mac, str) and len(encoded_mac) > 12:
+            decoded = binascii.unhexlify(encoded_mac).decode('ascii')
+            # Check if it starts with '0x' (hex prefix)
+            if decoded.startswith('0x'):
+                return decoded[2:]  # Return MAC without '0x' prefix
+        
+        # If decoding fails or doesn't match pattern, return original
+        return encoded_mac
+        
+    except Exception:
+        # If any decoding fails, return original
+        return encoded_mac
+
+
 def process(data, idx_zone, dns=True):
     """Process data received from a device.