dynamocomponentdeployment_controller.go 57.7 KB
Newer Older
1
/*
2
 * SPDX-FileCopyrightText: Copyright (c) 2022 Atalaya Tech. Inc
3
 * SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
4
5
6
7
8
9
10
11
12
13
14
15
16
 * 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.
17
 * Modifications Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES
18
19
20
21
22
23
24
 */

package controller

import (
	"context"
	"fmt"
25
	"maps"
26
27
28
29
30
31
32
33
34
35
	"os"
	"time"

	appsv1 "k8s.io/api/apps/v1"
	autoscalingv2 "k8s.io/api/autoscaling/v2"
	corev1 "k8s.io/api/core/v1"
	networkingv1 "k8s.io/api/networking/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

	"emperror.dev/errors"
36
	"github.com/ai-dynamo/dynamo/deploy/operator/api/v1alpha1"
37
	"github.com/ai-dynamo/dynamo/deploy/operator/internal/checkpoint"
38
39
40
41
	"github.com/ai-dynamo/dynamo/deploy/operator/internal/common"
	commonconsts "github.com/ai-dynamo/dynamo/deploy/operator/internal/consts"
	commonController "github.com/ai-dynamo/dynamo/deploy/operator/internal/controller_common"
	"github.com/ai-dynamo/dynamo/deploy/operator/internal/dynamo"
42
	"github.com/ai-dynamo/dynamo/deploy/operator/internal/observability"
43
	webhookvalidation "github.com/ai-dynamo/dynamo/deploy/operator/internal/webhook/validation"
44
45
46
47
48
49
50
51
52
53
	networkingv1beta1 "istio.io/client-go/pkg/apis/networking/v1beta1"
	k8serrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/meta"
	"k8s.io/apimachinery/pkg/api/resource"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/apimachinery/pkg/util/intstr"
	"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"
54
	"sigs.k8s.io/controller-runtime/pkg/event"
55
56
	"sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/predicate"
57
58
59

	leaderworkersetv1 "sigs.k8s.io/lws/api/leaderworkerset/v1"
	volcanov1beta1 "volcano.sh/apis/pkg/apis/scheduling/v1beta1"
60
61
62
)

const (
63
64
65
	DefaultClusterName                                   = "default"
	DefaultServiceAccountName                            = "default"
	KubeAnnotationDeploymentStrategy                     = "nvidia.com/deployment-strategy"
66
67
	KubeAnnotationDeploymentRollingUpdateMaxSurge        = "nvidia.com/deployment-rolling-update-max-surge"
	KubeAnnotationDeploymentRollingUpdateMaxUnavailable  = "nvidia.com/deployment-rolling-update-max-unavailable"
68
69
70
71
	KubeAnnotationEnableStealingTrafficDebugMode         = "nvidia.com/enable-stealing-traffic-debug-mode"
	KubeAnnotationEnableDebugMode                        = "nvidia.com/enable-debug-mode"
	KubeAnnotationEnableDebugPodReceiveProductionTraffic = "nvidia.com/enable-debug-pod-receive-production-traffic"
	DeploymentTargetTypeDebug                            = "debug"
72
73
)

74
75
// DynamoComponentDeploymentReconciler reconciles a DynamoComponentDeployment object
type DynamoComponentDeploymentReconciler struct {
76
	client.Client
77
	Recorder              record.EventRecorder
78
	Config                commonController.Config
79
80
	EtcdStorage           etcdStorage
	DockerSecretRetriever dockerSecretRetriever
81
82
}

83
84
85
// +kubebuilder:rbac:groups=nvidia.com,resources=dynamocomponentdeployments,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=nvidia.com,resources=dynamocomponentdeployments/status,verbs=get;update;patch
// +kubebuilder:rbac:groups=nvidia.com,resources=dynamocomponentdeployments/finalizers,verbs=update
86
87
88
89
90
91
92
93
94
95
96
97
98
99

//+kubebuilder:rbac:groups=apps,resources=deployments,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=core,resources=pods,verbs=get;list;watch
//+kubebuilder:rbac:groups=core,resources=services,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=core,resources=configmaps,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=core,resources=events,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=autoscaling,resources=horizontalpodautoscalers,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=networking.k8s.io,resources=ingressclasses,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=networking.k8s.io,resources=ingresses,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=events.k8s.io,resources=events,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=coordination.k8s.io,resources=leases,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=networking.istio.io,resources=virtualservices,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=core,resources=persistentvolumeclaims,verbs=get;list;create;delete

100
101
102
// +kubebuilder:rbac:groups=scheduling.volcano.sh,resources=podgroups,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=leaderworkerset.x-k8s.io,resources=leaderworkersets,verbs=get;list;watch;create;update;patch;delete

