Add strict deserialization for kubelet component config

CodecFactory is started with EnableStrict that throws an error when deserializing
a Kubelet component config that is malformed (e.g. unknown or duplicate keys).

When strict decoding a v1beta1 config fails, non-strict decoding is used and a warning is emitted.
For this, NewSchemeAndCodecs is now a variadic function that can take multiple
arguments for augmenting the returned codec factory. Strict decoding is
then explicitely enabled when decoding a kubelet config.
Additionally, decoding a RemoteConfigSource needs to be non-strict
to avoid an accidental error when it contains newer API fields that are not
yet known to the Kubelet.

DecodeKubeletConfiguration returns a wrapped error instead of a simple string
so its type can be tested.

Add unit tests for unhappy paths when loading a component config

Add keys for test cases struct fields, remove nil field initialization

Co-Authored-By: Jordan Liggitt <jordan@liggitt.net>

pkg/kubelet/apis/config/scheme/scheme.go

@@ -26,8 +26,9 @@ import (
 // Utility functions for the Kubelet's kubeletconfig API group
 
 // NewSchemeAndCodecs is a utility function that returns a Scheme and CodecFactory
-// that understand the types in the kubeletconfig API group.
+// that understand the types in the kubeletconfig API group. Passing mutators allows
-func NewSchemeAndCodecs() (*runtime.Scheme, *serializer.CodecFactory, error) {
+// for adjusting the behavior of the CodecFactory, for example enable strict decoding.
+func NewSchemeAndCodecs(mutators ...serializer.CodecFactoryOptionsMutator) (*runtime.Scheme, *serializer.CodecFactory, error) {
 	scheme := runtime.NewScheme()
 	if err := kubeletconfig.AddToScheme(scheme); err != nil {
 		return nil, nil, err
@@ -35,6 +36,6 @@ func NewSchemeAndCodecs() (*runtime.Scheme, *serializer.CodecFactory, error) {
 	if err := kubeletconfigv1beta1.AddToScheme(scheme); err != nil {
 		return nil, nil, err
 	}
-	codecs := serializer.NewCodecFactory(scheme)
+	codecs := serializer.NewCodecFactory(scheme, mutators...)
 	return scheme, &codecs, nil
 }

pkg/kubelet/kubeletconfig/checkpoint/BUILD

@@ -43,6 +43,7 @@ go_library(
         "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/fields:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/runtime:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/runtime/serializer:go_default_library",
         "//staging/src/k8s.io/client-go/kubernetes:go_default_library",
         "//staging/src/k8s.io/client-go/tools/cache:go_default_library",
         "//staging/src/k8s.io/kubelet/config/v1beta1:go_default_library",

pkg/kubelet/kubeletconfig/checkpoint/download.go

@@ -26,6 +26,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/serializer"
 	clientset "k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/cache"
 	kubeletconfigv1beta1 "k8s.io/kubelet/config/v1beta1"
@@ -106,8 +107,9 @@ func NewRemoteConfigSource(source *apiv1.NodeConfigSource) (RemoteConfigSource,
 // DecodeRemoteConfigSource is a helper for using the apimachinery to decode serialized RemoteConfigSources;
 // e.g. the metadata stored by checkpoint/store/fsstore.go
 func DecodeRemoteConfigSource(data []byte) (RemoteConfigSource, error) {
-	// decode the remote config source
+	// Decode the remote config source. We want this to be non-strict
-	_, codecs, err := scheme.NewSchemeAndCodecs()
+	// so we don't error out on newer API keys.
+	_, codecs, err := scheme.NewSchemeAndCodecs(serializer.DisableStrict)
 	if err != nil {
 		return nil, err
 	}

pkg/kubelet/kubeletconfig/configfiles/BUILD

@@ -43,6 +43,8 @@ go_test(
         "//pkg/kubelet/kubeletconfig/util/test:go_default_library",
         "//pkg/util/filesystem:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/api/equality:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/runtime:go_default_library",
         "//staging/src/k8s.io/kubelet/config/v1beta1:go_default_library",
+        "//vendor/github.com/pkg/errors:go_default_library",
     ],
 )

pkg/kubelet/kubeletconfig/configfiles/configfiles.go

@@ -45,7 +45,7 @@ type fsLoader struct {
 
 // NewFsLoader returns a Loader that loads a KubeletConfiguration from the `kubeletFile`
 func NewFsLoader(fs utilfs.Filesystem, kubeletFile string) (Loader, error) {
-	_, kubeletCodecs, err := kubeletscheme.NewSchemeAndCodecs()
+	_, kubeletCodecs, err := kubeletscheme.NewSchemeAndCodecs(serializer.EnableStrict)
 	if err != nil {
 		return nil, err
 	}

pkg/kubelet/kubeletconfig/configfiles/configfiles_test.go

@@ -21,7 +21,10 @@ import (
 	"path/filepath"
 	"testing"
 
+	"github.com/pkg/errors"
+
 	apiequality "k8s.io/apimachinery/pkg/api/equality"
+	"k8s.io/apimachinery/pkg/runtime"
 	kubeletconfigv1beta1 "k8s.io/kubelet/config/v1beta1"
 	kubeletconfig "k8s.io/kubernetes/pkg/kubelet/apis/config"
 	kubeletscheme "k8s.io/kubernetes/pkg/kubelet/apis/config/scheme"
@@ -36,101 +39,114 @@ const kubeletFile = "kubelet"
 
 func TestLoad(t *testing.T) {
 	cases := []struct {
-		desc   string
+		desc      string
-		file   *string
+		file      *string
-		expect *kubeletconfig.KubeletConfiguration
+		expect    *kubeletconfig.KubeletConfiguration
-		err    string
+		err       string
+		strictErr bool
 	}{
 		// missing file
 		{
-			"missing file",
+			desc: "missing file",
-			nil,
+			err:  "failed to read",
-			nil,
-			"failed to read",
 		},
 		// empty file
 		{
-			"empty file",
+			desc: "empty file",
-			newString(``),
+			file: newString(``),
-			nil,
+			err:  "was empty",
-			"was empty",
 		},
 		// invalid format
 		{
-			"invalid yaml",
+			desc: "invalid yaml",
-			newString(`*`),
+			file: newString(`*`),
-			nil,
+			err:  "failed to decode",
-			"failed to decode",
 		},
 		{
-			"invalid json",
+			desc: "invalid json",
-			newString(`{*`),
+			file: newString(`{*`),
-			nil,
+			err:  "failed to decode",
-			"failed to decode",
 		},
 		// invalid object
 		{
-			"missing kind",
+			desc: "missing kind",
-			newString(`{"apiVersion":"kubelet.config.k8s.io/v1beta1"}`),
+			file: newString(`{"apiVersion":"kubelet.config.k8s.io/v1beta1"}`),
-			nil,
+			err:  "failed to decode",
-			"failed to decode",
 		},
 		{
-			"missing version",
+			desc: "missing version",
-			newString(`{"kind":"KubeletConfiguration"}`),
+			file: newString(`{"kind":"KubeletConfiguration"}`),
-			nil,
+			err:  "failed to decode",
-			"failed to decode",
 		},
 		{
-			"unregistered kind",
+			desc: "unregistered kind",
-			newString(`{"kind":"BogusKind","apiVersion":"kubelet.config.k8s.io/v1beta1"}`),
+			file: newString(`{"kind":"BogusKind","apiVersion":"kubelet.config.k8s.io/v1beta1"}`),
-			nil,
+			err:  "failed to decode",
-			"failed to decode",
 		},
 		{
-			"unregistered version",
+			desc: "unregistered version",
-			newString(`{"kind":"KubeletConfiguration","apiVersion":"bogusversion"}`),
+			file: newString(`{"kind":"KubeletConfiguration","apiVersion":"bogusversion"}`),
-			nil,
+			err:  "failed to decode",
-			"failed to decode",
 		},
 
 		// empty object with correct kind and version should result in the defaults for that kind and version
 		{
-			"default from yaml",
+			desc: "default from yaml",
-			newString(`kind: KubeletConfiguration
+			file: newString(`kind: KubeletConfiguration
 apiVersion: kubelet.config.k8s.io/v1beta1`),
-			newConfig(t),
+			expect: newConfig(t),
-			"",
 		},
 		{
-			"default from json",
+			desc:   "default from json",
-			newString(`{"kind":"KubeletConfiguration","apiVersion":"kubelet.config.k8s.io/v1beta1"}`),
+			file:   newString(`{"kind":"KubeletConfiguration","apiVersion":"kubelet.config.k8s.io/v1beta1"}`),
-			newConfig(t),
+			expect: newConfig(t),
-			"",
 		},
 
 		// relative path
 		{
-			"yaml, relative path is resolved",
+			desc: "yaml, relative path is resolved",
-			newString(fmt.Sprintf(`kind: KubeletConfiguration
+			file: newString(fmt.Sprintf(`kind: KubeletConfiguration
 apiVersion: kubelet.config.k8s.io/v1beta1
 staticPodPath: %s`, relativePath)),
-			func() *kubeletconfig.KubeletConfiguration {
+			expect: func() *kubeletconfig.KubeletConfiguration {
 				kc := newConfig(t)
 				kc.StaticPodPath = filepath.Join(configDir, relativePath)
 				return kc
 			}(),
-			"",
 		},
 		{
-			"json, relative path is resolved",
+			desc: "json, relative path is resolved",
-			newString(fmt.Sprintf(`{"kind":"KubeletConfiguration","apiVersion":"kubelet.config.k8s.io/v1beta1","staticPodPath":"%s"}`, relativePath)),
+			file: newString(fmt.Sprintf(`{"kind":"KubeletConfiguration","apiVersion":"kubelet.config.k8s.io/v1beta1","staticPodPath":"%s"}`, relativePath)),
-			func() *kubeletconfig.KubeletConfiguration {
+			expect: func() *kubeletconfig.KubeletConfiguration {
 				kc := newConfig(t)
 				kc.StaticPodPath = filepath.Join(configDir, relativePath)
 				return kc
 			}(),
-			"",
+		},
+		{
+			// This should fail from v1beta2+
+			desc: "duplicate field",
+			file: newString(fmt.Sprintf(`kind: KubeletConfiguration
+apiVersion: kubelet.config.k8s.io/v1beta1
+staticPodPath: %s
+staticPodPath: %s/foo`, relativePath, relativePath)),
+			// err:       `key "staticPodPath" already set`,
+			// strictErr: true,
+			expect: func() *kubeletconfig.KubeletConfiguration {
+				kc := newConfig(t)
+				kc.StaticPodPath = filepath.Join(configDir, relativePath, "foo")
+				return kc
+			}(),
+		},
+		{
+			// This should fail from v1beta2+
+			desc: "unknown field",
+			file: newString(`kind: KubeletConfiguration
+apiVersion: kubelet.config.k8s.io/v1beta1
+foo: bar`),
+			// err:       "found unknown field: foo",
+			// strictErr: true,
+			expect: newConfig(t),
 		},
 	}
 
@@ -148,6 +164,10 @@ staticPodPath: %s`, relativePath)),
 				t.Fatalf("unexpected error: %v", err)
 			}
 			kc, err := loader.Load()
+
+			if c.strictErr && !runtime.IsStrictDecodingError(errors.Cause(err)) {
+				t.Fatalf("got error: %v, want strict decoding error", err)
+			}
 			if utiltest.SkipRest(t, c.desc, err, c.err) {
 				return
 			}

pkg/kubelet/kubeletconfig/util/codec/BUILD

@@ -14,9 +14,12 @@ go_library(
         "//pkg/apis/core/install:go_default_library",
         "//pkg/kubelet/apis/config:go_default_library",
         "//pkg/kubelet/apis/config/scheme:go_default_library",
+        "//pkg/kubelet/apis/config/v1beta1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/runtime:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/runtime/serializer:go_default_library",
+        "//vendor/github.com/pkg/errors:go_default_library",
+        "//vendor/k8s.io/klog:go_default_library",
     ],
 )
 

pkg/kubelet/kubeletconfig/util/codec/codec.go

@@ -19,6 +19,9 @@ package codec
 import (
 	"fmt"
 
+	"github.com/pkg/errors"
+	"k8s.io/klog"
+
 	// ensure the core apis are installed
 	_ "k8s.io/kubernetes/pkg/apis/core/install"
 
@@ -28,9 +31,10 @@ import (
 	"k8s.io/kubernetes/pkg/api/legacyscheme"
 	kubeletconfig "k8s.io/kubernetes/pkg/kubelet/apis/config"
 	"k8s.io/kubernetes/pkg/kubelet/apis/config/scheme"
+	kubeletconfigv1beta1 "k8s.io/kubernetes/pkg/kubelet/apis/config/v1beta1"
 )
 
-// EncodeKubeletConfig encodes an internal KubeletConfiguration to an external YAML representation
+// EncodeKubeletConfig encodes an internal KubeletConfiguration to an external YAML representation.
 func EncodeKubeletConfig(internal *kubeletconfig.KubeletConfiguration, targetVersion schema.GroupVersion) ([]byte, error) {
 	encoder, err := NewKubeletconfigYAMLEncoder(targetVersion)
 	if err != nil {
@@ -44,7 +48,7 @@ func EncodeKubeletConfig(internal *kubeletconfig.KubeletConfiguration, targetVer
 	return data, nil
 }
 
-// NewKubeletconfigYAMLEncoder returns an encoder that can write objects in the kubeletconfig API group to YAML
+// NewKubeletconfigYAMLEncoder returns an encoder that can write objects in the kubeletconfig API group to YAML.
 func NewKubeletconfigYAMLEncoder(targetVersion schema.GroupVersion) (runtime.Encoder, error) {
 	_, codecs, err := scheme.NewSchemeAndCodecs()
 	if err != nil {
@@ -58,7 +62,7 @@ func NewKubeletconfigYAMLEncoder(targetVersion schema.GroupVersion) (runtime.Enc
 	return codecs.EncoderForVersion(info.Serializer, targetVersion), nil
 }
 
-// NewYAMLEncoder generates a new runtime.Encoder that encodes objects to YAML
+// NewYAMLEncoder generates a new runtime.Encoder that encodes objects to YAML.
 func NewYAMLEncoder(groupName string) (runtime.Encoder, error) {
 	// encode to YAML
 	mediaType := "application/yaml"
@@ -72,16 +76,55 @@ func NewYAMLEncoder(groupName string) (runtime.Encoder, error) {
 		return nil, fmt.Errorf("no enabled versions for group %q", groupName)
 	}
 
-	// the "best" version supposedly comes first in the list returned from legacyscheme.Registry.EnabledVersionsForGroup
+	// the "best" version supposedly comes first in the list returned from legacyscheme.Registry.EnabledVersionsForGroup.
 	return legacyscheme.Codecs.EncoderForVersion(info.Serializer, versions[0]), nil
 }
 
-// DecodeKubeletConfiguration decodes a serialized KubeletConfiguration to the internal type
+// newLenientSchemeAndCodecs returns a scheme that has only v1beta1 registered into
+// it and a CodecFactory with strict decoding disabled.
+func newLenientSchemeAndCodecs() (*runtime.Scheme, *serializer.CodecFactory, error) {
+	lenientScheme := runtime.NewScheme()
+	if err := kubeletconfig.AddToScheme(lenientScheme); err != nil {
+		return nil, nil, fmt.Errorf("failed to add internal kubelet config API to lenient scheme: %v", err)
+	}
+	if err := kubeletconfigv1beta1.AddToScheme(lenientScheme); err != nil {
+		return nil, nil, fmt.Errorf("failed to add kubelet config v1beta1 API to lenient scheme: %v", err)
+	}
+	lenientCodecs := serializer.NewCodecFactory(lenientScheme, serializer.DisableStrict)
+	return lenientScheme, &lenientCodecs, nil
+}
+
+// DecodeKubeletConfiguration decodes a serialized KubeletConfiguration to the internal type.
 func DecodeKubeletConfiguration(kubeletCodecs *serializer.CodecFactory, data []byte) (*kubeletconfig.KubeletConfiguration, error) {
-	// the UniversalDecoder runs defaulting and returns the internal type by default
+	var (
+		obj runtime.Object
+		gvk *schema.GroupVersionKind
+	)
+
+	// The UniversalDecoder runs defaulting and returns the internal type by default.
 	obj, gvk, err := kubeletCodecs.UniversalDecoder().Decode(data, nil, nil)
 	if err != nil {
-		return nil, fmt.Errorf("failed to decode, error: %v", err)
+		// Try strict decoding first. If that fails decode with a lenient
+		// decoder, which has only v1beta1 registered, and log a warning.
+		// The lenient path is to be dropped when support for v1beta1 is dropped.
+		if !runtime.IsStrictDecodingError(err) {
+			return nil, errors.Wrap(err, "failed to decode")
+		}
+
+		var lenientErr error
+		_, lenientCodecs, lenientErr := newLenientSchemeAndCodecs()
+		if lenientErr != nil {
+			return nil, lenientErr
+		}
+
+		obj, gvk, lenientErr = lenientCodecs.UniversalDecoder().Decode(data, nil, nil)
+		if lenientErr != nil {
+			// Lenient decoding failed with the current version, return the
+			// original strict error.
+			return nil, fmt.Errorf("failed lenient decoding: %v", err)
+		}
+		// Continue with the v1beta1 object that was decoded leniently, but emit a warning.
+		klog.Warningf("using lenient decoding as strict decoding failed: %v", err)
 	}
 
 	internalKC, ok := obj.(*kubeletconfig.KubeletConfiguration)