Fix NetBox state change race conditions with unified locking and connection limiting (#1839)

Implemented critical fixes to prevent race conditions when multiple
state changes occur simultaneously for NetBox nodes:

1. Unified Lock Keys:
   - All three state-setting functions now use same lock key per device
   - Previously each function used different locks allowing parallel execution
   - Prevents race conditions when multiple states change in quick succession
   - Fixed incorrect lock key in set_power_state (was using provision_state key)
   - Lock key: lock_osism_tasks_netbox_{device_name}

2. NetBox Connection Limiting:
   - Implemented custom Redis-based semaphore (pottery lacks RedisSemaphore)
   - Default: maximum 5 concurrent operations per NetBox instance
   - Configurable via NETBOX_MAX_CONNECTIONS environment variable
   - Prevents overloading NetBox services during parallel operations
   - Works for both primary and secondary NetBox instances
   - Uses Redis sorted sets for distributed coordination

3. Increased Timeouts:
   - Lock acquisition timeout: 20s -> 120s (2 minutes)
   - Auto-release time: 60s -> 300s (5 minutes)
   - Handles high-concurrency scenarios during node provisioning and sync

Technical Implementation:
- Custom RedisSemaphore class using Redis ZSET operations
- Helper function _update_netbox_device_field() wraps all NetBox updates
- Per-URL semaphore with MD5 hash-based key generation
- Automatic cleanup of expired semaphore holders (60s TTL)

AI-assisted: Claude Code

Signed-off-by: Christian Berendt <berendt@osism.tech>

Author: berendt

osism/settings.py

@@ -59,3 +59,6 @@ def read_secret(secret_name):
 
 # Redfish connection timeout in seconds
 REDFISH_TIMEOUT = int(os.getenv("REDFISH_TIMEOUT", "20"))
+
+# NetBox connection limiting
+NETBOX_MAX_CONNECTIONS = int(os.getenv("NETBOX_MAX_CONNECTIONS", "5"))

osism/tasks/netbox.py

@@ -15,6 +15,28 @@ def setup_periodic_tasks(sender, **kwargs):
     pass
 
 
+def _update_netbox_device_field(nb, device_name, field_name, value):
+    """Helper to update a NetBox device field with semaphore limiting.
+
+    Args:
+        nb: NetBox API instance
+        device_name: Name of the device
+        field_name: Custom field name to update
+        value: Value to set
+
+    Returns:
+        bool: True if successful, False otherwise
+    """
+    semaphore = utils.create_netbox_semaphore(nb.base_url)
+    with semaphore:
+        device = nb.dcim.devices.get(name=device_name)
+        if device:
+            device.custom_fields.update({field_name: value})
+            device.save()
+            return True
+        return False
+
+
 def _matches_netbox_filter(nb, netbox_filter, is_primary=False):
     """Check if a NetBox instance matches the given filter.
 
@@ -84,21 +106,19 @@ def set_maintenance(
     utils.check_task_lock_and_exit()
 
     lock = utils.create_redlock(
-        key=f"lock_osism_tasks_netbox_set_maintenance_{device_name}",
-        auto_release_time=60,
+        key=f"lock_osism_tasks_netbox_{device_name}",
+        auto_release_time=300,
     )
-    if lock.acquire(timeout=20):
+    if lock.acquire(timeout=120):
         try:
             # Process primary NetBox
             if _matches_netbox_filter(utils.nb, netbox_filter, is_primary=True):
                 logger.info(
                     f"Set maintenance state of device {device_name} = {state} on {utils.nb.base_url}"
                 )
-                device = utils.nb.dcim.devices.get(name=device_name)
-                if device:
-                    device.custom_fields.update({"maintenance": state})
-                    device.save()
-                else:
+                if not _update_netbox_device_field(
+                    utils.nb, device_name, "maintenance", state
+                ):
                     logger.error(
                         f"Could not set maintenance for {device_name} on {utils.nb.base_url}"
                     )
@@ -123,11 +143,9 @@ def set_maintenance(
                 logger.info(
                     f"Set maintenance state of device {device_name} = {state} on {nb.base_url}"
                 )
-                device = nb.dcim.devices.get(name=device_name)
-                if device:
-                    device.custom_fields.update({"maintenance": state})
-                    device.save()
-                else:
+                if not _update_netbox_device_field(
+                    nb, device_name, "maintenance", state
+                ):
                     logger.error(
                         f"Could not set maintenance for {device_name} on {nb.base_url}"
                     )
@@ -156,21 +174,19 @@ def set_provision_state(
     utils.check_task_lock_and_exit()
 
     lock = utils.create_redlock(
-        key=f"lock_osism_tasks_netbox_set_provision_state_{device_name}",
-        auto_release_time=60,
+        key=f"lock_osism_tasks_netbox_{device_name}",
+        auto_release_time=300,
     )
-    if lock.acquire(timeout=20):
+    if lock.acquire(timeout=120):
         try:
             # Process primary NetBox
             if _matches_netbox_filter(utils.nb, netbox_filter, is_primary=True):
                 logger.info(
                     f"Set provision state of device {device_name} = {state} on {utils.nb.base_url}"
                 )
-                device = utils.nb.dcim.devices.get(name=device_name)
-                if device:
-                    device.custom_fields.update({"provision_state": state})
-                    device.save()
-                else:
+                if not _update_netbox_device_field(
+                    utils.nb, device_name, "provision_state", state
+                ):
                     logger.error(
                         f"Could not set provision state for {device_name} on {utils.nb.base_url}"
                     )
@@ -195,11 +211,9 @@ def set_provision_state(
                 logger.info(
                     f"Set provision state of device {device_name} = {state} on {nb.base_url}"
                 )
-                device = nb.dcim.devices.get(name=device_name)
-                if device:
-                    device.custom_fields.update({"provision_state": state})
-                    device.save()
-                else:
+                if not _update_netbox_device_field(
+                    nb, device_name, "provision_state", state
+                ):
                     logger.error(
                         f"Could not set provision state for {device_name} on {nb.base_url}"
                     )
@@ -232,21 +246,19 @@ def set_power_state(
     utils.check_task_lock_and_exit()
 
     lock = utils.create_redlock(
-        key=f"lock_osism_tasks_netbox_set_provision_state_{device_name}",
-        auto_release_time=60,
+        key=f"lock_osism_tasks_netbox_{device_name}",
+        auto_release_time=300,
     )
-    if lock.acquire(timeout=20):
+    if lock.acquire(timeout=120):
         try:
             # Process primary NetBox
             if _matches_netbox_filter(utils.nb, netbox_filter, is_primary=True):
                 logger.info(
                     f"Set power state of device {device_name} = {state} on {utils.nb.base_url}"
                 )
-                device = utils.nb.dcim.devices.get(name=device_name)
-                if device:
-                    device.custom_fields.update({"power_state": state})
-                    device.save()
-                else:
+                if not _update_netbox_device_field(
+                    utils.nb, device_name, "power_state", state
+                ):
                     logger.error(
                         f"Could not set power state for {device_name} on {utils.nb.base_url}"
                     )
@@ -271,11 +283,9 @@ def set_power_state(
                 logger.info(
                     f"Set power state of device {device_name} = {state} on {nb.base_url}"
                 )
-                device = nb.dcim.devices.get(name=device_name)
-                if device:
-                    device.custom_fields.update({"power_state": state})
-                    device.save()
-                else:
+                if not _update_netbox_device_field(
+                    nb, device_name, "power_state", state
+                ):
                     logger.error(
                         f"Could not set power state for {device_name} on {nb.base_url}"
                     )

osism/utils/__init__.py

@@ -1,6 +1,7 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import time
+import uuid
 import os
 from contextlib import redirect_stdout, redirect_stderr
 from cryptography.fernet import Fernet
@@ -18,6 +19,74 @@
 from osism import settings
 
 
+class RedisSemaphore:
+    """Redis-based distributed semaphore for limiting concurrent operations.
+
+    This implementation uses Redis sorted sets to track active holders and enforce
+    a maximum concurrency limit.
+    """
+
+    def __init__(self, redis_client, key, maxsize, timeout=None):
+        """Initialize the semaphore.
+
+        Args:
+            redis_client: Redis client instance
+            key: Redis key for this semaphore
+            maxsize: Maximum number of concurrent holders
+            timeout: Optional timeout for acquisition in seconds
+        """
+        self.redis = redis_client
+        self.key = f"semaphore:{key}"
+        self.maxsize = maxsize
+        self.timeout = timeout
+        self.identifier = None
+
+    def acquire(self, timeout=None):
+        """Acquire the semaphore.
+
+        Args:
+            timeout: Optional timeout in seconds (overrides instance timeout)
+
+        Returns:
+            bool: True if acquired, False if timeout
+        """
+        timeout = timeout or self.timeout or 10
+        identifier = str(uuid.uuid4())
+        now = time.time()
+        end_time = now + timeout
+
+        while time.time() < end_time:
+            # Clean up expired holders
+            self.redis.zremrangebyscore(self.key, 0, now - 60)
+
+            # Try to acquire
+            if self.redis.zcard(self.key) < self.maxsize:
+                self.redis.zadd(self.key, {identifier: now})
+                self.identifier = identifier
+                return True
+
+            time.sleep(0.01)  # Wait 10ms before retry
+
+        return False
+
+    def release(self):
+        """Release the semaphore."""
+        if self.identifier:
+            self.redis.zrem(self.key, self.identifier)
+            self.identifier = None
+
+    def __enter__(self):
+        """Context manager entry."""
+        if not self.acquire():
+            raise TimeoutError(f"Could not acquire semaphore {self.key}")
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        """Context manager exit."""
+        self.release()
+        return False
+
+
 class TimeoutHTTPAdapter(HTTPAdapter):
     """HTTPAdapter that sets a default timeout for all requests."""
 
@@ -291,6 +360,34 @@ def create_redlock(key, auto_release_time=3600):
             )
 
 
+def create_netbox_semaphore(netbox_url, max_connections=None):
+    """
+    Create a Redis semaphore for limiting concurrent NetBox connections.
+
+    Args:
+        netbox_url (str): The NetBox URL to create a semaphore for
+        max_connections (int): Maximum concurrent connections (default: from settings.NETBOX_MAX_CONNECTIONS)
+
+    Returns:
+        RedisSemaphore: The configured semaphore instance
+    """
+    if max_connections is None:
+        max_connections = settings.NETBOX_MAX_CONNECTIONS
+
+    # Create unique key per NetBox instance based on URL
+    import hashlib
+
+    url_hash = hashlib.md5(netbox_url.encode()).hexdigest()[:8]
+    key = f"netbox_semaphore_{url_hash}"
+
+    return RedisSemaphore(
+        key=key,
+        maxsize=max_connections,
+        redis_client=redis,
+        timeout=30,
+    )
+
+
 def set_task_lock(user=None, reason=None):
     """
     Set task lock to prevent new tasks from starting.