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

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
	clientv3 "go.etcd.io/etcd/client/v3"
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
	grovev1alpha1 "github.com/NVIDIA/grove/operator/api/core/v1alpha1"
58
	nvidiacomv1alpha1 "github.com/ai-dynamo/dynamo/deploy/cloud/operator/api/v1alpha1"
59
	"github.com/ai-dynamo/dynamo/deploy/cloud/operator/internal/consts"
60
61
62
	"github.com/ai-dynamo/dynamo/deploy/cloud/operator/internal/controller"
	commonController "github.com/ai-dynamo/dynamo/deploy/cloud/operator/internal/controller_common"
	"github.com/ai-dynamo/dynamo/deploy/cloud/operator/internal/etcd"
63
	"github.com/ai-dynamo/dynamo/deploy/cloud/operator/internal/modelendpoint"
64
	"github.com/ai-dynamo/dynamo/deploy/cloud/operator/internal/namespace_scope"
65
	"github.com/ai-dynamo/dynamo/deploy/cloud/operator/internal/rbac"
66
	"github.com/ai-dynamo/dynamo/deploy/cloud/operator/internal/secret"
67
	"github.com/ai-dynamo/dynamo/deploy/cloud/operator/internal/secrets"
68
69
70
71
72
73
74
75
76
	istioclientsetscheme "istio.io/client-go/pkg/clientset/versioned/scheme"
	//+kubebuilder:scaffold:imports
)

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

77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
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
}

105
106
107
108
func init() {
	utilruntime.Must(clientgoscheme.AddToScheme(scheme))

	utilruntime.Must(nvidiacomv1alpha1.AddToScheme(scheme))
109
110
111
112

	utilruntime.Must(lwsscheme.AddToScheme(scheme))

	utilruntime.Must(volcanoscheme.AddToScheme(scheme))
113
114

	utilruntime.Must(grovev1alpha1.AddToScheme(scheme))
115
116
117
118

	utilruntime.Must(apiextensionsv1.AddToScheme(scheme))

	utilruntime.Must(istioclientsetscheme.AddToScheme(scheme))
119
120
121
	//+kubebuilder:scaffold:scheme
}

122
//nolint:gocyclo
123
124
125
126
127
128
129
130
func main() {
	var metricsAddr string
	var enableLeaderElection bool
	var probeAddr string
	var secureMetrics bool
	var enableHTTP2 bool
	var restrictedNamespace string
	var leaderElectionID string
131
	var leaderElectionNamespace string
132
133
	var natsAddr string
	var etcdAddr string
134
	var istioVirtualServiceGateway string
135
	var virtualServiceSupportsHTTPS bool
136
	var ingressControllerClassName string
137
138
	var ingressControllerTLSSecretName string
	var ingressHostSuffix string
139
	var groveTerminationDelay time.Duration
140
	var modelExpressURL string
141
	var prometheusEndpoint string
142
143
	var mpiRunSecretName string
	var mpiRunSecretNamespace string
144
	var plannerClusterRoleName string
145
	var dgdrProfilingClusterRoleName string
146
147
148
	var namespaceScopeLeaseDuration time.Duration
	var namespaceScopeLeaseRenewInterval time.Duration
	var operatorVersion string
149
150
151
152
153
154
155
156
157
158
159
160
161
	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")
	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.")
162
163
164
	flag.StringVar(&leaderElectionNamespace,
		"leader-election-namespace", "",
		"Namespace where the leader election resource will be created (default: same as operator namespace)")
165
166
	flag.StringVar(&natsAddr, "natsAddr", "", "address of the NATS server")
	flag.StringVar(&etcdAddr, "etcdAddr", "", "address of the etcd server")
167
168
	flag.StringVar(&istioVirtualServiceGateway, "istio-virtual-service-gateway", "",
		"The name of the istio virtual service gateway to use")
169
170
	flag.BoolVar(&virtualServiceSupportsHTTPS, "virtual-service-supports-https", false,
		"If set, assume VirtualService endpoints are HTTPS")
171
172
	flag.StringVar(&ingressControllerClassName, "ingress-controller-class-name", "",
		"The name of the ingress controller class to use")
173
174
175
176
	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")
177
	flag.DurationVar(&groveTerminationDelay, "grove-termination-delay", consts.DefaultGroveTerminationDelay,
178
		"The termination delay for Grove PodCliqueSets")
179
180
	flag.StringVar(&modelExpressURL, "model-express-url", "",
		"URL of the Model Express server to inject into all pods")
181
182
	flag.StringVar(&prometheusEndpoint, "prometheus-endpoint", "",
		"URL of the Prometheus endpoint to use for metrics")
183
184
185
186
	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)")
