main.go 30.6 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
 */

package main

import (
23
	"context"
24
25
	"crypto/tls"
	"flag"
26
	"net/url"
27
	"os"
28
	"time"
29
30
31

	// Import all Kubernetes client auth plugins (e.g. Azure, GCP, OIDC, etc.)
	// to ensure that exec-entrypoint and run can make use of them.
32

33
	corev1 "k8s.io/api/core/v1"
34
	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
35
36
	"k8s.io/client-go/discovery/cached/memory"
	"k8s.io/client-go/dynamic"
37
38
	"k8s.io/client-go/informers"
	"k8s.io/client-go/kubernetes"
39
	_ "k8s.io/client-go/plugin/pkg/client/auth"
40
41
	"k8s.io/client-go/restmapper"
	"k8s.io/client-go/scale"
42
	k8sCache "k8s.io/client-go/tools/cache"
43
44
45
46
47
48
49
50
51
52
53
	"sigs.k8s.io/controller-runtime/pkg/cache"

	"k8s.io/apimachinery/pkg/runtime"
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/healthz"
	"sigs.k8s.io/controller-runtime/pkg/log/zap"
	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
	"sigs.k8s.io/controller-runtime/pkg/webhook"

54
55
56
	lwsscheme "sigs.k8s.io/lws/client-go/clientset/versioned/scheme"
	volcanoscheme "volcano.sh/apis/pkg/client/clientset/versioned/scheme"

57
	semver "github.com/Masterminds/semver/v3"
58
59
60
61
62
63
	nvidiacomv1alpha1 "github.com/ai-dynamo/dynamo/deploy/operator/api/v1alpha1"
	"github.com/ai-dynamo/dynamo/deploy/operator/internal/consts"
	"github.com/ai-dynamo/dynamo/deploy/operator/internal/controller"
	commonController "github.com/ai-dynamo/dynamo/deploy/operator/internal/controller_common"
	"github.com/ai-dynamo/dynamo/deploy/operator/internal/modelendpoint"
	"github.com/ai-dynamo/dynamo/deploy/operator/internal/namespace_scope"
64
	"github.com/ai-dynamo/dynamo/deploy/operator/internal/observability"
65
66
67
68
	"github.com/ai-dynamo/dynamo/deploy/operator/internal/rbac"
	"github.com/ai-dynamo/dynamo/deploy/operator/internal/secret"
	"github.com/ai-dynamo/dynamo/deploy/operator/internal/secrets"
	internalwebhook "github.com/ai-dynamo/dynamo/deploy/operator/internal/webhook"
69
	webhookdefaulting "github.com/ai-dynamo/dynamo/deploy/operator/internal/webhook/defaulting"
70
	webhookvalidation "github.com/ai-dynamo/dynamo/deploy/operator/internal/webhook/validation"
71
	grovev1alpha1 "github.com/ai-dynamo/grove/operator/api/core/v1alpha1"
72
	istioclientsetscheme "istio.io/client-go/pkg/clientset/versioned/scheme"
73
	gaiev1 "sigs.k8s.io/gateway-api-inference-extension/api/v1"
74
75
76
77
78
79
80
81
	//+kubebuilder:scaffold:imports
)

var (
	scheme   = runtime.NewScheme()
	setupLog = ctrl.Log.WithName("setup")
)

82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
func createScalesGetter(mgr ctrl.Manager) (scale.ScalesGetter, error) {
	config := mgr.GetConfig()

	// Create kubernetes client for discovery
	kubeClient, err := kubernetes.NewForConfig(config)
	if err != nil {
		return nil, err
	}

	// Create cached discovery client
	cachedDiscovery := memory.NewMemCacheClient(kubeClient.Discovery())

	// Create REST mapper
	restMapper := restmapper.NewDeferredDiscoveryRESTMapper(cachedDiscovery)

	scalesGetter, err := scale.NewForConfig(
		config,
		restMapper,
		dynamic.LegacyAPIPathResolverFunc,
		scale.NewDiscoveryScaleKindResolver(cachedDiscovery),
	)
	if err != nil {
		return nil, err
	}

	return scalesGetter, nil
}

110
111
112
113
func init() {
	utilruntime.Must(clientgoscheme.AddToScheme(scheme))

	utilruntime.Must(nvidiacomv1alpha1.AddToScheme(scheme))
114
115
116
117

	utilruntime.Must(lwsscheme.AddToScheme(scheme))

	utilruntime.Must(volcanoscheme.AddToScheme(scheme))
118
119

	utilruntime.Must(grovev1alpha1.AddToScheme(scheme))
120
121
122
123

	utilruntime.Must(apiextensionsv1.AddToScheme(scheme))

	utilruntime.Must(istioclientsetscheme.AddToScheme(scheme))
124
125

	utilruntime.Must(gaiev1.Install(scheme))
126
127
128
	//+kubebuilder:scaffold:scheme
}

