Force etcd2 to use application/json, add base64-wrapper decoder as fallback

Author: liggitt

pkg/api/testapi/testapi.go

@@ -349,7 +349,7 @@ func (g TestGroup) StorageCodec() runtime.Codec {
 	// etcd2 only supports string data - we must wrap any result before returning
 	// TODO: remove for etcd3 / make parameterizable
 	if !storageSerializer.EncodesAsText {
-		s = runtime.NewBase64Serializer(s)
+		s = runtime.NewBase64Serializer(s, s)
 	}
 	ds := recognizer.NewDecoder(s, api.Codecs.UniversalDeserializer())
 

staging/src/k8s.io/apimachinery/pkg/api/testing/codec.go

@@ -58,7 +58,7 @@ func TestStorageCodec(codecs runtimeserializer.CodecFactory, gvs ...schema.Group
 		// TODO: remove for etcd3 / make parameterizable
 		serializer := serializerInfo.Serializer
 		if !serializerInfo.EncodesAsText {
-			serializer = runtime.NewBase64Serializer(serializer)
+			serializer = runtime.NewBase64Serializer(serializer, serializer)
 		}
 
 		decoder := recognizer.NewDecoder(serializer, codecs.UniversalDeserializer())

staging/src/k8s.io/apimachinery/pkg/runtime/codec.go

@@ -195,16 +195,17 @@ func (c *parameterCodec) EncodeParameters(obj Object, to schema.GroupVersion) (u
 }
 
 type base64Serializer struct {
-	Serializer
+	Encoder
+	Decoder
 }
 
-func NewBase64Serializer(s Serializer) Serializer {
-	return &base64Serializer{s}
+func NewBase64Serializer(e Encoder, d Decoder) Serializer {
+	return &base64Serializer{e, d}
 }
 
 func (s base64Serializer) Encode(obj Object, stream io.Writer) error {
 	e := base64.NewEncoder(base64.StdEncoding, stream)
-	err := s.Serializer.Encode(obj, e)
+	err := s.Encoder.Encode(obj, e)
 	e.Close()
 	return err
 }
@@ -215,7 +216,7 @@ func (s base64Serializer) Decode(data []byte, defaults *schema.GroupVersionKind,
 	if err != nil {
 		return nil, nil, err
 	}
-	return s.Serializer.Decode(out[:n], defaults, into)
+	return s.Decoder.Decode(out[:n], defaults, into)
 }
 
 // SerializerInfoForMediaType returns the first info in types that has a matching media type (which cannot

staging/src/k8s.io/apiserver/pkg/server/options/etcd.go

@@ -68,8 +68,8 @@ func (s *EtcdOptions) AddFlags(fs *pflag.FlagSet) {
 		"format: group/resource#servers, where servers are http://ip:port, semicolon separated.")
 
 	fs.StringVar(&s.DefaultStorageMediaType, "storage-media-type", s.DefaultStorageMediaType, ""+
-		"The media type to use to store objects in storage. Defaults to application/json. "+
-		"Some resources may only support a specific media type and will ignore this setting.")
+		"The media type to use to store objects in storage. "+
+		"Some resources or storage backends may only support a specific media type and will ignore this setting.")
 	fs.IntVar(&s.DeleteCollectionWorkers, "delete-collection-workers", s.DeleteCollectionWorkers,
 		"Number of workers spawned for DeleteCollection call. These are used to speed up namespace cleanup.")
 

staging/src/k8s.io/apiserver/pkg/server/storage/storage_codec.go

@@ -47,18 +47,22 @@ func NewStorageCodec(opts StorageCodecConfig) (runtime.Codec, error) {
 	if err != nil {
 		return nil, fmt.Errorf("%q is not a valid mime-type", opts.StorageMediaType)
 	}
+
+	if opts.Config.Type == storagebackend.StorageTypeETCD2 && mediaType != "application/json" {
+		glog.Warningf(`storage type %q does not support media type %q, using "application/json"`, storagebackend.StorageTypeETCD2, mediaType)
+		mediaType = "application/json"
+	}
+
 	serializer, ok := runtime.SerializerInfoForMediaType(opts.StorageSerializer.SupportedMediaTypes(), mediaType)
 	if !ok {
-		return nil, fmt.Errorf("unable to find serializer for %q", opts.StorageMediaType)
+		return nil, fmt.Errorf("unable to find serializer for %q", mediaType)
 	}
 
 	s := serializer.Serializer
 
-	// etcd2 only supports string data - we must wrap any result before returning
-	// TODO: storagebackend should return a boolean indicating whether it supports binary data
+	// make sure the selected encoder supports string data
 	if !serializer.EncodesAsText && opts.Config.Type == storagebackend.StorageTypeETCD2 {
-		glog.V(4).Infof("Wrapping the underlying binary storage serializer with a base64 encoding for etcd2")
-		s = runtime.NewBase64Serializer(s)
+		return nil, fmt.Errorf("storage type %q does not support binary media type %q", storagebackend.StorageTypeETCD2, mediaType)
 	}
 
 	// Give callers the opportunity to wrap encoders and decoders.  For decoders, each returned decoder will
@@ -67,7 +71,17 @@ func NewStorageCodec(opts StorageCodecConfig) (runtime.Codec, error) {
 	if opts.EncoderDecoratorFn != nil {
 		encoder = opts.EncoderDecoratorFn(encoder)
 	}
-	decoders := []runtime.Decoder{s, opts.StorageSerializer.UniversalDeserializer()}
+	decoders := []runtime.Decoder{
+		// selected decoder as the primary
+		s,
+		// universal deserializer as a fallback
+		opts.StorageSerializer.UniversalDeserializer(),
+		// base64-wrapped universal deserializer as a last resort.
+		// this allows reading base64-encoded protobuf, which should only exist if etcd2+protobuf was used at some point.
+		// data written that way could exist in etcd2, or could have been migrated to etcd3.
+		// TODO: flag this type of data if we encounter it, require migration (read to decode, write to persist using a supported encoder), and remove in 1.8
+		runtime.NewBase64Serializer(nil, opts.StorageSerializer.UniversalDeserializer()),
+	}
 	if opts.DecoderDecoratorFn != nil {
 		decoders = opts.DecoderDecoratorFn(decoders)
 	}