(feat) block cluster upgrade when installed incompatible operators

Author: gallettilance

cmd/cluster-olm-operator/main.go

@@ -97,10 +97,12 @@ func runOperator(ctx context.Context, cc *controllercmd.ControllerContext) error
 		controllerList = append(controllerList, controller)
 	}
 
-	upgradeableConditionController := controller.NewStaticUpgradeableConditionController(
-		"OLMStaticUpgradeableConditionController",
+	dynamicUpgradeableConditionController := controller.NewDynamicUpgradeableConditionController(
+		cl.KubeClient,
+		cl.DynamicClient,
+		"OLMDynamicUpgradeableConditionController",
 		cl.OperatorClient,
-		cc.EventRecorder.ForComponent("OLMStaticUpgradeableConditionController"),
+		cc.EventRecorder.ForComponent("OLMDynamicUpgradeableConditionController"),
 		controllerNames,
 	)
 
@@ -119,7 +121,7 @@ func runOperator(ctx context.Context, cc *controllercmd.ControllerContext) error
 
 	cl.StartInformers(ctx)
 
-	for _, c := range append(controllerList, upgradeableConditionController, clusterOperatorController) {
+	for _, c := range append(controllerList, dynamicUpgradeableConditionController, clusterOperatorController) {
 		go func(c factory.Controller) {
 			defer runtime.HandleCrash()
 			c.Run(ctx, 1)

go.mod

@@ -6,6 +6,7 @@ toolchain go1.22.7
 
 require (
 	github.com/blang/semver/v4 v4.0.0
+	github.com/go-logr/logr v1.4.2
 	github.com/openshift/api v0.0.0-20240527133614-ba11c1587003
 	github.com/openshift/build-machinery-go v0.0.0-20240419090851-af9c868bcf52
 	github.com/openshift/client-go v0.0.0-20240528061634-b054aa794d87
@@ -68,7 +69,6 @@ require (
 	github.com/fxamacker/cbor/v2 v2.7.0 // indirect
 	github.com/go-errors/errors v1.4.2 // indirect
 	github.com/go-gorp/gorp/v3 v3.1.0 // indirect
-	github.com/go-logr/logr v1.4.2 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/go-openapi/jsonpointer v0.21.0 // indirect
 	github.com/go-openapi/jsonreference v0.21.0 // indirect

pkg/controller/dynamicupgradeableconditionscontroller.go

@@ -0,0 +1,337 @@
+package controller
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log"
+	"os"
+	"sort"
+	"strings"
+	"sync"
+	"time"
+
+	semver "github.com/blang/semver/v4"
+	"github.com/go-logr/logr"
+	operatorv1 "github.com/openshift/api/operator/v1"
+	"github.com/openshift/cluster-olm-operator/pkg/clients"
+	"github.com/openshift/library-go/pkg/controller/factory"
+	"github.com/openshift/library-go/pkg/operator/events"
+	"github.com/openshift/library-go/pkg/operator/v1helpers"
+	helmclient "github.com/operator-framework/helm-operator-plugins/pkg/client"
+	storage "github.com/operator-framework/helm-operator-plugins/pkg/storage"
+	ocv1alpha1 "github.com/operator-framework/operator-controller/api/v1alpha1"
+	"github.com/operator-framework/operator-registry/alpha/property"
+	helm "helm.sh/helm/v3/pkg/storage"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/dynamic"
+	"k8s.io/client-go/dynamic/dynamicinformer"
+	"k8s.io/client-go/informers"
+	"k8s.io/client-go/kubernetes"
+	v1 "k8s.io/client-go/kubernetes/typed/core/v1"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/klog/v2"
+)
+
+const (
+	incompatibleOperatorsInstalled = "IncompatibleOperatorsInstalled"
+	MaxOpenShiftVersionProperty    = "olm.maxOpenShiftVersion"
+	OwnerKindKey                   = "olm.operatorframework.io/owner-kind"
+	OwnerNameKey                   = "olm.operatorframework.io/owner-name"
+	PackageNameKey                 = "olm.operatorframework.io/package-name"
+	BundleNameKey                  = "olm.operatorframework.io/bundle-name"
+	BundleVersionKey               = "olm.operatorframework.io/bundle-version"
+)
+
+type dynamicUpgradeableConditionController struct {
+	kubeclient     kubernetes.Interface
+	informer       informers.GenericInformer
+	client         dynamic.Interface
+	name           string
+	operatorClient *clients.OperatorClient
+	prefixes       []string
+	logger         logr.Logger
+}
+
+type ChartMetadata struct {
+	Metadata struct {
+		Annotations map[string]string `yaml:"annotations"`
+	} `yaml:"metadata"`
+}
+
+type SecretData struct {
+	Chart struct {
+		ChartMetadata
+	} `yaml:"chart"`
+}
+
+func NewDynamicUpgradeableConditionController(kubeclient kubernetes.Interface, client dynamic.Interface, name string, operatorClient *clients.OperatorClient, eventRecorder events.Recorder, prefixes []string) factory.Controller {
+	infFact := dynamicinformer.NewDynamicSharedInformerFactory(client, 30*time.Minute)
+
+	clusterExtensionGVR := schema.GroupVersionResource{
+		Group:    ocv1alpha1.ClusterExtensionGVK.Group,
+		Version:  ocv1alpha1.ClusterExtensionGVK.Version,
+		Resource: "clusterextensions",
+	}
+
+	inf := infFact.ForResource(clusterExtensionGVR)
+
+	c := &dynamicUpgradeableConditionController{
+		informer:       inf,
+		client:         client,
+		kubeclient:     kubeclient,
+		name:           name,
+		operatorClient: operatorClient,
+		prefixes:       prefixes,
+		logger:         klog.FromContext(context.TODO()).WithName(name),
+	}
+
+	_, err := inf.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc:    c.handleClusterExtension,
+		UpdateFunc: func(oldObj, newObj interface{}) { c.handleClusterExtension(newObj) },
+		DeleteFunc: c.handleClusterExtension,
+	})
+	if err != nil {
+		c.logger.V(4).Error(nil, "could not add event handler to informer")
+		return nil
+	}
+
+	return factory.New().WithBareInformers(inf.Informer()).ToController(name, eventRecorder)
+}
+
+func (c *dynamicUpgradeableConditionController) handleClusterExtension(obj interface{}) {
+	_, ok := obj.(*ocv1alpha1.ClusterExtension)
+	if !ok {
+		c.logger.V(4).Error(nil, "could not cast to ClusterExtension object")
+		return
+	}
+
+	c.sync()
+}
+
+func (c *dynamicUpgradeableConditionController) sync() {
+	c.logger.V(4).Info("sync started")
+	defer c.logger.V(4).Info("done syncing")
+	updateStatusFuncs, err := c.ensureControllersManaged()
+	if err != nil {
+		c.logger.V(4).Error(err, "error verifying OLM state")
+		return
+	}
+
+	current, err := getCurrentOpenShiftVersion()
+	if err != nil {
+		c.logger.V(4).Error(err, "Error getting current OCP release")
+		return
+	}
+
+	if current == nil {
+		// Note: This shouldn't happen
+		err = fmt.Errorf("failed to determine current OpenShift Y-stream release")
+		c.logger.V(4).Error(err, "current release is nil")
+		return
+	}
+
+	next, err := nextY(*current)
+	if err != nil {
+		c.logger.V(4).Error(err, "Error finding next OCP Y-stream release")
+		return
+	}
+
+	var incompatibleOperators []string
+
+	ceList, err := c.informer.Lister().List(nil)
+	if err != nil {
+		c.logger.V(4).Error(err, "Error listing cluster extensions")
+		return
+	}
+
+	// Get all ClusterExtensions incompatible with next Y-stream
+	for _, obj := range ceList {
+		ce, ok := obj.(*ocv1alpha1.ClusterExtension)
+		if !ok {
+			c.logger.V(4).Error(nil, "Error: could not cast to ClusterExtension object")
+			return
+		}
+		store := buildHelmStore(ce, c.kubeclient.CoreV1().Secrets(ce.Namespace))
+
+		rel, _ := store.Deployed(ce.Name)
+		props, err := propertyListFromPropertiesAnnotation(rel.Chart.Metadata.Annotations["olm.properties"])
+		if err != nil {
+			c.logger.V(4).Error(err, "Error converting olm.properties")
+			return
+		}
+		for _, p := range props {
+			if p.Type == MaxOpenShiftVersionProperty {
+				version := string(p.Value)
+				maxOCPVersion, err := toSemver(version)
+				if err != nil {
+					c.logger.V(4).Info(fmt.Sprintf("Error converting to semver for version %s: %v", version, err))
+					continue
+				}
+				if maxOCPVersion != nil && !maxOCPVersion.GTE(next) {
+					// Incompatible
+					incompatibleOperators = append(incompatibleOperators, rel.Labels[BundleNameKey])
+				}
+			}
+		}
+	}
+
+	// deterministic ordering
+	sort.Strings(incompatibleOperators)
+
+	if len(incompatibleOperators) > 0 {
+		updateStatusFuncs = append(updateStatusFuncs, v1helpers.UpdateConditionFn(operatorv1.OperatorCondition{
+			Type:    "Upgradeable",
+			Status:  operatorv1.ConditionFalse,
+			Reason:  incompatibleOperatorsInstalled,
+			Message: strings.Join(incompatibleOperators, ","),
+		}))
+
+		if _, _, updateErr := v1helpers.UpdateStatus(context.TODO(), c.operatorClient, updateStatusFuncs...); updateErr != nil {
+			log.Printf("Error listing secrets: %v", err)
+			return
+		}
+	} else {
+		updateStatusFuncs = append(updateStatusFuncs, v1helpers.UpdateConditionFn(operatorv1.OperatorCondition{
+			Type:   "Upgradeable",
+			Status: operatorv1.ConditionTrue,
+		}))
+
+		if _, _, updateErr := v1helpers.UpdateStatus(context.TODO(), c.operatorClient, updateStatusFuncs...); updateErr != nil {
+			log.Printf("Error listing secrets: %v", err)
+			return
+		}
+	}
+}
+
+func (c *dynamicUpgradeableConditionController) ensureControllersManaged() ([]v1helpers.UpdateStatusFunc, error) {
+	c.logger.V(4).Info("verifying state of olm controllers")
+	defer c.logger.V(4).Info("done verifying state of olm controllers")
+	opSpec, _, _, err := c.operatorClient.GetOperatorState()
+	if err != nil {
+		return nil, err
+	}
+	if opSpec.ManagementState != operatorv1.Managed {
+		return nil, nil
+	}
+
+	updateStatusFuncs := make([]v1helpers.UpdateStatusFunc, 0, len(c.prefixes))
+	for _, prefix := range c.prefixes {
+		updateStatusFuncs = append(updateStatusFuncs, v1helpers.UpdateConditionFn(operatorv1.OperatorCondition{
+			Type:   fmt.Sprintf("%sUpgradeable", prefix),
+			Status: operatorv1.ConditionTrue,
+		}))
+	}
+	return updateStatusFuncs, nil
+}
+
+func propertyListFromPropertiesAnnotation(raw string) ([]property.Property, error) {
+	var props []property.Property
+	if err := json.Unmarshal([]byte(raw), &props); err != nil {
+		return nil, fmt.Errorf("failed to unmarshal properties annotation: %w", err)
+	}
+	return props, nil
+}
+
+func buildHelmStore(ce *ocv1alpha1.ClusterExtension, secretClient v1.SecretInterface) helm.Storage {
+	csConfig := storage.ChunkedSecretsConfig{
+		ChunkSize:      0,
+		MaxReadChunks:  0,
+		MaxWriteChunks: 0,
+		Log:            nil,
+	}
+
+	owner := ce.Name
+	ownerRefs := []metav1.OwnerReference{*metav1.NewControllerRef(ce, ce.GetObjectKind().GroupVersionKind())}
+	ownerRefSecretClient := helmclient.NewOwnerRefSecretClient(secretClient, ownerRefs, func(secret *corev1.Secret) bool {
+		return secret.Type == storage.SecretTypeChunkedIndex
+	})
+
+	return helm.Storage{
+		Driver:     storage.NewChunkedSecrets(ownerRefSecretClient, owner, csConfig),
+		MaxHistory: 0,
+		Log:        log.Printf,
+	}
+}
+
+type openshiftRelease struct {
+	version *semver.Version
+	mu      sync.Mutex
+}
+
+var (
+	currentRelease = &openshiftRelease{}
+)
+
+const (
+	releaseEnvVar = "RELEASE_VERSION" // OpenShift's env variable for defining the current release
+)
+
+func getCurrentOpenShiftVersion() (*semver.Version, error) {
+	currentRelease.mu.Lock()
+	defer currentRelease.mu.Unlock()
+
+	if currentRelease.version != nil {
+		/*
+			If the version is already set, we don't want to set it again as the currentRelease
+			is designed to be a singleton. If a new version is set, we are making an assumption
+			that this controller will be restarted and thus pull in the new version from the
+			environment into memory.
+
+			Note: sync.Once is not used here as it was difficult to reliably test without hitting
+			race conditions.
+		*/
+		return currentRelease.version, nil
+	}
+
+	// Get the raw version from the releaseEnvVar environment variable
+	raw, ok := os.LookupEnv(releaseEnvVar)
+	if !ok || raw == "" {
+		// No env var set, try again later
+		return nil, fmt.Errorf("desired release version missing from %v env variable", releaseEnvVar)
+	}
+
+	release, err := semver.ParseTolerant(raw)
+	if err != nil {
+		return nil, fmt.Errorf("cluster version has invalid desired release version: %w", err)
+	}
+
+	currentRelease.version = &release
+
+	return currentRelease.version, nil
+}
+
+func nextY(v semver.Version) (semver.Version, error) {
+	v.Build = nil // Builds are irrelevant
+
+	if len(v.Pre) > 0 {
+		// Dropping pre-releases is equivalent to incrementing Y
+		v.Pre = nil
+		v.Patch = 0
+
+		return v, nil
+	}
+
+	return v, v.IncrementMinor() // Sets Y=Y+1 and Z=0
+}
+
+func toSemver(max string) (*semver.Version, error) {
+	value := strings.Trim(max, "\"")
+	if value == "" {
+		// Handle "" separately, so parse doesn't treat it as a zero
+		return nil, fmt.Errorf(`value cannot be "" (an empty string)`)
+	}
+
+	version, err := semver.ParseTolerant(value)
+	if err != nil {
+		return nil, fmt.Errorf(`failed to parse "%q" as semver: %w`, value, err)
+	}
+
+	truncatedVersion := semver.Version{Major: version.Major, Minor: version.Minor}
+	if !version.EQ(truncatedVersion) {
+		return nil, fmt.Errorf("property %s must specify only <major>.<minor> version, got invalid value %s", MaxOpenShiftVersionProperty, version)
+	}
+	return &truncatedVersion, nil
+}