129
//nolint:gocyclo
130
131
132
133
134
135
136
137
func main() {
	var metricsAddr string
	var enableLeaderElection bool
	var probeAddr string
	var secureMetrics bool
	var enableHTTP2 bool
	var restrictedNamespace string
	var leaderElectionID string
138
	var leaderElectionNamespace string
139
140
	var natsAddr string
	var etcdAddr string
141
	var istioVirtualServiceGateway string
142
	var virtualServiceSupportsHTTPS bool
143
	var ingressControllerClassName string
144
145
	var ingressControllerTLSSecretName string
	var ingressHostSuffix string
146
	var groveTerminationDelay time.Duration
147
	var modelExpressURL string
148
	var prometheusEndpoint string
149
150
	var mpiRunSecretName string
	var mpiRunSecretNamespace string
151
	var plannerClusterRoleName string
152
	var dgdrProfilingClusterRoleName string
153
	var eppClusterRoleName string
154
155
156
	var namespaceScopeLeaseDuration time.Duration
	var namespaceScopeLeaseRenewInterval time.Duration
	var operatorVersion string
157
	var discoveryBackend string
158
	var enableWebhooks bool
159
160
161
162
163
164
165
166
167
168
169
	// Checkpoint configuration
	var checkpointEnabled bool
	var checkpointStorageType string
	var checkpointSignalHostPath string
	var checkpointPVCName string
	var checkpointPVCBasePath string
	var checkpointS3URI string
	var checkpointS3CredentialsSecret string
	var checkpointOCIURI string
	var checkpointOCICredentialsSecret string
	var checkpointInitContainerImage string
170
171
	var checkpointReadyForCheckpointFilePath string
	var checkpointRestoreMarkerFilePath string
172
173
174
175
176
177
178
179
180
	flag.StringVar(&metricsAddr, "metrics-bind-address", ":8080", "The address the metric endpoint binds to.")
	flag.StringVar(&probeAddr, "health-probe-bind-address", ":8081", "The address the probe endpoint binds to.")
	flag.BoolVar(&enableLeaderElection, "leader-elect", false,
		"Enable leader election for controller manager. "+
			"Enabling this will ensure there is only one active controller manager.")
	flag.BoolVar(&secureMetrics, "metrics-secure", false,
		"If set the metrics endpoint is served securely")
	flag.BoolVar(&enableHTTP2, "enable-http2", false,
		"If set, HTTP/2 will be enabled for the metrics and webhook servers")
181
182
183
	flag.BoolVar(&enableWebhooks, "enable-webhooks", false,
		"Enable admission webhooks for validation. When enabled, controllers skip validation "+
			"(webhooks handle it). When disabled, controllers perform validation.")
184
185
186
187
	flag.StringVar(&restrictedNamespace, "restrictedNamespace", "",
		"Enable resources filtering, only the resources belonging to the given namespace will be handled.")
	flag.StringVar(&leaderElectionID, "leader-election-id", "", "Leader election id"+
		"Id to use for the leader election.")
188
189
190
	flag.StringVar(&leaderElectionNamespace,
		"leader-election-namespace", "",
		"Namespace where the leader election resource will be created (default: same as operator namespace)")
191
192
	flag.StringVar(&natsAddr, "natsAddr", "", "address of the NATS server")
	flag.StringVar(&etcdAddr, "etcdAddr", "", "address of the etcd server")
193
194
	flag.StringVar(&istioVirtualServiceGateway, "istio-virtual-service-gateway", "",
		"The name of the istio virtual service gateway to use")
195
196
	flag.BoolVar(&virtualServiceSupportsHTTPS, "virtual-service-supports-https", false,
		"If set, assume VirtualService endpoints are HTTPS")
197
198
	flag.StringVar(&ingressControllerClassName, "ingress-controller-class-name", "",
		"The name of the ingress controller class to use")
199
200
201
202
	flag.StringVar(&ingressControllerTLSSecretName, "ingress-controller-tls-secret-name", "",
		"The name of the ingress controller TLS secret to use")
	flag.StringVar(&ingressHostSuffix, "ingress-host-suffix", "",
		"The suffix to use for the ingress host")
203
	flag.DurationVar(&groveTerminationDelay, "grove-termination-delay", consts.DefaultGroveTerminationDelay,
204
		"The termination delay for Grove PodCliqueSets")
205
206
	flag.StringVar(&modelExpressURL, "model-express-url", "",
		"URL of the Model Express server to inject into all pods")
207
208
	flag.StringVar(&prometheusEndpoint, "prometheus-endpoint", "",
		"URL of the Prometheus endpoint to use for metrics")
209
210
211
212
	flag.StringVar(&mpiRunSecretName, "mpi-run-ssh-secret-name", "",
		"Name of the secret containing the SSH key for MPI Run (required)")
	flag.StringVar(&mpiRunSecretNamespace, "mpi-run-ssh-secret-namespace", "",
		"Namespace where the MPI SSH secret is located (required)")
213
214
	flag.StringVar(&plannerClusterRoleName, "planner-cluster-role-name", "",
		"Name of the ClusterRole for planner (cluster-wide mode only)")
215
216
	flag.StringVar(&dgdrProfilingClusterRoleName, "dgdr-profiling-cluster-role-name", "",
		"Name of the ClusterRole for DGDR profiling jobs (cluster-wide mode only)")
217
218
	flag.StringVar(&eppClusterRoleName, "epp-cluster-role-name", "",
		"Name of the ClusterRole for EPP (cluster-wide mode only)")
219
220
221
222
223
224
	flag.DurationVar(&namespaceScopeLeaseDuration, "namespace-scope-lease-duration", 30*time.Second,
		"Duration of namespace scope marker lease before expiration (namespace-restricted mode only)")
	flag.DurationVar(&namespaceScopeLeaseRenewInterval, "namespace-scope-lease-renew-interval", 10*time.Second,
		"Interval for renewing namespace scope marker lease (namespace-restricted mode only)")
	flag.StringVar(&operatorVersion, "operator-version", "unknown",
		"Version of the operator (used in lease holder identity)")
225
226
	flag.StringVar(&discoveryBackend, "discovery-backend", "kubernetes",
		"Discovery backend to use: 'kubernetes' (default, uses Kubernetes API) or 'etcd' (uses ETCD)")
227
228
229
230
231
	// Checkpoint flags
	flag.BoolVar(&checkpointEnabled, "checkpoint-enabled", false,
		"Enable checkpoint/restore functionality")
	flag.StringVar(&checkpointStorageType, "checkpoint-storage-type", commonController.CheckpointStorageTypePVC,
		"Checkpoint storage backend type: pvc, s3, or oci")
232
	flag.StringVar(&checkpointSignalHostPath, "checkpoint-signal-host-path", "/var/lib/chrek/signals",
233
		"Host path for signal files used for checkpoint job coordination")
234
	flag.StringVar(&checkpointPVCName, "checkpoint-pvc-name", "chrek-pvc",
235
236
237
238
239
240
241
242
243
244
245
246
247
		"Name of the PVC for checkpoint storage (used when storage-type=pvc)")
	flag.StringVar(&checkpointPVCBasePath, "checkpoint-pvc-base-path", "/checkpoints",
		"Base path within the PVC for storing checkpoints (used when storage-type=pvc)")
	flag.StringVar(&checkpointS3URI, "checkpoint-s3-uri", "",
		"S3 URI for checkpoint storage: s3://[endpoint/]bucket/prefix (used when storage-type=s3)")
	flag.StringVar(&checkpointS3CredentialsSecret, "checkpoint-s3-credentials-secret", "",
		"Secret name containing AWS credentials (used when storage-type=s3)")
	flag.StringVar(&checkpointOCIURI, "checkpoint-oci-uri", "",
		"OCI URI for checkpoint storage: oci://registry/repository (used when storage-type=oci)")
	flag.StringVar(&checkpointOCICredentialsSecret, "checkpoint-oci-credentials-secret", "",
		"Docker config secret name for OCI registry auth (used when storage-type=oci)")
	flag.StringVar(&checkpointInitContainerImage, "checkpoint-init-container-image", "busybox:latest",
		"Image to use for checkpoint init containers (e.g., signal file cleanup)")
248
249
250
251
252
	flag.StringVar(&checkpointReadyForCheckpointFilePath,
		"checkpoint-ready-for-checkpoint-file-path", "/tmp/ready-for-checkpoint",
		"Path written by the worker container when the model is loaded and ready for checkpointing")
	flag.StringVar(&checkpointRestoreMarkerFilePath, "checkpoint-restore-marker-file-path", "/tmp/dynamo-restored",
		"Path written by restore-entrypoint after successful CRIU restore")
253
254
255
256
257
	opts := zap.Options{
		Development: true,
	}
	opts.BindFlags(flag.CommandLine)
	flag.Parse()
258
	ctrl.SetLogger(zap.New(zap.UseFlagOptions(&opts)))
259

260
261
262
263
264
	if restrictedNamespace == "" && plannerClusterRoleName == "" {
		setupLog.Error(nil, "planner-cluster-role-name is required in cluster-wide mode")
		os.Exit(1)
	}

265
266
267
268
269
270
271
272
	// Validate and normalize operator version to semver
	if _, err := semver.NewVersion(operatorVersion); err != nil {
		setupLog.Info("WARNING: operator-version is not valid semver, falling back to 0.0.0-unknown",
			"provided", operatorVersion, "error", err.Error())
		operatorVersion = "0.0.0-unknown"
	}
	setupLog.Info("Operator version configured", "version", operatorVersion)

273
274
275
	// Validate discoveryBackend value
	if discoveryBackend != "kubernetes" && discoveryBackend != "etcd" {
		setupLog.Error(nil, "invalid discovery-backend value, must be 'kubernetes' or 'etcd'", "value", discoveryBackend)
276
277
		os.Exit(1)
	}
278
	setupLog.Info("Discovery backend configured", "backend", discoveryBackend)
279

280
281
282
283
284
285
286
287
288
	// Validate modelExpressURL if provided
	if modelExpressURL != "" {
		if _, err := url.Parse(modelExpressURL); err != nil {
			setupLog.Error(err, "invalid model-express-url provided", "url", modelExpressURL)
			os.Exit(1)
		}
		setupLog.Info("Model Express URL configured", "url", modelExpressURL)
	}

289
290
291
292
293
294
295
296
297
298
	if mpiRunSecretName == "" {
		setupLog.Error(nil, "mpi-run-ssh-secret-name is required")
		os.Exit(1)
	}

	if mpiRunSecretNamespace == "" {
		setupLog.Error(nil, "mpi-run-ssh-secret-namespace is required")
		os.Exit(1)
	}

299
	ctrlConfig := commonController.Config{
300
		RestrictedNamespace: restrictedNamespace,
301
302
303
304
		Grove: commonController.GroveConfig{
			Enabled:          false, // Will be set after Grove discovery
			TerminationDelay: groveTerminationDelay,
		},
305
306
307
		LWS: commonController.LWSConfig{
			Enabled: false, // Will be set after LWS discovery
		},
308
309
310
		KaiScheduler: commonController.KaiSchedulerConfig{
			Enabled: false, // Will be set after Kai-scheduler discovery
		},
311
312
		EtcdAddress: etcdAddr,
		NatsAddress: natsAddr,
313
314
315
316
317
318
		IngressConfig: commonController.IngressConfig{
			VirtualServiceGateway:      istioVirtualServiceGateway,
			IngressControllerClassName: ingressControllerClassName,
			IngressControllerTLSSecret: ingressControllerTLSSecretName,
			IngressHostSuffix:          ingressHostSuffix,
		},
319
320
		ModelExpressURL:    modelExpressURL,
		PrometheusEndpoint: prometheusEndpoint,
321
322
323
		MpiRun: commonController.MpiRunConfig{
			SecretName: mpiRunSecretName,
		},
324
		RBAC: commonController.RBACConfig{
325
326
			PlannerClusterRoleName:       plannerClusterRoleName,
			DGDRProfilingClusterRoleName: dgdrProfilingClusterRoleName,
327
			EPPClusterRoleName:           eppClusterRoleName,
328
		},
329
		DiscoveryBackend: discoveryBackend,
330
		Checkpoint: commonController.CheckpointConfig{
331
332
333
334
			Enabled:                    checkpointEnabled,
			InitContainerImage:         checkpointInitContainerImage,
			ReadyForCheckpointFilePath: checkpointReadyForCheckpointFilePath,
			RestoreMarkerFilePath:      checkpointRestoreMarkerFilePath,
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
			Storage: commonController.CheckpointStorageConfig{
				Type:           checkpointStorageType,
				SignalHostPath: checkpointSignalHostPath,
				PVC: commonController.CheckpointPVCConfig{
					PVCName:  checkpointPVCName,
					BasePath: checkpointPVCBasePath,
				},
				S3: commonController.CheckpointS3Config{
					URI:                  checkpointS3URI,
					CredentialsSecretRef: checkpointS3CredentialsSecret,
				},
				OCI: commonController.CheckpointOCIConfig{
					URI:                  checkpointOCIURI,
					CredentialsSecretRef: checkpointOCICredentialsSecret,
				},
			},
		},
352
353
	}

354
	mainCtx := ctrl.SetupSignalHandler()
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372

	// if the enable-http2 flag is false (the default), http/2 should be disabled
	// due to its vulnerabilities. More specifically, disabling http/2 will
	// prevent from being vulnerable to the HTTP/2 Stream Cancellation and
	// Rapid Reset CVEs. For more information see:
	// - https://github.com/advisories/GHSA-qppj-fm5r-hxr3
	// - https://github.com/advisories/GHSA-4374-p667-p6c8
	disableHTTP2 := func(c *tls.Config) {
		setupLog.Info("disabling http/2")
		c.NextProtos = []string{"http/1.1"}
	}

	tlsOpts := []func(*tls.Config){}
	if !enableHTTP2 {
		tlsOpts = append(tlsOpts, disableHTTP2)
	}

	webhookServer := webhook.NewServer(webhook.Options{
373
374
375
376
377
378
		// Bind to all interfaces so the Service can reach the webhook server
		Host: "0.0.0.0",
		// Must match the port exposed by the manager container and targeted by the Service.
		Port: 9443,
		// Must match the mountPath of the webhook certificate secret in the Deployment.
		CertDir: "/tmp/k8s-webhook-server/serving-certs",
379
380
381
382
383
384
385
386
387
388
		TLSOpts: tlsOpts,
	})

	mgrOpts := ctrl.Options{
		Scheme: scheme,
		Metrics: metricsserver.Options{
			BindAddress:   metricsAddr,
			SecureServing: secureMetrics,
			TLSOpts:       tlsOpts,
		},
389
390
391
392
393
		WebhookServer:           webhookServer,
		HealthProbeBindAddress:  probeAddr,
		LeaderElection:          enableLeaderElection,
		LeaderElectionID:        leaderElectionID,
		LeaderElectionNamespace: leaderElectionNamespace,
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
		// LeaderElectionReleaseOnCancel defines if the leader should step down voluntarily
		// when the Manager ends. This requires the binary to immediately end when the
		// Manager is stopped, otherwise, this setting is unsafe. Setting this significantly
		// speeds up voluntary leader transitions as the new leader don't have to wait
		// LeaseDuration time first.
		//
		// In the default scaffold provided, the program ends immediately after
		// the manager stops, so would be fine to enable this option. However,
		// if you are doing or is intended to do any operation such as perform cleanups
		// after the manager stops then its usage might be unsafe.
		// LeaderElectionReleaseOnCancel: true,
	}
	if restrictedNamespace != "" {
		mgrOpts.Cache.DefaultNamespaces = map[string]cache.Config{
			restrictedNamespace: {},
		}
410
411
412
		setupLog.Info("Restricted namespace configured, launching in restricted mode", "namespace", restrictedNamespace)
	} else {
		setupLog.Info("No restricted namespace configured, launching in cluster-wide mode")
413
414
415
416
417
418
419
	}
	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), mgrOpts)
	if err != nil {
		setupLog.Error(err, "unable to start manager")
		os.Exit(1)
	}

