Put versioned API of cluster into state store

cmd/kops-controller/controllers/BUILD.bazel

@@ -8,9 +8,9 @@ go_library(
     deps = [
         "//pkg/apis/kops:go_default_library",
         "//pkg/apis/kops/registry:go_default_library",
+        "//pkg/kopscodecs:go_default_library",
         "//pkg/nodeidentity:go_default_library",
         "//pkg/nodelabels:go_default_library",
-        "//upup/pkg/fi/utils:go_default_library",
         "//util/pkg/vfs:go_default_library",
         "//vendor/github.com/go-logr/logr:go_default_library",
         "//vendor/k8s.io/api/core/v1:go_default_library",

cmd/kops-controller/controllers/node_controller.go

@@ -31,9 +31,9 @@ import (
 	"k8s.io/klog"
 	"k8s.io/kops/pkg/apis/kops"
 	"k8s.io/kops/pkg/apis/kops/registry"
+	"k8s.io/kops/pkg/kopscodecs"
 	"k8s.io/kops/pkg/nodeidentity"
 	"k8s.io/kops/pkg/nodelabels"
-	"k8s.io/kops/upup/pkg/fi/utils"
 	"k8s.io/kops/util/pkg/vfs"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -243,12 +243,14 @@ func (r *NodeReconciler) loadCluster(p vfs.Path) (*kops.Cluster, error) {
 		return nil, fmt.Errorf("error loading Cluster %q: %v", p, err)
 	}
 
-	cluster := &kops.Cluster{}
-	if err := utils.YamlUnmarshal(b, cluster); err != nil {
+	o, _, err := kopscodecs.Decode(b, nil)
+	if err != nil {
 		return nil, fmt.Errorf("error parsing Cluster %q: %v", p, err)
 	}
-
-	return cluster, nil
+	if cluster, ok := o.(*kops.Cluster); ok {
+		return cluster, nil
+	}
+	return nil, fmt.Errorf("unexpected object type for Cluster %q: %T", p, o)
 }
 
 // loadInstanceGroup loads a kops.InstanceGroup object from the vfs backing store
@@ -261,10 +263,12 @@ func (r *NodeReconciler) loadNamedInstanceGroup(name string) (*kops.InstanceGrou
 		return nil, fmt.Errorf("error loading InstanceGroup %q: %v", p, err)
 	}
 
-	instanceGroup := &kops.InstanceGroup{}
-	if err := utils.YamlUnmarshal(b, instanceGroup); err != nil {
+	o, _, err := kopscodecs.Decode(b, nil)
+	if err != nil {
 		return nil, fmt.Errorf("error parsing InstanceGroup %q: %v", p, err)
 	}
-
-	return instanceGroup, nil
+	if instanceGroup, ok := o.(*kops.InstanceGroup); ok {
+		return instanceGroup, nil
+	}
+	return nil, fmt.Errorf("unexpected object type for InstanceGroup %q: %T", p, o)
 }

cmd/kops/create_cluster.go

@@ -1298,7 +1298,7 @@ func RunCreateCluster(ctx context.Context, f *util.Factory, out io.Writer, c *Cr
 		return fmt.Errorf("error writing updated configuration: %v", err)
 	}
 
-	err = registry.WriteConfigDeprecated(cluster, configBase.Join(registry.PathClusterCompleted), fullCluster)
+	err = registry.WriteConfigDeprecated(cluster, configBase.Join(registry.PathLegacyClusterCompleted), fullCluster)
 	if err != nil {
 		return fmt.Errorf("error writing completed cluster spec: %v", err)
 	}

cmd/kops/edit_cluster.go

@@ -250,7 +250,7 @@ func RunEditCluster(ctx context.Context, f *util.Factory, cmd *cobra.Command, ar
 			return preservedFile(err, file, out)
 		}
 
-		err = registry.WriteConfigDeprecated(newCluster, configBase.Join(registry.PathClusterCompleted), fullCluster)
+		err = registry.WriteConfigDeprecated(newCluster, configBase.Join(registry.PathLegacyClusterCompleted), fullCluster)
 		if err != nil {
 			return preservedFile(fmt.Errorf("error writing completed cluster spec: %v", err), file, out)
 		}

cmd/kops/get_cluster.go

@@ -31,6 +31,7 @@ import (
 	"k8s.io/kops/cmd/kops/util"
 	kopsapi "k8s.io/kops/pkg/apis/kops"
 	"k8s.io/kops/pkg/apis/kops/registry"
+	"k8s.io/kops/pkg/kopscodecs"
 	"k8s.io/kops/util/pkg/tables"
 	"k8s.io/kubectl/pkg/util/i18n"
 	"k8s.io/kubectl/pkg/util/templates"
@@ -283,12 +284,21 @@ func fullClusterSpecs(clusters []*kopsapi.Cluster) ([]*kopsapi.Cluster, error) {
 		if err != nil {
 			return nil, fmt.Errorf("error reading full cluster spec for %q: %v", cluster.ObjectMeta.Name, err)
 		}
-		fullSpec := &kopsapi.Cluster{}
-		err = registry.ReadConfigDeprecated(configBase.Join(registry.PathClusterCompleted), fullSpec)
+		configPath := configBase.Join(registry.PathClusterCompleted)
+		b, err := configPath.ReadFile()
 		if err != nil {
-			return nil, fmt.Errorf("error reading full cluster spec for %q: %v", cluster.ObjectMeta.Name, err)
+			return nil, fmt.Errorf("error loading Cluster %q: %v", configPath, err)
+		}
+
+		o, _, err := kopscodecs.Decode(b, nil)
+		if err != nil {
+			return nil, fmt.Errorf("error parsing Cluster %q: %v", configPath, err)
+		}
+		if fullSpec, ok := o.(*kopsapi.Cluster); ok {
+			fullSpecs = append(fullSpecs, fullSpec)
+		} else {
+			return nil, fmt.Errorf("unexpected object type for Cluster %q: %T", configPath, o)
 		}
-		fullSpecs = append(fullSpecs, fullSpec)
 	}
 	return fullSpecs, nil
 }

pkg/apis/kops/registry/registry.go

@@ -28,7 +28,9 @@ const (
 	// Path for the user-specified cluster spec
 	PathCluster = "config"
 	// Path for completed cluster spec in the state store
-	PathClusterCompleted = "cluster.spec"
+	PathClusterCompleted = "cluster-completed.spec"
+	// Path for unversioned completed cluster spec in the state store
+	PathLegacyClusterCompleted = "cluster.spec"
 )
 
 func ConfigBase(c *api.Cluster) (vfs.Path, error) {

pkg/apis/kops/registry/statestore.go

@@ -20,37 +20,13 @@ import (
 	"bytes"
 	"fmt"
 	"os"
-	"strings"
 
 	"k8s.io/kops/pkg/acls"
 	"k8s.io/kops/pkg/apis/kops"
 	"k8s.io/kops/upup/pkg/fi/utils"
 	"k8s.io/kops/util/pkg/vfs"
 )
 
-func ReadConfigDeprecated(configPath vfs.Path, config interface{}) error {
-	data, err := configPath.ReadFile()
-	if err != nil {
-		if os.IsNotExist(err) {
-			return err
-		}
-		return fmt.Errorf("error reading configuration file %s: %v", configPath, err)
-	}
-
-	// Yaml can't parse empty strings
-	configString := string(data)
-	configString = strings.TrimSpace(configString)
-
-	if configString != "" {
-		err = utils.YamlUnmarshal([]byte(configString), config)
-		if err != nil {
-			return fmt.Errorf("error parsing configuration: %v", err)
-		}
-	}
-
-	return nil
-}
-
 // WriteConfigDeprecated writes a config file as yaml.
 // It is deprecated because it is unversioned, but it is still used, in particular for writing the completed config.
 func WriteConfigDeprecated(cluster *kops.Cluster, configPath vfs.Path, config interface{}, writeOptions ...vfs.WriteOption) error {

pkg/bundle/BUILD.bazel

@@ -15,7 +15,6 @@ go_library(
         "//pkg/model:go_default_library",
         "//upup/pkg/fi:go_default_library",
         "//upup/pkg/fi/cloudup:go_default_library",
-        "//upup/pkg/fi/utils:go_default_library",
         "//util/pkg/vfs:go_default_library",
         "//vendor/k8s.io/klog:go_default_library",
     ],

pkg/bundle/builder.go

@@ -32,7 +32,6 @@ import (
 	"k8s.io/kops/pkg/model"
 	"k8s.io/kops/upup/pkg/fi"
 	"k8s.io/kops/upup/pkg/fi/cloudup"
-	"k8s.io/kops/upup/pkg/fi/utils"
 	"k8s.io/kops/util/pkg/vfs"
 )
 
@@ -57,7 +56,7 @@ func (b *Builder) Build(cluster *kops.Cluster, ig *kops.InstanceGroup) (*Data, e
 		return nil, err
 	}
 
-	fullCluster := &kops.Cluster{}
+	var fullCluster *kops.Cluster
 	{
 		configBase, err := b.Clientset.ConfigBaseFor(cluster)
 		if err != nil {
@@ -71,10 +70,14 @@ func (b *Builder) Build(cluster *kops.Cluster, ig *kops.InstanceGroup) (*Data, e
 			return nil, fmt.Errorf("error loading Cluster %q: %v", p, err)
 		}
 
-		err = utils.YamlUnmarshal(b, fullCluster)
+		o, _, err := kopscodecs.Decode(b, nil)
 		if err != nil {
 			return nil, fmt.Errorf("error parsing Cluster %q: %v", p, err)
 		}
+		var ok bool
+		if fullCluster, ok = o.(*kops.Cluster); !ok {
+			return nil, fmt.Errorf("unexpected object type for Cluster %q: %T", p, o)
+		}
 	}
 
 	klog.Infof("fullCluster %v", fullCluster)
@@ -87,13 +90,13 @@ func (b *Builder) Build(cluster *kops.Cluster, ig *kops.InstanceGroup) (*Data, e
 	var files []*DataFile
 
 	{
-		data, err := utils.YamlMarshal(fullCluster)
+		data, err := kopscodecs.ToVersionedYaml(fullCluster)
 		if err != nil {
 			return nil, fmt.Errorf("error marshaling configuration: %v", err)
 		}
 
 		file := &DataFile{}
-		file.Header.Name = "cluster.spec"
+		file.Header.Name = "cluster-completed.spec"
 		file.Header.Size = int64(len(data))
 		file.Header.Mode = 0644
 		file.Data = data

pkg/client/simple/api/clientset.go

@@ -68,6 +68,12 @@ func (c *RESTClientset) UpdateCluster(ctx context.Context, cluster *kops.Cluster
 	return c.KopsClient.Clusters(namespace).Update(ctx, cluster, metav1.UpdateOptions{})
 }
 
+// UpdateCompletedCluster implements the UpdateCompletedCluster method of Clientset for a kubernetes-API state store
+func (c *RESTClientset) UpdateCompletedCluster(ctx context.Context, cluster *kops.Cluster) error {
+	// Not implemented
+	return nil
+}
+
 // ConfigBaseFor implements the ConfigBaseFor method of Clientset for a kubernetes-API state store
 func (c *RESTClientset) ConfigBaseFor(cluster *kops.Cluster) (vfs.Path, error) {
 	if cluster.Spec.ConfigBase != "" {

pkg/client/simple/clientset.go

@@ -36,6 +36,9 @@ type Clientset interface {
 	// UpdateCluster updates a cluster
 	UpdateCluster(ctx context.Context, cluster *kops.Cluster, status *kops.ClusterStatus) (*kops.Cluster, error)
 
+	// UpdateCompletedCluster updates a completed cluster.
+	UpdateCompletedCluster(ctx context.Context, cluster *kops.Cluster) error
+
 	// ListClusters returns all clusters
 	ListClusters(ctx context.Context, options metav1.ListOptions) (*kops.ClusterList, error)
 

pkg/client/simple/vfsclientset/clientset.go

@@ -51,6 +51,11 @@ func (c *VFSClientset) UpdateCluster(ctx context.Context, cluster *kops.Cluster,
 	return c.clusters().Update(cluster, status)
 }
 
+// UpdateCompletedCluster implements the UpdateCluster method of simple.Clientset for a VFS-backed state store.
+func (c *VFSClientset) UpdateCompletedCluster(ctx context.Context, cluster *kops.Cluster) error {
+	return c.clusters().UpdateCompleted(cluster)
+}
+
 // CreateCluster implements the CreateCluster method of simple.Clientset for a VFS-backed state store
 func (c *VFSClientset) CreateCluster(ctx context.Context, cluster *kops.Cluster) (*kops.Cluster, error) {
 	return c.clusters().Create(cluster)
@@ -117,7 +122,7 @@ func DeleteAllClusterState(basePath vfs.Path) error {
 			continue
 		}
 
-		if relativePath == "config" || relativePath == "cluster.spec" {
+		if relativePath == "config" || relativePath == "cluster.spec" || relativePath == "cluster-completed.spec" {
 			continue
 		}
 		if strings.HasPrefix(relativePath, "addons/") {

pkg/client/simple/vfsclientset/cluster.go

@@ -152,6 +152,22 @@ func (r *ClusterVFS) Update(c *api.Cluster, status *api.ClusterStatus) (*api.Clu
 	return c, nil
 }
 
+func (r *ClusterVFS) UpdateCompleted(c *api.Cluster) error {
+	clusterName := c.ObjectMeta.Name
+	if clusterName == "" {
+		return field.Required(field.NewPath("objectMeta", "name"), "clusterName is required")
+	}
+
+	if err := r.writeConfig(c, r.basePath.Join(clusterName, registry.PathClusterCompleted), c); err != nil {
+		if os.IsNotExist(err) {
+			return err
+		}
+		return fmt.Errorf("error writing Cluster: %v", err)
+	}
+
+	return nil
+}
+
 // List returns a slice containing all the cluster names
 // It skips directories that don't look like clusters
 func (r *ClusterVFS) listNames() ([]string, error) {

pkg/model/alimodel/policy_builder.go

@@ -276,7 +276,8 @@ func (b *PolicyBuilder) AddOSSPermissions(p *Policy) (*Policy, error) {
 			} else if b.Role == kops.InstanceGroupRoleNode {
 				resources := []string{
 					strings.Join([]string{b.RAMPrefix(), ":oss:*:*:", ramOSSPath, "/addons/*"}, ""),
-					strings.Join([]string{b.RAMPrefix(), ":oss:*:*:", ramOSSPath, "/cluster.spec"}, ""),
+					strings.Join([]string{b.RAMPrefix(), ":oss:*:*:", ramOSSPath, "/cluster-completed.spec"}, ""),
+					strings.Join([]string{b.RAMPrefix(), ":oss:*:*:", ramOSSPath, "/cluster.spec"}, ""), // todo remove in kops 1.20
 					strings.Join([]string{b.RAMPrefix(), ":oss:*:*:", ramOSSPath, "/config"}, ""),
 					strings.Join([]string{b.RAMPrefix(), ":oss:*:*:", ramOSSPath, "/instancegroup/*"}, ""),
 					strings.Join([]string{b.RAMPrefix(), ":oss:*:*:", ramOSSPath, "/pki/issued/*"}, ""),

pkg/model/iam/iam_builder.go

@@ -355,7 +355,8 @@ func (b *PolicyBuilder) AddS3Permissions(p *Policy) (*Policy, error) {
 				} else if b.Role == kops.InstanceGroupRoleNode {
 					resources := []string{
 						strings.Join([]string{b.IAMPrefix(), ":s3:::", iamS3Path, "/addons/*"}, ""),
-						strings.Join([]string{b.IAMPrefix(), ":s3:::", iamS3Path, "/cluster.spec"}, ""),
+						strings.Join([]string{b.IAMPrefix(), ":s3:::", iamS3Path, "/cluster-completed.spec"}, ""),
+						strings.Join([]string{b.IAMPrefix(), ":s3:::", iamS3Path, "/cluster.spec"}, ""), // todo remove in kops 1.20
 						strings.Join([]string{b.IAMPrefix(), ":s3:::", iamS3Path, "/config"}, ""),
 						strings.Join([]string{b.IAMPrefix(), ":s3:::", iamS3Path, "/instancegroup/*"}, ""),
 						strings.Join([]string{b.IAMPrefix(), ":s3:::", iamS3Path, "/pki/issued/*"}, ""),

pkg/model/iam/tests/iam_builder_node_strict.json

@@ -30,6 +30,7 @@
       ],
       "Resource": [
         "arn:aws:s3:::kops-tests/iam-builder-test.k8s.local/addons/*",
+        "arn:aws:s3:::kops-tests/iam-builder-test.k8s.local/cluster-completed.spec",
         "arn:aws:s3:::kops-tests/iam-builder-test.k8s.local/cluster.spec",
         "arn:aws:s3:::kops-tests/iam-builder-test.k8s.local/config",
         "arn:aws:s3:::kops-tests/iam-builder-test.k8s.local/instancegroup/*",

pkg/model/iam/tests/iam_builder_node_strict_ecr.json

@@ -30,6 +30,7 @@
       ],
       "Resource": [
         "arn:aws:s3:::kops-tests/iam-builder-test.k8s.local/addons/*",
+        "arn:aws:s3:::kops-tests/iam-builder-test.k8s.local/cluster-completed.spec",
         "arn:aws:s3:::kops-tests/iam-builder-test.k8s.local/cluster.spec",
         "arn:aws:s3:::kops-tests/iam-builder-test.k8s.local/config",
         "arn:aws:s3:::kops-tests/iam-builder-test.k8s.local/instancegroup/*",

upup/pkg/fi/cloudup/apply_cluster.go

@@ -886,11 +886,16 @@ func (c *ApplyClusterCmd) Run(ctx context.Context) error {
 	c.Target = target
 
 	if !dryRun {
-		err = registry.WriteConfigDeprecated(cluster, configBase.Join(registry.PathClusterCompleted), c.Cluster)
+		err = c.Clientset.UpdateCompletedCluster(ctx, c.Cluster)
 		if err != nil {
 			return fmt.Errorf("error writing completed cluster spec: %v", err)
 		}
 
+		err = registry.WriteConfigDeprecated(cluster, configBase.Join(registry.PathLegacyClusterCompleted), c.Cluster)
+		if err != nil {
+			return fmt.Errorf("error writing legacy completed cluster spec: %v", err)
+		}
+
 		vfsMirror := vfsclientset.NewInstanceGroupMirror(cluster, configBase)
 
 		for _, g := range c.InstanceGroups {

upup/pkg/fi/nodeup/BUILD.bazel

@@ -15,6 +15,7 @@ go_library(
         "//pkg/apis/kops/registry:go_default_library",
         "//pkg/apis/nodeup:go_default_library",
         "//pkg/assets:go_default_library",
+        "//pkg/kopscodecs:go_default_library",
         "//upup/pkg/fi:go_default_library",
         "//upup/pkg/fi/loader:go_default_library",
         "//upup/pkg/fi/nodeup/cloudinit:go_default_library",

upup/pkg/fi/nodeup/command.go

@@ -34,6 +34,7 @@ import (
 	"k8s.io/kops/pkg/apis/kops/registry"
 	"k8s.io/kops/pkg/apis/nodeup"
 	"k8s.io/kops/pkg/assets"
+	"k8s.io/kops/pkg/kopscodecs"
 	"k8s.io/kops/upup/pkg/fi"
 	"k8s.io/kops/upup/pkg/fi/nodeup/cloudinit"
 	"k8s.io/kops/upup/pkg/fi/nodeup/local"
@@ -120,7 +121,6 @@ func (c *NodeUpCommand) Run(out io.Writer) error {
 		return fmt.Errorf("ConfigBase is required")
 	}
 
-	c.cluster = &api.Cluster{}
 	{
 		clusterLocation := fi.StringValue(c.config.ClusterLocation)
 
@@ -140,24 +140,32 @@ func (c *NodeUpCommand) Run(out io.Writer) error {
 			return fmt.Errorf("error loading Cluster %q: %v", p, err)
 		}
 
-		err = utils.YamlUnmarshal(b, c.cluster)
+		o, _, err := kopscodecs.Decode(b, nil)
 		if err != nil {
 			return fmt.Errorf("error parsing Cluster %q: %v", p, err)
 		}
+		var ok bool
+		if c.cluster, ok = o.(*api.Cluster); !ok {
+			return fmt.Errorf("unexpected object type for Cluster %q: %T", p, o)
+		}
 	}
 
 	if c.config.InstanceGroupName != "" {
 		instanceGroupLocation := configBase.Join("instancegroup", c.config.InstanceGroupName)
 
-		c.instanceGroup = &api.InstanceGroup{}
 		b, err := instanceGroupLocation.ReadFile()
 		if err != nil {
 			return fmt.Errorf("error loading InstanceGroup %q: %v", instanceGroupLocation, err)
 		}
 
-		if err = utils.YamlUnmarshal(b, c.instanceGroup); err != nil {
+		o, _, err := kopscodecs.Decode(b, nil)
+		if err != nil {
 			return fmt.Errorf("error parsing InstanceGroup %q: %v", instanceGroupLocation, err)
 		}
+		var ok bool
+		if c.instanceGroup, ok = o.(*api.InstanceGroup); !ok {
+			return fmt.Errorf("unexpected object type for InstanceGroup %q: %T", instanceGroupLocation, o)
+		}
 	} else {
 		klog.Warningf("No instance group defined in nodeup config")
 	}