mirror of
https://github.com/argoproj/argo-cd.git
synced 2026-02-20 01:28:45 +01:00
607 lines
23 KiB
Go
607 lines
23 KiB
Go
package controller
|
|
|
|
import (
|
|
"context"
|
|
stderrors "errors"
|
|
"fmt"
|
|
"os"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"k8s.io/apimachinery/pkg/util/strategicpatch"
|
|
|
|
cdcommon "github.com/argoproj/argo-cd/v3/common"
|
|
|
|
gitopsDiff "github.com/argoproj/argo-cd/gitops-engine/pkg/diff"
|
|
"github.com/argoproj/argo-cd/gitops-engine/pkg/sync"
|
|
"github.com/argoproj/argo-cd/gitops-engine/pkg/sync/common"
|
|
"github.com/argoproj/argo-cd/gitops-engine/pkg/utils/kube"
|
|
jsonpatch "github.com/evanphx/json-patch"
|
|
log "github.com/sirupsen/logrus"
|
|
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/argoproj/argo-cd/v3/controller/metrics"
|
|
"github.com/argoproj/argo-cd/v3/controller/syncid"
|
|
"github.com/argoproj/argo-cd/v3/pkg/apis/application/v1alpha1"
|
|
applog "github.com/argoproj/argo-cd/v3/util/app/log"
|
|
"github.com/argoproj/argo-cd/v3/util/argo"
|
|
"github.com/argoproj/argo-cd/v3/util/argo/diff"
|
|
"github.com/argoproj/argo-cd/v3/util/glob"
|
|
kubeutil "github.com/argoproj/argo-cd/v3/util/kube"
|
|
logutils "github.com/argoproj/argo-cd/v3/util/log"
|
|
"github.com/argoproj/argo-cd/v3/util/lua"
|
|
)
|
|
|
|
const (
|
|
// EnvVarSyncWaveDelay is an environment variable which controls the delay in seconds between
|
|
// each sync-wave
|
|
EnvVarSyncWaveDelay = "ARGOCD_SYNC_WAVE_DELAY"
|
|
|
|
// serviceAccountDisallowedCharSet contains the characters that are not allowed to be present
|
|
// in a DefaultServiceAccount configured for a DestinationServiceAccount
|
|
serviceAccountDisallowedCharSet = "!*[]{}\\/"
|
|
)
|
|
|
|
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) {
|
|
clusterCache, err := m.liveStateCache.GetClusterCache(cluster)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("error getting cluster cache: %w", err)
|
|
}
|
|
|
|
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, clusterCache.GetOpenAPISchema(), 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(app *v1alpha1.Application, project *v1alpha1.AppProject, state *v1alpha1.OperationState) {
|
|
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(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 = argo.FormatAppConditions(errConditions)
|
|
return
|
|
}
|
|
|
|
destCluster, err := argo.GetDestinationCluster(context.Background(), 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
|
|
}
|
|
|
|
openAPISchema, err := m.getOpenAPISchema(destCluster)
|
|
if err != nil {
|
|
state.Phase = common.OperationError
|
|
state.Message = fmt.Sprintf("failed to load openAPISchema: %v", err)
|
|
return
|
|
}
|
|
|
|
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") {
|
|
patchedTargets, err := normalizeTargetResources(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 := deriveServiceAccountToImpersonate(project, app, destCluster)
|
|
if err != nil {
|
|
state.Phase = common.OperationError
|
|
state.Message = fmt.Sprintf("failed to find a matching service account to impersonate: %v", err)
|
|
return
|
|
}
|
|
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 {
|
|
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(), func(project string) ([]*v1alpha1.Cluster, error) {
|
|
return m.db.GetProjectClusters(context.TODO(), project)
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if !permitted {
|
|
return fmt.Errorf("namespace %v is not permitted in project '%s'", un.GetNamespace(), project.Name)
|
|
}
|
|
}
|
|
return nil
|
|
}),
|
|
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) ||
|
|
argo.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.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,
|
|
openAPISchema,
|
|
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()
|
|
} else {
|
|
syncCtx.Sync()
|
|
}
|
|
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 := argo.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(cr *comparisonResult) ([]*unstructured.Unstructured, error) {
|
|
// normalize live and target resources
|
|
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
|
|
}
|
|
originalTarget := cr.reconciliationResult.Target[idx]
|
|
if live == nil {
|
|
patchedTargets = append(patchedTargets, originalTarget)
|
|
continue
|
|
}
|
|
|
|
var lookupPatchMeta *strategicpatch.PatchMetaFromStruct
|
|
versionedObject, err := scheme.Scheme.New(normalizedTarget.GroupVersionKind())
|
|
if err == nil {
|
|
meta, err := strategicpatch.NewPatchMetaFromStruct(versionedObject)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
lookupPatchMeta = &meta
|
|
}
|
|
|
|
livePatch, err := getMergePatch(normalized.Lives[idx], live, lookupPatchMeta)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
normalizedTarget, err = applyMergePatch(normalizedTarget, livePatch, versionedObject)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
patchedTargets = append(patchedTargets, normalizedTarget)
|
|
}
|
|
return patchedTargets, nil
|
|
}
|
|
|
|
// getMergePatch calculates and returns the patch between the original and the
|
|
// modified unstructures.
|
|
func getMergePatch(original, modified *unstructured.Unstructured, lookupPatchMeta *strategicpatch.PatchMetaFromStruct) ([]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) (*unstructured.Unstructured, error) {
|
|
originalJSON, err := obj.MarshalJSON()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var patchedJSON []byte
|
|
if versionedObject == nil {
|
|
patchedJSON, err = jsonpatch.MergePatch(originalJSON, patch)
|
|
} else {
|
|
patchedJSON, err = strategicpatch.StrategicMergePatch(originalJSON, patch, versionedObject)
|
|
}
|
|
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
|
|
}
|
|
|
|
// 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, Argo 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
|
|
if app.Status.OperationState != nil {
|
|
isManual = !app.Status.OperationState.Operation.InitiatedBy.Automated
|
|
}
|
|
canSync, err := window.CanSync(isManual)
|
|
if err != nil {
|
|
// prevents sync because sync window has an error
|
|
return true, err
|
|
}
|
|
return !canSync, nil
|
|
}
|
|
|
|
// deriveServiceAccountToImpersonate determines the service account to be used for impersonation for the sync operation.
|
|
// The returned service account will be fully qualified including namespace and the service account name in the format system:serviceaccount:<namespace>:<service_account>
|
|
func deriveServiceAccountToImpersonate(project *v1alpha1.AppProject, application *v1alpha1.Application, destCluster *v1alpha1.Cluster) (string, error) {
|
|
// spec.Destination.Namespace is optional. If not specified, use the Application's
|
|
// namespace
|
|
serviceAccountNamespace := application.Spec.Destination.Namespace
|
|
if serviceAccountNamespace == "" {
|
|
serviceAccountNamespace = application.Namespace
|
|
}
|
|
// Loop through the destinationServiceAccounts and see if there is any destination that is a candidate.
|
|
// if so, return the service account specified for that destination.
|
|
for _, item := range project.Spec.DestinationServiceAccounts {
|
|
dstServerMatched, err := glob.MatchWithError(item.Server, destCluster.Server)
|
|
if err != nil {
|
|
return "", fmt.Errorf("invalid glob pattern for destination server: %w", err)
|
|
}
|
|
dstNamespaceMatched, err := glob.MatchWithError(item.Namespace, application.Spec.Destination.Namespace)
|
|
if err != nil {
|
|
return "", fmt.Errorf("invalid glob pattern for destination namespace: %w", err)
|
|
}
|
|
if dstServerMatched && dstNamespaceMatched {
|
|
if strings.Trim(item.DefaultServiceAccount, " ") == "" || strings.ContainsAny(item.DefaultServiceAccount, serviceAccountDisallowedCharSet) {
|
|
return "", fmt.Errorf("default service account contains invalid chars '%s'", item.DefaultServiceAccount)
|
|
} else if strings.Contains(item.DefaultServiceAccount, ":") {
|
|
// service account is specified along with its namespace.
|
|
return "system:serviceaccount:" + item.DefaultServiceAccount, nil
|
|
}
|
|
// service account needs to be prefixed with a namespace
|
|
return fmt.Sprintf("system:serviceaccount:%s:%s", serviceAccountNamespace, item.DefaultServiceAccount), nil
|
|
}
|
|
}
|
|
// if there is no match found in the AppProject.Spec.DestinationServiceAccounts, use the default service account of the destination namespace.
|
|
return "", fmt.Errorf("no matching service account found for destination server %s and namespace %s", application.Spec.Destination.Server, serviceAccountNamespace)
|
|
}
|