103
104
105
// Reconcile is part of the main kubernetes reconciliation loop which aims to
// move the current state of the cluster closer to the desired state.
// TODO(user): Modify the Reconcile function to compare the state specified by
106
// the DynamoComponentDeployment object against the actual cluster state, and then
107
108
109
110
111
112
113
// perform operations to make the cluster state reflect the state specified by
// the user.
//
// For more details, check Reconcile and its Result here:
// - https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.18.2/pkg/reconcile
//
//nolint:gocyclo,nakedret
114
func (r *DynamoComponentDeploymentReconciler) Reconcile(ctx context.Context, req ctrl.Request) (result ctrl.Result, err error) {
115
116
	logs := log.FromContext(ctx)

117
118
	dynamoComponentDeployment := &v1alpha1.DynamoComponentDeployment{}
	err = r.Get(ctx, req.NamespacedName, dynamoComponentDeployment)
119
120
121
122
	if err != nil {
		if k8serrors.IsNotFound(err) {
			// Object not found, return.  Created objects are automatically garbage collected.
			// For additional cleanup logic use finalizers.
123
			logs.Info("DynamoComponentDeployment resource not found. Ignoring since object must be deleted.")
124
125
126
127
			err = nil
			return
		}
		// Error reading the object - requeue the request.
128
		logs.Error(err, "Failed to get DynamoComponentDeployment.")
129
130
131
		return
	}

132
	logs = logs.WithValues("dynamoComponentDeployment", dynamoComponentDeployment.Name, "namespace", dynamoComponentDeployment.Namespace)
133

134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
	// Setup defer to handle errors and update status
	defer func() {
		if err == nil {
			return
		}
		reconcileErr := err
		logs.Error(reconcileErr, "Failed to reconcile DynamoComponentDeployment.")
		r.Recorder.Eventf(dynamoComponentDeployment, corev1.EventTypeWarning, "ReconcileError",
			"Failed to reconcile DynamoComponentDeployment: %v", reconcileErr)
		if _, statusErr := r.setStatusConditions(ctx, req,
			metav1.Condition{
				Type:    v1alpha1.DynamoGraphDeploymentConditionTypeAvailable,
				Status:  metav1.ConditionFalse,
				Reason:  "Reconciling",
				Message: fmt.Sprintf("Failed to reconcile DynamoComponentDeployment: %v", reconcileErr),
			},
		); statusErr != nil {
			logs.Error(statusErr, "Failed to update DynamoComponentDeployment status after reconcile error")
		}
	}()

	// Validate the DynamoComponentDeployment spec (defense in depth - only when webhooks are disabled)
	if !r.Config.WebhooksEnabled {
		validator := webhookvalidation.NewDynamoComponentDeploymentValidator(dynamoComponentDeployment)
158
		if _, validationErr := validator.Validate(ctx); validationErr != nil {
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
			logs.Error(validationErr, "DynamoComponentDeployment validation failed, refusing to reconcile")

			// Set validation error condition
			meta.SetStatusCondition(&dynamoComponentDeployment.Status.Conditions, metav1.Condition{
				Type:               "Valid",
				Status:             metav1.ConditionFalse,
				ObservedGeneration: dynamoComponentDeployment.Generation,
				Reason:             "ValidationFailed",
				Message:            fmt.Sprintf("Validation failed: %v", validationErr),
			})

			// Update status and don't requeue (user must fix the spec)
			if statusErr := r.Status().Update(ctx, dynamoComponentDeployment); statusErr != nil {
				logs.Error(statusErr, "Failed to update DynamoComponentDeployment status with validation error")
				err = statusErr
				return ctrl.Result{}, err
			}

			// Record event for visibility
			r.Recorder.Event(dynamoComponentDeployment, corev1.EventTypeWarning, "ValidationFailed", validationErr.Error())

			// Don't requeue - user must fix the spec
			logs.Info("DynamoComponentDeployment is invalid, not reconciling until spec is fixed")
			err = nil
			return ctrl.Result{}, nil
		}

		// Set Valid condition to True and persist it
		meta.SetStatusCondition(&dynamoComponentDeployment.Status.Conditions, metav1.Condition{
			Type:               "Valid",
			Status:             metav1.ConditionTrue,
			ObservedGeneration: dynamoComponentDeployment.Generation,
			Reason:             "ValidationPassed",
			Message:            "DynamoComponentDeployment spec is valid",
		})
		if statusErr := r.Status().Update(ctx, dynamoComponentDeployment); statusErr != nil {
			logs.Error(statusErr, "Failed to update DynamoComponentDeployment status with validation success")
			err = statusErr
			return ctrl.Result{}, err
		}
	}

201
	deleted, err := commonController.HandleFinalizer(ctx, dynamoComponentDeployment, r.Client, r)
202
203
204
205
206
207
208
209
	if err != nil {
		logs.Error(err, "Failed to handle finalizer")
		return ctrl.Result{}, err
	}
	if deleted {
		return ctrl.Result{}, nil
	}

210
211
212
213
214
	if len(dynamoComponentDeployment.Status.Conditions) == 0 {
		logs.Info("Starting to reconcile DynamoComponentDeployment")
		logs.Info("Initializing DynamoComponentDeployment status")
		r.Recorder.Event(dynamoComponentDeployment, corev1.EventTypeNormal, "Reconciling", "Starting to reconcile DynamoComponentDeployment")
		dynamoComponentDeployment, err = r.setStatusConditions(ctx, req,
215
			metav1.Condition{
216
				Type:    v1alpha1.DynamoGraphDeploymentConditionTypeAvailable,
217
218
				Status:  metav1.ConditionUnknown,
				Reason:  "Reconciling",
219
				Message: "Starting to reconcile DynamoComponentDeployment",
220
221
			},
			metav1.Condition{
222
				Type:    v1alpha1.DynamoGraphDeploymentConditionTypeDynamoComponentReady,
223
224
				Status:  metav1.ConditionUnknown,
				Reason:  "Reconciling",
225
				Message: "Starting to reconcile DynamoComponentDeployment",
226
227
228
229
230
231
232
			},
		)
		if err != nil {
			return
		}
	}

233
	// Create the appropriate workload resource based on deployment type
234
	var componentReconcileResult ComponentReconcileResult
235
	if r.Config.LWS.Enabled && dynamoComponentDeployment.IsMultinode() {
236
		componentReconcileResult, err = r.reconcileLeaderWorkerSetResources(ctx, dynamoComponentDeployment)
237
	} else {
238
		componentReconcileResult, err = r.reconcileDeploymentResources(ctx, dynamoComponentDeployment)
239
	}
240
241
242
243
	if err != nil {
		return ctrl.Result{}, fmt.Errorf("failed to reconcile the resources: %w", err)
	}
	modified := componentReconcileResult.modified
244
245

	// create or update api-server service
246
	serviceModified, err := r.createOrUpdateOrDeleteServices(ctx, generateResourceOption{
247
		dynamoComponentDeployment: dynamoComponentDeployment,
248
249
	})
	if err != nil {
250
		return ctrl.Result{}, fmt.Errorf("failed to create or update the service: %w", err)
251
252
	}

253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
	// create or update headless service for model endpoint discovery
	componentMap := map[string]*v1alpha1.DynamoComponentDeploymentSharedSpec{
		dynamoComponentDeployment.Name: &dynamoComponentDeployment.Spec.DynamoComponentDeploymentSharedSpec,
	}
	if err := dynamo.ReconcileModelServicesForComponents(
		ctx,
		r,
		dynamoComponentDeployment,
		componentMap,
		dynamoComponentDeployment.Namespace,
	); err != nil {
		logs.Error(err, "Failed to reconcile model service")
		return ctrl.Result{}, err
	}

268
	// create or update api-server ingresses
269
	ingressModified, err := r.createOrUpdateOrDeleteIngress(ctx, generateResourceOption{
270
		dynamoComponentDeployment: dynamoComponentDeployment,
271
	})
272
	if err != nil {
273
		return ctrl.Result{}, fmt.Errorf("failed to create or update the ingress: %w", err)
274
275
	}

276
	if serviceModified || ingressModified {
277
278
279
280
		modified = true
	}

	if !modified {
281
		r.Recorder.Eventf(dynamoComponentDeployment, corev1.EventTypeNormal, "UpdateDynamoGraphDeployment", "No changes to dynamo deployment %s", dynamoComponentDeployment.Name)
282
283
284
	}

	logs.Info("Finished reconciling.")
285
	r.Recorder.Eventf(dynamoComponentDeployment, corev1.EventTypeNormal, "Update", "All resources updated!")
286

287
288
289
	err = r.setStatusConditionAndServiceReplicaStatus(ctx, dynamoComponentDeployment, componentReconcileResult)
	if err != nil {
		return ctrl.Result{}, fmt.Errorf("failed to set status condition and service replica status: %w", err)
290
291
	}

292
293
294
	return
}

295
296
297
298
299
type ComponentReconcileResult struct {
	modified             bool
	status               metav1.ConditionStatus
	reason               string
	message              string
300
	serviceReplicaStatus *v1alpha1.ServiceReplicaStatus
301
302
303
304
305
306
307
308
309
310
311
}

func (r *DynamoComponentDeploymentReconciler) reconcileDeploymentResources(ctx context.Context, dynamoComponentDeployment *v1alpha1.DynamoComponentDeployment) (ComponentReconcileResult, error) {
	logger := log.FromContext(ctx)
	deploymentModified, deployment, err := r.createOrUpdateOrDeleteDeployments(ctx, generateResourceOption{
		dynamoComponentDeployment: dynamoComponentDeployment,
	})
	if err != nil {
		return ComponentReconcileResult{}, fmt.Errorf("failed to create or update the deployment: %w", err)
	}

312
313
314
315
316
317
318
319
320
321
	logger.V(1).Info("Deployment sync completed",
		"deploymentModified", deploymentModified,
		"deploymentName", deployment.Name,
		"deploymentGeneration", deployment.Generation,
		"deploymentObservedGeneration", deployment.Status.ObservedGeneration,
		"deploymentReplicas", deployment.Status.Replicas,
		"deploymentUpdatedReplicas", deployment.Status.UpdatedReplicas,
		"deploymentAvailableReplicas", deployment.Status.AvailableReplicas,
		"deploymentReadyReplicas", deployment.Status.ReadyReplicas)

322
	serviceReplicaStatus := &v1alpha1.ServiceReplicaStatus{
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
		ComponentKind:     v1alpha1.ComponentKindDeployment,
		ComponentName:     deployment.Name,
		Replicas:          deployment.Status.Replicas,
		UpdatedReplicas:   deployment.Status.UpdatedReplicas,
		ReadyReplicas:     &deployment.Status.ReadyReplicas,
		AvailableReplicas: &deployment.Status.AvailableReplicas,
	}

	if IsDeploymentReady(deployment) {
		return ComponentReconcileResult{
			modified:             deploymentModified,
			status:               metav1.ConditionTrue,
			reason:               "DeploymentReady",
			message:              "Deployment is ready",
			serviceReplicaStatus: serviceReplicaStatus,
		}, nil
	}
	return ComponentReconcileResult{
		modified:             deploymentModified,
		status:               metav1.ConditionFalse,
		reason:               "DeploymentNotReady",
		message:              "Deployment is not ready",
		serviceReplicaStatus: serviceReplicaStatus,
	}, nil
}