420
421
422
423
	// Initialize observability metrics
	setupLog.Info("Initializing observability metrics")
	observability.InitMetrics()

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
	// Initialize namespace scope mechanism
	var leaseManager *namespace_scope.LeaseManager
	var leaseWatcher *namespace_scope.LeaseWatcher

	if restrictedNamespace != "" {
		// Namespace-restricted mode: Create and maintain namespace scope marker lease
		setupLog.Info("Creating namespace scope marker lease manager",
			"namespace", restrictedNamespace,
			"leaseDuration", namespaceScopeLeaseDuration,
			"renewInterval", namespaceScopeLeaseRenewInterval)

		leaseManager, err = namespace_scope.NewLeaseManager(
			mgr.GetConfig(),
			restrictedNamespace,
			operatorVersion,
			namespaceScopeLeaseDuration,
			namespaceScopeLeaseRenewInterval,
		)
		if err != nil {
			setupLog.Error(err, "unable to create namespace scope marker lease manager")
			os.Exit(1)
		}

		// Start the lease manager
		if err = leaseManager.Start(mainCtx); err != nil {
			setupLog.Error(err, "unable to start namespace scope marker lease manager")
			os.Exit(1)
		}

		// Monitor for fatal lease errors
		// If lease renewal fails repeatedly, we must exit to prevent split-brain
		go func() {
			select {
			case err := <-leaseManager.Errors():
				setupLog.Error(err, "FATAL: Lease manager encountered unrecoverable error, shutting down to prevent split-brain")
				os.Exit(1)
			case <-mainCtx.Done():
				// Normal shutdown, error channel monitoring no longer needed
				return
			}
		}()

		// Ensure lease is released on shutdown
		defer func() {
			shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
			defer cancel()
			if err := leaseManager.Stop(shutdownCtx); err != nil {
				setupLog.Error(err, "failed to stop lease manager cleanly")
			}
		}()

		setupLog.Info("Namespace scope marker lease manager started successfully")
	} else {
		// Cluster-wide mode: Watch for namespace scope marker leases
		setupLog.Info("Setting up namespace scope marker lease watcher for cluster-wide mode")

		leaseWatcher, err = namespace_scope.NewLeaseWatcher(mgr.GetConfig())
		if err != nil {
			setupLog.Error(err, "unable to create namespace scope marker lease watcher")
			os.Exit(1)
		}

		// Start the lease watcher
		if err = leaseWatcher.Start(mainCtx); err != nil {
			setupLog.Error(err, "unable to start namespace scope marker lease watcher")
			os.Exit(1)
		}

		setupLog.Info("Namespace scope marker lease watcher started successfully")
493
494
495

		// Pass leaseWatcher to controller config for namespace exclusion filtering
		ctrlConfig.ExcludedNamespaces = leaseWatcher
496
497
	}

