feat: kubernetes discovery

Author: evansims

Cargo.toml

@@ -116,6 +116,10 @@ url = "2.5"
 # Validation
 validator = { version = "0.20", features = ["derive"] }
 
+# Kubernetes client
+kube = { version = "0.97", features = ["runtime", "client", "derive"] }
+k8s-openapi = { version = "0.23", features = ["v1_31"] }
+
 [profile.release]
 codegen-units = 1
 lto = "thin"

crates/infera-management-core/Cargo.toml

@@ -53,6 +53,10 @@ validator = { workspace = true }
 reqwest = { workspace = true }
 url = { workspace = true }
 
+# Kubernetes client (for service discovery)
+kube = { workspace = true }
+k8s-openapi = { workspace = true }
+
 # Metrics
 metrics = { workspace = true }
 

crates/infera-management-core/src/config.rs

@@ -255,6 +255,52 @@ pub struct CacheInvalidationConfig {
     /// Number of retry attempts on webhook failure
     #[serde(default = "default_webhook_retry_attempts")]
     pub retry_attempts: u8,
+
+    /// Service discovery configuration
+    #[serde(default)]
+    pub discovery: DiscoveryConfig,
+}
+
+/// Service discovery configuration
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct DiscoveryConfig {
+    /// Discovery mode (none or kubernetes)
+    #[serde(default)]
+    pub mode: DiscoveryMode,
+
+    /// Cache TTL for discovered endpoints (in seconds)
+    #[serde(default = "default_discovery_cache_ttl")]
+    pub cache_ttl_seconds: u64,
+
+    /// Whether to enable health checking of endpoints
+    #[serde(default = "default_discovery_health_check")]
+    pub enable_health_check: bool,
+
+    /// Health check interval (in seconds)
+    #[serde(default = "default_discovery_health_check_interval")]
+    pub health_check_interval_seconds: u64,
+}
+
+impl Default for DiscoveryConfig {
+    fn default() -> Self {
+        Self {
+            mode: DiscoveryMode::None,
+            cache_ttl_seconds: default_discovery_cache_ttl(),
+            enable_health_check: default_discovery_health_check(),
+            health_check_interval_seconds: default_discovery_health_check_interval(),
+        }
+    }
+}
+
+/// Service discovery mode
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
+#[serde(rename_all = "lowercase")]
+pub enum DiscoveryMode {
+    /// No service discovery - use service URL directly
+    #[default]
+    None,
+    /// Kubernetes service discovery - resolve to pod IPs
+    Kubernetes,
 }
 
 // Default value functions
@@ -379,6 +425,7 @@ fn default_cache_invalidation() -> CacheInvalidationConfig {
         http_endpoints: vec![], // No webhooks by default
         timeout_ms: 5000,       // 5 seconds
         retry_attempts: 0,      // Fire-and-forget (no retries)
+        discovery: DiscoveryConfig::default(),
     }
 }
 
@@ -402,6 +449,18 @@ fn default_webhook_retry_attempts() -> u8 {
     0 // Fire-and-forget
 }
 