func (r *DynamoComponentDeploymentReconciler) reconcileLeaderWorkerSetResources(ctx context.Context, dynamoComponentDeployment *v1alpha1.DynamoComponentDeployment) (ComponentReconcileResult, error) {
	logger := log.FromContext(ctx)

	desiredReplicas := int32(1)
	if dynamoComponentDeployment.Spec.Replicas != nil {
		desiredReplicas = *dynamoComponentDeployment.Spec.Replicas
	}

	anyModified := false
	leaderWorkerSets := make([]*leaderworkersetv1.LeaderWorkerSet, 0, desiredReplicas)
	for i := range int(desiredReplicas) {
		volcanoPodGroupModified, _, err := commonController.SyncResource(ctx, r, dynamoComponentDeployment, func(ctx context.Context) (*volcanov1beta1.PodGroup, bool, error) {
			return r.generateVolcanoPodGroup(ctx, generateResourceOption{
				dynamoComponentDeployment:               dynamoComponentDeployment,
				isStealingTrafficDebugModeEnabled:       false,
				containsStealingTrafficDebugModeEnabled: false,
				instanceID:                              &i,
			})
		})
		if err != nil {
			return ComponentReconcileResult{}, fmt.Errorf("failed to sync the PodGroup: %w", err)
		}

		leaderWorkerSetModified, lwsObj, err := commonController.SyncResource(ctx, r, dynamoComponentDeployment, func(ctx context.Context) (*leaderworkersetv1.LeaderWorkerSet, bool, error) {
			return r.generateLeaderWorkerSet(ctx, generateResourceOption{
				dynamoComponentDeployment:               dynamoComponentDeployment,
				isStealingTrafficDebugModeEnabled:       false,
				containsStealingTrafficDebugModeEnabled: false,
				instanceID:                              &i,
			})
		})
		if err != nil {
			return ComponentReconcileResult{}, fmt.Errorf("failed to sync the LeaderWorkerSet: %w", err)
		}

		if leaderWorkerSetModified || volcanoPodGroupModified {
			anyModified = true
		}
		leaderWorkerSets = append(leaderWorkerSets, lwsObj)
	}

	// Clean up any excess LeaderWorkerSets (if replicas were decreased)
	baseKubeName := r.getKubeName(dynamoComponentDeployment, false)
	for i := int(desiredReplicas); ; i++ {
		// Try to find a LeaderWorkerSet with the next index
		nextLWSName := fmt.Sprintf("%s-%d", baseKubeName, i)
		lwsToDelete := &leaderworkersetv1.LeaderWorkerSet{}
		err := r.Get(ctx, types.NamespacedName{
			Name:      nextLWSName,
			Namespace: dynamoComponentDeployment.Namespace,
		}, lwsToDelete)

		if err != nil {
			if k8serrors.IsNotFound(err) {
				break
			}
			return ComponentReconcileResult{}, fmt.Errorf("failed to get the LeaderWorkerSet for deletion: %w", err)
		}

		err = r.Delete(ctx, lwsToDelete)
		if err != nil {
			return ComponentReconcileResult{}, fmt.Errorf("failed to delete the LeaderWorkerSet: %w", err)
		}

		podGroupName := nextLWSName
		podGroupToDelete := &volcanov1beta1.PodGroup{}
		err = r.Get(ctx, types.NamespacedName{
			Name:      podGroupName,
			Namespace: dynamoComponentDeployment.Namespace,
		}, podGroupToDelete)

		if err != nil {
			if !k8serrors.IsNotFound(err) {
				logger.Error(err, "Failed to get PodGroup for deletion", "podGroupName", podGroupName)
			}
		} else {
			err = r.Delete(ctx, podGroupToDelete)
			if err != nil {
				logger.Error(err, "Failed to delete PodGroup", "podGroupName", podGroupName)
			}
		}

		anyModified = true
	}
433
434

	allReady := true
435
	lwsReplicaStatuses := []v1alpha1.ServiceReplicaStatus{}
436
437
438
439
	for _, leaderWorkerSet := range leaderWorkerSets {
		if !IsLeaderWorkerSetReady(leaderWorkerSet) {
			allReady = false
		}
440
		lwsReplicaStatuses = append(lwsReplicaStatuses, getLeaderWorkerSetReplicasStatus(leaderWorkerSet))
441
442
443
	}

	if allReady {
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
		return ComponentReconcileResult{
			modified:             anyModified,
			status:               metav1.ConditionTrue,
			reason:               "AllLeaderWorkerSetsReady",
			message:              "All LeaderWorkerSets are ready",
			serviceReplicaStatus: combineLWSReplicaStatuses(lwsReplicaStatuses),
		}, nil
	}
	return ComponentReconcileResult{
		modified:             anyModified,
		status:               metav1.ConditionFalse,
		reason:               "SomeLeaderWorkerSetsNotReady",
		message:              "Some LeaderWorkerSets are not ready",
		serviceReplicaStatus: combineLWSReplicaStatuses(lwsReplicaStatuses),
	}, nil

}

func (r *DynamoComponentDeploymentReconciler) setStatusConditionAndServiceReplicaStatus(ctx context.Context, dynamoComponentDeployment *v1alpha1.DynamoComponentDeployment, componentReconcileResult ComponentReconcileResult) error {
463
	availableCondition := metav1.Condition{
464
465
466
467
468
469
		Type:    v1alpha1.DynamoGraphDeploymentConditionTypeAvailable,
		Status:  componentReconcileResult.status,
		Reason:  componentReconcileResult.reason,
		Message: componentReconcileResult.message,
	}

470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
	var componentReadyReason, componentReadyMessage string
	if componentReconcileResult.status == metav1.ConditionTrue {
		componentReadyReason = "ComponentReady"
		componentReadyMessage = "DynamoComponent is ready"
	} else {
		componentReadyReason = "ComponentNotReady"
		componentReadyMessage = "DynamoComponent is not ready"
	}

	componentReadyCondition := metav1.Condition{
		Type:    v1alpha1.DynamoGraphDeploymentConditionTypeDynamoComponentReady,
		Status:  componentReconcileResult.status,
		Reason:  componentReadyReason,
		Message: componentReadyMessage,
	}

	meta.SetStatusCondition(&dynamoComponentDeployment.Status.Conditions, availableCondition)
	meta.SetStatusCondition(&dynamoComponentDeployment.Status.Conditions, componentReadyCondition)
488
	dynamoComponentDeployment.Status.Service = componentReconcileResult.serviceReplicaStatus
489
	dynamoComponentDeployment.Status.ObservedGeneration = dynamoComponentDeployment.Generation
490
491
492
493

	err := r.Status().Update(ctx, dynamoComponentDeployment)
	if err != nil {
		return fmt.Errorf("failed to update DynamoComponentDeployment status: %w", err)
494
	}
495
496
497
498
499
500
501
502
503
504
505
506
507
	return nil
}

func getLeaderWorkerSetReplicasStatus(leaderWorkerSet *leaderworkersetv1.LeaderWorkerSet) v1alpha1.ServiceReplicaStatus {
	return v1alpha1.ServiceReplicaStatus{
		ComponentKind:   v1alpha1.ComponentKindLeaderWorkerSet,
		ComponentName:   leaderWorkerSet.Name,
		Replicas:        leaderWorkerSet.Status.Replicas,
		UpdatedReplicas: leaderWorkerSet.Status.UpdatedReplicas,
		ReadyReplicas:   &leaderWorkerSet.Status.ReadyReplicas,
	}
}

508
func combineLWSReplicaStatuses(serviceReplicaStatuses []v1alpha1.ServiceReplicaStatus) *v1alpha1.ServiceReplicaStatus {
509
	if len(serviceReplicaStatuses) == 0 {
510
		return nil
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
	}

	firstServiceStatus := serviceReplicaStatuses[0]
	var readyReplicas int32 = 0
	if firstServiceStatus.ReadyReplicas != nil {
		readyReplicas = *firstServiceStatus.ReadyReplicas
	}
	for _, serviceReplicaStatus := range serviceReplicaStatuses[1:] {
		firstServiceStatus.Replicas += serviceReplicaStatus.Replicas
		firstServiceStatus.UpdatedReplicas += serviceReplicaStatus.UpdatedReplicas
		if serviceReplicaStatus.ReadyReplicas != nil {
			readyReplicas += *serviceReplicaStatus.ReadyReplicas
		}
	}

	firstServiceStatus.ReadyReplicas = &readyReplicas
527
	return &firstServiceStatus
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
}

// IsLeaderWorkerSetReady determines if a LeaderWorkerSet is fully ready and available
func IsLeaderWorkerSetReady(leaderWorkerSet *leaderworkersetv1.LeaderWorkerSet) bool {
	if leaderWorkerSet == nil {
		return false
	}

	desiredReplicas := int32(1)
	if leaderWorkerSet.Spec.Replicas != nil {
		desiredReplicas = *leaderWorkerSet.Spec.Replicas
	}

	// Special case: if no replicas are desired, the LeaderWorkerSet is considered ready
	if desiredReplicas == 0 {
		return true
	}

	status := leaderWorkerSet.Status

	if status.ReadyReplicas < desiredReplicas {
		return false
	}

	// Look for the Available condition specifically - this is defined in the CRD for LeaderWorkerSet
	for _, cond := range leaderWorkerSet.Status.Conditions {
		if cond.Type == string(leaderworkersetv1.LeaderWorkerSetAvailable) {
			return cond.Status == metav1.ConditionTrue
		}
	}

	return false
}

func (r *DynamoComponentDeploymentReconciler) generateVolcanoPodGroup(ctx context.Context, opt generateResourceOption) (*volcanov1beta1.PodGroup, bool, error) {
	logs := log.FromContext(ctx)
	logs.Info("Generating Volcano PodGroup")

	if opt.instanceID == nil {
		return nil, false, errors.New("generateVolcanoPodGroup: instanceID cannot be nil")
	}
	instanceID := *opt.instanceID

	if instanceID < 0 {
		return nil, false, fmt.Errorf("generateVolcanoPodGroup: instanceID cannot be negative, got %d", instanceID)
	}

575
	podGroupName := r.getKubeName(opt.dynamoComponentDeployment, opt.isStealingTrafficDebugModeEnabled)
576
577
578
579
580
581
582
	podGroupName = fmt.Sprintf("%s-%d", podGroupName, instanceID)

	kubeNs := opt.dynamoComponentDeployment.Namespace

	labels := make(map[string]string)
	labels["instance-id"] = fmt.Sprintf("%d", instanceID)

583
	minMember := opt.dynamoComponentDeployment.GetNumberOfNodes()
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599

	podGroup := &volcanov1beta1.PodGroup{
		ObjectMeta: metav1.ObjectMeta{
			Name:      podGroupName,
			Namespace: kubeNs,
			Labels:    labels,
		},
		Spec: volcanov1beta1.PodGroupSpec{
			MinMember: minMember,
		},
	}

	return podGroup, false, nil
}

