k8s源码学习-codec编码器

k8s源码学习-codec编码器

codec编码器是什么？

codec编解码器是一个通用的术语，包含编码器和解码器，表示将数据转换为特定格式的过程，在k8s系统中主要负责将etcd中数据进行编码和解码操作

Serializer是什么？

Serializer是序列化器，包含序列化和反序列化操作，序列化操作是将数据（数组，结构体，对象）转为字符串的操作，反序列化是将字符串转为数据过程

codec编码器和Serializer是什么关系？

Serializer算是codec编码器的一种，因为每种Serializer序列化器都实现了Encoder和Decoder接口，只要Encoder和Decoder接口方法的数据结构的都是序列化器，代码：vendor/k8s.io/apimachinery/pkg/runtime/interfaces.go

type Encoder interface{ Encode(obj Object, w io.Writer) error Identifier() Identifier } type Decoder interface{ Decode(data []byte, defaults *schema.GroupVersionKind, into Object) (Object, *schema.GroupVersionKind, error) } type Serializer interface{ Encoder Decoder } type Codec Serializer

Codec编解码器包含3种序Serializer列化器

在进行编解码操作时候，每一种序列化器都对资源对象的metav1.TypeMeta(即ApiVersion和Kind字段)进程验证，如果资源对象未提供这些字段，就会返回错误

jsonSerializer json格式的序列化和发序列化器，使用application/json的ContentType作为标识符 yamlSerializer yaml格式的序列化和发序列化器，使用application/yaml的ContentType作为标识符 protobufSerializer profobuf格式的序列化和发序列化器，使用application/vnd.kubernetes.protobuf的ContentType作为标识符

Codec 编码器通过 NewCodecFactory 函数实例化

实例化的过程中，将jsonSerializer、yamlSerializer、protobufSerializer序列化器全部实例化，NewCodecFactory调用newSerializersForScheme代码如下：vendor/k8s.io/apimachinery/pkg/runtime/serializer/codec_factory.go

