diff --git a/deploy/cloud/operator/internal/consts/consts.go b/deploy/cloud/operator/internal/consts/consts.go index da16238ab..df9560fd6 100644 --- a/deploy/cloud/operator/internal/consts/consts.go +++ b/deploy/cloud/operator/internal/consts/consts.go @@ -43,6 +43,7 @@ const ( KubeLabelDynamoBaseModelHash = "nvidia.com/dynamo-base-model-hash" KubeAnnotationDynamoBaseModel = "nvidia.com/dynamo-base-model" KubeLabelDynamoDiscoveryBackend = "nvidia.com/dynamo-discovery-backend" + KubeLabelDynamoDiscoveryEnabled = "nvidia.com/dynamo-discovery-enabled" KubeLabelValueFalse = "false" KubeLabelValueTrue = "true" diff --git a/deploy/cloud/operator/internal/controller/dynamocomponentdeployment_controller.go b/deploy/cloud/operator/internal/controller/dynamocomponentdeployment_controller.go index 0b9d50a4a..6acab7b8d 100644 --- a/deploy/cloud/operator/internal/controller/dynamocomponentdeployment_controller.go +++ b/deploy/cloud/operator/internal/controller/dynamocomponentdeployment_controller.go @@ -1298,6 +1298,10 @@ func (r *DynamoComponentDeploymentReconciler) generateService(opt generateResour if isK8sDiscovery { labels[commonconsts.KubeLabelDynamoDiscoveryBackend] = "kubernetes" } + // Discovery is enabled for non frontend components + if isK8sDiscovery && !opt.dynamoComponentDeployment.IsFrontendComponent() { + labels[commonconsts.KubeLabelDynamoDiscoveryEnabled] = commonconsts.KubeLabelValueTrue + } var servicePort corev1.ServicePort if opt.dynamoComponentDeployment.IsFrontendComponent() { diff --git a/deploy/cloud/operator/internal/controller/dynamographdeployment_controller.go b/deploy/cloud/operator/internal/controller/dynamographdeployment_controller.go index 7a8af3d9b..a944fa9c1 100644 --- a/deploy/cloud/operator/internal/controller/dynamographdeployment_controller.go +++ b/deploy/cloud/operator/internal/controller/dynamographdeployment_controller.go @@ -368,8 +368,9 @@ func (r *DynamoGraphDeploymentReconciler) reconcileGroveResources(ctx context.Co // if k8s discovery is enabled, create a service for each component // else, only create for the frontend component - if r.Config.IsK8sDiscoveryEnabled(dynamoDeployment.Annotations) || component.ComponentType == consts.ComponentTypeFrontend { - mainComponentService, err := dynamo.GenerateComponentService(ctx, dynamoDeployment, component, componentName) + isK8sDiscoveryEnabled := r.Config.IsK8sDiscoveryEnabled(dynamoDeployment.Annotations) + if isK8sDiscoveryEnabled || component.ComponentType == consts.ComponentTypeFrontend { + mainComponentService, err := dynamo.GenerateComponentService(ctx, dynamoDeployment, component, componentName, isK8sDiscoveryEnabled) if err != nil { logger.Error(err, "failed to generate the main component service") return "", "", "", fmt.Errorf("failed to generate the main component service: %w", err) diff --git a/deploy/cloud/operator/internal/dynamo/graph.go b/deploy/cloud/operator/internal/dynamo/graph.go index c5436df62..d3f6bdf57 100644 --- a/deploy/cloud/operator/internal/dynamo/graph.go +++ b/deploy/cloud/operator/internal/dynamo/graph.go @@ -396,7 +396,7 @@ func getCliqueStartupDependencies( return nil } -func GenerateComponentService(ctx context.Context, dynamoDeployment *v1alpha1.DynamoGraphDeployment, component *v1alpha1.DynamoComponentDeploymentSharedSpec, componentName string) (*corev1.Service, error) { +func GenerateComponentService(ctx context.Context, dynamoDeployment *v1alpha1.DynamoGraphDeployment, component *v1alpha1.DynamoComponentDeploymentSharedSpec, componentName string, isK8sDiscoveryEnabled bool) (*corev1.Service, error) { if component.DynamoNamespace == nil { return nil, fmt.Errorf("expected DynamoComponentDeployment %s to have a dynamoNamespace", componentName) } @@ -431,6 +431,15 @@ func GenerateComponentService(ctx context.Context, dynamoDeployment *v1alpha1.Dy Ports: []corev1.ServicePort{servicePort}, }, } + if isK8sDiscoveryEnabled { + service.Labels = map[string]string{ + commonconsts.KubeLabelDynamoDiscoveryBackend: "kubernetes", + } + // Discovery is enabled for non frontend components + if component.ComponentType != commonconsts.ComponentTypeFrontend { + service.Labels[commonconsts.KubeLabelDynamoDiscoveryEnabled] = commonconsts.KubeLabelValueTrue + } + } return service, nil } diff --git a/deploy/discovery/README.md b/deploy/discovery/README.md index b76c57712..e981da977 100644 --- a/deploy/discovery/README.md +++ b/deploy/discovery/README.md @@ -16,7 +16,7 @@ When deploying with the Dynamo operator, simply add the annotation to your DGD m ```yaml metadata: annotations: - nvidia.com/dynamo-discover-backend: kubernetes + nvidia.com/dynamo-discovery-backend: kubernetes ``` The operator will automatically configure the required EndpointSlices, labels, and pod environment variables. See [`dgd.yaml`](./dgd.yaml) for a complete example. diff --git a/deploy/discovery/dgd.yaml b/deploy/discovery/dgd.yaml index 1440fe99d..b55db6b3c 100644 --- a/deploy/discovery/dgd.yaml +++ b/deploy/discovery/dgd.yaml @@ -17,9 +17,6 @@ spec: dynamoNamespace: dynamo componentType: frontend replicas: 1 - envs: - - name: DYN_SYSTEM_PORT - value: "9090" extraPodSpec: mainContainer: image: ${IMAGE} @@ -30,29 +27,6 @@ spec: resources: limits: gpu: "1" - envs: - - name: DYN_SYSTEM_PORT - value: "9090" - - name: DYN_SYSTEM_STARTING_HEALTH_STATUS - value: "notready" - - name: DYN_SYSTEM_USE_ENDPOINT_HEALTH_STATUS - value: "[\"generate\", \"clear_kv_blocks\"]" - readinessProbe: - httpGet: - path: /health - port: 9090 - initialDelaySeconds: 60 - periodSeconds: 10 - timeoutSeconds: 5 - failureThreshold: 60 - livenessProbe: - httpGet: - path: /live - port: 9090 - initialDelaySeconds: 60 - periodSeconds: 10 - timeoutSeconds: 5 - failureThreshold: 60 extraPodSpec: terminationGracePeriodSeconds: 120 mainContainer: @@ -72,29 +46,6 @@ spec: resources: limits: gpu: "1" - envs: - - name: DYN_SYSTEM_PORT - value: "9090" - - name: DYN_SYSTEM_STARTING_HEALTH_STATUS - value: "notready" - - name: DYN_SYSTEM_USE_ENDPOINT_HEALTH_STATUS - value: "[\"generate\", \"clear_kv_blocks\"]" - readinessProbe: - httpGet: - path: /health - port: 9090 - initialDelaySeconds: 60 - periodSeconds: 10 - timeoutSeconds: 5 - failureThreshold: 60 - livenessProbe: - httpGet: - path: /live - port: 9090 - initialDelaySeconds: 60 - periodSeconds: 10 - timeoutSeconds: 5 - failureThreshold: 60 extraPodSpec: terminationGracePeriodSeconds: 120 mainContainer: diff --git a/lib/runtime/src/discovery/kube/daemon.rs b/lib/runtime/src/discovery/kube/daemon.rs index 4eee1287a..d65ad4024 100644 --- a/lib/runtime/src/discovery/kube/daemon.rs +++ b/lib/runtime/src/discovery/kube/daemon.rs @@ -68,11 +68,12 @@ impl DiscoveryDaemon { let (reader, writer) = reflector::store(); // Apply label selector to only watch discovery-enabled EndpointSlices - let watch_config = - Config::default().labels("nvidia.com/dynamo-discovery-backend=kubernetes"); + let watch_config = Config::default() + .labels("nvidia.com/dynamo-discovery-backend=kubernetes") + .labels("nvidia.com/dynamo-discovery-enabled=true"); tracing::info!( - "Daemon watching EndpointSlices with label: nvidia.com/dynamo-discovery-backend=kubernetes" + "Daemon watching EndpointSlices with labels: nvidia.com/dynamo-discovery-backend=kubernetes, nvidia.com/dynamo-discovery-enabled=true" ); // Spawn reflector task (runs independently) @@ -181,8 +182,8 @@ impl DiscoveryDaemon { ) -> Result { let start = std::time::Instant::now(); - // Extract ALL ready endpoints (instance_id, pod_name, pod_ip) directly from reflector - let all_endpoints: Vec<(u64, String, String)> = reader + // Extract ALL ready endpoints (instance_id, pod_name, pod_ip, system_port) directly from reflector + let all_endpoints: Vec<(u64, String, String, u16)> = reader .state() .iter() .flat_map(|arc_slice| extract_endpoint_info(arc_slice.as_ref())) @@ -194,25 +195,26 @@ impl DiscoveryDaemon { ); // Concurrent fetch: Fetch metadata for all endpoints in parallel - let fetch_futures = all_endpoints - .into_iter() - .map(|(instance_id, pod_name, pod_ip)| { - let daemon = self.clone(); - async move { - match daemon.fetch_metadata(&pod_name, &pod_ip).await { - Ok(metadata) => Some((instance_id, metadata)), - Err(e) => { - tracing::warn!( - "Failed to fetch metadata for pod {} (instance_id={:x}): {}", - pod_name, - instance_id, - e - ); - None + let fetch_futures = + all_endpoints + .into_iter() + .map(|(instance_id, pod_name, pod_ip, system_port)| { + let daemon = self.clone(); + async move { + match daemon.fetch_metadata(&pod_name, &pod_ip, system_port).await { + Ok(metadata) => Some((instance_id, metadata)), + Err(e) => { + tracing::warn!( + "Failed to fetch metadata for pod {} (instance_id={:x}): {}", + pod_name, + instance_id, + e + ); + None + } } } - } - }); + }); // Execute fetches concurrently with bounded parallelism let results: Vec<_> = futures::stream::iter(fetch_futures) @@ -241,7 +243,12 @@ impl DiscoveryDaemon { } /// Fetch metadata for a single pod (with caching) - async fn fetch_metadata(&self, pod_name: &str, pod_ip: &str) -> Result> { + async fn fetch_metadata( + &self, + pod_name: &str, + pod_ip: &str, + system_port: u16, + ) -> Result> { let instance_id = hash_pod_name(pod_name); // Check cache @@ -258,7 +265,7 @@ impl DiscoveryDaemon { } // Cache miss: fetch from HTTP - let url = format!("http://{}:{}/metadata", pod_ip, self.pod_info.system_port); + let url = format!("http://{}:{}/metadata", pod_ip, system_port); tracing::debug!("Fetching metadata from {url}"); diff --git a/lib/runtime/src/discovery/kube/utils.rs b/lib/runtime/src/discovery/kube/utils.rs index 6e1c5bee9..213da614e 100644 --- a/lib/runtime/src/discovery/kube/utils.rs +++ b/lib/runtime/src/discovery/kube/utils.rs @@ -13,11 +13,34 @@ pub fn hash_pod_name(pod_name: &str) -> u64 { hasher.finish() } -/// Extract endpoint information from an EndpointSlice -/// Returns (instance_id, pod_name, pod_ip) tuples for ready endpoints -pub(super) fn extract_endpoint_info(slice: &EndpointSlice) -> Vec<(u64, String, String)> { - let mut result = Vec::new(); +/// Extract the system port from an EndpointSlice's ports +/// Looks for a port with name "system", returns None if not found +fn extract_system_port(slice: &EndpointSlice) -> Option { + slice.ports.as_ref().and_then(|ports| { + ports + .iter() + .find(|p| p.name.as_deref() == Some("system")) + .and_then(|p| p.port.map(|port| port as u16)) + }) +} +/// Extract endpoint information from an EndpointSlice +/// Returns (instance_id, pod_name, pod_ip, system_port) tuples for ready endpoints +pub(super) fn extract_endpoint_info(slice: &EndpointSlice) -> Vec<(u64, String, String, u16)> { + // Extract system port from EndpointSlice ports + let system_port = match extract_system_port(slice) { + Some(port) => port, + None => { + let slice_name = slice.metadata.name.as_deref().unwrap_or("unknown"); + tracing::warn!( + "EndpointSlice '{}' did not have a system port defined", + slice_name + ); + return Vec::new(); + } + }; + + let mut result = Vec::new(); let endpoints = &slice.endpoints; for endpoint in endpoints { @@ -46,7 +69,7 @@ pub(super) fn extract_endpoint_info(slice: &EndpointSlice) -> Vec<(u64, String, // Get first IP only (avoid duplicate instance IDs) if let Some(ip) = endpoint.addresses.first() { - result.push((instance_id, pod_name.to_string(), ip.clone())); + result.push((instance_id, pod_name.to_string(), ip.clone(), system_port)); } }