test(e2e): Fix flaky test for ClusterCacheInfo validation (#537)

Signed-off-by: Jayendra Parsai <[email protected]>

Author: jparsai

test/e2e/appcache_test.go

@@ -29,12 +29,6 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 )
 
-const (
-	PrincipalName       = "principal"
-	AgentManagedName    = "agent-managed"
-	AgentAutonomousName = "agent-autonomous"
-)
-
 type CacheTestSuite struct {
 	fixture.BaseSuite
 }
@@ -48,22 +42,22 @@ func (suite *CacheTestSuite) TearDownTest() {
 }
 
 func (suite *CacheTestSuite) SetupTest() {
-	if !fixture.IsProcessRunning(PrincipalName) {
+	if !fixture.IsProcessRunning(fixture.PrincipalName) {
 		// Start the principal if it is not running and wait for it to be ready
-		suite.Require().NoError(fixture.StartProcess(PrincipalName))
-		fixture.CheckReadiness(suite.T(), PrincipalName)
+		suite.Require().NoError(fixture.StartProcess(fixture.PrincipalName))
+		fixture.CheckReadiness(suite.T(), fixture.PrincipalName)
 	} else {
 		// If principal is already running, verify that it is ready
-		fixture.CheckReadiness(suite.T(), PrincipalName)
+		fixture.CheckReadiness(suite.T(), fixture.PrincipalName)
 	}
 
-	if !fixture.IsProcessRunning(AgentManagedName) {
+	if !fixture.IsProcessRunning(fixture.AgentManagedName) {
 		// Start the agent if it is not running and wait for it to be ready
-		suite.Require().NoError(fixture.StartProcess(AgentManagedName))
-		fixture.CheckReadiness(suite.T(), AgentManagedName)
+		suite.Require().NoError(fixture.StartProcess(fixture.AgentManagedName))
+		fixture.CheckReadiness(suite.T(), fixture.AgentManagedName)
 	} else {
 		// If agent is already running, verify that it is ready
-		fixture.CheckReadiness(suite.T(), AgentManagedName)
+		fixture.CheckReadiness(suite.T(), fixture.AgentManagedName)
 	}
 
 	suite.BaseSuite.SetupTest()
@@ -113,17 +107,17 @@ func (suite *CacheTestSuite) Test_RevertDisconnectedManagedClusterChanges() {
 
 	// Case 1: Agent is disconnected with principal, now modify the application directly in the managed-cluster,
 	// but changes should be reverted, to be in sync with last known state of principal application
-	requires.NoError(fixture.StopProcess(PrincipalName))
+	requires.NoError(fixture.StopProcess(fixture.PrincipalName))
 	requires.Eventually(func() bool {
-		return !fixture.IsProcessRunning(PrincipalName)
+		return !fixture.IsProcessRunning(fixture.PrincipalName)
 	}, 30*time.Second, 1*time.Second)
 
 	updateAppInfo(suite.Ctx, suite.ManagedAgentClient, agentKey, []string{"a", "b"}, requires)
 	validateAppReverted(suite.Ctx, suite.ManagedAgentClient, &app, agentKey, requires, suite.T())
 
 	// Case 2: Agent is reconnected with principal and now changes done in principal should reflect in managed-cluster
-	requires.NoError(fixture.StartProcess(PrincipalName))
-	fixture.CheckReadiness(suite.T(), PrincipalName)
+	requires.NoError(fixture.StartProcess(fixture.PrincipalName))
+	fixture.CheckReadiness(suite.T(), fixture.PrincipalName)
 
 	updateAppInfo(suite.Ctx, suite.PrincipalClient, principalKey, []string{"a", "b"}, requires)
 	validateAppInfoUpdated(suite.Ctx, suite.ManagedAgentClient, agentKey, []string{"a", "b"}, requires)
@@ -143,8 +137,8 @@ func (suite *CacheTestSuite) Test_CacheRecreatedOnRestart() {
 
 	// Case 1: Agent is restarted, now make direct changes in the managed-cluster,
 	// but they should be reverted to be in sync with last known state of principal app
-	fixture.RestartAgent(suite.T(), AgentManagedName)
-	fixture.CheckReadiness(suite.T(), AgentManagedName)
+	fixture.RestartAgent(suite.T(), fixture.AgentManagedName)
+	fixture.CheckReadiness(suite.T(), fixture.AgentManagedName)
 
 	updateAppInfo(suite.Ctx, suite.ManagedAgentClient, agentKey, []string{"a", "b"}, requires)
 	validateAppReverted(suite.Ctx, suite.ManagedAgentClient, &app, agentKey, requires, suite.T())
@@ -169,18 +163,18 @@ func (suite *CacheTestSuite) Test_RevertManagedClusterOfflineChanges() {
 	// Agent in not running, but still make changes in the managed-cluster application manifest.
 	// When agent is restarted, these changes should be reverted to be in sync with principal.
 	requires.Eventually(func() bool {
-		return fixture.IsProcessRunning(AgentManagedName)
+		return fixture.IsProcessRunning(fixture.AgentManagedName)
 	}, 60*time.Second, 1*time.Second)
 
-	requires.NoError(fixture.StopProcess(AgentManagedName))
+	requires.NoError(fixture.StopProcess(fixture.AgentManagedName))
 	requires.Eventually(func() bool {
-		return !fixture.IsProcessRunning(AgentManagedName)
+		return !fixture.IsProcessRunning(fixture.AgentManagedName)
 	}, 60*time.Second, 1*time.Second)
 
 	updateAppInfo(suite.Ctx, suite.ManagedAgentClient, agentKey, []string{"a", "b"}, requires)
 
-	requires.NoError(fixture.StartProcess(AgentManagedName))
-	fixture.CheckReadiness(suite.T(), AgentManagedName)
+	requires.NoError(fixture.StartProcess(fixture.AgentManagedName))
+	fixture.CheckReadiness(suite.T(), fixture.AgentManagedName)
 	validateAppReverted(suite.Ctx, suite.ManagedAgentClient, &app, agentKey, requires, suite.T())
 }
 
@@ -196,7 +190,7 @@ func createApp(ctx context.Context, client fixture.KubeClient, requires *require
 	app := argoapp.Application{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      name,
-			Namespace: AgentManagedName,
+			Namespace: fixture.AgentManagedName,
 		},
 		Spec: argoapp.ApplicationSpec{
 			Project: "default",
@@ -206,7 +200,7 @@ func createApp(ctx context.Context, client fixture.KubeClient, requires *require
 				Path:           "kustomize-guestbook",
 			},
 			Destination: argoapp.ApplicationDestination{
-				Server:    "https://kubernetes.default.svc",
+				Server:    fixture.AgentClusterServerURL,
 				Namespace: namespace,
 			},
 			SyncPolicy: &argoapp.SyncPolicy{

test/e2e/clusterinfo_test.go

@@ -42,31 +42,31 @@ func TestClusterTestSuite(t *testing.T) {
 }
 
 func (suite *ClusterInfoTestSuite) SetupTest() {
-	if !fixture.IsProcessRunning(PrincipalName) {
+	if !fixture.IsProcessRunning(fixture.PrincipalName) {
 		// Start the principal if it is not running and wait for it to be ready
-		suite.Require().NoError(fixture.StartProcess(PrincipalName))
-		fixture.CheckReadiness(suite.T(), PrincipalName)
+		suite.Require().NoError(fixture.StartProcess(fixture.PrincipalName))
+		fixture.CheckReadiness(suite.T(), fixture.PrincipalName)
 	} else {
 		// If principal is already running, verify that it is ready
-		fixture.CheckReadiness(suite.T(), PrincipalName)
+		fixture.CheckReadiness(suite.T(), fixture.PrincipalName)
 	}
 
-	if !fixture.IsProcessRunning(AgentManagedName) {
+	if !fixture.IsProcessRunning(fixture.AgentManagedName) {
 		// Start the agent if it is not running and wait for it to be ready
-		suite.Require().NoError(fixture.StartProcess(AgentManagedName))
-		fixture.CheckReadiness(suite.T(), AgentManagedName)
+		suite.Require().NoError(fixture.StartProcess(fixture.AgentManagedName))
+		fixture.CheckReadiness(suite.T(), fixture.AgentManagedName)
 	} else {
 		// If agent is already running, verify that it is ready
-		fixture.CheckReadiness(suite.T(), AgentManagedName)
+		fixture.CheckReadiness(suite.T(), fixture.AgentManagedName)
 	}
 
-	if !fixture.IsProcessRunning(AgentAutonomousName) {
+	if !fixture.IsProcessRunning(fixture.AgentAutonomousName) {
 		// Start the agent if it is not running and wait for it to be ready
-		suite.Require().NoError(fixture.StartProcess(AgentAutonomousName))
-		fixture.CheckReadiness(suite.T(), AgentAutonomousName)
+		suite.Require().NoError(fixture.StartProcess(fixture.AgentAutonomousName))
+		fixture.CheckReadiness(suite.T(), fixture.AgentAutonomousName)
 	} else {
 		// If agent is already running, verify that it is ready
-		fixture.CheckReadiness(suite.T(), AgentAutonomousName)
+		fixture.CheckReadiness(suite.T(), fixture.AgentAutonomousName)
 	}
 
 	suite.BaseSuite.SetupTest()
@@ -81,33 +81,33 @@ func (suite *ClusterInfoTestSuite) Test_ClusterInfo_Managed() {
 	requires.Eventually(func() bool {
 		return fixture.HasConnectionStatus(os.Getenv(fixture.ManagedAgentServerKey), appv1.ConnectionState{
 			Status:  appv1.ConnectionStatusSuccessful,
-			Message: fmt.Sprintf(message, AgentManagedName, "connected"),
+			Message: fmt.Sprintf(message, fixture.AgentManagedName, "connected"),
 		})
 	}, 30*time.Second, 1*time.Second)
 
 	// Stop the agent
-	err := fixture.StopProcess(AgentManagedName)
+	err := fixture.StopProcess(fixture.AgentManagedName)
 	requires.NoError(err)
 
 	// Verify that connection status is updated when agent is disconnected
 	requires.Eventually(func() bool {
 		return fixture.HasConnectionStatus(os.Getenv(fixture.ManagedAgentServerKey), appv1.ConnectionState{
 			Status:     appv1.ConnectionStatusFailed,
-			Message:    fmt.Sprintf(message, AgentManagedName, "disconnected"),
+			Message:    fmt.Sprintf(message, fixture.AgentManagedName, "disconnected"),
 			ModifiedAt: &metav1.Time{Time: time.Now()},
 		})
 	}, 30*time.Second, 1*time.Second)
 
 	// Restart the agent
-	err = fixture.StartProcess(AgentManagedName)
+	err = fixture.StartProcess(fixture.AgentManagedName)
 	requires.NoError(err)
-	fixture.CheckReadiness(suite.T(), AgentManagedName)
+	fixture.CheckReadiness(suite.T(), fixture.AgentManagedName)
 
 	// Verify that connection status is updated again when agent is re-connected
 	requires.Eventually(func() bool {
 		return fixture.HasConnectionStatus(os.Getenv(fixture.ManagedAgentServerKey), appv1.ConnectionState{
 			Status:     appv1.ConnectionStatusSuccessful,
-			Message:    fmt.Sprintf(message, AgentManagedName, "connected"),
+			Message:    fmt.Sprintf(message, fixture.AgentManagedName, "connected"),
 			ModifiedAt: &metav1.Time{Time: time.Now()},
 		})
 	}, 30*time.Second, 1*time.Second)
@@ -120,33 +120,33 @@ func (suite *ClusterInfoTestSuite) Test_ClusterInfo_Autonomous() {
 	requires.Eventually(func() bool {
 		return fixture.HasConnectionStatus(os.Getenv(fixture.AutonomousAgentServerKey), appv1.ConnectionState{
 			Status:  appv1.ConnectionStatusSuccessful,
-			Message: fmt.Sprintf(message, AgentAutonomousName, "connected"),
+			Message: fmt.Sprintf(message, fixture.AgentAutonomousName, "connected"),
 		})
 	}, 30*time.Second, 1*time.Second)
 
 	// Stop the agent
-	err := fixture.StopProcess(AgentAutonomousName)
+	err := fixture.StopProcess(fixture.AgentAutonomousName)
 	requires.NoError(err)
 
 	// Verify that connection status is updated when agent is disconnected
 	requires.Eventually(func() bool {
 		return fixture.HasConnectionStatus(os.Getenv(fixture.AutonomousAgentServerKey), appv1.ConnectionState{
 			Status:     appv1.ConnectionStatusFailed,
-			Message:    fmt.Sprintf(message, AgentAutonomousName, "disconnected"),
+			Message:    fmt.Sprintf(message, fixture.AgentAutonomousName, "disconnected"),
 			ModifiedAt: &metav1.Time{Time: time.Now()},
 		})
 	}, 30*time.Second, 1*time.Second)
 
 	// Restart the agent
-	err = fixture.StartProcess(AgentAutonomousName)
+	err = fixture.StartProcess(fixture.AgentAutonomousName)
 	requires.NoError(err)
-	fixture.CheckReadiness(suite.T(), AgentAutonomousName)
+	fixture.CheckReadiness(suite.T(), fixture.AgentAutonomousName)
 
 	// Verify that connection status is updated again when agent is re-connected
 	requires.Eventually(func() bool {
 		return fixture.HasConnectionStatus(os.Getenv(fixture.AutonomousAgentServerKey), appv1.ConnectionState{
 			Status:     appv1.ConnectionStatusSuccessful,
-			Message:    fmt.Sprintf(message, AgentAutonomousName, "connected"),
+			Message:    fmt.Sprintf(message, fixture.AgentAutonomousName, "connected"),
 			ModifiedAt: &metav1.Time{Time: time.Now()},
 		})
 	}, 30*time.Second, 1*time.Second)
@@ -222,17 +222,17 @@ func (suite *ClusterInfoTestSuite) Test_ClusterCacheInfo() {
 	requires.Eventually(func() bool {
 		return fixture.HasConnectionStatus(os.Getenv(fixture.ManagedAgentServerKey), appv1.ConnectionState{
 			Status:  appv1.ConnectionStatusSuccessful,
-			Message: fmt.Sprintf(message, AgentManagedName, "connected"),
+			Message: fmt.Sprintf(message, fixture.AgentManagedName, "connected"),
 		})
 	}, 30*time.Second, 1*time.Second)
 
 	// Step 8:
 	// Disconnect agent and verify that connection status is changed to Failed
-	requires.NoError(fixture.StopProcess(AgentManagedName))
+	requires.NoError(fixture.StopProcess(fixture.AgentManagedName))
 	requires.Eventually(func() bool {
 		return fixture.HasConnectionStatus(os.Getenv(fixture.ManagedAgentServerKey), appv1.ConnectionState{
 			Status:     appv1.ConnectionStatusFailed,
-			Message:    fmt.Sprintf(message, AgentManagedName, "disconnected"),
+			Message:    fmt.Sprintf(message, fixture.AgentManagedName, "disconnected"),
 			ModifiedAt: &metav1.Time{Time: time.Now()},
 		})
 	}, 60*time.Second, 2*time.Second)
@@ -249,13 +249,13 @@ func (suite *ClusterInfoTestSuite) Test_ClusterCacheInfo() {
 
 	// Step 10:
 	// Reconnect agent and verify that connection status and cluster cache info are updated again with correct values
-	requires.NoError(fixture.StartProcess(AgentManagedName))
-	fixture.CheckReadiness(suite.T(), AgentManagedName)
+	requires.NoError(fixture.StartProcess(fixture.AgentManagedName))
+	fixture.CheckReadiness(suite.T(), fixture.AgentManagedName)
 
 	requires.Eventually(func() bool {
 		return fixture.HasConnectionStatus(os.Getenv(fixture.ManagedAgentServerKey), appv1.ConnectionState{
 			Status:     appv1.ConnectionStatusSuccessful,
-			Message:    fmt.Sprintf(message, AgentManagedName, "connected"),
+			Message:    fmt.Sprintf(message, fixture.AgentManagedName, "connected"),
 			ModifiedAt: &metav1.Time{Time: time.Now()},
 		})
 	}, 60*time.Second, 2*time.Second)

test/e2e/fixture/cluster.go

@@ -33,6 +33,11 @@ const (
 	PrincipalRedisServerPasswordKey    = "AGENT_E2E_PRINCIPAL_REDIS_PASSWORD"
 	AgentManagedRedisServerAddressKey  = "AGENT_E2E_AGENT_MANAGED_REDIS_SERVER_ADDRESS"
 	AgentManagedRedisServerPasswordKey = "AGENT_E2E_AGENT_MANAGED_REDIS_PASSWORD"
+
+	PrincipalName         = "principal"
+	AgentManagedName      = "agent-managed"
+	AgentAutonomousName   = "agent-autonomous"
+	AgentClusterServerURL = "https://kubernetes.default.svc"
 )
 
 // HasConnectionStatus checks if the connection info for a given server matches
@@ -57,7 +62,7 @@ func HasConnectionStatus(serverName string, expected appv1.ConnectionState) bool
 func HasClusterCacheInfoSynced(serverName string) bool {
 	principalClusterInfo, err := GetPrincipalClusterInfo(serverName)
 	if err != nil {
-		fmt.Println("HasConnectionStatus: error", err)
+		fmt.Println("HasClusterCacheInfoSynced: error", err)
 		return false
 	}
 
@@ -82,6 +87,7 @@ func HasApplicationsCount(expected int64) bool {
 		fmt.Println("HasApplicationsCount: error", err)
 		return false
 	}
+	fmt.Printf("HasApplicationsCount expected: %d, actual: %d\n", expected, agentCacheInfo.ApplicationsCount)
 	return agentCacheInfo.ApplicationsCount == expected
 }
 
@@ -100,19 +106,10 @@ func verifyConnectionModificationTime(actualTime *metav1.Time, expectedTime *met
 
 // GetManagedAgentClusterInfo retrieves cluster info from managed agent's Redis server
 func GetManagedAgentClusterInfo() (appv1.ClusterInfo, error) {
-	// Create Redis client and cache
-	redisOptions := &redis.Options{Addr: os.Getenv(AgentManagedRedisServerAddressKey),
-		Password: os.Getenv(AgentManagedRedisServerPasswordKey)}
-
-	redisClient := redis.NewClient(redisOptions)
-	cache := appstatecache.NewCache(cacheutil.NewCache(
-		cacheutil.NewRedisCache(redisClient, time.Minute, cacheutil.RedisCompressionGZip)), time.Minute)
-
-	clusterServer := "https://kubernetes.default.svc"
 
 	// Fetch cluster info from redis cache
 	clusterInfo := appv1.ClusterInfo{}
-	err := cache.GetClusterInfo(clusterServer, &clusterInfo)
+	err := getCacheInstance(AgentManagedName).GetClusterInfo(AgentClusterServerURL, &clusterInfo)
 	if err != nil {
 		// Treat missing cache key error (means no apps exist yet) as zero-value info
 		if err == cacheutil.ErrCacheMiss {
@@ -127,17 +124,28 @@ func GetManagedAgentClusterInfo() (appv1.ClusterInfo, error) {
 
 // GetPrincipalClusterInfo retrieves cluster info from principal's Redis server
 func GetPrincipalClusterInfo(server string) (appv1.ClusterInfo, error) {
-	// Create Redis client and cache
-	redisOptions := &redis.Options{Addr: os.Getenv(PrincipalRedisServerAddressKey),
-		Password: os.Getenv(PrincipalRedisServerPasswordKey)}
+	var clusterInfo appv1.ClusterInfo
+	err := getCacheInstance(PrincipalName).GetClusterInfo(server, &clusterInfo)
+	return clusterInfo, err
+}
+
+// getCacheInstance creates a new cache instance for the given source
+func getCacheInstance(source string) *appstatecache.Cache {
+	redisOptions := &redis.Options{}
+	switch source {
+	case PrincipalName:
+		redisOptions.Addr = os.Getenv(PrincipalRedisServerAddressKey)
+		redisOptions.Password = os.Getenv(PrincipalRedisServerPasswordKey)
+	case AgentManagedName:
+		redisOptions.Addr = os.Getenv(AgentManagedRedisServerAddressKey)
+		redisOptions.Password = os.Getenv(AgentManagedRedisServerPasswordKey)
+	default:
+		panic(fmt.Sprintf("invalid source: %s", source))
+	}
 
 	redisClient := redis.NewClient(redisOptions)
 	cache := appstatecache.NewCache(cacheutil.NewCache(
 		cacheutil.NewRedisCache(redisClient, 0, cacheutil.RedisCompressionGZip)), 0)
 
-	// fetch cluster info
-	var clusterInfo appv1.ClusterInfo
-	err := cache.GetClusterInfo(server, &clusterInfo)
-
-	return clusterInfo, err
+	return cache
 }

test/e2e/fixture/fixture.go

@@ -87,7 +87,7 @@ func (suite *BaseSuite) SetupTest() {
 			suite.T().Log(err)
 		}
 		return err == nil && len(project.Annotations) > 0 && project.Annotations["created"] == now
-	}, 10*time.Second, 1*time.Second)
+	}, 30*time.Second, 1*time.Second)
 
 	suite.T().Logf("Test begun at: %v", time.Now())
 }
@@ -280,5 +280,15 @@ func CleanUp(ctx context.Context, principalClient KubeClient, managedAgentClient
 		return err
 	}
 
+	return resetManagedAgentClusterInfo()
+}
+
+// resetManagedAgentClusterInfo resets the cluster info in the redis cache for the managed agent
+func resetManagedAgentClusterInfo() error {
+	// Reset cluster info in redis cache
+	if err := getCacheInstance(AgentManagedName).SetClusterInfo(AgentClusterServerURL, &argoapp.ClusterInfo{}); err != nil {
+		fmt.Println("resetManagedAgentClusterInfo: error", err)
+		return err
+	}
 	return nil
 }

test/e2e/fixture/toxyproxy.go

@@ -123,5 +123,5 @@ func CheckReadiness(t require.TestingT, compName string) {
 		}
 		defer resp.Body.Close()
 		return resp.StatusCode == http.StatusOK
-	}, 60*time.Second, 1*time.Second)
+	}, 120*time.Second, 2*time.Second)
 }