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..33854d7 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: # --------------------------------------------------------------------------- @@ -79,6 +79,33 @@ jobs: - name: Integration tests run: make dev-integration + # The monitoring test's disabled phase ran above without the CRD. The + # phases below cover enabled mode with the CRD installed, then disabled + # mode again to prove stale owned resources are removed. + - name: Install PodMonitor CRD + run: | + bash hack/retry.sh curl --fail --location --show-error --silent \ + --retry 3 --retry-all-errors --retry-delay 2 \ + --output /tmp/monitoring.coreos.com_podmonitors.yaml \ + https://raw.githubusercontent.com/prometheus-operator/prometheus-operator/v0.80.1/example/prometheus-operator-crd/monitoring.coreos.com_podmonitors.yaml + kubectl apply --server-side -f /tmp/monitoring.coreos.com_podmonitors.yaml + + - name: Monitoring enabled phase + env: + PRAXIS_MONITORING_ENABLED: "true" + run: | + kubectl -n praxis-system set env deployment/praxis-operator PRAXIS_MONITORING_ENABLED=true + kubectl -n praxis-system rollout status deployment/praxis-operator --timeout=120s + cargo test --features integration -- --ignored monitoring_reconciliation + + - name: Monitoring disabled cleanup phase + env: + PRAXIS_MONITORING_ENABLED: "false" + run: | + kubectl -n praxis-system set env deployment/praxis-operator PRAXIS_MONITORING_ENABLED=false + kubectl -n praxis-system rollout status deployment/praxis-operator --timeout=120s + cargo test --features integration -- --ignored monitoring_reconciliation + - name: Debug cluster state if: failure() run: bash hack/dump-debug.sh diff --git a/CHANGELOG.md b/CHANGELOG.md index 6968c1d..591f093 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,9 @@ The format follows [Keep a Changelog], and this project adheres to ### Added +- Opt-in PodMonitor generation and metrics ingress policy with ownership-checked + cleanup when monitoring is disabled or no usable listeners remain. + - 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 @@ -25,6 +28,13 @@ The format follows [Keep a Changelog], and this project adheres to ### Fixed +- 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 so health and + metrics bind to the dedicated admin endpoint. +- Gateway listeners without routes now answer HTTP 404 with a static response + instead of generating an empty router chain. + - 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/deployment.yaml b/deploy/deployment.yaml index 9d77635..5be1d7f 100644 --- a/deploy/deployment.yaml +++ b/deploy/deployment.yaml @@ -32,6 +32,8 @@ spec: env: - name: PRAXIS_IMAGE value: "__PRAXIS_IMAGE__" + - name: PRAXIS_MONITORING_ENABLED + value: "false" - name: POD_NAME valueFrom: fieldRef: diff --git a/deploy/rbac.yaml b/deploy/rbac.yaml index 32220a8..d80442b 100644 --- a/deploy/rbac.yaml +++ b/deploy/rbac.yaml @@ -59,6 +59,15 @@ rules: - apiGroups: ["policy"] resources: ["poddisruptionbudgets"] verbs: ["get", "list", "watch", "create", "update", "patch", "delete"] + # Required even when PRAXIS_MONITORING_ENABLED=false: reconciliation + # checks for stale owned NetworkPolicy and PodMonitor resources. A missing + # Prometheus Operator CRD is treated as already deleted. + - apiGroups: ["networking.k8s.io"] + resources: ["networkpolicies"] + verbs: ["get", "list", "watch", "create", "update", "patch", "delete"] + - apiGroups: ["monitoring.coreos.com"] + resources: ["podmonitors"] + verbs: ["get", "list", "watch", "create", "update", "patch", "delete"] - apiGroups: [""] resources: ["events"] verbs: ["create", "patch"] diff --git a/docs/architecture.md b/docs/architecture.md index b56b4f8..428dd03 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -101,7 +101,7 @@ container. Key properties: - **TLS volumes:** mounts each referenced TLS Secret as a read-only volume - **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 diff --git a/docs/configuration.md b/docs/configuration.md index 403ffa3..fe7b007 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -12,13 +12,51 @@ 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 with the image chosen + by the deployment tooling (`praxis:dev` for local development). ```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. + +### Optional monitoring resources + +Set `PRAXIS_MONITORING_ENABLED=true` to have the operator create a +`PodMonitor` and a metrics-only `NetworkPolicy` for each managed data plane. +The Prometheus Operator CRD must be installed before enabling this option; +with the option unset or false, reconciliation tolerates an absent CRD and +removes any owned monitoring resources left by an earlier enabled configuration. + +- **Default:** disabled +- **Namespace:** `PRAXIS_MONITORING_NAMESPACE` (default `monitoring`) +- **Prometheus selector:** `PRAXIS_MONITORING_POD_SELECTOR` as comma-separated + `key=value` labels (default `app.kubernetes.io/name=prometheus`). Malformed or + duplicate requirements cause a configuration error; no partial selector is applied. + +The generated policy allows only the selected Prometheus pods in the selected +namespace to reach port 9902. It explicitly allows TCP traffic to every configured +data listener port from any source; Kubernetes combines these allowances with +other policies additively. If the +cluster's CNI applies NetworkPolicy to node-originated kubelet probes, add the +cluster's node CIDRs in a separate policy overlay. + +```yaml +env: + - name: PRAXIS_MONITORING_ENABLED + value: "true" + - name: PRAXIS_MONITORING_NAMESPACE + value: observability + - name: PRAXIS_MONITORING_POD_SELECTOR + value: "app.kubernetes.io/name=prometheus,app.kubernetes.io/instance=main" ``` ### `RUST_LOG` @@ -61,6 +99,23 @@ permissions: | secrets | get, list, watch, create, update, patch, delete | | events | create, patch | +### Networking Resources + +| API group | Resource | Verbs | +|-----------|----------|-------| +| networking.k8s.io | networkpolicies | get, list, watch, create, update, patch, delete | + +### Monitoring Operator Resources + +| API group | Resource | Verbs | +|-----------|----------|-------| +| monitoring.coreos.com | podmonitors | get, list, watch, create, update, patch, delete | + +> **Upgrading:** apply the updated `deploy/rbac.yaml` before updating the +> operator image, even when monitoring is disabled. Disabled-mode +> reconciliation still checks for and removes stale owned `NetworkPolicy` and +> `PodMonitor` resources, and the previous ClusterRole lacks those permissions. + ### Discovery Resources | Resource | Verbs | @@ -153,7 +208,7 @@ directly. - **ConfigMap:** `praxis-{gateway-name}` with key `config.yaml` -- **Admin port:** 9901 (health probes, diagnostics) +- **Admin port:** 9902 (health and metrics; `admin.metrics_address`) - **Config changes:** a SHA-256 hash annotation (`praxis.sh/config-hash`) on the pod template triggers a rolling update when config changes diff --git a/docs/development.md b/docs/development.md index ecee4ba..421deba 100644 --- a/docs/development.md +++ b/docs/development.md @@ -101,7 +101,7 @@ make kind-down | Variable | Default | |---------------------|---------------------------------| | `KIND_CLUSTER_NAME` | `praxis-conformance` | -| `PRAXIS_IMAGE` | `ghcr.io/praxis-proxy/praxis:0.3.0` | +| `PRAXIS_IMAGE` | `praxis:dev` | | `OPERATOR_IMAGE` | `praxis-operator:dev` | ## Testing diff --git a/src/config/generate.rs b/src/config/generate.rs index 6764905..4ef585c 100644 --- a/src/config/generate.rs +++ b/src/config/generate.rs @@ -30,9 +30,6 @@ 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, } @@ -40,8 +37,8 @@ pub struct PraxisConfig { /// 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,13 +51,6 @@ pub struct PraxisFilterChain { pub filters: Vec, } -/// Insecure options (for container deployments). -#[derive(Debug, Clone, Serialize, PartialEq)] -pub struct PraxisInsecureOptions { - /// Allow admin endpoint on public interface. - pub allow_public_admin: bool, -} - // ----------------------------------------------------------------------------- // Config Assembly // ----------------------------------------------------------------------------- @@ -88,12 +78,9 @@ 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, }) } @@ -117,8 +104,33 @@ fn build_filter_chain( listener_hostnames: &std::collections::HashMap>, ) -> yaml_serde::Result { let name = &listener.name; + let scoped = scoped_routes(routes, &listener.merged_section_names, 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() { + return routeless_chain(name, filters); + } + filters.extend_from_slice(&route_filters.terminating); + filters.push(build_router_filter(&scoped_refs)?); + filters.extend_from_slice(&route_filters.transforming); + filters.push(build_lb_filter(clusters)?); + + Ok(PraxisFilterChain { + name: format!("{name}-chain"), + filters, + }) +} - let section_names = &listener.merged_section_names; +/// Selects and orders the routes that target one listener. +fn scoped_routes( + routes: &[PraxisRoute], + section_names: &[String], + listener_hostnames: &std::collections::HashMap>, +) -> Vec { let filtered: Vec<_> = routes .iter() .filter(|route| { @@ -132,17 +144,18 @@ fn build_filter_chain( .collect(); let mut scoped = inject_listener_hostnames(&filtered, listener_hostnames); sort_routes_by_specificity_owned(&mut scoped); - let scoped_refs: Vec<_> = scoped.iter().collect(); - - let mut filters = vec![PraxisFilterEntry { - filter: "request_id".to_owned(), - config: yaml_serde::Value::Null, - }]; - filters.extend_from_slice(&route_filters.terminating); - filters.push(build_router_filter(&scoped_refs)?); - filters.extend_from_slice(&route_filters.transforming); - filters.push(build_lb_filter(clusters)?); + scoped +} +/// Terminates a listener that has no routes with a 404 response. +/// +/// Praxis rejects an empty router route list, so a routeless listener needs a +/// terminating filter instead of a router. +fn routeless_chain(name: &str, mut filters: Vec) -> yaml_serde::Result { + filters.push(PraxisFilterEntry { + filter: "static_response".to_owned(), + config: yaml_serde::to_value(serde_json::json!({"status": 404, "body": "no route"}))?, + }); Ok(PraxisFilterChain { name: format!("{name}-chain"), filters, @@ -342,12 +355,11 @@ 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.filter_chains.len(), 1, "should have one filter chain"); let chain = &config.filter_chains[0]; @@ -405,16 +417,15 @@ 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" - ); assert!(yaml.contains("request_id"), "YAML should contain request_id filter"); assert!(yaml.contains("router"), "YAML should contain router filter"); assert!( @@ -557,10 +568,17 @@ 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"); + let names: Vec<_> = chain.filters.iter().map(|filter| filter.filter.as_str()).collect(); + assert_eq!( + names, + vec!["request_id", "static_response"], + "a routeless listener must terminate with a static response" + ); + assert_eq!( + chain.filters[1].config.get("body"), + Some(&yaml_serde::Value::String("no route".to_owned())), + "a routeless listener must answer with the no-route response" + ); } #[test] @@ -643,9 +661,29 @@ mod tests { tls: None, }; - let config = assemble_config(vec![listener], &[], &[], &RouteFilters::default(), &Default::default()).unwrap(); + let route = PraxisRoute { + path: None, + path_prefix: "/".to_owned(), + host: None, + headers: None, + 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 config = assemble_config( + vec![listener], + &[route], + &[cluster], + &RouteFilters::default(), + &Default::default(), + ) + .unwrap(); - let router = &config.filter_chains[0].filters[1]; + let router = config.filter_chains[0] + .filters + .iter() + .find(|filter| filter.filter == "router") + .expect("a routed listener has a router"); assert_eq!( router.config.get("multi_level_subdomain_matching"), Some(&yaml_serde::Value::Bool(true)), @@ -681,7 +719,23 @@ mod tests { transforming: vec![headers], }; - let config = assemble_config(vec![listener], &[], &[], &route_filters, &Default::default()).unwrap(); + let route = PraxisRoute { + path: None, + path_prefix: "/".to_owned(), + host: None, + headers: None, + 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 config = assemble_config( + vec![listener], + &[route], + &[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(); diff --git a/src/context.rs b/src/context.rs index 98f1ffb..942daa6 100644 --- a/src/context.rs +++ b/src/context.rs @@ -29,8 +29,11 @@ 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; + +/// Named health and metrics port on the Praxis data-plane container. +pub(crate) const ADMIN_PORT_NAME: &str = "admin"; /// Image used when `PRAXIS_IMAGE` is unset. /// @@ -38,7 +41,10 @@ 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"; +/// +/// The deploy manifest substitutes `__PRAXIS_IMAGE__` with the image the +/// deployment tooling selects; local development uses `praxis:dev`. +const DEFAULT_PRAXIS_IMAGE: &str = "ghcr.io/praxis-proxy/praxis:0.7.5"; // ----------------------------------------------------------------------------- // Data Plane Image @@ -128,6 +134,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..97a7eea --- /dev/null +++ b/src/controller/child_cleanup.rs @@ -0,0 +1,280 @@ +// 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}, + networking::v1::NetworkPolicy, + policy::v1::PodDisruptionBudget, +}; +use kube::{ + Api, Resource, ResourceExt as _, + api::{DeleteParams, DynamicObject, Preconditions}, +}; +use serde::de::DeserializeOwned; + +use crate::{ + error::{OperatorError, Result}, + resources::{labels::child_name, monitoring::pod_monitor_api_resource}, +}; + +/// Removes only children controlled by this Gateway, tolerating prior deletion. +/// +/// Every kind is attempted even when an earlier deletion fails: a +/// transient error on one resource must not strand the rest until a +/// later retry happens to succeed. The first failure is still returned, +/// so reconciliation keeps retrying until every child is gone. +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 data_plane = [ + 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, + ]; + let monitoring = delete_monitoring_children(client, gw).await; + + first_error(data_plane.into_iter().chain([monitoring])) +} + +/// Removes monitoring resources regardless of the current enablement setting. +/// +/// Both kinds are attempted even when the first one fails, for the same +/// reason [`delete_children`] does not stop early. +pub(super) async fn delete_monitoring_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 monitors = Api::::namespaced_with(client.clone(), &namespace, &pod_monitor_api_resource()); + let deletions = [ + delete_owned::(client, &namespace, &name, &owner_uid).await, + delete_from_api(&monitors, &name, &owner_uid).await, + ]; + + first_error(deletions) +} + +/// Returns the first failure from a batch whose entries all ran already. +fn first_error(results: impl IntoIterator>) -> Result<()> { + results + .into_iter() + .find_map(std::result::Result::err) + .map_or(Ok(()), Err) +} + +/// 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); + delete_from_api(&api, name, owner_uid).await +} + +/// Uses the same ownership and concurrency guards for static and dynamic kinds. +/// A missing object or absent optional CRD both return 404 and need no deletion. +async fn delete_from_api(api: &Api, name: &str, owner_uid: &str) -> Result<()> +where + K: Resource + DeserializeOwned + Clone + Debug, +{ + let Some(child) = ignore_not_found(api.get(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() + }; + ignore_not_found(api.delete(name, ¶ms).await).map(|_| ()) +} + +/// Treats any API 404 as absent, including API groups returning plain-text errors. +fn ignore_not_found(response: std::result::Result) -> Result> { + match response { + Ok(value) => Ok(Some(value)), + Err(kube::Error::Api(status)) if status.code == 404 => Ok(None), + Err(error) => Err(error.into()), + } +} + +#[cfg(test)] +mod tests { + use serde_json::json; + + use super::*; + use crate::testing; + + #[tokio::test] + /// Only Gateway-owned children are deleted, guarded by UID preconditions. + 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] + /// Foreign owners and non-controller references are left untouched. + 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] + /// Missing resources are ignored; other API failures still surface. + 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" + ); + } + #[tokio::test] + /// One failing kind does not stop the others from being reclaimed. + async fn cleanup_attempts_every_kind_after_a_failure() { + let (ctx, journal) = testing::fake_context( + vec![testing::Canned::server_error("/services/praxis-gw")], + testing::Cached::default(), + ); + let gateway: Gateway = serde_json::from_value(serde_json::json!({"metadata":{"name":"gw","namespace":"infra","uid":"gateway-uid"},"spec":{"gatewayClassName":"praxis","listeners":[]}})).unwrap(); + + assert!( + delete_children(&ctx.client, &gateway).await.is_err(), + "a failed deletion must still reach error_policy, or the stale child is never retried" + ); + for kind in [ + "/deployments/praxis-gw", + "/configmaps/praxis-gw", + "/poddisruptionbudgets/praxis-gw", + "/networkpolicies/praxis-gw", + "/podmonitors/praxis-gw", + ] { + assert!( + journal + .requests() + .iter() + .any(|request| request.method == "GET" && request.uri.contains(kind)), + "a failure on one kind must not strand the rest until a retry happens to succeed: {kind}" + ); + } + } + + #[tokio::test] + #[expect( + clippy::too_many_lines, + reason = "explicit monitoring fixtures cover policy and reconciliation contracts" + )] + /// Monitoring cleanup keeps the owner check and tolerates an absent CRD. + async fn monitoring_cleanup_obeys_ownership_and_tolerates_absent_crd() { + for owner in ["gateway-uid", "foreign-uid"] { + let mut monitor = serde_json::json!({"apiVersion":"monitoring.coreos.com/v1", "kind":"PodMonitor", "metadata": { + "name":"praxis-gw", "namespace":"infra", "uid":"monitor-uid", "resourceVersion":"9", + "ownerReferences":[{"apiVersion":"gateway.networking.k8s.io/v1", "kind":"Gateway", "name":"gw", "uid":owner, "controller":true}] + }}); + let mut policy = monitor.clone(); + policy["apiVersion"] = "networking.k8s.io/v1".into(); + policy["kind"] = "NetworkPolicy".into(); + monitor["spec"] = serde_json::json!({}); + let (ctx, journal) = testing::fake_context( + vec![ + testing::Canned::ok("/networkpolicies/praxis-gw", policy), + testing::Canned::ok("/podmonitors/praxis-gw", monitor), + ], + testing::Cached::default(), + ); + let gateway: Gateway = serde_json::from_value(serde_json::json!({"metadata":{"name":"gw","namespace":"infra","uid":"gateway-uid"},"spec":{"gatewayClassName":"praxis","listeners":[]}})).unwrap(); + delete_monitoring_children(&ctx.client, &gateway).await.unwrap(); + let deletes: Vec<_> = journal + .requests() + .into_iter() + .filter(|request| request.method == "DELETE") + .collect(); + assert_eq!( + deletes.len(), + if owner == "gateway-uid" { 2 } else { 0 }, + "only this Gateway's monitoring can be deleted" + ); + for request in deletes { + assert_eq!( + request.body.unwrap()["preconditions"]["resourceVersion"], + "9", + "deletion must guard replacement races" + ); + } + } + let (ctx, _) = testing::fake_context(vec![], testing::Cached::default()); + let monitors = Api::::namespaced_with(ctx.client.clone(), "infra", &pod_monitor_api_resource()); + delete_from_api(&monitors, "praxis-gw", "gateway-uid").await.unwrap(); + } + #[test] + /// A CRD-less API group is already deleted, not an error to report. + fn absent_api_group_does_not_require_a_not_found_reason() { + let response: kube::core::Status = serde_json::from_value(json!({"status":"Failure", "code":404, "message":"404 page not found\n", "reason":"Failed to parse error data"})).unwrap(); + assert_eq!( + ignore_not_found::<()>(Err(kube::Error::Api(Box::new(response)))).unwrap(), + None, + "absent API groups return unstructured 404 responses" + ); + } +} diff --git a/src/controller/gateway.rs b/src/controller/gateway.rs index fe14889..6478b58 100644 --- a/src/controller/gateway.rs +++ b/src/controller/gateway.rs @@ -8,7 +8,7 @@ use std::{fmt::Debug, sync::Arc, time::Duration}; use gateway_api::{gateways::Gateway, httproutes::HTTPRoute, referencegrants::ReferenceGrant}; use kube::{ Api, Resource, ResourceExt as _, - api::{Patch, PatchParams}, + api::{ApiResource, DynamicObject, Patch, PatchParams}, runtime::{ controller::Action, events::{Event, EventType}, @@ -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}, }; // ----------------------------------------------------------------------------- @@ -82,7 +82,8 @@ pub fn error_policy(_gw: Arc, error: &OperatorError, _ctx: Arc | OperatorError::LeadershipLost | OperatorError::CacheSync(_) | OperatorError::Serialization(_) - | OperatorError::YamlSerialization(_) => Duration::from_secs(30), + | OperatorError::YamlSerialization(_) + | OperatorError::InvalidMonitoringSelector => Duration::from_secs(30), }; error!( %error, "Gateway reconciliation failed, retrying in {delay:?}" @@ -119,7 +120,13 @@ async fn apply(gw: Arc, ctx: &Context) -> Result { route_parent_status::update_route_parent_statuses(&ctx.client, &gw, &attached, &grants).await?; } - let requeue_secs = if can_accept { 15 } else { 2 }; + let requeue_secs = if can_accept { + 15 + } else if praxis_config::usable_listeners(&gw.spec.listeners).is_empty() { + 60 + } else { + 2 + }; Ok(Action::requeue(Duration::from_secs(requeue_secs))) } @@ -134,12 +141,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); } @@ -315,6 +318,29 @@ where Ok(()) } +/// Applies a namespaced dynamic resource via server-side apply. +pub(super) async fn apply_dynamic_resource( + client: &kube::Client, + ns: &str, + resource: &DynamicObject, + api_resource: &ApiResource, +) -> Result<()> { + let api = Api::::namespaced_with(client.clone(), ns, api_resource); + let name = resource + .meta() + .name + .as_deref() + .ok_or(OperatorError::MissingObjectKey(".metadata.name"))?; + api.patch( + name, + &PatchParams::apply("praxis-operator").force(), + &Patch::Apply(resource), + ) + .await?; + debug!(%name, kind = %api_resource.kind, "applied dynamic resource"); + Ok(()) +} + // ----------------------------------------------------------------------------- // Validation // ----------------------------------------------------------------------------- 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..21fc37a 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, @@ -156,16 +156,18 @@ async fn build_listener_statuses( let mut any_rejected = false; for listener in listeners { - if let Some(reason) = conflicts.get(&listener.name) { + // The reserved port is checked first: it is why the listener cannot + // bind at all, and reporting a conflict between two listeners that + // both ask for it would bury the actionable reason. + if let Some(status) = rejected_listener_status(listener, generation) { any_rejected = true; - statuses.push(conflicted_listener_status(listener, generation, *reason)); + statuses.push(status); continue; } - let protocol_supported = ListenerProtocol::is_supported(&listener.protocol); - if !protocol_supported { + if let Some(reason) = conflicts.get(&listener.name) { any_rejected = true; - statuses.push(unsupported_listener_status(listener, generation)); + statuses.push(conflicted_listener_status(listener, generation, *reason)); continue; } @@ -178,6 +180,25 @@ 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(|| { + let message = format!("port {ADMIN_PORT} is reserved for health and metrics"); + ListenerStatus { + name: listener.name.clone(), + attached_routes: 0, + supported_kinds: vec![], + conditions: vec![ + conditions::not_accepted(generation, "PortUnavailable", &message), + conditions::not_programmed(generation, "Invalid", &message), + ], + } + }) +} + /// Builds a status entry for a listener conflicting with another. /// /// A conflicted listener is not accepted, not programmed, and attaches @@ -507,6 +528,74 @@ 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 listeners_sharing_the_reserved_port_report_the_reservation_not_a_conflict() { + let (ctx, journal) = testing::fake_context(vec![gateway_response()], testing::Cached::default()); + let mut first = http_listener(); + first.port = ADMIN_PORT; + let mut second = first.clone(); + second.name = "reserved-2".to_owned(); + + build_and_apply_gateway_status(&ctx, &gateway(), &[first, second], &[]) + .await + .expect("a reachable API server accepts the patch"); + + let status = patched_status(&journal); + let listeners = status + .pointer("/listeners") + .and_then(Value::as_array) + .expect("both reserved listeners need a status entry"); + assert_eq!(listeners.len(), 2, "every declared listener is reported"); + for listener in listeners { + assert_eq!( + listener.pointer("/conditions/0/reason").and_then(Value::as_str), + Some("PortUnavailable"), + "the reservation is why neither listener can bind, and a generic conflict hides it" + ); + let conflicted = listener + .pointer("/conditions") + .and_then(Value::as_array) + .is_some_and(|conditions| conditions.iter().any(|condition| condition["type"] == "Conflicted")); + assert!( + !conflicted, + "the reserved port is the reason both listeners are rejected, not a conflict \ + between them" + ); + } + } + #[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/namespace_filter.rs b/src/controller/namespace_filter.rs index 7e0b6b0..5067f30 100644 --- a/src/controller/namespace_filter.rs +++ b/src/controller/namespace_filter.rs @@ -8,7 +8,7 @@ //! Kubernetes label selector, so the evaluation is large enough to //! separate from the status writing that consumes its answer. -use std::collections::BTreeMap; +use std::{borrow::Borrow, collections::BTreeMap}; use gateway_api::{ gateways::{ @@ -33,12 +33,15 @@ use crate::{ /// /// A route is retained if at least one listener it targets allows its /// namespace. The default policy (when unspecified) is `Same`. -pub(super) fn filter_routes_by_allowed_namespaces<'route>( +pub(super) fn filter_routes_by_allowed_namespaces<'route, L>( attached: &[AttachedRoute<'route>], - listeners: &[GatewayListeners], + listeners: &[L], gateway_ns: &str, stores: &Stores, -) -> Vec> { +) -> Vec> +where + L: Borrow, +{ let all_namespaces = Some(stores.namespaces()); attached @@ -57,18 +60,25 @@ pub(super) fn filter_routes_by_allowed_namespaces<'route>( } /// Checks whether a route is allowed by at least one targeted listener. -fn route_allowed_by_any_listener( +fn route_allowed_by_any_listener( route: &HTTPRoute, section_names: &[Option], - listeners: &[GatewayListeners], + listeners: &[L], gateway_ns: &str, all_namespaces: Option<&[Namespace]>, -) -> bool { +) -> bool +where + L: Borrow, +{ let route_ns = route_status::route_namespace(route); section_names.iter().any(|section| { let matching: Vec<&GatewayListeners> = match section { - Some(name) => listeners.iter().filter(|ls| ls.name == *name).collect(), - None => listeners.iter().collect(), + Some(name) => listeners + .iter() + .map(Borrow::borrow) + .filter(|ls| ls.name == *name) + .collect(), + None => listeners.iter().map(Borrow::borrow).collect(), }; matching .iter() diff --git a/src/controller/ownership.rs b/src/controller/ownership.rs index 3897122..14eb4ee 100644 --- a/src/controller/ownership.rs +++ b/src/controller/ownership.rs @@ -14,7 +14,7 @@ use gateway_api::{gatewayclasses::GatewayClass, gateways::Gateway, httproutes::H use kube::{Api, ResourceExt as _}; use tracing::debug; -use super::namespace_filter; +use super::{namespace_filter, praxis_config}; use crate::{ context::CONTROLLER_NAME, error::{OperatorError, Result}, @@ -75,6 +75,12 @@ fn is_api_not_found(err: &kube::Error) -> bool { /// Collects `HTTPRoute` resources attached to the Gateway, filtered by /// namespace policies. +/// +/// Only listeners the data plane can actually serve take part: a +/// listener on the reserved port, with an unsupported protocol, or in a +/// binding conflict is rejected in status, so attaching routes to it +/// would report an attachment the Gateway does not have. The same +/// predicate as config generation is used so the two cannot drift. pub(super) fn collect_routes<'route>( gw: &Gateway, all_routes: &'route [Arc], @@ -83,8 +89,9 @@ 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 = praxis_config::usable_listeners(&gw.spec.listeners); + let attached = attachment::attached_routes(&name, &ns, &listeners, all_routes); + namespace_filter::filter_routes_by_allowed_namespaces(&attached, &listeners, &ns, stores) } // ----------------------------------------------------------------------------- @@ -193,6 +200,29 @@ mod tests { ); } + #[test] + fn test_collect_routes_ignores_listeners_that_are_never_programmed() { + use crate::context::ADMIN_PORT; + + let routes = vec![Arc::new(route("mine", "gw"))]; + let stores = Stores::fake(vec![], vec![], vec![]); + for (port, protocol) in [(ADMIN_PORT, "HTTP"), (80, "TCP")] { + let mut gw = gateway(); + gw.spec.listeners = vec![GatewayListeners { + name: "rejected".to_owned(), + port, + protocol: protocol.to_owned(), + ..Default::default() + }]; + + assert!( + collect_routes(&gw, &routes, &stores).is_empty(), + "a listener the data plane never serves must not attach routes, or route status \ + would report an attachment this Gateway does not have: {protocol} on {port}" + ); + } + } + // ----------------------------------------------------------------------- // Test Utilities // ----------------------------------------------------------------------- diff --git a/src/controller/praxis_config.rs b/src/controller/praxis_config.rs index 60fb718..e915d68 100644 --- a/src/controller/praxis_config.rs +++ b/src/controller/praxis_config.rs @@ -29,6 +29,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}, @@ -37,6 +38,10 @@ use crate::{ deployment::{DeploymentParams, build_deployment}, disruption::build_pod_disruption_budget, labels::child_name, + monitoring::{ + MonitoringConfig, build_metrics_network_policy, build_pod_monitor, monitoring_config, + pod_monitor_api_resource, + }, service::build_service, }, }; @@ -72,12 +77,7 @@ 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); @@ -99,6 +99,17 @@ pub(super) async fn build_praxis_config( }) } +/// 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() +} + /// 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 { @@ -295,6 +306,7 @@ pub(super) async fn apply_child_resources( gw: &Gateway, config_output: &PraxisConfigOutput, ) -> Result { + let monitoring = monitoring_config()?; let ns = gw.namespace().unwrap_or_default(); let name = gw.name_any(); let child = child_name(&name); @@ -320,9 +332,35 @@ pub(super) async fn apply_child_resources( let budget = build_pod_disruption_budget(&child, &ns, gw)?; super::gateway::apply_resource(client, &ns, &budget).await?; + reconcile_monitoring(client, gw, config_output, monitoring.as_ref()).await?; Ok(config_hash) } +/// Applies enabled monitoring or removes only this Gateway's stale monitoring. +/// +/// The ingress policy is applied before the `PodMonitor` is published. +/// Publishing the scrape target first would let Prometheus discover it +/// while the policy that scopes metrics access is still unapplied; a +/// rejected policy then leaves the target advertised with no settled +/// reachability. The policy alone is harmless: the data ports stay open +/// to every peer, only the metrics port is narrowed. +async fn reconcile_monitoring( + client: &kube::Client, + gw: &Gateway, + config_output: &PraxisConfigOutput, + monitoring: Option<&MonitoringConfig>, +) -> Result<()> { + let Some(config) = monitoring else { + return super::child_cleanup::delete_monitoring_children(client, gw).await; + }; + let ns = gw.namespace().unwrap_or_default(); + let child = child_name(&gw.name_any()); + let policy = build_metrics_network_policy(&child, &ns, gw, config, &config_output.listener_ports)?; + super::gateway::apply_resource(client, &ns, &policy).await?; + let monitor = build_pod_monitor(&child, &ns, gw)?; + super::gateway::apply_dynamic_resource(client, &ns, &monitor, &pod_monitor_api_resource()).await +} + /// Converts `(name, port)` pairs into Kubernetes `ServicePort` entries. fn build_service_ports(listener_ports: &[(String, i32)]) -> Vec { listener_ports @@ -630,6 +668,19 @@ mod tests { // ----------------------------------------------------------------------- /// Builds a plain HTTP listener on port 80. + /// Builds the generated-config output a reconciliation step receives. + fn test_config_output( + config_yaml: &str, + listener_ports: Vec<(String, i32)>, + tls_secret_names: Vec, + ) -> PraxisConfigOutput { + PraxisConfigOutput { + config_yaml: config_yaml.to_owned(), + listener_ports, + tls_secret_names, + } + } + fn http_listener() -> GatewayListeners { GatewayListeners { name: "http".to_owned(), @@ -717,4 +768,128 @@ mod tests { }), ) } + /// Supported listeners without conflicts are selectable, even next to bad ones. + #[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" + ); + } + + /// The reserved metrics port never becomes a listener or Service port. + #[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" + ); + } + + /// Enabled monitoring applies both resources; disabled monitoring cleans up. + #[tokio::test] + #[expect( + clippy::too_many_lines, + reason = "explicit monitoring fixtures cover policy and reconciliation contracts" + )] + async fn monitoring_apply_and_disabled_cleanup_use_real_reconciliation_paths() { + let output = test_config_output("listeners: []", vec![("http".to_owned(), 8080)], vec![]); + let (client, journal) = testing::fake_client(vec![ + testing::Canned::ok( + "/networkpolicies", + serde_json::json!({"apiVersion":"networking.k8s.io/v1", "kind":"NetworkPolicy"}), + ), + testing::Canned::ok( + "/podmonitors", + serde_json::json!({"apiVersion":"monitoring.coreos.com/v1", "kind":"PodMonitor"}), + ), + ]); + let config = MonitoringConfig { + namespace: "monitoring".to_owned(), + pod_selector: BTreeMap::from([("app".to_owned(), "prometheus".to_owned())]), + }; + reconcile_monitoring(&client, &gateway(), &output, Some(&config)) + .await + .unwrap(); + assert_eq!( + journal + .requests() + .iter() + .filter(|request| request.method == "PATCH") + .count(), + 2, + "both enabled resources must be applied" + ); + let (client, journal) = testing::fake_client(vec![]); + reconcile_monitoring(&client, &gateway(), &output, None).await.unwrap(); + for kind in ["/networkpolicies/", "/podmonitors/"] { + assert!( + journal + .requests() + .iter() + .any(|request| request.method == "GET" && request.uri.contains(kind)), + "disabled monitoring must look for stale owned resources: {kind}" + ); + } + let (client, journal) = testing::fake_client(vec![ + testing::Canned::ok( + "/networkpolicies", + serde_json::json!({"apiVersion":"networking.k8s.io/v1", "kind":"NetworkPolicy"}), + ), + testing::Canned::server_error("/podmonitors"), + ]); + let result = reconcile_monitoring(&client, &gateway(), &output, Some(&config)).await; + assert!( + result.is_err(), + "missing or failing required CRD must fail enabled reconciliation" + ); + let requests = journal.requests(); + let policy = requests + .iter() + .position(|request| request.uri.contains("/networkpolicies")) + .expect("the ingress policy is applied even when the PodMonitor cannot be"); + let monitor = requests + .iter() + .position(|request| request.uri.contains("/podmonitors")) + .expect("the PodMonitor is still attempted"); + assert!( + policy < monitor, + "the policy has to exist before Prometheus can discover the target, or a rejected \ + policy leaves a published scrape target with unsettled reachability" + ); + } } diff --git a/src/error.rs b/src/error.rs index eedde3c..9ebdbaa 100644 --- a/src/error.rs +++ b/src/error.rs @@ -26,6 +26,10 @@ pub enum OperatorError { #[error("gatewayclass not found: {0}")] GatewayClassNotFound(String), + /// A monitoring selector would grant ambiguous or incomplete access. + #[error("invalid PRAXIS_MONITORING_POD_SELECTOR: expected unique, non-empty key=value entries")] + InvalidMonitoringSelector, + /// Leadership was taken by another replica. #[error("leadership lost to another replica")] LeadershipLost, diff --git a/src/gateway_api/attachment.rs b/src/gateway_api/attachment.rs index 7bfb4e5..b288438 100644 --- a/src/gateway_api/attachment.rs +++ b/src/gateway_api/attachment.rs @@ -3,7 +3,7 @@ //! Route attachment logic for `HTTPRoute` `parentRefs`. -use std::sync::Arc; +use std::{borrow::Borrow, sync::Arc}; use gateway_api::{ gateways::GatewayListeners, @@ -81,13 +81,18 @@ pub fn listener_matches_parent_ref(listener: &GatewayListeners, parent: &HttpRou /// Gateway may have several, told apart by hostname. A ref naming /// neither targets all of them, which is spelled `None` so that a /// route does not have to be re-derived when listeners change. -fn targeted_sections(parent: &HttpRouteParentRefs, listeners: &[GatewayListeners]) -> Vec> { +/// Resolves the listener section names a parent reference targets. +fn targeted_sections(parent: &HttpRouteParentRefs, listeners: &[L]) -> Vec> +where + L: Borrow, +{ if parent.section_name.is_none() && parent.port.is_none() { return vec![None]; } listeners .iter() + .map(Borrow::borrow) .filter(|listener| listener_matches_parent_ref(listener, parent)) .map(|listener| Some(listener.name.clone())) .collect() @@ -98,12 +103,15 @@ fn targeted_sections(parent: &HttpRouteParentRefs, listeners: &[GatewayListeners /// Each entry pairs a route with the listener section names its /// `parentRefs` resolve to. A `None` section name means the route /// attaches to all listeners. -pub fn attached_routes<'route>( +pub fn attached_routes<'route, L>( gateway_name: &str, gateway_ns: &str, - listeners: &[GatewayListeners], + listeners: &[L], routes: &'route [Arc], -) -> Vec> { +) -> Vec> +where + L: Borrow, +{ let mut result = Vec::new(); for route in routes { diff --git a/src/resources.rs b/src/resources.rs index cca9bf0..4386174 100644 --- a/src/resources.rs +++ b/src/resources.rs @@ -7,4 +7,5 @@ pub mod configmap; pub mod deployment; pub mod disruption; pub mod labels; +pub mod monitoring; pub mod service; 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..52cd868 100644 --- a/src/resources/deployment.rs +++ b/src/resources/deployment.rs @@ -24,7 +24,7 @@ use k8s_openapi::{ use kube::ResourceExt as _; use super::labels::{descriptive_labels, infrastructure_annotations, owner_reference, standard_labels}; -use crate::context::{ADMIN_PORT, praxis_image}; +use crate::context::{ADMIN_PORT, ADMIN_PORT_NAME, praxis_image}; // ----------------------------------------------------------------------------- // Constants @@ -239,7 +239,7 @@ fn build_container_ports(listener_ports: &[(String, i32)]) -> Vec ); } else { ports.push(ContainerPort { - name: Some("admin".to_owned()), + name: Some(ADMIN_PORT_NAME.to_owned()), container_port: ADMIN_PORT, protocol: Some("TCP".to_owned()), ..Default::default() @@ -715,7 +715,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, "admin port should be 9902"); } #[test] @@ -853,8 +853,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 +869,8 @@ mod tests { ); assert_eq!( readiness_http.port, - IntOrString::Int(9901), - "readiness port should be 9901" + IntOrString::Int(9902), + "readiness port should be 9902" ); } diff --git a/src/resources/monitoring.rs b/src/resources/monitoring.rs new file mode 100644 index 0000000..65c5af9 --- /dev/null +++ b/src/resources/monitoring.rs @@ -0,0 +1,373 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright (c) 2026 Shane Utt + +//! Optional Prometheus monitoring resources for managed data planes. + +use std::collections::BTreeMap; + +use gateway_api::gateways::Gateway; +use k8s_openapi::{ + api::networking::v1::{NetworkPolicy, NetworkPolicyIngressRule, NetworkPolicyPeer, NetworkPolicyPort}, + apimachinery::pkg::{ + apis::meta::v1::{LabelSelector, ObjectMeta}, + util::intstr::IntOrString, + }, +}; +use kube::{ + ResourceExt as _, + api::{ApiResource, DynamicObject}, + core::GroupVersionKind, +}; +use serde_json::json; + +use super::labels::{owner_reference, standard_labels}; +use crate::context::{ADMIN_PORT, ADMIN_PORT_NAME}; + +/// Environment variable enabling monitoring resources. +const MONITORING_ENABLED_ENV: &str = "PRAXIS_MONITORING_ENABLED"; + +/// Environment variable selecting the monitoring namespace. +const MONITORING_NAMESPACE_ENV: &str = "PRAXIS_MONITORING_NAMESPACE"; + +/// Environment variable selecting Prometheus pods. +const MONITORING_SELECTOR_ENV: &str = "PRAXIS_MONITORING_POD_SELECTOR"; + +/// Optional monitoring resource configuration. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct MonitoringConfig { + /// Namespace containing the Prometheus pods. + pub namespace: String, + /// Labels selecting the Prometheus pods. + pub pod_selector: BTreeMap, +} + +/// Reads optional monitoring configuration from the operator environment. +/// +/// # Errors +/// +/// Rejects malformed selectors instead of broadening metrics access. +pub fn monitoring_config() -> crate::error::Result> { + let enabled = std::env::var(MONITORING_ENABLED_ENV) + .ok() + .is_some_and(|value| value.eq_ignore_ascii_case("true")); + monitoring_config_from_values( + enabled, + std::env::var(MONITORING_NAMESPACE_ENV).ok().as_deref(), + std::env::var(MONITORING_SELECTOR_ENV).ok().as_deref(), + ) +} + +/// Builds a namespaced `PodMonitor` as a dynamic resource so the operator can +/// run without the Prometheus Operator CRD installed when monitoring is off. +/// +/// # Errors +/// +/// Returns an error when the Gateway has no UID for its owner reference. +pub fn build_pod_monitor(name: &str, namespace: &str, gateway: &Gateway) -> crate::error::Result { + let resource = pod_monitor_api_resource(); + let mut monitor = DynamicObject::new(name, &resource).within(namespace); + monitor.metadata.labels = Some(standard_labels(&gateway.name_any())); + monitor.metadata.owner_references = Some(vec![owner_reference(gateway)?]); + Ok(monitor.data(json!({ + "spec": { + "selector": {"matchLabels": standard_labels(&gateway.name_any())}, + "namespaceSelector": {"matchNames": [namespace]}, + "podMetricsEndpoints": [{ + "port": ADMIN_PORT_NAME, + "path": "/metrics", + "interval": "30s" + }] + } + }))) +} + +/// Returns the API resource descriptor for the Prometheus Operator CRD. +#[must_use] +pub fn pod_monitor_api_resource() -> ApiResource { + ApiResource::from_gvk(&GroupVersionKind::gvk("monitoring.coreos.com", "v1", "PodMonitor")) +} + +/// Builds the optional metrics ingress policy for one managed data plane. +/// +/// # Errors +/// +/// Returns an error when the Gateway has no UID for its owner reference. +pub fn build_metrics_network_policy( + name: &str, + namespace: &str, + gateway: &Gateway, + config: &MonitoringConfig, + listener_ports: &[(String, i32)], +) -> crate::error::Result { + let labels = standard_labels(&gateway.name_any()); + let source = NetworkPolicyPeer { + namespace_selector: Some(LabelSelector { + match_labels: Some(BTreeMap::from([( + "kubernetes.io/metadata.name".to_owned(), + config.namespace.clone(), + )])), + ..Default::default() + }), + pod_selector: Some(LabelSelector { + match_labels: Some(config.pod_selector.clone()), + ..Default::default() + }), + ..Default::default() + }; + Ok(NetworkPolicy { + metadata: ObjectMeta { + name: Some(name.to_owned()), + namespace: Some(namespace.to_owned()), + labels: Some(labels.clone()), + owner_references: Some(vec![owner_reference(gateway)?]), + ..Default::default() + }, + spec: Some(build_network_policy_spec(labels, source, listener_ports)), + }) +} + +/// Builds the policy specification separately to keep the resource builder +/// within the project's function-size limit. +fn build_network_policy_spec( + labels: BTreeMap, + source: NetworkPolicyPeer, + listener_ports: &[(String, i32)], +) -> k8s_openapi::api::networking::v1::NetworkPolicySpec { + let mut rules = vec![NetworkPolicyIngressRule { + from: Some(vec![source]), + ports: Some(vec![policy_port(ADMIN_PORT)]), + }]; + let data_ports: Vec<_> = listener_ports + .iter() + .filter(|(_, port)| *port != ADMIN_PORT) + .map(|(_, port)| policy_port(*port)) + .collect(); + if !data_ports.is_empty() { + rules.insert( + 0, + NetworkPolicyIngressRule { + from: None, + ports: Some(data_ports), + }, + ); + } + k8s_openapi::api::networking::v1::NetworkPolicySpec { + pod_selector: Some(LabelSelector { + match_labels: Some(labels), + ..Default::default() + }), + policy_types: Some(vec!["Ingress".to_owned()]), + ingress: Some(rules), + ..Default::default() + } +} + +/// Builds a TCP policy port without a source restriction. +fn policy_port(port: i32) -> NetworkPolicyPort { + NetworkPolicyPort { + port: Some(IntOrString::Int(port)), + protocol: Some("TCP".to_owned()), + ..Default::default() + } +} + +/// Resolves monitoring settings from explicit values for deterministic tests. +fn monitoring_config_from_values( + enabled: bool, + namespace: Option<&str>, + selector: Option<&str>, +) -> crate::error::Result> { + if !enabled { + return Ok(None); + } + Ok(Some(MonitoringConfig { + namespace: namespace.unwrap_or("monitoring").to_owned(), + pod_selector: selector.map_or_else(|| Ok(default_selector()), parse_selector)?, + })) +} + +/// Returns the default Prometheus pod selector. +fn default_selector() -> BTreeMap { + BTreeMap::from([("app.kubernetes.io/name".to_owned(), "prometheus".to_owned())]) +} + +/// Parses every equality requirement, rejecting incomplete or duplicate entries. +fn parse_selector(selector: &str) -> crate::error::Result> { + let mut parsed = BTreeMap::new(); + for pair in selector.split(',') { + let (key, value) = pair + .split_once('=') + .ok_or(crate::error::OperatorError::InvalidMonitoringSelector)?; + let (key, value) = (key.trim(), value.trim()); + if key.is_empty() + || value.is_empty() + || value.contains('=') + || parsed.insert(key.to_owned(), value.to_owned()).is_some() + { + return Err(crate::error::OperatorError::InvalidMonitoringSelector); + } + } + Ok(parsed) +} + +#[cfg(test)] +mod tests { + use gateway_api::gateways::GatewaySpec; + + use super::*; + + /// Builds the Gateway used by the monitoring tests. + fn gateway() -> Gateway { + Gateway { + metadata: ObjectMeta { + name: Some("test-gateway".to_owned()), + namespace: Some("default".to_owned()), + uid: Some("test-uid".to_owned()), + ..Default::default() + }, + spec: GatewaySpec::default(), + status: None, + } + } + + /// The `PodMonitor` scrapes the reserved metrics port and is Gateway-owned. + #[test] + fn pod_monitor_uses_metrics_port_and_owner() { + let monitor = build_pod_monitor("praxis-test", "default", &gateway()).unwrap(); + assert_eq!( + monitor.types.as_ref().map(|types| types.kind.as_str()), + Some("PodMonitor"), + "use the monitoring CRD" + ); + assert_eq!( + monitor.metadata.namespace.as_deref(), + Some("default"), + "monitor lives with the gateway" + ); + assert_eq!( + monitor.metadata.owner_references.as_ref().unwrap()[0].uid, + "test-uid", + "gateway owns monitor" + ); + assert_eq!( + monitor.data["spec"]["podMetricsEndpoints"][0]["port"], ADMIN_PORT_NAME, + "scrape the named admin port" + ); + } + + #[test] + #[expect( + clippy::too_many_lines, + reason = "explicit monitoring fixtures cover policy and reconciliation contracts" + )] + /// Data ports stay open to every peer; metrics stay closed to all but Prometheus. + fn policy_preserves_data_ingress_and_restricts_metrics_to_selected_peer() { + let config = monitoring_config_from_values(true, Some("observability"), Some("team=platform,instance=main")) + .unwrap() + .unwrap(); + let ports = [ + ("http".to_owned(), 8080), + ("https".to_owned(), 8443), + ("reserved".to_owned(), ADMIN_PORT), + ]; + let policy = build_metrics_network_policy("praxis-test", "default", &gateway(), &config, &ports).unwrap(); + let spec = policy.spec.unwrap(); + assert_eq!( + spec.policy_types, + Some(vec!["Ingress".to_owned()]), + "egress remains unchanged" + ); + let rules = spec.ingress.unwrap(); + assert_eq!(rules.len(), 2, "data and metrics have separate rules"); + assert!(rules[0].from.is_none(), "clients from any source can reach data ports"); + assert_eq!( + rules[0].ports, + Some(vec![policy_port(8080), policy_port(8443)]), + "metrics must never appear in unrestricted data ports" + ); + assert_eq!( + rules[1].ports, + Some(vec![policy_port(ADMIN_PORT)]), + "only metrics uses restricted peer" + ); + let peers = rules[1].from.as_ref().unwrap(); + assert_eq!(peers.len(), 1, "namespace and pod selectors must be ANDed"); + assert_eq!( + peers[0] + .namespace_selector + .as_ref() + .unwrap() + .match_labels + .as_ref() + .unwrap()["kubernetes.io/metadata.name"], + "observability", + "restrict namespace" + ); + assert_eq!( + peers[0].pod_selector.as_ref().unwrap().match_labels.as_ref(), + Some(&config.pod_selector), + "preserve all selector requirements" + ); + } + + /// Monitoring is opt-in and the selector comes from the environment contract. + #[test] + fn monitoring_is_opt_in_and_selector_is_configurable() { + assert_eq!( + monitoring_config_from_values(false, None, Some("invalid")).unwrap(), + None, + "disabled monitoring needs no selector" + ); + let enabled = monitoring_config_from_values(true, Some("observability"), Some("team=platform")) + .unwrap() + .unwrap(); + assert_eq!(enabled.namespace, "observability", "use configured namespace"); + assert_eq!( + enabled.pod_selector.get("team"), + Some(&"platform".to_owned()), + "use configured selector" + ); + assert_eq!( + monitoring_config_from_values(true, None, None) + .unwrap() + .unwrap() + .pod_selector, + default_selector(), + "defaults apply only when unset" + ); + } + + /// A data plane with no listeners cannot open metrics to every peer. + #[test] + fn empty_data_ports_cannot_allow_unrestricted_metrics() { + let config = monitoring_config_from_values(true, None, None).unwrap().unwrap(); + for ports in [vec![], vec![("reserved".to_owned(), ADMIN_PORT)]] { + let policy = build_metrics_network_policy("praxis-test", "default", &gateway(), &config, &ports).unwrap(); + let rules = policy.spec.unwrap().ingress.unwrap(); + assert_eq!(rules.len(), 1, "empty ports must not produce an allow-all rule"); + assert!(rules[0].from.is_some(), "metrics always requires the Prometheus peer"); + } + } + + /// Malformed selectors fail closed instead of broadening access. + #[test] + fn malformed_selectors_never_produce_partial_or_default_access() { + for selector in [ + "team=monitoring,instance", + "", + "team=", + "=value", + "team=a,", + "team=a,team=b", + "team=a=b", + ] { + assert!( + matches!( + parse_selector(selector), + Err(crate::error::OperatorError::InvalidMonitoringSelector) + ), + "reject the entire selector: {selector}" + ); + } + } +} diff --git a/tests/integration.rs b/tests/integration.rs index 3f7854e..5e6592f 100644 --- a/tests/integration.rs +++ b/tests/integration.rs @@ -1830,3 +1830,181 @@ async fn retry_http_get(client: &reqwest::Client, url: &str, timeout_secs: u64) } } } + +// Run this test with PRAXIS_MONITORING_ENABLED matching the operator Deployment. +// Exercise disabled mode before installing the PodMonitor CRD, enabled mode after +// installing it, and disabled mode again to cover stale-resource cleanup. +#[cfg(test)] +mod monitoring_tests { + use k8s_openapi::api::{core::v1::Pod, networking::v1::NetworkPolicy}; + use kube::{ + ResourceExt as _, + api::{DynamicObject, ListParams, Patch, PatchParams}, + }; + use praxis_operator::resources::monitoring::{ + MonitoringConfig, build_metrics_network_policy, build_pod_monitor, monitoring_config, pod_monitor_api_resource, + }; + + use super::*; + + #[tokio::test] + #[ignore] + async fn monitoring_reconciliation() { + let ctx = TestContext::new().await; + let config = monitoring_config().expect("valid monitoring environment"); + let gc = ctx.create_gateway_class("monitoring-gc").await; + ctx.await_cluster_condition::(&gc, "Accepted", "True", RECONCILE_TIMEOUT_SECS) + .await; + let ns = ctx.create_test_namespace("monitoring").await; + let name = ctx.create_gateway("monitoring", &ns, &gc, 8080).await; + ctx.await_gateway_condition(&name, &ns, "Accepted", "True", RECONCILE_TIMEOUT_SECS) + .await; + let child = format!("praxis-{name}"); + let deployments = Api::::namespaced(ctx.client.clone(), &ns); + await_deployment_ready(&deployments, &child, DATA_PLANE_TIMEOUT_SECS).await; + let gateways = Api::::namespaced(ctx.client.clone(), &ns); + let gw = gateways.get(&name).await.unwrap(); + let policies = Api::::namespaced(ctx.client.clone(), &ns); + let monitors = Api::::namespaced_with(ctx.client.clone(), &ns, &pod_monitor_api_resource()); + if let Some(config) = config { + assert_monitoring_matches(&policies, &monitors, &child, &ns, &gw, &config).await; + let pods = Api::::namespaced(ctx.client.clone(), &ns) + .list(&ListParams::default().labels(&format!("app.kubernetes.io/instance={name}"))) + .await + .unwrap(); + let pod_name = pods.items.first().expect("proxy pod").name_any(); + let request = http::Request::get(format!("/api/v1/namespaces/{ns}/pods/{pod_name}:8080/proxy/")) + .body(kube::client::Body::empty()) + .unwrap(); + let response = ctx.client.send(request).await.expect("reach routeless proxy"); + assert_eq!( + response.status(), + http::StatusCode::NOT_FOUND, + "routeless published proxy returns HTTP 404" + ); + gateways + .patch( + &name, + &PatchParams::default(), + &Patch::Merge( + serde_json::json!({"spec":{"listeners":[{"name":"http","port":9902,"protocol":"HTTP"}]}}), + ), + ) + .await + .unwrap(); + await_monitoring_absent(&policies, &monitors, &child).await; + gateways + .patch( + &name, + &PatchParams::default(), + &Patch::Merge( + serde_json::json!({"spec":{"listeners":[{"name":"http","port":8080,"protocol":"HTTP"}]}}), + ), + ) + .await + .unwrap(); + assert_monitoring_matches(&policies, &monitors, &child, &ns, &gw, &config).await; + } else { + let crds = Api::< + k8s_openapi::apiextensions_apiserver::pkg::apis::apiextensions::v1::CustomResourceDefinition, + >::all(ctx.client.clone()); + let has_crd = crds + .get_opt("podmonitors.monitoring.coreos.com") + .await + .unwrap() + .is_some(); + let stale_config = MonitoringConfig { + namespace: "monitoring".to_owned(), + pod_selector: std::collections::BTreeMap::from([("app".to_owned(), "prometheus".to_owned())]), + }; + let policy = + build_metrics_network_policy(&child, &ns, &gw, &stale_config, &[("http".to_owned(), 8080)]).unwrap(); + policies + .patch(&child, &PatchParams::apply("monitoring-test"), &Patch::Apply(&policy)) + .await + .unwrap(); + if has_crd { + let monitor = build_pod_monitor(&child, &ns, &gw).unwrap(); + monitors + .patch(&child, &PatchParams::apply("monitoring-test"), &Patch::Apply(&monitor)) + .await + .unwrap(); + } + gateways.patch(&name, &PatchParams::default(), &Patch::Merge(serde_json::json!({"metadata":{"annotations":{"test.praxis.sh/reconcile":"monitoring-cleanup"}}}))).await.unwrap(); + await_monitoring_absent(&policies, &monitors, &child).await; + assert!( + deployments.get_opt(&child).await.unwrap().is_some(), + "disabled monitoring does not prevent data-plane reconciliation without the CRD" + ); + } + ctx.cleanup_gateway(&name, &ns).await; + ctx.cleanup_gateway_class(&gc).await; + ctx.cleanup_namespace(&ns).await; + } + + async fn await_monitoring_absent(policies: &Api, monitors: &Api, name: &str) { + let deadline = tokio::time::Instant::now() + .checked_add(tokio::time::Duration::from_secs(RECONCILE_TIMEOUT_SECS)) + .expect("test deadline fits the clock"); + loop { + if policies.get_opt(name).await.unwrap().is_none() && monitor_is_absent(monitors, name).await { + return; + } + assert!( + tokio::time::Instant::now() < deadline, + "owned monitoring resources must be reclaimed" + ); + tokio::time::sleep(tokio::time::Duration::from_millis(POLL_INTERVAL_MS)).await; + } + } + async fn await_monitoring_present( + policies: &Api, + monitors: &Api, + name: &str, + ) -> (NetworkPolicy, DynamicObject) { + let deadline = tokio::time::Instant::now() + .checked_add(tokio::time::Duration::from_secs(RECONCILE_TIMEOUT_SECS)) + .expect("test deadline fits the clock"); + loop { + if let (Some(policy), Some(monitor)) = ( + policies.get_opt(name).await.unwrap(), + monitors.get_opt(name).await.unwrap(), + ) { + return (policy, monitor); + } + assert!( + tokio::time::Instant::now() < deadline, + "enabled monitoring must reconcile both resources" + ); + tokio::time::sleep(tokio::time::Duration::from_millis(POLL_INTERVAL_MS)).await; + } + } + + /// Waits for both monitoring resources and checks their stored contents. + async fn assert_monitoring_matches( + policies: &Api, + monitors: &Api, + child: &str, + ns: &str, + gw: &Gateway, + config: &MonitoringConfig, + ) { + let (policy, monitor) = await_monitoring_present(policies, monitors, child).await; + let mut expected = build_metrics_network_policy(child, ns, gw, config, &[("http".to_owned(), 8080)]).unwrap(); + expected.metadata = policy.metadata.clone(); + assert_eq!(policy, expected, "the API stores the intended policy"); + assert_eq!( + monitor.data, + build_pod_monitor(child, ns, gw).unwrap().data, + "the API stores the intended monitor spec" + ); + } + + async fn monitor_is_absent(monitors: &Api, name: &str) -> bool { + match monitors.get(name).await { + Ok(_) => false, + Err(kube::Error::Api(status)) if status.code == 404 => true, + Err(error) => panic!("unexpected PodMonitor API failure: {error}"), + } + } +}