fix: bug

Makefile

@@ -7,6 +7,9 @@ build:
 push:
 	docker push code.whiteblossom.xyz/infra/node-external-ip-controller:latest
 
+build-n-push:
+	docker buildx build --platform=linux/amd64,linux/arm64 -t code.whiteblossom.xyz/infra/node-external-ip-controller:latest --push -f node-external-ip-controller.Dockerfile . --push
+
 builder:
 	docker buildx create \
   --name container-builder \

node-external-ip-controller.Dockerfile

@@ -2,7 +2,7 @@
 FROM python:3.13-alpine3.19
 
 # Install the Kubernetes Python client
-RUN pip install kubernetes kubernetes_asyncio
+RUN pip install kubernetes kubernetes_asyncio PyYAML
 
 # Copy the controller script into the container
 COPY node_external_ip_controller_async.py /app/node_external_ip_controller.py

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

@@ -1,128 +1,163 @@
 import asyncio
 from kubernetes_asyncio import client, config, watch
+import os
+import logging
+import ipaddress
 
 # Configuration
-SERVICE_NAME_PATTERN = (
-    "traefik"  # Define the service name pattern or label to identify services
-)
-NAMESPACE = "kube-system"
-ANNOTATION_KEY = "kube-vip.io/loadbalancerIPs"
-ZERO_GATEWAY_IP = "172.28.10.1"
-NODE_LABEL = "svccontroller.k3s.cattle.io/enablelb=true"
+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 = 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))
+
+# Logging configuration
+logging.basicConfig(level=logging.DEBUG)
+logger = logging.getLogger(__name__)
 
 
-async def update_service_annotation(v1, service_name, external_ip):
+def is_valid_ip(ip):
     try:
-        # Get the current service object
-        service = await v1.read_namespaced_service(service_name, NAMESPACE)
+        ipaddress.ip_address(ip)
+        return True
+    except ValueError:
+        return False
 
-        # Check if the annotation needs to be updated
-        current_annotation = service.metadata.annotations.get(ANNOTATION_KEY)
-        target_annotation = ",".join({external_ip, ZERO_GATEWAY_IP})
-        if current_annotation != target_annotation:
-            # Update the annotation
+
+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, "")
+        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}}}
-            await v1.patch_namespaced_service(service_name, NAMESPACE, body)
-            print(
+            logger.debug(f"Patching service {service_name} with body: {body}")
+            await v1.patch_namespaced_service(service_name, namespace, body)
+            logger.info(
                 f"Updated service {service_name} with new external IP: {target_annotation}"
             )
+        else:
+            logger.debug(f"No update required for service {service_name}")
 
     except client.exceptions.ApiException as e:
-        print(f"API Exception in update_service_annotation: {e}")
+        logger.error(f"API Exception in update_service_annotation: {e}")
 
 
-async def watch_nodes():
-    await config.load_incluster_config()
-    v1 = client.CoreV1Api()
+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")
             async for event in w.stream(
-                v1.list_node, label_selector=NODE_LABEL, _request_timeout=300
+                v1.list_node,
+                label_selector=NODE_LABEL,
+                _request_timeout=NODE_REQUEST_TIMEOUT,
             ):
                 node = event["object"]
-                node_name = node.metadata.name
-
-                # Check for external IP
-                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}")
-
-                    # Get all services that need to be updated with this external IP
-                    services = await v1.list_namespaced_service(
-                        NAMESPACE, label_selector=SERVICE_NAME_PATTERN
+                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}"
                     )
-                    for service in services.items:
-                        await update_service_annotation(
-                            v1, service.metadata.name, external_ip
-                        )
+                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:
-            print(f"API Exception in watch_nodes: {e}")
+            logger.error(f"API Exception in watch_nodes: {e}")
             await asyncio.sleep(5)
-
         except asyncio.CancelledError:
-            print("Watch task was cancelled.")
+            logger.info("Watch task was cancelled.")
             break
-
         except Exception as e:
-            print(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():
-    await config.load_incluster_config()
-    v1 = client.CoreV1Api()
+async def watch_services(v1, external_ips_update_queue):
     w = watch.Watch()
 
     while True:
         try:
+            logger.debug("Starting to watch services")
             async for event in w.stream(
-                v1.list_namespaced_service,
-                NAMESPACE,
-                label_selector=SERVICE_NAME_PATTERN,
-                _request_timeout=300,
+                v1.list_service_for_all_namespaces,
+                label_selector=SERVICE_LABEL,
+                _request_timeout=SERVICE_REQUEST_TIMEOUT,
             ):
-                if event["type"] == "ADDED":
-                    service = event["object"]
-                    service_name = service.metadata.name
-                    print(f"New service detected: {service_name}")
-
-                    # Fetch current external IP for nodes that match the label
-                    nodes = await v1.list_node(label_selector=NODE_LABEL)
-                    for node in nodes.items:
-                        for address in node.status.addresses:
-                            if address.type == "ExternalIP":
-                                external_ip = address.address
-                                await update_service_annotation(
-                                    v1, service_name, external_ip
-                                )
-                                break
+                service = event["object"]
+                if event["type"] in {"ADDED", "MODIFIED"}:
+                    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:
-            print(f"API Exception in watch_services: {e}")
+            logger.error(f"API Exception in watch_services: {e}")
             await asyncio.sleep(5)
-
         except asyncio.CancelledError:
-            print("Watch task was cancelled.")
+            logger.info("Watch task was cancelled.")
             break
-
         except Exception as e:
-            print(f"Unexpected error in watch_services: {e}")
+            logger.error(f"Unexpected error in watch_services: {e}", exc_info=True)
             await asyncio.sleep(5)
 
 
 async def main():
-    await config.load_incluster_config()
+    config.load_incluster_config()
+    v1 = client.CoreV1Api()
 
-    # Run watch_nodes and watch_services concurrently
-    await asyncio.gather(watch_nodes(), watch_services())
+    external_ips_update_queue = asyncio.Queue()
+    await asyncio.gather(
+        watch_nodes(v1, external_ips_update_queue),
+        watch_services(v1, external_ips_update_queue),
+    )
 
 
 if __name__ == "__main__":