Merge pull request #85879 from gongguan/cri-kubemark

hollow-node use remote CRI

cmd/kubelet/app/server.go

@@ -757,6 +757,17 @@ func run(s *options.KubeletServer, kubeDeps *kubelet.Dependencies, featureGate f
 		klog.Warning(err)
 	}
 
+	err = kubelet.PreInitRuntimeService(&s.KubeletConfiguration,
+		kubeDeps, &s.ContainerRuntimeOptions,
+		s.ContainerRuntime,
+		s.RuntimeCgroups,
+		s.RemoteRuntimeEndpoint,
+		s.RemoteImageEndpoint,
+		s.NonMasqueradeCIDR)
+	if err != nil {
+		return err
+	}
+
 	if err := RunKubelet(s, kubeDeps, s.RunOnce); err != nil {
 		return err
 	}
@@ -1062,7 +1073,6 @@ func RunKubelet(kubeServer *options.KubeletServer, kubeDeps *kubelet.Dependencie
 		kubeDeps,
 		&kubeServer.ContainerRuntimeOptions,
 		kubeServer.ContainerRuntime,
-		kubeServer.RuntimeCgroups,
 		kubeServer.HostnameOverride,
 		kubeServer.NodeIP,
 		kubeServer.ProviderID,
@@ -1072,8 +1082,6 @@ func RunKubelet(kubeServer *options.KubeletServer, kubeDeps *kubelet.Dependencie
 		kubeServer.RegisterNode,
 		kubeServer.RegisterWithTaints,
 		kubeServer.AllowedUnsafeSysctls,
-		kubeServer.RemoteRuntimeEndpoint,
-		kubeServer.RemoteImageEndpoint,
 		kubeServer.ExperimentalMounterPath,
 		kubeServer.ExperimentalKernelMemcgNotification,
 		kubeServer.ExperimentalCheckNodeCapabilitiesBeforeMount,
@@ -1083,7 +1091,6 @@ func RunKubelet(kubeServer *options.KubeletServer, kubeDeps *kubelet.Dependencie
 		kubeServer.MaxContainerCount,
 		kubeServer.MasterServiceNamespace,
 		kubeServer.RegisterSchedulable,
-		kubeServer.NonMasqueradeCIDR,
 		kubeServer.KeepTerminatedPodVolumes,
 		kubeServer.NodeLabels,
 		kubeServer.SeccompProfileRoot,
@@ -1138,7 +1145,6 @@ func createAndInitKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
 	kubeDeps *kubelet.Dependencies,
 	crOptions *config.ContainerRuntimeOptions,
 	containerRuntime string,
-	runtimeCgroups string,
 	hostnameOverride string,
 	nodeIP string,
 	providerID string,
@@ -1148,8 +1154,6 @@ func createAndInitKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
 	registerNode bool,
 	registerWithTaints []api.Taint,
 	allowedUnsafeSysctls []string,
-	remoteRuntimeEndpoint string,
-	remoteImageEndpoint string,
 	experimentalMounterPath string,
 	experimentalKernelMemcgNotification bool,
 	experimentalCheckNodeCapabilitiesBeforeMount bool,
@@ -1159,7 +1163,6 @@ func createAndInitKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
 	maxContainerCount int32,
 	masterServiceNamespace string,
 	registerSchedulable bool,
-	nonMasqueradeCIDR string,
 	keepTerminatedPodVolumes bool,
 	nodeLabels map[string]string,
 	seccompProfileRoot string,
@@ -1172,7 +1175,6 @@ func createAndInitKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
 		kubeDeps,
 		crOptions,
 		containerRuntime,
-		runtimeCgroups,
 		hostnameOverride,
 		nodeIP,
 		providerID,
@@ -1182,8 +1184,6 @@ func createAndInitKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
 		registerNode,
 		registerWithTaints,
 		allowedUnsafeSysctls,
-		remoteRuntimeEndpoint,
-		remoteImageEndpoint,
 		experimentalMounterPath,
 		experimentalKernelMemcgNotification,
 		experimentalCheckNodeCapabilitiesBeforeMount,
@@ -1193,7 +1193,6 @@ func createAndInitKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
 		maxContainerCount,
 		masterServiceNamespace,
 		registerSchedulable,
-		nonMasqueradeCIDR,
 		keepTerminatedPodVolumes,
 		nodeLabels,
 		seccompProfileRoot,

cmd/kubemark/BUILD

@@ -19,8 +19,7 @@ go_library(
         "//pkg/api/legacyscheme:go_default_library",
         "//pkg/kubelet/cadvisor/testing:go_default_library",
         "//pkg/kubelet/cm:go_default_library",
-        "//pkg/kubelet/dockershim:go_default_library",
+        "//pkg/kubelet/remote/fake:go_default_library",
-        "//pkg/kubelet/dockershim/libdocker:go_default_library",
         "//pkg/kubemark:go_default_library",
         "//pkg/master/ports:go_default_library",
         "//pkg/util/iptables/testing:go_default_library",

cmd/kubemark/hollow-node.go

@@ -43,8 +43,7 @@ import (
 	"k8s.io/kubernetes/pkg/api/legacyscheme"
 	cadvisortest "k8s.io/kubernetes/pkg/kubelet/cadvisor/testing"
 	"k8s.io/kubernetes/pkg/kubelet/cm"
-	"k8s.io/kubernetes/pkg/kubelet/dockershim"
+	fakeremote "k8s.io/kubernetes/pkg/kubelet/remote/fake"
-	"k8s.io/kubernetes/pkg/kubelet/dockershim/libdocker"
 	"k8s.io/kubernetes/pkg/kubemark"
 	"k8s.io/kubernetes/pkg/master/ports"
 	fakeiptables "k8s.io/kubernetes/pkg/util/iptables/testing"
@@ -194,19 +193,14 @@ func run(config *hollowNodeConfig) {
 			NodeName: config.NodeName,
 		}
 		containerManager := cm.NewStubContainerManager()
-
+		fakeRemoteRuntime := fakeremote.NewFakeRemoteRuntime()
-		fakeDockerClientConfig := &dockershim.ClientConfig{
-			DockerEndpoint:    libdocker.FakeDockerEndpoint,
-			EnableSleep:       true,
-			WithTraceDisabled: true,
-		}
 
 		hollowKubelet := kubemark.NewHollowKubelet(
 			f, c,
 			client,
 			heartbeatClient,
 			cadvisorInterface,
-			fakeDockerClientConfig,
+			fakeRemoteRuntime,
 			containerManager,
 		)
 		hollowKubelet.Run()

pkg/kubelet/dockershim/docker_service.go

@@ -179,8 +179,6 @@ func NewDockerClientFromConfig(config *ClientConfig) libdocker.Interface {
 			config.DockerEndpoint,
 			config.RuntimeRequestTimeout,
 			config.ImagePullProgressDeadline,
-			config.WithTraceDisabled,
-			config.EnableSleep,
 		)
 		return client
 	}

pkg/kubelet/dockershim/libdocker/client.go

@@ -84,19 +84,7 @@ func getDockerClient(dockerEndpoint string) (*dockerapi.Client, error) {
 // is the timeout for docker requests. If timeout is exceeded, the request
 // will be cancelled and throw out an error. If requestTimeout is 0, a default
 // value will be applied.
-func ConnectToDockerOrDie(dockerEndpoint string, requestTimeout, imagePullProgressDeadline time.Duration,
+func ConnectToDockerOrDie(dockerEndpoint string, requestTimeout, imagePullProgressDeadline time.Duration) Interface {
-	withTraceDisabled bool, enableSleep bool) Interface {
-	if dockerEndpoint == FakeDockerEndpoint {
-		fakeClient := NewFakeDockerClient()
-		if withTraceDisabled {
-			fakeClient = fakeClient.WithTraceDisabled()
-		}
-
-		if enableSleep {
-			fakeClient.EnableSleep = true
-		}
-		return fakeClient
-	}
 	client, err := getDockerClient(dockerEndpoint)
 	if err != nil {
 		klog.Fatalf("Couldn't connect to docker: %v", err)

pkg/kubelet/kubelet.go

@@ -264,6 +264,12 @@ type Dependencies struct {
 	DynamicPluginProber     volume.DynamicPluginProber
 	TLSOptions              *server.TLSOptions
 	KubeletConfigController *kubeletconfig.Controller
+	RemoteRuntimeService    internalapi.RuntimeService
+	RemoteImageService      internalapi.ImageManagerService
+	criHandler              http.Handler
+	dockerLegacyService     dockershim.DockerLegacyService
+	// remove it after cadvisor.UsingLegacyCadvisorStats dropped.
+	useLegacyCadvisorStats bool
 }
 
 // makePodSourceConfig creates a config.PodConfig from the given
@@ -317,16 +323,82 @@ func makePodSourceConfig(kubeCfg *kubeletconfiginternal.KubeletConfiguration, ku
 	return cfg, nil
 }
 
-func getRuntimeAndImageServices(remoteRuntimeEndpoint string, remoteImageEndpoint string, runtimeRequestTimeout metav1.Duration) (internalapi.RuntimeService, internalapi.ImageManagerService, error) {
+// PreInitRuntimeService will init runtime service before RunKubelet.
-	rs, err := remote.NewRemoteRuntimeService(remoteRuntimeEndpoint, runtimeRequestTimeout.Duration)
+func PreInitRuntimeService(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
-	if err != nil {
+	kubeDeps *Dependencies,
-		return nil, nil, err
+	crOptions *config.ContainerRuntimeOptions,
+	containerRuntime string,
+	runtimeCgroups string,
+	remoteRuntimeEndpoint string,
+	remoteImageEndpoint string,
+	nonMasqueradeCIDR string) error {
+	if remoteRuntimeEndpoint != "" {
+		// remoteImageEndpoint is same as remoteRuntimeEndpoint if not explicitly specified
+		if remoteImageEndpoint == "" {
+			remoteImageEndpoint = remoteRuntimeEndpoint
+		}
 	}
-	is, err := remote.NewRemoteImageService(remoteImageEndpoint, runtimeRequestTimeout.Duration)
+
-	if err != nil {
+	switch containerRuntime {
-		return nil, nil, err
+	case kubetypes.DockerContainerRuntime:
+		// TODO: These need to become arguments to a standalone docker shim.
+		pluginSettings := dockershim.NetworkPluginSettings{
+			HairpinMode:        kubeletconfiginternal.HairpinMode(kubeCfg.HairpinMode),
+			NonMasqueradeCIDR:  nonMasqueradeCIDR,
+			PluginName:         crOptions.NetworkPluginName,
+			PluginConfDir:      crOptions.CNIConfDir,
+			PluginBinDirString: crOptions.CNIBinDir,
+			PluginCacheDir:     crOptions.CNICacheDir,
+			MTU:                int(crOptions.NetworkPluginMTU),
+		}
+
+		// Create and start the CRI shim running as a grpc server.
+		streamingConfig := getStreamingConfig(kubeCfg, kubeDeps, crOptions)
+		ds, err := dockershim.NewDockerService(kubeDeps.DockerClientConfig, crOptions.PodSandboxImage, streamingConfig,
+			&pluginSettings, runtimeCgroups, kubeCfg.CgroupDriver, crOptions.DockershimRootDirectory, !crOptions.RedirectContainerStreaming)
+		if err != nil {
+			return err
+		}
+		if crOptions.RedirectContainerStreaming {
+			kubeDeps.criHandler = ds
+		}
+
+		// The unix socket for kubelet <-> dockershim communication, dockershim start before runtime service init.
+		klog.V(5).Infof("RemoteRuntimeEndpoint: %q, RemoteImageEndpoint: %q",
+			remoteRuntimeEndpoint,
+			remoteImageEndpoint)
+		klog.V(2).Infof("Starting the GRPC server for the docker CRI shim.")
+		dockerServer := dockerremote.NewDockerServer(remoteRuntimeEndpoint, ds)
+		if err := dockerServer.Start(); err != nil {
+			return err
+		}
+
+		// Create dockerLegacyService when the logging driver is not supported.
+		supported, err := ds.IsCRISupportedLogDriver()
+		if err != nil {
+			return err
+		}
+		if !supported {
+			kubeDeps.dockerLegacyService = ds
+		}
+	case kubetypes.RemoteContainerRuntime:
+		// No-op.
+		break
+	default:
+		return fmt.Errorf("unsupported CRI runtime: %q", containerRuntime)
 	}
-	return rs, is, err
+
+	var err error
+	if kubeDeps.RemoteRuntimeService, err = remote.NewRemoteRuntimeService(remoteRuntimeEndpoint, kubeCfg.RuntimeRequestTimeout.Duration); err != nil {
+		return err
+	}
+	if kubeDeps.RemoteImageService, err = remote.NewRemoteImageService(remoteImageEndpoint, kubeCfg.RuntimeRequestTimeout.Duration); err != nil {
+		return err
+	}
+
+	kubeDeps.useLegacyCadvisorStats = cadvisor.UsingLegacyCadvisorStats(containerRuntime, remoteRuntimeEndpoint)
+
+	return nil
 }
 
 // NewMainKubelet instantiates a new Kubelet object along with all the required internal modules.
@@ -335,7 +407,6 @@ func NewMainKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
 	kubeDeps *Dependencies,
 	crOptions *config.ContainerRuntimeOptions,
 	containerRuntime string,
-	runtimeCgroups string,
 	hostnameOverride string,
 	nodeIP string,
 	providerID string,
@@ -345,8 +416,6 @@ func NewMainKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
 	registerNode bool,
 	registerWithTaints []api.Taint,
 	allowedUnsafeSysctls []string,
-	remoteRuntimeEndpoint string,
-	remoteImageEndpoint string,
 	experimentalMounterPath string,
 	experimentalKernelMemcgNotification bool,
 	experimentalCheckNodeCapabilitiesBeforeMount bool,
@@ -356,7 +425,6 @@ func NewMainKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
 	maxContainerCount int32,
 	masterServiceNamespace string,
 	registerSchedulable bool,
-	nonMasqueradeCIDR string,
 	keepTerminatedPodVolumes bool,
 	nodeLabels map[string]string,
 	seccompProfileRoot string,
@@ -599,72 +667,11 @@ func NewMainKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
 
 	klet.statusManager = status.NewManager(klet.kubeClient, klet.podManager, klet)
 
-	if remoteRuntimeEndpoint != "" {
-		// remoteImageEndpoint is same as remoteRuntimeEndpoint if not explicitly specified
-		if remoteImageEndpoint == "" {
-			remoteImageEndpoint = remoteRuntimeEndpoint
-		}
-	}
-
-	// TODO: These need to become arguments to a standalone docker shim.
-	pluginSettings := dockershim.NetworkPluginSettings{
-		HairpinMode:        kubeletconfiginternal.HairpinMode(kubeCfg.HairpinMode),
-		NonMasqueradeCIDR:  nonMasqueradeCIDR,
-		PluginName:         crOptions.NetworkPluginName,
-		PluginConfDir:      crOptions.CNIConfDir,
-		PluginBinDirString: crOptions.CNIBinDir,
-		PluginCacheDir:     crOptions.CNICacheDir,
-		MTU:                int(crOptions.NetworkPluginMTU),
-	}
-
 	klet.resourceAnalyzer = serverstats.NewResourceAnalyzer(klet, kubeCfg.VolumeStatsAggPeriod.Duration)
 
-	// if left at nil, that means it is unneeded
+	klet.dockerLegacyService = kubeDeps.dockerLegacyService
-	var legacyLogProvider kuberuntime.LegacyLogProvider
+	klet.criHandler = kubeDeps.criHandler
-
+	klet.runtimeService = kubeDeps.RemoteRuntimeService
-	switch containerRuntime {
-	case kubetypes.DockerContainerRuntime:
-		// Create and start the CRI shim running as a grpc server.
-		streamingConfig := getStreamingConfig(kubeCfg, kubeDeps, crOptions)
-		ds, err := dockershim.NewDockerService(kubeDeps.DockerClientConfig, crOptions.PodSandboxImage, streamingConfig,
-			&pluginSettings, runtimeCgroups, kubeCfg.CgroupDriver, crOptions.DockershimRootDirectory, !crOptions.RedirectContainerStreaming)
-		if err != nil {
-			return nil, err
-		}
-		if crOptions.RedirectContainerStreaming {
-			klet.criHandler = ds
-		}
-
-		// The unix socket for kubelet <-> dockershim communication.
-		klog.V(5).Infof("RemoteRuntimeEndpoint: %q, RemoteImageEndpoint: %q",
-			remoteRuntimeEndpoint,
-			remoteImageEndpoint)
-		klog.V(2).Infof("Starting the GRPC server for the docker CRI shim.")
-		server := dockerremote.NewDockerServer(remoteRuntimeEndpoint, ds)
-		if err := server.Start(); err != nil {
-			return nil, err
-		}
-
-		// Create dockerLegacyService when the logging driver is not supported.
-		supported, err := ds.IsCRISupportedLogDriver()
-		if err != nil {
-			return nil, err
-		}
-		if !supported {
-			klet.dockerLegacyService = ds
-			legacyLogProvider = ds
-		}
-	case kubetypes.RemoteContainerRuntime:
-		// No-op.
-		break
-	default:
-		return nil, fmt.Errorf("unsupported CRI runtime: %q", containerRuntime)
-	}
-	runtimeService, imageService, err := getRuntimeAndImageServices(remoteRuntimeEndpoint, remoteImageEndpoint, kubeCfg.RuntimeRequestTimeout)
-	if err != nil {
-		return nil, err
-	}
-	klet.runtimeService = runtimeService
 
 	if utilfeature.DefaultFeatureGate.Enabled(features.RuntimeClass) && kubeDeps.KubeClient != nil {
 		klet.runtimeClassManager = runtimeclass.NewManager(kubeDeps.KubeClient)
@@ -687,10 +694,10 @@ func NewMainKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
 		int(kubeCfg.RegistryBurst),
 		kubeCfg.CPUCFSQuota,
 		kubeCfg.CPUCFSQuotaPeriod,
-		runtimeService,
+		kubeDeps.RemoteRuntimeService,
-		imageService,
+		kubeDeps.RemoteImageService,
 		kubeDeps.ContainerManager.InternalContainerLifecycle(),
-		legacyLogProvider,
+		kubeDeps.dockerLegacyService,
 		klet.runtimeClassManager,
 	)
 	if err != nil {
@@ -706,7 +713,7 @@ func NewMainKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
 	}
 	klet.runtimeCache = runtimeCache
 
-	if cadvisor.UsingLegacyCadvisorStats(containerRuntime, remoteRuntimeEndpoint) {
+	if kubeDeps.useLegacyCadvisorStats {
 		klet.StatsProvider = stats.NewCadvisorStatsProvider(
 			klet.cadvisor,
 			klet.resourceAnalyzer,
@@ -720,8 +727,8 @@ func NewMainKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
 			klet.resourceAnalyzer,
 			klet.podManager,
 			klet.runtimeCache,
-			runtimeService,
+			kubeDeps.RemoteRuntimeService,
-			imageService,
+			kubeDeps.RemoteImageService,
 			stats.NewLogMetricsService(),
 			kubecontainer.RealOS{})
 	}

pkg/kubemark/BUILD

@@ -22,7 +22,7 @@ go_library(
         "//pkg/kubelet/cadvisor:go_default_library",
         "//pkg/kubelet/cm:go_default_library",
         "//pkg/kubelet/container/testing:go_default_library",
-        "//pkg/kubelet/dockershim:go_default_library",
+        "//pkg/kubelet/remote/fake:go_default_library",
         "//pkg/kubelet/types:go_default_library",
         "//pkg/proxy:go_default_library",
         "//pkg/proxy/config:go_default_library",

pkg/kubemark/hollow_kubelet.go

@@ -32,7 +32,7 @@ import (
 	"k8s.io/kubernetes/pkg/kubelet/cadvisor"
 	"k8s.io/kubernetes/pkg/kubelet/cm"
 	containertest "k8s.io/kubernetes/pkg/kubelet/container/testing"
-	"k8s.io/kubernetes/pkg/kubelet/dockershim"
+	fakeremote "k8s.io/kubernetes/pkg/kubelet/remote/fake"
 	kubetypes "k8s.io/kubernetes/pkg/kubelet/types"
 	"k8s.io/kubernetes/pkg/util/oom"
 	"k8s.io/kubernetes/pkg/volume"
@@ -98,22 +98,23 @@ func NewHollowKubelet(
 	client *clientset.Clientset,
 	heartbeatClient *clientset.Clientset,
 	cadvisorInterface cadvisor.Interface,
-	dockerClientConfig *dockershim.ClientConfig,
+	remoteRuntime *fakeremote.RemoteRuntime,
 	containerManager cm.ContainerManager) *HollowKubelet {
 	d := &kubelet.Dependencies{
-		KubeClient:         client,
+		KubeClient:           client,
-		HeartbeatClient:    heartbeatClient,
+		HeartbeatClient:      heartbeatClient,
-		DockerClientConfig: dockerClientConfig,
+		RemoteRuntimeService: remoteRuntime.RuntimeService,
-		CAdvisorInterface:  cadvisorInterface,
+		RemoteImageService:   remoteRuntime.ImageService,
-		Cloud:              nil,
+		CAdvisorInterface:    cadvisorInterface,
-		OSInterface:        &containertest.FakeOS{},
+		Cloud:                nil,
-		ContainerManager:   containerManager,
+		OSInterface:          &containertest.FakeOS{},
-		VolumePlugins:      volumePlugins(),
+		ContainerManager:     containerManager,
-		TLSOptions:         nil,
+		VolumePlugins:        volumePlugins(),
-		OOMAdjuster:        oom.NewFakeOOMAdjuster(),
+		TLSOptions:           nil,
-		Mounter:            &mount.FakeMounter{},
+		OOMAdjuster:          oom.NewFakeOOMAdjuster(),
-		Subpather:          &subpath.FakeSubpath{},
+		Mounter:              &mount.FakeMounter{},
-		HostUtil:           hostutil.NewFakeHostUtil(nil),
+		Subpather:            &subpath.FakeSubpath{},
+		HostUtil:             hostutil.NewFakeHostUtil(nil),
 	}
 
 	return &HollowKubelet{
@@ -160,6 +161,7 @@ func GetHollowKubeletConfig(opt *HollowKubletOptions) (*options.KubeletFlags, *k
 	f.MaxContainerCount = 100
 	f.MaxPerPodContainerCount = 2
 	f.NodeLabels = opt.NodeLabels
+	f.ContainerRuntimeOptions.ContainerRuntime = kubetypes.RemoteContainerRuntime
 	f.RegisterNode = true
 	f.RegisterSchedulable = true
 	f.ProviderID = fmt.Sprintf("kubemark://%v", opt.NodeName)