Improve reliability of kvstore-related tests

daemon/cmd/daemon_test.go

@@ -57,8 +57,6 @@ type DaemonSuite struct {
 	// as returned by policy.GetPolicyEnabled().
 	oldPolicyEnabled string
 
-	kvstoreInit bool
-
 	// Owners interface mock
 	OnGetPolicyRepository  func() *policy.Repository
 	OnGetNamedPorts        func() (npm types.NamedPortMultiMap)
@@ -208,11 +206,6 @@ func (ds *DaemonSuite) TearDownTest(c *C) {
 		os.RemoveAll(option.Config.RunDir)
 	}
 
-	if ds.kvstoreInit {
-		kvstore.Client().DeletePrefix(ctx, kvstore.OperationalPath)
-		kvstore.Client().DeletePrefix(ctx, kvstore.BaseKeyPrefix)
-	}
-
 	// Restore the policy enforcement mode.
 	policy.SetPolicyEnabled(ds.oldPolicyEnabled)
 
@@ -230,12 +223,10 @@ var _ = Suite(&DaemonEtcdSuite{})
 
 func (e *DaemonEtcdSuite) SetUpSuite(c *C) {
 	testutils.IntegrationTest(c)
-
-	kvstore.SetupDummy("etcd")
-	e.DaemonSuite.kvstoreInit = true
 }
 
 func (e *DaemonEtcdSuite) SetUpTest(c *C) {
+	kvstore.SetupDummy(c, "etcd")
 	e.DaemonSuite.SetUpTest(c)
 }
 
@@ -251,12 +242,10 @@ var _ = Suite(&DaemonConsulSuite{})
 
 func (e *DaemonConsulSuite) SetUpSuite(c *C) {
 	testutils.IntegrationTest(c)
-
-	kvstore.SetupDummy("consul")
-	e.DaemonSuite.kvstoreInit = true
 }
 
 func (e *DaemonConsulSuite) SetUpTest(c *C) {
+	kvstore.SetupDummy(c, "consul")
 	e.DaemonSuite.SetUpTest(c)
 }
 

pkg/clustermesh/clustermesh_test.go

@@ -100,11 +100,7 @@ func (s *ClusterMeshTestSuite) TestClusterMesh(c *C) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	kvstore.SetupDummy("etcd")
-	defer func() {
-		kvstore.Client().DeletePrefix(context.TODO(), kvstore.BaseKeyPrefix)
-		kvstore.Client().Close(ctx)
-	}()
+	kvstore.SetupDummy(c, "etcd")
 
 	identity.InitWellKnownIdentities(&fakeConfig.Config{})
 	// The nils are only used by k8s CRD identities. We default to kvstore.

pkg/clustermesh/kvstoremesh/kvstoremesh_test.go

@@ -81,16 +81,12 @@ func (w *remoteEtcdClientWrapper) ListAndWatch(ctx context.Context, name, prefix
 }
 
 func TestRemoteClusterRun(t *testing.T) {
-	t.Skip("Skip: it appears to badly interact with the other clustermesh tests")
 	testutils.IntegrationTest(t)
 
-	kvstore.SetupDummyWithConfigOpts("etcd",
+	kvstore.SetupDummyWithConfigOpts(t, "etcd",
 		// Explicitly set higher QPS than the default to speedup the test
 		map[string]string{kvstore.EtcdRateLimitOption: "100"},
 	)
-	t.Cleanup(func() {
-		kvstore.Client().Close(context.Background())
-	})
 
 	tests := []struct {
 		name   string

pkg/clustermesh/remote_cluster_test.go

@@ -58,13 +58,10 @@ func (f *fakeIPCache) Upsert(string, net.IP, uint8, *ipcache.K8sMetadata, ipcach
 func TestRemoteClusterRun(t *testing.T) {
 	testutils.IntegrationTest(t)
 
-	kvstore.SetupDummyWithConfigOpts("etcd",
+	kvstore.SetupDummyWithConfigOpts(t, "etcd",
 		// Explicitly set higher QPS than the default to speedup the test
 		map[string]string{kvstore.EtcdRateLimitOption: "100"},
 	)
-	t.Cleanup(func() {
-		kvstore.Client().Close(context.Background())
-	})
 
 	tests := []struct {
 		name   string

pkg/clustermesh/services_test.go

@@ -62,7 +62,7 @@ func (s *ClusterMeshServicesTestSuite) SetUpTest(c *C) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	kvstore.SetupDummy("etcd")
+	kvstore.SetupDummy(c, "etcd")
 
 	s.randomName = rand.RandomString()
 	clusterName1 := s.randomName + "1"
@@ -121,8 +121,6 @@ func (s *ClusterMeshServicesTestSuite) SetUpTest(c *C) {
 
 func (s *ClusterMeshServicesTestSuite) TearDownTest(c *C) {
 	os.RemoveAll(s.testDir)
-	kvstore.Client().DeletePrefix(context.TODO(), kvstore.BaseKeyPrefix)
-	kvstore.Client().Close(context.TODO())
 }
 
 func (s *ClusterMeshServicesTestSuite) expectEvent(c *C, action k8s.CacheAction, id k8s.ServiceID, fn func(event k8s.ServiceEvent) bool) {

pkg/endpoint/endpoint_test.go

@@ -142,7 +142,7 @@ func NewCachingIdentityAllocator(owner cache.IdentityAllocatorOwner) fakeIdentit
 
 func (s *EndpointSuite) SetUpTest(c *C) {
 	/* Required to test endpoint CEP policy model */
-	kvstore.SetupDummy("etcd")
+	kvstore.SetupDummy(c, "etcd")
 	identity.InitWellKnownIdentities(&fakeConfig.Config{})
 	// The nils are only used by k8s CRD identities. We default to kvstore.
 	mgr := NewCachingIdentityAllocator(&testidentity.IdentityAllocatorOwnerMock{})
@@ -153,7 +153,6 @@ func (s *EndpointSuite) SetUpTest(c *C) {
 
 func (s *EndpointSuite) TearDownTest(c *C) {
 	s.mgr.Close()
-	kvstore.Client().Close(context.TODO())
 	node.UnsetTestLocalNodeStore()
 }
 

pkg/endpoint/redirect_test.go

@@ -131,8 +131,7 @@ func (d *DummyOwner) UpdateIdentities(added, deleted cache.IdentityCache) {}
 
 func (s *RedirectSuite) TestAddVisibilityRedirects(c *check.C) {
 	// Setup dependencies for endpoint.
-	kvstore.SetupDummy("etcd")
-	defer kvstore.Client().Close(context.TODO())
+	kvstore.SetupDummy(c, "etcd")
 
 	identity.InitWellKnownIdentities(&fakeConfig.Config{})
 	idAllocatorOwner := &DummyIdentityAllocatorOwner{}

pkg/identity/cache/allocation_test.go

@@ -94,11 +94,7 @@ func (e *IdentityAllocatorEtcdSuite) SetUpSuite(c *C) {
 }
 
 func (e *IdentityAllocatorEtcdSuite) SetUpTest(c *C) {
-	kvstore.SetupDummy("etcd")
-}
-
-func (e *IdentityAllocatorEtcdSuite) TearDownTest(c *C) {
-	kvstore.Client().Close(context.TODO())
+	kvstore.SetupDummy(c, "etcd")
 }
 
 type IdentityAllocatorConsulSuite struct {
@@ -112,11 +108,7 @@ func (e *IdentityAllocatorConsulSuite) SetUpSuite(c *C) {
 }
 
 func (e *IdentityAllocatorConsulSuite) SetUpTest(c *C) {
-	kvstore.SetupDummy("consul")
-}
-
-func (e *IdentityAllocatorConsulSuite) TearDownTest(c *C) {
-	kvstore.Client().Close(context.TODO())
+	kvstore.SetupDummy(c, "consul")
 }
 
 type dummyOwner struct {

pkg/kvstore/allocator/allocator_test.go

@@ -50,12 +50,7 @@ func (e *AllocatorEtcdSuite) SetUpSuite(c *C) {
 
 func (e *AllocatorEtcdSuite) SetUpTest(c *C) {
 	e.backend = "etcd"
-	kvstore.SetupDummy("etcd")
-}
-
-func (e *AllocatorEtcdSuite) TearDownTest(c *C) {
-	kvstore.Client().DeletePrefix(context.TODO(), testPrefix)
-	kvstore.Client().Close(context.TODO())
+	kvstore.SetupDummy(c, "etcd")
 }
 
 type AllocatorConsulSuite struct {
@@ -70,12 +65,7 @@ func (e *AllocatorConsulSuite) SetUpSuite(c *C) {
 
 func (e *AllocatorConsulSuite) SetUpTest(c *C) {
 	e.backend = "consul"
-	kvstore.SetupDummy("consul")
-}
-
-func (e *AllocatorConsulSuite) TearDownTest(c *C) {
-	kvstore.Client().DeletePrefix(context.TODO(), testPrefix)
-	kvstore.Client().Close(context.TODO())
+	kvstore.SetupDummy(c, "consul")
 }
 
 // FIXME: this should be named better, it implements pkg/allocator.Backend

pkg/kvstore/base_test.go

@@ -42,7 +42,7 @@ func testValue(i int) string {
 
 func (s *BaseTests) TestGetSet(c *C) {
 	prefix := "unit-test/"
-	maxID := 256
+	maxID := 8
 
 	Client().DeletePrefix(context.TODO(), prefix)
 	defer Client().DeletePrefix(context.TODO(), prefix)
@@ -68,8 +68,9 @@ func (s *BaseTests) TestGetSet(c *C) {
 		c.Assert(err, IsNil)
 		c.Assert(string(val), checker.DeepEquals, testValue(i))
 
-		val, err = Client().Get(context.TODO(), testKey(prefix, i))
+		key, val, err = Client().GetPrefix(context.TODO(), testKey(prefix, i))
 		c.Assert(err, IsNil)
+		c.Assert(key, checker.DeepEquals, testKey(prefix, i))
 		c.Assert(string(val), checker.DeepEquals, testValue(i))
 	}
 

pkg/kvstore/consul_test.go

@@ -28,11 +28,7 @@ func (e *ConsulSuite) SetUpSuite(c *C) {
 }
 
 func (e *ConsulSuite) SetUpTest(c *C) {
-	SetupDummy("consul")
-}
-
-func (e *ConsulSuite) TearDownTest(c *C) {
-	Client().Close(context.TODO())
+	SetupDummy(c, "consul")
 }
 
 var handler http.HandlerFunc

pkg/kvstore/dummy.go

@@ -3,31 +3,81 @@
 
 package kvstore
 
-import "context"
+import (
+	"context"
+	"testing"
+	"time"
 
-// SetupDummy sets up kvstore for tests
-func SetupDummy(dummyBackend string) {
-	SetupDummyWithConfigOpts(dummyBackend, nil)
+	"github.com/cilium/cilium/pkg/inctimer"
+)
+
+// SetupDummy sets up kvstore for tests. A lock mechanism it used to prevent
+// the creation of two clients at the same time, to avoid interferences in case
+// different tests are run in parallel. A cleanup function is automatically
+// registered to delete all keys and close the client when the test terminates.
+func SetupDummy(tb testing.TB, dummyBackend string) {
+	SetupDummyWithConfigOpts(tb, dummyBackend, nil)
 }
 
 // SetupDummyWithConfigOpts sets up the dummy kvstore for tests but also
-// configures the module with the provided opts.
-func SetupDummyWithConfigOpts(dummyBackend string, opts map[string]string) {
+// configures the module with the provided opts. A lock mechanism it used to
+// prevent the creation of two clients at the same time, to avoid interferences
+// in case different tests are run in parallel. A cleanup function is
+// automatically registered to delete all keys and close the client when the
+// test terminates.
+func SetupDummyWithConfigOpts(tb testing.TB, dummyBackend string, opts map[string]string) {
 	module := getBackend(dummyBackend)
 	if module == nil {
-		log.Panicf("Unknown dummy kvstore backend %s", dummyBackend)
+		tb.Fatalf("Unknown dummy kvstore backend %s", dummyBackend)
 	}
 
 	module.setConfigDummy()
 
 	if opts != nil {
 		err := module.setConfig(opts)
 		if err != nil {
-			log.WithError(err).Panic("Unable to set config options for kvstore backend module")
+			tb.Fatalf("Unable to set config options for kvstore backend module: %v", err)
 		}
 	}
 
-	if err := initClient(context.TODO(), module, nil); err != nil {
-		log.WithError(err).Panic("Unable to initialize kvstore client")
+	if err := initClient(context.Background(), module, nil); err != nil {
+		tb.Fatalf("Unable to initialize kvstore client: %v", err)
+	}
+
+	tb.Cleanup(func() {
+		if err := Client().DeletePrefix(context.Background(), ""); err != nil {
+			tb.Fatalf("Unable to delete all kvstore keys: %v", err)
+		}
+
+		Client().Close(context.Background())
+	})
+
+	ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
+	defer cancel()
+
+	timer, done := inctimer.New()
+	defer done()
+
+	// Multiple tests might be running in parallel by go test if they are part of
+	// different packages. Let's implement a locking mechanism to ensure that only
+	// one at a time can access the kvstore, to prevent that they interact with
+	// each other. Locking is implemented through CreateOnly (rather than using
+	// the locking abstraction), so that we can release it in the same atomic
+	// transaction that also removes all the other keys.
+	for {
+		succeeded, err := Client().CreateOnly(ctx, ".lock", []byte(""), true)
+		if err != nil {
+			tb.Fatalf("Unable to acquire the kvstore lock: %v", err)
+		}
+
+		if succeeded {
+			return
+		}
+
+		select {
+		case <-timer.After(100 * time.Millisecond):
+		case <-ctx.Done():
+			tb.Fatal("Timed out waiting to acquire the kvstore lock")
+		}
 	}
 }