[Misc]: move crd check in Initialize part (#949)

move crd check in Initialize

Signed-off-by: googs1025 <[email protected]>

Author: googs1025

cmd/controllers/main.go

@@ -246,7 +246,10 @@ func main() {
 	}
 
 	// Initialize controllers
-	controller.Initialize()
+	if err = controller.Initialize(mgr); err != nil {
+		setupLog.Error(err, "unable to initialize controllers")
+		os.Exit(1)
+	}
 
 	// Cert won't be ready until manager starts, so start a goroutine here which
 	// will block until the cert is ready before setting up the controllers.

pkg/controller/controller.go

@@ -17,6 +17,9 @@ limitations under the License.
 package controller
 
 import (
+	"context"
+	"fmt"
+
 	"github.com/vllm-project/aibrix/pkg/config"
 	"github.com/vllm-project/aibrix/pkg/controller/kvcache"
 	"github.com/vllm-project/aibrix/pkg/controller/modeladapter"
@@ -25,8 +28,13 @@ import (
 	"github.com/vllm-project/aibrix/pkg/controller/rayclusterfleet"
 	"github.com/vllm-project/aibrix/pkg/controller/rayclusterreplicaset"
 	"github.com/vllm-project/aibrix/pkg/features"
+
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/klog/v2"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 )
 
@@ -37,7 +45,7 @@ import (
 
 var controllerAddFuncs []func(manager.Manager, config.RuntimeConfig) error
 
-func Initialize() {
+func Initialize(mgr manager.Manager) error {
 	if features.IsControllerEnabled(features.PodAutoscalerController) {
 		controllerAddFuncs = append(controllerAddFuncs, podautoscaler.Add)
 	}
@@ -51,14 +59,22 @@ func Initialize() {
 	}
 
 	if features.IsControllerEnabled(features.DistributedInferenceController) {
-		// TODO: only enable them if KubeRay is installed (check RayCluster CRD exist)
+		// Check if the CRD (e.g., "rayclusters.ray.io") exists. If not, fail directly.
+		crdName := "rayclusters.ray.io"
+		if err := checkCRDExists(mgr.GetClient(), crdName); err != nil {
+			return fmt.Errorf("failed to validate CRD '%s': %v. "+
+				"Please ensure that the CRD is installed and available in the cluster. "+
+				"You can verify this by running 'kubectl get crd %s'",
+				crdName, err, crdName)
+		}
 		controllerAddFuncs = append(controllerAddFuncs, rayclusterreplicaset.Add)
 		controllerAddFuncs = append(controllerAddFuncs, rayclusterfleet.Add)
 	}
 
 	if features.IsControllerEnabled(features.KVCacheController) {
 		controllerAddFuncs = append(controllerAddFuncs, kvcache.Add)
 	}
+	return nil
 }
 
 // SetupWithManager sets up the controller with the Manager.
@@ -74,3 +90,25 @@ func SetupWithManager(m manager.Manager, runtimeConfig config.RuntimeConfig) err
 	}
 	return nil
 }
+
+// checkCRDExists checks if the specified CRD exists in the cluster.
+func checkCRDExists(c client.Client, crdName string) error {
+	gvk := schema.GroupVersionKind{
+		Group:   "apiextensions.k8s.io",
+		Version: "v1",
+		Kind:    "CustomResourceDefinition",
+	}
+	// Create an unstructured object to represent the CRD.
+	crd := &unstructured.Unstructured{}
+	crd.SetGroupVersionKind(gvk)
+	crd.SetName(crdName)
+
+	err := c.Get(context.TODO(), client.ObjectKey{Name: crdName}, crd)
+	if err != nil {
+		if apierrors.IsNotFound(err) {
+			return fmt.Errorf("CRD %q not found. Please ensure %q is installed", crdName, crdName)
+		}
+		return fmt.Errorf("error checking CRD %q: %v", crdName, err)
+	}
+	return nil
+}

pkg/controller/rayclusterfleet/rayclusterfleet_controller.go

@@ -22,24 +22,23 @@ import (
 	"reflect"
 	"time"
 
+	rayclusterv1 "github.com/ray-project/kuberay/ray-operator/apis/ray/v1"
+	orchestrationv1alpha1 "github.com/vllm-project/aibrix/api/orchestration/v1alpha1"
 	"github.com/vllm-project/aibrix/pkg/config"
 	"github.com/vllm-project/aibrix/pkg/controller/util/expectation"
+
 	appsv1 "k8s.io/api/apps/v1"
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
-
-	rayclusterv1 "github.com/ray-project/kuberay/ray-operator/apis/ray/v1"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/klog/v2"
-	"sigs.k8s.io/controller-runtime/pkg/manager"
-	"sigs.k8s.io/controller-runtime/pkg/reconcile"
-
-	orchestrationv1alpha1 "github.com/vllm-project/aibrix/api/orchestration/v1alpha1"
-	"k8s.io/apimachinery/pkg/runtime"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 )
 
 var (
@@ -51,8 +50,7 @@ var (
 // Add creates a new RayClusterFleet Controller and adds it to the Manager with default RBAC.
 // The Manager will set fields on the Controller and Start it when the Manager is Started.
 func Add(mgr manager.Manager, runtimeConfig config.RuntimeConfig) error {
-	// TODO: check crd exists or not. If not, we should fail here directly without moving forward.
-
+	klog.InfoS("Starting raycluster-fleet-controller")
 	r, err := newReconciler(mgr, runtimeConfig)
 	if err != nil {
 		return err
@@ -84,7 +82,7 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error {
 		return err
 	}
 
-	klog.V(4).InfoS("Finished to add model-adapter-controller")
+	klog.V(4).InfoS("Finished to add raycluster-fleet-controller")
 	return nil
 }
 

pkg/controller/rayclusterreplicaset/rayclusterreplicaset_controller.go

@@ -28,9 +28,7 @@ import (
 	"github.com/vllm-project/aibrix/pkg/controller/util/expectation"
 
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
-	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
-	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/klog/v2"
@@ -46,47 +44,17 @@ var (
 	controllerKind                              = orchestrationv1alpha1.GroupVersion.WithKind("RayClusterReplicaSet")
 )
 
-// Add first validates that the required Ray CRD (e.g., "rayclusters.ray.io") exists in the cluster.
-// If the CRD is not found, the function fails early with an error.
-// If the CRD exists, this function creates a new RayClusterReplicaSet Controller and adds it to the Manager with default RBAC.
+// Add creates a new RayClusterReplicaSet Controller and adds it to the Manager with default RBAC.
 // The Manager will set fields on the Controller and Start it when the Manager is Started.
 func Add(mgr manager.Manager, runtimeConfig config.RuntimeConfig) error {
-	// Check if the CRD exists. If not, fail directly.
-	crdName := "rayclusters.ray.io"
-	if err := checkCRDExists(mgr.GetClient(), crdName); err != nil {
-		return fmt.Errorf("failed to validate CRD: %v", err)
-	}
-
+	klog.InfoS("Starting raycluster-replicaset-controller")
 	r, err := newReconciler(mgr, runtimeConfig)
 	if err != nil {
 		return err
 	}
 	return add(mgr, r)
 }
 
-// checkCRDExists checks if the specified CRD exists in the cluster.
-func checkCRDExists(c client.Client, crdName string) error {
-	gvk := schema.GroupVersionKind{
-		Group:   "apiextensions.k8s.io",
-		Version: "v1",
-		Kind:    "CustomResourceDefinition",
-	}
-
-	// Create an unstructured object to represent the CRD.
-	crd := &unstructured.Unstructured{}
-	crd.SetGroupVersionKind(gvk)
-	crd.SetName(crdName)
-
-	err := c.Get(context.TODO(), client.ObjectKey{Name: crdName}, crd)
-	if err != nil {
-		if apierrors.IsNotFound(err) {
-			return fmt.Errorf("CRD %q not found. Please ensure %q is installed", crdName, crdName)
-		}
-		return fmt.Errorf("error checking CRD %q: %v", crdName, err)
-	}
-	return nil
-}
-
 // newReconciler returns a new reconcile.Reconciler
 func newReconciler(mgr manager.Manager, runtimeConfig config.RuntimeConfig) (reconcile.Reconciler, error) {
 	reconciler := &RayClusterReplicaSetReconciler{