feat: add 'argocd-util cluster shards' command that prints shards sta…

…tistics (#6353)

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

cmd/argocd-util/commands/cluster.go

@@ -3,6 +3,7 @@ package commands
 import (
 	"context"
 	"fmt"
+	"math"
 	"os"
 	"text/tabwriter"
 	"time"
@@ -43,11 +44,79 @@ func NewClusterCommand(pathOpts *clientcmd.PathOptions) *cobra.Command {
 	command.AddCommand(NewClusterConfig())
 	command.AddCommand(NewGenClusterConfigCommand(pathOpts))
 	command.AddCommand(NewClusterStatsCommand())
+	command.AddCommand(NewClusterShardsCommand())
 
 	return command
 }
 
-func NewClusterStatsCommand() *cobra.Command {
+type ClusterWithInfo struct {
+	argoappv1.Cluster
+	Shard int
+}
+
+func loadClusters(kubeClient *kubernetes.Clientset, replicas int, namespace string, portForwardRedis bool, cacheSrc func() (*appstatecache.Cache, error), shard int) ([]ClusterWithInfo, error) {
+	settingsMgr := settings.NewSettingsManager(context.Background(), kubeClient, namespace)
+
+	argoDB := db.NewDB(namespace, settingsMgr, kubeClient)
+	clustersList, err := argoDB.ListClusters(context.Background())
+	if err != nil {
+		return nil, err
+	}
+	var cache *appstatecache.Cache
+	if portForwardRedis {
+		overrides := clientcmd.ConfigOverrides{}
+		port, err := kubeutil.PortForward("app.kubernetes.io/name=argocd-redis-ha-haproxy", 6379, namespace, &overrides)
+		if err != nil {
+			return nil, err
+		}
+		client := redis.NewClient(&redis.Options{Addr: fmt.Sprintf("localhost:%d", port)})
+		cache = appstatecache.NewCache(cacheutil.NewCache(cacheutil.NewRedisCache(client, time.Hour)), time.Hour)
+	} else {
+		cache, err = cacheSrc()
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	clusters := make([]ClusterWithInfo, len(clustersList.Items))
+	batchSize := 10
+	batchesCount := int(math.Ceil(float64(len(clusters)) / float64(batchSize)))
+	for batchNum := 0; batchNum < batchesCount; batchNum++ {
+		batchStart := batchSize * batchNum
+		batchEnd := batchSize * (batchNum + 1)
+		if batchEnd > len(clustersList.Items) {
+			batchEnd = len(clustersList.Items)
+		}
+		batch := clustersList.Items[batchStart:batchEnd]
+		_ = kube.RunAllAsync(len(batch), func(i int) error {
+			cluster := batch[i]
+			clusterShard := 0
+			if replicas > 0 {
+				clusterShard = sharding.GetShardByID(cluster.ID, replicas)
+			}
+
+			if shard != -1 && clusterShard != shard {
+				return nil
+			}
+
+			_ = cache.GetClusterInfo(cluster.Server, &cluster.Info)
+			clusters[batchStart+i] = ClusterWithInfo{cluster, clusterShard}
+			return nil
+		})
+	}
+	return clusters, nil
+}
+
+func getControllerReplicas(kubeClient *kubernetes.Clientset, namespace string) (int, error) {
+	controllerPods, err := kubeClient.CoreV1().Pods(namespace).List(context.Background(), v1.ListOptions{
+		LabelSelector: "app.kubernetes.io/name=argocd-application-controller"})
+	if err != nil {
+		return 0, err
+	}
+	return len(controllerPods.Items), nil
+}
+
+func NewClusterShardsCommand() *cobra.Command {
 	var (
 		shard            int
 		replicas         int
@@ -56,57 +125,92 @@ func NewClusterStatsCommand() *cobra.Command {
 		portForwardRedis bool
 	)
 	var command = cobra.Command{
-		Use:   "stats",
-		Short: "Prints information cluster statistics and inferred shard number",
+		Use:   "shards",
+		Short: "Print information about each controller shard and portion of Kubernetes resources it is responsible for.",
 		Run: func(cmd *cobra.Command, args []string) {
 			log.SetLevel(log.WarnLevel)
 
 			clientCfg, err := clientConfig.ClientConfig()
 			errors.CheckError(err)
 			namespace, _, err := clientConfig.Namespace()
 			errors.CheckError(err)
-
 			kubeClient := kubernetes.NewForConfigOrDie(clientCfg)
+
 			if replicas == 0 {
-				controllerPods, err := kubeClient.CoreV1().Pods(namespace).List(context.Background(), v1.ListOptions{
-					LabelSelector: "app.kubernetes.io/name=argocd-application-controller"})
+				replicas, err = getControllerReplicas(kubeClient, namespace)
 				errors.CheckError(err)
-				replicas = len(controllerPods.Items)
+			}
+			if replicas == 0 {
+				return
 			}
 
-			settingsMgr := settings.NewSettingsManager(context.Background(), kubeClient, namespace)
+			clusters, err := loadClusters(kubeClient, replicas, namespace, portForwardRedis, cacheSrc, shard)
+			errors.CheckError(err)
+			if len(clusters) == 0 {
+				return
+			}
 
-			argoDB := db.NewDB(namespace, settingsMgr, kubeClient)
-			clusters, err := argoDB.ListClusters(context.Background())
+			printStatsSummary(clusters)
+		},
+	}
+	clientConfig = cli.AddKubectlFlagsToCmd(&command)
+	command.Flags().IntVar(&shard, "shard", -1, "Cluster shard filter")
+	command.Flags().IntVar(&replicas, "replicas", 0, "Application controller replicas count. Inferred from number of running controller pods if not specified")
+	command.Flags().BoolVar(&portForwardRedis, "port-forward-redis", true, "Automatically port-forward ha proxy redis from current namespace?")
+	cacheSrc = appstatecache.AddCacheFlagsToCmd(&command)
+	return &command
+}
+
+func printStatsSummary(clusters []ClusterWithInfo) {
+	totalResourcesCount := int64(0)
+	resourcesCountByShard := map[int]int64{}
+	for _, c := range clusters {
+		totalResourcesCount += c.Info.CacheInfo.ResourcesCount
+		resourcesCountByShard[c.Shard] += c.Info.CacheInfo.ResourcesCount
+	}
+
+	avgResourcesByShard := totalResourcesCount / int64(len(resourcesCountByShard))
+	w := tabwriter.NewWriter(os.Stdout, 0, 0, 2, ' ', 0)
+	_, _ = fmt.Fprintf(w, "SHARD\tRESOURCES COUNT\n")
+	for shard := 0; shard < len(resourcesCountByShard); shard++ {
+		cnt := resourcesCountByShard[shard]
+		percent := (float64(cnt) / float64(avgResourcesByShard)) * 100.0
+		_, _ = fmt.Fprintf(w, "%d\t%s\n", shard, fmt.Sprintf("%d (%.0f%%)", cnt, percent))
+	}
+	_ = w.Flush()
+}
+
+func NewClusterStatsCommand() *cobra.Command {
+	var (
+		shard            int
+		replicas         int
+		clientConfig     clientcmd.ClientConfig
+		cacheSrc         func() (*appstatecache.Cache, error)
+		portForwardRedis bool
+	)
+	var command = cobra.Command{
+		Use:   "stats",
+		Short: "Prints information cluster statistics and inferred shard number",
+		Run: func(cmd *cobra.Command, args []string) {
+			log.SetLevel(log.WarnLevel)
+
+			clientCfg, err := clientConfig.ClientConfig()
 			errors.CheckError(err)
-			var cache *appstatecache.Cache
-			if portForwardRedis {
-				overrides := clientcmd.ConfigOverrides{}
-				port, err := kubeutil.PortForward("app.kubernetes.io/name=argocd-redis-ha-haproxy", 6379, namespace, &overrides)
-				errors.CheckError(err)
-				client := redis.NewClient(&redis.Options{Addr: fmt.Sprintf("localhost:%d", port)})
-				cache = appstatecache.NewCache(cacheutil.NewCache(cacheutil.NewRedisCache(client, time.Hour)), time.Hour)
-			} else {
-				cache, err = cacheSrc()
+			namespace, _, err := clientConfig.Namespace()
+			errors.CheckError(err)
+
+			kubeClient := kubernetes.NewForConfigOrDie(clientCfg)
+			if replicas == 0 {
+				replicas, err = getControllerReplicas(kubeClient, namespace)
 				errors.CheckError(err)
 			}
+			clusters, err := loadClusters(kubeClient, replicas, namespace, portForwardRedis, cacheSrc, shard)
+			errors.CheckError(err)
 
 			w := tabwriter.NewWriter(os.Stdout, 0, 0, 2, ' ', 0)
 			_, _ = fmt.Fprintf(w, "SERVER\tSHARD\tCONNECTION\tAPPS COUNT\tRESOURCES COUNT\n")
-
-			for _, cluster := range clusters.Items {
-				clusterShard := 0
-				if replicas > 0 {
-					clusterShard = sharding.GetShardByID(cluster.ID, replicas)
-				}
-
-				if shard != -1 && clusterShard != shard {
-					continue
-				}
-
-				var info argoappv1.ClusterInfo
-				_ = cache.GetClusterInfo(cluster.Server, &info)
-				_, _ = fmt.Fprintf(w, "%s\t%d\t%s\t%d\t%d\n", cluster.Server, clusterShard, info.ConnectionState.Status, info.ApplicationsCount, info.CacheInfo.ResourcesCount)
+			for _, cluster := range clusters {
+				_, _ = fmt.Fprintf(w, "%s\t%d\t%s\t%d\t%d\n", cluster.Server, cluster.Shard, cluster.Info.ConnectionState.Status, cluster.Info.ApplicationsCount, cluster.Info.CacheInfo.ResourcesCount)
 			}
 			_ = w.Flush()
 		},
@@ -176,6 +280,7 @@ func NewGenClusterConfigCommand(pathOpts *clientcmd.PathOptions) *cobra.Command
 			clstContext := cfgAccess.Contexts[contextName]
 			if clstContext == nil {
 				log.Fatalf("Context %s does not exist in kubeconfig", contextName)
+				return
 			}
 
 			overrides := clientcmd.ConfigOverrides{

docs/operator-manual/server-commands/argocd-util_cluster.md

@@ -17,5 +17,6 @@ argocd-util cluster [flags]
 * [argocd-util](argocd-util.md)	 - argocd-util tools used by Argo CD
 * [argocd-util cluster generate-spec](argocd-util_cluster_generate-spec.md)	 - Generate declarative config for a cluster
 * [argocd-util cluster kubeconfig](argocd-util_cluster_kubeconfig.md)	 - Generates kubeconfig for the specified cluster
+* [argocd-util cluster shards](argocd-util_cluster_shards.md)	 - Print information about each controller shard and portion of Kubernetes resources it is responsible for.
 * [argocd-util cluster stats](argocd-util_cluster_stats.md)	 - Prints information cluster statistics and inferred shard number
 

docs/operator-manual/server-commands/argocd-util_cluster_shards.md

@@ -0,0 +1,44 @@
+## argocd-util cluster shards
+
+Print information about each controller shard and portion of Kubernetes resources it is responsible for.
+
+```
+argocd-util cluster shards [flags]
+```
+
+### Options
+
+```
+      --app-state-cache-expiration duration   Cache expiration for app state (default 1h0m0s)
+      --as string                             Username to impersonate for the operation
+      --as-group stringArray                  Group to impersonate for the operation, this flag can be repeated to specify multiple groups.
+      --certificate-authority string          Path to a cert file for the certificate authority
+      --client-certificate string             Path to a client certificate file for TLS
+      --client-key string                     Path to a client key file for TLS
+      --cluster string                        The name of the kubeconfig cluster to use
+      --context string                        The name of the kubeconfig context to use
+      --default-cache-expiration duration     Cache expiration default (default 24h0m0s)
+  -h, --help                                  help for shards
+      --insecure-skip-tls-verify              If true, the server's certificate will not be checked for validity. This will make your HTTPS connections insecure
+      --kubeconfig string                     Path to a kube config. Only required if out-of-cluster
+  -n, --namespace string                      If present, the namespace scope for this CLI request
+      --password string                       Password for basic authentication to the API server
+      --port-forward-redis                    Automatically port-forward ha proxy redis from current namespace? (default true)
+      --redis string                          Redis server hostname and port (e.g. argocd-redis:6379). 
+      --redisdb int                           Redis database.
+      --replicas int                          Application controller replicas count. Inferred from number of running controller pods if not specified
+      --request-timeout string                The length of time to wait before giving up on a single server request. Non-zero values should contain a corresponding time unit (e.g. 1s, 2m, 3h). A value of zero means don't timeout requests. (default "0")
+      --sentinel stringArray                  Redis sentinel hostname and port (e.g. argocd-redis-ha-announce-0:6379). 
+      --sentinelmaster string                 Redis sentinel master group name. (default "master")
+      --server string                         The address and port of the Kubernetes API server
+      --shard int                             Cluster shard filter (default -1)
+      --tls-server-name string                If provided, this name will be used to validate server certificate. If this is not provided, hostname used to contact the server is used.
+      --token string                          Bearer token for authentication to the API server
+      --user string                           The name of the kubeconfig user to use
+      --username string                       Username for basic authentication to the API server
+```
+
+### SEE ALSO
+
+* [argocd-util cluster](argocd-util_cluster.md)	 - Manage clusters configuration
+

util/kube/portforwarder.go

@@ -67,7 +67,7 @@ func PortForward(podSelector string, targetPort int, namespace string, overrides
 	out := new(bytes.Buffer)
 	errOut := new(bytes.Buffer)
 
-	ln, err := net.Listen("tcp", "[::]:0")
+	ln, err := net.Listen("tcp", "localhost:0")
 	if err != nil {
 		return -1, err
 	}