add kubelet skew check for MCO upgradeable

Co-authored-by: Qi Wang <[email protected]>
Signed-off-by: Qi Wang <[email protected]>

Author: deads2k

Diff for: cmd/machine-config-operator/start.go

@@ -75,6 +75,7 @@ func runStartCmd(cmd *cobra.Command, args []string) {
 			ctrlctx.ClientBuilder.APIExtClientOrDie(componentName),
 			ctrlctx.ClientBuilder.ConfigClientOrDie(componentName),
 			ctrlctx.OpenShiftKubeAPIServerKubeNamespacedInformerFactory.Core().V1().ConfigMaps(),
+			ctrlctx.KubeInformerFactory.Core().V1().Nodes(),
 		)
 
 		ctrlctx.NamespacedInformerFactory.Start(ctrlctx.Stop)

Diff for: pkg/operator/operator.go

@@ -81,6 +81,7 @@ type Operator struct {
 	clusterCmLister  corelisterv1.ConfigMapLister
 	proxyLister      configlistersv1.ProxyLister
 	oseKubeAPILister corelisterv1.ConfigMapLister
+	nodeLister       corelisterv1.NodeLister
 	dnsLister        configlistersv1.DNSLister
 
 	crdListerSynced                  cache.InformerSynced
@@ -98,6 +99,7 @@ type Operator struct {
 	clusterRoleBindingInformerSynced cache.InformerSynced
 	proxyListerSynced                cache.InformerSynced
 	oseKubeAPIListerSynced           cache.InformerSynced
+	nodeListerSynced                 cache.InformerSynced
 	dnsListerSynced                  cache.InformerSynced
 
 	// queue only ever has one item, but it has nice error handling backoff/retry semantics
@@ -131,6 +133,7 @@ func New(
 	apiExtClient apiextclientset.Interface,
 	configClient configclientset.Interface,
 	oseKubeAPIInformer coreinformersv1.ConfigMapInformer,
+	nodeInformer coreinformersv1.NodeInformer,
 ) *Operator {
 	eventBroadcaster := record.NewBroadcaster()
 	eventBroadcaster.StartLogging(glog.Infof)
@@ -163,6 +166,7 @@ func New(
 		mcpInformer.Informer(),
 		proxyInformer.Informer(),
 		oseKubeAPIInformer.Informer(),
+		nodeInformer.Informer(),
 		dnsInformer.Informer(),
 	} {
 		i.AddEventHandler(optr.eventHandler())
@@ -182,6 +186,8 @@ func New(
 	optr.proxyListerSynced = proxyInformer.Informer().HasSynced
 	optr.oseKubeAPILister = oseKubeAPIInformer.Lister()
 	optr.oseKubeAPIListerSynced = oseKubeAPIInformer.Informer().HasSynced
+	optr.nodeLister = nodeInformer.Lister()
+	optr.nodeListerSynced = nodeInformer.Informer().HasSynced
 
 	optr.serviceAccountInformerSynced = serviceAccountInfomer.Informer().HasSynced
 	optr.clusterRoleInformerSynced = clusterRoleInformer.Informer().HasSynced
@@ -235,6 +241,7 @@ func (optr *Operator) Run(workers int, stopCh <-chan struct{}) {
 		optr.networkListerSynced,
 		optr.proxyListerSynced,
 		optr.oseKubeAPIListerSynced,
+		optr.nodeListerSynced,
 		optr.mcpListerSynced,
 		optr.mcListerSynced,
 		optr.dnsListerSynced) {

Diff for: pkg/operator/status.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"strconv"
+	"strings"
 
 	"github.com/golang/glog"
 	configv1 "github.com/openshift/api/config/v1"
@@ -16,6 +18,7 @@ import (
 	"k8s.io/apimachinery/pkg/labels"
 
 	mcfgv1 "github.com/openshift/machine-config-operator/pkg/apis/machineconfiguration.openshift.io/v1"
+	v1 "github.com/openshift/machine-config-operator/pkg/apis/machineconfiguration.openshift.io/v1"
 	ctrlcommon "github.com/openshift/machine-config-operator/pkg/controller/common"
 )
 
@@ -243,6 +246,13 @@ func (optr *Operator) syncDegradedStatus(ierr syncError) (err error) {
 	return optr.updateStatus(co, coStatus)
 }
 
+const (
+	skewUnchecked   = "KubeletSkewUnchecked"
+	skewSupported   = "KubeletSkewSupported"
+	skewUnsupported = "KubeletSkewUnsupported"
+	skewPresent     = "KubeletSkewPresent"
+)
+
 // syncUpgradeableStatus applies the new condition to the mco's ClusterOperator object.
 func (optr *Operator) syncUpgradeableStatus() error {
 	co, err := optr.fetchClusterOperator()
@@ -289,9 +299,168 @@ func (optr *Operator) syncUpgradeableStatus() error {
 		coStatus.Message = "One or more machine config pools are updating, please see `oc get mcp` for further details"
 	}
 
+	// don't overwrite status if updating or degraded
+	if !updating && !degraded {
+		skewStatus, status, err := optr.isKubeletSkewSupported(pools)
+		if err != nil {
+			glog.Errorf("Error checking version skew: %v, kubelet skew status: %v, status reason: %v, status message: %v", err, skewStatus, status.Reason, status.Message)
+			coStatus.Reason = status.Reason
+			coStatus.Message = status.Message
+			return optr.updateStatus(co, coStatus)
+		}
+		switch skewStatus {
+		case skewUnchecked:
+			coStatus.Reason = status.Reason
+			coStatus.Message = status.Message
+			return optr.updateStatus(co, coStatus)
+		case skewUnsupported:
+			coStatus.Reason = status.Reason
+			coStatus.Message = status.Message
+			mcoObjectRef := &corev1.ObjectReference{
+				Kind:      co.Kind,
+				Name:      co.Name,
+				Namespace: co.Namespace,
+				UID:       co.GetUID(),
+			}
+			glog.Infof("kubelet skew status: %v, status reason: %v", skewStatus, status.Reason)
+			optr.eventRecorder.Eventf(mcoObjectRef, corev1.EventTypeWarning, coStatus.Reason, coStatus.Message)
+			return optr.updateStatus(co, coStatus)
+		case skewPresent:
+			coStatus.Reason = status.Reason
+			coStatus.Message = status.Message
+			glog.Infof("kubelet skew status: %v, status reason: %v", skewStatus, status.Reason)
+			return optr.updateStatus(co, coStatus)
+		}
+	}
 	return optr.updateStatus(co, coStatus)
 }
 
+// isKubeletSkewSupported checks the version skew of kube-apiserver and node kubelet version.
+// Returns the skew status. version skew > 2 is not supported.
+func (optr *Operator) isKubeletSkewSupported(pools []*v1.MachineConfigPool) (skewStatus string, coStatus configv1.ClusterOperatorStatusCondition, err error) {
+	coStatus = configv1.ClusterOperatorStatusCondition{}
+	kubeAPIServerStatus, err := optr.configClient.ConfigV1().ClusterOperators().Get(context.TODO(), "kube-apiserver", metav1.GetOptions{})
+	if err != nil {
+		coStatus.Reason = skewUnchecked
+		coStatus.Message = fmt.Sprintf("An error occurred when checking kubelet version skew: %v", err)
+		return skewUnchecked, coStatus, err
+	}
+	// looks like
+	//   - name: kube-apiserver
+	//     version: 1.21.0-rc.0
+	kubeAPIServerVersion := ""
+	for _, version := range kubeAPIServerStatus.Status.Versions {
+		if version.Name != "kube-apiserver" {
+			continue
+		}
+		kubeAPIServerVersion = version.Version
+		break
+	}
+	if kubeAPIServerVersion == "" {
+		err = fmt.Errorf("kube-apiserver does not yet have a version")
+		coStatus.Reason = skewUnchecked
+		coStatus.Message = fmt.Sprintf("An error occurred when checking kubelet version skew: %v", err.Error())
+		return skewUnchecked, coStatus, err
+	}
+	kubeAPIServerMinorVersion, err := getMinorKubeletVersion(kubeAPIServerVersion)
+	if err != nil {
+		coStatus.Reason = skewUnchecked
+		coStatus.Message = fmt.Sprintf("An error occurred when checking kubelet version skew: %v", err)
+		return skewUnchecked, coStatus, err
+	}
+	var (
+		lastError      error
+		kubeletVersion string
+	)
+	nodes, err := optr.GetAllManagedNodes(pools)
+	if err != nil {
+		err = fmt.Errorf("getting all managed nodes failed: %v", err)
+		coStatus.Reason = skewUnchecked
+		coStatus.Message = fmt.Sprintf("An error occurred when getting all the managed nodes: %v", err.Error())
+	}
+	for _, node := range nodes {
+		// looks like kubeletVersion: v1.21.0-rc.0+6143dea
+		kubeletVersion = node.Status.NodeInfo.KubeletVersion
+		if kubeletVersion == "" {
+			continue
+		}
+		nodeMinorVersion, err := getMinorKubeletVersion(kubeletVersion)
+		if err != nil {
+			lastError = err
+			continue
+		}
+		if nodeMinorVersion+2 < kubeAPIServerMinorVersion {
+			coStatus.Reason = skewUnsupported
+			coStatus.Message = fmt.Sprintf("One or more nodes have an unsupported kubelet version skew. Please see `oc get nodes` for details and upgrade all nodes so that they have a kubelet version of at least %v.", getMinimalSkewSupportNodeVersion(kubeAPIServerVersion))
+			return skewUnsupported, coStatus, nil
+		}
+		if nodeMinorVersion+2 == kubeAPIServerMinorVersion {
+			coStatus.Reason = skewPresent
+			coStatus.Message = fmt.Sprintf("Current kubelet version %v will not be supported by newer kube-apiserver. Please upgrade the kubelet first if plan to upgrade the kube-apiserver", kubeletVersion)
+			return skewPresent, coStatus, nil
+		}
+	}
+	if kubeletVersion == "" {
+		err = fmt.Errorf("kubelet does not yet have a version")
+		coStatus.Reason = skewUnchecked
+		coStatus.Message = fmt.Sprintf("An error occurred when checking kubelet version skew: %v", err.Error())
+		return skewUnchecked, coStatus, err
+	}
+	if lastError != nil {
+		coStatus.Reason = skewUnchecked
+		coStatus.Message = fmt.Sprintf("An error occurred when checking kubelet version skew: %v", err)
+		return skewUnchecked, coStatus, lastError
+	}
+	return skewSupported, coStatus, nil
+}
+
+// GetAllManagedNodes returns the nodes managed by MCO
+func (optr *Operator) GetAllManagedNodes(pools []*v1.MachineConfigPool) ([]*corev1.Node, error) {
+	nodes := []*corev1.Node{}
+	for _, pool := range pools {
+		selector, err := metav1.LabelSelectorAsSelector(pool.Spec.NodeSelector)
+		if err != nil {
+			return nil, fmt.Errorf("label selector for pool %v failed %v", pool.Name, err)
+		}
+		poolNodes, err := optr.nodeLister.List(selector)
+		if err != nil {
+			return nil, fmt.Errorf("could not list nodes for pool %v with error %v", pool.Name, err)
+		}
+		nodes = append(nodes, poolNodes...)
+	}
+	return nodes, nil
+}
+
+// getMinorKubeletVersion parses the minor version number of kubelet
+func getMinorKubeletVersion(version string) (int, error) {
+	tokens := strings.Split(version, ".")
+	if len(tokens) < 2 {
+		return 0, fmt.Errorf("incorrect version syntax: %q", version)
+	}
+	minorVersion, err := strconv.ParseInt(tokens[1], 10, 32)
+	if err != nil {
+		return 0, err
+	}
+	return int(minorVersion), nil
+}
+
+// getMinimalSkewSupportNodeVersion returns the minimal supported node kubelet version.
+func getMinimalSkewSupportNodeVersion(version string) string {
+	// drop the pre-release and commit hash
+	if strings.Contains(version, "-") {
+		version = version[:strings.Index(version, "-")]
+	}
+	if strings.Contains(version, "+") {
+		version = version[:strings.Index(version, "+")]
+	}
+	tokens := strings.Split(version, ".")
+	if minorVersion, err := strconv.ParseInt(tokens[1], 10, 32); err == nil {
+		tokens[1] = strconv.Itoa(int(minorVersion) - 2)
+		return strings.Join(tokens, ".")
+	}
+	return version
+}
+
 func (optr *Operator) fetchClusterOperator() (*configv1.ClusterOperator, error) {
 	co, err := optr.configClient.ConfigV1().ClusterOperators().Get(context.TODO(), optr.name, metav1.GetOptions{})
 	if meta.IsNoMatchError(err) {

Diff for: pkg/operator/status_test.go

@@ -7,6 +7,10 @@ import (
 	"reflect"
 	"testing"
 
+	corelisterv1 "k8s.io/client-go/listers/core/v1"
+	clientgotesting "k8s.io/client-go/testing"
+	"k8s.io/client-go/tools/cache"
+
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
@@ -20,6 +24,7 @@ import (
 	cov1helpers "github.com/openshift/library-go/pkg/config/clusteroperator/v1helpers"
 	mcfgv1 "github.com/openshift/machine-config-operator/pkg/apis/machineconfiguration.openshift.io/v1"
 	ctrlcommon "github.com/openshift/machine-config-operator/pkg/controller/common"
+	"github.com/openshift/machine-config-operator/test/helpers"
 )
 
 func TestIsMachineConfigPoolConfigurationValid(t *testing.T) {
@@ -152,12 +157,23 @@ func TestIsMachineConfigPoolConfigurationValid(t *testing.T) {
 	}
 }
 
-type mockMCPLister struct{}
+type mockMCPLister struct {
+	pools []*mcfgv1.MachineConfigPool
+}
 
 func (mcpl *mockMCPLister) List(selector labels.Selector) (ret []*mcfgv1.MachineConfigPool, err error) {
-	return nil, nil
+	return mcpl.pools, nil
 }
 func (mcpl *mockMCPLister) Get(name string) (ret *mcfgv1.MachineConfigPool, err error) {
+	if mcpl.pools == nil {
+		return nil, nil
+	}
+	for _, pool := range mcpl.pools {
+		if pool.Name == name {
+			return pool, nil
+		}
+
+	}
 	return nil, nil
 }
 
@@ -550,7 +566,24 @@ func TestOperatorSyncStatus(t *testing.T) {
 			eventRecorder: &record.FakeRecorder{},
 		}
 		optr.vStore = newVersionStore()
-		optr.mcpLister = &mockMCPLister{}
+		optr.mcpLister = &mockMCPLister{
+			pools: []*mcfgv1.MachineConfigPool{
+				helpers.NewMachineConfigPool("master", nil, helpers.MasterSelector, "v0"),
+				helpers.NewMachineConfigPool("workers", nil, helpers.WorkerSelector, "v0"),
+			},
+		}
+
+		nodeIndexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc})
+		optr.nodeLister = corelisterv1.NewNodeLister(nodeIndexer)
+		nodeIndexer.Add(&corev1.Node{
+			ObjectMeta: metav1.ObjectMeta{Name: "first-node", Labels: map[string]string{"node-role/worker": ""}},
+			Status: corev1.NodeStatus{
+				NodeInfo: corev1.NodeSystemInfo{
+					KubeletVersion: "v1.21",
+				},
+			},
+		})
+
 		coName := fmt.Sprintf("test-%s", uuid.NewUUID())
 		co := &configv1.ClusterOperator{ObjectMeta: metav1.ObjectMeta{Name: coName}}
 		cov1helpers.SetStatusCondition(&co.Status.Conditions, configv1.ClusterOperatorStatusCondition{Type: configv1.OperatorAvailable, Status: configv1.ConditionFalse})
@@ -559,6 +592,14 @@ func TestOperatorSyncStatus(t *testing.T) {
 		cov1helpers.SetStatusCondition(&co.Status.Conditions, configv1.ClusterOperatorStatusCondition{Type: configv1.OperatorUpgradeable, Status: configv1.ConditionUnknown})
 		co.Status.Versions = append(co.Status.Versions, configv1.OperandVersion{Name: "operator", Version: "test-version"})
 		optr.name = coName
+		kasOperator := &configv1.ClusterOperator{
+			ObjectMeta: metav1.ObjectMeta{Name: "kube-apiserver"},
+			Status: configv1.ClusterOperatorStatus{
+				Versions: []configv1.OperandVersion{
+					{Name: "kube-apiserver", Version: "1.21"},
+				},
+			},
+		}
 
 		for j, sync := range testCase.syncs {
 			optr.inClusterBringup = sync.inClusterBringUp
@@ -567,7 +608,7 @@ func TestOperatorSyncStatus(t *testing.T) {
 			} else {
 				optr.vStore.Set("operator", "test-version")
 			}
-			optr.configClient = fakeconfigclientset.NewSimpleClientset(co)
+			optr.configClient = fakeconfigclientset.NewSimpleClientset(co, kasOperator)
 			err := optr.syncAll(sync.syncFuncs)
 			if sync.expectOperatorFail {
 				assert.NotNil(t, err, "test case %d, sync call %d, expected an error", idx, j)
@@ -597,12 +638,179 @@ func TestInClusterBringUpStayOnErr(t *testing.T) {
 	}
 	optr.vStore = newVersionStore()
 	optr.vStore.Set("operator", "test-version")
-	optr.mcpLister = &mockMCPLister{}
+	optr.mcpLister = &mockMCPLister{
+		pools: []*mcfgv1.MachineConfigPool{
+			helpers.NewMachineConfigPool("master", nil, helpers.MasterSelector, "v0"),
+			helpers.NewMachineConfigPool("workers", nil, helpers.WorkerSelector, "v0"),
+		},
+	}
+	nodeIndexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc})
+	optr.nodeLister = corelisterv1.NewNodeLister(nodeIndexer)
+	nodeIndexer.Add(&corev1.Node{
+		ObjectMeta: metav1.ObjectMeta{Name: "first-node", Labels: map[string]string{"node-role/worker": ""}},
+		Status: corev1.NodeStatus{
+			NodeInfo: corev1.NodeSystemInfo{
+				KubeletVersion: "v1.21",
+			},
+		},
+	})
+	co := &configv1.ClusterOperator{}
+	kasOperator := &configv1.ClusterOperator{
+		ObjectMeta: metav1.ObjectMeta{Name: "kube-apiserver"},
+		Status: configv1.ClusterOperatorStatus{
+			Versions: []configv1.OperandVersion{
+				{Name: "kube-apiserver", Version: "1.21"},
+			},
+		},
+	}
+	cov1helpers.SetStatusCondition(&co.Status.Conditions, configv1.ClusterOperatorStatusCondition{Type: configv1.OperatorAvailable, Status: configv1.ConditionFalse})
+	cov1helpers.SetStatusCondition(&co.Status.Conditions, configv1.ClusterOperatorStatusCondition{Type: configv1.OperatorProgressing, Status: configv1.ConditionFalse})
+	cov1helpers.SetStatusCondition(&co.Status.Conditions, configv1.ClusterOperatorStatusCondition{Type: configv1.OperatorDegraded, Status: configv1.ConditionFalse})
+	optr.configClient = fakeconfigclientset.NewSimpleClientset(co, kasOperator)
+	optr.inClusterBringup = true
+
+	fn1 := func(config *renderConfig) error {
+		return errors.New("mocked fn1")
+	}
+	err := optr.syncAll([]syncFunc{{name: "mock1", fn: fn1}})
+	assert.NotNil(t, err, "expected syncAll to fail")
+
+	assert.True(t, optr.inClusterBringup)
+
+	fn1 = func(config *renderConfig) error {
+		return nil
+	}
+	err = optr.syncAll([]syncFunc{{name: "mock1", fn: fn1}})
+	assert.Nil(t, err, "expected syncAll to pass")
+
+	assert.False(t, optr.inClusterBringup)
+}
+
+func TestKubeletSkewUnSupported(t *testing.T) {
+	kasOperator := &configv1.ClusterOperator{
+		ObjectMeta: metav1.ObjectMeta{Name: "kube-apiserver"},
+		Status: configv1.ClusterOperatorStatus{
+			Versions: []configv1.OperandVersion{
+				{Name: "kube-apiserver", Version: "1.21"},
+			},
+		},
+	}
+	optr := &Operator{
+		eventRecorder: &record.FakeRecorder{},
+	}
+	optr.vStore = newVersionStore()
+	optr.vStore.Set("operator", "test-version")
+	optr.mcpLister = &mockMCPLister{
+		pools: []*mcfgv1.MachineConfigPool{
+			helpers.NewMachineConfigPool("master", nil, helpers.MasterSelector, "v0"),
+			helpers.NewMachineConfigPool("workers", nil, helpers.WorkerSelector, "v0"),
+		},
+	}
+	nodeIndexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc})
+	optr.nodeLister = corelisterv1.NewNodeLister(nodeIndexer)
+	nodeIndexer.Add(&corev1.Node{
+		ObjectMeta: metav1.ObjectMeta{Name: "first-node", Labels: map[string]string{"node-role/worker": ""}},
+		Status: corev1.NodeStatus{
+			NodeInfo: corev1.NodeSystemInfo{
+				KubeletVersion: "v1.18",
+			},
+		},
+	})
+
+	co := &configv1.ClusterOperator{}
+	cov1helpers.SetStatusCondition(&co.Status.Conditions, configv1.ClusterOperatorStatusCondition{Type: configv1.OperatorAvailable, Status: configv1.ConditionFalse})
+	cov1helpers.SetStatusCondition(&co.Status.Conditions, configv1.ClusterOperatorStatusCondition{Type: configv1.OperatorProgressing, Status: configv1.ConditionFalse})
+	cov1helpers.SetStatusCondition(&co.Status.Conditions, configv1.ClusterOperatorStatusCondition{Type: configv1.OperatorDegraded, Status: configv1.ConditionFalse})
+	fakeClient := fakeconfigclientset.NewSimpleClientset(co, kasOperator)
+	optr.configClient = fakeClient
+	optr.inClusterBringup = true
+
+	fn1 := func(config *renderConfig) error {
+		return errors.New("mocked fn1")
+	}
+	err := optr.syncAll([]syncFunc{{name: "mock1", fn: fn1}})
+	assert.NotNil(t, err, "expected syncAll to fail")
+
+	assert.True(t, optr.inClusterBringup)
+
+	fn1 = func(config *renderConfig) error {
+		return nil
+	}
+	err = optr.syncAll([]syncFunc{{name: "mock1", fn: fn1}})
+	assert.Nil(t, err, "expected syncAll to pass")
+
+	assert.False(t, optr.inClusterBringup)
+
+	var lastUpdate clientgotesting.UpdateAction
+	for _, action := range fakeClient.Actions() {
+		if action.GetVerb() == "update" {
+			lastUpdate = action.(clientgotesting.UpdateAction)
+		}
+	}
+	if lastUpdate == nil {
+		t.Fatal("missing update")
+	}
+	operatorStatus := lastUpdate.GetObject().(*configv1.ClusterOperator)
+	var upgradeable *configv1.ClusterOperatorStatusCondition
+	for _, condition := range operatorStatus.Status.Conditions {
+		if condition.Type == configv1.OperatorUpgradeable {
+			upgradeable = &condition
+			break
+		}
+	}
+	if upgradeable == nil {
+		t.Fatal("missing condition")
+	}
+	if upgradeable.Status != configv1.ConditionTrue {
+		t.Fatal(upgradeable)
+	}
+	if upgradeable.Message != "One or more nodes have an unsupported kubelet version skew. Please see `oc get nodes` for details and upgrade all nodes so that they have a kubelet version of at least 1.19." {
+		t.Fatal(upgradeable)
+	}
+	if upgradeable.Reason != "KubeletSkewUnsupported" {
+		t.Fatal(upgradeable)
+	}
+}
+
+func TestCustomPoolKubeletSkewUnSupported(t *testing.T) {
+	customSelector := metav1.AddLabelToSelector(&metav1.LabelSelector{}, "node-role/custom", "")
+	kasOperator := &configv1.ClusterOperator{
+		ObjectMeta: metav1.ObjectMeta{Name: "kube-apiserver"},
+		Status: configv1.ClusterOperatorStatus{
+			Versions: []configv1.OperandVersion{
+				{Name: "kube-apiserver", Version: "1.21"},
+			},
+		},
+	}
+	optr := &Operator{
+		eventRecorder: &record.FakeRecorder{},
+	}
+	optr.vStore = newVersionStore()
+	optr.vStore.Set("operator", "test-version")
+	optr.mcpLister = &mockMCPLister{
+		pools: []*mcfgv1.MachineConfigPool{
+			helpers.NewMachineConfigPool("master", nil, helpers.MasterSelector, "v0"),
+			helpers.NewMachineConfigPool("workers", nil, helpers.WorkerSelector, "v0"),
+			helpers.NewMachineConfigPool("custom", nil, customSelector, "v0"),
+		},
+	}
+	nodeIndexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc})
+	optr.nodeLister = corelisterv1.NewNodeLister(nodeIndexer)
+	nodeIndexer.Add(&corev1.Node{
+		ObjectMeta: metav1.ObjectMeta{Name: "custom", Labels: map[string]string{"node-role/custom": ""}},
+		Status: corev1.NodeStatus{
+			NodeInfo: corev1.NodeSystemInfo{
+				KubeletVersion: "v1.18",
+			},
+		},
+	})
+
 	co := &configv1.ClusterOperator{}
 	cov1helpers.SetStatusCondition(&co.Status.Conditions, configv1.ClusterOperatorStatusCondition{Type: configv1.OperatorAvailable, Status: configv1.ConditionFalse})
 	cov1helpers.SetStatusCondition(&co.Status.Conditions, configv1.ClusterOperatorStatusCondition{Type: configv1.OperatorProgressing, Status: configv1.ConditionFalse})
 	cov1helpers.SetStatusCondition(&co.Status.Conditions, configv1.ClusterOperatorStatusCondition{Type: configv1.OperatorDegraded, Status: configv1.ConditionFalse})
-	optr.configClient = fakeconfigclientset.NewSimpleClientset(co)
+	fakeClient := fakeconfigclientset.NewSimpleClientset(co, kasOperator)
+	optr.configClient = fakeClient
 	optr.inClusterBringup = true
 
 	fn1 := func(config *renderConfig) error {
@@ -620,4 +828,156 @@ func TestInClusterBringUpStayOnErr(t *testing.T) {
 	assert.Nil(t, err, "expected syncAll to pass")
 
 	assert.False(t, optr.inClusterBringup)
+
+	var lastUpdate clientgotesting.UpdateAction
+	for _, action := range fakeClient.Actions() {
+		if action.GetVerb() == "update" {
+			lastUpdate = action.(clientgotesting.UpdateAction)
+		}
+	}
+	if lastUpdate == nil {
+		t.Fatal("missing update")
+	}
+	operatorStatus := lastUpdate.GetObject().(*configv1.ClusterOperator)
+	var upgradeable *configv1.ClusterOperatorStatusCondition
+	for _, condition := range operatorStatus.Status.Conditions {
+		if condition.Type == configv1.OperatorUpgradeable {
+			upgradeable = &condition
+			break
+		}
+	}
+	if upgradeable == nil {
+		t.Fatal("missing condition")
+	}
+	if upgradeable.Status != configv1.ConditionTrue {
+		t.Fatal(upgradeable)
+	}
+	if upgradeable.Message != "One or more nodes have an unsupported kubelet version skew. Please see `oc get nodes` for details and upgrade all nodes so that they have a kubelet version of at least 1.19." {
+		t.Fatal(upgradeable)
+	}
+	if upgradeable.Reason != "KubeletSkewUnsupported" {
+		t.Fatal(upgradeable)
+	}
+}
+
+func TestKubeletSkewSupported(t *testing.T) {
+	kasOperator := &configv1.ClusterOperator{
+		ObjectMeta: metav1.ObjectMeta{Name: "kube-apiserver"},
+		Status: configv1.ClusterOperatorStatus{
+			Versions: []configv1.OperandVersion{
+				{Name: "kube-apiserver", Version: "1.21"},
+			},
+		},
+	}
+	optr := &Operator{
+		eventRecorder: &record.FakeRecorder{},
+	}
+	optr.vStore = newVersionStore()
+	optr.vStore.Set("operator", "test-version")
+	optr.mcpLister = &mockMCPLister{
+		pools: []*mcfgv1.MachineConfigPool{
+			helpers.NewMachineConfigPool("master", nil, helpers.MasterSelector, "v0"),
+			helpers.NewMachineConfigPool("workers", nil, helpers.WorkerSelector, "v0"),
+		},
+	}
+	nodeIndexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc})
+	optr.nodeLister = corelisterv1.NewNodeLister(nodeIndexer)
+	nodeIndexer.Add(&corev1.Node{
+		ObjectMeta: metav1.ObjectMeta{Name: "first-node", Labels: map[string]string{"node-role/worker": ""}},
+		Status: corev1.NodeStatus{
+			NodeInfo: corev1.NodeSystemInfo{
+				KubeletVersion: "v1.20",
+			},
+		},
+	})
+
+	co := &configv1.ClusterOperator{}
+	cov1helpers.SetStatusCondition(&co.Status.Conditions, configv1.ClusterOperatorStatusCondition{Type: configv1.OperatorAvailable, Status: configv1.ConditionFalse})
+	cov1helpers.SetStatusCondition(&co.Status.Conditions, configv1.ClusterOperatorStatusCondition{Type: configv1.OperatorProgressing, Status: configv1.ConditionFalse})
+	cov1helpers.SetStatusCondition(&co.Status.Conditions, configv1.ClusterOperatorStatusCondition{Type: configv1.OperatorDegraded, Status: configv1.ConditionFalse})
+	fakeClient := fakeconfigclientset.NewSimpleClientset(co, kasOperator)
+	optr.configClient = fakeClient
+	optr.inClusterBringup = true
+
+	fn1 := func(config *renderConfig) error {
+		return errors.New("mocked fn1")
+	}
+	err := optr.syncAll([]syncFunc{{name: "mock1", fn: fn1}})
+	assert.NotNil(t, err, "expected syncAll to fail")
+
+	assert.True(t, optr.inClusterBringup)
+
+	fn1 = func(config *renderConfig) error {
+		return nil
+	}
+	err = optr.syncAll([]syncFunc{{name: "mock1", fn: fn1}})
+	assert.Nil(t, err, "expected syncAll to pass")
+
+	assert.False(t, optr.inClusterBringup)
+
+	var lastUpdate clientgotesting.UpdateAction
+	for _, action := range fakeClient.Actions() {
+		if action.GetVerb() == "update" {
+			lastUpdate = action.(clientgotesting.UpdateAction)
+		}
+	}
+	if lastUpdate == nil {
+		t.Fatal("missing update")
+	}
+	operatorStatus := lastUpdate.GetObject().(*configv1.ClusterOperator)
+	var upgradeable *configv1.ClusterOperatorStatusCondition
+	for _, condition := range operatorStatus.Status.Conditions {
+		if condition.Type == configv1.OperatorUpgradeable {
+			upgradeable = &condition
+			break
+		}
+	}
+	if upgradeable == nil {
+		t.Fatal("missing condition")
+	}
+	if upgradeable.Status != configv1.ConditionTrue {
+		t.Fatal(upgradeable)
+	}
+	if upgradeable.Message != "" {
+		t.Fatal(upgradeable)
+	}
+	if upgradeable.Reason != "AsExpected" {
+		t.Fatal(upgradeable)
+	}
+}
+
+func TestGetMinorKubeletVersion(t *testing.T) {
+	tcs := []struct {
+		version      string
+		minor        int
+		expectNilErr bool
+	}{
+		{"v1.20.1", 20, true},
+		{"v1.20.1+abc0", 20, true},
+		{"v1.20.1+0123", 20, true},
+		{"v1.20.1-rc", 20, true},
+		{"v1.20.1-rc.1", 20, true},
+		{"v1.20.1-rc+abc123", 20, true},
+		{"v1.20.1-rc.0+abc123", 20, true},
+		{"v1.20.1", 20, true},
+		{"1.20.1", 20, true},
+		{"1.20", 20, true},
+		{"12", 0, false},
+		{".xy", 0, false},
+		{"1.xy.1", 0, false},
+	}
+	for _, tc := range tcs {
+		minorV, err := getMinorKubeletVersion(tc.version)
+		if tc.expectNilErr && err != nil {
+			t.Errorf("test %q failed: unexpected error %v", tc.version, err)
+			continue
+		}
+		if !tc.expectNilErr && err == nil {
+			t.Errorf("test %q failed: expected error, got nil ", tc.version)
+			continue
+		}
+		if tc.expectNilErr {
+			assert.Equal(t, tc.minor, minorV, fmt.Sprintf("failed test %q", tc.version))
+		}
+	}
 }