Merge pull request #549 from element-hq/bbz/large-numbers-of-sfu-udp-ports

Handle wide Matrix RTC SFU UDP port ranges

Author: benbz

charts/matrix-stack/templates/matrix-rtc/sfu_deployment.yaml

@@ -114,6 +114,8 @@ spec:
   {{- if .enabled }}
     {{- $portType := .type -}}
     {{- with .portRange }}
+    {{- /* container.port is metadata. Omit if a large range is provided so that we don't run into document size limits when submitting to the API server */ -}}
+    {{- if le (sub (.endPort | int) (.startPort | int)) 100 }}
     {{- range $port := untilStep (.startPort | int) (.endPort | int) 1 }}
         - containerPort: {{ $port }}
           name: rtc-udp-{{ $port }}
@@ -123,6 +125,7 @@ spec:
       {{- end }}
     {{- end }}
     {{- end }}
+    {{- end }}
   {{- end }}
   {{- end }}
 {{- end }}

charts/matrix-stack/templates/matrix-rtc/sfu_rtc_udp_range_service.yaml

@@ -1,5 +1,5 @@
 {{- /*
-Copyright 2024 New Vector Ltd
+Copyright 2024-2025 New Vector Ltd
 
 SPDX-License-Identifier: AGPL-3.0-only
 */ -}}
@@ -8,29 +8,36 @@ SPDX-License-Identifier: AGPL-3.0-only
 {{- if .enabled -}}
 {{- with .sfu -}}
 {{- if and .enabled .exposedServices.rtcUdp.enabled (eq .exposedServices.rtcUdp.portType "NodePort") -}}
+{{- $range_start := (.exposedServices.rtcUdp.portRange.startPort | int) -}}
+{{- $range_end := (.exposedServices.rtcUdp.portRange.endPort | int) -}}
+{{- $range_size := sub $range_end $range_start -}}
+{{- $number_of_ranges := ((add ((div $range_size 250) | floor | int) 1) | int) -}}
+{{- range $port_segment := until $number_of_ranges }}
+---
 apiVersion: v1
 kind: Service
 metadata:
   labels:
-    {{- include "element-io.matrix-rtc-sfu-rtc.labels" (dict "root" $ "context" .) | nindent 4 }}
-  name: {{ $.Release.Name }}-matrix-rtc-sfu-udp-range
+    {{- include "element-io.matrix-rtc-sfu-rtc.labels" (dict "root" $ "context" $.Values.matrixRTC.sfu) | nindent 4 }}
+  name: {{ $.Release.Name }}-matrix-rtc-sfu-udp-range-{{ $port_segment }}
   namespace: {{ $.Release.Namespace }}
 spec:
   type: NodePort
   externalTrafficPolicy: Local
   ports:
-{{- with .exposedServices.rtcUdp.portRange }}
-{{- range $port := untilStep (.startPort | int) (.endPort | int) 1 }}
+{{- $segment_start := ((add $range_start (mul 250 $port_segment)) | int) }}
+{{- $segment_end := ((min (add $range_end 1) (add $segment_start 250)) | int) }}
+{{- range $port := untilStep $segment_start $segment_end 1 }}
   - name: rtc-udp-{{ $port }}
     port: {{ $port }}
     targetPort: {{ $port }}
     nodePort: {{ $port }}
     protocol: UDP
-{{- end }}
 {{- end }}
   selector:
     app.kubernetes.io/instance: "{{ $.Release.Name }}-matrix-rtc-sfu"
 {{- end }}
 {{- end }}
 {{- end }}
 {{- end }}
+{{- end }}

newsfragments/549.changed.md

@@ -0,0 +1 @@
+Omit the UDP port range metadata for Matrix RTC's SFU if the range is larger than 100 ports.

newsfragments/549.fixed.md

@@ -0,0 +1 @@
+Fix Matrix RTC's SFU constructing an invalid Service if given too wide a nodePort range.

tests/manifests/test_matrix_rtc.py

@@ -5,6 +5,8 @@
 import pytest
 import yaml
 
+from .utils import template_id
+
 
 @pytest.mark.parametrize("values_file", ["matrix-rtc-minimal-values.yaml"])
 @pytest.mark.asyncio_cooperative
@@ -39,3 +41,68 @@ async def test_external_ip_underrides(values, make_templates):
             break
     else:
         raise RuntimeError("Could not find config.yaml")
