test loadbalancer with ingress and gateway

tests/integration/tests/conftest.py

@@ -216,7 +216,19 @@ def session_instance(
         bootstrap_config_path,
     )
 
+    instance_default_ip = util.get_default_ip(instance)
+
     instance.exec(["k8s", "bootstrap", "--file", bootstrap_config_path])
+    instance_default_cidr = util.get_default_cidr(instance, instance_default_ip)
+
+    lb_cidr = util.find_suitable_cidr(
+        parent_cidr=instance_default_cidr,
+        excluded_ips=[instance_default_ip],
+    )
+
+    instance.exec(
+        ["k8s", "set", f"load-balancer.cidrs={lb_cidr}", "load-balancer.l2-mode=true"]
+    )
     util.wait_until_k8s_ready(instance, [instance])
     util.wait_for_network(instance)
     util.wait_for_dns(instance)

tests/integration/tests/test_gateway.py

@@ -73,8 +73,17 @@ def test_gateway(session_instance: harness.Instance):
     )
     gateway_http_port = get_gateway_service_node_port(result)
 
-    assert gateway_http_port is not None, "No ingress nodePort found."
+    assert gateway_http_port is not None, "No Gateway nodePort found."
 
+    # Test the Gateway service via loadbalancer IP.
     util.stubbornly(retries=5, delay_s=5).on(session_instance).until(
         lambda p: "Welcome to nginx!" in p.stdout.decode()
     ).exec(["curl", f"localhost:{gateway_http_port}"])
+
+    gateway_ip = util.get_external_service_ip(
+        session_instance, ["ck-ingress-contour-envoy", "cilium-ingress"]
+    )
+    assert gateway_ip is not None, "No Gateway IP found."
+    util.stubbornly(retries=5, delay_s=5).on(session_instance).until(
+        lambda p: "Welcome to nginx!" in p.stdout.decode()
+    ).exec(["curl", f"{gateway_ip}", "-H", "Host: foo.bar.com"])

tests/integration/tests/test_ingress.py

@@ -3,6 +3,8 @@
 #
 import json
 import logging
+import subprocess
+import time
 from pathlib import Path
 from typing import List
 
@@ -35,6 +37,37 @@ def get_ingress_service_node_port(p):
     return None
 
 
