diff --git a/gateway_services.yaml b/gateway_services.yaml deleted file mode 100644 index fd831eb..0000000 --- a/gateway_services.yaml +++ /dev/null @@ -1,4 +0,0 @@ -gateway_services: - - name: traefik-tcp - namespace: kube-system - label_pattern: app.kubernetes.io/name=traefik diff --git a/node-external-ip-controller.Dockerfile b/node-external-ip-controller.Dockerfile index 69f4d9d..8dc32ba 100644 --- a/node-external-ip-controller.Dockerfile +++ b/node-external-ip-controller.Dockerfile @@ -6,7 +6,6 @@ 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 -COPY gateway_services.yaml /app/gateway_services.yaml # Set the working directory WORKDIR /app diff --git a/node_external_ip_controller_async.py b/node_external_ip_controller_async.py index 51a78f9..c3e6048 100644 --- a/node_external_ip_controller_async.py +++ b/node_external_ip_controller_async.py @@ -3,32 +3,24 @@ from kubernetes_asyncio import client, config, watch import os # Configuration -import yaml - -GATEWAY_SERVICES_FILE = os.getenv("GATEWAY_SERVICES_FILE", "gateway_services.yaml") -with open(GATEWAY_SERVICES_FILE, "r") as f: - gateway_services = yaml.safe_load(f) - -services = gateway_services["gateway_services"] 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_REQUEST_TIMEOUT = int(os.getenv("SERVICE_REQUEST_TIMEOUT", 300)) NODE_REQUEST_TIMEOUT = int(os.getenv("NODE_REQUEST_TIMEOUT", 30)) async def update_service_annotation(v1, service, external_ips): try: - # Get the current service object - service_name = service["name"] - namespace = service["namespace"] + service_name = service.metadata.name + namespace = service.metadata.namespace service_obj = await v1.read_namespaced_service(service_name, namespace) - # Check if the annotation needs to be updated current_annotation = service_obj.metadata.annotations.get(ANNOTATION_KEY) target_annotation = ",".join(external_ips) + "," + 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( @@ -40,97 +32,71 @@ async def update_service_annotation(v1, service, external_ips): print(f"API Exception in update_service_annotation: {e}", flush=True) -async def watch_nodes(): - config.load_incluster_config() - v1 = client.CoreV1Api() +async def watch_nodes(v1, external_ips_update_queue): w = watch.Watch() - while True: try: - # Check for external IP - external_ips, node_names = [], [] async for event in w.stream( v1.list_node, label_selector=NODE_LABEL, _request_timeout=NODE_REQUEST_TIMEOUT, ): node = event["object"] - node_name = node.metadata.name - - for address in node.status.addresses: - if address.type == "ExternalIP": - external_ips.append(address.address) - node_names.append(node_name) - if len(external_ips) > 0: - print( - f"Detected external IPs {str(external_ips)} for node {str(node_names)}", - flush=True, - ) - for service in services: - await update_service_annotation(v1, service, external_ips) + external_ips = [ + addr.address + for addr in node.status.addresses + if addr.type == "ExternalIP" + ] + if external_ips: + await external_ips_update_queue.put(external_ips) except client.exceptions.ApiException as e: print(f"API Exception in watch_nodes: {e}", flush=True) await asyncio.sleep(5) - except asyncio.CancelledError: print("Watch task was cancelled.", flush=True) break - except Exception as e: print(f"Unexpected error in watch_nodes: {e}", flush=True) await asyncio.sleep(5) -async def watch_service(service): - config.load_incluster_config() - v1 = client.CoreV1Api() +async def watch_services(v1, external_ips_update_queue): w = watch.Watch() - service_name = service["name"] - namespace = service["namespace"] - label_selector = service["label_pattern"] + label_selector = f"{SERVICE_LABEL_KEY}={SERVICE_LABEL_VALUE}" while True: try: async for event in w.stream( - v1.list_namespaced_service, - namespace, + v1.list_service_for_all_namespaces, label_selector=label_selector, _request_timeout=SERVICE_REQUEST_TIMEOUT, ): - service_obj = event["object"] - if ( - event["type"] == "ADDED" - and service_obj.metadata.name == service_name - ): - print(f"New service detected: {service_name}", flush=True) - - # Fetch current external IP for nodes that match the label - nodes = await v1.list_node(label_selector=NODE_LABEL) - external_ips = [] - for node in nodes.items: - for address in node.status.addresses: - if address.type == "ExternalIP": - external_ips.append(address.address) + 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) except client.exceptions.ApiException as e: print(f"API Exception in watch_services: {e}", flush=True) await asyncio.sleep(5) - except asyncio.CancelledError: print("Watch task was cancelled.", flush=True) break - except Exception as e: print(f"Unexpected error in watch_services: {e}", flush=True) await asyncio.sleep(5) async def main(): - # Run watch_nodes and watch_services concurrently - service_watchers = [watch_service(service) for service in services] - await asyncio.gather(watch_nodes(), *service_watchers) + config.load_incluster_config() + v1 = client.CoreV1Api() + + 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__":