func NewCodecFactory(scheme *runtime.Scheme, mutators ...CodecFactoryOptionsMutator) CodecFactory { options := CodecFactoryOptions{Pretty: true} for _, fn := range mutators { fn(&options) } serializers := newSerializersForScheme(scheme, json.DefaultMetaFactory, options) return newCodecFactory(scheme, serializers) } func newSerializersForScheme(scheme *runtime.Scheme, mf json.MetaFactory, options CodecFactoryOptions) []serializerType { jsonSerializer := json.NewSerializerWithOptions( mf, scheme, scheme, json.SerializerOptions{Yaml: false, Pretty: false, Strict: options.Strict}, ) jsonSerializerType := serializerType{ AcceptContentTypes: []string{runtime.ContentTypeJSON}, ContentType: runtime.ContentTypeJSON, FileExtensions: []string{"json"}, EncodesAsText: true, Serializer: jsonSerializer, Framer: json.Framer, StreamSerializer: jsonSerializer, } if options.Pretty { jsonSerializerType.PrettySerializer = json.NewSerializerWithOptions( mf, scheme, scheme, json.SerializerOptions{Yaml: false, Pretty: true, Strict: options.Strict}, ) } strictJSONSerializer := json.NewSerializerWithOptions( mf, scheme, scheme, json.SerializerOptions{Yaml: false, Pretty: false, Strict: true}, ) jsonSerializerType.StrictSerializer = strictJSONSerializer yamlSerializer := json.NewSerializerWithOptions( mf, scheme, scheme, json.SerializerOptions{Yaml: true, Pretty: false, Strict: options.Strict}, ) strictYAMLSerializer := json.NewSerializerWithOptions( mf, scheme, scheme, json.SerializerOptions{Yaml: true, Pretty: false, Strict: true}, ) protoSerializer := protobuf.NewSerializer(scheme, scheme) protoRawSerializer := protobuf.NewRawSerializer(scheme, scheme) serializers := []serializerType{ jsonSerializerType, { AcceptContentTypes: []string{runtime.ContentTypeYAML}, ContentType: runtime.ContentTypeYAML, FileExtensions: []string{"yaml"}, EncodesAsText: true, Serializer: yamlSerializer, StrictSerializer: strictYAMLSerializer, }, { AcceptContentTypes: []string{runtime.ContentTypeProtobuf}, ContentType: runtime.ContentTypeProtobuf, FileExtensions: []string{"pb"}, Serializer: protoSerializer, // note, strict decoding is unsupported for protobuf, // fall back to regular serializing StrictSerializer: protoSerializer, Framer: protobuf.LengthDelimitedFramer, StreamSerializer: protoRawSerializer, }, } for _, fn := range serializerExtensions { if serializer, ok := fn(scheme); ok { serializers = append(serializers, serializer) } } return serializers }

jsonSerializer 通过 json.NewSerializerWithOptions 函数实例化，Yaml字段=false，ContentTypeJSON string = "application/json", jsonSerializer使用Go语言标准库encoding/json实现序列化和反序列化yamlSerializer 通过 json.NewSerializerWithOptions 函数实例化，Yaml字段=true，ContentTypeYAML string = "application/yaml"，yamlSerializer使用第三方库通过 protobuf.NewSerializer(scheme, scheme) 函数实例化ContentTypeProtobuf string = "application/vnd.kubernetes.protobuf"

序列化Encode和反序列化Decode操作

序列化Encode：如果s.options.Yaml=true，会先通过json.Marshal将资源转为json格式，然后通过yaml.JSONToYAML再转为yaml格式，如果是json格式，通过json.NewEncoder转为json格式，如果s.options.Pretty=true，会做一些优化反序列化Decode：Decode函数支持两种格式的反序列化操作，分别是YAML和JSON格式。如果是YAML格式（s.options.Yaml=true），通过yaml.YAMLToJSON函数将JSON格式数据转为资源对象填充到data字段中，此时，不管反序列化操作的是YAML还是JSON格式，data字段都是JSON格式数据，然后通过s.meta.Interpret函授从josn格式数据中提取资源对象的*schema.GroupVersionKind,最后通过json.unmarshal将Json数据反序列化并返回。代码：vendor/k8s.io/apimachinery/pkg/runtime/serializer/json/json.go

// Encode serializes the provided object to the given writer. func (s *Serializer) Encode(obj runtime.Object, w io.Writer) error { if co, ok := obj.(runtime.CacheableObject); ok { return co.CacheEncode(s.Identifier(), s.doEncode, w) } return s.doEncode(obj, w) } func (s *Serializer) doEncode(obj runtime.Object, w io.Writer) error { if s.options.Yaml { json, err := json.Marshal(obj) if err != nil { return err } data, err := yaml.JSONToYAML(json) if err != nil { return err } _, err = w.Write(data) return err } if s.options.Pretty { data, err := json.MarshalIndent(obj, "", " ") if err != nil { return err } _, err = w.Write(data) return err } encoder := json.NewEncoder(w) return encoder.Encode(obj) } // Decode attempts to convert the provided data into YAML or JSON func (s *Serializer) Decode(originalData []byte, gvk *schema.GroupVersionKind, into runtime.Object) (runtime.Object, *schema.GroupVersionKind, error) { data := originalData if s.options.Yaml { altered, err := yaml.YAMLToJSON(data) if err != nil { return nil, nil, err } data = altered } actual, err := s.meta.Interpret(data) if err != nil { return nil, nil, err } if gvk != nil { *actual = gvkWithDefaults(*actual, *gvk) } if unk, ok := into.(*runtime.Unknown); ok && unk != nil { unk.Raw = originalData unk.ContentType = runtime.ContentTypeJSON unk.GetObjectKind().SetGroupVersionKind(*actual) return unk, actual, nil } if into != nil { _, isUnstructured := into.(runtime.Unstructured) types, _, err := s.typer.ObjectKinds(into) switch { case runtime.IsNotRegisteredError(err), isUnstructured: strictErrs, err := s.unmarshal(into, data, originalData) if err != nil { return nil, actual, err } else if len(strictErrs) > 0 { return into, actual, runtime.NewStrictDecodingError(strictErrs) } return into, actual, nil case err != nil: return nil, actual, err default: *actual = gvkWithDefaults(*actual, types[0]) } } if len(actual.Kind) == 0 { return nil, actual, runtime.NewMissingKindErr(string(originalData)) } if len(actual.Version) == 0 { return nil, actual, runtime.NewMissingVersionErr(string(originalData)) } // use the target if necessary obj, err := runtime.UseOrCreateObject(s.typer, s.creater, *actual, into) if err != nil { return nil, actual, err } strictErrs, err := s.unmarshal(obj, data, originalData) if err != nil { return nil, actual, err } else if len(strictErrs) > 0 { return obj, actual, runtime.NewStrictDecodingError(strictErrs) } return obj, actual, nil }

protobufSerializer序列化器

Protobuf（Google Protocol Buffer）是Google公司内部的混合语言数据标准，Protocol Buffers是一种轻便、高效的结构化数据存储格式，可以用于结构化数据序列化。它很适合做数据存储或成为RPC数据交换格式。它可用于通信协议、数据存储等领域，与语言无关、与平台无关、可扩展的序列化结构数据格式。Protobuf Example代码示例如下：

package lm; message helloworld { required int32 id = 1; required string str = 2; optional int32 opt = 3; }

protobufSerializer序列化器使用proto库来实现序列化和反序列操作

protobuf序列化操作

代码：vendor/k8s.io/apimachinery/pkg/runtime/serializer/protobuf/protobuf.go

// Encode serializes the provided object to the given writer. func (s *Serializer) Encode(obj runtime.Object, w io.Writer) error { if co, ok := obj.(runtime.CacheableObject); ok { return co.CacheEncode(s.Identifier(), s.doEncode, w) } return s.doEncode(obj, w) } func (s *Serializer) doEncode(obj runtime.Object, w io.Writer) error { prefixSize := uint64(len(s.prefix)) var unk runtime.Unknown switch t := obj.(type) { case *runtime.Unknown: estimatedSize := prefixSize + uint64(t.Size()) data := make([]byte, estimatedSize) i, err := t.MarshalTo(data[prefixSize:]) if err != nil { return err } copy(data, s.prefix) _, err = w.Write(data[:prefixSize+uint64(i)]) return err default: kind := obj.GetObjectKind().GroupVersionKind() unk = runtime.Unknown{ TypeMeta: runtime.TypeMeta{ Kind: kind.Kind, APIVersion: kind.GroupVersion().String(), }, } } switch t := obj.(type) { case bufferedMarshaller: // this path performs a single allocation during write but requires the caller to implement // the more efficient Size and MarshalToSizedBuffer methods encodedSize := uint64(t.Size()) estimatedSize := prefixSize + estimateUnknownSize(&unk, encodedSize) data := make([]byte, estimatedSize) i, err := unk.NestedMarshalTo(data[prefixSize:], t, encodedSize) if err != nil { return err } copy(data, s.prefix) _, err = w.Write(data[:prefixSize+uint64(i)]) return err case proto.Marshaler: // this path performs extra allocations data, err := t.Marshal() if err != nil { return err } unk.Raw = data estimatedSize := prefixSize + uint64(unk.Size()) data = make([]byte, estimatedSize) i, err := unk.MarshalTo(data[prefixSize:]) if err != nil { return err } copy(data, s.prefix) _, err = w.Write(data[:prefixSize+uint64(i)]) return err default: // TODO: marshal with a different content type and serializer (JSON for third party objects) return errNotMarshalable{reflect.TypeOf(obj)} } }

Encode函数首先验证资源对象是否为proto.Marshaler类型，proto.Marshaler是一个interface接口类型，该接口专门留给对象自定义实现的序列化操作。如果资源对象为proto.Marshaler类型，则通过t.Marshal序列化函数进行编码。通过unk.MarshalTo函数在编码后的数据前加上protoEncodingPrefix前缀，前缀为magic-number特殊标识，其用于标识一个包的完整性。所有通过protobufSerializer序列化器编码的数据都会有前缀。前缀数据共4字节，分别是0x6b、0x38、0x73、0x00，其中第4个字节是为编码样式保留的。

反序列化操作

代码：vendor/k8s.io/apimachinery/pkg/runtime/serializer/protobuf/protobuf.go

func (s *Serializer) Decode(originalData []byte, gvk *schema.GroupVersionKind, into runtime.Object) (runtime.Object, *schema.GroupVersionKind, error) { prefixLen := len(s.prefix) switch { case len(originalData) == 0: // TODO: treat like decoding {} from JSON with defaulting return nil, nil, fmt.Errorf("empty data") case len(originalData) < prefixLen || !bytes.Equal(s.prefix, originalData[:prefixLen]): return nil, nil, fmt.Errorf("provided data does not appear to be a protobuf message, expected prefix %v", s.prefix) case len(originalData) == prefixLen: // TODO: treat like decoding {} from JSON with defaulting return nil, nil, fmt.Errorf("empty body") } data := originalData[prefixLen:] unk := runtime.Unknown{} if err := unk.Unmarshal(data); err != nil { return nil, nil, err } actual := unk.GroupVersionKind() copyKindDefaults(&actual, gvk) if intoUnknown, ok := into.(*runtime.Unknown); ok && intoUnknown != nil { *intoUnknown = unk if ok, _, _ := s.RecognizesData(unk.Raw); ok { intoUnknown.ContentType = runtime.ContentTypeProtobuf } return intoUnknown, &actual, nil } if into != nil { types, _, err := s.typer.ObjectKinds(into) switch { case runtime.IsNotRegisteredError(err): pb, ok := into.(proto.Message) if !ok { return nil, &actual, errNotMarshalable{reflect.TypeOf(into)} } if err := proto.Unmarshal(unk.Raw, pb); err != nil { return nil, &actual, err } return into, &actual, nil case err != nil: return nil, &actual, err default: copyKindDefaults(&actual, &types[0]) // if the result of defaulting did not set a version or group, ensure that at least group is set // (copyKindDefaults will not assign Group if version is already set). This guarantees that the group // of into is set if there is no better information from the caller or object. if len(actual.Version) == 0 && len(actual.Group) == 0 { actual.Group = types[0].Group } } } if len(actual.Kind) == 0 { return nil, &actual, runtime.NewMissingKindErr(fmt.Sprintf("%#v", unk.TypeMeta)) } if len(actual.Version) == 0 { return nil, &actual, runtime.NewMissingVersionErr(fmt.Sprintf("%#v", unk.TypeMeta)) } return unmarshalToObject(s.typer, s.creater, &actual, into, unk.Raw) }

Decode函数首先验证protoEncodingPrefix前缀，前缀为magic-number特殊标识，其用于标识一个包的完整性，然后验证资源对象是否为proto.Message类型，最后通过proto.Unmarshal反序列化函数进行解码。

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。