package controller import ( "context" "encoding/json" stderrors "errors" "fmt" "os" "reflect" "strconv" "time" "k8s.io/apimachinery/pkg/util/strategicpatch" cdcommon "github.com/hanzoai/cd/common" gitopsDiff "github.com/hanzoai/cd/gitops-engine/pkg/diff" "github.com/hanzoai/cd/gitops-engine/pkg/sync" "github.com/hanzoai/cd/gitops-engine/pkg/sync/common" "github.com/hanzoai/cd/gitops-engine/pkg/utils/kube" jsonpatch "github.com/evanphx/json-patch" log "github.com/sirupsen/logrus" otel_codes "go.opentelemetry.io/otel/codes" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apimachinery/pkg/util/managedfields" "k8s.io/client-go/kubernetes/scheme" "k8s.io/client-go/rest" "k8s.io/kubectl/pkg/util/openapi" "github.com/hanzoai/cd/controller/metrics" "github.com/hanzoai/cd/controller/syncid" "github.com/hanzoai/cd/pkg/apis/application/v1alpha1" applog "github.com/hanzoai/cd/util/app/log" "github.com/hanzoai/cd/util/cd" "github.com/hanzoai/cd/util/cd/diff" kubeutil "github.com/hanzoai/cd/util/kube" logutils "github.com/hanzoai/cd/util/log" "github.com/hanzoai/cd/util/lua" "github.com/hanzoai/cd/util/settings" ) const ( // EnvVarSyncWaveDelay is an environment variable which controls the delay in seconds between // each sync-wave EnvVarSyncWaveDelay = "CD_SYNC_WAVE_DELAY" ) func (m *appStateManager) getOpenAPISchema(server *v1alpha1.Cluster) (openapi.Resources, error) { cluster, err := m.liveStateCache.GetClusterCache(server) if err != nil { return nil, err } return cluster.GetOpenAPISchema(), nil } func (m *appStateManager) getGVKParser(server *v1alpha1.Cluster) (*managedfields.GvkParser, error) { cluster, err := m.liveStateCache.GetClusterCache(server) if err != nil { return nil, err } return cluster.GetGVKParser(), nil } // getServerSideDiffDryRunApplier will return the kubectl implementation of the KubeApplier // interface that provides functionality to dry run apply kubernetes resources. Returns a // cleanup function that must be called to remove the generated kube config for this // server. func (m *appStateManager) getServerSideDiffDryRunApplier(cluster *v1alpha1.Cluster) (gitopsDiff.KubeApplier, func(), error) { rawConfig, err := cluster.RawRestConfig() if err != nil { return nil, nil, fmt.Errorf("error getting cluster REST config: %w", err) } ops, cleanup, err := kubeutil.ManageServerSideDiffDryRuns(rawConfig, m.onKubectlRun) if err != nil { return nil, nil, fmt.Errorf("error creating kubectl ResourceOperations: %w", err) } return ops, cleanup, nil } func NewOperationState(operation v1alpha1.Operation) *v1alpha1.OperationState { return &v1alpha1.OperationState{ Phase: common.OperationRunning, Operation: operation, StartedAt: metav1.Now(), } } func newSyncOperationResult(app *v1alpha1.Application, op v1alpha1.SyncOperation) *v1alpha1.SyncOperationResult { syncRes := &v1alpha1.SyncOperationResult{} if len(op.Sources) > 0 || op.Source != nil { // specific source specified in the SyncOperation if op.Source != nil { syncRes.Source = *op.Source } syncRes.Sources = op.Sources } else { // normal sync case, get sources from the spec syncRes.Sources = app.Spec.Sources syncRes.Source = app.Spec.GetSource() } // Sync requests might be requested with ambiguous revisions (e.g. master, HEAD, v1.2.3). // This can change meaning when resuming operations (e.g a hook sync). After calculating a // concrete git commit SHA, the revision of the SyncOperationResult will be updated with the SHA syncRes.Revision = op.Revision syncRes.Revisions = op.Revisions return syncRes } func (m *appStateManager) SyncAppState(ctx context.Context, app *v1alpha1.Application, project *v1alpha1.AppProject, state *v1alpha1.OperationState) { ctx, span := tracer.Start(ctx, "controller.SyncAppState") setAppTraceAttrs(span, app) // SyncAppState is void; it signals failure through state.Phase rather than a return value, so // map a terminal failed phase onto the span status at exit (mirroring traceutil.EndSpan). defer func() { if state.Phase.Failed() { span.SetStatus(otel_codes.Error, state.Message) } span.End() }() syncId, err := syncid.Generate() if err != nil { state.Phase = common.OperationError state.Message = fmt.Sprintf("Failed to generate sync ID: %v", err) return } logEntry := log.WithFields(applog.GetAppLogFields(app)).WithField("syncId", syncId) if state.Operation.Sync == nil { state.Phase = common.OperationError state.Message = "Invalid operation request: no operation specified" return } syncOp := *state.Operation.Sync if state.SyncResult == nil { state.SyncResult = newSyncOperationResult(app, syncOp) } if isBlocked, err := syncWindowPreventsSync(app, project); isBlocked { // If the operation is currently running, simply let the user know the sync is blocked by a current sync window if state.Phase == common.OperationRunning { state.Message = "Sync operation blocked by sync window" if err != nil { state.Message = fmt.Sprintf("%s: %v", state.Message, err) } } return } revisions := state.SyncResult.Revisions sources := state.SyncResult.Sources isMultiSourceSync := len(sources) > 0 if !isMultiSourceSync { sources = []v1alpha1.ApplicationSource{state.SyncResult.Source} revisions = []string{state.SyncResult.Revision} } // ignore error if CompareStateRepoError, this shouldn't happen as noRevisionCache is true compareResult, err := m.CompareAppState(ctx, app, project, revisions, sources, false, true, syncOp.Manifests, isMultiSourceSync) if err != nil && !stderrors.Is(err, ErrCompareStateRepo) { state.Phase = common.OperationError state.Message = err.Error() return } // We are now guaranteed to have a concrete commit SHA. Save this in the sync result revision so that we remember // what we should be syncing to when resuming operations. state.SyncResult.Revision = compareResult.syncStatus.Revision state.SyncResult.Revisions = compareResult.syncStatus.Revisions // validates if it should fail the sync on that revision if it finds shared resources hasSharedResource, sharedResourceMessage := hasSharedResourceCondition(app) if syncOp.SyncOptions.HasOption("FailOnSharedResource=true") && hasSharedResource { state.Phase = common.OperationFailed state.Message = "Shared resource found: " + sharedResourceMessage return } // If there are any comparison or spec errors error conditions do not perform the operation if errConditions := app.Status.GetConditions(map[v1alpha1.ApplicationConditionType]bool{ v1alpha1.ApplicationConditionComparisonError: true, v1alpha1.ApplicationConditionInvalidSpecError: true, }); len(errConditions) > 0 { state.Phase = common.OperationError state.Message = cd.FormatAppConditions(errConditions) return } destCluster, err := cd.GetDestinationCluster(ctx, app.Spec.Destination, m.db) if err != nil { state.Phase = common.OperationError state.Message = fmt.Sprintf("Failed to get destination cluster: %v", err) return } rawConfig, err := destCluster.RawRestConfig() if err != nil { state.Phase = common.OperationError state.Message = err.Error() return } clusterRESTConfig, err := destCluster.RESTConfig() if err != nil { state.Phase = common.OperationError state.Message = err.Error() return } restConfig := metrics.AddMetricsTransportWrapper(m.metricsServer, app, clusterRESTConfig) resourceOverrides, err := m.settingsMgr.GetResourceOverrides() if err != nil { state.Phase = common.OperationError state.Message = fmt.Sprintf("Failed to load resource overrides: %v", err) return } initialResourcesRes := make([]common.ResourceSyncResult, len(state.SyncResult.Resources)) for i, res := range state.SyncResult.Resources { key := kube.ResourceKey{Group: res.Group, Kind: res.Kind, Namespace: res.Namespace, Name: res.Name} initialResourcesRes[i] = common.ResourceSyncResult{ ResourceKey: key, Message: res.Message, Status: res.Status, HookPhase: res.HookPhase, HookType: res.HookType, SyncPhase: res.SyncPhase, Version: res.Version, Images: res.Images, Order: i + 1, } } prunePropagationPolicy := metav1.DeletePropagationForeground switch { case syncOp.SyncOptions.HasOption("PrunePropagationPolicy=background"): prunePropagationPolicy = metav1.DeletePropagationBackground case syncOp.SyncOptions.HasOption("PrunePropagationPolicy=foreground"): prunePropagationPolicy = metav1.DeletePropagationForeground case syncOp.SyncOptions.HasOption("PrunePropagationPolicy=orphan"): prunePropagationPolicy = metav1.DeletePropagationOrphan } clientSideApplyManager := common.DefaultClientSideApplyMigrationManager // Check for custom field manager from application annotation if managerValue := app.GetAnnotation(cdcommon.AnnotationClientSideApplyMigrationManager); managerValue != "" { clientSideApplyManager = managerValue } reconciliationResult := compareResult.reconciliationResult // if RespectIgnoreDifferences is enabled, it should normalize the target // resources which in this case applies the live values in the configured // ignore differences fields. if syncOp.SyncOptions.HasOption("RespectIgnoreDifferences=true") { openAPISchema, err := m.getOpenAPISchema(destCluster) if err != nil { state.Phase = common.OperationError state.Message = fmt.Sprintf("failed to load openAPISchema: %v", err) return } patchedTargets, err := normalizeTargetResources(openAPISchema, compareResult) if err != nil { state.Phase = common.OperationError state.Message = fmt.Sprintf("Failed to normalize target resources: %s", err) return } reconciliationResult.Target = patchedTargets } installationID, err := m.settingsMgr.GetInstallationID() if err != nil { log.Errorf("Could not get installation ID: %v", err) return } trackingMethod, err := m.settingsMgr.GetTrackingMethod() if err != nil { log.Errorf("Could not get trackingMethod: %v", err) return } impersonationEnabled, err := m.settingsMgr.IsImpersonationEnabled() if err != nil { log.Errorf("could not get impersonation feature flag: %v", err) return } if impersonationEnabled { serviceAccountToImpersonate, err := settings.DeriveServiceAccountToImpersonate(project, app, destCluster) if err != nil { state.Phase = common.OperationError state.Message = fmt.Sprintf("failed to derive service account to impersonate: %v", err) return } if serviceAccountToImpersonate == "" { // No matching service account found - check enforcement impersonationEnforced, enforcedErr := m.settingsMgr.IsImpersonationEnforced() if enforcedErr != nil { log.Errorf("could not get impersonation enforcement flag: %v", enforcedErr) state.Phase = common.OperationError state.Message = fmt.Sprintf("failed to check impersonation enforcement setting: %v", enforcedErr) return } if impersonationEnforced { state.Phase = common.OperationError state.Message = fmt.Sprintf("no matching service account found for destination server %s and namespace %s", destCluster.Server, app.Spec.Destination.Namespace) return } // Non-enforced mode: log info and continue with controller SA logEntry.Infof("no matching service account found for impersonation (project: %s, server: %s, namespace: %s), falling back to controller service account", project.Name, destCluster.Server, app.Spec.Destination.Namespace) } else { logEntry = logEntry.WithFields(log.Fields{"impersonationEnabled": "true", "serviceAccount": serviceAccountToImpersonate}) // set the impersonation headers. rawConfig.Impersonate = rest.ImpersonationConfig{ UserName: serviceAccountToImpersonate, } restConfig.Impersonate = rest.ImpersonationConfig{ UserName: serviceAccountToImpersonate, } } } opts := []sync.SyncOpt{ sync.WithLogr(logutils.NewLogrusLogger(logEntry)), sync.WithHealthOverride(lua.ResourceHealthOverrides(resourceOverrides)), sync.WithPermissionValidator(func(un *unstructured.Unstructured, res *metav1.APIResource) error { return validateSyncPermissions(project, destCluster, func(proj string) ([]*v1alpha1.Cluster, error) { return m.db.GetProjectClusters(ctx, proj) }, un, res) }), sync.WithOperationSettings(syncOp.DryRun, syncOp.Prune, syncOp.SyncStrategy.Force(), syncOp.IsApplyStrategy() || len(syncOp.Resources) > 0), sync.WithInitialState(state.Phase, state.Message, initialResourcesRes, state.StartedAt), sync.WithResourcesFilter(func(key kube.ResourceKey, target *unstructured.Unstructured, live *unstructured.Unstructured) bool { return (len(syncOp.Resources) == 0 || isPostDeleteHook(target) || isPreDeleteHook(target) || cd.ContainsSyncResource(key.Name, key.Namespace, schema.GroupVersionKind{Kind: key.Kind, Group: key.Group}, syncOp.Resources)) && m.isSelfReferencedObj(live, target, app.GetName(), v1alpha1.TrackingMethod(trackingMethod), installationID) }), sync.WithManifestValidation(!syncOp.SyncOptions.HasOption(common.SyncOptionsDisableValidation)), sync.WithSyncWaveHook(delayBetweenSyncWaves), sync.WithPruneLast(syncOp.SyncOptions.HasOption(common.SyncOptionPruneLast)), sync.WithResourceModificationChecker(syncOp.SyncOptions.HasOption("ApplyOutOfSyncOnly=true"), compareResult.diffResultList), sync.WithPrunePropagationPolicy(&prunePropagationPolicy), sync.WithReplace(syncOp.SyncOptions.HasOption(common.SyncOptionReplace)), sync.WithServerSideApply(syncOp.SyncOptions.HasOption(common.SyncOptionServerSideApply)), sync.WithServerSideApplyManager(cdcommon.ArgoCDSSAManager), sync.WithClientSideApplyMigration( !syncOp.SyncOptions.HasOption(common.SyncOptionDisableClientSideApplyMigration), clientSideApplyManager, ), sync.WithPruneConfirmed(app.IsDeletionConfirmed(state.StartedAt.Time)), sync.WithDefaultPruneOption(syncOp.SyncOptions.GetOptionValue(common.SyncOptionPrune)), sync.WithSkipDryRunOnMissingResource(syncOp.SyncOptions.HasOption(common.SyncOptionSkipDryRunOnMissingResource)), } if syncOp.SyncOptions.HasOption("CreateNamespace=true") { opts = append(opts, sync.WithNamespaceModifier(syncNamespace(app.Spec.SyncPolicy))) } syncCtx, cleanup, err := sync.NewSyncContext( compareResult.syncStatus.Revision, reconciliationResult, restConfig, rawConfig, m.kubectl, app.Spec.Destination.Namespace, opts..., ) if err != nil { state.Phase = common.OperationError state.Message = fmt.Sprintf("failed to initialize sync context: %v", err) return } defer cleanup() start := time.Now() if state.Phase == common.OperationTerminating { syncCtx.Terminate(ctx) } else { syncCtx.Sync(ctx) } var resState []common.ResourceSyncResult state.Phase, state.Message, resState = syncCtx.GetState() state.SyncResult.Resources = nil if app.Spec.SyncPolicy != nil { state.SyncResult.ManagedNamespaceMetadata = app.Spec.SyncPolicy.ManagedNamespaceMetadata } var apiVersion []kube.APIResourceInfo for _, res := range resState { augmentedMsg, err := cd.AugmentSyncMsg(res, func() ([]kube.APIResourceInfo, error) { if apiVersion == nil { _, apiVersion, err = m.liveStateCache.GetVersionsInfo(destCluster) if err != nil { return nil, fmt.Errorf("failed to get version info from the target cluster %q", destCluster.Server) } } return apiVersion, nil }) if err != nil { log.Errorf("using the original message since: %v", err) } else { res.Message = augmentedMsg } state.SyncResult.Resources = append(state.SyncResult.Resources, &v1alpha1.ResourceResult{ HookType: res.HookType, Group: res.ResourceKey.Group, Kind: res.ResourceKey.Kind, Namespace: res.ResourceKey.Namespace, Name: res.ResourceKey.Name, Version: res.Version, SyncPhase: res.SyncPhase, HookPhase: res.HookPhase, Status: res.Status, Message: res.Message, Images: res.Images, }) } logEntry.WithField("duration", time.Since(start)).Info("sync/terminate complete") if !syncOp.DryRun && len(syncOp.Resources) == 0 && state.Phase.Successful() { err := m.persistRevisionHistory(app, compareResult.syncStatus.Revision, compareResult.syncStatus.ComparedTo.Source, compareResult.syncStatus.Revisions, compareResult.syncStatus.ComparedTo.Sources, isMultiSourceSync, state.StartedAt, state.Operation.InitiatedBy) if err != nil { state.Phase = common.OperationError state.Message = fmt.Sprintf("failed to record sync to history: %v", err) } } } // normalizeTargetResources modifies target resources to ensure ignored fields are not touched during synchronization: // - applies normalization to the target resources based on the live resources // - copies ignored fields from the matching live resources: apply normalizer to the live resource, // calculates the patch performed by normalizer and applies the patch to the target resource func normalizeTargetResources(openAPISchema openapi.Resources, cr *comparisonResult) ([]*unstructured.Unstructured, error) { // Normalize live and target resources (cleaning or aligning them) normalized, err := diff.Normalize(cr.reconciliationResult.Live, cr.reconciliationResult.Target, cr.diffConfig) if err != nil { return nil, err } patchedTargets := []*unstructured.Unstructured{} for idx, live := range cr.reconciliationResult.Live { normalizedTarget := normalized.Targets[idx] if normalizedTarget == nil { patchedTargets = append(patchedTargets, nil) continue } gvk := normalizedTarget.GroupVersionKind() originalTarget := cr.reconciliationResult.Target[idx] if live == nil { // No live resource, just use target patchedTargets = append(patchedTargets, originalTarget) continue } var ( lookupPatchMeta strategicpatch.LookupPatchMeta versionedObject any ) // Load patch meta struct or OpenAPI schema for CRDs if versionedObject, err = scheme.Scheme.New(gvk); err == nil { if lookupPatchMeta, err = strategicpatch.NewPatchMetaFromStruct(versionedObject); err != nil { return nil, err } } else if crdSchema := openAPISchema.LookupResource(gvk); crdSchema != nil { lookupPatchMeta = strategicpatch.NewPatchMetaFromOpenAPI(crdSchema) } // RespectIgnoreDifferences preserves ignored fields by copying their live // values into the target that is applied during sync. `status` must be // excluded from that copy: it is owned by the resource's own controller, // never by the sync. Merging live `status` into the apply makes the sync // field manager (ArgoCDSSAManager, "cd-controller") a co-owner of // `status` under server-side apply. For resources without a /status // subresource (e.g. apps.hanzo.ai/Application) this freezes a stale // status.operationState.phase that the controller can no longer correct. liveForPatch, normalizedLiveForPatch := live, normalized.Lives[idx] liveForPatch = liveForPatch.DeepCopy() unstructured.RemoveNestedField(liveForPatch.Object, "status") if normalizedLiveForPatch != nil { normalizedLiveForPatch = normalizedLiveForPatch.DeepCopy() unstructured.RemoveNestedField(normalizedLiveForPatch.Object, "status") } // Calculate live patch livePatch, err := getMergePatch(normalizedLiveForPatch, liveForPatch, lookupPatchMeta) if err != nil { return nil, err } patchedTarget, err := applyMergePatch(normalizedTarget, livePatch, versionedObject, lookupPatchMeta) if err != nil { return nil, err } // Restore non-ignored fields that may have been overwritten due to // patchStrategy:"replace" on a parent field (e.g. policy/v1 PDB selector). // Strategic merge patch treats "replace" fields as atomic, so patching in // one ignored sub-field pulls the entire parent from live, clobbering // non-ignored sibling fields. We detect and undo this here. var normalizedLiveObj map[string]any if normalized.Lives[idx] != nil { normalizedLiveObj = normalized.Lives[idx].Object } restoreNonIgnoredFields(patchedTarget.Object, originalTarget.Object, normalizedTarget.Object, normalizedLiveObj) patchedTargets = append(patchedTargets, patchedTarget) } return patchedTargets, nil } // getMergePatch calculates and returns the patch between the original and the // modified unstructures. func getMergePatch(original, modified *unstructured.Unstructured, lookupPatchMeta strategicpatch.LookupPatchMeta) ([]byte, error) { originalJSON, err := original.MarshalJSON() if err != nil { return nil, err } modifiedJSON, err := modified.MarshalJSON() if err != nil { return nil, err } if lookupPatchMeta != nil { return strategicpatch.CreateThreeWayMergePatch(modifiedJSON, modifiedJSON, originalJSON, lookupPatchMeta, true) } return jsonpatch.CreateMergePatch(originalJSON, modifiedJSON) } // applyMergePatch will apply the given patch in the obj and return the patched unstructure. func applyMergePatch(obj *unstructured.Unstructured, patch []byte, versionedObject any, meta strategicpatch.LookupPatchMeta) (*unstructured.Unstructured, error) { originalJSON, err := obj.MarshalJSON() if err != nil { return nil, err } var patchedJSON []byte switch { case versionedObject != nil: patchedJSON, err = strategicpatch.StrategicMergePatch(originalJSON, patch, versionedObject) case meta != nil: var originalMap, patchMap map[string]any if err := json.Unmarshal(originalJSON, &originalMap); err != nil { return nil, err } if err := json.Unmarshal(patch, &patchMap); err != nil { return nil, err } patchedMap, err := strategicpatch.StrategicMergeMapPatchUsingLookupPatchMeta(originalMap, patchMap, meta) if err != nil { return nil, err } patchedJSON, err = json.Marshal(patchedMap) if err != nil { return nil, err } default: patchedJSON, err = jsonpatch.MergePatch(originalJSON, patch) } if err != nil { return nil, err } patchedObj := &unstructured.Unstructured{} _, _, err = unstructured.UnstructuredJSONScheme.Decode(patchedJSON, nil, patchedObj) if err != nil { return nil, err } return patchedObj, nil } // restoreNonIgnoredFields walks patched, original, normalized (target), and // normalizedLive in parallel. It corrects three classes of collateral damage // caused by patchStrategy:"replace" treating a parent field as atomic: // // 1. Overwrite — a non-ignored value was replaced with the live value. // 2. Drop — a non-ignored key was removed because live lacks it. // 3. Add — a non-ignored live-only key leaked into the patched target. // // A field is considered "not ignored" when normalizedTarget == originalTarget // for that field (the normalizer left it alone). For live-only keys (pass 2), // a key is non-ignored if it exists in normalizedLive (the normalizer did not // strip it from live). func restoreNonIgnoredFields(patched, original, normalizedTarget, normalizedLive map[string]any) { // Pass 1: restore non-ignored fields that were overwritten or dropped. for key, originalVal := range original { patchedVal, inPatched := patched[key] normalizedVal, inNormalized := normalizedTarget[key] patchedMap, patchedIsMap := patchedVal.(map[string]any) originalMap, originalIsMap := originalVal.(map[string]any) normalizedMap, normalizedIsMap := normalizedVal.(map[string]any) if inPatched && patchedIsMap && originalIsMap && normalizedIsMap { var normalizedLiveMap map[string]any if v, ok := normalizedLive[key].(map[string]any); ok { normalizedLiveMap = v } restoreNonIgnoredFields(patchedMap, originalMap, normalizedMap, normalizedLiveMap) continue } // Leaf, type-changed, or missing field. // If normalized == original, the normalizer did not touch this field, // so it is not ignored and should keep the original (target) value. if inNormalized && reflect.DeepEqual(normalizedVal, originalVal) && (!inPatched || !reflect.DeepEqual(patchedVal, originalVal)) { patched[key] = originalVal } } // Pass 2: remove non-ignored keys that were introduced into patched from // the live object via replace-strategy collateral but do not exist in the // original target. A key is non-ignored if it exists in normalizedLive // (the normalizer did not strip it). Ignored live-only keys are kept — // they were intentionally copied by the livePatch. for key := range patched { if _, inOriginal := original[key]; inOriginal { continue } if _, inNormalizedLive := normalizedLive[key]; inNormalizedLive { delete(patched, key) } } } // hasSharedResourceCondition will check if the Application has any resource that has already // been synced by another Application. If the resource is found in another Application it returns // true along with a human readable message of which specific resource has this condition. func hasSharedResourceCondition(app *v1alpha1.Application) (bool, string) { for _, condition := range app.Status.Conditions { if condition.Type == v1alpha1.ApplicationConditionSharedResourceWarning { return true, condition.Message } } return false, "" } // delayBetweenSyncWaves is a gitops-engine SyncWaveHook which introduces an artificial delay // between each sync wave. We introduce an artificial delay in order give other controllers a // _chance_ to react to the spec change that we just applied. This is important because without // this, Hanzo CD will likely assess resource health too quickly (against the stale object), causing // hooks to fire prematurely. See: https://github.com/argoproj/argo-cd/issues/4669. // Note, this is not foolproof, since a proper fix would require the CRD record // status.observedGeneration coupled with a health.lua that verifies // status.observedGeneration == metadata.generation func delayBetweenSyncWaves(_ common.SyncPhase, _ int, finalWave bool) error { if !finalWave { delaySec := 2 if delaySecStr := os.Getenv(EnvVarSyncWaveDelay); delaySecStr != "" { if val, err := strconv.Atoi(delaySecStr); err == nil { delaySec = val } } duration := time.Duration(delaySec) * time.Second time.Sleep(duration) } return nil } func syncWindowPreventsSync(app *v1alpha1.Application, proj *v1alpha1.AppProject) (bool, error) { window := proj.Spec.SyncWindows.Matches(app) isManual := false var operationStartTime *time.Time if app.Status.OperationState != nil { isManual = !app.Status.OperationState.Operation.InitiatedBy.Automated if !app.Status.OperationState.StartedAt.IsZero() { t := app.Status.OperationState.StartedAt.Time operationStartTime = &t } } canSync, err := window.CanSync(isManual, operationStartTime) if err != nil { // prevents sync because sync window has an error return true, err } return !canSync, nil } // validateSyncPermissions checks whether the given resource is permitted by the project's // allow/deny lists and destination rules. It returns an error if the API resource info is nil // (preventing a nil-pointer panic), if the resource's group/kind is not permitted, or if // the resource's namespace is not an allowed destination. func validateSyncPermissions( project *v1alpha1.AppProject, destCluster *v1alpha1.Cluster, getProjectClusters func(string) ([]*v1alpha1.Cluster, error), un *unstructured.Unstructured, res *metav1.APIResource, ) error { if res == nil { return fmt.Errorf("failed to get API resource info for %s/%s: unable to verify permissions", un.GroupVersionKind().Group, un.GroupVersionKind().Kind) } if !project.IsGroupKindNamePermitted(un.GroupVersionKind().GroupKind(), un.GetName(), res.Namespaced) { return fmt.Errorf("resource %s:%s is not permitted in project %s", un.GroupVersionKind().Group, un.GroupVersionKind().Kind, project.Name) } if res.Namespaced { permitted, err := project.IsDestinationPermitted(destCluster, un.GetNamespace(), getProjectClusters) if err != nil { return err } if !permitted { return fmt.Errorf("namespace %v is not permitted in project '%s'", un.GetNamespace(), project.Name) } } return nil }