+fn default_discovery_cache_ttl() -> u64 {
+    300 // 5 minutes
+}
+
+fn default_discovery_health_check() -> bool {
+    false
+}
+
+fn default_discovery_health_check_interval() -> u64 {
+    30 // 30 seconds
+}
+
 impl ManagementConfig {
     /// Load configuration from a file with environment variable overrides
     pub fn load<P: AsRef<Path>>(path: P) -> Result<Self> {

crates/infera-management-core/src/webhook_client.rs

@@ -369,6 +369,9 @@ fn is_kubernetes_service(endpoint: &str) -> bool {
 ///
 /// List of pod endpoints (e.g., ["http://10.0.1.2:8080", "http://10.0.1.3:8080"])
 async fn discover_k8s_service_endpoints(service_endpoint: &str) -> Result<Vec<String>, String> {
+    use k8s_openapi::api::core::v1::Endpoints as K8sEndpoints;
+    use kube::{Api, Client};
+
     // Parse the service URL
     let url =
         url::Url::parse(service_endpoint).map_err(|e| format!("Invalid service URL: {}", e))?;
@@ -379,7 +382,7 @@ async fn discover_k8s_service_endpoints(service_endpoint: &str) -> Result<Vec<St
     let service_port = url
         .port_or_known_default()
         .ok_or_else(|| "No port in service URL".to_string())?;
-    let _scheme = url.scheme();
+    let scheme = url.scheme();
 
     // Extract service name and namespace from hostname
     // Formats supported:
@@ -402,23 +405,58 @@ async fn discover_k8s_service_endpoints(service_endpoint: &str) -> Result<Vec<St
         "Discovering Kubernetes service endpoints"
     );
 
-    // In a real implementation, you would use the kube-rs library to query the Kubernetes API
-    // For now, we'll return a placeholder that falls back to the service URL
-    // TODO: Implement actual Kubernetes API integration using kube-rs
+    // Create Kubernetes client
+    let client = Client::try_default()
+        .await
+        .map_err(|e| format!("Failed to create Kubernetes client: {}", e))?;
+
+    // Get the Endpoints resource for this service
+    let endpoints_api: Api<K8sEndpoints> = Api::namespaced(client, namespace);
+
+    let endpoints = endpoints_api.get(service_name).await.map_err(|e| {
+        if e.to_string().contains("404") {
+            format!("Service {}.{} not found", service_name, namespace)
+        } else {
+            format!(
+                "Failed to get endpoints for {}.{}: {}",
+                service_name, namespace, e
+            )
+        }
+    })?;
+
+    // Extract pod IPs from the Endpoints resource
+    let mut pod_endpoints = Vec::new();
+
+    if let Some(subsets) = endpoints.subsets {
+        for subset in subsets {
+            // Only use ready addresses (healthy pods)
+            if let Some(addresses) = subset.addresses {
+                for address in addresses {
+                    let pod_ip = &address.ip;
+                    let endpoint_url = format!("{}://{}:{}", scheme, pod_ip, service_port);
+                    pod_endpoints.push(endpoint_url);
+                }
+            }
+        }
+    }
 
-    // Placeholder: In production, this would use kube-rs to:
-    // 1. Get the Endpoints resource for the service
-    // 2. Extract all pod IPs
-    // 3. Return list of "http://pod-ip:port" URLs
+    if pod_endpoints.is_empty() {
+        warn!(
+            service_name = %service_name,
+            namespace = %namespace,
+            "No ready endpoints found for service - falling back to service URL"
+        );
+        return Ok(vec![service_endpoint.to_string()]);
+    }
 
-    // For now, just return the original service endpoint as fallback
-    warn!(
+    info!(
         service_name = %service_name,
         namespace = %namespace,
-        "Kubernetes service discovery not yet fully implemented - using service URL as fallback"
+        endpoint_count = pod_endpoints.len(),
+        "Discovered Kubernetes service endpoints"
     );
 
-    Ok(vec![service_endpoint.to_string()])
+    Ok(pod_endpoints)
 }
 
 #[cfg(test)]

k8s/rbac.yaml

@@ -0,0 +1,43 @@
+---
+# ServiceAccount for InferaDB Management API
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: inferadb-management
+  namespace: inferadb
+  labels:
+    app.kubernetes.io/name: inferadb-management
+    app.kubernetes.io/component: control-plane
+
+---
+# ClusterRole for service discovery
+# Allows reading Endpoints resources across all namespaces
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRole
+metadata:
+  name: inferadb-management-discovery
+  labels:
+    app.kubernetes.io/name: inferadb-management
+    app.kubernetes.io/component: control-plane
+rules:
+  - apiGroups: [""]
+    resources: ["endpoints"]
+    verbs: ["get", "list", "watch"]
+
+---
+# ClusterRoleBinding for service discovery
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRoleBinding
+metadata:
+  name: inferadb-management-discovery
+  labels:
+    app.kubernetes.io/name: inferadb-management
+    app.kubernetes.io/component: control-plane
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: ClusterRole
+  name: inferadb-management-discovery
+subjects:
+  - kind: ServiceAccount
+    name: inferadb-management
+    namespace: inferadb