mirror of
https://github.com/hanzoai/deploy.git
synced 2026-10-07 08:44:50 +00:00
Two corrections on top of the naming pass, both found by checking the live cluster rather than the source. The rename had silently changed the user-facing annotation key from argocd.argoproj.io/sync-wave to cd.argoproj.io/sync-wave — 63 live Applications carry the old key, so deploying that would have ignored every sync-wave and broken fleet deploy ordering. The key is now cd.hanzo.ai/*, and universe 575650286 moves the ApplicationSet template that stamps it, so the controller regenerates all 63 consistently. The API group in code said argoproj.io/v1alpha1 in 157 places while the cluster has served apps.hanzo.ai/v1alpha1 since 2026-07-23 (the CRDs are applications.apps.hanzo.ai). Code now matches what is actually served. The legacy applications.argoproj.io/app-name label had ZERO live users, so it went with the rest. foo.argoproj.io test fixtures became foo.example.com — arbitrary test domains should not carry a vendor name. go build ./... exit 0. Co-authored-by: Hanzo Dev <[email protected]>
1392 lines
55 KiB
Go
1392 lines
55 KiB
Go
/*
|
|
Licensed under the Apache License, Version 2.0 (the "License");
|
|
you may not use this file except in compliance with the License.
|
|
You may obtain a copy of the License at
|
|
|
|
http://www.apache.org/licenses/LICENSE-2.0
|
|
|
|
Unless required by applicable law or agreed to in writing, software
|
|
distributed under the License is distributed on an "AS IS" BASIS,
|
|
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
See the License for the specific language governing permissions and
|
|
limitations under the License.
|
|
*/
|
|
|
|
package controllers
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"runtime/debug"
|
|
"sort"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/google/go-cmp/cmp"
|
|
"github.com/google/go-cmp/cmp/cmpopts"
|
|
log "github.com/sirupsen/logrus"
|
|
"golang.org/x/sync/errgroup"
|
|
corev1 "k8s.io/api/core/v1"
|
|
apierrors "k8s.io/apimachinery/pkg/api/errors"
|
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
|
"k8s.io/apimachinery/pkg/runtime"
|
|
"k8s.io/apimachinery/pkg/types"
|
|
"k8s.io/client-go/kubernetes"
|
|
"k8s.io/client-go/tools/record"
|
|
"k8s.io/client-go/util/retry"
|
|
ctrl "sigs.k8s.io/controller-runtime"
|
|
"sigs.k8s.io/controller-runtime/pkg/builder"
|
|
"sigs.k8s.io/controller-runtime/pkg/client"
|
|
"sigs.k8s.io/controller-runtime/pkg/controller"
|
|
"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
|
|
"sigs.k8s.io/controller-runtime/pkg/event"
|
|
"sigs.k8s.io/controller-runtime/pkg/handler"
|
|
"sigs.k8s.io/controller-runtime/pkg/predicate"
|
|
|
|
"github.com/hanzoai/deploy/applicationset/controllers/template"
|
|
"github.com/hanzoai/deploy/applicationset/generators"
|
|
"github.com/hanzoai/deploy/applicationset/metrics"
|
|
"github.com/hanzoai/deploy/applicationset/progressivesync"
|
|
"github.com/hanzoai/deploy/applicationset/status"
|
|
"github.com/hanzoai/deploy/applicationset/utils"
|
|
"github.com/hanzoai/deploy/common"
|
|
applog "github.com/hanzoai/deploy/util/app/log"
|
|
"github.com/hanzoai/deploy/util/db"
|
|
"github.com/hanzoai/deploy/util/settings"
|
|
|
|
argov1alpha1 "github.com/hanzoai/deploy/pkg/apis/application/v1alpha1"
|
|
argoutil "github.com/hanzoai/deploy/util/cd"
|
|
"github.com/hanzoai/deploy/util/cd/normalizers"
|
|
|
|
"github.com/hanzoai/deploy/pkg/apis/application"
|
|
)
|
|
|
|
const (
|
|
// Rather than importing the whole cd-notifications controller, just copying the const here
|
|
// https://github.com/argoproj-labs/cd-notifications/blob/33d345fa838829bb50fca5c08523aba380d2c12b/pkg/controller/subscriptions.go#L12
|
|
// https://github.com/argoproj-labs/cd-notifications/blob/33d345fa838829bb50fca5c08523aba380d2c12b/pkg/controller/state.go#L17
|
|
NotifiedAnnotationKey = "notified.notifications.apps.hanzo.ai"
|
|
ReconcileRequeueOnValidationError = time.Minute * 3
|
|
ReverseDeletionOrder = "Reverse"
|
|
AllAtOnceDeletionOrder = "AllAtOnce"
|
|
revisionAndSpecChangedMsg = "Application has pending changes (revision and spec differ), setting status to Waiting"
|
|
revisionChangedMsg = "Application has pending changes, setting status to Waiting"
|
|
specChangedMsg = "Application has pending changes (spec differs), setting status to Waiting"
|
|
)
|
|
|
|
var defaultPreservedFinalizers = []string{
|
|
argov1alpha1.PreDeleteFinalizerName,
|
|
argov1alpha1.PostDeleteFinalizerName,
|
|
}
|
|
|
|
var defaultPreservedAnnotations = []string{
|
|
NotifiedAnnotationKey,
|
|
argov1alpha1.AnnotationKeyRefresh,
|
|
argov1alpha1.AnnotationKeyHydrate,
|
|
}
|
|
|
|
// ApplicationSetReconciler reconciles a ApplicationSet object
|
|
type ApplicationSetReconciler struct {
|
|
client.Client
|
|
Scheme *runtime.Scheme
|
|
Recorder record.EventRecorder
|
|
Generators map[string]generators.Generator
|
|
DB db.DB
|
|
KubeClientset kubernetes.Interface
|
|
Policy argov1alpha1.ApplicationsSyncPolicy
|
|
EnablePolicyOverride bool
|
|
utils.Renderer
|
|
ArgoCDNamespace string
|
|
ApplicationSetNamespaces []string
|
|
EnableProgressiveSyncs bool
|
|
SCMRootCAPath string
|
|
GlobalPreservedAnnotations []string
|
|
GlobalPreservedLabels []string
|
|
Metrics *metrics.ApplicationsetMetrics
|
|
MaxResourcesStatusCount int
|
|
ClusterInformer *settings.ClusterInformer
|
|
ConcurrentApplicationUpdates int
|
|
ProgressiveSyncManager *progressivesync.Manager
|
|
}
|
|
|
|
var _ progressivesync.Dependencies = (*ApplicationSetReconciler)(nil)
|
|
|
|
// +kubebuilder:rbac:groups=apps.hanzo.ai,resources=applicationsets,verbs=get;list;watch;create;update;patch;delete
|
|
// +kubebuilder:rbac:groups=apps.hanzo.ai,resources=applicationsets/status,verbs=get;update;patch
|
|
|
|
func (r *ApplicationSetReconciler) Reconcile(ctx context.Context, req ctrl.Request) (result ctrl.Result, err error) {
|
|
startReconcile := time.Now()
|
|
logCtx := log.WithField("applicationset", req.NamespacedName)
|
|
|
|
defer func() {
|
|
if rec := recover(); rec != nil {
|
|
logCtx.Errorf("Recovered from panic: %+v\n%s", rec, debug.Stack())
|
|
result = ctrl.Result{}
|
|
var ok bool
|
|
err, ok = rec.(error)
|
|
if !ok {
|
|
err = fmt.Errorf("%v", r)
|
|
}
|
|
}
|
|
}()
|
|
|
|
var applicationSetInfo argov1alpha1.ApplicationSet
|
|
parametersGenerated := false
|
|
startTime := time.Now()
|
|
if err := r.Get(ctx, req.NamespacedName, &applicationSetInfo); err != nil {
|
|
if client.IgnoreNotFound(err) != nil {
|
|
logCtx.WithError(err).Infof("unable to get ApplicationSet: '%v' ", err)
|
|
}
|
|
return ctrl.Result{}, client.IgnoreNotFound(err)
|
|
}
|
|
|
|
defer func() {
|
|
r.Metrics.ObserveReconcile(&applicationSetInfo, time.Since(startTime))
|
|
}()
|
|
|
|
// Do not attempt to further reconcile the ApplicationSet if it is being deleted.
|
|
if applicationSetInfo.DeletionTimestamp != nil {
|
|
appsetName := applicationSetInfo.Name
|
|
logCtx.Debugf("DeletionTimestamp is set on %s", appsetName)
|
|
deleteAllowed := utils.DefaultPolicy(applicationSetInfo.Spec.SyncPolicy, r.Policy, r.EnablePolicyOverride).AllowDelete()
|
|
if !deleteAllowed {
|
|
logCtx.Debugf("ApplicationSet policy does not allow to delete")
|
|
if err := r.removeOwnerReferencesOnDeleteAppSet(ctx, applicationSetInfo); err != nil {
|
|
return ctrl.Result{}, err
|
|
}
|
|
logCtx.Debugf("ownerReferences referring %s is deleted from generated applications", appsetName)
|
|
}
|
|
if progressivesync.IsDeletionOrderReversed(&applicationSetInfo) {
|
|
logCtx.Debugf("DeletionOrder is set as Reverse on %s", appsetName)
|
|
currentApplications, err := r.getCurrentApplications(ctx, applicationSetInfo)
|
|
if err != nil {
|
|
return ctrl.Result{}, err
|
|
}
|
|
requeueTime, err := r.ProgressiveSyncManager.PerformReverseDeletion(ctx, logCtx, applicationSetInfo, currentApplications)
|
|
if err != nil {
|
|
return ctrl.Result{}, err
|
|
} else if requeueTime > 0 {
|
|
return ctrl.Result{RequeueAfter: requeueTime}, err
|
|
}
|
|
}
|
|
controllerutil.RemoveFinalizer(&applicationSetInfo, argov1alpha1.ResourcesFinalizerName)
|
|
if err := r.Update(ctx, &applicationSetInfo); err != nil {
|
|
return ctrl.Result{}, err
|
|
}
|
|
return ctrl.Result{}, nil
|
|
}
|
|
|
|
if err := r.migrateStatus(ctx, &applicationSetInfo); err != nil {
|
|
logCtx.Errorf("failed to migrate status subresource %v", err)
|
|
return ctrl.Result{}, err
|
|
}
|
|
|
|
// ensure finalizer exists if deletionOrder is set as Reverse
|
|
if r.EnableProgressiveSyncs && progressivesync.IsDeletionOrderReversed(&applicationSetInfo) {
|
|
if !controllerutil.ContainsFinalizer(&applicationSetInfo, argov1alpha1.ResourcesFinalizerName) {
|
|
controllerutil.AddFinalizer(&applicationSetInfo, argov1alpha1.ResourcesFinalizerName)
|
|
if err := r.Update(ctx, &applicationSetInfo); err != nil {
|
|
return ctrl.Result{}, err
|
|
}
|
|
}
|
|
}
|
|
|
|
// Log a warning if there are unrecognized generators
|
|
_ = utils.CheckInvalidGenerators(&applicationSetInfo)
|
|
// desiredApplications is the main list of all expected Applications from all generators in this appset.
|
|
generatedApplications, applicationSetReason, err := template.GenerateApplications(logCtx, applicationSetInfo, r.Generators, r.Renderer, r.Client)
|
|
if err != nil {
|
|
logCtx.Errorf("unable to generate applications: %v", err)
|
|
_ = r.setApplicationSetStatusCondition(ctx,
|
|
&applicationSetInfo,
|
|
[]argov1alpha1.ApplicationSetCondition{
|
|
{
|
|
Type: argov1alpha1.ApplicationSetConditionErrorOccurred,
|
|
Message: err.Error(),
|
|
Reason: string(applicationSetReason),
|
|
Status: argov1alpha1.ApplicationSetConditionStatusTrue,
|
|
},
|
|
}, parametersGenerated,
|
|
)
|
|
// In order for the controller SDK to respect RequeueAfter, the error must be nil
|
|
return ctrl.Result{RequeueAfter: ReconcileRequeueOnValidationError}, nil
|
|
}
|
|
|
|
parametersGenerated = true
|
|
|
|
validateErrors, err := r.validateGeneratedApplications(ctx, generatedApplications, applicationSetInfo)
|
|
if err != nil {
|
|
// While some generators may return an error that requires user intervention,
|
|
// other generators reference external resources that may change to cause
|
|
// the error to no longer occur. We thus log the error and requeue
|
|
// with a timeout to give this another shot at a later time.
|
|
//
|
|
// Changes to watched resources will cause this to be reconciled sooner than
|
|
// the RequeueAfter time.
|
|
logCtx.Errorf("error occurred during application validation: %s", err.Error())
|
|
|
|
_ = r.setApplicationSetStatusCondition(ctx,
|
|
&applicationSetInfo,
|
|
[]argov1alpha1.ApplicationSetCondition{
|
|
{
|
|
Type: argov1alpha1.ApplicationSetConditionErrorOccurred,
|
|
Message: err.Error(),
|
|
Reason: argov1alpha1.ApplicationSetReasonApplicationValidationError,
|
|
Status: argov1alpha1.ApplicationSetConditionStatusTrue,
|
|
},
|
|
}, parametersGenerated,
|
|
)
|
|
return ctrl.Result{RequeueAfter: ReconcileRequeueOnValidationError}, nil
|
|
}
|
|
|
|
currentApplications, err := r.getCurrentApplications(ctx, applicationSetInfo)
|
|
if err != nil {
|
|
return ctrl.Result{}, fmt.Errorf("failed to get current applications for application set: %w", err)
|
|
}
|
|
|
|
// appSyncMap tracks which apps will be synced during this reconciliation.
|
|
appSyncMap := map[string]bool{}
|
|
|
|
if r.EnableProgressiveSyncs {
|
|
if !progressivesync.IsRollingSyncStrategy(&applicationSetInfo) && len(applicationSetInfo.Status.ApplicationStatus) > 0 {
|
|
// If an appset was previously syncing with a `RollingSync` strategy but it has switched to the default strategy, clean up the progressive sync application statuses
|
|
logCtx.Infof("Removing %v unnecessary AppStatus entries from ApplicationSet %v", len(applicationSetInfo.Status.ApplicationStatus), applicationSetInfo.Name)
|
|
|
|
err := r.SetAppSetApplicationStatus(ctx, logCtx, &applicationSetInfo, []argov1alpha1.ApplicationSetApplicationStatus{})
|
|
if err != nil {
|
|
return ctrl.Result{}, fmt.Errorf("failed to clear previous AppSet application statuses for %v: %w", applicationSetInfo.Name, err)
|
|
}
|
|
} else if progressivesync.IsRollingSyncStrategy(&applicationSetInfo) {
|
|
// before starting progressive sync, checks if steps
|
|
if progressivesync.IsStepsEmpty(&applicationSetInfo) {
|
|
_ = r.setApplicationSetStatusCondition(ctx,
|
|
&applicationSetInfo,
|
|
[]argov1alpha1.ApplicationSetCondition{
|
|
{
|
|
Type: argov1alpha1.ApplicationSetConditionErrorOccurred,
|
|
Message: "No steps defined for rollout",
|
|
Reason: argov1alpha1.ApplicationSetReasonApplicationSetRolloutError,
|
|
Status: argov1alpha1.ApplicationSetConditionStatusTrue,
|
|
},
|
|
}, parametersGenerated,
|
|
)
|
|
return ctrl.Result{RequeueAfter: ReconcileRequeueOnValidationError}, nil
|
|
}
|
|
appSyncMap, err = r.ProgressiveSyncManager.PerformProgressiveSyncs(ctx, logCtx, applicationSetInfo, currentApplications, generatedApplications)
|
|
if err != nil {
|
|
return ctrl.Result{}, fmt.Errorf("failed to perform progressive sync reconciliation for application set: %w", err)
|
|
}
|
|
}
|
|
} else {
|
|
// Progressive Sync is disabled, clear any existing applicationStatus to prevent stale data
|
|
if len(applicationSetInfo.Status.ApplicationStatus) > 0 {
|
|
logCtx.Infof("Progressive Sync disabled, removing %v AppStatus entries from ApplicationSet %v", len(applicationSetInfo.Status.ApplicationStatus), applicationSetInfo.Name)
|
|
|
|
err := r.SetAppSetApplicationStatus(ctx, logCtx, &applicationSetInfo, []argov1alpha1.ApplicationSetApplicationStatus{})
|
|
if err != nil {
|
|
return ctrl.Result{}, fmt.Errorf("failed to clear AppSet application statuses when Progressive Sync is disabled for %v: %w", applicationSetInfo.Name, err)
|
|
}
|
|
}
|
|
}
|
|
|
|
if len(validateErrors) > 0 {
|
|
errorApps := make([]string, 0, len(validateErrors))
|
|
for key := range validateErrors {
|
|
errorApps = append(errorApps, key)
|
|
}
|
|
sort.Strings(errorApps)
|
|
|
|
var message string
|
|
for _, appName := range errorApps {
|
|
message = validateErrors[appName].Error()
|
|
logCtx.WithField("application", appName).Errorf("validation error found during application validation: %s", message)
|
|
}
|
|
if len(validateErrors) > 1 {
|
|
// Only the last message gets added to the appset status, to keep the size reasonable.
|
|
message = fmt.Sprintf("%s (and %d more)", message, len(validateErrors)-1)
|
|
}
|
|
_ = r.setApplicationSetStatusCondition(ctx,
|
|
&applicationSetInfo,
|
|
[]argov1alpha1.ApplicationSetCondition{
|
|
{
|
|
Type: argov1alpha1.ApplicationSetConditionErrorOccurred,
|
|
Message: message,
|
|
Reason: argov1alpha1.ApplicationSetReasonApplicationValidationError,
|
|
Status: argov1alpha1.ApplicationSetConditionStatusTrue,
|
|
},
|
|
}, parametersGenerated,
|
|
)
|
|
}
|
|
|
|
var validApps []argov1alpha1.Application
|
|
for i := range generatedApplications {
|
|
if validateErrors[generatedApplications[i].QualifiedName()] == nil {
|
|
validApps = append(validApps, generatedApplications[i])
|
|
}
|
|
}
|
|
|
|
if r.EnableProgressiveSyncs {
|
|
// trigger appropriate application syncs if RollingSync strategy is enabled
|
|
if progressivesync.RollingSyncStrategyEnabled(&applicationSetInfo) {
|
|
validApps = r.ProgressiveSyncManager.SyncDesiredApplications(logCtx, &applicationSetInfo, appSyncMap, validApps)
|
|
}
|
|
}
|
|
|
|
// Sort apps by name so they are updated/created in the same order, and condition errors are the same
|
|
sort.Slice(validApps, func(i, j int) bool {
|
|
return validApps[i].Name < validApps[j].Name
|
|
})
|
|
|
|
if utils.DefaultPolicy(applicationSetInfo.Spec.SyncPolicy, r.Policy, r.EnablePolicyOverride).AllowUpdate() {
|
|
err = r.createOrUpdateInCluster(ctx, logCtx, applicationSetInfo, validApps)
|
|
if err != nil {
|
|
_ = r.setApplicationSetStatusCondition(ctx,
|
|
&applicationSetInfo,
|
|
[]argov1alpha1.ApplicationSetCondition{
|
|
{
|
|
Type: argov1alpha1.ApplicationSetConditionErrorOccurred,
|
|
Message: err.Error(),
|
|
Reason: argov1alpha1.ApplicationSetReasonUpdateApplicationError,
|
|
Status: argov1alpha1.ApplicationSetConditionStatusTrue,
|
|
},
|
|
}, parametersGenerated,
|
|
)
|
|
return ctrl.Result{}, err
|
|
}
|
|
} else {
|
|
err = r.createInCluster(ctx, logCtx, applicationSetInfo, validApps)
|
|
if err != nil {
|
|
_ = r.setApplicationSetStatusCondition(ctx,
|
|
&applicationSetInfo,
|
|
[]argov1alpha1.ApplicationSetCondition{
|
|
{
|
|
Type: argov1alpha1.ApplicationSetConditionErrorOccurred,
|
|
Message: err.Error(),
|
|
Reason: argov1alpha1.ApplicationSetReasonCreateApplicationError,
|
|
Status: argov1alpha1.ApplicationSetConditionStatusTrue,
|
|
},
|
|
}, parametersGenerated,
|
|
)
|
|
return ctrl.Result{}, err
|
|
}
|
|
}
|
|
|
|
if utils.DefaultPolicy(applicationSetInfo.Spec.SyncPolicy, r.Policy, r.EnablePolicyOverride).AllowDelete() {
|
|
// Delete the generatedApplications instead of the validApps because we want to be able to delete applications in error/invalid state
|
|
err = r.deleteInCluster(ctx, logCtx, applicationSetInfo, generatedApplications)
|
|
if err != nil {
|
|
_ = r.setApplicationSetStatusCondition(ctx,
|
|
&applicationSetInfo,
|
|
[]argov1alpha1.ApplicationSetCondition{
|
|
{
|
|
Type: argov1alpha1.ApplicationSetConditionErrorOccurred,
|
|
Message: err.Error(),
|
|
Reason: argov1alpha1.ApplicationSetReasonDeleteApplicationError,
|
|
Status: argov1alpha1.ApplicationSetConditionStatusTrue,
|
|
},
|
|
}, parametersGenerated,
|
|
)
|
|
return ctrl.Result{}, err
|
|
}
|
|
}
|
|
|
|
// Update resources status after create/update/delete so it reflects the actual cluster state.
|
|
currentApplications, err = r.getCurrentApplications(ctx, applicationSetInfo)
|
|
if err != nil {
|
|
return ctrl.Result{}, fmt.Errorf("failed to get current applications for application set: %w", err)
|
|
}
|
|
err = r.updateResourcesStatus(ctx, logCtx, &applicationSetInfo, currentApplications)
|
|
if err != nil {
|
|
return ctrl.Result{}, fmt.Errorf("failed to update resources status for application set: %w", err)
|
|
}
|
|
|
|
if applicationSetInfo.RefreshRequired() {
|
|
delete(applicationSetInfo.Annotations, common.AnnotationApplicationSetRefresh)
|
|
err := r.Update(ctx, &applicationSetInfo)
|
|
if err != nil {
|
|
logCtx.Warnf("error occurred while updating ApplicationSet: %v", err)
|
|
_ = r.setApplicationSetStatusCondition(ctx,
|
|
&applicationSetInfo,
|
|
[]argov1alpha1.ApplicationSetCondition{
|
|
{
|
|
Type: argov1alpha1.ApplicationSetConditionErrorOccurred,
|
|
Message: err.Error(),
|
|
Reason: argov1alpha1.ApplicationSetReasonRefreshApplicationError,
|
|
Status: argov1alpha1.ApplicationSetConditionStatusTrue,
|
|
},
|
|
}, parametersGenerated,
|
|
)
|
|
return ctrl.Result{}, err
|
|
}
|
|
}
|
|
|
|
requeueAfter := r.getMinRequeueAfter(&applicationSetInfo)
|
|
|
|
if len(validateErrors) == 0 {
|
|
if err := r.setApplicationSetStatusCondition(ctx,
|
|
&applicationSetInfo,
|
|
[]argov1alpha1.ApplicationSetCondition{
|
|
{
|
|
Type: argov1alpha1.ApplicationSetConditionResourcesUpToDate,
|
|
Message: "All applications have been generated successfully",
|
|
Reason: argov1alpha1.ApplicationSetReasonApplicationSetUpToDate,
|
|
Status: argov1alpha1.ApplicationSetConditionStatusTrue,
|
|
},
|
|
}, parametersGenerated,
|
|
); err != nil {
|
|
return ctrl.Result{}, err
|
|
}
|
|
} else if requeueAfter == time.Duration(0) {
|
|
// Ensure that the request is requeued if there are validation errors.
|
|
requeueAfter = ReconcileRequeueOnValidationError
|
|
}
|
|
|
|
logCtx.WithField("requeueAfter", requeueAfter).Info("end reconcile in ", time.Since(startReconcile))
|
|
|
|
return ctrl.Result{
|
|
RequeueAfter: requeueAfter,
|
|
}, nil
|
|
}
|
|
|
|
func getParametersGeneratedCondition(parametersGenerated bool, message string) argov1alpha1.ApplicationSetCondition {
|
|
var parametersGeneratedCondition argov1alpha1.ApplicationSetCondition
|
|
if parametersGenerated {
|
|
parametersGeneratedCondition = argov1alpha1.ApplicationSetCondition{
|
|
Type: argov1alpha1.ApplicationSetConditionParametersGenerated,
|
|
Message: "Successfully generated parameters for all Applications",
|
|
Reason: argov1alpha1.ApplicationSetReasonParametersGenerated,
|
|
Status: argov1alpha1.ApplicationSetConditionStatusTrue,
|
|
}
|
|
} else {
|
|
parametersGeneratedCondition = argov1alpha1.ApplicationSetCondition{
|
|
Type: argov1alpha1.ApplicationSetConditionParametersGenerated,
|
|
Message: message,
|
|
Reason: argov1alpha1.ApplicationSetReasonErrorOccurred,
|
|
Status: argov1alpha1.ApplicationSetConditionStatusFalse,
|
|
}
|
|
}
|
|
return parametersGeneratedCondition
|
|
}
|
|
|
|
func (r *ApplicationSetReconciler) setApplicationSetStatusCondition(ctx context.Context, applicationSet *argov1alpha1.ApplicationSet, conditions []argov1alpha1.ApplicationSetCondition, parametersGenerated bool) error {
|
|
// Initialize the default condition types that this method evaluates
|
|
evaluatedTypes := map[argov1alpha1.ApplicationSetConditionType]bool{
|
|
argov1alpha1.ApplicationSetConditionParametersGenerated: true,
|
|
argov1alpha1.ApplicationSetConditionErrorOccurred: false,
|
|
argov1alpha1.ApplicationSetConditionResourcesUpToDate: false,
|
|
argov1alpha1.ApplicationSetConditionRolloutProgressing: false,
|
|
argov1alpha1.ApplicationSetConditionInvalidRolloutConfig: false,
|
|
}
|
|
|
|
if !progressivesync.IsRollingSyncStrategy(applicationSet) {
|
|
// Progressing sync is always evaluated so conditions are removed when it is not enabled
|
|
evaluatedTypes[argov1alpha1.ApplicationSetConditionInvalidRolloutConfig] = true
|
|
evaluatedTypes[argov1alpha1.ApplicationSetConditionRolloutProgressing] = true
|
|
}
|
|
|
|
var newConditions []argov1alpha1.ApplicationSetCondition
|
|
hasParamsGenerated := false
|
|
paramsMsg := "progressive sync"
|
|
for _, condition := range conditions {
|
|
if condition.Type == argov1alpha1.ApplicationSetConditionParametersGenerated {
|
|
hasParamsGenerated = true
|
|
break
|
|
}
|
|
if condition.Type == argov1alpha1.ApplicationSetConditionErrorOccurred {
|
|
paramsMsg = condition.Message
|
|
}
|
|
}
|
|
// Evaluate ParametersGenerated since it is always provided
|
|
if !hasParamsGenerated {
|
|
newConditions = append(newConditions, getParametersGeneratedCondition(parametersGenerated, paramsMsg))
|
|
}
|
|
for _, condition := range conditions {
|
|
// Evaluate current condition
|
|
evaluatedTypes[condition.Type] = true
|
|
newConditions = append(newConditions, condition)
|
|
|
|
// Evaluate dependencies between conditions.
|
|
switch condition.Type {
|
|
case argov1alpha1.ApplicationSetConditionResourcesUpToDate:
|
|
if condition.Status == argov1alpha1.ApplicationSetConditionStatusTrue {
|
|
// If the resources are up to date, we know there was no errors
|
|
evaluatedTypes[argov1alpha1.ApplicationSetConditionErrorOccurred] = true
|
|
newConditions = append(newConditions, argov1alpha1.ApplicationSetCondition{
|
|
Type: argov1alpha1.ApplicationSetConditionErrorOccurred,
|
|
Status: argov1alpha1.ApplicationSetConditionStatusFalse,
|
|
Reason: condition.Reason,
|
|
Message: condition.Message,
|
|
})
|
|
}
|
|
case argov1alpha1.ApplicationSetConditionErrorOccurred:
|
|
if condition.Status == argov1alpha1.ApplicationSetConditionStatusTrue {
|
|
// If there is an error anywhere in the reconciliation, we cannot consider the resources up to date
|
|
evaluatedTypes[argov1alpha1.ApplicationSetConditionResourcesUpToDate] = true
|
|
newConditions = append(newConditions, argov1alpha1.ApplicationSetCondition{
|
|
Type: argov1alpha1.ApplicationSetConditionResourcesUpToDate,
|
|
Status: argov1alpha1.ApplicationSetConditionStatusFalse,
|
|
Reason: argov1alpha1.ApplicationSetReasonErrorOccurred,
|
|
Message: condition.Message,
|
|
})
|
|
}
|
|
case argov1alpha1.ApplicationSetConditionRolloutProgressing:
|
|
if !progressivesync.IsRollingSyncStrategy(applicationSet) {
|
|
// if the condition is a rolling sync and it is disabled, ignore it
|
|
evaluatedTypes[condition.Type] = false
|
|
}
|
|
|
|
case argov1alpha1.ApplicationSetConditionInvalidRolloutConfig:
|
|
if !progressivesync.IsRollingSyncStrategy(applicationSet) {
|
|
evaluatedTypes[condition.Type] = false
|
|
}
|
|
}
|
|
}
|
|
|
|
// Update the applicationSet conditions
|
|
previousConditions := applicationSet.Status.Conditions
|
|
applicationSet.Status.SetConditions(newConditions, evaluatedTypes)
|
|
|
|
// Try to not call get/update if nothing has changed
|
|
needToUpdateConditions := len(applicationSet.Status.Conditions) != len(previousConditions)
|
|
if !needToUpdateConditions {
|
|
for i, c := range applicationSet.Status.Conditions {
|
|
previous := previousConditions[i]
|
|
if c.Type != previous.Type || c.Reason != previous.Reason || c.Status != previous.Status || c.Message != previous.Message {
|
|
needToUpdateConditions = true
|
|
break
|
|
}
|
|
}
|
|
}
|
|
|
|
if !needToUpdateConditions {
|
|
return nil
|
|
}
|
|
// DefaultRetry will retry 5 times with a backoff factor of 1, jitter of 0.1 and a duration of 10ms
|
|
err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
|
|
updatedAppset := &argov1alpha1.ApplicationSet{}
|
|
if err := r.Get(ctx, types.NamespacedName{Namespace: applicationSet.Namespace, Name: applicationSet.Name}, updatedAppset); err != nil {
|
|
if client.IgnoreNotFound(err) != nil {
|
|
return nil
|
|
}
|
|
return fmt.Errorf("error fetching updated application set: %w", err)
|
|
}
|
|
|
|
updatedAppset.Status.SetConditions(newConditions, evaluatedTypes)
|
|
|
|
// Update the newly fetched object with new set of conditions
|
|
err := r.Client.Status().Update(ctx, updatedAppset)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
updatedAppset.DeepCopyInto(applicationSet)
|
|
return nil
|
|
})
|
|
if err != nil && !apierrors.IsNotFound(err) {
|
|
return fmt.Errorf("unable to set application set condition: %w", err)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// validateGeneratedApplications uses the Hanzo CD validation functions to verify the correctness of the
|
|
// generated applications.
|
|
func (r *ApplicationSetReconciler) validateGeneratedApplications(ctx context.Context, desiredApplications []argov1alpha1.Application, applicationSetInfo argov1alpha1.ApplicationSet) (map[string]error, error) {
|
|
errorsByApp := map[string]error{}
|
|
namesSet := map[string]bool{}
|
|
for i := range desiredApplications {
|
|
app := &desiredApplications[i]
|
|
if namesSet[app.Name] {
|
|
errorsByApp[app.QualifiedName()] = fmt.Errorf("ApplicationSet %s contains applications with duplicate name: %s", applicationSetInfo.Name, app.Name)
|
|
continue
|
|
}
|
|
namesSet[app.Name] = true
|
|
appProject := &argov1alpha1.AppProject{}
|
|
err := r.Get(ctx, types.NamespacedName{Name: app.Spec.Project, Namespace: r.ArgoCDNamespace}, appProject)
|
|
if err != nil {
|
|
if apierrors.IsNotFound(err) {
|
|
errorsByApp[app.QualifiedName()] = fmt.Errorf("application references project %s which does not exist", app.Spec.Project)
|
|
continue
|
|
}
|
|
return nil, err
|
|
}
|
|
|
|
if _, err = argoutil.GetDestinationCluster(ctx, app.Spec.Destination, r.DB); err != nil {
|
|
errorsByApp[app.QualifiedName()] = fmt.Errorf("application destination spec is invalid: %s", err.Error())
|
|
continue
|
|
}
|
|
}
|
|
|
|
return errorsByApp, nil
|
|
}
|
|
|
|
func (r *ApplicationSetReconciler) getMinRequeueAfter(applicationSetInfo *argov1alpha1.ApplicationSet) time.Duration {
|
|
var res time.Duration
|
|
for _, requestedGenerator := range applicationSetInfo.Spec.Generators {
|
|
relevantGenerators := generators.GetRelevantGenerators(&requestedGenerator, r.Generators)
|
|
|
|
for _, g := range relevantGenerators {
|
|
t := g.GetRequeueAfter(&requestedGenerator)
|
|
|
|
if res == 0 {
|
|
res = t
|
|
} else if t != 0 && t < res {
|
|
res = t
|
|
}
|
|
}
|
|
}
|
|
|
|
return res
|
|
}
|
|
|
|
func ignoreNotAllowedNamespaces(namespaces []string) predicate.Predicate {
|
|
return predicate.NewPredicateFuncs(func(object client.Object) bool {
|
|
return utils.IsNamespaceAllowed(namespaces, object.GetNamespace())
|
|
})
|
|
}
|
|
|
|
func appControllerIndexer(rawObj client.Object) []string {
|
|
// grab the job object, extract the owner...
|
|
app := rawObj.(*argov1alpha1.Application)
|
|
owner := metav1.GetControllerOf(app)
|
|
if owner == nil {
|
|
return nil
|
|
}
|
|
// ...make sure it's a application set...
|
|
if owner.APIVersion != argov1alpha1.SchemeGroupVersion.String() || owner.Kind != "ApplicationSet" {
|
|
return nil
|
|
}
|
|
|
|
// ...and if so, return it
|
|
return []string{owner.Name}
|
|
}
|
|
|
|
func (r *ApplicationSetReconciler) SetupWithManager(mgr ctrl.Manager, enableProgressiveSyncs bool, maxConcurrentReconciliations int) error {
|
|
if err := mgr.GetFieldIndexer().IndexField(context.TODO(), &argov1alpha1.Application{}, ".metadata.controller", appControllerIndexer); err != nil {
|
|
return fmt.Errorf("error setting up with manager: %w", err)
|
|
}
|
|
|
|
appOwnsHandler := getApplicationOwnsHandler(enableProgressiveSyncs)
|
|
appSetOwnsHandler := getApplicationSetOwnsHandler(enableProgressiveSyncs)
|
|
|
|
return ctrl.NewControllerManagedBy(mgr).WithOptions(controller.Options{
|
|
MaxConcurrentReconciles: maxConcurrentReconciliations,
|
|
}).For(&argov1alpha1.ApplicationSet{}, builder.WithPredicates(appSetOwnsHandler)).
|
|
Owns(&argov1alpha1.Application{}, builder.WithPredicates(appOwnsHandler)).
|
|
WithEventFilter(ignoreNotAllowedNamespaces(r.ApplicationSetNamespaces)).
|
|
Watches(
|
|
&corev1.Secret{},
|
|
&clusterSecretEventHandler{
|
|
Client: mgr.GetClient(),
|
|
Log: log.WithField("type", "createSecretEventHandler"),
|
|
ApplicationSetNamespaces: r.ApplicationSetNamespaces,
|
|
}).
|
|
Complete(r)
|
|
}
|
|
|
|
// createOrUpdateInCluster will create / update application resources in the cluster.
|
|
// - For new applications, it will call create
|
|
// - For existing application, it will call update
|
|
// The function also adds owner reference to all applications, and uses it to delete them.
|
|
func (r *ApplicationSetReconciler) createOrUpdateInCluster(ctx context.Context, logCtx *log.Entry, applicationSet argov1alpha1.ApplicationSet, desiredApplications []argov1alpha1.Application) error {
|
|
// Build the diff config once per reconcile.
|
|
// Diff config is per applicationset, so generate it once for all applications
|
|
diffConfig, err := utils.BuildIgnoreDiffConfig(applicationSet.Spec.IgnoreApplicationDifferences, normalizers.IgnoreNormalizerOpts{})
|
|
if err != nil {
|
|
return fmt.Errorf("failed to build ignore diff config: %w", err)
|
|
}
|
|
|
|
g, ctx := errgroup.WithContext(ctx)
|
|
concurrency := r.concurrency()
|
|
g.SetLimit(concurrency)
|
|
|
|
var appErrorsMu sync.Mutex
|
|
appErrors := map[string]error{}
|
|
|
|
for _, generatedApp := range desiredApplications {
|
|
// Normalize to avoid fighting with the application controller.
|
|
generatedApp.Spec = *argoutil.NormalizeApplicationSpec(&generatedApp.Spec)
|
|
g.Go(func() error {
|
|
appLog := logCtx.WithFields(applog.GetAppLogFields(&generatedApp))
|
|
|
|
found := &argov1alpha1.Application{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: generatedApp.Name,
|
|
Namespace: generatedApp.Namespace,
|
|
},
|
|
TypeMeta: metav1.TypeMeta{
|
|
Kind: application.ApplicationKind,
|
|
APIVersion: "apps.hanzo.ai/v1alpha1",
|
|
},
|
|
}
|
|
|
|
action, err := utils.CreateOrUpdate(ctx, appLog, r.Client, diffConfig, found, func() error {
|
|
// Copy only the Application/ObjectMeta fields that are significant, from the generatedApp
|
|
found.Spec = generatedApp.Spec
|
|
|
|
// allow setting the Operation field to trigger a sync operation on an Application
|
|
if generatedApp.Operation != nil {
|
|
found.Operation = generatedApp.Operation
|
|
}
|
|
|
|
preservedAnnotations := make([]string, 0)
|
|
preservedLabels := make([]string, 0)
|
|
|
|
if applicationSet.Spec.PreservedFields != nil {
|
|
preservedAnnotations = append(preservedAnnotations, applicationSet.Spec.PreservedFields.Annotations...)
|
|
preservedLabels = append(preservedLabels, applicationSet.Spec.PreservedFields.Labels...)
|
|
}
|
|
|
|
if len(r.GlobalPreservedAnnotations) > 0 {
|
|
preservedAnnotations = append(preservedAnnotations, r.GlobalPreservedAnnotations...)
|
|
}
|
|
|
|
if len(r.GlobalPreservedLabels) > 0 {
|
|
preservedLabels = append(preservedLabels, r.GlobalPreservedLabels...)
|
|
}
|
|
|
|
// Preserve specially treated argo cd annotations:
|
|
// * https://github.com/argoproj/applicationset/issues/180
|
|
// * https://github.com/argoproj/argo-cd/issues/10500
|
|
preservedAnnotations = append(preservedAnnotations, defaultPreservedAnnotations...)
|
|
|
|
for _, key := range preservedAnnotations {
|
|
if state, exists := found.Annotations[key]; exists {
|
|
if generatedApp.Annotations == nil {
|
|
generatedApp.Annotations = map[string]string{}
|
|
}
|
|
generatedApp.Annotations[key] = state
|
|
}
|
|
}
|
|
|
|
for _, key := range preservedLabels {
|
|
if state, exists := found.Labels[key]; exists {
|
|
if generatedApp.Labels == nil {
|
|
generatedApp.Labels = map[string]string{}
|
|
}
|
|
generatedApp.Labels[key] = state
|
|
}
|
|
}
|
|
|
|
// Preserve deleting finalizers and avoid diff conflicts
|
|
for _, finalizer := range defaultPreservedFinalizers {
|
|
for _, f := range found.Finalizers {
|
|
// For finalizers, use prefix matching in case it contains "/" stages
|
|
if strings.HasPrefix(f, finalizer) {
|
|
generatedApp.Finalizers = append(generatedApp.Finalizers, f)
|
|
}
|
|
}
|
|
}
|
|
|
|
found.Annotations = generatedApp.Annotations
|
|
found.Labels = generatedApp.Labels
|
|
found.Finalizers = generatedApp.Finalizers
|
|
|
|
return controllerutil.SetControllerReference(&applicationSet, found, r.Scheme)
|
|
})
|
|
if err != nil {
|
|
appLog.WithError(err).WithField("action", action).Errorf("failed to %s Application", action)
|
|
// If the context was canceled or its deadline exceeded, return the error so it propagates through g.Wait().
|
|
if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
|
|
return err
|
|
}
|
|
// For backwards compatibility with sequential behavior: continue processing other applications
|
|
// but record the error keyed by app name so we can deterministically return the error from
|
|
// the lexicographically first failing app, regardless of goroutine scheduling order.
|
|
appErrorsMu.Lock()
|
|
appErrors[generatedApp.Name] = err
|
|
appErrorsMu.Unlock()
|
|
return nil
|
|
}
|
|
|
|
if action != controllerutil.OperationResultNone {
|
|
// Don't pollute etcd with "unchanged Application" events
|
|
r.Recorder.Eventf(&applicationSet, corev1.EventTypeNormal, fmt.Sprint(action), "%s Application %q", action, generatedApp.Name)
|
|
appLog.Logf(log.InfoLevel, "%s Application", action)
|
|
} else {
|
|
// "unchanged Application" can be inferred by Reconcile Complete with no action being listed
|
|
// Or enable debug logging
|
|
appLog.Logf(log.DebugLevel, "%s Application", action)
|
|
}
|
|
return nil
|
|
})
|
|
}
|
|
|
|
if err := g.Wait(); errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
|
|
return err
|
|
}
|
|
return firstAppError(appErrors)
|
|
}
|
|
|
|
// createInCluster will filter from the desiredApplications only the application that needs to be created
|
|
// Then it will call createOrUpdateInCluster to do the actual create
|
|
func (r *ApplicationSetReconciler) createInCluster(ctx context.Context, logCtx *log.Entry, applicationSet argov1alpha1.ApplicationSet, desiredApplications []argov1alpha1.Application) error {
|
|
var createApps []argov1alpha1.Application
|
|
current, err := r.getCurrentApplications(ctx, applicationSet)
|
|
if err != nil {
|
|
return fmt.Errorf("error getting current applications: %w", err)
|
|
}
|
|
|
|
m := make(map[string]bool) // Will holds the app names that are current in the cluster
|
|
|
|
for _, app := range current {
|
|
m[app.Name] = true
|
|
}
|
|
|
|
// filter applications that are not in m[string]bool (new to the cluster)
|
|
for _, app := range desiredApplications {
|
|
_, exists := m[app.Name]
|
|
|
|
if !exists {
|
|
createApps = append(createApps, app)
|
|
}
|
|
}
|
|
|
|
return r.createOrUpdateInCluster(ctx, logCtx, applicationSet, createApps)
|
|
}
|
|
|
|
func (r *ApplicationSetReconciler) getCurrentApplications(ctx context.Context, applicationSet argov1alpha1.ApplicationSet) ([]argov1alpha1.Application, error) {
|
|
var current argov1alpha1.ApplicationList
|
|
err := r.List(ctx, ¤t, client.MatchingFields{".metadata.controller": applicationSet.Name}, client.InNamespace(applicationSet.Namespace))
|
|
if err != nil {
|
|
return nil, fmt.Errorf("error retrieving applications: %w", err)
|
|
}
|
|
|
|
return current.Items, nil
|
|
}
|
|
|
|
// deleteInCluster will delete Applications that are currently on the cluster, but not in appList.
|
|
// The function must be called after all generators had been called and generated applications
|
|
func (r *ApplicationSetReconciler) deleteInCluster(ctx context.Context, logCtx *log.Entry, applicationSet argov1alpha1.ApplicationSet, desiredApplications []argov1alpha1.Application) error {
|
|
clusterList, err := utils.ListClusters(r.ClusterInformer)
|
|
if err != nil {
|
|
return fmt.Errorf("error listing clusters: %w", err)
|
|
}
|
|
|
|
// Save current applications to be able to delete the ones that are not in appList
|
|
current, err := r.getCurrentApplications(ctx, applicationSet)
|
|
if err != nil {
|
|
return fmt.Errorf("error getting current applications: %w", err)
|
|
}
|
|
|
|
m := make(map[string]bool) // will hold the app names in appList for the deletion process
|
|
|
|
for _, app := range desiredApplications {
|
|
m[app.Name] = true
|
|
}
|
|
|
|
g, ctx := errgroup.WithContext(ctx)
|
|
concurrency := r.concurrency()
|
|
g.SetLimit(concurrency)
|
|
|
|
var appErrorsMu sync.Mutex
|
|
appErrors := map[string]error{}
|
|
|
|
// Delete apps that are not in m[string]bool
|
|
for _, app := range current {
|
|
_, exists := m[app.Name]
|
|
if exists {
|
|
continue
|
|
}
|
|
appLogCtx := logCtx.WithFields(applog.GetAppLogFields(&app))
|
|
g.Go(func() error {
|
|
// Removes the Hanzo CD resources finalizer if the application contains an invalid target (eg missing cluster)
|
|
err := r.removeFinalizerOnInvalidDestination(ctx, applicationSet, &app, clusterList, appLogCtx)
|
|
if err != nil {
|
|
appLogCtx.WithError(err).Error("failed to update Application")
|
|
// If the context was canceled or its deadline exceeded, return the error so it propagates through g.Wait().
|
|
if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
|
|
return err
|
|
}
|
|
// For backwards compatibility with sequential behavior: continue processing other applications
|
|
// but record the error keyed by app name so we can deterministically return the error from
|
|
// the lexicographically first failing app, regardless of goroutine scheduling order.
|
|
appErrorsMu.Lock()
|
|
appErrors[app.Name] = err
|
|
appErrorsMu.Unlock()
|
|
return nil
|
|
}
|
|
|
|
err = r.Delete(ctx, &app)
|
|
if err != nil {
|
|
appLogCtx.WithError(err).Error("failed to delete Application")
|
|
// If the context was canceled or its deadline exceeded, return the error so it propagates through g.Wait().
|
|
if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
|
|
return err
|
|
}
|
|
appErrorsMu.Lock()
|
|
appErrors[app.Name] = err
|
|
appErrorsMu.Unlock()
|
|
return nil
|
|
}
|
|
r.Recorder.Eventf(&applicationSet, corev1.EventTypeNormal, "Deleted", "Deleted Application %q", app.Name)
|
|
appLogCtx.Log(log.InfoLevel, "Deleted application")
|
|
return nil
|
|
})
|
|
}
|
|
|
|
if err := g.Wait(); errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
|
|
return err
|
|
}
|
|
return firstAppError(appErrors)
|
|
}
|
|
|
|
// concurrency returns the configured number of concurrent application updates, defaulting to 1.
|
|
func (r *ApplicationSetReconciler) concurrency() int {
|
|
if r.ConcurrentApplicationUpdates <= 0 {
|
|
return 1
|
|
}
|
|
return r.ConcurrentApplicationUpdates
|
|
}
|
|
|
|
// firstAppError returns the error associated with the lexicographically smallest application name
|
|
// in the provided map. This gives a deterministic result when multiple goroutines may have
|
|
// recorded errors concurrently, matching the behavior of the original sequential loop where the
|
|
// first application in iteration order would determine the returned error.
|
|
func firstAppError(appErrors map[string]error) error {
|
|
if len(appErrors) == 0 {
|
|
return nil
|
|
}
|
|
names := make([]string, 0, len(appErrors))
|
|
for name := range appErrors {
|
|
names = append(names, name)
|
|
}
|
|
sort.Strings(names)
|
|
return appErrors[names[0]]
|
|
}
|
|
|
|
// removeFinalizerOnInvalidDestination removes the Hanzo 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 []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
|
|
}
|
|
|
|
var validDestination bool
|
|
|
|
// Detect if the destination is invalid (name doesn't correspond to a matching cluster)
|
|
if destCluster, err := argoutil.GetDestinationCluster(ctx, app.Spec.Destination, r.DB); err != nil {
|
|
appLog.Warnf("The destination cluster for %s could not be found: %v", app.Name, err)
|
|
validDestination = false
|
|
} else {
|
|
// Detect if the destination's server field does not match an existing cluster
|
|
matchingCluster := false
|
|
for _, cluster := range clusterList {
|
|
// A cluster matches if either the server matches OR the name matches
|
|
// This handles cases where:
|
|
// 1. The cluster is the in-cluster (server=https://kubernetes.default.svc, name=in-cluster)
|
|
// 2. A custom cluster has the same server as in-cluster but a different name
|
|
if destCluster.Server == cluster.Server || (destCluster.Name != "" && cluster.Name != "" && destCluster.Name == cluster.Name) {
|
|
matchingCluster = true
|
|
break
|
|
}
|
|
}
|
|
|
|
if !matchingCluster {
|
|
appLog.Warnf("A match for the destination cluster for %s, by server url, could not be found", app.Name)
|
|
}
|
|
|
|
validDestination = matchingCluster
|
|
}
|
|
// If the destination is invalid (for example the cluster is no longer defined), then remove
|
|
// the application finalizers to avoid triggering Hanzo CD bug #5817
|
|
if !validDestination {
|
|
// Filter out the Hanzo CD finalizer from the finalizer list
|
|
var newFinalizers []string
|
|
for _, existingFinalizer := range app.Finalizers {
|
|
if existingFinalizer != argov1alpha1.ResourcesFinalizerName { // only remove this one
|
|
newFinalizers = append(newFinalizers, existingFinalizer)
|
|
}
|
|
}
|
|
|
|
// If the finalizer length changed (due to filtering out an Argo finalizer), update the finalizer list on the app
|
|
if len(newFinalizers) != len(app.Finalizers) {
|
|
updated := app.DeepCopy()
|
|
updated.Finalizers = newFinalizers
|
|
patch := client.MergeFrom(app)
|
|
if log.IsLevelEnabled(log.DebugLevel) {
|
|
utils.LogPatch(appLog, patch, updated)
|
|
}
|
|
if err := r.Patch(ctx, updated, patch); err != nil {
|
|
return fmt.Errorf("error updating finalizers: %w", err)
|
|
}
|
|
// Application must have updated list of finalizers
|
|
updated.DeepCopyInto(app)
|
|
|
|
r.Recorder.Eventf(&applicationSet, corev1.EventTypeNormal, "Updated", "Updated Application %q finalizer before deletion, because application has an invalid destination", app.Name)
|
|
appLog.Log(log.InfoLevel, "Updating application finalizer before deletion, because application has an invalid destination")
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (r *ApplicationSetReconciler) removeOwnerReferencesOnDeleteAppSet(ctx context.Context, applicationSet argov1alpha1.ApplicationSet) error {
|
|
applications, err := r.getCurrentApplications(ctx, applicationSet)
|
|
if err != nil {
|
|
return fmt.Errorf("error getting current applications for ApplicationSet: %w", err)
|
|
}
|
|
|
|
for _, app := range applications {
|
|
app.SetOwnerReferences([]metav1.OwnerReference{})
|
|
err := r.Update(ctx, &app)
|
|
if err != nil {
|
|
return fmt.Errorf("error updating application: %w", err)
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// migrateStatus run migrations on the status subresource of ApplicationSet early during the run of ApplicationSetReconciler.Reconcile
|
|
// this handles any defaulting of values - which would otherwise cause the references to r.Client.Status().Update to fail given missing required fields.
|
|
func (r *ApplicationSetReconciler) migrateStatus(ctx context.Context, appset *argov1alpha1.ApplicationSet) error {
|
|
update := false
|
|
if statusList := appset.Status.ApplicationStatus; statusList != nil {
|
|
for idx := range statusList {
|
|
if statusList[idx].TargetRevisions == nil {
|
|
statusList[idx].TargetRevisions = []string{}
|
|
update = true
|
|
}
|
|
}
|
|
}
|
|
|
|
if update {
|
|
// DefaultRetry will retry 5 times with a backoff factor of 1, jitter of 0.1 and a duration of 10ms
|
|
err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
|
|
namespacedName := types.NamespacedName{Namespace: appset.Namespace, Name: appset.Name}
|
|
updatedAppset := &argov1alpha1.ApplicationSet{}
|
|
if err := r.Get(ctx, namespacedName, updatedAppset); err != nil {
|
|
if client.IgnoreNotFound(err) != nil {
|
|
return nil
|
|
}
|
|
return fmt.Errorf("error fetching updated application set: %w", err)
|
|
}
|
|
|
|
updatedAppset.Status.ApplicationStatus = appset.Status.ApplicationStatus
|
|
|
|
// Update the newly fetched object with new set of ApplicationStatus
|
|
err := r.Client.Status().Update(ctx, updatedAppset)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
updatedAppset.DeepCopyInto(appset)
|
|
return nil
|
|
})
|
|
if err != nil && !apierrors.IsNotFound(err) {
|
|
return fmt.Errorf("unable to set application set condition: %w", err)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (r *ApplicationSetReconciler) updateResourcesStatus(ctx context.Context, logCtx *log.Entry, appset *argov1alpha1.ApplicationSet, apps []argov1alpha1.Application) error {
|
|
statusMap := status.GetResourceStatusMap(appset)
|
|
statusMap = status.BuildResourceStatus(statusMap, apps)
|
|
|
|
statuses := []argov1alpha1.ResourceStatus{}
|
|
for _, status := range statusMap {
|
|
statuses = append(statuses, status)
|
|
}
|
|
sort.Slice(statuses, func(i, j int) bool {
|
|
return statuses[i].Name < statuses[j].Name
|
|
})
|
|
resourcesCount := int64(len(statuses))
|
|
if r.MaxResourcesStatusCount > 0 && len(statuses) > r.MaxResourcesStatusCount {
|
|
logCtx.Warnf("Truncating ApplicationSet %s resource status from %d to max allowed %d entries", appset.Name, len(statuses), r.MaxResourcesStatusCount)
|
|
statuses = statuses[:r.MaxResourcesStatusCount]
|
|
}
|
|
appset.Status.Resources = statuses
|
|
appset.Status.ResourcesCount = resourcesCount
|
|
// DefaultRetry will retry 5 times with a backoff factor of 1, jitter of 0.1 and a duration of 10ms
|
|
err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
|
|
namespacedName := types.NamespacedName{Namespace: appset.Namespace, Name: appset.Name}
|
|
updatedAppset := &argov1alpha1.ApplicationSet{}
|
|
if err := r.Get(ctx, namespacedName, updatedAppset); err != nil {
|
|
if client.IgnoreNotFound(err) != nil {
|
|
return nil
|
|
}
|
|
return fmt.Errorf("error fetching updated application set: %w", err)
|
|
}
|
|
|
|
updatedAppset.Status.Resources = appset.Status.Resources
|
|
updatedAppset.Status.ResourcesCount = resourcesCount
|
|
|
|
// Update the newly fetched object with new status resources
|
|
err := r.Client.Status().Update(ctx, updatedAppset)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
updatedAppset.DeepCopyInto(appset)
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
logCtx.Errorf("unable to set application set status: %v", err)
|
|
return fmt.Errorf("unable to set application set status: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// setAppSetApplicationStatus updates the ApplicationSet's status field
|
|
// with any new/changed Application statuses.
|
|
func (r *ApplicationSetReconciler) setAppSetApplicationStatus(ctx context.Context, logCtx *log.Entry, applicationSet *argov1alpha1.ApplicationSet, applicationStatuses []argov1alpha1.ApplicationSetApplicationStatus) error {
|
|
needToUpdateStatus := false
|
|
|
|
if len(applicationStatuses) != len(applicationSet.Status.ApplicationStatus) {
|
|
logCtx.WithFields(log.Fields{
|
|
"current_count": len(applicationSet.Status.ApplicationStatus),
|
|
"expected_count": len(applicationStatuses),
|
|
}).Debug("application status count changed")
|
|
needToUpdateStatus = true
|
|
} else {
|
|
for i := range applicationStatuses {
|
|
appStatus := applicationStatuses[i]
|
|
idx := utils.FindApplicationStatusIndex(applicationSet.Status.ApplicationStatus, appStatus.Application)
|
|
if idx == -1 {
|
|
logCtx.WithFields(log.Fields{"application": appStatus.Application}).Debug("application not found in current status")
|
|
needToUpdateStatus = true
|
|
break
|
|
}
|
|
currentStatus := applicationSet.Status.ApplicationStatus[idx]
|
|
statusChanged := currentStatus.Status != appStatus.Status
|
|
stepChanged := currentStatus.Step != appStatus.Step
|
|
messageChanged := currentStatus.Message != appStatus.Message
|
|
|
|
if statusChanged || stepChanged || messageChanged {
|
|
if statusChanged {
|
|
logCtx.WithFields(log.Fields{"application": appStatus.Application, "previous_status": currentStatus.Status, "new_status": appStatus.Status}).
|
|
Debug("application status changed")
|
|
}
|
|
if stepChanged {
|
|
logCtx.WithFields(log.Fields{"application": appStatus.Application, "previous_step": currentStatus.Step, "new_step": appStatus.Step}).
|
|
Debug("application step changed")
|
|
}
|
|
if messageChanged {
|
|
logCtx.WithFields(log.Fields{"application": appStatus.Application}).Debug("application message changed")
|
|
}
|
|
needToUpdateStatus = true
|
|
break
|
|
}
|
|
}
|
|
}
|
|
|
|
if needToUpdateStatus {
|
|
// sort to make sure the array is always in the same order
|
|
applicationSet.Status.ApplicationStatus = make([]argov1alpha1.ApplicationSetApplicationStatus, len(applicationStatuses))
|
|
copy(applicationSet.Status.ApplicationStatus, applicationStatuses)
|
|
sort.Slice(applicationSet.Status.ApplicationStatus, func(i, j int) bool {
|
|
return applicationSet.Status.ApplicationStatus[i].Application < applicationSet.Status.ApplicationStatus[j].Application
|
|
})
|
|
|
|
// DefaultRetry will retry 5 times with a backoff factor of 1, jitter of 0.1 and a duration of 10ms
|
|
err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
|
|
updatedAppset := &argov1alpha1.ApplicationSet{}
|
|
if err := r.Get(ctx, types.NamespacedName{Namespace: applicationSet.Namespace, Name: applicationSet.Name}, updatedAppset); err != nil {
|
|
if client.IgnoreNotFound(err) != nil {
|
|
return nil
|
|
}
|
|
return fmt.Errorf("error fetching updated application set: %w", err)
|
|
}
|
|
|
|
updatedAppset.Status.ApplicationStatus = applicationSet.Status.ApplicationStatus
|
|
|
|
// Update the newly fetched object with new set of ApplicationStatus
|
|
err := r.Client.Status().Update(ctx, updatedAppset)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
updatedAppset.DeepCopyInto(applicationSet)
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
logCtx.Errorf("unable to set application set status: %v", err)
|
|
return fmt.Errorf("unable to set application set status: %w", err)
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func getApplicationOwnsHandler(enableProgressiveSyncs bool) predicate.Funcs {
|
|
return predicate.Funcs{
|
|
CreateFunc: func(e event.CreateEvent) bool {
|
|
// if we are the owner and there is a create event, we most likely created it and do not need to
|
|
// re-reconcile
|
|
if log.IsLevelEnabled(log.DebugLevel) {
|
|
logFields := log.Fields{"app": ""}
|
|
app, isApp := e.Object.(*argov1alpha1.Application)
|
|
if isApp {
|
|
logFields = applog.GetAppLogFields(app)
|
|
}
|
|
log.WithFields(logFields).Debugln("received create event from owning an application")
|
|
}
|
|
return false
|
|
},
|
|
DeleteFunc: func(e event.DeleteEvent) bool {
|
|
if log.IsLevelEnabled(log.DebugLevel) {
|
|
logFields := log.Fields{"app": ""}
|
|
app, isApp := e.Object.(*argov1alpha1.Application)
|
|
if isApp {
|
|
logFields = applog.GetAppLogFields(app)
|
|
}
|
|
log.WithFields(logFields).Debugln("received delete event from owning an application")
|
|
}
|
|
return true
|
|
},
|
|
UpdateFunc: func(e event.UpdateEvent) bool {
|
|
appOld, isApp := e.ObjectOld.(*argov1alpha1.Application)
|
|
if !isApp {
|
|
return false
|
|
}
|
|
logCtx := log.WithFields(applog.GetAppLogFields(appOld))
|
|
logCtx.Debugln("received update event from owning an application")
|
|
appNew, isApp := e.ObjectNew.(*argov1alpha1.Application)
|
|
if !isApp {
|
|
return false
|
|
}
|
|
requeue := shouldRequeueForApplication(appOld, appNew, enableProgressiveSyncs)
|
|
logCtx.WithField("requeue", requeue).Debugf("requeue caused by application %s", appNew.Name)
|
|
return requeue
|
|
},
|
|
GenericFunc: func(e event.GenericEvent) bool {
|
|
if log.IsLevelEnabled(log.DebugLevel) {
|
|
logFields := log.Fields{}
|
|
app, isApp := e.Object.(*argov1alpha1.Application)
|
|
if isApp {
|
|
logFields = applog.GetAppLogFields(app)
|
|
}
|
|
log.WithFields(logFields).Debugln("received generic event from owning an application")
|
|
}
|
|
return true
|
|
},
|
|
}
|
|
}
|
|
|
|
// shouldRequeueForApplication determines when we want to requeue an ApplicationSet for reconciling based on an owned
|
|
// application change
|
|
// The applicationset controller owns a subset of the Application CR.
|
|
// We do not need to re-reconcile if parts of the application change outside the applicationset's control.
|
|
// An example being, Application.ApplicationStatus.ReconciledAt which gets updated by the application controller.
|
|
// Additionally, Application.ObjectMeta.ResourceVersion and Application.ObjectMeta.Generation which are set by K8s.
|
|
func shouldRequeueForApplication(appOld *argov1alpha1.Application, appNew *argov1alpha1.Application, enableProgressiveSyncs bool) bool {
|
|
if appOld == nil || appNew == nil {
|
|
return false
|
|
}
|
|
|
|
// the applicationset controller owns the application spec, labels, annotations, and finalizers on the applications
|
|
// reflect.DeepEqual considers nil slices/maps not equal to empty slices/maps
|
|
// https://pkg.go.dev/reflect#DeepEqual
|
|
// ApplicationDestination has an unexported field so we can just use the == for comparison
|
|
if !cmp.Equal(appOld.Spec, appNew.Spec, cmpopts.EquateEmpty(), cmpopts.EquateComparable(argov1alpha1.ApplicationDestination{})) ||
|
|
!cmp.Equal(appOld.GetAnnotations(), appNew.GetAnnotations(), cmpopts.EquateEmpty()) ||
|
|
!cmp.Equal(appOld.GetLabels(), appNew.GetLabels(), cmpopts.EquateEmpty()) ||
|
|
!cmp.Equal(appOld.GetFinalizers(), appNew.GetFinalizers(), cmpopts.EquateEmpty()) {
|
|
return true
|
|
}
|
|
|
|
// progressive syncs use the application status for updates. if they differ, requeue to trigger the next progression
|
|
if enableProgressiveSyncs {
|
|
if appOld.Status.Health.Status != appNew.Status.Health.Status || appOld.Status.Sync.Status != appNew.Status.Sync.Status {
|
|
return true
|
|
}
|
|
|
|
if appOld.Status.OperationState != nil && appNew.Status.OperationState != nil {
|
|
if appOld.Status.OperationState.Phase != appNew.Status.OperationState.Phase ||
|
|
appOld.Status.OperationState.StartedAt != appNew.Status.OperationState.StartedAt {
|
|
return true
|
|
}
|
|
}
|
|
}
|
|
|
|
return false
|
|
}
|
|
|
|
func getApplicationSetOwnsHandler(enableProgressiveSyncs bool) predicate.Funcs {
|
|
return predicate.Funcs{
|
|
CreateFunc: func(e event.CreateEvent) bool {
|
|
appSet, isApp := e.Object.(*argov1alpha1.ApplicationSet)
|
|
if !isApp {
|
|
return false
|
|
}
|
|
log.WithField("applicationset", appSet.QualifiedName()).Debugln("received create event")
|
|
// Always queue a new applicationset
|
|
return true
|
|
},
|
|
DeleteFunc: func(e event.DeleteEvent) bool {
|
|
appSet, isApp := e.Object.(*argov1alpha1.ApplicationSet)
|
|
if !isApp {
|
|
return false
|
|
}
|
|
log.WithField("applicationset", appSet.QualifiedName()).Debugln("received delete event")
|
|
// Always queue for the deletion of an applicationset
|
|
return true
|
|
},
|
|
UpdateFunc: func(e event.UpdateEvent) bool {
|
|
appSetOld, isAppSet := e.ObjectOld.(*argov1alpha1.ApplicationSet)
|
|
if !isAppSet {
|
|
return false
|
|
}
|
|
appSetNew, isAppSet := e.ObjectNew.(*argov1alpha1.ApplicationSet)
|
|
if !isAppSet {
|
|
return false
|
|
}
|
|
requeue := shouldRequeueForApplicationSet(appSetOld, appSetNew, enableProgressiveSyncs)
|
|
log.WithField("applicationset", appSetNew.QualifiedName()).
|
|
WithField("requeue", requeue).Debugln("received update event")
|
|
return requeue
|
|
},
|
|
GenericFunc: func(e event.GenericEvent) bool {
|
|
appSet, isApp := e.Object.(*argov1alpha1.ApplicationSet)
|
|
if !isApp {
|
|
return false
|
|
}
|
|
log.WithField("applicationset", appSet.QualifiedName()).Debugln("received generic event")
|
|
// Always queue for the generic of an applicationset
|
|
return true
|
|
},
|
|
}
|
|
}
|
|
|
|
// shouldRequeueForApplicationSet determines when we need to requeue an applicationset
|
|
func shouldRequeueForApplicationSet(appSetOld, appSetNew *argov1alpha1.ApplicationSet, enableProgressiveSyncs bool) bool {
|
|
if appSetOld == nil || appSetNew == nil {
|
|
return false
|
|
}
|
|
|
|
// Requeue if any ApplicationStatus.Status changed for Progressive sync strategy
|
|
if enableProgressiveSyncs {
|
|
if !cmp.Equal(appSetOld.Status.ApplicationStatus, appSetNew.Status.ApplicationStatus, cmpopts.EquateEmpty()) {
|
|
return true
|
|
}
|
|
}
|
|
|
|
// only compare the applicationset spec, annotations, labels and finalizers, deletionTimestamp, specifically avoiding
|
|
// the status field. status is owned by the applicationset controller,
|
|
// and we do not need to requeue when it does bookkeeping
|
|
// NB: the ApplicationDestination comes from the ApplicationSpec being embedded
|
|
// in the ApplicationSetTemplate from the generators
|
|
if !cmp.Equal(appSetOld.Spec, appSetNew.Spec, cmpopts.EquateEmpty(), cmpopts.EquateComparable(argov1alpha1.ApplicationDestination{})) ||
|
|
!cmp.Equal(appSetOld.GetLabels(), appSetNew.GetLabels(), cmpopts.EquateEmpty()) ||
|
|
!cmp.Equal(appSetOld.GetFinalizers(), appSetNew.GetFinalizers(), cmpopts.EquateEmpty()) ||
|
|
!cmp.Equal(appSetOld.DeletionTimestamp, appSetNew.DeletionTimestamp, cmpopts.EquateEmpty()) {
|
|
return true
|
|
}
|
|
|
|
// Requeue only when the refresh annotation is newly added to the ApplicationSet.
|
|
// Changes to other annotations made simultaneously might be missed, but such cases are rare.
|
|
if !cmp.Equal(appSetOld.GetAnnotations(), appSetNew.GetAnnotations(), cmpopts.EquateEmpty()) {
|
|
_, oldHasRefreshAnnotation := appSetOld.Annotations[common.AnnotationApplicationSetRefresh]
|
|
_, newHasRefreshAnnotation := appSetNew.Annotations[common.AnnotationApplicationSetRefresh]
|
|
|
|
if oldHasRefreshAnnotation && !newHasRefreshAnnotation {
|
|
return false
|
|
}
|
|
return true
|
|
}
|
|
|
|
return false
|
|
}
|
|
|
|
var _ handler.EventHandler = &clusterSecretEventHandler{}
|