ensure cache is consistent before reading data

Signed-off-by: zhangzujian <zhangzujian.7@gmail.com>

Author: zhangzujian

client/api.go

@@ -21,6 +21,11 @@ type API interface {
 	// If it has a capacity != 0, only 'capacity' elements will be filled in
 	List(ctx context.Context, result any) error
 
+	// Create a Conditional API from a Function that is used to filter cached data
+	// The function must accept a Model implementation and return a boolean. E.g:
+	// ConditionFromFunc(func(l *LogicalSwitch) bool { return l.Enabled })
+	WherePredict(ctx context.Context, predicate any) (ConditionalAPI, error)
+
 	// Create a Conditional API from a Function that is used to filter cached data
 	// The function must accept a Model implementation and return a boolean. E.g:
 	// ConditionFromFunc(func(l *LogicalSwitch) bool { return l.Enabled })
@@ -139,8 +144,10 @@ func (a api) List(_ context.Context, result any) error {
 	}
 
 	if a.cond != nil && a.cond.Table() != table {
-		return &ErrWrongType{resultPtr.Type(),
-			fmt.Sprintf("Table derived from input type (%s) does not match Table from Condition (%s)", table, a.cond.Table())}
+		return &ErrWrongType{
+			resultPtr.Type(),
+			fmt.Sprintf("Table derived from input type (%s) does not match Table from Condition (%s)", table, a.cond.Table()),
+		}
 	}
 
 	tableCache := a.cache.Table(table)
@@ -194,6 +201,11 @@ func (a api) WhereAll(m model.Model, cond ...model.Condition) ConditionalAPI {
 	return newConditionalAPI(a.cache, a.conditionFromExplicitConditions(true, m, cond...), a.logger, a.validateModel)
 }
 
+// WherePredict returns a conditionalAPI based a Predicate
+func (a api) WherePredict(ctx context.Context, predicate interface{}) (ConditionalAPI, error) {
+	return newConditionalAPI(a.cache, a.conditionFromFunc(predicate), a.logger, a.validateModel), nil
+}
+
 // WhereCache returns a conditionalAPI based a Predicate
 func (a api) WhereCache(predicate any) ConditionalAPI {
 	return newConditionalAPI(a.cache, a.conditionFromFunc(predicate), a.logger, a.validateModel)
@@ -321,7 +333,6 @@ func (a api) Create(models ...model.Model) ([]ovsdb.Operation, error) {
 				namedUUID = tmpUUID
 			} else if ovsdb.IsValidUUID(tmpUUID) {
 				realUUID = tmpUUID
-
 			}
 		} else {
 			return nil, fmt.Errorf("error accessing _uuid field: %w", err)
@@ -607,14 +618,18 @@ func (a api) getTableFromFunc(predicate any) (string, error) {
 	modelInterface := reflect.TypeOf((*model.Model)(nil)).Elem()
 	modelType := predType.In(0)
 	if !modelType.Implements(modelInterface) {
-		return "", &ErrWrongType{predType,
-			fmt.Sprintf("Type %s does not implement Model interface", modelType.String())}
+		return "", &ErrWrongType{
+			predType,
+			fmt.Sprintf("Type %s does not implement Model interface", modelType.String()),
+		}
 	}
 
 	table := a.cache.DatabaseModel().FindTable(modelType)
 	if table == "" {
-		return "", &ErrWrongType{predType,
-			fmt.Sprintf("Model %s not found in Database Model", modelType.String())}
+		return "", &ErrWrongType{
+			predType,
+			fmt.Sprintf("Model %s not found in Database Model", modelType.String()),
+		}
 	}
 	return table, nil
 }

client/client.go

@@ -1084,7 +1084,7 @@ func (o *ovsdbClient) monitor(ctx context.Context, cookie MonitorCookie, reconne
 	return err
 }
 
-// Echo tests the liveness of the OVSDB connetion
+// Echo tests the liveness of the OVSDB connection
 func (o *ovsdbClient) Echo(ctx context.Context) error {
 	args := ovsdb.NewEchoArgs()
 	var reply []any
@@ -1369,15 +1369,21 @@ func isCacheConsistent(db *database) bool {
 
 // best effort to ensure cache is in a good state for reading. RLocks the
 // database's cache before returning; caller must always unlock.
-func waitForCacheConsistent(ctx context.Context, db *database, logger *logr.Logger, dbName string) {
+func (o *ovsdbClient) waitForCacheConsistent(ctx context.Context, db *database, logger *logr.Logger, dbName string) error {
 	if !hasMonitors(db) {
 		db.cacheMutex.RLock()
-		return
+		return nil
+	}
+
+	if err := o.Echo(ctx); err != nil {
+		db.cacheMutex.RLock()
+		return err
 	}
+
 	// Check immediately as a fastpath
 	db.cacheMutex.RLock()
 	if isCacheConsistent(db) {
-		return
+		return nil
 	}
 	db.cacheMutex.RUnlock()
 
@@ -1389,11 +1395,11 @@ func waitForCacheConsistent(ctx context.Context, db *database, logger *logr.Logg
 			logger.V(3).Info("warning: unable to ensure cache consistency for reading",
 				"database", dbName)
 			db.cacheMutex.RLock()
-			return
+			return nil
 		case <-ticker.C:
 			db.cacheMutex.RLock()
 			if isCacheConsistent(db) {
-				return
+				return nil
 			}
 			db.cacheMutex.RUnlock()
 		}
@@ -1413,8 +1419,11 @@ func hasMonitors(db *database) bool {
 // Get implements the API interface's Get function
 func (o *ovsdbClient) Get(ctx context.Context, model model.Model) error {
 	primaryDB := o.primaryDB()
-	waitForCacheConsistent(ctx, primaryDB, o.logger, o.primaryDBName)
+	err := o.waitForCacheConsistent(ctx, primaryDB, o.logger, o.primaryDBName)
 	defer primaryDB.cacheMutex.RUnlock()
+	if err != nil {
+		return err
+	}
 	return primaryDB.api.Get(ctx, model)
 }
 
@@ -1426,8 +1435,11 @@ func (o *ovsdbClient) Create(models ...model.Model) ([]ovsdb.Operation, error) {
 // List implements the API interface's List function
 func (o *ovsdbClient) List(ctx context.Context, result any) error {
 	primaryDB := o.primaryDB()
-	waitForCacheConsistent(ctx, primaryDB, o.logger, o.primaryDBName)
+	err := o.waitForCacheConsistent(ctx, primaryDB, o.logger, o.primaryDBName)
 	defer primaryDB.cacheMutex.RUnlock()
+	if err != nil {
+		return err
+	}
 	return primaryDB.api.List(ctx, result)
 }
 
@@ -1446,6 +1458,17 @@ func (o *ovsdbClient) WhereAll(m model.Model, conditions ...model.Condition) Con
 	return o.primaryDB().api.WhereAll(m, conditions...)
 }
 
+// WherePredict implements the API interface's WherePredict function
+func (o *ovsdbClient) WherePredict(ctx context.Context, predicate interface{}) (ConditionalAPI, error) {
+	primaryDB := o.primaryDB()
+	err := o.waitForCacheConsistent(ctx, primaryDB, o.logger, o.primaryDBName)
+	defer primaryDB.cacheMutex.RUnlock()
+	if err != nil {
+		return nil, err
+	}
+	return o.primaryDB().api.WhereCache(predicate), nil
+}
+
 // WhereCache implements the API interface's WhereCache function
 func (o *ovsdbClient) WhereCache(predicate any) ConditionalAPI {
 	return o.primaryDB().api.WhereCache(predicate)