Changed ovs.Transaction from pseudo to real atomic transaction

- Leverages ovs.bundle() to perform atomic transactions
- Now ovs.Transaction interface has AddFlow(), DeleteFlows() and Commit() methods
  General usage:
  otx := ovs.NewTransaction()
  otx.AddFlow(flow1) // No execution, only caches flow context
  ...
  otx.DeleteFlows(flowN) // No execution, only caches flow context
  ...
  err := otx.Commit() // Executes all cached flows as single atomic transaction
- With this change, most of the operations in ovs controller like setup,
  addHostSubnet, addService, etc. has only one ovs bundle call. So there
  won't be partial commited changes and it reduces the downtime during
  watch resync events.

Author: Ravi Sankar Penta

pkg/network/node/multitenant.go

@@ -49,7 +49,7 @@ func (mp *multiTenantPlugin) Start(node *OsdnNode) error {
 	otx := node.oc.NewTransaction()
 	otx.AddFlow("table=80, priority=200, reg0=0, actions=output:NXM_NX_REG2[]")
 	otx.AddFlow("table=80, priority=200, reg1=0, actions=output:NXM_NX_REG2[]")
-	if err := otx.EndTransaction(); err != nil {
+	if err := otx.Commit(); err != nil {
 		return err
 	}
 
@@ -141,7 +141,7 @@ func (mp *multiTenantPlugin) EnsureVNIDRules(vnid uint32) {
 
 	otx := mp.node.oc.NewTransaction()
 	otx.AddFlow("table=80, priority=100, reg0=%d, reg1=%d, actions=output:NXM_NX_REG2[]", vnid, vnid)
-	if err := otx.EndTransaction(); err != nil {
+	if err := otx.Commit(); err != nil {
 		utilruntime.HandleError(fmt.Errorf("Error adding OVS flow for VNID: %v", err))
 	}
 }
@@ -158,7 +158,7 @@ func (mp *multiTenantPlugin) SyncVNIDRules() {
 		mp.vnidInUse[uint32(vnid)] = false
 		otx.DeleteFlows("table=80, reg1=%d", vnid)
 	}
-	if err := otx.EndTransaction(); err != nil {
+	if err := otx.Commit(); err != nil {
 		utilruntime.HandleError(fmt.Errorf("Error deleting syncing OVS VNID rules: %v", err))
 	}
 }

pkg/network/node/networkpolicy.go

@@ -91,7 +91,7 @@ func (np *networkPolicyPlugin) Start(node *OsdnNode) error {
 		otx.AddFlow("table=21, priority=200, ip, nw_dst=%s, actions=ct(commit,table=30)", cn.ClusterCIDR.String())
 	}
 	otx.AddFlow("table=80, priority=200, ip, ct_state=+rpl, actions=output:NXM_NX_REG2[]")
-	if err := otx.EndTransaction(); err != nil {
+	if err := otx.Commit(); err != nil {
 		return err
 	}
 
@@ -229,7 +229,7 @@ func (np *networkPolicyPlugin) syncNamespace(npns *npNamespace) {
 			otx.AddFlow("table=80, priority=50, reg1=%d, actions=output:NXM_NX_REG2[]", npns.vnid)
 		}
 	}
-	if err := otx.EndTransaction(); err != nil {
+	if err := otx.Commit(); err != nil {
 		utilruntime.HandleError(fmt.Errorf("Error syncing OVS flows for VNID: %v", err))
 	}
 }

pkg/network/node/ovscontroller.go

@@ -217,12 +217,7 @@ func (oc *ovsController) SetupOVS(clusterNetworkCIDR []string, serviceNetworkCID
 	// Table 253: rule version note
 	otx.AddFlow("table=%d, actions=note:%s", ruleVersionTable, oc.getVersionNote())
 
-	err = otx.EndTransaction()
-	if err != nil {
-		return err
-	}
-
-	return nil
+	return otx.Commit()
 }
 
 func (oc *ovsController) NewTransaction() ovs.Transaction {
@@ -255,7 +250,7 @@ func (oc *ovsController) setupPodFlows(ofport int, podIP net.IP, vnid uint32) er
 	// IP traffic to container
 	otx.AddFlow("table=70, priority=100, ip, nw_dst=%s, actions=load:%d->NXM_NX_REG1[], load:%d->NXM_NX_REG2[], goto_table:80", ipstr, vnid, ofport)
 
-	return otx.EndTransaction()
+	return otx.Commit()
 }
 
 func (oc *ovsController) cleanupPodFlows(podIP net.IP) error {
@@ -266,7 +261,7 @@ func (oc *ovsController) cleanupPodFlows(podIP net.IP) error {
 	otx.DeleteFlows("ip, nw_src=%s", ipstr)
 	otx.DeleteFlows("arp, nw_dst=%s", ipstr)
 	otx.DeleteFlows("arp, nw_src=%s", ipstr)
-	return otx.EndTransaction()
+	return otx.Commit()
 }
 
 func (oc *ovsController) SetUpPod(sandboxID, hostVeth string, podIP net.IP, vnid uint32) (int, error) {
@@ -501,7 +496,7 @@ func (oc *ovsController) UpdateEgressNetworkPolicyRules(policies []networkapi.Eg
 		otx.DeleteFlows("table=101, reg0=%d, cookie=1/1", vnid)
 	}
 
-	if txErr := otx.EndTransaction(); txErr != nil {
+	if txErr := otx.Commit(); txErr != nil {
 		errs = append(errs, txErr)
 	}
 
@@ -526,7 +521,7 @@ func (oc *ovsController) AddHostSubnetRules(subnet *networkapi.HostSubnet) error
 		otx.AddFlow("table=90, priority=100, cookie=0x%08x, ip, nw_dst=%s, actions=move:NXM_NX_REG0[]->NXM_NX_TUN_ID[0..31],set_field:%s->tun_dst,output:1", cookie, subnet.Subnet, subnet.HostIP)
 	}
 
-	return otx.EndTransaction()
+	return otx.Commit()
 }
 
 func (oc *ovsController) DeleteHostSubnetRules(subnet *networkapi.HostSubnet) error {
@@ -536,7 +531,7 @@ func (oc *ovsController) DeleteHostSubnetRules(subnet *networkapi.HostSubnet) er
 	otx.DeleteFlows("table=10, cookie=0x%08x/0xffffffff, tun_src=%s", cookie, subnet.HostIP)
 	otx.DeleteFlows("table=50, cookie=0x%08x/0xffffffff, arp, nw_dst=%s", cookie, subnet.Subnet)
 	otx.DeleteFlows("table=90, cookie=0x%08x/0xffffffff, ip, nw_dst=%s", cookie, subnet.Subnet)
-	return otx.EndTransaction()
+	return otx.Commit()
 }
 
 func (oc *ovsController) AddServiceRules(service *kapi.Service, netID uint32) error {
@@ -555,13 +550,13 @@ func (oc *ovsController) AddServiceRules(service *kapi.Service, netID uint32) er
 		otx.AddFlow(baseRule + action)
 	}
 
-	return otx.EndTransaction()
+	return otx.Commit()
 }
 
 func (oc *ovsController) DeleteServiceRules(service *kapi.Service) error {
 	otx := oc.ovs.NewTransaction()
 	otx.DeleteFlows(generateBaseServiceRule(service.Spec.ClusterIP))
-	return otx.EndTransaction()
+	return otx.Commit()
 }
 
 func generateBaseServiceRule(IP string) string {
@@ -601,7 +596,7 @@ func (oc *ovsController) UpdateLocalMulticastFlows(vnid uint32, enabled bool, of
 		otx.DeleteFlows("table=120, reg0=%d", vnid)
 	}
 
-	return otx.EndTransaction()
+	return otx.Commit()
 }
 
 func (oc *ovsController) UpdateVXLANMulticastFlows(remoteIPs []string) error {
@@ -618,7 +613,7 @@ func (oc *ovsController) UpdateVXLANMulticastFlows(remoteIPs []string) error {
 		otx.AddFlow("table=111, priority=100, actions=goto_table:120")
 	}
 
-	return otx.EndTransaction()
+	return otx.Commit()
 }
 
 // FindUnusedVNIDs returns a list of VNIDs for which there are table 80 "check" rules,
@@ -702,14 +697,14 @@ func (oc *ovsController) ensureTunMAC() error {
 func (oc *ovsController) SetNamespaceEgressNormal(vnid uint32) error {
 	otx := oc.ovs.NewTransaction()
 	otx.DeleteFlows("table=100, reg0=%d", vnid)
-	return otx.EndTransaction()
+	return otx.Commit()
 }
 
 func (oc *ovsController) SetNamespaceEgressDropped(vnid uint32) error {
 	otx := oc.ovs.NewTransaction()
 	otx.DeleteFlows("table=100, reg0=%d", vnid)
 	otx.AddFlow("table=100, priority=100, reg0=%d, actions=drop", vnid)
-	return otx.EndTransaction()
+	return otx.Commit()
 }
 
 func (oc *ovsController) SetNamespaceEgressViaEgressIP(vnid uint32, nodeIP, mark string) error {
@@ -726,5 +721,5 @@ func (oc *ovsController) SetNamespaceEgressViaEgressIP(vnid uint32, nodeIP, mark
 	} else {
 		otx.AddFlow("table=100, priority=100, reg0=%d, ip, actions=move:NXM_NX_REG0[]->NXM_NX_TUN_ID[0..31],set_field:%s->tun_dst,output:1", vnid, nodeIP)
 	}
-	return otx.EndTransaction()
+	return otx.Commit()
 }

pkg/network/node/ovscontroller_test.go

@@ -805,7 +805,7 @@ func TestAlreadySetUp(t *testing.T) {
 
 		otx := ovsif.NewTransaction()
 		otx.AddFlow(tc.flow)
-		if err := otx.EndTransaction(); err != nil {
+		if err := otx.Commit(); err != nil {
 			t.Fatalf("(%d) unexpected error from AddFlow: %v", i, err)
 		}
 		if success := oc.AlreadySetUp(); success != tc.success {
@@ -919,7 +919,7 @@ func TestSyncVNIDRules(t *testing.T) {
 		for _, flow := range tc.flows {
 			otx.AddFlow(flow)
 		}
-		if err := otx.EndTransaction(); err != nil {
+		if err := otx.Commit(); err != nil {
 			t.Fatalf("(%d) unexpected error from AddFlow: %v", i, err)
 		}
 

pkg/network/node/singletenant.go

@@ -24,7 +24,7 @@ func (sp *singleTenantPlugin) SupportsVNIDs() bool {
 func (sp *singleTenantPlugin) Start(node *OsdnNode) error {
 	otx := node.oc.NewTransaction()
 	otx.AddFlow("table=80, priority=200, actions=output:NXM_NX_REG2[]")
-	return otx.EndTransaction()
+	return otx.Commit()
 }
 
 func (sp *singleTenantPlugin) AddNetNamespace(netns *networkapi.NetNamespace) {

pkg/util/ovs/fake_ovs_test.go

@@ -165,7 +165,7 @@ func TestFakeDumpFlows(t *testing.T) {
 	otx.AddFlow("table=30, priority=300, ip, nw_dst=%s, actions=output:2", localSubnetGateway)
 	otx.AddFlow("table=35, priority=300, ip, nw_dst=%s, actions=ct(commit,exec(set_field:1->ct_mark),table=70)", localSubnetGateway)
 
-	err = otx.EndTransaction()
+	err = otx.Commit()
 	if err != nil {
 		t.Fatalf("unexpected error from AddFlow: %v", err)
 	}
@@ -280,7 +280,7 @@ func TestFlowMatchesMasked(t *testing.T) {
 	otx.AddFlow("table=100, priority=200, reg0=2, actions=two")
 	otx.AddFlow("table=100, priority=300, reg0=3, cookie=1, actions=three")
 	otx.AddFlow("table=100, priority=400, reg0=4, cookie=0xe, actions=four")
-	err = otx.EndTransaction()
+	err = otx.Commit()
 	if err != nil {
 		t.Fatalf("unexpected error from AddFlow: %v", err)
 	}
@@ -294,7 +294,7 @@ func TestFlowMatchesMasked(t *testing.T) {
 
 	otx = ovsif.NewTransaction()
 	otx.DeleteFlows("table=100, cookie=0/0xFFFF")
-	err = otx.EndTransaction()
+	err = otx.Commit()
 	if err != nil {
 		t.Fatalf("unexpected error from AddFlow: %v", err)
 	}
@@ -308,7 +308,7 @@ func TestFlowMatchesMasked(t *testing.T) {
 
 	otx = ovsif.NewTransaction()
 	otx.DeleteFlows("table=100, cookie=2/2")
-	err = otx.EndTransaction()
+	err = otx.Commit()
 	if err != nil {
 		t.Fatalf("unexpected error from AddFlow: %v", err)
 	}

pkg/util/ovs/ovs.go

@@ -73,26 +73,25 @@ type Interface interface {
 	// strings, one per flow. If flow is not "" then it describes the flows to dump.
 	DumpFlows(flow string, args ...interface{}) ([]string, error)
 
-	// NewTransaction begins a new OVS transaction. If an error occurs at
-	// any step in the transaction, it will be recorded until
-	// EndTransaction(), and any further calls on the transaction will be
-	// ignored.
+	// NewTransaction begins a new OVS transaction.
 	NewTransaction() Transaction
 }
 
 // Transaction manages a single set of OVS flow modifications
 type Transaction interface {
-	// AddFlow adds a flow to the bridge. The arguments are passed to fmt.Sprintf().
+	// AddFlow prepares adding a flow to the bridge.
+	// Given flow is cached but not executed at this time.
+	// The arguments are passed to fmt.Sprintf().
 	AddFlow(flow string, args ...interface{})
 
-	// DeleteFlows deletes all matching flows from the bridge. The arguments are
-	// passed to fmt.Sprintf().
+	// DeleteFlows prepares deleting all matching flows from the bridge.
+	// Given flow is cached but not executed at this time.
+	// The arguments are passed to fmt.Sprintf().
 	DeleteFlows(flow string, args ...interface{})
 
-	// EndTransaction ends an OVS transaction and returns any error that occurred
-	// during the transaction. You should not use the transaction again after
-	// calling this function.
-	EndTransaction() error
+	// Commit executes all cached flows as a single atomic transaction and
+	// returns any error that occurred during the transaction.
+	Commit() error
 }
 
 const (
@@ -297,43 +296,6 @@ func (ovsif *ovsExec) Clear(table, record string, columns ...string) error {
 	return err
 }
 
-type ovsExecTx struct {
-	ovsif *ovsExec
-	err   error
-}
-
-func (tx *ovsExecTx) exec(cmd string, args ...string) (string, error) {
-	out := ""
-	if tx.err == nil {
-		out, tx.err = tx.ovsif.exec(cmd, args...)
-	}
-	return out, tx.err
-}
-
-func (ovsif *ovsExec) NewTransaction() Transaction {
-	return &ovsExecTx{ovsif: ovsif}
-}
-
-func (tx *ovsExecTx) AddFlow(flow string, args ...interface{}) {
-	if len(args) > 0 {
-		flow = fmt.Sprintf(flow, args...)
-	}
-	tx.exec(OVS_OFCTL, "add-flow", tx.ovsif.bridge, flow)
-}
-
-func (tx *ovsExecTx) DeleteFlows(flow string, args ...interface{}) {
-	if len(args) > 0 {
-		flow = fmt.Sprintf(flow, args...)
-	}
-	tx.exec(OVS_OFCTL, "del-flows", tx.ovsif.bridge, flow)
-}
-
-func (tx *ovsExecTx) EndTransaction() error {
-	err := tx.err
-	tx.err = nil
-	return err
-}
-
 func (ovsif *ovsExec) DumpFlows(flow string, args ...interface{}) ([]string, error) {
 	if len(args) > 0 {
 		flow = fmt.Sprintf(flow, args...)
@@ -353,6 +315,10 @@ func (ovsif *ovsExec) DumpFlows(flow string, args ...interface{}) ([]string, err
 	return flows, nil
 }
 
+func (ovsif *ovsExec) NewTransaction() Transaction {
+	return &ovsExecTx{ovsif: ovsif, flows: []string{}}
+}
+
 // bundle executes all given flows as a single atomic transaction
 func (ovsif *ovsExec) bundle(flows []string) error {
 	if len(flows) == 0 {
@@ -362,3 +328,31 @@ func (ovsif *ovsExec) bundle(flows []string) error {
 	_, err := ovsif.execWithStdin(OVS_OFCTL, flows, "bundle", ovsif.bridge, "-")
 	return err
 }
+
+// ovsExecTx implements ovs.Transaction and maintains current flow context
+type ovsExecTx struct {
+	ovsif *ovsExec
+	flows []string
+}
+
+func (tx *ovsExecTx) AddFlow(flow string, args ...interface{}) {
+	if len(args) > 0 {
+		flow = fmt.Sprintf(flow, args...)
+	}
+	tx.flows = append(tx.flows, fmt.Sprintf("flow add %s", flow))
+}
+
+func (tx *ovsExecTx) DeleteFlows(flow string, args ...interface{}) {
+	if len(args) > 0 {
+		flow = fmt.Sprintf(flow, args...)
+	}
+	tx.flows = append(tx.flows, fmt.Sprintf("flow delete %s", flow))
+}
+
+func (tx *ovsExecTx) Commit() error {
+	err := tx.ovsif.bundle(tx.flows)
+
+	// Reset flow context
+	tx.flows = []string{}
+	return err
+}