/* Copyright 2022 The KubeVela Authors. 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. */ package app import ( "context" "flag" "fmt" "io" "os" "path/filepath" "strconv" "time" velaclient "github.com/kubevela/pkg/controller/client" "github.com/kubevela/pkg/controller/sharding" "github.com/kubevela/pkg/meta" "github.com/kubevela/pkg/util/profiling" "github.com/pkg/errors" "github.com/spf13/cobra" utilfeature "k8s.io/apiserver/pkg/util/feature" "k8s.io/client-go/rest" "k8s.io/klog/v2" "k8s.io/klog/v2/textlogger" ctrl "sigs.k8s.io/controller-runtime" ctrlcache "sigs.k8s.io/controller-runtime/pkg/cache" ctrlclient "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/healthz" "sigs.k8s.io/controller-runtime/pkg/manager" "sigs.k8s.io/controller-runtime/pkg/manager/signals" metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server" ctrlwebhook "sigs.k8s.io/controller-runtime/pkg/webhook" "github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1" "github.com/oam-dev/kubevela/apis/types" "github.com/oam-dev/kubevela/cmd/core/app/config" "github.com/oam-dev/kubevela/cmd/core/app/hooks" "github.com/oam-dev/kubevela/cmd/core/app/hooks/crdvalidation" "github.com/oam-dev/kubevela/cmd/core/app/options" "github.com/oam-dev/kubevela/pkg/auth" "github.com/oam-dev/kubevela/pkg/cache" commonconfig "github.com/oam-dev/kubevela/pkg/controller/common" oamv1beta1 "github.com/oam-dev/kubevela/pkg/controller/core.oam.dev/v1beta1" "github.com/oam-dev/kubevela/pkg/controller/core.oam.dev/v1beta1/application" "github.com/oam-dev/kubevela/pkg/features" "github.com/oam-dev/kubevela/pkg/logging" "github.com/oam-dev/kubevela/pkg/monitor/watcher" "github.com/oam-dev/kubevela/pkg/multicluster" "github.com/oam-dev/kubevela/pkg/oam" "github.com/oam-dev/kubevela/pkg/utils/common" "github.com/oam-dev/kubevela/pkg/utils/util" oamwebhook "github.com/oam-dev/kubevela/pkg/webhook/core.oam.dev" "github.com/oam-dev/kubevela/version" ) var ( scheme = common.Scheme waitSecretTimeout = 90 * time.Second waitSecretInterval = 2 * time.Second ) // NewCoreCommand creates a *cobra.Command object with default parameters func NewCoreCommand() *cobra.Command { coreOptions := options.NewCoreOptions() cmd := &cobra.Command{ Use: "vela-core", Long: `The KubeVela controller manager is a daemon that embeds the core control loops shipped with KubeVela`, RunE: func(cmd *cobra.Command, args []string) error { return run(signals.SetupSignalHandler(), coreOptions) }, SilenceUsage: true, FParseErrWhitelist: cobra.FParseErrWhitelist{ // Allow unknown flags for backward-compatibility. UnknownFlags: true, }, } flags := cmd.Flags() namedFlagSets := coreOptions.Flags() for _, flagSet := range namedFlagSets.FlagSets { flags.AddFlagSet(flagSet) } meta.Name = types.VelaCoreName klog.InfoS("KubeVela information", "version", version.VelaVersion, "revision", version.GitRevision) klog.InfoS("Vela-Core init", "definition namespace", oam.SystemDefinitionNamespace) return cmd } func run(ctx context.Context, coreOptions *options.CoreOptions) error { klog.InfoS("Starting KubeVela core controller", "context", "initialization", "leaderElection", coreOptions.Server.EnableLeaderElection, "webhookEnabled", coreOptions.Webhook.UseWebhook) // Sync configurations klog.V(2).InfoS("Syncing configurations to global variables") syncConfigurations(ctx, coreOptions) klog.InfoS("Configuration sync completed successfully") // Setup logging klog.V(2).InfoS("Setting up logging configuration", "debug", coreOptions.Observability.LogDebug, "devLogs", coreOptions.Observability.DevLogs, "logFilePath", coreOptions.Observability.LogFilePath) setupLogging(coreOptions.Observability) // Configure Kubernetes client klog.InfoS("Configuring Kubernetes client", "QPS", coreOptions.Kubernetes.QPS, "burst", coreOptions.Kubernetes.Burst) kubeConfig, err := configureKubernetesClient(coreOptions.Kubernetes) if err != nil { klog.ErrorS(err, "Failed to configure Kubernetes client") return fmt.Errorf("failed to configure Kubernetes client: %w", err) } // Start profiling server klog.V(2).InfoS("Starting profiling server in background") go profiling.StartProfilingServer(nil) // Setup multi-cluster if enabled if coreOptions.MultiCluster.EnableClusterGateway { klog.InfoS("Multi-cluster gateway enabled, setting up multi-cluster capability", "enableMetrics", coreOptions.MultiCluster.EnableClusterMetrics, "metricsInterval", coreOptions.MultiCluster.ClusterMetricsInterval) if err := setupMultiCluster(ctx, kubeConfig, coreOptions.MultiCluster); err != nil { klog.ErrorS(err, "Failed to setup multi-cluster") return fmt.Errorf("failed to setup multi-cluster: %w", err) } klog.InfoS("Multi-cluster setup completed successfully") } // Configure feature gates klog.V(2).InfoS("Configuring feature gates") configureFeatureGates(coreOptions) // Create controller manager klog.InfoS("Creating controller manager", "metricsAddr", coreOptions.Observability.MetricsAddr, "healthAddr", coreOptions.Server.HealthAddr, "webhookPort", coreOptions.Webhook.WebhookPort) manager, err := createControllerManager(ctx, kubeConfig, coreOptions) if err != nil { klog.ErrorS(err, "Failed to create controller manager") return fmt.Errorf("failed to create controller manager: %w", err) } klog.InfoS("Controller manager created successfully") // Register health checks klog.V(2).InfoS("Registering health and readiness checks") if err := registerHealthChecks(manager); err != nil { klog.ErrorS(err, "Failed to register health checks") return fmt.Errorf("failed to register health checks: %w", err) } // Setup controllers based on sharding mode klog.InfoS("Setting up controllers", "shardingEnabled", sharding.EnableSharding, "shardID", sharding.ShardID) if err := setupControllers(ctx, manager, coreOptions); err != nil { klog.ErrorS(err, "Failed to setup controllers") return fmt.Errorf("failed to setup controllers: %w", err) } klog.InfoS("Controllers setup completed successfully") // Start application monitor klog.InfoS("Starting application metrics monitor") if err := startApplicationMonitor(ctx, manager); err != nil { klog.ErrorS(err, "Failed to start application monitor") return fmt.Errorf("failed to start application monitor: %w", err) } // Start the manager klog.InfoS("Starting controller manager") if err := manager.Start(ctx); err != nil { klog.ErrorS(err, "Failed to run manager") return err } // Cleanup performCleanup(coreOptions) klog.InfoS("Program safely stopped") return nil } // syncConfigurations syncs parsed config values to external package global variables func syncConfigurations(ctx context.Context, coreOptions *options.CoreOptions) { if coreOptions.Workflow != nil { klog.V(3).InfoS("Syncing workflow configuration") coreOptions.Workflow.SyncToWorkflowGlobals() } if coreOptions.CUE != nil { klog.V(3).InfoS("Syncing CUE configuration") coreOptions.CUE.SyncToCUEGlobals(ctx) } if coreOptions.Application != nil { klog.V(3).InfoS("Syncing application configuration") coreOptions.Application.SyncToApplicationGlobals() } if coreOptions.Performance != nil { klog.V(3).InfoS("Syncing performance configuration") coreOptions.Performance.SyncToPerformanceGlobals() } if coreOptions.Resource != nil { klog.V(3).InfoS("Syncing resource configuration") coreOptions.Resource.SyncToResourceGlobals() } if coreOptions.OAM != nil { klog.V(3).InfoS("Syncing OAM configuration") coreOptions.OAM.SyncToOAMGlobals() } } // setupLogging configures klog based on parsed observability settings func setupLogging(observabilityConfig *config.ObservabilityConfig) { // Configure klog verbosity if observabilityConfig.LogDebug { _ = flag.Set("v", strconv.Itoa(int(commonconfig.LogDebug))) } // Configure log file output if observabilityConfig.LogFilePath != "" { _ = flag.Set("logtostderr", "false") _ = flag.Set("log_file", observabilityConfig.LogFilePath) _ = flag.Set("log_file_max_size", strconv.FormatUint(observabilityConfig.LogFileMaxSize, 10)) } // Set logger (use --dev-logs=true for local development) if observabilityConfig.DevLogs { logOutput := logging.NewColorWriter(os.Stdout) klog.LogToStderr(false) klog.SetOutput(logOutput) ctrl.SetLogger(textlogger.NewLogger(textlogger.NewConfig(textlogger.Output(logOutput)))) } else { ctrl.SetLogger(textlogger.NewLogger(textlogger.NewConfig())) } } // ConfigProvider is a function type that provides a Kubernetes REST config type ConfigProvider func() (*rest.Config, error) // configureKubernetesClient creates and configures the Kubernetes REST config func configureKubernetesClient(kubernetesConfig *config.KubernetesConfig) (*rest.Config, error) { return configureKubernetesClientWithProvider(kubernetesConfig, ctrl.GetConfig) } // configureKubernetesClientWithProvider creates and configures the Kubernetes REST config // using a provided config provider function. This allows for dependency injection in tests. func configureKubernetesClientWithProvider(kubernetesConfig *config.KubernetesConfig, configProvider ConfigProvider) (*rest.Config, error) { // Gracefully handle error returns instead of panicking kubeConfig, err := configProvider() if err != nil { return nil, err } kubeConfig.UserAgent = types.KubeVelaName + "/" + version.GitRevision kubeConfig.QPS = float32(kubernetesConfig.QPS) kubeConfig.Burst = kubernetesConfig.Burst kubeConfig.Wrap(auth.NewImpersonatingRoundTripper) klog.InfoS("Kubernetes Config Loaded", "UserAgent", kubeConfig.UserAgent, "QPS", kubeConfig.QPS, "Burst", kubeConfig.Burst, ) return kubeConfig, nil } // setupMultiCluster initializes multi-cluster capability func setupMultiCluster(ctx context.Context, kubeConfig *rest.Config, multiClusterConfig *config.MultiClusterConfig) error { klog.V(2).InfoS("Initializing multi-cluster client") clusterClient, err := multicluster.Initialize(kubeConfig, true) if err != nil { klog.ErrorS(err, "Failed to enable multi-cluster capability") return err } klog.InfoS("Multi-cluster client initialized successfully") if multiClusterConfig.EnableClusterMetrics { klog.InfoS("Enabling cluster metrics collection", "interval", multiClusterConfig.ClusterMetricsInterval) _, err := multicluster.NewClusterMetricsMgr(ctx, clusterClient, multiClusterConfig.ClusterMetricsInterval) if err != nil { klog.ErrorS(err, "Failed to enable multi-cluster-metrics capability") return err } klog.InfoS("Cluster metrics manager initialized successfully") } return nil } // configureFeatureGates sets up feature-dependent configurations func configureFeatureGates(coreOptions *options.CoreOptions) { if utilfeature.DefaultMutableFeatureGate.Enabled(features.ApplyOnce) { klog.V(2).InfoS("ApplyOnce feature gate enabled, configuring application re-sync period", "period", coreOptions.Kubernetes.InformerSyncPeriod) commonconfig.ApplicationReSyncPeriod = coreOptions.Kubernetes.InformerSyncPeriod } } // buildManagerOptions constructs ctrl.Options from CoreOptions for creating a controller manager. // This function is extracted for testability - it contains the option construction logic // without the side effects of creating a manager or starting background processes. func buildManagerOptions(ctx context.Context, coreOptions *options.CoreOptions) ctrl.Options { leaderElectionID := util.GenerateLeaderElectionID(types.KubeVelaName, coreOptions.Controller.IgnoreAppWithoutControllerRequirement) leaderElectionID += sharding.GetShardIDSuffix() return ctrl.Options{ Scheme: scheme, Metrics: metricsserver.Options{ BindAddress: coreOptions.Observability.MetricsAddr, }, LeaderElection: coreOptions.Server.EnableLeaderElection, LeaderElectionNamespace: coreOptions.Server.LeaderElectionNamespace, LeaderElectionID: leaderElectionID, WebhookServer: ctrlwebhook.NewServer(ctrlwebhook.Options{ Port: coreOptions.Webhook.WebhookPort, CertDir: coreOptions.Webhook.CertDir, }), HealthProbeBindAddress: coreOptions.Server.HealthAddr, LeaseDuration: &coreOptions.Server.LeaseDuration, RenewDeadline: &coreOptions.Server.RenewDeadline, RetryPeriod: &coreOptions.Server.RetryPeriod, NewClient: velaclient.DefaultNewControllerClient, NewCache: cache.BuildCache(ctx, ctrlcache.Options{ Scheme: scheme, SyncPeriod: &coreOptions.Kubernetes.InformerSyncPeriod, // SyncPeriod is configured with default value, aka. 10h. First, controller-runtime does not // recommend use it as a time trigger, instead, it is expected to work for failure tolerance // of controller-runtime. Additionally, set this value will affect not only application // controller but also all other controllers like definition controller. Therefore, for // functionalities like state-keep, they should be invented in other ways. }, &v1beta1.Application{}, &v1beta1.ApplicationRevision{}, &v1beta1.ResourceTracker{}, ), Client: ctrlclient.Options{ Cache: &ctrlclient.CacheOptions{ DisableFor: cache.NewResourcesToDisableCache(), }, }, } } // createControllerManager creates and configures the controller-runtime manager func createControllerManager(ctx context.Context, kubeConfig *rest.Config, coreOptions *options.CoreOptions) (ctrl.Manager, error) { leaderElectionID := util.GenerateLeaderElectionID(types.KubeVelaName, coreOptions.Controller.IgnoreAppWithoutControllerRequirement) leaderElectionID += sharding.GetShardIDSuffix() klog.V(2).InfoS("Creating controller manager with configuration", "leaderElectionID", leaderElectionID, "leaderElection", coreOptions.Server.EnableLeaderElection, "leaderElectionNamespace", coreOptions.Server.LeaderElectionNamespace, "leaseDuration", coreOptions.Server.LeaseDuration, "renewDeadline", coreOptions.Server.RenewDeadline) managerOptions := buildManagerOptions(ctx, coreOptions) manager, err := ctrl.NewManager(kubeConfig, managerOptions) if err != nil { klog.ErrorS(err, "Unable to create a controller manager") return nil, err } return manager, nil } // setupControllers sets up controllers based on sharding configuration func setupControllers(ctx context.Context, manager ctrl.Manager, coreOptions *options.CoreOptions) error { if !sharding.EnableSharding { return prepareRun(ctx, manager, coreOptions) } return prepareRunInShardingMode(ctx, manager, coreOptions) } // startApplicationMonitor starts the application metrics watcher func startApplicationMonitor(ctx context.Context, manager ctrl.Manager) error { klog.InfoS("Starting vela application monitor") applicationInformer, err := manager.GetCache().GetInformer(ctx, &v1beta1.Application{}) if err != nil { klog.ErrorS(err, "Unable to get informer for application") return err } watcher.StartApplicationMetricsWatcher(applicationInformer) klog.V(2).InfoS("Application metrics watcher started successfully") return nil } // performCleanup handles any necessary cleanup operations func performCleanup(coreOptions *options.CoreOptions) { klog.V(2).InfoS("Performing cleanup operations") if coreOptions.Observability.LogFilePath != "" { klog.V(3).InfoS("Flushing log file", "path", coreOptions.Observability.LogFilePath) klog.Flush() } } // prepareRunInShardingMode initializes the controller manager in sharding mode where workload // is distributed across multiple controller instances. In sharding mode: // - Master shard handles webhooks, scheduling, and full controller setup // - Non-master shards only run the Application controller for their assigned Applications // This enables horizontal scaling of the KubeVela control plane across multiple pods. func prepareRunInShardingMode(ctx context.Context, manager manager.Manager, coreOptions *options.CoreOptions) error { if sharding.IsMaster() { klog.InfoS("Controller running in sharding mode", "shardType", "master", "webhookAutoSchedule", !utilfeature.DefaultMutableFeatureGate.Enabled(features.DisableWebhookAutoSchedule)) if !utilfeature.DefaultMutableFeatureGate.Enabled(features.DisableWebhookAutoSchedule) { klog.V(2).InfoS("Starting webhook auto-scheduler in background") go sharding.DefaultScheduler.Get().Start(ctx) } if err := prepareRun(ctx, manager, coreOptions); err != nil { return err } } else { klog.InfoS("Controller running in sharding mode", "shardType", "worker", "shardID", sharding.ShardID) klog.V(2).InfoS("Setting up application controller for worker shard") if err := application.Setup(manager, coreOptions.Controller.Args); err != nil { klog.ErrorS(err, "Failed to setup application controller in sharding mode") return err } klog.InfoS("Application controller setup completed for worker shard") } return nil } // prepareRun sets up the complete KubeVela controller manager with all necessary components: // - Configures and registers OAM webhooks if enabled // - Sets up all OAM controllers (Application, ComponentDefinition, WorkflowStepDefinition, PolicyDefinition, and TraitDefinition) // - Initializes multi-cluster capabilities and cluster info // - Runs pre-start validation hooks to ensure system readiness // This function is used in single-instance mode or by the master shard in sharding mode. func prepareRun(ctx context.Context, manager manager.Manager, coreOptions *options.CoreOptions) error { // Bootstrap provider registry early before other initialization klog.V(2).InfoS("Initializing provider registry") bootstrapProviderRegistry() if coreOptions.Webhook.UseWebhook { klog.InfoS("Webhook enabled, registering OAM webhooks", "port", coreOptions.Webhook.WebhookPort, "certDir", coreOptions.Webhook.CertDir) oamwebhook.Register(manager, coreOptions.Controller.Args) klog.V(2).InfoS("Waiting for webhook secret volume", "timeout", waitSecretTimeout, "checkInterval", waitSecretInterval) if err := waitWebhookSecretVolume(coreOptions.Webhook.CertDir, waitSecretTimeout, waitSecretInterval); err != nil { klog.ErrorS(err, "Unable to get webhook secret") return err } klog.InfoS("Webhook secret volume ready, webhooks registered successfully") } klog.InfoS("Setting up OAM controllers") if err := oamv1beta1.Setup(manager, coreOptions.Controller.Args); err != nil { klog.ErrorS(err, "Unable to setup the OAM controller") return err } klog.InfoS("OAM controllers setup completed successfully") klog.V(2).InfoS("Initializing control plane cluster info") if err := multicluster.InitClusterInfo(manager.GetConfig()); err != nil { klog.ErrorS(err, "Failed to init control plane cluster info") return err } klog.InfoS("Starting vela controller manager with pre-start validation") for _, hook := range []hooks.PreStartHook{crdvalidation.NewHook()} { hookName := hook.Name() klog.InfoS("Running pre-start hook", "hook", hookName) if err := hook.Run(ctx); err != nil { klog.ErrorS(err, "Failed to run pre-start hook", "hook", hookName) return fmt.Errorf("failed to run hook %s: %w", hookName, err) } klog.InfoS("Pre-start hook completed successfully", "hook", hookName) } klog.InfoS("All pre-start validation hooks completed successfully") return nil } // registerHealthChecks is used to create readiness&liveness probes func registerHealthChecks(manager ctrl.Manager) error { klog.InfoS("Registering readiness and health checks") if err := manager.AddReadyzCheck("ping", healthz.Ping); err != nil { klog.ErrorS(err, "Failed to add readiness check") return err } klog.V(3).InfoS("Readiness check registered", "check", "ping") // TODO: change the health check to be different from readiness check if err := manager.AddHealthzCheck("ping", healthz.Ping); err != nil { klog.ErrorS(err, "Failed to add health check") return err } klog.V(3).InfoS("Health check registered", "check", "ping") return nil } // waitWebhookSecretVolume waits for webhook secret ready to avoid manager running crash func waitWebhookSecretVolume(certDir string, timeout, interval time.Duration) error { start := time.Now() for { time.Sleep(interval) if time.Since(start) > timeout { return fmt.Errorf("getting webhook secret timeout after %s", timeout.String()) } klog.InfoS("Wait webhook secret", "time consumed(second)", int64(time.Since(start).Seconds()), "timeout(second)", int64(timeout.Seconds())) if _, err := os.Stat(certDir); !os.IsNotExist(err) { ready := func() bool { certDirectory, err := os.Open(filepath.Clean(certDir)) if err != nil { return false } defer func() { if err := certDirectory.Close(); err != nil { klog.ErrorS(err, "Failed to close directory") } }() // check if dir is empty if _, err := certDirectory.Readdir(1); errors.Is(err, io.EOF) { return false } // check if secret files are empty err = filepath.Walk(certDir, func(path string, fileInfo os.FileInfo, err error) error { // even Cert dir is created, cert files are still empty for a while if fileInfo.Size() == 0 { return errors.New("secret is not ready") } return nil }) if err == nil { klog.InfoS("Webhook secret is ready", "time consumed(second)", int64(time.Since(start).Seconds())) return true } return false }() if ready { return nil } } } }