use round robin to distribute requests to all repo server pods (#353)

* use round robin to distribute requests to all repo server pods

* try to open repo server connections when building manifests, not before

if we avoid sharing, this should let k8s build new connections to new pods

* add comment, check error

pkg/argo_client/client.go

@@ -23,13 +23,10 @@ import (
 )
 
 type ArgoClient struct {
-	client apiclient.Client
-
-	repoClient         repoapiclient.RepoServerServiceClient
-	namespace          string
-	k8s                kubernetes.Interface
-	k8sConfig          *rest.Config
-	sendFullRepository bool
+	client    apiclient.Client
+	k8s       kubernetes.Interface
+	k8sConfig *rest.Config
+	cfg       config.ServerConfig
 }
 
 func NewArgoClient(
@@ -56,25 +53,27 @@ func NewArgoClient(
 		return nil, err
 	}
 
+	return &ArgoClient{
+		cfg:       cfg,
+		client:    argo,
+		k8s:       k8s.ClientSet(),
+		k8sConfig: k8s.Config(),
+	}, nil
+}
+
+func (a *ArgoClient) createRepoServerClient() (repoapiclient.RepoServerServiceClient, *grpc.ClientConn, error) {
 	log.Info().Msg("creating client")
-	tlsConfig := tls.Config{InsecureSkipVerify: cfg.ArgoCDRepositoryInsecure}
-	conn, err := grpc.NewClient(cfg.ArgoCDRepositoryEndpoint,
+	tlsConfig := tls.Config{InsecureSkipVerify: a.cfg.ArgoCDRepositoryInsecure}
+	conn, err := grpc.NewClient(a.cfg.ArgoCDRepositoryEndpoint,
 		grpc.WithTransportCredentials(
 			credentials.NewTLS(&tlsConfig),
 		),
 	)
 	if err != nil {
-		return nil, errors.Wrap(err, "failed to create client")
+		return nil, nil, errors.Wrap(err, "failed to create client")
 	}
 
-	return &ArgoClient{
-		repoClient:         repoapiclient.NewRepoServerServiceClient(conn),
-		client:             argo,
-		namespace:          cfg.ArgoCDNamespace,
-		k8s:                k8s.ClientSet(),
-		k8sConfig:          k8s.Config(),
-		sendFullRepository: cfg.ArgoCDSendFullRepository,
-	}, nil
+	return repoapiclient.NewRepoServerServiceClient(conn), conn, nil
 }
 
 // GetApplicationClient has related argocd diff code https://github.com/argoproj/argo-cd/blob/d3ff9757c460ae1a6a11e1231251b5d27aadcdd1/cmd/argocd/commands/app.go#L899

pkg/argo_client/manifests.go

@@ -126,8 +126,8 @@ func (a *ArgoClient) generateManifests(ctx context.Context, app v1alpha1.Applica
 		return nil, errors.Wrap(err, "failed to get settings")
 	}
 
-	settingsMgr := argosettings.NewSettingsManager(ctx, a.k8s, a.namespace)
-	argoDB := db.NewDB(a.namespace, settingsMgr, a.k8s)
+	settingsMgr := argosettings.NewSettingsManager(ctx, a.k8s, a.cfg.ArgoCDNamespace)
+	argoDB := db.NewDB(a.cfg.ArgoCDNamespace, settingsMgr, a.k8s)
 
 	repoTarget := source.TargetRevision
 	if pkg.AreSameRepos(source.RepoURL, pullRequest.CloneURL) && areSameTargetRef(source.TargetRevision, pullRequest.BaseRef) {
@@ -141,7 +141,7 @@ func (a *ArgoClient) generateManifests(ctx context.Context, app v1alpha1.Applica
 	}
 
 	var packageDir string
-	if a.sendFullRepository {
+	if a.cfg.ArgoCDSendFullRepository {
 		log.Info().Msg("sending full repository")
 		packageDir = repo.Directory
 	} else {
@@ -226,11 +226,25 @@ func (a *ArgoClient) generateManifests(ctx context.Context, app v1alpha1.Applica
 		RefSources:         refSources,
 	}
 
+	// creating a new client forces grpc to create a new connection, which causes
+	//the k8s load balancer to select a new pod, balancing requests among all repo-server pods.
+	repoClient, conn, err := a.createRepoServerClient()
+	if err != nil {
+		return nil, errors.Wrap(err, "error creating repo client")
+	}
+	defer conn.Close()
+
 	log.Info().Msg("generating manifest with files")
-	stream, err := a.repoClient.GenerateManifestWithFiles(ctx)
+	stream, err := repoClient.GenerateManifestWithFiles(ctx)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to get manifests with files")
 	}
+	defer func() {
+		err := stream.CloseSend()
+		if err != nil {
+			log.Error().Err(err).Msg("failed to close stream")
+		}
+	}()
 
 	log.Info().Msg("sending request")
 	if err := stream.Send(&repoapiclient.ManifestRequestWithFiles{