OCPBUGS-76353: Ensure dbus calls are sync

go.mod

@@ -26,6 +26,7 @@ require (
 	github.com/distribution/reference v0.6.0
 	github.com/fsnotify/fsnotify v1.9.0
 	github.com/ghodss/yaml v1.0.1-0.20190212211648-25d852aebe32
+	github.com/godbus/dbus/v5 v5.1.1-0.20240921181615-a817f3cc4a9e
 	github.com/golangci/golangci-lint v1.62.0
 	github.com/google/go-cmp v0.7.0
 	github.com/google/goexpect v0.0.0-20210430020637-ab937bf7fd6f
@@ -125,7 +126,6 @@ require (
 	github.com/go-openapi/validate v0.24.0 // indirect
 	github.com/go-task/slim-sprig/v3 v3.0.0 // indirect
 	github.com/go-viper/mapstructure/v2 v2.2.1 // indirect
-	github.com/godbus/dbus/v5 v5.1.1-0.20240921181615-a817f3cc4a9e // indirect
 	github.com/golangci/go-printf-func-name v0.1.0 // indirect
 	github.com/golangci/modinfo v0.3.4 // indirect
 	github.com/golangci/plugin-module-register v0.1.1 // indirect

pkg/daemon/certificate_writer.go

@@ -295,8 +295,14 @@ func (dn *Daemon) syncControllerConfigHandler(key string) error {
 	if controllerConfig.Annotations[ctrlcommon.ServiceCARotateAnnotation] == ctrlcommon.ServiceCARotateTrue && oldAnno != controllerConfig.Annotations[ctrlcommon.ServiceCARotateAnnotation] && cmErr == nil && kubeConfigDiff && !allCertsThere && !dn.deferKubeletRestart {
 		if len(onDiskKC.Clusters[0].Cluster.CertificateAuthorityData) > 0 {
 			logSystem("restarting kubelet due to server-ca rotation")
-			if err := runCmdSync("systemctl", "stop", "kubelet"); err != nil {
-				return err
+			systemdConnection, err := dn.systemdManager.NewConnection(context.Background())
+			if err != nil {
+				return fmt.Errorf("error creating connection to systemd: %w", err)
+			}
+			defer systemdConnection.Close()
+
+			if err := systemdConnection.Stop(context.Background(), "kubelet"); err != nil {
+				return fmt.Errorf("could not stop kubelet for server-ca rotation. Error: %v", err)
 			}
 			f, err := os.ReadFile("/var/lib/kubelet/kubeconfig")
 			if err != nil && os.IsNotExist(err) {
@@ -323,12 +329,12 @@ func (dn *Daemon) syncControllerConfigHandler(key string) error {
 				return err
 			}
 
-			if err := runCmdSync("systemctl", "daemon-reload"); err != nil {
-				return err
+			if err := systemdConnection.ReloadDaemon(context.Background()); err != nil {
+				return fmt.Errorf("could not reload systemd after kubelet stop for server-ca rotation. Error: %v", err)
 			}
 
-			if err := runCmdSync("systemctl", "start", "kubelet"); err != nil {
-				return err
+			if err := systemdConnection.Start(context.Background(), "kubelet"); err != nil {
+				return fmt.Errorf("could not start kubelet after stopping it for server-ca rotation. Error: %v", err)
 			}
 		}
 

pkg/daemon/config_drift_monitor_test.go

@@ -496,9 +496,18 @@ func (tc *configDriftMonitorTestCase) writeIgnitionConfig(t *testing.T, ignConfi
 		return fmt.Errorf("could not write ignition config files: %w", err)
 	}
 
-	// Write systemd units the same way the MCD does.
-	if err := writeUnits(ignConfig.Systemd.Units, tc.systemdPath, true); err != nil {
-		return fmt.Errorf("could not write systemd units: %w", err)
+	// Use a mock systemd connection for testing
+	// Create mock units for all units in the config
+	mockUnits := make(map[string]*mockUnitState)
+	for _, unit := range ignConfig.Systemd.Units {
+		mockUnits[unit.Name] = newMockUnitState(unit.Name)
+	}
+	mockConn := newMockSystemdConnection(mockUnits)
+
+	for _, unit := range ignConfig.Systemd.Units {
+		if err := writeUnit(mockConn, unit, tc.systemdPath, true); err != nil {
+			return fmt.Errorf("could not write unit: %w", err)
+		}
 	}
 
 	return nil

pkg/daemon/constants/constants.go

@@ -119,7 +119,7 @@ const (
 	RHCOS9SSHKeyPath = CoreUserSSHPath + "/authorized_keys.d/ignition"
 
 	// CRIOServiceName is used to specify reloads and restarts of the CRI-O service
-	CRIOServiceName = "crio"
+	CRIOServiceName = "crio.service"
 
 	// DaemonReloadCommand is used to specify reloads and restarts of the systemd manager configuration
 	DaemonReloadCommand = "daemon-reload"

pkg/daemon/daemon.go

@@ -169,6 +169,9 @@ type Daemon struct {
 	// Abstraction for running commands against the OS
 	cmdRunner CommandRunner
 
+	// Abstraction for interacting with systemd
+	systemdManager SystemdManager
+
 	// Bare minimal podman client
 	podmanInterface PodmanInterface
 }
@@ -315,9 +318,10 @@ func New(
 	var nodeUpdaterClient *RpmOstreeClient
 	cmdRunner := &CommandRunnerOS{}
 	podmanInterface := NewPodmanExec(cmdRunner)
+	systemdManager := NewSystemdManagerDefault()
 	// Only pull the osImageURL from OSTree when we are on RHCOS or FCOS
 	if hostos.IsCoreOSVariant() {
-		nodeUpdaterClientVal := NewNodeUpdaterClient(cmdRunner, podmanInterface)
+		nodeUpdaterClientVal := NewNodeUpdaterClient(cmdRunner, podmanInterface, systemdManager)
 		nodeUpdaterClient = &nodeUpdaterClientVal
 		err := nodeUpdaterClient.Initialize()
 		if err != nil {
@@ -357,6 +361,7 @@ func New(
 		irreconcilableReporter: NewNoOpIrreconcilableReporterImpl(),
 		cmdRunner:              cmdRunner,
 		podmanInterface:        podmanInterface,
+		systemdManager:         systemdManager,
 	}, nil
 }
 
@@ -1223,7 +1228,7 @@ func (dn *Daemon) syncNodeHypershift(key string) error {
 
 	if ctrlcommon.InSlice(postConfigChangeActionReloadCrio, actions) {
 		serviceName := constants.CRIOServiceName
-		if err := reloadService(serviceName); err != nil {
+		if err := dn.systemdManager.DoConnection(context.Background(), SystemdReload(serviceName)); err != nil {
 			return fmt.Errorf("could not apply update: reloading %s configuration failed. Error: %w", serviceName, err)
 		}
 		klog.Infof("%s config reloaded successfully! Desired config %s has been applied, skipping reboot", serviceName, desiredConfig.Name)

pkg/daemon/file_writers.go

@@ -2,13 +2,15 @@ package daemon
 
 import (
 	"bufio"
+	"context"
 	"fmt"
 	"os"
 	"os/exec"
 	"os/user"
 	"path/filepath"
 	"strconv"
 	"strings"
+	"time"
 
 	ign3types "github.com/coreos/ignition/v2/config/v3_5/types"
 	"github.com/google/renameio"
@@ -265,7 +267,7 @@ func writeFiles(files []ign3types.File, skipCertificateWrite bool) error {
 }
 
 // writeUnit writes a systemd unit and its dropins to disk
-func writeUnit(u ign3types.Unit, systemdRoot string, isCoreOSVariant bool) error {
+func writeUnit(systemdConnection SystemdConnection, u ign3types.Unit, systemdRoot string, isCoreOSVariant bool) error {
 	if err := writeDropins(u, systemdRoot, isCoreOSVariant); err != nil {
 		return err
 	}
@@ -302,13 +304,16 @@ func writeUnit(u ign3types.Unit, systemdRoot string, isCoreOSVariant bool) error
 		// If the unit is currently enabled, disable it before overwriting since we might be
 		// changing its WantedBy= or RequiredBy= directive (see OCPBUGS-33694). Later code will
 		// re-enable the new unit as directed by the MachineConfig.
-		cmd := exec.Command("systemctl", "is-enabled", u.Name)
-		out, _ := cmd.CombinedOutput()
-		if cmd.ProcessState.ExitCode() == 0 && strings.TrimSpace(string(out)) == "enabled" {
+		ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+		defer cancel()
+		isEnabled, err := systemdConnection.IsEnabled(ctx, u.Name)
+		if err != nil {
+			return fmt.Errorf("failed to check if %q is enabled: %w", u.Name, err)
+		}
+		if isEnabled {
 			klog.Infof("Disabling systemd unit %s before re-writing it", u.Name)
-			disableOut, err := exec.Command("systemctl", "disable", u.Name).CombinedOutput()
-			if err != nil {
-				return fmt.Errorf("disabling %s failed: %w (output: %s)", u.Name, err, string(disableOut))
+			if err := systemdConnection.Disable(ctx, u.Name); err != nil {
+				return fmt.Errorf("failed to disable %q: %w", u.Name, err)
 			}
 		}
 		if err := writeFileAtomicallyWithDefaults(fpath, []byte(*u.Contents)); err != nil {
@@ -340,17 +345,6 @@ func unitHasContent(u ign3types.Unit) bool {
 	return u.Contents != nil && *u.Contents != ""
 }
 
-// writeUnits writes systemd units and their dropins to disk
-func writeUnits(units []ign3types.Unit, systemdRoot string, isCoreOSVariant bool) error {
-	for _, u := range units {
-		if err := writeUnit(u, systemdRoot, isCoreOSVariant); err != nil {
-			return err
-		}
-	}
-
-	return nil
-}
-
 func lookupUID(username string) (int, error) {
 	osUser, err := user.Lookup(username)
 	if err != nil {

pkg/daemon/pinned_image_set.go

@@ -119,6 +119,9 @@ type PinnedImageSetManager struct {
 	queue                    workqueue.TypedRateLimitingInterface[string]
 	fgHandler                ctrlcommon.FeatureGatesHandler
 
+	// Abstraction for interacting with systemd
+	systemdManager SystemdManager
+
 	// mutex protects cancelFn
 	mu       sync.Mutex
 	cancelFn context.CancelFunc
@@ -150,6 +153,7 @@ func NewPinnedImageSetManager(
 		registryCfgPath:          registryCfgPath,
 		prefetchTimeout:          prefetchTimeout,
 		minStorageAvailableBytes: minStorageAvailableBytes,
+		systemdManager:           NewSystemdManagerDefault(),
 		fgHandler:                fgHandler,
 		queue: workqueue.NewTypedRateLimitingQueueWithConfig[string](
 			workqueue.DefaultTypedControllerRateLimiter[string](),
@@ -285,7 +289,7 @@ func (p *PinnedImageSetManager) syncMachineConfigPools(ctx context.Context, pool
 
 	// write config and reload crio last to allow a window for kubelet to gc
 	// images in an emergency
-	if err := ensureCrioPinnedImagesConfigFile(crioPinnedImagesDropInFilePath, imageNames); err != nil {
+	if err := p.ensureCrioPinnedImagesConfigFile(crioPinnedImagesDropInFilePath, imageNames); err != nil {
 		klog.Errorf("failed to write crio config file: %v", err)
 		return err
 	}
@@ -492,7 +496,7 @@ func (p *PinnedImageSetManager) checkImagePayloadStorage(ctx context.Context, im
 }
 
 // ensureCrioPinnedImagesConfigFile ensures the crio config file is up to date with the pinned images.
-func ensureCrioPinnedImagesConfigFile(path string, imageNames []string) error {
+func (p *PinnedImageSetManager) ensureCrioPinnedImagesConfigFile(path string, imageNames []string) error {
 	cfgExists, err := hasConfigFile(path)
 	if err != nil {
 		return fmt.Errorf("failed to check crio config file: %w", err)
@@ -502,7 +506,7 @@ func ensureCrioPinnedImagesConfigFile(path string, imageNames []string) error {
 		if err := deleteCrioConfigFile(); err != nil {
 			return fmt.Errorf("failed to remove CRI-O config file: %w", err)
 		}
-		return crioReload()
+		return p.crioReload()
 	} else if len(imageNames) == 0 {
 		return nil
 	}
@@ -526,7 +530,7 @@ func ensureCrioPinnedImagesConfigFile(path string, imageNames []string) error {
 		if err := writeFiles([]ign3types.File{ignFile}, true); err != nil {
 			return fmt.Errorf("failed to write CRIO config file: %w", err)
 		}
-		return crioReload()
+		return p.crioReload()
 	}
 	klog.Infof("CRI-O config file is up to date, no reload required")
 
@@ -1080,7 +1084,7 @@ func (p *PinnedImageSetManager) deleteMachineConfigPool(obj interface{}) {
 		return
 	}
 
-	crioReload()
+	p.crioReload()
 }
 
 func (p *PinnedImageSetManager) enqueue(pool *mcfgv1.MachineConfigPool) {
@@ -1164,6 +1168,15 @@ func (p *PinnedImageSetManager) getImageSize(ctx context.Context, imageName, aut
 	return totalSize, nil
 }
 
+func (p *PinnedImageSetManager) crioReload() error {
+	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+	defer cancel()
+	if err := p.systemdManager.DoConnection(ctx, SystemdReload(constants.CRIOServiceName)); err != nil {
+		return fmt.Errorf("could not apply update: reloading %s configuration failed. Error: %w", constants.CRIOServiceName, err)
+	}
+	return nil
+}
+
 // ensurePullImage first checks if the image exists locally and then will attempt to pull
 // the image from the container runtime with a retry/backoff.
 func ensurePullImage(ctx context.Context, client *cri.Client, backoff wait.Backoff, image string, authConfig *runtimeapi.AuthConfig) error {
@@ -1232,14 +1245,6 @@ func uniqueSortedImageNames(images []mcfgv1.PinnedImageRef) []string {
 	return unique
 }
 
-func crioReload() error {
-	serviceName := constants.CRIOServiceName
-	if err := reloadService(serviceName); err != nil {
-		return fmt.Errorf("could not apply update: reloading %s configuration failed. Error: %w", serviceName, err)
-	}
-	return nil
-}
-
 func deleteCrioConfigFile() error {
 	// remove the crio config file
 	if err := os.Remove(crioPinnedImagesDropInFilePath); err != nil {

pkg/daemon/pinned_image_set_test.go

@@ -571,11 +571,16 @@ func TestEnsureCrioPinnedImagesConfigFile(t *testing.T) {
 	testCfgPath := filepath.Join(tmpDir, "50-pinned-images")
 	err = os.WriteFile(testCfgPath, newCfgBytes, 0644)
 	require.NoError(err)
-	err = ensureCrioPinnedImagesConfigFile(testCfgPath, imageNames)
+
+	pinnedImageSetManager := PinnedImageSetManager{
+		systemdManager: newMockSystemdManager(nil),
+	}
+
+	err = pinnedImageSetManager.ensureCrioPinnedImagesConfigFile(testCfgPath, imageNames)
 	require.NoError(err)
 
 	imageNames = append(imageNames, "quay.io/openshift-release-dev/ocp-v4.0-art-dev@sha256:dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd")
-	err = ensureCrioPinnedImagesConfigFile(testCfgPath, imageNames)
+	err = pinnedImageSetManager.ensureCrioPinnedImagesConfigFile(testCfgPath, imageNames)
 	require.ErrorIs(err, os.ErrPermission) // this error is from atomic writer attempting to write.
 }