+def get_external_service_ip(instance: harness.Instance, svcs) -> str:
+    try_count = 0
+    ingress_ip = None
+    while ingress_ip is None and try_count < 5:
+        try_count += 1
+        for svc in svcs:
+            try:
+                ingress_ip = (
+                    instance.exec(
+                        [
+                            "k8s",
+                            "kubectl",
+                            "--namespace",
+                            "kube-system",
+                            "get",
+                            "service",
+                            svc,
+                            "-o=jsonpath='{.status.loadBalancer.ingress[0].ip}'",
+                        ],
+                        capture_output=True,
+                    )
+                    .stdout.decode()
+                    .replace("'", "")
+                )
+            except subprocess.CalledProcessError:
+                ingress_ip = None
+                pass
+        time.sleep(3)
+    return ingress_ip
+
+
 def test_ingress(session_instance: List[harness.Instance]):
 
     result = (
@@ -77,3 +110,12 @@ def test_ingress(session_instance: List[harness.Instance]):
     util.stubbornly(retries=5, delay_s=5).on(session_instance).until(
         lambda p: "Welcome to nginx!" in p.stdout.decode()
     ).exec(["curl", f"localhost:{ingress_http_port}", "-H", "Host: foo.bar.com"])
+
+    # Test the ingress service via loadbalancer IP
+    ingress_ip = get_external_service_ip(
+        session_instance, ["ck-ingress-contour-envoy", "cilium-ingress"]
+    )
+    assert ingress_ip is not None, "No ingress IP found."
+    util.stubbornly(retries=5, delay_s=5).on(session_instance).until(
+        lambda p: "Welcome to nginx!" in p.stdout.decode()
+    ).exec(["curl", f"{ingress_ip}", "-H", "Host: foo.bar.com"])

tests/integration/tests/test_loadbalancer.py

@@ -1,10 +1,7 @@
 #
 # Copyright 2024 Canonical, Ltd.
 #
-import ipaddress
 import logging
-import subprocess
-import time
 from pathlib import Path
 from typing import List
 
@@ -15,29 +12,6 @@
 LOG = logging.getLogger(__name__)
 
 
-def find_suitable_cidr(parent_cidr: str, excluded_ips: List[str]):
-    net = ipaddress.IPv4Network(parent_cidr, False)
-
-    # Starting from the first IP address from the parent cidr,
-    # we search for a /30 cidr block(4 total ips, 2 available)
-    # that doesn't contain the excluded ips to avoid collisions
-    # /30 because this is the smallest CIDR cilium hands out IPs from
-    for i in range(4, 255, 4):
-        lb_net = ipaddress.IPv4Network(f"{str(net[0]+i)}/30", False)
-
-        contains_excluded = False
-        for excluded in excluded_ips:
-            if ipaddress.ip_address(excluded) in lb_net:
-                contains_excluded = True
-                break
-
-        if contains_excluded:
-            continue
-
-        return str(lb_net)
-    raise RuntimeError("Could not find a suitable CIDR for LoadBalancer services")
-
-
 @pytest.mark.node_count(2)
 def test_loadbalancer(instances: List[harness.Instance]):
     instance = instances[0]
@@ -49,7 +23,7 @@ def test_loadbalancer(instances: List[harness.Instance]):
 
     instance_default_cidr = util.get_default_cidr(instance, instance_default_ip)
 
-    lb_cidr = find_suitable_cidr(
+    lb_cidr = util.find_suitable_cidr(
         parent_cidr=instance_default_cidr,
         excluded_ips=[instance_default_ip, tester_instance_default_ip],
     )
@@ -115,56 +89,3 @@ def test_loadbalancer(instances: List[harness.Instance]):
     )
 
     assert "Welcome to nginx!" in p.stdout.decode()
-
-    # Try to access the service via Ingress
-    instance.exec(["k8s", "enable", "ingress"])
-    instance.exec(
-        ["k8s", "kubectl", "apply", "-f", "-"],
-        input=Path(MANIFESTS_DIR / "ingress-test.yaml").read_bytes(),
-    )
-    instance.exec(
-        [
-            "k8s",
-            "kubectl",
-            "wait",
-            "--for=condition=ready",
-            "pod",
-            "-l",
-            "run=my-nginx",
-            "--timeout",
-            "180s",
-        ]
-    )
-
-    try_count = 0
-    ingress_ip = None
-    while ingress_ip is None and try_count < 5:
-        try_count += 1
-        for svc in ["ck-ingress-contour-envoy", "cilium-ingress"]:
-            try:
-                ingress_ip = (
-                    instance.exec(
-                        [
-                            "k8s",
-                            "kubectl",
-                            "--namespace",
-                            "kube-system",
-                            "get",
-                            "service",
-                            svc,
-                            "-o=jsonpath='{.status.loadBalancer.ingress[0].ip}'",
-                        ],
-                        capture_output=True,
-                    )
-                    .stdout.decode()
-                    .replace("'", "")
-                )
-            except subprocess.CalledProcessError:
-                ingress_ip = None
-                pass
-        time.sleep(3)
-
-    assert ingress_ip is not None, "No ingress IP found."
-    util.stubbornly(retries=5, delay_s=5).on(tester_instance).until(
-        lambda p: "Welcome to nginx!" in p.stdout.decode()
-    ).exec(["curl", f"{ingress_ip}", "-H", "Host: foo.bar.com"])

tests/integration/tests/test_util/util.py

@@ -1,6 +1,7 @@
 #
 # Copyright 2024 Canonical, Ltd.
 #
+import ipaddress
 import json
 import logging
 import re
@@ -217,7 +218,7 @@ def wait_for_dns(instance: harness.Instance):
 
 def wait_for_network(instance: harness.Instance):
     LOG.info("Waiting for network to be ready")
-    instance.exec(["k8s", "x-wait-for", "network"])
+    instance.exec(["k8s", "x-wait-for", "network", "--timeout", "10m"])
 
 
 def hostname(instance: harness.Instance) -> str:
@@ -428,3 +429,27 @@ def _maj_min(version: str):
     track = f"{maj_min[0]}.{maj_min[1]}" + (flavor_track and f"-{flavor_track}")
     LOG.info("Previous track for %s is from track: %s", snap_version, track)
     return track
+
+
+def find_suitable_cidr(parent_cidr: str, excluded_ips: List[str]):
+    """Find a suitable CIDR for LoadBalancer services"""
+    net = ipaddress.IPv4Network(parent_cidr, False)
+
+    # Starting from the first IP address from the parent cidr,
+    # we search for a /30 cidr block(4 total ips, 2 available)
+    # that doesn't contain the excluded ips to avoid collisions
+    # /30 because this is the smallest CIDR cilium hands out IPs from
+    for i in range(4, 255, 4):
+        lb_net = ipaddress.IPv4Network(f"{str(net[0]+i)}/30", False)
+
+        contains_excluded = False
+        for excluded in excluded_ips:
+            if ipaddress.ip_address(excluded) in lb_net:
+                contains_excluded = True
+                break
+
+        if contains_excluded:
+            continue
+
+        return str(lb_net)
+    raise RuntimeError("Could not find a suitable CIDR for LoadBalancer services")