feat: inject DGD id in planner env variables

deploy/cloud/operator/api/v1alpha1/dynamocomponentdeployment_types.go

@@ -260,3 +260,16 @@ func (s *DynamoComponentDeploymentSharedSpec) GetNumberOfNodes() int32 {
 	}
 	return 1
 }
+
+func (s *DynamoComponentDeployment) GetParentGraphDeploymentName() string {
+	for _, ownerRef := range s.ObjectMeta.OwnerReferences {
+		if ownerRef.Kind == "DynamoGraphDeployment" {
+			return ownerRef.Name
+		}
+	}
+	return ""
+}
+
+func (s *DynamoComponentDeployment) GetParentGraphDeploymentNamespace() string {
+	return s.GetNamespace()
+}

deploy/cloud/operator/api/v1alpha1/dynamocomponentdeployment_types_test.go

@@ -210,3 +210,54 @@ func TestDynamoComponentDeployment_SetDynamoDeploymentConfig(t *testing.T) {
 		})
 	}
 }
+
+func TestDynamoComponentDeployment_GetParentGraphDeploymentName(t *testing.T) {
+	type fields struct {
+		TypeMeta   metav1.TypeMeta
+		ObjectMeta metav1.ObjectMeta
+		Spec       DynamoComponentDeploymentSpec
+		Status     DynamoComponentDeploymentStatus
+	}
+	tests := []struct {
+		name   string
+		fields fields
+		want   string
+	}{
+		{
+			name: "test",
+			fields: fields{
+				ObjectMeta: metav1.ObjectMeta{
+					OwnerReferences: []metav1.OwnerReference{
+						{
+							Kind: "DynamoGraphDeployment",
+							Name: "name",
+						},
+					},
+				},
+			},
+			want: "name",
+		},
+		{
+			name: "no owner reference",
+			fields: fields{
+				ObjectMeta: metav1.ObjectMeta{
+					OwnerReferences: []metav1.OwnerReference{},
+				},
+			},
+			want: "",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := &DynamoComponentDeployment{
+				TypeMeta:   tt.fields.TypeMeta,
+				ObjectMeta: tt.fields.ObjectMeta,
+				Spec:       tt.fields.Spec,
+				Status:     tt.fields.Status,
+			}
+			if got := s.GetParentGraphDeploymentName(); got != tt.want {
+				t.Errorf("DynamoComponentDeployment.GetParentGraphDeploymentName() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}

deploy/cloud/operator/internal/dynamo/component_common.go

@@ -14,7 +14,7 @@ import (
 type ComponentDefaults interface {
 	// GetBaseContainer returns the base container configuration for this component type
 	// The numberOfNodes parameter indicates the total number of nodes in the deployment
-	GetBaseContainer(numberOfNodes int32) (corev1.Container, error)
+	GetBaseContainer(numberOfNodes int32, parentGraphDeploymentName, parentGraphDeploymentNamespace string) (corev1.Container, error)
 
 	// GetBasePodSpec returns the base pod spec configuration for this component type
 	// The numberOfNodes parameter indicates the total number of nodes in the deployment
@@ -38,7 +38,7 @@ func ComponentDefaultsFactory(componentType string, numberOfNodes int32) Compone
 // BaseComponentDefaults provides common defaults shared by all components
 type BaseComponentDefaults struct{}
 
-func (b *BaseComponentDefaults) GetBaseContainer(numberOfNodes int32) (corev1.Container, error) {
+func (b *BaseComponentDefaults) GetBaseContainer(numberOfNodes int32, parentGraphDeploymentName, parentGraphDeploymentNamespace string) (corev1.Container, error) {
 	return b.getCommonContainer(), nil
 }
 

deploy/cloud/operator/internal/dynamo/component_frontend.go

@@ -23,7 +23,7 @@ func NewFrontendDefaults() *FrontendDefaults {
 	return &FrontendDefaults{&BaseComponentDefaults{}}
 }
 
-func (f *FrontendDefaults) GetBaseContainer(numberOfNodes int32) (corev1.Container, error) {
+func (f *FrontendDefaults) GetBaseContainer(numberOfNodes int32, parentGraphDeploymentName, parentGraphDeploymentNamespace string) (corev1.Container, error) {
 	// Frontend doesn't need backend-specific config
 	container := f.getCommonContainer()
 

deploy/cloud/operator/internal/dynamo/component_planner.go

@@ -20,7 +20,7 @@ func NewPlannerDefaults() *PlannerDefaults {
 	return &PlannerDefaults{&BaseComponentDefaults{}}
 }
 
-func (p *PlannerDefaults) GetBaseContainer(numberOfNodes int32) (corev1.Container, error) {
+func (p *PlannerDefaults) GetBaseContainer(numberOfNodes int32, parentGraphDeploymentName, parentGraphDeploymentNamespace string) (corev1.Container, error) {
 	container := p.getCommonContainer()
 
 	// Add planner-specific defaults
@@ -35,6 +35,17 @@ func (p *PlannerDefaults) GetBaseContainer(numberOfNodes int32) (corev1.Containe
 		},
 	}
 
+	container.Env = []corev1.EnvVar{
+		{
+			Name:  "K8S_PARENT_NAME",
+			Value: parentGraphDeploymentName,
+		},
+		{
+			Name:  "K8S_PARENT_NAMESPACE",
+			Value: parentGraphDeploymentNamespace,
+		},
+	}
+
 	return container, nil
 }
 

deploy/cloud/operator/internal/dynamo/component_planner_test.go

@@ -0,0 +1,81 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package dynamo
+
+import (
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
+)
+
+func TestPlannerDefaults_GetBaseContainer(t *testing.T) {
+	type fields struct {
+		BaseComponentDefaults *BaseComponentDefaults
+	}
+	type args struct {
+		numberOfNodes                  int32
+		parentGraphDeploymentName      string
+		parentGraphDeploymentNamespace string
+	}
+	tests := []struct {
+		name    string
+		fields  fields
+		args    args
+		want    corev1.Container
+		wantErr bool
+	}{
+		{
+			name: "test",
+			fields: fields{
+				BaseComponentDefaults: &BaseComponentDefaults{},
+			},
+			args: args{
+				numberOfNodes:                  1,
+				parentGraphDeploymentName:      "name",
+				parentGraphDeploymentNamespace: "namespace",
+			},
+			want: corev1.Container{
+				Name: "main",
+				Command: []string{
+					"/bin/sh",
+					"-c",
+				},
+				Resources: corev1.ResourceRequirements{
+					Requests: corev1.ResourceList{
+						corev1.ResourceCPU:    resource.MustParse("2"),
+						corev1.ResourceMemory: resource.MustParse("2Gi"),
+					},
+					Limits: corev1.ResourceList{
+						corev1.ResourceCPU:    resource.MustParse("2"),
+						corev1.ResourceMemory: resource.MustParse("2Gi"),
+					},
+				},
+				Env: []corev1.EnvVar{
+					{Name: "K8S_PARENT_NAME", Value: "name"},
+					{Name: "K8S_PARENT_NAMESPACE", Value: "namespace"},
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			p := &PlannerDefaults{
+				BaseComponentDefaults: tt.fields.BaseComponentDefaults,
+			}
+			got, err := p.GetBaseContainer(tt.args.numberOfNodes, tt.args.parentGraphDeploymentName, tt.args.parentGraphDeploymentNamespace)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("PlannerDefaults.GetBaseContainer() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			diff := cmp.Diff(got, tt.want)
+			if diff != "" {
+				t.Errorf("PlannerDefaults.GetBaseContainer() = %v, want %v", diff, tt.want)
+			}
+		})
+	}
+}

deploy/cloud/operator/internal/dynamo/component_worker.go

@@ -23,7 +23,7 @@ func NewWorkerDefaults() *WorkerDefaults {
 	return &WorkerDefaults{&BaseComponentDefaults{}}
 }
 
-func (w *WorkerDefaults) GetBaseContainer(numberOfNodes int32) (corev1.Container, error) {
+func (w *WorkerDefaults) GetBaseContainer(numberOfNodes int32, parentGraphDeploymentName, parentGraphDeploymentNamespace string) (corev1.Container, error) {
 	container := w.getCommonContainer()
 
 	// Add system port

deploy/cloud/operator/internal/dynamo/graph.go

@@ -667,6 +667,7 @@ func GenerateBasePodSpec(
 	component *v1alpha1.DynamoComponentDeploymentOverridesSpec,
 	backendFramework BackendFramework,
 	secretsRetriever SecretsRetriever,
+	parentGraphDeploymentName string,
 	namespace string,
 	role Role,
 	numberOfNodes int32,
@@ -676,7 +677,7 @@ func GenerateBasePodSpec(
 ) (corev1.PodSpec, error) {
 	// Start with base container generated per component type
 	componentDefaults := ComponentDefaultsFactory(component.ComponentType, numberOfNodes)
-	container, err := componentDefaults.GetBaseContainer(numberOfNodes)
+	container, err := componentDefaults.GetBaseContainer(numberOfNodes, parentGraphDeploymentName, namespace)
 	if err != nil {
 		return corev1.PodSpec{}, fmt.Errorf("failed to get base container: %w", err)
 	}
@@ -823,7 +824,7 @@ func GeneratePodSpecForComponent(
 	if len(dynamoDeployment.Spec.Envs) > 0 {
 		component.Envs = MergeEnvs(dynamoDeployment.Spec.Envs, component.Envs)
 	}
-	podSpec, err := GenerateBasePodSpec(component, backendFramework, secretsRetriever, dynamoDeployment.Namespace, role, numberOfNodes, controllerConfig, multinodeDeploymentType, serviceName)
+	podSpec, err := GenerateBasePodSpec(component, backendFramework, secretsRetriever, dynamoDeployment.Name, dynamoDeployment.Namespace, role, numberOfNodes, controllerConfig, multinodeDeploymentType, serviceName)
 	if err != nil {
 		return corev1.PodSpec{}, err
 	}
@@ -1129,6 +1130,7 @@ func GenerateBasePodSpecForController(
 		componentSpec,
 		backendFramework,
 		secretsRetriever,
+		dynComponent.GetParentGraphDeploymentName(),
 		dynComponent.Namespace,
 		role,
 		numberOfNodes,

deploy/cloud/operator/internal/dynamo/graph_test.go

@@ -4126,6 +4126,7 @@ func TestGenerateBasePodSpec_PlannerServiceAccount(t *testing.T) {
 				tt.component,
 				BackendFrameworkSGLang,
 				secretsRetriever,
+				"test-deployment",
 				"default",
 				RoleMain,
 				1,