diff --git a/.github/workflows/conformance.yaml b/.github/workflows/conformance.yaml index f6cb044..4ce8704 100644 --- a/.github/workflows/conformance.yaml +++ b/.github/workflows/conformance.yaml @@ -21,7 +21,7 @@ env: CARGO_TERM_COLOR: always CONTAINER_ENGINE: docker KIND_CLUSTER_NAME: praxis-conformance - PRAXIS_IMAGE: ghcr.io/praxis-proxy/praxis:0.5.2 + PRAXIS_IMAGE: ghcr.io/praxis-proxy/praxis:0.7.5 OPERATOR_IMAGE: praxis-operator:dev jobs: diff --git a/.github/workflows/integration.yaml b/.github/workflows/integration.yaml index f20597d..acea72b 100644 --- a/.github/workflows/integration.yaml +++ b/.github/workflows/integration.yaml @@ -29,7 +29,7 @@ env: KIND_CLUSTER_NAME: ci V: ${{ inputs.debug && '1' || '' }} OPERATOR_IMAGE: praxis-operator:dev - PRAXIS_IMAGE: ghcr.io/praxis-proxy/praxis:0.5.2 + PRAXIS_IMAGE: ghcr.io/praxis-proxy/praxis:0.7.5 jobs: # --------------------------------------------------------------------------- diff --git a/.github/workflows/rollout.yaml b/.github/workflows/rollout.yaml new file mode 100644 index 0000000..d09e595 --- /dev/null +++ b/.github/workflows/rollout.yaml @@ -0,0 +1,123 @@ +name: Rollout + +on: + schedule: + - cron: "23 3 * * *" + workflow_dispatch: + push: + tags: + - "v*" + pull_request: + branches: [main] + paths: + - ".github/workflows/rollout.yaml" + - "Cargo.toml" + - "Cargo.lock" + - "Containerfile" + - "Containerfile.*" + - "src/**" + - "deploy/**" + +concurrency: + group: rollout-${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: ${{ github.ref != 'refs/heads/main' }} + +permissions: {} + +env: + CARGO_TERM_COLOR: always + CONTAINER_ENGINE: docker + +jobs: + kind: + runs-on: ubuntu-24.04 + timeout-minutes: 75 + permissions: + contents: read + steps: + - name: Check out operator + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + + - name: Check out praxis + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + repository: praxis-proxy/praxis + path: praxis + ref: f5a79aa36bebbaf542ce109c81144354926f52a9 + persist-credentials: false + + - name: Set up Rust + uses: praxis-proxy/conventions/.github/actions/setup-rust@f671c1215f89562d286310a6cabdcbaa5da9605c # v0.4.0 + + - name: Install KIND + run: | + curl --fail --location --show-error --silent \ + --output ./kind-linux-amd64.sha256sum \ + https://kind.sigs.k8s.io/dl/v0.27.0/kind-linux-amd64.sha256sum + curl --fail --location --show-error --silent \ + --output ./kind-linux-amd64 \ + https://kind.sigs.k8s.io/dl/v0.27.0/kind-linux-amd64 + sha256sum --check ./kind-linux-amd64.sha256sum + chmod +x ./kind-linux-amd64 + sudo mv ./kind-linux-amd64 /usr/local/bin/kind + + - name: Install kubectl + run: | + curl --fail --location --show-error --silent \ + --output ./kubectl.sha256 \ + "https://dl.k8s.io/release/v1.33.3/bin/linux/amd64/kubectl.sha256" + curl --fail --location --show-error --silent \ + --output ./kubectl \ + "https://dl.k8s.io/release/v1.33.3/bin/linux/amd64/kubectl" + echo "$(cat ./kubectl.sha256) kubectl" | sha256sum --check --strict - + chmod +x ./kubectl + sudo mv ./kubectl /usr/local/bin/kubectl + + - name: Run rollout suite (source image) + working-directory: praxis + run: >- + cargo xtask rollout + --manifest-dir ../deploy/rollout-test + --operator-dir .. + --keep-cluster + + - name: Pull the deployed default Praxis image + id: published + run: | + published_praxis_image=$(sed -n 's/^const DEFAULT_PRAXIS_IMAGE: &str = "\(.*\)";$/\1/p' src/context.rs) + test -n "$published_praxis_image" + docker pull "$published_praxis_image" + echo "image=$published_praxis_image" >> "$GITHUB_OUTPUT" + + - name: Start the published-image case with a fresh cluster + run: kind delete cluster --name praxis-rollout + + - name: Run rollout suite (published default image) + working-directory: praxis + env: + PRAXIS_RELEASE_IMAGE: ${{ steps.published.outputs.image }} + run: >- + cargo xtask rollout + --manifest-dir ../deploy/rollout-test + --operator-dir .. + --image "$PRAXIS_RELEASE_IMAGE" + --operator-image localhost/praxis-operator:rollout + --skip-build + --keep-cluster + + - name: Dump cluster state + if: failure() + run: | + kubectl --request-timeout=30s --context kind-praxis-rollout -n praxis get pods -o wide || true + kubectl --request-timeout=30s --context kind-praxis-rollout get nodes -o wide || true + kubectl --request-timeout=30s --context kind-praxis-rollout get events -A --sort-by=.lastTimestamp || true + kubectl --request-timeout=30s --context kind-praxis-rollout get crd gatewayclasses.gateway.networking.k8s.io gateways.gateway.networking.k8s.io httproutes.gateway.networking.k8s.io || true + + - name: Delete test cluster + if: always() + run: | + if command -v kind >/dev/null 2>&1; then + kind delete cluster --name praxis-rollout + fi diff --git a/CHANGELOG.md b/CHANGELOG.md index 6968c1d..2eb1fec 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,19 +12,35 @@ The format follows [Keep a Changelog], and this project adheres to ### Added +- A five-second preStop delay for data-plane pods so Service endpoints can + be removed before the proxy stops accepting connections. + - Health, readiness and metrics endpoints on port 8080, with matching probes on the operator Deployment. - Leader election over a coordination `Lease`, so the operator can run more than one replica safely. - Kubernetes events explaining why a Gateway was rejected. -- Configurable data-plane replicas via the `praxis.sh/replicas` - annotation, a `PodDisruptionBudget`, and pod spread across nodes. +- Configurable positive data-plane replica counts via the + `praxis.sh/replicas` annotation, a `PodDisruptionBudget`, zero-unavailable + rolling updates, and zone and hostname pod spread. +- Kubernetes reference and rollout-test assets, plus the GitHub Actions + workflow that runs the Praxis rollout acceptance suite against them. - `method` and `queryParams` route matches are now detected and reported rather than silently ignored. - Listener protocol and hostname conflict detection. ### Fixed +- Return HTTP 404 for unmatched requests on Gateway listeners without routes. + +- Remove stale proxy children when no usable Gateway listeners remain, including + conflict-only configurations, and refresh HTTPRoute parent rejection statuses + after listeners move to or from the reserved metrics port. + +- Gateway listeners on the reserved health/metrics port 9902 now report + `PortUnavailable` without blocking other valid listeners. +- Generated data-plane configuration now targets Praxis 0.7.5, including its + metrics-only admin endpoint and routeless-listener support. - Status writes no longer restamp unchanged conditions, which had kept the operator in a permanent reconcile loop. - Endpoint weight distribution no longer overflows `i32` and aborts the diff --git a/deploy/rollout-test/backend.yaml b/deploy/rollout-test/backend.yaml new file mode 100644 index 0000000..e249843 --- /dev/null +++ b/deploy/rollout-test/backend.yaml @@ -0,0 +1,126 @@ +apiVersion: apps/v1 +kind: Deployment +metadata: + name: backend + namespace: praxis + labels: + app.kubernetes.io/name: backend +spec: + replicas: 2 + selector: + matchLabels: + app.kubernetes.io/name: backend + template: + metadata: + labels: + app.kubernetes.io/name: backend + spec: + # Keep fixture backends stable while worker nodes hosting proxies drain. + nodeSelector: + node-role.kubernetes.io/control-plane: "" + tolerations: + - key: node-role.kubernetes.io/control-plane + operator: Exists + effect: NoSchedule + containers: + - name: backend + image: busybox:1.37.0 + args: [httpd, -f, -p, "8080", -h, /www] + ports: + - name: http + containerPort: 8080 + readinessProbe: + httpGet: + path: /healthy + port: http + periodSeconds: 2 + volumeMounts: + - name: www + mountPath: /www + readOnly: true + volumes: + - name: www + configMap: + name: backend-content +--- +apiVersion: v1 +kind: ConfigMap +metadata: + name: backend-content + namespace: praxis +data: + config-v1: config-v1 + config-v2: config-v2 + healthy: ok + index.html: ok +--- +apiVersion: v1 +kind: Service +metadata: + name: backend + namespace: praxis +spec: + selector: + app.kubernetes.io/name: backend + ports: + - name: http + port: 8080 + targetPort: http +--- +apiVersion: apps/v1 +kind: Deployment +metadata: + name: healthy-backend + namespace: praxis + labels: + app.kubernetes.io/name: healthy-backend +spec: + replicas: 2 + selector: + matchLabels: + app.kubernetes.io/name: healthy-backend + template: + metadata: + labels: + app.kubernetes.io/name: healthy-backend + spec: + # Keep fixture backends stable while worker nodes hosting proxies drain. + nodeSelector: + node-role.kubernetes.io/control-plane: "" + tolerations: + - key: node-role.kubernetes.io/control-plane + operator: Exists + effect: NoSchedule + containers: + - name: backend + image: busybox:1.37.0 + args: [httpd, -f, -p, "8080", -h, /www] + ports: + - name: http + containerPort: 8080 + readinessProbe: + httpGet: + path: /healthy + port: http + periodSeconds: 2 + volumeMounts: + - name: www + mountPath: /www + readOnly: true + volumes: + - name: www + configMap: + name: backend-content +--- +apiVersion: v1 +kind: Service +metadata: + name: healthy-backend + namespace: praxis +spec: + selector: + app.kubernetes.io/name: healthy-backend + ports: + - name: http + port: 8080 + targetPort: http diff --git a/deploy/rollout-test/gateway.yaml b/deploy/rollout-test/gateway.yaml new file mode 100644 index 0000000..3953ae8 --- /dev/null +++ b/deploy/rollout-test/gateway.yaml @@ -0,0 +1,25 @@ +apiVersion: gateway.networking.k8s.io/v1 +kind: Gateway +metadata: + name: rollout + namespace: praxis + annotations: + praxis.sh/replicas: "2" +spec: + gatewayClassName: praxis + listeners: + - name: http + port: 8080 + protocol: HTTP + allowedRoutes: + namespaces: + from: All + - name: https + port: 8443 + protocol: HTTPS + tls: + certificateRefs: + - name: praxis-tls + allowedRoutes: + namespaces: + from: All diff --git a/deploy/rollout-test/httproute.yaml b/deploy/rollout-test/httproute.yaml new file mode 100644 index 0000000..946872d --- /dev/null +++ b/deploy/rollout-test/httproute.yaml @@ -0,0 +1,32 @@ +apiVersion: gateway.networking.k8s.io/v1 +kind: HTTPRoute +metadata: + name: rollout + namespace: praxis +spec: + parentRefs: + - name: rollout + rules: + - matches: + - path: + type: Exact + value: /config-version + filters: + - type: URLRewrite + urlRewrite: + path: + type: ReplaceFullPath + replaceFullPath: /config-v1 + backendRefs: + - name: backend + port: 8080 + - matches: + - path: + type: Exact + value: /healthy + backendRefs: + - name: healthy-backend + port: 8080 + - backendRefs: + - name: backend + port: 8080 diff --git a/deploy/rollout-test/kustomization.yaml b/deploy/rollout-test/kustomization.yaml new file mode 100644 index 0000000..21ad875 --- /dev/null +++ b/deploy/rollout-test/kustomization.yaml @@ -0,0 +1,11 @@ +apiVersion: kustomize.config.k8s.io/v1beta1 +kind: Kustomization +resources: + - ../rbac.yaml + - ../deployment.yaml + - ../gatewayclass.yaml + - backend.yaml + - gateway.yaml + - httproute.yaml + - namespace.yaml + - service.yaml diff --git a/deploy/rollout-test/namespace.yaml b/deploy/rollout-test/namespace.yaml new file mode 100644 index 0000000..d144684 --- /dev/null +++ b/deploy/rollout-test/namespace.yaml @@ -0,0 +1,7 @@ +apiVersion: v1 +kind: Namespace +metadata: + name: praxis + labels: + app.kubernetes.io/name: praxis + praxis-rollout/test: "true" diff --git a/deploy/rollout-test/service.yaml b/deploy/rollout-test/service.yaml new file mode 100644 index 0000000..3f1950b --- /dev/null +++ b/deploy/rollout-test/service.yaml @@ -0,0 +1,18 @@ +apiVersion: v1 +kind: Service +metadata: + name: praxis-rollout-test + namespace: praxis +spec: + type: NodePort + selector: + app.kubernetes.io/instance: rollout + ports: + - name: http + port: 8080 + targetPort: 8080 + nodePort: 30080 + - name: https + port: 8443 + targetPort: 8443 + nodePort: 30443 diff --git a/docs/architecture.md b/docs/architecture.md index b56b4f8..354a743 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -91,8 +91,9 @@ and filters. ### Deployment -A single-replica Deployment running the Praxis proxy -container. Key properties: +A Deployment running one Praxis container per pod. It has one replica by +default. The `praxis.sh/replicas` Gateway annotation sets another positive +replica count. Key properties: - **Image:** controlled by the `PRAXIS_IMAGE` environment variable on the operator @@ -100,14 +101,16 @@ container. Key properties: `/etc/praxis/config.yaml` - **TLS volumes:** mounts each referenced TLS Secret as a read-only volume +- **Shutdown:** a five-second native preStop sleep lets Service endpoint + removal propagate before SIGTERM, within the 15-second grace period. - **Health probes:** liveness (`/healthy`) and - readiness (`/ready`) on admin port 9901 + readiness (`/ready`) on metrics port 9902 - **Security:** runs as UID 100, read-only filesystem, drops all capabilities except `NET_BIND_SERVICE`, RuntimeDefault seccomp profile - **Config hash:** a `praxis.sh/config-hash` annotation - on the pod template triggers a rolling update - whenever the configuration changes + on the pod template covers listener and TLS topology. Topology changes + trigger a rolling update; route and filter changes hot-reload. ### Service @@ -133,9 +136,9 @@ Praxis's native YAML configuration: | Request redirect | `redirect` filter | The config always includes a `request_id` filter -(generates trace IDs) and a `router` filter (handles -routing decisions). Backend load balancing defaults to -round-robin. +(generates trace IDs). Routed listeners also include a +`router` filter for routing decisions; routeless listeners +omit it and terminate with a 404 static response. Backend load balancing defaults to round-robin. ## Labeling diff --git a/docs/configuration.md b/docs/configuration.md index 403ffa3..d8ffc61 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -12,15 +12,23 @@ files for the operator itself. Container image used for data-plane pods. Set on the operator Deployment. -- **Default:** `ghcr.io/praxis-proxy/praxis:latest` -- **Deploy manifest:** `ghcr.io/praxis-proxy/praxis:0.3.0` +- **Default:** `ghcr.io/praxis-proxy/praxis:0.7.5` +- **Deploy manifest:** `__PRAXIS_IMAGE__` (substituted during deployment) ```yaml env: - name: PRAXIS_IMAGE - value: "ghcr.io/praxis-proxy/praxis:0.3.0" + value: "ghcr.io/praxis-proxy/praxis:0.7.5" ``` +The data plane reserves port 9902 for health and metrics. A Gateway listener +requesting that port reports `Accepted=False` with reason `PortUnavailable`; +other valid listeners remain supported. If no usable listeners remain, including +when all listeners conflict, the operator removes the Gateway’s managed proxy +resources and recreates them when a usable listener is restored. Attached +HTTPRoutes update their parent status when listeners move to or from the +reserved port. + ### `RUST_LOG` Controls operator log verbosity using the standard @@ -92,7 +100,7 @@ permissions: | Resource | Request | Limit | |----------|---------|-------| -| CPU | 100m | none | +| CPU | 100m | 1 | | Memory | 64Mi | 256Mi | These are defaults in the deploy manifests and @@ -153,10 +161,11 @@ directly. - **ConfigMap:** `praxis-{gateway-name}` with key `config.yaml` -- **Admin port:** 9901 (health probes, diagnostics) -- **Config changes:** a SHA-256 hash annotation - (`praxis.sh/config-hash`) on the pod template - triggers a rolling update when config changes +- **Metrics port:** 9902 (health probes, metrics) +- **Config changes:** route and filter changes update + the ConfigMap for hot reload; a topology hash + (`praxis.sh/config-hash`) rolls pods only when + listener or TLS Secret topology changes See [Architecture](architecture.md) for details on how Gateway API resources are translated to Praxis diff --git a/src/config/cluster.rs b/src/config/cluster.rs index 7ce714b..7015487 100644 --- a/src/config/cluster.rs +++ b/src/config/cluster.rs @@ -48,24 +48,18 @@ pub enum PraxisEndpoint { /// Generates a cluster name from namespace, service, and port. /// -/// Returns a cluster name in the format `{namespace}~{service}~{port}`. +/// Returns a cluster name in the format `{namespace}_{service}_{port}`. /// -/// Uses `~` as separator because it cannot appear in Kubernetes namespace -/// or service names (DNS subdomain charset), preventing ambiguity. +/// Underscores are accepted by Praxis but cannot occur in Kubernetes +/// namespace or Service names, so different backends cannot collide. /// /// ``` /// use praxis_operator::config::cluster::cluster_name; /// -/// assert_eq!(cluster_name("default", "echo", 8080), "default~echo~8080"); -/// -/// // The format is only injective because `~` is outside the DNS -/// // subdomain charset. Feed it a name containing the separator and -/// // two distinct backends do collide — which is why the API server's -/// // own validation is what makes this key safe: -/// assert_eq!(cluster_name("a", "b~c", 80), cluster_name("a~b", "c", 80)); +/// assert_eq!(cluster_name("default", "echo", 8080), "default_echo_8080"); /// ``` pub fn cluster_name(namespace: &str, service: &str, port: i32) -> String { - format!("{namespace}~{service}~{port}") + format!("{namespace}_{service}_{port}") } /// Builds a Praxis cluster configuration. @@ -102,13 +96,22 @@ mod tests { #[test] fn test_cluster_name() { let name = cluster_name("default", "my-svc", 8080); - assert_eq!(name, "default~my-svc~8080", "cluster name format should match"); + assert_eq!(name, "default_my-svc_8080", "cluster name format should match"); } #[test] fn test_cluster_name_different_namespace() { let name = cluster_name("kube-system", "dns", 53); - assert_eq!(name, "kube-system~dns~53", "cluster name should include namespace"); + assert_eq!(name, "kube-system_dns_53", "cluster name should include namespace"); + } + + #[test] + fn cluster_names_distinguish_hyphenated_namespaces_and_services() { + assert_ne!( + cluster_name("a-b", "c", 80), + cluster_name("a", "b-c", 80), + "backends in different namespaces must not share traffic through a name collision" + ); } #[test] diff --git a/src/config/generate.rs b/src/config/generate.rs index 6764905..e79aed1 100644 --- a/src/config/generate.rs +++ b/src/config/generate.rs @@ -30,18 +30,18 @@ pub struct PraxisConfig { /// Filter chains with routing and processing filters. pub filter_chains: Vec, - /// Insecure options for container deployments. - pub insecure_options: PraxisInsecureOptions, - /// Listeners (proxy entry points). pub listeners: Vec, + + /// Runtime configuration. + pub runtime: PraxisRuntime, } /// Admin endpoint configuration. #[derive(Debug, Clone, Serialize, PartialEq)] pub struct PraxisAdmin { - /// Admin bind address. - pub address: String, + /// Health and metrics bind address. + pub metrics_address: String, } /// Named filter chain. @@ -54,11 +54,11 @@ pub struct PraxisFilterChain { pub filters: Vec, } -/// Insecure options (for container deployments). +/// Runtime configuration. #[derive(Debug, Clone, Serialize, PartialEq)] -pub struct PraxisInsecureOptions { - /// Allow admin endpoint on public interface. - pub allow_public_admin: bool, +pub struct PraxisRuntime { + /// Worker threads per service. + pub threads: usize, } // ----------------------------------------------------------------------------- @@ -88,13 +88,11 @@ pub fn assemble_config( Ok(PraxisConfig { admin: PraxisAdmin { - address: "0.0.0.0:9901".to_owned(), + metrics_address: "0.0.0.0:9902".to_owned(), }, filter_chains, - insecure_options: PraxisInsecureOptions { - allow_public_admin: true, - }, listeners, + runtime: PraxisRuntime { threads: 1 }, }) } @@ -103,8 +101,9 @@ pub fn assemble_config( /// The chain runs `request_id`, the route filters that answer a request /// without an upstream, `router` (with matching routes), the route /// filters that reshape a routed request, and finally `load_balancer` -/// (with embedded clusters). [`RouteFilters`] explains why the router -/// splits the two groups. +/// (with embedded clusters). Routeless listeners terminate with a +/// 404 response because Praxis rejects an empty router route list. [`RouteFilters`] explains why the router splits the +/// two groups. /// /// # Errors /// @@ -117,27 +116,23 @@ fn build_filter_chain( listener_hostnames: &std::collections::HashMap>, ) -> yaml_serde::Result { let name = &listener.name; - - let section_names = &listener.merged_section_names; - let filtered: Vec<_> = routes - .iter() - .filter(|route| { - route.listener_names.is_empty() - || route - .listener_names - .iter() - .any(|ln| ln.is_none() || ln.as_deref().is_some_and(|sn| section_names.contains(&sn.to_owned()))) - }) - .cloned() - .collect(); - let mut scoped = inject_listener_hostnames(&filtered, listener_hostnames); - sort_routes_by_specificity_owned(&mut scoped); + let scoped = scoped_routes(listener, routes, listener_hostnames); let scoped_refs: Vec<_> = scoped.iter().collect(); let mut filters = vec![PraxisFilterEntry { filter: "request_id".to_owned(), config: yaml_serde::Value::Null, }]; + if scoped_refs.is_empty() { + filters.push(PraxisFilterEntry { + filter: "static_response".to_owned(), + config: yaml_serde::to_value(serde_json::json!({"status": 404}))?, + }); + return Ok(PraxisFilterChain { + name: format!("{name}-chain"), + filters, + }); + } filters.extend_from_slice(&route_filters.terminating); filters.push(build_router_filter(&scoped_refs)?); filters.extend_from_slice(&route_filters.transforming); @@ -149,19 +144,26 @@ fn build_filter_chain( }) } -/// Injects listener hostnames into routes that lack one. -/// -/// Each route's `listener_names` determines which Gateway listener it -/// targets. If that listener has a hostname constraint, routes without -/// an HTTPRoute-level hostname inherit it. -fn inject_listener_hostnames( +/// Returns the routes scoped to one listener, ordered by specificity. +fn scoped_routes( + listener: &PraxisListener, routes: &[PraxisRoute], listener_hostnames: &std::collections::HashMap>, ) -> Vec { - routes + let section_names = &listener.merged_section_names; + let mut scoped: Vec<_> = routes .iter() + .filter(|route| { + route.listener_names.is_empty() + || route.listener_names.iter().any(|name| { + name.as_deref() + .is_none_or(|section| section_names.iter().any(|candidate| candidate == section)) + }) + }) .flat_map(|route| expand_route_hostnames(route, listener_hostnames)) - .collect() + .collect(); + sort_routes_by_specificity_owned(&mut scoped); + scoped } /// Expands a route into one copy per targeted listener hostname. @@ -327,11 +329,11 @@ mod tests { path_prefix: "/api/".to_owned(), host: None, headers: None, - cluster: "default~my-svc~8080".to_owned(), + cluster: "default-my-svc-8080".to_owned(), listener_names: vec![], }; - let cluster = build_cluster("default~my-svc~8080", vec!["10.0.0.1:8080".to_owned()], None); + let cluster = build_cluster("default-my-svc-8080", vec!["10.0.0.1:8080".to_owned()], None); let config = assemble_config( vec![listener], @@ -342,12 +344,12 @@ mod tests { ) .unwrap(); - assert_eq!(config.admin.address, "0.0.0.0:9901", "admin address should be set"); - assert_eq!(config.listeners.len(), 1, "should have one listener"); - assert!( - config.insecure_options.allow_public_admin, - "allow_public_admin should be true" + assert_eq!( + config.admin.metrics_address, "0.0.0.0:9902", + "health and metrics should bind to the dedicated metrics address" ); + assert_eq!(config.listeners.len(), 1, "should have one listener"); + assert_eq!(config.runtime.threads, 1, "runtime should use one worker thread"); assert_eq!(config.filter_chains.len(), 1, "should have one filter chain"); let chain = &config.filter_chains[0]; @@ -387,11 +389,11 @@ mod tests { path_prefix: "/".to_owned(), host: None, headers: None, - cluster: "default~svc~80".to_owned(), + cluster: "default-svc-80".to_owned(), listener_names: vec![], }; - let cluster = build_cluster("default~svc~80", vec!["10.0.0.1:80".to_owned()], None); + let cluster = build_cluster("default-svc-80", vec!["10.0.0.1:80".to_owned()], None); let config = assemble_config( vec![listener], @@ -405,16 +407,20 @@ mod tests { let yaml = yaml_serde::to_string(&config).expect("config should serialize to YAML"); assert!(yaml.contains("admin:"), "YAML should contain admin section"); - assert!(yaml.contains("0.0.0.0:9901"), "YAML should contain admin address"); + assert!( + yaml.contains("metrics_address: 0.0.0.0:9902"), + "YAML should contain metrics address" + ); assert!(yaml.contains("listeners:"), "YAML should contain listeners section"); assert!( yaml.contains("filter_chains:"), "YAML should contain filter_chains section" ); assert!( - yaml.contains("insecure_options:"), - "YAML should contain insecure_options section" + !yaml.contains("insecure_options"), + "public metrics must not require an insecure option" ); + assert!(yaml.contains("threads: 1"), "YAML should contain runtime thread count"); assert!(yaml.contains("request_id"), "YAML should contain request_id filter"); assert!(yaml.contains("router"), "YAML should contain router filter"); assert!( @@ -454,11 +460,11 @@ mod tests { path_prefix: "/".to_owned(), host: None, headers: None, - cluster: "default~svc~80".to_owned(), + cluster: "default-svc-80".to_owned(), listener_names: vec![], }; - let cluster = build_cluster("default~svc~80", vec!["10.0.0.1:80".to_owned()], None); + let cluster = build_cluster("default-svc-80", vec!["10.0.0.1:80".to_owned()], None); let config = assemble_config( vec![http_listener, https_listener], @@ -557,10 +563,24 @@ mod tests { let config = assemble_config(vec![listener], &[], &[], &RouteFilters::default(), &Default::default()).unwrap(); let chain = &config.filter_chains[0]; - let lb_config = &chain.filters[2].config; - let clusters_value = lb_config.get("clusters").expect("lb config should have clusters key"); - let clusters_seq = clusters_value.as_sequence().expect("clusters should be a sequence"); - assert!(clusters_seq.is_empty(), "clusters should be empty"); + assert_eq!( + chain.filters.len(), + 2, + "a routeless listener should omit the router and load balancer" + ); + assert_eq!( + chain.filters[0].filter, "request_id", + "the routeless chain should retain request IDs" + ); + assert_eq!( + chain.filters[1].filter, "static_response", + "unmatched traffic must terminate" + ); + assert_eq!( + chain.filters[1].config["status"].as_u64(), + Some(404), + "routeless listeners return 404" + ); } #[test] @@ -633,19 +653,16 @@ mod tests { #[test] fn test_router_matches_wildcards_at_any_depth() { - let listener = PraxisListener { - name: "http".to_owned(), - address: "0.0.0.0:80".to_owned(), - protocol: None, - filter_chains: vec!["http-chain".to_owned()], - hostname: None, - merged_section_names: vec![], - tls: None, + let route = PraxisRoute { + path: None, + path_prefix: "/".to_owned(), + host: Some("*.example.com".to_owned()), + headers: None, + cluster: "cluster".to_owned(), + listener_names: vec![], }; + let router = build_router_filter(&[&route]).unwrap(); - let config = assemble_config(vec![listener], &[], &[], &RouteFilters::default(), &Default::default()).unwrap(); - - let router = &config.filter_chains[0].filters[1]; assert_eq!( router.config.get("multi_level_subdomain_matching"), Some(&yaml_serde::Value::Bool(true)), @@ -680,8 +697,24 @@ mod tests { terminating: vec![redirect], transforming: vec![headers], }; + let route = PraxisRoute { + path: None, + path_prefix: "/".to_owned(), + host: None, + headers: None, + cluster: "cluster".to_owned(), + listener_names: vec![], + }; + let cluster = build_cluster("cluster", vec!["10.0.0.1:80".to_owned()], None); - let config = assemble_config(vec![listener], &[], &[], &route_filters, &Default::default()).unwrap(); + let config = assemble_config( + vec![listener], + std::slice::from_ref(&route), + std::slice::from_ref(&cluster), + &route_filters, + &Default::default(), + ) + .unwrap(); let chain = &config.filter_chains[0]; let names: Vec<_> = chain.filters.iter().map(|f| f.filter.as_str()).collect(); @@ -694,6 +727,65 @@ mod tests { ); } + #[test] + fn routeless_listeners_do_not_inherit_terminating_filters() { + let routed = PraxisListener { + name: "routed".to_owned(), + address: "0.0.0.0:80".to_owned(), + protocol: None, + filter_chains: vec!["routed-chain".to_owned()], + hostname: None, + merged_section_names: vec!["routed".to_owned()], + tls: None, + }; + let routeless = PraxisListener { + name: "routeless".to_owned(), + address: "0.0.0.0:8081".to_owned(), + protocol: None, + filter_chains: vec!["routeless-chain".to_owned()], + hostname: None, + merged_section_names: vec!["routeless".to_owned()], + tls: None, + }; + let route = PraxisRoute { + path: None, + path_prefix: "/".to_owned(), + host: None, + headers: None, + cluster: "cluster".to_owned(), + listener_names: vec![Some("routed".to_owned())], + }; + let cluster = build_cluster("cluster", vec!["10.0.0.1:80".to_owned()], None); + let route_filters = RouteFilters { + terminating: vec![PraxisFilterEntry { + filter: "redirect".to_owned(), + config: yaml_serde::Value::Null, + }], + transforming: Vec::new(), + }; + + let config = assemble_config( + vec![routed, routeless], + std::slice::from_ref(&route), + std::slice::from_ref(&cluster), + &route_filters, + &Default::default(), + ) + .unwrap(); + let routeless = config + .filter_chains + .iter() + .find(|chain| chain.name == "routeless-chain") + .expect("routeless chain"); + let names: Vec<_> = routeless.filters.iter().map(|filter| filter.filter.as_str()).collect(); + + assert_eq!( + names, + vec!["request_id", "static_response"], + "terminating filters from another listener's route must not answer routeless traffic" + ); + } + #[test] fn test_route_expanded_for_multiple_listener_hostnames() { let mut hostnames = std::collections::HashMap::new(); diff --git a/src/config/routing.rs b/src/config/routing.rs index 4b6545c..84002ac 100644 --- a/src/config/routing.rs +++ b/src/config/routing.rs @@ -12,6 +12,7 @@ use gateway_api::{ use serde::Serialize; use tracing::warn; +use super::cluster::cluster_name as backend_cluster_name; use crate::gateway_api::{ attachment::AttachedRoute, hostname::intersect_hostnames, reference_grant::is_reference_allowed, route_validation::validate_route, @@ -81,7 +82,7 @@ pub struct PraxisFilterEntry { /// `Service` endpoints into cluster `IP:port` pairs. #[derive(Debug, Clone, PartialEq)] pub struct BackendRef { - /// Cluster name (format: `namespace~service~port`). + /// Cluster name (format: `namespace_service_port`). pub cluster_name: String, /// Kubernetes namespace of the backend Service. @@ -321,9 +322,15 @@ fn merged_cluster_name_from_refs(backends: &[&HttpRouteRulesBackendRefs], route_ } backends .iter() - .map(|backend| single_cluster_name(backend, route_ns)) + .map(|backend| { + format!( + "{}_w{}", + single_cluster_name(backend, route_ns), + backend.weight.unwrap_or(1) + ) + }) .collect::>() - .join("+") + .join("__") } /// Builds a cluster name for a single backend ref. @@ -331,7 +338,7 @@ fn single_cluster_name(backend: &HttpRouteRulesBackendRefs, route_ns: &str) -> S let namespace = backend.namespace.as_deref().unwrap_or(route_ns); let service = backend.name.as_str(); let port = backend.port.unwrap_or(80); - format!("{namespace}~{service}~{port}") + backend_cluster_name(namespace, service, port) } /// Resolves a single backend ref into a [`BackendRef`]. @@ -355,7 +362,7 @@ fn process_backend_ref( let namespace = backend.namespace.as_deref().unwrap_or(route_ns); let service = backend.name.as_str(); let port = backend.port.unwrap_or(80); - let dedup_key = format!("{cluster_name}:{namespace}~{service}~{port}"); + let dedup_key = format!("{cluster_name}:{}", backend_cluster_name(namespace, service, port)); if !seen.insert(dedup_key) { return None; @@ -538,6 +545,39 @@ mod tests { use super::*; + #[test] + fn merged_cluster_names_are_valid_and_distinguish_weights() { + let first = HttpRouteRulesBackendRefs { + name: "api".to_owned(), + port: Some(80), + weight: Some(1), + ..Default::default() + }; + let mut second = HttpRouteRulesBackendRefs { + name: "canary".to_owned(), + ..first.clone() + }; + let equal_split = merged_cluster_name_from_refs(&[&first, &second], "default"); + second.weight = Some(3); + let unequal_split = merged_cluster_name_from_refs(&[&first, &second], "default"); + + assert!( + equal_split + .chars() + .all(|character| character.is_ascii_alphanumeric() || matches!(character, '_' | '-')), + "multi-backend names must pass Praxis name validation: {equal_split}" + ); + assert_ne!( + equal_split, unequal_split, + "rules with different traffic splits must not merge their endpoints" + ); + assert_ne!( + equal_split, + single_cluster_name(&first, "default"), + "merged and single clusters must be distinct" + ); + } + #[test] fn test_convert_routes_path_prefix() { let route = HTTPRoute { @@ -594,14 +634,14 @@ mod tests { for r in &praxis_routes { assert_eq!( - r.cluster, "default~my-svc~8080", + r.cluster, "default_my-svc_8080", "cluster should match namespace-service-port" ); } assert_eq!(backend_refs.len(), 1, "should have one backend ref"); assert_eq!( - backend_refs[0].cluster_name, "default~my-svc~8080", + backend_refs[0].cluster_name, "default_my-svc_8080", "backend ref cluster name should match" ); assert_eq!(backend_refs[0].service, "my-svc", "backend ref service should match"); @@ -714,11 +754,11 @@ mod tests { ); assert!( - backend_refs.iter().any(|b| b.cluster_name == "default~api-svc~8080"), + backend_refs.iter().any(|b| b.cluster_name == "default_api-svc_8080"), "should have api-svc backend ref" ); assert!( - backend_refs.iter().any(|b| b.cluster_name == "default~web-svc~80"), + backend_refs.iter().any(|b| b.cluster_name == "default_web-svc_80"), "should have web-svc backend ref" ); } @@ -806,12 +846,12 @@ mod tests { assert_eq!(praxis_routes.len(), 1, "should produce one route"); assert_eq!( - praxis_routes[0].cluster, "other-ns~svc~80", + praxis_routes[0].cluster, "other-ns_svc_80", "cluster should use backend namespace" ); assert!( - backend_refs.iter().any(|b| b.cluster_name == "other-ns~svc~80"), + backend_refs.iter().any(|b| b.cluster_name == "other-ns_svc_80"), "backend refs should use backend namespace" ); assert_eq!( @@ -1383,7 +1423,7 @@ mod tests { let (praxis_routes, backend_refs) = convert_routes(&routes, &HashMap::new(), &[]); assert_eq!( - praxis_routes[0].cluster, "infra~local~80", + praxis_routes[0].cluster, "infra_local_80", "the cluster name must describe only the backends that actually resolve, or the \ generated name promises endpoints the cluster will never contain" ); diff --git a/src/context.rs b/src/context.rs index 98f1ffb..d4d4702 100644 --- a/src/context.rs +++ b/src/context.rs @@ -29,8 +29,8 @@ pub const GATEWAY_FINALIZER: &str = "gateway.praxis.sh/finalizer"; /// them. pub const FIELD_MANAGER: &str = "praxis-operator"; -/// Admin port on the Praxis data-plane container. -pub const ADMIN_PORT: i32 = 9901; +/// Health and metrics port on the Praxis data-plane container. +pub const ADMIN_PORT: i32 = 9902; /// Image used when `PRAXIS_IMAGE` is unset. /// @@ -38,7 +38,7 @@ pub const ADMIN_PORT: i32 = 9901; /// every data plane it creates, so `latest` would let an unreviewed /// proxy release change the behaviour of an unchanged operator, and the /// version CI exercises would drift from the version users run. -const DEFAULT_PRAXIS_IMAGE: &str = "ghcr.io/praxis-proxy/praxis:0.5.2"; +const DEFAULT_PRAXIS_IMAGE: &str = "ghcr.io/praxis-proxy/praxis:0.7.5"; // ----------------------------------------------------------------------------- // Data Plane Image @@ -128,6 +128,6 @@ mod tests { #[test] fn test_admin_port_matches_the_praxis_default() { - assert_eq!(ADMIN_PORT, 9901, "probes target the Praxis admin endpoint"); + assert_eq!(ADMIN_PORT, 9902, "probes target the Praxis metrics endpoint"); } } diff --git a/src/controller.rs b/src/controller.rs index f6743a0..58497cb 100644 --- a/src/controller.rs +++ b/src/controller.rs @@ -3,6 +3,7 @@ //! Kubernetes controllers for Gateway API resources. +mod child_cleanup; #[cfg(test)] mod fixtures; pub mod gateway; diff --git a/src/controller/child_cleanup.rs b/src/controller/child_cleanup.rs new file mode 100644 index 0000000..e6ec9c4 --- /dev/null +++ b/src/controller/child_cleanup.rs @@ -0,0 +1,177 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright (c) 2026 Praxis Contributors + +//! Reclaims owned data-plane resources when a Gateway has no usable listeners. + +use std::fmt::Debug; + +use gateway_api::gateways::Gateway; +use k8s_openapi::api::{ + apps::v1::Deployment, + core::v1::{ConfigMap, Service}, + policy::v1::PodDisruptionBudget, +}; +use kube::{ + Api, Resource, ResourceExt as _, + api::{DeleteParams, Preconditions}, +}; +use serde::de::DeserializeOwned; + +use crate::{ + error::{OperatorError, Result}, + resources::labels::child_name, +}; + +/// Removes only children controlled by this Gateway, tolerating prior deletion. +/// +/// Every child is attempted even when one delete fails. Stopping at the +/// first error would leave the Gateway's children half-dismantled until +/// the retry, and a second failure would extend that window again; the +/// first error is still returned so reconciliation repeats. +pub(super) async fn delete_children(client: &kube::Client, gw: &Gateway) -> Result<()> { + let namespace = gw.namespace().unwrap_or_default(); + let name = child_name(&gw.name_any()); + let owner_uid = gw.uid().ok_or(OperatorError::MissingObjectKey(".metadata.uid"))?; + let deletions = [ + delete_owned::(client, &namespace, &name, &owner_uid).await, + Box::pin(delete_owned::(client, &namespace, &name, &owner_uid)).await, + delete_owned::(client, &namespace, &name, &owner_uid).await, + delete_owned::(client, &namespace, &name, &owner_uid).await, + ]; + deletions.into_iter().collect::>>().map(|_| ()) +} + +/// Checks controller ownership and guards deletion against concurrent replacement. +async fn delete_owned(client: &kube::Client, namespace: &str, name: &str, owner_uid: &str) -> Result<()> +where + K: Resource + DeserializeOwned + Clone + Debug, + ::DynamicType: Default, +{ + let api = Api::::namespaced(client.clone(), namespace); + let Some(child) = api.get_opt(name).await? else { + return Ok(()); + }; + if !child + .owner_references() + .iter() + .any(|owner| owner.controller == Some(true) && owner.uid == owner_uid) + { + return Ok(()); + } + let params = DeleteParams { + preconditions: Some(Preconditions { + uid: Some(child.uid().ok_or(OperatorError::MissingObjectKey(".metadata.uid"))?), + resource_version: child.resource_version(), + }), + ..DeleteParams::default() + }; + match api.delete(name, ¶ms).await { + Ok(_) => Ok(()), + Err(kube::Error::Api(response)) if response.code == 404 => Ok(()), + Err(error) => Err(error.into()), + } +} + +#[cfg(test)] +mod tests { + use serde_json::json; + + use super::*; + use crate::testing; + + #[tokio::test] + async fn cleanup_deletes_only_owned_children_with_identity_preconditions() { + let service = json!({"apiVersion":"v1", "kind":"Service", "metadata": { + "name":"praxis-gw", "namespace":"infra", "uid":"service-uid", "resourceVersion":"7", + "ownerReferences":[{"apiVersion":"gateway.networking.k8s.io/v1", "kind":"Gateway", + "name":"gw", "uid":"gateway-uid", "controller":true}] + }}); + let (ctx, journal) = testing::fake_context( + vec![testing::Canned::ok("/services/praxis-gw", service)], + testing::Cached::default(), + ); + delete_owned::(&ctx.client, "infra", "praxis-gw", "gateway-uid") + .await + .unwrap(); + let requests: Vec<_> = journal + .requests() + .into_iter() + .filter(|request| request.method == "DELETE") + .collect(); + assert_eq!(requests.len(), 1, "owned resources must be removed"); + assert_eq!( + requests[0].body.as_ref().unwrap()["preconditions"], + json!({"uid":"service-uid", "resourceVersion":"7"}), + "replacement and ownership races must fail deletion" + ); + } + + #[tokio::test] + async fn cleanup_preserves_foreign_and_non_controller_resources() { + for (owner, controller) in [("someone-else", true), ("gateway-uid", false)] { + let service = json!({"apiVersion":"v1", "kind":"Service", "metadata": { + "name":"praxis-gw", "uid":"service-uid", "ownerReferences":[{ + "apiVersion":"gateway.networking.k8s.io/v1", "kind":"Gateway", "name":"gw", + "uid":owner, "controller":controller}] + }}); + let (ctx, journal) = testing::fake_context( + vec![testing::Canned::ok("/services/praxis-gw", service)], + testing::Cached::default(), + ); + delete_owned::(&ctx.client, "infra", "praxis-gw", "gateway-uid") + .await + .unwrap(); + assert!( + journal.requests().iter().all(|request| request.method != "DELETE"), + "a matching resource name is not ownership" + ); + } + } + + #[tokio::test] + async fn cleanup_attempts_every_child_even_after_a_failure() { + let (ctx, journal) = testing::fake_context(vec![testing::Canned::server_error("")], testing::Cached::default()); + let gateway = Gateway { + metadata: k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta { + name: Some("gw".to_owned()), + namespace: Some("infra".to_owned()), + uid: Some("gateway-uid".to_owned()), + ..Default::default() + }, + spec: gateway_api::gateways::GatewaySpec::default(), + status: None, + }; + + let result = delete_children(&ctx.client, &gateway).await; + + assert!( + result.is_err(), + "a delete the API server refused has to fail the reconcile so it is retried" + ); + for kind in ["/services/", "/deployments/", "/configmaps/", "/poddisruptionbudgets/"] { + assert!( + !journal.matching(kind).is_empty(), + "every owned child must be attempted, or one transient error leaves the Gateway \ + half-dismantled until the next retry: {kind}" + ); + } + } + + #[tokio::test] + async fn cleanup_tolerates_missing_resources_and_propagates_api_failures() { + let (ctx, _) = testing::fake_context(vec![], testing::Cached::default()); + delete_owned::(&ctx.client, "infra", "praxis-gw", "gateway-uid") + .await + .unwrap(); + let (ctx, _) = testing::fake_context( + vec![testing::Canned::server_error("/services/praxis-gw")], + testing::Cached::default(), + ); + assert!( + delete_owned::(&ctx.client, "infra", "praxis-gw", "gateway-uid") + .await + .is_err(), + "cleanup failures must be retried instead of reporting stale resources removed" + ); + } +} diff --git a/src/controller/gateway.rs b/src/controller/gateway.rs index fe14889..c013f8e 100644 --- a/src/controller/gateway.rs +++ b/src/controller/gateway.rs @@ -19,11 +19,11 @@ use kube::{ use serde::{Serialize, de::DeserializeOwned}; use tracing::{debug, error, info}; -use super::{gateway_status, ownership, praxis_config, rollout, route_parent_status}; +use super::{child_cleanup, gateway_status, ownership, praxis_config, rollout, route_parent_status}; use crate::{ context::{Context, GATEWAY_FINALIZER}, error::{OperatorError, Result}, - gateway_api::{attachment::AttachedRoute, conditions, protocol::ListenerProtocol, route_status}, + gateway_api::{attachment::AttachedRoute, conditions, route_status}, }; // ----------------------------------------------------------------------------- @@ -98,9 +98,8 @@ pub fn error_policy(_gw: Arc, error: &OperatorError, _ctx: Arc /// update Gateway and route statuses. /// /// Route parent statuses are set **after** child resources are applied -/// and the Deployment rollout is verified, preventing the conformance -/// test from sending traffic before the data plane has the latest -/// configuration. +/// and the initial Deployment rollout is verified. Accepted means the +/// configuration is valid; `ConfigMap` projection and hot reload are asynchronous. async fn apply(gw: Arc, ctx: &Context) -> Result { if !ownership::validate_gateway_class(&ctx.client, &gw).await? || reject_unsupported_spec(ctx, &gw).await? { return Ok(Action::await_change()); @@ -134,12 +133,8 @@ async fn apply_config_if_supported( ns: &str, grants: &[ReferenceGrant], ) -> Result { - let has_supported = gw - .spec - .listeners - .iter() - .any(|listener| ListenerProtocol::is_supported(&listener.protocol)); - if !has_supported { + if praxis_config::usable_listeners(&gw.spec.listeners).is_empty() { + child_cleanup::delete_children(client, gw).await?; return Ok(false); } diff --git a/src/controller/gateway_class.rs b/src/controller/gateway_class.rs index 05892a8..ce411f9 100644 --- a/src/controller/gateway_class.rs +++ b/src/controller/gateway_class.rs @@ -53,7 +53,7 @@ use crate::{ /// rather than appending, so conformance asked for /// `append-val-1,header-val-2` and got `header-val-2`. /// -/// Praxis 0.5.2, now the pinned data plane, appends: `request_add` +/// Praxis 0.7.5, now the pinned data plane, appends: `request_add` /// reads the existing values and combines them. Both header-modification /// features are claimed again on that basis, and the conformance suites /// they gate are no longer skipped. diff --git a/src/controller/gateway_status.rs b/src/controller/gateway_status.rs index 0f60aeb..d4afeb4 100644 --- a/src/controller/gateway_status.rs +++ b/src/controller/gateway_status.rs @@ -22,7 +22,7 @@ use tracing::{debug, info}; use super::listener_validation; use crate::{ - context::{Context, FIELD_MANAGER}, + context::{ADMIN_PORT, Context, FIELD_MANAGER}, error::Result, gateway_api::{ attachment::AttachedRoute, @@ -162,10 +162,9 @@ async fn build_listener_statuses( continue; } - let protocol_supported = ListenerProtocol::is_supported(&listener.protocol); - if !protocol_supported { + if let Some(status) = rejected_listener_status(listener, generation) { any_rejected = true; - statuses.push(unsupported_listener_status(listener, generation)); + statuses.push(status); continue; } @@ -178,6 +177,26 @@ async fn build_listener_statuses( (statuses, any_accepted, any_rejected) } +/// Reports unsupported protocols and ports reserved by the data plane. +fn rejected_listener_status(listener: &GatewayListeners, generation: i64) -> Option { + if !ListenerProtocol::is_supported(&listener.protocol) { + return Some(unsupported_listener_status(listener, generation)); + } + (listener.port == ADMIN_PORT).then(|| ListenerStatus { + name: listener.name.clone(), + attached_routes: 0, + supported_kinds: vec![], + conditions: vec![ + conditions::not_accepted( + generation, + "PortUnavailable", + "port 9902 is reserved for health and metrics", + ), + conditions::not_programmed(generation, "Invalid", "port 9902 is reserved for health and metrics"), + ], + }) +} + /// Builds a status entry for a listener conflicting with another. /// /// A conflicted listener is not accepted, not programmed, and attaches @@ -507,6 +526,38 @@ mod tests { ); } + #[tokio::test] + async fn metrics_port_is_rejected_while_other_listeners_remain_accepted() { + let (ctx, journal) = testing::fake_context(vec![gateway_response()], testing::Cached::default()); + let valid = http_listener(); + let mut reserved = valid.clone(); + reserved.name = "reserved".to_owned(); + reserved.port = ADMIN_PORT; + build_and_apply_gateway_status(&ctx, &gateway(), &[valid, reserved], &[]) + .await + .expect("status must describe the reserved listener"); + let status = patched_status(&journal); + assert_eq!( + status + .pointer("/listeners/1/conditions/0/reason") + .and_then(Value::as_str), + Some("PortUnavailable"), + "the conflicting listener must explain the reservation" + ); + assert_eq!( + status + .pointer("/listeners/0/conditions/0/status") + .and_then(Value::as_str), + Some("True"), + "other listeners must remain accepted" + ); + assert_eq!( + condition_status(&status, "Accepted"), + Some("True"), + "one valid listener keeps the Gateway accepted" + ); + } + #[tokio::test] async fn test_an_unsupported_protocol_listener_is_rejected_in_status() { let (ctx, journal) = testing::fake_context(vec![gateway_response()], testing::Cached::default()); diff --git a/src/controller/httproute.rs b/src/controller/httproute.rs index d6e0187..92a7286 100644 --- a/src/controller/httproute.rs +++ b/src/controller/httproute.rs @@ -16,12 +16,15 @@ use gateway_api::{ httproutes::{HTTPRoute, HttpRouteParentRefs}, }; use k8s_openapi::{api::core::v1::Namespace, apimachinery::pkg::apis::meta::v1::Condition}; -use kube::{Api, ResourceExt as _, runtime::controller::Action}; +use kube::{ + Api, ResourceExt as _, + runtime::{controller::Action, reflector::ObjectRef}, +}; use tracing::{debug, error, info}; use super::namespace_filter; use crate::{ - context::{CONTROLLER_NAME, Context}, + context::{ADMIN_PORT, CONTROLLER_NAME, Context}, error::{OperatorError, Result}, gateway_api::{ attachment::listener_matches_parent_ref, conditions, hostname, route_status, status_types::RouteParentStatus, @@ -155,6 +158,29 @@ async fn lookup_parent_gateway(gw_name: &str, gw_ns: &str, route_ns: &str, ctx: None } +/// Enqueues routes referencing a Gateway when its listener configuration changes. +#[must_use] +pub(crate) fn map_gateway_to_routes(gateway: &Gateway, routes: &[Arc]) -> Vec> { + let namespace = gateway.namespace().unwrap_or_default(); + let name = gateway.name_any(); + routes + .iter() + .filter(|route| { + route.spec.parent_refs.as_ref().is_some_and(|parents| { + parents.iter().any(|parent| { + route_status::is_ref_targeting_gateway( + parent, + &name, + &namespace, + route_status::route_namespace(route), + ) + }) + }) + }) + .map(|route| ObjectRef::from_obj(route.as_ref())) + .collect() +} + // ----------------------------------------------------------------------------- // Validation // ----------------------------------------------------------------------------- @@ -199,11 +225,11 @@ fn validate_listener_attachment( /// same listener. A ref naming a listener and a port that listener /// does not serve selects nothing, exactly as one naming a listener /// that does not exist does. -fn parent_ref_selects_a_listener(gw: &Gateway, parent_ref: &HttpRouteParentRefs) -> bool { +pub(super) fn parent_ref_selects_a_listener(gw: &Gateway, parent_ref: &HttpRouteParentRefs) -> bool { gw.spec .listeners .iter() - .any(|listener| listener_matches_parent_ref(listener, parent_ref)) + .any(|listener| listener.port != ADMIN_PORT && listener_matches_parent_ref(listener, parent_ref)) } /// Returns `true` when the route's namespace is allowed by listeners. @@ -238,10 +264,11 @@ fn targeted_listeners<'listen>( listeners: &'listen [GatewayListeners], section_name: Option<&str>, ) -> Vec<&'listen GatewayListeners> { - match section_name { - Some(name) => listeners.iter().filter(|ls| ls.name == name).collect(), - None => listeners.iter().collect(), - } + listeners + .iter() + .filter(|listener| listener.port != ADMIN_PORT) + .filter(|listener| section_name.is_none_or(|name| listener.name == name)) + .collect() } /// Checks if any route hostname intersects with a matching listener. @@ -293,6 +320,54 @@ mod tests { use super::*; use crate::testing; + #[test] + fn reserved_listeners_cannot_match_route_parents() { + let mut reserved = listener("reserved", Some("reserved.example.com")); + reserved.port = ADMIN_PORT; + let gw = gateway(vec![reserved, listener("valid", Some("valid.example.com"))]); + assert!( + !parent_ref_selects_a_listener(&gw, &parent_ref(Some("reserved"))), + "reserved section must be rejected" + ); + let mut parent = parent_ref(None); + parent.port = Some(ADMIN_PORT); + assert!( + !parent_ref_selects_a_listener(&gw, &parent), + "reserved port must be rejected" + ); + assert!( + parent_ref_selects_a_listener(&gw, &parent_ref(None)), + "unqualified parents retain usable siblings" + ); + assert!( + !hostnames_intersect(&route(&["reserved.example.com"]), &gw, None), + "reserved listeners cannot supply hostname matches" + ); + assert_eq!( + targeted_listeners(&gw.spec.listeners, None).len(), + 1, + "namespace validation must also exclude reserved listeners" + ); + } + + #[test] + fn gateway_changes_enqueue_only_referencing_routes() { + let gw = gateway(vec![listener("http", None)]); + let mut parent = parent_ref(None); + parent.namespace = Some("infra".to_owned()); + let matching = Arc::new(parented_route(parent.clone())); + parent.namespace = Some("elsewhere".to_owned()); + let foreign = Arc::new(parented_route(parent)); + let refs = map_gateway_to_routes(&gw, &[matching, foreign]); + assert_eq!(refs.len(), 1, "only routes naming this exact Gateway are reconciled"); + assert_eq!(refs[0].name, "route", "enqueue the route"); + assert_eq!( + refs[0].namespace.as_deref(), + Some("apps"), + "retain cross-namespace route identity" + ); + } + #[test] fn test_a_parent_ref_naming_nothing_selects_every_listener() { let gw = gateway(vec![listener("http", None)]); diff --git a/src/controller/ownership.rs b/src/controller/ownership.rs index 3897122..414c62b 100644 --- a/src/controller/ownership.rs +++ b/src/controller/ownership.rs @@ -16,7 +16,7 @@ use tracing::debug; use super::namespace_filter; use crate::{ - context::CONTROLLER_NAME, + context::{ADMIN_PORT, CONTROLLER_NAME}, error::{OperatorError, Result}, gateway_api::attachment::{self, AttachedRoute}, stores::Stores, @@ -83,8 +83,15 @@ pub(super) fn collect_routes<'route>( let ns = gw.namespace().unwrap_or_default(); let name = gw.name_any(); - let attached = attachment::attached_routes(&name, &ns, &gw.spec.listeners, all_routes); - namespace_filter::filter_routes_by_allowed_namespaces(&attached, &gw.spec.listeners, &ns, stores) + let listeners: Vec<_> = gw + .spec + .listeners + .iter() + .filter(|listener| listener.port != ADMIN_PORT) + .cloned() + .collect(); + let attached = attachment::attached_routes(&name, &ns, &listeners, all_routes); + namespace_filter::filter_routes_by_allowed_namespaces(&attached, &listeners, &ns, stores) } // ----------------------------------------------------------------------------- diff --git a/src/controller/praxis_config.rs b/src/controller/praxis_config.rs index 60fb718..21b37b8 100644 --- a/src/controller/praxis_config.rs +++ b/src/controller/praxis_config.rs @@ -6,8 +6,9 @@ //! Turns the listeners and attached routes of one Gateway into the //! configuration document the data plane reads, then applies the //! `ConfigMap`, `Deployment`, `Service`, and `PodDisruptionBudget` that carry -//! it. The config hash computed here is what tells the Gateway -//! controller whether a rollout is still in flight. +//! it. The rollout hash computed here covers only listener and TLS topology, +//! so hot-reloadable route changes update the `ConfigMap` without restarting +//! the data plane. use std::collections::{BTreeMap, HashMap, HashSet}; @@ -29,6 +30,7 @@ use crate::{ routing::{BackendRef, PraxisRoute, convert_routes, rule_has_authorized_backend}, weights::{ResolvedBackend, distribute_service_weights, sort_service_endpoints}, }, + context::ADMIN_PORT, endpoints, error::Result, gateway_api::{attachment::AttachedRoute, listener_conflict, protocol::ListenerProtocol}, @@ -62,6 +64,9 @@ pub(super) struct PraxisConfigOutput { /// TLS secret names referenced by HTTPS listeners (deduplicated). pub(super) tls_secret_names: Vec, + + /// TLS topology used to decide whether a rollout is required. + pub(super) listener_tls: Vec<(String, Option)>, } /// Converts Gateway listeners, attached routes, and resolved endpoints @@ -72,18 +77,14 @@ pub(super) async fn build_praxis_config( attached: &[AttachedRoute<'_>], grants: &[ReferenceGrant], ) -> Result { - let conflicts = listener_conflict::detect_conflicts(listeners); - let supported: Vec<_> = listeners - .iter() - .filter(|listener| ListenerProtocol::is_supported(&listener.protocol)) - .filter(|listener| !conflicts.contains_key(&listener.name)) - .collect(); + let supported = usable_listeners(listeners); let listener_hostnames = build_listener_hostname_map(&supported); let praxis_listeners = merge_listeners_by_port(&supported); let (praxis_routes, backend_refs) = convert_attached_routes(attached, &listener_hostnames, grants); let route_filters = collect_filters(attached, grants); let clusters = resolve_clusters(client, &backend_refs).await?; + let listener_tls = collect_listener_tls(&praxis_listeners); let config = assemble_config( praxis_listeners, &praxis_routes, @@ -96,9 +97,29 @@ pub(super) async fn build_praxis_config( config_yaml: yaml_serde::to_string(&config)?, listener_ports: collect_listener_ports(&supported), tls_secret_names: collect_tls_secret_names(&supported), + listener_tls, }) } +/// Selects supported, unreserved listeners that have no binding conflicts. +pub(super) fn usable_listeners(listeners: &[GatewayListeners]) -> Vec<&GatewayListeners> { + let conflicts = listener_conflict::detect_conflicts(listeners); + listeners + .iter() + .filter(|listener| ListenerProtocol::is_supported(&listener.protocol)) + .filter(|listener| listener.port != ADMIN_PORT) + .filter(|listener| !conflicts.contains_key(&listener.name)) + .collect() +} + +/// Collects the TLS definitions that cannot hot-reload in Praxis. +fn collect_listener_tls(listeners: &[PraxisListener]) -> Vec<(String, Option)> { + listeners + .iter() + .map(|listener| (listener.address.clone(), listener.tls.clone())) + .collect() +} + /// Merges Gateway listeners on the same port into a single Praxis /// listener, combining TLS certificates from all listeners in the group. fn merge_listeners_by_port(supported: &[&GatewayListeners]) -> Vec { @@ -302,7 +323,7 @@ pub(super) async fn apply_child_resources( let cm = build_configmap(&child, &ns, gw, &config_output.config_yaml)?; super::gateway::apply_resource(client, &ns, &cm).await?; - let config_hash = sha256_hex(&config_output.config_yaml); + let config_hash = rollout_hash(config_output)?; let deploy = build_deployment(&DeploymentParams { name: &child, config_hash: &config_hash, @@ -323,6 +344,16 @@ pub(super) async fn apply_child_resources( Ok(config_hash) } +/// Hashes the parts of the generated config that require a Deployment rollout. +fn rollout_hash(config_output: &PraxisConfigOutput) -> Result { + let identity = serde_json::json!({ + "listeners": config_output.listener_ports, + "tlsSecrets": config_output.tls_secret_names, + "listenerTls": config_output.listener_tls, + }); + Ok(sha256_hex(&serde_json::to_string(&identity)?)) +} + /// Converts `(name, port)` pairs into Kubernetes `ServicePort` entries. fn build_service_ports(listener_ports: &[(String, i32)]) -> Vec { listener_ports @@ -519,6 +550,56 @@ mod tests { ); } + #[test] + fn usable_listeners_exclude_conflicts_and_preserve_a_valid_sibling() { + let http = http_listener(); + let mut https = http.clone(); + https.name = "https".to_owned(); + https.protocol = "HTTPS".to_owned(); + assert!( + usable_listeners(&[http.clone(), https.clone()]).is_empty(), + "a protocol-conflicting pair contributes no usable listeners" + ); + let mut valid = http.clone(); + valid.name = "valid".to_owned(); + valid.port = 8085; + let listeners = [http, https, valid]; + let selected: Vec<_> = usable_listeners(&listeners) + .into_iter() + .map(|listener| listener.name.as_str()) + .collect(); + assert_eq!( + selected, + vec!["valid"], + "conflicting listeners must not remove an independent listener" + ); + } + + #[tokio::test] + async fn reserved_metrics_port_is_excluded_without_discarding_other_listeners() { + let (client, _) = testing::fake_client(vec![]); + let valid = http_listener(); + let mut reserved = valid.clone(); + reserved.name = "reserved".to_owned(); + reserved.port = ADMIN_PORT; + let output = build_praxis_config(&client, &[valid, reserved], &[], &[]) + .await + .expect("the valid listener must still generate configuration"); + assert_eq!( + output.listener_ports, + vec![("http".to_owned(), 80)], + "the metrics port must not become a proxy listener or Service port" + ); + let yaml: yaml_serde::Value = yaml_serde::from_str(&output.config_yaml).expect("generated YAML"); + assert_eq!( + yaml.get("listeners") + .and_then(yaml_serde::Value::as_sequence) + .map(Vec::len), + Some(1), + "only the valid proxy listener may be configured" + ); + } + #[tokio::test] async fn test_an_unsupported_listener_contributes_nothing() { let (client, _) = testing::fake_client(vec![]); @@ -571,6 +652,7 @@ mod tests { config_yaml: "listeners: []\n".to_owned(), listener_ports: vec![("http".to_owned(), 80)], tls_secret_names: vec![], + listener_tls: vec![], }; let hash = Box::pin(apply_child_resources(&client, &gateway(), &output)) @@ -579,9 +661,8 @@ mod tests { assert_eq!( hash, - sha256_hex(&output.config_yaml), - "the returned hash is what the pod template is annotated with, so it has to be the \ - hash of the config that was actually applied" + rollout_hash(&output).expect("rollout hash"), + "the returned hash must be the topology hash applied to the Deployment" ); for kind in ["/configmaps", "/deployments", "/services", "/poddisruptionbudgets"] { assert_eq!( @@ -592,6 +673,114 @@ mod tests { } } + #[test] + fn rollout_hash_keeps_hot_reload_changes_stable() { + let output = test_config_output("route: v1\n", vec![("http".to_owned(), 80)], vec![], vec![]); + let route_changed = test_config_output("route: v2\n", output.listener_ports.clone(), vec![], vec![]); + + assert_eq!( + rollout_hash(&output).expect("base hash"), + rollout_hash(&route_changed).expect("route hash"), + "route and filter changes must hot-reload without changing the pod template" + ); + } + + #[test] + fn rollout_hash_changes_for_topology_changes() { + let output = test_config_output("route: v1\n", vec![("http".to_owned(), 80)], vec![], vec![]); + let mut listener_added = test_config_output("route: v2\n", output.listener_ports.clone(), vec![], vec![]); + listener_added.listener_ports.push(("extra".to_owned(), 8081)); + let mut tls_added = test_config_output("route: v2\n", output.listener_ports.clone(), vec![], vec![]); + tls_added.tls_secret_names.push("praxis-tls".to_owned()); + + assert_ne!( + rollout_hash(&output).expect("base hash"), + rollout_hash(&listener_added).expect("listener hash"), + "a new listener port cannot hot-reload and must roll the pods" + ); + assert_ne!( + rollout_hash(&output).expect("base hash"), + rollout_hash(&tls_added).expect("TLS hash"), + "a new TLS Secret mount cannot hot-reload and must roll the pods" + ); + } + + #[tokio::test] + async fn rollout_hash_changes_when_https_hostname_changes() { + let (client, _) = testing::fake_client(vec![]); + let mut https = https_listener("https", 8443, "praxis-tls"); + https.hostname = Some("before.example.com".to_owned()); + let before = build_praxis_config(&client, &[https.clone()], &[], &[]) + .await + .expect("initial HTTPS config"); + https.hostname = Some("after.example.com".to_owned()); + let after = build_praxis_config(&client, &[https], &[], &[]) + .await + .expect("updated HTTPS config"); + assert_ne!( + rollout_hash(&before).expect("initial hash"), + rollout_hash(&after).expect("updated hash"), + "changing only an HTTPS hostname must roll pods to update SNI" + ); + assert_eq!(before.listener_ports, after.listener_ports, "ports are unchanged"); + assert_eq!( + before.tls_secret_names, after.tls_secret_names, + "Secret mounts are unchanged" + ); + } + + #[test] + fn rollout_hash_changes_when_tls_moves_between_listeners() { + let listener_ports = vec![("http".to_owned(), 80), ("https".to_owned(), 8443)]; + let tls = test_listener_tls("example.com"); + let output = test_config_output( + "route: v1\n", + listener_ports.clone(), + vec!["praxis-tls".to_owned()], + vec![ + ("0.0.0.0:80".to_owned(), None), + ("0.0.0.0:8443".to_owned(), Some(tls.clone())), + ], + ); + let tls_moved = test_config_output( + "route: v1\n", + listener_ports, + vec!["praxis-tls".to_owned()], + vec![("0.0.0.0:80".to_owned(), Some(tls)), ("0.0.0.0:8443".to_owned(), None)], + ); + + assert_ne!( + rollout_hash(&output).expect("base hash"), + rollout_hash(&tls_moved).expect("moved TLS hash"), + "moving TLS between unchanged listener ports changes topology and must roll the pods" + ); + } + + fn test_config_output( + config_yaml: &str, + listener_ports: Vec<(String, i32)>, + tls_secret_names: Vec, + listener_tls: Vec<(String, Option)>, + ) -> PraxisConfigOutput { + PraxisConfigOutput { + config_yaml: config_yaml.to_owned(), + listener_ports, + tls_secret_names, + listener_tls, + } + } + + fn test_listener_tls(hostname: &str) -> PraxisTls { + PraxisTls { + certificates: vec![PraxisCertificate { + cert_path: "/tls/praxis-tls/tls.crt".to_owned(), + key_path: "/tls/praxis-tls/tls.key".to_owned(), + server_names: Some(vec![hostname.to_owned()]), + default: None, + }], + } + } + #[tokio::test] async fn test_a_refused_apply_stops_the_rest() { let (client, journal) = testing::failing_client(); @@ -599,6 +788,7 @@ mod tests { config_yaml: "listeners: []\n".to_owned(), listener_ports: vec![], tls_secret_names: vec![], + listener_tls: vec![], }; Box::pin(apply_child_resources(&client, &gateway(), &output)) diff --git a/src/controller/route_parent_status.rs b/src/controller/route_parent_status.rs index 15e117b..e36e673 100644 --- a/src/controller/route_parent_status.rs +++ b/src/controller/route_parent_status.rs @@ -3,10 +3,9 @@ //! Writing `Accepted` and `ResolvedRefs` onto attached routes. //! -//! The Gateway controller owns route acceptance, but only once the data -//! plane is actually serving the matching configuration — otherwise a -//! client following the status would send traffic to a proxy that has -//! not caught up yet. +//! The Gateway controller owns route acceptance after applying valid configuration. +//! Acceptance does not acknowledge completion of asynchronous `ConfigMap` projection +//! or data-plane hot reload. use gateway_api::{ gateways::Gateway, @@ -31,9 +30,8 @@ use crate::{ /// /// Sets `Accepted = True` and evaluates `ResolvedRefs` for each route /// that targets this Gateway. Called by the Gateway controller **after** -/// child resources are applied and the Deployment rollout is verified, -/// so the conformance test cannot send traffic before the data plane -/// is serving the matching configuration. +/// child resources are applied and the initial Deployment rollout is verified. +/// This does not acknowledge that a hot-reloaded configuration is already serving. pub(super) async fn update_route_parent_statuses( client: &kube::Client, gw: &Gateway, @@ -41,7 +39,6 @@ pub(super) async fn update_route_parent_statuses( grants: &[ReferenceGrant], ) -> Result<()> { let gw_ns = gw.namespace().unwrap_or_default(); - let gw_name = gw.name_any(); for AttachedRoute { route, .. } in attached { let route_ns = route_status::route_namespace(route); @@ -50,17 +47,7 @@ pub(super) async fn update_route_parent_statuses( continue; }; - let statuses = build_route_statuses( - route, - parent_refs, - route_ns, - &gw_name, - &gw_ns, - generation, - client, - grants, - ) - .await; + let statuses = build_route_statuses(route, parent_refs, route_ns, gw, &gw_ns, generation, client, grants).await; if !statuses.is_empty() { route_status::apply_parent_statuses(client, route, &statuses).await?; @@ -75,7 +62,7 @@ async fn build_route_statuses( route: &HTTPRoute, parent_refs: &[HttpRouteParentRefs], route_ns: &str, - gw_name: &str, + gateway: &Gateway, gw_ns: &str, generation: i64, client: &kube::Client, @@ -85,7 +72,9 @@ async fn build_route_statuses( let mut statuses = Vec::new(); for parent_ref in parent_refs { - if !route_status::is_ref_targeting_gateway(parent_ref, gw_name, gw_ns, route_ns) { + if !route_status::is_ref_targeting_gateway(parent_ref, &gateway.name_any(), gw_ns, route_ns) + || !super::httproute::parent_ref_selects_a_listener(gateway, parent_ref) + { continue; } @@ -132,6 +121,75 @@ mod tests { use super::*; use crate::{controller::fixtures::regex_route, testing}; + #[tokio::test] + #[expect( + clippy::too_many_lines, + reason = "explicit listener fixtures verify mixed-parent status isolation" + )] + async fn acceptance_skips_a_reserved_parent_when_the_route_also_has_a_usable_parent() { + let gateway = Gateway { + metadata: ObjectMeta { + name: Some("gw".to_owned()), + namespace: Some("default".to_owned()), + ..Default::default() + }, + spec: gateway_api::gateways::GatewaySpec { + gateway_class_name: "praxis".to_owned(), + listeners: vec![ + gateway_api::gateways::GatewayListeners { + name: "reserved".to_owned(), + port: 9902, + protocol: "HTTP".to_owned(), + ..Default::default() + }, + gateway_api::gateways::GatewayListeners { + name: "valid".to_owned(), + port: 8080, + protocol: "HTTP".to_owned(), + ..Default::default() + }, + ], + ..Default::default() + }, + status: None, + }; + let route = HTTPRoute { + metadata: ObjectMeta::default(), + spec: gateway_api::httproutes::HttpRouteSpec::default(), + status: None, + }; + let parents: Vec<_> = ["reserved", "valid"] + .into_iter() + .map(|section| HttpRouteParentRefs { + name: "gw".to_owned(), + section_name: Some(section.to_owned()), + ..Default::default() + }) + .collect(); + let (context, _) = testing::fake_context(vec![], testing::Cached::default()); + let statuses = build_route_statuses( + &route, + &parents, + "default", + &gateway, + "default", + 1, + &context.client, + &[], + ) + .await; + assert_eq!( + statuses.len(), + 1, + "Gateway acceptance must not overwrite reserved-parent rejection" + ); + assert_eq!( + statuses[0].parent_ref.section_name.as_deref(), + Some("valid"), + "only the usable parent is accepted" + ); + } + #[test] fn test_validation_conditions_accepts_a_supported_route() { let conds = validation_conditions(&route_validation::RouteValidation::default(), 1); @@ -313,7 +371,12 @@ mod tests { }, spec: GatewaySpec { gateway_class_name: "praxis".to_owned(), - listeners: vec![], + listeners: vec![gateway_api::gateways::GatewayListeners { + name: "http".to_owned(), + port: 80, + protocol: "HTTP".to_owned(), + ..Default::default() + }], ..Default::default() }, status: None, diff --git a/src/lib.rs b/src/lib.rs index 81d53cd..71cb09d 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -235,9 +235,15 @@ fn managed_children() -> watcher::Config { /// Wires the `HTTPRoute` controller. /// -/// Watches all `HTTPRoute` resources and reconciles parent status entries. +/// Watches routes and their parent Gateways to reconcile attachment status changes. fn build_route_controller(client: &Client, ctx: Arc) -> impl Future { + let stores = ctx.stores.clone(); Controller::new(Api::::all(client.clone()), watcher::Config::default()) + .watches( + Api::::all(client.clone()), + watcher::Config::default(), + move |gateway| controller::httproute::map_gateway_to_routes(&gateway, &stores.routes()), + ) .shutdown_on_signal() .run( controller::httproute::reconcile, diff --git a/src/resources/configmap.rs b/src/resources/configmap.rs index a871e5e..5cf30e2 100644 --- a/src/resources/configmap.rs +++ b/src/resources/configmap.rs @@ -71,7 +71,7 @@ mod tests { "praxis-config", "default", &gateway, - "admin:\n address: 0.0.0.0:9901\n", + "admin:\n metrics_address: 0.0.0.0:9902\n", ) .unwrap(); @@ -127,7 +127,7 @@ mod tests { status: None, }; - let yaml_content = "admin:\n address: 0.0.0.0:9901\nlisteners: []\n"; + let yaml_content = "admin:\n metrics_address: 0.0.0.0:9902\nlisteners: []\n"; let config = build_configmap("praxis-config", "default", &gateway, yaml_content).unwrap(); let data = config.data.expect("data should be set"); diff --git a/src/resources/deployment.rs b/src/resources/deployment.rs index 873677f..231baa5 100644 --- a/src/resources/deployment.rs +++ b/src/resources/deployment.rs @@ -8,11 +8,11 @@ use std::collections::BTreeMap; use gateway_api::gateways::Gateway; use k8s_openapi::{ api::{ - apps::v1::{Deployment, DeploymentSpec, DeploymentStrategy}, + apps::v1::{Deployment, DeploymentSpec, DeploymentStrategy, RollingUpdateDeployment}, core::v1::{ Capabilities, ConfigMapVolumeSource, Container, ContainerPort, EmptyDirVolumeSource, HTTPGetAction, - PodSpec, PodTemplateSpec, Probe, ResourceRequirements, SeccompProfile, SecretVolumeSource, SecurityContext, - TopologySpreadConstraint, Volume, VolumeMount, + Lifecycle, LifecycleHandler, PodSpec, PodTemplateSpec, Probe, ResourceRequirements, SeccompProfile, + SecretVolumeSource, SecurityContext, SleepAction, TopologySpreadConstraint, Volume, VolumeMount, }, }, apimachinery::pkg::{ @@ -33,28 +33,25 @@ use crate::context::{ADMIN_PORT, praxis_image}; /// UID the Praxis proxy container runs as (nobody/nfsnobody). const PROXY_UID: i64 = 100; +/// Time for Service endpoint removal to propagate before SIGTERM. +const ENDPOINT_PROPAGATION_SECONDS: i64 = 5; + +/// Seconds a rollout may make no progress before the `Deployment` +/// controller reports `ProgressDeadlineExceeded`. +/// +/// With `maxUnavailable: 0` a replacement that cannot schedule or start +/// leaves the old pod serving indefinitely, so the failure has to be +/// bounded somewhere: this is the Deployment controller's own default, +/// stated explicitly because the rollout strategy below depends on it. +const PROGRESS_DEADLINE_SECONDS: i32 = 600; + /// Annotation overriding the data-plane replica count. const REPLICAS_ANNOTATION: &str = "praxis.sh/replicas"; /// Replicas run when the Gateway does not ask for a specific count. /// -/// One, matching the behaviour every existing Gateway already has. -/// -/// This was forced rather than chosen. Praxis 0.3.1 registered its KV -/// admin endpoints on the health port via `SO_REUSEPORT`, so probe -/// connections landed on whichever listener the kernel picked and -/// roughly half of them 404'd; every extra replica was another pod -/// whose liveness probe flapped, and a Gateway whose pods never settle -/// never reports Programmed. -/// -/// The pinned data plane is now 0.5.2, which no longer registers that -/// endpoint, so the constraint is gone. Two is the better availability -/// default — a single-replica Gateway is a single point of failure for -/// every route attached to it — and raising it is now a live option -/// rather than a blocked one. It is deliberately not part of the -/// version bump: that is a behaviour change deserving its own -/// conformance run, and bundling it would make a red run ambiguous. -/// `praxis.sh/replicas` remains the per-Gateway override meanwhile. +/// One, preserving the default footprint while allowing high-availability +/// Gateways to opt into a second replica. const DEFAULT_REPLICAS: i32 = 1; // ----------------------------------------------------------------------------- @@ -66,12 +63,11 @@ pub struct DeploymentParams<'gw> { /// Child resource name. pub name: &'gw str, - /// SHA-256 hex digest of the `ConfigMap` contents. + /// SHA-256 hex digest of listener and TLS topology. /// - /// Stored as a pod-template annotation so that config changes trigger a - /// rolling restart. Required because Kubernetes `ConfigMap` volume mounts - /// use atomic symlink swaps that `inotify`-based file watchers cannot - /// detect. + /// Stored as a pod-template annotation so only changes that cannot hot + /// reload trigger a rolling restart. Route and filter changes update the + /// mounted `ConfigMap` without changing this topology hash. pub config_hash: &'gw str, /// Parent Gateway. @@ -89,10 +85,10 @@ pub struct DeploymentParams<'gw> { /// Builds a Deployment for the Praxis data-plane. /// -/// Creates a Deployment with a single replica running the Praxis proxy -/// container. The pod mounts the configuration `ConfigMap` and TLS secrets. -/// Health probes use the admin endpoint on [`ADMIN_PORT`]. Config updates -/// are picked up via the proxy's file watcher (no pod restart needed). +/// Creates a Deployment with one replica by default. Each pod +/// mounts the configuration `ConfigMap` and TLS secrets. Health probes use +/// the metrics endpoint on [`ADMIN_PORT`]. Config hash changes trigger a +/// rolling update with zero unavailable replicas. /// /// The `listener_ports` field provides `(name, port)` pairs from `Gateway` /// listeners; the admin port is appended unless a listener already occupies @@ -259,7 +255,10 @@ fn build_praxis_container(ports: Vec, volume_mounts: Vec, volume_mounts: Vec, volume_mounts: Vec Lifecycle { + Lifecycle { + pre_stop: Some(LifecycleHandler { + sleep: Some(SleepAction { + seconds: ENDPOINT_PROPAGATION_SECONDS, + }), + ..Default::default() + }), + ..Default::default() + } +} + /// Creates an HTTP health probe against the admin port. fn admin_probe(path: &str, period_seconds: i32, failure_threshold: i32, initial_delay: i32) -> Probe { Probe { @@ -375,16 +388,28 @@ fn desired_replicas(gateway: &Gateway) -> i32 { /// Scheduling stays best-effort: on a single-node cluster a `DoNotSchedule` /// constraint would leave every replica after the first pending forever. fn spread_constraints(labels: &BTreeMap) -> Vec { - vec![TopologySpreadConstraint { - label_selector: Some(LabelSelector { - match_labels: Some(labels.clone()), + vec![ + TopologySpreadConstraint { + label_selector: Some(LabelSelector { + match_labels: Some(labels.clone()), + ..Default::default() + }), + max_skew: 1, + topology_key: "topology.kubernetes.io/zone".to_owned(), + when_unsatisfiable: "ScheduleAnyway".to_owned(), ..Default::default() - }), - max_skew: 1, - topology_key: "kubernetes.io/hostname".to_owned(), - when_unsatisfiable: "ScheduleAnyway".to_owned(), - ..Default::default() - }] + }, + TopologySpreadConstraint { + label_selector: Some(LabelSelector { + match_labels: Some(labels.clone()), + ..Default::default() + }), + max_skew: 1, + topology_key: "kubernetes.io/hostname".to_owned(), + when_unsatisfiable: "ScheduleAnyway".to_owned(), + ..Default::default() + }, + ] } /// The label sets and pod template a [`Deployment`] is assembled from. @@ -420,13 +445,17 @@ fn build_deployment_object( }, spec: Some(DeploymentSpec { replicas: Some(desired_replicas(gateway)), + progress_deadline_seconds: Some(PROGRESS_DEADLINE_SECONDS), selector: LabelSelector { match_labels: Some(meta.selector), ..Default::default() }, strategy: Some(DeploymentStrategy { type_: Some("RollingUpdate".to_owned()), - ..Default::default() + rolling_update: Some(RollingUpdateDeployment { + max_unavailable: Some(IntOrString::Int(0)), + max_surge: Some(IntOrString::Int(1)), + }), }), template: meta.pod_template, ..Default::default() @@ -621,14 +650,29 @@ mod tests { .and_then(|pod| pod.topology_spread_constraints) .expect("spread constraints should be set"); - assert_eq!( - constraints[0].topology_key, "kubernetes.io/hostname", - "replicas should be spread across nodes" - ); - assert_eq!( - constraints[0].when_unsatisfiable, "ScheduleAnyway", - "a single-node cluster must still schedule every replica" - ); + assert_eq!(constraints.len(), 2, "both zone and hostname spread must be configured"); + for (constraint, key) in constraints + .iter() + .zip(["topology.kubernetes.io/zone", "kubernetes.io/hostname"]) + { + assert_eq!( + constraint.topology_key, key, + "replicas must spread across each failure domain" + ); + assert_eq!(constraint.max_skew, 1, "replicas should be evenly distributed"); + assert_eq!( + constraint.when_unsatisfiable, "ScheduleAnyway", + "single-node scheduling must still work" + ); + assert_eq!( + constraint + .label_selector + .as_ref() + .and_then(|selector| selector.match_labels.as_ref()), + Some(&standard_labels("test-gateway")), + "spread must apply only to this Gateway's pods" + ); + } } #[test] @@ -641,7 +685,7 @@ mod tests { assert_eq!( spec.replicas, Some(DEFAULT_REPLICAS), - "a Gateway that asks for nothing should get the redundant default" + "a Gateway that asks for nothing should get one replica" ); let selector = spec.selector; @@ -666,6 +710,33 @@ mod tests { Some("RollingUpdate".to_owned()), "strategy should be RollingUpdate to avoid traffic blackouts" ); + let rolling = strategy.rolling_update.expect("rolling update options should be set"); + assert_eq!( + rolling.max_unavailable, + Some(IntOrString::Int(0)), + "a rolling update must not remove the last ready replica" + ); + assert_eq!( + rolling.max_surge, + Some(IntOrString::Int(1)), + "a rolling update should add one replacement at a time" + ); + } + + #[test] + fn test_build_deployment_bounds_a_stuck_rollout() { + let gateway = test_gateway(); + let ports = vec![("http".to_owned(), 8080)]; + let deployment = build_deployment(&test_params(&gateway, &ports)).unwrap(); + + let spec = deployment.spec.expect("spec should be set"); + assert_eq!( + spec.progress_deadline_seconds, + Some(PROGRESS_DEADLINE_SECONDS), + "maxUnavailable: 0 keeps the old pod serving a replacement that never becomes ready, \ + so the progress deadline is the only thing that turns a stuck rollout into a \ + reported failure instead of waiting on it forever" + ); } #[test] @@ -715,7 +786,7 @@ mod tests { Some("admin".to_owned()), "last port name should be admin" ); - assert_eq!(ports[2].container_port, 9901, "admin port should be 9901"); + assert_eq!(ports[2].container_port, 9902, "metrics port should be 9902"); } #[test] @@ -853,8 +924,8 @@ mod tests { ); assert_eq!( liveness_http.port, - IntOrString::Int(9901), - "liveness port should be 9901" + IntOrString::Int(9902), + "liveness port should be 9902" ); let readiness = container @@ -869,8 +940,8 @@ mod tests { ); assert_eq!( readiness_http.port, - IntOrString::Int(9901), - "readiness port should be 9901" + IntOrString::Int(9902), + "readiness port should be 9902" ); } @@ -917,6 +988,23 @@ mod tests { assert_eq!(seccomp.localhost_profile, None, "localhost_profile should be None"); } + #[test] + fn pod_shutdown_waits_for_service_endpoint_propagation() { + let gateway = test_gateway(); + let deployment = build_deployment(&test_params(&gateway, &[("http".to_owned(), 8080)])).unwrap(); + let pod = deployment.spec.unwrap().template.spec.unwrap(); + let hook = pod.containers[0].lifecycle.as_ref().unwrap().pre_stop.as_ref().unwrap(); + assert_eq!( + hook.sleep.as_ref().unwrap().seconds, + 5, + "endpoint removal must precede SIGTERM" + ); + assert!( + pod.termination_grace_period_seconds.unwrap() > 10, + "the grace period must cover the hook and Pingora's five-second request grace" + ); + } + #[test] fn test_build_deployment_pod_hardening() { let gateway = test_gateway(); @@ -968,7 +1056,11 @@ mod tests { Some(&Quantity("256Mi".to_owned())), "memory limit should be 256Mi" ); - assert!(limits.get("cpu").is_none(), "cpu limit should not be set"); + assert_eq!( + limits.get("cpu"), + Some(&Quantity("1".to_owned())), + "cpu limit should match one runtime worker thread" + ); } #[test] diff --git a/tests/integration.rs b/tests/integration.rs index 3f7854e..ac6bca4 100644 --- a/tests/integration.rs +++ b/tests/integration.rs @@ -40,13 +40,14 @@ use gateway_api::{ use k8s_openapi::{ api::{ apps::v1::Deployment, - core::v1::{ConfigMap, Namespace, Service, ServicePort, ServiceSpec}, + core::v1::{ConfigMap, Namespace, Pod, Service, ServicePort, ServiceSpec}, + policy::v1::PodDisruptionBudget, }, apimachinery::pkg::apis::meta::v1::{Condition, ObjectMeta}, }; use kube::{ Api, Client, - api::{DeleteParams, PostParams}, + api::{DeleteParams, ListParams, PostParams}, }; // --------------------------------------------------------------------------- @@ -326,6 +327,321 @@ async fn gateway_invalid_gateway_class_not_accepted() { ctx.cleanup_namespace(&ns).await; } +#[tokio::test] +#[ignore] +async fn reserved_or_conflicting_listeners_do_not_leave_stale_children() { + let ctx = TestContext::new().await; + let gc = ctx.create_gateway_class("reserved-port-gc").await; + ctx.await_cluster_condition::(&gc, "Accepted", "True", RECONCILE_TIMEOUT_SECS) + .await; + let ns = ctx.create_test_namespace("reserved-port").await; + let name = ctx.create_gateway("reserved-port", &ns, &gc, 9902).await; + ctx.await_gateway_condition(&name, &ns, "Accepted", "False", RECONCILE_TIMEOUT_SECS) + .await; + let deployments: Api = Api::namespaced(ctx.client.clone(), &ns); + assert!( + matches!(deployments.get(&format!("praxis-{name}")).await, + Err(kube::Error::Api(response)) if response.code == 404), + "a Gateway with only the reserved listener must not create a broken data plane" + ); + let gateways: Api = Api::namespaced(ctx.client.clone(), &ns); + let patch = serde_json::json!({"spec": {"listeners": [ + {"name": "http", "port": 9902, "protocol": "HTTP"}, + {"name": "valid", "port": 8085, "protocol": "HTTP"} + ]}}); + gateways + .patch( + &name, + &kube::api::PatchParams::default(), + &kube::api::Patch::Merge(patch), + ) + .await + .expect("add a usable listener"); + ctx.await_gateway_condition(&name, &ns, "Accepted", "True", RECONCILE_TIMEOUT_SECS) + .await; + let gateway = gateways.get(&name).await.expect("Gateway status"); + let listeners = gateway.status.expect("status").listeners.expect("listener statuses"); + let reserved = listeners + .iter() + .find(|listener| listener.name == "http") + .expect("reserved listener status"); + assert!( + reserved.conditions.iter().any(|condition| condition.type_ == "Accepted" + && condition.status == "False" + && condition.reason == "PortUnavailable"), + "the reserved listener must report its port conflict" + ); + let configmaps: Api = Api::namespaced(ctx.client.clone(), &ns); + let config = configmaps + .get(&format!("praxis-{name}")) + .await + .expect("valid listener config"); + let yaml = config + .data + .expect("config data") + .get("config.yaml") + .expect("config YAML") + .clone(); + let config: yaml_serde::Value = yaml_serde::from_str(&yaml).expect("generated YAML"); + let listeners = config + .get("listeners") + .and_then(yaml_serde::Value::as_sequence) + .expect("proxy listeners"); + assert_eq!(listeners.len(), 1, "only the supported listener may reach the proxy"); + assert_eq!( + listeners + .first() + .and_then(|listener| listener.get("address")) + .and_then(yaml_serde::Value::as_str), + Some("0.0.0.0:8085"), + "the metrics port must remain separate from Gateway listeners" + ); + let child = format!("praxis-{name}"); + await_deployment_ready(&deployments, &child, DATA_PLANE_TIMEOUT_SECS).await; + for rejected_listeners in [ + serde_json::json!({"spec":{"listeners":[{"name":"http", "port":9902, "protocol":"HTTP"}]}}), + serde_json::json!({"spec":{"listeners":[ + {"name":"http", "port":8085, "protocol":"HTTP"}, + {"name":"https", "port":8085, "protocol":"HTTPS", "tls":{ + "mode":"Terminate", "certificateRefs":[{"kind":"Secret", "name":"unused-during-conflict"}] + }} + ]}}), + ] { + let previous_uid = deployments.get(&child).await.unwrap().metadata.uid; + gateways + .patch( + &name, + &kube::api::PatchParams::default(), + &kube::api::Patch::Merge(rejected_listeners), + ) + .await + .unwrap(); + ctx.await_gateway_condition(&name, &ns, "Accepted", "False", RECONCILE_TIMEOUT_SECS) + .await; + let services: Api = Api::namespaced(ctx.client.clone(), &ns); + let budgets: Api = Api::namespaced(ctx.client.clone(), &ns); + let pods: Api = Api::namespaced(ctx.client.clone(), &ns); + let deadline = tokio::time::Instant::now() + .checked_add(tokio::time::Duration::from_secs(DATA_PLANE_TIMEOUT_SECS)) + .expect("test timeout fits the clock"); + loop { + if deployments.get_opt(&child).await.unwrap().is_none() + && services.get_opt(&child).await.unwrap().is_none() + && configmaps.get_opt(&child).await.unwrap().is_none() + && budgets.get_opt(&child).await.unwrap().is_none() + && pods + .list(&ListParams::default().labels(&format!("app.kubernetes.io/instance={name}"))) + .await + .unwrap() + .items + .is_empty() + { + break; + } + assert!( + tokio::time::Instant::now() < deadline, + "rejecting the final usable listener must remove stale children and proxies" + ); + tokio::time::sleep(tokio::time::Duration::from_millis(POLL_INTERVAL_MS)).await; + } + let restored_listener = + serde_json::json!({"spec":{"listeners":[{"name":"valid", "port":8085, "protocol":"HTTP"}]}}); + gateways + .patch( + &name, + &kube::api::PatchParams::default(), + &kube::api::Patch::Merge(restored_listener), + ) + .await + .unwrap(); + ctx.await_gateway_condition(&name, &ns, "Accepted", "True", RECONCILE_TIMEOUT_SECS) + .await; + await_deployment_ready(&deployments, &child, DATA_PLANE_TIMEOUT_SECS).await; + assert_ne!( + deployments.get(&child).await.unwrap().metadata.uid, + previous_uid, + "restoring a usable listener must recreate the data plane" + ); + } + ctx.cleanup_gateway(&name, &ns).await; + ctx.cleanup_gateway_class(&gc).await; + ctx.cleanup_namespace(&ns).await; +} + +#[cfg(test)] +mod reserved_listener_tests { + use super::*; + + #[tokio::test] + #[ignore] + async fn reserved_listener_transition_rejects_existing_route_parents() { + let ctx = TestContext::new().await; + let gc = ctx.create_gateway_class("reserved-route-gc").await; + ctx.await_cluster_condition::(&gc, "Accepted", "True", RECONCILE_TIMEOUT_SECS) + .await; + let ns = ctx.create_test_namespace("reserved-route").await; + let gateway = ctx.create_gateway("reserved-route", &ns, &gc, 8080).await; + let gateways: Api = Api::namespaced(ctx.client.clone(), &ns); + let listeners = |port| { + serde_json::json!({"spec":{"listeners":[ + {"name":"http", "port":port, "protocol":"HTTP"}, + {"name":"sibling", "port":8085, "protocol":"HTTP"} + ]}}) + }; + gateways + .patch( + &gateway, + &kube::api::PatchParams::default(), + &kube::api::Patch::Merge(listeners(8080)), + ) + .await + .unwrap(); + let backend = ctx.create_backend_service("route-backend", &ns, 80).await; + ctx.deploy_echo_server("route-backend", &ns, 80).await; + let backend_deadline = tokio::time::Instant::now() + .checked_add(tokio::time::Duration::from_secs(DATA_PLANE_TIMEOUT_SECS)) + .expect("test timeout fits the clock"); + while praxis_operator::endpoints::resolve_endpoints(&ctx.client, &ns, &backend, 80) + .await + .unwrap() + .is_empty() + { + assert!( + tokio::time::Instant::now() < backend_deadline, + "the route fixture must have a ready backend before configuration" + ); + tokio::time::sleep(tokio::time::Duration::from_millis(POLL_INTERVAL_MS)).await; + } + let routes: Api = Api::namespaced(ctx.client.clone(), &ns); + let mut created = Vec::new(); + for (prefix, sections) in [("blocked", vec!["http"]), ("both", vec!["http", "sibling"])] { + let name = unique_name(prefix); + let route = HTTPRoute { + metadata: ObjectMeta { + name: Some(name.clone()), + namespace: Some(ns.clone()), + ..Default::default() + }, + spec: HttpRouteSpec { + parent_refs: Some( + sections + .iter() + .map(|section| HttpRouteParentRefs { + name: gateway.clone(), + section_name: Some((*section).to_owned()), + ..Default::default() + }) + .collect(), + ), + rules: Some(vec![HttpRouteRules { + matches: Some(vec![HttpRouteRulesMatches { + path: Some(HttpRouteRulesMatchesPath { + r#type: Some(HttpRouteRulesMatchesPathType::PathPrefix), + value: Some(format!("/{prefix}")), + }), + ..Default::default() + }]), + backend_refs: Some(vec![HttpRouteRulesBackendRefs { + name: backend.clone(), + port: Some(80), + ..Default::default() + }]), + ..Default::default() + }]), + ..Default::default() + }, + status: None, + }; + routes.create(&PostParams::default(), &route).await.unwrap(); + let expected: Vec<_> = sections.iter().map(|section| (*section, "True")).collect(); + await_route_section_statuses(&routes, &name, &expected).await; + created.push(name); + } + gateways + .patch( + &gateway, + &kube::api::PatchParams::default(), + &kube::api::Patch::Merge(listeners(9902)), + ) + .await + .unwrap(); + let blocked = created.first().unwrap(); + let both = created.get(1).unwrap(); + let rejected = await_route_section_statuses(&routes, blocked, &[("http", "False")]).await; + assert_eq!( + find_httproute_parent_condition(&rejected, "Accepted", "False") + .unwrap() + .reason, + "NoMatchingParent", + "the reserved parent is no longer eligible" + ); + await_route_section_statuses(&routes, both, &[("http", "False"), ("sibling", "True")]).await; + let maps: Api = Api::namespaced(ctx.client.clone(), &ns); + let deadline = tokio::time::Instant::now() + .checked_add(tokio::time::Duration::from_secs(RECONCILE_TIMEOUT_SECS)) + .expect("test timeout fits the clock"); + loop { + let map = maps.get(&format!("praxis-{gateway}")).await.unwrap(); + let yaml = map.data.as_ref().unwrap().get("config.yaml").unwrap(); + let config: yaml_serde::Value = yaml_serde::from_str(yaml).unwrap(); + let reserved = config + .get("listeners") + .and_then(yaml_serde::Value::as_sequence) + .unwrap() + .iter() + .any(|listener| listener.get("address").and_then(yaml_serde::Value::as_str) == Some("0.0.0.0:9902")); + if !yaml.contains("/blocked") && yaml.contains("/both") && !reserved { + break; + } + assert!( + tokio::time::Instant::now() < deadline, + "only usable sibling routing may remain configured" + ); + tokio::time::sleep(tokio::time::Duration::from_millis(POLL_INTERVAL_MS)).await; + } + gateways + .patch( + &gateway, + &kube::api::PatchParams::default(), + &kube::api::Patch::Merge(listeners(8080)), + ) + .await + .unwrap(); + await_route_section_statuses(&routes, blocked, &[("http", "True")]).await; + await_route_section_statuses(&routes, both, &[("http", "True"), ("sibling", "True")]).await; + ctx.cleanup_gateway(&gateway, &ns).await; + ctx.cleanup_gateway_class(&gc).await; + ctx.cleanup_namespace(&ns).await; + } + + async fn await_route_section_statuses(api: &Api, name: &str, expected: &[(&str, &str)]) -> HTTPRoute { + let deadline = tokio::time::Instant::now() + .checked_add(tokio::time::Duration::from_secs(DATA_PLANE_TIMEOUT_SECS)) + .expect("test timeout fits the clock"); + loop { + let route = api.get(name).await.unwrap(); + if route.status.as_ref().is_some_and(|status| { + expected.iter().all(|(section, value)| { + status.parents.iter().any(|parent| { + parent.parent_ref.section_name.as_deref() == Some(*section) + && parent + .conditions + .iter() + .any(|condition| condition.type_ == "Accepted" && condition.status == *value) + }) + }) + }) { + return route; + } + assert!( + tokio::time::Instant::now() < deadline, + "route {name} must converge to parent statuses {expected:?}" + ); + tokio::time::sleep(tokio::time::Duration::from_millis(POLL_INTERVAL_MS)).await; + } + } +} + #[tokio::test] #[ignore] async fn gateway_parameters_ref_rejected() { @@ -1190,14 +1506,9 @@ async fn httproute_attached_count_reflected_in_listener() { // Data-Plane Verification Tests // --------------------------------------------------------------------------- -// Requires MetalLB and stable proxy pods (no concurrent rolling updates). #[tokio::test] #[ignore] async fn data_plane_health_probes_respond() { - if std::env::var("RUN_DATAPLANE_TESTS").is_err() { - eprintln!("skipped: set RUN_DATAPLANE_TESTS=1 (requires MetalLB)"); - return; - } let ctx = TestContext::new().await; let gc = ctx.create_gateway_class("dp-health-gc").await; ctx.await_cluster_condition::(&gc, "Accepted", "True", RECONCILE_TIMEOUT_SECS) @@ -1208,30 +1519,52 @@ async fn data_plane_health_probes_respond() { ctx.await_gateway_condition(&gw, &ns, "Programmed", "True", RECONCILE_TIMEOUT_SECS) .await; - let gw_api: Api = Api::namespaced(ctx.client.clone(), &ns); - let addr = await_gateway_address(&gw_api, &gw, DATA_PLANE_TIMEOUT_SECS).await; + let deployments: Api = Api::namespaced(ctx.client.clone(), &ns); + await_deployment_replicas_ready(&deployments, &format!("praxis-{gw}"), 1, DATA_PLANE_TIMEOUT_SECS).await; - let http = reqwest::Client::builder() - .danger_accept_invalid_certs(true) - .timeout(std::time::Duration::from_secs(5)) - .build() - .expect("failed to build reqwest client"); + let pods: Api = Api::namespaced(ctx.client.clone(), &ns); + let replicas = pods + .list(&ListParams::default().labels(&format!("app.kubernetes.io/instance={gw}"))) + .await + .expect("data-plane pods should be listed"); + assert_eq!(replicas.items.len(), 1, "Gateways should run one replica by default"); + for pod in replicas.items { + let name = pod.metadata.name.expect("pod name should be set"); + for path in ["healthy", "ready", "metrics"] { + let request = http::Request::get(format!("/api/v1/namespaces/{ns}/pods/{name}:9902/proxy/{path}")) + .body(Vec::new()) + .expect("pod proxy request should be valid"); + ctx.client + .request_text(request) + .await + .expect("each replica should serve health and metrics on port 9902"); + } + } - let healthy_url = format!("http://{addr}:9901/healthy"); - let healthy_resp = retry_http_get(&http, &healthy_url, DATA_PLANE_TIMEOUT_SECS).await; - assert!( - healthy_resp.status().is_success(), - "admin /healthy endpoint should return 2xx" - ); + ctx.cleanup_gateway(&gw, &ns).await; + ctx.cleanup_gateway_class(&gc).await; + ctx.cleanup_namespace(&ns).await; +} - let ready_url = format!("http://{addr}:9901/ready"); - let ready_resp = retry_http_get(&http, &ready_url, DATA_PLANE_TIMEOUT_SECS).await; - assert!( - ready_resp.status().is_success(), - "admin /ready endpoint should return 2xx" - ); +#[tokio::test] +#[ignore] +async fn gateway_replica_annotation_creates_two_instances() { + let ctx = TestContext::new().await; + let gc = ctx.create_gateway_class("dp-replicas-gc").await; + ctx.await_cluster_condition::(&gc, "Accepted", "True", RECONCILE_TIMEOUT_SECS) + .await; - ctx.cleanup_gateway(&gw, &ns).await; + let ns = ctx.create_test_namespace("dp-replicas").await; + let gw_name = ctx + .create_gateway_with_replicas("dp-replicas-gw", &ns, &gc, 8103, 2) + .await; + + ctx.await_gateway_condition(&gw_name, &ns, "Programmed", "True", RECONCILE_TIMEOUT_SECS) + .await; + let deployments: Api = Api::namespaced(ctx.client.clone(), &ns); + await_deployment_replicas_ready(&deployments, &format!("praxis-{gw_name}"), 2, DATA_PLANE_TIMEOUT_SECS).await; + + ctx.cleanup_gateway(&gw_name, &ns).await; ctx.cleanup_gateway_class(&gc).await; ctx.cleanup_namespace(&ns).await; } @@ -1270,11 +1603,11 @@ async fn data_plane_routes_traffic_to_backend() { .expect("failed to build reqwest client"); let url = format!("http://{addr}:8102/test"); - let resp = retry_http_get(&http, &url, DATA_PLANE_TIMEOUT_SECS).await; - assert!( - resp.status().is_success() || resp.status().as_u16() == 404, - "proxy should forward traffic (got status {status})", - status = resp.status() + let resp = await_echo_response(&http, &url, DATA_PLANE_TIMEOUT_SECS).await; + assert_eq!( + resp.text().await.expect("echo response body").trim(), + "ok", + "the hot-reloaded route must reach the echo backend" ); ctx.cleanup_httproute(&route, &ns).await; @@ -1451,6 +1784,18 @@ impl TestContext { } async fn create_gateway(&self, prefix: &str, namespace: &str, gc_name: &str, port: i32) -> String { + self.create_gateway_with_annotations(prefix, namespace, gc_name, port, None) + .await + } + + async fn create_gateway_with_annotations( + &self, + prefix: &str, + namespace: &str, + gc_name: &str, + port: i32, + annotations: Option>, + ) -> String { let name = unique_name(prefix); let gw_api: Api = Api::namespaced(self.client.clone(), namespace); @@ -1458,6 +1803,7 @@ impl TestContext { metadata: ObjectMeta { name: Some(name.clone()), namespace: Some(namespace.to_owned()), + annotations, ..Default::default() }, spec: GatewaySpec { @@ -1481,6 +1827,27 @@ impl TestContext { name } + async fn create_gateway_with_replicas( + &self, + prefix: &str, + namespace: &str, + gc_name: &str, + port: i32, + replicas: i32, + ) -> String { + self.create_gateway_with_annotations( + prefix, + namespace, + gc_name, + port, + Some(std::collections::BTreeMap::from([( + "praxis.sh/replicas".to_owned(), + replicas.to_string(), + )])), + ) + .await + } + async fn create_backend_service(&self, name: &str, namespace: &str, port: i32) -> String { let svc_api: Api = Api::namespaced(self.client.clone(), namespace); let svc = Service { @@ -1801,14 +2168,25 @@ async fn await_listener_count(api: &Api, name: &str, expected: usize, t } async fn await_deployment_ready(api: &Api, name: &str, timeout_secs: u64) { + await_deployment_replicas_ready(api, name, 1, timeout_secs).await; +} + +async fn await_deployment_replicas_ready(api: &Api, name: &str, expected: i32, timeout_secs: u64) { let deadline = tokio::time::Instant::now() + tokio::time::Duration::from_secs(timeout_secs); loop { if tokio::time::Instant::now() > deadline { - panic!("timed out waiting for Deployment {name} to be ready"); + panic!("timed out waiting for {expected} replicas on Deployment {name}"); } if let Ok(deploy) = api.get(name).await { - let ready = deploy.status.as_ref().and_then(|s| s.ready_replicas).unwrap_or(0); - if ready >= 1 { + let desired = deploy.spec.as_ref().and_then(|spec| spec.replicas).unwrap_or_default(); + let generation = deploy.metadata.generation.unwrap_or_default(); + if let Some(status) = deploy.status.as_ref() + && desired == expected + && status.observed_generation.unwrap_or_default() >= generation + && status.updated_replicas.unwrap_or_default() == expected + && status.ready_replicas.unwrap_or_default() == expected + && status.available_replicas.unwrap_or_default() == expected + { return; } } @@ -1816,17 +2194,20 @@ async fn await_deployment_ready(api: &Api, name: &str, timeout_secs: } } -async fn retry_http_get(client: &reqwest::Client, url: &str, timeout_secs: u64) -> reqwest::Response { +/// Waits for the route to reach the echo backend after ConfigMap propagation. +async fn await_echo_response(client: &reqwest::Client, url: &str, timeout_secs: u64) -> reqwest::Response { let deadline = tokio::time::Instant::now() + tokio::time::Duration::from_secs(timeout_secs); + let mut last_response = "no request completed".to_owned(); loop { - if tokio::time::Instant::now() > deadline { - panic!("timed out waiting for HTTP response from {url}"); - } + assert!( + tokio::time::Instant::now() <= deadline, + "route did not hot-reload at {url}: {last_response}" + ); match client.get(url).send().await { - Ok(resp) => return resp, - Err(_) => { - tokio::time::sleep(tokio::time::Duration::from_secs(2)).await; - }, + Ok(response) if response.status() == reqwest::StatusCode::OK => return response, + Ok(response) => last_response = format!("HTTP {}", response.status()), + Err(error) => last_response = error.to_string(), } + tokio::time::sleep(tokio::time::Duration::from_secs(2)).await; } }