rkt: Refactor the systemd interface.

Replace shell out calls with dbus API calls.
Remove unused 'Reload()'.

pkg/kubelet/rkt/fake_rkt_interface_test.go

@@ -107,9 +107,10 @@ func (f *fakeRktInterface) GetLogs(ctx context.Context, in *rktapi.GetLogsReques
 // See https://github.com/coreos/rkt/issues/1769.
 type fakeSystemd struct {
 	sync.Mutex
-	called  []string
-	version string
-	err     error
+	called           []string
+	resetFailedUnits []string
+	version          string
+	err              error
 }
 
 func newFakeSystemd() *fakeSystemd {
@@ -143,15 +144,9 @@ func (f *fakeSystemd) RestartUnit(name string, mode string, ch chan<- string) (i
 	return 0, fmt.Errorf("Not implemented")
 }
 
-func (f *fakeSystemd) Reload() error {
-	return fmt.Errorf("Not implemented")
-}
-
-func (f *fakeSystemd) ResetFailed() error {
-	f.Lock()
-	defer f.Unlock()
-
-	f.called = append(f.called, "ResetFailed")
+func (f *fakeSystemd) ResetFailedUnit(name string) error {
+	f.called = append(f.called, "ResetFailedUnit")
+	f.resetFailedUnits = append(f.resetFailedUnits, name)
 	return f.err
 }
 

pkg/kubelet/rkt/rkt.go

@@ -1415,7 +1415,7 @@ func (r *Runtime) convertRktPod(rktpod *rktapi.Pod) (*kubecontainer.Pod, error)
 	return kubepod, nil
 }
 
-// GetPods runs 'systemctl list-unit' and 'rkt list' to get the list of rkt pods.
+// GetPods runs 'rkt list' to get the list of rkt pods.
 // Then it will use the result to construct a list of container runtime pods.
 // If all is false, then only running pods will be returned, otherwise all pods will be
 // returned.
@@ -1741,10 +1741,6 @@ func (r *Runtime) GarbageCollect(gcPolicy kubecontainer.ContainerGCPolicy) error
 
 	glog.V(4).Infof("rkt: Garbage collecting triggered with policy %v", gcPolicy)
 
-	if err := r.systemd.ResetFailed(); err != nil {
-		glog.Errorf("rkt: Failed to reset failed systemd services: %v, continue to gc anyway...", err)
-	}
-
 	// GC all inactive systemd service files and pods.
 	files, err := r.os.ReadDir(systemdServiceDir)
 	if err != nil {
@@ -1787,13 +1783,16 @@ func (r *Runtime) GarbageCollect(gcPolicy kubecontainer.ContainerGCPolicy) error
 			if _, ok := allPods[rktUUID]; !ok {
 				glog.V(4).Infof("rkt: No rkt pod found for service file %q, will remove it", serviceName)
 
+				if err := r.systemd.ResetFailedUnit(serviceName); err != nil {
+					glog.Warningf("rkt: Failed to reset the failed systemd service %q: %v", serviceName, err)
+				}
 				serviceFile := serviceFilePath(serviceName)
 
 				// Network may not be around anymore so errors are ignored
 				r.cleanupPodNetworkFromServiceFile(serviceFile)
 
 				if err := r.os.Remove(serviceFile); err != nil {
-					errlist = append(errlist, fmt.Errorf("rkt: Failed to remove service file %q: %v", serviceName, err))
+					errlist = append(errlist, fmt.Errorf("rkt: Failed to remove service file %q: %v", serviceFile, err))
 				}
 			}
 		}
@@ -1857,8 +1856,11 @@ func (r *Runtime) removePod(uuid string) error {
 	}
 
 	// GC systemd service files as well.
+	if err := r.systemd.ResetFailedUnit(serviceName); err != nil {
+		glog.Warningf("rkt: Failed to reset the failed systemd service %q: %v", serviceName, err)
+	}
 	if err := r.os.Remove(serviceFile); err != nil {
-		errlist = append(errlist, fmt.Errorf("rkt: Failed to remove service file %q for pod %q: %v", serviceName, uuid, err))
+		errlist = append(errlist, fmt.Errorf("rkt: Failed to remove service file %q for pod %q: %v", serviceFile, uuid, err))
 	}
 
 	return errors.NewAggregate(errlist)

pkg/kubelet/rkt/rkt_test.go

@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"net"
 	"os"
+	"path/filepath"
 	"sort"
 	"testing"
 	"time"
@@ -1626,13 +1627,20 @@ func TestGarbageCollect(t *testing.T) {
 
 		sort.Sort(sortedStringList(tt.expectedServiceFiles))
 		sort.Sort(sortedStringList(fakeOS.Removes))
+		sort.Sort(sortedStringList(fs.resetFailedUnits))
 
 		assert.Equal(t, tt.expectedServiceFiles, fakeOS.Removes, testCaseHint)
+		var expectedService []string
+		for _, f := range tt.expectedServiceFiles {
+			expectedService = append(expectedService, filepath.Base(f))
+		}
+		assert.Equal(t, expectedService, fs.resetFailedUnits, testCaseHint)
 
 		// Cleanup after each test.
 		cli.Reset()
 		ctrl.Finish()
 		fakeOS.Removes = []string{}
+		fs.resetFailedUnits = []string{}
 		getter.pods = make(map[kubetypes.UID]*api.Pod)
 	}
 }

pkg/kubelet/rkt/systemd.go

@@ -18,7 +18,6 @@ package rkt
 
 import (
 	"fmt"
-	"os/exec"
 	"strconv"
 	"strings"
 
@@ -60,10 +59,8 @@ type systemdInterface interface {
 	StopUnit(name string, mode string, ch chan<- string) (int, error)
 	// StopUnits restarts the unit with the given name.
 	RestartUnit(name string, mode string, ch chan<- string) (int, error)
-	// Reload is equivalent to 'systemctl daemon-reload'.
-	Reload() error
-	// ResetFailed is equivalent to 'systemctl reset-failed'.
-	ResetFailed() error
+	// ResetFailedUnit resets the "failed" state of a specific unit.
+	ResetFailedUnit(name string) error
 }
 
 // systemd implements the systemdInterface using dbus and systemctl.
@@ -83,28 +80,13 @@ func newSystemd() (*systemd, error) {
 
 // Version returns the version of the systemd.
 func (s *systemd) Version() (systemdVersion, error) {
-	output, err := exec.Command("systemctl", "--version").Output()
+	versionStr, err := s.Conn.GetManagerProperty("Version")
 	if err != nil {
 		return -1, err
 	}
-	// Example output of 'systemctl --version':
-	//
-	// systemd 215
-	// +PAM +AUDIT +SELINUX +IMA +SYSVINIT +LIBCRYPTSETUP +GCRYPT +ACL +XZ -SECCOMP -APPARMOR
-	//
-	lines := strings.Split(string(output), "\n")
-	tuples := strings.Split(lines[0], " ")
-	if len(tuples) != 2 {
-		return -1, fmt.Errorf("rkt: Failed to parse version %v", lines)
-	}
-	result, err := strconv.Atoi(string(tuples[1]))
+	result, err := strconv.Atoi(strings.Trim(versionStr, "\""))
 	if err != nil {
 		return -1, err
 	}
 	return systemdVersion(result), nil
 }
-
-// ResetFailed calls 'systemctl reset failed'
-func (s *systemd) ResetFailed() error {
-	return exec.Command("systemctl", "reset-failed").Run()
-}