func (r *DynamoComponentDeploymentReconciler) generateLeaderPodTemplateSpec(ctx context.Context, opt generateResourceOption, kubeName string, labels map[string]string, instanceID int) (*corev1.PodTemplateSpec, error) {
600
	leaderPodTemplateSpec, err := r.generatePodTemplateSpec(ctx, opt, dynamo.RoleLeader)
601
602
603
604
	if err != nil {
		return nil, errors.Wrap(err, "failed to generate leader pod template")
	}

605
	maps.Copy(leaderPodTemplateSpec.ObjectMeta.Labels, labels)
606
607
608
609
610
611
612
613
614
615
616
	leaderPodTemplateSpec.ObjectMeta.Labels["role"] = "leader"
	leaderPodTemplateSpec.ObjectMeta.Labels["instance-id"] = fmt.Sprintf("%d", instanceID)
	delete(leaderPodTemplateSpec.ObjectMeta.Labels, commonconsts.KubeLabelDynamoSelector)

	if leaderPodTemplateSpec.ObjectMeta.Annotations == nil {
		leaderPodTemplateSpec.ObjectMeta.Annotations = make(map[string]string)
	}
	leaderPodTemplateSpec.ObjectMeta.Annotations["scheduling.k8s.io/group-name"] = kubeName

	leaderPodTemplateSpec.Spec.SchedulerName = "volcano"

617
	err = checkMainContainer(&leaderPodTemplateSpec.Spec)
618

619
620
	if err != nil {
		return nil, errors.Wrap(err, "generateLeaderPodTemplateSpec: failed to check main container")
621
622
623
624
625
	}

	return leaderPodTemplateSpec, nil
}

626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
func checkMainContainer(spec *corev1.PodSpec) error {

	if len(spec.Containers) == 0 {
		return errors.New("No containers found in pod spec")
	}

	mainContainerFound := false
	for _, container := range spec.Containers {
		if container.Name != commonconsts.MainContainerName {
			continue
		}

		if len(container.Command) == 0 {
			return errors.New("container Command cannot be nil for LWS pod")
		}

		if len(container.Args) == 0 {
			return errors.New("container Args cannot be empty for LWS pod")
		}

		mainContainerFound = true
		break
	}

	if !mainContainerFound {
		return errors.New("main container not found in pod spec")
	}

	return nil
}

657
func (r *DynamoComponentDeploymentReconciler) generateWorkerPodTemplateSpec(ctx context.Context, opt generateResourceOption, kubeName string, labels map[string]string, instanceID int) (*corev1.PodTemplateSpec, error) {
658
	workerPodTemplateSpec, err := r.generatePodTemplateSpec(ctx, opt, dynamo.RoleWorker)
659
660
661
662
	if err != nil {
		return nil, errors.Wrap(err, "failed to generate worker pod template")
	}

663
	maps.Copy(workerPodTemplateSpec.ObjectMeta.Labels, labels)
664
665
666
667
668
669
670
671
672
673
674
	workerPodTemplateSpec.ObjectMeta.Labels["role"] = "worker"
	workerPodTemplateSpec.ObjectMeta.Labels["instance-id"] = fmt.Sprintf("%d", instanceID)
	delete(workerPodTemplateSpec.ObjectMeta.Labels, commonconsts.KubeLabelDynamoSelector)

	workerPodTemplateSpec.Spec.SchedulerName = "volcano"

	if workerPodTemplateSpec.ObjectMeta.Annotations == nil {
		workerPodTemplateSpec.ObjectMeta.Annotations = make(map[string]string)
	}
	workerPodTemplateSpec.ObjectMeta.Annotations["scheduling.k8s.io/group-name"] = kubeName

675
	err = checkMainContainer(&workerPodTemplateSpec.Spec)
676

677
678
	if err != nil {
		return nil, errors.Wrap(err, "generateWorkerPodTemplateSpec: failed to check LWS worker main container")
679
680
681
	}

	if opt.dynamoComponentDeployment.Spec.Resources == nil || opt.dynamoComponentDeployment.Spec.Resources.Limits == nil || opt.dynamoComponentDeployment.Spec.Resources.Limits.GPU == "" {
682
		return nil, fmt.Errorf("generateWorkerPodTemplateSpec: GPU limit is not set for LWS worker pod")
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
	}

	return workerPodTemplateSpec, nil
}

// generateLeaderWorkerSet creates a LeaderWorkerSet resource from the DynamoComponentDeployment
func (r *DynamoComponentDeploymentReconciler) generateLeaderWorkerSet(ctx context.Context, opt generateResourceOption) (*leaderworkersetv1.LeaderWorkerSet, bool, error) {
	logs := log.FromContext(ctx)
	logs.Info("Generating LeaderWorkerSet")

	if opt.instanceID == nil {
		return nil, false, errors.New("generateLeaderWorkerSet: instanceID cannot be nil")
	}
	instanceID := *opt.instanceID

	if instanceID < 0 {
		return nil, false, fmt.Errorf("generateLeaderWorkerSet: instanceID cannot be negative, got %d", instanceID)
	}

702
	kubeName := r.getKubeName(opt.dynamoComponentDeployment, opt.isStealingTrafficDebugModeEnabled)
703
704
705
	kubeName = fmt.Sprintf("%s-%d", kubeName, instanceID)

	kubeNs := opt.dynamoComponentDeployment.Namespace
706
	labels := r.getKubeLabels(opt.dynamoComponentDeployment)
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740

	if labels == nil {
		labels = make(map[string]string)
	}
	labels["instance-id"] = fmt.Sprintf("%d", instanceID)

	leaderWorkerSet := &leaderworkersetv1.LeaderWorkerSet{
		ObjectMeta: metav1.ObjectMeta{
			Name:      kubeName,
			Namespace: kubeNs,
			Labels:    labels,
		},
	}

	leaderPodLabels := make(map[string]string)
	for k, v := range labels {
		leaderPodLabels[k] = v
	}
	leaderPodTemplateSpec, err := r.generateLeaderPodTemplateSpec(ctx, opt, kubeName, leaderPodLabels, instanceID)
	if err != nil {
		return nil, false, errors.Wrap(err, "generateLeaderWorkerSet: failed to generate leader pod template")
	}

	workerPodLabels := make(map[string]string)
	for k, v := range labels {
		workerPodLabels[k] = v
	}
	workerPodTemplateSpec, err := r.generateWorkerPodTemplateSpec(ctx, opt, kubeName, workerPodLabels, instanceID)
	if err != nil {
		return nil, false, errors.Wrap(err, "generateLeaderWorkerSet: failed to generate worker pod template")
	}

	// Each individual LeaderWorkerSet always has exactly 1 replica
	singleReplica := int32(1)
741
	groupSize := opt.dynamoComponentDeployment.GetNumberOfNodes()
742
743
744
745
746
747
748
749
750
751
752
753
754
755

	leaderWorkerSet.Spec = leaderworkersetv1.LeaderWorkerSetSpec{
		Replicas:      &singleReplica,
		StartupPolicy: leaderworkersetv1.LeaderCreatedStartupPolicy,
		LeaderWorkerTemplate: leaderworkersetv1.LeaderWorkerTemplate{
			LeaderTemplate: leaderPodTemplateSpec,
			WorkerTemplate: *workerPodTemplateSpec,
			Size:           &groupSize,
		},
	}

	return leaderWorkerSet, false, nil
}

756
func (r *DynamoComponentDeploymentReconciler) FinalizeResource(ctx context.Context, dynamoComponentDeployment *v1alpha1.DynamoComponentDeployment) error {
757
	logger := log.FromContext(ctx)
758
	logger.Info("Finalizing the DynamoComponentDeployment", "dynamoComponentDeployment", dynamoComponentDeployment)
759
760
761
762
763
764
765

	// Only delete etcd keys if using etcd discovery backend
	// When using Kubernetes discovery (the default), skip etcd cleanup to avoid hangs
	if r.Config.DiscoveryBackend != "etcd" {
		return nil
	}

766
767
768
	if dynamoComponentDeployment.Spec.ServiceName != "" && dynamoComponentDeployment.Spec.DynamoNamespace != nil && *dynamoComponentDeployment.Spec.DynamoNamespace != "" {
		logger.Info("Deleting the etcd keys for the service", "service", dynamoComponentDeployment.Spec.ServiceName, "dynamoNamespace", *dynamoComponentDeployment.Spec.DynamoNamespace)
		err := r.EtcdStorage.DeleteKeys(ctx, fmt.Sprintf("/%s/components/%s", *dynamoComponentDeployment.Spec.DynamoNamespace, dynamoComponentDeployment.Spec.ServiceName))
769
		if err != nil {
770
			logger.Error(err, "Failed to delete the etcd keys for the service", "service", dynamoComponentDeployment.Spec.ServiceName, "dynamoNamespace", *dynamoComponentDeployment.Spec.DynamoNamespace)
771
772
773
774
775
776
			return err
		}
	}
	return nil
}

