Refactor follower behavior for ClusterRepos handler (rancher#54161)

Author: aruiz14

pkg/catalogv2/git/download.go

@@ -8,7 +8,8 @@ import (
 	corev1 "k8s.io/api/core/v1"
 )
 
-// Ensure runs git clone, clean DIRTY contents and fetch the latest commit
+// Ensure the repository is checked out at the provided commit reference.
+// Cloning/fetching and cleaning a dirty state is performed if necessary
 func Ensure(secret *corev1.Secret, namespace, name, gitURL, commit string, insecureSkipTLS bool, caBundle []byte) error {
 	git, err := gitForRepo(secret, namespace, name, gitURL, insecureSkipTLS, caBundle)
 	if err != nil {

pkg/controllers/dashboard/helm/repo.go

@@ -8,8 +8,10 @@ import (
 	"errors"
 	"fmt"
 	"os"
+	"slices"
 	"time"
 
+	"github.com/rancher/wrangler/v3/pkg/genericcondition"
 	"k8s.io/apimachinery/pkg/api/equality"
 
 	catalog "github.com/rancher/rancher/pkg/apis/catalog.cattle.io/v1"
@@ -70,6 +72,7 @@ func RegisterRepos(ctx context.Context,
 
 }
 
+// RegisterReposForFollowers need to run on every replica (including the leader), see comment on EnsureRepositoryCheckout
 func RegisterReposForFollowers(ctx context.Context,
 	secrets corev1controllers.SecretCache,
 	clusterRepos catalogcontrollers.ClusterRepoController) {
@@ -78,23 +81,26 @@ func RegisterReposForFollowers(ctx context.Context,
 		clusterRepos: clusterRepos,
 	}
 
-	catalogcontrollers.RegisterClusterRepoStatusHandler(ctx, clusterRepos,
-		condition.Cond(catalog.FollowerRepoDownloaded), "helm-clusterrepo-ensure", h.ClusterRepoDownloadEnsureStatusHandler)
-
+	clusterRepos.OnChange(ctx, "helm-clusterrepo-ensure", h.EnsureRepositoryCheckout)
 }
 
-func (r *repoHandler) ClusterRepoDownloadEnsureStatusHandler(repo *catalog.ClusterRepo, status catalog.RepoStatus) (catalog.RepoStatus, error) {
+// EnsureRepositoryCheckout will fetch and/or initialize the necessary data to ensure the corresponding git repositories match commit in the status
+// .status.commit is only updated by the leader, but this needs to run in all replicas, including the leader despite being redundant, to ensure repositories are correctly initialized after a restart
+func (r *repoHandler) EnsureRepositoryCheckout(_ string, repo *catalog.ClusterRepo) (*catalog.ClusterRepo, error) {
+	if repo == nil {
+		return nil, nil
+	}
 	if registry.IsOCI(repo.Spec.URL) {
-		return status, nil
+		return repo, nil
 	}
 
-	interval := defaultInterval
-	if repo.Spec.RefreshInterval > 0 {
-		interval = time.Duration(repo.Spec.RefreshInterval) * time.Second
+	if isDisabled(repo) || repo.Status.Commit == "" {
+		return repo, nil
 	}
 
-	r.clusterRepos.EnqueueAfter(repo.Name, interval)
-	return r.ensure(&repo.Spec, status, &repo.ObjectMeta)
+	// Sync the followers' repository copy to the commit observed by the leader
+	err := r.ensure(repo)
+	return repo, err
 }
 
 func (r *repoHandler) ClusterRepoOnChange(key string, repo *catalog.ClusterRepo) (*catalog.ClusterRepo, error) {
@@ -119,6 +125,8 @@ func (r *repoHandler) ClusterRepoOnChange(key string, repo *catalog.ClusterRepo)
 	}
 
 	newStatus := repo.Status.DeepCopy()
+	removeLegacyCondition(newStatus)
+
 	retryPolicy, err := getRetryPolicy(repo)
 	if err != nil {
 		err = fmt.Errorf("failed to get retry policy: %w", err)
@@ -137,9 +145,9 @@ func (r *repoHandler) ClusterRepoOnChange(key string, repo *catalog.ClusterRepo)
 	newStatus.ShouldNotSkip = false
 
 	// If repo is disabled, then don't update the clusterrepo
-	if repo.Spec.Enabled != nil && !*repo.Spec.Enabled {
+	if isDisabled(repo) {
 		logrus.Infof("skipping repo %s because it is disabled", repo.Name)
-		return setErrorCondition(repo, err, newStatus, interval, ociCondition, r.clusterRepos)
+		return setErrorCondition(repo, nil, newStatus, interval, repoCondition, r.clusterRepos)
 	}
 
 	return r.download(repo, newStatus, metav1.OwnerReference{
@@ -150,6 +158,18 @@ func (r *repoHandler) ClusterRepoOnChange(key string, repo *catalog.ClusterRepo)
 	}, interval, retryPolicy)
 }
 
+func isDisabled(repo *catalog.ClusterRepo) bool {
+	return repo != nil && repo.Spec.Enabled != nil && !*repo.Spec.Enabled
+}
+
+func removeLegacyCondition(status *catalog.RepoStatus) {
+	if n := slices.IndexFunc(status.Conditions, func(cond genericcondition.GenericCondition) bool {
+		return cond.Type == "FollowerDownloaded"
+	}); n >= 0 {
+		status.Conditions = append(status.Conditions[:n], status.Conditions[n+1:]...)
+	}
+}
+
 func toOwnerObject(namespace string, owner metav1.OwnerReference) runtime.Object {
 	return &metav1.PartialObjectMetadata{
 		TypeMeta: metav1.TypeMeta{
@@ -232,22 +252,17 @@ func createOrUpdateMap(namespace string, index *repo.IndexFile, owner metav1.Own
 	return objs[0].(*corev1.ConfigMap), err
 }
 
-func (r *repoHandler) ensure(repoSpec *catalog.RepoSpec, status catalog.RepoStatus, metadata *metav1.ObjectMeta) (catalog.RepoStatus, error) {
-	if status.Commit == "" {
-		return status, nil
-	}
-
-	// If repo is disabled, skip updating the replicas.
-	if repoSpec.Enabled != nil && !*repoSpec.Enabled {
-		return status, nil
+func (r *repoHandler) ensure(repo *catalog.ClusterRepo) error {
+	if repo.Status.Commit == "" {
+		return nil
 	}
 
-	secret, err := catalogv2.GetSecret(r.secrets, repoSpec, metadata.Namespace)
+	secret, err := catalogv2.GetSecret(r.secrets, &repo.Spec, repo.Namespace)
 	if err != nil {
-		return status, err
+		return err
 	}
 
-	return status, git.Ensure(secret, metadata.Namespace, metadata.Name, status.URL, status.Commit, repoSpec.InsecureSkipTLSverify, repoSpec.CABundle)
+	return git.Ensure(secret, repo.Namespace, repo.Name, repo.Status.URL, repo.Status.Commit, repo.Spec.InsecureSkipTLSverify, repo.Spec.CABundle)
 }
 
 func (r *repoHandler) download(repository *catalog.ClusterRepo, newStatus *catalog.RepoStatus, owner metav1.OwnerReference, interval time.Duration, retryPolicy retryPolicy) (*catalog.ClusterRepo, error) {

pkg/controllers/dashboard/helm/repo_oci.go

@@ -121,8 +121,8 @@ func (o *OCIRepohandler) onClusterRepoChange(key string, clusterRepo *catalog.Cl
 	var index *repo.IndexFile
 
 	// If repo is disabled, then don't update the clusterrepo
-	if clusterRepo.Spec.Enabled != nil && !*clusterRepo.Spec.Enabled {
-		return setErrorCondition(clusterRepo, err, newStatus, ociInterval, ociCondition, o.clusterRepoController)
+	if isDisabled(clusterRepo) {
+		return setErrorCondition(clusterRepo, nil, newStatus, ociInterval, ociCondition, o.clusterRepoController)
 	}
 
 	secret, err := catalogv2.GetSecret(o.secretCacheController, &clusterRepo.Spec, clusterRepo.Namespace)