Repository: cortexlabs/cortex (language: python)

Issue:
Support creating and deleting nodegroups in an existing cluster



Discussion / hints:

Fix patch:
diff --git a/cli/cmd/cluster.go b/cli/cmd/cluster.go
index c1b50923d6..0a6b48db53 100644
--- a/cli/cmd/cluster.go
+++ b/cli/cmd/cluster.go
@@ -56,13 +56,11 @@ import (
 var (
 	_flagClusterUpEnv                string
 	_flagClusterInfoEnv              string
-	_flagClusterScaleNodeGroup       string
-	_flagClusterScaleMinInstances    int64
-	_flagClusterScaleMaxInstances    int64
 	_flagClusterConfig               string
 	_flagClusterName                 string
 	_flagClusterRegion               string
 	_flagClusterInfoDebug            bool
+	_flagClusterInfoPrintConfig      bool
 	_flagClusterDisallowPrompt       bool
 	_flagClusterDownKeepAWSResources bool
 )
@@ -79,18 +77,16 @@ func clusterInit() {
 	addClusterConfigFlag(_clusterInfoCmd)
 	addClusterNameFlag(_clusterInfoCmd)
 	addClusterRegionFlag(_clusterInfoCmd)
-	_clusterInfoCmd.Flags().VarP(&_flagOutput, "output", "o", fmt.Sprintf("output format: one of %s", strings.Join(flags.UserOutputTypeStrings(), "|")))
+	_clusterInfoCmd.Flags().VarP(&_flagOutput, "output", "o", fmt.Sprintf("output format: one of %s", strings.Join(flags.OutputTypeStrings(), "|")))
 	_clusterInfoCmd.Flags().StringVarP(&_flagClusterInfoEnv, "configure-env", "e", "", "name of environment to configure")
 	_clusterInfoCmd.Flags().BoolVarP(&_flagClusterInfoDebug, "debug", "d", false, "save the current cluster state to a file")
+	_clusterInfoCmd.Flags().BoolVarP(&_flagClusterInfoPrintConfig, "print-config", "", false, "print the cluster config")
 	_clusterInfoCmd.Flags().BoolVarP(&_flagClusterDisallowPrompt, "yes", "y", false, "skip prompts")
 	_clusterCmd.AddCommand(_clusterInfoCmd)
 
-	_clusterScaleCmd.Flags().SortFlags = false
-	addClusterNameFlag(_clusterScaleCmd)
-	addClusterRegionFlag(_clusterScaleCmd)
-	addClusterScaleFlags(_clusterScaleCmd)
-	_clusterScaleCmd.Flags().BoolVarP(&_flagClusterDisallowPrompt, "yes", "y", false, "skip prompts")
-	_clusterCmd.AddCommand(_clusterScaleCmd)
+	_clusterConfigureCmd.Flags().SortFlags = false
+	_clusterConfigureCmd.Flags().BoolVarP(&_flagClusterDisallowPrompt, "yes", "y", false, "skip prompts")
+	_clusterCmd.AddCommand(_clusterConfigureCmd)
 
 	_clusterDownCmd.Flags().SortFlags = false
 	addClusterConfigFlag(_clusterDownCmd)
@@ -120,13 +116,6 @@ func addClusterRegionFlag(cmd *cobra.Command) {
 	cmd.Flags().StringVarP(&_flagClusterRegion, "region", "r", "", "aws region of the cluster")
 }
 
-func addClusterScaleFlags(cmd *cobra.Command) {
-	cmd.Flags().StringVar(&_flagClusterScaleNodeGroup, "node-group", "", "name of the node group to scale")
-	cmd.MarkFlagRequired("node-group")
-	cmd.Flags().Int64Var(&_flagClusterScaleMinInstances, "min-instances", 0, "minimum number of instances")
-	cmd.Flags().Int64Var(&_flagClusterScaleMaxInstances, "max-instances", 0, "maximum number of instances")
-}
-
 var _clusterCmd = &cobra.Command{
 	Use:   "cluster",
 	Short: "manage cortex clusters (contains subcommands)",
@@ -172,21 +161,25 @@ var _clusterUpCmd = &cobra.Command{
 			exit.Error(err)
 		}
 
-		clusterConfig, err := getInstallClusterConfig(awsClient, clusterConfigFile, _flagClusterDisallowPrompt)
+		stacks, err := clusterstate.GetClusterStacks(awsClient, accessConfig)
 		if err != nil {
 			exit.Error(err)
 		}
 
-		clusterState, err := clusterstate.GetClusterState(awsClient, accessConfig)
-		if err != nil {
+		state := clusterstate.GetClusterState(stacks)
+		if err := clusterstate.AssertClusterState(stacks, state, clusterstate.StateClusterDoesntExist); err != nil {
 			exit.Error(err)
 		}
 
-		err = clusterstate.AssertClusterStatus(accessConfig.ClusterName, accessConfig.Region, clusterState.Status, clusterstate.StatusNotFound, clusterstate.StatusDeleteComplete)
+		promptIfNotAdmin(awsClient, _flagClusterDisallowPrompt)
+
+		clusterConfig, err := getInstallClusterConfig(awsClient, clusterConfigFile)
 		if err != nil {
 			exit.Error(err)
 		}
 
+		confirmInstallClusterConfig(clusterConfig, awsClient, _flagClusterDisallowPrompt)
+
 		err = createS3BucketIfNotFound(awsClient, clusterConfig.Bucket, clusterConfig.Tags)
 		if err != nil {
 			exit.Error(err)
@@ -313,29 +306,20 @@ var _clusterUpCmd = &cobra.Command{
 	},
 }
 
-var _clusterScaleCmd = &cobra.Command{
-	Use:   "scale [flags]",
-	Short: "update the min/max instances for a nodegroup",
-	Args:  cobra.NoArgs,
+var _clusterConfigureCmd = &cobra.Command{
+	Use:   "configure CLUSTER_CONFIG_FILE",
+	Short: "update the cluster's configuration",
+	Args:  cobra.ExactArgs(1),
 	Run: func(cmd *cobra.Command, args []string) {
 		telemetry.Event("cli.cluster.configure")
 
-		var scaleMinIntances, scaleMaxInstances *int64
-		if wasFlagProvided(cmd, "min-instances") {
-			scaleMinIntances = pointer.Int64(_flagClusterScaleMinInstances)
-		}
-		if wasFlagProvided(cmd, "max-instances") {
-			scaleMaxInstances = pointer.Int64(_flagClusterScaleMaxInstances)
-		}
-		if scaleMinIntances == nil && scaleMaxInstances == nil {
-			exit.Error(ErrorSpecifyAtLeastOneFlag("--min-instances", "--max-instances"))
-		}
+		clusterConfigFile := args[0]
 
 		if _, err := docker.GetDockerClient(); err != nil {
 			exit.Error(err)
 		}
 
-		accessConfig, err := getClusterAccessConfigWithCache()
+		accessConfig, err := getClusterAccessConfigWithCache(false)
 		if err != nil {
 			exit.Error(err)
 		}
@@ -345,35 +329,50 @@ var _clusterScaleCmd = &cobra.Command{
 			exit.Error(err)
 		}
 
-		clusterState, err := clusterstate.GetClusterState(awsClient, accessConfig)
+		stacks, err := clusterstate.GetClusterStacks(awsClient, accessConfig)
 		if err != nil {
 			exit.Error(err)
 		}
 
-		err = clusterstate.AssertClusterStatus(accessConfig.ClusterName, accessConfig.Region, clusterState.Status, clusterstate.StatusCreateComplete, clusterstate.StatusUpdateComplete, clusterstate.StatusUpdateRollbackComplete)
-		if err != nil {
+		state := clusterstate.GetClusterState(stacks)
+		if err := clusterstate.AssertClusterState(stacks, state, clusterstate.StateClusterExists); err != nil {
 			exit.Error(err)
 		}
 
-		clusterConfig := refreshCachedClusterConfig(*awsClient, accessConfig, true)
-		clusterConfig, ngIndex, err := updateNodeGroupScale(clusterConfig, _flagClusterScaleNodeGroup, scaleMinIntances, scaleMaxInstances, _flagClusterDisallowPrompt)
+		oldClusterConfig := refreshCachedClusterConfig(awsClient, accessConfig, true)
+
+		promptIfNotAdmin(awsClient, _flagClusterDisallowPrompt)
+
+		newClusterConfig, configureChanges, err := getConfigureClusterConfig(awsClient, stacks, oldClusterConfig, clusterConfigFile)
 		if err != nil {
 			exit.Error(err)
 		}
 
-		out, exitCode, err := runManagerWithClusterConfig("/root/install.sh --update", &clusterConfig, awsClient, nil, nil, []string{
-			"CORTEX_SCALING_NODEGROUP=" + _flagClusterScaleNodeGroup,
-			"CORTEX_SCALING_MIN_INSTANCES=" + s.Int64(clusterConfig.NodeGroups[ngIndex].MinInstances),
-			"CORTEX_SCALING_MAX_INSTANCES=" + s.Int64(clusterConfig.NodeGroups[ngIndex].MaxInstances),
+		if !configureChanges.HasChanges() {
+			fmt.Println("your cluster is already up to date")
+			exit.Ok()
+		}
+
+		confirmConfigureClusterConfig(configureChanges, oldClusterConfig, *newClusterConfig, _flagClusterDisallowPrompt)
+
+		out, exitCode, err := runManagerWithClusterConfig("/root/install.sh --configure", newClusterConfig, awsClient, nil, nil, []string{
+			"CORTEX_NODEGROUP_NAMES_TO_SCALE=" + strings.Join(configureChanges.NodeGroupsToScale, " "),          // NodeGroupsToScale contain the cluster config node-group names
+			"CORTEX_NODEGROUP_NAMES_TO_ADD=" + strings.Join(configureChanges.NodeGroupsToAdd, " "),              // NodeGroupsToAdd contain the cluster config node-group names
+			"CORTEX_EKS_NODEGROUP_NAMES_TO_REMOVE=" + strings.Join(configureChanges.EKSNodeGroupsToRemove, " "), // EKSNodeGroupsToRemove contain the EKS node-group names
 		})
 		if err != nil {
 			exit.Error(err)
 		}
 		if exitCode == nil || *exitCode != 0 {
 			helpStr := "\ndebugging tips (may or may not apply to this error):"
-			helpStr += fmt.Sprintf("\n* if your cluster was unable to provision instances, additional error information may be found in the activity history of your cluster's autoscaling groups (select each autoscaling group and click the  \"Activity\" or \"Activity History\" tab): https://console.aws.amazon.com/ec2/autoscaling/home?region=%s#AutoScalingGroups:", clusterConfig.Region)
+			helpStr += fmt.Sprintf(
+				"\n* if your cluster was unable to provision/remove/scale some nodegroups, additional error information may be found in the description of your cloudformation stack (https://console.aws.amazon.com/cloudformation/home?region=%s#/stacks)"+
+					" or in the activity history of your cluster's autoscaling groups (select each autoscaling group and click the  \"Activity\" or \"Activity History\" tab) (https://console.aws.amazon.com/ec2/autoscaling/home?region=%s#AutoScalingGroups)",
+				oldClusterConfig.Region,
+				oldClusterConfig.Region,
+			)
 			fmt.Println(helpStr)
-			exit.Error(ErrorClusterScale(out + helpStr))
+			exit.Error(ErrorClusterConfigure(out + helpStr))
 		}
 	},
 }
@@ -389,23 +388,41 @@ var _clusterInfoCmd = &cobra.Command{
 			exit.Error(err)
 		}
 
-		accessConfig, err := getClusterAccessConfigWithCache()
+		accessConfig, err := getClusterAccessConfigWithCache(true)
 		if err != nil {
 			exit.Error(err)
 		}
 
+		if _flagClusterInfoPrintConfig && _flagOutput == flags.PrettyOutputType {
+			_flagOutput = flags.YAMLOutputType
+		}
+
 		awsClient, err := newAWSClient(accessConfig.Region, _flagOutput == flags.PrettyOutputType)
 		if err != nil {
 			exit.Error(err)
 		}
 
+		if _flagClusterInfoPrintConfig && _flagClusterInfoDebug {
+			exit.Error(ErrorMutuallyExclusiveFlags("--print-config", "--debug"))
+		}
+		if _flagClusterInfoDebug && _flagOutput != flags.PrettyOutputType {
+			exit.Error(ErrorMutuallyExclusiveFlags("--debug", "--output"))
+		}
+
+		stacks, err := clusterstate.GetClusterStacks(awsClient, accessConfig)
+		if err != nil {
+			exit.Error(err)
+		}
+
+		state := clusterstate.GetClusterState(stacks)
+		if err := clusterstate.AssertClusterState(stacks, state, clusterstate.StateClusterExists); err != nil {
+			exit.Error(err)
+		}
+
 		if _flagClusterInfoDebug {
-			if _flagOutput != flags.PrettyOutputType {
-				exit.Error(ErrorJSONOutputNotSupportedWithFlag("--debug"))
-			}
 			cmdDebug(awsClient, accessConfig)
 		} else {
-			cmdInfo(awsClient, accessConfig, _flagOutput, _flagClusterDisallowPrompt)
+			cmdInfo(awsClient, accessConfig, stacks, _flagClusterInfoPrintConfig, _flagOutput, _flagClusterDisallowPrompt)
 		}
 	},
 }
@@ -421,7 +438,7 @@ var _clusterDownCmd = &cobra.Command{
 			exit.Error(err)
 		}
 
-		accessConfig, err := getClusterAccessConfigWithCache()
+		accessConfig, err := getClusterAccessConfigWithCache(true)
 		if err != nil {
 			exit.Error(err)
 		}
@@ -440,17 +457,17 @@ var _clusterDownCmd = &cobra.Command{
 
 		warnIfNotAdmin(awsClient)
 
-		errorsList := []error{}
-
 		if _flagClusterDisallowPrompt {
 			fmt.Printf("your cluster named \"%s\" in %s will be spun down and all apis will be deleted\n\n", accessConfig.ClusterName, accessConfig.Region)
 		} else {
 			prompt.YesOrExit(fmt.Sprintf("your cluster named \"%s\" in %s will be spun down and all apis will be deleted, are you sure you want to continue?", accessConfig.ClusterName, accessConfig.Region), "", "")
 		}
 
-		fmt.Print("￮ retrieving cluster ... ")
 		var clusterExists bool
-		clusterState, err := clusterstate.GetClusterState(awsClient, accessConfig)
+		errorsList := []error{}
+
+		fmt.Print("￮ retrieving cluster ... ")
+		stacks, err := clusterstate.GetClusterStacks(awsClient, accessConfig)
 		if err != nil {
 			errorsList = append(errorsList, err)
 			fmt.Print("failed ✗")
@@ -458,10 +475,8 @@ var _clusterDownCmd = &cobra.Command{
 			errors.PrintError(err)
 			fmt.Println()
 		} else {
-			switch clusterState.Status {
-			case clusterstate.StatusNotFound:
-				fmt.Println("cluster doesn't exist ✓")
-			case clusterstate.StatusDeleteComplete:
+			state := clusterstate.GetClusterState(stacks)
+			if err := clusterstate.AssertClusterState(stacks, state, clusterstate.StateClusterDoesntExist); err != nil {
 				awsClient.DeleteQueuesWithPrefix(clusterconfig.SQSNamePrefix(accessConfig.ClusterName))
 				awsClient.DeletePolicy(clusterconfig.DefaultPolicyARN(accountID, accessConfig.ClusterName, accessConfig.Region))
 				if !_flagClusterDownKeepAWSResources {
@@ -472,10 +487,10 @@ var _clusterDownCmd = &cobra.Command{
 						}
 					}
 				}
-				fmt.Println("already deleted ✓")
-			default:
 				fmt.Println("✓")
 				clusterExists = true
+			} else {
+				fmt.Println("already deleted ✓")
 			}
 		}
 
@@ -671,7 +686,7 @@ var _clusterExportCmd = &cobra.Command{
 	Run: func(cmd *cobra.Command, args []string) {
 		telemetry.Event("cli.cluster.export")
 
-		accessConfig, err := getClusterAccessConfigWithCache()
+		accessConfig, err := getClusterAccessConfigWithCache(true)
 		if err != nil {
 			exit.Error(err)
 		}
@@ -683,13 +698,13 @@ var _clusterExportCmd = &cobra.Command{
 		}
 		warnIfNotAdmin(awsClient)
 
-		clusterState, err := clusterstate.GetClusterState(awsClient, accessConfig)
+		stacks, err := clusterstate.GetClusterStacks(awsClient, accessConfig)
 		if err != nil {
 			exit.Error(err)
 		}
 
-		err = clusterstate.AssertClusterStatus(accessConfig.ClusterName, accessConfig.Region, clusterState.Status, clusterstate.StatusCreateComplete, clusterstate.StatusUpdateComplete, clusterstate.StatusUpdateRollbackComplete)
-		if err != nil {
+		state := clusterstate.GetClusterState(stacks)
+		if err := clusterstate.AssertClusterState(stacks, state, clusterstate.StateClusterExists); err != nil {
 			exit.Error(err)
 		}
 
@@ -739,14 +754,8 @@ var _clusterExportCmd = &cobra.Command{
 	},
 }
 
-func cmdInfo(awsClient *aws.Client, accessConfig *clusterconfig.AccessConfig, outputType flags.OutputType, disallowPrompt bool) {
-	if outputType == flags.PrettyOutputType {
-		if err := printInfoClusterState(awsClient, accessConfig); err != nil {
-			exit.Error(err)
-		}
-	}
-
-	clusterConfig := refreshCachedClusterConfig(*awsClient, accessConfig, outputType == flags.PrettyOutputType)
+func cmdInfo(awsClient *aws.Client, accessConfig *clusterconfig.AccessConfig, stacks clusterstate.ClusterStacks, printConfig bool, outputType flags.OutputType, disallowPrompt bool) {
+	clusterConfig := refreshCachedClusterConfig(awsClient, accessConfig, outputType == flags.PrettyOutputType)
 
 	operatorLoadBalancer, err := getLoadBalancer(accessConfig.ClusterName, OperatorLoadBalancer, awsClient)
 	if err != nil {
@@ -760,25 +769,36 @@ func cmdInfo(awsClient *aws.Client, accessConfig *clusterconfig.AccessConfig, ou
 	operatorEndpoint := s.EnsurePrefix(*operatorLoadBalancer.DNSName, "https://")
 	apiEndpoint := *apiLoadBalancer.DNSName
 
-	if outputType == flags.JSONOutputType {
+	if outputType == flags.JSONOutputType || outputType == flags.YAMLOutputType {
 		infoResponse, err := getInfoOperatorResponse(operatorEndpoint)
 		if err != nil {
 			exit.Error(err)
 		}
 		infoResponse.ClusterConfig.Config = clusterConfig
 
-		jsonBytes, err := libjson.Marshal(map[string]interface{}{
-			"cluster_config":    infoResponse.ClusterConfig.Config,
-			"cluster_metadata":  infoResponse.ClusterConfig.OperatorMetadata,
-			"node_infos":        infoResponse.NodeInfos,
-			"endpoint_operator": operatorEndpoint,
-			"endpoint_api":      apiEndpoint,
-		})
+		var infoInterface interface{}
+		if printConfig {
+			infoInterface = infoResponse.ClusterConfig.Config
+		} else {
+			infoInterface = map[string]interface{}{
+				"cluster_config":    infoResponse.ClusterConfig.Config,
+				"cluster_metadata":  infoResponse.ClusterConfig.OperatorMetadata,
+				"node_infos":        infoResponse.NodeInfos,
+				"endpoint_operator": operatorEndpoint,
+				"endpoint_api":      apiEndpoint,
+			}
+		}
+
+		var outputBytes []byte
+		if outputType == flags.JSONOutputType {
+			outputBytes, err = libjson.Marshal(infoInterface)
+		} else {
+			outputBytes, err = yaml.Marshal(infoInterface)
+		}
 		if err != nil {
 			exit.Error(err)
 		}
-
-		fmt.Println(string(jsonBytes))
+		fmt.Println(string(outputBytes))
 	}
 	if outputType == flags.PrettyOutputType {
 		fmt.Println(console.Bold("endpoints:"))
@@ -786,7 +806,7 @@ func cmdInfo(awsClient *aws.Client, accessConfig *clusterconfig.AccessConfig, ou
 		fmt.Println("api load balancer:", apiEndpoint)
 		fmt.Println()
 
-		if err := printInfoOperatorResponse(clusterConfig, operatorEndpoint); err != nil {
+		if err := printInfoOperatorResponse(clusterConfig, stacks, operatorEndpoint); err != nil {
 			exit.Error(err)
 		}
 	}
@@ -798,29 +818,11 @@ func cmdInfo(awsClient *aws.Client, accessConfig *clusterconfig.AccessConfig, ou
 	}
 }
 
-func printInfoClusterState(awsClient *aws.Client, accessConfig *clusterconfig.AccessConfig) error {
-	clusterState, err := clusterstate.GetClusterState(awsClient, accessConfig)
-	if err != nil {
-		return err
-	}
-
-	fmt.Println(clusterState.TableString())
-	if clusterState.Status == clusterstate.StatusCreateFailed || clusterState.Status == clusterstate.StatusDeleteFailed {
-		fmt.Println(fmt.Sprintf("more information can be found in your AWS console: %s", clusterstate.CloudFormationURL(accessConfig.ClusterName, accessConfig.Region)))
-		fmt.Println()
-	}
-
-	err = clusterstate.AssertClusterStatus(accessConfig.ClusterName, accessConfig.Region, clusterState.Status, clusterstate.StatusCreateComplete, clusterstate.StatusUpdateComplete, clusterstate.StatusUpdateRollbackComplete)
-	if err != nil {
-		return err
-	}
-
-	return nil
-}
-
-func printInfoOperatorResponse(clusterConfig clusterconfig.Config, operatorEndpoint string) error {
+func printInfoOperatorResponse(clusterConfig clusterconfig.Config, stacks clusterstate.ClusterStacks, operatorEndpoint string) error {
 	fmt.Print("fetching cluster status ...\n\n")
 
+	fmt.Println(stacks.TableString())
+
 	yamlBytes, err := yaml.Marshal(clusterConfig)
 	if err != nil {
 		return err
@@ -1063,7 +1065,7 @@ func cmdDebug(awsClient *aws.Client, accessConfig *clusterconfig.AccessConfig) {
 	return
 }
 
-func refreshCachedClusterConfig(awsClient aws.Client, accessConfig *clusterconfig.AccessConfig, printToStdout bool) clusterconfig.Config {
+func refreshCachedClusterConfig(awsClient *aws.Client, accessConfig *clusterconfig.AccessConfig, printToStdout bool) clusterconfig.Config {
 	// add empty file if cached cluster doesn't exist so that the file output by manager container maintains current user permissions
 	cachedClusterConfigPath := cachedClusterConfigPath(accessConfig.ClusterName, accessConfig.Region)
 	containerConfigPath := fmt.Sprintf("/out/%s", filepath.Base(cachedClusterConfigPath))
@@ -1078,7 +1080,7 @@ func refreshCachedClusterConfig(awsClient aws.Client, accessConfig *clusterconfi
 	if printToStdout {
 		fmt.Print("syncing cluster configuration ...\n\n")
 	}
-	out, exitCode, err := runManagerAccessCommand("/root/refresh.sh "+containerConfigPath, *accessConfig, &awsClient, nil, copyFromPaths)
+	out, exitCode, err := runManagerAccessCommand("/root/refresh.sh "+containerConfigPath, *accessConfig, awsClient, nil, copyFromPaths)
 	if err != nil {
 		exit.Error(err)
 	}
@@ -1094,69 +1096,6 @@ func refreshCachedClusterConfig(awsClient aws.Client, accessConfig *clusterconfi
 	return *refreshedClusterConfig
 }
 
-func updateNodeGroupScale(clusterConfig clusterconfig.Config, targetNg string, desiredMinReplicas, desiredMaxReplicas *int64, disallowPrompt bool) (clusterconfig.Config, int, error) {
-	clusterName := clusterConfig.ClusterName
-	region := clusterConfig.Region
-
-	availableNodeGroups := []string{}
-	for idx, ng := range clusterConfig.NodeGroups {
-		if ng == nil {
-			continue
-		}
-		availableNodeGroups = append(availableNodeGroups, ng.Name)
-		if ng.Name == targetNg {
-			var minReplicas, maxReplicas int64
-			if desiredMinReplicas == nil {
-				minReplicas = ng.MinInstances
-			} else {
-				minReplicas = *desiredMinReplicas
-			}
-			if desiredMaxReplicas == nil {
-				maxReplicas = ng.MaxInstances
-			} else {
-				maxReplicas = *desiredMaxReplicas
-			}
-
-			if minReplicas < 0 {
-				return clusterconfig.Config{}, 0, ErrorMinInstancesLowerThan(0)
-			}
-			if maxReplicas < 0 {
-				return clusterconfig.Config{}, 0, ErrorMaxInstancesLowerThan(0)
-			}
-			if minReplicas > maxReplicas {
-				return clusterconfig.Config{}, 0, ErrorMinInstancesGreaterThanMaxInstances(minReplicas, maxReplicas)
-			}
-
-			if ng.MinInstances == minReplicas && ng.MaxInstances == maxReplicas {
-				fmt.Printf("the %s nodegroup in the %s cluster in %s already has min instances set to %d and max instances set to %d\n", ng.Name, clusterName, region, minReplicas, maxReplicas)
-				exit.Ok()
-			}
-
-			if !disallowPrompt {
-				promptMessage := ""
-				if ng.MinInstances != minReplicas && ng.MaxInstances != maxReplicas {
-					promptMessage = fmt.Sprintf("your nodegroup named %s in your %s cluster in %s will update its %s from %d to %d and update its %s from %d to %d", ng.Name, clusterName, region, clusterconfig.MinInstancesKey, ng.MinInstances, minReplicas, clusterconfig.MaxInstancesKey, ng.MaxInstances, maxReplicas)
-				}
-				if ng.MinInstances == minReplicas && ng.MaxInstances != maxReplicas {
-					promptMessage = fmt.Sprintf("your nodegroup named %s in your %s cluster in %s will update its %s from %d to %d", ng.Name, clusterName, region, clusterconfig.MaxInstancesKey, ng.MaxInstances, maxReplicas)
-				}
-				if ng.MinInstances != minReplicas && ng.MaxInstances == maxReplicas {
-					promptMessage = fmt.Sprintf("your nodegroup named %s in your %s cluster in %s will update its %s from %d to %d", ng.Name, clusterName, region, clusterconfig.MinInstancesKey, ng.MinInstances, minReplicas)
-				}
-				if !prompt.YesOrNo(promptMessage, "", "") {
-					exit.Ok()
-				}
-			}
-
-			clusterConfig.NodeGroups[idx].MinInstances = minReplicas
-			clusterConfig.NodeGroups[idx].MaxInstances = maxReplicas
-			return clusterConfig, idx, nil
-		}
-	}
-
-	return clusterconfig.Config{}, 0, ErrorNodeGroupNotFound(targetNg, clusterName, region, availableNodeGroups)
-}
-
 func createS3BucketIfNotFound(awsClient *aws.Client, bucket string, tags map[string]string) error {
 	bucketFound, err := awsClient.DoesBucketExist(bucket)
 	if err != nil {
diff --git a/cli/cmd/delete.go b/cli/cmd/delete.go
index 84e46b3adb..062aaf5c38 100644
--- a/cli/cmd/delete.go
+++ b/cli/cmd/delete.go
@@ -42,7 +42,7 @@ func deleteInit() {
 
 	_deleteCmd.Flags().BoolVarP(&_flagDeleteForce, "force", "f", false, "delete the api without confirmation")
 	_deleteCmd.Flags().BoolVarP(&_flagDeleteKeepCache, "keep-cache", "c", false, "keep cached data for the api")
-	_deleteCmd.Flags().VarP(&_flagOutput, "output", "o", fmt.Sprintf("output format: one of %s", strings.Join(flags.UserOutputTypeStrings(), "|")))
+	_deleteCmd.Flags().VarP(&_flagOutput, "output", "o", fmt.Sprintf("output format: one of %s", strings.Join(flags.OutputTypeStringsExcluding(flags.YAMLOutputType), "|")))
 }
 
 var _deleteCmd = &cobra.Command{
diff --git a/cli/cmd/deploy.go b/cli/cmd/deploy.go
index e263fe0cda..254f932e71 100644
--- a/cli/cmd/deploy.go
+++ b/cli/cmd/deploy.go
@@ -49,7 +49,7 @@ func deployInit() {
 	_deployCmd.Flags().StringVarP(&_flagDeployEnv, "env", "e", "", "environment to use")
 	_deployCmd.Flags().BoolVarP(&_flagDeployForce, "force", "f", false, "override the in-progress api update")
 	_deployCmd.Flags().BoolVarP(&_flagDeployDisallowPrompt, "yes", "y", false, "skip prompts")
-	_deployCmd.Flags().VarP(&_flagOutput, "output", "o", fmt.Sprintf("output format: one of %s", strings.Join(flags.UserOutputTypeStrings(), "|")))
+	_deployCmd.Flags().VarP(&_flagOutput, "output", "o", fmt.Sprintf("output format: one of %s", strings.Join(flags.OutputTypeStringsExcluding(flags.YAMLOutputType), "|")))
 }
 
 var _deployCmd = &cobra.Command{
diff --git a/cli/cmd/env.go b/cli/cmd/env.go
index 7a736bf467..6f1611b0e3 100644
--- a/cli/cmd/env.go
+++ b/cli/cmd/env.go
@@ -39,7 +39,7 @@ func envInit() {
 	_envCmd.AddCommand(_envConfigureCmd)
 
 	_envListCmd.Flags().SortFlags = false
-	_envListCmd.Flags().VarP(&_flagOutput, "output", "o", fmt.Sprintf("output format: one of %s", strings.Join(flags.UserOutputTypeStrings(), "|")))
+	_envListCmd.Flags().VarP(&_flagOutput, "output", "o", fmt.Sprintf("output format: one of %s", strings.Join(flags.OutputTypeStringsExcluding(flags.YAMLOutputType), "|")))
 	_envCmd.AddCommand(_envListCmd)
 
 	_envDefaultCmd.Flags().SortFlags = false
diff --git a/cli/cmd/errors.go b/cli/cmd/errors.go
index bbc196d61b..c8bdeddd49 100644
--- a/cli/cmd/errors.go
+++ b/cli/cmd/errors.go
@@ -52,7 +52,7 @@ const (
 	ErrMissingAWSCredentials               = "cli.missing_aws_credentials"
 	ErrCredentialsInClusterConfig          = "cli.credentials_in_cluster_config"
 	ErrClusterUp                           = "cli.cluster_up"
-	ErrClusterScale                        = "cli.cluster_scale"
+	ErrClusterConfigure                    = "cli.cluster_configure"
 	ErrClusterDebug                        = "cli.cluster_debug"
 	ErrClusterRefresh                      = "cli.cluster_refresh"
 	ErrClusterDown                         = "cli.cluster_down"
@@ -61,7 +61,7 @@ const (
 	ErrMaxInstancesLowerThan               = "cli.max_instances_lower_than"
 	ErrMinInstancesGreaterThanMaxInstances = "cli.min_instances_greater_than_max_instances"
 	ErrNodeGroupNotFound                   = "cli.nodegroup_not_found"
-	ErrJSONOutputNotSupportedWithFlag      = "cli.json_output_not_supported_with_flag"
+	ErrMutuallyExclusiveFlags              = "cli.mutually_exclusive_flags"
 	ErrClusterAccessConfigRequired         = "cli.cluster_access_config_or_prompts_required"
 	ErrShellCompletionNotSupported         = "cli.shell_completion_not_supported"
 	ErrNoTerminalWidth                     = "cli.no_terminal_width"
@@ -162,9 +162,9 @@ func ErrorClusterUp(out string) error {
 	})
 }
 
-func ErrorClusterScale(out string) error {
+func ErrorClusterConfigure(out string) error {
 	return errors.WithStack(&errors.Error{
-		Kind:    ErrClusterScale,
+		Kind:    ErrClusterConfigure,
 		Message: out,
 		NoPrint: true,
 	})
@@ -229,10 +229,10 @@ func ErrorNodeGroupNotFound(scalingNodeGroupName, clusterName, clusterRegion str
 	})
 }
 
-func ErrorJSONOutputNotSupportedWithFlag(flag string) error {
+func ErrorMutuallyExclusiveFlags(flagA, flagB string) error {
 	return errors.WithStack(&errors.Error{
-		Kind:    ErrJSONOutputNotSupportedWithFlag,
-		Message: fmt.Sprintf("flag %s cannot be used when output type is set to json", flag),
+		Kind:    ErrMutuallyExclusiveFlags,
+		Message: fmt.Sprintf("flags %s and %s cannot be used at the same time", flagA, flagB),
 	})
 }
 
diff --git a/cli/cmd/get.go b/cli/cmd/get.go
index 99cfc0bb7e..c48512db9c 100644
--- a/cli/cmd/get.go
+++ b/cli/cmd/get.go
@@ -62,7 +62,7 @@ func getInit() {
 	_getCmd.Flags().SortFlags = false
 	_getCmd.Flags().StringVarP(&_flagGetEnv, "env", "e", "", "environment to use")
 	_getCmd.Flags().BoolVarP(&_flagWatch, "watch", "w", false, "re-run the command every 2 seconds")
-	_getCmd.Flags().VarP(&_flagOutput, "output", "o", fmt.Sprintf("output format: one of %s", strings.Join(flags.UserOutputTypeStrings(), "|")))
+	_getCmd.Flags().VarP(&_flagOutput, "output", "o", fmt.Sprintf("output format: one of %s", strings.Join(flags.OutputTypeStringsExcluding(flags.YAMLOutputType), "|")))
 	addVerboseFlag(_getCmd)
 }
 
diff --git a/cli/cmd/lib_cluster_config.go b/cli/cmd/lib_cluster_config.go
index 29304a4f38..57ef843dc9 100644
--- a/cli/cmd/lib_cluster_config.go
+++ b/cli/cmd/lib_cluster_config.go
@@ -34,6 +34,7 @@ import (
 	s "github.com/cortexlabs/cortex/pkg/lib/strings"
 	"github.com/cortexlabs/cortex/pkg/lib/table"
 	"github.com/cortexlabs/cortex/pkg/types/clusterconfig"
+	"github.com/cortexlabs/cortex/pkg/types/clusterstate"
 )
 
 var _cachedClusterConfigRegex = regexp.MustCompile(`^cluster_\S+\.yaml$`)
@@ -87,7 +88,7 @@ func getNewClusterAccessConfig(clusterConfigFile string) (*clusterconfig.AccessC
 	return accessConfig, nil
 }
 
-func getClusterAccessConfigWithCache() (*clusterconfig.AccessConfig, error) {
+func getClusterAccessConfigWithCache(hasClusterFlags bool) (*clusterconfig.AccessConfig, error) {
 	accessConfig := &clusterconfig.AccessConfig{
 		ImageManager: consts.DefaultRegistry() + "/manager:" + consts.CortexVersion,
 	}
@@ -115,13 +116,12 @@ func getClusterAccessConfigWithCache() (*clusterconfig.AccessConfig, error) {
 	}
 
 	if accessConfig.ClusterName == "" || accessConfig.Region == "" {
-		cliFlagsOnly := _flagClusterScaleNodeGroup != ""
-		return nil, ErrorClusterAccessConfigRequired(cliFlagsOnly)
+		return nil, ErrorClusterAccessConfigRequired(hasClusterFlags)
 	}
 	return accessConfig, nil
 }
 
-func getInstallClusterConfig(awsClient *aws.Client, clusterConfigFile string, disallowPrompt bool) (*clusterconfig.Config, error) {
+func getInstallClusterConfig(awsClient *aws.Client, clusterConfigFile string) (*clusterconfig.Config, error) {
 	clusterConfig := &clusterconfig.Config{}
 
 	err := readUserClusterConfigFile(clusterConfig, clusterConfigFile)
@@ -129,22 +129,35 @@ func getInstallClusterConfig(awsClient *aws.Client, clusterConfigFile string, di
 		return nil, err
 	}
 
-	promptIfNotAdmin(awsClient, disallowPrompt)
+	clusterConfig.Telemetry = isTelemetryEnabled()
 
-	clusterConfig.Telemetry, err = readTelemetryConfig()
+	err = clusterConfig.ValidateOnInstall(awsClient)
 	if err != nil {
-		return nil, err
+		err = errors.Append(err, fmt.Sprintf("\n\ncluster configuration schema can be found at https://docs.cortex.dev/v/%s/", consts.CortexVersionMinor))
+		return nil, errors.Wrap(err, clusterConfigFile)
+	}
+
+	return clusterConfig, nil
+}
+
+func getConfigureClusterConfig(awsClient *aws.Client, stacks clusterstate.ClusterStacks, cachedClusterConfig clusterconfig.Config, newClusterConfigFile string) (*clusterconfig.Config, clusterconfig.ConfigureChanges, error) {
+	newUserClusterConfig := &clusterconfig.Config{}
+
+	err := readUserClusterConfigFile(newUserClusterConfig, newClusterConfigFile)
+	if err != nil {
+		return nil, clusterconfig.ConfigureChanges{}, err
 	}
 
-	err = clusterConfig.Validate(awsClient)
+	newUserClusterConfig.Telemetry = isTelemetryEnabled()
+	cachedClusterConfig.Telemetry = newUserClusterConfig.Telemetry
+
+	configureChanges, err := newUserClusterConfig.ValidateOnConfigure(awsClient, cachedClusterConfig, stacks.NodeGroupsStacks)
 	if err != nil {
 		err = errors.Append(err, fmt.Sprintf("\n\ncluster configuration schema can be found at https://docs.cortex.dev/v/%s/", consts.CortexVersionMinor))
-		return nil, errors.Wrap(err, clusterConfigFile)
+		return nil, clusterconfig.ConfigureChanges{}, errors.Wrap(err, newClusterConfigFile)
 	}
 
-	confirmInstallClusterConfig(clusterConfig, awsClient, disallowPrompt)
-
-	return clusterConfig, nil
+	return newUserClusterConfig, configureChanges, nil
 }
 
 func confirmInstallClusterConfig(clusterConfig *clusterconfig.Config, awsClient *aws.Client, disallowPrompt bool) {
@@ -264,3 +277,43 @@ func confirmInstallClusterConfig(clusterConfig *clusterconfig.Config, awsClient
 		prompt.YesOrExit("would you like to continue?", "", exitMessage)
 	}
 }
+
+func confirmConfigureClusterConfig(configureChanges clusterconfig.ConfigureChanges, oldCc, newCc clusterconfig.Config, disallowPrompt bool) {
+	fmt.Printf("your %s cluster in region %s will be updated as follows:\n\n", newCc.ClusterName, newCc.Region)
+	if len(configureChanges.NodeGroupsToScale) > 0 {
+		fmt.Printf("￮ %d %s will be scaled\n", len(configureChanges.NodeGroupsToScale), s.PluralS("nodegroup", len(configureChanges.NodeGroupsToScale)))
+		for _, ngName := range configureChanges.NodeGroupsToScale {
+			var output string
+			ngOld := oldCc.GetNodeGroupByName(ngName)
+			ngScaled := newCc.GetNodeGroupByName(ngName)
+			if ngOld.MinInstances != ngScaled.MinInstances && ngOld.MaxInstances != ngScaled.MaxInstances {
+				output = fmt.Sprintf("nodegroup %s will update its %s from %d to %d and update its %s from %d to %d", ngName, clusterconfig.MinInstancesKey, ngOld.MinInstances, ngScaled.MinInstances, clusterconfig.MaxInstancesKey, ngOld.MaxInstances, ngScaled.MaxInstances)
+			}
+			if ngOld.MinInstances == ngScaled.MinInstances && ngOld.MaxInstances != ngScaled.MaxInstances {
+				output = fmt.Sprintf("nodegroup %s will update its %s from %d to %d", ngName, clusterconfig.MaxInstancesKey, ngOld.MaxInstances, ngScaled.MaxInstances)
+			}
+			if ngOld.MinInstances != ngScaled.MinInstances && ngOld.MaxInstances == ngScaled.MaxInstances {
+				output = fmt.Sprintf("nodegroup %s will update its %s from %d to %d", ngName, clusterconfig.MinInstancesKey, ngOld.MinInstances, ngScaled.MinInstances)
+			}
+			fmt.Printf(" ￮ %s\n", output)
+		}
+	}
+	if len(configureChanges.NodeGroupsToAdd) > 0 {
+		fmt.Printf("￮ %d %s will be added: %s\n", len(configureChanges.NodeGroupsToAdd), s.PluralS("nodegroup", len(configureChanges.NodeGroupsToAdd)), s.StrsAnd(configureChanges.NodeGroupsToAdd))
+	}
+	if len(configureChanges.NodeGroupsToRemove) > 0 {
+		fmt.Printf("￮ %d %s will be removed: %s\n", len(configureChanges.NodeGroupsToRemove), s.PluralS("nodegroup", len(configureChanges.NodeGroupsToRemove)), s.StrsAnd(configureChanges.NodeGroupsToRemove))
+	}
+	// EKS node groups that don't appear in the old/new cluster config
+	// this is unlikely but can happen
+	ghostEKSNodeGroups := configureChanges.GetGhostEKSNodeGroups()
+	if len(ghostEKSNodeGroups) > 0 {
+		fmt.Printf("￮ %d %s will be removed: %s\n", len(ghostEKSNodeGroups), s.PluralS("EKS nodegroup", len(ghostEKSNodeGroups)), s.StrsAnd(ghostEKSNodeGroups))
+	}
+	fmt.Println()
+
+	if !disallowPrompt {
+		exitMessage := fmt.Sprintf("cluster configuration can be modified via the cluster config file; see https://docs.cortex.dev/v/%s/ for more information", consts.CortexVersionMinor)
+		prompt.YesOrExit(fmt.Sprintf("your cluster named \"%s\" in %s will be updated according to the configuration above, are you sure you want to continue?", newCc.ClusterName, newCc.Region), "", exitMessage)
+	}
+}
diff --git a/cli/cmd/refresh.go b/cli/cmd/refresh.go
index fb73a29bb4..7dfbe81357 100644
--- a/cli/cmd/refresh.go
+++ b/cli/cmd/refresh.go
@@ -38,7 +38,7 @@ func refreshInit() {
 	_refreshCmd.Flags().SortFlags = false
 	_refreshCmd.Flags().StringVarP(&_flagRefreshEnv, "env", "e", "", "environment to use")
 	_refreshCmd.Flags().BoolVarP(&_flagRefreshForce, "force", "f", false, "override the in-progress api update")
-	_refreshCmd.Flags().VarP(&_flagOutput, "output", "o", fmt.Sprintf("output format: one of %s", strings.Join(flags.UserOutputTypeStrings(), "|")))
+	_refreshCmd.Flags().VarP(&_flagOutput, "output", "o", fmt.Sprintf("output format: one of %s", strings.Join(flags.OutputTypeStringsExcluding(flags.YAMLOutputType), "|")))
 }
 
 var _refreshCmd = &cobra.Command{
diff --git a/cli/types/flags/output_type.go b/cli/types/flags/output_type.go
index 1427747b86..984fb20eb1 100644
--- a/cli/types/flags/output_type.go
+++ b/cli/types/flags/output_type.go
@@ -22,12 +22,14 @@ const (
 	UnknownOutputType OutputType = iota
 	PrettyOutputType
 	JSONOutputType
+	YAMLOutputType
 )
 
 var _outputTypes = []string{
 	"unknown",
 	"pretty",
 	"json",
+	"yaml",
 }
 
 func OutputTypeFromString(s string) OutputType {
@@ -39,12 +41,18 @@ func OutputTypeFromString(s string) OutputType {
 	return UnknownOutputType
 }
 
-func UserOutputTypeStrings() []string {
+func OutputTypeStrings() []string {
 	return _outputTypes[1:]
 }
 
-func OutputTypeStrings() []string {
-	return _outputTypes[1:]
+func OutputTypeStringsExcluding(outputType OutputType) []string {
+	var outputTypes []string
+	for _, _outputType := range _outputTypes[1:] {
+		if OutputTypeFromString(_outputType) != outputType {
+			outputTypes = append(outputTypes, _outputType)
+		}
+	}
+	return outputTypes
 }
 
 func (t OutputType) String() string {
diff --git a/dev/generate_cli_md.sh b/dev/generate_cli_md.sh
index 63f9f40d0d..0980c737f1 100755
--- a/dev/generate_cli_md.sh
+++ b/dev/generate_cli_md.sh
@@ -38,7 +38,7 @@ commands=(
   "delete"
   "cluster up"
   "cluster info"
-  "cluster scale"
+  "cluster configure"
   "cluster down"
   "cluster export"
   "env configure"
diff --git a/docs/clients/cli.md b/docs/clients/cli.md
index 1e060cda88..c8c264c51f 100644
--- a/docs/clients/cli.md
+++ b/docs/clients/cli.md
@@ -104,29 +104,25 @@ Flags:
   -c, --config string          path to a cluster configuration file
   -n, --name string            name of the cluster
   -r, --region string          aws region of the cluster
-  -o, --output string          output format: one of pretty|json (default "pretty")
+  -o, --output string          output format: one of pretty|json|yaml (default "pretty")
   -e, --configure-env string   name of environment to configure
   -d, --debug                  save the current cluster state to a file
+      --print-config           print the cluster config
   -y, --yes                    skip prompts
   -h, --help                   help for info
 ```
 
-## cluster scale
+## cluster configure
 
 ```text
-update the min/max instances for a nodegroup
+update the cluster's configuration
 
 Usage:
-  cortex cluster scale [flags]
+  cortex cluster configure CLUSTER_CONFIG_FILE [flags]
 
 Flags:
-  -n, --name string         name of the cluster
-  -r, --region string       aws region of the cluster
-      --node-group string   name of the node group to scale
-      --min-instances int   minimum number of instances
-      --max-instances int   maximum number of instances
-  -y, --yes                 skip prompts
-  -h, --help                help for scale
+  -y, --yes    skip prompts
+  -h, --help   help for configure
 ```
 
 ## cluster down
diff --git a/docs/clusters/instances/multi.md b/docs/clusters/instances/multi.md
index e34194ed88..6e62326fbc 100644
--- a/docs/clusters/instances/multi.md
+++ b/docs/clusters/instances/multi.md
@@ -4,11 +4,9 @@ Cortex can be configured to provision different instance types to improve worklo
 
 ## Best practices
 
-**Node groups with lower indices have higher priority.**
-
-1. Spot node groups should be listed before on-demand node groups.
-1. CPU node groups should be listed before GPU/Inferentia node groups.
-1. Node groups with small instance types should be listed before node groups with large instance types.
+1. Spot node groups should have a higher priority than on-demand node groups.
+1. CPU node groups should have higher priorities than GPU/Inferentia node groups.
+1. Node groups with small instance types should have higher priorities than node groups with large instance types.
 
 ## Examples
 
@@ -22,6 +20,7 @@ node_groups:
     instance_type: m5.large
     min_instances: 0
     max_instances: 5
+    priority: 100
     spot: true
     spot_config:
       instance_distribution: [m5a.large, m5d.large, m5n.large, m5ad.large, m5dn.large, m4.large, t3.large, t3a.large, t2.large]
@@ -41,6 +40,7 @@ node_groups:
     instance_type: m5.large
     min_instances: 0
     max_instances: 5
+    priority: 100
   - name: gpu
     instance_type: g4dn.xlarge
     min_instances: 0
@@ -61,6 +61,7 @@ node_groups:
     instance_type: m5.large
     min_instances: 0
     max_instances: 5
+    priority: 100
     spot: true
     spot_config:
       instance_distribution: [m5a.large, m5d.large, m5n.large, m5ad.large, m5dn.large, m4.large, t3.large, t3a.large, t2.large]
@@ -68,10 +69,12 @@ node_groups:
     instance_type: m5.large
     min_instances: 0
     max_instances: 5
+    priority: 50
   - name: gpu-spot
     instance_type: g4dn.xlarge
     min_instances: 0
     max_instances: 5
+    priority: 20
     spot: true
   - name: gpu-on-demand
     instance_type: g4dn.xlarge
@@ -89,16 +92,19 @@ node_groups:
     instance_type: t3.medium
     min_instances: 0
     max_instances: 5
+    priority: 100
     spot: true
   - name: cpu-2
     instance_type: m5.2xlarge
     min_instances: 0
     max_instances: 5
+    priority: 70
     spot: true
   - name: cpu-3
     instance_type: m5.8xlarge
     min_instances: 0
     max_instances: 5
+    priority: 30
     spot: true
   - name: cpu-4
     instance_type: m5.24xlarge
diff --git a/docs/clusters/management/create.md b/docs/clusters/management/create.md
index 5fe3a5d51e..f60145b5b5 100644
--- a/docs/clusters/management/create.md
+++ b/docs/clusters/management/create.md
@@ -29,12 +29,13 @@ region: us-east-1
 # list of availability zones for your region
 availability_zones:  # default: 3 random availability zones in your region, e.g. [us-east-1a, us-east-1b, us-east-1c]
 
-# list of cluster node groups; the smaller index, the higher the priority of the node group
+# list of cluster node groups;
 node_groups:
   - name: ng-cpu # name of the node group
     instance_type: m5.large # instance type
     min_instances: 1 # minimum number of instances
     max_instances: 5 # maximum number of instances
+    priority: 1 # priority of the node group; the higher the value, the higher the priority [1-100]
     instance_volume_size: 50 # disk storage size per instance (GB)
     instance_volume_type: gp3 # instance volume type [gp2 | gp3 | io1 | st1 | sc1]
     # instance_volume_iops: 3000 # instance volume iops (only applicable to io1/gp3)
diff --git a/manager/generate_eks.py b/manager/generate_eks.py
index a2caa74df3..c54c18a216 100644
--- a/manager/generate_eks.py
+++ b/manager/generate_eks.py
@@ -14,6 +14,7 @@
 
 import json
 import sys
+import click
 
 import yaml
 
@@ -170,26 +171,51 @@ def get_inf_resources(instance_type):
 
 
 def get_all_worker_nodegroups(ami_map: dict, cluster_config: dict) -> list:
+    """
+    Gets all node groups in EKS-dict format.
+    """
     worker_nodegroups = []
     for ng in cluster_config["node_groups"]:
-        worker_nodegroup = default_nodegroup(cluster_config)
-        worker_nodegroup["ami"] = get_ami(ami_map, ng["instance_type"])
+        worker_nodegroups.append(get_worker_nodegroup(ami_map, ng, cluster_config))
+    return worker_nodegroups
 
-        apply_worker_settings(worker_nodegroup, ng)
-        apply_clusterconfig(worker_nodegroup, ng)
 
-        if ng["spot"]:
-            apply_spot_settings(worker_nodegroup, ng)
+def get_worker_nodegroup(ami_map: dict, nodegroup_config: dict, cluster_config: dict) -> dict:
+    """
+    Converts Cortex-dict nodegroup config to EKS-dict format.
+    """
+    worker_nodegroup = default_nodegroup(cluster_config)
+    worker_nodegroup["ami"] = get_ami(ami_map, nodegroup_config["instance_type"])
 
-        if is_gpu(ng["instance_type"]):
-            apply_gpu_settings(worker_nodegroup)
+    apply_worker_settings(worker_nodegroup, nodegroup_config)
+    apply_clusterconfig(worker_nodegroup, nodegroup_config)
 
-        if is_inf(ng["instance_type"]):
-            apply_inf_settings(worker_nodegroup, ng)
+    if nodegroup_config["spot"]:
+        apply_spot_settings(worker_nodegroup, nodegroup_config)
 
-        worker_nodegroups.append(worker_nodegroup)
+    if is_gpu(nodegroup_config["instance_type"]):
+        apply_gpu_settings(worker_nodegroup)
 
-    return worker_nodegroups
+    if is_inf(nodegroup_config["instance_type"]):
+        apply_inf_settings(worker_nodegroup, nodegroup_config)
+
+    return worker_nodegroup
+
+
+def get_nodegroup_config_by_name(cluster_config: dict, ng_name: str) -> dict:
+    """
+    Gets a nodegroup in Cortex-dict format from Cortex cluster config.
+    """
+    for ng in cluster_config["node_groups"]:
+        if ng["name"] == ng_name:
+            return ng
+
+
+def get_empty_eks_nodegroup(name: str) -> dict:
+    """
+    Gets an empty nodegroup in EKS-dict format that only has the Cortex nodegroup name filled out.
+    """
+    return {"name": name}
 
 
 def get_ami(ami_map: dict, instance_type: str) -> str:
@@ -198,14 +224,55 @@ def get_ami(ami_map: dict, instance_type: str) -> str:
     return ami_map["cpu"]
 
 
-def generate_eks(cluster_config_path, ami_json_path):
-    with open(cluster_config_path, "r") as f:
-        cluster_config = yaml.safe_load(f)
-
+@click.command()
+@click.argument("cluster-config_file", type=click.File("r"))
+@click.argument("ami-json-file", type=click.File("r"))
+@click.option(
+    "--add-cortex-node-groups",
+    type=str,
+    help="specific cortex nodegroups to add to the generated eks file; use this for existing clusters",
+)
+@click.option(
+    "--remove-eks-node-groups",
+    type=str,
+    help="specific eks nodegroup stacks to add to the generated eks file; use this for existing clusters",
+)
+def generate_eks(
+    cluster_config_file, ami_json_file, add_cortex_node_groups: str, remove_eks_node_groups: str
+):
+    cluster_config = yaml.safe_load(cluster_config_file)
     region = cluster_config["region"]
+    name = cluster_config["cluster_name"]
+    ami_map = json.load(ami_json_file)[K8S_VERSION][region]
+
+    eks = {
+        "apiVersion": "eksctl.io/v1alpha5",
+        "kind": "ClusterConfig",
+        "metadata": {
+            "name": name,
+            "region": region,
+            "version": K8S_VERSION,
+        },
+    }
 
-    with open(ami_json_path, "r") as f:
-        ami_map = json.load(f)[K8S_VERSION][region]
+    if add_cortex_node_groups:
+        node_group_names = add_cortex_node_groups.split(",")
+        eks["nodeGroups"] = []
+        for node_group_name in node_group_names:
+            nodegroup_config = get_nodegroup_config_by_name(cluster_config, node_group_name)
+            eks["nodeGroups"].append(
+                get_worker_nodegroup(ami_map, nodegroup_config, cluster_config)
+            )
+        click.echo(yaml.dump(eks, Dumper=IgnoreAliases, default_flow_style=False, default_style=""))
+        return
+
+    if remove_eks_node_groups:
+        stacks_names = remove_eks_node_groups.split(",")
+        eks["nodeGroups"] = []
+        for stack_name in stacks_names:
+            eks["nodeGroups"].append(get_empty_eks_nodegroup(stack_name))
+        click.echo(yaml.dump(eks, Dumper=IgnoreAliases, default_flow_style=False, default_style=""))
+        return
 
     operator_nodegroup = default_nodegroup(cluster_config)
     operator_settings = {
@@ -234,8 +301,8 @@ def generate_eks(cluster_config_path, ami_json_path):
         "apiVersion": "eksctl.io/v1alpha5",
         "kind": "ClusterConfig",
         "metadata": {
-            "name": cluster_config["cluster_name"],
-            "region": cluster_config["region"],
+            "name": name,
+            "region": region,
             "version": K8S_VERSION,
             "tags": cluster_config["tags"],
         },
@@ -270,7 +337,7 @@ def generate_eks(cluster_config_path, ami_json_path):
     if cluster_config.get("vpc_cidr", "") != "":
         eks["vpc"]["cidr"] = cluster_config["vpc_cidr"]
 
-    print(yaml.dump(eks, Dumper=IgnoreAliases, default_flow_style=False, default_style=""))
+    click.echo(yaml.dump(eks, Dumper=IgnoreAliases, default_flow_style=False, default_style=""))
 
 
 class IgnoreAliases(yaml.Dumper):
@@ -284,4 +351,4 @@ def ignore_aliases(self, data):
 
 
 if __name__ == "__main__":
-    generate_eks(cluster_config_path=sys.argv[1], ami_json_path=sys.argv[2])
+    generate_eks()
diff --git a/manager/install.sh b/manager/install.sh
index c9403b7c2b..2294fe3b01 100755
--- a/manager/install.sh
+++ b/manager/install.sh
@@ -18,13 +18,14 @@ set -eo pipefail
 
 export CORTEX_VERSION=master
 export CORTEX_VERSION_MINOR=master
-EKSCTL_TIMEOUT=45m
+EKSCTL_CLUSTER_TIMEOUT=45m
+EKSCTL_NODEGROUP_TIMEOUT=30m
 mkdir /workspace
 
 arg1="$1"
 
 function main() {
-  if [ "$arg1" = "--update" ]; then
+  if [ "$arg1" = "--configure" ]; then
     cluster_configure
   else
     cluster_up
@@ -85,7 +86,9 @@ function cluster_up() {
 function cluster_configure() {
   check_eks
 
-  resize_nodegroup
+  resize_nodegroups
+  add_nodegroups
+  remove_nodegroups
 
   echo -n "￮ updating cluster configuration "
   setup_configmap
@@ -143,7 +146,7 @@ function create_eks() {
 
   echo -e "￮ spinning up the cluster (this will take about 45 minutes) ...\n"
   python generate_eks.py $CORTEX_CLUSTER_CONFIG_FILE manifests/ami.json > /workspace/eks.yaml
-  eksctl create cluster --timeout=$EKSCTL_TIMEOUT --install-neuron-plugin=false --install-nvidia-plugin=false -f /workspace/eks.yaml
+  eksctl create cluster --timeout=$EKSCTL_CLUSTER_TIMEOUT --install-neuron-plugin=false --install-nvidia-plugin=false -f /workspace/eks.yaml
   echo
 
   write_kubeconfig
@@ -256,56 +259,97 @@ function start_controller_manager() {
   echo "✓"
 }
 
-function resize_nodegroup() {
-  eksctl get nodegroup --cluster=$CORTEX_CLUSTER_NAME --region=$CORTEX_REGION -o json > nodegroups.json
-  ng_len=$(cat nodegroups.json | jq -r length)
-  config_ng="$CORTEX_SCALING_NODEGROUP"
+function resize_nodegroups() {
+  if [ -z "$CORTEX_NODEGROUP_NAMES_TO_SCALE" ]; then
+    return
+  fi
 
-  has_ng="false"
-  for eks_idx in $(seq 0 $(($ng_len-1))); do
-    stack_ng=$(cat nodegroups.json | jq -r .[$eks_idx].Name)
-    if [ "$stack_ng" = "cx-operator" ]; then
-      continue
+  eksctl get nodegroup --cluster=$CORTEX_CLUSTER_NAME --region=$CORTEX_REGION -v 0 -o json > nodegroups.json
+  eks_ng_len=$(cat nodegroups.json | jq -r length)
+  cfg_ng_len=$(cat $CORTEX_CLUSTER_CONFIG_FILE | yq -r .node_groups | yq -r length)
+
+  for cfg_ng_name in $CORTEX_NODEGROUP_NAMES_TO_SCALE; do
+    has_ng="false"
+    for eks_idx in $(seq 0 $(($eks_ng_len-1))); do
+      stack_ng=$(cat nodegroups.json | jq -r .[$eks_idx].Name)
+      if [ "$stack_ng" = "cx-operator" ]; then
+        continue
+      fi
+      if [[ "$stack_ng" == *"$cfg_ng_name" ]]; then
+        has_ng="true"
+        break
+      fi
+    done
+
+    if [ "$has_ng" == "false" ]; then
+      echo -e "error: \"$cfg_ng_name\" nodegroup (\"cx-*-$cfg_ng_name\" on aws) couldn't be scaled because stack couldn't be found\n"
+      exit 1
     fi
-    if [[ "$stack_ng" == *"$config_ng" ]]; then
-      has_ng="true"
-      break
+
+    for cfg_idx in $(seq 0 $(($cfg_ng_len-1))); do
+      cfg_ng=$(cat $CORTEX_CLUSTER_CONFIG_FILE | yq -r .node_groups[$cfg_idx].name)
+      if [ "$cfg_ng" = "$cfg_ng_name" ]; then
+        break
+      fi
+    done
+
+    desired=$(cat nodegroups.json | jq -r .[$eks_idx].DesiredCapacity)
+    existing_min=$(cat nodegroups.json | jq -r .[$eks_idx].MinSize)
+    existing_max=$(cat nodegroups.json | jq -r .[$eks_idx].MaxSize)
+    updating_min=$(cat $CORTEX_CLUSTER_CONFIG_FILE | yq -r .node_groups[$cfg_idx].min_instances)
+    updating_max=$(cat $CORTEX_CLUSTER_CONFIG_FILE | yq -r .node_groups[$cfg_idx].max_instances)
+
+    if [ "$desired" -lt $updating_min ]; then
+      desired=$updating_min
+    fi
+    if [ "$desired" -gt $updating_max ]; then
+      desired=$updating_max
+    fi
+
+    if [ "$existing_min" != "$updating_min" ] && [ "$existing_max" != "$updating_max" ]; then
+      echo "￮ nodegroup $cfg_ng_name: updating min instances to $updating_min and max instances to $updating_max"
+      eksctl scale nodegroup --cluster=$CORTEX_CLUSTER_NAME --region=$CORTEX_REGION $stack_ng --nodes $desired --nodes-min $updating_min --nodes-max $updating_max --timeout "60m"
+      echo
+    elif [ "$existing_min" != "$updating_min" ]; then
+      echo "￮ nodegroup $cfg_ng_name: updating min instances to $updating_min"
+      eksctl scale nodegroup --cluster=$CORTEX_CLUSTER_NAME --region=$CORTEX_REGION $stack_ng --nodes $desired --nodes-min $updating_min --timeout "60m"
+      echo
+    elif [ "$existing_max" != "$updating_max" ]; then
+      echo "￮ nodegroup $cfg_ng_name: updating max instances to $updating_max"
+      eksctl scale nodegroup --cluster=$CORTEX_CLUSTER_NAME --region=$CORTEX_REGION $stack_ng --nodes $desired --nodes-max $updating_max --timeout "60m"
+      echo
     fi
   done
 
-  if [ "$has_ng" == "false" ]; then
-    echo "error: \"cx-*-$config_ng\" node group couldn't be found"
-    exit 1
+  rm nodegroups.json
+}
+
+function add_nodegroups() {
+  if [ -z "$CORTEX_NODEGROUP_NAMES_TO_ADD" ]; then
+    return
   fi
 
-  desired=$(cat nodegroups.json | jq -r .[$eks_idx].DesiredCapacity)
-  existing_min=$(cat nodegroups.json | jq -r .[$eks_idx].MinSize)
-  existing_max=$(cat nodegroups.json | jq -r .[$eks_idx].MaxSize)
-  updating_min="$CORTEX_SCALING_MIN_INSTANCES"
-  updating_max="$CORTEX_SCALING_MAX_INSTANCES"
+  nodegroup_names="$(join_by , $CORTEX_NODEGROUP_NAMES_TO_ADD)"
 
-  if [ "$desired" -lt $updating_min ]; then
-    desired=$updating_min
-  fi
-  if [ "$desired" -gt $updating_max ]; then
-    desired=$updating_max
-  fi
+  echo "￮ adding new nodegroup(s) to the cluster ..."
+  python generate_eks.py $CORTEX_CLUSTER_CONFIG_FILE manifests/ami.json --add-cortex-node-groups="$nodegroup_names" > /workspace/nodegroups.yaml
+  eksctl create nodegroup --timeout=$EKSCTL_NODEGROUP_TIMEOUT --install-neuron-plugin=false --install-nvidia-plugin=false -f /workspace/nodegroups.yaml
+  rm /workspace/nodegroups.yaml
+  echo
+}
 
-  if [ "$existing_min" != "$updating_min" ] && [ "$existing_max" != "$updating_max" ]; then
-    echo "￮ nodegroup $config_ng: updating min instances to $updating_min and max instances to $updating_max"
-    eksctl scale nodegroup --cluster=$CORTEX_CLUSTER_NAME --region=$CORTEX_REGION $stack_ng --nodes $desired --nodes-min $updating_min --nodes-max $updating_max --timeout "60m"
-    echo
-  elif [ "$existing_min" != "$updating_min" ]; then
-    echo "￮ nodegroup $config_ng: updating min instances to $updating_min"
-    eksctl scale nodegroup --cluster=$CORTEX_CLUSTER_NAME --region=$CORTEX_REGION $stack_ng --nodes $desired --nodes-min $updating_min --timeout "60m"
-    echo
-  elif [ "$existing_max" != "$updating_max" ]; then
-    echo "￮ nodegroup $config_ng: updating max instances to $updating_max"
-    eksctl scale nodegroup --cluster=$CORTEX_CLUSTER_NAME --region=$CORTEX_REGION $stack_ng --nodes $desired --nodes-max $updating_max --timeout "60m"
-    echo
+function remove_nodegroups() {
+  if [ -z "$CORTEX_EKS_NODEGROUP_NAMES_TO_REMOVE" ]; then
+    return
   fi
 
-  rm nodegroups.json
+  eks_nodegroup_names="$(join_by , $CORTEX_EKS_NODEGROUP_NAMES_TO_REMOVE)"
+
+  echo "￮ removing nodegroup(s) from the cluster ..."
+  python generate_eks.py $CORTEX_CLUSTER_CONFIG_FILE manifests/ami.json --remove-eks-node-groups="$eks_nodegroup_names" > /workspace/nodegroups.yaml
+  eksctl delete nodegroup --timeout=$EKSCTL_NODEGROUP_TIMEOUT --approve -f /workspace/nodegroups.yaml
+  rm /workspace/nodegroups.yaml
+  echo
 }
 
 function setup_istio() {
@@ -529,4 +573,10 @@ function output_if_error() {
   set -e
 }
 
+function join_by {
+  local IFS="$1"
+  shift
+  echo "$*"
+}
+
 main
diff --git a/manager/requirements.txt b/manager/requirements.txt
index b9c8c79b97..291bc91710 100644
--- a/manager/requirements.txt
+++ b/manager/requirements.txt
@@ -2,3 +2,4 @@ boto3
 jinja2
 pyyaml
 yq
+click
diff --git a/pkg/lib/aws/cloudformation.go b/pkg/lib/aws/cloudformation.go
index 105c295895..7b2d5f04b5 100644
--- a/pkg/lib/aws/cloudformation.go
+++ b/pkg/lib/aws/cloudformation.go
@@ -22,17 +22,24 @@ import (
 	"github.com/cortexlabs/cortex/pkg/lib/sets/strset"
 )
 
-func (c *Client) ListEKSStacks(controlPlaneStackName string, nodegroupStackNames strset.Set) ([]*cloudformation.StackSummary, error) {
-	var stackSummaries []*cloudformation.StackSummary
-	stackSet := strset.Union(nodegroupStackNames, strset.New(controlPlaneStackName))
+func (c *Client) ListEKSStacks(controlPlaneStackName string, nodeGroupStackNamePrefixes strset.Set) ([]*cloudformation.StackSummary, error) {
+	mostRecentStackRecordsByName := map[string]*cloudformation.StackSummary{}
+	stackSet := strset.Union(nodeGroupStackNamePrefixes, strset.New(controlPlaneStackName))
 
 	err := c.CloudFormation().ListStacksPages(
 		&cloudformation.ListStacksInput{},
 		func(listStackOutput *cloudformation.ListStacksOutput, lastPage bool) bool {
 			for _, stackSummary := range listStackOutput.StackSummaries {
-
-				if stackSet.HasWithPrefix(*stackSummary.StackName) {
-					stackSummaries = append(stackSummaries, stackSummary)
+				if stackSummary == nil || stackSummary.StackName == nil || !stackSet.HasWithPrefix(*stackSummary.StackName) {
+					continue
+				}
+				if _, ok := mostRecentStackRecordsByName[*stackSummary.StackName]; !ok {
+					mostRecentStackRecordsByName[*stackSummary.StackName] = stackSummary
+				} else {
+					created := mostRecentStackRecordsByName[*stackSummary.StackName].CreationTime
+					if created != nil && stackSummary.CreationTime != nil && stackSummary.CreationTime.After(*created) {
+						mostRecentStackRecordsByName[*stackSummary.StackName] = stackSummary
+					}
 				}
 
 				if *stackSummary.StackName == controlPlaneStackName {
@@ -47,5 +54,13 @@ func (c *Client) ListEKSStacks(controlPlaneStackName string, nodegroupStackNames
 		return nil, errors.WithStack(err)
 	}
 
-	return stackSummaries, nil
+	return getStackSummariesFromMap(mostRecentStackRecordsByName), nil
+}
+
+func getStackSummariesFromMap(stackSummaries map[string]*cloudformation.StackSummary) []*cloudformation.StackSummary {
+	var stackSummariesSlice []*cloudformation.StackSummary
+	for _, stack := range stackSummaries {
+		stackSummariesSlice = append(stackSummariesSlice, stack)
+	}
+	return stackSummariesSlice
 }
diff --git a/pkg/lib/aws/errors.go b/pkg/lib/aws/errors.go
index f66761135e..8ded3da324 100644
--- a/pkg/lib/aws/errors.go
+++ b/pkg/lib/aws/errors.go
@@ -247,7 +247,7 @@ func ErrorSecurityGroupRulesExceeded(currentLimit, additionalQuotaRequired int,
 	url := "https://console.aws.amazon.com/servicequotas/home?#!/services/vpc/quotas"
 	return errors.WithStack(&errors.Error{
 		Kind:    ErrSecurityGroupRulesExceeded,
-		Message: fmt.Sprintf("security group rules limit of %d exceeded in region %s; use fewer availability zones, remove some node groups from your cluster config, reduce the number of CIDR white lists (if you have any), or increase your quota for inbound/outbound rules per security group by at least %d here: %s (if your request was recently approved, please allow ~30 minutes for AWS to reflect this change)", currentLimit, region, additionalQuotaRequired, url),
+		Message: fmt.Sprintf("security group rules limit of %d exceeded in region %s; remove some node groups, use fewer availability zones, reduce the number of CIDR white lists, or increase your quota for inbound/outbound rules per security group by at least %d here: %s (if your request was recently approved, please allow ~30 minutes for AWS to reflect this change)", currentLimit, region, additionalQuotaRequired, url),
 	})
 }
 
@@ -255,6 +255,6 @@ func ErrorSecurityGroupLimitExceeded(currentLimit, additionalQuotaRequired int,
 	url := "https://console.aws.amazon.com/servicequotas/home?#!/services/vpc/quotas"
 	return errors.WithStack(&errors.Error{
 		Kind:    ErrSecurityGroupLimitExceeded,
-		Message: fmt.Sprintf("security group limit of %d exceeded in region %s; remove some node groups from your cluster config or increase your quota for security groups by at least %d here: %s (if your request was recently approved, please allow ~30 minutes for AWS to reflect this change)", currentLimit, region, additionalQuotaRequired, url),
+		Message: fmt.Sprintf("security group limit of %d exceeded in region %s; remove some node groups or increase your quota for security groups by at least %d here: %s (if your request was recently approved, please allow ~30 minutes for AWS to reflect this change)", currentLimit, region, additionalQuotaRequired, url),
 	})
 }
diff --git a/pkg/lib/aws/servicequotas.go b/pkg/lib/aws/servicequotas.go
index 480e92db32..ef99e1787b 100644
--- a/pkg/lib/aws/servicequotas.go
+++ b/pkg/lib/aws/servicequotas.go
@@ -278,8 +278,8 @@ func (c *Client) VerifyVPCQuota(vpcQuota int, requiredVPCs int) error {
 	return nil
 }
 
-func (c *Client) VerifySecurityGroupQuota(securifyGroupsQuota int, numNodeGroups int) error {
-	requiredSecurityGroups := requiredSecurityGroups(numNodeGroups)
+func (c *Client) VerifySecurityGroupQuota(securifyGroupsQuota int, numNodeGroups int, clusterAlreadyExists bool) error {
+	requiredSecurityGroups := requiredSecurityGroups(numNodeGroups, clusterAlreadyExists)
 	sgs, err := c.DescribeSecurityGroups()
 	if err != nil {
 		return err
@@ -323,6 +323,7 @@ func requiredRulesForNodeGroupSecurityGroup(numAZs, whitelistLength int) int {
 	} else if whitelistLength > 1 {
 		whitelistRuleCount = 1 + 5*(whitelistLength-1)
 	}
+
 	return _baseInboundRulesForNodeGroup + numAZs*_inboundRulesPerAZ + whitelistRuleCount
 }
 
@@ -332,7 +333,10 @@ func requiredRulesForControlPlaneSecurityGroup(numNodeGroups int) int {
 	return 2 * (numNodeGroups + 1)
 }
 
-func requiredSecurityGroups(numNodeGroups int) int {
+func requiredSecurityGroups(numNodeGroups int, clusterAlreadyExists bool) int {
+	if clusterAlreadyExists {
+		return numNodeGroups
+	}
 	// each node group requires a security group
 	return _baseNumberOfSecurityGroups + numNodeGroups
 }
diff --git a/pkg/lib/structs/deepcopy.go b/pkg/lib/structs/deepcopy.go
new file mode 100644
index 0000000000..bfef819e1f
--- /dev/null
+++ b/pkg/lib/structs/deepcopy.go
@@ -0,0 +1,30 @@
+/*
+Copyright 2021 Cortex Labs, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package structs
+
+import (
+	"bytes"
+	"encoding/gob"
+)
+
+func DeepCopy(dst, src interface{}) error {
+	var buf bytes.Buffer
+	if err := gob.NewEncoder(&buf).Encode(src); err != nil {
+		return err
+	}
+	return gob.NewDecoder(bytes.NewBuffer(buf.Bytes())).Decode(dst)
+}
diff --git a/pkg/operator/schema/schema.go b/pkg/operator/schema/schema.go
index 8a896d4d10..401686fee3 100644
--- a/pkg/operator/schema/schema.go
+++ b/pkg/operator/schema/schema.go
@@ -25,22 +25,22 @@ import (
 )
 
 type InfoResponse struct {
-	ClusterConfig      clusterconfig.InternalConfig `json:"cluster_config"`
-	NodeInfos          []NodeInfo                   `json:"node_infos"`
-	NumPendingReplicas int                          `json:"num_pending_replicas"`
+	ClusterConfig      clusterconfig.InternalConfig `json:"cluster_config" yaml:"cluster_config"`
+	NodeInfos          []NodeInfo                   `json:"node_infos" yaml:"node_infos"`
+	NumPendingReplicas int                          `json:"num_pending_replicas" yaml:"num_pending_replicas"`
 }
 
 type NodeInfo struct {
-	Name                    string             `json:"name"`
-	NodeGroupName           string             `json:"nodegroup_name"`
-	InstanceType            string             `json:"instance_type"`
-	IsSpot                  bool               `json:"is_spot"`
-	Price                   float64            `json:"price"`
-	NumReplicas             int                `json:"num_replicas"`
-	NumAsyncGatewayReplicas int                `json:"num_async_gateway_replicas"`
-	ComputeUserCapacity     userconfig.Compute `json:"compute_user_capacity"`  // the total resources available to the user on a node
-	ComputeAvailable        userconfig.Compute `json:"compute_available"`      // unused resources on a node
-	ComputeUserRequested    userconfig.Compute `json:"compute_user_requested"` // total resources requested by user on a node
+	Name                    string             `json:"name" yaml:"name"`
+	NodeGroupName           string             `json:"nodegroup_name" yaml:"nodegroup_name"`
+	InstanceType            string             `json:"instance_type" yaml:"instance_type"`
+	IsSpot                  bool               `json:"is_spot" yaml:"is_spot"`
+	Price                   float64            `json:"price" yaml:"price"`
+	NumReplicas             int                `json:"num_replicas" yaml:"num_replicas"`
+	NumAsyncGatewayReplicas int                `json:"num_async_gateway_replicas" yaml:"num_async_gateway_replicas"`
+	ComputeUserCapacity     userconfig.Compute `json:"compute_user_capacity" yaml:"compute_user_capacity"`   // the total resources available to the user on a node
+	ComputeAvailable        userconfig.Compute `json:"compute_available" yaml:"compute_unavailable"`         // unused resources on a node
+	ComputeUserRequested    userconfig.Compute `json:"compute_user_requested" yaml:"compute_user_requested"` // total resources requested by user on a node
 }
 
 type DeployResult struct {
diff --git a/pkg/types/clusterconfig/cluster_config.go b/pkg/types/clusterconfig/cluster_config.go
index d82d2fa627..f8f10d7f83 100644
--- a/pkg/types/clusterconfig/cluster_config.go
+++ b/pkg/types/clusterconfig/cluster_config.go
@@ -28,6 +28,7 @@ import (
 	"time"
 
 	"github.com/aws/amazon-vpc-cni-k8s/pkg/awsutils"
+	"github.com/aws/aws-sdk-go/service/cloudformation"
 	"github.com/aws/aws-sdk-go/service/iam"
 	"github.com/cortexlabs/cortex/pkg/consts"
 	"github.com/cortexlabs/cortex/pkg/lib/aws"
@@ -38,6 +39,7 @@ import (
 	"github.com/cortexlabs/cortex/pkg/lib/pointer"
 	"github.com/cortexlabs/cortex/pkg/lib/sets/strset"
 	"github.com/cortexlabs/cortex/pkg/lib/slices"
+	"github.com/cortexlabs/cortex/pkg/lib/structs"
 	"github.com/cortexlabs/yaml"
 )
 
@@ -135,6 +137,7 @@ type NodeGroup struct {
 	InstanceType             string      `json:"instance_type" yaml:"instance_type"`
 	MinInstances             int64       `json:"min_instances" yaml:"min_instances"`
 	MaxInstances             int64       `json:"max_instances" yaml:"max_instances"`
+	Priority                 int64       `json:"priority" yaml:"priority"`
 	InstanceVolumeSize       int64       `json:"instance_volume_size" yaml:"instance_volume_size"`
 	InstanceVolumeType       VolumeType  `json:"instance_volume_type" yaml:"instance_volume_type"`
 	InstanceVolumeIOPS       *int64      `json:"instance_volume_iops" yaml:"instance_volume_iops"`
@@ -162,10 +165,10 @@ type Config struct {
 }
 
 type OperatorMetadata struct {
-	APIVersion          string `json:"api_version"`
-	OperatorID          string `json:"operator_id"`
-	ClusterID           string `json:"cluster_id"`
-	IsOperatorInCluster bool   `json:"is_operator_in_cluster"`
+	APIVersion          string `json:"api_version" yaml:"api_version"`
+	OperatorID          string `json:"operator_id" yaml:"operator_id"`
+	ClusterID           string `json:"cluster_id" yaml:"cluster_id"`
+	IsOperatorInCluster bool   `json:"is_operator_in_cluster" yaml:"is_operator_in_cluster"`
 }
 
 type InternalConfig struct {
@@ -184,6 +187,33 @@ type AccessConfig struct {
 	ImageManager string `json:"image_manager" yaml:"image_manager"`
 }
 
+type ConfigureChanges struct {
+	NodeGroupsToAdd       []string
+	NodeGroupsToRemove    []string
+	NodeGroupsToScale     []string
+	EKSNodeGroupsToRemove []string // EKS node group names of (NodeGroupsToRemove ∩ Cortex-converted EKS node groups) ∪ (Cortex-converted EKS node groups - the new cluster config's nodegroups)
+}
+
+func (c *ConfigureChanges) HasChanges() bool {
+	return len(c.NodeGroupsToAdd) != 0 || len(c.NodeGroupsToRemove) != 0 || len(c.NodeGroupsToScale) != 0 || len(c.EKSNodeGroupsToRemove) != 0
+}
+
+// GetGhostEKSNodeGroups returns the set difference between EKSNodeGroupsToRemove and the EKS-converted NodeGroupsToRemove
+func (c *ConfigureChanges) GetGhostEKSNodeGroups() []string {
+	if len(c.EKSNodeGroupsToRemove) <= len(c.NodeGroupsToRemove) {
+		return nil
+	}
+
+	eksNodeGroupPrefix := "cx-wx-"
+	var ghostEKSNodeGroups []string
+	for _, eksNgToRemove := range c.EKSNodeGroupsToRemove {
+		if !slices.HasString(c.NodeGroupsToRemove, eksNgToRemove[len(eksNodeGroupPrefix):]) {
+			ghostEKSNodeGroups = append(ghostEKSNodeGroups, eksNgToRemove)
+		}
+	}
+	return ghostEKSNodeGroups
+}
+
 // NewForFile initializes and validates the cluster config from the YAML config file
 func NewForFile(clusterConfigPath string) (*Config, error) {
 	coreConfig := CoreConfig{}
@@ -219,13 +249,8 @@ func RegionValidator(region string) (string, error) {
 }
 
 func (cc *Config) DeepCopy() (Config, error) {
-	bytes, err := yaml.Marshal(cc)
-	if err != nil {
-		return Config{}, err
-	}
-
 	deepCopied := Config{}
-	err = yaml.Unmarshal(bytes, &deepCopied)
+	err := structs.DeepCopy(&deepCopied, cc)
 	if err != nil {
 		return Config{}, err
 	}
@@ -494,126 +519,8 @@ var ManagedConfigStructFieldValidations = []*cr.StructFieldValidation{
 	{
 		StructField: "NodeGroups",
 		StructListValidation: &cr.StructListValidation{
-			Required: true,
-			StructValidation: &cr.StructValidation{
-				StructFieldValidations: []*cr.StructFieldValidation{
-					{
-						StructField: "Name",
-						StringValidation: &cr.StringValidation{
-							Required:                   true,
-							AlphaNumericDashUnderscore: true,
-							MaxLength:                  _maxNodeGroupLength,
-						},
-					},
-					{
-						StructField: "InstanceType",
-						StringValidation: &cr.StringValidation{
-							Required:  true,
-							MinLength: 1,
-							Validator: validateInstanceType,
-						},
-					},
-					{
-						StructField: "MinInstances",
-						Int64Validation: &cr.Int64Validation{
-							Default:              int64(1),
-							GreaterThanOrEqualTo: pointer.Int64(0),
-						},
-					},
-					{
-						StructField: "MaxInstances",
-						Int64Validation: &cr.Int64Validation{
-							Default:              int64(5),
-							GreaterThanOrEqualTo: pointer.Int64(0), // this will be validated to be > 0 during cluster up (can be scaled down later)
-						},
-					},
-					{
-						StructField: "InstanceVolumeSize",
-						Int64Validation: &cr.Int64Validation{
-							Default:              50,
-							GreaterThanOrEqualTo: pointer.Int64(20), // large enough to fit docker images and any other overhead
-							LessThanOrEqualTo:    pointer.Int64(16384),
-						},
-					},
-					{
-						StructField: "InstanceVolumeType",
-						StringValidation: &cr.StringValidation{
-							AllowedValues: VolumeTypesStrings(),
-							Default:       GP3VolumeType.String(),
-						},
-						Parser: func(str string) (interface{}, error) {
-							return VolumeTypeFromString(str), nil
-						},
-					},
-					{
-						StructField: "InstanceVolumeIOPS",
-						Int64PtrValidation: &cr.Int64PtrValidation{
-							AllowExplicitNull: true,
-						},
-					},
-					{
-						StructField: "InstanceVolumeThroughput",
-						Int64PtrValidation: &cr.Int64PtrValidation{
-							GreaterThanOrEqualTo: pointer.Int64(125),
-							LessThanOrEqualTo:    pointer.Int64(1000),
-							AllowExplicitNull:    true,
-						},
-					},
-					{
-						StructField: "Spot",
-						BoolValidation: &cr.BoolValidation{
-							Default: false,
-						},
-					},
-					{
-						StructField: "SpotConfig",
-						StructValidation: &cr.StructValidation{
-							DefaultNil:        true,
-							AllowExplicitNull: true,
-							StructFieldValidations: []*cr.StructFieldValidation{
-								{
-									StructField: "InstanceDistribution",
-									StringListValidation: &cr.StringListValidation{
-										DisallowDups:      true,
-										Validator:         validateInstanceDistribution,
-										AllowExplicitNull: true,
-									},
-								},
-								{
-									StructField: "OnDemandBaseCapacity",
-									Int64PtrValidation: &cr.Int64PtrValidation{
-										GreaterThanOrEqualTo: pointer.Int64(0),
-										AllowExplicitNull:    true,
-									},
-								},
-								{
-									StructField: "OnDemandPercentageAboveBaseCapacity",
-									Int64PtrValidation: &cr.Int64PtrValidation{
-										GreaterThanOrEqualTo: pointer.Int64(0),
-										LessThanOrEqualTo:    pointer.Int64(100),
-										AllowExplicitNull:    true,
-									},
-								},
-								{
-									StructField: "MaxPrice",
-									Float64PtrValidation: &cr.Float64PtrValidation{
-										GreaterThan:       pointer.Float64(0),
-										AllowExplicitNull: true,
-									},
-								},
-								{
-									StructField: "InstancePools",
-									Int64PtrValidation: &cr.Int64PtrValidation{
-										GreaterThanOrEqualTo: pointer.Int64(1),
-										LessThanOrEqualTo:    pointer.Int64(int64(_maxInstancePools)),
-										AllowExplicitNull:    true,
-									},
-								},
-							},
-						},
-					},
-				},
-			},
+			Required:         true,
+			StructValidation: nodeGroupsFieldValidation,
 		},
 	},
 	{
@@ -786,6 +693,134 @@ var ManagedConfigStructFieldValidations = []*cr.StructFieldValidation{
 	},
 }
 
+var nodeGroupsFieldValidation *cr.StructValidation = &cr.StructValidation{
+	StructFieldValidations: []*cr.StructFieldValidation{
+		{
+			StructField: "Name",
+			StringValidation: &cr.StringValidation{
+				Required:                   true,
+				AlphaNumericDashUnderscore: true,
+				MaxLength:                  _maxNodeGroupLength,
+			},
+		},
+		{
+			StructField: "InstanceType",
+			StringValidation: &cr.StringValidation{
+				Required:  true,
+				MinLength: 1,
+				Validator: validateInstanceType,
+			},
+		},
+		{
+			StructField: "MinInstances",
+			Int64Validation: &cr.Int64Validation{
+				Default:              int64(1),
+				GreaterThanOrEqualTo: pointer.Int64(0),
+			},
+		},
+		{
+			StructField: "MaxInstances",
+			Int64Validation: &cr.Int64Validation{
+				Default:              int64(5),
+				GreaterThanOrEqualTo: pointer.Int64(0), // this will be validated to be > 0 during cluster up (can be scaled down later)
+			},
+		},
+		{
+			StructField: "Priority",
+			Int64Validation: &cr.Int64Validation{
+				Default:              int64(1),
+				GreaterThanOrEqualTo: pointer.Int64(1),
+				LessThanOrEqualTo:    pointer.Int64(100),
+			},
+		},
+		{
+			StructField: "InstanceVolumeSize",
+			Int64Validation: &cr.Int64Validation{
+				Default:              50,
+				GreaterThanOrEqualTo: pointer.Int64(20), // large enough to fit docker images and any other overhead
+				LessThanOrEqualTo:    pointer.Int64(16384),
+			},
+		},
+		{
+			StructField: "InstanceVolumeType",
+			StringValidation: &cr.StringValidation{
+				AllowedValues: VolumeTypesStrings(),
+				Default:       GP3VolumeType.String(),
+			},
+			Parser: func(str string) (interface{}, error) {
+				return VolumeTypeFromString(str), nil
+			},
+		},
+		{
+			StructField: "InstanceVolumeIOPS",
+			Int64PtrValidation: &cr.Int64PtrValidation{
+				AllowExplicitNull: true,
+			},
+		},
+		{
+			StructField: "InstanceVolumeThroughput",
+			Int64PtrValidation: &cr.Int64PtrValidation{
+				GreaterThanOrEqualTo: pointer.Int64(125),
+				LessThanOrEqualTo:    pointer.Int64(1000),
+				AllowExplicitNull:    true,
+			},
+		},
+		{
+			StructField: "Spot",
+			BoolValidation: &cr.BoolValidation{
+				Default: false,
+			},
+		},
+		{
+			StructField: "SpotConfig",
+			StructValidation: &cr.StructValidation{
+				DefaultNil:        true,
+				AllowExplicitNull: true,
+				StructFieldValidations: []*cr.StructFieldValidation{
+					{
+						StructField: "InstanceDistribution",
+						StringListValidation: &cr.StringListValidation{
+							DisallowDups:      true,
+							Validator:         validateInstanceDistribution,
+							AllowExplicitNull: true,
+						},
+					},
+					{
+						StructField: "OnDemandBaseCapacity",
+						Int64PtrValidation: &cr.Int64PtrValidation{
+							GreaterThanOrEqualTo: pointer.Int64(0),
+							AllowExplicitNull:    true,
+						},
+					},
+					{
+						StructField: "OnDemandPercentageAboveBaseCapacity",
+						Int64PtrValidation: &cr.Int64PtrValidation{
+							GreaterThanOrEqualTo: pointer.Int64(0),
+							LessThanOrEqualTo:    pointer.Int64(100),
+							AllowExplicitNull:    true,
+						},
+					},
+					{
+						StructField: "MaxPrice",
+						Float64PtrValidation: &cr.Float64PtrValidation{
+							GreaterThan:       pointer.Float64(0),
+							AllowExplicitNull: true,
+						},
+					},
+					{
+						StructField: "InstancePools",
+						Int64PtrValidation: &cr.Int64PtrValidation{
+							GreaterThanOrEqualTo: pointer.Int64(1),
+							LessThanOrEqualTo:    pointer.Int64(int64(_maxInstancePools)),
+							AllowExplicitNull:    true,
+						},
+					},
+				},
+			},
+		},
+	},
+}
+
 func CoreConfigValidations(allowExtraFields bool) *cr.StructValidation {
 	return &cr.StructValidation{
 		Required:               true,
@@ -855,10 +890,7 @@ func (cc *CoreConfig) SQSNamePrefix() string {
 	return SQSNamePrefix(cc.ClusterName)
 }
 
-// this validates the user-provided cluster config
-func (cc *Config) Validate(awsClient *aws.Client) error {
-	fmt.Print("verifying your configuration ...\n\n")
-
+func (cc *Config) validate(awsClient *aws.Client) error {
 	numNodeGroups := len(cc.NodeGroups)
 	if numNodeGroups == 0 {
 		return ErrorNoNodeGroupSpecified()
@@ -931,11 +963,6 @@ func (cc *Config) Validate(awsClient *aws.Client) error {
 		return ErrorS3RegionDiffersFromCluster(cc.Bucket, bucketRegion, cc.Region)
 	}
 
-	if cc.ClusterUID != "" {
-		return ErrorDisallowedField(ClusterUIDKey)
-	}
-	cc.ClusterUID = strconv.FormatInt(time.Now().Unix(), 10)
-
 	if cc.CortexPolicyARN != "" {
 		return ErrorDisallowedField(CortexPolicyARNKey)
 	}
@@ -995,12 +1022,101 @@ func (cc *Config) Validate(awsClient *aws.Client) error {
 		}
 	}
 
+	return nil
+}
+
+func (cc *Config) validateConfigDiff(oldConfig Config) error {
+	err := cc.validateTopLevelSectionDiff(oldConfig)
+	if err != nil {
+		return err
+	}
+
+	return cc.validateSharedNodeGroupsDiff(oldConfig)
+}
+
+func (cc *Config) validateTopLevelSectionDiff(oldConfig Config) error {
+	// validate actionable changes
+	newClusterConfigCopy, err := cc.DeepCopy()
+	if err != nil {
+		return err
+	}
+
+	oldClusterConfigCopy, err := oldConfig.DeepCopy()
+	if err != nil {
+		return err
+	}
+
+	newClusterConfigCopy.NodeGroups = []*NodeGroup{}
+	oldClusterConfigCopy.NodeGroups = []*NodeGroup{}
+
+	h1, err := newClusterConfigCopy.Hash()
+	if err != nil {
+		return err
+	}
+	h2, err := oldClusterConfigCopy.Hash()
+	if err != nil {
+		return err
+	}
+	if h1 != h2 {
+		return ErrorConfigCannotBeChangedOnConfigure()
+	}
+
+	return nil
+}
+
+func (cc *Config) validateSharedNodeGroupsDiff(oldConfig Config) error {
+	sharedNgsFromNewConfig, sharedNgsFromOldConfig := cc.getCommonNodeGroups(oldConfig)
+	for i := range sharedNgsFromNewConfig {
+		newNgCopy, err := sharedNgsFromNewConfig[i].DeepCopy()
+		if err != nil {
+			return errors.Wrap(err, NodeGroupsKey)
+		}
+		oldNgCopy, err := sharedNgsFromOldConfig[i].DeepCopy()
+		if err != nil {
+			return errors.Wrap(err, NodeGroupsKey)
+		}
+
+		newNgCopy.MinInstances = 0
+		newNgCopy.MaxInstances = 0
+		oldNgCopy.MinInstances = 0
+		oldNgCopy.MaxInstances = 0
+
+		newHash, err := newNgCopy.Hash()
+		if err != nil {
+			return errors.Wrap(err, NodeGroupsKey)
+		}
+		oldHash, err := oldNgCopy.Hash()
+		if err != nil {
+			return errors.Wrap(err, NodeGroupsKey)
+		}
+
+		if newHash != oldHash {
+			return errors.Wrap(ErrorNodeGroupCanOnlyBeScaled(), NodeGroupsKey, newNgCopy.Name)
+		}
+	}
+	return nil
+}
+
+// this validates the user-provided cluster config
+func (cc *Config) ValidateOnInstall(awsClient *aws.Client) error {
+	fmt.Print("verifying your configuration ...\n\n")
+
+	err := cc.validate(awsClient)
+	if err != nil {
+		return err
+	}
+
+	if cc.ClusterUID != "" {
+		return ErrorDisallowedField(ClusterUIDKey)
+	}
+	cc.ClusterUID = strconv.FormatInt(time.Now().Unix(), 10)
+
 	var requiredVPCs int
 	if len(cc.Subnets) == 0 {
 		requiredVPCs = 1
 	}
 	longestCIDRWhiteList := libmath.MaxInt(len(cc.APILoadBalancerCIDRWhiteList), len(cc.OperatorLoadBalancerCIDRWhiteList))
-	if err := VerifyNetworkQuotas(awsClient, 1, cc.NATGateway != NoneNATGateway, cc.NATGateway == HighlyAvailableNATGateway, requiredVPCs, strset.FromSlice(cc.AvailabilityZones), len(cc.NodeGroups), longestCIDRWhiteList); err != nil {
+	if err := VerifyNetworkQuotas(awsClient, 1, cc.NATGateway != NoneNATGateway, cc.NATGateway == HighlyAvailableNATGateway, requiredVPCs, strset.FromSlice(cc.AvailabilityZones), len(cc.NodeGroups), len(cc.NodeGroups), longestCIDRWhiteList, false); err != nil {
 		// Skip AWS errors, since some regions (e.g. eu-north-1) do not support this API
 		if !aws.IsAWSError(err) {
 			return err
@@ -1010,6 +1126,49 @@ func (cc *Config) Validate(awsClient *aws.Client) error {
 	return nil
 }
 
+func (cc *Config) ValidateOnConfigure(awsClient *aws.Client, oldConfig Config, eksNodeGroupStacks []*cloudformation.StackSummary) (ConfigureChanges, error) {
+	fmt.Print("verifying your configuration ...\n\n")
+
+	cc.ClusterUID = oldConfig.ClusterUID
+	err := cc.validate(awsClient)
+	if err != nil {
+		return ConfigureChanges{}, err
+	}
+
+	err = cc.validateConfigDiff(oldConfig)
+	if err != nil {
+		return ConfigureChanges{}, err
+	}
+
+	ngsToBeAdded := cc.getNewNodeGroups(oldConfig)
+	ngsToBeRemoved := cc.getRemovedNodeGroups(oldConfig)
+
+	tempMaxNodeGroupCount := len(cc.NodeGroups) + len(ngsToBeRemoved)
+	tempNetAdditionOfNodeGroupCount := tempMaxNodeGroupCount - len(oldConfig.NodeGroups)
+	longestCIDRWhiteList := libmath.MaxInt(len(cc.APILoadBalancerCIDRWhiteList), len(cc.OperatorLoadBalancerCIDRWhiteList))
+	if err := VerifyNetworkQuotasOnConfigure(awsClient, strset.FromSlice(cc.AvailabilityZones), tempMaxNodeGroupCount, tempNetAdditionOfNodeGroupCount, longestCIDRWhiteList); err != nil {
+		// Skip AWS errors, since some regions (e.g. eu-north-1) do not support this API
+		if !aws.IsAWSError(err) {
+			return ConfigureChanges{}, errors.Wrap(err, NodeGroupsKey)
+		}
+	}
+
+	sharedNgsFromNewConfig, sharedNgsFromOldConfig := cc.getCommonNodeGroups(oldConfig)
+	ngNamesToBeScaled := []*NodeGroup{}
+	for i := range sharedNgsFromNewConfig {
+		if sharedNgsFromNewConfig[i].MinInstances != sharedNgsFromOldConfig[i].MinInstances || sharedNgsFromNewConfig[i].MaxInstances != sharedNgsFromOldConfig[i].MaxInstances {
+			ngNamesToBeScaled = append(ngNamesToBeScaled, sharedNgsFromNewConfig[i])
+		}
+	}
+
+	return ConfigureChanges{
+		NodeGroupsToAdd:       GetNodeGroupNames(ngsToBeAdded),
+		NodeGroupsToRemove:    GetNodeGroupNames(ngsToBeRemoved),
+		NodeGroupsToScale:     GetNodeGroupNames(ngNamesToBeScaled),
+		EKSNodeGroupsToRemove: getStaleEksNodeGroups(cc.ClusterName, eksNodeGroupStacks, cc.NodeGroups, ngsToBeRemoved),
+	}, nil
+}
+
 func (ng *NodeGroup) validateNodeGroup(awsClient *aws.Client, region string) error {
 	if ng.MinInstances > ng.MaxInstances {
 		return ErrorMinInstancesGreaterThanMax(ng.MinInstances, ng.MaxInstances)
@@ -1116,6 +1275,76 @@ func (ng *NodeGroup) validateNodeGroup(awsClient *aws.Client, region string) err
 	return nil
 }
 
+func (cc *Config) GetNodeGroupByName(name string) *NodeGroup {
+	for _, ng := range cc.NodeGroups {
+		if ng.Name == name {
+			return ng
+		}
+	}
+	return nil
+}
+
+func (cc *Config) getNewNodeGroups(oldConfig Config) []*NodeGroup {
+	var newNodeGroups []*NodeGroup
+	for _, updatingNg := range cc.NodeGroups {
+		isNewNg := true
+		for _, previousNg := range oldConfig.NodeGroups {
+			if previousNg.Name == updatingNg.Name {
+				isNewNg = false
+				break
+			}
+		}
+		if isNewNg {
+			ngCopy := *updatingNg
+			newNodeGroups = append(newNodeGroups, &ngCopy)
+		}
+	}
+	return newNodeGroups
+}
+
+func (cc *Config) getRemovedNodeGroups(oldConfig Config) []*NodeGroup {
+	var removedNodeGroups []*NodeGroup
+	for _, previousNg := range oldConfig.NodeGroups {
+		isRemovedNg := true
+		for _, updatingNg := range cc.NodeGroups {
+			if previousNg.Name == updatingNg.Name {
+				isRemovedNg = false
+				break
+			}
+		}
+		if isRemovedNg {
+			ngCopy := *previousNg
+			removedNodeGroups = append(removedNodeGroups, &ngCopy)
+		}
+	}
+	return removedNodeGroups
+}
+
+func (cc *Config) getCommonNodeGroups(oldConfig Config) ([]*NodeGroup, []*NodeGroup) {
+	var commonNewNodeGroups []*NodeGroup
+	var commonOldNodeGroups []*NodeGroup
+	for _, previousNg := range oldConfig.NodeGroups {
+		for _, updatingNg := range cc.NodeGroups {
+			if previousNg.Name == updatingNg.Name {
+				ngNewCopy := *updatingNg
+				ngOldCopy := *previousNg
+				commonNewNodeGroups = append(commonNewNodeGroups, &ngNewCopy)
+				commonOldNodeGroups = append(commonOldNodeGroups, &ngOldCopy)
+				break
+			}
+		}
+	}
+	return commonNewNodeGroups, commonOldNodeGroups
+}
+
+func GetNodeGroupNames(nodeGroups []*NodeGroup) []string {
+	ngNames := make([]string, len(nodeGroups))
+	for i := range nodeGroups {
+		ngNames[i] = nodeGroups[i].Name
+	}
+	return ngNames
+}
+
 func CheckSpotInstanceCompatibility(target aws.InstanceMetadata, suggested aws.InstanceMetadata) error {
 	if target.Inf > 0 && suggested.Inf == 0 {
 		return ErrorIncompatibleSpotInstanceTypeInf(suggested)
@@ -1249,6 +1478,27 @@ func validateInstanceDistribution(instances []string) ([]string, error) {
 	return instances, nil
 }
 
+func (ng *NodeGroup) DeepCopy() (NodeGroup, error) {
+	deepCopied := NodeGroup{}
+	err := structs.DeepCopy(&deepCopied, ng)
+	if err != nil {
+		return NodeGroup{}, err
+	}
+
+	return deepCopied, nil
+}
+
+func (ng *NodeGroup) Hash() (string, error) {
+	bytes, err := yaml.Marshal(ng)
+	if err != nil {
+		return "", err
+	}
+
+	hash := sha256.New()
+	hash.Write(bytes)
+	return hex.EncodeToString(hash.Sum(nil)), nil
+}
+
 func (ng *NodeGroup) MaxPossibleOnDemandInstances() int64 {
 	if !ng.Spot || ng.SpotConfig == nil {
 		return ng.MaxInstances
@@ -1287,6 +1537,72 @@ func (ng *NodeGroup) SpotConfigOnDemandValues() (int64, int64) {
 	return onDemandBaseCapacity, onDemandPercentageAboveBaseCapacity
 }
 
+func doesStackExist(stack *cloudformation.StackSummary) bool {
+	if stack == nil || stack.StackName == nil || slices.HasString([]string{
+		cloudformation.StackStatusDeleteComplete,
+		cloudformation.StackStatusDeleteInProgress,
+	}, *stack.StackStatus) {
+		return false
+	}
+	if slices.HasString([]string{
+		cloudformation.StackStatusCreateComplete,
+		cloudformation.StackStatusUpdateInProgress,
+		cloudformation.StackStatusUpdateComplete,
+		cloudformation.StackStatusUpdateRollbackComplete,
+		cloudformation.StackStatusRollbackComplete,
+	}, *stack.StackStatus) {
+		return true
+	}
+	return false
+}
+
+func getStaleEksNodeGroups(clusterName string, eksNodeGroupStacks []*cloudformation.StackSummary, ngsToExist, ngsMarkedForRemoval []*NodeGroup) []string {
+	eksNodeGroupsToRemove := strset.New()
+	for _, ng := range ngsMarkedForRemoval {
+		lifecycle := "d"
+		if ng.Spot {
+			lifecycle = "s"
+		}
+
+		eksNgName := fmt.Sprintf("cx-w%s-%s", lifecycle, ng.Name)
+		eksStackName := fmt.Sprintf("eksctl-%s-nodegroup-cx-w%s-%s", clusterName, lifecycle, ng.Name)
+		for _, eksNgStack := range eksNodeGroupStacks {
+			if eksNgStack == nil || eksNgStack.StackName == nil {
+				continue
+			}
+			if *eksNgStack.StackName == eksStackName {
+				eksNodeGroupsToRemove.Add(eksNgName)
+				break
+			}
+		}
+	}
+
+	for _, eksNgStack := range eksNodeGroupStacks {
+		if !doesStackExist(eksNgStack) {
+			continue
+		}
+
+		foundNg := false
+		for _, ng := range ngsToExist {
+			lifecycle := "d"
+			if ng.Spot {
+				lifecycle = "s"
+			}
+			eksStackName := fmt.Sprintf("eksctl-%s-nodegroup-cx-w%s-%s", clusterName, lifecycle, ng.Name)
+			if *eksNgStack.StackName == eksStackName {
+				foundNg = true
+				break
+			}
+		}
+		if !foundNg {
+			eksNgName := (*eksNgStack.StackName)[len(fmt.Sprintf("eksctl-%s-nodegroup-", clusterName)):]
+			eksNodeGroupsToRemove.Add(eksNgName)
+		}
+	}
+
+	return eksNodeGroupsToRemove.Slice()
+}
+
 func (cc *CoreConfig) TelemetryEvent() map[string]interface{} {
 	event := make(map[string]interface{})
 
@@ -1434,6 +1750,7 @@ func (mc *ManagedConfig) TelemetryEvent() map[string]interface{} {
 		event[nodeGroupKey("instance_type")] = ng.InstanceType
 		event[nodeGroupKey("min_instances")] = ng.MinInstances
 		event[nodeGroupKey("max_instances")] = ng.MaxInstances
+		event[nodeGroupKey("priority")] = ng.Priority
 		event[nodeGroupKey("instance_volume_size")] = ng.InstanceVolumeSize
 		event[nodeGroupKey("instance_volume_type")] = ng.InstanceVolumeType
 		if ng.InstanceVolumeIOPS != nil {
@@ -1511,9 +1828,9 @@ func (mc *ManagedConfig) GetNodeGroupByName(name string) *NodeGroup {
 }
 
 func (mc *ManagedConfig) GetNodeGroupNames() []string {
-	allNodeGroupNames := []string{}
-	for _, ng := range mc.NodeGroups {
-		allNodeGroupNames = append(allNodeGroupNames, ng.Name)
+	allNodeGroupNames := make([]string, len(mc.NodeGroups))
+	for i := range mc.NodeGroups {
+		allNodeGroupNames[i] = mc.NodeGroups[i].Name
 	}
 
 	return allNodeGroupNames
diff --git a/pkg/types/clusterconfig/errors.go b/pkg/types/clusterconfig/errors.go
index fa69e95c23..c6c13113ba 100644
--- a/pkg/types/clusterconfig/errors.go
+++ b/pkg/types/clusterconfig/errors.go
@@ -57,6 +57,8 @@ const (
 	ErrUnsupportedAvailabilityZone            = "clusterconfig.unsupported_availability_zone"
 	ErrNotEnoughValidDefaultAvailibilityZones = "clusterconfig.not_enough_valid_default_availability_zones"
 	ErrNoNATGatewayWithSubnets                = "clusterconfig.no_nat_gateway_with_subnets"
+	ErrConfigCannotBeChangedOnConfigure       = "clusterconfig.config_cannot_be_changed_on_configure"
+	ErrNodeGroupCanOnlyBeScaled               = "clusterconfig.node_group_can_only_be_scaled"
 	ErrSpecifyOneOrNone                       = "clusterconfig.specify_one_or_none"
 	ErrSpecifyTwoOrNone                       = "clusterconfig.specify_two_or_none"
 	ErrDependentFieldMustBeSpecified          = "clusterconfig.dependent_field_must_be_specified"
@@ -113,7 +115,7 @@ func ErrorNoNodeGroupSpecified() error {
 func ErrorNodeGroupMaxInstancesIsZero() error {
 	return errors.WithStack(&errors.Error{
 		Kind:    ErrNodeGroupMaxInstancesIsZero,
-		Message: fmt.Sprintf("nodegroups cannot be created with `%s` set to 0 (but `%s` can be scaled to 0 after the cluster has been created)", MaxInstancesKey, MaxInstancesKey),
+		Message: fmt.Sprintf("nodegroups cannot be created with `%s` set to 0 (but `%s` can be scaled to 0 after the nodegroup has been created)", MaxInstancesKey, MaxInstancesKey),
 	})
 }
 
@@ -283,6 +285,20 @@ func ErrorNoNATGatewayWithSubnets() error {
 	})
 }
 
+func ErrorConfigCannotBeChangedOnConfigure() error {
+	return errors.WithStack(&errors.Error{
+		Kind:    ErrConfigCannotBeChangedOnConfigure,
+		Message: fmt.Sprintf("in a running cluster, only the %s field can be modified", NodeGroupsKey),
+	})
+}
+
+func ErrorNodeGroupCanOnlyBeScaled() error {
+	return errors.WithStack(&errors.Error{
+		Kind:    ErrNodeGroupCanOnlyBeScaled,
+		Message: "in a running cluster, nodegroups can only be scaled, added, or deleted",
+	})
+}
+
 func ErrorSpecifyOneOrNone(fieldName1 string, fieldName2 string, fieldNames ...string) error {
 	fieldNames = append([]string{fieldName1, fieldName2}, fieldNames...)
 
diff --git a/pkg/types/clusterconfig/network_validations.go b/pkg/types/clusterconfig/network_validations.go
index 70ed4e3f04..91f3d112e8 100644
--- a/pkg/types/clusterconfig/network_validations.go
+++ b/pkg/types/clusterconfig/network_validations.go
@@ -25,12 +25,12 @@ import (
 )
 
 const (
-	_elasticIPsQuotaCode         = "L-0263D0A3"
-	_internetGatewayQuotaCode    = "L-A4707A72"
-	_natGatewayQuotaCode         = "L-FE5A380F"
-	_vpcQuotaCode                = "L-F678F1CE"
-	_securityGroupsQuotaCode     = "L-E79EC296"
-	_securityGroupRulesQuotaCode = "L-0EA8095F"
+	_elasticIPsQuotaCode         = "L-0263D0A3" // from EC2 service
+	_internetGatewayQuotaCode    = "L-A4707A72" // from EC2 service
+	_natGatewayQuotaCode         = "L-FE5A380F" // from EC2 service
+	_vpcQuotaCode                = "L-F678F1CE" // from VPC service
+	_securityGroupsQuotaCode     = "L-E79EC296" // from VPC service
+	_securityGroupRulesQuotaCode = "L-0EA8095F" // from VPC service
 )
 
 func VerifyNetworkQuotas(
@@ -41,18 +41,26 @@ func VerifyNetworkQuotas(
 	requiredVPCs int,
 	availabilityZones strset.Set,
 	numNodeGroups int,
-	longestCIDRWhiteList int) error {
+	netAdditionOfNodeGroups int,
+	longestCIDRWhiteList int,
+	clusterAlreadyExists bool,
+) error {
 
 	desiredQuotaCodes := []string{
-		_elasticIPsQuotaCode,
-		_internetGatewayQuotaCode,
-		_natGatewayQuotaCode,
 		_vpcQuotaCode,
 		_securityGroupsQuotaCode,
 		_securityGroupRulesQuotaCode,
 	}
-
-	serviceCodes := []string{"ec2", "vpc"}
+	serviceCodes := []string{"vpc"}
+
+	if !clusterAlreadyExists {
+		desiredQuotaCodes = append(desiredQuotaCodes,
+			_elasticIPsQuotaCode,
+			_internetGatewayQuotaCode,
+			_natGatewayQuotaCode,
+		)
+		serviceCodes = append(serviceCodes, "ec2")
+	}
 
 	quotaCodeToValueMap, err := awsClient.ListServiceQuotas(desiredQuotaCodes, serviceCodes)
 	if err != nil {
@@ -66,49 +74,53 @@ func VerifyNetworkQuotas(
 		}
 	}()
 
-	// check internet GW quota
-	if requiredInternetGateways > 0 {
-		if internetGatewayQuota, found := quotaCodeToValueMap[_internetGatewayQuotaCode]; found {
-			err := awsClient.VerifyInternetGatewayQuota(internetGatewayQuota, requiredInternetGateways)
-			if err != nil {
-				return err
+	if !clusterAlreadyExists {
+		// check internet GW quota
+		if requiredInternetGateways > 0 {
+			if internetGatewayQuota, found := quotaCodeToValueMap[_internetGatewayQuotaCode]; found {
+				err := awsClient.VerifyInternetGatewayQuota(internetGatewayQuota, requiredInternetGateways)
+				if err != nil {
+					return err
+				}
+			} else {
+				skippedValidations = append(skippedValidations, fmt.Sprintf("skipping internet gateway quota verification: unable to find internet gateway quota (%s)", _internetGatewayQuotaCode))
 			}
-		} else {
-			skippedValidations = append(skippedValidations, fmt.Sprintf("skipping internet gateway quota verification: unable to find internet gateway quota (%s)", _internetGatewayQuotaCode))
 		}
-	}
 
-	if natGatewayRequired {
-		if natGatewayQuota, found := quotaCodeToValueMap[_natGatewayQuotaCode]; found {
-			err := awsClient.VerifyNATGatewayQuota(natGatewayQuota, availabilityZones, highlyAvailableNATGateway)
-			if err != nil {
-				return err
+		// check nat GW quota
+		if natGatewayRequired {
+			if natGatewayQuota, found := quotaCodeToValueMap[_natGatewayQuotaCode]; found {
+				err := awsClient.VerifyNATGatewayQuota(natGatewayQuota, availabilityZones, highlyAvailableNATGateway)
+				if err != nil {
+					return err
+				}
+			} else {
+				skippedValidations = append(skippedValidations, fmt.Sprintf("skipping nat gateway quota verification: unable to find nat gateway quota (%s)\n", _natGatewayQuotaCode))
 			}
-		} else {
-			skippedValidations = append(skippedValidations, fmt.Sprintf("skipping nat gateway quota verification: unable to find nat gateway quota (%s)\n", _natGatewayQuotaCode))
 		}
-	}
 
-	// check EIP quota
-	if natGatewayRequired {
-		if eipQuota, found := quotaCodeToValueMap[_elasticIPsQuotaCode]; found {
-			err := awsClient.VerifyEIPQuota(eipQuota, availabilityZones, highlyAvailableNATGateway)
-			if err != nil {
-				return err
+		// check EIP quota
+		if natGatewayRequired {
+			if eipQuota, found := quotaCodeToValueMap[_elasticIPsQuotaCode]; found {
+				err := awsClient.VerifyEIPQuota(eipQuota, availabilityZones, highlyAvailableNATGateway)
+				if err != nil {
+					return err
+				}
+			} else {
+				skippedValidations = append(skippedValidations, fmt.Sprintf("skipping elastic ip quota verification: unable to find elastic ip quota (%s)\n", _elasticIPsQuotaCode))
 			}
-		} else {
-			skippedValidations = append(skippedValidations, fmt.Sprintf("skipping elastic ip quota verification: unable to find elastic ip quota (%s)\n", _elasticIPsQuotaCode))
 		}
-	}
 
-	if requiredVPCs > 0 {
-		if vpcQuota, found := quotaCodeToValueMap[_vpcQuotaCode]; found {
-			err := awsClient.VerifyVPCQuota(vpcQuota, requiredVPCs)
-			if err != nil {
-				return err
+		// check required VPC quota
+		if requiredVPCs > 0 {
+			if vpcQuota, found := quotaCodeToValueMap[_vpcQuotaCode]; found {
+				err := awsClient.VerifyVPCQuota(vpcQuota, requiredVPCs)
+				if err != nil {
+					return err
+				}
+			} else {
+				skippedValidations = append(skippedValidations, fmt.Sprintf("skipping vpc quota verification: unable to find vpc quota (%s)\n", _vpcQuotaCode))
 			}
-		} else {
-			skippedValidations = append(skippedValidations, fmt.Sprintf("skipping vpc quota verification: unable to find vpc quota (%s)\n", _vpcQuotaCode))
 		}
 	}
 
@@ -122,7 +134,7 @@ func VerifyNetworkQuotas(
 	}
 
 	if securityGroupsQuota, found := quotaCodeToValueMap[_securityGroupsQuotaCode]; found {
-		err := awsClient.VerifySecurityGroupQuota(securityGroupsQuota, numNodeGroups)
+		err := awsClient.VerifySecurityGroupQuota(securityGroupsQuota, netAdditionOfNodeGroups, clusterAlreadyExists)
 		if err != nil {
 			return err
 		}
@@ -132,3 +144,13 @@ func VerifyNetworkQuotas(
 
 	return nil
 }
+
+func VerifyNetworkQuotasOnConfigure(
+	awsClient *aws.Client,
+	availabilityZones strset.Set,
+	numNodeGroups int,
+	netAdditionOfNodeGroups int,
+	longestCIDRWhiteList int) error {
+
+	return VerifyNetworkQuotas(awsClient, 0, false, false, 0, availabilityZones, numNodeGroups, netAdditionOfNodeGroups, longestCIDRWhiteList, true)
+}
diff --git a/pkg/types/clusterstate/clusterstate.go b/pkg/types/clusterstate/clusterstate.go
index 103f1821c9..60e859d52c 100644
--- a/pkg/types/clusterstate/clusterstate.go
+++ b/pkg/types/clusterstate/clusterstate.go
@@ -18,12 +18,13 @@ package clusterstate
 
 import (
 	"fmt"
-	"time"
+	"strings"
 
 	"github.com/aws/aws-sdk-go/service/cloudformation"
 	"github.com/cortexlabs/cortex/pkg/lib/aws"
 	"github.com/cortexlabs/cortex/pkg/lib/errors"
 	"github.com/cortexlabs/cortex/pkg/lib/sets/strset"
+	"github.com/cortexlabs/cortex/pkg/lib/slices"
 	"github.com/cortexlabs/cortex/pkg/lib/table"
 	"github.com/cortexlabs/cortex/pkg/types/clusterconfig"
 )
@@ -36,56 +37,42 @@ const (
 	onDemandTemplatePrefix = "eksctl-%s-nodegroup-cx-wd"
 )
 
-type ClusterState struct {
-	StatusMap    map[string]string // cloudformation stackname to cloudformation stackstatus
-	ControlPlane string
-	NodeGroups   []string
-	Status       Status
+type ClusterStacks struct {
+	clusterName       string
+	region            string
+	ControlPlaneStack *cloudformation.StackSummary
+	OperatorStack     *cloudformation.StackSummary
+	NodeGroupsStacks  []*cloudformation.StackSummary
 }
 
-func is(status string, allowedStatus string, allowedStatuses ...string) bool {
-	statusSet := strset.New(allowedStatuses...)
-	statusSet.Add(allowedStatus)
-
-	return statusSet.Has(status)
-}
-
-func any(statuses []string, allowedStatus string, allowedStatuses ...string) bool {
-	statusSet := strset.New(allowedStatuses...)
-	statusSet.Add(allowedStatus)
-	for _, stackStatus := range statuses {
-		if statusSet.Has(stackStatus) {
-			return true
-		}
+func (cs ClusterStacks) TableString() string {
+	numStacks := len(cs.NodeGroupsStacks)
+	if cs.ControlPlaneStack != nil {
+		numStacks++
+	}
+	if cs.OperatorStack != nil {
+		numStacks++
 	}
 
-	return false
-}
+	idx := 0
+	rows := make([][]interface{}, numStacks)
 
-func all(statuses []string, allowedStatus string, allowedStatuses ...string) bool {
-	statusSet := strset.New(allowedStatuses...)
-	statusSet.Add(allowedStatus)
-	for _, stackStatus := range statuses {
-		if !statusSet.Has(stackStatus) {
-			return false
+	if cs.ControlPlaneStack != nil {
+		rows[idx] = []interface{}{
+			*cs.ControlPlaneStack.StackName, *cs.ControlPlaneStack.StackStatus,
 		}
+		idx++
 	}
-
-	return true
-}
-
-func (cs ClusterState) TableString() string {
-	rows := make([][]interface{}, len(cs.StatusMap))
-	rows[0] = []interface{}{
-		cs.ControlPlane, cs.StatusMap[cs.ControlPlane],
+	if cs.OperatorStack != nil {
+		rows[idx] = []interface{}{
+			*cs.OperatorStack.StackName, *cs.OperatorStack.StackStatus,
+		}
+		idx++
 	}
 
-	idx := 1
-	for _, nodeGroupName := range cs.NodeGroups {
-		if status, ok := cs.StatusMap[nodeGroupName]; ok {
-			rows[idx] = []interface{}{nodeGroupName, status}
-			idx++
-		}
+	for _, stack := range cs.NodeGroupsStacks {
+		rows[idx] = []interface{}{*stack.StackName, *stack.StackStatus}
+		idx++
 	}
 
 	t := table.Table{
@@ -99,132 +86,73 @@ func (cs ClusterState) TableString() string {
 		},
 		Rows: rows,
 	}
-	var items table.KeyValuePairs
-	items.Add(cs.ControlPlane, cs.StatusMap[cs.ControlPlane])
-
 	return t.MustFormat()
 }
 
-func getStatus(statusMap map[string]string, controlPlane string, clusterName string, region string) (Status, error) {
-	// the order matters
-	allStatuses := []string{}
-	controlPlaneStatus := statusMap[controlPlane]
-	nodeGroupStatuses := []string{}
-
-	for stackName, status := range statusMap {
-		allStatuses = append(allStatuses, status)
-		if stackName != controlPlane {
-			nodeGroupStatuses = append(nodeGroupStatuses, status)
-		}
-	}
-
-	if any(allStatuses, string(StatusCreateFailedTimedOut)) {
-		return StatusNotFound, ErrorUnexpectedCloudFormationStatus(clusterName, region, statusMap)
-	}
-
-	if len(nodeGroupStatuses) == 0 && controlPlaneStatus == string(StatusNotFound) {
-		return StatusNotFound, nil
-	}
-
-	// controlplane stack may be created while nodegroup stacks aren't listed in cloudformation stacks during cluster spin up
-	if len(nodeGroupStatuses) == 0 && is(controlPlaneStatus, cloudformation.StackStatusCreateComplete, cloudformation.StackStatusCreateInProgress) {
-		return StatusCreateInProgress, nil
-	}
-
-	if any(allStatuses, cloudformation.StackStatusCreateFailed) {
-		return StatusCreateFailed, nil
-	}
-
-	if any(allStatuses, cloudformation.StackStatusDeleteFailed) {
-		return StatusDeleteFailed, nil
-	}
-
-	if any(allStatuses, cloudformation.StackStatusDeleteInProgress) {
-		return StatusDeleteInProgress, nil
-	}
-
-	if all(allStatuses, cloudformation.StackStatusCreateComplete) {
-		return StatusCreateComplete, nil
-	}
-
-	if all(allStatuses, cloudformation.StackStatusUpdateComplete) {
-		return StatusUpdateComplete, nil
-	}
-
-	if all(allStatuses, cloudformation.StackStatusUpdateRollbackComplete) {
-		return StatusUpdateRollbackComplete, nil
-	}
-
-	if all(allStatuses, cloudformation.StackStatusDeleteComplete) {
-		return StatusDeleteComplete, nil
-	}
-
-	// nodegroup stacks are deleted first while control plane stack is still in create complete state
-	if controlPlaneStatus == cloudformation.StackStatusCreateComplete &&
-		all(nodeGroupStatuses, cloudformation.StackStatusDeleteInProgress, cloudformation.StackStatusDeleteComplete) {
-		return StatusDeleteInProgress, nil
-	}
-
-	// controlplane stack may be in complete state while nodegroup stacks are still in creating or one nodegroup finishes before the other
-	if controlPlaneStatus == cloudformation.StackStatusCreateComplete &&
-		all(nodeGroupStatuses, cloudformation.StackStatusCreateInProgress, cloudformation.StackStatusCreateComplete) {
-		return StatusCreateInProgress, nil
-	}
-
-	if controlPlaneStatus == cloudformation.StackStatusCreateComplete &&
-		all(nodeGroupStatuses, cloudformation.StackStatusCreateComplete, cloudformation.StackStatusUpdateComplete, cloudformation.StackStatusUpdateRollbackComplete) {
-		return StatusUpdateComplete, nil
-	}
-
-	return StatusNotFound, ErrorUnexpectedCloudFormationStatus(clusterName, region, statusMap)
-}
-
-func GetClusterState(awsClient *aws.Client, accessConfig *clusterconfig.AccessConfig) (*ClusterState, error) {
+func GetClusterStacks(awsClient *aws.Client, accessConfig *clusterconfig.AccessConfig) (ClusterStacks, error) {
 	controlPlaneStackName := fmt.Sprintf(controlPlaneTemplate, accessConfig.ClusterName)
 	operatorStackName := fmt.Sprintf(operatorTemplate, accessConfig.ClusterName)
 	spotStackNamePrefix := fmt.Sprintf(spotTemplatePrefix, accessConfig.ClusterName)
 	onDemandStackNamePrefix := fmt.Sprintf(onDemandTemplatePrefix, accessConfig.ClusterName)
-
 	nodeGroupStackPrefixesSet := strset.New(operatorStackName, spotStackNamePrefix, onDemandStackNamePrefix)
 
 	stackSummaries, err := awsClient.ListEKSStacks(controlPlaneStackName, nodeGroupStackPrefixesSet)
 	if err != nil {
-		return nil, errors.Wrap(err, "unable to get cluster state from cloudformation")
+		return ClusterStacks{}, errors.Wrap(err, "unable to get cluster state from cloudformation")
 	}
 
-	statusMap := map[string]string{}
-	nodeGroupStackNames := []string{}
-	var controlPlaneCreationTime time.Time
-
-	for _, stackSummary := range stackSummaries {
-		statusMap[*stackSummary.StackName] = *stackSummary.StackStatus
-		if *stackSummary.StackName == controlPlaneStackName {
-			controlPlaneCreationTime = *stackSummary.CreationTime
-		} else {
-			nodeGroupStackNames = append(nodeGroupStackNames, *stackSummary.StackName)
+	var controlPlaneStack, operatorStack *cloudformation.StackSummary
+	var ngStacks []*cloudformation.StackSummary
+	for _, stack := range stackSummaries {
+		if stack == nil || stack.StackName == nil {
+			continue
+		}
+		if strings.HasPrefix(*stack.StackName, spotStackNamePrefix) || strings.HasPrefix(*stack.StackName, onDemandStackNamePrefix) {
+			ngStacks = append(ngStacks, stack)
+		} else if *stack.StackName == controlPlaneStackName {
+			controlPlaneStack = stack
+		} else if *stack.StackName == operatorStackName {
+			operatorStack = stack
 		}
 	}
 
-	if _, ok := statusMap[controlPlaneStackName]; !ok {
-		statusMap[controlPlaneStackName] = string(StatusNotFound)
-	}
+	return ClusterStacks{
+		clusterName:       accessConfig.ClusterName,
+		region:            accessConfig.Region,
+		ControlPlaneStack: controlPlaneStack,
+		OperatorStack:     operatorStack,
+		NodeGroupsStacks:  ngStacks,
+	}, nil
+}
 
-	// add a timeout for situations where the control plane is listed in the cloudformation stacks but not the nodegroup stacks
-	if !is(statusMap[controlPlaneStackName], string(StatusNotFound), cloudformation.StackStatusDeleteComplete) && len(nodeGroupStackNames) == 0 && time.Now().After(controlPlaneCreationTime.Add(30*time.Minute)) {
-		statusMap[operatorStackName] = string(StatusCreateFailedTimedOut)
-	}
+func GetClusterState(stacks ClusterStacks) State {
+	controlPlaneStackName := fmt.Sprintf(controlPlaneTemplate, stacks.clusterName)
 
-	status, err := getStatus(statusMap, controlPlaneStackName, accessConfig.ClusterName, accessConfig.Region)
-	if err != nil {
-		return nil, err
+	if stacks.ControlPlaneStack == nil || stacks.ControlPlaneStack.StackName == nil {
+		return StateClusterDoesntExist
 	}
+	if *stacks.ControlPlaneStack.StackName == controlPlaneStackName {
+		controlPlaneStatus := *stacks.ControlPlaneStack.StackStatus
 
-	return &ClusterState{
-		ControlPlane: controlPlaneStackName,
-		NodeGroups:   nodeGroupStackNames,
-		StatusMap:    statusMap,
-		Status:       status,
-	}, nil
+		if slices.HasString([]string{
+			cloudformation.StackStatusDeleteComplete,
+			cloudformation.StackStatusDeleteInProgress,
+		}, controlPlaneStatus) {
+			return StateClusterDoesntExist
+		}
+
+		if slices.HasString([]string{
+			cloudformation.StackStatusCreateComplete,
+			cloudformation.StackStatusUpdateComplete,
+			cloudformation.StackStatusRollbackComplete,
+			cloudformation.StackStatusUpdateRollbackComplete,
+		}, controlPlaneStatus) {
+			return StateClusterExists
+		}
+
+		return StateClusterInUnexpectedState
+	}
+	return StateClusterDoesntExist
 }
 
 func CloudFormationURL(clusterName string, region string) string {
diff --git a/pkg/types/clusterstate/errors.go b/pkg/types/clusterstate/errors.go
index 5a3f1e94ba..012f14566e 100644
--- a/pkg/types/clusterstate/errors.go
+++ b/pkg/types/clusterstate/errors.go
@@ -23,85 +23,46 @@ import (
 )
 
 const (
-	ErrClusterDoesNotExist            = "clusterstatus.cluster_does_not_exist"
-	ErrClusterUpInProgress            = "clusterstatus.cluster_up_in_progress"
-	ErrClusterCreateFailed            = "clusterstatus.cluster_create_failed"
-	ErrClusterCreateFailedTimeout     = "clusterstatus.cluster_create_failed_timeout"
-	ErrClusterAlreadyCreated          = "clusterstatus.cluster_already_created"
-	ErrClusterAlreadyUpdated          = "clusterstatus.cluster_already_updated"
-	ErrClusterDownInProgress          = "clusterstatus.cluster_down_in_progress"
-	ErrClusterAlreadyDeleted          = "clusterstatus.cluster_already_deleted"
-	ErrClusterDeleteFailed            = "clusterstatus.cluster_delete_failed"
-	ErrUnexpectedCloudFormationStatus = "clusterstatus.unexpected_cloud_formation_status"
+	ErrClusterDoesNotExist    = "clusterstate.cluster_does_not_exist"
+	ErrClusterAlreadyExists   = "clusterstate.cluster_already_exists"
+	ErrUnexpectedClusterState = "clusterstate.unexpected_cluster_state"
 )
 
-func ErrorClusterDoesNotExist(clusterName string, region string) error {
+func ErrorClusterDoesNotExist(stacks ClusterStacks) error {
 	return errors.WithStack(&errors.Error{
 		Kind:    ErrClusterDoesNotExist,
-		Message: fmt.Sprintf("there is no cluster named \"%s\" in %s", clusterName, region),
+		Message: fmt.Sprintf("there is no cluster named \"%s\" in %s", stacks.clusterName, stacks.region),
 	})
 }
 
-func ErrorClusterUpInProgress(clusterName string, region string) error {
+func ErrorClusterAlreadyExists(stacks ClusterStacks) error {
 	return errors.WithStack(&errors.Error{
-		Kind:    ErrClusterUpInProgress,
-		Message: fmt.Sprintf("creation of cluster \"%s\" in %s is currently in progress", clusterName, region),
+		Kind:    ErrClusterAlreadyExists,
+		Message: fmt.Sprintf("a cluster named \"%s\" already exists in %s", stacks.clusterName, stacks.region),
 	})
 }
 
-func ErrorClusterCreateFailed(clusterName string, region string) error {
+func ErrorUnexpectedClusterState(stacks ClusterStacks) error {
+	msg := fmt.Sprintf("cluster named \"%s\" in %s is in an unexpected state; if your CloudFormation stacks are updating, please wait for them to complete. Otherwise, run `cortex cluster down` to delete the cluster\n\n", stacks.clusterName, stacks.region)
+	msg += fmt.Sprintf(stacks.TableString())
 	return errors.WithStack(&errors.Error{
-		Kind:    ErrClusterCreateFailed,
-		Message: fmt.Sprintf("cluster \"%s\" in %s could not be created; please view error information and delete the CloudFormation stacks directly from your AWS console: %s", clusterName, region, CloudFormationURL(clusterName, region)),
+		Kind:     ErrUnexpectedClusterState,
+		Message:  msg,
+		Metadata: stacks,
 	})
 }
 
-func ErrorClusterCreateFailedTimeout(clusterName string, region string) error {
-	return errors.WithStack(&errors.Error{
-		Kind:    ErrClusterCreateFailedTimeout,
-		Message: fmt.Sprintf("cluster \"%s\" in %s could not be created; please view error information and delete the CloudFormation stacks directly from your AWS console: %s", clusterName, region, CloudFormationURL(clusterName, region)),
-	})
-}
-
-func ErrorClusterAlreadyCreated(clusterName string, region string) error {
-	return errors.WithStack(&errors.Error{
-		Kind:    ErrClusterAlreadyCreated,
-		Message: fmt.Sprintf("a cluster named \"%s\" already exists in %s", clusterName, region),
-	})
-}
-
-func ErrorClusterAlreadyUpdated(clusterName string, region string) error {
-	return errors.WithStack(&errors.Error{
-		Kind:    ErrClusterAlreadyUpdated,
-		Message: fmt.Sprintf("a cluster named \"%s\" already created and updated in %s", clusterName, region),
-	})
-}
-
-func ErrorClusterDownInProgress(clusterName string, region string) error {
-	return errors.WithStack(&errors.Error{
-		Kind:    ErrClusterDownInProgress,
-		Message: fmt.Sprintf("deletion of cluster \"%s\" in %s is currently in progress", clusterName, region),
-	})
-}
-
-func ErrorClusterAlreadyDeleted(clusterName string, region string) error {
-	return errors.WithStack(&errors.Error{
-		Kind:    ErrClusterAlreadyDeleted,
-		Message: fmt.Sprintf("cluster \"%s\" in %s has already been deleted (or does not exist)", clusterName, region),
-	})
-}
-
-func ErrorClusterDeleteFailed(clusterName string, region string) error {
-	return errors.WithStack(&errors.Error{
-		Kind:    ErrClusterDeleteFailed,
-		Message: fmt.Sprintf("deletion of cluster \"%s\" in %s failed; please run `cortex cluster down` to delete the cluster, or if that fails, delete the CloudFormation stacks directly from your AWS console: %s", clusterName, region, CloudFormationURL(clusterName, region)),
-	})
-}
-
-func ErrorUnexpectedCloudFormationStatus(clusterName string, region string, metadata interface{}) error {
-	return errors.WithStack(&errors.Error{
-		Kind:     ErrUnexpectedCloudFormationStatus,
-		Message:  fmt.Sprintf("cluster named \"%s\" in %s is in an unexpected state; if your CloudFormation stacks are updating, please wait for them to complete. Otherwise, run `cortex cluster down` to delete the cluster, or if that fails, delete the CloudFormation stacks directly from your AWS console: %s", clusterName, region, CloudFormationURL(clusterName, region)),
-		Metadata: metadata,
-	})
+func AssertClusterState(stacks ClusterStacks, currentState, allowedState State) error {
+	if currentState == allowedState {
+		return nil
+	}
+	switch currentState {
+	case StateClusterDoesntExist:
+		return ErrorClusterDoesNotExist(stacks)
+	case StateClusterExists:
+		return ErrorClusterAlreadyExists(stacks)
+	case StateClusterInUnexpectedState:
+		return ErrorUnexpectedClusterState(stacks)
+	}
+	return nil
 }
diff --git a/pkg/types/clusterstate/state.go b/pkg/types/clusterstate/state.go
new file mode 100644
index 0000000000..f20984b364
--- /dev/null
+++ b/pkg/types/clusterstate/state.go
@@ -0,0 +1,25 @@
+/*
+Copyright 2021 Cortex Labs, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package clusterstate
+
+type State string
+
+const (
+	StateClusterExists            State = "cluster_exists"
+	StateClusterDoesntExist       State = "cluster_doesnt_exist"
+	StateClusterInUnexpectedState State = "cluster_in_unexpected_state"
+)
diff --git a/pkg/types/clusterstate/status.go b/pkg/types/clusterstate/status.go
deleted file mode 100644
index f3b53aaae6..0000000000
--- a/pkg/types/clusterstate/status.go
+++ /dev/null
@@ -1,65 +0,0 @@
-/*
-Copyright 2021 Cortex Labs, Inc.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package clusterstate
-
-type Status string
-
-const (
-	StatusNotFound               Status = "not_found"
-	StatusCreateInProgress       Status = "create_in_progress"
-	StatusCreateFailed           Status = "create_failed"
-	StatusCreateFailedTimedOut   Status = "create_failed_timed_out"
-	StatusCreateComplete         Status = "create_complete"
-	StatusUpdateComplete         Status = "update_complete"
-	StatusUpdateRollbackComplete Status = "update_rollback_complete"
-	StatusDeleteInProgress       Status = "delete_in_progress"
-	StatusDeleteComplete         Status = "delete_complete"
-	StatusDeleteFailed           Status = "delete_failed"
-)
-
-func AssertClusterStatus(clusterName string, region string, status Status, allowedStatuses ...Status) error {
-	for _, allowedStatus := range allowedStatuses {
-		if status == allowedStatus {
-			return nil
-		}
-	}
-
-	switch status {
-	case StatusNotFound:
-		return ErrorClusterDoesNotExist(clusterName, region)
-	case StatusCreateInProgress:
-		return ErrorClusterUpInProgress(clusterName, region)
-	case StatusCreateFailed:
-		return ErrorClusterCreateFailed(clusterName, region)
-	case StatusCreateFailedTimedOut:
-		return ErrorClusterCreateFailedTimeout(clusterName, region)
-	case StatusCreateComplete:
-		return ErrorClusterAlreadyCreated(clusterName, region)
-	case StatusUpdateComplete:
-		return ErrorClusterAlreadyUpdated(clusterName, region)
-	case StatusUpdateRollbackComplete:
-		return ErrorClusterAlreadyUpdated(clusterName, region)
-	case StatusDeleteInProgress:
-		return ErrorClusterDownInProgress(clusterName, region)
-	case StatusDeleteComplete:
-		return ErrorClusterAlreadyDeleted(clusterName, region)
-	case StatusDeleteFailed:
-		return ErrorClusterDeleteFailed(clusterName, region)
-	default:
-		return ErrorUnexpectedCloudFormationStatus(clusterName, region, status)
-	}
-}
diff --git a/pkg/workloads/k8s.go b/pkg/workloads/k8s.go
index 172221efe5..590eac0e03 100644
--- a/pkg/workloads/k8s.go
+++ b/pkg/workloads/k8s.go
@@ -465,7 +465,6 @@ func GenerateResourceTolerations() []kcore.Toleration {
 }
 
 func GenerateNodeAffinities(apiNodeGroups []string) *kcore.Affinity {
-	// node groups are ordered according to how the cluster config node groups are ordered
 	var nodeGroups []*clusterconfig.NodeGroup
 	for _, clusterNodeGroup := range config.ClusterConfig.NodeGroups {
 		for _, apiNodeGroupName := range apiNodeGroups {
@@ -475,16 +474,14 @@ func GenerateNodeAffinities(apiNodeGroups []string) *kcore.Affinity {
 		}
 	}
 
-	numNodeGroups := len(apiNodeGroups)
 	if apiNodeGroups == nil {
 		nodeGroups = config.ClusterConfig.NodeGroups
-		numNodeGroups = len(config.ClusterConfig.NodeGroups)
 	}
 
 	var requiredNodeGroups []string
 	var preferredAffinities []kcore.PreferredSchedulingTerm
 
-	for idx, nodeGroup := range nodeGroups {
+	for _, nodeGroup := range nodeGroups {
 		var nodeGroupPrefix string
 		if nodeGroup.Spot {
 			nodeGroupPrefix = "cx-ws-"
@@ -493,7 +490,7 @@ func GenerateNodeAffinities(apiNodeGroups []string) *kcore.Affinity {
 		}
 
 		preferredAffinities = append(preferredAffinities, kcore.PreferredSchedulingTerm{
-			Weight: int32(100 * (1 - float64(idx)/float64(numNodeGroups))),
+			Weight: int32(nodeGroup.Priority),
 			Preference: kcore.NodeSelectorTerm{
 				MatchExpressions: []kcore.NodeSelectorRequirement{
 					{


Test patch:
diff --git a/pkg/lib/structs/deepcopy_test.go b/pkg/lib/structs/deepcopy_test.go
new file mode 100644
index 0000000000..952f99c636
--- /dev/null
+++ b/pkg/lib/structs/deepcopy_test.go
@@ -0,0 +1,55 @@
+/*
+Copyright 2021 Cortex Labs, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package structs
+
+import (
+	"testing"
+
+	"github.com/cortexlabs/cortex/pkg/lib/pointer"
+	"github.com/stretchr/testify/require"
+)
+
+type SampleStruct struct {
+	AField string
+	BField *string
+	CField *bool
+}
+
+func TestDeepCopy(t *testing.T) {
+	t.Parallel()
+
+	var a SampleStruct
+	b := SampleStruct{
+		AField: "fox",
+		BField: pointer.String("bull"),
+	}
+
+	err := DeepCopy(&a, &b)
+	require.NoError(t, err)
+
+	require.EqualValues(t, b.AField, "fox")
+	require.EqualValues(t, b.AField, a.AField)
+
+	require.True(t, a.BField != nil)
+	require.True(t, b.BField != nil)
+	require.True(t, a.BField != b.BField)
+	require.True(t, a.CField == nil)
+	require.True(t, b.CField == nil)
+
+	require.EqualValues(t, *a.BField, "bull")
+	require.EqualValues(t, *a.BField, *b.BField)
+}