Skip to content

Issue #155 - Application update failes due to concurrent access #156

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
alexmt merged 1 commit into from
May 3, 2018
Merged

Issue #155 - Application update failes due to concurrent access #156

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion cmd/argocd/commands/app.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -217,7 +217,10 @@ func NewApplicationSetCommand(clientOpts *argocdclient.ClientOptions) *cobra.Com
os.Exit(1)
}
setParameterOverrides(app, appOpts.parameters)
_, err = appIf.Update(context.Background(), app)
_, err = appIf.UpdateSpec(context.Background(), &application.ApplicationSpecRequest{
AppName: app.Name,
Spec: &app.Spec,
})
errors.CheckError(err)
},
}
Expand Down
59 changes: 29 additions & 30 deletions controller/controller.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ import (
"k8s.io/apimachinery/pkg/fields"
"k8s.io/apimachinery/pkg/labels"
"k8s.io/apimachinery/pkg/selection"
"k8s.io/apimachinery/pkg/types"
"k8s.io/apimachinery/pkg/util/runtime"
"k8s.io/apimachinery/pkg/util/wait"
"k8s.io/apimachinery/pkg/watch"
Expand Down Expand Up @@ -222,26 +223,26 @@ func (ctrl *ApplicationController) processNextItem() bool {
if isForceRefreshed || app.NeedRefreshAppStatus(ctrl.statusRefreshTimeout) {
log.Infof("Refreshing application '%s' status (force refreshed: %v)", app.Name, isForceRefreshed)

status, err := ctrl.tryRefreshAppStatus(app.DeepCopy())
comparisonResult, parameters, err := ctrl.tryRefreshAppStatus(app.DeepCopy())
if err != nil {
status = app.Status.DeepCopy()
status.ComparisonResult = appv1.ComparisonResult{
comparisonResult = &appv1.ComparisonResult{
Status: appv1.ComparisonStatusError,
Error: fmt.Sprintf("Failed to get application status for application '%s': %v", app.Name, err),
ComparedTo: app.Spec.Source,
ComparedAt: metav1.Time{Time: time.Now().UTC()},
}
parameters = nil
}
ctrl.updateAppStatus(app.Name, app.Namespace, status)
ctrl.updateAppStatus(app.Name, app.Namespace, comparisonResult, parameters)
}

return true
}

func (ctrl *ApplicationController) tryRefreshAppStatus(app *appv1.Application) (*appv1.ApplicationStatus, error) {
func (ctrl *ApplicationController) tryRefreshAppStatus(app *appv1.Application) (*appv1.ComparisonResult, *[]appv1.ComponentParameter, error) {
conn, client, err := ctrl.repoClientset.NewRepositoryClient()
if err != nil {
return nil, err
return nil, nil, err
}
defer util.Close(conn)
repo, err := ctrl.apiRepoService.Get(context.Background(), &apireposerver.RepoQuery{Repo: app.Spec.Source.RepoURL})
Expand Down Expand Up @@ -271,14 +272,14 @@ func (ctrl *ApplicationController) tryRefreshAppStatus(app *appv1.Application) (
})
if err != nil {
log.Errorf("Failed to load application manifest %v", err)
return nil, err
return nil, nil, err
}
targetObjs := make([]*unstructured.Unstructured, len(manifestInfo.Manifests))
for i, manifestStr := range manifestInfo.Manifests {
var obj unstructured.Unstructured
if err := json.Unmarshal([]byte(manifestStr), &obj); err != nil {
if err != nil {
return nil, err
return nil, nil, err
}
}
targetObjs[i] = &obj
Expand All @@ -287,11 +288,10 @@ func (ctrl *ApplicationController) tryRefreshAppStatus(app *appv1.Application) (
server, namespace := argoutil.ResolveServerNamespace(app.Spec.Destination, manifestInfo)
comparisonResult, err := ctrl.appComparator.CompareAppState(server, namespace, targetObjs, app)
if err != nil {
return nil, err
return nil, nil, err
}
log.Infof("App %s comparison result: prev: %s. current: %s", app.Name, app.Status.ComparisonResult.Status, comparisonResult.Status)
newStatus := app.Status
newStatus.ComparisonResult = *comparisonResult

paramsReq := repository.EnvParamsRequest{
Repo: repo,
Revision: revision,
Expand All @@ -300,37 +300,36 @@ func (ctrl *ApplicationController) tryRefreshAppStatus(app *appv1.Application) (
}
params, err := client.GetEnvParams(context.Background(), &paramsReq)
if err != nil {
return nil, err
return nil, nil, err
}
newStatus.Parameters = make([]appv1.ComponentParameter, len(params.Params))
parameters := make([]appv1.ComponentParameter, len(params.Params))
for i := range params.Params {
newStatus.Parameters[i] = *params.Params[i]
parameters[i] = *params.Params[i]
}
return &newStatus, nil
return comparisonResult, &parameters, nil
}

func (ctrl *ApplicationController) runWorker() {
for ctrl.processNextItem() {
}
}

func (ctrl *ApplicationController) updateAppStatus(appName string, namespace string, status *appv1.ApplicationStatus) {
appKey := fmt.Sprintf("%s/%s", namespace, appName)
obj, exists, err := ctrl.appInformer.GetIndexer().GetByKey(appKey)
func (ctrl *ApplicationController) updateAppStatus(appName string, namespace string, comparisonResult *appv1.ComparisonResult, parameters *[]appv1.ComponentParameter) {
statusPatch := make(map[string]interface{})
statusPatch["comparisonResult"] = comparisonResult
statusPatch["parameters"] = parameters
patch, err := json.Marshal(map[string]interface{}{
"status": statusPatch,
})

if err == nil {
appClient := ctrl.applicationClientset.ArgoprojV1alpha1().Applications(namespace)
_, err = appClient.Patch(appName, types.MergePatchType, patch)
}
if err != nil {
log.Warnf("Failed to get application '%s' from informer index: %+v", appKey, err)
log.Warnf("Error updating application: %v", err)
} else {
if exists {
app := obj.(*appv1.Application).DeepCopy()
app.Status = *status
appClient := ctrl.applicationClientset.ArgoprojV1alpha1().Applications(namespace)
_, err := appClient.Update(app)
if err != nil {
log.Warnf("Error updating application: %v", err)
} else {
log.Info("Application update successful")
}
}
log.Info("Application update successful")
}
}

Expand Down
42 changes: 29 additions & 13 deletions server/application/application.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,7 +75,7 @@ func (s *Server) List(ctx context.Context, q *ApplicationQuery) (*appv1.Applicat

// Create creates an application
func (s *Server) Create(ctx context.Context, a *appv1.Application) (*appv1.Application, error) {
err := s.validateApp(ctx, a)
err := s.validateApp(ctx, &a.Spec)
if err != nil {
return nil, err
}
Expand All @@ -89,13 +89,29 @@ func (s *Server) Get(ctx context.Context, q *ApplicationQuery) (*appv1.Applicati

// Update updates an application
func (s *Server) Update(ctx context.Context, a *appv1.Application) (*appv1.Application, error) {
err := s.validateApp(ctx, a)
err := s.validateApp(ctx, &a.Spec)
if err != nil {
return nil, err
}
return s.appclientset.ArgoprojV1alpha1().Applications(s.ns).Update(a)
}

// UpdateSpec updates an application spec
func (s *Server) UpdateSpec(ctx context.Context, q *ApplicationSpecRequest) (*appv1.ApplicationSpec, error) {
err := s.validateApp(ctx, q.Spec)
if err != nil {
return nil, err
}
patch, err := json.Marshal(map[string]appv1.ApplicationSpec{
"spec": *q.Spec,
})
if err != nil {
return nil, err
}
_, err = s.appclientset.ArgoprojV1alpha1().Applications(s.ns).Patch(q.AppName, types.MergePatchType, patch)
return q.Spec, err
}

// Delete removes an application and all associated resources
func (s *Server) Delete(ctx context.Context, q *DeleteApplicationRequest) (*ApplicationResponse, error) {
var err error
Expand Down Expand Up @@ -164,20 +180,20 @@ func (s *Server) Watch(q *ApplicationQuery, ws ApplicationService_WatchServer) e
// * the git path contains a valid app.yaml
// * the specified environment exists
// * the referenced cluster has been added to ArgoCD
func (s *Server) validateApp(ctx context.Context, a *appv1.Application) error {
func (s *Server) validateApp(ctx context.Context, spec *appv1.ApplicationSpec) error {
// Test the repo
conn, repoClient, err := s.repoClientset.NewRepositoryClient()
if err != nil {
return err
}
defer util.Close(conn)
repoRes, err := s.repoService.Get(ctx, &apirepository.RepoQuery{Repo: a.Spec.Source.RepoURL})
repoRes, err := s.repoService.Get(ctx, &apirepository.RepoQuery{Repo: spec.Source.RepoURL})
if err != nil {
if errStatus, ok := status.FromError(err); ok && errStatus.Code() == codes.NotFound {
// The repo has not been added to ArgoCD so we do not have credentials to access it.
// We support the mode where apps can be created from public repositories. Test the
// repo to make sure it is publically accessible
err = git.TestRepo(a.Spec.Source.RepoURL, "", "", "")
// repo to make sure it is publicly accessible
err = git.TestRepo(spec.Source.RepoURL, "", "", "")
if err != nil {
return err
}
Expand All @@ -189,10 +205,10 @@ func (s *Server) validateApp(ctx context.Context, a *appv1.Application) error {
// Verify app.yaml is functional
req := repository.KsonnetAppRequest{
Repo: &appv1.Repository{
Repo: a.Spec.Source.RepoURL,
Repo: spec.Source.RepoURL,
},
Revision: a.Spec.Source.TargetRevision,
Path: a.Spec.Source.Path,
Revision: spec.Source.TargetRevision,
Path: spec.Source.Path,
}
if repoRes != nil {
req.Repo.Username = repoRes.Username
Expand All @@ -205,15 +221,15 @@ func (s *Server) validateApp(ctx context.Context, a *appv1.Application) error {
}

// Verify the specified environment is defined in it
envSpec, ok := ksAppRes.Environments[a.Spec.Source.Environment]
envSpec, ok := ksAppRes.Environments[spec.Source.Environment]
if !ok {
return status.Errorf(codes.InvalidArgument, "environment '%s' does not exist in app", a.Spec.Source.Environment)
return status.Errorf(codes.InvalidArgument, "environment '%s' does not exist in app", spec.Source.Environment)
}
// Ensure the k8s cluster the app is referencing, is configured in ArgoCD
// NOTE: need to check if it was overridden in the destination spec
clusterURL := envSpec.Destination.Server
if a.Spec.Destination != nil && a.Spec.Destination.Server != "" {
clusterURL = a.Spec.Destination.Server
if spec.Destination != nil && spec.Destination.Server != "" {
clusterURL = spec.Destination.Server
}
_, err = s.clusterService.Get(ctx, &cluster.ClusterQuery{Server: clusterURL})
if err != nil {
Expand Down
Loading