From 4ca6797ce87dff2d7923bfe9b334763208853e10 Mon Sep 17 00:00:00 2001 From: Bartek Plotka Date: Tue, 10 Nov 2026 17:05:50 +0000 Subject: [PATCH 1/2] discovery/kubernetes: add KubernetesSdConfig and pod target group builder Add feature = "discovery-kubernetes", KubernetesSdConfig validation and path resolution, #[non_exhaustive] Kubernetes resource models, and build_pod_target_group matching Prometheus pod.go. Signed-off-by: Bartek Plotka --- .github/workflows/ci.yml | 6 + Cargo.toml | 1 + src/discovery.rs | 55 ++- src/discovery/kubernetes.rs | 870 ++++++++++++++++++++++++++++++++++ tests/discovery_kubernetes.rs | 581 +++++++++++++++++++++++ 5 files changed, 1512 insertions(+), 1 deletion(-) create mode 100644 src/discovery/kubernetes.rs create mode 100644 tests/discovery_kubernetes.rs diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ed079a6..c20b05a 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -36,6 +36,8 @@ jobs: run: cargo clippy --all-targets --no-default-features --features discovery -- -D warnings - name: Run clippy (http-config only) run: cargo clippy --all-targets --no-default-features --features http-config -- -D warnings + - name: Run clippy (discovery-kubernetes only) + run: cargo clippy --all-targets --no-default-features --features discovery-kubernetes -- -D warnings - name: Run tests (all features) run: cargo test --all-features - name: Run tests (no default features) @@ -44,6 +46,8 @@ jobs: run: cargo test --no-default-features --features discovery - name: Run tests (http-config only) run: cargo test --no-default-features --features http-config + - name: Run tests (discovery-kubernetes only) + run: cargo test --no-default-features --features discovery-kubernetes - name: Run clippy (OTAP compatibility suite) env: CARGO_TARGET_DIR: target @@ -68,6 +72,8 @@ jobs: run: cargo check --all-targets --no-default-features --features discovery - name: Check MSRV (http-config only) run: cargo check --all-targets --no-default-features --features http-config + - name: Check MSRV (discovery-kubernetes only) + run: cargo check --all-targets --no-default-features --features discovery-kubernetes go-fixtures: name: Go testdata sync check diff --git a/Cargo.toml b/Cargo.toml index b2eae54..eecf9a0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -16,6 +16,7 @@ exclude = ["tests/otap-compat"] default = ["arrow", "dto"] arrow = ["dep:arrow-array", "dep:arrow-buffer", "dep:arrow-schema"] discovery = ["dep:serde", "dep:serde_json", "dep:tokio"] +discovery-kubernetes = ["discovery", "http-config"] dto = ["dep:prost", "dep:prost-types"] http-config = ["dep:reqwest", "dep:serde", "dep:serde_json"] diff --git a/src/discovery.rs b/src/discovery.rs index aacef43..7d7e14d 100644 --- a/src/discovery.rs +++ b/src/discovery.rs @@ -11,16 +11,30 @@ // See the License for the specific language governing permissions and // limitations under the License. -//! Prometheus-compatible target discovery (`static_configs` and `dns_sd_configs`). +//! Prometheus-compatible target discovery (`static_configs`, `dns_sd_configs`, +//! and `reqwest`-based `kubernetes_sd_configs` for `role: pod`). pub mod dns; +#[cfg(feature = "discovery-kubernetes")] +pub mod kubernetes; use std::collections::BTreeMap; use std::time::Duration; use serde::{Deserialize, Deserializer, Serialize, Serializer}; +#[cfg(feature = "http-config")] +pub use crate::http_config::{ + Authorization, BasicAuth, HttpClientConfig, HttpConfigError, HttpHeader, HttpHeaders, OAuth2, + ProxyConfig, TlsConfig, TlsVersion, +}; pub use dns::{DnsDiscovery, DnsRecord, DnsRecordType, DnsSdConfig, ResolvConf}; +#[cfg(feature = "discovery-kubernetes")] +pub use kubernetes::{ + KubernetesAttachMetadataConfig, KubernetesMetadataContext, KubernetesNamespaceDiscovery, + KubernetesRole, KubernetesSdConfig, KubernetesSelectorConfig, Pod, build_pod_target_group, + build_pod_target_groups, +}; /// Reserved Prometheus label holding the `:` scrape target address. pub const ADDRESS_LABEL: &str = "__address__"; @@ -37,6 +51,10 @@ pub enum DiscoveryError { /// I/O or filesystem error. #[error("I/O error: {0}")] Io(#[from] std::io::Error), + /// HTTP client or transport error during discovery. + #[cfg(feature = "http-config")] + #[error("HTTP error: {0}")] + Http(#[from] reqwest::Error), /// DNS protocol or lookup error. #[error("DNS error: {0}")] Dns(String), @@ -45,6 +63,17 @@ pub enum DiscoveryError { Json(#[from] serde_json::Error), } +#[cfg(feature = "http-config")] +impl From for DiscoveryError { + fn from(err: crate::http_config::HttpConfigError) -> Self { + match err { + crate::http_config::HttpConfigError::InvalidConfig(msg) => Self::InvalidConfig(msg), + crate::http_config::HttpConfigError::Io(e) => Self::Io(e), + crate::http_config::HttpConfigError::Http(e) => Self::Http(e), + } + } +} + /// Static target group configuration matching Prometheus `static_configs` entries. #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] @@ -139,6 +168,10 @@ pub struct DiscoveryConfigs { /// DNS service discovery configurations (`dns_sd_configs`). #[serde(default, skip_serializing_if = "Vec::is_empty")] pub dns_sd_configs: Vec, + /// Kubernetes service discovery configurations (`kubernetes_sd_configs`). + #[cfg(feature = "discovery-kubernetes")] + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub kubernetes_sd_configs: Vec, } impl DiscoveryConfigs { @@ -147,6 +180,10 @@ impl DiscoveryConfigs { for dns_cfg in &mut self.dns_sd_configs { dns_cfg.validate()?; } + #[cfg(feature = "discovery-kubernetes")] + for k8s_cfg in &mut self.kubernetes_sd_configs { + k8s_cfg.validate()?; + } Ok(()) } @@ -165,6 +202,22 @@ pub(crate) fn join_host_port(host: &str, port: &str) -> String { } } +/// Sanitizes an arbitrary string into a valid Prometheus ASCII label name suffix by replacing +/// any character outside `[a-zA-Z0-9_]` with `'_'`, matching Prometheus `strutil.SanitizeLabelName`. +#[cfg(feature = "discovery-kubernetes")] +pub(crate) fn sanitize_label_name(input: &str) -> String { + input + .chars() + .map(|c| { + if c.is_ascii_alphanumeric() || c == '_' { + c + } else { + '_' + } + }) + .collect() +} + /// Parses a Prometheus duration string (such as `"30s"`, `"1m"`, `"500ms"`, `"1h30m"`, `"7d"`, /// `"1w"`, `"1y"`, or `"0"`) into a [`Duration`], matching `prometheus/common/model.ParseDuration`. pub fn parse_duration(s: &str) -> Result { diff --git a/src/discovery/kubernetes.rs b/src/discovery/kubernetes.rs new file mode 100644 index 0000000..3e086d0 --- /dev/null +++ b/src/discovery/kubernetes.rs @@ -0,0 +1,870 @@ +// Copyright 2026 The Prometheus Authors +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Prometheus-compatible Kubernetes service discovery (`kubernetes_sd_configs`) schema and +//! pod target group builder (`role: pod`). + +use std::collections::{BTreeMap, BTreeSet}; +use std::path::Path; + +use serde::{Deserialize, Deserializer, Serialize, Serializer}; + +use crate::http_config::HttpClientConfig; + +use super::{ + ADDRESS_LABEL, DiscoveryError, LabelSet, TargetGroup, join_host_port, sanitize_label_name, +}; + +/// Meta label prefix for all Kubernetes discovery labels (`"__meta_kubernetes_"`). +pub const META_LABEL_PREFIX: &str = "__meta_kubernetes_"; +/// Meta label storing the Kubernetes object namespace. +pub const NAMESPACE_LABEL: &str = "__meta_kubernetes_namespace"; +/// Meta label storing the Kubernetes node name. +pub const NODE_NAME_LABEL: &str = "__meta_kubernetes_node_name"; +/// Meta label storing the Kubernetes pod name. +pub const POD_NAME_LABEL: &str = "__meta_kubernetes_pod_name"; +/// Meta label storing the Kubernetes pod IP address. +pub const POD_IP_LABEL: &str = "__meta_kubernetes_pod_ip"; +/// Meta label storing the Kubernetes pod ready condition status (`"true"`, `"false"`, `"unknown"`). +pub const POD_READY_LABEL: &str = "__meta_kubernetes_pod_ready"; +/// Meta label storing the Kubernetes pod lifecycle phase. +pub const POD_PHASE_LABEL: &str = "__meta_kubernetes_pod_phase"; +/// Meta label storing the scheduled node name on the pod spec. +pub const POD_NODE_NAME_LABEL: &str = "__meta_kubernetes_pod_node_name"; +/// Meta label storing the host IP where the pod is running. +pub const POD_HOST_IP_LABEL: &str = "__meta_kubernetes_pod_host_ip"; +/// Meta label storing the pod UID. +pub const POD_UID_LABEL: &str = "__meta_kubernetes_pod_uid"; +/// Meta label storing the pod controller kind. +pub const POD_CONTROLLER_KIND_LABEL: &str = "__meta_kubernetes_pod_controller_kind"; +/// Meta label storing the pod controller name. +pub const POD_CONTROLLER_NAME_LABEL: &str = "__meta_kubernetes_pod_controller_name"; +/// Meta label storing the owning Deployment name when `attach_metadata.deployment` is enabled. +pub const POD_DEPLOYMENT_NAME_LABEL: &str = "__meta_kubernetes_pod_deployment_name"; +/// Meta label storing the owning Job name when `attach_metadata.job` is enabled. +pub const POD_JOB_NAME_LABEL: &str = "__meta_kubernetes_pod_job_name"; +/// Meta label storing the owning CronJob name when `attach_metadata.cronjob` is enabled. +pub const POD_CRONJOB_NAME_LABEL: &str = "__meta_kubernetes_pod_cronjob_name"; +/// Meta label storing the pod container name. +pub const POD_CONTAINER_NAME_LABEL: &str = "__meta_kubernetes_pod_container_name"; +/// Meta label storing the pod container runtime ID. +pub const POD_CONTAINER_ID_LABEL: &str = "__meta_kubernetes_pod_container_id"; +/// Meta label storing the pod container image. +pub const POD_CONTAINER_IMAGE_LABEL: &str = "__meta_kubernetes_pod_container_image"; +/// Meta label storing the pod container port name. +pub const POD_CONTAINER_PORT_NAME_LABEL: &str = "__meta_kubernetes_pod_container_port_name"; +/// Meta label storing the pod container port number. +pub const POD_CONTAINER_PORT_NUMBER_LABEL: &str = "__meta_kubernetes_pod_container_port_number"; +/// Meta label storing the pod container port protocol (`"TCP"`, `"UDP"`, `"SCTP"`). +pub const POD_CONTAINER_PORT_PROTOCOL_LABEL: &str = "__meta_kubernetes_pod_container_port_protocol"; +/// Meta label indicating whether the container is an `initContainer` (`"true"` or `"false"`). +pub const POD_CONTAINER_IS_INIT_LABEL: &str = "__meta_kubernetes_pod_container_init"; + +/// Kubernetes service discovery role matching Prometheus `discovery/kubernetes.Role`. +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub enum KubernetesRole { + /// Node discovery role (`"node"`). + Node, + /// Pod discovery role (`"pod"`). + Pod, + /// Service discovery role (`"service"`). + Service, + /// Endpoints discovery role (`"endpoints"`). + Endpoints, + /// EndpointSlice discovery role (`"endpointslice"`). + EndpointSlice, + /// Ingress discovery role (`"ingress"`). + Ingress, +} + +impl KubernetesRole { + /// Returns the lowercase Prometheus string representation of this role. + pub const fn as_str(self) -> &'static str { + match self { + Self::Node => "node", + Self::Pod => "pod", + Self::Service => "service", + Self::Endpoints => "endpoints", + Self::EndpointSlice => "endpointslice", + Self::Ingress => "ingress", + } + } + + /// Parses a role name matching Prometheus `Role.UnmarshalYAML`. + pub fn parse(s: &str) -> Result { + match s { + "node" => Ok(Self::Node), + "pod" => Ok(Self::Pod), + "service" => Ok(Self::Service), + "endpoints" => Ok(Self::Endpoints), + "endpointslice" => Ok(Self::EndpointSlice), + "ingress" => Ok(Self::Ingress), + _ => Err(DiscoveryError::InvalidConfig(format!( + "unknown Kubernetes SD role {s:?}" + ))), + } + } + + fn allowed_selector_roles(self) -> &'static [Self] { + match self { + Self::Pod => &[Self::Pod, Self::Node], + Self::Service => &[Self::Service], + Self::EndpointSlice => &[Self::Pod, Self::Service, Self::EndpointSlice], + Self::Endpoints => &[Self::Pod, Self::Service, Self::Endpoints], + Self::Node => &[Self::Node], + Self::Ingress => &[Self::Ingress], + } + } + + fn allowed_selector_error_msg(self) -> &'static str { + match self { + Self::Pod => "pod role supports only pod, node selectors", + Self::Service => "service role supports only service selectors", + Self::EndpointSlice => { + "endpointslice role supports only pod, service, endpointslice selectors" + } + Self::Endpoints => "endpoints role supports only pod, service, endpoints selectors", + Self::Node => "node role supports only node selectors", + Self::Ingress => "ingress role supports only ingress selectors", + } + } +} + +impl Serialize for KubernetesRole { + fn serialize(&self, serializer: S) -> Result { + serializer.serialize_str(self.as_str()) + } +} + +impl<'de> Deserialize<'de> for KubernetesRole { + fn deserialize>(deserializer: D) -> Result { + let s = String::deserialize(deserializer)?; + Self::parse(&s).map_err(serde::de::Error::custom) + } +} + +/// Namespace discovery filtering options matching Prometheus `kubernetes.NamespaceDiscovery`. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct KubernetesNamespaceDiscovery { + /// Includes the pod's own in-cluster namespace. + #[serde(default, skip_serializing_if = "is_false")] + pub own_namespace: bool, + /// Explicit list of namespaces to query. If empty and `own_namespace` is `false`, + /// queries across all namespaces. + #[serde( + default, + deserialize_with = "deserialize_null_default", + skip_serializing_if = "Vec::is_empty" + )] + pub names: Vec, +} + +/// Resource label/field selector matching Prometheus `kubernetes.SelectorConfig`. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct KubernetesSelectorConfig { + /// Kubernetes role to which this selector applies. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub role: Option, + /// Label selector string (`labelSelector` query parameter). + #[serde(default, skip_serializing_if = "String::is_empty")] + pub label: String, + /// Field selector string (`fieldSelector` query parameter). + #[serde(default, skip_serializing_if = "String::is_empty")] + pub field: String, +} + +/// Optional metadata attachment flags matching Prometheus `kubernetes.AttachMetadataConfig`. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct KubernetesAttachMetadataConfig { + /// Attaches node labels and annotations (`__meta_kubernetes_node_*`) and filters pods + /// by the node selector if configured. + #[serde(default, skip_serializing_if = "is_false")] + pub node: bool, + /// Attaches namespace labels and annotations (`__meta_kubernetes_namespace_*`). + #[serde(default, skip_serializing_if = "is_false")] + pub namespace: bool, + /// Resolves owning `ReplicaSet` -> `Deployment` name (`__meta_kubernetes_pod_deployment_name`). + #[serde(default, skip_serializing_if = "is_false")] + pub deployment: bool, + /// Attaches owning `Job` name (`__meta_kubernetes_pod_job_name`). + #[serde(default, skip_serializing_if = "is_false")] + pub job: bool, + /// Resolves owning `Job` -> `CronJob` name (`__meta_kubernetes_pod_cronjob_name`). + #[serde(default, skip_serializing_if = "is_false")] + pub cronjob: bool, +} + +/// Kubernetes service discovery configuration matching Prometheus `kubernetes.SDConfig`. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct KubernetesSdConfig { + /// Kubernetes API server URL. If omitted, in-cluster environment variables are used. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub api_server: Option, + /// Kubernetes discovery role (`"pod"`, `"service"`, `"endpoints"`, `"endpointslice"`, + /// `"node"`, or `"ingress"`). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub role: Option, + /// Path to a kubeconfig file (mutually exclusive with `api_server` and custom HTTP config). + #[serde(default, skip_serializing_if = "String::is_empty")] + pub kubeconfig_file: String, + /// Inlined Prometheus `HTTPClientConfig`. + #[serde(flatten)] + pub http_client_config: HttpClientConfig, + /// Namespace discovery configuration. + #[serde(default, skip_serializing_if = "is_default_namespaces")] + pub namespaces: KubernetesNamespaceDiscovery, + /// Label and field selectors per Kubernetes resource role. + #[serde( + default, + deserialize_with = "deserialize_null_default", + skip_serializing_if = "Vec::is_empty" + )] + pub selectors: Vec, + /// Additional metadata attachment flags. + #[serde(default, skip_serializing_if = "is_default_attach_metadata")] + pub attach_metadata: KubernetesAttachMetadataConfig, +} + +impl KubernetesSdConfig { + /// Joins relative file paths in `kubeconfig_file` and `http_client_config` with `dir`. + pub fn set_directory(&mut self, dir: &Path) { + self.http_client_config.set_directory(dir); + if !self.kubeconfig_file.is_empty() { + let p = Path::new(&self.kubeconfig_file); + if !p.is_absolute() { + self.kubeconfig_file = dir.join(p).to_string_lossy().into_owned(); + } + } + } + + /// Validates this configuration according to Prometheus `kubernetes.SDConfig.UnmarshalYAML`. + pub fn validate(&mut self) -> Result<(), DiscoveryError> { + let Some(role) = self.role else { + return Err(DiscoveryError::InvalidConfig( + "role missing (one of: pod, service, endpoints, endpointslice, node, ingress)" + .to_owned(), + )); + }; + + if self.api_server.as_deref() == Some("") { + self.api_server = None; + } + if let Some(api_server) = &self.api_server { + let parsed = reqwest::Url::parse(api_server).map_err(|e| { + DiscoveryError::InvalidConfig(format!("invalid api_server URL {api_server:?}: {e}")) + })?; + if !matches!(parsed.scheme(), "http" | "https") { + return Err(DiscoveryError::InvalidConfig(format!( + "invalid api_server URL scheme {:?}, expected \"http\" or \"https\"", + parsed.scheme() + ))); + } + if parsed.query().is_some() || parsed.fragment().is_some() { + return Err(DiscoveryError::InvalidConfig(format!( + "api_server URL {api_server:?} must not contain query parameters or fragments" + ))); + } + } + + for ns in &self.namespaces.names { + validate_k8s_namespace_name(ns)?; + } + + let is_default_http = self.http_client_config.is_default(); + self.http_client_config.validate()?; + + if self.api_server.is_some() && !self.kubeconfig_file.is_empty() { + return Err(DiscoveryError::InvalidConfig( + "cannot use 'kubeconfig_file' and 'api_server' simultaneously".to_owned(), + )); + } + if !self.kubeconfig_file.is_empty() && !is_default_http { + return Err(DiscoveryError::InvalidConfig( + "cannot use a custom HTTP client configuration together with 'kubeconfig_file'" + .to_owned(), + )); + } + if self.api_server.is_none() && !is_default_http { + return Err(DiscoveryError::InvalidConfig( + "to use custom HTTP client configuration please provide the 'api_server' URL explicitly".to_owned(), + )); + } + if self.api_server.is_some() && self.namespaces.own_namespace { + return Err(DiscoveryError::InvalidConfig( + "cannot use 'api_server' and 'namespaces.own_namespace' simultaneously".to_owned(), + )); + } + if !self.kubeconfig_file.is_empty() && self.namespaces.own_namespace { + return Err(DiscoveryError::InvalidConfig( + "cannot use 'kubeconfig_file' and 'namespaces.own_namespace' simultaneously" + .to_owned(), + )); + } + + let allowed = role.allowed_selector_roles(); + let mut seen_selector_roles = BTreeSet::new(); + for selector in &self.selectors { + let Some(sel_role) = selector.role else { + return Err(DiscoveryError::InvalidConfig( + "role missing in selector (one of: pod, service, endpoints, endpointslice, node, ingress)".to_owned(), + )); + }; + if !seen_selector_roles.insert(sel_role) { + return Err(DiscoveryError::InvalidConfig(format!( + "duplicated selector role: {}", + sel_role.as_str() + ))); + } + if !allowed.contains(&sel_role) { + return Err(DiscoveryError::InvalidConfig( + role.allowed_selector_error_msg().to_owned(), + )); + } + } + + Ok(()) + } +} + +fn is_false(b: &bool) -> bool { + !*b +} + +fn is_default_namespaces(ns: &KubernetesNamespaceDiscovery) -> bool { + *ns == KubernetesNamespaceDiscovery::default() +} + +fn is_default_attach_metadata(am: &KubernetesAttachMetadataConfig) -> bool { + *am == KubernetesAttachMetadataConfig::default() +} + +fn deserialize_null_default<'de, D, T>(deserializer: D) -> Result +where + D: Deserializer<'de>, + T: Default + Deserialize<'de>, +{ + Ok(Option::deserialize(deserializer)?.unwrap_or_default()) +} + +/// Kubernetes `metav1.OwnerReference` representation. +#[non_exhaustive] +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct OwnerReference { + /// Kind of the owning Kubernetes resource (e.g. `"ReplicaSet"`, `"DaemonSet"`, `"Job"`). + #[serde(default)] + pub kind: String, + /// Name of the owning Kubernetes resource. + #[serde(default)] + pub name: String, + /// Whether this reference points to the managing controller. + #[serde(default)] + pub controller: Option, +} + +/// Kubernetes `metav1.ObjectMeta` representation. +#[non_exhaustive] +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ObjectMeta { + /// Object name within its namespace. + #[serde(default)] + pub name: String, + /// Object namespace. + #[serde(default)] + pub namespace: String, + /// Unique identifier (`metadata.uid`). + #[serde(default)] + pub uid: String, + /// Map of string keys and values (`metadata.labels`). + #[serde(default, deserialize_with = "deserialize_null_default")] + pub labels: BTreeMap, + /// Map of unstructured annotations (`metadata.annotations`). + #[serde(default, deserialize_with = "deserialize_null_default")] + pub annotations: BTreeMap, + /// List of objects depended by this object (`metadata.ownerReferences`). + #[serde(default, deserialize_with = "deserialize_null_default")] + pub owner_references: Vec, +} + +impl ObjectMeta { + /// Returns the first [`OwnerReference`] with `controller == Some(true)`, matching + /// Kubernetes `metav1.GetControllerOf`. + pub fn controller_of(&self) -> Option<&OwnerReference> { + self.owner_references + .iter() + .find(|r| r.controller == Some(true)) + } +} + +/// Kubernetes container port definition (`corev1.ContainerPort`). +#[non_exhaustive] +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ContainerPort { + /// Optional port name. + #[serde(default)] + pub name: String, + /// Port number exposed on the pod's IP address. + #[serde(default)] + pub container_port: u32, + /// Protocol for port (`"TCP"`, `"UDP"`, or `"SCTP"`). + #[serde(default)] + pub protocol: String, +} + +/// Kubernetes container definition (`corev1.Container`). +#[non_exhaustive] +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct Container { + /// Container name. + #[serde(default)] + pub name: String, + /// Container image name. + #[serde(default)] + pub image: String, + /// List of ports to expose from the container. + #[serde(default, deserialize_with = "deserialize_null_default")] + pub ports: Vec, +} + +/// Kubernetes pod specification (`corev1.PodSpec`). +#[non_exhaustive] +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct PodSpec { + /// Scheduled node name (`spec.nodeName`). + #[serde(default)] + pub node_name: String, + /// Application containers belonging to the pod (`spec.containers`). + #[serde(default, deserialize_with = "deserialize_null_default")] + pub containers: Vec, + /// Initialization containers belonging to the pod (`spec.initContainers`). + #[serde(default, deserialize_with = "deserialize_null_default")] + pub init_containers: Vec, +} + +/// Kubernetes pod condition (`corev1.PodCondition`). +#[non_exhaustive] +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct PodCondition { + /// Type of pod condition (e.g. `"Ready"`). + #[serde(rename = "type", default)] + pub condition_type: String, + /// Status of the condition (`"True"`, `"False"`, `"Unknown"`). + #[serde(default)] + pub status: String, +} + +/// Kubernetes container runtime status (`corev1.ContainerStatus`). +#[non_exhaustive] +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ContainerStatus { + /// Container name. + #[serde(default)] + pub name: String, + /// Container ID in the format `://`. + #[serde(rename = "containerID", alias = "containerId", default)] + pub container_id: String, +} + +/// Kubernetes pod status (`corev1.PodStatus`). +#[non_exhaustive] +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct PodStatus { + /// Current lifecycle phase of the pod (`"Pending"`, `"Running"`, `"Succeeded"`, `"Failed"`, `"Unknown"`). + #[serde(default)] + pub phase: String, + /// IP address allocated to the pod (`status.podIP`). + #[serde(rename = "podIP", default)] + pub pod_ip: String, + /// IP address of the host to which the pod is assigned (`status.hostIP`). + #[serde(rename = "hostIP", default)] + pub host_ip: String, + /// Current service state conditions of the pod (`status.conditions`). + #[serde(default, deserialize_with = "deserialize_null_default")] + pub conditions: Vec, + /// Statuses of regular containers (`status.containerStatuses`). + #[serde(default, deserialize_with = "deserialize_null_default")] + pub container_statuses: Vec, + /// Statuses of init containers (`status.initContainerStatuses`). + #[serde(default, deserialize_with = "deserialize_null_default")] + pub init_container_statuses: Vec, +} + +/// Kubernetes `corev1.Pod` resource. +#[non_exhaustive] +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct Pod { + /// Standard object metadata. + #[serde(default)] + pub metadata: ObjectMeta, + /// Specification of the desired behavior of the pod. + #[serde(default)] + pub spec: PodSpec, + /// Most recently observed status of the pod. + #[serde(default)] + pub status: PodStatus, +} + +/// Kubernetes `corev1.Node` resource. +#[non_exhaustive] +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct Node { + /// Standard object metadata. + #[serde(default)] + pub metadata: ObjectMeta, +} + +/// Kubernetes `corev1.Namespace` resource. +#[non_exhaustive] +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct Namespace { + /// Standard object metadata. + #[serde(default)] + pub metadata: ObjectMeta, +} + +/// Kubernetes `appsv1.ReplicaSet` resource. +#[non_exhaustive] +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct ReplicaSet { + /// Standard object metadata. + #[serde(default)] + pub metadata: ObjectMeta, +} + +/// Kubernetes `batchv1.Job` resource. +#[non_exhaustive] +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct Job { + /// Standard object metadata. + #[serde(default)] + pub metadata: ObjectMeta, +} + +/// Auxiliary cluster metadata indexed for `attach_metadata` enrichment during pod discovery. +#[non_exhaustive] +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct KubernetesMetadataContext { + /// Nodes indexed by `metadata.name`. + pub nodes: BTreeMap, + /// Namespaces indexed by `metadata.name`. + pub namespaces: BTreeMap, + /// ReplicaSets indexed by `"{namespace}/{name}"`. + pub replicasets: BTreeMap, + /// Jobs indexed by `"{namespace}/{name}"`. + pub jobs: BTreeMap, +} + +impl KubernetesMetadataContext { + /// Indexes slices of nodes, namespaces, ReplicaSets, and Jobs into a lookup context. + pub fn from_items( + nodes: impl IntoIterator, + namespaces: impl IntoIterator, + replicasets: impl IntoIterator, + jobs: impl IntoIterator, + ) -> Self { + let nodes = nodes + .into_iter() + .map(|n| (n.metadata.name.clone(), n)) + .collect(); + let namespaces = namespaces + .into_iter() + .map(|ns| (ns.metadata.name.clone(), ns)) + .collect(); + let replicasets = replicasets + .into_iter() + .map(|rs| { + ( + namespaced_name(&rs.metadata.namespace, &rs.metadata.name), + rs, + ) + }) + .collect(); + let jobs = jobs + .into_iter() + .map(|j| (namespaced_name(&j.metadata.namespace, &j.metadata.name), j)) + .collect(); + Self { + nodes, + namespaces, + replicasets, + jobs, + } + } +} + +/// Builds a Prometheus [`TargetGroup`] for a single Kubernetes [`Pod`], matching +/// `discovery/kubernetes.Pod.buildPod` in Prometheus. +pub fn build_pod_target_group( + pod: &Pod, + metadata: &KubernetesMetadataContext, + attach: &KubernetesAttachMetadataConfig, +) -> TargetGroup { + let source = pod_source(&pod.metadata.namespace, &pod.metadata.name); + let mut tg = TargetGroup { + targets: Vec::new(), + labels: LabelSet::new(), + source, + }; + + // PodIP can be empty when a pod is starting or has been evicted. + if pod.status.pod_ip.is_empty() { + return tg; + } + + // Filter out pods scheduled on nodes that are not in the node store, as + // these were filtered out by node selectors. + if attach.node && !metadata.nodes.contains_key(&pod.spec.node_name) { + return tg; + } + + let mut labels = pod_labels(pod, metadata, attach); + labels.insert(NAMESPACE_LABEL.to_owned(), pod.metadata.namespace.clone()); + if attach.node { + add_node_labels(&mut labels, metadata, &pod.spec.node_name); + } + if attach.namespace { + add_namespace_labels(&mut labels, metadata, &pod.metadata.namespace); + } + tg.labels = labels; + + let regular_len = pod.spec.containers.len(); + for (idx, container) in pod + .spec + .containers + .iter() + .chain(pod.spec.init_containers.iter()) + .enumerate() + { + let is_init = idx >= regular_len; + let is_init_str = if is_init { "true" } else { "false" }; + let statuses = if is_init { + &pod.status.init_container_statuses + } else { + &pod.status.container_statuses + }; + let container_id = find_pod_container_id(statuses, &container.name); + + if container.ports.is_empty() { + let mut target = LabelSet::new(); + target.insert(ADDRESS_LABEL.to_owned(), pod.status.pod_ip.clone()); + target.insert(POD_CONTAINER_NAME_LABEL.to_owned(), container.name.clone()); + target.insert(POD_CONTAINER_ID_LABEL.to_owned(), container_id.to_owned()); + target.insert( + POD_CONTAINER_IMAGE_LABEL.to_owned(), + container.image.clone(), + ); + target.insert( + POD_CONTAINER_IS_INIT_LABEL.to_owned(), + is_init_str.to_owned(), + ); + tg.targets.push(target); + continue; + } + + for port in &container.ports { + let port_str = port.container_port.to_string(); + let addr = join_host_port(&pod.status.pod_ip, &port_str); + + let mut target = LabelSet::new(); + target.insert(ADDRESS_LABEL.to_owned(), addr); + target.insert(POD_CONTAINER_NAME_LABEL.to_owned(), container.name.clone()); + target.insert(POD_CONTAINER_ID_LABEL.to_owned(), container_id.to_owned()); + target.insert( + POD_CONTAINER_IMAGE_LABEL.to_owned(), + container.image.clone(), + ); + target.insert(POD_CONTAINER_PORT_NUMBER_LABEL.to_owned(), port_str); + target.insert(POD_CONTAINER_PORT_NAME_LABEL.to_owned(), port.name.clone()); + target.insert( + POD_CONTAINER_PORT_PROTOCOL_LABEL.to_owned(), + port.protocol.clone(), + ); + target.insert( + POD_CONTAINER_IS_INIT_LABEL.to_owned(), + is_init_str.to_owned(), + ); + tg.targets.push(target); + } + } + + tg +} + +/// Builds Prometheus [`TargetGroup`]s for a slice of Kubernetes [`Pod`]s. +pub fn build_pod_target_groups( + pods: &[Pod], + metadata: &KubernetesMetadataContext, + attach: &KubernetesAttachMetadataConfig, +) -> Vec { + pods.iter() + .map(|pod| build_pod_target_group(pod, metadata, attach)) + .collect() +} + +fn pod_labels( + pod: &Pod, + metadata: &KubernetesMetadataContext, + attach: &KubernetesAttachMetadataConfig, +) -> LabelSet { + let mut ls = LabelSet::new(); + ls.insert(POD_IP_LABEL.to_owned(), pod.status.pod_ip.clone()); + ls.insert(POD_READY_LABEL.to_owned(), pod_ready(pod)); + ls.insert(POD_PHASE_LABEL.to_owned(), pod.status.phase.clone()); + ls.insert(POD_NODE_NAME_LABEL.to_owned(), pod.spec.node_name.clone()); + ls.insert(POD_HOST_IP_LABEL.to_owned(), pod.status.host_ip.clone()); + ls.insert(POD_UID_LABEL.to_owned(), pod.metadata.uid.clone()); + + add_object_meta_labels(&mut ls, &pod.metadata, "pod"); + + if let Some(created_by) = pod.metadata.controller_of() { + if !created_by.kind.is_empty() { + ls.insert( + POD_CONTROLLER_KIND_LABEL.to_owned(), + created_by.kind.clone(), + ); + } + if !created_by.name.is_empty() { + ls.insert( + POD_CONTROLLER_NAME_LABEL.to_owned(), + created_by.name.clone(), + ); + } + match created_by.kind.as_str() { + "ReplicaSet" => { + if attach.deployment { + let key = namespaced_name(&pod.metadata.namespace, &created_by.name); + if let Some(rs) = metadata.replicasets.get(&key) + && let Some(rs_owner) = rs.metadata.controller_of() + && rs_owner.kind == "Deployment" + { + ls.insert(POD_DEPLOYMENT_NAME_LABEL.to_owned(), rs_owner.name.clone()); + } + } + } + "Job" => { + if attach.job { + ls.insert(POD_JOB_NAME_LABEL.to_owned(), created_by.name.clone()); + } + if attach.cronjob { + let key = namespaced_name(&pod.metadata.namespace, &created_by.name); + if let Some(job) = metadata.jobs.get(&key) + && let Some(job_owner) = job.metadata.controller_of() + && job_owner.kind == "CronJob" + { + ls.insert(POD_CRONJOB_NAME_LABEL.to_owned(), job_owner.name.clone()); + } + } + } + _ => {} + } + } + + ls +} + +fn add_node_labels(ls: &mut LabelSet, metadata: &KubernetesMetadataContext, node_name: &str) { + if node_name.is_empty() { + return; + } + ls.insert(NODE_NAME_LABEL.to_owned(), node_name.to_owned()); + if let Some(node) = metadata.nodes.get(node_name) { + add_object_meta_labels(ls, &node.metadata, "node"); + } +} + +fn add_namespace_labels(ls: &mut LabelSet, metadata: &KubernetesMetadataContext, namespace: &str) { + if let Some(ns) = metadata.namespaces.get(namespace) { + add_object_meta_labels(ls, &ns.metadata, "namespace"); + } +} + +fn add_object_meta_labels(ls: &mut LabelSet, meta: &ObjectMeta, role: &str) { + // Namespace name is already populated from `pod.metadata.namespace` (matching Prometheus + // `addNamespaceMetaLabels`), and empty auxiliary `meta.name` must not clobber `node_name`. + if role != "namespace" && !meta.name.is_empty() { + ls.insert(format!("{META_LABEL_PREFIX}{role}_name"), meta.name.clone()); + } + + for (k, v) in &meta.labels { + let ln = sanitize_label_name(k); + ls.insert(format!("{META_LABEL_PREFIX}{role}_label_{ln}"), v.clone()); + ls.insert( + format!("{META_LABEL_PREFIX}{role}_labelpresent_{ln}"), + "true".to_owned(), + ); + } + + for (k, v) in &meta.annotations { + let ln = sanitize_label_name(k); + ls.insert( + format!("{META_LABEL_PREFIX}{role}_annotation_{ln}"), + v.clone(), + ); + ls.insert( + format!("{META_LABEL_PREFIX}{role}_annotationpresent_{ln}"), + "true".to_owned(), + ); + } +} + +fn find_pod_container_id<'a>(statuses: &'a [ContainerStatus], container_name: &str) -> &'a str { + statuses + .iter() + .find(|s| s.name == container_name) + .map_or("", |s| s.container_id.as_str()) +} + +fn pod_ready(pod: &Pod) -> String { + for cond in &pod.status.conditions { + if cond.condition_type == "Ready" { + return cond.status.to_ascii_lowercase(); + } + } + "unknown".to_owned() +} + +fn namespaced_name(namespace: &str, name: &str) -> String { + format!("{namespace}/{name}") +} + +fn pod_source(namespace: &str, name: &str) -> String { + format!("pod/{}", namespaced_name(namespace, name)) +} + +fn validate_k8s_namespace_name(ns: &str) -> Result<(), DiscoveryError> { + let bytes = ns.as_bytes(); + let is_valid = !bytes.is_empty() + && bytes.len() <= 63 + && bytes[0].is_ascii_lowercase() | bytes[0].is_ascii_digit() + && bytes[bytes.len() - 1].is_ascii_lowercase() | bytes[bytes.len() - 1].is_ascii_digit() + && bytes + .iter() + .all(|&b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-'); + if !is_valid { + return Err(DiscoveryError::InvalidConfig(format!( + "invalid Kubernetes namespace name {ns:?}" + ))); + } + Ok(()) +} diff --git a/tests/discovery_kubernetes.rs b/tests/discovery_kubernetes.rs new file mode 100644 index 0000000..0fdf169 --- /dev/null +++ b/tests/discovery_kubernetes.rs @@ -0,0 +1,581 @@ +// Copyright 2026 The Prometheus Authors +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Integration tests for Kubernetes pod service discovery (`kubernetes_sd_configs` for `role: pod`). + +#![cfg(feature = "discovery-kubernetes")] + +use std::collections::BTreeMap; + +use expfmt::discovery::kubernetes::{ + Job, KubernetesAttachMetadataConfig, KubernetesMetadataContext, KubernetesNamespaceDiscovery, + KubernetesRole, KubernetesSdConfig, KubernetesSelectorConfig, Namespace, Node, OwnerReference, + ReplicaSet, build_pod_target_group, +}; +use expfmt::discovery::{ + ADDRESS_LABEL, Authorization, BasicAuth, DiscoveryConfigs, HttpClientConfig, HttpHeader, + HttpHeaders, LabelSet, OAuth2, Pod, TlsConfig, TlsVersion, +}; + +fn label_set(pairs: &[(&str, &str)]) -> LabelSet { + pairs + .iter() + .map(|(k, v)| ((*k).to_owned(), (*v).to_owned())) + .collect() +} + +#[test] +fn kubernetes_sd_config_validation_matches_prometheus() { + // Unknown fields on KubernetesSdConfig and DiscoveryConfigs are rejected even with flattened HttpClientConfig. + assert!( + serde_json::from_str::( + r#"{"role": "pod", "api_server": "https://127.0.0.1:6443", "bearer_tokens": "abc"}"# + ) + .is_err() + ); + let mut top_cfg: DiscoveryConfigs = serde_json::from_str( + r#"{"kubernetes_sd_configs": [{"role": "pod", "api_server": "https://127.0.0.1:6443", "bearer_token": "abc"}]}"#, + ) + .unwrap(); + top_cfg.validate().unwrap(); + + // Missing role. + let mut cfg = KubernetesSdConfig::default(); + assert_eq!( + cfg.validate().unwrap_err().to_string(), + "role missing (one of: pod, service, endpoints, endpointslice, node, ingress)" + ); + + // Both api_server and kubeconfig_file. + let mut cfg: KubernetesSdConfig = serde_json::from_str( + r#"{ + "role": "pod", + "api_server": "https://127.0.0.1:6443", + "kubeconfig_file": "/etc/kubeconfig" + }"#, + ) + .unwrap(); + assert_eq!( + cfg.validate().unwrap_err().to_string(), + "cannot use 'kubeconfig_file' and 'api_server' simultaneously" + ); + + // Custom HTTP client config without api_server. + let mut cfg: KubernetesSdConfig = serde_json::from_str( + r#"{ + "role": "pod", + "bearer_token": "secret-token" + }"#, + ) + .unwrap(); + assert_eq!( + cfg.validate().unwrap_err().to_string(), + "to use custom HTTP client configuration please provide the 'api_server' URL explicitly" + ); + + // api_server with namespaces.own_namespace. + let mut cfg: KubernetesSdConfig = serde_json::from_str( + r#"{ + "role": "pod", + "api_server": "https://127.0.0.1:6443", + "namespaces": { "own_namespace": true } + }"#, + ) + .unwrap(); + assert_eq!( + cfg.validate().unwrap_err().to_string(), + "cannot use 'api_server' and 'namespaces.own_namespace' simultaneously" + ); + + // Duplicated selector role. + let mut cfg: KubernetesSdConfig = serde_json::from_str( + r#"{ + "role": "pod", + "selectors": [ + { "role": "pod", "label": "app=web" }, + { "role": "pod", "field": "status.phase=Running" } + ] + }"#, + ) + .unwrap(); + assert_eq!( + cfg.validate().unwrap_err().to_string(), + "duplicated selector role: pod" + ); + + // Disallowed selector role for pod. + let mut cfg: KubernetesSdConfig = serde_json::from_str( + r#"{ + "role": "pod", + "selectors": [ + { "role": "service", "label": "app=web" } + ] + }"#, + ) + .unwrap(); + assert_eq!( + cfg.validate().unwrap_err().to_string(), + "pod role supports only pod, node selectors" + ); + + // HTTPClientConfig validation: bearer_token + bearer_token_file. + let mut cfg: KubernetesSdConfig = serde_json::from_str( + r#"{ + "role": "pod", + "api_server": "https://127.0.0.1:6443", + "bearer_token": "tok", + "bearer_token_file": "/path/to/tok" + }"#, + ) + .unwrap(); + assert_eq!( + cfg.validate().unwrap_err().to_string(), + "at most one of bearer_token & bearer_token_file must be configured" + ); + + // Reserved HTTP header. + let mut cfg: KubernetesSdConfig = serde_json::from_str( + r#"{ + "role": "pod", + "api_server": "https://127.0.0.1:6443", + "http_headers": { + "Authorization": { "values": ["Bearer foo"] } + } + }"#, + ) + .unwrap(); + assert_eq!( + cfg.validate().unwrap_err().to_string(), + "setting header \"Authorization\" is not allowed" + ); + + // Valid pod config with pod + node selectors, TLS, and custom header. + let mut cfg = KubernetesSdConfig { + api_server: Some("https://127.0.0.1:6443".to_owned()), + role: Some(KubernetesRole::Pod), + kubeconfig_file: String::new(), + http_client_config: HttpClientConfig { + basic_auth: Some(BasicAuth { + username: "prom".to_owned(), + password: "secret".to_owned(), + ..BasicAuth::default() + }), + tls_config: TlsConfig { + min_version: Some(TlsVersion::Tls12), + max_version: Some(TlsVersion::Tls13), + ..TlsConfig::default() + }, + http_headers: Some(HttpHeaders { + headers: BTreeMap::from([( + "X-Custom-Scope".to_owned(), + HttpHeader { + values: vec!["otap".to_owned()], + ..HttpHeader::default() + }, + )]), + }), + ..HttpClientConfig::default() + }, + namespaces: KubernetesNamespaceDiscovery { + own_namespace: false, + names: vec!["default".to_owned(), "monitoring".to_owned()], + }, + selectors: vec![ + KubernetesSelectorConfig { + role: Some(KubernetesRole::Pod), + label: "app=prometheus".to_owned(), + field: "status.phase=Running".to_owned(), + }, + KubernetesSelectorConfig { + role: Some(KubernetesRole::Node), + label: "region=eu-west-1".to_owned(), + field: String::new(), + }, + ], + attach_metadata: KubernetesAttachMetadataConfig { + node: true, + namespace: true, + deployment: true, + job: true, + cronjob: true, + }, + }; + cfg.validate().expect("valid KubernetesSdConfig"); + + // set_directory resolves relative paths across kubeconfig_file and http_client_config. + let mut dir_cfg = KubernetesSdConfig { + kubeconfig_file: "certs/kubeconfig".to_owned(), + http_client_config: HttpClientConfig { + bearer_token_file: "secrets/token".to_owned(), + authorization: Some(Authorization { + credentials_file: "secrets/creds".to_owned(), + ..Authorization::default() + }), + oauth2: Some(OAuth2 { + client_secret_file: "oauth/secret".to_owned(), + ..OAuth2::default() + }), + ..HttpClientConfig::default() + }, + ..KubernetesSdConfig::default() + }; + dir_cfg.set_directory(std::path::Path::new("/etc/prometheus")); + assert_eq!(dir_cfg.kubeconfig_file, "/etc/prometheus/certs/kubeconfig"); + assert_eq!( + dir_cfg.http_client_config.bearer_token_file, + "/etc/prometheus/secrets/token" + ); +} + +#[test] +fn kubernetes_pod_target_groups_match_prometheus_pod_test() { + // Ported from TestPodDiscoveryAdd in discovery/kubernetes/pod_test.go. + let empty_meta = KubernetesMetadataContext::default(); + let no_attach = KubernetesAttachMetadataConfig::default(); + + // 1. Single container + multiple ports + portless container + initContainer. + let pod: Pod = serde_json::from_value(serde_json::json!({ + "metadata": { + "name": "testpod", + "namespace": "default", + "uid": "deadbeef", + "labels": { "test/label": "value1" }, + "annotations": { "test/annotation": "value2" }, + "ownerReferences": [{ + "kind": "ReplicaSet", + "name": "testdeployment-4228282135", + "controller": true + }] + }, + "spec": { + "nodeName": "testnode", + "containers": [ + { + "name": "c1", + "image": "c1:latest", + "ports": [ + { "name": "mainport", "containerPort": 9000, "protocol": "TCP" }, + { "name": "mgmtport", "containerPort": 9001, "protocol": "UDP" } + ] + }, + { + "name": "c2_portless", + "image": "c2:latest", + "ports": [] + } + ], + "initContainers": [{ + "name": "init_c", + "image": "busybox:latest", + "ports": [{ "name": "initport", "containerPort": 9100, "protocol": "TCP" }] + }] + }, + "status": { + "phase": "Running", + "podIP": "10.0.0.1", + "hostIP": "192.168.1.10", + "conditions": [{ "type": "Ready", "status": "True" }], + "containerStatuses": [ + { "name": "c1", "containerID": "docker://c1-id" }, + { "name": "c2_portless", "containerID": "docker://c2-id" } + ], + "initContainerStatuses": [ + { "name": "init_c", "containerID": "docker://init-id" } + ] + } + })) + .unwrap(); + + let node: Node = serde_json::from_value(serde_json::json!({ + "metadata": { + "name": "testnode", + "labels": { "node.k8s.io/zone": "us-east1-b" }, + "annotations": { "node.k8s.io/ann": "nval" } + } + })) + .unwrap(); + let ns: Namespace = serde_json::from_value(serde_json::json!({ + "metadata": { + "name": "default", + "labels": { "ns-label": "nsval" }, + "annotations": { "ns-ann": "nsannval" } + } + })) + .unwrap(); + let rs: ReplicaSet = serde_json::from_value(serde_json::json!({ + "metadata": { + "name": "testdeployment-4228282135", + "namespace": "default", + "ownerReferences": [{ + "kind": "Deployment", + "name": "testdeployment", + "controller": true + }] + } + })) + .unwrap(); + let meta = KubernetesMetadataContext::from_items(vec![node], vec![ns], vec![rs], vec![]); + let attach_all = KubernetesAttachMetadataConfig { + node: true, + namespace: true, + deployment: true, + job: true, + cronjob: true, + }; + + let tg = build_pod_target_group(&pod, &meta, &attach_all); + assert_eq!(tg.source, "pod/default/testpod"); + assert_eq!( + tg.labels, + label_set(&[ + ("__meta_kubernetes_namespace", "default"), + ("__meta_kubernetes_namespace_annotation_ns_ann", "nsannval"), + ( + "__meta_kubernetes_namespace_annotationpresent_ns_ann", + "true" + ), + ("__meta_kubernetes_namespace_label_ns_label", "nsval"), + ("__meta_kubernetes_namespace_labelpresent_ns_label", "true"), + ("__meta_kubernetes_node_annotation_node_k8s_io_ann", "nval"), + ( + "__meta_kubernetes_node_annotationpresent_node_k8s_io_ann", + "true" + ), + ( + "__meta_kubernetes_node_label_node_k8s_io_zone", + "us-east1-b" + ), + ( + "__meta_kubernetes_node_labelpresent_node_k8s_io_zone", + "true" + ), + ("__meta_kubernetes_node_name", "testnode"), + ("__meta_kubernetes_pod_annotation_test_annotation", "value2"), + ( + "__meta_kubernetes_pod_annotationpresent_test_annotation", + "true" + ), + ("__meta_kubernetes_pod_controller_kind", "ReplicaSet"), + ( + "__meta_kubernetes_pod_controller_name", + "testdeployment-4228282135" + ), + ("__meta_kubernetes_pod_deployment_name", "testdeployment"), + ("__meta_kubernetes_pod_host_ip", "192.168.1.10"), + ("__meta_kubernetes_pod_ip", "10.0.0.1"), + ("__meta_kubernetes_pod_label_test_label", "value1"), + ("__meta_kubernetes_pod_labelpresent_test_label", "true"), + ("__meta_kubernetes_pod_name", "testpod"), + ("__meta_kubernetes_pod_node_name", "testnode"), + ("__meta_kubernetes_pod_phase", "Running"), + ("__meta_kubernetes_pod_ready", "true"), + ("__meta_kubernetes_pod_uid", "deadbeef"), + ]) + ); + assert_eq!( + tg.targets, + vec![ + label_set(&[ + ("__address__", "10.0.0.1:9000"), + ("__meta_kubernetes_pod_container_id", "docker://c1-id"), + ("__meta_kubernetes_pod_container_image", "c1:latest"), + ("__meta_kubernetes_pod_container_init", "false"), + ("__meta_kubernetes_pod_container_name", "c1"), + ("__meta_kubernetes_pod_container_port_name", "mainport"), + ("__meta_kubernetes_pod_container_port_number", "9000"), + ("__meta_kubernetes_pod_container_port_protocol", "TCP"), + ]), + label_set(&[ + ("__address__", "10.0.0.1:9001"), + ("__meta_kubernetes_pod_container_id", "docker://c1-id"), + ("__meta_kubernetes_pod_container_image", "c1:latest"), + ("__meta_kubernetes_pod_container_init", "false"), + ("__meta_kubernetes_pod_container_name", "c1"), + ("__meta_kubernetes_pod_container_port_name", "mgmtport"), + ("__meta_kubernetes_pod_container_port_number", "9001"), + ("__meta_kubernetes_pod_container_port_protocol", "UDP"), + ]), + label_set(&[ + ("__address__", "10.0.0.1"), + ("__meta_kubernetes_pod_container_id", "docker://c2-id"), + ("__meta_kubernetes_pod_container_image", "c2:latest"), + ("__meta_kubernetes_pod_container_init", "false"), + ("__meta_kubernetes_pod_container_name", "c2_portless"), + ]), + label_set(&[ + ("__address__", "10.0.0.1:9100"), + ("__meta_kubernetes_pod_container_id", "docker://init-id"), + ("__meta_kubernetes_pod_container_image", "busybox:latest"), + ("__meta_kubernetes_pod_container_init", "true"), + ("__meta_kubernetes_pod_container_name", "init_c"), + ("__meta_kubernetes_pod_container_port_name", "initport"), + ("__meta_kubernetes_pod_container_port_number", "9100"), + ("__meta_kubernetes_pod_container_port_protocol", "TCP"), + ]), + ] + ); + + // 2. Pending / Evicted pod without PodIP returns empty TargetGroup. + let mut pending_pod = pod.clone(); + pending_pod.status.pod_ip.clear(); + let pending_tg = build_pod_target_group(&pending_pod, &empty_meta, &no_attach); + assert_eq!(pending_tg.source, "pod/default/testpod"); + assert!(pending_tg.targets.is_empty()); + assert!(pending_tg.labels.is_empty()); + + // 3. Node filtering when attach_metadata.node = true and pod's node is absent from store. + let filtered_tg = build_pod_target_group(&pod, &empty_meta, &attach_all); + assert_eq!(filtered_tg.source, "pod/default/testpod"); + assert!(filtered_tg.targets.is_empty()); + assert!(filtered_tg.labels.is_empty()); + + // 4. Job -> CronJob resolution when attach_metadata.job and attach_metadata.cronjob are true. + let mut job_owner = OwnerReference::default(); + job_owner.kind = "Job".to_owned(); + job_owner.name = "nightly-job-123".to_owned(); + job_owner.controller = Some(true); + let mut job_pod = pod; + job_pod.metadata.owner_references = vec![job_owner]; + let job: Job = serde_json::from_value(serde_json::json!({ + "metadata": { + "name": "nightly-job-123", + "namespace": "default", + "ownerReferences": [{ + "kind": "CronJob", + "name": "nightly-job", + "controller": true + }] + } + })) + .unwrap(); + let job_meta = KubernetesMetadataContext::from_items(vec![], vec![], vec![], vec![job]); + let job_attach = KubernetesAttachMetadataConfig { + job: true, + cronjob: true, + ..KubernetesAttachMetadataConfig::default() + }; + let job_tg = build_pod_target_group(&job_pod, &job_meta, &job_attach); + assert_eq!( + job_tg.labels.get("__meta_kubernetes_pod_controller_kind"), + Some(&"Job".to_owned()) + ); + assert_eq!( + job_tg.labels.get("__meta_kubernetes_pod_job_name"), + Some(&"nightly-job-123".to_owned()) + ); + assert_eq!( + job_tg.labels.get("__meta_kubernetes_pod_cronjob_name"), + Some(&"nightly-job".to_owned()) + ); +} + +#[test] +fn kubernetes_model_and_validation_edge_cases() { + assert!(serde_json::from_str::(r#"{"role": "POD"}"#).is_err()); + let mut bad_scheme: KubernetesSdConfig = + serde_json::from_str(r#"{"role": "pod", "api_server": "ftp://k8s.local"}"#).unwrap(); + assert!(bad_scheme.validate().is_err()); + let mut bad_query: KubernetesSdConfig = + serde_json::from_str(r#"{"role": "pod", "api_server": "https://k8s.local?watch=true"}"#) + .unwrap(); + assert!(bad_query.validate().is_err()); + let mut bad_ns: KubernetesSdConfig = + serde_json::from_str(r#"{"role": "pod", "namespaces": {"names": ["default?watch=true"]}}"#) + .unwrap(); + assert!(bad_ns.validate().is_err()); + + let ipv6_null_pod: Pod = serde_json::from_str( + r#"{ + "metadata": { + "name": "v6pod", + "namespace": "prod", + "labels": null, + "annotations": null, + "ownerReferences": [ + {"kind": "Job", "name": "non-controller", "controller": false}, + {"kind": "Job", "name": "cron-child-1", "controller": true} + ] + }, + "spec": { + "nodeName": "n1", + "containers": [ + {"name": "app", "image": "app:v1", "ports": [{"name": "http", "containerPort": 8080, "protocol": "TCP"}]}, + {"name": "sidecar", "image": "sc:v1", "ports": null} + ], + "initContainers": null + }, + "status": { + "phase": "Running", + "podIP": "2001:db8::9", + "conditions": [{"type": "Ready", "status": "False"}], + "containerStatuses": null, + "initContainerStatuses": null + } + }"#, + ) + .unwrap(); + + // Nameless Namespace entry in KubernetesMetadataContext must NOT clobber __meta_kubernetes_namespace. + let ns_without_name: Namespace = + serde_json::from_str(r#"{"metadata": {"labels": {"team": "obs"}}}"#).unwrap(); + let job_item: Job = serde_json::from_str( + r#"{"metadata": {"name": "cron-child-1", "namespace": "prod", "ownerReferences": [{"kind": "CronJob", "name": "hourly", "controller": true}]}}"#, + ) + .unwrap(); + let mut ctx = KubernetesMetadataContext::from_items(vec![], vec![], vec![], vec![job_item]); + ctx.namespaces.insert("prod".to_owned(), ns_without_name); + + let tg_cron_only = build_pod_target_group( + &ipv6_null_pod, + &ctx, + &KubernetesAttachMetadataConfig { + namespace: true, + job: false, + cronjob: true, + ..KubernetesAttachMetadataConfig::default() + }, + ); + assert_eq!( + tg_cron_only.labels.get("__meta_kubernetes_namespace"), + Some(&"prod".to_owned()) + ); + assert_eq!( + tg_cron_only + .labels + .get("__meta_kubernetes_namespace_label_team"), + Some(&"obs".to_owned()) + ); + assert_eq!( + tg_cron_only.labels.get("__meta_kubernetes_pod_ready"), + Some(&"false".to_owned()) + ); + assert!( + !tg_cron_only + .labels + .contains_key("__meta_kubernetes_pod_job_name") + ); + assert_eq!( + tg_cron_only + .labels + .get("__meta_kubernetes_pod_cronjob_name"), + Some(&"hourly".to_owned()) + ); + assert_eq!( + tg_cron_only.targets[0].get(ADDRESS_LABEL), + Some(&"[2001:db8::9]:8080".to_owned()) + ); + assert_eq!( + tg_cron_only.targets[1].get(ADDRESS_LABEL), + Some(&"2001:db8::9".to_owned()) + ); +} From b87ab0ef6fe6a0a587f94311309e9769f4728d8c Mon Sep 17 00:00:00 2001 From: Bartek Plotka Date: Tue, 10 Nov 2026 17:09:26 +0000 Subject: [PATCH 2/2] discovery/kubernetes: add reqwest-based KubernetesPodDiscovery runner Query the Kubernetes API server over reqwest for pods and optional attach_metadata resources (nodes, namespaces with 404 tolerance, ReplicaSets, Jobs), supporting explicit api_server, in-cluster env/token/CA resolution, and !Send LocalSet execution. Signed-off-by: Bartek Plotka --- src/discovery.rs | 4 +- src/discovery/kubernetes.rs | 533 +++++++++++++++++++++++++++++++++- tests/discovery_kubernetes.rs | 421 ++++++++++++++++++++++++++- 3 files changed, 952 insertions(+), 6 deletions(-) diff --git a/src/discovery.rs b/src/discovery.rs index 7d7e14d..34f3752 100644 --- a/src/discovery.rs +++ b/src/discovery.rs @@ -32,8 +32,8 @@ pub use dns::{DnsDiscovery, DnsRecord, DnsRecordType, DnsSdConfig, ResolvConf}; #[cfg(feature = "discovery-kubernetes")] pub use kubernetes::{ KubernetesAttachMetadataConfig, KubernetesMetadataContext, KubernetesNamespaceDiscovery, - KubernetesRole, KubernetesSdConfig, KubernetesSelectorConfig, Pod, build_pod_target_group, - build_pod_target_groups, + KubernetesPodDiscovery, KubernetesRole, KubernetesSdConfig, KubernetesSelectorConfig, Pod, + build_pod_target_group, build_pod_target_groups, }; /// Reserved Prometheus label holding the `:` scrape target address. diff --git a/src/discovery/kubernetes.rs b/src/discovery/kubernetes.rs index 3e086d0..a733713 100644 --- a/src/discovery/kubernetes.rs +++ b/src/discovery/kubernetes.rs @@ -11,10 +11,12 @@ // See the License for the specific language governing permissions and // limitations under the License. -//! Prometheus-compatible Kubernetes service discovery (`kubernetes_sd_configs`) schema and -//! pod target group builder (`role: pod`). +//! Prometheus-compatible Kubernetes service discovery (`kubernetes_sd_configs`), +//! with `reqwest`-based discovery for `role: pod`. use std::collections::{BTreeMap, BTreeSet}; +use std::env; +use std::fs; use std::path::Path; use serde::{Deserialize, Deserializer, Serialize, Serializer}; @@ -25,6 +27,10 @@ use super::{ ADDRESS_LABEL, DiscoveryError, LabelSet, TargetGroup, join_host_port, sanitize_label_name, }; +const IN_CLUSTER_TOKEN_FILE: &str = "/var/run/secrets/kubernetes.io/serviceaccount/token"; +const IN_CLUSTER_CA_FILE: &str = "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt"; +const IN_CLUSTER_NAMESPACE_FILE: &str = "/var/run/secrets/kubernetes.io/serviceaccount/namespace"; + /// Meta label prefix for all Kubernetes discovery labels (`"__meta_kubernetes_"`). pub const META_LABEL_PREFIX: &str = "__meta_kubernetes_"; /// Meta label storing the Kubernetes object namespace. @@ -338,6 +344,10 @@ impl KubernetesSdConfig { Ok(()) } + + fn selector_for(&self, role: KubernetesRole) -> Option<&KubernetesSelectorConfig> { + self.selectors.iter().find(|s| s.role == Some(role)) + } } fn is_false(b: &bool) -> bool { @@ -525,6 +535,14 @@ pub struct Pod { pub status: PodStatus, } +/// Kubernetes `corev1.PodList` response payload. +#[derive(Clone, Debug, Default, PartialEq, Eq, Deserialize)] +pub(crate) struct PodList { + /// List of pods. + #[serde(default, deserialize_with = "deserialize_null_default")] + pub items: Vec, +} + /// Kubernetes `corev1.Node` resource. #[non_exhaustive] #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] @@ -534,6 +552,14 @@ pub struct Node { pub metadata: ObjectMeta, } +/// Kubernetes `corev1.NodeList` response payload. +#[derive(Clone, Debug, Default, PartialEq, Eq, Deserialize)] +pub(crate) struct NodeList { + /// List of nodes. + #[serde(default, deserialize_with = "deserialize_null_default")] + pub items: Vec, +} + /// Kubernetes `corev1.Namespace` resource. #[non_exhaustive] #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] @@ -543,6 +569,14 @@ pub struct Namespace { pub metadata: ObjectMeta, } +/// Kubernetes `corev1.NamespaceList` response payload. +#[derive(Clone, Debug, Default, PartialEq, Eq, Deserialize)] +pub(crate) struct NamespaceList { + /// List of namespaces. + #[serde(default, deserialize_with = "deserialize_null_default")] + pub items: Vec, +} + /// Kubernetes `appsv1.ReplicaSet` resource. #[non_exhaustive] #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] @@ -552,6 +586,14 @@ pub struct ReplicaSet { pub metadata: ObjectMeta, } +/// Kubernetes `appsv1.ReplicaSetList` response payload. +#[derive(Clone, Debug, Default, PartialEq, Eq, Deserialize)] +pub(crate) struct ReplicaSetList { + /// List of ReplicaSets. + #[serde(default, deserialize_with = "deserialize_null_default")] + pub items: Vec, +} + /// Kubernetes `batchv1.Job` resource. #[non_exhaustive] #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] @@ -561,6 +603,14 @@ pub struct Job { pub metadata: ObjectMeta, } +/// Kubernetes `batchv1.JobList` response payload. +#[derive(Clone, Debug, Default, PartialEq, Eq, Deserialize)] +pub(crate) struct JobList { + /// List of Jobs. + #[serde(default, deserialize_with = "deserialize_null_default")] + pub items: Vec, +} + /// Auxiliary cluster metadata indexed for `attach_metadata` enrichment during pod discovery. #[non_exhaustive] #[derive(Clone, Debug, Default, PartialEq, Eq)] @@ -852,6 +902,330 @@ fn pod_source(namespace: &str, name: &str) -> String { format!("pod/{}", namespaced_name(namespace, name)) } +/// `reqwest`-based Kubernetes Pod service discovery runner (`role: pod`). +#[derive(Clone, Debug)] +pub struct KubernetesPodDiscovery { + config: KubernetesSdConfig, + client: Option, + own_namespace_file: String, + in_cluster_token_file: String, + in_cluster_ca_file: String, +} + +impl KubernetesPodDiscovery { + /// Creates a new [`KubernetesPodDiscovery`] after validating `config` (`role` must be `Pod`) + /// and building a [`reqwest::Client`] when `api_server` is configured. + /// + /// In-cluster mode (`api_server: None`) builds its TLS client using the service account CA + /// bundle and token when [`KubernetesPodDiscovery::discover`] runs. + pub fn new(mut config: KubernetesSdConfig) -> Result { + validate_pod_runner_config(&mut config)?; + let client = if config.api_server.is_some() { + Some(config.http_client_config.build_client()?) + } else { + None + }; + Ok(Self { + config, + client, + own_namespace_file: IN_CLUSTER_NAMESPACE_FILE.to_owned(), + in_cluster_token_file: IN_CLUSTER_TOKEN_FILE.to_owned(), + in_cluster_ca_file: IN_CLUSTER_CA_FILE.to_owned(), + }) + } + + /// Creates a new [`KubernetesPodDiscovery`] using a caller-supplied [`reqwest::Client`]. + pub fn with_client( + mut config: KubernetesSdConfig, + client: reqwest::Client, + ) -> Result { + validate_pod_runner_config(&mut config)?; + Ok(Self { + config, + client: Some(client), + own_namespace_file: IN_CLUSTER_NAMESPACE_FILE.to_owned(), + in_cluster_token_file: IN_CLUSTER_TOKEN_FILE.to_owned(), + in_cluster_ca_file: IN_CLUSTER_CA_FILE.to_owned(), + }) + } + + /// Overrides the path used to read the in-cluster service account namespace when + /// `namespaces.own_namespace` is `true` (defaults to + /// `"/var/run/secrets/kubernetes.io/serviceaccount/namespace"`). + pub fn with_own_namespace_file(mut self, path: impl Into) -> Self { + self.own_namespace_file = path.into(); + self + } + + /// Returns the validated [`KubernetesSdConfig`]. + pub fn config(&self) -> &KubernetesSdConfig { + &self.config + } + + /// Queries the Kubernetes API server over HTTP(S) using `reqwest` and returns the discovered + /// pod [`TargetGroup`]s. + pub async fn discover(&self) -> Result, DiscoveryError> { + let (base_url, client, effective_http) = self.resolve_endpoint_and_auth()?; + let namespaces = self.resolve_namespaces()?; + + let pod_selector = self.config.selector_for(KubernetesRole::Pod); + let node_selector = self.config.selector_for(KubernetesRole::Node); + + let mut all_pods = Vec::new(); + if namespaces.is_empty() { + let url = build_resource_url(&base_url, &["api", "v1", "pods"], pod_selector)?; + let list: PodList = self.get_json(&client, url, &effective_http).await?; + all_pods.extend(list.items); + } else { + for ns in &namespaces { + let url = build_resource_url( + &base_url, + &["api", "v1", "namespaces", ns, "pods"], + pod_selector, + )?; + let list: PodList = self.get_json(&client, url, &effective_http).await?; + all_pods.extend(list.items); + } + } + + let mut metadata = KubernetesMetadataContext::default(); + let attach = &self.config.attach_metadata; + + if attach.node { + let url = build_resource_url(&base_url, &["api", "v1", "nodes"], node_selector)?; + let list: NodeList = self.get_json(&client, url, &effective_http).await?; + for node in list.items { + metadata.nodes.insert(node.metadata.name.clone(), node); + } + } + + if attach.namespace { + if namespaces.is_empty() { + let url = build_resource_url(&base_url, &["api", "v1", "namespaces"], None)?; + let list: NamespaceList = self.get_json(&client, url, &effective_http).await?; + for ns in list.items { + metadata.namespaces.insert(ns.metadata.name.clone(), ns); + } + } else { + for ns_name in &namespaces { + let url = + build_resource_url(&base_url, &["api", "v1", "namespaces", ns_name], None)?; + if let Some(ns) = self + .get_optional_json::(&client, url, &effective_http) + .await? + { + metadata.namespaces.insert(ns.metadata.name.clone(), ns); + } + } + } + } + + if attach.deployment { + if namespaces.is_empty() { + let url = + build_resource_url(&base_url, &["apis", "apps", "v1", "replicasets"], None)?; + let list: ReplicaSetList = self.get_json(&client, url, &effective_http).await?; + for rs in list.items { + metadata.replicasets.insert( + namespaced_name(&rs.metadata.namespace, &rs.metadata.name), + rs, + ); + } + } else { + for ns in &namespaces { + let url = build_resource_url( + &base_url, + &["apis", "apps", "v1", "namespaces", ns, "replicasets"], + None, + )?; + let list: ReplicaSetList = self.get_json(&client, url, &effective_http).await?; + for rs in list.items { + metadata.replicasets.insert( + namespaced_name(&rs.metadata.namespace, &rs.metadata.name), + rs, + ); + } + } + } + } + + if attach.cronjob { + if namespaces.is_empty() { + let url = build_resource_url(&base_url, &["apis", "batch", "v1", "jobs"], None)?; + let list: JobList = self.get_json(&client, url, &effective_http).await?; + for job in list.items { + metadata.jobs.insert( + namespaced_name(&job.metadata.namespace, &job.metadata.name), + job, + ); + } + } else { + for ns in &namespaces { + let url = build_resource_url( + &base_url, + &["apis", "batch", "v1", "namespaces", ns, "jobs"], + None, + )?; + let list: JobList = self.get_json(&client, url, &effective_http).await?; + for job in list.items { + metadata.jobs.insert( + namespaced_name(&job.metadata.namespace, &job.metadata.name), + job, + ); + } + } + } + } + + Ok(build_pod_target_groups(&all_pods, &metadata, attach)) + } + + async fn get_json Deserialize<'de>>( + &self, + client: &reqwest::Client, + url: reqwest::Url, + http_cfg: &HttpClientConfig, + ) -> Result { + let req = client + .get(url) + .header("Accept", "application/json") + .header("User-Agent", "expfmt-rs/0.1"); + let req = http_cfg.apply_to_request(req)?; + let resp = req.send().await?.error_for_status()?; + let body = resp.bytes().await?; + Ok(serde_json::from_slice(&body)?) + } + + async fn get_optional_json Deserialize<'de>>( + &self, + client: &reqwest::Client, + url: reqwest::Url, + http_cfg: &HttpClientConfig, + ) -> Result, DiscoveryError> { + let req = client + .get(url) + .header("Accept", "application/json") + .header("User-Agent", "expfmt-rs/0.1"); + let req = http_cfg.apply_to_request(req)?; + let resp = req.send().await?; + if resp.status() == reqwest::StatusCode::NOT_FOUND { + return Ok(None); + } + let resp = resp.error_for_status()?; + let body = resp.bytes().await?; + Ok(Some(serde_json::from_slice(&body)?)) + } + + fn resolve_endpoint_and_auth( + &self, + ) -> Result<(String, reqwest::Client, HttpClientConfig), DiscoveryError> { + if let Some(api_server) = &self.config.api_server { + let client = match &self.client { + Some(c) => c.clone(), + None => self.config.http_client_config.build_client()?, + }; + return Ok(( + api_server.trim_end_matches('/').to_owned(), + client, + self.config.http_client_config.clone(), + )); + } + + let host = env::var("KUBERNETES_SERVICE_HOST").unwrap_or_default(); + let port = env::var("KUBERNETES_SERVICE_PORT").unwrap_or_default(); + self.resolve_in_cluster_endpoint_and_auth(&host, &port) + } + + fn resolve_in_cluster_endpoint_and_auth( + &self, + host: &str, + port: &str, + ) -> Result<(String, reqwest::Client, HttpClientConfig), DiscoveryError> { + // In-cluster configuration matching client-go rest.InClusterConfig(). + if host.is_empty() || port.is_empty() { + return Err(DiscoveryError::InvalidConfig( + "unable to load in-cluster configuration, KUBERNETES_SERVICE_HOST and KUBERNETES_SERVICE_PORT must be defined".to_owned(), + )); + } + let base_url = format!("https://{}", join_host_port(host, port)); + let mut http_cfg = self.config.http_client_config.clone(); + if http_cfg.basic_auth.is_none() + && http_cfg.authorization.is_none() + && http_cfg.bearer_token.is_empty() + && http_cfg.bearer_token_file.is_empty() + { + let token = fs::read_to_string(&self.in_cluster_token_file).map_err(|e| { + DiscoveryError::InvalidConfig(format!( + "unable to read in-cluster token file {}: {e}", + self.in_cluster_token_file + )) + })?; + if token.trim().is_empty() { + return Err(DiscoveryError::InvalidConfig(format!( + "in-cluster token file {} is empty", + self.in_cluster_token_file + ))); + } + http_cfg + .bearer_token_file + .clone_from(&self.in_cluster_token_file); + } + let client = match &self.client { + Some(c) => c.clone(), + None => { + if http_cfg.tls_config.ca.is_empty() + && http_cfg.tls_config.ca_file.is_empty() + && !http_cfg.tls_config.insecure_skip_verify + { + if !Path::new(&self.in_cluster_ca_file).exists() { + return Err(DiscoveryError::InvalidConfig(format!( + "unable to read in-cluster CA certificate file {}", + self.in_cluster_ca_file + ))); + } + http_cfg + .tls_config + .ca_file + .clone_from(&self.in_cluster_ca_file); + } + http_cfg.build_client()? + } + }; + Ok((base_url, client, http_cfg)) + } + + fn resolve_namespaces(&self) -> Result, DiscoveryError> { + let mut names = Vec::with_capacity(self.config.namespaces.names.len() + 1); + for ns in &self.config.namespaces.names { + validate_k8s_namespace_name(ns)?; + if !names.contains(ns) { + names.push(ns.clone()); + } + } + if self.config.namespaces.own_namespace { + let own_ns = fs::read_to_string(&self.own_namespace_file) + .map_err(|e| { + DiscoveryError::InvalidConfig(format!( + "unable to read in-cluster namespace file {}: {e}", + self.own_namespace_file + )) + })? + .trim() + .to_owned(); + if own_ns.is_empty() { + return Err(DiscoveryError::InvalidConfig( + "could not read own namespace name (empty file)".to_owned(), + )); + } + validate_k8s_namespace_name(&own_ns)?; + if !names.contains(&own_ns) { + names.push(own_ns); + } + } + Ok(names) + } +} + fn validate_k8s_namespace_name(ns: &str) -> Result<(), DiscoveryError> { let bytes = ns.as_bytes(); let is_valid = !bytes.is_empty() @@ -868,3 +1242,158 @@ fn validate_k8s_namespace_name(ns: &str) -> Result<(), DiscoveryError> { } Ok(()) } + +fn validate_pod_runner_config(config: &mut KubernetesSdConfig) -> Result<(), DiscoveryError> { + config.validate()?; + if config.role != Some(KubernetesRole::Pod) { + return Err(DiscoveryError::InvalidConfig(format!( + "KubernetesPodDiscovery requires role \"pod\", got {:?}", + config.role.map(KubernetesRole::as_str) + ))); + } + if !config.kubeconfig_file.is_empty() { + return Err(DiscoveryError::InvalidConfig( + "kubeconfig_file is not supported by KubernetesPodDiscovery; configure api_server and HTTP credentials instead".to_owned(), + )); + } + Ok(()) +} + +fn build_resource_url( + base_url: &str, + segments: &[&str], + selector: Option<&KubernetesSelectorConfig>, +) -> Result { + let mut url = reqwest::Url::parse(base_url).map_err(|e| { + DiscoveryError::InvalidConfig(format!("invalid Kubernetes base URL {base_url:?}: {e}")) + })?; + if !matches!(url.scheme(), "http" | "https") { + return Err(DiscoveryError::InvalidConfig(format!( + "invalid Kubernetes base URL scheme {:?}, expected \"http\" or \"https\"", + url.scheme() + ))); + } + { + let mut path_segs = url.path_segments_mut().map_err(|()| { + DiscoveryError::InvalidConfig(format!( + "Kubernetes base URL {base_url:?} cannot be a base" + )) + })?; + path_segs.pop_if_empty(); + path_segs.extend(segments); + } + if let Some(sel) = selector { + let mut pairs = url.query_pairs_mut(); + if !sel.label.is_empty() { + pairs.append_pair("labelSelector", &sel.label); + } + if !sel.field.is_empty() { + pairs.append_pair("fieldSelector", &sel.field); + } + } + Ok(url) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn resolve_namespaces_rejects_empty_own_namespace_file() { + let cfg = KubernetesSdConfig { + role: Some(KubernetesRole::Pod), + namespaces: KubernetesNamespaceDiscovery { + own_namespace: true, + names: Vec::new(), + }, + ..KubernetesSdConfig::default() + }; + let disc = KubernetesPodDiscovery::new(cfg) + .unwrap() + .with_own_namespace_file("/dev/null"); + let err = disc.resolve_namespaces().unwrap_err().to_string(); + assert_eq!(err, "could not read own namespace name (empty file)"); + } + + #[test] + fn in_cluster_auth_ca_and_namespace_validation() { + let kubeconfig_cfg = KubernetesSdConfig { + role: Some(KubernetesRole::Pod), + kubeconfig_file: "/etc/kubeconfig".to_owned(), + ..KubernetesSdConfig::default() + }; + assert!(KubernetesPodDiscovery::new(kubeconfig_cfg).is_err()); + + let node_cfg = KubernetesSdConfig { + role: Some(KubernetesRole::Node), + ..KubernetesSdConfig::default() + }; + assert!(KubernetesPodDiscovery::new(node_cfg).is_err()); + + let dir = std::env::temp_dir().join(format!("scrape-rs-k8s-test-{}", std::process::id())); + fs::create_dir_all(&dir).unwrap(); + let ns_file = dir.join("namespace"); + let tok_file = dir.join("token"); + let ca_file = dir.join("ca.crt"); + fs::write(&ns_file, " monitoring \n").unwrap(); + fs::write(&tok_file, " in-cluster-jwt \n").unwrap(); + fs::write( + &ca_file, + "-----BEGIN CERTIFICATE-----\n\ +MIIBSzCB8qADAgECAgEBMAoGCCqGSM49BAMCMBwxGjAYBgNVBAMMEVRlc3QgRGlz\n\ +Y292ZXJ5IENBMCAXDTIwMDEwMTAwMDAwMFoYDzIxMjAwMTAxMDAwMDAwWjAcMRow\n\ +GAYDVQQDDBFUZXN0IERpc2NvdmVyeSBDQTBZMBMGByqGSM49AgEGCCqGSM49AwEH\n\ +A0IABIh8inO1e0XFBqDBtcxv3BmM0MHgV2PZeF+BswiHPsQ63K6Xt/xGZS9ziRh8\n\ +bErqSNtv7W9shFuHxVmIuf+ROo+jIzAhMA8GA1UdEwEB/wQFMAMBAf8wDgYDVR0P\n\ +AQH/BAQDAgGGMAoGCCqGSM49BAMCA0gAMEUCIQCXh/u32CC/j81T+3oEJnRR0P8A\n\ +ZCXcq24++2iU8RQHuAIgTvD813EN4j4Wd++SFXpG+evSLLk+fyJse98OIktvI/E=\n\ +-----END CERTIFICATE-----\n", + ) + .unwrap(); + + let mut disc = KubernetesPodDiscovery::new(KubernetesSdConfig { + role: Some(KubernetesRole::Pod), + namespaces: KubernetesNamespaceDiscovery { + own_namespace: true, + names: vec![ + "default".to_owned(), + "monitoring".to_owned(), + "default".to_owned(), + ], + }, + ..KubernetesSdConfig::default() + }) + .unwrap() + .with_own_namespace_file(ns_file.to_string_lossy()); + + assert_eq!( + disc.resolve_namespaces().unwrap(), + vec!["default".to_owned(), "monitoring".to_owned()] + ); + + assert!( + disc.resolve_in_cluster_endpoint_and_auth("", "443") + .is_err() + ); + disc.in_cluster_token_file = dir.join("missing-token").to_string_lossy().into_owned(); + assert!( + disc.resolve_in_cluster_endpoint_and_auth("10.96.0.1", "443") + .is_err() + ); + disc.in_cluster_token_file = tok_file.to_string_lossy().into_owned(); + disc.in_cluster_ca_file = dir.join("missing-ca").to_string_lossy().into_owned(); + assert!( + disc.resolve_in_cluster_endpoint_and_auth("10.96.0.1", "443") + .is_err() + ); + disc.in_cluster_ca_file = ca_file.to_string_lossy().into_owned(); + let (base_url, _client, eff_http) = disc + .resolve_in_cluster_endpoint_and_auth("fd00::1", "6443") + .unwrap(); + assert_eq!(base_url, "https://[fd00::1]:6443"); + assert_eq!(eff_http.bearer_token_file, disc.in_cluster_token_file); + assert_eq!(eff_http.tls_config.ca_file, disc.in_cluster_ca_file); + + let _ = fs::remove_dir_all(&dir); + } +} diff --git a/tests/discovery_kubernetes.rs b/tests/discovery_kubernetes.rs index 0fdf169..1a99894 100644 --- a/tests/discovery_kubernetes.rs +++ b/tests/discovery_kubernetes.rs @@ -16,16 +16,19 @@ #![cfg(feature = "discovery-kubernetes")] use std::collections::BTreeMap; +use std::net::{IpAddr, Ipv4Addr, SocketAddr}; use expfmt::discovery::kubernetes::{ Job, KubernetesAttachMetadataConfig, KubernetesMetadataContext, KubernetesNamespaceDiscovery, - KubernetesRole, KubernetesSdConfig, KubernetesSelectorConfig, Namespace, Node, OwnerReference, - ReplicaSet, build_pod_target_group, + KubernetesPodDiscovery, KubernetesRole, KubernetesSdConfig, KubernetesSelectorConfig, + Namespace, Node, OwnerReference, ReplicaSet, build_pod_target_group, }; use expfmt::discovery::{ ADDRESS_LABEL, Authorization, BasicAuth, DiscoveryConfigs, HttpClientConfig, HttpHeader, HttpHeaders, LabelSet, OAuth2, Pod, TlsConfig, TlsVersion, }; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpListener; fn label_set(pairs: &[(&str, &str)]) -> LabelSet { pairs @@ -579,3 +582,417 @@ fn kubernetes_model_and_validation_edge_cases() { Some(&"2001:db8::9".to_owned()) ); } + +const TEST_CA_CERT_PEM: &str = "-----BEGIN CERTIFICATE-----\n\ +MIIBSzCB8qADAgECAgEBMAoGCCqGSM49BAMCMBwxGjAYBgNVBAMMEVRlc3QgRGlz\n\ +Y292ZXJ5IENBMCAXDTIwMDEwMTAwMDAwMFoYDzIxMjAwMTAxMDAwMDAwWjAcMRow\n\ +GAYDVQQDDBFUZXN0IERpc2NvdmVyeSBDQTBZMBMGByqGSM49AgEGCCqGSM49AwEH\n\ +A0IABIh8inO1e0XFBqDBtcxv3BmM0MHgV2PZeF+BswiHPsQ63K6Xt/xGZS9ziRh8\n\ +bErqSNtv7W9shFuHxVmIuf+ROo+jIzAhMA8GA1UdEwEB/wQFMAMBAf8wDgYDVR0P\n\ +AQH/BAQDAgGGMAoGCCqGSM49BAMCA0gAMEUCIQCXh/u32CC/j81T+3oEJnRR0P8A\n\ +ZCXcq24++2iU8RQHuAIgTvD813EN4j4Wd++SFXpG+evSLLk+fyJse98OIktvI/E=\n\ +-----END CERTIFICATE-----\n"; + +const TEST_SERVER_CERT_PEM: &str = "-----BEGIN CERTIFICATE-----\n\ +MIIBVzCB/aADAgECAgECMAoGCCqGSM49BAMCMBwxGjAYBgNVBAMMEVRlc3QgRGlz\n\ +Y292ZXJ5IENBMCAXDTIwMDEwMTAwMDAwMFoYDzIxMjAwMTAxMDAwMDAwWjAUMRIw\n\ +EAYDVQQDDAkxMjcuMC4wLjEwWTATBgcqhkjOPQIBBggqhkjOPQMBBwNCAASgpvT7\n\ +pxvHJ9b0hfVl1OSzVw4wXXNbMgOcQuzTJyWIdPLOgN4uV3uYNUCGBoOmzZiSHQNV\n\ +kxosSi48IIZKN66CozYwNDAMBgNVHRMBAf8EAjAAMA8GA1UdEQQIMAaHBH8AAAEw\n\ +EwYDVR0lBAwwCgYIKwYBBQUHAwEwCgYIKoZIzj0EAwIDSQAwRgIhAL8MTVC7DzUW\n\ +S6c9vBHZmQpTHRttDCHL5pJuJjYfD6UIAiEA4Svgdadp5oYsNbiccD3xiOxd2f0C\n\ +gcCEe0UeZFy+qgs=\n\ +-----END CERTIFICATE-----\n"; + +const TEST_SERVER_KEY_PEM: &str = "-----BEGIN PRIVATE KEY-----\n\ +MIGHAgEAMBMGByqGSM49AgEGCCqGSM49AwEHBG0wawIBAQQglO/xnuT6hNqAa1Au\n\ +8mCUOEYEuik/fdi8Fp1fJTvPI/mhRANCAASgpvT7pxvHJ9b0hfVl1OSzVw4wXXNb\n\ +MgOcQuzTJyWIdPLOgN4uV3uYNUCGBoOmzZiSHQNVkxosSi48IIZKN66C\n\ +-----END PRIVATE KEY-----\n"; + +const TEST_CLIENT_CERT_PEM: &str = "-----BEGIN CERTIFICATE-----\n\ +MIIBRzCB7qADAgECAgEDMAoGCCqGSM49BAMCMBwxGjAYBgNVBAMMEVRlc3QgRGlz\n\ +Y292ZXJ5IENBMCAXDTIwMDEwMTAwMDAwMFoYDzIxMjAwMTAxMDAwMDAwWjAWMRQw\n\ +EgYDVQQDDAtwcm9tLWNsaWVudDBZMBMGByqGSM49AgEGCCqGSM49AwEHA0IABPaV\n\ +o6u67B6wO8kr/rbYTvsvDXIxURJaLinfvVeqsPW3SDYaUTM9f8tSa7xbvmBwLbmr\n\ +y1s7cN7v0Pfo5lYFU5CjJTAjMAwGA1UdEwEB/wQCMAAwEwYDVR0lBAwwCgYIKwYB\n\ +BQUHAwIwCgYIKoZIzj0EAwIDSAAwRQIgNdIcz83GyYDyCadBNBtP+jxPeCBBpQY4\n\ +c/Mzu1lyEr8CIQDZMeAQYUPozmcVavNzujS3wddD9dNSIpZBGsOd9p4Kdg==\n\ +-----END CERTIFICATE-----\n"; + +const TEST_CLIENT_KEY_PEM: &str = "-----BEGIN PRIVATE KEY-----\n\ +MIGHAgEAMBMGByqGSM49AgEGCCqGSM49AwEHBG0wawIBAQQgOcr6T6T9duo1QH4u\n\ +nR6BDYL+R0s2UDkIdWN9ry1Z60ihRANCAAT2laOruuwesDvJK/622E77Lw1yMVES\n\ +Wi4p371XqrD1t0g2GlEzPX/LUmu8W75gcC25q8tbO3De79D36OZWBVOQ\n\ +-----END PRIVATE KEY-----\n"; + +#[tokio::test] +async fn kubernetes_pod_discovery_reqwest_end_to_end() { + let listener = TcpListener::bind(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0)) + .await + .unwrap(); + let addr = listener.local_addr().unwrap(); + + tokio::spawn(async move { + loop { + let Ok((mut stream, _)) = listener.accept().await else { + break; + }; + tokio::spawn(async move { + let mut req_buf = vec![0u8; 4096]; + let n = stream.read(&mut req_buf).await.unwrap_or(0); + if n == 0 { + return; + } + let req_str = String::from_utf8_lossy(&req_buf[..n]); + let first_line = req_str.lines().next().unwrap_or(""); + assert!( + req_str + .lines() + .any(|l| l.eq_ignore_ascii_case("authorization: Bearer test-k8s-token")), + "missing bearer token header in:\n{req_str}" + ); + + let body = if first_line.starts_with( + "GET /api/v1/namespaces/monitoring/pods?labelSelector=app%3Dmetrics&fieldSelector=status.phase%3DRunning ", + ) { + r#"{ + "items": [{ + "metadata": { + "name": "metrics-pod-0", + "namespace": "monitoring", + "uid": "uid-123", + "labels": {"app": "metrics"}, + "ownerReferences": [{ + "kind": "ReplicaSet", + "name": "metrics-rs", + "controller": true + }] + }, + "spec": { + "nodeName": "node-1", + "containers": [{ + "name": "exporter", + "image": "prom/exporter:v1", + "ports": [{ + "name": "http-metrics", + "containerPort": 9100, + "protocol": "TCP" + }] + }] + }, + "status": { + "phase": "Running", + "podIP": "10.244.0.9", + "hostIP": "192.168.10.5", + "conditions": [{"type": "Ready", "status": "True"}], + "containerStatuses": [{ + "name": "exporter", + "containerID": "containerd://abc123" + }] + } + }] + }"# + } else if first_line.starts_with("GET /api/v1/nodes?labelSelector=role%3Dworker ") { + r#"{ + "items": [{ + "metadata": { + "name": "node-1", + "labels": {"role": "worker"} + } + }] + }"# + } else if first_line + .starts_with("GET /apis/apps/v1/namespaces/monitoring/replicasets ") + { + r#"{ + "items": [{ + "metadata": { + "name": "metrics-rs", + "namespace": "monitoring", + "ownerReferences": [{ + "kind": "Deployment", + "name": "metrics-deploy", + "controller": true + }] + } + }] + }"# + } else { + panic!("unexpected request line: {first_line}"); + }; + + let resp = format!( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", + body.len(), + body + ); + let _ = stream.write_all(resp.as_bytes()).await; + }); + } + }); + + let cfg: KubernetesSdConfig = serde_json::from_str(&format!( + r#"{{ + "api_server": "http://{addr}", + "role": "pod", + "bearer_token": "test-k8s-token", + "namespaces": {{ + "names": ["monitoring"] + }}, + "selectors": [ + {{ + "role": "pod", + "label": "app=metrics", + "field": "status.phase=Running" + }}, + {{ + "role": "node", + "label": "role=worker" + }} + ], + "attach_metadata": {{ + "node": true, + "deployment": true + }} + }}"# + )) + .unwrap(); + + let discovery = KubernetesPodDiscovery::new(cfg).unwrap(); + let groups = discovery.discover().await.unwrap(); + assert_eq!(groups.len(), 1); + assert_eq!(groups[0].source, "pod/monitoring/metrics-pod-0"); + assert_eq!( + groups[0] + .labels + .get("__meta_kubernetes_pod_deployment_name"), + Some(&"metrics-deploy".to_owned()) + ); + assert_eq!( + groups[0].labels.get("__meta_kubernetes_node_label_role"), + Some(&"worker".to_owned()) + ); + assert_eq!( + groups[0].targets, + vec![label_set(&[ + ("__address__", "10.244.0.9:9100"), + ("__meta_kubernetes_pod_container_id", "containerd://abc123"), + ("__meta_kubernetes_pod_container_image", "prom/exporter:v1"), + ("__meta_kubernetes_pod_container_init", "false"), + ("__meta_kubernetes_pod_container_name", "exporter"), + ("__meta_kubernetes_pod_container_port_name", "http-metrics"), + ("__meta_kubernetes_pod_container_port_number", "9100"), + ("__meta_kubernetes_pod_container_port_protocol", "TCP"), + ])] + ); +} + +#[test] +fn https_mtls_all_namespaces_404_tolerance_and_localset_execution() { + use rustls::pki_types::pem::PemObject; + use rustls::pki_types::{CertificateDer, PrivateKeyDer}; + + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + let local = tokio::task::LocalSet::new(); + + local.block_on(&rt, async { + let dir = + std::env::temp_dir().join(format!("scrape-rs-https-k8s-test-{}", std::process::id())); + std::fs::create_dir_all(&dir).unwrap(); + let user_file = dir.join("user.txt"); + let pass_file = dir.join("pass.txt"); + let hdr_file = dir.join("hdr.txt"); + std::fs::write(&user_file, "prom-user\n").unwrap(); + std::fs::write(&pass_file, "prom-pass\n").unwrap(); + std::fs::write(&hdr_file, "from-file\n").unwrap(); + + let mut client_roots = rustls::RootCertStore::empty(); + client_roots + .add(CertificateDer::from_pem_slice(TEST_CA_CERT_PEM.as_bytes()).unwrap()) + .unwrap(); + let verifier = rustls::server::WebPkiClientVerifier::builder(std::sync::Arc::new( + client_roots, + )) + .build() + .unwrap(); + let server_cert = CertificateDer::from_pem_slice(TEST_SERVER_CERT_PEM.as_bytes()).unwrap(); + let server_key = PrivateKeyDer::from_pem_slice(TEST_SERVER_KEY_PEM.as_bytes()).unwrap(); + let tls_cfg = rustls::ServerConfig::builder() + .with_client_cert_verifier(verifier) + .with_single_cert(vec![server_cert], server_key) + .unwrap(); + let acceptor = tokio_rustls::TlsAcceptor::from(std::sync::Arc::new(tls_cfg)); + + let listener = TcpListener::bind(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0)) + .await + .unwrap(); + let addr = listener.local_addr().unwrap(); + + tokio::task::spawn_local(async move { + while let Ok((tcp, _)) = listener.accept().await { + let acceptor = acceptor.clone(); + tokio::task::spawn_local(async move { + let Ok(mut tls_stream) = acceptor.accept(tcp).await else { + return; + }; + let mut req_buf = vec![0u8; 4096]; + let n = tls_stream.read(&mut req_buf).await.unwrap_or(0); + if n == 0 { + return; + } + let req_str = String::from_utf8_lossy(&req_buf[..n]); + let first_line = req_str.lines().next().unwrap_or(""); + assert!( + req_str.lines().any(|l| l.eq_ignore_ascii_case( + "authorization: Basic cHJvbS11c2VyOnByb20tcGFzcw==" + )), + "missing basic auth in:\n{req_str}" + ); + assert!( + req_str + .lines() + .any(|l| l.eq_ignore_ascii_case("x-trace-file: from-file")), + "missing file header in:\n{req_str}" + ); + + let (status, body) = if first_line.starts_with("GET /api/v1/pods ") { + ( + "200 OK", + r#"{"items":[{"metadata":{"name":"job-pod","namespace":"batch-ns","ownerReferences":[{"kind":"Job","name":"job-1","controller":true}]},"spec":{"containers":[{"name":"worker","image":"worker:1","ports":[{"containerPort":9091,"protocol":"TCP"}]}]},"status":{"phase":"Running","podIP":"10.9.0.2"}}]}"#, + ) + } else if first_line.starts_with("GET /api/v1/namespaces ") { + ( + "200 OK", + r#"{"items":[{"metadata":{"name":"batch-ns","labels":{"tier":"batch"}}}]}"#, + ) + } else if first_line.starts_with("GET /apis/batch/v1/jobs ") { + ( + "200 OK", + r#"{"items":[{"metadata":{"name":"job-1","namespace":"batch-ns","ownerReferences":[{"kind":"CronJob","name":"nightly","controller":true}]}}]}"#, + ) + } else if first_line.starts_with("GET /api/v1/namespaces/missing-ns ") { + ("404 Not Found", r#"{"status":"Failure","code":404}"#) + } else if first_line.starts_with("GET /api/v1/namespaces/missing-ns/pods ") { + ("200 OK", r#"{"items":null}"#) + } else if first_line.starts_with("GET /api/v1/namespaces/batch-ns ") { + ( + "200 OK", + r#"{"metadata":{"name":"batch-ns","labels":{"tier":"batch"}}}"#, + ) + } else if first_line.starts_with("GET /api/v1/namespaces/batch-ns/pods ") { + ( + "200 OK", + r#"{"items":[{"metadata":{"name":"job-pod","namespace":"batch-ns"},"spec":{"containers":[{"name":"worker","image":"worker:1","ports":null}]},"status":{"phase":"Running","podIP":"10.9.0.2"}}]}"#, + ) + } else { + panic!("unexpected HTTPS request: {first_line}"); + }; + + let resp = format!( + "HTTP/1.1 {status}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", + body.len(), + body + ); + let _ = tls_stream.write_all(resp.as_bytes()).await; + let _ = tls_stream.shutdown().await; + }); + } + }); + + let base_http = HttpClientConfig { + basic_auth: Some(BasicAuth { + username_file: user_file.to_string_lossy().into_owned(), + password_file: pass_file.to_string_lossy().into_owned(), + ..BasicAuth::default() + }), + enable_http2: false, + follow_redirects: false, + tls_config: TlsConfig { + ca: TEST_CA_CERT_PEM.to_owned(), + cert: TEST_CLIENT_CERT_PEM.to_owned(), + key: TEST_CLIENT_KEY_PEM.to_owned(), + min_version: Some(TlsVersion::Tls12), + max_version: Some(TlsVersion::Tls13), + ..TlsConfig::default() + }, + http_headers: Some(HttpHeaders { + headers: BTreeMap::from([( + "X-Trace-File".to_owned(), + HttpHeader { + files: vec![hdr_file.to_string_lossy().into_owned()], + ..HttpHeader::default() + }, + )]), + }), + ..HttpClientConfig::default() + }; + + // 1. All-namespaces HTTPS mTLS discovery with attach_metadata.{namespace, job, cronjob}. + let all_ns_disc = KubernetesPodDiscovery::new(KubernetesSdConfig { + api_server: Some(format!("https://{addr}")), + role: Some(KubernetesRole::Pod), + http_client_config: base_http.clone(), + attach_metadata: KubernetesAttachMetadataConfig { + namespace: true, + job: true, + cronjob: true, + ..KubernetesAttachMetadataConfig::default() + }, + ..KubernetesSdConfig::default() + }) + .unwrap(); + + let groups = tokio::task::spawn_local(async move { all_ns_disc.discover().await }) + .await + .unwrap() + .unwrap(); + assert_eq!(groups.len(), 1); + assert_eq!( + groups[0] + .labels + .get("__meta_kubernetes_namespace_label_tier"), + Some(&"batch".to_owned()) + ); + assert_eq!( + groups[0].labels.get("__meta_kubernetes_pod_cronjob_name"), + Some(&"nightly".to_owned()) + ); + + // 2. Multi-namespace HTTPS mTLS discovery where `missing-ns` returns `items: null` for pods + // and `404 Not Found` for `GET /api/v1/namespaces/missing-ns` without aborting `batch-ns`. + let multi_ns_disc = KubernetesPodDiscovery::new(KubernetesSdConfig { + api_server: Some(format!("https://{addr}")), + role: Some(KubernetesRole::Pod), + http_client_config: base_http, + namespaces: KubernetesNamespaceDiscovery { + own_namespace: false, + names: vec!["missing-ns".to_owned(), "batch-ns".to_owned()], + }, + attach_metadata: KubernetesAttachMetadataConfig { + namespace: true, + ..KubernetesAttachMetadataConfig::default() + }, + ..KubernetesSdConfig::default() + }) + .unwrap(); + + let multi_groups = tokio::task::spawn_local(async move { multi_ns_disc.discover().await }) + .await + .unwrap() + .unwrap(); + assert_eq!(multi_groups.len(), 1); + assert_eq!(multi_groups[0].source, "pod/batch-ns/job-pod"); + assert_eq!( + multi_groups[0] + .labels + .get("__meta_kubernetes_namespace_label_tier"), + Some(&"batch".to_owned()) + ); + + let _ = std::fs::remove_dir_all(&dir); + }); +}