Merge pull request #30212 from feiskyer/kuberuntime-flag

Automatic merge from submit-queue

Kubelet: add --container-runtime-endpoint and --image-service-endpoint

Flag `--container-runtime-endpoint` (overrides `--container-runtime`) is introduced to identify the unix socket file of the remote runtime service. And flag `--image-service-endpoint` is introduced to identify the unix socket file of the image service.

This PR is part of #28789 Milestone 0. 

CC @yujuhong @Random-Liu

cmd/kubelet/app/options/options.go

@@ -184,6 +184,8 @@ func (s *KubeletServer) AddFlags(fs *pflag.FlagSet) {
 	fs.Int32Var(&s.IPTablesDropBit, "iptables-drop-bit", s.IPTablesDropBit, "The bit of the fwmark space to mark packets for dropping. Must be within the range [0, 31].")
 
 	// Flags intended for testing, not recommended used in production environments.
+	fs.StringVar(&s.RemoteRuntimeEndpoint, "container-runtime-endpoint", s.RemoteRuntimeEndpoint, "The unix socket endpoint of remote runtime service. If not empty, this option will override --container-runtime. This is an experimental feature. Intended for testing only.")
+	fs.StringVar(&s.RemoteImageEndpoint, "image-service-endpoint", s.RemoteImageEndpoint, "The unix socket endpoint of remote image service. If not specified, it will be the same with container-runtime-endpoint by default. This is an experimental feature. Intended for testing only.")
 	fs.BoolVar(&s.ReallyCrashForTesting, "really-crash-for-testing", s.ReallyCrashForTesting, "If true, when panics occur crash. Intended for testing.")
 	fs.Float64Var(&s.ChaosChance, "chaos-chance", s.ChaosChance, "If > 0.0, introduce random client errors and latency. Intended for testing. [default=0.0]")
 	fs.BoolVar(&s.Containerized, "containerized", s.Containerized, "Experimental support for running kubelet in a container.  Intended for testing. [default=false]")

cmd/kubelet/app/server.go

@@ -226,6 +226,8 @@ func UnsecuredKubeletConfig(s *options.KubeletServer) (*KubeletConfig, error) {
 		ConfigureCBR0:                s.ConfigureCBR0,
 		ContainerManager:             nil,
 		ContainerRuntime:             s.ContainerRuntime,
+		RemoteRuntimeEndpoint:        s.RemoteRuntimeEndpoint,
+		RemoteImageEndpoint:          s.RemoteImageEndpoint,
 		RuntimeRequestTimeout:        s.RuntimeRequestTimeout.Duration,
 		CPUCFSQuota:                  s.CPUCFSQuota,
 		DiskSpacePolicy:              diskSpacePolicy,
@@ -877,6 +879,8 @@ type KubeletConfig struct {
 	ConfigureCBR0                  bool
 	ContainerManager               cm.ContainerManager
 	ContainerRuntime               string
+	RemoteRuntimeEndpoint          string
+	RemoteImageEndpoint            string
 	RuntimeRequestTimeout          time.Duration
 	CPUCFSQuota                    bool
 	DiskSpacePolicy                kubelet.DiskSpacePolicy
@@ -1025,6 +1029,8 @@ func CreateAndInitKubelet(kc *KubeletConfig) (k KubeletBootstrap, pc *config.Pod
 		kc.CgroupsPerQOS,
 		kc.CgroupRoot,
 		kc.ContainerRuntime,
+		kc.RemoteRuntimeEndpoint,
+		kc.RemoteImageEndpoint,
 		kc.RuntimeRequestTimeout,
 		kc.RktPath,
 		kc.RktAPIEndpoint,

hack/verify-flags/known-flags.txt

@@ -90,6 +90,7 @@ consumer-service-namespace
 contain-pod-resources
 container-port
 container-runtime
+container-runtime-endpoint
 controller-start-interval
 cors-allowed-origins
 cpu-cfs-quota
@@ -223,6 +224,7 @@ image-gc-high-threshold
 image-gc-low-threshold
 image-project
 image-pull-policy
+image-service-endpoint
 include-extended-apis
 included-types-overrides
 input-base

pkg/apis/componentconfig/types.go

@@ -279,6 +279,10 @@ type KubeletConfiguration struct {
 	CgroupRoot string `json:"cgroupRoot,omitempty"`
 	// containerRuntime is the container runtime to use.
 	ContainerRuntime string `json:"containerRuntime"`
+	// remoteRuntimeEndpoint is the endpoint of remote runtime service
+	RemoteRuntimeEndpoint string `json:"remoteRuntimeEndpoint"`
+	// remoteImageEndpoint is the endpoint of remote image service
+	RemoteImageEndpoint string `json:"remoteImageEndpoint"`
 	// runtimeRequestTimeout is the timeout for all runtime requests except long running
 	// requests - pull, logs, exec and attach.
 	RuntimeRequestTimeout unversioned.Duration `json:"runtimeRequestTimeout,omitempty"`

pkg/apis/componentconfig/v1alpha1/types.go

@@ -334,6 +334,10 @@ type KubeletConfiguration struct {
 	CgroupsPerQOS *bool `json:"CgroupsPerQOS,omitempty"`
 	// containerRuntime is the container runtime to use.
 	ContainerRuntime string `json:"containerRuntime"`
+	// remoteRuntimeEndpoint is the endpoint of remote runtime service
+	RemoteRuntimeEndpoint string `json:"remoteRuntimeEndpoint"`
+	// remoteImageEndpoint is the endpoint of remote image service
+	RemoteImageEndpoint string `json:"remoteImageEndpoint"`
 	// runtimeRequestTimeout is the timeout for all runtime requests except long running
 	// requests - pull, logs, exec and attach.
 	RuntimeRequestTimeout unversioned.Duration `json:"runtimeRequestTimeout"`

pkg/apis/componentconfig/v1alpha1/zz_generated.conversion.go

@@ -244,6 +244,8 @@ func autoConvert_v1alpha1_KubeletConfiguration_To_componentconfig_KubeletConfigu
 		return err
 	}
 	out.ContainerRuntime = in.ContainerRuntime
+	out.RemoteRuntimeEndpoint = in.RemoteRuntimeEndpoint
+	out.RemoteImageEndpoint = in.RemoteImageEndpoint
 	out.RuntimeRequestTimeout = in.RuntimeRequestTimeout
 	out.RktPath = in.RktPath
 	out.RktAPIEndpoint = in.RktAPIEndpoint
@@ -418,6 +420,8 @@ func autoConvert_componentconfig_KubeletConfiguration_To_v1alpha1_KubeletConfigu
 	out.SystemCgroups = in.SystemCgroups
 	out.CgroupRoot = in.CgroupRoot
 	out.ContainerRuntime = in.ContainerRuntime
+	out.RemoteRuntimeEndpoint = in.RemoteRuntimeEndpoint
+	out.RemoteImageEndpoint = in.RemoteImageEndpoint
 	out.RuntimeRequestTimeout = in.RuntimeRequestTimeout
 	out.RktPath = in.RktPath
 	out.RktAPIEndpoint = in.RktAPIEndpoint

pkg/apis/componentconfig/v1alpha1/zz_generated.deepcopy.go

@@ -254,6 +254,8 @@ func DeepCopy_v1alpha1_KubeletConfiguration(in interface{}, out interface{}, c *
 			out.CgroupsPerQOS = nil
 		}
 		out.ContainerRuntime = in.ContainerRuntime
+		out.RemoteRuntimeEndpoint = in.RemoteRuntimeEndpoint
+		out.RemoteImageEndpoint = in.RemoteImageEndpoint
 		out.RuntimeRequestTimeout = in.RuntimeRequestTimeout
 		out.RktPath = in.RktPath
 		out.RktAPIEndpoint = in.RktAPIEndpoint

pkg/apis/componentconfig/zz_generated.deepcopy.go

@@ -266,6 +266,8 @@ func DeepCopy_componentconfig_KubeletConfiguration(in interface{}, out interface
 		out.SystemCgroups = in.SystemCgroups
 		out.CgroupRoot = in.CgroupRoot
 		out.ContainerRuntime = in.ContainerRuntime
+		out.RemoteRuntimeEndpoint = in.RemoteRuntimeEndpoint
+		out.RemoteImageEndpoint = in.RemoteImageEndpoint
 		out.RuntimeRequestTimeout = in.RuntimeRequestTimeout
 		out.RktPath = in.RktPath
 		out.RktAPIEndpoint = in.RktAPIEndpoint

pkg/kubelet/kubelet.go

@@ -54,6 +54,7 @@ import (
 	"k8s.io/kubernetes/pkg/kubelet/events"
 	"k8s.io/kubernetes/pkg/kubelet/eviction"
 	"k8s.io/kubernetes/pkg/kubelet/images"
+	"k8s.io/kubernetes/pkg/kubelet/kuberuntime"
 	"k8s.io/kubernetes/pkg/kubelet/lifecycle"
 	"k8s.io/kubernetes/pkg/kubelet/metrics"
 	"k8s.io/kubernetes/pkg/kubelet/network"
@@ -61,6 +62,7 @@ import (
 	kubepod "k8s.io/kubernetes/pkg/kubelet/pod"
 	"k8s.io/kubernetes/pkg/kubelet/prober"
 	proberesults "k8s.io/kubernetes/pkg/kubelet/prober/results"
+	"k8s.io/kubernetes/pkg/kubelet/remote"
 	"k8s.io/kubernetes/pkg/kubelet/rkt"
 	"k8s.io/kubernetes/pkg/kubelet/server"
 	"k8s.io/kubernetes/pkg/kubelet/server/stats"
@@ -212,6 +214,8 @@ func NewMainKubelet(
 	CgroupsPerQOS bool,
 	cgroupRoot string,
 	containerRuntime string,
+	remoteRuntimeEndpoint string,
+	remoteImageEndpoint string,
 	runtimeRequestTimeout time.Duration,
 	rktPath string,
 	rktAPIEndpoint string,
@@ -415,6 +419,15 @@ func NewMainKubelet(
 	klet.podCache = kubecontainer.NewCache()
 	klet.podManager = kubepod.NewBasicPodManager(kubepod.NewBasicMirrorClient(klet.kubeClient))
 
+	if remoteRuntimeEndpoint != "" {
+		containerRuntime = "remote"
+
+		// remoteImageEndpoint is same as remoteRuntimeEndpoint if not explicitly specified
+		if remoteImageEndpoint == "" {
+			remoteImageEndpoint = remoteRuntimeEndpoint
+		}
+	}
+
 	// Initialize the runtime.
 	switch containerRuntime {
 	case "docker":
@@ -479,6 +492,32 @@ func NewMainKubelet(
 			return nil, err
 		}
 		klet.containerRuntime = rktRuntime
+	case "remote":
+		remoteRuntimeService, err := remote.NewRemoteRuntimeService(remoteRuntimeEndpoint, runtimeRequestTimeout)
+		if err != nil {
+			return nil, err
+		}
+		remoteImageService, err := remote.NewRemoteImageService(remoteImageEndpoint, runtimeRequestTimeout)
+		if err != nil {
+			return nil, err
+		}
+		klet.containerRuntime, err = kuberuntime.NewKubeGenericRuntimeManager(
+			kubecontainer.FilterEventRecorder(recorder),
+			klet.livenessManager,
+			containerRefManager,
+			osInterface,
+			klet.networkPlugin,
+			klet,
+			klet.httpClient,
+			imageBackOff,
+			serializeImagePulls,
+			klet.cpuCFSQuota,
+			remoteRuntimeService,
+			remoteImageService,
+		)
+		if err != nil {
+			return nil, err
+		}
 	default:
 		return nil, fmt.Errorf("unsupported container runtime %q specified", containerRuntime)
 	}

pkg/kubelet/kuberuntime/fake_kuberuntime_manager.go

@@ -22,13 +22,13 @@ import (
 	"time"
 
 	"k8s.io/kubernetes/pkg/api"
-	"k8s.io/kubernetes/pkg/apis/componentconfig"
 	"k8s.io/kubernetes/pkg/client/record"
+	"k8s.io/kubernetes/pkg/credentialprovider"
 	internalApi "k8s.io/kubernetes/pkg/kubelet/api"
 	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
-	containertest "k8s.io/kubernetes/pkg/kubelet/container/testing"
+	"k8s.io/kubernetes/pkg/kubelet/images"
+	"k8s.io/kubernetes/pkg/kubelet/lifecycle"
 	"k8s.io/kubernetes/pkg/kubelet/network"
-	nettest "k8s.io/kubernetes/pkg/kubelet/network/testing"
 	proberesults "k8s.io/kubernetes/pkg/kubelet/prober/results"
 	kubetypes "k8s.io/kubernetes/pkg/types"
 	"k8s.io/kubernetes/pkg/util/flowcontrol"
@@ -76,27 +76,36 @@ func (f *fakeRuntimeHelper) GetExtraSupplementalGroupsForPod(pod *api.Pod) []int
 	return nil
 }
 
-func NewFakeKubeRuntimeManager(runtimeService internalApi.RuntimeService, imageService internalApi.ImageManagerService) (*kubeGenericRuntimeManager, error) {
+func NewFakeKubeRuntimeManager(runtimeService internalApi.RuntimeService, imageService internalApi.ImageManagerService, networkPlugin network.NetworkPlugin, osInterface kubecontainer.OSInterface) (*kubeGenericRuntimeManager, error) {
-	networkPlugin, _ := network.InitNetworkPlugin(
+	recorder := &record.FakeRecorder{}
-		[]network.NetworkPlugin{},
+	kubeRuntimeManager := &kubeGenericRuntimeManager{
-		"",
+		recorder:            recorder,
-		nettest.NewFakeHost(nil),
+		cpuCFSQuota:         false,
-		componentconfig.HairpinNone,
+		livenessManager:     proberesults.NewManager(),
-		"10.0.0.0/8",
+		containerRefManager: kubecontainer.NewRefManager(),
-	)
+		osInterface:         osInterface,
-
+		networkPlugin:       networkPlugin,
-	return NewKubeGenericRuntimeManager(
+		runtimeHelper:       &fakeRuntimeHelper{},
-		&record.FakeRecorder{},
+		runtimeService:      runtimeService,
-		proberesults.NewManager(),
+		imageService:        imageService,
-		kubecontainer.NewRefManager(),
+		keyring:             credentialprovider.NewDockerKeyring(),
-		&containertest.FakeOS{},
+	}
-		networkPlugin,
+
-		&fakeRuntimeHelper{},
+	typedVersion, err := runtimeService.Version(kubeRuntimeAPIVersion)
-		&fakeHTTP{},
+	if err != nil {
-		flowcontrol.NewBackOff(time.Second, 300*time.Second),
+		return nil, err
-		false,
+	}
-		false,
+
-		runtimeService,
+	kubeRuntimeManager.runtimeName = typedVersion.GetRuntimeName()
-		imageService,
+	kubeRuntimeManager.imagePuller = images.NewImageManager(
-	)
+		kubecontainer.FilterEventRecorder(recorder),
+		kubeRuntimeManager,
+		flowcontrol.NewBackOff(time.Second, 300*time.Second),
+		false)
+	kubeRuntimeManager.runner = lifecycle.NewHandlerRunner(
+		&fakeHTTP{},
+		kubeRuntimeManager,
+		kubeRuntimeManager)
+
+	return kubeRuntimeManager, nil
 }

pkg/kubelet/kuberuntime/kuberuntime_manager.go

@@ -96,7 +96,7 @@ func NewKubeGenericRuntimeManager(
 	cpuCFSQuota bool,
 	runtimeService internalApi.RuntimeService,
 	imageService internalApi.ImageManagerService,
-) (*kubeGenericRuntimeManager, error) {
+) (kubecontainer.Runtime, error) {
 	kubeRuntimeManager := &kubeGenericRuntimeManager{
 		recorder:            recorder,
 		cpuCFSQuota:         cpuCFSQuota,

pkg/kubelet/kuberuntime/kuberuntime_manager_test.go

@@ -20,13 +20,25 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"k8s.io/kubernetes/pkg/apis/componentconfig"
 	apitest "k8s.io/kubernetes/pkg/kubelet/api/testing"
+	containertest "k8s.io/kubernetes/pkg/kubelet/container/testing"
+	"k8s.io/kubernetes/pkg/kubelet/network"
+	nettest "k8s.io/kubernetes/pkg/kubelet/network/testing"
 )
 
 func createTestFakeRuntimeManager() (*apitest.FakeRuntimeService, *apitest.FakeImageService, *kubeGenericRuntimeManager, error) {
 	fakeRuntimeService := apitest.NewFakeRuntimeService()
 	fakeImageService := apitest.NewFakeImageService()
-	manager, err := NewFakeKubeRuntimeManager(fakeRuntimeService, fakeImageService)
+	networkPlugin, _ := network.InitNetworkPlugin(
+		[]network.NetworkPlugin{},
+		"",
+		nettest.NewFakeHost(nil),
+		componentconfig.HairpinNone,
+		"10.0.0.0/8",
+	)
+	osInterface := &containertest.FakeOS{}
+	manager, err := NewFakeKubeRuntimeManager(fakeRuntimeService, fakeImageService, networkPlugin, osInterface)
 	return fakeRuntimeService, fakeImageService, manager, err
 }