Merge pull request #64749 from Random-Liu/fix-standalone-dockershim

cmd/kubelet/app/server.go

@@ -721,7 +721,7 @@ func run(s *options.KubeletServer, kubeDeps *kubelet.Dependencies, stopCh <-chan
 		glog.Warning(err)
 	}
 
-	if err := RunKubelet(&s.KubeletFlags, &s.KubeletConfiguration, kubeDeps, s.RunOnce, stopCh); err != nil {
+	if err := RunKubelet(&s.KubeletFlags, &s.KubeletConfiguration, kubeDeps, s.RunOnce); err != nil {
 		return err
 	}
 
@@ -887,7 +887,7 @@ func addChaosToClientConfig(s *options.KubeletServer, config *restclient.Config)
 //   2 Kubelet binary
 //   3 Standalone 'kubernetes' binary
 // Eventually, #2 will be replaced with instances of #3
-func RunKubelet(kubeFlags *options.KubeletFlags, kubeCfg *kubeletconfiginternal.KubeletConfiguration, kubeDeps *kubelet.Dependencies, runOnce bool, stopCh <-chan struct{}) error {
+func RunKubelet(kubeFlags *options.KubeletFlags, kubeCfg *kubeletconfiginternal.KubeletConfiguration, kubeDeps *kubelet.Dependencies, runOnce bool) error {
 	hostname := nodeutil.GetHostname(kubeFlags.HostnameOverride)
 	// Query the cloud provider for our node name, default to hostname if kubeDeps.Cloud == nil
 	nodeName, err := getNodeName(kubeDeps.Cloud, hostname)
@@ -961,8 +961,7 @@ func RunKubelet(kubeFlags *options.KubeletFlags, kubeCfg *kubeletconfiginternal.
 		kubeFlags.NodeLabels,
 		kubeFlags.SeccompProfileRoot,
 		kubeFlags.BootstrapCheckpointPath,
-		kubeFlags.NodeStatusMaxImages,
-		stopCh)
+		kubeFlags.NodeStatusMaxImages)
 	if err != nil {
 		return fmt.Errorf("failed to create kubelet: %v", err)
 	}
@@ -1047,8 +1046,7 @@ func CreateAndInitKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
 	nodeLabels map[string]string,
 	seccompProfileRoot string,
 	bootstrapCheckpointPath string,
-	nodeStatusMaxImages int32,
-	stopCh <-chan struct{}) (k kubelet.Bootstrap, err error) {
+	nodeStatusMaxImages int32) (k kubelet.Bootstrap, err error) {
 	// TODO: block until all sources have delivered at least one update to the channel, or break the sync loop
 	// up into "per source" synchronizations
 
@@ -1082,8 +1080,7 @@ func CreateAndInitKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
 		nodeLabels,
 		seccompProfileRoot,
 		bootstrapCheckpointPath,
-		nodeStatusMaxImages,
-		stopCh)
+		nodeStatusMaxImages)
 	if err != nil {
 		return nil, err
 	}
@@ -1184,5 +1181,9 @@ func RunDockershim(f *options.KubeletFlags, c *kubeletconfiginternal.KubeletConf
 	}
 	glog.V(2).Infof("Starting the GRPC server for the docker CRI shim.")
 	server := dockerremote.NewDockerServer(f.RemoteRuntimeEndpoint, ds)
-	return server.Start(stopCh)
+	if err := server.Start(); err != nil {
+		return err
+	}
+	<-stopCh
+	return nil
 }

pkg/kubelet/dockershim/docker_service.go

