MCO-1009: MCO-1008: MCO-905: Implement NodeDisruptionPolicy API

cmd/machine-config-daemon/start.go

@@ -182,6 +182,7 @@ func runStartCmd(_ *cobra.Command, _ []string) {
 		ctrlctx.InformerFactory.Machineconfiguration().V1().MachineConfigs(),
 		ctrlctx.KubeInformerFactory.Core().V1().Nodes(),
 		ctrlctx.InformerFactory.Machineconfiguration().V1().ControllerConfigs(),
+		ctrlctx.ClientBuilder.OperatorClientOrDie(componentName),
 		startOpts.kubeletHealthzEnabled,
 		startOpts.kubeletHealthzEndpoint,
 		ctrlctx.FeatureGateAccess,
@@ -194,6 +195,7 @@ func runStartCmd(_ *cobra.Command, _ []string) {
 	ctrlctx.KubeInformerFactory.Start(stopCh)
 	ctrlctx.KubeNamespacedInformerFactory.Start(stopCh)
 	ctrlctx.InformerFactory.Start(stopCh)
+	ctrlctx.OperatorInformerFactory.Start(stopCh)
 	close(ctrlctx.InformersStarted)
 
 	select {

manifests/machineconfigdaemon/clusterrole.yaml

@@ -19,6 +19,9 @@ rules:
   resourceNames: ["privileged"]
   resources: ["securitycontextconstraints"]
   verbs: ["use"]
+- apiGroups: ["operator.openshift.io"]
+  resources: ["machineconfigurations"]
+  verbs: ["get", "list", "watch"]
 - apiGroups:
   - authentication.k8s.io
   resources:

pkg/apihelpers/apihelpers.go

@@ -8,8 +8,73 @@ import (
 	"fmt"
 
 	mcfgv1 "github.com/openshift/api/machineconfiguration/v1"
+	opv1 "github.com/openshift/api/operator/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/klog/v2"
+)
+
+var (
+	// This is the list of MCO's default node disruption policies.
+	defaultClusterPolicies = opv1.NodeDisruptionPolicyClusterStatus{
+		Files: []opv1.NodeDisruptionPolicyStatusFile{
+			{
+				Path: "/etc/mco/internal-registry-pull-secret.json",
+				Actions: []opv1.NodeDisruptionPolicyStatusAction{
+					{
+						Type: opv1.NoneStatusAction,
+					},
+				},
+			},
+			{
+				Path: "/var/lib/kubelet/config.json",
+				Actions: []opv1.NodeDisruptionPolicyStatusAction{
+					{
+						Type: opv1.NoneStatusAction,
+					},
+				},
+			},
+			{
+				Path: "/etc/machine-config-daemon/no-reboot/containers-gpg.pub",
+				Actions: []opv1.NodeDisruptionPolicyStatusAction{
+					{
+						Type: opv1.ReloadStatusAction,
+						Reload: &opv1.ReloadService{
+							ServiceName: "crio.service",
+						},
+					},
+				},
+			},
+			{
+				Path: "/etc/containers/policy.json",
+				Actions: []opv1.NodeDisruptionPolicyStatusAction{
+					{
+						Type: opv1.ReloadStatusAction,
+						Reload: &opv1.ReloadService{
+							ServiceName: "crio.service",
+						},
+					},
+				},
+			},
+			{
+				Path: "/etc/containers/registries.conf",
+				Actions: []opv1.NodeDisruptionPolicyStatusAction{
+					{
+						Type: opv1.SpecialStatusAction,
+					},
+				},
+			},
+		},
+		SSHKey: opv1.NodeDisruptionPolicyStatusSSHKey{
+			Actions: []opv1.NodeDisruptionPolicyStatusAction{
+				{
+					Type: opv1.NoneStatusAction,
+				},
+			},
+		},
+	}
 )
 
 // NewMachineConfigPoolCondition creates a new MachineConfigPool condition.
@@ -214,3 +279,120 @@ func IsControllerConfigCompleted(ccName string, ccGetter func(string) (*mcfgv1.C
 	}
 	return fmt.Errorf("ControllerConfig has not completed: completed(%v) running(%v) failing(%v)", completed, running, failing)
 }
+
+// Merges the cluster's default node disruption policies with the user defined policies, if any.
+func MergeClusterPolicies(userDefinedClusterPolicies opv1.NodeDisruptionPolicyConfig) opv1.NodeDisruptionPolicyClusterStatus {
+
+	mergedClusterPolicies := opv1.NodeDisruptionPolicyClusterStatus{}
+
+	// Add default file policies to the merged list.
+	mergedClusterPolicies.Files = append(mergedClusterPolicies.Files, defaultClusterPolicies.Files...)
+
+	// Iterate through user file policies.
+	// If there is a conflict with default policy, replace that entry in the merged list with the user defined policy.
+	// If there was no conflict, add the user defined policy as a new entry to the merged list.
+	for _, userDefinedPolicyFile := range userDefinedClusterPolicies.Files {
+		override := false
+		for i, defaultPolicyFile := range defaultClusterPolicies.Files {
+			if defaultPolicyFile.Path == userDefinedPolicyFile.Path {
+				mergedClusterPolicies.Files[i] = convertSpecFileToStatusFile(userDefinedPolicyFile)
+				override = true
+				break
+			}
+		}
+		if !override {
+			mergedClusterPolicies.Files = append(mergedClusterPolicies.Files, convertSpecFileToStatusFile(userDefinedPolicyFile))
+		}
+	}
+
+	// Add default service unit policies to the merged list.
+	mergedClusterPolicies.Units = append(mergedClusterPolicies.Units, defaultClusterPolicies.Units...)
+
+	// Iterate through user service unit policies.
+	// If there is a conflict with default policy, replace that entry in the merged list with the user defined policy.
+	// If there was no conflict, add the user defined policy as a new entry to the merged list.
+	for _, userDefinedPolicyUnit := range userDefinedClusterPolicies.Units {
+		override := false
+		for i, defaultPolicyUnit := range defaultClusterPolicies.Units {
+			if defaultPolicyUnit.Name == userDefinedPolicyUnit.Name {
+				mergedClusterPolicies.Units[i] = convertSpecUnitToStatusUnit(userDefinedPolicyUnit)
+				override = true
+				break
+			}
+		}
+		if !override {
+			mergedClusterPolicies.Units = append(mergedClusterPolicies.Units, convertSpecUnitToStatusUnit(userDefinedPolicyUnit))
+		}
+	}
+
+	// If no user defined SSH policy exists, use the cluster defaults.
+	if len(userDefinedClusterPolicies.SSHKey.Actions) == 0 {
+		mergedClusterPolicies.SSHKey = *defaultClusterPolicies.SSHKey.DeepCopy()
+	} else {
+		mergedClusterPolicies.SSHKey = convertSpecSSHKeyToStatusSSHKey(*userDefinedClusterPolicies.SSHKey.DeepCopy())
+	}
+	return mergedClusterPolicies
+}
+
+// converts NodeDisruptionPolicySpecFile -> NodeDisruptionPolicyStatusFile
+func convertSpecFileToStatusFile(specFile opv1.NodeDisruptionPolicySpecFile) opv1.NodeDisruptionPolicyStatusFile {
+	statusFile := opv1.NodeDisruptionPolicyStatusFile{Path: specFile.Path, Actions: []opv1.NodeDisruptionPolicyStatusAction{}}
+	for _, action := range specFile.Actions {
+		statusFile.Actions = append(statusFile.Actions, convertSpecActiontoStatusAction(action))
+	}
+	return statusFile
+}
+
+// converts NodeDisruptionPolicySpecUnit -> NodeDisruptionPolicyStatusUnit
+func convertSpecUnitToStatusUnit(specUnit opv1.NodeDisruptionPolicySpecUnit) opv1.NodeDisruptionPolicyStatusUnit {
+	statusUnit := opv1.NodeDisruptionPolicyStatusUnit{Name: specUnit.Name, Actions: []opv1.NodeDisruptionPolicyStatusAction{}}
+	for _, action := range specUnit.Actions {
+		statusUnit.Actions = append(statusUnit.Actions, convertSpecActiontoStatusAction(action))
+	}
+	return statusUnit
+}
+
+// converts NodeDisruptionPolicySpecSSHKey -> NodeDisruptionPolicyStatusSSHKey
+func convertSpecSSHKeyToStatusSSHKey(specSSHKey opv1.NodeDisruptionPolicySpecSSHKey) opv1.NodeDisruptionPolicyStatusSSHKey {
+	statusSSHKey := opv1.NodeDisruptionPolicyStatusSSHKey{Actions: []opv1.NodeDisruptionPolicyStatusAction{}}
+	for _, action := range specSSHKey.Actions {
+		statusSSHKey.Actions = append(statusSSHKey.Actions, convertSpecActiontoStatusAction(action))
+	}
+	return statusSSHKey
+}
+
+// converts NodeDisruptionPolicySpecAction -> NodeDisruptionPolicyStatusAction
+func convertSpecActiontoStatusAction(action opv1.NodeDisruptionPolicySpecAction) opv1.NodeDisruptionPolicyStatusAction {
+	switch action.Type {
+	case opv1.DaemonReloadSpecAction:
+		return opv1.NodeDisruptionPolicyStatusAction{Type: opv1.DaemonReloadStatusAction}
+	case opv1.DrainSpecAction:
+		return opv1.NodeDisruptionPolicyStatusAction{Type: opv1.DrainStatusAction}
+	case opv1.NoneSpecAction:
+		return opv1.NodeDisruptionPolicyStatusAction{Type: opv1.NoneStatusAction}
+	case opv1.RebootSpecAction:
+		return opv1.NodeDisruptionPolicyStatusAction{Type: opv1.RebootStatusAction}
+	case opv1.ReloadSpecAction:
+		return opv1.NodeDisruptionPolicyStatusAction{Type: opv1.ReloadStatusAction, Reload: &opv1.ReloadService{
+			ServiceName: action.Reload.ServiceName,
+		}}
+	case opv1.RestartSpecAction:
+		return opv1.NodeDisruptionPolicyStatusAction{Type: opv1.RestartStatusAction, Restart: &opv1.RestartService{
+			ServiceName: action.Restart.ServiceName,
+		}}
+	default: // We should never be here as this is guarded by API validation. The return statement is to silence errors.
+		klog.Fatal("Unexpected action type found in Node Disruption Status calculation")
+		return opv1.NodeDisruptionPolicyStatusAction{Type: opv1.RebootStatusAction}
+	}
+}
+
+// Checks if a list of NodeDisruptionActions contain any action from the set of target actions
+func CheckNodeDisruptionActionsForTargetActions(actions []opv1.NodeDisruptionPolicyStatusAction, targetActions ...opv1.NodeDisruptionPolicyStatusActionType) bool {
+
+	currentActions := sets.New[opv1.NodeDisruptionPolicyStatusActionType]()
+	for _, action := range actions {
+		currentActions.Insert(action.Type)
+	}
+
+	return currentActions.HasAny(targetActions...)
+}

pkg/controller/common/helpers.go

@@ -970,6 +970,8 @@ func dedupePasswdUserSSHKeys(passwdUser ign2types.PasswdUser) ign2types.PasswdUs
 
 // CalculateConfigFileDiffs compares the files present in two ignition configurations and returns the list of files
 // that are different between them
+//
+//nolint:dupl
 func CalculateConfigFileDiffs(oldIgnConfig, newIgnConfig *ign3types.Config) []string {
 	// Go through the files and see what is new or different
 	oldFileSet := make(map[string]ign3types.File)
@@ -986,8 +988,6 @@ func CalculateConfigFileDiffs(oldIgnConfig, newIgnConfig *ign3types.Config) []st
 	for path := range oldFileSet {
 		_, ok := newFileSet[path]
 		if !ok {
-			// debug: remove
-			klog.Infof("File diff: %v was deleted", path)
 			diffFileSet = append(diffFileSet, path)
 		}
 	}
@@ -996,18 +996,50 @@ func CalculateConfigFileDiffs(oldIgnConfig, newIgnConfig *ign3types.Config) []st
 	for path, newFile := range newFileSet {
 		oldFile, ok := oldFileSet[path]
 		if !ok {
-			// debug: remove
-			klog.Infof("File diff: %v was added", path)
 			diffFileSet = append(diffFileSet, path)
 		} else if !reflect.DeepEqual(oldFile, newFile) {
-			// debug: remove
-			klog.Infof("File diff: detected change to %v", newFile.Path)
 			diffFileSet = append(diffFileSet, path)
 		}
 	}
 	return diffFileSet
 }
 
+// CalculateConfigUnitDiffs compares the units present in two ignition configurations and returns the list of units
+// that are different between them
+//
+//nolint:dupl
+func CalculateConfigUnitDiffs(oldIgnConfig, newIgnConfig *ign3types.Config) []string {
+	// Go through the units and see what is new or different
+	oldUnitSet := make(map[string]ign3types.Unit)
+	for _, u := range oldIgnConfig.Systemd.Units {
+		oldUnitSet[u.Name] = u
+	}
+	newUnitSet := make(map[string]ign3types.Unit)
+	for _, u := range newIgnConfig.Systemd.Units {
+		newUnitSet[u.Name] = u
+	}
+	diffUnitSet := []string{}
+
+	// First check if any units were removed
+	for unit := range oldUnitSet {
+		_, ok := newUnitSet[unit]
+		if !ok {
+			diffUnitSet = append(diffUnitSet, unit)
+		}
+	}
+
+	// Now check if any units were added/changed
+	for name, newUnit := range newUnitSet {
+		oldUnit, ok := oldUnitSet[name]
+		if !ok {
+			diffUnitSet = append(diffUnitSet, name)
+		} else if !reflect.DeepEqual(oldUnit, newUnit) {
+			diffUnitSet = append(diffUnitSet, name)
+		}
+	}
+	return diffUnitSet
+}
+
 // NewIgnFile returns a simple ignition3 file from just path and file contents.
 // It also ensures the compression field is set to the empty string, which is
 // currently required for ensuring child configs that may be merged layer

pkg/daemon/constants/constants.go

@@ -108,4 +108,7 @@ const (
 
 	// CRIOServiceName is used to specify reloads and restarts of the CRI-O service
 	CRIOServiceName = "crio"
+
+	// DaemonReloadCommand is used to specify reloads and restarts of the systemd manager configuration
+	DaemonReloadCommand = "daemon-reload"
 )

pkg/daemon/daemon.go

@@ -20,6 +20,7 @@ import (
 	"time"
 
 	mcfgclientset "github.com/openshift/client-go/machineconfiguration/clientset/versioned"
+	mcopclientset "github.com/openshift/client-go/operator/clientset/versioned"
 	"github.com/openshift/library-go/pkg/operator/configobserver/featuregates"
 
 	ign3types "github.com/coreos/ignition/v2/config/v3_4/types"
@@ -44,6 +45,7 @@ import (
 	mcfgalphav1 "github.com/openshift/api/machineconfiguration/v1alpha1"
 	mcfginformersv1 "github.com/openshift/client-go/machineconfiguration/informers/externalversions/machineconfiguration/v1"
 	mcfglistersv1 "github.com/openshift/client-go/machineconfiguration/listers/machineconfiguration/v1"
+
 	mcoResourceRead "github.com/openshift/machine-config-operator/lib/resourceread"
 	ctrlcommon "github.com/openshift/machine-config-operator/pkg/controller/common"
 	"github.com/openshift/machine-config-operator/pkg/daemon/constants"
@@ -86,6 +88,9 @@ type Daemon struct {
 
 	mcfgClient mcfgclientset.Interface
 
+	// mcopClient allows interaction with Openshift operator level objects, such as MachineConfiguration
+	mcopClient mcopclientset.Interface
+
 	// nodeLister is used to watch for updates via the informer
 	nodeLister       corev1lister.NodeLister
 	nodeListerSynced cache.InformerSynced
@@ -348,14 +353,15 @@ func (dn *Daemon) ClusterConnect(
 	mcInformer mcfginformersv1.MachineConfigInformer,
 	nodeInformer coreinformersv1.NodeInformer,
 	ccInformer mcfginformersv1.ControllerConfigInformer,
+	mcopClient mcopclientset.Interface,
 	kubeletHealthzEnabled bool,
 	kubeletHealthzEndpoint string,
 	featureGatesAccessor featuregates.FeatureGateAccess,
 ) error {
 	dn.name = name
 	dn.kubeClient = kubeClient
 	dn.mcfgClient = mcfgClient
-
+	dn.mcopClient = mcopClient
 	// Other controllers start out with the default controller limiter which retries
 	// in milliseconds; since any change here will involve rebooting the node
 	// we don't need to react in milliseconds.  See also updateDelay above.

pkg/daemon/daemon_test.go

@@ -22,6 +22,8 @@ import (
 	"k8s.io/apimachinery/pkg/util/diff"
 	kubeinformers "k8s.io/client-go/informers"
 	k8sfake "k8s.io/client-go/kubernetes/fake"
+
+	mcopfake "github.com/openshift/client-go/operator/clientset/versioned/fake"
 	core "k8s.io/client-go/testing"
 	"k8s.io/client-go/tools/cache"
 
@@ -115,6 +117,7 @@ type fixture struct {
 
 	client     *fake.Clientset
 	kubeclient *k8sfake.Clientset
+	oclient    *mcopfake.Clientset
 
 	mcLister   []*mcfgv1.MachineConfig
 	nodeLister []*corev1.Node
@@ -124,6 +127,7 @@ type fixture struct {
 
 	objects     []runtime.Object
 	kubeobjects []runtime.Object
+	oObjects    []runtime.Object
 }
 
 func newFixture(t *testing.T) *fixture {
@@ -142,6 +146,7 @@ var (
 func (f *fixture) newController() *Daemon {
 	f.client = fake.NewSimpleClientset(f.objects...)
 	f.kubeclient = k8sfake.NewSimpleClientset(f.kubeobjects...)
+	f.oclient = mcopfake.NewSimpleClientset(f.oObjects...)
 
 	i := informers.NewSharedInformerFactory(f.client, noResyncPeriodFunc())
 	k8sI := kubeinformers.NewSharedInformerFactory(f.kubeclient, noResyncPeriodFunc())
@@ -156,6 +161,7 @@ func (f *fixture) newController() *Daemon {
 		i.Machineconfiguration().V1().MachineConfigs(),
 		k8sI.Core().V1().Nodes(),
 		i.Machineconfiguration().V1().ControllerConfigs(),
+		f.oclient,
 		false,
 		"",
 		d.featureGatesAccessor,