187
188
	flag.StringVar(&plannerClusterRoleName, "planner-cluster-role-name", "",
		"Name of the ClusterRole for planner (cluster-wide mode only)")
189
190
	flag.StringVar(&dgdrProfilingClusterRoleName, "dgdr-profiling-cluster-role-name", "",
		"Name of the ClusterRole for DGDR profiling jobs (cluster-wide mode only)")
191
192
193
194
195
196
	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)")
197
198
199
200
201
202
	opts := zap.Options{
		Development: true,
	}
	opts.BindFlags(flag.CommandLine)
	flag.Parse()

203
204
205
206
207
	if restrictedNamespace == "" && plannerClusterRoleName == "" {
		setupLog.Error(nil, "planner-cluster-role-name is required in cluster-wide mode")
		os.Exit(1)
	}

208
209
210
211
212
213
214
215
216
	// 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)
	}

217
218
219
220
221
222
223
224
225
226
	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)
	}

227
	ctrlConfig := commonController.Config{
228
		RestrictedNamespace: restrictedNamespace,
229
230
231
232
		Grove: commonController.GroveConfig{
			Enabled:          false, // Will be set after Grove discovery
			TerminationDelay: groveTerminationDelay,
		},
233
234
235
		LWS: commonController.LWSConfig{
			Enabled: false, // Will be set after LWS discovery
		},
236
237
238
		KaiScheduler: commonController.KaiSchedulerConfig{
			Enabled: false, // Will be set after Kai-scheduler discovery
		},
239
240
		EtcdAddress: etcdAddr,
		NatsAddress: natsAddr,
241
242
243
244
245
246
		IngressConfig: commonController.IngressConfig{
			VirtualServiceGateway:      istioVirtualServiceGateway,
			IngressControllerClassName: ingressControllerClassName,
			IngressControllerTLSSecret: ingressControllerTLSSecretName,
			IngressHostSuffix:          ingressHostSuffix,
		},
247
248
		ModelExpressURL:    modelExpressURL,
		PrometheusEndpoint: prometheusEndpoint,
249
250
251
		MpiRun: commonController.MpiRunConfig{
			SecretName: mpiRunSecretName,
		},
252
		RBAC: commonController.RBACConfig{
253
254
			PlannerClusterRoleName:       plannerClusterRoleName,
			DGDRProfilingClusterRoleName: dgdrProfilingClusterRoleName,
255
		},
256
257
	}

258
	mainCtx := ctrl.SetupSignalHandler()
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
	ctrl.SetLogger(zap.New(zap.UseFlagOptions(&opts)))

	// 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{
		TLSOpts: tlsOpts,
	})

	mgrOpts := ctrl.Options{
		Scheme: scheme,
		Metrics: metricsserver.Options{
			BindAddress:   metricsAddr,
			SecureServing: secureMetrics,
			TLSOpts:       tlsOpts,
		},
288
289
290
291
292
		WebhookServer:           webhookServer,
		HealthProbeBindAddress:  probeAddr,
		LeaderElection:          enableLeaderElection,
		LeaderElectionID:        leaderElectionID,
		LeaderElectionNamespace: leaderElectionNamespace,
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
		// 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: {},
		}
309
310
311
		setupLog.Info("Restricted namespace configured, launching in restricted mode", "namespace", restrictedNamespace)
	} else {
		setupLog.Info("No restricted namespace configured, launching in cluster-wide mode")
312
313
314
315
316
317
318
	}
	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), mgrOpts)
	if err != nil {
		setupLog.Error(err, "unable to start manager")
		os.Exit(1)
	}

319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
	// 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")
	}

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

393
	// Detect orchestrators availability using discovery client
