Added Bundle() to ovs interface that executes all the given flows as a single atomic transaction

Author: Ravi Sankar Penta

pkg/util/ovs/ovs.go

@@ -1,6 +1,7 @@
 package ovs
 
 import (
+	"bytes"
 	"fmt"
 	"strconv"
 	"strings"
@@ -72,6 +73,9 @@ 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)
 
+	// Bundle executes all given flows as a single atomic transaction
+	Bundle(flows []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
@@ -138,13 +142,23 @@ func New(execer exec.Interface, bridge string, minVersion string) (Interface, er
 	return ovsif, nil
 }
 
-func (ovsif *ovsExec) exec(cmd string, args ...string) (string, error) {
+func (ovsif *ovsExec) execWithStdin(cmd string, stdinArgs []string, args ...string) (string, error) {
 	if cmd == OVS_OFCTL {
 		args = append([]string{"-O", "OpenFlow13"}, args...)
 	}
-	glog.V(4).Infof("Executing: %s %s", cmd, strings.Join(args, " "))
 
-	output, err := ovsif.execer.Command(cmd, args...).CombinedOutput()
+	kcmd := ovsif.execer.Command(cmd, args...)
+	if stdinArgs != nil {
+		stdinString := strings.Join(stdinArgs, "\n")
+		stdin := bytes.NewBufferString(stdinString)
+		kcmd.SetStdin(stdin)
+
+		glog.V(4).Infof("Executing: %s %s |\n%s", cmd, strings.Join(args, " "), stdinString)
+	} else {
+		glog.V(4).Infof("Executing: %s %s", cmd, strings.Join(args, " "))
+	}
+
+	output, err := kcmd.CombinedOutput()
 	if err != nil {
 		glog.V(2).Infof("Error executing %s: %s", cmd, string(output))
 		return "", err
@@ -161,6 +175,10 @@ func (ovsif *ovsExec) exec(cmd string, args ...string) (string, error) {
 	return outStr, nil
 }
 
+func (ovsif *ovsExec) exec(cmd string, args ...string) (string, error) {
+	return ovsif.execWithStdin(cmd, nil, args...)
+}
+
 func (ovsif *ovsExec) AddBridge(properties ...string) error {
 	args := []string{"add-br", ovsif.bridge}
 	if len(properties) > 0 {
@@ -334,3 +352,28 @@ func (ovsif *ovsExec) DumpFlows(flow string, args ...interface{}) ([]string, err
 	}
 	return flows, nil
 }
+
+func (ovsif *ovsExec) Bundle(flows []string) error {
+	if len(flows) == 0 {
+		return nil
+	}
+
+	_, err := ovsif.execWithStdin(OVS_OFCTL, flows, "bundle", ovsif.bridge, "-")
+	return err
+}
+
+// BundleAddFlowRepr converts regular flow add rule to bundle compatible flow rule
+func BundleAddFlowRepr(flow string, args ...interface{}) string {
+	if len(args) > 0 {
+		flow = fmt.Sprintf(flow, args...)
+	}
+	return fmt.Sprintf("flow add %s", flow)
+}
+
+// BundleDeleteFlowRepr converts regular flow delete rule to bundle compatible flow rule
+func BundleDeleteFlowRepr(flow string, args ...interface{}) string {
+	if len(args) > 0 {
+		flow = fmt.Sprintf(flow, args...)
+	}
+	return fmt.Sprintf("flow delete %s", flow)
+}

pkg/util/ovs/ovs_test.go

@@ -92,6 +92,44 @@ func TestTransactionFailure(t *testing.T) {
 	ensureTestResults(t, fexec)
 }
 
+func TestBundle(t *testing.T) {
+	fexec := normalSetup()
+	ovsif, err := New(fexec, "br0", "")
+	if err != nil {
+		t.Fatalf("Unexpected error from ovs.New(): %v", err)
+	}
+
+	// Empty bundle flows
+	bundleFlows := []string{}
+	if err = ovsif.Bundle(bundleFlows); err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	ensureTestResults(t, fexec)
+
+	// Invalid bundle flows
+	err = fmt.Errorf("ovs-ofctl: -:1: Unsupported bundle message type: flow2")
+	addTestResult(t, fexec, "ovs-ofctl -O OpenFlow13 bundle br0 -", "", err)
+	bundleFlows = []string{
+		"flow add flow1",
+		"flow2",
+	}
+	if err = ovsif.Bundle(bundleFlows); err == nil {
+		t.Fatalf("Unexpectedly failed to get error")
+	}
+	ensureTestResults(t, fexec)
+
+	// Valid bundle flows
+	addTestResult(t, fexec, "ovs-ofctl -O OpenFlow13 bundle br0 -", "", nil)
+	bundleFlows = []string{
+		"flow add flow1",
+		"flow delete flow2",
+	}
+	if err = ovsif.Bundle(bundleFlows); err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	ensureTestResults(t, fexec)
+}
+
 func TestDumpFlows(t *testing.T) {
 	fexec := normalSetup()
 	addTestResult(t, fexec, "ovs-ofctl -O OpenFlow13 dump-flows br0 ", `OFPST_FLOW reply (OF1.3) (xid=0x2):