Merge branch 'master' of https://github.com/argoproj/argo-cd into bb-bearer-token

Author: reggie-k

.golangci.yaml

@@ -66,6 +66,8 @@ linters-settings:
         pkg: k8s.io/api/rbac/v1
       - alias: apierrors
         pkg: k8s.io/apimachinery/pkg/api/errors
+      - alias: apiextensionsv1
+        pkg: k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1
       - alias: metav1
         pkg: k8s.io/apimachinery/pkg/apis/meta/v1
       - alias: informersv1

applicationset/controllers/applicationset_controller.go

@@ -768,7 +768,7 @@ func (r *ApplicationSetReconciler) deleteInCluster(ctx context.Context, logCtx *
 }
 
 // removeFinalizerOnInvalidDestination removes the Argo CD resources finalizer if the application contains an invalid target (eg missing cluster)
-func (r *ApplicationSetReconciler) removeFinalizerOnInvalidDestination(ctx context.Context, applicationSet argov1alpha1.ApplicationSet, app *argov1alpha1.Application, clusterList *argov1alpha1.ClusterList, appLog *log.Entry) error {
+func (r *ApplicationSetReconciler) removeFinalizerOnInvalidDestination(ctx context.Context, applicationSet argov1alpha1.ApplicationSet, app *argov1alpha1.Application, clusterList []utils.ClusterSpecifier, appLog *log.Entry) error {
 	// Only check if the finalizers need to be removed IF there are finalizers to remove
 	if len(app.Finalizers) == 0 {
 		return nil
@@ -783,7 +783,7 @@ func (r *ApplicationSetReconciler) removeFinalizerOnInvalidDestination(ctx conte
 	} else {
 		// Detect if the destination's server field does not match an existing cluster
 		matchingCluster := false
-		for _, cluster := range clusterList.Items {
+		for _, cluster := range clusterList {
 			if destCluster.Server != cluster.Server {
 				continue
 			}

applicationset/generators/cluster.go

@@ -70,7 +70,7 @@ func (g *ClusterGenerator) GenerateParams(appSetGenerator *argoappsetv1alpha1.Ap
 	// - Since local clusters do not have secrets, they do not have labels to match against
 	ignoreLocalClusters := len(appSetGenerator.Clusters.Selector.MatchExpressions) > 0 || len(appSetGenerator.Clusters.Selector.MatchLabels) > 0
 
-	// ListCluster from Argo CD's util/db package will include the local cluster in the list of clusters
+	// ListCluster will include the local cluster in the list of clusters
 	clustersFromArgoCD, err := utils.ListClusters(g.ctx, g.clientset, g.namespace)
 	if err != nil {
 		return nil, fmt.Errorf("error listing clusters: %w", err)
@@ -93,7 +93,7 @@ func (g *ClusterGenerator) GenerateParams(appSetGenerator *argoappsetv1alpha1.Ap
 	logCtx.Debugf("Using flat mode = %t for cluster generator", isFlatMode)
 	clustersParams := make([]map[string]any, 0)
 
-	for _, cluster := range clustersFromArgoCD.Items {
+	for _, cluster := range clustersFromArgoCD {
 		// If there is a secret for this cluster, then it's a non-local cluster, so it will be
 		// handled by the next step.
 		if secretForCluster, exists := clusterSecrets[cluster.Name]; exists {

applicationset/generators/duck_type.go

@@ -165,9 +165,6 @@ func (g *DuckTypeGenerator) GenerateParams(appSetGenerator *argoprojiov1alpha1.A
 	}
 	log.Infof("Number of decisions found: %v", len(clusterDecisions))
 
-	// Read this outside the loop to improve performance
-	argoClusters := clustersFromArgoCD.Items
-
 	if len(clusterDecisions) == 0 {
 		log.Warningf("clusterDecisionResource status.%s missing", statusListKey)
 		return nil, nil
@@ -188,7 +185,7 @@ func (g *DuckTypeGenerator) GenerateParams(appSetGenerator *argoprojiov1alpha1.A
 
 		found := false
 
-		for _, argoCluster := range argoClusters {
+		for _, argoCluster := range clustersFromArgoCD {
 			if argoCluster.Name == strMatchValue {
 				log.WithField(matchKey, argoCluster.Name).Info("matched cluster in ArgoCD")
 				params["name"] = argoCluster.Name

applicationset/utils/clusterUtils.go

@@ -3,7 +3,6 @@ package utils
 import (
 	"context"
 	"fmt"
-	"sync"
 
 	"github.com/argoproj/argo-cd/v3/common"
 	appv1 "github.com/argoproj/argo-cd/v3/pkg/apis/application/v1alpha1"
@@ -13,16 +12,14 @@ import (
 	"k8s.io/client-go/kubernetes"
 )
 
-var (
-	localCluster = appv1.Cluster{
-		Name:            "in-cluster",
-		Server:          appv1.KubernetesInternalAPIServerAddr,
-		ConnectionState: appv1.ConnectionState{Status: appv1.ConnectionStatusSuccessful},
-	}
-	initLocalCluster sync.Once
-)
+// ClusterSpecifier contains only the name and server URL of a cluster. We use this struct to avoid partially-populating
+// the full Cluster struct, which would be misleading.
+type ClusterSpecifier struct {
+	Name   string
+	Server string
+}
 
-func ListClusters(ctx context.Context, clientset kubernetes.Interface, namespace string) (*appv1.ClusterList, error) {
+func ListClusters(ctx context.Context, clientset kubernetes.Interface, namespace string) ([]ClusterSpecifier, error) {
 	clusterSecretsList, err := clientset.CoreV1().Secrets(namespace).List(ctx,
 		metav1.ListOptions{LabelSelector: common.LabelKeySecretType + "=" + common.LabelValueSecretTypeCluster})
 	if err != nil {
@@ -35,54 +32,29 @@ func ListClusters(ctx context.Context, clientset kubernetes.Interface, namespace
 
 	clusterSecrets := clusterSecretsList.Items
 
-	clusterList := appv1.ClusterList{
-		Items: make([]appv1.Cluster, len(clusterSecrets)),
-	}
+	clusterList := make([]ClusterSpecifier, len(clusterSecrets))
+
 	hasInClusterCredentials := false
 	for i, clusterSecret := range clusterSecrets {
-		// This line has changed from the original Argo CD code: now receives an error, and handles it
 		cluster, err := db.SecretToCluster(&clusterSecret)
 		if err != nil || cluster == nil {
 			return nil, fmt.Errorf("unable to convert cluster secret to cluster object '%s': %w", clusterSecret.Name, err)
 		}
-
-		// db.SecretToCluster populates these, but they're not meant to be available to the caller.
-		cluster.Labels = nil
-		cluster.Annotations = nil
-
-		clusterList.Items[i] = *cluster
+		clusterList[i] = ClusterSpecifier{
+			Name:   cluster.Name,
+			Server: cluster.Server,
+		}
 		if cluster.Server == appv1.KubernetesInternalAPIServerAddr {
 			hasInClusterCredentials = true
 		}
 	}
 	if !hasInClusterCredentials {
-		localCluster := getLocalCluster(clientset)
-		if localCluster != nil {
-			clusterList.Items = append(clusterList.Items, *localCluster)
-		}
+		// There was no secret for the in-cluster config, so we add it here. We don't fully-populate the Cluster struct,
+		// since only the name and server fields are used by the generator.
+		clusterList = append(clusterList, ClusterSpecifier{
+			Name:   "in-cluster",
+			Server: appv1.KubernetesInternalAPIServerAddr,
+		})
 	}
-	return &clusterList, nil
-}
-
-func getLocalCluster(clientset kubernetes.Interface) *appv1.Cluster {
-	initLocalCluster.Do(func() {
-		info, err := clientset.Discovery().ServerVersion()
-		if err == nil {
-			//nolint:staticcheck
-			localCluster.ServerVersion = fmt.Sprintf("%s.%s", info.Major, info.Minor)
-			//nolint:staticcheck
-			localCluster.ConnectionState = appv1.ConnectionState{Status: appv1.ConnectionStatusSuccessful}
-		} else {
-			//nolint:staticcheck
-			localCluster.ConnectionState = appv1.ConnectionState{
-				Status:  appv1.ConnectionStatusFailed,
-				Message: err.Error(),
-			}
-		}
-	})
-	cluster := localCluster.DeepCopy()
-	now := metav1.Now()
-	//nolint:staticcheck
-	cluster.ConnectionState.ModifiedAt = &now
-	return cluster
+	return clusterList, nil
 }

controller/appcontroller.go

@@ -288,9 +288,32 @@ func NewApplicationController(
 					return fmt.Errorf("application controller deployment replicas is not set or is less than 0, replicas: %d", appControllerDeployment.Spec.Replicas)
 				}
 				shard := env.ParseNumFromEnv(common.EnvControllerShard, -1, -math.MaxInt32, math.MaxInt32)
-				if _, err := sharding.GetOrUpdateShardFromConfigMap(kubeClientset.(*kubernetes.Clientset), settingsMgr, int(*appControllerDeployment.Spec.Replicas), shard); err != nil {
+				shard, err := sharding.GetOrUpdateShardFromConfigMap(kubeClientset.(*kubernetes.Clientset), settingsMgr, int(*appControllerDeployment.Spec.Replicas), shard)
+				if err != nil {
 					return fmt.Errorf("error while updating the heartbeat for to the Shard Mapping ConfigMap: %w", err)
 				}
+
+				// update the shard number in the clusterSharding, and resync all applications if the shard number is updated
+				if ctrl.clusterSharding.UpdateShard(shard) {
+					// update shard number in stateCache
+					ctrl.stateCache.UpdateShard(shard)
+
+					// resync all applications
+					apps, err := ctrl.appLister.List(labels.Everything())
+					if err != nil {
+						return err
+					}
+					for _, app := range apps {
+						if !ctrl.canProcessApp(app) {
+							continue
+						}
+						key, err := cache.MetaNamespaceKeyFunc(app)
+						if err == nil {
+							ctrl.appRefreshQueue.AddRateLimited(key)
+							ctrl.clusterSharding.AddApp(app)
+						}
+					}
+				}
 			}
 		}
 		return nil

controller/cache/cache.go

@@ -153,6 +153,8 @@ type LiveStateCache interface {
 	GetClustersInfo() []clustercache.ClusterInfo
 	// Init must be executed before cache can be used
 	Init() error
+	// UpdateShard will update the shard of ClusterSharding when the shard has changed.
+	UpdateShard(shard int) bool
 }
 
 type ObjectUpdatedHandler = func(managedByApp map[string]bool, ref corev1.ObjectReference)
@@ -906,3 +908,8 @@ func (c *liveStateCache) GetClustersInfo() []clustercache.ClusterInfo {
 func (c *liveStateCache) GetClusterCache(server *appv1.Cluster) (clustercache.ClusterCache, error) {
 	return c.getSyncedCluster(server)
 }
+
+// UpdateShard will update the shard of ClusterSharding when the shard has changed.
+func (c *liveStateCache) UpdateShard(shard int) bool {
+	return c.clusterSharding.UpdateShard(shard)
+}

controller/cache/mocks/LiveStateCache.go

@@ -19,6 +19,8 @@ type ClusterShardingCache interface {
 	UpdateApp(a *v1alpha1.Application)
 	IsManagedCluster(c *v1alpha1.Cluster) bool
 	GetDistribution() map[string]int
+	GetAppDistribution() map[string]int
+	UpdateShard(shard int) bool
 }
 
 type ClusterSharding struct {
@@ -244,3 +246,33 @@ func (sharding *ClusterSharding) UpdateApp(a *v1alpha1.Application) {
 		log.Debugf("Skipping sharding distribution update. No relevant changes")
 	}
 }
+
+// GetAppDistribution should be not be called from a DestributionFunction because
+// it could cause a deadlock when updateDistribution is called.
+func (sharding *ClusterSharding) GetAppDistribution() map[string]int {
+	sharding.lock.RLock()
+	clusters := sharding.Clusters
+	apps := sharding.Apps
+	sharding.lock.RUnlock()
+
+	appDistribution := make(map[string]int, len(clusters))
+
+	for _, a := range apps {
+		if _, ok := appDistribution[a.Spec.Destination.Server]; !ok {
+			appDistribution[a.Spec.Destination.Server] = 0
+		}
+		appDistribution[a.Spec.Destination.Server]++
+	}
+	return appDistribution
+}
+
+// UpdateShard will update the shard of ClusterSharding when the shard has changed.
+func (sharding *ClusterSharding) UpdateShard(shard int) bool {
+	if shard != sharding.Shard {
+		sharding.lock.RLock()
+		sharding.Shard = shard
+		sharding.lock.RUnlock()
+		return true
+	}
+	return false
+}

controller/sharding/cache.go

@@ -485,7 +485,10 @@ func GetClusterSharding(kubeClient kubernetes.Interface, settingsMgr *settings.S
 					err = fmt.Errorf("unable to get shard due to error updating the sharding config map: %w", err)
 					break
 				}
-				log.Warnf("conflict when getting shard from shard mapping configMap. Retrying (%d/3)", i)
+				// if `err == nil`, should not log the following warning message
+				if err != nil {
+					log.Warnf("conflict when getting shard from shard mapping configMap. Retrying (%d/3)", i)
+				}
 			}
 			errors.CheckError(err)
 		} else {

controller/sharding/sharding.go

@@ -94,7 +94,7 @@ require (
 	golang.org/x/time v0.10.0
 	google.golang.org/genproto/googleapis/api v0.0.0-20241202173237-19429a94021a
 	google.golang.org/grpc v1.70.0
-	google.golang.org/protobuf v1.36.4
+	google.golang.org/protobuf v1.36.5
 	gopkg.in/yaml.v2 v2.4.0
 	gopkg.in/yaml.v3 v3.0.1
 	k8s.io/api v0.31.0

go.mod

@@ -1382,8 +1382,8 @@ google.golang.org/protobuf v1.27.1/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQ
 google.golang.org/protobuf v1.28.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
 google.golang.org/protobuf v1.33.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=
 google.golang.org/protobuf v1.34.2/go.mod h1:qYOHts0dSfpeUzUFpOMr/WGzszTmLH+DiWniOlNbLDw=
-google.golang.org/protobuf v1.36.4 h1:6A3ZDJHn/eNqc1i+IdefRzy/9PokBTPvcqMySR7NNIM=
-google.golang.org/protobuf v1.36.4/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=
+google.golang.org/protobuf v1.36.5 h1:tPhr+woSbjfYvY6/GPufUoYizxw1cF/yFoxJ2fmpwlM=
+google.golang.org/protobuf v1.36.5/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=
 gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
 gopkg.in/alexcesaro/quotedprintable.v3 v3.0.0-20150716171945-2caba252f4dc h1:2gGKlE2+asNV9m7xrywl36YYNnBG5ZQ0r/BOOxqPpmk=
 gopkg.in/alexcesaro/quotedprintable.v3 v3.0.0-20150716171945-2caba252f4dc/go.mod h1:m7x9LTH6d71AHyAX77c9yqWCCa3UKHcVEj9y7hAtKDk=

go.sum

@@ -11,7 +11,7 @@ import (
 	"github.com/argoproj/argo-cd/v3/pkg/apis/application"
 
 	"github.com/argoproj/gitops-engine/pkg/utils/kube"
-	extensionsobj "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
+	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"sigs.k8s.io/yaml"
 )
@@ -22,7 +22,7 @@ var kindToCRDPath = map[string]string{
 	application.ApplicationSetFullName: "manifests/crds/applicationset-crd.yaml",
 }
 
-func getCustomResourceDefinitions() map[string]*extensionsobj.CustomResourceDefinition {
+func getCustomResourceDefinitions() map[string]*apiextensionsv1.CustomResourceDefinition {
 	crdYamlBytes, err := exec.Command(
 		"controller-gen",
 		"paths=./pkg/apis/application/...",
@@ -41,7 +41,7 @@ func getCustomResourceDefinitions() map[string]*extensionsobj.CustomResourceDefi
 
 	objs, err := kube.SplitYAML(crdYamlBytes)
 	checkErr(err)
-	crds := make(map[string]*extensionsobj.CustomResourceDefinition)
+	crds := make(map[string]*apiextensionsv1.CustomResourceDefinition)
 	for i := range objs {
 		un := objs[i]
 
@@ -80,14 +80,14 @@ func removeValidation(un *unstructured.Unstructured, path string) {
 	unstructured.RemoveNestedField(un.Object, schemaPath...)
 }
 
-func toCRD(un *unstructured.Unstructured, removeDesc bool) *extensionsobj.CustomResourceDefinition {
+func toCRD(un *unstructured.Unstructured, removeDesc bool) *apiextensionsv1.CustomResourceDefinition {
 	if removeDesc {
 		removeDescription(un.Object)
 	}
 	unBytes, err := json.Marshal(un)
 	checkErr(err)
 
-	var crd extensionsobj.CustomResourceDefinition
+	var crd apiextensionsv1.CustomResourceDefinition
 	err = json.Unmarshal(unBytes, &crd)
 	checkErr(err)
 
@@ -134,7 +134,7 @@ func main() {
 	}
 }
 
-func writeCRDintoFile(crd *extensionsobj.CustomResourceDefinition, path string) {
+func writeCRDintoFile(crd *apiextensionsv1.CustomResourceDefinition, path string) {
 	jsonBytes, err := json.Marshal(crd)
 	checkErr(err)
 

hack/gen-crd-spec/main.go

@@ -104,13 +104,15 @@ RUN userdel -r ubuntu && \
     HOME=/home/user git config --global user.name "ArgoCD Test User" && \
     HOME=/home/user git config --global user.email "[email protected]" && \
     HOME=/home/user git config --global --add safe.directory '*' && \
+    mkdir -p /go/src && \
     mkdir -p /go/pkg && \
+    chown -R user:user /go && \
     mkdir -p /var/run/sshd && \
     mkdir -p /root/.ssh && \
-    mkdir -p /go && \
     chown root /etc/ssh/ssh_host_*_key* && \
     chmod 0600 /etc/ssh/ssh_host_*_key && \
     mkdir -p /tmp/go-build-cache && \
+    chown -R user:user /tmp/go-build-cache && \
     ln -s /usr/local/bin/node /usr/local/bin/nodejs && \
     ln -s /usr/local/lib/node_modules/npm/bin/npm-cli.js /usr/local/bin/npm && \
     ln -s /usr/local/lib/node_modules/npm/bin/npx-cli.js /usr/local/bin/npx && \

test/container/Dockerfile

@@ -1,6 +1,6 @@
 ARG BASE_IMAGE=docker.io/library/ubuntu:24.04@sha256:80dd3c3b9c6cecb9f1667e9290b3bc61b78c2678c02cbdae5f0fea92cc6734ab
 
-FROM docker.io/library/golang:1.23.5@sha256:51a6466e8dbf3e00e422eb0f7a97ac450b2d57b33617bbe8d2ee0bddcd9d0d37 AS go
+FROM docker.io/library/golang:1.23.6@sha256:927112936d6b496ed95f55f362cc09da6e3e624ef868814c56d55bd7323e0959 AS go
 
 RUN go install github.com/mattn/goreman@latest && \
     go install github.com/kisielk/godepgraph@latest