Move sync logic to contoller

Dockerfile-ci-builder

@@ -11,7 +11,8 @@ RUN curl -O https://get.docker.com/builds/Linux/x86_64/docker-1.13.1.tgz && \
   gometalinter.v2 --install
 
 # Install kubectl
-RUN curl -o /usr/local/bin/kubectl -LO https://storage.googleapis.com/kubernetes-release/release/$(curl -s https://storage.googleapis.com/kubernetes-release/release/stable.txt)/bin/linux/amd64/kubectl
+RUN curl -o /kubectl -LO https://storage.googleapis.com/kubernetes-release/release/$(curl -s https://storage.googleapis.com/kubernetes-release/release/stable.txt)/bin/linux/amd64/kubectl && \
+    chmod +x /kubectl && mv /kubectl /usr/local/bin/kubectl
 
 # Install ksonnet
 env KSONNET_VERSION=0.10.1

cmd/argocd-application-controller/main.go

@@ -8,11 +8,10 @@ import (
 	"strconv"
 	"time"
 
-	argocd "github.com/argoproj/argo-cd"
+	"github.com/argoproj/argo-cd"
 	"github.com/argoproj/argo-cd/controller"
 	"github.com/argoproj/argo-cd/errors"
 	appclientset "github.com/argoproj/argo-cd/pkg/client/clientset/versioned"
-	"github.com/argoproj/argo-cd/reposerver"
 	"github.com/argoproj/argo-cd/server/cluster"
 	apirepository "github.com/argoproj/argo-cd/server/repository"
 	"github.com/argoproj/argo-cd/util/cli"
@@ -24,6 +23,7 @@ import (
 	// load the gcp plugin (required to authenticate against GKE clusters).
 	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
 	// load the oidc plugin (required to authenticate with OpenID Connect).
+	"github.com/argoproj/argo-cd/reposerver"
 	_ "k8s.io/client-go/plugin/pkg/client/auth/oidc"
 )
 
@@ -36,12 +36,13 @@ const (
 
 func newCommand() *cobra.Command {
 	var (
-		clientConfig      clientcmd.ClientConfig
-		appResyncPeriod   int64
-		repoServerAddress string
-		workers           int
-		logLevel          string
-		glogLevel         int
+		clientConfig        clientcmd.ClientConfig
+		appResyncPeriod     int64
+		repoServerAddress   string
+		statusProcessors    int
+		operationProcessors int
+		logLevel            string
+		glogLevel           int
 	)
 	var command = cobra.Command{
 		Use:   cliName,
@@ -71,27 +72,28 @@ func newCommand() *cobra.Command {
 				InstanceID: "",
 			}
 			resyncDuration := time.Duration(appResyncPeriod) * time.Second
-			apiRepoServer := apirepository.NewServer(namespace, kubeClient, appClient)
 			apiClusterServer := cluster.NewServer(namespace, kubeClient, appClient)
 			clusterService := cluster.NewServer(namespace, kubeClient, appClient)
-			appComparator := controller.NewKsonnetAppComparator(clusterService)
+			apiRepoServer := apirepository.NewServer(namespace, kubeClient, appClient)
+			appStateManager := controller.NewAppStateManager(
+				clusterService, apiRepoServer, appClient, reposerver.NewRepositoryServerClientset(repoServerAddress), namespace)
+			appHealthManager := controller.NewAppHealthManager(clusterService, namespace)
 
 			appController := controller.NewApplicationController(
 				namespace,
 				kubeClient,
 				appClient,
-				reposerver.NewRepositoryServerClientset(repoServerAddress),
-				apiRepoServer,
 				apiClusterServer,
-				appComparator,
+				appStateManager,
+				appHealthManager,
 				resyncDuration,
 				&controllerConfig)
 
 			ctx, cancel := context.WithCancel(context.Background())
 			defer cancel()
 
 			log.Infof("Application Controller (version: %s) starting (namespace: %s)", argocd.GetVersion(), namespace)
-			go appController.Run(ctx, workers)
+			go appController.Run(ctx, statusProcessors, operationProcessors)
 			// Wait forever
 			select {}
 		},
@@ -100,7 +102,8 @@ func newCommand() *cobra.Command {
 	clientConfig = cli.AddKubectlFlagsToCmd(&command)
 	command.Flags().Int64Var(&appResyncPeriod, "app-resync", defaultAppResyncPeriod, "Time period in seconds for application resync.")
 	command.Flags().StringVar(&repoServerAddress, "repo-server", "localhost:8081", "Repo server address.")
-	command.Flags().IntVar(&workers, "workers", 1, "Number of application workers")
+	command.Flags().IntVar(&statusProcessors, "status-processors", 1, "Number of application status processors")
+	command.Flags().IntVar(&operationProcessors, "operation-processors", 1, "Number of application operation processors")
 	command.Flags().StringVar(&logLevel, "loglevel", "info", "Set the logging level. One of: debug|info|warn|error")
 	command.Flags().IntVar(&glogLevel, "gloglevel", 0, "Set the glog logging level")
 	return &command

cmd/argocd-server/commands/root.go

@@ -1,6 +1,8 @@
 package commands
 
 import (
+	"context"
+
 	"github.com/argoproj/argo-cd/errors"
 	appclientset "github.com/argoproj/argo-cd/pkg/client/clientset/versioned"
 	"github.com/argoproj/argo-cd/reposerver"
@@ -51,7 +53,10 @@ func NewCommand() *cobra.Command {
 				DisableAuth:     disableAuth,
 			}
 			argocd := server.NewServer(argoCDOpts)
-			argocd.Run()
+			ctx := context.Background()
+			ctx, cancel := context.WithCancel(ctx)
+			defer cancel()
+			argocd.Run(ctx, 8080)
 		},
 	}
 

cmd/argocd/commands/app.go

@@ -366,8 +366,13 @@ func NewApplicationSyncCommand(clientOpts *argocdclient.ClientOptions) *cobra.Co
 				Revision: revision,
 				Prune:    prune,
 			}
-			syncRes, err := appIf.Sync(context.Background(), &syncReq)
+			_, err := appIf.Sync(context.Background(), &syncReq)
 			errors.CheckError(err)
+
+			status, err := waitUntilOperationCompleted(appIf, appName)
+			errors.CheckError(err)
+			syncRes := status.SyncResult
+
 			fmt.Printf("%s %s\n", appName, syncRes.Message)
 			w := tabwriter.NewWriter(os.Stdout, 0, 0, 2, ' ', 0)
 			fmt.Fprintf(w, "NAME\tKIND\tMESSAGE\n")
@@ -383,6 +388,33 @@ func NewApplicationSyncCommand(clientOpts *argocdclient.ClientOptions) *cobra.Co
 	return command
 }
 
+func waitUntilOperationCompleted(appClient application.ApplicationServiceClient, appName string) (*argoappv1.OperationState, error) {
+	wc, err := appClient.Watch(context.Background(), &application.ApplicationQuery{
+		Name: appName,
+	})
+	if err != nil {
+		return nil, err
+	}
+	appEvent, err := wc.Recv()
+	if err != nil {
+		return nil, err
+	}
+	for {
+		if appEvent.Application.Status.OperationState != nil && appEvent.Application.Status.OperationState.Status != argoappv1.OperationStatusInProgress {
+			if appEvent.Application.Status.OperationState.Status == argoappv1.OperationStatusFailed {
+				return nil, fmt.Errorf("operation failed: %s", appEvent.Application.Status.OperationState.ErrorDetails)
+			} else {
+				return appEvent.Application.Status.OperationState, nil
+			}
+		} else {
+			appEvent, err = wc.Recv()
+			if err != nil {
+				return nil, err
+			}
+		}
+	}
+}
+
 // setParameterOverrides updates an existing or appends a new parameter override in the application
 func setParameterOverrides(app *argoappv1.Application, parameters []string) {
 	if len(parameters) == 0 {
@@ -489,12 +521,18 @@ func NewApplicationRollbackCommand(clientOpts *argocdclient.ClientOptions) *cobr
 			if depInfo == nil {
 				log.Fatalf("Application '%s' does not have deployment id '%d' in history\n", app.ObjectMeta.Name, depID)
 			}
-			syncRes, err := appIf.Rollback(ctx, &application.ApplicationRollbackRequest{
+
+			_, err = appIf.Rollback(ctx, &application.ApplicationRollbackRequest{
 				Name:  appName,
 				ID:    int64(depID),
 				Prune: prune,
 			})
 			errors.CheckError(err)
+
+			status, err := waitUntilOperationCompleted(appIf, appName)
+			errors.CheckError(err)
+			syncRes := status.SyncResult
+
 			fmt.Printf("%s %s\n", appName, syncRes.Message)
 			w := tabwriter.NewWriter(os.Stdout, 0, 0, 2, ' ', 0)
 			fmt.Fprintf(w, "NAME\tKIND\tMESSAGE\n")