777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
// IsDeploymentReady determines if a Kubernetes Deployment is fully ready and available.
// It checks various status fields to ensure all replicas are available and the deployment
// configuration has been fully applied.
func IsDeploymentReady(deployment *appsv1.Deployment) bool {
	if deployment == nil {
		return false
	}
	// Paused deployments should not be considered ready
	if deployment.Spec.Paused {
		return false
	}
	// Default to 1 replica if not specified
	desiredReplicas := int32(1)
	if deployment.Spec.Replicas != nil {
		desiredReplicas = *deployment.Spec.Replicas
	}
	// Special case: if no replicas are desired, the deployment is considered ready
	if desiredReplicas == 0 {
		return true
	}
	status := deployment.Status
	// Check all basic status requirements:
	// 1. ObservedGeneration: Deployment controller has observed the latest configuration
	// 2. UpdatedReplicas: All replicas have been updated to the latest version
	// 3. AvailableReplicas: All desired replicas are available (schedulable and healthy)
802
	// 4. Replicas: Total replicas equals desired (no surge pods remaining from rolling update)
803
804
	if status.ObservedGeneration < deployment.Generation ||
		status.UpdatedReplicas < desiredReplicas ||
805
806
		status.AvailableReplicas < desiredReplicas ||
		status.Replicas != desiredReplicas {
807
808
809
810
811
812
813
814
815
816
817
818
819
		return false
	}
	// Finally, check for the DeploymentAvailable condition
	// This is Kubernetes' own assessment that the deployment is available
	for _, cond := range deployment.Status.Conditions {
		if cond.Type == appsv1.DeploymentAvailable && cond.Status == corev1.ConditionTrue {
			return true
		}
	}
	// If we get here, the basic checks passed but the Available condition wasn't found
	return false
}

820
821
func (r *DynamoComponentDeploymentReconciler) setStatusConditions(ctx context.Context, req ctrl.Request, conditions ...metav1.Condition) (dynamoComponentDeployment *v1alpha1.DynamoComponentDeployment, err error) {
	dynamoComponentDeployment = &v1alpha1.DynamoComponentDeployment{}
822
823
	maxRetries := 3
	for range maxRetries - 1 {
824
825
		if err = r.Get(ctx, req.NamespacedName, dynamoComponentDeployment); err != nil {
			err = errors.Wrap(err, "Failed to re-fetch DynamoComponentDeployment")
826
827
828
			return
		}
		for _, condition := range conditions {
829
			meta.SetStatusCondition(&dynamoComponentDeployment.Status.Conditions, condition)
830
		}
831
		if err = r.Status().Update(ctx, dynamoComponentDeployment); err != nil {
832
833
834
835
836
			if k8serrors.IsConflict(err) {
				time.Sleep(100 * time.Millisecond)
				continue
			}
			break
837
838
839
840
841
		} else {
			break
		}
	}
	if err != nil {
842
		err = errors.Wrap(err, "Failed to update DynamoComponentDeployment status")
843
844
		return
	}
845
846
	if err = r.Get(ctx, req.NamespacedName, dynamoComponentDeployment); err != nil {
		err = errors.Wrap(err, "Failed to re-fetch DynamoComponentDeployment")
847
848
849
850
851
852
		return
	}
	return
}

//nolint:nakedret
853
854
func (r *DynamoComponentDeploymentReconciler) createOrUpdateOrDeleteDeployments(ctx context.Context, opt generateResourceOption) (modified bool, depl *appsv1.Deployment, err error) {
	containsStealingTrafficDebugModeEnabled := checkIfContainsStealingTrafficDebugModeEnabled(opt.dynamoComponentDeployment)
855
	// create the main deployment
856
857
858
859
860
861
862
	modified, depl, err = commonController.SyncResource(ctx, r, opt.dynamoComponentDeployment, func(ctx context.Context) (*appsv1.Deployment, bool, error) {
		return r.generateDeployment(ctx, generateResourceOption{
			dynamoComponentDeployment:               opt.dynamoComponentDeployment,
			isStealingTrafficDebugModeEnabled:       false,
			containsStealingTrafficDebugModeEnabled: containsStealingTrafficDebugModeEnabled,
		})
	})
863
864
865
866
	if err != nil {
		err = errors.Wrap(err, "create or update deployment")
		return
	}
867
	// create the debug deployment
868
869
870
871
872
873
874
	modified2, _, err := commonController.SyncResource(ctx, r, opt.dynamoComponentDeployment, func(ctx context.Context) (*appsv1.Deployment, bool, error) {
		return r.generateDeployment(ctx, generateResourceOption{
			dynamoComponentDeployment:               opt.dynamoComponentDeployment,
			isStealingTrafficDebugModeEnabled:       true,
			containsStealingTrafficDebugModeEnabled: containsStealingTrafficDebugModeEnabled,
		})
	})
875
876
	if err != nil {
		err = errors.Wrap(err, "create or update debug deployment")
877
	}
878
	modified = modified || modified2
879
880
881
	return
}

882
883
func getResourceAnnotations(dynamoComponentDeployment *v1alpha1.DynamoComponentDeployment) map[string]string {
	resourceAnnotations := dynamoComponentDeployment.Spec.Annotations
884
885
886
887
888
889
890
891
892
893
894
895
	if resourceAnnotations == nil {
		resourceAnnotations = map[string]string{}
	}

	return resourceAnnotations
}

func checkIfIsDebugModeEnabled(annotations map[string]string) bool {
	if annotations == nil {
		return false
	}

896
	return annotations[KubeAnnotationEnableDebugMode] == commonconsts.KubeLabelValueTrue
897
898
899
900
901
902
903
}

func checkIfIsStealingTrafficDebugModeEnabled(annotations map[string]string) bool {
	if annotations == nil {
		return false
	}

904
	return annotations[KubeAnnotationEnableStealingTrafficDebugMode] == commonconsts.KubeLabelValueTrue
905
906
907
908
909
910
911
}

func checkIfIsDebugPodReceiveProductionTrafficEnabled(annotations map[string]string) bool {
	if annotations == nil {
		return false
	}

912
	return annotations[KubeAnnotationEnableDebugPodReceiveProductionTraffic] == commonconsts.KubeLabelValueTrue
913
914
}

915
916
func checkIfContainsStealingTrafficDebugModeEnabled(dynamoComponentDeployment *v1alpha1.DynamoComponentDeployment) bool {
	return checkIfIsStealingTrafficDebugModeEnabled(dynamoComponentDeployment.Spec.Annotations)
917
918
919
}

//nolint:nakedret
920
921
func (r *DynamoComponentDeploymentReconciler) createOrUpdateOrDeleteServices(ctx context.Context, opt generateResourceOption) (modified bool, err error) {
	resourceAnnotations := getResourceAnnotations(opt.dynamoComponentDeployment)
922
	isDebugPodReceiveProductionTrafficEnabled := checkIfIsDebugPodReceiveProductionTrafficEnabled(resourceAnnotations)
923
	containsStealingTrafficDebugModeEnabled := checkIfContainsStealingTrafficDebugModeEnabled(opt.dynamoComponentDeployment)
924
	// main generic service
925
	modified, _, err = commonController.SyncResource(ctx, r, opt.dynamoComponentDeployment, func(ctx context.Context) (*corev1.Service, bool, error) {
926
		return r.generateService(generateResourceOption{
927
928
929
930
931
932
933
			dynamoComponentDeployment:               opt.dynamoComponentDeployment,
			isStealingTrafficDebugModeEnabled:       false,
			isDebugPodReceiveProductionTraffic:      isDebugPodReceiveProductionTrafficEnabled,
			containsStealingTrafficDebugModeEnabled: containsStealingTrafficDebugModeEnabled,
			isGenericService:                        true,
		})
	})
934
935
936
937
	if err != nil {
		return
	}

938
	// debug production service (if enabled)
939
	modified_, _, err := commonController.SyncResource(ctx, r, opt.dynamoComponentDeployment, func(ctx context.Context) (*corev1.Service, bool, error) {
940
		return r.generateService(generateResourceOption{
941
942
943
944
945
946
947
			dynamoComponentDeployment:               opt.dynamoComponentDeployment,
			isStealingTrafficDebugModeEnabled:       false,
			isDebugPodReceiveProductionTraffic:      isDebugPodReceiveProductionTrafficEnabled,
			containsStealingTrafficDebugModeEnabled: containsStealingTrafficDebugModeEnabled,
			isGenericService:                        false,
		})
	})
948
949
950
	if err != nil {
		return
	}
951
952
	modified = modified || modified_
	// debug service (if enabled)
953
	modified_, _, err = commonController.SyncResource(ctx, r, opt.dynamoComponentDeployment, func(ctx context.Context) (*corev1.Service, bool, error) {
954
		return r.generateService(generateResourceOption{
955
956
957
958
959
960
961
			dynamoComponentDeployment:               opt.dynamoComponentDeployment,
			isStealingTrafficDebugModeEnabled:       true,
			isDebugPodReceiveProductionTraffic:      isDebugPodReceiveProductionTrafficEnabled,
			containsStealingTrafficDebugModeEnabled: containsStealingTrafficDebugModeEnabled,
			isGenericService:                        false,
		})
	})
962
	if err != nil {
963
964
		return
	}
965
	modified = modified || modified_
966
967
968
	return
}