394
395
396
	setupLog.Info("Detecting Grove availability...")
	groveEnabled := commonController.DetectGroveAvailability(mainCtx, mgr)
	ctrlConfig.Grove.Enabled = groveEnabled
397
398
	setupLog.Info("Detecting LWS availability...")
	lwsEnabled := commonController.DetectLWSAvailability(mainCtx, mgr)
399
400
401
402
	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
403
404
405
406
407
	// Detect Kai-scheduler availability using discovery client
	setupLog.Info("Detecting Kai-scheduler availability...")
	kaiSchedulerEnabled := commonController.DetectKaiSchedulerAvailability(mainCtx, mgr)
	ctrlConfig.KaiScheduler.Enabled = kaiSchedulerEnabled

408
409
410
411
412
413
414
	setupLog.Info("Detected orchestrators availability",
		"grove", groveEnabled,
		"lws", lwsEnabled,
		"volcano", volcanoEnabled,
		"kai-scheduler", kaiSchedulerEnabled,
	)

415
416
417
418
419
420
421
422
423
424
425
	// Create etcd client
	cli, err := clientv3.New(clientv3.Config{
		Endpoints:            []string{etcdAddr},
		DialTimeout:          5 * time.Second,
		DialKeepAliveTime:    10 * time.Second,
		DialKeepAliveTimeout: 3 * time.Second,
	})
	if err != nil {
		setupLog.Error(err, "unable to create etcd client")
		os.Exit(1)
	}
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

	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")
490
491
		os.Exit(1)
	}
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
	// 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")
			}
		}
	}()
513
514
515
516
517
518
519
520

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

521
522
523
524
525
526
	if err = (&controller.DynamoComponentDeploymentReconciler{
		Client:                mgr.GetClient(),
		Recorder:              mgr.GetEventRecorderFor("dynamocomponentdeployment"),
		Config:                ctrlConfig,
		EtcdStorage:           etcd.NewStorage(cli),
		DockerSecretRetriever: dockerSecretRetriever,
527
	}).SetupWithManager(mgr); err != nil {
528
		setupLog.Error(err, "unable to create controller", "controller", "DynamoComponentDeployment")
529
530
		os.Exit(1)
	}
531
532
533
534
535
536
537
	// 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)
	}

538
539
540
	// Initialize RBAC manager for cross-namespace resource management
	rbacManager := rbac.NewManager(mgr.GetClient())

541
	if err = (&controller.DynamoGraphDeploymentReconciler{
542
543
544
545
		Client:                mgr.GetClient(),
		Recorder:              mgr.GetEventRecorderFor("dynamographdeployment"),
		Config:                ctrlConfig,
		DockerSecretRetriever: dockerSecretRetriever,
546
		ScaleClient:           scaleClient,
547
		MPISecretReplicator:   mpiSecretReplicator,
548
		RBACManager:           rbacManager,
549
	}).SetupWithManager(mgr); err != nil {
550
		setupLog.Error(err, "unable to create controller", "controller", "DynamoGraphDeployment")
551
552
		os.Exit(1)
	}
553
554

	if err = (&controller.DynamoGraphDeploymentRequestReconciler{
555
556
557
558
		Client:      mgr.GetClient(),
		Recorder:    mgr.GetEventRecorderFor("dynamographdeploymentrequest"),
		Config:      ctrlConfig,
		RBACManager: rbacManager,
559
560
561
562
	}).SetupWithManager(mgr); err != nil {
		setupLog.Error(err, "unable to create controller", "controller", "DynamoGraphDeploymentRequest")
		os.Exit(1)
	}
563
564
565
566
567
568
569
570
571

	if err = (&controller.DynamoModelReconciler{
		Client:         mgr.GetClient(),
		Recorder:       mgr.GetEventRecorderFor("dynamomodel"),
		EndpointClient: modelendpoint.NewClient(),
	}).SetupWithManager(mgr); err != nil {
		setupLog.Error(err, "unable to create controller", "controller", "DynamoModel")
		os.Exit(1)
	}
572
573
574
575
576
577
578
579
580
581
582
583
	//+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")
584
	if err := mgr.Start(mainCtx); err != nil {
585
586
587
588
		setupLog.Error(err, "problem running manager")
		os.Exit(1)
	}
}