498
499
500
	// Start resource counter background goroutine (after ExcludedNamespaces is set)
	setupLog.Info("Starting resource counter")
	go observability.StartResourceCounter(mainCtx, mgr.GetClient(), ctrlConfig.ExcludedNamespaces)
501

502
	// Detect orchestrators availability using discovery client
503
504
505
	setupLog.Info("Detecting Grove availability...")
	groveEnabled := commonController.DetectGroveAvailability(mainCtx, mgr)
	ctrlConfig.Grove.Enabled = groveEnabled
506
507
	setupLog.Info("Detecting LWS availability...")
	lwsEnabled := commonController.DetectLWSAvailability(mainCtx, mgr)
508
509
510
511
	setupLog.Info("Detecting Volcano availability...")
	volcanoEnabled := commonController.DetectVolcanoAvailability(mainCtx, mgr)
	// LWS for multinode deployment usage depends on both LWS and Volcano availability
	ctrlConfig.LWS.Enabled = lwsEnabled && volcanoEnabled
512
513
514
515
516
	// Detect Kai-scheduler availability using discovery client
	setupLog.Info("Detecting Kai-scheduler availability...")
	kaiSchedulerEnabled := commonController.DetectKaiSchedulerAvailability(mainCtx, mgr)
	ctrlConfig.KaiScheduler.Enabled = kaiSchedulerEnabled

