dynamocomponentdeployment_controller.go 53.2 KB
Newer Older
1
/*
2
 * SPDX-FileCopyrightText: Copyright (c) 2022 Atalaya Tech. Inc
3
4
5
6
7
8
9
10
11
12
13
14
15
16
 * 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.
17
 * Modifications Copyright (c) 2025 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/cloud/operator/api/v1alpha1"
37
	"github.com/ai-dynamo/dynamo/deploy/cloud/operator/internal/common"
38
39
	commonconsts "github.com/ai-dynamo/dynamo/deploy/cloud/operator/internal/consts"
	commonController "github.com/ai-dynamo/dynamo/deploy/cloud/operator/internal/controller_common"
40
	"github.com/ai-dynamo/dynamo/deploy/cloud/operator/internal/dynamo"
41
	webhookvalidation "github.com/ai-dynamo/dynamo/deploy/cloud/operator/internal/webhook/validation"
42
43
44
45
46
47
48
49
50
51
	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"
52
	"sigs.k8s.io/controller-runtime/pkg/event"
53
54
	"sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/predicate"
55
56
57

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

const (
61
62
63
64
65
66
67
	DefaultClusterName                                   = "default"
	DefaultServiceAccountName                            = "default"
	KubeAnnotationDeploymentStrategy                     = "nvidia.com/deployment-strategy"
	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"
68
69
)

70
71
// DynamoComponentDeploymentReconciler reconciles a DynamoComponentDeployment object
type DynamoComponentDeploymentReconciler struct {
72
	client.Client
73
	Recorder              record.EventRecorder
74
	Config                commonController.Config
75
76
	EtcdStorage           etcdStorage
	DockerSecretRetriever dockerSecretRetriever
77
78
}

79
80
81
// +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
82
83
84
85
86
87
88
89
90
91
92
93
94
95

//+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

96
97
98
// +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

99
100
101
// 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
102
// the DynamoComponentDeployment object against the actual cluster state, and then
103
104
105
106
107
108
109
// 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
110
func (r *DynamoComponentDeploymentReconciler) Reconcile(ctx context.Context, req ctrl.Request) (result ctrl.Result, err error) {
111
112
	logs := log.FromContext(ctx)

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

128
	logs = logs.WithValues("dynamoComponentDeployment", dynamoComponentDeployment.Name, "namespace", dynamoComponentDeployment.Namespace)
129

130
131
132
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
158
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
	// 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)
		if _, validationErr := validator.Validate(); validationErr != nil {
			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
		}
	}

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

206
207
208
209
210
	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,
211
			metav1.Condition{
212
				Type:    v1alpha1.DynamoGraphDeploymentConditionTypeAvailable,
213
214
				Status:  metav1.ConditionUnknown,
				Reason:  "Reconciling",
215
				Message: "Starting to reconcile DynamoComponentDeployment",
216
217
			},
			metav1.Condition{
218
				Type:    v1alpha1.DynamoGraphDeploymentConditionTypeDynamoComponentReady,
219
220
				Status:  metav1.ConditionUnknown,
				Reason:  "Reconciling",
221
				Message: "Starting to reconcile DynamoComponentDeployment",
222
223
224
225
226
227
228
229
230
			},
		)
		if err != nil {
			return
		}
	}

	modified := false

231
232
233
	// Create the appropriate workload resource based on deployment type
	var leaderWorkerSets []*leaderworkersetv1.LeaderWorkerSet
	var deployment *appsv1.Deployment
234
	if r.Config.LWS.Enabled && dynamoComponentDeployment.IsMultinode() {
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
		desiredReplicas := int32(1)
		if dynamoComponentDeployment.Spec.Replicas != nil {
			desiredReplicas = *dynamoComponentDeployment.Spec.Replicas
		}

		anyModified := false

		for i := range int(desiredReplicas) {

			modified_, _, 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 ctrl.Result{}, err
			}

			if modified_ {
				anyModified = true
			}

			modified_, 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 ctrl.Result{}, err
			}

			if modified_ {
				anyModified = true
			}
277

278
279
280
281
			leaderWorkerSets = append(leaderWorkerSets, lwsObj)
		}

		// Clean up any excess LeaderWorkerSets (if replicas were decreased)
282
		baseKubeName := r.getKubeName(dynamoComponentDeployment, false)
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
		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 ctrl.Result{}, err
			}

			err = r.Delete(ctx, lwsToDelete)
			if err != nil {
				return ctrl.Result{}, 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) {
					logs.Error(err, "Failed to get PodGroup for deletion", "podGroupName", podGroupName)
				}
			} else {
				err = r.Delete(ctx, podGroupToDelete)
				if err != nil {
					logs.Error(err, "Failed to delete PodGroup", "podGroupName", podGroupName)
				}
			}

			anyModified = true
		}

		modified = anyModified

	} else {
		modified_, obj, err := r.createOrUpdateOrDeleteDeployments(ctx, generateResourceOption{
329
330
			dynamoComponentDeployment: dynamoComponentDeployment,
		})
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
		if err != nil {
			return ctrl.Result{}, err
		}

		if modified_ {
			modified = true
		}

		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
		}

356
357
358
	}

	// create or update api-server service
359
	modified_, err := r.createOrUpdateOrDeleteServices(ctx, generateResourceOption{
360
		dynamoComponentDeployment: dynamoComponentDeployment,
361
362
363
364
365
366
367
368
369
	})
	if err != nil {
		return
	}

	if modified_ {
		modified = true
	}

370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
	// 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
	}

385
	// create or update api-server ingresses
386
	modified_, err = r.createOrUpdateOrDeleteIngress(ctx, generateResourceOption{
387
		dynamoComponentDeployment: dynamoComponentDeployment,
388
	})
389
390
391
392
393
394
395
396
397
	if err != nil {
		return
	}

	if modified_ {
		modified = true
	}

	if !modified {
398
		r.Recorder.Eventf(dynamoComponentDeployment, corev1.EventTypeNormal, "UpdateDynamoGraphDeployment", "No changes to dynamo deployment %s", dynamoComponentDeployment.Name)
399
400
401
	}

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

404
	if dynamoComponentDeployment.IsMultinode() {
405
406
407
408
409
		err = r.computeAvailableStatusConditionForLeaderWorkerSets(ctx, req, leaderWorkerSets)
	} else {
		err = r.computeAvailableStatusCondition(ctx, req, deployment)
	}

410
411
412
	return
}

413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
// computeAvailableStatusConditionForLeaderWorkerSet updates the status condition based on LeaderWorkerSet readiness
func (r *DynamoComponentDeploymentReconciler) computeAvailableStatusConditionForLeaderWorkerSets(ctx context.Context, req ctrl.Request, leaderWorkerSets []*leaderworkersetv1.LeaderWorkerSet) error {
	logs := log.FromContext(ctx)

	allReady := true
	for _, leaderWorkerSet := range leaderWorkerSets {
		if !IsLeaderWorkerSetReady(leaderWorkerSet) {
			allReady = false
			break
		}
	}

	if allReady {
		logs.Info("All LeaderWorkerSets are ready. Setting available status condition to true.")
		_, err := r.setStatusConditions(ctx, req,
			metav1.Condition{
				Type:    v1alpha1.DynamoGraphDeploymentConditionTypeAvailable,
				Status:  metav1.ConditionTrue,
				Reason:  "AllLeaderWorkerSetsReady",
				Message: "All LeaderWorkerSets are ready",
			},
		)
		return err
	} else {
		logs.Info("Not all LeaderWorkerSets are ready. Setting available status condition to false.")
		_, err := r.setStatusConditions(ctx, req,
			metav1.Condition{
				Type:    v1alpha1.DynamoGraphDeploymentConditionTypeAvailable,
				Status:  metav1.ConditionFalse,
				Reason:  "LeaderWorkerSetsNotReady",
				Message: "Not all LeaderWorkerSets are ready",
			},
		)
		return err
	}
}

// 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)
	}

495
	podGroupName := r.getKubeName(opt.dynamoComponentDeployment, opt.isStealingTrafficDebugModeEnabled)
496
497
498
499
500
501
502
	podGroupName = fmt.Sprintf("%s-%d", podGroupName, instanceID)

	kubeNs := opt.dynamoComponentDeployment.Namespace

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

503
	minMember := opt.dynamoComponentDeployment.GetNumberOfNodes()
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519

	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) {
520
	leaderPodTemplateSpec, err := r.generatePodTemplateSpec(ctx, opt, dynamo.RoleLeader)
521
522
523
524
	if err != nil {
		return nil, errors.Wrap(err, "failed to generate leader pod template")
	}

525
	maps.Copy(leaderPodTemplateSpec.ObjectMeta.Labels, labels)
526
527
528
529
530
531
532
533
534
535
536
	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"

537
	err = checkMainContainer(&leaderPodTemplateSpec.Spec)
538

539
540
	if err != nil {
		return nil, errors.Wrap(err, "generateLeaderPodTemplateSpec: failed to check main container")
541
542
543
544
545
	}

	return leaderPodTemplateSpec, nil
}

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
575
576
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
}

577
func (r *DynamoComponentDeploymentReconciler) generateWorkerPodTemplateSpec(ctx context.Context, opt generateResourceOption, kubeName string, labels map[string]string, instanceID int) (*corev1.PodTemplateSpec, error) {
578
	workerPodTemplateSpec, err := r.generatePodTemplateSpec(ctx, opt, dynamo.RoleWorker)
579
580
581
582
	if err != nil {
		return nil, errors.Wrap(err, "failed to generate worker pod template")
	}

583
	maps.Copy(workerPodTemplateSpec.ObjectMeta.Labels, labels)
584
585
586
587
588
589
590
591
592
593
594
	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

595
	err = checkMainContainer(&workerPodTemplateSpec.Spec)
596

597
598
	if err != nil {
		return nil, errors.Wrap(err, "generateWorkerPodTemplateSpec: failed to check LWS worker main container")
599
600
601
	}

	if opt.dynamoComponentDeployment.Spec.Resources == nil || opt.dynamoComponentDeployment.Spec.Resources.Limits == nil || opt.dynamoComponentDeployment.Spec.Resources.Limits.GPU == "" {
602
		return nil, fmt.Errorf("generateWorkerPodTemplateSpec: GPU limit is not set for LWS worker pod")
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
	}

	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)
	}

622
	kubeName := r.getKubeName(opt.dynamoComponentDeployment, opt.isStealingTrafficDebugModeEnabled)
623
624
625
	kubeName = fmt.Sprintf("%s-%d", kubeName, instanceID)

	kubeNs := opt.dynamoComponentDeployment.Namespace
626
	labels := r.getKubeLabels(opt.dynamoComponentDeployment)
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
657
658
659
660

	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)
661
	groupSize := opt.dynamoComponentDeployment.GetNumberOfNodes()
662
663
664
665
666
667
668
669
670
671
672
673
674
675

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

	return leaderWorkerSet, false, nil
}

676
func (r *DynamoComponentDeploymentReconciler) FinalizeResource(ctx context.Context, dynamoComponentDeployment *v1alpha1.DynamoComponentDeployment) error {
677
	logger := log.FromContext(ctx)
678
679
680
681
	logger.Info("Finalizing the DynamoComponentDeployment", "dynamoComponentDeployment", dynamoComponentDeployment)
	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))
682
		if err != nil {
683
			logger.Error(err, "Failed to delete the etcd keys for the service", "service", dynamoComponentDeployment.Spec.ServiceName, "dynamoNamespace", *dynamoComponentDeployment.Spec.DynamoNamespace)
684
685
686
687
688
689
			return err
		}
	}
	return nil
}

690
func (r *DynamoComponentDeploymentReconciler) computeAvailableStatusCondition(ctx context.Context, req ctrl.Request, deployment *appsv1.Deployment) error {
691
692
693
694
695
	logs := log.FromContext(ctx)
	if IsDeploymentReady(deployment) {
		logs.Info("Deployment is ready. Setting available status condition to true.")
		_, err := r.setStatusConditions(ctx, req,
			metav1.Condition{
696
				Type:    v1alpha1.DynamoGraphDeploymentConditionTypeAvailable,
697
698
699
700
701
702
703
704
705
706
				Status:  metav1.ConditionTrue,
				Reason:  "DeploymentReady",
				Message: "Deployment is ready",
			},
		)
		return err
	} else {
		logs.Info("Deployment is not ready. Setting available status condition to false.")
		_, err := r.setStatusConditions(ctx, req,
			metav1.Condition{
707
				Type:    v1alpha1.DynamoGraphDeploymentConditionTypeAvailable,
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
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
				Status:  metav1.ConditionFalse,
				Reason:  "DeploymentNotReady",
				Message: "Deployment is not ready",
			},
		)
		return err
	}
}

// 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)
	if status.ObservedGeneration < deployment.Generation ||
		status.UpdatedReplicas < desiredReplicas ||
		status.AvailableReplicas < desiredReplicas {
		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
}

758
759
func (r *DynamoComponentDeploymentReconciler) setStatusConditions(ctx context.Context, req ctrl.Request, conditions ...metav1.Condition) (dynamoComponentDeployment *v1alpha1.DynamoComponentDeployment, err error) {
	dynamoComponentDeployment = &v1alpha1.DynamoComponentDeployment{}
760
761
	maxRetries := 3
	for range maxRetries - 1 {
762
763
		if err = r.Get(ctx, req.NamespacedName, dynamoComponentDeployment); err != nil {
			err = errors.Wrap(err, "Failed to re-fetch DynamoComponentDeployment")
764
765
766
			return
		}
		for _, condition := range conditions {
767
			meta.SetStatusCondition(&dynamoComponentDeployment.Status.Conditions, condition)
768
		}
769
		if err = r.Status().Update(ctx, dynamoComponentDeployment); err != nil {
770
771
772
773
774
			if k8serrors.IsConflict(err) {
				time.Sleep(100 * time.Millisecond)
				continue
			}
			break
775
776
777
778
779
		} else {
			break
		}
	}
	if err != nil {
780
		err = errors.Wrap(err, "Failed to update DynamoComponentDeployment status")
781
782
		return
	}
783
784
	if err = r.Get(ctx, req.NamespacedName, dynamoComponentDeployment); err != nil {
		err = errors.Wrap(err, "Failed to re-fetch DynamoComponentDeployment")
785
786
787
788
789
790
		return
	}
	return
}

//nolint:nakedret
791
792
func (r *DynamoComponentDeploymentReconciler) createOrUpdateOrDeleteDeployments(ctx context.Context, opt generateResourceOption) (modified bool, depl *appsv1.Deployment, err error) {
	containsStealingTrafficDebugModeEnabled := checkIfContainsStealingTrafficDebugModeEnabled(opt.dynamoComponentDeployment)
793
	// create the main deployment
794
795
796
797
798
799
800
	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,
		})
	})
801
802
803
804
	if err != nil {
		err = errors.Wrap(err, "create or update deployment")
		return
	}
805
	// create the debug deployment
806
807
808
809
810
811
812
	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,
		})
	})
813
814
	if err != nil {
		err = errors.Wrap(err, "create or update debug deployment")
815
	}
816
	modified = modified || modified2
817
818
819
	return
}

820
821
func getResourceAnnotations(dynamoComponentDeployment *v1alpha1.DynamoComponentDeployment) map[string]string {
	resourceAnnotations := dynamoComponentDeployment.Spec.Annotations
822
823
824
825
826
827
828
829
830
831
832
833
	if resourceAnnotations == nil {
		resourceAnnotations = map[string]string{}
	}

	return resourceAnnotations
}

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

834
	return annotations[KubeAnnotationEnableDebugMode] == commonconsts.KubeLabelValueTrue
835
836
837
838
839
840
841
}

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

842
	return annotations[KubeAnnotationEnableStealingTrafficDebugMode] == commonconsts.KubeLabelValueTrue
843
844
845
846
847
848
849
}

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

850
	return annotations[KubeAnnotationEnableDebugPodReceiveProductionTraffic] == commonconsts.KubeLabelValueTrue
851
852
}

853
854
func checkIfContainsStealingTrafficDebugModeEnabled(dynamoComponentDeployment *v1alpha1.DynamoComponentDeployment) bool {
	return checkIfIsStealingTrafficDebugModeEnabled(dynamoComponentDeployment.Spec.Annotations)
855
856
857
}

//nolint:nakedret
858
859
func (r *DynamoComponentDeploymentReconciler) createOrUpdateOrDeleteServices(ctx context.Context, opt generateResourceOption) (modified bool, err error) {
	resourceAnnotations := getResourceAnnotations(opt.dynamoComponentDeployment)
860
	isDebugPodReceiveProductionTrafficEnabled := checkIfIsDebugPodReceiveProductionTrafficEnabled(resourceAnnotations)
861
	containsStealingTrafficDebugModeEnabled := checkIfContainsStealingTrafficDebugModeEnabled(opt.dynamoComponentDeployment)
862
	// main generic service
863
	modified, _, err = commonController.SyncResource(ctx, r, opt.dynamoComponentDeployment, func(ctx context.Context) (*corev1.Service, bool, error) {
864
		return r.generateService(generateResourceOption{
865
866
867
868
869
870
871
			dynamoComponentDeployment:               opt.dynamoComponentDeployment,
			isStealingTrafficDebugModeEnabled:       false,
			isDebugPodReceiveProductionTraffic:      isDebugPodReceiveProductionTrafficEnabled,
			containsStealingTrafficDebugModeEnabled: containsStealingTrafficDebugModeEnabled,
			isGenericService:                        true,
		})
	})
872
873
874
875
	if err != nil {
		return
	}

876
	// debug production service (if enabled)
877
	modified_, _, err := commonController.SyncResource(ctx, r, opt.dynamoComponentDeployment, func(ctx context.Context) (*corev1.Service, bool, error) {
878
		return r.generateService(generateResourceOption{
879
880
881
882
883
884
885
			dynamoComponentDeployment:               opt.dynamoComponentDeployment,
			isStealingTrafficDebugModeEnabled:       false,
			isDebugPodReceiveProductionTraffic:      isDebugPodReceiveProductionTrafficEnabled,
			containsStealingTrafficDebugModeEnabled: containsStealingTrafficDebugModeEnabled,
			isGenericService:                        false,
		})
	})
886
887
888
	if err != nil {
		return
	}
889
890
	modified = modified || modified_
	// debug service (if enabled)
891
	modified_, _, err = commonController.SyncResource(ctx, r, opt.dynamoComponentDeployment, func(ctx context.Context) (*corev1.Service, bool, error) {
892
		return r.generateService(generateResourceOption{
893
894
895
896
897
898
899
			dynamoComponentDeployment:               opt.dynamoComponentDeployment,
			isStealingTrafficDebugModeEnabled:       true,
			isDebugPodReceiveProductionTraffic:      isDebugPodReceiveProductionTrafficEnabled,
			containsStealingTrafficDebugModeEnabled: containsStealingTrafficDebugModeEnabled,
			isGenericService:                        false,
		})
	})
900
	if err != nil {
901
902
		return
	}
903
	modified = modified || modified_
904
905
906
	return
}

907
908
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) {
909
910
		return r.generateIngress(ctx, opt)
	})
911
	if err != nil {
912
		return false, err
913
	}
914
	if r.Config.IngressConfig.UseVirtualService() {
915
916
917
918
919
920
921
		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
922
	}
923
	return modified, nil
924
925
}

926
func (r *DynamoComponentDeploymentReconciler) generateIngress(ctx context.Context, opt generateResourceOption) (*networkingv1.Ingress, bool, error) {
927
	log := log.FromContext(ctx)
928
929
930
931
	log.Info("Starting generateIngress")

	ingress := &networkingv1.Ingress{
		ObjectMeta: metav1.ObjectMeta{
932
933
			Name:      opt.dynamoComponentDeployment.Name,
			Namespace: opt.dynamoComponentDeployment.Namespace,
934
935
		},
	}
936

937
	if opt.dynamoComponentDeployment.Spec.Ingress == nil || !opt.dynamoComponentDeployment.Spec.Ingress.Enabled || opt.dynamoComponentDeployment.Spec.Ingress.IngressControllerClassName == nil {
938
939
		log.Info("Ingress is not enabled")
		return ingress, true, nil
940
	}
941
	return dynamo.GenerateComponentIngress(ctx, opt.dynamoComponentDeployment.Name, opt.dynamoComponentDeployment.Namespace, *opt.dynamoComponentDeployment.Spec.Ingress), false, nil
942
943
}

944
func (r *DynamoComponentDeploymentReconciler) generateVirtualService(ctx context.Context, opt generateResourceOption) (*networkingv1beta1.VirtualService, bool, error) {
945
946
947
	log := log.FromContext(ctx)
	log.Info("Starting generateVirtualService")

948
949
	vs := &networkingv1beta1.VirtualService{
		ObjectMeta: metav1.ObjectMeta{
950
951
			Name:      opt.dynamoComponentDeployment.Name,
			Namespace: opt.dynamoComponentDeployment.Namespace,
952
		},
953
954
	}

955
	if !opt.dynamoComponentDeployment.Spec.Ingress.IsVirtualServiceEnabled() {
956
957
958
		log.Info("VirtualService is not enabled")
		return vs, true, nil
	}
959
	return dynamo.GenerateComponentVirtualService(ctx, opt.dynamoComponentDeployment.Name, opt.dynamoComponentDeployment.Namespace, *opt.dynamoComponentDeployment.Spec.Ingress), false, nil
960
961
}

962
func (r *DynamoComponentDeploymentReconciler) getKubeName(dynamoComponentDeployment *v1alpha1.DynamoComponentDeployment, debug bool) string {
963
	if debug {
964
		return fmt.Sprintf("%s-d", dynamoComponentDeployment.Name)
965
	}
966
	return dynamoComponentDeployment.Name
967
968
}

969
func (r *DynamoComponentDeploymentReconciler) getServiceName(dynamoComponentDeployment *v1alpha1.DynamoComponentDeployment, debug bool) string {
970
971
	var kubeName string
	if debug {
972
		kubeName = fmt.Sprintf("%s-d", dynamoComponentDeployment.Name)
973
	} else {
974
		kubeName = fmt.Sprintf("%s-p", dynamoComponentDeployment.Name)
975
976
977
978
	}
	return kubeName
}

979
980
func (r *DynamoComponentDeploymentReconciler) getGenericServiceName(dynamoComponentDeployment *v1alpha1.DynamoComponentDeployment) string {
	return r.getKubeName(dynamoComponentDeployment, false)
981
982
}

983
func (r *DynamoComponentDeploymentReconciler) getKubeLabels(dynamoComponentDeployment *v1alpha1.DynamoComponentDeployment) map[string]string {
984
985
986
987
988
989
990
991
992
	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)
993
	}
994
	return labels
995
996
}

997
998
func (r *DynamoComponentDeploymentReconciler) getKubeAnnotations(dynamoComponentDeployment *v1alpha1.DynamoComponentDeployment) map[string]string {
	annotations := map[string]string{}
999
1000
1001
1002
1003
1004
1005
1006
	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)
1007
1008
1009
1010
1011
	}
	return annotations
}

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

1015
	labels := r.getKubeLabels(opt.dynamoComponentDeployment)
1016

1017
	annotations := r.getKubeAnnotations(opt.dynamoComponentDeployment)
1018

1019
	kubeName := r.getKubeName(opt.dynamoComponentDeployment, opt.isStealingTrafficDebugModeEnabled)
1020

1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
	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
1036
	podTemplateSpec, err := r.generatePodTemplateSpec(ctx, opt, dynamo.RoleMain)
1037
1038
1039
1040
	if err != nil {
		return
	}

1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
	defaultMaxSurge := intstr.FromString("25%")
	defaultMaxUnavailable := intstr.FromString("25%")

	strategy := appsv1.DeploymentStrategy{
		Type: appsv1.RollingUpdateDeploymentStrategyType,
		RollingUpdate: &appsv1.RollingUpdateDeployment{
			MaxSurge:       &defaultMaxSurge,
			MaxUnavailable: &defaultMaxUnavailable,
		},
	}

1052
	resourceAnnotations := getResourceAnnotations(opt.dynamoComponentDeployment)
1053
1054
	strategyStr := resourceAnnotations[KubeAnnotationDeploymentStrategy]
	if strategyStr != "" {
1055
		strategyType := common.DeploymentStrategy(strategyStr)
1056
		switch strategyType {
1057
		case common.DeploymentStrategyRollingUpdate:
1058
1059
1060
1061
1062
1063
1064
			strategy = appsv1.DeploymentStrategy{
				Type: appsv1.RollingUpdateDeploymentStrategyType,
				RollingUpdate: &appsv1.RollingUpdateDeployment{
					MaxSurge:       &defaultMaxSurge,
					MaxUnavailable: &defaultMaxUnavailable,
				},
			}
1065
		case common.DeploymentStrategyRecreate:
1066
1067
1068
			strategy = appsv1.DeploymentStrategy{
				Type: appsv1.RecreateDeploymentStrategyType,
			}
1069
		case common.DeploymentStrategyRampedSlowRollout:
1070
1071
1072
1073
1074
1075
1076
			strategy = appsv1.DeploymentStrategy{
				Type: appsv1.RollingUpdateDeploymentStrategyType,
				RollingUpdate: &appsv1.RollingUpdateDeployment{
					MaxSurge:       &[]intstr.IntOrString{intstr.FromInt(1)}[0],
					MaxUnavailable: &[]intstr.IntOrString{intstr.FromInt(0)}[0],
				},
			}
1077
		case common.DeploymentStrategyBestEffortControlledRollout:
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
			strategy = appsv1.DeploymentStrategy{
				Type: appsv1.RollingUpdateDeploymentStrategyType,
				RollingUpdate: &appsv1.RollingUpdateDeployment{
					MaxSurge:       &[]intstr.IntOrString{intstr.FromInt(0)}[0],
					MaxUnavailable: &[]intstr.IntOrString{intstr.FromString("20%")}[0],
				},
			}
		}
	}

	var replicas *int32
1089
	replicas = opt.dynamoComponentDeployment.Spec.Replicas
1090
1091
1092
1093
	if opt.isStealingTrafficDebugModeEnabled {
		replicas = &[]int32{int32(1)}[0]
	}

1094
1095
1096
1097
	kubeDeployment.Spec = appsv1.DeploymentSpec{
		Replicas: replicas,
		Selector: &metav1.LabelSelector{
			MatchLabels: map[string]string{
1098
				commonconsts.KubeLabelDynamoSelector: kubeName,
1099
1100
			},
		},
1101
1102
		Template: *podTemplateSpec,
		Strategy: strategy,
1103
1104
1105
1106
1107
	}

	return
}

1108
type generateResourceOption struct {
1109
	dynamoComponentDeployment               *v1alpha1.DynamoComponentDeployment
1110
1111
1112
1113
	isStealingTrafficDebugModeEnabled       bool
	containsStealingTrafficDebugModeEnabled bool
	isDebugPodReceiveProductionTraffic      bool
	isGenericService                        bool
1114
	instanceID                              *int
1115
}
1116

1117
func (r *DynamoComponentDeploymentReconciler) generateHPA(opt generateResourceOption) (*autoscalingv2.HorizontalPodAutoscaler, bool, error) {
1118
	labels := r.getKubeLabels(opt.dynamoComponentDeployment)
1119

1120
	annotations := r.getKubeAnnotations(opt.dynamoComponentDeployment)
1121

1122
	kubeName := r.getKubeName(opt.dynamoComponentDeployment, false)
1123

1124
	kubeNs := opt.dynamoComponentDeployment.Namespace
1125

1126
	hpaConf := opt.dynamoComponentDeployment.Spec.Autoscaling
1127
1128
1129
1130
1131
1132
1133
1134

	kubeHpa := &autoscalingv2.HorizontalPodAutoscaler{
		ObjectMeta: metav1.ObjectMeta{
			Name:        kubeName,
			Namespace:   kubeNs,
			Labels:      labels,
			Annotations: annotations,
		},
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
	}

	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,
1151
		},
1152
		Metrics: hpaConf.Metrics,
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
	}

	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,
					},
				},
			},
		}
	}

1171
	return kubeHpa, false, nil
1172
1173
1174
}

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

1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
	// 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
	}

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

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

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

1207
	podAnnotations := make(map[string]string)
1208

1209
	kubeName := r.getKubeName(opt.dynamoComponentDeployment, opt.isStealingTrafficDebugModeEnabled)
1210

1211
	resourceAnnotations := opt.dynamoComponentDeployment.Spec.Annotations
1212
1213
1214
1215
1216
1217
1218

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

	isDebugModeEnabled := checkIfIsDebugModeEnabled(resourceAnnotations)

1219
	podSpec, err := dynamo.GenerateBasePodSpecForController(opt.dynamoComponentDeployment, r.DockerSecretRetriever, r.Config, role, commonconsts.MultinodeDeploymentTypeLWS)
1220
	if err != nil {
1221
		err = errors.Wrap(err, "failed to generate base pod spec")
1222
1223
1224
		return nil, err
	}

1225
	// Ensure we have at least one container (the main container should be there from GenerateBasePodSpec)
1226
	if len(podSpec.Containers) == 0 {
1227
		return nil, errors.New("no containers found in base pod spec")
1228
1229
	}

1230
	debuggerImage := "python:3.12-slim"
1231
1232
1233
1234
1235
1236
	debuggerImage_ := os.Getenv("INTERNAL_IMAGES_DEBUGGER")
	if debuggerImage_ != "" {
		debuggerImage = debuggerImage_
	}

	if opt.isStealingTrafficDebugModeEnabled || isDebugModeEnabled {
1237
		podSpec.Containers = append(podSpec.Containers, corev1.Container{
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
			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,
		})
	}

1264
	podLabels[commonconsts.KubeLabelDynamoSelector] = kubeName
1265

1266
	extraPodMetadata := opt.dynamoComponentDeployment.Spec.ExtraPodMetadata
1267
1268

	if extraPodMetadata != nil {
1269
1270
		maps.Copy(podAnnotations, extraPodMetadata.Annotations)
		maps.Copy(podLabels, extraPodMetadata.Labels)
1271
1272
1273
1274
	}

	if podSpec.ServiceAccountName == "" {
		serviceAccounts := &corev1.ServiceAccountList{}
1275
		err = r.List(ctx, serviceAccounts, client.InNamespace(opt.dynamoComponentDeployment.Namespace), client.MatchingLabels{
1276
			commonconsts.KubeLabelDynamoComponentPod: commonconsts.KubeLabelValueTrue,
1277
1278
		})
		if err != nil {
1279
			err = errors.Wrapf(err, "failed to list service accounts in namespace %s", opt.dynamoComponentDeployment.Namespace)
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
			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,
		},
1298
		Spec: *podSpec,
1299
1300
1301
1302
1303
	}

	return
}

1304
func (r *DynamoComponentDeploymentReconciler) generateService(opt generateResourceOption) (*corev1.Service, bool, error) {
1305
1306
	var kubeName string
	if opt.isGenericService {
1307
		kubeName = r.getGenericServiceName(opt.dynamoComponentDeployment)
1308
	} else {
1309
		kubeName = r.getServiceName(opt.dynamoComponentDeployment, opt.isStealingTrafficDebugModeEnabled)
1310
1311
	}

1312
	kubeNs := opt.dynamoComponentDeployment.Namespace
1313

1314
	kubeService := &corev1.Service{
1315
1316
1317
1318
1319
1320
		ObjectMeta: metav1.ObjectMeta{
			Name:      kubeName,
			Namespace: kubeNs,
		},
	}

1321
1322
1323
1324
1325
	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()) {
1326
		// 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
1327
1328
1329
		return kubeService, true, nil
	}

1330
	labels := r.getKubeLabels(opt.dynamoComponentDeployment)
1331

1332
1333
	if opt.dynamoComponentDeployment.Spec.DynamoNamespace == nil {
		return nil, false, fmt.Errorf("expected DynamoComponentDeployment %s to have a dynamoNamespace", opt.dynamoComponentDeployment.Name)
1334
1335
	}

1336
1337
1338
1339
1340
	selector := map[string]string{
		commonconsts.KubeLabelDynamoComponentType: opt.dynamoComponentDeployment.Spec.ComponentType,
		commonconsts.KubeLabelDynamoNamespace:     *opt.dynamoComponentDeployment.Spec.DynamoNamespace,
	}
	// // If using LeaderWorkerSet, modify selector to only target leaders
1341
	if opt.dynamoComponentDeployment.IsMultinode() {
1342
1343
		selector["role"] = "leader"
	}
1344
	if opt.isStealingTrafficDebugModeEnabled {
1345
		selector[commonconsts.KubeLabelDynamoDeploymentTargetType] = DeploymentTargetTypeDebug
1346
	}
1347
1348
1349
	if isK8sDiscovery {
		labels[commonconsts.KubeLabelDynamoDiscoveryBackend] = "kubernetes"
	}
1350
1351
1352
1353
	// Discovery is enabled for non frontend components
	if isK8sDiscovery && !opt.dynamoComponentDeployment.IsFrontendComponent() {
		labels[commonconsts.KubeLabelDynamoDiscoveryEnabled] = commonconsts.KubeLabelValueTrue
	}
1354

1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
	var servicePort corev1.ServicePort
	if opt.dynamoComponentDeployment.IsFrontendComponent() {
		servicePort = corev1.ServicePort{
			Name:       commonconsts.DynamoServicePortName,
			Port:       commonconsts.DynamoServicePort,
			TargetPort: intstr.FromString(commonconsts.DynamoContainerPortName),
			Protocol:   corev1.ProtocolTCP,
		}
	} else { // TODO: only for worker components
		servicePort = corev1.ServicePort{
			Name:       commonconsts.DynamoSystemPortName,
			Port:       commonconsts.DynamoSystemPort,
			TargetPort: intstr.FromString(commonconsts.DynamoSystemPortName),
			Protocol:   corev1.ProtocolTCP,
		}
	}
1371
1372
1373

	spec := corev1.ServiceSpec{
		Selector: selector,
1374
		Ports:    []corev1.ServicePort{servicePort},
1375
1376
	}

1377
	annotations := r.getKubeAnnotations(opt.dynamoComponentDeployment)
1378

1379
1380
1381
	kubeService.ObjectMeta.Annotations = annotations
	kubeService.ObjectMeta.Labels = labels
	kubeService.Spec = spec
1382

1383
	return kubeService, false, nil
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
}

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.
1404
func (r *DynamoComponentDeploymentReconciler) SetupWithManager(mgr ctrl.Manager) error {
1405
	m := ctrl.NewControllerManagedBy(mgr).
1406
		For(&v1alpha1.DynamoComponentDeployment{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})).
1407
1408
1409
1410
1411
1412
1413
		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 },
		})).
1414
1415
1416
		Owns(&corev1.Service{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})).
		Owns(&networkingv1.Ingress{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})).
		Owns(&corev1.PersistentVolumeClaim{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})).
1417
		WithEventFilter(commonController.EphemeralDeploymentEventFilter(r.Config))
1418

1419
	if r.Config.LWS.Enabled {
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
		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 },
			}))
	}

1436
	if r.Config.IngressConfig.UseVirtualService() {
1437
1438
		m.Owns(&networkingv1beta1.VirtualService{}, builder.WithPredicates(predicate.GenerationChangedPredicate{}))
	}
1439
1440
1441
	m.Owns(&autoscalingv2.HorizontalPodAutoscaler{})
	return m.Complete(r)
}
1442
1443
1444
1445

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