Repository: ovn-org/libovsdb
Pull request #344 (2022-12-02 20:08:29)

Client: Support conditional monitoring
Today the WithConditionalTable interface exists for condition monitor
but it doesn't really work.

This patches fix the interface WithConditionalTable to support multiple
conditions, properly convert the model.Condition to ovsdb.Condition, and
apply the conditions in client.monitor implementation.

Signed-off-by: Han Zhou <hzhou@ovn.org>


Patch:
diff --git a/client/client.go b/client/client.go
index 1385379..4b2ef8f 100644
--- a/client/client.go
+++ b/client/client.go
@@ -840,6 +840,20 @@ func (o *ovsdbClient) Monitor(ctx context.Context, monitor *Monitor) (MonitorCoo
 	return cookie, o.monitor(ctx, cookie, false, monitor)
 }
 
+// If fields is provided, the request will be constrained to the provided columns
+// If no fields are provided, all columns will be used
+func newMonitorRequest(data *mapper.Info, fields []string, conditions []ovsdb.Condition) (*ovsdb.MonitorRequest, error) {
+	var columns []string
+	if len(fields) > 0 {
+		columns = append(columns, fields...)
+	} else {
+		for c := range data.Metadata.TableSchema.Columns {
+			columns = append(columns, c)
+		}
+	}
+	return &ovsdb.MonitorRequest{Columns: columns, Where: conditions, Select: ovsdb.NewDefaultMonitorSelect()}, nil
+}
+
 //gocyclo:ignore
 // monitor must only be called with a lock on monitorsMutex
 func (o *ovsdbClient) monitor(ctx context.Context, cookie MonitorCookie, reconnecting bool, monitor *Monitor) error {
@@ -851,9 +865,6 @@ func (o *ovsdbClient) monitor(ctx context.Context, cookie MonitorCookie, reconne
 	if o.rpcClient == nil {
 		return ErrNotConnected
 	}
-	if len(monitor.Tables) == 0 {
-		return fmt.Errorf("at least one table should be monitored")
-	}
 	if len(monitor.Errors) != 0 {
 		var errString []string
 		for _, err := range monitor.Errors {
@@ -861,10 +872,12 @@ func (o *ovsdbClient) monitor(ctx context.Context, cookie MonitorCookie, reconne
 		}
 		return fmt.Errorf(strings.Join(errString, ". "))
 	}
+	if len(monitor.Tables) == 0 {
+		return fmt.Errorf("at least one table should be monitored")
+	}
 	dbName := cookie.DatabaseName
 	db := o.databases[dbName]
 	db.modelMutex.RLock()
-	mmapper := db.model.Mapper
 	typeMap := db.model.Types()
 	requests := make(map[string]ovsdb.MonitorRequest)
 	for _, o := range monitor.Tables {
@@ -880,7 +893,7 @@ func (o *ovsdbClient) monitor(ctx context.Context, cookie MonitorCookie, reconne
 		if err != nil {
 			return err
 		}
-		request, err := mmapper.NewMonitorRequest(info, o.Fields)
+		request, err := newMonitorRequest(info, o.Fields, o.Conditions)
 		if err != nil {
 			return err
 		}
diff --git a/client/monitor.go b/client/monitor.go
index 65b38ac..4a0270a 100644
--- a/client/monitor.go
+++ b/client/monitor.go
@@ -61,19 +61,31 @@ func newMonitorCookie(dbName string) MonitorCookie {
 type TableMonitor struct {
 	// Table is the table to be monitored
 	Table string
-	// Condition is the condition under which the table should be monitored
-	Condition model.Condition
+	// Conditions are the conditions under which the table should be monitored
+	Conditions []ovsdb.Condition
 	// Fields are the fields in the model to monitor
 	// If none are supplied, all fields will be used
 	Fields []string
 }
 
-func fieldsAsStrings(dbModel model.DatabaseModel, m model.Model, fields []interface{}) ([]string, error) {
+func newTableMonitor(o *ovsdbClient, m model.Model, conditions []model.Condition, fields []interface{}) (*TableMonitor, error) {
+	dbModel := o.primaryDB().model
+	tableName := dbModel.FindTable(reflect.TypeOf(m))
+	if tableName == "" {
+		return nil, fmt.Errorf("object of type %s is not part of the ClientDBModel", reflect.TypeOf(m))
+	}
+
 	var columns []string
+	var ovsdbConds []ovsdb.Condition
 
-	if len(fields) == 0 {
-		return columns, nil
+	if len(fields) == 0 && len(conditions) == 0 {
+		return &TableMonitor{
+			Table:      tableName,
+			Conditions: ovsdbConds,
+			Fields:     columns,
+		}, nil
 	}
+
 	data, err := dbModel.NewModelInfo(m)
 	if err != nil {
 		return nil, fmt.Errorf("unable to obtain info from model %v: %v", m, err)
@@ -85,46 +97,40 @@ func fieldsAsStrings(dbModel model.DatabaseModel, m model.Model, fields []interf
 		}
 		columns = append(columns, column)
 	}
-	return columns, nil
+	db := o.databases[o.primaryDBName]
+	mmapper := db.model.Mapper
+	for _, modelCond := range conditions {
+		ovsdbCond, err := mmapper.NewCondition(data, modelCond.Field, modelCond.Function, modelCond.Value)
+		if err != nil {
+			return nil, fmt.Errorf("unable to convert condition %v: %v", modelCond, err)
+		}
+		ovsdbConds = append(ovsdbConds, *ovsdbCond)
+	}
+	return &TableMonitor{
+		Table:      tableName,
+		Conditions: ovsdbConds,
+		Fields:     columns,
+	}, nil
 }
 
 func WithTable(m model.Model, fields ...interface{}) MonitorOption {
 	return func(o *ovsdbClient, monitor *Monitor) error {
-		dbModel := o.primaryDB().model
-		tableName := dbModel.FindTable(reflect.TypeOf(m))
-		if tableName == "" {
-			return fmt.Errorf("object of type %s is not part of the ClientDBModel", reflect.TypeOf(m))
-		}
-		fieldsStr, err := fieldsAsStrings(dbModel, m, fields)
+		tableMonitor, err := newTableMonitor(o, m, []model.Condition{}, fields)
 		if err != nil {
 			return err
 		}
-		tableMonitor := TableMonitor{
-			Table:  tableName,
-			Fields: fieldsStr,
-		}
-		monitor.Tables = append(monitor.Tables, tableMonitor)
+		monitor.Tables = append(monitor.Tables, *tableMonitor)
 		return nil
 	}
 }
 
-func WithConditionalTable(m model.Model, condition model.Condition, fields ...interface{}) MonitorOption {
+func WithConditionalTable(m model.Model, conditions []model.Condition, fields ...interface{}) MonitorOption {
 	return func(o *ovsdbClient, monitor *Monitor) error {
-		dbModel := o.primaryDB().model
-		tableName := dbModel.FindTable(reflect.TypeOf(m))
-		if tableName == "" {
-			return fmt.Errorf("object of type %s is not part of the ClientDBModel", reflect.TypeOf(m))
-		}
-		fieldsStr, err := fieldsAsStrings(dbModel, m, fields)
+		tableMonitor, err := newTableMonitor(o, m, conditions, fields)
 		if err != nil {
 			return err
 		}
-		tableMonitor := TableMonitor{
-			Table:     tableName,
-			Condition: condition,
-			Fields:    fieldsStr,
-		}
-		monitor.Tables = append(monitor.Tables, tableMonitor)
+		monitor.Tables = append(monitor.Tables, *tableMonitor)
 		return nil
 	}
 }
diff --git a/mapper/mapper.go b/mapper/mapper.go
index 7b76200..5ca7a41 100644
--- a/mapper/mapper.go
+++ b/mapper/mapper.go
@@ -315,18 +315,3 @@ func (m Mapper) equalIndexes(one, other *Info, indexes ...string) (bool, error)
 	}
 	return false, nil
 }
-
-// NewMonitorRequest returns a monitor request for the provided tableName
-// If fields is provided, the request will be constrained to the provided columns
-// If no fields are provided, all columns will be used
-func (m *Mapper) NewMonitorRequest(data *Info, fields []string) (*ovsdb.MonitorRequest, error) {
-	var columns []string
-	if len(fields) > 0 {
-		columns = append(columns, fields...)
-	} else {
-		for c := range data.Metadata.TableSchema.Columns {
-			columns = append(columns, c)
-		}
-	}
-	return &ovsdb.MonitorRequest{Columns: columns, Select: ovsdb.NewDefaultMonitorSelect()}, nil
-}


Test patch:
diff --git a/client/client_test.go b/client/client_test.go
index b4a5e0a..91eff98 100644
--- a/client/client_test.go
+++ b/client/client_test.go
@@ -17,6 +17,7 @@ import (
 	"github.com/google/uuid"
 	"github.com/ovn-org/libovsdb/cache"
 	db "github.com/ovn-org/libovsdb/database"
+	"github.com/ovn-org/libovsdb/mapper"
 	"github.com/ovn-org/libovsdb/model"
 	"github.com/ovn-org/libovsdb/ovsdb"
 	"github.com/ovn-org/libovsdb/ovsdb/serverdb"
@@ -1014,3 +1015,69 @@ func TestClientReconnectLeaderOnly(t *testing.T) {
 		return atomic.LoadInt32(&connected2) > 2
 	}, 2*time.Second, 10*time.Millisecond)
 }
+
+func TestNewMonitorRequest(t *testing.T) {
+	var testSchema = []byte(`{
+  "cksum": "223619766 22548",
+  "name": "TestSchema",
+  "tables": {
+    "TestTable": {
+      "indexes": [["name"],["composed_1","composed_2"]],
+      "columns": {
+        "name": {
+          "type": "string"
+        },
+        "composed_1": {
+          "type": {
+            "key": "string"
+          }
+        },
+        "composed_2": {
+          "type": {
+            "key": "string"
+          }
+        },
+        "int1": {
+          "type": {
+            "key": "integer"
+          }
+        },
+        "int2": {
+          "type": {
+            "key": "integer"
+          }
+        },
+        "config": {
+          "type": {
+            "key": "string",
+            "max": "unlimited",
+            "min": 0,
+            "value": "string"
+          }
+	}
+      }
+    }
+  }
+}`)
+	type testType struct {
+		ID     string            `ovsdb:"_uuid"`
+		MyName string            `ovsdb:"name"`
+		Config map[string]string `ovsdb:"config"`
+		Comp1  string            `ovsdb:"composed_1"`
+		Comp2  string            `ovsdb:"composed_2"`
+		Int1   int               `ovsdb:"int1"`
+		Int2   int               `ovsdb:"int2"`
+	}
+	var schema ovsdb.DatabaseSchema
+	err := json.Unmarshal(testSchema, &schema)
+	require.NoError(t, err)
+	testTable := &testType{}
+	info, err := mapper.NewInfo("TestTable", schema.Table("TestTable"), testTable)
+	assert.NoError(t, err)
+	mr, err := newMonitorRequest(info, nil, nil)
+	require.NoError(t, err)
+	assert.ElementsMatch(t, mr.Columns, []string{"name", "config", "composed_1", "composed_2", "int1", "int2"})
+	mr2, err := newMonitorRequest(info, []string{"int1", "name"}, nil)
+	require.NoError(t, err)
+	assert.ElementsMatch(t, mr2.Columns, []string{"int1", "name"})
+}
diff --git a/client/monitor_test.go b/client/monitor_test.go
index a5336f6..206436a 100644
--- a/client/monitor_test.go
+++ b/client/monitor_test.go
@@ -50,3 +50,34 @@ func TestWithTableAndFields(t *testing.T) {
 	assert.Equal(t, 1, len(m.Tables))
 	assert.ElementsMatch(t, []string{"bridges", "cur_cfg"}, m.Tables[0].Fields)
 }
+
+func TestWithTableAndFieldsAndConditions(t *testing.T) {
+	client, err := newOVSDBClient(defDB)
+	assert.NoError(t, err)
+	populateClientModel(t, client)
+
+	m := newMonitor()
+	bridge := Bridge{}
+
+	conditions := []model.Condition{
+		{
+			Field:    &bridge.Name,
+			Function: ovsdb.ConditionEqual,
+			Value:    "foo",
+		},
+	}
+
+	opt := WithConditionalTable(&bridge, conditions, &bridge.Name, &bridge.DatapathType)
+	err = opt(client, m)
+	assert.NoError(t, err)
+
+	assert.Equal(t, 1, len(m.Tables))
+	assert.ElementsMatch(t, []string{"name", "datapath_type"}, m.Tables[0].Fields)
+	assert.ElementsMatch(t, []ovsdb.Condition{
+		{
+			Column:   "name",
+			Function: ovsdb.ConditionEqual,
+			Value:    "foo",
+		},
+	}, m.Tables[0].Conditions)
+}
diff --git a/mapper/mapper_test.go b/mapper/mapper_test.go
index 2cefede..fe208d0 100644
--- a/mapper/mapper_test.go
+++ b/mapper/mapper_test.go
@@ -7,7 +7,6 @@ import (
 
 	"github.com/ovn-org/libovsdb/ovsdb"
 	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
 )
 
 var (
@@ -1061,70 +1060,3 @@ func testOvsMap(t *testing.T, set interface{}) ovsdb.OvsMap {
 	assert.Nil(t, err)
 	return oMap
 }
-
-func TestNewMonitorRequest(t *testing.T) {
-	var testSchema = []byte(`{
-  "cksum": "223619766 22548",
-  "name": "TestSchema",
-  "tables": {
-    "TestTable": {
-      "indexes": [["name"],["composed_1","composed_2"]],
-      "columns": {
-        "name": {
-          "type": "string"
-        },
-        "composed_1": {
-          "type": {
-            "key": "string"
-          }
-        },
-        "composed_2": {
-          "type": {
-            "key": "string"
-          }
-        },
-        "int1": {
-          "type": {
-            "key": "integer"
-          }
-        },
-        "int2": {
-          "type": {
-            "key": "integer"
-          }
-        },
-        "config": {
-          "type": {
-            "key": "string",
-            "max": "unlimited",
-            "min": 0,
-            "value": "string"
-          }
-	}
-      }
-    }
-  }
-}`)
-	type testType struct {
-		ID     string            `ovsdb:"_uuid"`
-		MyName string            `ovsdb:"name"`
-		Config map[string]string `ovsdb:"config"`
-		Comp1  string            `ovsdb:"composed_1"`
-		Comp2  string            `ovsdb:"composed_2"`
-		Int1   int               `ovsdb:"int1"`
-		Int2   int               `ovsdb:"int2"`
-	}
-	var schema ovsdb.DatabaseSchema
-	err := json.Unmarshal(testSchema, &schema)
-	require.NoError(t, err)
-	mapper := NewMapper(schema)
-	testTable := &testType{}
-	info, err := NewInfo("TestTable", schema.Table("TestTable"), testTable)
-	assert.NoError(t, err)
-	mr, err := mapper.NewMonitorRequest(info, nil)
-	require.NoError(t, err)
-	assert.ElementsMatch(t, mr.Columns, []string{"name", "config", "composed_1", "composed_2", "int1", "int2"})
-	mr2, err := mapper.NewMonitorRequest(info, []string{"int1", "name"})
-	require.NoError(t, err)
-	assert.ElementsMatch(t, mr2.Columns, []string{"int1", "name"})
-}
diff --git a/test/ovs/ovs_integration_test.go b/test/ovs/ovs_integration_test.go
index 7f1b012..c263084 100644
--- a/test/ovs/ovs_integration_test.go
+++ b/test/ovs/ovs_integration_test.go
@@ -591,6 +591,57 @@ func (suite *OVSIntegrationSuite) TestMonitorCancelIntegration() {
 	}, 2*time.Second, 500*time.Millisecond)
 }
 
+func (suite *OVSIntegrationSuite) TestMonitorConditionIntegration() {
+	// Monitor table Queue rows with dscp == 1 or 2.
+	queue := queueType{}
+	dscp1 := 1
+	dscp2 := 2
+	conditions := []model.Condition{
+		{
+			Field:    &queue.DSCP,
+			Function: ovsdb.ConditionEqual,
+			Value:    &dscp1,
+		},
+		{
+			Field:    &queue.DSCP,
+			Function: ovsdb.ConditionEqual,
+			Value:    &dscp2,
+		},
+	}
+
+	_, err := suite.client.Monitor(
+		context.TODO(),
+		suite.client.NewMonitor(
+			client.WithConditionalTable(&queue, conditions),
+		),
+	)
+	require.NoError(suite.T(), err)
+
+	uuid, err := suite.createQueue("test1", 1)
+	require.NoError(suite.T(), err)
+	require.Eventually(suite.T(), func() bool {
+		q := &queueType{UUID: uuid}
+		err = suite.client.Get(context.Background(), q)
+		return err == nil
+	}, 2*time.Second, 500*time.Millisecond)
+
+	uuid, err = suite.createQueue("test2", 3)
+	require.NoError(suite.T(), err)
+	assert.Never(suite.T(), func() bool {
+		q := &queueType{UUID: uuid}
+		err = suite.client.Get(context.Background(), q)
+		return err == nil
+	}, 2*time.Second, 500*time.Millisecond)
+
+	uuid, err = suite.createQueue("test3", 2)
+	require.NoError(suite.T(), err)
+	require.Eventually(suite.T(), func() bool {
+		q := &queueType{UUID: uuid}
+		err = suite.client.Get(context.Background(), q)
+		return err == nil
+	}, 2*time.Second, 500*time.Millisecond)
+}
+
 func (suite *OVSIntegrationSuite) TestInsertDuplicateTransactIntegration() {
 	uuid, err := suite.createBridge("br-dup")
 	require.NoError(suite.T(), err)