fix: bug

node_external_ip_controller.py

@@ -1,76 +0,0 @@
-from kubernetes import client, config, watch
-import os
-import time
-
-# Load in-cluster config
-config.load_incluster_config()
-
-# Set up Kubernetes API client
-v1 = client.CoreV1Api()
-
-# Configuration
-SERVICE_NAME = "traefik"
-NAMESPACE = "kube-system"
-ANNOTATION_KEY = "kube-vip.io/loadbalancerIPs"
-NODE_LABEL = "svccontroller.k3s.cattle.io/enablelb=true"
-
-
-def update_service_annotation(external_ip):
-    # Get the current service object
-    service = v1.read_namespaced_service(SERVICE_NAME, NAMESPACE)
-
-    # Check if the annotation needs to be updated
-    current_annotation = service.metadata.annotations.get(ANNOTATION_KEY)
-    ipam_address = service.spec.load_balancer_ip
-    if ipam_address:
-        print(
-            f"service {SERVICE_NAME} has existing ipam IP: {ipam_address}", flush=True
-        )
-        target_annotation = ",".join({external_ip, ipam_address})
-    else:
-        target_annotation = external_ip
-    if current_annotation != target_annotation:
-        # Update the annotation
-        body = {"metadata": {"annotations": {ANNOTATION_KEY: target_annotation}}}
-        v1.patch_namespaced_service(SERVICE_NAME, NAMESPACE, body)
-        print(
-            f"Updated service {SERVICE_NAME} with new external IPs: {target_annotation}",
-            flush=True,
-        )
-
-
-def main():
-    w = watch.Watch()
-    while True:
-        try:
-            for event in w.stream(
-                v1.list_node, label_selector=NODE_LABEL, _request_timeout=300
-            ):
-                node = event["object"]
-                node_name = node.metadata.name
-
-                # Extract the external IP if it exists
-                external_ip = None
-                for address in node.status.addresses:
-                    if address.type == "ExternalIP":
-                        external_ip = address.address
-                        break
-
-                if external_ip:
-                    print(
-                        f"Detected external IP {external_ip} for node {node_name}",
-                        flush=True,
-                    )
-                    update_service_annotation(external_ip)
-
-        except client.exceptions.ApiException as e:
-            print(f"API Exception: {e}", flush=True)
-            time.sleep(5)  # Wait before retrying
-
-        except Exception as e:
-            print(f"Unexpected error: {e}", flush=True)
-            time.sleep(5)
-
-
-if __name__ == "__main__":
-    main()

node_external_ip_controller_async.py

@@ -2,13 +2,14 @@ import asyncio
 from kubernetes_asyncio import client, config, watch
 import os
 import logging
+import ipaddress
 
 # Configuration
 ANNOTATION_KEY = os.getenv("ANNOTATION_KEY", "kube-vip.io/loadbalancerIPs")
 ZERO_GATEWAY_IP = os.getenv("ZERO_GATEWAY_IP", "172.28.10.1")
 NODE_LABEL = os.getenv("NODE_LABEL", "svccontroller.k3s.cattle.io/enablelb=true")
-SERVICE_LABEL_KEY = os.getenv("SERVICE_LABEL_KEY", "zlanservice")
-SERVICE_LABEL_VALUE = os.getenv("SERVICE_LABEL_VALUE", "true")
+SERVICE_LABEL = os.getenv("SERVICE_LABEL", "enablezlan=true")
+ZLAN_GATEWAY_IP_KEY = os.getenv("ZLAN_GATEWAY_IP_KEY", "zlanip")
 SERVICE_REQUEST_TIMEOUT = int(os.getenv("SERVICE_REQUEST_TIMEOUT", 300))
 NODE_REQUEST_TIMEOUT = int(os.getenv("NODE_REQUEST_TIMEOUT", 30))
 
@@ -17,18 +18,38 @@ logging.basicConfig(level=logging.DEBUG)
 logger = logging.getLogger(__name__)
 
 
-async def update_service_annotation(v1, service, external_ips):
+def is_valid_ip(ip):
+    try:
+        ipaddress.ip_address(ip)
+        return True
+    except ValueError:
+        return False
+
+
+async def update_service_annotation(v1, service, external_ipset):
     try:
         service_name = service.metadata.name
         namespace = service.metadata.namespace
         logger.debug(f"Fetching service {service_name} in namespace {namespace}")
         service_obj = await v1.read_namespaced_service(service_name, namespace)
 
