feat: add scaling adapter (#4699)

Signed-off-by: Julien Mancuso <jmancuso@nvidia.com>

deploy/cloud/helm/crds/templates/nvidia.com_dynamocomponentdeployments.yaml

@@ -77,12 +77,13 @@ spec:
                     (such as Pod, Service, and Ingress when applicable).
                   type: object
                 autoscaling:
-                  description: Autoscaling config for this component (replica range, target utilization, etc.).
+                  description: |-
+                    Deprecated: This field is deprecated and ignored. Use DynamoGraphDeploymentScalingAdapter
+                    with HPA, KEDA, or Planner for autoscaling instead. See docs/kubernetes/autoscaling.md
+                    for migration guidance. This field will be removed in a future API version.
                   properties:
                     behavior:
-                      description: |-
-                        HorizontalPodAutoscalerBehavior configures the scaling behavior of the target
-                        in both Up and Down directions (scaleUp and scaleDown fields respectively).
+                      description: 'Deprecated: This field is ignored.'
                       properties:
                         scaleDown:
                           description: |-
@@ -231,10 +232,13 @@ spec:
                           type: object
                       type: object
                     enabled:
+                      description: 'Deprecated: This field is ignored.'
                       type: boolean
                     maxReplicas:
+                      description: 'Deprecated: This field is ignored.'
                       type: integer
                     metrics:
+                      description: 'Deprecated: This field is ignored.'
                       items:
                         description: |-
                           MetricSpec specifies how to scale based on a single metric
@@ -665,6 +669,7 @@ spec:
                         type: object
                       type: array
                     minReplicas:
+                      description: 'Deprecated: This field is ignored.'
                       type: integer
                   type: object
                 backendFramework:
@@ -10184,8 +10189,12 @@ spec:
                       type: integer
                   type: object
                 replicas:
-                  description: Replicas is the desired number of Pods for this component when autoscaling is not used.
+                  description: |-
+                    Replicas is the desired number of Pods for this component.
+                    When scalingAdapter is enabled (default), this field is managed by the
+                    DynamoGraphDeploymentScalingAdapter and should not be modified directly.
                   format: int32
+                  minimum: 0
                   type: integer
                 resources:
                   description: |-
@@ -10264,6 +10273,20 @@ spec:
                           type: string
                       type: object
                   type: object
+                scalingAdapter:
+                  description: |-
+                    ScalingAdapter configures whether this service uses the DynamoGraphDeploymentScalingAdapter.
+                    When enabled (default), replicas are managed via DGDSA and external autoscalers can scale
+                    the service using the Scale subresource. When disabled, replicas can be modified directly.
+                  properties:
+                    disable:
+                      default: false
+                      description: |-
+                        Disable indicates whether the ScalingAdapter should be disabled for this service.
+                        When false (default), a DGDSA is created and owns the replicas field.
+                        When true, no DGDSA is created and replicas can be modified directly in the DGD.
+                      type: boolean
+                  type: object
                 serviceName:
                   description: The name of the component
                   type: string

deploy/cloud/helm/crds/templates/nvidia.com_dynamographdeployments.yaml

@@ -219,12 +219,13 @@ spec:
                           (such as Pod, Service, and Ingress when applicable).
                         type: object
                       autoscaling:
-                        description: Autoscaling config for this component (replica range, target utilization, etc.).
+                        description: |-
+                          Deprecated: This field is deprecated and ignored. Use DynamoGraphDeploymentScalingAdapter
+                          with HPA, KEDA, or Planner for autoscaling instead. See docs/kubernetes/autoscaling.md
+                          for migration guidance. This field will be removed in a future API version.
                         properties:
                           behavior:
-                            description: |-
-                              HorizontalPodAutoscalerBehavior configures the scaling behavior of the target
-                              in both Up and Down directions (scaleUp and scaleDown fields respectively).
+                            description: 'Deprecated: This field is ignored.'
                             properties:
                               scaleDown:
                                 description: |-
@@ -373,10 +374,13 @@ spec:
                                 type: object
                             type: object
                           enabled:
+                            description: 'Deprecated: This field is ignored.'
                             type: boolean
                           maxReplicas:
+                            description: 'Deprecated: This field is ignored.'
                             type: integer
                           metrics:
+                            description: 'Deprecated: This field is ignored.'
                             items:
                               description: |-
                                 MetricSpec specifies how to scale based on a single metric
@@ -807,6 +811,7 @@ spec:
                               type: object
                             type: array
                           minReplicas:
+                            description: 'Deprecated: This field is ignored.'
                             type: integer
                         type: object
                       componentType:
@@ -10319,8 +10324,12 @@ spec:
                             type: integer
                         type: object
                       replicas:
-                        description: Replicas is the desired number of Pods for this component when autoscaling is not used.
+                        description: |-
+                          Replicas is the desired number of Pods for this component.
+                          When scalingAdapter is enabled (default), this field is managed by the
+                          DynamoGraphDeploymentScalingAdapter and should not be modified directly.
                         format: int32
+                        minimum: 0
                         type: integer
                       resources:
                         description: |-
@@ -10399,6 +10408,20 @@ spec:
                                 type: string
                             type: object
                         type: object
+                      scalingAdapter:
+                        description: |-
+                          ScalingAdapter configures whether this service uses the DynamoGraphDeploymentScalingAdapter.
+                          When enabled (default), replicas are managed via DGDSA and external autoscalers can scale
+                          the service using the Scale subresource. When disabled, replicas can be modified directly.
+                        properties:
+                          disable:
+                            default: false
+                            description: |-
+                              Disable indicates whether the ScalingAdapter should be disabled for this service.
+                              When false (default), a DGDSA is created and owns the replicas field.
+                              When true, no DGDSA is created and replicas can be modified directly in the DGD.
+                            type: boolean
+                        type: object
                       serviceName:
                         description: The name of the component
                         type: string

deploy/cloud/helm/crds/templates/nvidia.com_dynamographdeploymentscalingadapters.yaml

+# SPDX-FileCopyrightText: Copyright (c) 2024-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# 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.
+
+---
+apiVersion: apiextensions.k8s.io/v1
+kind: CustomResourceDefinition
+metadata:
+  annotations:
+    controller-gen.kubebuilder.io/version: v0.16.4
+    helm.sh/resource-policy: keep
+  name: dynamographdeploymentscalingadapters.nvidia.com
+spec:
+  group: nvidia.com
+  names:
+    kind: DynamoGraphDeploymentScalingAdapter
+    listKind: DynamoGraphDeploymentScalingAdapterList
+    plural: dynamographdeploymentscalingadapters
+    shortNames:
+      - dgdsa
+    singular: dynamographdeploymentscalingadapter
+  scope: Namespaced
+  versions:
+    - additionalPrinterColumns:
+        - description: DynamoGraphDeployment name
+          jsonPath: .spec.dgdRef.name
+          name: DGD
+          type: string
+        - description: Service name
+          jsonPath: .spec.dgdRef.serviceName
+          name: SERVICE
+          type: string
+        - description: Current replicas
+          jsonPath: .status.replicas
+          name: REPLICAS
+          type: integer
+        - jsonPath: .metadata.creationTimestamp
+          name: AGE
+          type: date
+      name: v1alpha1
+      schema:
+        openAPIV3Schema:
+          description: |-
+            DynamoGraphDeploymentScalingAdapter provides a scaling interface for individual services
+            within a DynamoGraphDeployment. It implements the Kubernetes scale
+            subresource, enabling integration with HPA, KEDA, and custom autoscalers.
+
+            The adapter acts as an intermediary between autoscalers and the DGD,
+            ensuring that only the adapter controller modifies the DGD's service replicas.
+            This prevents conflicts when multiple autoscaling mechanisms are in play.
+          properties:
+            apiVersion:
+              description: |-
+                APIVersion defines the versioned schema of this representation of an object.
+                Servers should convert recognized schemas to the latest internal value, and
+                may reject unrecognized values.
+                More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources
+              type: string
+            kind:
+              description: |-
+                Kind is a string value representing the REST resource this object represents.
+                Servers may infer this from the endpoint the client submits requests to.
+                Cannot be updated.
+                In CamelCase.
+                More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds
+              type: string
+            metadata:
+              type: object
+            spec:
+              description: DynamoGraphDeploymentScalingAdapterSpec defines the desired state of DynamoGraphDeploymentScalingAdapter
+              properties:
+                dgdRef:
+                  description: DGDRef references the DynamoGraphDeployment and the specific service to scale.
+                  properties:
+                    name:
+                      description: Name of the DynamoGraphDeployment
+                      minLength: 1
+                      type: string
+                    serviceName:
+                      description: ServiceName is the key name of the service within the DGD's spec.services map to scale
+                      minLength: 1
+                      type: string
+                  required:
+                    - name
+                    - serviceName
+                  type: object
+                replicas:
+                  description: |-
+                    Replicas is the desired number of replicas for the target service.
+                    This field is modified by external autoscalers (HPA/KEDA/Planner) or manually by users.
+                  format: int32
+                  minimum: 0
+                  type: integer
+              required:
+                - dgdRef
+                - replicas
+              type: object
+            status:
+              description: DynamoGraphDeploymentScalingAdapterStatus defines the observed state of DynamoGraphDeploymentScalingAdapter
+              properties:
+                lastScaleTime:
+                  description: LastScaleTime is the last time the adapter scaled the target service.
+                  format: date-time
+                  type: string
+                replicas:
+                  description: |-
+                    Replicas is the current number of replicas for the target service.
+                    This is synced from the DGD's service replicas and is required for the scale subresource.
+                  format: int32
+                  type: integer
+                selector:
+                  description: |-
+                    Selector is a label selector string for the pods managed by this adapter.
+                    Required for HPA compatibility via the scale subresource.
+                  type: string
+              type: object
+          type: object
+      served: true
+      storage: true
+      subresources:
+        scale:
+          labelSelectorPath: .status.selector
+          specReplicasPath: .spec.replicas
+          statusReplicasPath: .status.replicas
+        status: {}

deploy/cloud/helm/platform/components/operator/templates/manager-rbac.yaml

@@ -369,6 +369,7 @@ rules:
   - dynamocomponentdeployments
   - dynamographdeploymentrequests
   - dynamographdeployments
+  - dynamographdeploymentscalingadapters
   - dynamomodels
   verbs:
   - create
@@ -393,6 +394,7 @@ rules:
   - dynamocomponentdeployments/status
   - dynamographdeploymentrequests/status
   - dynamographdeployments/status
+  - dynamographdeploymentscalingadapters/status
   - dynamomodels/status
   verbs:
   - get

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

@@ -53,11 +53,19 @@ type VolumeMount struct {
 	UseAsCompilationCache bool `json:"useAsCompilationCache,omitempty"`
 }
 
+// Deprecated: This field is deprecated and ignored. Use DynamoGraphDeploymentScalingAdapter
+// with HPA, KEDA, or Planner for autoscaling instead. See docs/kubernetes/autoscaling.md
+// for migration guidance. This field will be removed in a future API version.
 type Autoscaling struct {
+	// Deprecated: This field is ignored.
 	Enabled bool `json:"enabled,omitempty"`
+	// Deprecated: This field is ignored.
 	MinReplicas int `json:"minReplicas,omitempty"`
+	// Deprecated: This field is ignored.
 	MaxReplicas int `json:"maxReplicas,omitempty"`
+	// Deprecated: This field is ignored.
 	Behavior *autoscalingv2.HorizontalPodAutoscalerBehavior `json:"behavior,omitempty"`
+	// Deprecated: This field is ignored.
 	Metrics []autoscalingv2.MetricSpec `json:"metrics,omitempty"`
 }
 
@@ -115,3 +123,15 @@ type ExtraPodSpec struct {
 	*corev1.PodSpec `json:",inline"`
 	MainContainer   *corev1.Container `json:"mainContainer,omitempty"`
 }
+
+// ScalingAdapter configures whether a service uses the DynamoGraphDeploymentScalingAdapter
+// for replica management. When enabled (default), the DGDSA owns the replicas field and
+// external autoscalers (HPA, KEDA, Planner) can control scaling via the Scale subresource.
+type ScalingAdapter struct {
+	// Disable indicates whether the ScalingAdapter should be disabled for this service.
+	// When false (default), a DGDSA is created and owns the replicas field.
+	// When true, no DGDSA is created and replicas can be modified directly in the DGD.
+	// +optional
+	// +kubebuilder:default=false
+	Disable bool `json:"disable,omitempty"`
+}

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

@@ -74,7 +74,9 @@ type DynamoComponentDeploymentSharedSpec struct {
 	// Resources requested and limits for this component, including CPU, memory,
 	// GPUs/devices, and any runtime-specific resources.
 	Resources *Resources `json:"resources,omitempty"`
-	// Autoscaling config for this component (replica range, target utilization, etc.).
+	// Deprecated: This field is deprecated and ignored. Use DynamoGraphDeploymentScalingAdapter
+	// with HPA, KEDA, or Planner for autoscaling instead. See docs/kubernetes/autoscaling.md
+	// for migration guidance. This field will be removed in a future API version.
 	Autoscaling *Autoscaling `json:"autoscaling,omitempty"`
 	// Envs defines additional environment variables to inject into the component containers.
 	Envs []corev1.EnvVar `json:"envs,omitempty"`
@@ -108,10 +110,18 @@ type DynamoComponentDeploymentSharedSpec struct {
 	LivenessProbe *corev1.Probe `json:"livenessProbe,omitempty"`
 	// ReadinessProbe to signal when the container is ready to receive traffic.
 	ReadinessProbe *corev1.Probe `json:"readinessProbe,omitempty"`
-	// Replicas is the desired number of Pods for this component when autoscaling is not used.
+	// Replicas is the desired number of Pods for this component.
+	// When scalingAdapter is enabled (default), this field is managed by the
+	// DynamoGraphDeploymentScalingAdapter and should not be modified directly.
+	// +kubebuilder:validation:Minimum=0
 	Replicas *int32 `json:"replicas,omitempty"`
 	// Multinode is the configuration for multinode components.
 	Multinode *MultinodeSpec `json:"multinode,omitempty"`
+	// ScalingAdapter configures whether this service uses the DynamoGraphDeploymentScalingAdapter.
+	// When enabled (default), replicas are managed via DGDSA and external autoscalers can scale
+	// the service using the Scale subresource. When disabled, replicas can be modified directly.
+	// +optional
+	ScalingAdapter *ScalingAdapter `json:"scalingAdapter,omitempty"`
 }
 
 type MultinodeSpec struct {

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

+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * 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.
+ */
+
+package v1alpha1
+
+import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+// DynamoGraphDeploymentScalingAdapterSpec defines the desired state of DynamoGraphDeploymentScalingAdapter
+type DynamoGraphDeploymentScalingAdapterSpec struct {
+	// Replicas is the desired number of replicas for the target service.
+	// This field is modified by external autoscalers (HPA/KEDA/Planner) or manually by users.
+	// +kubebuilder:validation:Required
+	// +kubebuilder:validation:Minimum=0
+	Replicas int32 `json:"replicas"`
+
+	// DGDRef references the DynamoGraphDeployment and the specific service to scale.
+	// +kubebuilder:validation:Required
+	DGDRef DynamoGraphDeploymentServiceRef `json:"dgdRef"`
+}
+
+// DynamoGraphDeploymentServiceRef identifies a specific service within a DynamoGraphDeployment
+type DynamoGraphDeploymentServiceRef struct {
+	// Name of the DynamoGraphDeployment
+	// +kubebuilder:validation:Required
+	// +kubebuilder:validation:MinLength=1
+	Name string `json:"name"`
+
+	// ServiceName is the key name of the service within the DGD's spec.services map to scale
+	// +kubebuilder:validation:Required
+	// +kubebuilder:validation:MinLength=1
+	ServiceName string `json:"serviceName"`
+}
+
+// DynamoGraphDeploymentScalingAdapterStatus defines the observed state of DynamoGraphDeploymentScalingAdapter
+type DynamoGraphDeploymentScalingAdapterStatus struct {
+	// Replicas is the current number of replicas for the target service.
+	// This is synced from the DGD's service replicas and is required for the scale subresource.
+	// +optional
+	Replicas int32 `json:"replicas,omitempty"`
+
+	// Selector is a label selector string for the pods managed by this adapter.
+	// Required for HPA compatibility via the scale subresource.
+	// +optional
+	Selector string `json:"selector,omitempty"`
+
+	// LastScaleTime is the last time the adapter scaled the target service.
+	// +optional
+	LastScaleTime *metav1.Time `json:"lastScaleTime,omitempty"`
+}
+
+// +kubebuilder:object:root=true
+// +kubebuilder:subresource:status
+// +kubebuilder:subresource:scale:specpath=.spec.replicas,statuspath=.status.replicas,selectorpath=.status.selector
+// +kubebuilder:printcolumn:name="DGD",type="string",JSONPath=".spec.dgdRef.name",description="DynamoGraphDeployment name"
+// +kubebuilder:printcolumn:name="SERVICE",type="string",JSONPath=".spec.dgdRef.serviceName",description="Service name"
+// +kubebuilder:printcolumn:name="REPLICAS",type="integer",JSONPath=".status.replicas",description="Current replicas"
+// +kubebuilder:printcolumn:name="AGE",type="date",JSONPath=".metadata.creationTimestamp"
+// +kubebuilder:resource:shortName={dgdsa}
+
+// DynamoGraphDeploymentScalingAdapter provides a scaling interface for individual services
+// within a DynamoGraphDeployment. It implements the Kubernetes scale
+// subresource, enabling integration with HPA, KEDA, and custom autoscalers.
+//
+// The adapter acts as an intermediary between autoscalers and the DGD,
+// ensuring that only the adapter controller modifies the DGD's service replicas.
+// This prevents conflicts when multiple autoscaling mechanisms are in play.
+type DynamoGraphDeploymentScalingAdapter struct {
+	metav1.TypeMeta   `json:",inline"`
+	metav1.ObjectMeta `json:"metadata,omitempty"`
+
+	Spec   DynamoGraphDeploymentScalingAdapterSpec   `json:"spec,omitempty"`
+	Status DynamoGraphDeploymentScalingAdapterStatus `json:"status,omitempty"`
+}
+
+// +kubebuilder:object:root=true
+
+// DynamoGraphDeploymentScalingAdapterList contains a list of DynamoGraphDeploymentScalingAdapter
+type DynamoGraphDeploymentScalingAdapterList struct {
+	metav1.TypeMeta `json:",inline"`
+	metav1.ListMeta `json:"metadata,omitempty"`
+	Items           []DynamoGraphDeploymentScalingAdapter `json:"items"`
+}
+
+func init() {
+	SchemeBuilder.Register(&DynamoGraphDeploymentScalingAdapter{}, &DynamoGraphDeploymentScalingAdapterList{})
+}

deploy/cloud/operator/api/v1alpha1/zz_generated.deepcopy.go

@@ -371,6 +371,11 @@ func (in *DynamoComponentDeploymentSharedSpec) DeepCopyInto(out *DynamoComponent
 		*out = new(MultinodeSpec)
 		**out = **in
 	}
+	if in.ScalingAdapter != nil {
+		in, out := &in.ScalingAdapter, &out.ScalingAdapter
+		*out = new(ScalingAdapter)
+		**out = **in
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new DynamoComponentDeploymentSharedSpec.
@@ -599,6 +604,115 @@ func (in *DynamoGraphDeploymentRequestStatus) DeepCopy() *DynamoGraphDeploymentR
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *DynamoGraphDeploymentScalingAdapter) DeepCopyInto(out *DynamoGraphDeploymentScalingAdapter) {
+	*out = *in
+	out.TypeMeta = in.TypeMeta
+	in.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
+	out.Spec = in.Spec
+	in.Status.DeepCopyInto(&out.Status)
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new DynamoGraphDeploymentScalingAdapter.
+func (in *DynamoGraphDeploymentScalingAdapter) DeepCopy() *DynamoGraphDeploymentScalingAdapter {
+	if in == nil {
+		return nil
+	}
+	out := new(DynamoGraphDeploymentScalingAdapter)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
+func (in *DynamoGraphDeploymentScalingAdapter) DeepCopyObject() runtime.Object {
+	if c := in.DeepCopy(); c != nil {
+		return c
+	}
+	return nil
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *DynamoGraphDeploymentScalingAdapterList) DeepCopyInto(out *DynamoGraphDeploymentScalingAdapterList) {
+	*out = *in
+	out.TypeMeta = in.TypeMeta
+	in.ListMeta.DeepCopyInto(&out.ListMeta)
+	if in.Items != nil {
+		in, out := &in.Items, &out.Items
+		*out = make([]DynamoGraphDeploymentScalingAdapter, len(*in))
+		for i := range *in {
+			(*in)[i].DeepCopyInto(&(*out)[i])
+		}
+	}
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new DynamoGraphDeploymentScalingAdapterList.
+func (in *DynamoGraphDeploymentScalingAdapterList) DeepCopy() *DynamoGraphDeploymentScalingAdapterList {
+	if in == nil {
+		return nil
+	}
+	out := new(DynamoGraphDeploymentScalingAdapterList)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
+func (in *DynamoGraphDeploymentScalingAdapterList) DeepCopyObject() runtime.Object {
+	if c := in.DeepCopy(); c != nil {
+		return c
+	}
+	return nil
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *DynamoGraphDeploymentScalingAdapterSpec) DeepCopyInto(out *DynamoGraphDeploymentScalingAdapterSpec) {
+	*out = *in
+	out.DGDRef = in.DGDRef
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new DynamoGraphDeploymentScalingAdapterSpec.
+func (in *DynamoGraphDeploymentScalingAdapterSpec) DeepCopy() *DynamoGraphDeploymentScalingAdapterSpec {
+	if in == nil {
+		return nil
+	}
+	out := new(DynamoGraphDeploymentScalingAdapterSpec)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *DynamoGraphDeploymentScalingAdapterStatus) DeepCopyInto(out *DynamoGraphDeploymentScalingAdapterStatus) {
+	*out = *in
+	if in.LastScaleTime != nil {
+		in, out := &in.LastScaleTime, &out.LastScaleTime
+		*out = (*in).DeepCopy()
+	}
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new DynamoGraphDeploymentScalingAdapterStatus.
+func (in *DynamoGraphDeploymentScalingAdapterStatus) DeepCopy() *DynamoGraphDeploymentScalingAdapterStatus {
+	if in == nil {
+		return nil
+	}
+	out := new(DynamoGraphDeploymentScalingAdapterStatus)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *DynamoGraphDeploymentServiceRef) DeepCopyInto(out *DynamoGraphDeploymentServiceRef) {
+	*out = *in
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new DynamoGraphDeploymentServiceRef.
+func (in *DynamoGraphDeploymentServiceRef) DeepCopy() *DynamoGraphDeploymentServiceRef {
+	if in == nil {
+		return nil
+	}
+	out := new(DynamoGraphDeploymentServiceRef)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *DynamoGraphDeploymentSpec) DeepCopyInto(out *DynamoGraphDeploymentSpec) {
 	*out = *in
@@ -1085,6 +1199,21 @@ func (in *Resources) DeepCopy() *Resources {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *ScalingAdapter) DeepCopyInto(out *ScalingAdapter) {
+	*out = *in
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ScalingAdapter.
+func (in *ScalingAdapter) DeepCopy() *ScalingAdapter {
+	if in == nil {
+		return nil
+	}
+	out := new(ScalingAdapter)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *SharedMemorySpec) DeepCopyInto(out *SharedMemorySpec) {
 	*out = *in

deploy/cloud/operator/cmd/main.go

@@ -578,6 +578,16 @@ func main() {
 		os.Exit(1)
 	}
 
+	if err = (&controller.DynamoGraphDeploymentScalingAdapterReconciler{
+		Client:   mgr.GetClient(),
+		Scheme:   mgr.GetScheme(),
+		Recorder: mgr.GetEventRecorderFor("dgdscalingadapter"),
+		Config:   ctrlConfig,
+	}).SetupWithManager(mgr); err != nil {
+		setupLog.Error(err, "unable to create controller", "controller", "DGDScalingAdapter")
+		os.Exit(1)
+	}
+
 	if err = (&controller.DynamoGraphDeploymentRequestReconciler{
 		Client:      mgr.GetClient(),
 		Recorder:    mgr.GetEventRecorderFor("dynamographdeploymentrequest"),

deploy/cloud/operator/config/crd/bases/nvidia.com_dynamocomponentdeployments.yaml

@@ -77,12 +77,13 @@ spec:
                     (such as Pod, Service, and Ingress when applicable).
                   type: object
                 autoscaling:
-                  description: Autoscaling config for this component (replica range, target utilization, etc.).
+                  description: |-
+                    Deprecated: This field is deprecated and ignored. Use DynamoGraphDeploymentScalingAdapter
+                    with HPA, KEDA, or Planner for autoscaling instead. See docs/kubernetes/autoscaling.md
+                    for migration guidance. This field will be removed in a future API version.
                   properties:
                     behavior:
-                      description: |-
-                        HorizontalPodAutoscalerBehavior configures the scaling behavior of the target
-                        in both Up and Down directions (scaleUp and scaleDown fields respectively).
+                      description: 'Deprecated: This field is ignored.'
                       properties:
                         scaleDown:
                           description: |-
@@ -231,10 +232,13 @@ spec:
                           type: object
                       type: object
                     enabled:
+                      description: 'Deprecated: This field is ignored.'
                       type: boolean
                     maxReplicas:
+                      description: 'Deprecated: This field is ignored.'
                       type: integer
                     metrics:
+                      description: 'Deprecated: This field is ignored.'
                       items:
                         description: |-
                           MetricSpec specifies how to scale based on a single metric
@@ -665,6 +669,7 @@ spec:
                         type: object
                       type: array
                     minReplicas:
+                      description: 'Deprecated: This field is ignored.'
                       type: integer
                   type: object
                 backendFramework:
@@ -10184,8 +10189,12 @@ spec:
                       type: integer
                   type: object
                 replicas:
-                  description: Replicas is the desired number of Pods for this component when autoscaling is not used.
+                  description: |-
+                    Replicas is the desired number of Pods for this component.
+                    When scalingAdapter is enabled (default), this field is managed by the
+                    DynamoGraphDeploymentScalingAdapter and should not be modified directly.
                   format: int32
+                  minimum: 0
                   type: integer
                 resources:
                   description: |-
@@ -10264,6 +10273,20 @@ spec:
                           type: string
                       type: object
                   type: object
+                scalingAdapter:
+                  description: |-
+                    ScalingAdapter configures whether this service uses the DynamoGraphDeploymentScalingAdapter.
+                    When enabled (default), replicas are managed via DGDSA and external autoscalers can scale
+                    the service using the Scale subresource. When disabled, replicas can be modified directly.
+                  properties:
+                    disable:
+                      default: false
+                      description: |-
+                        Disable indicates whether the ScalingAdapter should be disabled for this service.
+                        When false (default), a DGDSA is created and owns the replicas field.
+                        When true, no DGDSA is created and replicas can be modified directly in the DGD.
+                      type: boolean
+                  type: object
                 serviceName:
                   description: The name of the component
                   type: string

deploy/cloud/operator/config/crd/bases/nvidia.com_dynamographdeployments.yaml

@@ -219,12 +219,13 @@ spec:
                           (such as Pod, Service, and Ingress when applicable).
                         type: object
                       autoscaling:
-                        description: Autoscaling config for this component (replica range, target utilization, etc.).
+                        description: |-
+                          Deprecated: This field is deprecated and ignored. Use DynamoGraphDeploymentScalingAdapter
+                          with HPA, KEDA, or Planner for autoscaling instead. See docs/kubernetes/autoscaling.md
+                          for migration guidance. This field will be removed in a future API version.
                         properties:
                           behavior:
-                            description: |-
-                              HorizontalPodAutoscalerBehavior configures the scaling behavior of the target
-                              in both Up and Down directions (scaleUp and scaleDown fields respectively).
+                            description: 'Deprecated: This field is ignored.'
                             properties:
                               scaleDown:
                                 description: |-
@@ -373,10 +374,13 @@ spec:
                                 type: object
                             type: object
                           enabled:
+                            description: 'Deprecated: This field is ignored.'
                             type: boolean
                           maxReplicas:
+                            description: 'Deprecated: This field is ignored.'
                             type: integer
                           metrics:
+                            description: 'Deprecated: This field is ignored.'
                             items:
                               description: |-
                                 MetricSpec specifies how to scale based on a single metric
@@ -807,6 +811,7 @@ spec:
                               type: object
                             type: array
                           minReplicas:
+                            description: 'Deprecated: This field is ignored.'
                             type: integer
                         type: object
                       componentType:
@@ -10319,8 +10324,12 @@ spec:
                             type: integer
                         type: object
                       replicas:
-                        description: Replicas is the desired number of Pods for this component when autoscaling is not used.
+                        description: |-
+                          Replicas is the desired number of Pods for this component.
+                          When scalingAdapter is enabled (default), this field is managed by the
+                          DynamoGraphDeploymentScalingAdapter and should not be modified directly.
                         format: int32
+                        minimum: 0
                         type: integer
                       resources:
                         description: |-
@@ -10399,6 +10408,20 @@ spec:
                                 type: string
                             type: object
                         type: object
+                      scalingAdapter:
+                        description: |-
+                          ScalingAdapter configures whether this service uses the DynamoGraphDeploymentScalingAdapter.
+                          When enabled (default), replicas are managed via DGDSA and external autoscalers can scale
+                          the service using the Scale subresource. When disabled, replicas can be modified directly.
+                        properties:
+                          disable:
+                            default: false
+                            description: |-
+                              Disable indicates whether the ScalingAdapter should be disabled for this service.
+                              When false (default), a DGDSA is created and owns the replicas field.
+                              When true, no DGDSA is created and replicas can be modified directly in the DGD.
+                            type: boolean
+                        type: object
                       serviceName:
                         description: The name of the component
                         type: string

deploy/cloud/operator/config/crd/bases/nvidia.com_dynamographdeploymentscalingadapters.yaml

+# SPDX-FileCopyrightText: Copyright (c) 2024-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# 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.
+
+---
+apiVersion: apiextensions.k8s.io/v1
+kind: CustomResourceDefinition
+metadata:
+  annotations:
+    controller-gen.kubebuilder.io/version: v0.16.4
+    helm.sh/resource-policy: keep
+  name: dynamographdeploymentscalingadapters.nvidia.com
+spec:
+  group: nvidia.com
+  names:
+    kind: DynamoGraphDeploymentScalingAdapter
+    listKind: DynamoGraphDeploymentScalingAdapterList
+    plural: dynamographdeploymentscalingadapters
+    shortNames:
+      - dgdsa
+    singular: dynamographdeploymentscalingadapter
+  scope: Namespaced
+  versions:
+    - additionalPrinterColumns:
+        - description: DynamoGraphDeployment name
+          jsonPath: .spec.dgdRef.name
+          name: DGD
+          type: string
+        - description: Service name
+          jsonPath: .spec.dgdRef.serviceName
+          name: SERVICE
+          type: string
+        - description: Current replicas
+          jsonPath: .status.replicas
+          name: REPLICAS
+          type: integer
+        - jsonPath: .metadata.creationTimestamp
+          name: AGE
+          type: date
+      name: v1alpha1
+      schema:
+        openAPIV3Schema:
+          description: |-
+            DynamoGraphDeploymentScalingAdapter provides a scaling interface for individual services
+            within a DynamoGraphDeployment. It implements the Kubernetes scale
+            subresource, enabling integration with HPA, KEDA, and custom autoscalers.
+
+            The adapter acts as an intermediary between autoscalers and the DGD,
+            ensuring that only the adapter controller modifies the DGD's service replicas.
+            This prevents conflicts when multiple autoscaling mechanisms are in play.
+          properties:
+            apiVersion:
+              description: |-
+                APIVersion defines the versioned schema of this representation of an object.
+                Servers should convert recognized schemas to the latest internal value, and
+                may reject unrecognized values.
+                More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources
+              type: string
+            kind:
+              description: |-
+                Kind is a string value representing the REST resource this object represents.
+                Servers may infer this from the endpoint the client submits requests to.
+                Cannot be updated.
+                In CamelCase.
+                More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds
+              type: string
+            metadata:
+              type: object
+            spec:
+              description: DynamoGraphDeploymentScalingAdapterSpec defines the desired state of DynamoGraphDeploymentScalingAdapter
+              properties:
+                dgdRef:
+                  description: DGDRef references the DynamoGraphDeployment and the specific service to scale.
+                  properties:
+                    name:
+                      description: Name of the DynamoGraphDeployment
+                      minLength: 1
+                      type: string
+                    serviceName:
+                      description: ServiceName is the key name of the service within the DGD's spec.services map to scale
+                      minLength: 1
+                      type: string
+                  required:
+                    - name
+                    - serviceName
+                  type: object
+                replicas:
+                  description: |-
+                    Replicas is the desired number of replicas for the target service.
+                    This field is modified by external autoscalers (HPA/KEDA/Planner) or manually by users.
+                  format: int32
+                  minimum: 0
+                  type: integer
+              required:
+                - dgdRef
+                - replicas
+              type: object
+            status:
+              description: DynamoGraphDeploymentScalingAdapterStatus defines the observed state of DynamoGraphDeploymentScalingAdapter
+              properties:
+                lastScaleTime:
+                  description: LastScaleTime is the last time the adapter scaled the target service.
+                  format: date-time
+                  type: string
+                replicas:
+                  description: |-
+                    Replicas is the current number of replicas for the target service.
+                    This is synced from the DGD's service replicas and is required for the scale subresource.
+                  format: int32
+                  type: integer
+                selector:
+                  description: |-
+                    Selector is a label selector string for the pods managed by this adapter.
+                    Required for HPA compatibility via the scale subresource.
+                  type: string
+              type: object
+          type: object
+      served: true
+      storage: true
+      subresources:
+        scale:
+          labelSelectorPath: .status.selector
+          specReplicasPath: .spec.replicas
+          statusReplicasPath: .status.replicas
+        status: {}

deploy/cloud/operator/config/rbac/role.yaml

@@ -182,6 +182,7 @@ rules:
   - dynamocomponentdeployments
   - dynamographdeploymentrequests
   - dynamographdeployments
+  - dynamographdeploymentscalingadapters
   - dynamomodels
   verbs:
   - create
@@ -206,6 +207,7 @@ rules:
   - dynamocomponentdeployments/status
   - dynamographdeploymentrequests/status
   - dynamographdeployments/status
+  - dynamographdeploymentscalingadapters/status
   - dynamomodels/status
   verbs:
   - get

deploy/cloud/operator/internal/consts/consts.go

@@ -7,8 +7,6 @@ import (
 )
 
 const (
-	HPACPUDefaultAverageUtilization = 80
-
 	DefaultUserId = "default"
 	DefaultOrgId  = "default"
 

deploy/cloud/operator/internal/controller/common.go

@@ -53,3 +53,43 @@ type dockerSecretRetriever interface {
 	// returns a list of secret names associated with the docker registry
 	GetSecrets(namespace, registry string) ([]string, error)
 }
+
+// getServiceKeys returns the keys of the services map for logging purposes
+func getServiceKeys(services map[string]*v1alpha1.DynamoComponentDeploymentSharedSpec) []string {
+	keys := make([]string, 0, len(services))
+	for k := range services {
+		keys = append(keys, k)
+	}
+	return keys
+}
+
+// servicesEqual compares two services maps to detect changes in replica counts
+func servicesEqual(old, new map[string]*v1alpha1.DynamoComponentDeploymentSharedSpec) bool {
+	if len(old) != len(new) {
+		return false
+	}
+
+	for key, oldSvc := range old {
+		newSvc, exists := new[key]
+		if !exists {
+			return false
+		}
+
+		// Compare replicas
+		oldReplicas := int32(1)
+		if oldSvc.Replicas != nil {
+			oldReplicas = *oldSvc.Replicas
+		}
+
+		newReplicas := int32(1)
+		if newSvc.Replicas != nil {
+			newReplicas = *newSvc.Replicas
+		}
+
+		if oldReplicas != newReplicas {
+			return false
+		}
+	}
+
+	return true
+}

deploy/cloud/operator/internal/controller/dynamocomponentdeployment_controller.go

@@ -338,21 +338,6 @@ func (r *DynamoComponentDeploymentReconciler) Reconcile(ctx context.Context, req
 		}
 
 		deployment = obj
-
-		// create or update api-server hpa
-		modified_, _, err = commonController.SyncResource(ctx, r, dynamoComponentDeployment, func(ctx context.Context) (*autoscalingv2.HorizontalPodAutoscaler, bool, error) {
-			return r.generateHPA(generateResourceOption{
-				dynamoComponentDeployment: dynamoComponentDeployment,
-			})
-		})
-		if err != nil {
-			return ctrl.Result{}, err
-		}
-
-		if modified_ {
-			modified = true
-		}
-
 	}
 
 	// create or update api-server service
@@ -1114,63 +1099,6 @@ type generateResourceOption struct {
 	instanceID                              *int
 }
 
-func (r *DynamoComponentDeploymentReconciler) generateHPA(opt generateResourceOption) (*autoscalingv2.HorizontalPodAutoscaler, bool, error) {
-	labels := r.getKubeLabels(opt.dynamoComponentDeployment)
-
-	annotations := r.getKubeAnnotations(opt.dynamoComponentDeployment)
-
-	kubeName := r.getKubeName(opt.dynamoComponentDeployment, false)
-
-	kubeNs := opt.dynamoComponentDeployment.Namespace
-
-	hpaConf := opt.dynamoComponentDeployment.Spec.Autoscaling
-
-	kubeHpa := &autoscalingv2.HorizontalPodAutoscaler{
-		ObjectMeta: metav1.ObjectMeta{
-			Name:        kubeName,
-			Namespace:   kubeNs,
-			Labels:      labels,
-			Annotations: annotations,
-		},
-	}
-
-	if hpaConf == nil || !hpaConf.Enabled {
-		// if hpa is not enabled, we need to delete the hpa
-		return kubeHpa, true, nil
-	}
-
-	minReplica := int32(hpaConf.MinReplicas)
-
-	kubeHpa.Spec = autoscalingv2.HorizontalPodAutoscalerSpec{
-		MinReplicas: &minReplica,
-		MaxReplicas: int32(hpaConf.MaxReplicas),
-		ScaleTargetRef: autoscalingv2.CrossVersionObjectReference{
-			APIVersion: "apps/v1",
-			Kind:       "Deployment",
-			Name:       kubeName,
-		},
-		Metrics: hpaConf.Metrics,
-	}
-
-	if len(kubeHpa.Spec.Metrics) == 0 {
-		averageUtilization := int32(commonconsts.HPACPUDefaultAverageUtilization)
-		kubeHpa.Spec.Metrics = []autoscalingv2.MetricSpec{
-			{
-				Type: autoscalingv2.ResourceMetricSourceType,
-				Resource: &autoscalingv2.ResourceMetricSource{
-					Name: corev1.ResourceCPU,
-					Target: autoscalingv2.MetricTarget{
-						Type:               autoscalingv2.UtilizationMetricType,
-						AverageUtilization: &averageUtilization,
-					},
-				},
-			},
-		}
-	}
-
-	return kubeHpa, false, nil
-}
-
 //nolint:gocyclo,nakedret
 func (r *DynamoComponentDeploymentReconciler) generatePodTemplateSpec(ctx context.Context, opt generateResourceOption, role dynamo.Role) (podTemplateSpec *corev1.PodTemplateSpec, err error) {
 	podLabels := r.getKubeLabels(opt.dynamoComponentDeployment)

deploy/cloud/operator/internal/controller/dynamographdeployment_controller.go

@@ -86,6 +86,7 @@ type DynamoGraphDeploymentReconciler struct {
 // +kubebuilder:rbac:groups=nvidia.com,resources=dynamographdeployments,verbs=get;list;watch;create;update;patch;delete
 // +kubebuilder:rbac:groups=nvidia.com,resources=dynamographdeployments/status,verbs=get;update;patch
 // +kubebuilder:rbac:groups=nvidia.com,resources=dynamographdeployments/finalizers,verbs=update
+// +kubebuilder:rbac:groups=nvidia.com,resources=dynamographdeploymentscalingadapters,verbs=get;list;watch;create;update;patch;delete
 // +kubebuilder:rbac:groups=grove.io,resources=podcliquesets,verbs=get;list;watch;create;update;patch;delete
 // +kubebuilder:rbac:groups=grove.io,resources=podcliques/scale,verbs=get;update;patch
 // +kubebuilder:rbac:groups=grove.io,resources=podcliquescalinggroups/scale,verbs=get;update;patch
@@ -225,6 +226,13 @@ func (r *DynamoGraphDeploymentReconciler) reconcileResources(ctx context.Context
 		return "", "", "", fmt.Errorf("failed to reconcile top-level PVCs: %w", err)
 	}
 
+	// Reconcile DynamoGraphDeploymentScalingAdapters for each service
+	err = r.reconcileScalingAdapters(ctx, dynamoDeployment)
+	if err != nil {
+		logger.Error(err, "Failed to reconcile scaling adapters")
+		return "", "", "", fmt.Errorf("failed to reconcile scaling adapters: %w", err)
+	}
+
 	// Reconcile the SA, Role and RoleBinding if k8s discovery is enabled
 	err = r.reconcileK8sDiscoveryResources(ctx, dynamoDeployment)
 	if err != nil {
@@ -607,6 +615,89 @@ func (r *DynamoGraphDeploymentReconciler) reconcilePVCs(ctx context.Context, dyn
 	return nil
 }
 
+// reconcileScalingAdapters ensures a DynamoGraphDeploymentScalingAdapter exists for each service in the DGD
+// that has scaling adapter enabled (default). Services with scalingAdapter.disable=true will not have a DGDSA.
+// This enables pluggable autoscaling via HPA, KEDA, or Planner.
+func (r *DynamoGraphDeploymentReconciler) reconcileScalingAdapters(ctx context.Context, dynamoDeployment *nvidiacomv1alpha1.DynamoGraphDeployment) error {
+	logger := log.FromContext(ctx)
+
+	// Process each service - SyncResource handles create, update, and delete via toDelete flag
+	for serviceName, component := range dynamoDeployment.Spec.Services {
+		// Check if scaling adapter is disabled for this service
+		scalingAdapterDisabled := component.ScalingAdapter != nil && component.ScalingAdapter.Disable
+
+		// Get current replicas (default to 1 if not set)
+		currentReplicas := int32(1)
+		if component.Replicas != nil {
+			currentReplicas = *component.Replicas
+		}
+
+		// Use SyncResource to handle creation/updates/deletion
+		// When toDelete=true, SyncResource will delete the existing resource if it exists
+		_, _, err := commonController.SyncResource(ctx, r, dynamoDeployment, func(ctx context.Context) (*nvidiacomv1alpha1.DynamoGraphDeploymentScalingAdapter, bool, error) {
+			adapterName := generateAdapterName(dynamoDeployment.Name, serviceName)
+			adapter := &nvidiacomv1alpha1.DynamoGraphDeploymentScalingAdapter{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      adapterName,
+					Namespace: dynamoDeployment.Namespace,
+					Labels: map[string]string{
+						consts.KubeLabelDynamoGraphDeploymentName: dynamoDeployment.Name,
+						consts.KubeLabelDynamoComponent:           serviceName,
+					},
+				},
+				Spec: nvidiacomv1alpha1.DynamoGraphDeploymentScalingAdapterSpec{
+					Replicas: currentReplicas,
+					DGDRef: nvidiacomv1alpha1.DynamoGraphDeploymentServiceRef{
+						Name:        dynamoDeployment.Name,
+						ServiceName: serviceName,
+					},
+				},
+			}
+			// Return toDelete=true if scaling adapter is disabled
+			return adapter, scalingAdapterDisabled, nil
+		})
+
+		if err != nil {
+			logger.Error(err, "Failed to sync DynamoGraphDeploymentScalingAdapter", "service", serviceName)
+			return err
+		}
+	}
+
+	// Clean up adapters for services that were removed from DGD entirely
+	adapterList := &nvidiacomv1alpha1.DynamoGraphDeploymentScalingAdapterList{}
+	if err := r.List(ctx, adapterList,
+		client.InNamespace(dynamoDeployment.Namespace),
+		client.MatchingLabels{consts.KubeLabelDynamoGraphDeploymentName: dynamoDeployment.Name},
+	); err != nil {
+		logger.Error(err, "Failed to list DynamoGraphDeploymentScalingAdapters")
+		return err
+	}
+
+	for i := range adapterList.Items {
+		adapter := &adapterList.Items[i]
+		serviceName := adapter.Spec.DGDRef.ServiceName
+
+		// Delete adapter if service no longer exists in DGD
+		if _, exists := dynamoDeployment.Spec.Services[serviceName]; !exists {
+			logger.Info("Deleting orphaned DynamoGraphDeploymentScalingAdapter", "adapter", adapter.Name, "service", serviceName)
+			if err := r.Delete(ctx, adapter); err != nil && !errors.IsNotFound(err) {
+				logger.Error(err, "Failed to delete orphaned adapter", "adapter", adapter.Name)
+				return err
+			}
+			r.Recorder.Eventf(dynamoDeployment, corev1.EventTypeNormal, "AdapterDeleted",
+				"Deleted orphaned scaling adapter %s for removed service %s", adapter.Name, serviceName)
+		}
+	}
+
+	return nil
+}
+
+// generateAdapterName creates a consistent name for a DynamoGraphDeploymentScalingAdapter
+// Service names are lowercased to comply with Kubernetes DNS subdomain naming requirements
+func generateAdapterName(dgdName, serviceName string) string {
+	return fmt.Sprintf("%s-%s", dgdName, strings.ToLower(serviceName))
+}
+
 func (r *DynamoGraphDeploymentReconciler) FinalizeResource(ctx context.Context, dynamoDeployment *nvidiacomv1alpha1.DynamoGraphDeployment) error {
 	// for now doing nothing
 	return nil
@@ -626,6 +717,13 @@ func (r *DynamoGraphDeploymentReconciler) SetupWithManager(mgr ctrl.Manager) err
 			UpdateFunc:  func(de event.UpdateEvent) bool { return true },
 			GenericFunc: func(ge event.GenericEvent) bool { return true },
 		})).
+		Owns(&nvidiacomv1alpha1.DynamoGraphDeploymentScalingAdapter{}, builder.WithPredicates(predicate.Funcs{
+			// ignore creation cause we don't want to be called again after we create the adapter
+			CreateFunc:  func(ce event.CreateEvent) bool { return false },
+			DeleteFunc:  func(de event.DeleteEvent) bool { return true },
+			UpdateFunc:  func(de event.UpdateEvent) bool { return false }, // Adapter updates are handled by adapter controller
+			GenericFunc: func(ge event.GenericEvent) bool { return false },
+		})).
 		Owns(&corev1.PersistentVolumeClaim{}, builder.WithPredicates(predicate.Funcs{
 			// ignore creation cause we don't want to be called again after we create the PVC
 			CreateFunc:  func(ce event.CreateEvent) bool { return false },

deploy/cloud/operator/internal/controller/dynamographdeployment_controller_test.go

+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * 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.
+ */
+
+package controller
+
+import (
+	"context"
+	"testing"
+
+	"github.com/ai-dynamo/dynamo/deploy/cloud/operator/api/v1alpha1"
+	"github.com/ai-dynamo/dynamo/deploy/cloud/operator/internal/consts"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/tools/record"
+	"k8s.io/utils/ptr"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+)
+
+func TestDynamoGraphDeploymentReconciler_reconcileScalingAdapters(t *testing.T) {
+	// Register custom types with the scheme
+	if err := v1alpha1.AddToScheme(scheme.Scheme); err != nil {
+		t.Fatalf("Failed to add v1alpha1 to scheme: %v", err)
+	}
+
+	tests := []struct {
+		name                 string
+		dgd                  *v1alpha1.DynamoGraphDeployment
+		existingAdapters     []v1alpha1.DynamoGraphDeploymentScalingAdapter
+		expectedAdapterCount int
+		expectedAdapters     map[string]int32 // map of adapter name to expected replicas
+		expectDeleted        []string         // adapter names that should be deleted
+	}{
+		{
+			name: "creates adapters for all services",
+			dgd: &v1alpha1.DynamoGraphDeployment{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-dgd",
+					Namespace: "default",
+				},
+				Spec: v1alpha1.DynamoGraphDeploymentSpec{
+					Services: map[string]*v1alpha1.DynamoComponentDeploymentSharedSpec{
+						"Frontend": {
+							Replicas: ptr.To(int32(2)),
+						},
+						"decode": {
+							Replicas: ptr.To(int32(3)),
+						},
+					},
+				},
+			},
+			expectedAdapterCount: 2,
+			expectedAdapters: map[string]int32{
+				"test-dgd-frontend": 2,
+				"test-dgd-decode":   3,
+			},
+		},
+		{
+			name: "uses default replicas when not specified",
+			dgd: &v1alpha1.DynamoGraphDeployment{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-dgd",
+					Namespace: "default",
+				},
+				Spec: v1alpha1.DynamoGraphDeploymentSpec{
+					Services: map[string]*v1alpha1.DynamoComponentDeploymentSharedSpec{
+						"worker": {},
+					},
+				},
+			},
+			expectedAdapterCount: 1,
+			expectedAdapters: map[string]int32{
+				"test-dgd-worker": 1, // default replicas
+			},
+		},
+		{
+			name: "skips adapter creation when disabled",
+			dgd: &v1alpha1.DynamoGraphDeployment{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-dgd",
+					Namespace: "default",
+				},
+				Spec: v1alpha1.DynamoGraphDeploymentSpec{
+					Services: map[string]*v1alpha1.DynamoComponentDeploymentSharedSpec{
+						"Frontend": {
+							Replicas: ptr.To(int32(2)),
+						},
+						"decode": {
+							Replicas: ptr.To(int32(3)),
+							ScalingAdapter: &v1alpha1.ScalingAdapter{
+								Disable: true,
+							},
+						},
+					},
+				},
+			},
+			expectedAdapterCount: 1,
+			expectedAdapters: map[string]int32{
+				"test-dgd-frontend": 2,
+			},
+		},
+		{
+			name: "deletes adapter when service is removed",
+			dgd: &v1alpha1.DynamoGraphDeployment{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-dgd",
+					Namespace: "default",
+					UID:       "test-uid",
+				},
+				Spec: v1alpha1.DynamoGraphDeploymentSpec{
+					Services: map[string]*v1alpha1.DynamoComponentDeploymentSharedSpec{
+						"Frontend": {
+							Replicas: ptr.To(int32(2)),
+						},
+					},
+				},
+			},
+			existingAdapters: []v1alpha1.DynamoGraphDeploymentScalingAdapter{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "test-dgd-frontend",
+						Namespace: "default",
+						Labels: map[string]string{
+							consts.KubeLabelDynamoGraphDeploymentName: "test-dgd",
+						},
+						OwnerReferences: []metav1.OwnerReference{
+							{
+								APIVersion: "nvidia.com/v1alpha1",
+								Kind:       "DynamoGraphDeployment",
+								Name:       "test-dgd",
+								UID:        "test-uid",
+							},
+						},
+					},
+					Spec: v1alpha1.DynamoGraphDeploymentScalingAdapterSpec{
+						Replicas: 2,
+						DGDRef: v1alpha1.DynamoGraphDeploymentServiceRef{
+							Name:        "test-dgd",
+							ServiceName: "Frontend",
+						},
+					},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "test-dgd-removed",
+						Namespace: "default",
+						Labels: map[string]string{
+							consts.KubeLabelDynamoGraphDeploymentName: "test-dgd",
+						},
+						OwnerReferences: []metav1.OwnerReference{
+							{
+								APIVersion: "nvidia.com/v1alpha1",
+								Kind:       "DynamoGraphDeployment",
+								Name:       "test-dgd",
+								UID:        "test-uid",
+							},
+						},
+					},
+					Spec: v1alpha1.DynamoGraphDeploymentScalingAdapterSpec{
+						Replicas: 1,
+						DGDRef: v1alpha1.DynamoGraphDeploymentServiceRef{
+							Name:        "test-dgd",
+							ServiceName: "removed",
+						},
+					},
+				},
+			},
+			expectedAdapterCount: 1,
+			expectedAdapters: map[string]int32{
+				"test-dgd-frontend": 2,
+			},
+			expectDeleted: []string{"test-dgd-removed"},
+		},
+		{
+			name: "deletes adapter when scalingAdapter.disable is set to true",
+			dgd: &v1alpha1.DynamoGraphDeployment{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-dgd",
+					Namespace: "default",
+					UID:       "test-uid",
+				},
+				Spec: v1alpha1.DynamoGraphDeploymentSpec{
+					Services: map[string]*v1alpha1.DynamoComponentDeploymentSharedSpec{
+						"Frontend": {
+							Replicas: ptr.To(int32(2)),
+							ScalingAdapter: &v1alpha1.ScalingAdapter{
+								Disable: true,
+							},
+						},
+					},
+				},
+			},
+			existingAdapters: []v1alpha1.DynamoGraphDeploymentScalingAdapter{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "test-dgd-frontend",
+						Namespace: "default",
+						Labels: map[string]string{
+							consts.KubeLabelDynamoGraphDeploymentName: "test-dgd",
+						},
+						OwnerReferences: []metav1.OwnerReference{
+							{
+								APIVersion: "nvidia.com/v1alpha1",
+								Kind:       "DynamoGraphDeployment",
+								Name:       "test-dgd",
+								UID:        "test-uid",
+							},
+						},
+					},
+					Spec: v1alpha1.DynamoGraphDeploymentScalingAdapterSpec{
+						Replicas: 2,
+						DGDRef: v1alpha1.DynamoGraphDeploymentServiceRef{
+							Name:        "test-dgd",
+							ServiceName: "Frontend",
+						},
+					},
+				},
+			},
+			expectedAdapterCount: 0,
+			expectedAdapters:     map[string]int32{},
+			expectDeleted:        []string{"test-dgd-frontend"},
+		},
+		{
+			name: "adapter name uses lowercase service name",
+			dgd: &v1alpha1.DynamoGraphDeployment{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "my-dgd",
+					Namespace: "default",
+				},
+				Spec: v1alpha1.DynamoGraphDeploymentSpec{
+					Services: map[string]*v1alpha1.DynamoComponentDeploymentSharedSpec{
+						"MyService": {
+							Replicas: ptr.To(int32(1)),
+						},
+					},
+				},
+			},
+			expectedAdapterCount: 1,
+			expectedAdapters: map[string]int32{
+				"my-dgd-myservice": 1, // lowercase
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Build initial objects
+			var initObjs []client.Object
+			initObjs = append(initObjs, tt.dgd)
+			for i := range tt.existingAdapters {
+				initObjs = append(initObjs, &tt.existingAdapters[i])
+			}
+
+			// Create fake client
+			fakeClient := fake.NewClientBuilder().
+				WithScheme(scheme.Scheme).
+				WithObjects(initObjs...).
+				Build()
+
+			// Create reconciler
+			r := &DynamoGraphDeploymentReconciler{
+				Client:   fakeClient,
+				Recorder: record.NewFakeRecorder(10),
+			}
+
+			// Run reconcileScalingAdapters
+			ctx := context.Background()
+			err := r.reconcileScalingAdapters(ctx, tt.dgd)
+			if err != nil {
+				t.Fatalf("reconcileScalingAdapters() error = %v", err)
+			}
+
+			// Verify adapters
+			adapterList := &v1alpha1.DynamoGraphDeploymentScalingAdapterList{}
+			if err := fakeClient.List(ctx, adapterList, client.InNamespace("default")); err != nil {
+				t.Fatalf("Failed to list adapters: %v", err)
+			}
+
+			if len(adapterList.Items) != tt.expectedAdapterCount {
+				t.Errorf("Expected %d adapters, got %d", tt.expectedAdapterCount, len(adapterList.Items))
+			}
+
+			// Check expected adapters exist with correct replicas
+			for name, expectedReplicas := range tt.expectedAdapters {
+				adapter := &v1alpha1.DynamoGraphDeploymentScalingAdapter{}
+				err := fakeClient.Get(ctx, types.NamespacedName{Name: name, Namespace: "default"}, adapter)
+				if err != nil {
+					t.Errorf("Expected adapter %s to exist, but got error: %v", name, err)
+					continue
+				}
+				if adapter.Spec.Replicas != expectedReplicas {
+					t.Errorf("Adapter %s has replicas=%d, expected %d", name, adapter.Spec.Replicas, expectedReplicas)
+				}
+			}
+
+			// Check that deleted adapters don't exist
+			for _, name := range tt.expectDeleted {
+				adapter := &v1alpha1.DynamoGraphDeploymentScalingAdapter{}
+				err := fakeClient.Get(ctx, types.NamespacedName{Name: name, Namespace: "default"}, adapter)
+				if err == nil {
+					t.Errorf("Expected adapter %s to be deleted, but it still exists", name)
+				}
+			}
+		})
+	}
+}

deploy/cloud/operator/internal/controller/dynamographdeploymentscalingadapter_controller.go

+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * 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.
+ */
+
+package controller
+
+import (
+	"context"
+	"fmt"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/tools/record"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/builder"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+	"sigs.k8s.io/controller-runtime/pkg/handler"
+	"sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/predicate"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+
+	nvidiacomv1alpha1 "github.com/ai-dynamo/dynamo/deploy/cloud/operator/api/v1alpha1"
+	"github.com/ai-dynamo/dynamo/deploy/cloud/operator/internal/consts"
+	commonController "github.com/ai-dynamo/dynamo/deploy/cloud/operator/internal/controller_common"
+)
+
+// DynamoGraphDeploymentScalingAdapterReconciler reconciles a DynamoGraphDeploymentScalingAdapter object
+type DynamoGraphDeploymentScalingAdapterReconciler struct {
+	client.Client
+	Scheme   *runtime.Scheme
+	Recorder record.EventRecorder
+	Config   commonController.Config
+}
+
+// +kubebuilder:rbac:groups=nvidia.com,resources=dynamographdeploymentscalingadapters,verbs=get;list;watch;create;update;patch;delete
+// +kubebuilder:rbac:groups=nvidia.com,resources=dynamographdeploymentscalingadapters/status,verbs=get;update;patch
+// +kubebuilder:rbac:groups=nvidia.com,resources=dynamographdeployments,verbs=get;list;watch;update;patch
+
+// Reconcile implements the reconciliation loop for DynamoGraphDeploymentScalingAdapter
+func (r *DynamoGraphDeploymentScalingAdapterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+	logger := log.FromContext(ctx)
+
+	// 1. Fetch the DynamoGraphDeploymentScalingAdapter
+	adapter := &nvidiacomv1alpha1.DynamoGraphDeploymentScalingAdapter{}
+	if err := r.Get(ctx, req.NamespacedName, adapter); err != nil {
+		return ctrl.Result{}, client.IgnoreNotFound(err)
+	}
+
+	// Skip reconciliation if being deleted
+	if !adapter.GetDeletionTimestamp().IsZero() {
+		logger.V(1).Info("Adapter is being deleted, skipping reconciliation")
+		return ctrl.Result{}, nil
+	}
+
+	// 2. Fetch the referenced DGD
+	dgd := &nvidiacomv1alpha1.DynamoGraphDeployment{}
+	dgdKey := types.NamespacedName{
+		Name:      adapter.Spec.DGDRef.Name,
+		Namespace: adapter.Namespace,
+	}
+	if err := r.Get(ctx, dgdKey, dgd); err != nil {
+		if errors.IsNotFound(err) {
+			logger.Error(err, "Referenced DGD not found", "dgd", dgdKey)
+			// DGD doesn't exist, can't proceed
+			return ctrl.Result{}, err
+		}
+		return ctrl.Result{}, err
+	}
+
+	// 3. Find the target service in DGD's spec.services map
+	component, exists := dgd.Spec.Services[adapter.Spec.DGDRef.ServiceName]
+	if !exists || component == nil {
+		logger.Error(nil, "Service not found in DGD",
+			"service", adapter.Spec.DGDRef.ServiceName,
+			"dgd", dgd.Name,
+			"availableServices", getServiceKeys(dgd.Spec.Services))
+		return ctrl.Result{}, fmt.Errorf("service %s not found in DGD", adapter.Spec.DGDRef.ServiceName)
+	}
+
+	// Get current replicas from DGD (default to 1 if not set)
+	currentReplicas := int32(1)
+	if component.Replicas != nil {
+		currentReplicas = *component.Replicas
+	}
+
+	// 4. Update DGD if replicas changed (DGDSA is the source of truth)
+	if currentReplicas != adapter.Spec.Replicas {
+		// Update the service's replicas in DGD
+		component.Replicas = &adapter.Spec.Replicas
+		dgd.Spec.Services[adapter.Spec.DGDRef.ServiceName] = component
+
+		if err := r.Update(ctx, dgd); err != nil {
+			logger.Error(err, "Failed to update DGD")
+			r.Recorder.Eventf(adapter, corev1.EventTypeWarning, "UpdateFailed",
+				"Failed to update DGD %s: %v", dgd.Name, err)
+			return ctrl.Result{}, err
+		}
+
+		logger.Info("Scaled service",
+			"dgd", dgd.Name,
+			"service", adapter.Spec.DGDRef.ServiceName,
+			"from", currentReplicas,
+			"to", adapter.Spec.Replicas)
+
+		r.Recorder.Eventf(adapter, corev1.EventTypeNormal, "Scaled",
+			"Scaled service %s from %d to %d replicas", adapter.Spec.DGDRef.ServiceName, currentReplicas, adapter.Spec.Replicas)
+
+		// Record scaling event
+		now := metav1.Now()
+		adapter.Status.LastScaleTime = &now
+	}
+
+	// 5. Update adapter status
+	adapter.Status.Replicas = adapter.Spec.Replicas
+	adapter.Status.Selector = r.buildPodSelector(dgd, adapter.Spec.DGDRef.ServiceName)
+
+	if err := r.Status().Update(ctx, adapter); err != nil {
+		logger.Error(err, "Failed to update adapter status")
+		return ctrl.Result{}, err
+	}
+
+	return ctrl.Result{}, nil
+}
+
+// buildPodSelector constructs a label selector for the pods managed by this service
+func (r *DynamoGraphDeploymentScalingAdapterReconciler) buildPodSelector(dgd *nvidiacomv1alpha1.DynamoGraphDeployment, serviceName string) string {
+	// Pods are labeled with:
+	// - nvidia.com/dynamo-graph-deployment-name = dgd.Name
+	// - nvidia.com/dynamo-component = serviceName (the key from spec.services map)
+	return fmt.Sprintf("%s=%s,%s=%s",
+		consts.KubeLabelDynamoGraphDeploymentName, dgd.Name,
+		consts.KubeLabelDynamoComponent, serviceName)
+}
+
+// SetupWithManager sets up the controller with the Manager
+func (r *DynamoGraphDeploymentScalingAdapterReconciler) SetupWithManager(mgr ctrl.Manager) error {
+	return ctrl.NewControllerManagedBy(mgr).
+		For(&nvidiacomv1alpha1.DynamoGraphDeploymentScalingAdapter{}, builder.WithPredicates(
+			predicate.GenerationChangedPredicate{},
+		)).
+		Named("dgdscalingadapter").
+		// Watch DGDs to sync status when DGD service replicas change
+		Watches(
+			&nvidiacomv1alpha1.DynamoGraphDeployment{},
+			handler.EnqueueRequestsFromMapFunc(r.findAdaptersForDGD),
+			builder.WithPredicates(predicate.Funcs{
+				CreateFunc: func(ce event.CreateEvent) bool { return false },
+				DeleteFunc: func(de event.DeleteEvent) bool { return true },
+				UpdateFunc: func(ue event.UpdateEvent) bool {
+					// Only trigger on spec changes (not status)
+					oldDGD, okOld := ue.ObjectOld.(*nvidiacomv1alpha1.DynamoGraphDeployment)
+					newDGD, okNew := ue.ObjectNew.(*nvidiacomv1alpha1.DynamoGraphDeployment)
+					if !okOld || !okNew {
+						return false
+					}
+					// Trigger if services map changed
+					return !servicesEqual(oldDGD.Spec.Services, newDGD.Spec.Services)
+				},
+				GenericFunc: func(ge event.GenericEvent) bool { return false },
+			}),
+		).
+		WithEventFilter(commonController.EphemeralDeploymentEventFilter(r.Config)).
+		Complete(r)
+}
+
+// findAdaptersForDGD maps DGD changes to adapter reconcile requests
+// Uses label selector to efficiently query only adapters for this specific DGD
+func (r *DynamoGraphDeploymentScalingAdapterReconciler) findAdaptersForDGD(ctx context.Context, obj client.Object) []reconcile.Request {
+	dgd, ok := obj.(*nvidiacomv1alpha1.DynamoGraphDeployment)
+	if !ok {
+		return nil
+	}
+
+	// Use label selector to filter at API level (more efficient than in-memory filtering)
+	adapterList := &nvidiacomv1alpha1.DynamoGraphDeploymentScalingAdapterList{}
+	if err := r.List(ctx, adapterList,
+		client.InNamespace(dgd.Namespace),
+		client.MatchingLabels{consts.KubeLabelDynamoGraphDeploymentName: dgd.Name},
+	); err != nil {
+		log.FromContext(ctx).Error(err, "Failed to list adapters for DGD", "dgd", dgd.Name)
+		return nil
+	}
+
+	// All returned adapters are guaranteed to belong to this DGD
+	requests := make([]reconcile.Request, 0, len(adapterList.Items))
+	for i := range adapterList.Items {
+		requests = append(requests, reconcile.Request{
+			NamespacedName: types.NamespacedName{
+				Name:      adapterList.Items[i].Name,
+				Namespace: adapterList.Items[i].Namespace,
+			},
+		})
+	}
+
+	return requests
+}

deploy/cloud/operator/internal/controller/dynamographdeploymentscalingadapter_controller_test.go

+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * 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.
+ */
+
+package controller
+
+import (
+	"context"
+	"testing"
+
+	"github.com/ai-dynamo/dynamo/deploy/cloud/operator/api/v1alpha1"
+	"github.com/ai-dynamo/dynamo/deploy/cloud/operator/internal/consts"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/tools/record"
+	"k8s.io/utils/ptr"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+)
+
+func TestDynamoGraphDeploymentScalingAdapterReconciler_Reconcile(t *testing.T) {
+	// Register custom types with the scheme
+	if err := v1alpha1.AddToScheme(scheme.Scheme); err != nil {
+		t.Fatalf("Failed to add v1alpha1 to scheme: %v", err)
+	}
+
+	tests := []struct {
+		name                   string
+		adapter                *v1alpha1.DynamoGraphDeploymentScalingAdapter
+		dgd                    *v1alpha1.DynamoGraphDeployment
+		expectedDGDReplicas    int32
+		expectedStatusReplicas int32
+		expectError            bool
+		expectRequeue          bool
+	}{
+		{
+			name: "updates DGD replicas when DGDSA spec differs",
+			adapter: &v1alpha1.DynamoGraphDeploymentScalingAdapter{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-dgd-frontend",
+					Namespace: "default",
+				},
+				Spec: v1alpha1.DynamoGraphDeploymentScalingAdapterSpec{
+					Replicas: 5,
+					DGDRef: v1alpha1.DynamoGraphDeploymentServiceRef{
+						Name:        "test-dgd",
+						ServiceName: "Frontend",
+					},
+				},
+			},
+			dgd: &v1alpha1.DynamoGraphDeployment{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-dgd",
+					Namespace: "default",
+				},
+				Spec: v1alpha1.DynamoGraphDeploymentSpec{
+					Services: map[string]*v1alpha1.DynamoComponentDeploymentSharedSpec{
+						"Frontend": {
+							Replicas: ptr.To(int32(2)),
+						},
+					},
+				},
+			},
+			expectedDGDReplicas:    5,
+			expectedStatusReplicas: 5,
+			expectError:            false,
+		},
+		{
+			name: "no update when replicas already match",
+			adapter: &v1alpha1.DynamoGraphDeploymentScalingAdapter{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-dgd-frontend",
+					Namespace: "default",
+				},
+				Spec: v1alpha1.DynamoGraphDeploymentScalingAdapterSpec{
+					Replicas: 3,
+					DGDRef: v1alpha1.DynamoGraphDeploymentServiceRef{
+						Name:        "test-dgd",
+						ServiceName: "Frontend",
+					},
+				},
+			},
+			dgd: &v1alpha1.DynamoGraphDeployment{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-dgd",
+					Namespace: "default",
+				},
+				Spec: v1alpha1.DynamoGraphDeploymentSpec{
+					Services: map[string]*v1alpha1.DynamoComponentDeploymentSharedSpec{
+						"Frontend": {
+							Replicas: ptr.To(int32(3)),
+						},
+					},
+				},
+			},
+			expectedDGDReplicas:    3,
+			expectedStatusReplicas: 3,
+			expectError:            false,
+		},
+		{
+			name: "uses default replicas (1) when DGD service has no replicas set",
+			adapter: &v1alpha1.DynamoGraphDeploymentScalingAdapter{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-dgd-worker",
+					Namespace: "default",
+				},
+				Spec: v1alpha1.DynamoGraphDeploymentScalingAdapterSpec{
+					Replicas: 4,
+					DGDRef: v1alpha1.DynamoGraphDeploymentServiceRef{
+						Name:        "test-dgd",
+						ServiceName: "worker",
+					},
+				},
+			},
+			dgd: &v1alpha1.DynamoGraphDeployment{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-dgd",
+					Namespace: "default",
+				},
+				Spec: v1alpha1.DynamoGraphDeploymentSpec{
+					Services: map[string]*v1alpha1.DynamoComponentDeploymentSharedSpec{
+						"worker": {}, // no replicas set
+					},
+				},
+			},
+			expectedDGDReplicas:    4,
+			expectedStatusReplicas: 4,
+			expectError:            false,
+		},
+		{
+			name: "error when service not found in DGD",
+			adapter: &v1alpha1.DynamoGraphDeploymentScalingAdapter{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-dgd-missing",
+					Namespace: "default",
+				},
+				Spec: v1alpha1.DynamoGraphDeploymentScalingAdapterSpec{
+					Replicas: 2,
+					DGDRef: v1alpha1.DynamoGraphDeploymentServiceRef{
+						Name:        "test-dgd",
+						ServiceName: "nonexistent",
+					},
+				},
+			},
+			dgd: &v1alpha1.DynamoGraphDeployment{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-dgd",
+					Namespace: "default",
+				},
+				Spec: v1alpha1.DynamoGraphDeploymentSpec{
+					Services: map[string]*v1alpha1.DynamoComponentDeploymentSharedSpec{
+						"Frontend": {
+							Replicas: ptr.To(int32(1)),
+						},
+					},
+				},
+			},
+			expectError: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Build initial objects
+			var initObjs []client.Object
+			initObjs = append(initObjs, tt.adapter, tt.dgd)
+
+			// Create fake client with status subresource support
+			fakeClient := fake.NewClientBuilder().
+				WithScheme(scheme.Scheme).
+				WithObjects(initObjs...).
+				WithStatusSubresource(&v1alpha1.DynamoGraphDeploymentScalingAdapter{}).
+				Build()
+
+			// Create reconciler
+			r := &DynamoGraphDeploymentScalingAdapterReconciler{
+				Client:   fakeClient,
+				Scheme:   scheme.Scheme,
+				Recorder: record.NewFakeRecorder(10),
+			}
+
+			// Run Reconcile
+			ctx := context.Background()
+			req := ctrl.Request{
+				NamespacedName: types.NamespacedName{
+					Name:      tt.adapter.Name,
+					Namespace: tt.adapter.Namespace,
+				},
+			}
+
+			result, err := r.Reconcile(ctx, req)
+
+			// Check error expectation
+			if tt.expectError && err == nil {
+				t.Errorf("Expected error, but got none")
+			}
+			if !tt.expectError && err != nil {
+				t.Errorf("Unexpected error: %v", err)
+			}
+
+			// Skip further checks if error was expected
+			if tt.expectError {
+				return
+			}
+
+			// Check requeue
+			if tt.expectRequeue && result.RequeueAfter == 0 {
+				t.Errorf("Expected requeue, but got none")
+			}
+
+			// Verify DGD replicas were updated
+			updatedDGD := &v1alpha1.DynamoGraphDeployment{}
+			if err := fakeClient.Get(ctx, types.NamespacedName{Name: tt.dgd.Name, Namespace: tt.dgd.Namespace}, updatedDGD); err != nil {
+				t.Fatalf("Failed to get updated DGD: %v", err)
+			}
+
+			service, exists := updatedDGD.Spec.Services[tt.adapter.Spec.DGDRef.ServiceName]
+			if !exists {
+				t.Fatalf("Service %s not found in updated DGD", tt.adapter.Spec.DGDRef.ServiceName)
+			}
+
+			actualReplicas := int32(1)
+			if service.Replicas != nil {
+				actualReplicas = *service.Replicas
+			}
+
+			if actualReplicas != tt.expectedDGDReplicas {
+				t.Errorf("DGD service replicas = %d, expected %d", actualReplicas, tt.expectedDGDReplicas)
+			}
+
+			// Verify adapter status was updated
+			updatedAdapter := &v1alpha1.DynamoGraphDeploymentScalingAdapter{}
+			if err := fakeClient.Get(ctx, types.NamespacedName{Name: tt.adapter.Name, Namespace: tt.adapter.Namespace}, updatedAdapter); err != nil {
+				t.Fatalf("Failed to get updated adapter: %v", err)
+			}
+
+			if updatedAdapter.Status.Replicas != tt.expectedStatusReplicas {
+				t.Errorf("Adapter status.replicas = %d, expected %d", updatedAdapter.Status.Replicas, tt.expectedStatusReplicas)
+			}
+
+			// Verify selector is set
+			if updatedAdapter.Status.Selector == "" {
+				t.Errorf("Adapter status.selector is empty, expected non-empty")
+			}
+		})
+	}
+}
+
+func TestDynamoGraphDeploymentScalingAdapterReconciler_Reconcile_NotFound(t *testing.T) {
+	// Register custom types with the scheme
+	if err := v1alpha1.AddToScheme(scheme.Scheme); err != nil {
+		t.Fatalf("Failed to add v1alpha1 to scheme: %v", err)
+	}
+
+	// Create fake client with no objects
+	fakeClient := fake.NewClientBuilder().
+		WithScheme(scheme.Scheme).
+		Build()
+
+	r := &DynamoGraphDeploymentScalingAdapterReconciler{
+		Client:   fakeClient,
+		Scheme:   scheme.Scheme,
+		Recorder: record.NewFakeRecorder(10),
+	}
+
+	ctx := context.Background()
+	req := ctrl.Request{
+		NamespacedName: types.NamespacedName{
+			Name:      "nonexistent",
+			Namespace: "default",
+		},
+	}
+
+	// Should return no error when adapter not found (client.IgnoreNotFound)
+	result, err := r.Reconcile(ctx, req)
+	if err != nil {
+		t.Errorf("Expected no error for not found adapter, got: %v", err)
+	}
+	if result.RequeueAfter != 0 {
+		t.Errorf("Expected no requeueAfter for not found adapter, got: %v", result.RequeueAfter)
+	}
+}
+
+func TestDynamoGraphDeploymentScalingAdapterReconciler_Reconcile_DGDNotFound(t *testing.T) {
+	// Register custom types with the scheme
+	if err := v1alpha1.AddToScheme(scheme.Scheme); err != nil {
+		t.Fatalf("Failed to add v1alpha1 to scheme: %v", err)
+	}
+
+	adapter := &v1alpha1.DynamoGraphDeploymentScalingAdapter{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-dgd-frontend",
+			Namespace: "default",
+		},
+		Spec: v1alpha1.DynamoGraphDeploymentScalingAdapterSpec{
+			Replicas: 5,
+			DGDRef: v1alpha1.DynamoGraphDeploymentServiceRef{
+				Name:        "nonexistent-dgd",
+				ServiceName: "Frontend",
+			},
+		},
+	}
+
+	fakeClient := fake.NewClientBuilder().
+		WithScheme(scheme.Scheme).
+		WithObjects(adapter).
+		Build()
+
+	r := &DynamoGraphDeploymentScalingAdapterReconciler{
+		Client:   fakeClient,
+		Scheme:   scheme.Scheme,
+		Recorder: record.NewFakeRecorder(10),
+	}
+
+	ctx := context.Background()
+	req := ctrl.Request{
+		NamespacedName: types.NamespacedName{
+			Name:      adapter.Name,
+			Namespace: adapter.Namespace,
+		},
+	}
+
+	// Should return error when DGD not found
+	_, err := r.Reconcile(ctx, req)
+	if err == nil {
+		t.Errorf("Expected error when DGD not found, got none")
+	}
+}
+
+func TestDynamoGraphDeploymentScalingAdapterReconciler_Reconcile_BeingDeleted(t *testing.T) {
+	// Register custom types with the scheme
+	if err := v1alpha1.AddToScheme(scheme.Scheme); err != nil {
+		t.Fatalf("Failed to add v1alpha1 to scheme: %v", err)
+	}
+
+	now := metav1.Now()
+	adapter := &v1alpha1.DynamoGraphDeploymentScalingAdapter{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:              "test-dgd-frontend",
+			Namespace:         "default",
+			DeletionTimestamp: &now,
+			Finalizers:        []string{"test-finalizer"}, // Required for deletion timestamp to be set
+		},
+		Spec: v1alpha1.DynamoGraphDeploymentScalingAdapterSpec{
+			Replicas: 5,
+			DGDRef: v1alpha1.DynamoGraphDeploymentServiceRef{
+				Name:        "test-dgd",
+				ServiceName: "Frontend",
+			},
+		},
+	}
+
+	dgd := &v1alpha1.DynamoGraphDeployment{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-dgd",
+			Namespace: "default",
+		},
+		Spec: v1alpha1.DynamoGraphDeploymentSpec{
+			Services: map[string]*v1alpha1.DynamoComponentDeploymentSharedSpec{
+				"Frontend": {
+					Replicas: ptr.To(int32(2)),
+				},
+			},
+		},
+	}
+
+	fakeClient := fake.NewClientBuilder().
+		WithScheme(scheme.Scheme).
+		WithObjects(adapter, dgd).
+		Build()
+
+	r := &DynamoGraphDeploymentScalingAdapterReconciler{
+		Client:   fakeClient,
+		Scheme:   scheme.Scheme,
+		Recorder: record.NewFakeRecorder(10),
+	}
+
+	ctx := context.Background()
+	req := ctrl.Request{
+		NamespacedName: types.NamespacedName{
+			Name:      adapter.Name,
+			Namespace: adapter.Namespace,
+		},
+	}
+
+	// Should return no error and skip reconciliation
+	result, err := r.Reconcile(ctx, req)
+	if err != nil {
+		t.Errorf("Expected no error for deleting adapter, got: %v", err)
+	}
+	if result.RequeueAfter != 0 {
+		t.Errorf("Expected no requeueAfter for deleting adapter, got: %v", result.RequeueAfter)
+	}
+
+	// DGD replicas should NOT be updated (still 2)
+	updatedDGD := &v1alpha1.DynamoGraphDeployment{}
+	if err := fakeClient.Get(ctx, types.NamespacedName{Name: dgd.Name, Namespace: dgd.Namespace}, updatedDGD); err != nil {
+		t.Fatalf("Failed to get DGD: %v", err)
+	}
+
+	if *updatedDGD.Spec.Services["Frontend"].Replicas != 2 {
+		t.Errorf("DGD replicas should remain unchanged, got %d", *updatedDGD.Spec.Services["Frontend"].Replicas)
+	}
+}
+
+func TestDynamoGraphDeploymentScalingAdapterReconciler_findAdaptersForDGD(t *testing.T) {
+	// Register custom types with the scheme
+	if err := v1alpha1.AddToScheme(scheme.Scheme); err != nil {
+		t.Fatalf("Failed to add v1alpha1 to scheme: %v", err)
+	}
+
+	dgd := &v1alpha1.DynamoGraphDeployment{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-dgd",
+			Namespace: "default",
+		},
+	}
+
+	// Adapters belonging to test-dgd
+	adapter1 := &v1alpha1.DynamoGraphDeploymentScalingAdapter{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-dgd-frontend",
+			Namespace: "default",
+			Labels: map[string]string{
+				consts.KubeLabelDynamoGraphDeploymentName: "test-dgd",
+			},
+		},
+		Spec: v1alpha1.DynamoGraphDeploymentScalingAdapterSpec{
+			DGDRef: v1alpha1.DynamoGraphDeploymentServiceRef{
+				Name:        "test-dgd",
+				ServiceName: "Frontend",
+			},
+		},
+	}
+
+	adapter2 := &v1alpha1.DynamoGraphDeploymentScalingAdapter{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-dgd-decode",
+			Namespace: "default",
+			Labels: map[string]string{
+				consts.KubeLabelDynamoGraphDeploymentName: "test-dgd",
+			},
+		},
+		Spec: v1alpha1.DynamoGraphDeploymentScalingAdapterSpec{
+			DGDRef: v1alpha1.DynamoGraphDeploymentServiceRef{
+				Name:        "test-dgd",
+				ServiceName: "decode",
+			},
+		},
+	}
+
+	// Adapter belonging to different DGD
+	adapterOther := &v1alpha1.DynamoGraphDeploymentScalingAdapter{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "other-dgd-frontend",
+			Namespace: "default",
+			Labels: map[string]string{
+				consts.KubeLabelDynamoGraphDeploymentName: "other-dgd",
+			},
+		},
+		Spec: v1alpha1.DynamoGraphDeploymentScalingAdapterSpec{
+			DGDRef: v1alpha1.DynamoGraphDeploymentServiceRef{
+				Name:        "other-dgd",
+				ServiceName: "Frontend",
+			},
+		},
+	}
+
+	fakeClient := fake.NewClientBuilder().
+		WithScheme(scheme.Scheme).
+		WithObjects(adapter1, adapter2, adapterOther).
+		Build()
+
+	r := &DynamoGraphDeploymentScalingAdapterReconciler{
+		Client: fakeClient,
+	}
+
+	ctx := context.Background()
+	requests := r.findAdaptersForDGD(ctx, dgd)
+
+	// Should return 2 requests (for test-dgd adapters only)
+	if len(requests) != 2 {
+		t.Errorf("findAdaptersForDGD() returned %d requests, expected 2", len(requests))
+	}
+
+	// Verify correct adapters are returned
+	expectedNames := map[string]bool{
+		"test-dgd-frontend": true,
+		"test-dgd-decode":   true,
+	}
+
+	for _, req := range requests {
+		if !expectedNames[req.Name] {
+			t.Errorf("Unexpected adapter in results: %s", req.Name)
+		}
+	}
+}