ovsdb: initial Transact RPC and types

Author: mdlayher

ovsdb/client_integration_test.go

@@ -19,6 +19,7 @@ import (
 	"fmt"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/digitalocean/go-openvswitch/ovsdb"
 	"github.com/google/go-cmp/cmp"
@@ -28,11 +29,16 @@ func TestClientIntegration(t *testing.T) {
 	c := dialOVSDB(t)
 	defer c.Close()
 
+	// Cancel RPCs if they take too long.
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
 	t.Run("echo", func(t *testing.T) {
-		testClientEcho(t, c)
+		testClientEcho(ctx, t, c)
 	})
+
 	t.Run("databases", func(t *testing.T) {
-		testClientDatabases(t, c)
+		testClientDatabases(ctx, t, c)
 	})
 }
 
@@ -75,8 +81,8 @@ func TestClientIntegrationConcurrent(t *testing.T) {
 	doneWG.Wait()
 }
 
-func testClientDatabases(t *testing.T, c *ovsdb.Client) {
-	dbs, err := c.ListDatabases(context.Background())
+func testClientDatabases(ctx context.Context, t *testing.T, c *ovsdb.Client) {
+	dbs, err := c.ListDatabases(ctx)
 	if err != nil {
 		t.Fatalf("failed to list databases: %v", err)
 	}
@@ -86,10 +92,25 @@ func testClientDatabases(t *testing.T, c *ovsdb.Client) {
 	if diff := cmp.Diff(want, dbs); diff != "" {
 		t.Fatalf("unexpected databases (-want +got):\n%s", diff)
 	}
+
+	for _, d := range dbs {
+		rows, err := c.Transact(ctx, d, []ovsdb.TransactOp{
+			ovsdb.Select{
+				Table: "Bridge",
+			},
+		})
+		if err != nil {
+			t.Fatalf("failed to perform transaction: %v", err)
+		}
+
+		for i, r := range rows {
+			t.Logf("[%02d] %v", i, r)
+		}
+	}
 }
 
-func testClientEcho(t *testing.T, c *ovsdb.Client) {
-	if err := c.Echo(context.Background()); err != nil {
+func testClientEcho(ctx context.Context, t *testing.T, c *ovsdb.Client) {
+	if err := c.Echo(ctx); err != nil {
 		t.Fatalf("failed to echo: %v", err)
 	}
 }

ovsdb/rpc.go

@@ -45,3 +45,36 @@ func (c *Client) Echo(ctx context.Context) error {
 
 	return nil
 }
+
+// A Row is a database row.  Its keys are database column names, and its values
+// are database column values.
+type Row map[string]interface{}
+
+// TODO(mdlayher): try to make concrete types for row values.
+
+// Transact creates and executes a transaction on the specified database.
+// Each operation is applied in the order they appear in ops.
+func (c *Client) Transact(ctx context.Context, db string, ops []TransactOp) ([]Row, error) {
+	// Required because transact uses an unusual syntax for its arguments.
+	arg := transactArg{
+		Database: db,
+		Ops:      ops,
+	}
+
+	// TODO(mdlayher): deal with non-select ops too.
+	var out []struct {
+		Rows []Row `json:"rows"`
+	}
+
+	if err := c.rpc(ctx, "transact", &out, arg); err != nil {
+		return nil, err
+	}
+
+	// Flatten results from all selects into one slice of rows.
+	var rows []Row
+	for _, o := range out {
+		rows = append(rows, o.Rows...)
+	}
+
+	return rows, nil
+}

ovsdb/rpc_test.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"testing"
 
+	"github.com/digitalocean/go-openvswitch/ovsdb"
 	"github.com/digitalocean/go-openvswitch/ovsdb/internal/jsonrpc"
 	"github.com/google/go-cmp/cmp"
 )
@@ -91,3 +92,63 @@ func TestClientEchoOK(t *testing.T) {
 		t.Fatalf("failed to echo: %v", err)
 	}
 }
+
+func TestClientTransactSelect(t *testing.T) {
+	const db = "Open_vSwitch"
+
+	c, _, done := testClient(t, func(req jsonrpc.Request) jsonrpc.Response {
+		if diff := cmp.Diff("transact", req.Method); diff != "" {
+			panicf("unexpected RPC method (-want +got):\n%s", diff)
+		}
+
+		// TODO(mdlayher): clean up with JSON unmarshaler implementations.
+		params := []interface{}{
+			db,
+			map[string]interface{}{
+				"op":    "select",
+				"table": "Bridge",
+				"where": []interface{}{
+					[]interface{}{"name", "==", "ovsbr0"},
+				},
+			},
+		}
+
+		if diff := cmp.Diff(params, req.Params); diff != "" {
+			panicf("unexpected RPC parameters (-want +got):\n%s", diff)
+		}
+
+		type result struct {
+			Rows []ovsdb.Row `json:"rows"`
+		}
+
+		return jsonrpc.Response{
+			ID: strPtr("1"),
+			Result: mustMarshalJSON(t, []result{{
+				Rows: []ovsdb.Row{{
+					"name": "ovsbr0",
+				}},
+			}}),
+		}
+	})
+	defer done()
+
+	ops := []ovsdb.TransactOp{ovsdb.Select{
+		Table: "Bridge",
+		Where: []ovsdb.Cond{
+			ovsdb.Equal("name", "ovsbr0"),
+		},
+	}}
+
+	rows, err := c.Transact(context.Background(), db, ops)
+	if err != nil {
+		t.Fatalf("failed to perform transaction: %v", err)
+	}
+
+	want := []ovsdb.Row{{
+		"name": "ovsbr0",
+	}}
+
+	if diff := cmp.Diff(want, rows); diff != "" {
+		t.Fatalf("unexpected rows (-want +got):\n%s", diff)
+	}
+}

ovsdb/transact.go

@@ -0,0 +1,104 @@
+// Copyright 2017 DigitalOcean.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package ovsdb
+
+import "encoding/json"
+
+// A Cond is a conditional expression which is evaluated by the OVSDB server
+// in a transaction.
+type Cond struct {
+	Column, Function, Value string
+}
+
+// TODO(mdlayher): more helper functions?  Cond as an interface?
+
+// Equal creates a Cond that ensures a column's value equals the
+// specified value.
+func Equal(column, value string) Cond {
+	return Cond{
+		Column:   column,
+		Function: "==",
+		Value:    value,
+	}
+}
+
+// MarshalJSON implements json.Marshaler.
+func (c Cond) MarshalJSON() ([]byte, error) {
+	// Conditionals are expected in three element arrays.
+	return json.Marshal([3]string{
+		c.Column,
+		c.Function,
+		c.Value,
+	})
+}
+
+// A TransactOp is an operation that can be applied with Client.Transact.
+type TransactOp interface {
+	json.Marshaler
+}
+
+var _ TransactOp = Select{}
+
+// Select is a TransactOp which fetches information from a database.
+type Select struct {
+	// The name of the table to select from.
+	Table string
+
+	// Zero or more Conds for conditional select.
+	Where []Cond
+
+	// TODO(mdlayher): specify columns.
+}
+
+// MarshalJSON implements json.Marshaler.
+func (s Select) MarshalJSON() ([]byte, error) {
+	// Send an empty array instead of nil if no where clause.
+	where := s.Where
+	if where == nil {
+		where = []Cond{}
+	}
+
+	sel := struct {
+		Op    string `json:"op"`
+		Table string `json:"table"`
+		Where []Cond `json:"where"`
+	}{
+		Op:    "select",
+		Table: s.Table,
+		Where: where,
+	}
+
+	return json.Marshal(sel)
+}
+
+// A transactArg is used to properly JSON marshal the arguments for a
+// transact RPC.
+type transactArg struct {
+	Database string
+	Ops      []TransactOp
+}
+
+// MarshalJSON implements json.Marshaler.
+func (t transactArg) MarshalJSON() ([]byte, error) {
+	out := []interface{}{
+		t.Database,
+	}
+
+	for _, op := range t.Ops {
+		out = append(out, op)
+	}
+
+	return json.Marshal(out)
+}