From 0482661cbccf442df0ff821f277c5a3091b0e5ca Mon Sep 17 00:00:00 2001 From: Ruslan Date: Tue, 14 Jul 2026 15:40:11 +0300 Subject: [PATCH 1/4] feat(k8s): expose UDP ports on cloud-compose via mixed-protocol LB sc parses the compose file and generates its own k8s Deployment + Service via Pulumi; during that translation the port protocol was discarded. toRunPorts mapped every compose port to a bare int, so "7882:7882/udp" silently became a TCP port and UDP never reached the pod. This blocks deploying a WebRTC media server (e.g. LiveKit SFU), where signaling is TCP (wss) but media is UDP. Thread the port protocol end-to-end: - CloudRunContainer.Ports is now []k8s.ContainerPort{Port, Protocol} instead of []int; toRunPorts preserves the compose protocol (udp -> "UDP"; tcp/unspecified -> ""). - New ports.go helpers set Protocol on ContainerPortArgs and ServicePortArgs, but only for non-TCP protocols, so TCP-only specs stay byte-for-byte identical (TCP keeps http- names; UDP gets udp-). - autoTCPReadinessProbe no longer attaches a TCP/HTTP probe to a UDP-only port (resolves the old "// TODO: support non-http ports"). - serviceType is now expressible via cloudExtras and threaded CloudExtras -> DeploymentConfig -> kubeArgs, letting a service turn its own Service into a LoadBalancer (required for protocols Caddy cannot proxy). A service exposing both TCP and UDP emits ONE mixed-protocol LoadBalancer Service (GA since k8s 1.26, supported on recent GKE) rather than splitting into two Services, so a single external IP fronts the whole service. Announcing that IP in ICE candidates (LiveKit rtc.use_external_ip / node_ip) is the app's job, not sc's. Backward compatible: ports with no protocol behave exactly as before, and serviceType defaults to ClusterIP. Co-Authored-By: Claude Opus 4.8 (1M context) Signed-off-by: Ruslan --- .../service-available-deployment-schemas.md | 57 ++++++ pkg/clouds/gcloud/gke_autopilot.go | 1 + pkg/clouds/k8s/containers_test.go | 16 +- pkg/clouds/k8s/kube_run.go | 5 + pkg/clouds/k8s/types.go | 58 ++++-- pkg/clouds/k8s/types_conversion_test.go | 9 +- pkg/clouds/pulumi/gcp/gke_autopilot_stack.go | 1 + pkg/clouds/pulumi/kubernetes/caddy.go | 2 +- .../kubernetes/custom_stack_vpa_test.go | 2 +- pkg/clouds/pulumi/kubernetes/deployment.go | 72 ++++--- .../pulumi/kubernetes/deployment_test.go | 26 +-- pkg/clouds/pulumi/kubernetes/kube_run.go | 1 + pkg/clouds/pulumi/kubernetes/ports.go | 88 +++++++++ pkg/clouds/pulumi/kubernetes/ports_test.go | 184 ++++++++++++++++++ .../pulumi/kubernetes/simple_container.go | 9 +- .../simple_container_advanced_test.go | 12 +- .../simple_container_edge_cases_test.go | 12 +- .../kubernetes/simple_container_final_test.go | 10 +- .../simple_container_parentenv_test.go | 8 +- .../kubernetes/simple_container_test.go | 2 +- .../kubernetes/vpa_container_policies_test.go | 2 +- 21 files changed, 481 insertions(+), 96 deletions(-) create mode 100644 pkg/clouds/pulumi/kubernetes/ports.go create mode 100644 pkg/clouds/pulumi/kubernetes/ports_test.go diff --git a/docs/docs/reference/service-available-deployment-schemas.md b/docs/docs/reference/service-available-deployment-schemas.md index 65379f42..952fd484 100644 --- a/docs/docs/reference/service-available-deployment-schemas.md +++ b/docs/docs/reference/service-available-deployment-schemas.md @@ -86,6 +86,63 @@ services: DATABASE_URL: ${DATABASE_HOST} ``` +### **Exposing UDP ports (e.g. WebRTC / LiveKit)** + +By default a `cloud-compose` service is created as a `ClusterIP` Service and +exposed through the shared Caddy ingress, which reverse-proxies **HTTP/TCP** +traffic only. Protocols that Caddy cannot proxy — most notably **UDP** — must be +exposed directly by turning the service's own Kubernetes Service into a +`LoadBalancer`. + +Two things are required: + +1. **Declare the UDP port in `docker-compose.yaml`** using the compose + `/udp` suffix. The port protocol is now preserved end-to-end and emitted on + both the container port and the Service port. Ports without a suffix stay + `TCP`, exactly as before. +2. **Set `serviceType: LoadBalancer`** under `cloudExtras`. This makes the + service's own Service a `LoadBalancer` with its own external IP, instead of + the default `ClusterIP` fronted by Caddy. + +```yaml +# docker-compose.yaml — a LiveKit SFU media server +services: + livekit: + image: livekit/livekit-server:latest + ports: + - "7880:7880" # signaling (wss) → TCP + - "7881:7881" # ICE/TCP → TCP + - "7882:7882/udp" # RTC media → UDP +``` + +```yaml +# client.yaml +stacks: + production: + type: cloud-compose + parent: myproject/devops + config: + dockerComposeFile: ./docker-compose.yaml + runs: + - livekit + cloudExtras: + serviceType: LoadBalancer +``` + +**Mixed TCP + UDP on one Service.** When a service exposes both TCP and UDP +ports, `sc` emits a **single mixed-protocol `LoadBalancer` Service** carrying all +ports (rather than splitting into separate TCP and UDP Services). This keeps a +single external IP for the whole service, which is what UDP media servers need +in order to advertise one address to clients. Mixed-protocol Services are GA in +Kubernetes since 1.26 (the `MixedProtocolLBService` feature) and are supported +by recent GKE — your cluster must be on a version that supports them. + +> **Announcing the external IP is the application's job.** `sc` provisions the +> `LoadBalancer` and the UDP port, but a media server such as LiveKit must be +> told its public address to put into ICE candidates (e.g. LiveKit's +> `rtc.use_external_ip` / `rtc.node_ip`). Wiring that address into your app's +> configuration is out of scope for `sc`. + ### **Deploying to Staging** ```sh sc deploy -s myservice -e staging diff --git a/pkg/clouds/gcloud/gke_autopilot.go b/pkg/clouds/gcloud/gke_autopilot.go index be909177..bf2a2f12 100644 --- a/pkg/clouds/gcloud/gke_autopilot.go +++ b/pkg/clouds/gcloud/gke_autopilot.go @@ -112,6 +112,7 @@ func ToGkeAutopilotConfig(tpl any, composeCfg compose.Config, stackCfg *api.Stac deployCfg.LivenessProbe = k8sCloudExtras.LivenessProbe // Extract global liveness probe configuration deployCfg.StartupProbe = k8sCloudExtras.StartupProbe // Extract global startup probe configuration deployCfg.PriorityClassName = k8sCloudExtras.PriorityClassName // Extract PriorityClass for pod scheduling and preemption + deployCfg.ServiceType = k8sCloudExtras.ServiceType // Extract Service type override (e.g. LoadBalancer for UDP) deployCfg.TopologySpreadConstraints = k8sCloudExtras.TopologySpreadConstraints // Process affinity rules and merge with existing NodeSelector if needed diff --git a/pkg/clouds/k8s/containers_test.go b/pkg/clouds/k8s/containers_test.go index f2f8f65d..f806295b 100644 --- a/pkg/clouds/k8s/containers_test.go +++ b/pkg/clouds/k8s/containers_test.go @@ -67,7 +67,7 @@ func TestConvertComposeToContainers_FullMapping(t *testing.T) { Expect(c.Command).To(Equal([]string{"/bin/entry"})) Expect(c.Args).To(Equal([]string{"serve", "--port", "8080"})) Expect(c.Env).To(HaveKeyWithValue("ENV", "prod")) - Expect(c.Ports).To(Equal([]int{8080})) + Expect(c.Ports).To(Equal(ContainerPorts(8080))) Expect(c.ComposeDir).To(Equal("/work")) Expect(c.Image.Context).To(Equal("./api")) // Context set, dockerfile empty => defaults to "Dockerfile". @@ -176,7 +176,7 @@ func TestFindIngressContainer(t *testing.T) { }, }} containers := []CloudRunContainer{ - {Name: "web", Ports: []int{3000}}, + {Name: "web", Ports: ContainerPorts(3000)}, {Name: "worker"}, } @@ -199,7 +199,7 @@ func TestFindIngressContainer(t *testing.T) { }}, }, }} - containers := []CloudRunContainer{{Name: "web", Ports: []int{3000}}} + containers := []CloudRunContainer{{Name: "web", Ports: ContainerPorts(3000)}} ic, err := FindIngressContainer(composeCfg, containers) Expect(err).ToNot(HaveOccurred()) @@ -218,7 +218,7 @@ func TestFindIngressContainer(t *testing.T) { }}, }, }} - containers := []CloudRunContainer{{Name: "web", Ports: []int{3000}}} + containers := []CloudRunContainer{{Name: "web", Ports: ContainerPorts(3000)}} ic, err := FindIngressContainer(composeCfg, containers) Expect(err).ToNot(HaveOccurred()) @@ -251,7 +251,7 @@ func TestFindIngressContainer(t *testing.T) { composeCfg := compose.Config{Project: &types.Project{ Services: []types.ServiceConfig{{Name: "solo"}}, }} - containers := []CloudRunContainer{{Name: "solo", Ports: []int{5000}}} + containers := []CloudRunContainer{{Name: "solo", Ports: ContainerPorts(5000)}} ic, err := FindIngressContainer(composeCfg, containers) Expect(err).ToNot(HaveOccurred()) @@ -267,8 +267,8 @@ func TestFindIngressContainer(t *testing.T) { Services: []types.ServiceConfig{{Name: "a"}, {Name: "b"}}, }} containers := []CloudRunContainer{ - {Name: "a", Ports: []int{1000}}, - {Name: "b", Ports: []int{2000}}, + {Name: "a", Ports: ContainerPorts(1000)}, + {Name: "b", Ports: ContainerPorts(2000)}, } ic, err := FindIngressContainer(composeCfg, containers) @@ -281,7 +281,7 @@ func TestFindIngressContainer(t *testing.T) { composeCfg := compose.Config{Project: &types.Project{ Services: []types.ServiceConfig{{Name: "solo"}}, }} - containers := []CloudRunContainer{{Name: "solo", Ports: []int{5000, 6000}}} + containers := []CloudRunContainer{{Name: "solo", Ports: ContainerPorts(5000, 6000)}} ic, err := FindIngressContainer(composeCfg, containers) Expect(err).ToNot(HaveOccurred()) diff --git a/pkg/clouds/k8s/kube_run.go b/pkg/clouds/k8s/kube_run.go index f8e799a4..5a539ddd 100644 --- a/pkg/clouds/k8s/kube_run.go +++ b/pkg/clouds/k8s/kube_run.go @@ -31,6 +31,10 @@ type CloudExtras struct { StartupProbe *CloudRunProbe `json:"startupProbe" yaml:"startupProbe"` EphemeralVolumes []GenericEphemeralVolume `json:"ephemeralVolumes" yaml:"ephemeralVolumes"` // Generic ephemeral volumes for large temp storage PriorityClassName *string `json:"priorityClassName" yaml:"priorityClassName"` // Kubernetes PriorityClass for pod scheduling and preemption + // ServiceType overrides the Kubernetes Service type for the service itself (default: ClusterIP, + // fronted by the shared Caddy ingress). Set to "LoadBalancer" to expose the service's ports + // directly, which is required for non-HTTP protocols such as UDP that Caddy cannot proxy. + ServiceType *string `json:"serviceType,omitempty" yaml:"serviceType,omitempty"` TopologySpreadConstraints []TopologySpreadConstraint `json:"topologySpreadConstraints" yaml:"topologySpreadConstraints"` } @@ -204,6 +208,7 @@ func ToKubernetesRunConfig(tpl any, composeCfg compose.Config, stackCfg *api.Sta deployCfg.LivenessProbe = k8sCloudExtras.LivenessProbe // Extract global liveness probe configuration deployCfg.EphemeralVolumes = k8sCloudExtras.EphemeralVolumes // Extract generic ephemeral volumes configuration deployCfg.PriorityClassName = k8sCloudExtras.PriorityClassName // Extract PriorityClass for pod scheduling and preemption + deployCfg.ServiceType = k8sCloudExtras.ServiceType // Extract Service type override (e.g. LoadBalancer for UDP) deployCfg.TopologySpreadConstraints = k8sCloudExtras.TopologySpreadConstraints // Process affinity rules and merge with existing NodeSelector if needed diff --git a/pkg/clouds/k8s/types.go b/pkg/clouds/k8s/types.go index 279bb178..45cbc0da 100644 --- a/pkg/clouds/k8s/types.go +++ b/pkg/clouds/k8s/types.go @@ -30,12 +30,13 @@ type DeploymentConfig struct { NodeSelector map[string]string `json:"nodeSelector" yaml:"nodeSelector"` Affinity *AffinityRules `json:"affinity" yaml:"affinity"` Tolerations []Toleration `json:"tolerations" yaml:"tolerations"` - VPA *VPAConfig `json:"vpa" yaml:"vpa"` // Vertical Pod Autoscaler configuration - ReadinessProbe *CloudRunProbe `json:"readinessProbe" yaml:"readinessProbe"` // Global readiness probe configuration - LivenessProbe *CloudRunProbe `json:"livenessProbe" yaml:"livenessProbe"` // Global liveness probe configuration - StartupProbe *CloudRunProbe `json:"startupProbe" yaml:"startupProbe"` // Global startup probe configuration - EphemeralVolumes []GenericEphemeralVolume `json:"ephemeralVolumes" yaml:"ephemeralVolumes"` // Generic ephemeral volumes for large temp storage - PriorityClassName *string `json:"priorityClassName" yaml:"priorityClassName"` // Kubernetes PriorityClass for pod scheduling and preemption + VPA *VPAConfig `json:"vpa" yaml:"vpa"` // Vertical Pod Autoscaler configuration + ReadinessProbe *CloudRunProbe `json:"readinessProbe" yaml:"readinessProbe"` // Global readiness probe configuration + LivenessProbe *CloudRunProbe `json:"livenessProbe" yaml:"livenessProbe"` // Global liveness probe configuration + StartupProbe *CloudRunProbe `json:"startupProbe" yaml:"startupProbe"` // Global startup probe configuration + EphemeralVolumes []GenericEphemeralVolume `json:"ephemeralVolumes" yaml:"ephemeralVolumes"` // Generic ephemeral volumes for large temp storage + PriorityClassName *string `json:"priorityClassName" yaml:"priorityClassName"` // Kubernetes PriorityClass for pod scheduling and preemption + ServiceType *string `json:"serviceType,omitempty" yaml:"serviceType,omitempty"` // Kubernetes Service type for the service (default: ClusterIP; LoadBalancer to expose UDP/non-HTTP ports directly) TopologySpreadConstraints []TopologySpreadConstraint `json:"topologySpreadConstraints" yaml:"topologySpreadConstraints"` } @@ -170,6 +171,23 @@ type ProbeHttpGet struct { HTTPHeaders []HTTPHeader `json:"httpHeaders,omitempty" yaml:"httpHeaders,omitempty"` } +// ContainerPort is a single exposed container port together with its L4 +// protocol. Protocol is the Kubernetes protocol name ("TCP" or "UDP"); an empty +// value means the Kubernetes default (TCP) and is emitted as no protocol at all, +// so ports without an explicit protocol stay byte-for-byte identical to before +// protocol support was added. +type ContainerPort struct { + Port int `json:"port" yaml:"port"` + Protocol string `json:"protocol,omitempty" yaml:"protocol,omitempty"` +} + +// ContainerPorts builds ContainerPort entries with the default (TCP) protocol. +func ContainerPorts(ports ...int) []ContainerPort { + return lo.Map(ports, func(p int, _ int) ContainerPort { + return ContainerPort{Port: p} + }) +} + type CloudRunContainer struct { Name string `json:"name" yaml:"name"` Command []string `json:"command" yaml:"command"` @@ -177,7 +195,7 @@ type CloudRunContainer struct { Image api.ContainerImage `json:"image" yaml:"image"` Env map[string]string `json:"env" yaml:"env"` Secrets map[string]string `json:"secrets" yaml:"secrets"` - Ports []int `json:"ports" yaml:"ports"` + Ports []ContainerPort `json:"ports" yaml:"ports"` MainPort *int `json:"mainPort" yaml:"mainPort"` ReadinessProbe *CloudRunProbe `json:"readinessProbe" yaml:"readinessProbe"` LivenessProbe *CloudRunProbe `json:"livenessProbe" yaml:"livenessProbe"` @@ -490,7 +508,7 @@ func ConvertComposeToContainers(composeCfg compose.Config, stackCfg *api.StackCo if container.MainPort == nil && len(container.Ports) > 1 { container.Warnings = append(container.Warnings, fmt.Sprintf("container %q has multiple ports and no main port specified", container.Name)) } else if len(container.Ports) > 0 { - container.MainPort = lo.ToPtr(container.Ports[0]) + container.MainPort = lo.ToPtr(container.Ports[0].Port) } containers = append(containers, container) } @@ -514,7 +532,7 @@ func FindIngressContainer(composeCfg compose.Config, contaniers []CloudRunContai }) if !found && len(contaniers) == 1 && len(contaniers[0].Ports) == 1 { iContainer = contaniers[0] - iContainer.MainPort = lo.ToPtr(iContainer.Ports[0]) + iContainer.MainPort = lo.ToPtr(iContainer.Ports[0].Port) found = true } if !found { @@ -530,17 +548,31 @@ func FindIngressContainer(composeCfg compose.Config, contaniers []CloudRunContai } } if iContainer.MainPort == nil && len(iContainer.Ports) == 1 { - iContainer.MainPort = lo.ToPtr(iContainer.Ports[0]) + iContainer.MainPort = lo.ToPtr(iContainer.Ports[0].Port) } return &iContainer, nil } -func toRunPorts(ports []types.ServicePortConfig) []int { - return lo.Map(ports, func(p types.ServicePortConfig, _ int) int { - return int(p.Target) +func toRunPorts(ports []types.ServicePortConfig) []ContainerPort { + return lo.Map(ports, func(p types.ServicePortConfig, _ int) ContainerPort { + return ContainerPort{ + Port: int(p.Target), + Protocol: normalizeComposeProtocol(p.Protocol), + } }) } +// normalizeComposeProtocol maps a docker-compose port protocol ("tcp"/"udp"/"") +// to the Kubernetes protocol stored on ContainerPort. TCP (the default) is +// stored as an empty string so downstream resources omit the protocol field and +// remain byte-for-byte identical to specs generated before UDP support. +func normalizeComposeProtocol(proto string) string { + if strings.EqualFold(proto, "udp") { + return "UDP" + } + return "" +} + func toStartupProbe(check *types.HealthCheckConfig) *CloudRunProbe { if check == nil { return nil diff --git a/pkg/clouds/k8s/types_conversion_test.go b/pkg/clouds/k8s/types_conversion_test.go index 9e24b100..8be8da73 100644 --- a/pkg/clouds/k8s/types_conversion_test.go +++ b/pkg/clouds/k8s/types_conversion_test.go @@ -474,8 +474,15 @@ func TestToRunPorts(t *testing.T) { RegisterTestingT(t) ports := []types.ServicePortConfig{{Target: 8080}, {Target: 9090}} - Expect(toRunPorts(ports)).To(Equal([]int{8080, 9090})) + Expect(toRunPorts(ports)).To(Equal(ContainerPorts(8080, 9090))) Expect(toRunPorts(nil)).To(BeEmpty()) + + // A UDP port keeps its protocol; TCP/unspecified ports normalise to "". + udpPorts := []types.ServicePortConfig{{Target: 7880}, {Target: 7882, Protocol: "udp"}} + Expect(toRunPorts(udpPorts)).To(Equal([]ContainerPort{ + {Port: 7880}, + {Port: 7882, Protocol: "UDP"}, + })) } func TestToRunEnv(t *testing.T) { diff --git a/pkg/clouds/pulumi/gcp/gke_autopilot_stack.go b/pkg/clouds/pulumi/gcp/gke_autopilot_stack.go index 0edec3a6..2e9aa468 100644 --- a/pkg/clouds/pulumi/gcp/gke_autopilot_stack.go +++ b/pkg/clouds/pulumi/gcp/gke_autopilot_stack.go @@ -193,6 +193,7 @@ func GkeAutopilotStack(ctx *sdk.Context, stack api.Stack, input api.ResourceInpu ReadinessProbe: gkeAutopilotInput.Deployment.ReadinessProbe, // Pass global readiness probe configuration LivenessProbe: gkeAutopilotInput.Deployment.LivenessProbe, // Pass global liveness probe configuration StartupProbe: gkeAutopilotInput.Deployment.StartupProbe, // Pass global startup probe configuration + ServiceType: gkeAutopilotInput.Deployment.ServiceType, // Pass Service type override (e.g. LoadBalancer for UDP) EphemeralSize: ephemeralSize, } diff --git a/pkg/clouds/pulumi/kubernetes/caddy.go b/pkg/clouds/pulumi/kubernetes/caddy.go index 9355b6d3..db68680f 100644 --- a/pkg/clouds/pulumi/kubernetes/caddy.go +++ b/pkg/clouds/pulumi/kubernetes/caddy.go @@ -168,7 +168,7 @@ func DeployCaddyService(ctx *sdk.Context, caddy CaddyDeployment, input api.Resou Name: caddyImage, Platform: api.ImagePlatformLinuxAmd64, }, - Ports: []int{443, 80}, + Ports: k8s.ContainerPorts(443, 80), MainPort: lo.ToPtr(80), Resources: caddy.Resources, // Use custom resources if specified, otherwise defaults will be applied } diff --git a/pkg/clouds/pulumi/kubernetes/custom_stack_vpa_test.go b/pkg/clouds/pulumi/kubernetes/custom_stack_vpa_test.go index 4ff06d62..10b71a01 100644 --- a/pkg/clouds/pulumi/kubernetes/custom_stack_vpa_test.go +++ b/pkg/clouds/pulumi/kubernetes/custom_stack_vpa_test.go @@ -35,7 +35,7 @@ func TestNewSimpleContainer_CustomStackVPATargetsDeployment(t *testing.T) { IngressContainer: &k8s.CloudRunContainer{ Name: "myapp-tenant-a", MainPort: lo.ToPtr(8080), - Ports: []int{8080}, + Ports: k8s.ContainerPorts(8080), }, Log: logger.New(), Containers: []corev1.ContainerArgs{ diff --git a/pkg/clouds/pulumi/kubernetes/deployment.go b/pkg/clouds/pulumi/kubernetes/deployment.go index 066f8f5b..41e32846 100644 --- a/pkg/clouds/pulumi/kubernetes/deployment.go +++ b/pkg/clouds/pulumi/kubernetes/deployment.go @@ -128,15 +128,8 @@ func DeploySimpleContainer(ctx *sdk.Context, args Args, opts ...sdk.ResourceOpti Value: sdk.String(containerEnvVars[k]), }) } - var ports corev1.ContainerPortArray - var readinessProbe *corev1.ProbeArgs - for _, p := range c.Container.Ports { - portName := toPortName(p) // TODO: support non-http ports - ports = append(ports, corev1.ContainerPortArgs{ - Name: sdk.String(portName), - ContainerPort: sdk.Int(p), - }) - } + ports := toContainerPorts(c.Container.Ports) + cReadyProbe := c.Container.ReadinessProbe // Use global readiness probe if container doesn't have one AND it's the ingress container // This prevents applying HTTP/TCP probes to worker containers that don't expose ports @@ -145,26 +138,15 @@ func DeploySimpleContainer(ctx *sdk.Context, args Args, opts ...sdk.ResourceOpti cReadyProbe = args.ReadinessProbe } - if cReadyProbe == nil && len(c.Container.Ports) == 1 { - readinessProbe = &corev1.ProbeArgs{ - TcpSocket: corev1.TCPSocketActionArgs{ - Port: sdk.String(toPortName(c.Container.Ports[0])), - }, - PeriodSeconds: sdk.IntPtr(10), - InitialDelaySeconds: sdk.IntPtr(5), - } - } else if cReadyProbe == nil && c.Container.MainPort != nil { - readinessProbe = &corev1.ProbeArgs{ - TcpSocket: corev1.TCPSocketActionArgs{ - Port: sdk.String(toPortName(lo.FromPtr(c.Container.MainPort))), - }, - PeriodSeconds: sdk.IntPtr(10), - InitialDelaySeconds: sdk.IntPtr(5), - } - } else if cReadyProbe != nil { + var readinessProbe *corev1.ProbeArgs + if cReadyProbe != nil { readinessProbe = toProbeArgs(c, cReadyProbe) - } else if len(c.Container.Ports) > 1 { - return corev1.ContainerArgs{}, errors.Errorf("container %q has multiple ports and no readiness probe specified", c.Container.Name) + } else { + probe, probeErr := autoTCPReadinessProbe(c.Container) + if probeErr != nil { + return corev1.ContainerArgs{}, probeErr + } + readinessProbe = probe } // Handle liveness probe @@ -333,7 +315,7 @@ func toProbeArgs(c *ContainerImage, probe *k8s.CloudRunProbe) *corev1.ProbeArgs } else if c.Container.MainPort != nil && *c.Container.MainPort > 0 { probePort = *c.Container.MainPort } else if len(c.Container.Ports) > 0 { - probePort = c.Container.Ports[0] + probePort = c.Container.Ports[0].Port } // periodSeconds (k8s-native) wins over the legacy duration-typed interval @@ -381,3 +363,35 @@ func toProbeArgs(c *ContainerImage, probe *k8s.CloudRunProbe) *corev1.ProbeArgs func toPortName(p int) string { return fmt.Sprintf("http-%d", p) } + +func tcpSocketProbe(portName string) *corev1.ProbeArgs { + return &corev1.ProbeArgs{ + TcpSocket: corev1.TCPSocketActionArgs{ + Port: sdk.String(portName), + }, + PeriodSeconds: sdk.IntPtr(10), + InitialDelaySeconds: sdk.IntPtr(5), + } +} + +// autoTCPReadinessProbe derives the default TCP readiness probe for a container +// that has no explicit (or global ingress) probe configured. UDP-only ports +// have no HTTP/TCP health surface, so no probe is attached to them. +func autoTCPReadinessProbe(container k8s.CloudRunContainer) (*corev1.ProbeArgs, error) { + switch { + case len(container.Ports) == 1: + if isUDP(container.Ports[0].Protocol) { + return nil, nil + } + return tcpSocketProbe(toPortName(container.Ports[0].Port)), nil + case container.MainPort != nil: + if isUDP(portProtocol(container, *container.MainPort)) { + return nil, nil + } + return tcpSocketProbe(toPortName(*container.MainPort)), nil + case len(container.Ports) > 1: + return nil, errors.Errorf("container %q has multiple ports and no readiness probe specified", container.Name) + default: + return nil, nil + } +} diff --git a/pkg/clouds/pulumi/kubernetes/deployment_test.go b/pkg/clouds/pulumi/kubernetes/deployment_test.go index 481d8b06..716c4c64 100644 --- a/pkg/clouds/pulumi/kubernetes/deployment_test.go +++ b/pkg/clouds/pulumi/kubernetes/deployment_test.go @@ -27,7 +27,7 @@ func TestToProbeArgs_WithHeaders(t *testing.T) { container: &ContainerImage{ Container: k8s.CloudRunContainer{ Name: "test-container", - Ports: []int{8080}, + Ports: k8s.ContainerPorts(8080), MainPort: lo.ToPtr(8080), }, }, @@ -49,7 +49,7 @@ func TestToProbeArgs_WithHeaders(t *testing.T) { container: &ContainerImage{ Container: k8s.CloudRunContainer{ Name: "test-container", - Ports: []int{8080}, + Ports: k8s.ContainerPorts(8080), MainPort: lo.ToPtr(8080), }, }, @@ -73,7 +73,7 @@ func TestToProbeArgs_WithHeaders(t *testing.T) { container: &ContainerImage{ Container: k8s.CloudRunContainer{ Name: "test-container", - Ports: []int{8080}, + Ports: k8s.ContainerPorts(8080), MainPort: lo.ToPtr(8080), }, }, @@ -92,7 +92,7 @@ func TestToProbeArgs_WithHeaders(t *testing.T) { container: &ContainerImage{ Container: k8s.CloudRunContainer{ Name: "test-container", - Ports: []int{8080}, + Ports: k8s.ContainerPorts(8080), MainPort: lo.ToPtr(8080), }, }, @@ -112,7 +112,7 @@ func TestToProbeArgs_WithHeaders(t *testing.T) { container: &ContainerImage{ Container: k8s.CloudRunContainer{ Name: "test-container", - Ports: []int{8080}, + Ports: k8s.ContainerPorts(8080), MainPort: lo.ToPtr(8080), }, }, @@ -129,7 +129,7 @@ func TestToProbeArgs_WithHeaders(t *testing.T) { container: &ContainerImage{ Container: k8s.CloudRunContainer{ Name: "test-container", - Ports: []int{8080}, + Ports: k8s.ContainerPorts(8080), MainPort: lo.ToPtr(8080), }, }, @@ -150,7 +150,7 @@ func TestToProbeArgs_WithHeaders(t *testing.T) { container: &ContainerImage{ Container: k8s.CloudRunContainer{ Name: "test-container", - Ports: []int{8080}, + Ports: k8s.ContainerPorts(8080), MainPort: lo.ToPtr(8080), }, }, @@ -208,7 +208,7 @@ func TestToProbeArgs_PortResolution(t *testing.T) { container: &ContainerImage{ Container: k8s.CloudRunContainer{ Name: "test-container", - Ports: []int{8080, 9090}, + Ports: k8s.ContainerPorts(8080, 9090), }, }, probe: &k8s.CloudRunProbe{ @@ -224,7 +224,7 @@ func TestToProbeArgs_PortResolution(t *testing.T) { container: &ContainerImage{ Container: k8s.CloudRunContainer{ Name: "test-container", - Ports: []int{8080, 9090}, + Ports: k8s.ContainerPorts(8080, 9090), MainPort: lo.ToPtr(9090), }, }, @@ -241,7 +241,7 @@ func TestToProbeArgs_PortResolution(t *testing.T) { container: &ContainerImage{ Container: k8s.CloudRunContainer{ Name: "test-container", - Ports: []int{8080, 9090}, + Ports: k8s.ContainerPorts(8080, 9090), // No MainPort set }, }, @@ -277,7 +277,7 @@ func TestToProbeArgs_BackwardCompatibility(t *testing.T) { container := &ContainerImage{ Container: k8s.CloudRunContainer{ Name: "test-container", - Ports: []int{8080}, + Ports: k8s.ContainerPorts(8080), MainPort: lo.ToPtr(8080), }, } @@ -308,7 +308,7 @@ func TestToProbeArgs_HeaderPreservation(t *testing.T) { container := &ContainerImage{ Container: k8s.CloudRunContainer{ Name: "test-container", - Ports: []int{8080}, + Ports: k8s.ContainerPorts(8080), MainPort: lo.ToPtr(8080), }, } @@ -337,7 +337,7 @@ func TestToProbeArgs_PeriodSeconds(t *testing.T) { container := &ContainerImage{ Container: k8s.CloudRunContainer{ Name: "test-container", - Ports: []int{8080}, + Ports: k8s.ContainerPorts(8080), MainPort: lo.ToPtr(8080), }, } diff --git a/pkg/clouds/pulumi/kubernetes/kube_run.go b/pkg/clouds/pulumi/kubernetes/kube_run.go index 6e0c90d5..5aaac5db 100644 --- a/pkg/clouds/pulumi/kubernetes/kube_run.go +++ b/pkg/clouds/pulumi/kubernetes/kube_run.go @@ -149,6 +149,7 @@ func KubeRun(ctx *sdk.Context, stack api.Stack, input api.ResourceInput, params ReadinessProbe: kubeRunInput.Deployment.ReadinessProbe, // Pass global readiness probe configuration LivenessProbe: kubeRunInput.Deployment.LivenessProbe, // Pass global liveness probe configuration StartupProbe: kubeRunInput.Deployment.StartupProbe, // Pass global startup probe configuration + ServiceType: kubeRunInput.Deployment.ServiceType, // Pass Service type override (e.g. LoadBalancer for UDP) EphemeralSize: lo.FromPtr(kubeRunInput.Deployment.StackConfig).Size.Ephemeral, } diff --git a/pkg/clouds/pulumi/kubernetes/ports.go b/pkg/clouds/pulumi/kubernetes/ports.go new file mode 100644 index 00000000..2594dfca --- /dev/null +++ b/pkg/clouds/pulumi/kubernetes/ports.go @@ -0,0 +1,88 @@ +// SPDX-License-Identifier: MIT +// Copyright (c) Simple Container + +package kubernetes + +import ( + "fmt" + "strings" + + corev1 "github.com/pulumi/pulumi-kubernetes/sdk/v4/go/kubernetes/core/v1" + sdk "github.com/pulumi/pulumi/sdk/v3/go/pulumi" + + "github.com/simple-container-com/api/pkg/clouds/k8s" +) + +func isUDP(proto string) bool { + return strings.EqualFold(proto, "UDP") +} + +// portProtocol returns the protocol declared for the given port number on the +// container, or "" (TCP) when the port is not declared. +func portProtocol(container k8s.CloudRunContainer, port int) string { + for _, p := range container.Ports { + if p.Port == port { + return p.Protocol + } + } + return "" +} + +// k8sProtocol returns the protocol value to emit on a generated Kubernetes port. +// TCP is the Kubernetes default, so it is returned as an empty string (protocol +// omitted) to keep TCP-only specs byte-for-byte identical to those generated +// before UDP support. Only non-default protocols (UDP) are emitted explicitly. +func k8sProtocol(proto string) string { + if isUDP(proto) { + return "UDP" + } + return "" +} + +// toPortName derives the Kubernetes port name for a container port. TCP and +// unspecified ports keep the historical "http-" name so existing specs do +// not change; UDP ports use a "udp-" name so mixed-protocol Services have +// unique, meaningful port names. +func toContainerPortName(p k8s.ContainerPort) string { + if isUDP(p.Protocol) { + return fmt.Sprintf("udp-%d", p.Port) + } + return toPortName(p.Port) +} + +// toContainerPorts converts the internal container-port model into Kubernetes +// container ports, carrying the protocol through for non-TCP ports. +func toContainerPorts(ports []k8s.ContainerPort) corev1.ContainerPortArray { + var result corev1.ContainerPortArray + for _, p := range ports { + args := corev1.ContainerPortArgs{ + Name: sdk.String(toContainerPortName(p)), + ContainerPort: sdk.Int(p.Port), + } + if proto := k8sProtocol(p.Protocol); proto != "" { + args.Protocol = sdk.String(proto) + } + result = append(result, args) + } + return result +} + +// toServicePorts converts the internal container-port model into Kubernetes +// Service ports. A Service that carries both TCP and UDP ports is a +// mixed-protocol Service (GA since Kubernetes 1.26 via the +// MixedProtocolLBService feature) and requires a cluster/cloud LoadBalancer that +// supports it (recent GKE does). +func toServicePorts(ports []k8s.ContainerPort) corev1.ServicePortArray { + result := corev1.ServicePortArray{} + for _, p := range ports { + args := corev1.ServicePortArgs{ + Name: sdk.String(toContainerPortName(p)), + Port: sdk.Int(p.Port), + } + if proto := k8sProtocol(p.Protocol); proto != "" { + args.Protocol = sdk.String(proto) + } + result = append(result, args) + } + return result +} diff --git a/pkg/clouds/pulumi/kubernetes/ports_test.go b/pkg/clouds/pulumi/kubernetes/ports_test.go new file mode 100644 index 00000000..a0a3aad9 --- /dev/null +++ b/pkg/clouds/pulumi/kubernetes/ports_test.go @@ -0,0 +1,184 @@ +// SPDX-License-Identifier: MIT +// Copyright (c) Simple Container + +package kubernetes + +import ( + "testing" + + . "github.com/onsi/gomega" + corev1 "github.com/pulumi/pulumi-kubernetes/sdk/v4/go/kubernetes/core/v1" + "github.com/pulumi/pulumi/sdk/v3/go/pulumi" + sdk "github.com/pulumi/pulumi/sdk/v3/go/pulumi" + "github.com/samber/lo" + + "github.com/simple-container-com/api/pkg/api/logger" + "github.com/simple-container-com/api/pkg/clouds/k8s" +) + +func TestK8sProtocol(t *testing.T) { + RegisterTestingT(t) + + // TCP and unspecified stay empty so the protocol field is omitted (byte-for-byte compat). + Expect(k8sProtocol("")).To(Equal("")) + Expect(k8sProtocol("tcp")).To(Equal("")) + Expect(k8sProtocol("TCP")).To(Equal("")) + // UDP is emitted explicitly, case-insensitively. + Expect(k8sProtocol("udp")).To(Equal("UDP")) + Expect(k8sProtocol("UDP")).To(Equal("UDP")) +} + +func TestToContainerPortName(t *testing.T) { + RegisterTestingT(t) + + // TCP/unspecified keep the historical "http-" name. + Expect(toContainerPortName(k8s.ContainerPort{Port: 7880})).To(Equal("http-7880")) + Expect(toContainerPortName(k8s.ContainerPort{Port: 7881, Protocol: "TCP"})).To(Equal("http-7881")) + // UDP ports use a distinct name so mixed-protocol Services have unique names. + Expect(toContainerPortName(k8s.ContainerPort{Port: 7882, Protocol: "UDP"})).To(Equal("udp-7882")) +} + +// livekitPorts models a LiveKit SFU ingress container: TCP signaling (7880), +// TCP ICE/TURN (7881) and a UDP media port (7882). +func livekitPorts() []k8s.ContainerPort { + return []k8s.ContainerPort{ + {Port: 7880}, + {Port: 7881}, + {Port: 7882, Protocol: "UDP"}, + } +} + +func TestToContainerPorts_ThreadsProtocol(t *testing.T) { + RegisterTestingT(t) + + ports := toContainerPorts(livekitPorts()) + Expect(ports).To(HaveLen(3)) + + tcp0 := ports[0].(corev1.ContainerPortArgs) + Expect(tcp0.ContainerPort).To(Equal(sdk.Int(7880))) + Expect(tcp0.Name).To(Equal(sdk.String("http-7880"))) + Expect(tcp0.Protocol).To(BeNil(), "TCP ports must omit protocol to stay byte-for-byte compatible") + + tcp1 := ports[1].(corev1.ContainerPortArgs) + Expect(tcp1.Protocol).To(BeNil()) + + udp := ports[2].(corev1.ContainerPortArgs) + Expect(udp.ContainerPort).To(Equal(sdk.Int(7882))) + Expect(udp.Name).To(Equal(sdk.String("udp-7882"))) + Expect(udp.Protocol).To(Equal(sdk.String("UDP"))) +} + +func TestToServicePorts_ThreadsProtocol(t *testing.T) { + RegisterTestingT(t) + + ports := toServicePorts(livekitPorts()) + Expect(ports).To(HaveLen(3)) + + tcp0 := ports[0].(corev1.ServicePortArgs) + Expect(tcp0.Port).To(Equal(sdk.Int(7880))) + Expect(tcp0.Name).To(Equal(sdk.String("http-7880"))) + Expect(tcp0.Protocol).To(BeNil()) + + tcp1 := ports[1].(corev1.ServicePortArgs) + Expect(tcp1.Port).To(Equal(sdk.Int(7881))) + Expect(tcp1.Protocol).To(BeNil()) + + udp := ports[2].(corev1.ServicePortArgs) + Expect(udp.Port).To(Equal(sdk.Int(7882))) + Expect(udp.Name).To(Equal(sdk.String("udp-7882"))) + Expect(udp.Protocol).To(Equal(sdk.String("UDP"))) +} + +func TestAutoTCPReadinessProbe(t *testing.T) { + RegisterTestingT(t) + + t.Run("single TCP port gets a TCP probe", func(t *testing.T) { + RegisterTestingT(t) + probe, err := autoTCPReadinessProbe(k8s.CloudRunContainer{ + Name: "web", + Ports: k8s.ContainerPorts(8080), + MainPort: lo.ToPtr(8080), + }) + Expect(err).ToNot(HaveOccurred()) + Expect(probe).ToNot(BeNil()) + Expect(probe.TcpSocket).ToNot(BeNil()) + Expect(probe.HttpGet).To(BeNil()) + }) + + t.Run("single UDP port gets no probe", func(t *testing.T) { + RegisterTestingT(t) + probe, err := autoTCPReadinessProbe(k8s.CloudRunContainer{ + Name: "media", + Ports: []k8s.ContainerPort{{Port: 7882, Protocol: "UDP"}}, + MainPort: lo.ToPtr(7882), + }) + Expect(err).ToNot(HaveOccurred()) + Expect(probe).To(BeNil(), "a UDP-only port has no HTTP/TCP health surface") + }) + + t.Run("mixed TCP+UDP probes the TCP main port", func(t *testing.T) { + RegisterTestingT(t) + probe, err := autoTCPReadinessProbe(k8s.CloudRunContainer{ + Name: "livekit", + Ports: livekitPorts(), + MainPort: lo.ToPtr(7880), + }) + Expect(err).ToNot(HaveOccurred()) + Expect(probe).ToNot(BeNil()) + Expect(probe.TcpSocket).ToNot(BeNil()) + }) + + t.Run("mixed ports with a UDP main port gets no probe", func(t *testing.T) { + RegisterTestingT(t) + probe, err := autoTCPReadinessProbe(k8s.CloudRunContainer{ + Name: "media-main", + Ports: livekitPorts(), + MainPort: lo.ToPtr(7882), + }) + Expect(err).ToNot(HaveOccurred()) + Expect(probe).To(BeNil()) + }) +} + +// TestSimpleContainer_UDPLoadBalancer exercises the full ingress path for a +// LiveKit-style container that exposes both TCP and UDP ports behind a +// LoadBalancer Service (a mixed-protocol Service). +func TestSimpleContainer_UDPLoadBalancer(t *testing.T) { + RegisterTestingT(t) + + mocks := NewSimpleContainerMocks() + + err := pulumi.RunErr(func(ctx *pulumi.Context) error { + args := &SimpleContainerArgs{ + Namespace: "livekit-test", + Service: "livekit", + ScEnv: "test", + Deployment: "livekit-deployment", + Replicas: 1, + Log: logger.New(), + + IngressContainer: &k8s.CloudRunContainer{ + Name: "livekit", + Ports: livekitPorts(), + MainPort: lo.ToPtr(7880), + }, + ServiceType: lo.ToPtr("LoadBalancer"), + + Containers: []corev1.ContainerArgs{ + { + Name: sdk.String("livekit"), + Image: sdk.String("livekit/livekit-server:latest"), + Ports: toContainerPorts(livekitPorts()), + }, + }, + } + + sc, err := NewSimpleContainer(ctx, args) + Expect(err).ToNot(HaveOccurred(), "mixed-protocol LoadBalancer container should be created successfully") + Expect(sc).ToNot(BeNil()) + Expect(sc.Service).ToNot(BeNil(), "a Service must be provisioned for the exposed ports") + return nil + }, pulumi.WithMocks("project", "stack", mocks)) + + Expect(err).ToNot(HaveOccurred()) +} diff --git a/pkg/clouds/pulumi/kubernetes/simple_container.go b/pkg/clouds/pulumi/kubernetes/simple_container.go index d32a8bfb..ab6da796 100644 --- a/pkg/clouds/pulumi/kubernetes/simple_container.go +++ b/pkg/clouds/pulumi/kubernetes/simple_container.go @@ -219,7 +219,7 @@ func NewSimpleContainer(ctx *sdk.Context, args *SimpleContainerArgs, opts ...sdk if args.IngressContainer != nil && args.IngressContainer.MainPort != nil { mainPort = args.IngressContainer.MainPort } else if len(lo.FromPtr(args.IngressContainer).Ports) == 1 { - mainPort = lo.ToPtr(lo.FromPtr(args.IngressContainer).Ports[0]) + mainPort = lo.ToPtr(lo.FromPtr(args.IngressContainer).Ports[0].Port) } if mainPort != nil { appAnnotations[AnnotationPort] = strconv.Itoa(*mainPort) @@ -829,12 +829,7 @@ ${proto}://${domain} { servicePorts := corev1.ServicePortArray{} if args.IngressContainer != nil { - for _, p := range lo.FromPtr(args.IngressContainer).Ports { - servicePorts = append(servicePorts, corev1.ServicePortArgs{ - Name: sdk.String(toPortName(p)), - Port: sdk.Int(p), - }) - } + servicePorts = toServicePorts(lo.FromPtr(args.IngressContainer).Ports) } // Build the Pulumi-input annotation map. The caddyfile-entry value, if // any, is an Output that resolves the namespace placeholder against the diff --git a/pkg/clouds/pulumi/kubernetes/simple_container_advanced_test.go b/pkg/clouds/pulumi/kubernetes/simple_container_advanced_test.go index c83ffa13..3df4641b 100644 --- a/pkg/clouds/pulumi/kubernetes/simple_container_advanced_test.go +++ b/pkg/clouds/pulumi/kubernetes/simple_container_advanced_test.go @@ -36,7 +36,7 @@ func TestSimpleContainer_MultiContainerDeployment(t *testing.T) { IngressContainer: &k8s.CloudRunContainer{ Name: "main-container", - Ports: []int{8080}, + Ports: k8s.ContainerPorts(8080), MainPort: lo.ToPtr(8080), }, ServiceType: lo.ToPtr("ClusterIP"), @@ -117,7 +117,7 @@ func TestSimpleContainer_ComplexVolumeConfiguration(t *testing.T) { IngressContainer: &k8s.CloudRunContainer{ Name: "app-container", - Ports: []int{8080}, + Ports: k8s.ContainerPorts(8080), MainPort: lo.ToPtr(8080), }, ServiceType: lo.ToPtr("ClusterIP"), @@ -220,7 +220,7 @@ func TestSimpleContainer_SecurityAndNetworkingConfiguration(t *testing.T) { IngressContainer: &k8s.CloudRunContainer{ Name: "secure-app", - Ports: []int{8443}, + Ports: k8s.ContainerPorts(8443), MainPort: lo.ToPtr(8443), }, ServiceType: lo.ToPtr("LoadBalancer"), @@ -417,7 +417,7 @@ func TestSimpleContainer_AutoscalingCombinations(t *testing.T) { IngressContainer: &k8s.CloudRunContainer{ Name: "test-container", - Ports: []int{8080}, + Ports: k8s.ContainerPorts(8080), MainPort: lo.ToPtr(8080), }, ServiceType: lo.ToPtr("ClusterIP"), @@ -551,7 +551,7 @@ func TestSimpleContainer_ResourceLimitsAndRequests(t *testing.T) { IngressContainer: &k8s.CloudRunContainer{ Name: "resource-container", - Ports: []int{8080}, + Ports: k8s.ContainerPorts(8080), MainPort: lo.ToPtr(8080), }, ServiceType: lo.ToPtr("ClusterIP"), @@ -628,7 +628,7 @@ func TestSimpleContainer_ServiceTypeVariations(t *testing.T) { IngressContainer: &k8s.CloudRunContainer{ Name: "service-container", - Ports: []int{8080}, + Ports: k8s.ContainerPorts(8080), MainPort: lo.ToPtr(8080), }, ServiceType: lo.ToPtr(st.serviceType), diff --git a/pkg/clouds/pulumi/kubernetes/simple_container_edge_cases_test.go b/pkg/clouds/pulumi/kubernetes/simple_container_edge_cases_test.go index 515c67e8..c01b3e75 100644 --- a/pkg/clouds/pulumi/kubernetes/simple_container_edge_cases_test.go +++ b/pkg/clouds/pulumi/kubernetes/simple_container_edge_cases_test.go @@ -75,7 +75,7 @@ func TestSimpleContainer_ExtremelyLongNames(t *testing.T) { IngressContainer: &k8s.CloudRunContainer{ Name: "test-container", - Ports: []int{8080}, + Ports: k8s.ContainerPorts(8080), MainPort: lo.ToPtr(8080), }, ServiceType: lo.ToPtr("ClusterIP"), @@ -157,7 +157,7 @@ func TestSimpleContainer_SpecialCharactersInNames(t *testing.T) { IngressContainer: &k8s.CloudRunContainer{ Name: "test-container", - Ports: []int{8080}, + Ports: k8s.ContainerPorts(8080), MainPort: lo.ToPtr(8080), }, ServiceType: lo.ToPtr("ClusterIP"), @@ -286,7 +286,7 @@ func TestSimpleContainer_ExtremeResourceValues(t *testing.T) { IngressContainer: &k8s.CloudRunContainer{ Name: "resource-container", - Ports: []int{8080}, + Ports: k8s.ContainerPorts(8080), MainPort: lo.ToPtr(8080), }, ServiceType: lo.ToPtr("ClusterIP"), @@ -384,7 +384,7 @@ func TestSimpleContainer_ExtremeScalingValues(t *testing.T) { IngressContainer: &k8s.CloudRunContainer{ Name: "test-container", - Ports: []int{8080}, + Ports: k8s.ContainerPorts(8080), MainPort: lo.ToPtr(8080), }, ServiceType: lo.ToPtr("ClusterIP"), @@ -485,7 +485,7 @@ func TestSimpleContainer_LargeVolumeConfiguration(t *testing.T) { IngressContainer: &k8s.CloudRunContainer{ Name: "volume-container", - Ports: []int{8080}, + Ports: k8s.ContainerPorts(8080), MainPort: lo.ToPtr(8080), }, ServiceType: lo.ToPtr("ClusterIP"), @@ -544,7 +544,7 @@ func TestSimpleContainer_EmptyStringFields(t *testing.T) { IngressContainer: &k8s.CloudRunContainer{ Name: "test-container", - Ports: []int{8080}, + Ports: k8s.ContainerPorts(8080), MainPort: lo.ToPtr(8080), }, ServiceType: lo.ToPtr("ClusterIP"), diff --git a/pkg/clouds/pulumi/kubernetes/simple_container_final_test.go b/pkg/clouds/pulumi/kubernetes/simple_container_final_test.go index fb7ccc36..196fcfb1 100644 --- a/pkg/clouds/pulumi/kubernetes/simple_container_final_test.go +++ b/pkg/clouds/pulumi/kubernetes/simple_container_final_test.go @@ -39,7 +39,7 @@ func TestSimpleContainer_CreationSuccess(t *testing.T) { IngressContainer: &k8s.CloudRunContainer{ Name: "test-container", - Ports: []int{8080}, + Ports: k8s.ContainerPorts(8080), MainPort: lo.ToPtr(8080), }, ServiceType: lo.ToPtr("ClusterIP"), @@ -94,7 +94,7 @@ func TestSimpleContainer_HPAIntegration(t *testing.T) { IngressContainer: &k8s.CloudRunContainer{ Name: "test-container", - Ports: []int{8080}, + Ports: k8s.ContainerPorts(8080), MainPort: lo.ToPtr(8080), }, ServiceType: lo.ToPtr("ClusterIP"), @@ -154,7 +154,7 @@ func TestSimpleContainer_VPAIntegration(t *testing.T) { IngressContainer: &k8s.CloudRunContainer{ Name: "test-container", - Ports: []int{8080}, + Ports: k8s.ContainerPorts(8080), MainPort: lo.ToPtr(8080), }, ServiceType: lo.ToPtr("ClusterIP"), @@ -212,7 +212,7 @@ func TestSimpleContainer_IngressIntegration(t *testing.T) { IngressContainer: &k8s.CloudRunContainer{ Name: "test-container", - Ports: []int{8080}, + Ports: k8s.ContainerPorts(8080), MainPort: lo.ToPtr(8080), }, ServiceType: lo.ToPtr("ClusterIP"), @@ -266,7 +266,7 @@ func TestSimpleContainer_PersistentVolumeIntegration(t *testing.T) { IngressContainer: &k8s.CloudRunContainer{ Name: "test-container", - Ports: []int{8080}, + Ports: k8s.ContainerPorts(8080), MainPort: lo.ToPtr(8080), }, ServiceType: lo.ToPtr("ClusterIP"), diff --git a/pkg/clouds/pulumi/kubernetes/simple_container_parentenv_test.go b/pkg/clouds/pulumi/kubernetes/simple_container_parentenv_test.go index 5428e360..346cb96e 100644 --- a/pkg/clouds/pulumi/kubernetes/simple_container_parentenv_test.go +++ b/pkg/clouds/pulumi/kubernetes/simple_container_parentenv_test.go @@ -95,7 +95,7 @@ func TestNewSimpleContainer_WithParentEnv(t *testing.T) { IngressContainer: &k8s.CloudRunContainer{ Name: tt.serviceName, MainPort: lo.ToPtr(8080), - Ports: []int{8080}, + Ports: k8s.ContainerPorts(8080), }, GenerateCaddyfileEntry: false, KubeProvider: nil, @@ -155,7 +155,7 @@ func TestNewSimpleContainer_WithHPAAndParentEnv(t *testing.T) { IngressContainer: &k8s.CloudRunContainer{ Name: "api", MainPort: lo.ToPtr(8080), - Ports: []int{8080}, + Ports: k8s.ContainerPorts(8080), Resources: &k8s.Resources{ Requests: map[string]string{ "cpu": "100m", @@ -216,7 +216,7 @@ func TestNewSimpleContainer_WithVPAAndParentEnv(t *testing.T) { IngressContainer: &k8s.CloudRunContainer{ Name: "web", MainPort: lo.ToPtr(8080), - Ports: []int{8080}, + Ports: k8s.ContainerPorts(8080), }, VPA: &k8s.VPAConfig{ Enabled: true, @@ -280,7 +280,7 @@ func TestNewSimpleContainer_MultipleCustomStacks(t *testing.T) { IngressContainer: &k8s.CloudRunContainer{ Name: "api", MainPort: lo.ToPtr(8080), - Ports: []int{8080}, + Ports: k8s.ContainerPorts(8080), }, GenerateCaddyfileEntry: false, KubeProvider: nil, diff --git a/pkg/clouds/pulumi/kubernetes/simple_container_test.go b/pkg/clouds/pulumi/kubernetes/simple_container_test.go index 7212b645..efc17ea8 100644 --- a/pkg/clouds/pulumi/kubernetes/simple_container_test.go +++ b/pkg/clouds/pulumi/kubernetes/simple_container_test.go @@ -47,7 +47,7 @@ func createBasicTestArgs() *SimpleContainerArgs { NodeSelector: map[string]string{}, IngressContainer: &k8s.CloudRunContainer{ Name: "test-container", - Ports: []int{8080}, + Ports: k8s.ContainerPorts(8080), MainPort: lo.ToPtr(8080), }, ServiceType: lo.ToPtr("ClusterIP"), diff --git a/pkg/clouds/pulumi/kubernetes/vpa_container_policies_test.go b/pkg/clouds/pulumi/kubernetes/vpa_container_policies_test.go index 1015ed6f..e0e030f6 100644 --- a/pkg/clouds/pulumi/kubernetes/vpa_container_policies_test.go +++ b/pkg/clouds/pulumi/kubernetes/vpa_container_policies_test.go @@ -30,7 +30,7 @@ func renderVPAContainerPolicies(t *testing.T, vpa *k8s.VPAConfig) []resource.Pro Prefix: "/", Replicas: 2, IngressContainer: &k8s.CloudRunContainer{ - Name: "myapp", MainPort: lo.ToPtr(8080), Ports: []int{8080}, + Name: "myapp", MainPort: lo.ToPtr(8080), Ports: k8s.ContainerPorts(8080), }, Log: logger.New(), Containers: []corev1.ContainerArgs{ From 3a45d05744e147366fd82f4e14559b5e46428ce1 Mon Sep 17 00:00:00 2001 From: Ruslan Date: Wed, 15 Jul 2026 15:44:12 +0300 Subject: [PATCH 2/4] feat(k8s): expose non-ingress ports + fix GKE mixed-protocol LB PR #362 threaded UDP protocol + a stack-level serviceType, but the Service was still built solely from the ingress container's ports, and on GKE a mixed TCP+UDP LoadBalancer failed to provision (LoadBalancerMixedProtocolNotSupported). Both blocked self-hosting a LiveKit SFU alongside the HTTP backend in one stack. - Service ports now include non-ingress run containers' ports (they share the pod's network namespace and the Service selects the pod), so a sidecar SFU's 7880/7882 land on the same Service as backend:3000. Ingress ports win on a port-number clash; ports are deduped. - When the emitted LoadBalancer Service mixes TCP and UDP, annotate it with cloud.google.com/l4-rbs: "enabled" so GKE uses the RBS/NEG-based NetLB, which supports mixed protocols (verified live: one external IP fronting 7880/7881 TCP + 7882/UDP). No config passthrough exists for Service annotations, so this must be set by sc. Tests: dedupePorts, hasMixedProtocols, and a non-ingress-UDP Service case. Co-Authored-By: Claude Opus 4.8 (1M context) Signed-off-by: Ilya Sadykov --- pkg/clouds/pulumi/kubernetes/deployment.go | 16 ++++ pkg/clouds/pulumi/kubernetes/ports.go | 30 ++++++++ pkg/clouds/pulumi/kubernetes/ports_test.go | 75 +++++++++++++++++++ .../pulumi/kubernetes/simple_container.go | 46 ++++++++---- 4 files changed, 154 insertions(+), 13 deletions(-) diff --git a/pkg/clouds/pulumi/kubernetes/deployment.go b/pkg/clouds/pulumi/kubernetes/deployment.go index 41e32846..2800b78b 100644 --- a/pkg/clouds/pulumi/kubernetes/deployment.go +++ b/pkg/clouds/pulumi/kubernetes/deployment.go @@ -210,6 +210,21 @@ func DeploySimpleContainer(ctx *sdk.Context, args Args, opts ...sdk.ResourceOpti // Merge secret environment variables from Args with those from Deployment config mergedSecretEnvs := lo.Assign(secretEnvs, args.SecretEnvs) + // Ports declared by non-ingress run containers, so the Service can publish + // them too (they share the pod's network namespace). Lets serviceType: + // LoadBalancer expose e.g. a self-hosted SFU sidecar's UDP media port. + ingressContainerName := "" + if args.Deployment.IngressContainer != nil { + ingressContainerName = args.Deployment.IngressContainer.Name + } + var extraServicePorts []k8s.ContainerPort + for _, c := range args.Images { + if c.Container.Name == ingressContainerName { + continue + } + extraServicePorts = append(extraServicePorts, c.Container.Ports...) + } + args.Params.Log.Warn(ctx.Context(), "configure simple container deployment for %q in %q", stackName, stackEnv) sc, err := NewSimpleContainer(ctx, &SimpleContainerArgs{ KubeProvider: args.KubeProvider, @@ -223,6 +238,7 @@ func DeploySimpleContainer(ctx *sdk.Context, args Args, opts ...sdk.ResourceOpti Deployment: deploymentName, ScEnv: stackEnv, IngressContainer: args.Deployment.IngressContainer, + ExtraServicePorts: extraServicePorts, Domain: lo.FromPtr(args.Deployment.StackConfig).Domain, Prefix: lo.FromPtr(args.Deployment.StackConfig).Prefix, ProxyKeepPrefix: lo.FromPtr(args.Deployment.StackConfig).ProxyKeepPrefix, diff --git a/pkg/clouds/pulumi/kubernetes/ports.go b/pkg/clouds/pulumi/kubernetes/ports.go index 2594dfca..002c39c6 100644 --- a/pkg/clouds/pulumi/kubernetes/ports.go +++ b/pkg/clouds/pulumi/kubernetes/ports.go @@ -17,6 +17,36 @@ func isUDP(proto string) bool { return strings.EqualFold(proto, "UDP") } +// dedupePorts removes duplicate ports by port number, keeping the first +// occurrence (callers pass ingress ports first so they win over ports +// contributed by sibling containers). +func dedupePorts(ports []k8s.ContainerPort) []k8s.ContainerPort { + seen := make(map[int]bool, len(ports)) + result := make([]k8s.ContainerPort, 0, len(ports)) + for _, p := range ports { + if seen[p.Port] { + continue + } + seen[p.Port] = true + result = append(result, p) + } + return result +} + +// hasMixedProtocols reports whether the port set contains both a TCP (default) +// and a UDP port, i.e. it would produce a mixed-protocol Service. +func hasMixedProtocols(ports []k8s.ContainerPort) bool { + var tcp, udp bool + for _, p := range ports { + if isUDP(p.Protocol) { + udp = true + } else { + tcp = true + } + } + return tcp && udp +} + // portProtocol returns the protocol declared for the given port number on the // container, or "" (TCP) when the port is not declared. func portProtocol(container k8s.CloudRunContainer, port int) string { diff --git a/pkg/clouds/pulumi/kubernetes/ports_test.go b/pkg/clouds/pulumi/kubernetes/ports_test.go index a0a3aad9..d5770307 100644 --- a/pkg/clouds/pulumi/kubernetes/ports_test.go +++ b/pkg/clouds/pulumi/kubernetes/ports_test.go @@ -48,6 +48,32 @@ func livekitPorts() []k8s.ContainerPort { } } +func TestDedupePorts(t *testing.T) { + RegisterTestingT(t) + + // Duplicate port numbers are dropped; the first occurrence wins. + out := dedupePorts([]k8s.ContainerPort{ + {Port: 3000}, + {Port: 7880}, + {Port: 7880, Protocol: "UDP"}, // dup port number -> dropped + {Port: 7882, Protocol: "UDP"}, + }) + Expect(out).To(HaveLen(3)) + Expect(out[0].Port).To(Equal(3000)) + Expect(out[1].Port).To(Equal(7880)) + Expect(out[1].Protocol).To(Equal("")) // first (TCP) kept over the later UDP dup + Expect(out[2].Port).To(Equal(7882)) +} + +func TestHasMixedProtocols(t *testing.T) { + RegisterTestingT(t) + + Expect(hasMixedProtocols(nil)).To(BeFalse()) + Expect(hasMixedProtocols([]k8s.ContainerPort{{Port: 80}, {Port: 443}})).To(BeFalse()) + Expect(hasMixedProtocols([]k8s.ContainerPort{{Port: 7882, Protocol: "UDP"}})).To(BeFalse()) + Expect(hasMixedProtocols([]k8s.ContainerPort{{Port: 7880}, {Port: 7882, Protocol: "UDP"}})).To(BeTrue()) +} + func TestToContainerPorts_ThreadsProtocol(t *testing.T) { RegisterTestingT(t) @@ -182,3 +208,52 @@ func TestSimpleContainer_UDPLoadBalancer(t *testing.T) { Expect(err).ToNot(HaveOccurred()) } + +func TestSimpleContainer_ExtraServicePorts_NonIngressUDP(t *testing.T) { + RegisterTestingT(t) + + mocks := NewSimpleContainerMocks() + + err := pulumi.RunErr(func(ctx *pulumi.Context) error { + args := &SimpleContainerArgs{ + Namespace: "voice-test", + Service: "voice", + ScEnv: "test", + Deployment: "voice-deployment", + Replicas: 1, + Log: logger.New(), + + // The HTTP backend is the ingress container (TCP only). + IngressContainer: &k8s.CloudRunContainer{ + Name: "backend", + Ports: []k8s.ContainerPort{{Port: 3000}}, + MainPort: lo.ToPtr(3000), + }, + ServiceType: lo.ToPtr("LoadBalancer"), + // The SFU sidecar's ports (incl. UDP media) come from a non-ingress + // container and must still land on the Service. + ExtraServicePorts: livekitPorts(), + + Containers: []corev1.ContainerArgs{ + { + Name: sdk.String("backend"), + Image: sdk.String("backend:latest"), + Ports: toContainerPorts([]k8s.ContainerPort{{Port: 3000}}), + }, + { + Name: sdk.String("livekit"), + Image: sdk.String("livekit/livekit-server:latest"), + Ports: toContainerPorts(livekitPorts()), + }, + }, + } + + sc, err := NewSimpleContainer(ctx, args) + Expect(err).ToNot(HaveOccurred(), "a Service mixing the ingress TCP port and a sidecar's UDP port should be created") + Expect(sc).ToNot(BeNil()) + Expect(sc.Service).ToNot(BeNil(), "the non-ingress UDP port must still produce a Service") + return nil + }, pulumi.WithMocks("project", "stack", mocks)) + + Expect(err).ToNot(HaveOccurred()) +} diff --git a/pkg/clouds/pulumi/kubernetes/simple_container.go b/pkg/clouds/pulumi/kubernetes/simple_container.go index ab6da796..537ed3ca 100644 --- a/pkg/clouds/pulumi/kubernetes/simple_container.go +++ b/pkg/clouds/pulumi/kubernetes/simple_container.go @@ -44,6 +44,9 @@ const ( AnnotationPrefix = "simple-container.com/prefix" AnnotationPort = "simple-container.com/port" AnnotationEnv = "simple-container.com/env" + // AnnotationGCPL4RBS opts a Service into GKE's RBS/NEG-based external NetLB, + // which (unlike the default target-pool NetLB) supports mixed TCP+UDP ports. + AnnotationGCPL4RBS = "cloud.google.com/l4-rbs" // Standard Kubernetes labels - using hyphens instead of dots for GCP compatibility // Kubernetes allows dots in label prefixes, but GCP labels do not @@ -119,15 +122,20 @@ type SimpleContainerArgs struct { PriorityClassName *string `json:"priorityClassName" yaml:"priorityClassName"` // Kubernetes PriorityClass for pod scheduling and preemption IngressContainer *k8s.CloudRunContainer `json:"ingressContainer" yaml:"ingressContainer"` ServiceType *string `json:"serviceType" yaml:"serviceType"` - ExternalTrafficPolicy *string `json:"externalTrafficPolicy" yaml:"externalTrafficPolicy"` - ProvisionIngress bool `json:"provisionIngress" yaml:"provisionIngress"` - Headers *k8s.Headers `json:"headers" yaml:"headers"` - Volumes []k8s.SimpleTextVolume `json:"volumes" yaml:"volumes"` - SecretVolumes []k8s.SimpleTextVolume `json:"secretVolumes" yaml:"secretVolumes"` - PersistentVolumes []k8s.PersistentVolume `json:"persistentVolumes" yaml:"persistentVolumes"` - EphemeralVolumes []k8s.GenericEphemeralVolume `json:"ephemeralVolumes" yaml:"ephemeralVolumes"` // Generic ephemeral volumes for large temp storage - VPA *k8s.VPAConfig `json:"vpa" yaml:"vpa"` - Scale *k8s.Scale `json:"scale" yaml:"scale"` + // ExtraServicePorts are ports declared by non-ingress run containers. They + // share the pod's network namespace, so publishing them on the Service (which + // selects the pod) exposes them too — e.g. a self-hosted SFU's UDP media port + // in a sidecar alongside the HTTP ingress container. + ExtraServicePorts []k8s.ContainerPort `json:"extraServicePorts" yaml:"extraServicePorts"` + ExternalTrafficPolicy *string `json:"externalTrafficPolicy" yaml:"externalTrafficPolicy"` + ProvisionIngress bool `json:"provisionIngress" yaml:"provisionIngress"` + Headers *k8s.Headers `json:"headers" yaml:"headers"` + Volumes []k8s.SimpleTextVolume `json:"volumes" yaml:"volumes"` + SecretVolumes []k8s.SimpleTextVolume `json:"secretVolumes" yaml:"secretVolumes"` + PersistentVolumes []k8s.PersistentVolume `json:"persistentVolumes" yaml:"persistentVolumes"` + EphemeralVolumes []k8s.GenericEphemeralVolume `json:"ephemeralVolumes" yaml:"ephemeralVolumes"` // Generic ephemeral volumes for large temp storage + VPA *k8s.VPAConfig `json:"vpa" yaml:"vpa"` + Scale *k8s.Scale `json:"scale" yaml:"scale"` Log logger.Logger // ... @@ -827,9 +835,21 @@ ${proto}://${domain} { serviceAnnotations[AnnotationCaddyfileEntry] = caddyfileEntry } - servicePorts := corev1.ServicePortArray{} - if args.IngressContainer != nil { - servicePorts = toServicePorts(lo.FromPtr(args.IngressContainer).Ports) + // Publish the ingress container's ports plus any extra ports declared by + // sibling run containers (same pod / shared network namespace), so a + // non-ingress service (e.g. a self-hosted SFU) can be exposed on the same + // Service. Ingress ports come first so they win on a port-number clash. + svcPortSpecs := append([]k8s.ContainerPort{}, lo.FromPtr(args.IngressContainer).Ports...) + svcPortSpecs = append(svcPortSpecs, args.ExtraServicePorts...) + svcPortSpecs = dedupePorts(svcPortSpecs) + servicePorts := toServicePorts(svcPortSpecs) + + // GKE's default target-pool NetLB rejects a Service mixing TCP and UDP + // (LoadBalancerMixedProtocolNotSupported); its RBS/NEG-based NetLB supports + // it. Opt in only when we actually emit a mixed-protocol LoadBalancer, so one + // external IP fronts both TCP signaling and UDP media. + if lo.FromPtr(args.ServiceType) == "LoadBalancer" && hasMixedProtocols(svcPortSpecs) { + serviceAnnotations[AnnotationGCPL4RBS] = "enabled" } // Build the Pulumi-input annotation map. The caddyfile-entry value, if // any, is an Output that resolves the namespace placeholder against the @@ -844,7 +864,7 @@ ${proto}://${domain} { serviceAnnotationsInput[AnnotationCaddyfileEntry] = caddyfileEntryAnnotation } var service *corev1.Service - if len(lo.FromPtr(args.IngressContainer).Ports) > 0 { + if len(svcPortSpecs) > 0 { service, err = corev1.NewService(ctx, sanitizedService, &corev1.ServiceArgs{ Metadata: &metav1.ObjectMetaArgs{ Name: sdk.String(sanitizedService), From 889aeb42425a289dcfb370f5d4066db06825987a Mon Sep 17 00:00:00 2001 From: Ruslan Date: Wed, 15 Jul 2026 20:18:52 +0300 Subject: [PATCH 3/4] feat(k8s): thread externalTrafficPolicy from cloudExtras to the service MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Service applied args.ExternalTrafficPolicy (simple_container.go) and the DeploymentArgs carried it, but nothing populated it for a client stack — the field only existed on CaddyConfig, and CloudExtras had no way to set it. So `cloudExtras.externalTrafficPolicy` was silently dropped and the Service stayed Cluster. Thread it like serviceType: CloudExtras -> DeploymentConfig -> kubeArgs -> Service. Needed for self-hosted WebRTC media behind the LB: externalTrafficPolicy: Local preserves the LB IP (and client source IP) on the return path, so clients accept the SFU downlink. Cluster SNATs to a node IP, which WebRTC rejects (dtls timeout, call drops ~2 min in). Co-Authored-By: Claude Opus 4.8 (1M context) Signed-off-by: Ilya Sadykov --- pkg/clouds/gcloud/gke_autopilot.go | 13 +++++++------ pkg/clouds/k8s/kube_run.go | 18 ++++++++++++------ pkg/clouds/k8s/types.go | 4 ++++ pkg/clouds/pulumi/kubernetes/kube_run.go | 17 +++++++++-------- 4 files changed, 32 insertions(+), 20 deletions(-) diff --git a/pkg/clouds/gcloud/gke_autopilot.go b/pkg/clouds/gcloud/gke_autopilot.go index bf2a2f12..2897cd4d 100644 --- a/pkg/clouds/gcloud/gke_autopilot.go +++ b/pkg/clouds/gcloud/gke_autopilot.go @@ -107,12 +107,13 @@ func ToGkeAutopilotConfig(tpl any, composeCfg compose.Config, stackCfg *api.Stac deployCfg.DisruptionBudget = k8sCloudExtras.DisruptionBudget deployCfg.NodeSelector = k8sCloudExtras.NodeSelector deployCfg.Tolerations = k8sCloudExtras.Tolerations - deployCfg.VPA = k8sCloudExtras.VPA // Extract VPA configuration from CloudExtras - deployCfg.ReadinessProbe = k8sCloudExtras.ReadinessProbe // Extract global readiness probe configuration - deployCfg.LivenessProbe = k8sCloudExtras.LivenessProbe // Extract global liveness probe configuration - deployCfg.StartupProbe = k8sCloudExtras.StartupProbe // Extract global startup probe configuration - deployCfg.PriorityClassName = k8sCloudExtras.PriorityClassName // Extract PriorityClass for pod scheduling and preemption - deployCfg.ServiceType = k8sCloudExtras.ServiceType // Extract Service type override (e.g. LoadBalancer for UDP) + deployCfg.VPA = k8sCloudExtras.VPA // Extract VPA configuration from CloudExtras + deployCfg.ReadinessProbe = k8sCloudExtras.ReadinessProbe // Extract global readiness probe configuration + deployCfg.LivenessProbe = k8sCloudExtras.LivenessProbe // Extract global liveness probe configuration + deployCfg.StartupProbe = k8sCloudExtras.StartupProbe // Extract global startup probe configuration + deployCfg.PriorityClassName = k8sCloudExtras.PriorityClassName // Extract PriorityClass for pod scheduling and preemption + deployCfg.ServiceType = k8sCloudExtras.ServiceType // Extract Service type override (e.g. LoadBalancer for UDP) + deployCfg.ExternalTrafficPolicy = k8sCloudExtras.ExternalTrafficPolicy // e.g. Local, required for WebRTC return media deployCfg.TopologySpreadConstraints = k8sCloudExtras.TopologySpreadConstraints // Process affinity rules and merge with existing NodeSelector if needed diff --git a/pkg/clouds/k8s/kube_run.go b/pkg/clouds/k8s/kube_run.go index 5a539ddd..c3cbbd40 100644 --- a/pkg/clouds/k8s/kube_run.go +++ b/pkg/clouds/k8s/kube_run.go @@ -35,6 +35,11 @@ type CloudExtras struct { // fronted by the shared Caddy ingress). Set to "LoadBalancer" to expose the service's ports // directly, which is required for non-HTTP protocols such as UDP that Caddy cannot proxy. ServiceType *string `json:"serviceType,omitempty" yaml:"serviceType,omitempty"` + // ExternalTrafficPolicy sets the LoadBalancer Service's externalTrafficPolicy. "Local" is + // required for WebRTC media behind the LB: it preserves the LB IP (and the client source IP) + // on the return path, so clients accept the SFU's downlink. The default ("Cluster") SNATs to + // a node IP, which WebRTC rejects (dtls timeout, call drops). + ExternalTrafficPolicy *string `json:"externalTrafficPolicy,omitempty" yaml:"externalTrafficPolicy,omitempty"` TopologySpreadConstraints []TopologySpreadConstraint `json:"topologySpreadConstraints" yaml:"topologySpreadConstraints"` } @@ -203,12 +208,13 @@ func ToKubernetesRunConfig(tpl any, composeCfg compose.Config, stackCfg *api.Sta deployCfg.RollingUpdate = k8sCloudExtras.RollingUpdate deployCfg.DisruptionBudget = k8sCloudExtras.DisruptionBudget deployCfg.NodeSelector = k8sCloudExtras.NodeSelector - deployCfg.VPA = k8sCloudExtras.VPA // Extract VPA configuration from CloudExtras - deployCfg.ReadinessProbe = k8sCloudExtras.ReadinessProbe // Extract global readiness probe configuration - deployCfg.LivenessProbe = k8sCloudExtras.LivenessProbe // Extract global liveness probe configuration - deployCfg.EphemeralVolumes = k8sCloudExtras.EphemeralVolumes // Extract generic ephemeral volumes configuration - deployCfg.PriorityClassName = k8sCloudExtras.PriorityClassName // Extract PriorityClass for pod scheduling and preemption - deployCfg.ServiceType = k8sCloudExtras.ServiceType // Extract Service type override (e.g. LoadBalancer for UDP) + deployCfg.VPA = k8sCloudExtras.VPA // Extract VPA configuration from CloudExtras + deployCfg.ReadinessProbe = k8sCloudExtras.ReadinessProbe // Extract global readiness probe configuration + deployCfg.LivenessProbe = k8sCloudExtras.LivenessProbe // Extract global liveness probe configuration + deployCfg.EphemeralVolumes = k8sCloudExtras.EphemeralVolumes // Extract generic ephemeral volumes configuration + deployCfg.PriorityClassName = k8sCloudExtras.PriorityClassName // Extract PriorityClass for pod scheduling and preemption + deployCfg.ServiceType = k8sCloudExtras.ServiceType // Extract Service type override (e.g. LoadBalancer for UDP) + deployCfg.ExternalTrafficPolicy = k8sCloudExtras.ExternalTrafficPolicy // e.g. Local, required for WebRTC return media deployCfg.TopologySpreadConstraints = k8sCloudExtras.TopologySpreadConstraints // Process affinity rules and merge with existing NodeSelector if needed diff --git a/pkg/clouds/k8s/types.go b/pkg/clouds/k8s/types.go index 45cbc0da..33a14fab 100644 --- a/pkg/clouds/k8s/types.go +++ b/pkg/clouds/k8s/types.go @@ -37,6 +37,10 @@ type DeploymentConfig struct { EphemeralVolumes []GenericEphemeralVolume `json:"ephemeralVolumes" yaml:"ephemeralVolumes"` // Generic ephemeral volumes for large temp storage PriorityClassName *string `json:"priorityClassName" yaml:"priorityClassName"` // Kubernetes PriorityClass for pod scheduling and preemption ServiceType *string `json:"serviceType,omitempty" yaml:"serviceType,omitempty"` // Kubernetes Service type for the service (default: ClusterIP; LoadBalancer to expose UDP/non-HTTP ports directly) + // ExternalTrafficPolicy for the service's LoadBalancer (default: Cluster). "Local" preserves + // the LB IP and client source IP on the return path — required for WebRTC media behind the LB + // (Cluster SNATs to a node IP, which clients reject: dtls timeout). + ExternalTrafficPolicy *string `json:"externalTrafficPolicy,omitempty" yaml:"externalTrafficPolicy,omitempty"` TopologySpreadConstraints []TopologySpreadConstraint `json:"topologySpreadConstraints" yaml:"topologySpreadConstraints"` } diff --git a/pkg/clouds/pulumi/kubernetes/kube_run.go b/pkg/clouds/pulumi/kubernetes/kube_run.go index 5aaac5db..7549e7a5 100644 --- a/pkg/clouds/pulumi/kubernetes/kube_run.go +++ b/pkg/clouds/pulumi/kubernetes/kube_run.go @@ -143,14 +143,15 @@ func KubeRun(ctx *sdk.Context, stack api.Stack, input api.ResourceInput, params Annotations: map[string]string{ "pulumi.com/patchForce": "true", }, - NodeSelector: nodeSelector, - Affinity: kubeRunInput.Deployment.Affinity, - VPA: kubeRunInput.Deployment.VPA, // Pass VPA configuration from DeploymentConfig - ReadinessProbe: kubeRunInput.Deployment.ReadinessProbe, // Pass global readiness probe configuration - LivenessProbe: kubeRunInput.Deployment.LivenessProbe, // Pass global liveness probe configuration - StartupProbe: kubeRunInput.Deployment.StartupProbe, // Pass global startup probe configuration - ServiceType: kubeRunInput.Deployment.ServiceType, // Pass Service type override (e.g. LoadBalancer for UDP) - EphemeralSize: lo.FromPtr(kubeRunInput.Deployment.StackConfig).Size.Ephemeral, + NodeSelector: nodeSelector, + Affinity: kubeRunInput.Deployment.Affinity, + VPA: kubeRunInput.Deployment.VPA, // Pass VPA configuration from DeploymentConfig + ReadinessProbe: kubeRunInput.Deployment.ReadinessProbe, // Pass global readiness probe configuration + LivenessProbe: kubeRunInput.Deployment.LivenessProbe, // Pass global liveness probe configuration + StartupProbe: kubeRunInput.Deployment.StartupProbe, // Pass global startup probe configuration + ServiceType: kubeRunInput.Deployment.ServiceType, // Pass Service type override (e.g. LoadBalancer for UDP) + ExternalTrafficPolicy: kubeRunInput.Deployment.ExternalTrafficPolicy, // e.g. Local, required for WebRTC return media + EphemeralSize: lo.FromPtr(kubeRunInput.Deployment.StackConfig).Size.Ephemeral, } params.Log.Info(ctx.Context(), "🔍 DEBUG: kubeArgs.Affinity passed to DeploySimpleContainer: %+v", kubeArgs.Affinity) From 509b914ecc3920c018b978d1fa1f8065168ebc8f Mon Sep 17 00:00:00 2001 From: universe-ops Date: Wed, 15 Jul 2026 21:38:59 +0300 Subject: [PATCH 4/4] fix(gke): thread externalTrafficPolicy on the autopilot path too MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The preceding commit wired CloudExtras.ExternalTrafficPolicy through the custom-stack path (kube_run.go) but the GKE Autopilot path (gke_autopilot_stack.go) only forwarded ServiceType, so the Service's externalTrafficPolicy was silently dropped to the default "Cluster" on Autopilot. That SNATs to a node IP and breaks WebRTC return media — exactly what "Local" fixes — so the setting must reach the Autopilot kubeArgs too. Mirror the ServiceType line; no other behavior change. Co-Authored-By: Claude Opus 4.8 Signed-off-by: Ilya Sadykov --- pkg/clouds/pulumi/gcp/gke_autopilot_stack.go | 19 ++++++++++--------- 1 file changed, 10 insertions(+), 9 deletions(-) diff --git a/pkg/clouds/pulumi/gcp/gke_autopilot_stack.go b/pkg/clouds/pulumi/gcp/gke_autopilot_stack.go index 2e9aa468..28f99e10 100644 --- a/pkg/clouds/pulumi/gcp/gke_autopilot_stack.go +++ b/pkg/clouds/pulumi/gcp/gke_autopilot_stack.go @@ -186,15 +186,16 @@ func GkeAutopilotStack(ctx *sdk.Context, stack api.Stack, input api.ResourceInpu Annotations: map[string]string{ "pulumi.com/patchForce": "true", }, - NodeSelector: gkeAutopilotInput.Deployment.NodeSelector, - Affinity: gkeAutopilotInput.Deployment.Affinity, - Tolerations: gkeAutopilotInput.Deployment.Tolerations, - VPA: gkeAutopilotInput.Deployment.VPA, // Pass VPA configuration to Kubernetes deployment - ReadinessProbe: gkeAutopilotInput.Deployment.ReadinessProbe, // Pass global readiness probe configuration - LivenessProbe: gkeAutopilotInput.Deployment.LivenessProbe, // Pass global liveness probe configuration - StartupProbe: gkeAutopilotInput.Deployment.StartupProbe, // Pass global startup probe configuration - ServiceType: gkeAutopilotInput.Deployment.ServiceType, // Pass Service type override (e.g. LoadBalancer for UDP) - EphemeralSize: ephemeralSize, + NodeSelector: gkeAutopilotInput.Deployment.NodeSelector, + Affinity: gkeAutopilotInput.Deployment.Affinity, + Tolerations: gkeAutopilotInput.Deployment.Tolerations, + VPA: gkeAutopilotInput.Deployment.VPA, // Pass VPA configuration to Kubernetes deployment + ReadinessProbe: gkeAutopilotInput.Deployment.ReadinessProbe, // Pass global readiness probe configuration + LivenessProbe: gkeAutopilotInput.Deployment.LivenessProbe, // Pass global liveness probe configuration + StartupProbe: gkeAutopilotInput.Deployment.StartupProbe, // Pass global startup probe configuration + ServiceType: gkeAutopilotInput.Deployment.ServiceType, // Pass Service type override (e.g. LoadBalancer for UDP) + ExternalTrafficPolicy: gkeAutopilotInput.Deployment.ExternalTrafficPolicy, // e.g. Local, required for WebRTC return media on Autopilot + EphemeralSize: ephemeralSize, } params.Log.Info(ctx.Context(), "🔍 DEBUG: kubeArgs.Affinity passed to DeploySimpleContainer: %+v", kubeArgs.Affinity)