Resolve the product variant once at startup

cmd/main.go

@@ -57,19 +57,22 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apiextensions-apiserver/pkg/apis/apiextensions"
 	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/client-go/kubernetes"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
+	clientgocache "k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/tools/clientcmd"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/apiutil"
 	"sigs.k8s.io/controller-runtime/pkg/client/config"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
 	ctrlmetrics "sigs.k8s.io/controller-runtime/pkg/metrics"
 	"sigs.k8s.io/controller-runtime/pkg/metrics/server"
 	"sigs.k8s.io/controller-runtime/pkg/webhook"
@@ -127,7 +130,7 @@ func main() {
 	var sgSetup bool
 	var manageCRDs bool
 	var preDelete bool
-	var variant string
+	var bootstrapVariant string
 
 	// bootstrapCRDs is a flag that can be used to install the CRDs and exit. This is useful for
 	// workflows that use an init container to install CustomResources prior to the operator starting.
@@ -166,14 +169,27 @@ If a value other than 'all' is specified, the first CRD with a prefix of the spe
 	flag.BoolVar(&manageCRDs, "manage-crds", false, "Operator should manage the projectcalico.org and operator.tigera.io CRDs.")
 	flag.BoolVar(&preDelete, "pre-delete", false, "Run helm pre-deletion hook logic, then exit.")
 	flag.BoolVar(&bootstrapCRDs, "bootstrap-crds", false, "Install CRDs and exit")
-	flag.StringVar(&variant, "variant", string(operatortigeraiov1.Calico), "Default product variant to assume during boostrapping.")
+	flag.StringVar(
+		&bootstrapVariant, "variant", string(operatortigeraiov1.Calico),
+		`Product variant to install CRDs for before an Installation exists. Only affects CRD and
+admission policy installation; once an Installation exists it is the authority on the variant.`,
+	)
 
 	opts := zap.Options{}
 	opts.BindFlags(flag.CommandLine)
 	flag.Parse()
 
 	ctrl.SetLogger(zap.New(zap.WriteTo(os.Stdout), zap.UseFlagOptions(&opts)))
 
+	// An unrecognised variant would quietly install the Calico CRDs, which can't be corrected
+	// later because CRDs are only ever created, never updated.
+	switch v := operatortigeraiov1.ProductVariant(bootstrapVariant); {
+	case v == operatortigeraiov1.Calico, v.IsEnterprise():
+	default:
+		fmt.Printf("Invalid -variant %q\n", bootstrapVariant)
+		os.Exit(1)
+	}
+
 	if showVersion {
 		// If the following line is updated then it might be necessary to update the assertOperatorImageVersion in hack/release/build.go
 		fmt.Println("Operator:", version.VERSION)
@@ -392,17 +408,17 @@ If a value other than 'all' is specified, the first CRD with a prefix of the spe
 	if bootstrapCRDs || manageCRDs {
 		setupLog.WithValues("v3", v3CRDs).Info("Ensuring CRDs are installed")
 
-		if err := crds.Ensure(mgr.GetClient(), variant, v3CRDs, setupLog); err != nil {
+		if err := crds.Ensure(mgr.GetClient(), bootstrapVariant, v3CRDs, setupLog); err != nil {
 			setupLog.Error(err, "Failed to ensure CRDs are created")
 			os.Exit(1)
 		}
 
-		if err := admission.Ensure(mgr.GetClient(), variant, v3CRDs, apiDiscovery.ServedVersion(admission.APIGroup, admission.KindPolicy), setupLog); err != nil {
+		if err := admission.Ensure(mgr.GetClient(), bootstrapVariant, v3CRDs, apiDiscovery.ServedVersion(admission.APIGroup, admission.KindPolicy), setupLog); err != nil {
 			setupLog.Error(err, "Failed to ensure MutatingAdmissionPolicies are created")
 			os.Exit(1)
 		}
 
-		if err := admission.EnsureValidating(mgr.GetClient(), variant, v3CRDs, apiDiscovery.ServedVersion(admission.APIGroup, admission.KindValidatingPolicy), setupLog); err != nil {
+		if err := admission.EnsureValidating(mgr.GetClient(), bootstrapVariant, v3CRDs, apiDiscovery.ServedVersion(admission.APIGroup, admission.KindValidatingPolicy), setupLog); err != nil {
 			setupLog.Error(err, "Failed to ensure ValidatingAdmissionPolicies are created")
 			os.Exit(1)
 		}
@@ -413,6 +429,41 @@ If a value other than 'all' is specified, the first CRD with a prefix of the spe
 		}
 	}
 
+	// Resolve the variant now that the operator CRDs exist.
+	bootVariant, err := resolveBootVariant(ctx, c, operatortigeraiov1.ProductVariant(bootstrapVariant))
+	if err != nil {
+		setupLog.Error(err, "Failed to resolve the product variant")
+		os.Exit(1)
+	}
+	setupLog.WithValues("variant", bootVariant).Info("Resolved product variant")
+
+	// The bootstrap pass above used the flag default, which doesn't cover the enterprise APIs.
+	if manageCRDs && bootVariant != operatortigeraiov1.ProductVariant(bootstrapVariant) {
+		setupLog.WithValues("variant", bootVariant).Info("Ensuring CRDs are installed for the resolved variant")
+
+		if err := crds.Ensure(mgr.GetClient(), string(bootVariant), v3CRDs, setupLog); err != nil {
+			setupLog.Error(err, "Failed to ensure CRDs are created")
+			os.Exit(1)
+		}
+	}
+
+	// The enterprise controllers can't register without their APIs. Exiting lets the kubelet
+	// retry us once the CRDs are installed.
+	if bootVariant.IsEnterprise() {
+		enterpriseAPIs, err := discovery.EnterpriseAPIsExist(cs)
+		if err != nil {
+			setupLog.Error(err, "Failed to determine whether the Enterprise APIs are available")
+			os.Exit(1)
+		}
+		if !enterpriseAPIs {
+			setupLog.Error(
+				fmt.Errorf("the Calico Enterprise CRDs are not installed"),
+				"Cannot run as Calico Enterprise",
+			)
+			os.Exit(1)
+		}
+	}
+
 	// Start a goroutine to handle termination.
 	go func() {
 		// Cancel the main context when we are done.
@@ -497,14 +548,6 @@ If a value other than 'all' is specified, the first CRD with a prefix of the spe
 	}
 	setupLog.WithValues("tenancy", multiTenant).Info("Checking tenancy mode")
 
-	// Determine if we need to start the Enterprise specific controllers.
-	enterpriseCRDExists, err := discovery.RequiresTigeraSecure(clientset)
-	if err != nil {
-		setupLog.Error(err, "Failed to determine if Enterprise controllers are required")
-		os.Exit(1)
-	}
-	setupLog.WithValues("required", enterpriseCRDExists).Info("Checking if Enterprise controllers are required")
-
 	clusterDomain, err := dns.GetClusterDomain(dns.DefaultResolveConfPath)
 	if err != nil {
 		clusterDomain = dns.DefaultClusterDomain
@@ -563,25 +606,31 @@ If a value other than 'all' is specified, the first CRD with a prefix of the spe
 	}
 
 	// Start a watch on our bootstrap configmap so we can restart if it changes.
-	if err = utils.MonitorConfigMap(clientset, bootstrapConfigMapName, bootConfig.Data); err != nil {
+	if err = utils.MonitorConfigMap(ctx, mgr.GetCache(), bootstrapConfigMapName, bootConfig.Data); err != nil {
 		log.Error(err, "Failed to monitor bootstrap configmap")
 		os.Exit(1)
 	}
 
+	// Same for the variant, which the process can only change by restarting.
+	if err = monitorVariant(ctx, mgr, bootVariant); err != nil {
+		log.Error(err, "Failed to monitor the product variant")
+		os.Exit(1)
+	}
+
 	options := options.ControllerOptions{
-		DetectedProvider:    provider,
-		EnterpriseCRDExists: enterpriseCRDExists,
-		ClusterDomain:       clusterDomain,
-		KubernetesVersion:   kubernetesVersion,
-		ManageCRDs:          manageCRDs,
-		ShutdownContext:     ctx,
-		K8sClientset:        clientset,
-		MultiTenant:         multiTenant,
-		ElasticExternal:     useExternalElastic,
-		Cloud:               isCloudBuild(),
-		ESMigration:         elasticIsMigrating,
-		UseV3CRDs:           v3CRDs,
-		APIDiscovery:        apiDiscovery,
+		DetectedProvider:  provider,
+		Variant:           bootVariant,
+		ClusterDomain:     clusterDomain,
+		KubernetesVersion: kubernetesVersion,
+		ManageCRDs:        manageCRDs,
+		ShutdownContext:   ctx,
+		K8sClientset:      clientset,
+		MultiTenant:       multiTenant,
+		ElasticExternal:   useExternalElastic,
+		Cloud:             isCloudBuild(),
+		ESMigration:       elasticIsMigrating,
+		UseV3CRDs:         v3CRDs,
+		APIDiscovery:      apiDiscovery,
 	}
 
 	// Before we start any controllers, make sure our options are valid.
@@ -598,7 +647,7 @@ If a value other than 'all' is specified, the first CRD with a prefix of the spe
 
 	// Register custom Prometheus metrics collector.
 	if common.MetricsEnabled() {
-		collector := metrics.NewOperatorCollector(mgr.GetClient(), enterpriseCRDExists)
+		collector := metrics.NewOperatorCollector(mgr.GetClient(), bootVariant.IsEnterprise())
 		ctrlmetrics.Registry.MustRegister(collector)
 	}
 
@@ -664,6 +713,77 @@ func setKubernetesServiceEnv(kubeconfigFile string) error {
 	return nil
 }
 
+// resolveBootVariant returns the variant the operator should run as. It falls back to the
+// bootstrap default rather than waiting, so a fresh cluster isn't blocked before the
+// Installation exists; the variant watch restarts us once it appears.
+func resolveBootVariant(ctx context.Context, c client.Client, def operatortigeraiov1.ProductVariant) (operatortigeraiov1.ProductVariant, error) {
+	spec := operatortigeraiov1.InstallationSpec{}
+
+	instance := &operatortigeraiov1.Installation{}
+	if err := c.Get(ctx, utils.DefaultInstanceKey, instance); err != nil {
+		if !errors.IsNotFound(err) && !meta.IsNoMatchError(err) {
+			return "", err
+		}
+	} else {
+		spec = instance.Spec
+	}
+
+	// The overlay can set the variant like any other field, so it has to be merged in.
+	overlay := &operatortigeraiov1.Installation{}
+	if err := c.Get(ctx, utils.OverlayInstanceKey, overlay); err != nil {
+		if !errors.IsNotFound(err) && !meta.IsNoMatchError(err) {
+			return "", err
+		}
+	} else {
+		spec = utils.OverrideInstallationSpec(spec, overlay.Spec)
+	}
+
+	if spec.Variant == "" {
+		return def, nil
+	}
+	return spec.Variant, nil
+}
+
+// monitorVariant restarts the operator when the effective variant moves off the one this
+// process booted with.
+func monitorVariant(ctx context.Context, mgr manager.Manager, booted operatortigeraiov1.ProductVariant) error {
+	// The cache isn't running yet, so don't wait on a sync that can't happen.
+	informer, err := mgr.GetCache().GetInformer(ctx, &operatortigeraiov1.Installation{}, cache.BlockUntilSynced(false))
+	if err != nil {
+		return err
+	}
+
+	// Re-resolve rather than reading the event's object, since the effective variant is the
+	// merge of the default Installation and the overlay.
+	c := mgr.GetClient()
+	check := func() {
+		// Exiting mid-uninstall would skip the graceful termination wait in main, which
+		// holds the process open so controllers can run their finalizers.
+		instance := &operatortigeraiov1.Installation{}
+		if err := c.Get(ctx, utils.DefaultInstanceKey, instance); err == nil && instance.DeletionTimestamp != nil {
+			return
+		}
+
+		requested, err := resolveBootVariant(ctx, c, booted)
+		if err != nil {
+			log.Error(err, "Failed to resolve the requested variant")
+			return
+		}
+
+		if requested != booted {
+			log.Info("Requested variant changed, rebooting", "booted", booted, "requested", requested)
+			os.Exit(0)
+		}
+	}
+
+	_, err = informer.AddEventHandler(clientgocache.ResourceEventHandlerFuncs{
+		AddFunc:    func(any) { check() },
+		UpdateFunc: func(_, _ any) { check() },
+		DeleteFunc: func(any) { check() },
+	})
+	return err
+}
+
 func showCRDs(variant operatortigeraiov1.ProductVariant, outputType string) error {
 	first := true
 	for _, v := range crds.GetCRDs(variant, os.Getenv("CALICO_API_GROUP") == "projectcalico.org/v3") {

pkg/common/discovery/discovery.go

@@ -31,10 +31,8 @@ import (
 
 const gkeNodeLabelPrefix = "cloud.google.com/gke-"
 
-// RequiresTigeraSecure determines if the configuration requires we start the tigera secure
-// controllers.
-func RequiresTigeraSecure(clientset *kubernetes.Clientset) (bool, error) {
-	// Use the discovery client to determine if the tigera secure specific APIs exist.
+// EnterpriseAPIsExist reports whether the cluster serves the Calico Enterprise APIs.
+func EnterpriseAPIsExist(clientset *kubernetes.Clientset) (bool, error) {
 	resources, err := clientset.Discovery().ServerResourcesForGroupVersion("operator.tigera.io/v1")
 	if err != nil {
 		return false, err

pkg/controller/apiserver/apiserver_controller.go

@@ -103,7 +103,7 @@ func Add(mgr manager.Manager, opts options.ControllerOptions) error {
 		return fmt.Errorf("apiserver-controller failed to watch ConfigMap %s: %w", render.K8sSvcEndpointConfigMapName, err)
 	}
 
-	if opts.EnterpriseCRDExists {
+	if opts.Variant.IsEnterprise() {
 		// Watch for changes to ApplicationLayer
 		err = c.WatchObject(&operatorv1.ApplicationLayer{ObjectMeta: metav1.ObjectMeta{Name: utils.DefaultEnterpriseInstanceKey.Name}}, &handler.EnqueueRequestForObject{})
 		if err != nil {
@@ -310,7 +310,7 @@ func (r *ReconcileAPIServer) Reconcile(ctx context.Context, request reconcile.Re
 	}
 
 	// Query for the installation object.
-	_, installationSpec, err := utils.GetInstallationSpec(context.Background(), r.client)
+	installationSpec, err := utils.GetInstallationSpec(context.Background(), r.client)
 	if err != nil {
 		if errors.IsNotFound(err) {
 			r.status.SetDegraded(operatorv1.ResourceNotFound, "Installation not found", err, reqLogger)