969
970
func (r *DynamoComponentDeploymentReconciler) createOrUpdateOrDeleteIngress(ctx context.Context, opt generateResourceOption) (bool, error) {
	modified, _, err := commonController.SyncResource(ctx, r, opt.dynamoComponentDeployment, func(ctx context.Context) (*networkingv1.Ingress, bool, error) {
971
972
		return r.generateIngress(ctx, opt)
	})
973
	if err != nil {
974
		return false, err
975
	}
976
	if r.Config.IngressConfig.UseVirtualService() {
977
978
979
980
981
982
983
		modified_, _, err := commonController.SyncResource(ctx, r, opt.dynamoComponentDeployment, func(ctx context.Context) (*networkingv1beta1.VirtualService, bool, error) {
			return r.generateVirtualService(ctx, opt)
		})
		if err != nil {
			return false, err
		}
		return modified || modified_, nil
984
	}
985
	return modified, nil
986
987
}

988
func (r *DynamoComponentDeploymentReconciler) generateIngress(ctx context.Context, opt generateResourceOption) (*networkingv1.Ingress, bool, error) {
989
	log := log.FromContext(ctx)
990
991
992
993
	log.Info("Starting generateIngress")

	ingress := &networkingv1.Ingress{
		ObjectMeta: metav1.ObjectMeta{
994
995
			Name:      opt.dynamoComponentDeployment.Name,
			Namespace: opt.dynamoComponentDeployment.Namespace,
996
997
		},
	}
998

999
	if opt.dynamoComponentDeployment.Spec.Ingress == nil || !opt.dynamoComponentDeployment.Spec.Ingress.Enabled || opt.dynamoComponentDeployment.Spec.Ingress.IngressControllerClassName == nil {
1000
1001
		log.Info("Ingress is not enabled")
		return ingress, true, nil
1002
	}
1003
	return dynamo.GenerateComponentIngress(ctx, opt.dynamoComponentDeployment.Name, opt.dynamoComponentDeployment.Namespace, *opt.dynamoComponentDeployment.Spec.Ingress), false, nil
1004
1005
}

1006
func (r *DynamoComponentDeploymentReconciler) generateVirtualService(ctx context.Context, opt generateResourceOption) (*networkingv1beta1.VirtualService, bool, error) {
1007
1008
1009
	log := log.FromContext(ctx)
	log.Info("Starting generateVirtualService")

1010
1011
	vs := &networkingv1beta1.VirtualService{
		ObjectMeta: metav1.ObjectMeta{
1012
1013
			Name:      opt.dynamoComponentDeployment.Name,
			Namespace: opt.dynamoComponentDeployment.Namespace,
1014
		},
1015
1016
	}

1017
	if !opt.dynamoComponentDeployment.Spec.Ingress.IsVirtualServiceEnabled() {
1018
1019
1020
		log.Info("VirtualService is not enabled")
		return vs, true, nil
	}
1021
	return dynamo.GenerateComponentVirtualService(ctx, opt.dynamoComponentDeployment.Name, opt.dynamoComponentDeployment.Namespace, *opt.dynamoComponentDeployment.Spec.Ingress), false, nil
1022
1023
}

1024
func (r *DynamoComponentDeploymentReconciler) getKubeName(dynamoComponentDeployment *v1alpha1.DynamoComponentDeployment, debug bool) string {
1025
	if debug {
1026
		return fmt.Sprintf("%s-d", dynamoComponentDeployment.Name)
1027
	}
1028
	return dynamoComponentDeployment.Name
1029
1030
}

1031
func (r *DynamoComponentDeploymentReconciler) getServiceName(dynamoComponentDeployment *v1alpha1.DynamoComponentDeployment, debug bool) string {
1032
1033
	var kubeName string
	if debug {
1034
		kubeName = fmt.Sprintf("%s-d", dynamoComponentDeployment.Name)
1035
	} else {
1036
		kubeName = fmt.Sprintf("%s-p", dynamoComponentDeployment.Name)
1037
1038
1039
1040
	}
	return kubeName
}

1041
1042
func (r *DynamoComponentDeploymentReconciler) getGenericServiceName(dynamoComponentDeployment *v1alpha1.DynamoComponentDeployment) string {
	return r.getKubeName(dynamoComponentDeployment, false)
1043
1044
}

1045
func (r *DynamoComponentDeploymentReconciler) getKubeLabels(dynamoComponentDeployment *v1alpha1.DynamoComponentDeployment) map[string]string {
1046
1047
1048
1049
1050
1051
1052
1053
1054
	labels := map[string]string{}
	if dynamoComponentDeployment != nil {
		if dynamoComponentDeployment.Spec.Labels != nil {
			maps.Copy(labels, dynamoComponentDeployment.Spec.Labels)
		}
		if dynamoComponentDeployment.Labels != nil {
			maps.Copy(labels, dynamoComponentDeployment.Labels)
		}
		dynamo.AddBaseModelLabel(labels, dynamoComponentDeployment.Spec.ModelRef)
1055
	}
1056
	return labels
1057
1058
}

1059
1060
func (r *DynamoComponentDeploymentReconciler) getKubeAnnotations(dynamoComponentDeployment *v1alpha1.DynamoComponentDeployment) map[string]string {
	annotations := map[string]string{}
1061
1062
1063
1064
1065
1066
1067
1068
	if dynamoComponentDeployment != nil {
		if dynamoComponentDeployment.Spec.Annotations != nil {
			maps.Copy(annotations, dynamoComponentDeployment.Spec.Annotations)
		}
		if dynamoComponentDeployment.Spec.ExtraPodMetadata != nil && dynamoComponentDeployment.Spec.ExtraPodMetadata.Annotations != nil {
			maps.Copy(annotations, dynamoComponentDeployment.Spec.ExtraPodMetadata.Annotations)
		}
		dynamo.AddBaseModelAnnotation(annotations, dynamoComponentDeployment.Spec.ModelRef)
1069
1070
1071
1072
1073
	}
	return annotations
}

//nolint:nakedret
1074
1075
func (r *DynamoComponentDeploymentReconciler) generateDeployment(ctx context.Context, opt generateResourceOption) (kubeDeployment *appsv1.Deployment, toDelete bool, err error) {
	kubeNs := opt.dynamoComponentDeployment.Namespace
1076

1077
	labels := r.getKubeLabels(opt.dynamoComponentDeployment)
1078

1079
	annotations := r.getKubeAnnotations(opt.dynamoComponentDeployment)
1080

1081
	kubeName := r.getKubeName(opt.dynamoComponentDeployment, opt.isStealingTrafficDebugModeEnabled)
1082

1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
	kubeDeployment = &appsv1.Deployment{
		ObjectMeta: metav1.ObjectMeta{
			Name:        kubeName,
			Namespace:   kubeNs,
			Labels:      labels,
			Annotations: annotations,
		},
	}

	if opt.isStealingTrafficDebugModeEnabled && !opt.containsStealingTrafficDebugModeEnabled {
		// if stealing traffic debug mode is enabked but disabled in the deployment, we need to delete the deployment
		return kubeDeployment, true, nil
	}

	// nolint: gosimple
1098
	podTemplateSpec, err := r.generatePodTemplateSpec(ctx, opt, dynamo.RoleMain)
1099
1100
1101
1102
	if err != nil {
		return
	}

1103
	maxSurge, maxUnavailable := getDeploymentRollingUpdateMaxSurgeAndMaxUnavailable(annotations)
1104
1105
1106
1107

	strategy := appsv1.DeploymentStrategy{
		Type: appsv1.RollingUpdateDeploymentStrategyType,
		RollingUpdate: &appsv1.RollingUpdateDeployment{
1108
1109
			MaxSurge:       &maxSurge,
			MaxUnavailable: &maxUnavailable,
1110
1111
1112
		},
	}

1113
	resourceAnnotations := getResourceAnnotations(opt.dynamoComponentDeployment)
1114
1115
	strategyStr := resourceAnnotations[KubeAnnotationDeploymentStrategy]
	if strategyStr != "" {
1116
		strategyType := common.DeploymentStrategy(strategyStr)
1117
		switch strategyType {
1118
		case common.DeploymentStrategyRollingUpdate:
1119
1120
1121
			strategy = appsv1.DeploymentStrategy{
				Type: appsv1.RollingUpdateDeploymentStrategyType,
				RollingUpdate: &appsv1.RollingUpdateDeployment{
1122
1123
					MaxSurge:       &maxSurge,
					MaxUnavailable: &maxUnavailable,
1124
1125
				},
			}
1126
		case common.DeploymentStrategyRecreate:
1127
1128
1129
1130
1131
1132
1133
			strategy = appsv1.DeploymentStrategy{
				Type: appsv1.RecreateDeploymentStrategyType,
			}
		}
	}

	var replicas *int32
1134
	replicas = opt.dynamoComponentDeployment.Spec.Replicas
1135
1136
1137
1138
	if opt.isStealingTrafficDebugModeEnabled {
		replicas = &[]int32{int32(1)}[0]
	}

1139
1140
1141
1142
	kubeDeployment.Spec = appsv1.DeploymentSpec{
		Replicas: replicas,
		Selector: &metav1.LabelSelector{
			MatchLabels: map[string]string{
1143
				commonconsts.KubeLabelDynamoSelector: kubeName,
1144
1145
			},
		},
1146
1147
		Template: *podTemplateSpec,
		Strategy: strategy,
1148
1149
1150
1151
1152
	}

	return
}

