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

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...)