517
518
519
520
521
522
523
	setupLog.Info("Detected orchestrators availability",
		"grove", groveEnabled,
		"lws", lwsEnabled,
		"volcano", volcanoEnabled,
		"kai-scheduler", kaiSchedulerEnabled,
	)

524
525
526
527
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
575
576
577
578
579
580
581
582
583
584
585
586
	dockerSecretRetriever := secrets.NewDockerSecretIndexer(mgr.GetClient())
	// refresh whenever a secret is created/deleted/updated
	// Set up informer
	var factory informers.SharedInformerFactory
	if restrictedNamespace == "" {
		factory = informers.NewSharedInformerFactory(kubernetes.NewForConfigOrDie(mgr.GetConfig()), time.Hour*24)
	} else {
		factory = informers.NewFilteredSharedInformerFactory(
			kubernetes.NewForConfigOrDie(mgr.GetConfig()),
			time.Hour*24,
			restrictedNamespace,
			nil,
		)
	}
	secretInformer := factory.Core().V1().Secrets().Informer()
	// Start the informer factory
	go factory.Start(mainCtx.Done())
	// Wait for the initial sync
	if !k8sCache.WaitForCacheSync(mainCtx.Done(), secretInformer.HasSynced) {
		setupLog.Error(nil, "Failed to sync informer cache")
		os.Exit(1)
	}
	setupLog.Info("Secret informer cache synced and ready")
	_, err = secretInformer.AddEventHandler(k8sCache.ResourceEventHandlerFuncs{
		AddFunc: func(obj interface{}) {
			secret := obj.(*corev1.Secret)
			if secret.Type == corev1.SecretTypeDockerConfigJson {
				setupLog.Info("refreshing docker secrets index after secret creation...")
				err := dockerSecretRetriever.RefreshIndex(context.Background())
				if err != nil {
					setupLog.Error(err, "unable to refresh docker secrets index after secret creation")
				} else {
					setupLog.Info("docker secrets index refreshed after secret creation")
				}
			}
		},
		UpdateFunc: func(old, new interface{}) {
			newSecret := new.(*corev1.Secret)
			if newSecret.Type == corev1.SecretTypeDockerConfigJson {
				setupLog.Info("refreshing docker secrets index after secret update...")
				err := dockerSecretRetriever.RefreshIndex(context.Background())
				if err != nil {
					setupLog.Error(err, "unable to refresh docker secrets index after secret update")
				} else {
					setupLog.Info("docker secrets index refreshed after secret update")
				}
			}
		},
		DeleteFunc: func(obj interface{}) {
			secret := obj.(*corev1.Secret)
			if secret.Type == corev1.SecretTypeDockerConfigJson {
				setupLog.Info("refreshing docker secrets index after secret deletion...")
				err := dockerSecretRetriever.RefreshIndex(context.Background())
				if err != nil {
					setupLog.Error(err, "unable to refresh docker secrets index after secret deletion")
				} else {
					setupLog.Info("docker secrets index refreshed after secret deletion")
				}
			}
		},
	})
	if err != nil {
		setupLog.Error(err, "unable to add event handler to secret informer")
587
588
		os.Exit(1)
	}
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
	// launch a goroutine to refresh the docker secret indexer in any case every minute
	go func() {
		// Initial refresh
		if err := dockerSecretRetriever.RefreshIndex(context.Background()); err != nil {
			setupLog.Error(err, "initial docker secrets index refresh failed")
		}
		ticker := time.NewTicker(60 * time.Second)
		defer ticker.Stop()
		for {
			select {
			case <-mainCtx.Done():
				return
			case <-ticker.C:
				setupLog.Info("refreshing docker secrets index...")
				if err := dockerSecretRetriever.RefreshIndex(mainCtx); err != nil {
					setupLog.Error(err, "unable to refresh docker secrets index")
				}
				setupLog.Info("docker secrets index refreshed")
			}
		}
	}()
610
611
612
613
614
615
616
617

	// Create MPI SSH SecretReplicator for cross-namespace secret replication
	mpiSecretReplicator := secret.NewSecretReplicator(
		mgr.GetClient(),
		mpiRunSecretNamespace,
		mpiRunSecretName,
	)

618
619
620
621
622
	if err = (&controller.DynamoComponentDeploymentReconciler{
		Client:                mgr.GetClient(),
		Recorder:              mgr.GetEventRecorderFor("dynamocomponentdeployment"),
		Config:                ctrlConfig,
		DockerSecretRetriever: dockerSecretRetriever,
623
	}).SetupWithManager(mgr); err != nil {
624
		setupLog.Error(err, "unable to create controller", "controller", "DynamoComponentDeployment")
625
626
		os.Exit(1)
	}
627
628
629
630
631
632
633
	// Create scale client for Grove resource scaling
	scaleClient, err := createScalesGetter(mgr)
	if err != nil {
		setupLog.Error(err, "unable to create scale client")
		os.Exit(1)
	}

634
635
636
	// Initialize RBAC manager for cross-namespace resource management
	rbacManager := rbac.NewManager(mgr.GetClient())

637
	if err = (&controller.DynamoGraphDeploymentReconciler{
638
639
640
641
		Client:                mgr.GetClient(),
		Recorder:              mgr.GetEventRecorderFor("dynamographdeployment"),
		Config:                ctrlConfig,
		DockerSecretRetriever: dockerSecretRetriever,
642
		ScaleClient:           scaleClient,
643
		MPISecretReplicator:   mpiSecretReplicator,
644
		RBACManager:           rbacManager,
645
	}).SetupWithManager(mgr); err != nil {
646
		setupLog.Error(err, "unable to create controller", "controller", "DynamoGraphDeployment")
647
648
		os.Exit(1)
	}
