feat: reduce redis traffic caused by app resource tree updates in red…

…is (argoproj#19722)

Signed-off-by: Alexander Matyushentsev <[email protected]>

docs/operator-manual/high_availability.md

@@ -130,6 +130,10 @@ stringData:
   count (grouped by k8s api version, the granule of parallelism for list operations). In this case, all resources will
   be buffered in memory -- no api server request will be blocked by processing.
 
+* `ARGOCD_APPLICATION_TREE_SHARD_SIZE` - environment variable controlling the max number of resources stored in one Redis
+  key. Splitting application tree into multiple keys helps to reduce the amount of traffic between the controller and Redis.
+  The default value is 0, which means that the application tree is stored in a single Redis key. The reasonable value is 100.
+
 **metrics**
 
 * `argocd_app_reconcile` - reports application reconciliation duration in seconds. Can be used to build reconciliation duration heat map to get a high-level reconciliation performance picture.

pkg/apis/application/v1alpha1/types.go

@@ -1640,6 +1640,60 @@ type ApplicationTree struct {
 	OrphanedNodes []ResourceNode `json:"orphanedNodes,omitempty" protobuf:"bytes,2,rep,name=orphanedNodes"`
 	// Hosts holds list of Kubernetes nodes that run application related pods
 	Hosts []HostInfo `json:"hosts,omitempty" protobuf:"bytes,3,rep,name=hosts"`
+	// ShardsCount contains total number of shards the application tree is split into
+	ShardsCount int64 `json:"shardsCount,omitempty" protobuf:"bytes,4,opt,name=shardsCount"`
+}
+
+func (t *ApplicationTree) Merge(other *ApplicationTree) {
+	t.Nodes = append(t.Nodes, other.Nodes...)
+	t.OrphanedNodes = append(t.OrphanedNodes, other.OrphanedNodes...)
+	t.Hosts = append(t.Hosts, other.Hosts...)
+	t.Normalize()
+}
+
+// GetShards split application tree into shards with populated metadata
+func (t *ApplicationTree) GetShards(size int64) []*ApplicationTree {
+	t.Normalize()
+	if size == 0 {
+		return []*ApplicationTree{t}
+	}
+
+	var items []func(*ApplicationTree)
+	for i := range t.Nodes {
+		item := t.Nodes[i]
+		items = append(items, func(shard *ApplicationTree) {
+			shard.Nodes = append(shard.Nodes, item)
+		})
+	}
+	for i := range t.OrphanedNodes {
+		item := t.OrphanedNodes[i]
+		items = append(items, func(shard *ApplicationTree) {
+			shard.OrphanedNodes = append(shard.OrphanedNodes, item)
+		})
+	}
+	for i := range t.Hosts {
+		item := t.Hosts[i]
+		items = append(items, func(shard *ApplicationTree) {
+			shard.Hosts = append(shard.Hosts, item)
+		})
+	}
+	var shards []*ApplicationTree
+	for len(items) > 0 {
+		shard := &ApplicationTree{}
+		shards = append(shards, shard)
+		cnt := 0
+		for i := int64(0); i < size && i < int64(len(items)); i++ {
+			items[i](shard)
+			cnt++
+		}
+		items = items[cnt:]
+	}
+	if len(shards) > 0 {
+		shards[0].ShardsCount = int64(len(shards))
+	} else {
+		shards = []*ApplicationTree{{ShardsCount: 0}}
+	}
+	return shards
 }
 
 // Normalize sorts application tree nodes and hosts. The persistent order allows to

pkg/apis/application/v1alpha1/types_test.go

@@ -3889,3 +3889,73 @@ func TestApplicationSpec_GetSourcePtrByIndex(t *testing.T) {
 		})
 	}
 }
