Merge pull request openshift#18166 from dcbw/ovs-cleanup-when-sandbox…

…-is-gone

Automatic merge from submit-queue.

sdn: try cleaning up OVS rules even if sandbox is gone

If a sandbox is deleted underneath kubernetes its netns will
be gone and its veth interface will be deleted by the kernel.
That means we can't inspect the veth for its IP address and
other details, which are used to remove OVS flows for the
interface.

But we've already got code to find out the IP using the
sandbox ID which kubelet passes down to us.  Let's use
that code to at least delete the stale OVS flows.

Fixes: https://bugzilla.redhat.com/show_bug.cgi?id=1518684

@danwinship @openshift/networking @knobunc

pkg/network/node/ovscontroller.go

@@ -34,7 +34,7 @@ const (
 	Vxlan0 = "vxlan0"
 
 	// rule versioning; increment each time flow rules change
-	ruleVersion = 5
+	ruleVersion = 6
 
 	ruleVersionTable = 253
 )
@@ -223,8 +223,8 @@ func (oc *ovsController) NewTransaction() ovs.Transaction {
 	return oc.ovs.NewTransaction()
 }
 
-func (oc *ovsController) ensureOvsPort(hostVeth string) (int, error) {
-	return oc.ovs.AddPort(hostVeth, -1)
+func (oc *ovsController) ensureOvsPort(hostVeth, sandboxID string) (int, error) {
+	return oc.ovs.AddPort(hostVeth, -1, "external-ids=sandbox="+sandboxID)
 }
 
 func (oc *ovsController) setupPodFlows(ofport int, podIP, podMAC, note string, vnid uint32) error {
@@ -278,33 +278,54 @@ func (oc *ovsController) SetUpPod(hostVeth, podIP, podMAC, sandboxID string, vni
 	if err != nil {
 		return -1, err
 	}
-	ofport, err := oc.ensureOvsPort(hostVeth)
+	ofport, err := oc.ensureOvsPort(hostVeth, sandboxID)
 	if err != nil {
 		return -1, err
 	}
 	return ofport, oc.setupPodFlows(ofport, podIP, podMAC, note, vnid)
 }
 
-func (oc *ovsController) SetPodBandwidth(hostVeth string, ingressBPS, egressBPS int64) error {
-	// note pod ingress == OVS egress and vice versa
+// Returned list can also be used for port names
+func (oc *ovsController) getInterfacesForSandbox(sandboxID string) ([]string, error) {
+	return oc.ovs.Find("interface", "name", "external-ids:sandbox="+sandboxID)
+}
 
-	qos, err := oc.ovs.Get("port", hostVeth, "qos")
+func (oc *ovsController) ClearPodBandwidth(portList []string, sandboxID string) error {
+	// Clear the QoS for any ports of this sandbox
+	for _, port := range portList {
+		if err := oc.ovs.Clear("port", port, "qos"); err != nil {
+			return err
+		}
+	}
+
+	// Now that the QoS is unused remove it
+	qosList, err := oc.ovs.Find("qos", "_uuid", "external-ids:sandbox="+sandboxID)
 	if err != nil {
 		return err
 	}
-	if qos != "[]" {
-		err = oc.ovs.Clear("port", hostVeth, "qos")
-		if err != nil {
-			return err
-		}
-		err = oc.ovs.Destroy("qos", qos)
-		if err != nil {
+	for _, qos := range qosList {
+		if err := oc.ovs.Destroy("qos", qos); err != nil {
 			return err
 		}
 	}
 
+	return nil
+}
+
+func (oc *ovsController) SetPodBandwidth(hostVeth, sandboxID string, ingressBPS, egressBPS int64) error {
+	// note pod ingress == OVS egress and vice versa
+
+	ports, err := oc.getInterfacesForSandbox(sandboxID)
+	if err != nil {
+		return err
+	}
+
+	if err := oc.ClearPodBandwidth(ports, sandboxID); err != nil {
+		return err
+	}
+
 	if ingressBPS > 0 {
-		qos, err := oc.ovs.Create("qos", "type=linux-htb", fmt.Sprintf("other-config:max-rate=%d", ingressBPS))
+		qos, err := oc.ovs.Create("qos", "type=linux-htb", fmt.Sprintf("other-config:max-rate=%d", ingressBPS), "external-ids=sandbox="+sandboxID)
 		if err != nil {
 			return err
 		}
@@ -382,22 +403,37 @@ func (oc *ovsController) UpdatePod(sandboxID string, vnid uint32) error {
 	return oc.setupPodFlows(ofport, podIP, podMAC, note, vnid)
 }
 
-func (oc *ovsController) TearDownPod(hostVeth, podIP, sandboxID string) error {
+func (oc *ovsController) TearDownPod(podIP, sandboxID string) error {
 	if podIP == "" {
-		_, ip, _, _, err := oc.getPodDetailsBySandboxID(sandboxID)
+		var err error
+		_, podIP, _, _, err = oc.getPodDetailsBySandboxID(sandboxID)
 		if err != nil {
 			// OVS flows related to sandboxID not found
 			// Nothing needs to be done in that case
 			return nil
 		}
-		podIP = ip
 	}
 
 	if err := oc.cleanupPodFlows(podIP); err != nil {
 		return err
 	}
-	_ = oc.SetPodBandwidth(hostVeth, -1, -1)
-	return oc.ovs.DeletePort(hostVeth)
+
+	ports, err := oc.getInterfacesForSandbox(sandboxID)
+	if err != nil {
+		return err
+	}
+
+	if err := oc.ClearPodBandwidth(ports, sandboxID); err != nil {
+		return err
+	}
+
+	for _, port := range ports {
+		if err := oc.ovs.DeletePort(port); err != nil {
+			return err
+		}
+	}
+
+	return nil
 }
 
 func policyNames(policies []networkapi.EgressNetworkPolicy) string {

pkg/network/node/ovscontroller_test.go

@@ -303,7 +303,7 @@ func TestOVSPod(t *testing.T) {
 	}
 
 	// Delete
-	err = oc.TearDownPod("veth1", "10.128.0.2", sandboxID)
+	err = oc.TearDownPod("10.128.0.2", sandboxID)
 	if err != nil {
 		t.Fatalf("Unexpected error deleting pod rules: %v", err)
 	}
@@ -903,12 +903,17 @@ func TestAlreadySetUp(t *testing.T) {
 	}{
 		{
 			// Good note
-			flow:    "cookie=0x0, duration=4.796s, table=253, n_packets=0, n_bytes=0, actions=note:00.05.00.00.00.00",
+			flow:    fmt.Sprintf("cookie=0x0, duration=4.796s, table=253, n_packets=0, n_bytes=0, actions=note:00.%02x.00.00.00.00", ruleVersion),
 			success: true,
 		},
+		{
+			// Wrong version
+			flow:    fmt.Sprintf("cookie=0x0, duration=4.796s, table=253, n_packets=0, n_bytes=0, actions=note:00.%02x.00.00.00.00", ruleVersion-1),
+			success: false,
+		},
 		{
 			// Wrong table
-			flow:    "cookie=0x0, duration=4.796s, table=10, n_packets=0, n_bytes=0, actions=note:00.05.00.00.00.00",
+			flow:    fmt.Sprintf("cookie=0x0, duration=4.796s, table=10, n_packets=0, n_bytes=0, actions=note:00.%02x.00.00.00.00", ruleVersion),
 			success: false,
 		},
 		{

pkg/network/node/pod.go

@@ -503,7 +503,7 @@ func (m *podManager) ipamDel(id string) error {
 	return nil
 }
 
-func setupPodBandwidth(ovs *ovsController, pod *kapi.Pod, hostVeth string) error {
+func setupPodBandwidth(ovs *ovsController, pod *kapi.Pod, hostVeth, sandboxID string) error {
 	ingressVal, egressVal, err := kbandwidth.ExtractPodBandwidthResources(pod.Annotations)
 	if err != nil {
 		return fmt.Errorf("failed to parse pod bandwidth: %v", err)
@@ -527,7 +527,7 @@ func setupPodBandwidth(ovs *ovsController, pod *kapi.Pod, hostVeth string) error
 		egressBPS = egressVal.Value()
 	}
 
-	return ovs.SetPodBandwidth(hostVeth, ingressBPS, egressBPS)
+	return ovs.SetPodBandwidth(hostVeth, sandboxID, ingressBPS, egressBPS)
 }
 
 func vnidToString(vnid uint32) string {
@@ -652,7 +652,7 @@ func (m *podManager) setup(req *cniserver.PodRequest) (cnitypes.Result, *running
 	if err != nil {
 		return nil, nil, err
 	}
-	if err := setupPodBandwidth(m.ovs, pod, hostVethName); err != nil {
+	if err := setupPodBandwidth(m.ovs, pod, hostVethName, req.SandboxID); err != nil {
 		return nil, nil, err
 	}
 
@@ -678,24 +678,21 @@ func (m *podManager) update(req *cniserver.PodRequest) (uint32, error) {
 func (m *podManager) teardown(req *cniserver.PodRequest) error {
 	defer PodOperationsLatency.WithLabelValues(PodOperationTeardown).Observe(sinceInMicroseconds(time.Now()))
 
-	netnsValid := true
+	var podIP string
+	errList := []error{}
+
 	if err := ns.IsNSorErr(req.Netns); err != nil {
-		if _, ok := err.(ns.NSPathNotExistErr); ok {
-			glog.V(3).Infof("teardown called on already-destroyed pod %s/%s; only cleaning up IPAM", req.PodNamespace, req.PodName)
-			netnsValid = false
+		if _, ok := err.(ns.NSPathNotExistErr); !ok {
+			// Namespace still exists, get pod IP from the veth
+			_, _, podIP, err = getVethInfo(req.Netns, podInterfaceName)
+			if err != nil {
+				errList = append(errList, err)
+			}
 		}
 	}
 
-	errList := []error{}
-	if netnsValid {
-		hostVethName, _, podIP, err := getVethInfo(req.Netns, podInterfaceName)
-		if err != nil {
-			return err
-		}
-
-		if err := m.ovs.TearDownPod(hostVethName, podIP, req.SandboxID); err != nil {
-			errList = append(errList, err)
-		}
+	if err := m.ovs.TearDownPod(podIP, req.SandboxID); err != nil {
+		errList = append(errList, err)
 	}
 
 	if err := m.ipamDel(req.SandboxID); err != nil {

pkg/util/ovs/fake_ovs.go

@@ -113,6 +113,10 @@ func (fake *ovsFake) Set(table, record string, values ...string) error {
 	return nil
 }
 
+func (fake *ovsFake) Find(table, column, condition string) ([]string, error) {
+	return make([]string, 0), nil
+}
+
 func (fake *ovsFake) Clear(table, record string, columns ...string) error {
 	return nil
 }

pkg/util/ovs/ovs.go

@@ -62,6 +62,10 @@ type Interface interface {
 	// the value is already unset
 	Clear(table, record string, columns ...string) error
 
+	// Find finds records in the OVS database that match the given condition.
+	// It returns the value of the given column of matching records.
+	Find(table, column, condition string) ([]string, error)
+
 	// DumpFlows dumps the flow table for the bridge and returns it as an array of
 	// strings, one per flow. If flow is not "" then it describes the flows to dump.
 	DumpFlows(flow string, args ...interface{}) ([]string, error)
@@ -244,6 +248,15 @@ func (ovsif *ovsExec) Set(table, record string, values ...string) error {
 	return err
 }
 
+// Returns the given column of records that match the condition
+func (ovsif *ovsExec) Find(table, column, condition string) ([]string, error) {
+	output, err := ovsif.exec(OVS_VSCTL, "--no-heading", "--data=bare", "--columns="+column, "find", table, condition)
+	if err != nil {
+		return nil, err
+	}
+	return strings.Fields(output), nil
+}
+
 func (ovsif *ovsExec) Clear(table, record string, columns ...string) error {
 	args := append([]string{"--if-exists", "clear", table, record}, columns...)
 	_, err := ovsif.exec(OVS_VSCTL, args...)