1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
func getDeploymentRollingUpdateMaxSurgeAndMaxUnavailable(annotations map[string]string) (intstr.IntOrString, intstr.IntOrString) {
	maxSurge := intstr.FromString("25%")
	maxUnavailable := intstr.FromString("25%")

	if annotations[KubeAnnotationDeploymentRollingUpdateMaxSurge] != "" {
		maxSurge = intstr.Parse(annotations[KubeAnnotationDeploymentRollingUpdateMaxSurge])
	}
	if annotations[KubeAnnotationDeploymentRollingUpdateMaxUnavailable] != "" {
		maxUnavailable = intstr.Parse(annotations[KubeAnnotationDeploymentRollingUpdateMaxUnavailable])
	}

	return maxSurge, maxUnavailable
}

1167
type generateResourceOption struct {
1168
	dynamoComponentDeployment               *v1alpha1.DynamoComponentDeployment
1169
1170
1171
1172
	isStealingTrafficDebugModeEnabled       bool
	containsStealingTrafficDebugModeEnabled bool
	isDebugPodReceiveProductionTraffic      bool
	isGenericService                        bool
1173
	instanceID                              *int
1174
}
1175
1176

//nolint:gocyclo,nakedret
1177
func (r *DynamoComponentDeploymentReconciler) generatePodTemplateSpec(ctx context.Context, opt generateResourceOption, role dynamo.Role) (podTemplateSpec *corev1.PodTemplateSpec, err error) {
1178
	podLabels := r.getKubeLabels(opt.dynamoComponentDeployment)
1179
	if opt.isStealingTrafficDebugModeEnabled {
1180
		podLabels[commonconsts.KubeLabelDynamoDeploymentTargetType] = DeploymentTargetTypeDebug
1181
1182
	}

1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
	// Convert user-provided metrics annotation into controller-managed label
	// By default (no annotation), metrics are enabled
	metricsAnnotationValue := ""
	if opt.dynamoComponentDeployment.Spec.Annotations != nil {
		metricsAnnotationValue = opt.dynamoComponentDeployment.Spec.Annotations[commonconsts.KubeAnnotationEnableMetrics]
	}
	switch metricsAnnotationValue {
	case commonconsts.KubeLabelValueFalse:
		// Explicitly disabled, don't add the label
	default:
		// Any other value (including empty) enables metrics
		podLabels[commonconsts.KubeLabelMetricsEnabled] = commonconsts.KubeLabelValueTrue
	}

1197
1198
1199
	// Add label for the dynamo graph deployment on the pods themselves
	podLabels[commonconsts.KubeLabelDynamoGraphDeploymentName] = opt.dynamoComponentDeployment.Spec.Labels[commonconsts.KubeLabelDynamoGraphDeploymentName]

1200
1201
1202
1203
1204
	// Add component type label if specified
	if opt.dynamoComponentDeployment.Spec.ComponentType != "" {
		podLabels[commonconsts.KubeLabelDynamoComponentType] = opt.dynamoComponentDeployment.Spec.ComponentType
	}

1205
1206
1207
1208
	if opt.dynamoComponentDeployment.Spec.SubComponentType != "" {
		podLabels[commonconsts.KubeLabelDynamoSubComponentType] = opt.dynamoComponentDeployment.Spec.SubComponentType
	}

1209
	podAnnotations := make(map[string]string)
1210

1211
	kubeName := r.getKubeName(opt.dynamoComponentDeployment, opt.isStealingTrafficDebugModeEnabled)
1212

1213
	resourceAnnotations := opt.dynamoComponentDeployment.Spec.Annotations
1214
1215
1216
1217
1218
1219
1220

	if resourceAnnotations == nil {
		resourceAnnotations = make(map[string]string)
	}

	isDebugModeEnabled := checkIfIsDebugModeEnabled(resourceAnnotations)

1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
	// Resolve checkpoint for this component
	var checkpointInfo *checkpoint.CheckpointInfo
	if opt.dynamoComponentDeployment.Spec.Checkpoint != nil && opt.dynamoComponentDeployment.Spec.Checkpoint.Enabled {
		info, err := checkpoint.ResolveCheckpointForService(ctx, r.Client, opt.dynamoComponentDeployment.Namespace, opt.dynamoComponentDeployment.Spec.Checkpoint)
		if err != nil {
			return nil, errors.Wrap(err, "failed to resolve checkpoint")
		}
		checkpointInfo = info
	}

	podSpec, err := dynamo.GenerateBasePodSpecForController(opt.dynamoComponentDeployment, r.DockerSecretRetriever, r.Config, role, commonconsts.MultinodeDeploymentTypeLWS, checkpointInfo)
1232
	if err != nil {
1233
		err = errors.Wrap(err, "failed to generate base pod spec")
1234
1235
1236
		return nil, err
	}

1237
	// Ensure we have at least one container (the main container should be there from GenerateBasePodSpec)
1238
	if len(podSpec.Containers) == 0 {
1239
		return nil, errors.New("no containers found in base pod spec")
1240
1241
	}

1242
	debuggerImage := "python:3.12-slim"
1243
1244
1245
1246
1247
1248
	debuggerImage_ := os.Getenv("INTERNAL_IMAGES_DEBUGGER")
	if debuggerImage_ != "" {
		debuggerImage = debuggerImage_
	}

	if opt.isStealingTrafficDebugModeEnabled || isDebugModeEnabled {
1249
		podSpec.Containers = append(podSpec.Containers, corev1.Container{
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
			Name:  "debugger",
			Image: debuggerImage,
			Command: []string{
				"sleep",
				"infinity",
			},
			SecurityContext: &corev1.SecurityContext{
				Capabilities: &corev1.Capabilities{
					Add: []corev1.Capability{"SYS_PTRACE"},
				},
			},
			Resources: corev1.ResourceRequirements{
				Requests: corev1.ResourceList{
					corev1.ResourceCPU:    resource.MustParse("100m"),
					corev1.ResourceMemory: resource.MustParse("100Mi"),
				},
				Limits: corev1.ResourceList{
					corev1.ResourceCPU:    resource.MustParse("1000m"),
					corev1.ResourceMemory: resource.MustParse("1000Mi"),
				},
			},
			Stdin: true,
			TTY:   true,
		})
	}

1276
	podLabels[commonconsts.KubeLabelDynamoSelector] = kubeName
1277

1278
	extraPodMetadata := opt.dynamoComponentDeployment.Spec.ExtraPodMetadata
1279
1280

	if extraPodMetadata != nil {
1281
1282
		maps.Copy(podAnnotations, extraPodMetadata.Annotations)
		maps.Copy(podLabels, extraPodMetadata.Labels)
1283
1284
	}

1285
1286
1287
1288
1289
1290
	// Propagate restart annotation to pod template to trigger rolling restart
	// This is the same mechanism used by kubectl rollout restart
	if restartAt, exists := resourceAnnotations[commonconsts.RestartAnnotation]; exists {
		podAnnotations[commonconsts.RestartAnnotation] = restartAt
	}

1291
1292
	if podSpec.ServiceAccountName == "" {
		serviceAccounts := &corev1.ServiceAccountList{}
1293
		err = r.List(ctx, serviceAccounts, client.InNamespace(opt.dynamoComponentDeployment.Namespace), client.MatchingLabels{
1294
			commonconsts.KubeLabelDynamoComponentPod: commonconsts.KubeLabelValueTrue,
1295
1296
		})
		if err != nil {
1297
			err = errors.Wrapf(err, "failed to list service accounts in namespace %s", opt.dynamoComponentDeployment.Namespace)
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
			return
		}
		if len(serviceAccounts.Items) > 0 {
			podSpec.ServiceAccountName = serviceAccounts.Items[0].Name
		} else {
			podSpec.ServiceAccountName = DefaultServiceAccountName
		}
	}

	if opt.isStealingTrafficDebugModeEnabled || isDebugModeEnabled {
		podSpec.ShareProcessNamespace = &[]bool{true}[0]
	}

	podTemplateSpec = &corev1.PodTemplateSpec{
		ObjectMeta: metav1.ObjectMeta{
			Labels:      podLabels,
			Annotations: podAnnotations,
		},
1316
		Spec: *podSpec,
1317
1318
1319
1320
1321
	}

	return
}

1322
func (r *DynamoComponentDeploymentReconciler) generateService(opt generateResourceOption) (*corev1.Service, bool, error) {
1323
1324
	var kubeName string
	if opt.isGenericService {
1325
		kubeName = r.getGenericServiceName(opt.dynamoComponentDeployment)
1326
	} else {
1327
		kubeName = r.getServiceName(opt.dynamoComponentDeployment, opt.isStealingTrafficDebugModeEnabled)
1328
1329
	}

1330
	kubeNs := opt.dynamoComponentDeployment.Namespace
1331

1332
	kubeService := &corev1.Service{
1333
1334
1335
1336
1337
1338
		ObjectMeta: metav1.ObjectMeta{
			Name:      kubeName,
			Namespace: kubeNs,
		},
	}

1339
1340
1341
1342
1343
	isK8sDiscovery := r.Config.IsK8sDiscoveryEnabled(opt.dynamoComponentDeployment.Spec.Annotations)

	// if discovery backend is k8s we want to create a service for each component
	// else, only create for the frontend component
	if !opt.isGenericService && !opt.containsStealingTrafficDebugModeEnabled && !(isK8sDiscovery || opt.dynamoComponentDeployment.IsFrontendComponent()) {
1344
		// if it's not the main component or if it's not a generic service and not contains stealing traffic debug mode enabled, we don't need to create the service
1345
1346
1347
		return kubeService, true, nil
	}

1348
	labels := r.getKubeLabels(opt.dynamoComponentDeployment)
1349

1350
1351
	if opt.dynamoComponentDeployment.Spec.DynamoNamespace == nil {
		return nil, false, fmt.Errorf("expected DynamoComponentDeployment %s to have a dynamoNamespace", opt.dynamoComponentDeployment.Name)
1352
1353
	}

1354
	selector := map[string]string{
1355
1356
1357
1358
1359
		commonconsts.KubeLabelDynamoComponentType: opt.dynamoComponentDeployment.Spec.ComponentType,    // e.g. "worker"
		commonconsts.KubeLabelDynamoNamespace:     *opt.dynamoComponentDeployment.Spec.DynamoNamespace, // result of ComputeDynamoNamespace(k8sNamespace, dgdName)
		// The original user provided component name (the service map key, e.g. "VllmDecodeWorker" in the DGD).
		// Needed to disambiguate amongst distinct components with the same component type within a DGD (e.g prefill/decode workers).
		commonconsts.KubeLabelDynamoComponent: opt.dynamoComponentDeployment.Spec.ServiceName,
1360
1361
	}
	// // If using LeaderWorkerSet, modify selector to only target leaders
1362
	if opt.dynamoComponentDeployment.IsMultinode() {
1363
1364
		selector["role"] = "leader"
	}
1365
	if opt.isStealingTrafficDebugModeEnabled {
1366
		selector[commonconsts.KubeLabelDynamoDeploymentTargetType] = DeploymentTargetTypeDebug
1367
	}
1368
1369
	if isK8sDiscovery {
		labels[commonconsts.KubeLabelDynamoDiscoveryBackend] = "kubernetes"
1370
1371
		labels[commonconsts.KubeLabelDynamoDiscoveryEnabled] = commonconsts.KubeLabelValueTrue
	}
1372

1373
	var servicePort corev1.ServicePort
1374
1375
	switch opt.dynamoComponentDeployment.Spec.ComponentType {
	case commonconsts.ComponentTypeFrontend:
1376
1377
1378
1379
1380
1381
		servicePort = corev1.ServicePort{
			Name:       commonconsts.DynamoServicePortName,
			Port:       commonconsts.DynamoServicePort,
			TargetPort: intstr.FromString(commonconsts.DynamoContainerPortName),
			Protocol:   corev1.ProtocolTCP,
		}
1382
1383
1384
1385
1386
1387
1388
1389
1390
1391
	case commonconsts.ComponentTypeEPP:
		// EPP exposes the gRPC endpoint for InferencePool communication
		servicePort = corev1.ServicePort{
			Name:       commonconsts.EPPGRPCPortName,
			Port:       commonconsts.EPPGRPCPort,
			TargetPort: intstr.FromInt(commonconsts.EPPGRPCPort),
			Protocol:   corev1.ProtocolTCP,
		}
	default:
		// Worker and other components use the system port for metrics/health
1392
1393
1394
1395
1396
1397
1398
		servicePort = corev1.ServicePort{
			Name:       commonconsts.DynamoSystemPortName,
			Port:       commonconsts.DynamoSystemPort,
			TargetPort: intstr.FromString(commonconsts.DynamoSystemPortName),
			Protocol:   corev1.ProtocolTCP,
		}
	}
1399
1400
1401

	spec := corev1.ServiceSpec{
		Selector: selector,
1402
		Ports:    []corev1.ServicePort{servicePort},
1403
1404
	}

1405
	annotations := r.getKubeAnnotations(opt.dynamoComponentDeployment)
1406

1407
1408
1409
	kubeService.ObjectMeta.Annotations = annotations
	kubeService.ObjectMeta.Labels = labels
	kubeService.Spec = spec
1410

1411
	return kubeService, false, nil
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
}

type TLSModeOpt string

const (
	TLSModeNone   TLSModeOpt = "none"
	TLSModeAuto   TLSModeOpt = "auto"
	TLSModeStatic TLSModeOpt = "static"
)

type IngressConfig struct {
	ClassName           *string
	Annotations         map[string]string
	Path                string
	PathType            networkingv1.PathType
	TLSMode             TLSModeOpt
	StaticTLSSecretName string
}

// SetupWithManager sets up the controller with the Manager.
1432
func (r *DynamoComponentDeploymentReconciler) SetupWithManager(mgr ctrl.Manager) error {
1433
	m := ctrl.NewControllerManagedBy(mgr).
1434
		For(&v1alpha1.DynamoComponentDeployment{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})).
1435
		Named(commonconsts.ResourceTypeDynamoComponentDeployment).
1436
1437
1438
1439
1440
1441
1442
		Owns(&appsv1.Deployment{}, builder.WithPredicates(predicate.Funcs{
			// ignore creation cause we don't want to be called again after we create the deployment
			CreateFunc:  func(ce event.CreateEvent) bool { return false },
			DeleteFunc:  func(de event.DeleteEvent) bool { return true },
			UpdateFunc:  func(de event.UpdateEvent) bool { return true },
			GenericFunc: func(ge event.GenericEvent) bool { return true },
		})).
1443
1444
1445
		Owns(&corev1.Service{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})).
		Owns(&networkingv1.Ingress{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})).
		Owns(&corev1.PersistentVolumeClaim{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})).
1446
		WithEventFilter(commonController.EphemeralDeploymentEventFilter(r.Config))
1447

1448
	if r.Config.LWS.Enabled {
1449
1450
1451
1452
1453
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463
1464
		m.Owns(&leaderworkersetv1.LeaderWorkerSet{}, builder.WithPredicates(predicate.Funcs{
			// ignore creation cause we don't want to be called again after we create the LeaderWorkerSet
			CreateFunc:  func(ce event.CreateEvent) bool { return false },
			DeleteFunc:  func(de event.DeleteEvent) bool { return true },
			UpdateFunc:  func(de event.UpdateEvent) bool { return true },
			GenericFunc: func(ge event.GenericEvent) bool { return true },
		})).
			Owns(&volcanov1beta1.PodGroup{}, builder.WithPredicates(predicate.Funcs{
				// ignore creation cause we don't want to be called again after we create the LeaderWorkerSet
				CreateFunc:  func(ce event.CreateEvent) bool { return false },
				DeleteFunc:  func(de event.DeleteEvent) bool { return true },
				UpdateFunc:  func(de event.UpdateEvent) bool { return true },
				GenericFunc: func(ge event.GenericEvent) bool { return true },
			}))
	}

1465
	if r.Config.IngressConfig.UseVirtualService() {
1466
1467
		m.Owns(&networkingv1beta1.VirtualService{}, builder.WithPredicates(predicate.GenerationChangedPredicate{}))
	}
1468
	m.Owns(&autoscalingv2.HorizontalPodAutoscaler{})
1469
1470
1471
	// Wrap with metrics collection
	observedReconciler := observability.NewObservedReconciler(r, commonconsts.ResourceTypeDynamoComponentDeployment)
	return m.Complete(observedReconciler)
1472
}
1473
1474
1475
1476

func (r *DynamoComponentDeploymentReconciler) GetRecorder() record.EventRecorder {
	return r.Recorder
}