Repository: ovn-org/libovsdb
Pull request #258 (2021-11-03 21:44:01)

Protect cache integrity during reads
During an invalid cache state, the client will disconnect, then attempt
reconnect to each endpoint. During this process reads against the cache
is are not prohibited. This means a client could be reading stale data
from the cache.

During reconnect we use the meta db.cacheMutex (not the cache mutex) to
control resetting the db's cache. This patch leverages that to guard
reads to the cache based on the same mutex. Additionally, it tries to
ensure that the cache will be in a consistent state when the read takes
place. The db.cacheMutex is not held for the entire reconnect process,
so we need to make some attempt to wait for a signal that a reconnect is
complete... a best effort attempt to give the client an accurate cache
read.

Signed-off-by: Tim Rozet <trozet@redhat.com>

coveralls: ## Pull Request Test Coverage Report for [Build 1418828115](https://coveralls.io/builds/43997274)

* **36** of **41**   **(87.8%)**  changed or added relevant lines in **1** file are covered.
* **1** unchanged line in **1** file lost coverage.
* Overall coverage increased (+**0.1%**) to **72.963%**

---

|  Changes Missing Coverage | Covered Lines | Changed/Added Lines | % |
| :-----|--------------|--------|---: |
| [client/client.go](https://coveralls.io/builds/43997274/source?filename=client%2Fclient.go#L1149) | 36 | 41 | 87.8%
<!-- | **Total:** | **36** | **41** | **87.8%** | -->

|  Files with Coverage Reduction | New Missed Lines | % |
| :-----|--------------|--: |
| [client/client.go](https://coveralls.io/builds/43997274/source?filename=client%2Fclient.go#L1154) | 1 | 64.83% |
<!-- | **Total:** | **1** |  | -->

|  Totals | [![Coverage Status](https://coveralls.io/builds/43997274/badge)](https://coveralls.io/builds/43997274) |
| :-- | --: |
| Change from base [Build 1399758238](https://coveralls.io/builds/43879239): |  0.1% |
| Covered Lines: | 4191 |
| Relevant Lines: | 5744 |

---
##### 💛  - [Coveralls](https://coveralls.io)

trozet: /assign @dave-tucker 
jcaamano: Should we immediately return an error if the client is used while reconnecting and keep track of that with it's own guarded flag?
Even if we attempt to block and wait for it, should we still keep track of it and return an error on timeout?
Isn't a 20 seconds timeout too much?? 

trozet: > Should we immediately return an error if the client is used while reconnecting and keep track of that with it's own guarded flag? 

How would that work? Ovn-kube would be responsible for retrying the operation? We block on transactions until we are connected and ready to send. That timeout looks like it would be 10 seconds.

Even if we attempt to block and wait for it, should we still keep track of it and return an error on timeout?
Probably. Or at least warn the cache may be inconsistent. Most of our ovnkube handling isnt smart enough to retry the operation again later. Maybe that is something we need to fix in ovnk.

Isn't a 20 seconds timeout too much??
I just did connection + monitor timeout. I'm open to changing it. What do you think is more reasonable?

dave-tucker: @trozet rather than hard-coding a timeout here, it would be better to propagate a `ctx context.Context` through the `Get` and `List` functions and down in to `waitForCacheConsistent`. The caller can implement whatever retry/backoff strategy they like.
dave-tucker: > Should we immediately return an error if the client is used while reconnecting and keep track of that with it's own guarded flag?

We currently have the `rpcMutex` which is held by the reconnect logic. This stops any rpc calls being made until after the connection is established. Unfortunately it also prevents us checking if we're connected until after you've acquired the mutex as the connection test is `rpcClient !=  nil`. We could add a separate bool to track connection state and return an error to the client earlier, but that also needs to be mutex protected :cry: 
jcaamano: > > Should we immediately return an error if the client is used while reconnecting and keep track of that with it's own guarded flag?
> 
> We currently have the `rpcMutex` which is held by the reconnect logic. This stops any rpc calls being made until after the connection is established. Unfortunately it also prevents us checking if we're connected until after you've acquired the mutex as the connection test is `rpcClient != nil`. We could add a separate bool to track connection state and return an error to the client earlier, but that also needs to be mutex protected cry

This is what I meant. Yes, it's an additional mutex but at least is very focused and specific.
jcaamano: > > Should we immediately return an error if the client is used while reconnecting and keep track of that with it's own guarded flag?
> 
> How would that work? Ovn-kube would be responsible for retrying the operation? We block on transactions until we are connected and ready to send. That timeout looks like it would be 10 seconds.
> 
> Even if we attempt to block and wait for it, should we still keep track of it and return an error on timeout? Probably. Or at least warn the cache may be inconsistent. Most of our ovnkube handling isnt smart enough to retry the operation again later. Maybe that is something we need to fix in ovnk.
> 
> Isn't a 20 seconds timeout too much?? I just did connection + monitor timeout. I'm open to changing it. What do you think is more reasonable?

We have a constant that defines the timeout to 10 secs and I think there was a plan to bring it back down. Maybe we can use it for this as well.

I guess if that we are not going to handle the error on ovn-k side returning an error makes no sense. Should we add a log if we are waiting here just show if we see gaps in logs we know this is the reason?

What I keep thinking though is that we eventually have controllers that handle events, and for each event there is 1 transaction and upon any error that transaction is not committed and the event is retried from the beginning. Then we can handle errors more gracefully.
trozet: @jcaamano deferredUpdates (plus the release of cacheMutex lock) is kind of our signal that the client has reconnected and the monitor is completely setup. I know that is not as obvious as if it had its own lock/name, but I prefer to not add another mutex (there are already too many in here). I can add some comments around isCacheConsistent to better explain why deferredUpdates works?
trozet: ok updated and added the missing docs

Patch:
diff --git a/client/api.go b/client/api.go
index db40259..f486bd4 100644
--- a/client/api.go
+++ b/client/api.go
@@ -1,6 +1,7 @@
 package client
 
 import (
+	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -18,7 +19,7 @@ type API interface {
 	// The function parameter must be a pointer to a slice of Models
 	// If the slice is null, the entire cache will be copied into the slice
 	// If it has a capacity != 0, only 'capacity' elements will be filled in
-	List(result interface{}) error
+	List(ctx context.Context, result interface{}) 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:
@@ -40,7 +41,7 @@ type API interface {
 	// provided model and the indexes defined in the associated schema
 	// For more complex ways of searching for elements in the cache, the
 	// preferred way is Where({condition}).List()
-	Get(model.Model) error
+	Get(context.Context, model.Model) error
 
 	// Create returns the operation needed to add the model(s) to the Database
 	// Only fields with non-default values will be added to the transaction
@@ -53,7 +54,7 @@ type API interface {
 type ConditionalAPI interface {
 	// List uses the condition to search on the cache and populates
 	// the slice of Models objects based on their type
-	List(result interface{}) error
+	List(ctx context.Context, result interface{}) error
 
 	// Mutate returns the operations needed to perform the mutation specified
 	// By the model and the list of Mutation objects
@@ -93,7 +94,7 @@ type api struct {
 }
 
 // List populates a slice of Models given as parameter based on the configured Condition
-func (a api) List(result interface{}) error {
+func (a api) List(ctx context.Context, result interface{}) error {
 	resultPtr := reflect.ValueOf(result)
 	if resultPtr.Type().Kind() != reflect.Ptr {
 		return &ErrWrongType{resultPtr.Type(), "Expected pointer to slice of valid Models"}
@@ -206,7 +207,7 @@ func (a api) conditionFromModel(any bool, model model.Model, cond ...model.Condi
 //
 // The way the cache is searched depends on the fields already populated in 'result'
 // Any table index (including _uuid) will be used for comparison
-func (a api) Get(m model.Model) error {
+func (a api) Get(ctx context.Context, m model.Model) error {
 	table, err := a.getTableFromModel(m)
 	if err != nil {
 		return err
diff --git a/client/client.go b/client/client.go
index de9b1c7..e8f8069 100644
--- a/client/client.go
+++ b/client/client.go
@@ -13,6 +13,7 @@ import (
 	"reflect"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/cenkalti/backoff/v4"
 	"github.com/cenkalti/rpc2"
@@ -1095,13 +1096,56 @@ func (o *ovsdbClient) Close() {
 	o.rpcClient.Close()
 }
 
+// Ensures the cache is consistent by evaluating that the client is connected
+// and the monitor is fully setup, with the cache populated
+func isCacheConsistent(db *database) bool {
+	// This works because when a client is disconnected the deferUpdates variable
+	// will be set to true. deferUpdates is also protected by the db.cacheMutex.
+	// When the client reconnects and then re-establishes the monitor; the final step
+	// is to process all deferred updates, set deferUpdates back to false, and unlock cacheMutex
+	db.cacheMutex.RLock()
+	defer db.cacheMutex.RUnlock()
+	return !db.deferUpdates
+}
+
+// best effort to ensure cache is in a good state for reading
+func waitForCacheConsistent(ctx context.Context, db *database, logger *logr.Logger, dbName string) {
+	if !hasMonitors(db) {
+		return
+	}
+	ticker := time.NewTicker(50 * time.Millisecond)
+	for {
+		select {
+		case <-ctx.Done():
+			logger.V(3).Info("warning: unable to ensure cache consistency for reading",
+				"database", dbName)
+			return
+		case <-ticker.C:
+			if isCacheConsistent(db) {
+				return
+			}
+
+		}
+	}
+}
+
+func hasMonitors(db *database) bool {
+	db.monitorsMutex.Lock()
+	defer db.monitorsMutex.Unlock()
+	return len(db.monitors) > 0
+}
+
 // Client API interface wrapper functions
 // We add this wrapper to allow users to access the API directly on the
 // client object
 
 //Get implements the API interface's Get function
-func (o *ovsdbClient) Get(model model.Model) error {
-	return o.primaryDB().api.Get(model)
+func (o *ovsdbClient) Get(ctx context.Context, model model.Model) error {
+	primaryDB := o.primaryDB()
+	waitForCacheConsistent(ctx, primaryDB, o.logger, o.primaryDBName)
+	primaryDB.cacheMutex.RLock()
+	defer primaryDB.cacheMutex.RUnlock()
+	return primaryDB.api.Get(ctx, model)
 }
 
 //Create implements the API interface's Create function
@@ -1110,8 +1154,12 @@ func (o *ovsdbClient) Create(models ...model.Model) ([]ovsdb.Operation, error) {
 }
 
 //List implements the API interface's List function
-func (o *ovsdbClient) List(result interface{}) error {
-	return o.primaryDB().api.List(result)
+func (o *ovsdbClient) List(ctx context.Context, result interface{}) error {
+	primaryDB := o.primaryDB()
+	waitForCacheConsistent(ctx, primaryDB, o.logger, o.primaryDBName)
+	primaryDB.cacheMutex.RLock()
+	defer primaryDB.cacheMutex.RUnlock()
+	return primaryDB.api.List(ctx, result)
 }
 
 //Where implements the API interface's Where function
diff --git a/cmd/stress/stress.go b/cmd/stress/stress.go
index e044386..2ebbdcb 100644
--- a/cmd/stress/stress.go
+++ b/cmd/stress/stress.go
@@ -77,7 +77,7 @@ func cleanup(ctx context.Context) {
 
 	// Remove all existing bridges
 	var bridges []bridgeType
-	if err := ovs.List(&bridges); err == nil {
+	if err := ovs.List(context.Background(), &bridges); err == nil {
 		log.Printf("%d existing bridges found", len(bridges))
 		for _, bridge := range bridges {
 			deleteBridge(ctx, ovs, rootUUID, &bridge)
diff --git a/example/play_with_ovs/main.go b/example/play_with_ovs/main.go
index 93a8253..6c7110f 100644
--- a/example/play_with_ovs/main.go
+++ b/example/play_with_ovs/main.go
@@ -37,7 +37,7 @@ func play(ovs client.Client) {
 		} else {
 			fmt.Printf("Current list of bridges:\n")
 			var bridges []vswitchd.Bridge
-			if err := ovs.List(&bridges); err != nil {
+			if err := ovs.List(context.Background(), &bridges); err != nil {
 				log.Fatal(err)
 			}
 			for _, b := range bridges {


Test patch:
diff --git a/client/api_test.go b/client/api_test.go
index 0095188..ed3487a 100644
--- a/client/api_test.go
+++ b/client/api_test.go
@@ -1,6 +1,7 @@
 package client
 
 import (
+	"context"
 	"fmt"
 	"strings"
 	"testing"
@@ -95,7 +96,7 @@ func TestAPIListSimple(t *testing.T) {
 				result = make([]testLogicalSwitch, tt.initialCap)
 			}
 			api := newAPI(tcache, &discardLogger)
-			err := api.List(&result)
+			err := api.List(context.Background(), &result)
 			if tt.err {
 				assert.NotNil(t, err)
 			} else {
@@ -111,14 +112,14 @@ func TestAPIListSimple(t *testing.T) {
 	t.Run("ApiList: Error wrong type", func(t *testing.T) {
 		var result []string
 		api := newAPI(tcache, &discardLogger)
-		err := api.List(&result)
+		err := api.List(context.Background(), &result)
 		assert.NotNil(t, err)
 	})
 
 	t.Run("ApiList: Type Selection", func(t *testing.T) {
 		var result []testLogicalSwitchPort
 		api := newAPI(tcache, &discardLogger)
-		err := api.List(&result)
+		err := api.List(context.Background(), &result)
 		assert.Nil(t, err)
 		assert.Len(t, result, 0, "Should be empty since cache is empty")
 	})
@@ -126,7 +127,7 @@ func TestAPIListSimple(t *testing.T) {
 	t.Run("ApiList: Empty List", func(t *testing.T) {
 		result := []testLogicalSwitch{}
 		api := newAPI(tcache, &discardLogger)
-		err := api.List(&result)
+		err := api.List(context.Background(), &result)
 		assert.Nil(t, err)
 		assert.Len(t, result, len(lscacheList))
 	})
@@ -213,7 +214,7 @@ func TestAPIListPredicate(t *testing.T) {
 			var result []testLogicalSwitch
 			api := newAPI(tcache, &discardLogger)
 			cond := api.WhereCache(tt.predicate)
-			err := cond.List(&result)
+			err := cond.List(context.Background(), &result)
 			if tt.err {
 				assert.NotNil(t, err)
 			} else {
@@ -301,7 +302,7 @@ func TestAPIListFields(t *testing.T) {
 			// Clean object
 			testObj = testLogicalSwitchPort{}
 			api := newAPI(tcache, &discardLogger)
-			err := api.Where(&testObj).List(&result)
+			err := api.Where(&testObj).List(context.Background(), &result)
 			if tt.err {
 				assert.NotNil(t, err)
 			} else {
@@ -319,7 +320,7 @@ func TestAPIListFields(t *testing.T) {
 			UUID: aUUID0,
 		}
 
-		err := api.Where(&obj).List(&result)
+		err := api.Where(&obj).List(context.Background(), &result)
 		assert.NotNil(t, err)
 	})
 }
@@ -502,7 +503,7 @@ func TestAPIGet(t *testing.T) {
 			var result testLogicalSwitchPort
 			tt.prepare(&result)
 			api := newAPI(tcache, &discardLogger)
-			err := api.Get(&result)
+			err := api.Get(context.Background(), &result)
 			if tt.err {
 				assert.NotNil(t, err)
 			} else {
diff --git a/server/server_integration_test.go b/server/server_integration_test.go
index b3cd5e5..3d51409 100644
--- a/server/server_integration_test.go
+++ b/server/server_integration_test.go
@@ -145,12 +145,12 @@ func TestClientServerInsert(t *testing.T) {
 	uuid := reply[0].UUID.GoUUID
 	require.Eventually(t, func() bool {
 		br := &bridgeType{UUID: uuid}
-		err := ovs.Get(br)
+		err := ovs.Get(context.Background(), br)
 		return err == nil
 	}, 2*time.Second, 500*time.Millisecond)
 
 	br := &bridgeType{UUID: uuid}
-	err = ovs.Get(br)
+	err = ovs.Get(context.Background(), br)
 	require.NoError(t, err)
 
 	assert.Equal(t, bridgeRow.Name, br.Name)
@@ -335,7 +335,7 @@ func TestClientServerInsertAndDelete(t *testing.T) {
 	uuid := reply[0].UUID.GoUUID
 	assert.Eventually(t, func() bool {
 		br := &bridgeType{UUID: uuid}
-		err := ovs.Get(br)
+		err := ovs.Get(context.Background(), br)
 		return err == nil
 	}, 2*time.Second, 500*time.Millisecond)
 
@@ -457,7 +457,7 @@ func TestClientServerInsertAndUpdate(t *testing.T) {
 	uuid := reply[0].UUID.GoUUID
 	assert.Eventually(t, func() bool {
 		br := &bridgeType{UUID: uuid}
-		err := ovs.Get(br)
+		err := ovs.Get(context.Background(), br)
 		return err == nil
 	}, 2*time.Second, 500*time.Millisecond)
 
@@ -482,7 +482,7 @@ func TestClientServerInsertAndUpdate(t *testing.T) {
 
 	require.Eventually(t, func() bool {
 		br := &bridgeType{UUID: uuid}
-		err = ovs.Get(br)
+		err = ovs.Get(context.Background(), br)
 		if err != nil {
 			return false
 		}
@@ -500,7 +500,7 @@ func TestClientServerInsertAndUpdate(t *testing.T) {
 
 	assert.Eventually(t, func() bool {
 		br := &bridgeType{UUID: uuid}
-		err = ovs.Get(br)
+		err = ovs.Get(context.Background(), br)
 		if err != nil {
 			return false
 		}
@@ -508,7 +508,7 @@ func TestClientServerInsertAndUpdate(t *testing.T) {
 	}, 2*time.Second, 500*time.Millisecond)
 
 	br := &bridgeType{UUID: uuid}
-	err = ovs.Get(br)
+	err = ovs.Get(context.Background(), br)
 	assert.NoError(t, err)
 
 	assert.Equal(t, bridgeRow, br)
diff --git a/test/ovs/ovs_integration_test.go b/test/ovs/ovs_integration_test.go
index 4579c11..00b375a 100644
--- a/test/ovs/ovs_integration_test.go
+++ b/test/ovs/ovs_integration_test.go
@@ -195,7 +195,7 @@ func (suite *OVSIntegrationSuite) TestConnectReconnect() {
 
 	bridgeUUID, err := suite.createBridge(bridgeName)
 	require.NoError(suite.T(), err)
-	br := <-brChan
+	<-brChan
 
 	// make another connect call, this should return without error as we're already connected
 	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
@@ -234,12 +234,12 @@ func (suite *OVSIntegrationSuite) TestConnectReconnect() {
 	err = suite.client.Connect(ctx)
 	require.NoError(suite.T(), err)
 
-	br = &bridgeType{
+	br := &bridgeType{
 		UUID: bridgeUUID,
 	}
 
 	// assert cache has been purged
-	err = suite.client.Get(br)
+	err = suite.client.Get(ctx, br)
 	require.Error(suite.T(), err, client.ErrNotFound)
 
 	err = suite.client.Echo(context.TODO())
@@ -254,10 +254,7 @@ func (suite *OVSIntegrationSuite) TestConnectReconnect() {
 	require.NoError(suite.T(), err)
 
 	// assert cache has been re-populated
-	require.Eventually(suite.T(), func() bool {
-		err := suite.client.Get(br)
-		return err == nil
-	}, 2*time.Second, 500*time.Millisecond)
+	require.NoError(suite.T(), suite.client.Get(ctx, br))
 
 }
 
@@ -335,7 +332,7 @@ func (suite *OVSIntegrationSuite) TestWithReconnect() {
 	require.NoError(suite.T(), err)
 
 	// check our original bridge is in the cache
-	err = suite.client.Get(br)
+	err = suite.client.Get(ctx, br)
 	require.NoError(suite.T(), err)
 
 	// create a new bridge to ensure the monitor and cache handler is still working
@@ -400,7 +397,7 @@ func (suite *OVSIntegrationSuite) TestInsertTransactIntegration() {
 	require.NoError(suite.T(), err)
 	require.Eventually(suite.T(), func() bool {
 		br := &bridgeType{UUID: uuid}
-		err := suite.client.Get(br)
+		err := suite.client.Get(context.Background(), br)
 		return err == nil
 	}, 2*time.Second, 500*time.Millisecond)
 }
@@ -411,7 +408,7 @@ func (suite *OVSIntegrationSuite) TestMultipleOpsTransactIntegration() {
 	require.NoError(suite.T(), err)
 	require.Eventually(suite.T(), func() bool {
 		br := &bridgeType{UUID: uuid}
-		err := suite.client.Get(br)
+		err := suite.client.Get(context.Background(), br)
 		return err == nil
 	}, 2*time.Second, 500*time.Millisecond)
 
@@ -456,7 +453,7 @@ func (suite *OVSIntegrationSuite) TestMultipleOpsTransactIntegration() {
 	require.NoError(suite.T(), err)
 
 	require.Eventually(suite.T(), func() bool {
-		err := suite.client.Get(br)
+		err := suite.client.Get(context.Background(), br)
 		return err == nil
 	}, 2*time.Second, 500*time.Millisecond)
 
@@ -477,7 +474,7 @@ func (suite *OVSIntegrationSuite) TestInsertAndDeleteTransactIntegration() {
 
 	require.Eventually(suite.T(), func() bool {
 		br := &bridgeType{UUID: bridgeUUID}
-		err := suite.client.Get(br)
+		err := suite.client.Get(context.Background(), br)
 		return err == nil
 	}, 2*time.Second, 500*time.Millisecond)
 
@@ -508,7 +505,7 @@ func (suite *OVSIntegrationSuite) TestInsertAndDeleteTransactIntegration() {
 
 	require.Eventually(suite.T(), func() bool {
 		br := &bridgeType{UUID: bridgeUUID}
-		err := suite.client.Get(br)
+		err := suite.client.Get(context.Background(), br)
 		return err != nil
 	}, 2*time.Second, 500*time.Millisecond)
 }
@@ -571,7 +568,7 @@ func (suite *OVSIntegrationSuite) TestMonitorCancelIntegration() {
 	require.NoError(suite.T(), err)
 	require.Eventually(suite.T(), func() bool {
 		q := &queueType{UUID: uuid}
-		err = suite.client.Get(q)
+		err = suite.client.Get(context.Background(), q)
 		return err == nil
 	}, 2*time.Second, 500*time.Millisecond)
 
@@ -582,7 +579,7 @@ func (suite *OVSIntegrationSuite) TestMonitorCancelIntegration() {
 	require.NoError(suite.T(), err)
 	assert.Never(suite.T(), func() bool {
 		q := &queueType{UUID: uuid}
-		err = suite.client.Get(q)
+		err = suite.client.Get(context.Background(), q)
 		return err == nil
 	}, 2*time.Second, 500*time.Millisecond)
 }
@@ -593,7 +590,7 @@ func (suite *OVSIntegrationSuite) TestInsertDuplicateTransactIntegration() {
 
 	require.Eventually(suite.T(), func() bool {
 		br := &bridgeType{UUID: uuid}
-		err := suite.client.Get(br)
+		err := suite.client.Get(context.Background(), br)
 		return err == nil
 	}, 2*time.Second, 500*time.Millisecond)
 
@@ -608,12 +605,12 @@ func (suite *OVSIntegrationSuite) TestUpdate() {
 
 	require.Eventually(suite.T(), func() bool {
 		br := &bridgeType{UUID: uuid}
-		err := suite.client.Get(br)
+		err := suite.client.Get(context.Background(), br)
 		return err == nil
 	}, 2*time.Second, 500*time.Millisecond)
 
 	bridgeRow := &bridgeType{UUID: uuid}
-	err = suite.client.Get(bridgeRow)
+	err = suite.client.Get(context.Background(), bridgeRow)
 	require.NoError(suite.T(), err)
 
 	// try to modify immutable field
@@ -633,7 +630,7 @@ func (suite *OVSIntegrationSuite) TestUpdate() {
 
 	require.Eventually(suite.T(), func() bool {
 		br := &bridgeType{UUID: uuid}
-		err = suite.client.Get(br)
+		err = suite.client.Get(context.Background(), br)
 		if err != nil {
 			return false
 		}
@@ -651,7 +648,7 @@ func (suite *OVSIntegrationSuite) TestUpdate() {
 
 	assert.Eventually(suite.T(), func() bool {
 		br := &bridgeType{UUID: uuid}
-		err = suite.client.Get(br)
+		err = suite.client.Get(context.Background(), br)
 		if err != nil {
 			return false
 		}
@@ -737,7 +734,7 @@ func (suite *OVSIntegrationSuite) TestCreateIPFIX() {
 
 	//Assert the IPFIX table is empty
 	ipfixes := []ipfixType{}
-	err = suite.client.List(&ipfixes)
+	err = suite.client.List(context.Background(), &ipfixes)
 	require.NoError(suite.T(), err)
 	require.Empty(suite.T(), ipfixes)