@@ -85,7 +85,7 @@ const (
 type CRIService interface {
 	runtimeapi.RuntimeServiceServer
 	runtimeapi.ImageServiceServer
-	Start(<-chan struct{}) error
+	Start() error
 }
 
 // DockerService is an interface that embeds the new RuntimeService and
@@ -400,17 +400,11 @@ func (ds *dockerService) GetPodPortMappings(podSandboxID string) ([]*hostport.Po
 }
 
 // Start initializes and starts components in dockerService.
-func (ds *dockerService) Start(stopCh <-chan struct{}) error {
+func (ds *dockerService) Start() error {
 	// Initialize the legacy cleanup flag.
 	if ds.startLocalStreamingServer {
 		go func() {
-			<-stopCh
-			if err := ds.streamingServer.Stop(); err != nil {
-				glog.Errorf("Failed to stop streaming server: %v", err)
-			}
-		}()
-		go func() {
-			if err := ds.streamingServer.Start(true); err != nil && err != http.ErrServerClosed {
+			if err := ds.streamingServer.Start(true); err != nil {
 				glog.Fatalf("Streaming server stopped unexpectedly: %v", err)
 			}
 		}()

pkg/kubelet/dockershim/remote/docker_server.go

@@ -49,9 +49,9 @@ func NewDockerServer(endpoint string, s dockershim.CRIService) *DockerServer {
 }
 
 // Start starts the dockershim grpc server.
-func (s *DockerServer) Start(stopCh <-chan struct{}) error {
+func (s *DockerServer) Start() error {
 	// Start the internal service.
-	if err := s.service.Start(stopCh); err != nil {
+	if err := s.service.Start(); err != nil {
 		glog.Errorf("Unable to start docker service")
 		return err
 	}
@@ -70,18 +70,8 @@ func (s *DockerServer) Start(stopCh <-chan struct{}) error {
 	runtimeapi.RegisterImageServiceServer(s.server, s.service)
 	go func() {
 		if err := s.server.Serve(l); err != nil {
-			glog.Errorf("Failed to serve connections: %v", err)
+			glog.Fatalf("Failed to serve connections: %v", err)
 		}
 	}()
-	go func() {
-		<-stopCh
-		s.Stop()
-	}()
 	return nil
 }
-
-// Stop stops the dockershim grpc server.
-func (s *DockerServer) Stop() {
-	glog.V(2).Infof("Stop docker server")
-	s.server.Stop()
-}

pkg/kubelet/kubelet.go

@@ -223,8 +223,7 @@ type Builder func(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
 	nodeLabels map[string]string,
 	seccompProfileRoot string,
 	bootstrapCheckpointPath string,
-	nodeStatusMaxImages int32,
-	stopCh <-chan struct{}) (Bootstrap, error)
+	nodeStatusMaxImages int32) (Bootstrap, error)
 
 // Dependencies is a bin for things we might consider "injected dependencies" -- objects constructed
 // at runtime that are necessary for running the Kubelet. This is a temporary solution for grouping
@@ -350,8 +349,7 @@ func NewMainKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
 	nodeLabels map[string]string,
 	seccompProfileRoot string,
 	bootstrapCheckpointPath string,
-	nodeStatusMaxImages int32,
-	stopCh <-chan struct{}) (*Kubelet, error) {
+	nodeStatusMaxImages int32) (*Kubelet, error) {
 	if rootDirectory == "" {
 		return nil, fmt.Errorf("invalid root directory %q", rootDirectory)
 	}
@@ -632,7 +630,7 @@ func NewMainKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
 			remoteImageEndpoint)
 		glog.V(2).Infof("Starting the GRPC server for the docker CRI shim.")
 		server := dockerremote.NewDockerServer(remoteRuntimeEndpoint, ds)
-		if err := server.Start(stopCh); err != nil {
+		if err := server.Start(); err != nil {
 			return nil, err
 		}
 

pkg/kubemark/BUILD

@@ -46,7 +46,6 @@ go_library(
         "//vendor/k8s.io/apimachinery/pkg/fields:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/labels:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/types:go_default_library",
-        "//vendor/k8s.io/apimachinery/pkg/util/wait:go_default_library",
         "//vendor/k8s.io/client-go/informers:go_default_library",
         "//vendor/k8s.io/client-go/informers/core/v1:go_default_library",
         "//vendor/k8s.io/client-go/kubernetes:go_default_library",

pkg/kubemark/hollow_kubelet.go

@@ -20,7 +20,6 @@ import (
 	"time"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/util/wait"
 	clientset "k8s.io/client-go/kubernetes"
 	kubeletapp "k8s.io/kubernetes/cmd/kubelet/app"
 	"k8s.io/kubernetes/cmd/kubelet/app/options"
@@ -90,7 +89,7 @@ func NewHollowKubelet(
 
 // Starts this HollowKubelet and blocks.
 func (hk *HollowKubelet) Run() {
-	if err := kubeletapp.RunKubelet(hk.KubeletFlags, hk.KubeletConfiguration, hk.KubeletDeps, false, wait.NeverStop); err != nil {
+	if err := kubeletapp.RunKubelet(hk.KubeletFlags, hk.KubeletConfiguration, hk.KubeletDeps, false); err != nil {
 		glog.Fatalf("Failed to run HollowKubelet: %v. Exiting.", err)
 	}
 	select {}