+
+func TestApplicationTree_GetShards(t *testing.T) {
+	tree := &ApplicationTree{
+		Nodes: []ResourceNode{
+			{ResourceRef: ResourceRef{Name: "node 1"}}, {ResourceRef: ResourceRef{Name: "node 2"}}, {ResourceRef: ResourceRef{Name: "node 3"}},
+		},
+		OrphanedNodes: []ResourceNode{
+			{ResourceRef: ResourceRef{Name: "orph-node 1"}}, {ResourceRef: ResourceRef{Name: "orph-node 2"}}, {ResourceRef: ResourceRef{Name: "orph-node 3"}},
+		},
+		Hosts: []HostInfo{
+			{Name: "host 1"}, {Name: "host 2"}, {Name: "host 3"},
+		},
+	}
+
+	shards := tree.GetShards(2)
+	require.Len(t, shards, 5)
+	require.Equal(t, &ApplicationTree{
+		ShardsCount: 5,
+		Nodes: []ResourceNode{
+			{ResourceRef: ResourceRef{Name: "node 1"}}, {ResourceRef: ResourceRef{Name: "node 2"}},
+		},
+	}, shards[0])
+	require.Equal(t, &ApplicationTree{
+		Nodes:         []ResourceNode{{ResourceRef: ResourceRef{Name: "node 3"}}},
+		OrphanedNodes: []ResourceNode{{ResourceRef: ResourceRef{Name: "orph-node 1"}}},
+	}, shards[1])
+	require.Equal(t, &ApplicationTree{
+		OrphanedNodes: []ResourceNode{{ResourceRef: ResourceRef{Name: "orph-node 2"}}, {ResourceRef: ResourceRef{Name: "orph-node 3"}}},
+	}, shards[2])
+	require.Equal(t, &ApplicationTree{
+		Hosts: []HostInfo{{Name: "host 1"}, {Name: "host 2"}},
+	}, shards[3])
+	require.Equal(t, &ApplicationTree{
+		Hosts: []HostInfo{{Name: "host 3"}},
+	}, shards[4])
+}
+
+func TestApplicationTree_Merge(t *testing.T) {
+	tree := &ApplicationTree{}
+	tree.Merge(&ApplicationTree{
+		ShardsCount: 5,
+		Nodes: []ResourceNode{
+			{ResourceRef: ResourceRef{Name: "node 1"}}, {ResourceRef: ResourceRef{Name: "node 2"}},
+		},
+	})
+	tree.Merge(&ApplicationTree{
+		Nodes:         []ResourceNode{{ResourceRef: ResourceRef{Name: "node 3"}}},
+		OrphanedNodes: []ResourceNode{{ResourceRef: ResourceRef{Name: "orph-node 1"}}},
+	})
+	tree.Merge(&ApplicationTree{
+		OrphanedNodes: []ResourceNode{{ResourceRef: ResourceRef{Name: "orph-node 2"}}, {ResourceRef: ResourceRef{Name: "orph-node 3"}}},
+	})
+	tree.Merge(&ApplicationTree{
+		Hosts: []HostInfo{{Name: "host 1"}, {Name: "host 2"}},
+	})
+	tree.Merge(&ApplicationTree{
+		Hosts: []HostInfo{{Name: "host 3"}},
+	})
+	require.Equal(t, &ApplicationTree{
+		Nodes: []ResourceNode{
+			{ResourceRef: ResourceRef{Name: "node 1"}}, {ResourceRef: ResourceRef{Name: "node 2"}}, {ResourceRef: ResourceRef{Name: "node 3"}},
+		},
+		OrphanedNodes: []ResourceNode{
+			{ResourceRef: ResourceRef{Name: "orph-node 1"}}, {ResourceRef: ResourceRef{Name: "orph-node 2"}}, {ResourceRef: ResourceRef{Name: "orph-node 3"}},
+		},
+		Hosts: []HostInfo{
+			{Name: "host 1"}, {Name: "host 2"}, {Name: "host 3"},
+		},
+	}, tree)
+}

util/cache/appstate/cache.go

@@ -13,7 +13,10 @@ import (
 	"github.com/argoproj/argo-cd/v2/util/env"
 )
 
-var ErrCacheMiss = cacheutil.ErrCacheMiss
+var (
+	ErrCacheMiss  = cacheutil.ErrCacheMiss
+	treeShardSize = env.ParseInt64FromEnv("ARGOCD_APPLICATION_TREE_SHARD_SIZE", 0, 0, 1000)
+)
 
 const (
 	clusterInfoCacheExpiration = 10 * time.Minute
@@ -68,16 +71,29 @@ func (c *Cache) SetAppManagedResources(appName string, managedResources []*appv1
 	return c.SetItem(appManagedResourcesKey(appName), managedResources, c.appStateCacheExpiration, managedResources == nil)
 }
 
-func appResourcesTreeKey(appName string) string {
-	return fmt.Sprintf("app|resources-tree|%s", appName)
+func appResourcesTreeKey(appName string, shard int64) string {
+	key := fmt.Sprintf("app|resources-tree|%s", appName)
+	if shard > 0 {
+		key = fmt.Sprintf("%s|%d", key, shard)
+	}
+	return key
 }
 
 func clusterInfoKey(server string) string {
 	return fmt.Sprintf("cluster|info|%s", server)
 }
 
 func (c *Cache) GetAppResourcesTree(appName string, res *appv1.ApplicationTree) error {
-	err := c.GetItem(appResourcesTreeKey(appName), &res)
+	err := c.GetItem(appResourcesTreeKey(appName, 0), &res)
+	if res.ShardsCount > 1 {
+		for i := int64(1); i < res.ShardsCount; i++ {
+			var shard appv1.ApplicationTree
+			if err = c.GetItem(appResourcesTreeKey(appName, i), &shard); err != nil {
+				return err
+			}
+			res.Merge(&shard)
+		}
+	}
 	return err
 }
 
@@ -86,13 +102,21 @@ func (c *Cache) OnAppResourcesTreeChanged(ctx context.Context, appName string, c
 }
 
 func (c *Cache) SetAppResourcesTree(appName string, resourcesTree *appv1.ApplicationTree) error {
-	if resourcesTree != nil {
-		resourcesTree.Normalize()
-	}
-	err := c.SetItem(appResourcesTreeKey(appName), resourcesTree, c.appStateCacheExpiration, resourcesTree == nil)
-	if err != nil {
-		return err
+	if resourcesTree == nil {
+		if err := c.SetItem(appResourcesTreeKey(appName, 0), resourcesTree, c.appStateCacheExpiration, true); err != nil {
+			return err
+		}
+	} else {
+		// Splitting resource tree into shards reduces number of Redis SET calls and therefore amount of traffic sent
+		// from controller to Redis. Controller still stores each shard in cache but util/cache/twolevelclient.go
+		// forwards request to Redis only if shard actually changes.
+		for i, shard := range resourcesTree.GetShards(treeShardSize) {
+			if err := c.SetItem(appResourcesTreeKey(appName, int64(i)), shard, c.appStateCacheExpiration, false); err != nil {
+				return err
+			}
+		}
 	}
+
 	return c.Cache.NotifyUpdated(appManagedResourcesKey(appName))
 }