fix: bug

node_external_ip_controller_async.py

@@ -8,8 +8,8 @@ import ipaddress
 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 = os.getenv("SERVICE_LABEL", "enablezlan=true")
-SERVICE_LABEL_VALUE = os.getenv("SERVICE_LABEL_VALUE", "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))
 
@@ -26,28 +26,30 @@ def is_valid_ip(ip):
         return False
 
 
-async def update_service_annotation(v1, service, external_ips):
+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)
+        current_annotation = service_obj.metadata.annotations.get(ANNOTATION_KEY, "")
-        zlan_gateway_ip = service_obj.metadata.labels.get(SERVICE_LABEL_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):
-            target_annotation = ",".join(external_ips) + "," + zlan_gateway_ip
+            external_ipset = set(external_ipset)
+            external_ipset.add(zlan_gateway_ip)
         else:
-            target_annotation = ",".join(external_ips)
             logger.debug(
                 f"Invalid Zlan Gateway IP: {zlan_gateway_ip}, excluding from target annotation"
             )
-
+        external_ips = list(external_ipset)
-        logger.debug(f"Current annotation: {current_annotation}")
+        target_annotation = ",".join(external_ips)
-        logger.debug(f"Target annotation: {target_annotation}")
+        if annotated_ipset != external_ipset:
-        if current_annotation != target_annotation:
+            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)
@@ -95,7 +97,10 @@ async def watch_nodes(v1, external_ips_update_queue):
                         f"External IPs for node {node.metadata.name}: {external_ips}"
                     )
                 external_node_ips = list(external_node_ipset)
-                await external_ips_update_queue.put(external_node_ips)
+                # 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:
@@ -105,19 +110,19 @@ 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"]
@@ -125,9 +130,12 @@ async def watch_services(v1, external_ips_update_queue):
                     logger.debug(
                         f"Processing event type: {event['type']} for service: {service.metadata.name}"
                     )
-                    external_ips = await external_ips_update_queue.get()
+                    external_ipset = await external_ips_update_queue.get()
-                    logger.debug(f"Retrieved external IPs: {external_ips}")
+                    # put back the last one if it turns empty
-                    await update_service_annotation(v1, service, external_ips)
+                    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:
@@ -137,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)