refactor(schedualr): split node collectors group controller to files (#…

…1818)

This PR aligns how the scheduler controllers are implemented, making it
similar to the existing pattern used in other places in the repo.

scheduler/controllers/nodecollectorsgroup/clustercollectorsgroup_controller.go

@@ -0,0 +1,19 @@
+package nodecollectorsgroup
+
+import (
+	"context"
+
+	"k8s.io/apimachinery/pkg/runtime"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+type clusterCollectorsGroupController struct {
+	client.Client
+	Scheme *runtime.Scheme
+}
+
+func (r *clusterCollectorsGroupController) Reconcile(ctx context.Context, _ ctrl.Request) (ctrl.Result, error) {
+	err := sync(ctx, r.Client)
+	return ctrl.Result{}, err
+}

scheduler/controllers/nodecollectorsgroup/common.go

@@ -0,0 +1,72 @@
+package nodecollectorsgroup
+
+import (
+	"context"
+	"errors"
+
+	odigosv1 "github.com/odigos-io/odigos/api/odigos/v1alpha1"
+	"github.com/odigos-io/odigos/common"
+	"github.com/odigos-io/odigos/k8sutils/pkg/consts"
+	k8sutilsconsts "github.com/odigos-io/odigos/k8sutils/pkg/consts"
+	"github.com/odigos-io/odigos/k8sutils/pkg/env"
+	"github.com/odigos-io/odigos/k8sutils/pkg/utils"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+func newNodeCollectorGroup(odigosConfig common.OdigosConfiguration) *odigosv1.CollectorsGroup {
+
+	ownMetricsPort := consts.OdigosNodeCollectorOwnTelemetryPortDefault
+	if odigosConfig.CollectorNode != nil && odigosConfig.CollectorNode.CollectorOwnMetricsPort != 0 {
+		ownMetricsPort = odigosConfig.CollectorNode.CollectorOwnMetricsPort
+	}
+
+	return &odigosv1.CollectorsGroup{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "CollectorsGroup",
+			APIVersion: "odigos.io/v1alpha1",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      consts.OdigosNodeCollectorDaemonSetName,
+			Namespace: env.GetCurrentNamespace(),
+		},
+		Spec: odigosv1.CollectorsGroupSpec{
+			Role:                    odigosv1.CollectorsGroupRoleNodeCollector,
+			CollectorOwnMetricsPort: ownMetricsPort,
+		},
+	}
+}
+
+func sync(ctx context.Context, c client.Client) error {
+
+	namespace := env.GetCurrentNamespace()
+
+	var instrumentedConfigs odigosv1.InstrumentationConfigList
+	err := c.List(ctx, &instrumentedConfigs)
+	if err != nil {
+		return errors.Join(errors.New("failed to list InstrumentationConfigs"), err)
+	}
+	numberOfInstrumentedApps := len(instrumentedConfigs.Items)
+
+	if numberOfInstrumentedApps == 0 {
+		// TODO: should we delete the collector group if cluster collector is not ready?
+		return utils.DeleteCollectorGroup(ctx, c, namespace, k8sutilsconsts.OdigosNodeCollectorCollectorGroupName)
+	}
+
+	clusterCollectorGroup, err := utils.GetCollectorGroup(ctx, c, namespace, k8sutilsconsts.OdigosClusterCollectorCollectorGroupName)
+	if err != nil {
+		return client.IgnoreNotFound(err)
+	}
+
+	odigosConfig, err := utils.GetCurrentOdigosConfig(ctx, c)
+	if err != nil {
+		return err
+	}
+
+	clusterCollectorReady := clusterCollectorGroup.Status.Ready
+	if clusterCollectorReady {
+		return utils.ApplyCollectorGroup(ctx, c, newNodeCollectorGroup(odigosConfig))
+	}
+
+	return nil
+}

scheduler/controllers/nodecollectorsgroup/instrumentationconfig_controller.go

@@ -0,0 +1,19 @@
+package nodecollectorsgroup
+
+import (
+	"context"
+
+	"k8s.io/apimachinery/pkg/runtime"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+type instrumentationConfigController struct {
+	client.Client
+	Scheme *runtime.Scheme
+}
+
+func (r *instrumentationConfigController) Reconcile(ctx context.Context, _ ctrl.Request) (ctrl.Result, error) {
+	err := sync(ctx, r.Client)
+	return ctrl.Result{}, err
+}

scheduler/controllers/nodecollectorsgroup/manager.go

@@ -0,0 +1,50 @@
+package nodecollectorsgroup
+
+import (
+	odigosv1 "github.com/odigos-io/odigos/api/odigos/v1alpha1"
+	odigospredicates "github.com/odigos-io/odigos/k8sutils/pkg/predicate"
+	corev1 "k8s.io/api/core/v1"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/predicate"
+)
+
+func SetupWithManager(mgr ctrl.Manager) error {
+
+	err := ctrl.NewControllerManagedBy(mgr).
+		For(&odigosv1.InstrumentationConfig{}).
+		Named("nodecollectorgroup-instrumentationconfig").
+		WithEventFilter(&odigospredicates.ExistencePredicate{}).
+		Complete(&instrumentationConfigController{
+			Client: mgr.GetClient(),
+			Scheme: mgr.GetScheme(),
+		})
+	if err != nil {
+		return err
+	}
+
+	err = ctrl.NewControllerManagedBy(mgr).
+		For(&corev1.ConfigMap{}).
+		Named("nodecollectorgroup-odigosconfig").
+		WithEventFilter(&odigospredicates.OdigosConfigMapPredicate).
+		Complete(&odigosConfigController{
+			Client: mgr.GetClient(),
+			Scheme: mgr.GetScheme(),
+		})
+	if err != nil {
+		return err
+	}
+
+	err = ctrl.NewControllerManagedBy(mgr).
+		For(&odigosv1.CollectorsGroup{}).
+		Named("nodecollectorgroup-clustercollectorsgroup").
+		WithEventFilter(predicate.And(&odigospredicates.OdigosCollectorsGroupClusterPredicate, &odigospredicates.CgBecomesReadyPredicate{})).
+		Complete(&clusterCollectorsGroupController{
+			Client: mgr.GetClient(),
+			Scheme: mgr.GetScheme(),
+		})
+	if err != nil {
+		return err
+	}
+
+	return nil
+}

scheduler/controllers/nodecollectorsgroup/odigosconfig_controller.go

@@ -0,0 +1,19 @@
+package nodecollectorsgroup
+
+import (
+	"context"
+
+	"k8s.io/apimachinery/pkg/runtime"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+type odigosConfigController struct {
+	client.Client
+	Scheme *runtime.Scheme
+}
+
+func (r *odigosConfigController) Reconcile(ctx context.Context, _ ctrl.Request) (ctrl.Result, error) {
+	err := sync(ctx, r.Client)
+	return ctrl.Result{}, err
+}

scheduler/main.go

@@ -43,6 +43,7 @@ import (
 	ctrlzap "sigs.k8s.io/controller-runtime/pkg/log/zap"
 
 	"github.com/odigos-io/odigos/scheduler/controllers"
+	"github.com/odigos-io/odigos/scheduler/controllers/nodecollectorsgroup"
 	//+kubebuilder:scaffold:imports
 )
 
@@ -111,14 +112,11 @@ func main() {
 		setupLog.Error(err, "unable to create controller", "controller", "Destination")
 		os.Exit(1)
 	}
-	if err = (&controllers.NodeCollectorsGroupReconciler{
-		Client: mgr.GetClient(),
-		Scheme: mgr.GetScheme(),
-	}).SetupWithManager(mgr); err != nil {
-		setupLog.Error(err, "unable to create controller", "controller", "InstrumentationConfig")
+	err = nodecollectorsgroup.SetupWithManager(mgr)
+	if err != nil {
+		setupLog.Error(err, "unable to create controllers for node collectors group")
 		os.Exit(1)
 	}
-	//+kubebuilder:scaffold:builder
 
 	if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {
 		setupLog.Error(err, "unable to set up health check")