Commit a519e8a4 authored by Dan Williams's avatar Dan Williams

kubenet: clean up networking when setup errors occur

Relying on the runtime to later call cleanup is fragile, so make sure that everything gets nicely cleaned up when setup errors occur.
parent 965492fd
...@@ -31,6 +31,7 @@ import ( ...@@ -31,6 +31,7 @@ import (
"github.com/golang/glog" "github.com/golang/glog"
"github.com/vishvananda/netlink" "github.com/vishvananda/netlink"
"github.com/vishvananda/netlink/nl" "github.com/vishvananda/netlink/nl"
"k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/apis/componentconfig" "k8s.io/kubernetes/pkg/apis/componentconfig"
kubecontainer "k8s.io/kubernetes/pkg/kubelet/container" kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
"k8s.io/kubernetes/pkg/kubelet/dockertools" "k8s.io/kubernetes/pkg/kubelet/dockertools"
...@@ -290,29 +291,7 @@ func (plugin *kubenetNetworkPlugin) Capabilities() utilsets.Int { ...@@ -290,29 +291,7 @@ func (plugin *kubenetNetworkPlugin) Capabilities() utilsets.Int {
return utilsets.NewInt(network.NET_PLUGIN_CAPABILITY_SHAPING) return utilsets.NewInt(network.NET_PLUGIN_CAPABILITY_SHAPING)
} }
func (plugin *kubenetNetworkPlugin) SetUpPod(namespace string, name string, id kubecontainer.ContainerID) error { func (plugin *kubenetNetworkPlugin) setup(namespace string, name string, id kubecontainer.ContainerID, pod *api.Pod) error {
plugin.mu.Lock()
defer plugin.mu.Unlock()
start := time.Now()
defer func() {
glog.V(4).Infof("SetUpPod took %v for %s/%s", time.Since(start), namespace, name)
}()
pod, ok := plugin.host.GetPodByName(namespace, name)
if !ok {
return fmt.Errorf("pod %q cannot be found", name)
}
ingress, egress, err := bandwidth.ExtractPodBandwidthResources(pod.Annotations)
if err != nil {
return fmt.Errorf("Error reading pod bandwidth annotations: %v", err)
}
if err := plugin.Status(); err != nil {
return fmt.Errorf("Kubenet cannot SetUpPod: %v", err)
}
// Bring up container loopback interface // Bring up container loopback interface
if _, err := plugin.addContainerToNetwork(plugin.loConfig, "lo", namespace, name, id); err != nil { if _, err := plugin.addContainerToNetwork(plugin.loConfig, "lo", namespace, name, id); err != nil {
return err return err
...@@ -348,6 +327,10 @@ func (plugin *kubenetNetworkPlugin) SetUpPod(namespace string, name string, id k ...@@ -348,6 +327,10 @@ func (plugin *kubenetNetworkPlugin) SetUpPod(namespace string, name string, id k
// initialization // initialization
shaper := plugin.shaper() shaper := plugin.shaper()
ingress, egress, err := bandwidth.ExtractPodBandwidthResources(pod.Annotations)
if err != nil {
return fmt.Errorf("Error reading pod bandwidth annotations: %v", err)
}
if egress != nil || ingress != nil { if egress != nil || ingress != nil {
if err := shaper.ReconcileCIDR(fmt.Sprintf("%s/32", ip4.String()), egress, ingress); err != nil { if err := shaper.ReconcileCIDR(fmt.Sprintf("%s/32", ip4.String()), egress, ingress); err != nil {
return fmt.Errorf("Failed to add pod to shaper: %v", err) return fmt.Errorf("Failed to add pod to shaper: %v", err)
...@@ -358,62 +341,113 @@ func (plugin *kubenetNetworkPlugin) SetUpPod(namespace string, name string, id k ...@@ -358,62 +341,113 @@ func (plugin *kubenetNetworkPlugin) SetUpPod(namespace string, name string, id k
// Open any hostports the pod's containers want // Open any hostports the pod's containers want
runningPods, err := plugin.getRunningPods() runningPods, err := plugin.getRunningPods()
if err == nil { if err != nil {
err = plugin.hostportHandler.OpenPodHostportsAndSync(pod, BridgeName, runningPods) return err
} }
if err := plugin.hostportHandler.OpenPodHostportsAndSync(pod, BridgeName, runningPods); err != nil {
// Need to SNAT outbound traffic from cluster return err
if err = plugin.ensureMasqRule(); err != nil {
glog.Errorf("Failed to ensure MASQ rule: %v", err)
} }
return err return nil
} }
func (plugin *kubenetNetworkPlugin) TearDownPod(namespace string, name string, id kubecontainer.ContainerID) error { func (plugin *kubenetNetworkPlugin) SetUpPod(namespace string, name string, id kubecontainer.ContainerID) error {
plugin.mu.Lock() plugin.mu.Lock()
defer plugin.mu.Unlock() defer plugin.mu.Unlock()
start := time.Now() start := time.Now()
defer func() { defer func() {
glog.V(4).Infof("TearDownPod took %v for %s/%s", time.Since(start), namespace, name) glog.V(4).Infof("SetUpPod took %v for %s/%s", time.Since(start), namespace, name)
}() }()
if plugin.netConfig == nil { pod, ok := plugin.host.GetPodByName(namespace, name)
return fmt.Errorf("Kubenet needs a PodCIDR to tear down pods") if !ok {
return fmt.Errorf("pod %q cannot be found", name)
} }
// no cached IP is Ok during teardown if err := plugin.Status(); err != nil {
podIP, hasIP := plugin.podIPs[id] return fmt.Errorf("Kubenet cannot SetUpPod: %v", err)
if hasIP { }
if err := plugin.setup(namespace, name, id, pod); err != nil {
// Make sure everything gets cleaned up on errors
podIP, _ := plugin.podIPs[id]
if err := plugin.teardown(namespace, name, id, podIP); err != nil {
// Not a hard error or warning
glog.V(4).Infof("Failed to clean up %s/%s after SetUpPod failure: %v", namespace, name, err)
}
return err
}
// Need to SNAT outbound traffic from cluster
if err := plugin.ensureMasqRule(); err != nil {
glog.Errorf("Failed to ensure MASQ rule: %v", err)
}
return nil
}
// Tears down as much of a pod's network as it can even if errors occur. Returns
// an aggregate error composed of all errors encountered during the teardown.
func (plugin *kubenetNetworkPlugin) teardown(namespace string, name string, id kubecontainer.ContainerID, podIP string) error {
errList := []error{}
if podIP != "" {
glog.V(5).Infof("Removing pod IP %s from shaper", podIP) glog.V(5).Infof("Removing pod IP %s from shaper", podIP)
// shaper wants /32 // shaper wants /32
if err := plugin.shaper().Reset(fmt.Sprintf("%s/32", podIP)); err != nil { if err := plugin.shaper().Reset(fmt.Sprintf("%s/32", podIP)); err != nil {
// Possible bandwidth shaping wasn't enabled for this pod anyways // Possible bandwidth shaping wasn't enabled for this pod anyways
glog.V(4).Infof("Failed to remove pod IP %s from shaper: %v", podIP, err) glog.V(4).Infof("Failed to remove pod IP %s from shaper: %v", podIP, err)
} }
delete(plugin.podIPs, id)
} }
if err := plugin.delContainerFromNetwork(plugin.netConfig, network.DefaultInterfaceName, namespace, name, id); err != nil { if err := plugin.delContainerFromNetwork(plugin.netConfig, network.DefaultInterfaceName, namespace, name, id); err != nil {
// This is to prevent returning error when TearDownPod is called twice on the same pod. This helps to reduce event pollution. // This is to prevent returning error when TearDownPod is called twice on the same pod. This helps to reduce event pollution.
if !hasIP { if podIP != "" {
glog.Warningf("Failed to delete container from kubenet: %v", err) glog.Warningf("Failed to delete container from kubenet: %v", err)
return nil } else {
errList = append(errList, err)
} }
return err
} }
delete(plugin.podIPs, id)
runningPods, err := plugin.getRunningPods() runningPods, err := plugin.getRunningPods()
if err == nil { if err == nil {
err = plugin.hostportHandler.SyncHostports(BridgeName, runningPods) err = plugin.hostportHandler.SyncHostports(BridgeName, runningPods)
} }
if err != nil {
errList = append(errList, err)
}
return utilerrors.NewAggregate(errList)
}
func (plugin *kubenetNetworkPlugin) TearDownPod(namespace string, name string, id kubecontainer.ContainerID) error {
plugin.mu.Lock()
defer plugin.mu.Unlock()
start := time.Now()
defer func() {
glog.V(4).Infof("TearDownPod took %v for %s/%s", time.Since(start), namespace, name)
}()
if plugin.netConfig == nil {
return fmt.Errorf("Kubenet needs a PodCIDR to tear down pods")
}
// no cached IP is Ok during teardown
podIP, _ := plugin.podIPs[id]
if err := plugin.teardown(namespace, name, id, podIP); err != nil {
return err
}
// Need to SNAT outbound traffic from cluster // Need to SNAT outbound traffic from cluster
if err := plugin.ensureMasqRule(); err != nil { if err := plugin.ensureMasqRule(); err != nil {
glog.Errorf("Failed to ensure MASQ rule: %v", err) glog.Errorf("Failed to ensure MASQ rule: %v", err)
} }
return err return nil
} }
// TODO: Use the addToNetwork function to obtain the IP of the Pod. That will assume idempotent ADD call to the plugin. // TODO: Use the addToNetwork function to obtain the IP of the Pod. That will assume idempotent ADD call to the plugin.
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment