fix(#4948): refactored common logic between operator an platformcontr…

…oller

pkg/cmd/manager/controller.go

@@ -0,0 +1,310 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements.  See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You 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.
+*/
+
+package manager
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/apache/camel-k/v2/pkg/apis"
+	"github.com/apache/camel-k/v2/pkg/client"
+	"github.com/apache/camel-k/v2/pkg/event"
+	"github.com/apache/camel-k/v2/pkg/platform"
+	"github.com/apache/camel-k/v2/pkg/util/kubernetes"
+	logutil "github.com/apache/camel-k/v2/pkg/util/log"
+	"go.uber.org/automaxprocs/maxprocs"
+	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
+	coordination "k8s.io/api/coordination/v1"
+	corev1 "k8s.io/api/core/v1"
+	k8serrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/leaderelection/resourcelock"
+	"k8s.io/client-go/tools/record"
+	"k8s.io/klog/v2"
+	"sigs.k8s.io/controller-runtime/pkg/cache"
+	ctrl "sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/config"
+	"sigs.k8s.io/controller-runtime/pkg/healthz"
+	logf "sigs.k8s.io/controller-runtime/pkg/log"
+	zapctrl "sigs.k8s.io/controller-runtime/pkg/log/zap"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+	"sigs.k8s.io/controller-runtime/pkg/manager/signals"
+)
+
+type Manager interface {
+	LoggerSetup() (logutil.Logger, error, string)
+	PrintVersion()
+	GetWatchNamespace() (string, error)
+	CreateBootstrapClient(cfg *rest.Config) (client.Client, error, string)
+	CreateEventBroadcaster(bootstrapClient client.Client, watchNamespace string, ctx context.Context) (record.EventBroadcaster, error, string)
+	GetControllerNamespaceAndLeaderElection(watchNamespace string, bootstrapClient client.Client, leaderElection bool, ctx context.Context) (string, bool, error, string)
+	SetOperatorImage(bootstrapClient client.Client, ctx context.Context) (error, string)
+	GetManagerOptions(bootstrapClient client.Client, watchNamespace string) (cache.Options, error, string)
+	CreateManager(healthPort int32, monitoringPort int32, leaderElection bool, leaderElectionID string, cfg *rest.Config, eventBroadcaster record.EventBroadcaster, controllerNamespace string, options cache.Options, ctx context.Context) (manager.Manager, client.Client, error, string)
+	ControllerPreStartResourcesInit(initCtx context.Context, bootstrapClient client.Client, watchNamespace string, controllerNamespace string, ctx context.Context, ctrlClient client.Client, mgr manager.Manager) (error, string)
+}
+
+type ControllerCmd struct {
+	ControllerManager Manager
+}
+
+func (c ControllerCmd) Run(healthPort, monitoringPort int32, leaderElection bool, leaderElectionID string) (error, string) {
+	log, err, errMessage := c.ControllerManager.LoggerSetup()
+	if err != nil {
+		log.Error(err, errMessage)
+	}
+
+	c.ControllerManager.PrintVersion()
+	// Will only appear if DEBUG level has been enabled using the env var LOG_LEVEL
+	log.Debug("*** DEBUG level messages will be logged ***")
+
+	watchNamespace, err := c.ControllerManager.GetWatchNamespace()
+	if err != nil {
+		return err, "failed to get watch namespace"
+	}
+
+	cfg, err := config.GetConfig()
+	if err != nil {
+		return err, "cannot get client config"
+	}
+
+	bootstrapClient, err, errMessage := c.ControllerManager.CreateBootstrapClient(cfg)
+	if err != nil {
+		return err, errMessage
+	}
+
+	ctx := signals.SetupSignalHandler()
+
+	eventBroadcaster, err, errMessage := c.ControllerManager.CreateEventBroadcaster(bootstrapClient, watchNamespace, ctx)
+	if eventBroadcaster != nil {
+		defer eventBroadcaster.Shutdown()
+	}
+	if err != nil {
+		return err, errMessage
+	}
+
+	controllerNamespace, leaderElection, err, errMessage := c.ControllerManager.GetControllerNamespaceAndLeaderElection(watchNamespace, bootstrapClient, leaderElection, ctx)
+	if err != nil {
+		return err, errMessage
+	}
+	if !leaderElection {
+		log.Info("Leader election is disabled!")
+	}
+
+	err, errMessage = c.ControllerManager.SetOperatorImage(bootstrapClient, ctx)
+	if err != nil {
+		return err, errMessage
+	}
+
+	options, err, errMessage := c.ControllerManager.GetManagerOptions(bootstrapClient, watchNamespace)
+	if err != nil {
+		return err, errMessage
+	}
+
+	mgr, ctrlClient, err, errMessage := c.ControllerManager.CreateManager(healthPort, monitoringPort, leaderElection, leaderElectionID, cfg, eventBroadcaster, controllerNamespace, options, ctx)
+
+	initCtx, initCancel := context.WithTimeout(ctx, 1*time.Minute)
+	defer initCancel()
+
+	err, errMessage = c.ControllerManager.ControllerPreStartResourcesInit(initCtx, bootstrapClient, watchNamespace, controllerNamespace, ctx, ctrlClient, mgr)
+	if err != nil {
+		return err, errMessage
+	}
+
+	log.Info("Starting the manager")
+	return mgr.Start(ctx), "manager exited non-zero"
+}
+
+type BaseManager struct {
+	Log                  logutil.Logger
+	WatchNamespaceEnvVar string
+	ControllerNamespace  string
+	ControllerPodName    string
+	AddToManager         func(ctx context.Context, manager manager.Manager, client client.Client) error
+}
+
+func (bm BaseManager) LoggerSetup() (logutil.Logger, error, string) {
+	// The logger instantiated here can be changed to any logger
+	// implementing the logr.Logger interface. This logger will
+	// be propagated through the whole operator, generating
+	// uniform and structured logs.
+
+	// The constants specified here are zap specific
+	var logLevel zapcore.Level
+	logLevelVal, ok := os.LookupEnv("LOG_LEVEL")
+	if ok {
+		switch strings.ToLower(logLevelVal) {
+		case "error":
+			logLevel = zapcore.ErrorLevel
+		case "info":
+			logLevel = zapcore.InfoLevel
+		case "debug":
+			logLevel = zapcore.DebugLevel
+		default:
+			customLevel, err := strconv.Atoi(strings.ToLower(logLevelVal))
+			if err != nil {
+				return bm.Log, err, "Invalid log-level"
+			}
+			// Need to multiply by -1 to turn logr expected level into zap level
+			logLevel = zapcore.Level(int8(customLevel) * -1)
+		}
+	} else {
+		logLevel = zapcore.InfoLevel
+	}
+
+	// Use and set atomic level that all following log events are compared with
+	// in order to evaluate if a given log level on the event is enabled.
+	logf.SetLogger(zapctrl.New(func(o *zapctrl.Options) {
+		o.Development = false
+		o.Level = zap.NewAtomicLevelAt(logLevel)
+	}))
+
+	klog.SetLogger(bm.Log.AsLogger())
+
+	_, err := maxprocs.Set(maxprocs.Logger(func(f string, a ...interface{}) { bm.Log.Info(fmt.Sprintf(f, a)) }))
+
+	return bm.Log, err, "failed to set GOMAXPROCS from cgroups"
+}
+
+// getWatchNamespace returns the Namespace the operator should be watching for changes.
+func (bm BaseManager) GetWatchNamespace() (string, error) {
+	ns, found := os.LookupEnv(bm.WatchNamespaceEnvVar)
+	if !found {
+		return "", fmt.Errorf("%s must be set", bm.WatchNamespaceEnvVar)
+	}
+	return ns, nil
+}
+
+func (bm BaseManager) CreateBootstrapClient(cfg *rest.Config) (client.Client, error, string) {
+	// Increase maximum burst that is used by client-side throttling,
+	// to prevent the requests made to apply the bundled Kamelets
+	// from being throttled.
+	cfg.QPS = 20
+	cfg.Burst = 200
+	bootstrapClient, err := client.NewClientWithConfig(false, cfg)
+	if err != nil {
+		return nil, err, "cannot initialize client"
+	}
+
+	return bootstrapClient, nil, ""
+}
+
+func (bm BaseManager) CreateEventBroadcaster(bootstrapClient client.Client, watchNamespace string, ctx context.Context) (record.EventBroadcaster, error, string) {
+	// We do not rely on the event broadcaster managed by controller runtime,
+	// so that we can check the operator has been granted permission to create
+	// Events. This is required for the operator to be installable by standard
+	// admin users, that are not granted create permission on Events by default.
+	broadcaster := record.NewBroadcaster()
+
+	if ok, err := kubernetes.CheckPermission(ctx, bootstrapClient, corev1.GroupName, "events", watchNamespace, "", "create"); err != nil || !ok {
+		// Do not sink Events to the server as they'll be rejected
+		broadcaster = event.NewSinkLessBroadcaster(broadcaster)
+		return broadcaster, err, "cannot check permissions for creating Events"
+		bm.Log.Info("Event broadcasting is disabled because of missing permissions to create Events")
+	}
+
+	return broadcaster, nil, ""
+}
+
+func (bm BaseManager) GetControllerNamespaceAndLeaderElection(watchNamespace string, bootstrapClient client.Client, leaderElection bool, ctx context.Context) (string, bool, error, string) {
+	operatorNamespace := platform.GetOperatorNamespace()
+	if operatorNamespace == "" {
+		// Fallback to using the watch namespace when the operator is not in-cluster.
+		// It does not support local (off-cluster) operator watching resources globally,
+		// in which case it's not possible to determine a namespace.
+		operatorNamespace = watchNamespace
+		if operatorNamespace == "" {
+			leaderElection = false
+			bm.Log.Info("unable to determine namespace for leader election")
+		}
+	}
+
+	if ok, err := kubernetes.CheckPermission(ctx, bootstrapClient, coordination.GroupName, "leases", operatorNamespace, "", "create"); err != nil || !ok {
+		leaderElection = false
+		if err != nil {
+			return operatorNamespace, leaderElection, err, "cannot check permissions for creating Leases"
+		}
+		bm.Log.Info("The operator is not granted permissions to create Leases")
+	}
+
+	return operatorNamespace, leaderElection, nil, ""
+}
+
+// SetOperatorImage set the operator container image if it runs in-container
+func (bm BaseManager) SetOperatorImage(bootstrapClient client.Client, ctx context.Context) (error, string) {
+	var err error
+	platform.OperatorImage, err = getOperatorImage(bm.ControllerNamespace, bm.ControllerPodName, ctx, bootstrapClient)
+	return err, "cannot get operator container image"
+}
+
+// getOperatorImage returns the image currently used by the running operator if present (when running out of cluster, it may be absent).
+func getOperatorImage(namespace string, podName string, ctx context.Context, c ctrl.Reader) (string, error) {
+	if namespace == "" || podName == "" {
+		return "", nil
+	}
+
+	pod := corev1.Pod{}
+	if err := c.Get(ctx, ctrl.ObjectKey{Namespace: namespace, Name: podName}, &pod); err != nil && k8serrors.IsNotFound(err) {
+		return "", nil
+	} else if err != nil {
+		return "", err
+	}
+	if len(pod.Spec.Containers) == 0 {
+		return "", fmt.Errorf("no containers found in operator pod")
+	}
+	return pod.Spec.Containers[0].Image, nil
+}
+
+func (bm BaseManager) CreateManager(healthPort int32, monitoringPort int32, leaderElection bool, leaderElectionID string, cfg *rest.Config, eventBroadcaster record.EventBroadcaster, controllerNamespace string, options cache.Options, ctx context.Context) (manager.Manager, client.Client, error, string) {
+	mgr, err := manager.New(cfg, manager.Options{
+		EventBroadcaster:              eventBroadcaster,
+		LeaderElection:                leaderElection,
+		LeaderElectionNamespace:       controllerNamespace,
+		LeaderElectionID:              leaderElectionID,
+		LeaderElectionResourceLock:    resourcelock.LeasesResourceLock,
+		LeaderElectionReleaseOnCancel: true,
+		HealthProbeBindAddress:        ":" + strconv.Itoa(int(healthPort)),
+		MetricsBindAddress:            ":" + strconv.Itoa(int(monitoringPort)),
+		Cache:                         options,
+	})
+	if err != nil {
+		return nil, nil, err, ""
+	}
+
+	bm.Log.Info("Configuring manager")
+	if err := mgr.AddHealthzCheck("health-probe", healthz.Ping); err != nil {
+		return nil, nil, err, "Unable add liveness check"
+	}
+	if err := apis.AddToScheme(mgr.GetScheme()); err != nil {
+		return nil, nil, err, ""
+	}
+	ctrlClient, err := client.FromManager(mgr)
+	if err != nil {
+		return nil, nil, err, ""
+	}
+	if err := bm.AddToManager(ctx, mgr, ctrlClient); err != nil {
+		return nil, nil, err, ""
+	}
+
+	return mgr, ctrlClient, nil, ""
+}