Support other openflow protocols better

Fixes #64

Author: sjafferali

ovs/openflow.go

@@ -186,24 +186,33 @@ func (o *OpenFlowService) AddFlowBundle(bridge string, fn func(tx *FlowTransacti
 //
 // If flow is nil, all flows will be deleted from the specified bridge.
 func (o *OpenFlowService) DelFlows(bridge string, flow *MatchFlow) error {
+	args := []string{"del-flows"}
+	args = append(args, o.c.ofctlFlags...)
+	args = append(args, bridge)
+
 	if flow == nil {
 		// This means we'll flush the entire flows
-		// from the specifided bridge.
-		_, err := o.exec("del-flows", bridge)
+		// from the specified bridge.
+		_, err := o.exec(args...)
 		return err
 	}
 	fb, err := flow.MarshalText()
 	if err != nil {
 		return err
 	}
 
-	_, err = o.exec("del-flows", bridge, string(fb))
+	args = append(args, string(fb))
+	_, err = o.exec(args...)
 	return err
 }
 
 // ModPort modifies the specified characteristics for the specified port.
 func (o *OpenFlowService) ModPort(bridge string, port string, action PortAction) error {
-	_, err := o.exec("mod-port", bridge, string(port), string(action))
+	args := []string{"mod-port"}
+	args = append(args, o.c.ofctlFlags...)
+	args = append(args, []string{bridge, port, string(action)}...)
+
+	_, err := o.exec(args...)
 	return err
 }
 
@@ -231,7 +240,10 @@ func (o *OpenFlowService) DumpPorts(bridge string) ([]*PortStats, error) {
 // If a table has no active flows and has not been used for a lookup or matched
 // by an incoming packet, it is filtered from the output.
 func (o *OpenFlowService) DumpTables(bridge string) ([]*Table, error) {
-	out, err := o.exec("dump-tables", bridge)
+	args := []string{"dump-tables", bridge}
+	args = append(args, o.c.ofctlFlags...)
+
+	out, err := o.exec(args...)
 	if err != nil {
 		return nil, err
 	}
@@ -259,15 +271,18 @@ func (o *OpenFlowService) DumpTables(bridge string) ([]*Table, error) {
 // If a table has no active flows and has not been used for a lookup or matched
 // by an incoming packet, it is filtered from the output.
 func (o *OpenFlowService) DumpFlows(bridge string) ([]*Flow, error) {
-	out, err := o.exec("dump-flows", bridge)
+	args := []string{"dump-flows", bridge}
+	args = append(args, o.c.ofctlFlags...)
+
+	out, err := o.exec(args...)
 	if err != nil {
 		return nil, err
 	}
 
 	var flows []*Flow
 	err = parseEachLine(out, dumpFlowsPrefix, func(b []byte) error {
-		// Do not attempt to parse NXST_FLOW messages.
-		if bytes.HasPrefix(b, dumpFlowsPrefix) {
+		// Do not attempt to parse ST_FLOW messages.
+		if bytes.Contains(b, dumpFlowsPrefix) {
 			return nil
 		}
 
@@ -303,9 +318,12 @@ var (
 	// the output from 'ovs-ofctl dump-tables'.
 	dumpTablesPrefix = []byte("OFPST_TABLE reply")
 
-	// dumpFlowsPrefix is a sentinel value returned at the beginning of
-	// the output from 'ovs-ofctl dump-flows'.
-	dumpFlowsPrefix = []byte("NXST_FLOW reply")
+	// dumpFlowsPrefix is a sentinel value returned at the beginning of the output
+	// from 'ovs-ofctl dump-flows'. The value returned when using protocol version
+	// 1.0 is "NXST_FLOW reply". The value returned when using protocol version > 1.1
+	// is "OFPST_FLOW reply". However, we use ST_FLOW here to be able to match both
+	// of these.
+	dumpFlowsPrefix = []byte("ST_FLOW reply")
 
 	// dumpAggregatePrefix is a sentinel value returned at the beginning of
 	// the output from "ovs-ofctl dump-aggregate"
@@ -387,8 +405,8 @@ func parseEachLine(in []byte, prefix []byte, fn func(b []byte) error) error {
 		return io.ErrUnexpectedEOF
 	}
 
-	// First line must contain prefix returned by OVS
-	if !bytes.HasPrefix(scanner.Bytes(), prefix) {
+	// First line must contain the prefix returned by OVS
+	if !bytes.Contains(scanner.Bytes(), prefix) {
 		return io.ErrUnexpectedEOF
 	}
 

ovs/openflow_test.go

@@ -1058,6 +1058,120 @@ NXST_FLOW reply (xid=0x4):
 	}
 }
 
+func TestClientOpenFlowDumpFlows15(t *testing.T) {
+	tests := []struct {
+		name  string
+		input string
+		flows string
+		want  []*Flow
+		err   error
+	}{
+		{
+			name:  "test single flow",
+			input: "br0",
+			flows: `OFPST_FLOW reply (OF1.5) (xid=0x2):
+ cookie=0x0, duration=82301.183s, table=0, n_packets=1127501, n_bytes=1595250938, idle_age=0, priority=100,ct_state=-trk,ip actions=ct(table=1)
+`,
+			want: []*Flow{
+				{
+					Priority: 100,
+					Protocol: ProtocolIPv4,
+					Matches: []Match{
+						ConnectionTrackingState(UnsetState(CTStateTracked)),
+					},
+					Table: 0,
+					Actions: []Action{
+						ConnectionTracking("table=1"),
+					},
+				},
+			},
+			err: nil,
+		},
+		{
+			name:  "test multiple flows",
+			input: "br0",
+			flows: `OFPST_FLOW reply (OF1.5) (xid=0x2):
+ cookie=0x0, duration=82301.183s, table=0, n_packets=1127501, n_bytes=1595250938, idle_age=0, priority=100,ct_state=-trk,ip actions=ct(table=1)
+ cookie=0x0, duration=82301.183s, table=0, n_packets=7370490, n_bytes=893401420, idle_age=0, priority=0 actions=NORMAL
+ cookie=0x0, duration=82301.183s, table=100, n_packets=1068, n_bytes=388186, idle_age=3191, priority=1000,udp,tp_dst=53 actions=ct(commit),NORMAL
+`,
+			want: []*Flow{
+				{
+					Priority: 100,
+					Protocol: ProtocolIPv4,
+					Matches: []Match{
+						ConnectionTrackingState(UnsetState(CTStateTracked)),
+					},
+					Table: 0,
+					Actions: []Action{
+						ConnectionTracking("table=1"),
+					},
+				},
+				{
+					Priority: 0,
+					Matches:  []Match{},
+					Table:    0,
+					Actions: []Action{
+						Normal(),
+					},
+				},
+				{
+					Priority: 1000,
+					Protocol: ProtocolUDPv4,
+					Matches: []Match{
+						TransportDestinationPort(53),
+					},
+					Table: 100,
+					Actions: []Action{
+						ConnectionTracking("commit"),
+						Normal(),
+					},
+				},
+			},
+			err: nil,
+		},
+	}
+
+	options := []OptionFunc{
+		Protocols([]string{ProtocolOpenFlow15}),
+		Timeout(1),
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, _ := testClient(options, func(cmd string, args ...string) ([]byte, error) {
+				if want, got := "ovs-ofctl", cmd; want != got {
+					t.Fatalf("incorrect command:\n- want: %v\n-  got: %v",
+						want, got)
+				}
+				wantArgs := []string{
+					"--timeout=1",
+					"dump-flows",
+					string(tt.input),
+					"--protocols=OpenFlow15",
+				}
+				if want, got := wantArgs, args; !reflect.DeepEqual(want, got) {
+					t.Fatalf("incorrect arguments\n- want: %v\n-  got: %v",
+						want, got)
+				}
+				return []byte(tt.flows), tt.err
+			}).OpenFlow.DumpFlows(tt.input)
+			if len(tt.want) != len(got) {
+				t.Errorf("got  %d", len(got))
+				t.Errorf("want %d", len(tt.want))
+				t.Fatal("expected return value to be equal")
+			}
+			for i := range tt.want {
+				if !flowsEqual(tt.want[i], got[i]) {
+					t.Errorf("got  %v", got[i])
+					t.Errorf("want %v", tt.want[i])
+					t.Fatal("expected return value to be equal")
+				}
+			}
+		})
+	}
+}
+
 func mustVerifyFlowBundle(t *testing.T, stdin io.Reader, flows []*Flow, matchFlows []*MatchFlow) {
 	s := bufio.NewScanner(stdin)
 	var gotFlows []*Flow