Merge pull request #33836 from feiskyer/portforward

Automatic merge from submit-queue

Kubelet: fix port forward for dockershim

This PR fixes port forward for dockershim and also adds a `kubecontainer.FormatPod`.

Locally cluster has passed `--ginkgo.focus=Port\sforwarding'` tests.

cc/ @Random-Liu @yujuhong

pkg/kubelet/container/helpers.go

@@ -17,6 +17,7 @@ limitations under the License.
 package container
 
 import (
+	"fmt"
 	"hash/adler32"
 	"strings"
 
@@ -214,3 +215,11 @@ func SandboxToContainerState(state runtimeApi.PodSandBoxState) ContainerState {
 	}
 	return ContainerStateUnknown
 }
+
+// FormatPod returns a string representing a pod in a human readable format,
+// with pod UID as part of the string.
+func FormatPod(pod *Pod) string {
+	// Use underscore as the delimiter because it is not allowed in pod name
+	// (DNS subdomain format), while allowed in the container name format.
+	return fmt.Sprintf("%s_%s(%s)", pod.Name, pod.Namespace, pod.ID)
+}

pkg/kubelet/dockershim/docker_service.go

@@ -71,7 +71,7 @@ type DockerLegacyService interface {
 	// Supporting legacy methods for docker.
 	GetContainerLogs(pod *api.Pod, containerID kubecontainer.ContainerID, logOptions *api.PodLogOptions, stdout, stderr io.Writer) (err error)
 	kubecontainer.ContainerAttacher
-	PortForward(pod *kubecontainer.Pod, port uint16, stream io.ReadWriteCloser) error
+	PortForward(sandboxID string, port uint16, stream io.ReadWriteCloser) error
 
 	// TODO: Remove this once exec is properly defined in CRI.
 	ExecInContainer(containerID kubecontainer.ContainerID, cmd []string, stdin io.Reader, stdout, stderr io.WriteCloser, tty bool, resize <-chan term.Size) error

pkg/kubelet/dockershim/legacy.go

@@ -39,8 +39,8 @@ func (ds *dockerService) GetContainerLogs(pod *api.Pod, containerID kubecontaine
 	return dockertools.GetContainerLogs(ds.client, pod, containerID, logOptions, stdout, stderr)
 }
 
-func (ds *dockerService) PortForward(pod *kubecontainer.Pod, port uint16, stream io.ReadWriteCloser) error {
-	return dockertools.PortForward(ds.client, pod, port, stream)
+func (ds *dockerService) PortForward(sandboxID string, port uint16, stream io.ReadWriteCloser) error {
+	return dockertools.PortForward(ds.client, sandboxID, port, stream)
 }
 
 func (ds *dockerService) ExecInContainer(containerID kubecontainer.ContainerID, cmd []string, stdin io.Reader, stdout, stderr io.WriteCloser, tty bool, resize <-chan term.Size) error {

pkg/kubelet/dockertools/docker_manager.go

@@ -1290,16 +1290,17 @@ func noPodInfraContainerError(podName, podNamespace string) error {
 //  - should we support nsenter + socat on the host? (current impl)
 //  - should we support nsenter + socat in a container, running with elevated privs and --pid=host?
 func (dm *DockerManager) PortForward(pod *kubecontainer.Pod, port uint16, stream io.ReadWriteCloser) error {
-	return PortForward(dm.client, pod, port, stream)
-}
-
-// Temporarily export this function to share with dockershim.
-func PortForward(client DockerInterface, pod *kubecontainer.Pod, port uint16, stream io.ReadWriteCloser) error {
 	podInfraContainer := pod.FindContainerByName(PodInfraContainerName)
 	if podInfraContainer == nil {
 		return noPodInfraContainerError(pod.Name, pod.Namespace)
 	}
-	container, err := client.InspectContainer(podInfraContainer.ID.ID)
+
+	return PortForward(dm.client, podInfraContainer.ID.ID, port, stream)
+}
+
+// Temporarily export this function to share with dockershim.
+func PortForward(client DockerInterface, podInfraContainerID string, port uint16, stream io.ReadWriteCloser) error {
+	container, err := client.InspectContainer(podInfraContainerID)
 	if err != nil {
 		return err
 	}

pkg/kubelet/kuberuntime/kuberuntime_manager.go

@@ -833,17 +833,10 @@ func (m *kubeGenericRuntimeManager) GarbageCollect(gcPolicy kubecontainer.Contai
 
 // GetPodContainerID gets pod sandbox ID
 func (m *kubeGenericRuntimeManager) GetPodContainerID(pod *kubecontainer.Pod) (kubecontainer.ContainerID, error) {
-	// TODO: add a format function for kubecontainer.Pod
-	podFullName := format.Pod(&api.Pod{
-		ObjectMeta: api.ObjectMeta{
-			Name:      pod.Name,
-			Namespace: pod.Namespace,
-			UID:       pod.ID,
-		},
-	})
+	formattedPod := kubecontainer.FormatPod(pod)
 	if len(pod.Sandboxes) == 0 {
-		glog.Errorf("No sandboxes are found for pod %q", podFullName)
-		return kubecontainer.ContainerID{}, fmt.Errorf("sandboxes for pod %q not found", podFullName)
+		glog.Errorf("No sandboxes are found for pod %q", formattedPod)
+		return kubecontainer.ContainerID{}, fmt.Errorf("sandboxes for pod %q not found", formattedPod)
 	}
 
 	// return sandboxID of the first sandbox since it is the latest one
@@ -852,11 +845,17 @@ func (m *kubeGenericRuntimeManager) GetPodContainerID(pod *kubecontainer.Pod) (k
 
 // Forward the specified port from the specified pod to the stream.
 func (m *kubeGenericRuntimeManager) PortForward(pod *kubecontainer.Pod, port uint16, stream io.ReadWriteCloser) error {
+	formattedPod := kubecontainer.FormatPod(pod)
+	if len(pod.Sandboxes) == 0 {
+		glog.Errorf("No sandboxes are found for pod %q", formattedPod)
+		return fmt.Errorf("sandbox for pod %q not found", formattedPod)
+	}
+
 	// Use docker portforward directly for in-process docker integration
 	// now to unblock other tests.
 	// TODO: remove this hack after portforward is defined in CRI.
 	if ds, ok := m.runtimeService.(dockershim.DockerLegacyService); ok {
-		return ds.PortForward(pod, port, stream)
+		return ds.PortForward(pod.Sandboxes[0].ID.ID, port, stream)
 	}
 
 	return fmt.Errorf("not implemented")