649

650
651
652
653
654
655
656
657
658
659
	if err = (&controller.DynamoGraphDeploymentScalingAdapterReconciler{
		Client:   mgr.GetClient(),
		Scheme:   mgr.GetScheme(),
		Recorder: mgr.GetEventRecorderFor("dgdscalingadapter"),
		Config:   ctrlConfig,
	}).SetupWithManager(mgr); err != nil {
		setupLog.Error(err, "unable to create controller", "controller", "DGDScalingAdapter")
		os.Exit(1)
	}

660
	if err = (&controller.DynamoGraphDeploymentRequestReconciler{
661
662
663
664
		Client:      mgr.GetClient(),
		Recorder:    mgr.GetEventRecorderFor("dynamographdeploymentrequest"),
		Config:      ctrlConfig,
		RBACManager: rbacManager,
665
666
667
668
	}).SetupWithManager(mgr); err != nil {
		setupLog.Error(err, "unable to create controller", "controller", "DynamoGraphDeploymentRequest")
		os.Exit(1)
	}
669
670
671
672
673

	if err = (&controller.DynamoModelReconciler{
		Client:         mgr.GetClient(),
		Recorder:       mgr.GetEventRecorderFor("dynamomodel"),
		EndpointClient: modelendpoint.NewClient(),
674
		Config:         ctrlConfig,
675
676
677
678
	}).SetupWithManager(mgr); err != nil {
		setupLog.Error(err, "unable to create controller", "controller", "DynamoModel")
		os.Exit(1)
	}
