From 56023cf3997332776b3aec4c1a6251c60edd43c7 Mon Sep 17 00:00:00 2001
From: Neel Dalsania <neel.dalsania@astronomer.io>
Date: Fri, 15 Jul 2022 22:17:17 +0530
Subject: [PATCH] updated triggerer replicas defaults in create & update
 deployment flows (#648)

---
 cmd/software/deployment.go        | 11 ++++++-----
 software/deployment/deployment.go | 22 ++++++++++++++++++----
 2 files changed, 24 insertions(+), 9 deletions(-)

diff --git a/cmd/software/deployment.go b/cmd/software/deployment.go
index 67e66354d..4f29f001a 100644
--- a/cmd/software/deployment.go
+++ b/cmd/software/deployment.go
@@ -35,7 +35,8 @@ var (
 	releaseName             string
 	nfsLocation             string
 	dagDeploymentType       string
-	triggererReplicas       int
+	createTriggererReplicas int
+	updateTriggererReplicas int
 	gitRevision             string
 	gitRepoURL              string
 	gitBranchName           string
@@ -141,7 +142,7 @@ func newDeploymentCreateCmd(out io.Writer) *cobra.Command {
 	}
 
 	if triggererEnabled {
-		cmd.Flags().IntVarP(&triggererReplicas, "triggerer-replicas", "", 0, "Number of replicas to use for triggerer airflow component, valid 0-2")
+		cmd.Flags().IntVarP(&createTriggererReplicas, "triggerer-replicas", "", 1, "Number of replicas to use for triggerer airflow component, valid 0-2")
 	}
 
 	if runtimeEnabled {
@@ -228,7 +229,7 @@ $ astro deployment update [deployment ID] --dag-deployment-type=volume --nfs-loc
 	}
 
 	if triggererEnabled {
-		cmd.Flags().IntVarP(&triggererReplicas, "triggerer-replicas", "", 0, "Number of replicas to use for triggerer airflow component, valid 0-2")
+		cmd.Flags().IntVarP(&updateTriggererReplicas, "triggerer-replicas", "", -1, "Number of replicas to use for triggerer airflow component, valid 0-2")
 	}
 
 	//noline:dupl
@@ -388,7 +389,7 @@ func deploymentCreate(cmd *cobra.Command, out io.Writer) error {
 		SSHKey:            sshKey,
 		KnownHosts:        knowHosts,
 		GitSyncInterval:   gitSyncInterval,
-		TriggererReplicas: triggererReplicas,
+		TriggererReplicas: createTriggererReplicas,
 	}
 	return deployment.Create(req, houstonClient, out)
 }
@@ -459,7 +460,7 @@ func deploymentUpdate(cmd *cobra.Command, args []string, dagDeploymentType, nfsL
 		}
 	}
 
-	return deployment.Update(args[0], cloudRole, argsMap, dagDeploymentType, nfsLocation, gitRepoURL, gitRevision, gitBranchName, gitDAGDir, sshKey, knowHosts, executorType, gitSyncInterval, triggererReplicas, houstonClient, out)
+	return deployment.Update(args[0], cloudRole, argsMap, dagDeploymentType, nfsLocation, gitRepoURL, gitRevision, gitBranchName, gitDAGDir, sshKey, knowHosts, executorType, gitSyncInterval, updateTriggererReplicas, houstonClient, out)
 }
 
 func deploymentAirflowUpgrade(cmd *cobra.Command, out io.Writer) error {
diff --git a/software/deployment/deployment.go b/software/deployment/deployment.go
index 48a91f7bb..cfe71669b 100644
--- a/software/deployment/deployment.go
+++ b/software/deployment/deployment.go
@@ -35,6 +35,8 @@ var (
 	errDeploymentAlreadyOnRuntime = errors.New("deployment is already using runtime image")
 	errRuntimeUpdateFailed        = errors.New("failed to update the deployment runtime version")
 	errInvalidAirflowVersion      = errors.New("invalid Airflow version to migrate the deployment to Runtime, please upgrade the deployment to 2.2.4 Airflow version before trying to migrate to Runtime image")
+
+	triggererAllowedAirflowVersion = semver.MustParse("2.2.0")
 )
 
 const (
@@ -138,6 +140,19 @@ func CheckTriggererEnabled(client houston.ClientInterface) bool {
 	return config.Flags.TriggererEnabled
 }
 
+func addTriggererReplicasArg(vars map[string]interface{}, client houston.ClientInterface, airflowVersion string, triggererReplicas int) {
+	// reset triggerer count to zero in case airflowVersion < 2.2.0
+	if airflowVersion != "" {
+		if version := semver.MustParse(airflowVersion); version != nil && version.LessThan(triggererAllowedAirflowVersion) {
+			triggererReplicas = 0
+		}
+	}
+
+	if CheckTriggererEnabled(client) {
+		vars["triggererReplicas"] = triggererReplicas
+	}
+}
+
 // Create airflow deployment
 func Create(req *CreateDeploymentRequest, client houston.ClientInterface, out io.Writer) error {
 	vars := map[string]interface{}{"label": req.Label, "workspaceId": req.WS, "executor": req.Executor, "cloudRole": req.CloudRole}
@@ -173,9 +188,8 @@ func Create(req *CreateDeploymentRequest, client houston.ClientInterface, out io
 		return err
 	}
 
-	if CheckTriggererEnabled(client) {
-		vars["triggererReplicas"] = req.TriggererReplicas
-	}
+	addTriggererReplicasArg(vars, client, req.AirflowVersion, req.TriggererReplicas)
+
 	d, err := client.CreateDeployment(vars)
 	if err != nil {
 		return err
@@ -338,7 +352,7 @@ func Update(id, cloudRole string, args map[string]string, dagDeploymentType, nfs
 		return err
 	}
 
-	if CheckTriggererEnabled(client) {
+	if CheckTriggererEnabled(client) && triggererReplicas != -1 {
 		vars["triggererReplicas"] = triggererReplicas
 	}