feat: block openshift upgrades for incompatible operators

- Refactor OpenShift ClusterOperator monitoring for olm-operator
  into an independent controller
- Set the OLM operator's ClusterOperator to upgradeable=false whenever
  incompatible operators, installed by OLM, exist on cluster;
  effectively blocking OpenShift upgrades

Signed-off-by: Nick Hale <[email protected]>

Author: njhale

cmd/olm/main.go

@@ -16,11 +16,12 @@ import (
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/pflag"
 	v1 "k8s.io/api/core/v1"
-
 	"k8s.io/klog"
+	ctrl "sigs.k8s.io/controller-runtime"
 
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/api/client/clientset/versioned"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/operators/olm"
+	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/operators/openshift"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/feature"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/filemonitor"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/operatorclient"
@@ -223,7 +224,24 @@ func main() {
 	<-op.Ready()
 
 	if *writeStatusName != "" {
-		operatorstatus.MonitorClusterStatus(*writeStatusName, op.AtLevel(), ctx.Done(), opClient, configClient, crClient)
+		reconciler, err := openshift.NewClusterOperatorReconciler(
+			openshift.WithClient(mgr.GetClient()),
+			openshift.WithScheme(mgr.GetScheme()),
+			openshift.WithLog(ctrl.Log.WithName("controllers").WithName("clusteroperator")),
+			openshift.WithName(*writeStatusName),
+			openshift.WithNamespace(*namespace),
+			openshift.WithSyncChannel(op.AtLevel()),
+			openshift.WithOLMOperator(),
+		)
+		if err != nil {
+			logger.WithError(err).Fatalf("error configuring openshift integration")
+			return
+		}
+
+		if err := reconciler.SetupWithManager(mgr); err != nil {
+			logger.WithError(err).Fatalf("error configuring openshift integration")
+			return
+		}
 	}
 
 	if *writePackageServerStatusName != "" {

cmd/olm/manager.go

@@ -47,9 +47,8 @@ func Manager(ctx context.Context, debug bool) (ctrl.Manager, error) {
 		return nil, err
 	}
 
-	// Setup a new controller to reconcile Operators
-	setupLog.Info("configuring controller")
 	if feature.Gate.Enabled(feature.OperatorLifecycleManagerV1) {
+		// Setup a new controller to reconcile Operators
 		operatorReconciler, err := operators.NewOperatorReconciler(
 			mgr.GetClient(),
 			ctrl.Log.WithName("controllers").WithName("operator"),
@@ -77,7 +76,6 @@ func Manager(ctx context.Context, debug bool) (ctrl.Manager, error) {
 		}
 
 	}
-
 	setupLog.Info("manager configured")
 
 	return mgr, nil

pkg/controller/operators/openshift/clusteroperator.go

@@ -0,0 +1,73 @@
+package openshift
+
+import (
+	"context"
+
+	configv1 "github.com/openshift/api/config/v1"
+	utilerrors "k8s.io/apimachinery/pkg/util/errors"
+)
+
+func NewClusterOperator(name string) *ClusterOperator {
+	co := &ClusterOperator{ClusterOperator: &configv1.ClusterOperator{}}
+	co.SetName(name)
+	return co
+}
+
+type ClusterOperator struct {
+	*configv1.ClusterOperator
+}
+
+func (c *ClusterOperator) GetOperatorVersion() string {
+	for _, v := range c.Status.Versions {
+		if v.Name == "operator" {
+			return v.Version
+		}
+	}
+
+	return ""
+}
+
+func (c *ClusterOperator) GetCondition(conditionType configv1.ClusterStatusConditionType) *configv1.ClusterOperatorStatusCondition {
+	for _, cond := range c.Status.Conditions {
+		if cond.Type == conditionType {
+			return &cond
+		}
+	}
+
+	return nil
+}
+
+func (c *ClusterOperator) SetCondition(condition *configv1.ClusterOperatorStatusCondition) {
+	// Filter dups
+	conditions := []configv1.ClusterOperatorStatusCondition{}
+	for _, c := range c.Status.Conditions {
+		if c.Type != condition.Type {
+			conditions = append(conditions, c)
+		}
+	}
+
+	c.Status.Conditions = append(conditions, *condition)
+}
+
+type Mutator interface {
+	Mutate(context.Context, *ClusterOperator) error
+}
+
+type MutateFunc func(context.Context, *ClusterOperator) error
+
+func (m MutateFunc) Mutate(ctx context.Context, co *ClusterOperator) error {
+	return m(ctx, co)
+}
+
+type SerialMutations []Mutator
+
+func (s SerialMutations) Mutate(ctx context.Context, co *ClusterOperator) error {
+	var errs []error
+	for _, m := range s {
+		if err := m.Mutate(ctx, co); err != nil {
+			errs = append(errs, err)
+		}
+	}
+
+	return utilerrors.NewAggregate(errs)
+}

pkg/controller/operators/openshift/clusteroperator_controller.go

@@ -0,0 +1,247 @@
+package openshift
+
+import (
+	"context"
+	"fmt"
+	"reflect"
+
+	configv1 "github.com/openshift/api/config/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/runtime"
+	utilerrors "k8s.io/apimachinery/pkg/util/errors"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/builder"
+	"sigs.k8s.io/controller-runtime/pkg/handler"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+	"sigs.k8s.io/controller-runtime/pkg/source"
+
+	operatorsv1alpha1 "github.com/operator-framework/api/pkg/operators/v1alpha1"
+	olmversion "github.com/operator-framework/operator-lifecycle-manager/pkg/version"
+)
+
+var (
+	localSchemeBuilder = runtime.NewSchemeBuilder(
+		configv1.AddToScheme,
+		operatorsv1alpha1.AddToScheme,
+	)
+
+	// AddToScheme adds all types necessary for the controller to operate.
+	AddToScheme = localSchemeBuilder.AddToScheme
+)
+
+type ClusterOperatorReconciler struct {
+	*ReconcilerConfig
+
+	delayRequeue reconcile.Result
+	mutate       MutateFunc
+	syncTracker  *SyncTracker
+	co           *ClusterOperator
+}
+
+func NewClusterOperatorReconciler(opts ...ReconcilerOption) (*ClusterOperatorReconciler, error) {
+	config := new(ReconcilerConfig)
+	config.apply(opts)
+	if err := config.complete(); err != nil {
+		return nil, err
+	}
+
+	co := NewClusterOperator(config.Name)
+	r := &ClusterOperatorReconciler{
+		ReconcilerConfig: config,
+		delayRequeue:     reconcile.Result{RequeueAfter: config.RequeueDelay},
+		co:               co,
+		syncTracker:      NewSyncTracker(config.SyncCh, co),
+	}
+
+	var mutations SerialMutations
+	if config.Mutator != nil {
+		mutations = append(mutations, config.Mutator)
+	}
+	mutations = append(mutations,
+		MutateFunc(r.setVersions),
+		MutateFunc(r.setProgressing),
+		MutateFunc(r.setAvailable),
+		MutateFunc(r.setDegraded),
+		MutateFunc(r.setUpgradeable),
+	)
+	r.mutate = mutations.Mutate
+
+	return r, nil
+}
+
+func (r *ClusterOperatorReconciler) SetupWithManager(mgr ctrl.Manager) error {
+	if err := mgr.Add(r.syncTracker); err != nil {
+		return fmt.Errorf("failed to add %T to manager: %s", r.syncTracker, err)
+	}
+
+	bldr := ctrl.NewControllerManagedBy(mgr).
+		For(&configv1.ClusterOperator{}, builder.WithPredicates(watchName(&r.Name))).
+		Watches(&source.Channel{Source: r.syncTracker.Events()}, &handler.EnqueueRequestForObject{})
+
+	return r.TweakBuilder(bldr).Complete(r)
+}
+
+func (r *ClusterOperatorReconciler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
+	log := r.Log.WithValues("request", req)
+	noRequeue := reconcile.Result{}
+	if req.NamespacedName.Name != r.co.GetName() {
+		// Throw away requests to reconcile all ClusterOperators but our own
+		// These should already be filtered by controller-runtime by this point
+		return noRequeue, nil
+	}
+
+	// Get the ClusterOperator
+	in := &configv1.ClusterOperator{}
+	if err := r.Client.Get(ctx, req.NamespacedName, in); err != nil {
+		if apierrors.IsNotFound(err) {
+			// The ClusterOperator is missing, let's create it
+			stripObject(r.co.ClusterOperator)
+			err = r.Client.Create(ctx, r.co.ClusterOperator)
+		}
+
+		// Transient error or successful creation, requeue
+		return r.delayRequeue, err
+	}
+
+	r.co.ClusterOperator = in.DeepCopy()
+
+	var errs []error
+	res := reconcile.Result{}
+	if err := r.mutate(ctx, r.co); err != nil {
+		// Transitive error, requeue
+		log.Error(err, "Error mutating ClusterOperator")
+		errs = append(errs, err)
+		res = r.delayRequeue
+	}
+
+	if !reflect.DeepEqual(r.co.Status, in.Status) {
+		// Status change detected, update
+		if err := r.Client.Status().Update(ctx, r.co.ClusterOperator); err != nil {
+			// Transitive error, requeue
+			errs = append(errs, err)
+			res = r.delayRequeue
+		}
+	}
+
+	return res, utilerrors.NewAggregate(errs)
+}
+
+func (r *ClusterOperatorReconciler) setVersions(_ context.Context, co *ClusterOperator) error {
+	// If we've successfully synced, we know our operator is working properly, so we can update the version
+	if r.syncTracker.SuccessfulSyncs() > 0 && !reflect.DeepEqual(co.Status.Versions, r.TargetVersions) {
+		co.Status.Versions = r.TargetVersions
+	}
+
+	return nil
+}
+
+func (r *ClusterOperatorReconciler) setProgressing(_ context.Context, co *ClusterOperator) error {
+	desired := &configv1.ClusterOperatorStatusCondition{
+		Type:               configv1.OperatorProgressing,
+		LastTransitionTime: r.Now(),
+	}
+
+	if r.syncTracker.SuccessfulSyncs() > 0 && reflect.DeepEqual(co.Status.Versions, r.TargetVersions) {
+		desired.Status = configv1.ConditionFalse
+		desired.Message = fmt.Sprintf("Deployed %s", olmversion.OLMVersion)
+	} else {
+		desired.Status = configv1.ConditionTrue
+		desired.Message = fmt.Sprintf("Waiting to see update %s succeed", olmversion.OLMVersion)
+	}
+
+	current := co.GetCondition(configv1.OperatorProgressing)
+	if conditionsEqual(current, desired) { // Comparison ignores lastUpdated
+		return nil
+	}
+
+	co.SetCondition(desired)
+
+	return nil
+}
+
+func (r *ClusterOperatorReconciler) setAvailable(_ context.Context, co *ClusterOperator) error {
+	desired := &configv1.ClusterOperatorStatusCondition{
+		Type:               configv1.OperatorAvailable,
+		LastTransitionTime: r.Now(),
+	}
+
+	if r.syncTracker.SuccessfulSyncs() > 0 && reflect.DeepEqual(co.Status.Versions, r.TargetVersions) {
+		desired.Status = configv1.ConditionTrue
+	} else {
+		desired.Status = configv1.ConditionFalse
+	}
+
+	current := co.GetCondition(configv1.OperatorAvailable)
+	if conditionsEqual(current, desired) { // Comparison ignores lastUpdated
+		return nil
+	}
+
+	co.SetCondition(desired)
+
+	return nil
+}
+
+func (r *ClusterOperatorReconciler) setDegraded(_ context.Context, co *ClusterOperator) error {
+	desired := &configv1.ClusterOperatorStatusCondition{
+		Type:               configv1.OperatorDegraded,
+		LastTransitionTime: r.Now(),
+	}
+
+	if r.syncTracker.SuccessfulSyncs() > 0 && reflect.DeepEqual(co.Status.Versions, r.TargetVersions) {
+		desired.Status = configv1.ConditionFalse
+	} else {
+		desired.Status = configv1.ConditionTrue
+		desired.Message = "Waiting for updates to take effect"
+	}
+
+	current := co.GetCondition(configv1.OperatorDegraded)
+	if conditionsEqual(current, desired) { // Comparison ignores lastUpdated
+		return nil
+	}
+
+	co.SetCondition(desired)
+
+	return nil
+}
+
+const (
+	IncompatibleOperatorsInstalled = "IncompatibleOperatorsInstalled"
+)
+
+func (r *ClusterOperatorReconciler) setUpgradeable(ctx context.Context, co *ClusterOperator) error {
+	desired := &configv1.ClusterOperatorStatusCondition{
+		Type:               configv1.OperatorUpgradeable,
+		Status:             configv1.ConditionTrue,
+		LastTransitionTime: r.Now(),
+	}
+
+	// Set upgradeable=false if (either/or):
+	// 1. OLM currently upgrading (takes priorty in the status message)
+	// 2. Operators currently installed that are incompatible with the next OCP minor version
+	if r.syncTracker.SuccessfulSyncs() < 1 || !reflect.DeepEqual(co.Status.Versions, r.TargetVersions) {
+		// OLM is still upgrading
+		desired.Status = configv1.ConditionFalse
+		desired.Message = "Waiting for updates to take effect"
+	} else {
+		incompatible, err := incompatibleOperators(ctx, r.Client)
+		if err != nil {
+			return err
+		}
+
+		if len(incompatible) > 0 {
+			// Some incompatible operator is installed
+			desired.Status = configv1.ConditionFalse
+			desired.Reason = IncompatibleOperatorsInstalled
+			desired.Message = incompatible.String() // TODO: Truncate message to field length
+		}
+	}
+
+	current := co.GetCondition(configv1.OperatorUpgradeable)
+	if conditionsEqual(current, desired) { // Comparison ignores lastUpdated
+		return nil
+	}
+
+	co.SetCondition(desired)
+
+	return nil
+}