sdn: kill containers that fail to update on node restart

With the move to remote runtimes, we can no longer get the pod's
network namespace from kubelet (since we cannot insert ourselves
into the remote runtime's plugin list and intercept network plugin
calls).  As kubelet does not call network plugins in any way on
startup if a container is already running, we have no way to ensure
the container is using the correct NetNamespace (as it may have
changed while openshift-node was down) at startup, unless we encode
the required information into OVS flows.

But if OVS was restarted around the same time OpenShift was,
those flows are lost, and we have no information with which to
recover the pod's networking on node startup.  In this case, kill
the infra container underneath kubelet so that it will be restarted
and we can set its network up again.

NOTE: this is somewhat hacky and will not work with other remote
runtimes like CRI-O, but OpenShift 3.6 hardcodes dockershim so this
isn't a problem yet.  The "correct" solution is to either checkpoint
our network configuration at container setup time and recover that
ourselves, or to add a GET/STATUS call to CNI and make Kubelet call
that operation on startup when recovering running containers.

Author: dcbw

pkg/sdn/plugin/node.go

@@ -20,7 +20,7 @@ import (
 	"github.com/openshift/origin/pkg/util/netutils"
 	"github.com/openshift/origin/pkg/util/ovs"
 
-	docker "github.com/fsouza/go-dockerclient"
+	dockertypes "github.com/docker/engine-api/types"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
@@ -32,7 +32,10 @@ import (
 	kclientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
 	kinternalinformers "k8s.io/kubernetes/pkg/client/informers/informers_generated/internalversion"
 	knetwork "k8s.io/kubernetes/pkg/kubelet/network"
+	kcontainer "k8s.io/kubernetes/pkg/kubelet/container"
 	kexec "k8s.io/kubernetes/pkg/util/exec"
+	"k8s.io/kubernetes/pkg/kubelet/leaky"
+	"k8s.io/kubernetes/pkg/kubelet/dockertools"
 )
 
 type osdnPolicy interface {
@@ -187,29 +190,123 @@ func (node *OsdnNode) dockerPreCNICleanup() error {
 	}
 
 	// Wait until docker has restarted since kubelet will exit it docker isn't running
-	dockerClient, err := docker.NewClientFromEnv()
-	if err != nil {
-		return fmt.Errorf("failed to get docker client: %v", err)
+	if _, err := ensureDockerClient(); err != nil {
+		return err
 	}
-	err = kwait.ExponentialBackoff(
+
+	log.Infof("Cleaned up left-over openshift-sdn docker bridge and interfaces")
+
+	return nil
+}
+
+func ensureDockerClient() (dockertools.DockerInterface, error) {
+	endpoint := os.Getenv("DOCKER_HOST")
+	if endpoint == "" {
+		endpoint = "unix:///var/run/docker.sock"
+	}
+	dockerClient := dockertools.ConnectToDockerOrDie(endpoint, time.Minute, time.Minute)
+
+	// Wait until docker has restarted since kubelet will exit it docker isn't running
+	if err := kwait.ExponentialBackoff(
 		kwait.Backoff{
 			Duration: 100 * time.Millisecond,
 			Factor:   1.2,
 			Steps:    6,
 		},
 		func() (bool, error) {
-			if err := dockerClient.Ping(); err != nil {
+			if _, err := dockerClient.Version(); err != nil {
 				// wait longer
 				return false, nil
 			}
 			return true, nil
-		})
+	}); err != nil {
+		return nil, fmt.Errorf("failed to connect to docker after SDN cleanup restart: %v", err)
+	}
+	return dockerClient, nil
+}
+
+func formatPod(pod *kapi.Pod) string {
+	return fmt.Sprintf("%s/%s", pod.Namespace, pod.Name)
+}
+
+func dockerSandboxNameToInfraPodNamePrefix(name string) (string, error) {
+	// Docker adds a "/" prefix to names. so trim it.
+	name = strings.TrimPrefix(name, "/")
+
+	parts := strings.Split(name, "_")
+	// Tolerate the random suffix.
+	// TODO(random-liu): Remove 7 field case when docker 1.11 is deprecated.
+	if len(parts) != 6 && len(parts) != 7 {
+		return "", fmt.Errorf("failed to parse the sandbox name: %q", name)
+	}
+	if parts[0] != "k8s" {
+		return "", fmt.Errorf("container is not managed by kubernetes: %q", name)
+	}
+
+	// Return /k8s_POD_name_namespace_uid
+	return fmt.Sprintf("/k8s_%s_%s_%s_%s", leaky.PodInfraContainerName, parts[2], parts[3], parts[4]), nil
+}
+
+func (node *OsdnNode) killInfraContainerForPod(docker dockertools.DockerInterface, containers []dockertypes.Container, cid kcontainer.ContainerID) error {
+	if cid.Type != "docker" {
+		return fmt.Errorf("unhandled runtime %q")
+	}
+
+	var err error
+	var infraPrefix string
+	for _, c := range containers {
+		if c.ID == cid.ID {
+			infraPrefix, err = dockerSandboxNameToInfraPodNamePrefix(c.Names[0])
+			if err != nil {
+				return fmt.Errorf("unparsable container ID %q", c.ID)
+			}
+			break
+		}
+	}
+	if infraPrefix == "" {
+		return fmt.Errorf("failed to generate infra container prefix from %q", cid.ID)
+	}
+	// Find and kill the infra container
+	for _, c := range containers {
+		if strings.HasPrefix(c.Names[0], infraPrefix) {
+			if err := docker.StopContainer(c.ID, 10); err != nil {
+				log.Warningf("failed to stop infra container %q", c.ID)
+			}
+		}
+	}
+
+	return nil
+}
+
+func (node *OsdnNode) killUpdateFailedPods(pods []kapi.Pod) error {
+	docker, err := ensureDockerClient()
 	if err != nil {
-		return fmt.Errorf("failed to connect to docker after SDN cleanup restart: %v", err)
+		return fmt.Errorf("failed to get docker client: %v", err)
 	}
 
-	log.Infof("Cleaned up left-over openshift-sdn docker bridge and interfaces")
+	containers, err := docker.ListContainers(dockertypes.ContainerListOptions{All: true})
+	if err != nil {
+		return fmt.Errorf("failed to list docker containers: %v", err)
+	}
 
+	for _, pod := range pods {
+		// Find the first ready container in the pod and use it to find the infra container
+		var cid kcontainer.ContainerID
+		for i := range pod.Status.ContainerStatuses {
+			if pod.Status.ContainerStatuses[i].State.Running != nil && pod.Status.ContainerStatuses[i].ContainerID != "" {
+				cid = kcontainer.ParseContainerID(pod.Status.ContainerStatuses[i].ContainerID)
+				break
+			}
+		}
+		if cid.IsEmpty() {
+			continue
+		}
+		log.V(5).Infof("Killing pod %q sandbox on restart", formatPod(&pod))
+		if err := node.killInfraContainerForPod(docker, containers, cid); err != nil {
+			log.Warningf("Failed to kill pod %q sandbox: %v", formatPod(&pod), err)
+			continue
+		}
+	}
 	return nil
 }
 
@@ -259,24 +356,33 @@ func (node *OsdnNode) Start() error {
 		return err
 	}
 
+	var podsToKill []kapi.Pod
 	if networkChanged {
 		var pods []kapi.Pod
 		pods, err = node.GetLocalPods(metav1.NamespaceAll)
 		if err != nil {
 			return err
 		}
 		for _, p := range pods {
-			err = node.UpdatePod(p)
-			if err != nil {
-				log.Warningf("Could not update pod %q: %s", p.Name, err)
+			// Ignore HostNetwork pods since they don't go through OVS
+			if p.Spec.SecurityContext != nil && p.Spec.SecurityContext.HostNetwork {
 				continue
 			}
-			if vnid, err := node.policy.GetVNID(p.Namespace); err == nil {
+			if err := node.UpdatePod(p); err != nil {
+				log.Warningf("will restart pod '%s/%s' due to update failure on restart: %s", p.Namespace, p.Name, err)
+				podsToKill = append(podsToKill, p)
+			} else if vnid, err := node.policy.GetVNID(p.Namespace); err == nil {
 				node.policy.EnsureVNIDRules(vnid)
 			}
 		}
 	}
 
+	// Kill pods we couldn't recover; they will get restarted and then
+	// we'll be able to set them up correctly
+	if err := node.killUpdateFailedPods(podsToKill); err != nil {
+		log.Warningf("Failed to restart pods that failed to update on restart: %v", err)
+	}
+
 	go kwait.Forever(node.policy.SyncVNIDRules, time.Hour)
 
 	log.V(5).Infof("openshift-sdn network plugin ready")