-        current_annotation = service_obj.metadata.annotations.get(ANNOTATION_KEY)
-        target_annotation = ",".join(external_ips) + "," + ZERO_GATEWAY_IP
-        logger.debug(f"Current annotation: {current_annotation}")
-        logger.debug(f"Target annotation: {target_annotation}")
-        if current_annotation != target_annotation:
+        current_annotation = service_obj.metadata.annotations.get(ANNOTATION_KEY, "")
+        annotated_ipset = set(current_annotation.split(","))
+        zlan_gateway_ip = service_obj.metadata.labels.get(ZLAN_GATEWAY_IP_KEY)
+        logger.debug(f"Zlan Gateway IP: {zlan_gateway_ip}")
+
+        if is_valid_ip(zlan_gateway_ip):
+            external_ipset = set(external_ipset)
+            external_ipset.add(zlan_gateway_ip)
+        else:
+            logger.debug(
+                f"Invalid Zlan Gateway IP: {zlan_gateway_ip}, excluding from target annotation"
+            )
+        external_ips = list(external_ipset)
+        target_annotation = ",".join(external_ips)
+        if annotated_ipset != external_ipset:
+            logger.debug(f"Current annotation: {current_annotation}: {annotated_ipset}")
+            logger.debug(f"Target annotation: {target_annotation}: {external_ipset}")
             body = {"metadata": {"annotations": {ANNOTATION_KEY: target_annotation}}}
             logger.debug(f"Patching service {service_name} with body: {body}")
             await v1.patch_namespaced_service(service_name, namespace, body)
@@ -44,6 +65,7 @@ async def update_service_annotation(v1, service, external_ips):
 
 async def watch_nodes(v1, external_ips_update_queue):
     w = watch.Watch()
+    external_node_ipset = set()
     while True:
         try:
             logger.debug("Starting to watch nodes")
@@ -53,17 +75,33 @@ async def watch_nodes(v1, external_ips_update_queue):
                 _request_timeout=NODE_REQUEST_TIMEOUT,
             ):
                 node = event["object"]
-                logger.debug(f"Received event for node: {node.metadata.name}")
+                event_type = event["type"]
+                logger.debug(
+                    f"Received {event_type} event for node: {node.metadata.name}"
+                )
                 external_ips = [
                     addr.address
                     for addr in node.status.addresses
                     if addr.type == "ExternalIP"
                 ]
+                if event_type in {"ADDED", "MODIFIED"}:
+                    external_node_ipset.update(external_ips)
                     logger.debug(
                         f"External IPs for node {node.metadata.name}: {external_ips}"
                     )
-                await external_ips_update_queue.put(external_ips)
-                logger.debug(f"Added external IPs to update queue: {external_ips}")
+                elif event_type == "REMOVED":
+                    for ip in external_ips:
+                        if ip in external_node_ipset:
+                            external_node_ipset.remove(ip)
+                    logger.debug(
+                        f"External IPs for node {node.metadata.name}: {external_ips}"
+                    )
+                external_node_ips = list(external_node_ipset)
+                # remove old ip lists before populating
+                while not external_ips_update_queue.empty():
+                    _previous_external_ips = await external_ips_update_queue.get()
+                await external_ips_update_queue.put(set(external_node_ips))
+                logger.debug(f"Added external IPs to update queue: {external_node_ips}")
 
         except client.exceptions.ApiException as e:
             logger.error(f"API Exception in watch_nodes: {e}")
@@ -72,25 +110,33 @@ async def watch_nodes(v1, external_ips_update_queue):
             logger.info("Watch task was cancelled.")
             break
         except Exception as e:
-            logger.error(f"Unexpected error in watch_nodes: {e}")
+            logger.error(f"Unexpected error in watch_nodes: {e}", exc_info=True)
             await asyncio.sleep(5)
 
 
 async def watch_services(v1, external_ips_update_queue):
     w = watch.Watch()
-    label_selector = f"{SERVICE_LABEL_KEY}={SERVICE_LABEL_VALUE}"
 
     while True:
         try:
+            logger.debug("Starting to watch services")
             async for event in w.stream(
                 v1.list_service_for_all_namespaces,
-                label_selector=label_selector,
+                label_selector=SERVICE_LABEL,
                 _request_timeout=SERVICE_REQUEST_TIMEOUT,
             ):
                 service = event["object"]
                 if event["type"] in {"ADDED", "MODIFIED"}:
-                    external_ips = await external_ips_update_queue.get()
-                    await update_service_annotation(v1, service, external_ips)
+                    logger.debug(
+                        f"Processing event type: {event['type']} for service: {service.metadata.name}"
+                    )
+                    external_ipset = await external_ips_update_queue.get()
+                    # put back the last one if it turns empty
+                    if external_ips_update_queue.empty():
+                        await external_ips_update_queue.put(external_ipset)
+                    logger.debug(f"Retrieved external IPs: {list(external_ipset)}")
+                    await update_service_annotation(v1, service, external_ipset)
+                    logger.debug(f"Service updated: {service.metadata.name}")
 
         except client.exceptions.ApiException as e:
             logger.error(f"API Exception in watch_services: {e}")
@@ -99,7 +145,7 @@ async def watch_services(v1, external_ips_update_queue):
             logger.info("Watch task was cancelled.")
             break
         except Exception as e:
-            logger.error(f"Unexpected error in watch_services: {e}")
+            logger.error(f"Unexpected error in watch_services: {e}", exc_info=True)
             await asyncio.sleep(5)