Merge pull request #22467 from Random-Liu/push-down-runtime-version-check

pkg/kubelet/container/runtime.go

@@ -57,7 +57,10 @@ type Runtime interface {
 	Version() (Version, error)
 	// APIVersion returns the API version information of the container
 	// runtime. This may be different from the runtime engine's version.
+	// TODO(random-liu): We should fold this into Version()
 	APIVersion() (Version, error)
+	// Status returns error if the runtime is unhealthy; nil otherwise.
+	Status() error
 	// GetPods returns a list containers group by pods. The boolean parameter
 	// specifies whether the runtime returns all containers including those already
 	// exited and dead containers (used for garbage collection).

pkg/kubelet/container/testing/fake_runtime.go

@@ -48,6 +48,7 @@ type FakeRuntime struct {
 	RuntimeType       string
 	Err               error
 	InspectErr        error
+	StatusErr         error
 }
 
 // FakeRuntime should implement Runtime.
@@ -108,6 +109,7 @@ func (f *FakeRuntime) ClearCalls() {
 	f.RuntimeType = ""
 	f.Err = nil
 	f.InspectErr = nil
+	f.StatusErr = nil
 }
 
 func (f *FakeRuntime) assertList(expect []string, test []string) error {
@@ -168,6 +170,14 @@ func (f *FakeRuntime) APIVersion() (Version, error) {
 	return &FakeVersion{Version: f.APIVersionInfo}, f.Err
 }
 
+func (f *FakeRuntime) Status() error {
+	f.Lock()
+	defer f.Unlock()
+
+	f.CalledFunctions = append(f.CalledFunctions, "Status")
+	return f.StatusErr
+}
+
 func (f *FakeRuntime) GetPods(all bool) ([]*Pod, error) {
 	f.Lock()
 	defer f.Unlock()

pkg/kubelet/container/testing/runtime_mock.go

@@ -53,6 +53,11 @@ func (r *Mock) APIVersion() (Version, error) {
 	return args.Get(0).(Version), args.Error(1)
 }
 
+func (r *Mock) Status() error {
+	args := r.Called()
+	return args.Error(0)
+}
+
 func (r *Mock) GetPods(all bool) ([]*Pod, error) {
 	args := r.Called(all)
 	return args.Get(0).([]*Pod), args.Error(1)

pkg/kubelet/dockertools/fake_docker_client.go

@@ -68,6 +68,26 @@ func NewFakeDockerClientWithVersion(version, apiVersion string) *FakeDockerClien
 	}
 }
 
+func (f *FakeDockerClient) InjectError(fn string, err error) {
+	f.Lock()
+	defer f.Unlock()
+	f.Errors[fn] = err
+}
+
+func (f *FakeDockerClient) InjectErrors(errs map[string]error) {
+	f.Lock()
+	defer f.Unlock()
+	for fn, err := range errs {
+		f.Errors[fn] = err
+	}
+}
+
+func (f *FakeDockerClient) ClearErrors() {
+	f.Lock()
+	defer f.Unlock()
+	f.Errors = map[string]error{}
+}
+
 func (f *FakeDockerClient) ClearCalls() {
 	f.Lock()
 	defer f.Unlock()
@@ -382,7 +402,7 @@ func (f *FakeDockerClient) PullImage(opts docker.PullImageOptions, auth docker.A
 }
 
 func (f *FakeDockerClient) Version() (*docker.Env, error) {
-	return &f.VersionInfo, nil
+	return &f.VersionInfo, f.popError("version")
 }
 
 func (f *FakeDockerClient) Info() (*docker.Env, error) {

pkg/kubelet/dockertools/manager.go

@@ -60,7 +60,7 @@ import (
 const (
 	DockerType = "docker"
 
-	MinimumDockerAPIVersion = "1.18"
+	minimumDockerAPIVersion = "1.18"
 
 	// ndots specifies the minimum number of dots that a domain name must contain for the resolver to consider it as FQDN (fully-qualified)
 	// we want to able to consider SRV lookup names like _dns._udp.kube-dns.default.svc to be considered relative.
@@ -938,6 +938,30 @@ func (dm *DockerManager) APIVersion() (kubecontainer.Version, error) {
 	return dockerAPIVersion(version), nil
 }
 
+// Status returns error if docker daemon is unhealthy, nil otherwise.
+// Now we do this by checking whether:
+// 1) `docker version` works
+// 2) docker version is compatible with minimum requirement
+func (dm *DockerManager) Status() error {
+	return dm.checkVersionCompatibility()
+}
+
+func (dm *DockerManager) checkVersionCompatibility() error {
+	version, err := dm.APIVersion()
+	if err != nil {
+		return err
+	}
+	// Verify the docker version.
+	result, err := version.Compare(minimumDockerAPIVersion)
+	if err != nil {
+		return fmt.Errorf("failed to compare current docker version %v with minimum support Docker version %q - %v", version, minimumDockerAPIVersion, err)
+	}
+	if result < 0 {
+		return fmt.Errorf("container runtime version is older than %s", minimumDockerAPIVersion)
+	}
+	return nil
+}
+
 // The first version of docker that supports exec natively is 1.3.0 == API 1.15
 var dockerAPIVersionWithExec = "1.15"
 

pkg/kubelet/dockertools/manager_test.go

@@ -546,7 +546,7 @@ func TestKillContainerInPodWithError(t *testing.T) {
 		},
 	}
 	fakeDocker.SetFakeRunningContainers(containers)
-	fakeDocker.Errors["stop"] = fmt.Errorf("sample error")
+	fakeDocker.InjectError("stop", fmt.Errorf("sample error"))
 
 	if err := manager.KillContainerInPod(kubecontainer.ContainerID{}, &pod.Spec.Containers[0], pod, "test kill container with error."); err == nil {
 		t.Errorf("expected error, found nil")
@@ -1725,7 +1725,7 @@ func TestSyncPodWithFailure(t *testing.T) {
 			ID:   "9876",
 			Name: "/k8s_POD." + strconv.FormatUint(generatePodInfraContainerHash(pod), 16) + "_foo_new_12345678_0",
 		}})
-		fakeDocker.Errors = test.dockerError
+		fakeDocker.InjectErrors(test.dockerError)
 		puller.ErrorsToInject = test.pullerError
 		pod.Spec.Containers = []api.Container{test.container}
 		result := runSyncPod(t, dm, fakeDocker, pod, nil, true)
@@ -1846,3 +1846,44 @@ func TestSecurityOptsAreNilWithDockerV19(t *testing.T) {
 	}
 	assert.NotContains(t, newContainer.HostConfig.SecurityOpt, "seccomp:unconfined", "Pods with Docker versions < 1.10 must not have seccomp disabled by default")
 }
+
+func TestCheckVersionCompatibility(t *testing.T) {
+	apiVersion, err := docker.NewAPIVersion(minimumDockerAPIVersion)
+	if err != nil {
+		t.Fatalf("unexpected error %v", err)
+	}
+	type test struct {
+		version    string
+		compatible bool
+	}
+	tests := []test{
+		// Minimum apiversion
+		{minimumDockerAPIVersion, true},
+		// Invalid apiversion
+		{"invalid_api_version", false},
+	}
+	for i := range apiVersion {
+		apiVersion[i]++
+		// Newer apiversion
+		tests = append(tests, test{apiVersion.String(), true})
+		apiVersion[i] -= 2
+		// Older apiversion
+		if apiVersion[i] >= 0 {
+			tests = append(tests, test{apiVersion.String(), false})
+		}
+		apiVersion[i]++
+	}
+
+	for i, tt := range tests {
+		testCase := fmt.Sprintf("test case #%d test version %q", i, tt.version)
+		dm, fakeDocker := newTestDockerManagerWithHTTPClientWithVersion(&fakeHTTP{}, "", tt.version)
+		err := dm.checkVersionCompatibility()
+		assert.Equal(t, tt.compatible, err == nil, testCase)
+		if tt.compatible == true {
+			// Get docker version error
+			fakeDocker.InjectError("version", fmt.Errorf("injected version error"))
+			err := dm.checkVersionCompatibility()
+			assert.NotNil(t, err, testCase+" version error check")
+		}
+	}
+}

pkg/kubelet/kubelet.go

@@ -458,7 +458,7 @@ func NewMainKubelet(
 	}
 
 	klet.pleg = pleg.NewGenericPLEG(klet.containerRuntime, plegChannelCapacity, plegRelistPeriod, klet.podCache, util.RealClock{})
-	klet.runtimeState = newRuntimeState(maxWaitForContainerRuntime, configureCBR0, klet.isContainerRuntimeVersionCompatible)
+	klet.runtimeState = newRuntimeState(maxWaitForContainerRuntime, configureCBR0)
 	klet.updatePodCIDR(podCIDR)
 
 	// setup containerGC
@@ -2692,7 +2692,7 @@ func (kl *Kubelet) GetPodByName(namespace, name string) (*api.Pod, bool) {
 }
 
 func (kl *Kubelet) updateRuntimeUp() {
-	if _, err := kl.containerRuntime.Version(); err != nil {
+	if err := kl.containerRuntime.Status(); err != nil {
 		glog.Errorf("Container runtime sanity check failed: %v", err)
 		return
 	}
@@ -3115,26 +3115,6 @@ func SetNodeStatus(f func(*api.Node) error) Option {
 	}
 }
 
-// FIXME: Why not combine this with container runtime health check?
-func (kl *Kubelet) isContainerRuntimeVersionCompatible() error {
-	switch kl.GetRuntime().Type() {
-	case "docker":
-		version, err := kl.GetRuntime().APIVersion()
-		if err != nil {
-			return nil
-		}
-		// Verify the docker version.
-		result, err := version.Compare(dockertools.MinimumDockerAPIVersion)
-		if err != nil {
-			return fmt.Errorf("failed to compare current docker version %v with minimum support Docker version %q - %v", version, dockertools.MinimumDockerAPIVersion, err)
-		}
-		if result < 0 {
-			return fmt.Errorf("container runtime version is older than %s", dockertools.MinimumDockerAPIVersion)
-		}
-	}
-	return nil
-}
-
 // tryUpdateNodeStatus tries to update node status to master. If ReconcileCBR0
 // is set, this function will also confirm that cbr0 is configured correctly.
 func (kl *Kubelet) tryUpdateNodeStatus() error {

pkg/kubelet/rkt/rkt.go

@@ -184,13 +184,6 @@ func New(config *Config,
 		rkt.imagePuller = kubecontainer.NewImagePuller(recorder, rkt, imageBackOff)
 	}
 
-	if err := rkt.checkVersion(minimumRktBinVersion, recommendedRktBinVersion, minimumAppcVersion, minimumRktApiVersion, minimumSystemdVersion); err != nil {
-		// TODO(yifan): Latest go-systemd version have the ability to close the
-		// dbus connection. However the 'docker/libcontainer' package is using
-		// the older go-systemd version, so we can't update the go-systemd version.
-		rkt.apisvcConn.Close()
-		return nil, err
-	}
 	return rkt, nil
 }
 
@@ -1062,7 +1055,12 @@ func (r *Runtime) Version() (kubecontainer.Version, error) {
 }
 
 func (r *Runtime) APIVersion() (kubecontainer.Version, error) {
-	return r.binVersion, nil
+	return r.apiVersion, nil
+}
+
+// Status returns error if rkt is unhealthy, nil otherwise.
+func (r *Runtime) Status() error {
+	return r.checkVersion(minimumRktBinVersion, recommendedRktBinVersion, minimumAppcVersion, minimumRktApiVersion, minimumSystemdVersion)
 }
 
 // SyncPod syncs the running pod to match the specified desired pod.

pkg/kubelet/runtime.go

@@ -30,7 +30,6 @@ type runtimeState struct {
 	internalError            error
 	cidr                     string
 	initError                error
-	runtimeCompatibility     func() error
 }
 
 func (s *runtimeState) setRuntimeSync(t time.Time) {
@@ -85,16 +84,12 @@ func (s *runtimeState) errors() []string {
 	if s.internalError != nil {
 		ret = append(ret, s.internalError.Error())
 	}
-	if err := s.runtimeCompatibility(); err != nil {
-		ret = append(ret, err.Error())
-	}
 	return ret
 }
 
 func newRuntimeState(
 	runtimeSyncThreshold time.Duration,
 	configureNetwork bool,
-	runtimeCompatibility func() error,
 ) *runtimeState {
 	var networkError error = nil
 	if configureNetwork {
@@ -105,6 +100,5 @@ func newRuntimeState(
 		baseRuntimeSyncThreshold: runtimeSyncThreshold,
 		networkError:             networkError,
 		internalError:            nil,
-		runtimeCompatibility:     runtimeCompatibility,
 	}
 }