feat: add periodic reconciliation to detect and fix service drift

Add full reconciliation every 6 hours to compare known LoadBalancer
services against actual cluster state. Track known services in memory
and trigger webhooks for any detected drift. Include API connectivity
verification before watch cycles to improve reliability.

Author: elasticroentgen

service_monitor.py

@@ -41,12 +41,17 @@
 last_event_time = time.time()
 HEALTH_CHECK_INTERVAL = 300  # 5 minutes in seconds
 MAX_WATCH_TIME = 3600  # 1 hour in seconds - force reconnect after this time
+RECONCILE_INTERVAL = 21600  # 6 hours in seconds - full reconciliation interval
 
 # Reconnection settings
 INITIAL_RETRY_DELAY = 5  # seconds
 MAX_RETRY_DELAY = 300  # 5 minutes in seconds
 RETRY_BACKOFF_FACTOR = 1.5
 
+# Track known LB services for reconciliation
+known_lb_services: set = set()
+last_reconcile_time = 0.0
+
 # Running flag for graceful shutdown
 running = True
 
@@ -101,16 +106,79 @@ def get_initial_resource_version(v1) -> Optional[str]:
     """
     Get current resourceVersion without triggering workflow.
     This ensures we start watching from the current state.
+    Also builds the initial snapshot of known LB services.
     """
+    global known_lb_services, last_reconcile_time
     try:
         services = v1.list_service_for_all_namespaces()
-        lb_count = sum(1 for s in services.items if s.spec.type == 'LoadBalancer')
-        logger.info(f"Found {lb_count} existing LoadBalancer services")
+        known_lb_services = {
+            f"{s.metadata.namespace}/{s.metadata.name}"
+            for s in services.items if s.spec.type == 'LoadBalancer'
+        }
+        last_reconcile_time = time.time()
+        logger.info(f"Found {len(known_lb_services)} existing LoadBalancer services")
         return services.metadata.resource_version
     except Exception as e:
         logger.error(f"Failed to get initial resourceVersion: {e}")
         return None
 
+
+def verify_api_connection(v1) -> bool:
+    """Lightweight API server connectivity check."""
+    try:
+        v1.read_namespace(name='default', _request_timeout=10)
+        return True
+    except Exception as e:
+        logger.warning(f"API server connectivity check failed: {e}")
+        return False
+
+
+def reconcile_services(v1) -> bool:
+    """
+    Full reconciliation: list all LB services and compare against known state.
+    If the set of services changed, trigger the webhook to ensure consistency.
+    Returns True if reconciliation succeeded.
+    """
+    global known_lb_services, last_reconcile_time
+    try:
+        services = v1.list_service_for_all_namespaces()
+        current_lb_services = {
+            f"{s.metadata.namespace}/{s.metadata.name}"
+            for s in services.items if s.spec.type == 'LoadBalancer'
+        }
+
+        added = current_lb_services - known_lb_services
+        removed = known_lb_services - current_lb_services
+
+        if added or removed:
+            logger.warning(
+                f"Reconciliation detected drift: "
+                f"{len(added)} added, {len(removed)} removed"
+            )
+            for svc in added:
+                logger.info(f"Reconciliation: new service {svc}")
+                trigger_semaphore_webhook('RECONCILE_ADDED', svc)
+            for svc in removed:
+                logger.info(f"Reconciliation: removed service {svc}")
+                trigger_semaphore_webhook('RECONCILE_DELETED', svc)
+        else:
+            logger.info(
+                f"Reconciliation OK: {len(current_lb_services)} LB services, no drift"
+            )
+
+        known_lb_services = current_lb_services
+        last_reconcile_time = time.time()
+
+        # Also update resourceVersion from the list response
+        rv = services.metadata.resource_version
+        if rv:
+            set_resource_version(rv)
+
+        return True
+    except Exception as e:
+        logger.error(f"Reconciliation failed: {e}")
+        return False
+
 def send_mattermost_notification(event_type: str, namespace: str, service_name: str) -> bool:
     """
     Send a notification to Mattermost about the triggered Semaphore webhook.
@@ -254,8 +322,7 @@ def watch_services(timeout_seconds: Optional[int] = None):
     resource_version = get_resource_version()
     logger.info(f"Starting to watch LoadBalancer services with timeout: {timeout_seconds or 'None'} seconds, resourceVersion: {resource_version or 'None'}...")
     start_time = time.time()
-    update_last_event_time()  # Initialize the last event time
-    touch_health_file()  # Initialize health file
+    touch_health_file()
 
     try:
         # Use resourceVersion for resumable watches and enable bookmarks
@@ -308,10 +375,16 @@ def watch_services(timeout_seconds: Optional[int] = None):
             # Only process LoadBalancer services
             if service_type == 'LoadBalancer':
                 logger.info(f"LoadBalancer service event: {event_type} - {namespace}/{name}")
+                service_key = f"{namespace}/{name}"
+
+                # Keep known_lb_services in sync
+                if event_type in ['ADDED', 'MODIFIED']:
+                    known_lb_services.add(service_key)
+                elif event_type == 'DELETED':
+                    known_lb_services.discard(service_key)
 
                 # Trigger webhook for relevant events
                 if event_type in ['ADDED', 'MODIFIED', 'DELETED']:
-                    service_key = f"{namespace}/{name}"
                     trigger_semaphore_webhook(event_type, service_key)
 
             # Check if we've been watching for too long and should restart
@@ -389,37 +462,53 @@ def main():
 
     while running:
         try:
+            # Verify API server connectivity before starting a watch cycle
+            probe_v1 = initialize_kubernetes_client()
+            if not verify_api_connection(probe_v1):
+                logger.warning("API server unreachable, will retry...")
+                attempt += 1
+                delay = exponential_backoff(attempt)
+                start_wait = time.time()
+                while running and time.time() - start_wait < delay:
+                    time.sleep(1)
+                continue
+
+            # Run periodic full reconciliation
+            if time.time() - last_reconcile_time > RECONCILE_INTERVAL:
+                logger.info("Starting periodic reconciliation...")
+                reconcile_services(probe_v1)
+
             # Use MAX_WATCH_TIME for longer watch periods (1 hour)
             watch_services(timeout_seconds=MAX_WATCH_TIME)
-            
+
             # Reset attempt counter on successful completion
             attempt = 0
-            
+
         except Exception as e:
             logger.error(f"Error in watch_services: {str(e)}")
             attempt += 1
             delay = exponential_backoff(attempt)
             logger.info(f"Retrying in {delay:.1f} seconds (attempt {attempt})...")
-            
+
             # Wait with interruption support
             start_wait = time.time()
             while running and time.time() - start_wait < delay:
                 time.sleep(1)
-        
+
         # Perform periodic health check
         if time.time() - last_health_check > HEALTH_CHECK_INTERVAL:
             healthy = perform_health_check()
             last_health_check = time.time()
-            
+
             # If not healthy, force a reconnection
             if not healthy:
                 logger.warning("Forcing reconnection due to unhealthy state")
                 continue
-        
+
         # Small delay to prevent tight loop if watch_services returns immediately
         if running:
             time.sleep(1)
-    
+
     logger.info("Service watcher shutting down")
 
 if __name__ == "__main__":