+
+
+async def get_sfu_udp_port_range_services(start_port, end_point, values, make_templates):
+    values["matrixRTC"]["sfu"]["exposedServices"]["rtcUdp"]["portRange"]["startPort"] = start_port
+    values["matrixRTC"]["sfu"]["exposedServices"]["rtcUdp"]["portRange"]["endPort"] = end_point
+
+    services = []
+    for template in await make_templates(values):
+        if template["kind"] == "Service" and "matrix-rtc-sfu-udp-range" in template["metadata"]["name"]:
+            services.append(template)
+    return services
+
+
+def assert_sharded_udp_range_ports(start_port, end_port, service):
+    id = template_id(service)
+    service_ports = service["spec"]["ports"]
+    assert len(service_ports) == (end_port - start_port + 1), f"{id} doesn't have the correct number of ports in it"
+
+    assert service_ports[0]["port"] == start_port, f"{id} doesn't start with port {start_port}"
+    expected_port = start_port
+    for index, port in enumerate(service_ports):
+        assert port["port"] == expected_port, (
+            f"{id}.port[{index}]['port'] isn't {expected_port} ({start_port} to {end_port})"
+        )
+        assert port["targetPort"] == expected_port, (
+            f"{id}.port[{index}]['targetPort'] isn't {expected_port} ({start_port} to {end_port})"
+        )
+        assert port["nodePort"] == expected_port, (
+            f"{id}.port[{index}]['nodePort'] isn't {expected_port} ({start_port} to {end_port})"
+        )
+        assert port["name"] == f"rtc-udp-{expected_port}", (
+            f"{id}.port[{index}]['name'] isn't rtc-udp-{expected_port} ({start_port} to {end_port})"
+        )
+        expected_port += 1
+    assert port["port"] == end_port, f"{id} doesn't end with port {end_port}"
+
+
+@pytest.mark.parametrize("values_file", ["matrix-rtc-exposed-services-values.yaml"])
+@pytest.mark.asyncio_cooperative
+async def test_udp_range_services_are_sharded(values, make_templates):
+    start_port = 32000
+
+    services = await get_sfu_udp_port_range_services(start_port, start_port, values, make_templates)
+    assert len(services) == 1, "SFU UDP Range service is incorrectly sharded for 1 port"
+    assert_sharded_udp_range_ports(start_port, start_port, services[0])
+
+    services = await get_sfu_udp_port_range_services(start_port, start_port + 1, values, make_templates)
+    assert len(services) == 1, "SFU UDP Range service is incorrectly sharded for 2 ports"
+    assert_sharded_udp_range_ports(start_port, start_port + 1, services[0])
+
+    services = await get_sfu_udp_port_range_services(start_port, start_port + 249, values, make_templates)
+    assert len(services) == 1, "SFU UDP Range service is incorrectly sharded for 250 ports"
+    assert_sharded_udp_range_ports(start_port, start_port + 249, services[0])
+
+    services = await get_sfu_udp_port_range_services(start_port, start_port + 250, values, make_templates)
+    assert len(services) == 2, "SFU UDP Range service is incorrectly sharded for 251 ports"
+    assert_sharded_udp_range_ports(start_port, start_port + 249, services[0])
+    assert_sharded_udp_range_ports(start_port + 250, start_port + 250, services[1])
+
+    services = await get_sfu_udp_port_range_services(start_port, start_port + 999, values, make_templates)
+    assert len(services) == 4, "SFU UDP Range service is incorrectly sharded for 1000 ports"
+    assert_sharded_udp_range_ports(start_port, start_port + 249, services[0])
+    assert_sharded_udp_range_ports(start_port + 250, start_port + 499, services[1])
+    assert_sharded_udp_range_ports(start_port + 500, start_port + 749, services[2])
+    assert_sharded_udp_range_ports(start_port + 750, start_port + 999, services[3])

tests/manifests/test_pod_containers_ports.py

@@ -23,7 +23,7 @@ async def test_unique_ports_in_containers(templates):
 @pytest.mark.asyncio_cooperative
 async def test_ports_in_containers_are_named(templates):
     for template in templates:
-        if template["kind"] in ["Deployment", "StatefulSet", "Job"]:
+        if template["kind"] in ["Deployment", "StatefulSet"]:
             port_names = []
             for container in template["spec"]["template"]["spec"]["containers"]:
                 for port in container.get("ports", []):
@@ -45,3 +45,21 @@ async def test_no_ports_in_jobs(templates):
             for container in template["spec"]["template"]["spec"]["containers"]:
                 ports += [port["containerPort"] for port in container.get("ports", [])]
             assert len(ports) == 0, f"Ports are present in job: {template_id(template)}, {ports}"
+
+
+@pytest.mark.parametrize("values_file", values_files_to_test)
+@pytest.mark.asyncio_cooperative
+async def test_not_too_many_container_ports(templates):
+    for template in templates:
+        if template["kind"] in ["Deployment", "StatefulSet"]:
+            for container in template["spec"]["template"]["spec"]["containers"]:
+                number_of_ports = len(container.get("ports", []))
+                # This limit is fairly arbitrary. Unlike with Services (which have a limit of 250 ports),
+                # there doesn't appear to be a hard limit of number of ports on a Pod/container. However if
+                # you go wild you hit maximum document size when attempting to put the manifest into the
+                # cluster. 100 is chosen as anything more quickly makes `kubectl {describe,get}` unusable.
+                # Container ports are "just" metadata, albeit one which helps the scheduler if `hostPorts`
+                # are involved
+                assert number_of_ports < 100, (
+                    f"{template_id(template)}/{container['name']} has too many ports ({number_of_ports} >= 100)"
+                )

tests/manifests/test_services.py

@@ -13,12 +13,22 @@
 async def test_ports_in_services_are_named(templates):
     for template in templates:
         if template["kind"] == "Service":
-            id = template_id(template)
-            assert "ports" in template["spec"], f"{id} does not specify a ports list"
-            assert len(template["spec"]["ports"]) > 0, f"{id} does not include any ports"
-
             port_names = []
             for port in template["spec"]["ports"]:
-                assert "name" in port, f"{id} has a port without a name: {port}"
+                assert "name" in port, f"{template_id(template)} has a port without a name: {port}"
                 port_names.append(port["name"])
-            assert len(port_names) == len(set(port_names)), f"Port names are not unique: {id}, {port_names}"
+            assert len(port_names) == len(set(port_names)), (
+                f"Port names are not unique: {template_id(template)}, {port_names}"
+            )
+
+
+@pytest.mark.parametrize("values_file", services_values_files_to_test)
+@pytest.mark.asyncio_cooperative
+async def test_not_too_many_ports_in_services(templates):
+    for template in templates:
+        if template["kind"] == "Service":
+            assert "ports" in template["spec"], f"{template_id(template)} does not specify a ports list"
+
+            number_of_ports = len(template["spec"]["ports"])
+            assert number_of_ports > 0, f"{template_id(template)} does not include any ports"
+            assert number_of_ports <= 250, f"{template_id(template)} has more than 250 ports"