Allow updating the cluster one instance group at a time

cmd/kops/update_cluster.go

@@ -27,13 +27,15 @@ import (
 
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
+	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/client-go/tools/clientcmd"
 	"k8s.io/klog/v2"
 	"k8s.io/kops/cmd/kops/util"
 	"k8s.io/kops/pkg/apis/kops"
 	"k8s.io/kops/pkg/assets"
 	"k8s.io/kops/pkg/commands/commandutils"
 	"k8s.io/kops/pkg/kubeconfig"
+	"k8s.io/kops/pkg/predicates"
 	"k8s.io/kops/upup/pkg/fi"
 	"k8s.io/kops/upup/pkg/fi/cloudup"
 	"k8s.io/kops/upup/pkg/fi/utils"
@@ -75,6 +77,14 @@ type UpdateClusterOptions struct {
 	user          string
 	internal      bool
 
+	// InstanceGroups is the list of instance groups to update;
+	// if not specified, all instance groups will be updated
+	InstanceGroups []string
+
+	// InstanceGroupRoles is the list of roles we should update
+	// if not specified, all instance groups will be updated
+	InstanceGroupRoles []string
+
 	Phase string
 
 	// LifecycleOverrides is a slice of taskName=lifecycle name values.  This slice is used
@@ -106,6 +116,11 @@ func NewCmdUpdateCluster(f *util.Factory, out io.Writer) *cobra.Command {
 	options := &UpdateClusterOptions{}
 	options.InitDefaults()
 
+	allRoles := make([]string, 0, len(kops.AllInstanceGroupRoles))
+	for _, r := range kops.AllInstanceGroupRoles {
+		allRoles = append(allRoles, r.ToLowerString())
+	}
+
 	cmd := &cobra.Command{
 		Use:               "cluster [CLUSTER]",
 		Short:             updateClusterShort,
@@ -132,6 +147,12 @@ func NewCmdUpdateCluster(f *util.Factory, out io.Writer) *cobra.Command {
 	cmd.RegisterFlagCompletionFunc("user", completeKubecfgUser)
 	cmd.Flags().BoolVar(&options.internal, "internal", options.internal, "Use the cluster's internal DNS name. Implies --create-kube-config")
 	cmd.Flags().BoolVar(&options.AllowKopsDowngrade, "allow-kops-downgrade", options.AllowKopsDowngrade, "Allow an older version of kOps to update the cluster than last used")
+	cmd.Flags().StringSliceVar(&options.InstanceGroups, "instance-group", options.InstanceGroups, "Instance groups to update (defaults to all if not specified)")
+	cmd.RegisterFlagCompletionFunc("instance-group", completeInstanceGroup(f, &options.InstanceGroups, &options.InstanceGroupRoles))
+	cmd.Flags().StringSliceVar(&options.InstanceGroupRoles, "instance-group-roles", options.InstanceGroupRoles, "Instance group roles to update ("+strings.Join(allRoles, ",")+")")
+	cmd.RegisterFlagCompletionFunc("instance-group-roles", func(cmd *cobra.Command, args []string, toComplete string) ([]string, cobra.ShellCompDirective) {
+		return sets.NewString(allRoles...).Delete(options.InstanceGroupRoles...).List(), cobra.ShellCompDirectiveNoFileComp
+	})
 	cmd.Flags().StringVar(&options.Phase, "phase", options.Phase, "Subset of tasks to run: "+strings.Join(cloudup.Phases.List(), ", "))
 	cmd.RegisterFlagCompletionFunc("phase", func(cmd *cobra.Command, args []string, toComplete string) ([]string, cobra.ShellCompDirective) {
 		return cloudup.Phases.List(), cobra.ShellCompDirectiveNoFileComp
@@ -285,24 +306,32 @@ func RunUpdateCluster(ctx context.Context, f *util.Factory, out io.Writer, c *Up
 		lifecycleOverrideMap[taskName] = lifecycleOverride
 	}
 
+	var instanceGroupFilters []predicates.Predicate[*kops.InstanceGroup]
+	if len(c.InstanceGroups) != 0 {
+		instanceGroupFilters = append(instanceGroupFilters, matchInstanceGroupNames(c.InstanceGroups))
+	} else if len(c.InstanceGroupRoles) != 0 {
+		instanceGroupFilters = append(instanceGroupFilters, matchInstanceGroupRoles(c.InstanceGroupRoles))
+	}
+
 	cloud, err := cloudup.BuildCloud(cluster)
 	if err != nil {
 		return nil, err
 	}
 
 	applyCmd := &cloudup.ApplyClusterCmd{
-		Cloud:              cloud,
-		Clientset:          clientset,
-		Cluster:            cluster,
-		DryRun:             isDryrun,
-		AllowKopsDowngrade: c.AllowKopsDowngrade,
-		RunTasksOptions:    &c.RunTasksOptions,
-		OutDir:             c.OutDir,
-		Phase:              phase,
-		TargetName:         targetName,
-		LifecycleOverrides: lifecycleOverrideMap,
-		GetAssets:          c.GetAssets,
-		DeletionProcessing: deletionProcessing,
+		Cloud:               cloud,
+		Clientset:           clientset,
+		Cluster:             cluster,
+		DryRun:              isDryrun,
+		AllowKopsDowngrade:  c.AllowKopsDowngrade,
+		RunTasksOptions:     &c.RunTasksOptions,
+		OutDir:              c.OutDir,
+		InstanceGroupFilter: predicates.AllOf(instanceGroupFilters...),
+		Phase:               phase,
+		TargetName:          targetName,
+		LifecycleOverrides:  lifecycleOverrideMap,
+		GetAssets:           c.GetAssets,
+		DeletionProcessing:  deletionProcessing,
 	}
 
 	applyResults, err := applyCmd.Run(ctx)
@@ -518,3 +547,31 @@ func completeLifecycleOverrides(cmd *cobra.Command, args []string, toComplete st
 	}
 	return completions, cobra.ShellCompDirectiveNoFileComp
 }
+
+// matchInstanceGroupNames returns a predicate that matches instance groups by name
+func matchInstanceGroupNames(names []string) predicates.Predicate[*kops.InstanceGroup] {
+	return func(ig *kops.InstanceGroup) bool {
+		for _, name := range names {
+			if ig.ObjectMeta.Name == name {
+				return true
+			}
+		}
+		return false
+	}
+}
+
+// matchInstanceGroupRoles returns a predicate that matches instance groups by role
+func matchInstanceGroupRoles(roles []string) predicates.Predicate[*kops.InstanceGroup] {
+	return func(ig *kops.InstanceGroup) bool {
+		for _, role := range roles {
+			instanceGroupRole, ok := kops.ParseInstanceGroupRole(role, true)
+			if !ok {
+				continue
+			}
+			if ig.Spec.Role == instanceGroupRole {
+				return true
+			}
+		}
+		return false
+	}
+}

pkg/model/awsmodel/iam.go

@@ -65,7 +65,7 @@ func (b *IAMModelBuilder) Build(c *fi.CloudupModelBuilderContext) error {
 
 	// Collect Instance Profile ARNs and their associated Instance Group roles
 	sharedProfileARNsToIGRole := make(map[string]kops.InstanceGroupRole)
-	for _, ig := range b.InstanceGroups {
+	for _, ig := range b.AllInstanceGroups {
 		if ig.Spec.IAM != nil && ig.Spec.IAM.Profile != nil {
 			specProfile := fi.ValueOf(ig.Spec.IAM.Profile)
 			if matchingRole, ok := sharedProfileARNsToIGRole[specProfile]; ok {

pkg/model/context.go

@@ -47,9 +47,18 @@ const (
 // KopsModelContext is the kops model
 type KopsModelContext struct {
 	iam.IAMModelContext
+
+	// AllInstanceGroups is the list of instance groups in the cluster.
+	// Generally most tasks should use InstanceGroups instead,
+	// but we sometimes need the full list for example when configuring cluster-wide IAM.
+	AllInstanceGroups []*kops.InstanceGroup
+
+	// InstanceGroups is the list of instance groups in the cluster that are being processed.
+	// This is a filtered list of AllInstanceGroups.
 	InstanceGroups []*kops.InstanceGroup
-	Region         string
-	SSHPublicKeys  [][]byte
+
+	Region        string
+	SSHPublicKeys [][]byte
 
 	// AdditionalObjects holds cluster-asssociated configuration objects, other than the Cluster and InstanceGroups.
 	AdditionalObjects kubemanifest.ObjectList
@@ -92,7 +101,7 @@ func (b *KopsModelContext) GatherSubnets(ig *kops.InstanceGroup) ([]*kops.Cluste
 
 // FindInstanceGroup returns the instance group with the matching Name (or nil if not found)
 func (b *KopsModelContext) FindInstanceGroup(name string) *kops.InstanceGroup {
-	for _, ig := range b.InstanceGroups {
+	for _, ig := range b.AllInstanceGroups {
 		if ig.ObjectMeta.Name == name {
 			return ig
 		}

pkg/predicates/function.go

@@ -0,0 +1,31 @@
+package predicates
+
+// Predicate is a predicate function for a type.
+type Predicate[T any] func(T) bool
+
+// AllOf returns a predicate that is true if all of the given predicates are true.
+func AllOf[T any](predicates ...Predicate[T]) Predicate[T] {
+	return func(t T) bool {
+		for _, predicate := range predicates {
+			if !predicate(t) {
+				return false
+			}
+		}
+		return true
+	}
+}
+
+// Filter returns a slice of elements that match the predicate.
+func Filter[T any](slice []T, predicate Predicate[T]) []T {
+	if predicate == nil {
+		return slice
+	}
+
+	var result []T
+	for _, element := range slice {
+		if predicate(element) {
+			result = append(result, element)
+		}
+	}
+	return result
+}

upup/pkg/fi/cloudup/apply_cluster.go

@@ -50,6 +50,7 @@ import (
 	"k8s.io/kops/pkg/model/openstackmodel"
 	"k8s.io/kops/pkg/model/scalewaymodel"
 	"k8s.io/kops/pkg/nodemodel"
+	"k8s.io/kops/pkg/predicates"
 	"k8s.io/kops/pkg/templates"
 	"k8s.io/kops/upup/models"
 	"k8s.io/kops/upup/pkg/fi"
@@ -133,6 +134,9 @@ type ApplyClusterCmd struct {
 
 	// DeletionProcessing controls whether we process deletions.
 	DeletionProcessing fi.DeletionProcessingMode
+
+	// InstanceGroupFilter is a predicate that restricts which instance groups we will update.
+	InstanceGroupFilter predicates.Predicate[*kops.InstanceGroup]
 }
 
 // ApplyResults holds information about an ApplyClusterCmd operation.
@@ -401,9 +405,13 @@ func (c *ApplyClusterCmd) Run(ctx context.Context) (*ApplyResults, error) {
 		}
 	}
 
+	allInstanceGroups := c.InstanceGroups
+	filteredInstanceGroups := predicates.Filter(allInstanceGroups, c.InstanceGroupFilter)
+
 	modelContext := &model.KopsModelContext{
 		IAMModelContext:   iam.IAMModelContext{Cluster: cluster},
-		InstanceGroups:    c.InstanceGroups,
+		InstanceGroups:    filteredInstanceGroups,
+		AllInstanceGroups: allInstanceGroups,
 		AdditionalObjects: c.AdditionalObjects,
 	}
 

upup/pkg/fi/cloudup/template_functions.go

@@ -708,7 +708,7 @@ func (tf *TemplateFunctions) KopsControllerConfig() (string, error) {
 		switch cluster.GetCloudProvider() {
 		case kops.CloudProviderAWS:
 			nodesRoles := sets.String{}
-			for _, ig := range tf.InstanceGroups {
+			for _, ig := range tf.AllInstanceGroups {
 				if ig.Spec.Role == kops.InstanceGroupRoleNode || ig.Spec.Role == kops.InstanceGroupRoleAPIServer {
 					profile, err := tf.LinkToIAMInstanceProfile(ig)
 					if err != nil {