fix dependency

node-external-ip-controller.Dockerfile

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

node_external_ip_controller.py

@@ -21,11 +21,22 @@ def update_service_annotation(external_ip):
 
     # Check if the annotation needs to be updated
     current_annotation = service.metadata.annotations.get(ANNOTATION_KEY)
-    if current_annotation != external_ip:
+    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: external_ip}}}
+        body = {"metadata": {"annotations": {ANNOTATION_KEY: target_annotation}}}
         v1.patch_namespaced_service(SERVICE_NAME, NAMESPACE, body)
-        print(f"Updated service {SERVICE_NAME} with new external IP: {external_ip}")
+        print(
+            f"Updated service {SERVICE_NAME} with new external IPs: {target_annotation}",
+            flush=True,
+        )
 
 
 def main():
@@ -46,15 +57,18 @@ def main():
                         break
 
                 if external_ip:
-                    print(f"Detected external IP {external_ip} for node {node_name}")
+                    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}")
+            print(f"API Exception: {e}", flush=True)
             time.sleep(5)  # Wait before retrying
 
         except Exception as e:
-            print(f"Unexpected error: {e}")
+            print(f"Unexpected error: {e}", flush=True)
             time.sleep(5)
 
 

node_external_ip_controller_async.py

@@ -0,0 +1,129 @@
+import asyncio
+from kubernetes_asyncio import client, config, watch
+
+# 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"
+
+
+async def update_service_annotation(v1, service_name, external_ip):
+    try:
+        # Get the current service object
+        service = await v1.read_namespaced_service(service_name, NAMESPACE)
+
+        # 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
+            body = {"metadata": {"annotations": {ANNOTATION_KEY: target_annotation}}}
+            await v1.patch_namespaced_service(service_name, NAMESPACE, body)
+            print(
+                f"Updated service {service_name} with new external IP: {target_annotation}"
+            )
+
+    except client.exceptions.ApiException as e:
+        print(f"API Exception in update_service_annotation: {e}")
+
+
+async def watch_nodes():
+    await config.load_incluster_config()
+    v1 = client.CoreV1Api()
+    w = watch.Watch()
+
+    while True:
+        try:
+            async for event in w.stream(
+                v1.list_node, label_selector=NODE_LABEL, _request_timeout=300
+            ):
+                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
+                    )
+                    for service in services.items:
+                        await update_service_annotation(
+                            v1, service.metadata.name, external_ip
+                        )
+
+        except client.exceptions.ApiException as e:
+            print(f"API Exception in watch_nodes: {e}")
+            await asyncio.sleep(5)
+
+        except asyncio.CancelledError:
+            print("Watch task was cancelled.")
+            break
+
+        except Exception as e:
+            print(f"Unexpected error in watch_nodes: {e}")
+            await asyncio.sleep(5)
+
+
+async def watch_services():
+    await config.load_incluster_config()
+    v1 = client.CoreV1Api()
+    w = watch.Watch()
+
+    while True:
+        try:
+            async for event in w.stream(
+                v1.list_namespaced_service,
+                NAMESPACE,
+                label_selector=SERVICE_NAME_PATTERN,
+                _request_timeout=300,
+            ):
+                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
+
+        except client.exceptions.ApiException as e:
+            print(f"API Exception in watch_services: {e}")
+            await asyncio.sleep(5)
+
+        except asyncio.CancelledError:
+            print("Watch task was cancelled.")
+            break
+
+        except Exception as e:
+            print(f"Unexpected error in watch_services: {e}")
+            await asyncio.sleep(5)
+
+
+async def main():
+    await config.load_incluster_config()
+
+    # Run watch_nodes and watch_services concurrently
+    await asyncio.gather(watch_nodes(), watch_services())
+
+
+if __name__ == "__main__":
+    asyncio.run(main())