679

680
681
682
683
684
685
686
687
688
	if err = (&controller.CheckpointReconciler{
		Client:   mgr.GetClient(),
		Config:   ctrlConfig,
		Recorder: mgr.GetEventRecorderFor("checkpoint"),
	}).SetupWithManager(mgr); err != nil {
		setupLog.Error(err, "unable to create controller", "controller", "DynamoCheckpoint")
		os.Exit(1)
	}

689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
	// Set webhooks enabled flag in config
	ctrlConfig.WebhooksEnabled = enableWebhooks

	if enableWebhooks {
		setupLog.Info("Webhooks are enabled - webhooks will validate, controllers will skip validation")
	} else {
		setupLog.Info("Webhooks are disabled - controllers will validate (defense in depth)")
	}

	// Configure webhooks with lease-based namespace exclusion (only if enabled)
	// In cluster-wide mode, inject ctrlConfig.ExcludedNamespaces (leaseWatcher) so webhooks can defer
	// to namespace-restricted operators. In namespace-restricted mode, webhooks validate without checking
	// leases (ExcludedNamespaces is nil). The webhooks use LeaseAwareValidator wrapper to add coordination.
	if enableWebhooks {
		if ctrlConfig.RestrictedNamespace == "" {
			// Cluster-wide mode: inject the same ExcludedNamespaces used by controllers
			setupLog.Info("Configuring webhooks with lease-based namespace exclusion for cluster-wide mode")
			internalwebhook.SetExcludedNamespaces(ctrlConfig.ExcludedNamespaces)
		} else {
			// Namespace-restricted mode: no exclusion checking needed (validators not wrapped)
			setupLog.Info("Configuring webhooks for namespace-restricted mode (no lease checking)",
				"restrictedNamespace", ctrlConfig.RestrictedNamespace)
			internalwebhook.SetExcludedNamespaces(nil)
		}

		// Register validation webhook handlers
		setupLog.Info("Registering validation webhooks")

		dcdHandler := webhookvalidation.NewDynamoComponentDeploymentHandler()
		if err = dcdHandler.RegisterWithManager(mgr); err != nil {
			setupLog.Error(err, "unable to register webhook", "webhook", "DynamoComponentDeployment")
			os.Exit(1)
		}

723
		dgdHandler := webhookvalidation.NewDynamoGraphDeploymentHandler(mgr)
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
		if err = dgdHandler.RegisterWithManager(mgr); err != nil {
			setupLog.Error(err, "unable to register webhook", "webhook", "DynamoGraphDeployment")
			os.Exit(1)
		}

		dmHandler := webhookvalidation.NewDynamoModelHandler()
		if err = dmHandler.RegisterWithManager(mgr); err != nil {
			setupLog.Error(err, "unable to register webhook", "webhook", "DynamoModel")
			os.Exit(1)
		}

		isClusterWide := ctrlConfig.RestrictedNamespace == ""
		dgdrHandler := webhookvalidation.NewDynamoGraphDeploymentRequestHandler(isClusterWide)
		if err = dgdrHandler.RegisterWithManager(mgr); err != nil {
			setupLog.Error(err, "unable to register webhook", "webhook", "DynamoGraphDeploymentRequest")
			os.Exit(1)
		}

		setupLog.Info("Validation webhooks registered successfully")
743
744
745
746
747
748
749
750
751
752
753

		// Register defaulting (mutating) webhook handlers
		setupLog.Info("Registering defaulting webhooks")

		dgdDefaulter := webhookdefaulting.NewDGDDefaulter(operatorVersion)
		if err = dgdDefaulter.RegisterWithManager(mgr); err != nil {
			setupLog.Error(err, "unable to register webhook", "webhook", "DynamoGraphDeployment-defaulting")
			os.Exit(1)
		}

		setupLog.Info("Defaulting webhooks registered successfully")
754
	}
755
756
757
758
759
760
761
762
763
764
765
766
	//+kubebuilder:scaffold:builder

	if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {
		setupLog.Error(err, "unable to set up health check")
		os.Exit(1)
	}
	if err := mgr.AddReadyzCheck("readyz", healthz.Ping); err != nil {
		setupLog.Error(err, "unable to set up ready check")
		os.Exit(1)
	}

	setupLog.Info("starting manager")
767
	if err := mgr.Start(mainCtx); err != nil {
768
769
770
771
		setupLog.Error(err, "problem running manager")
		os.Exit(1)
	}
}