K8s源码分析(12)-资源的数据访问层

TA码字

发布于 2021-12-07 14:27:24

6150

发布于 2021-12-07 14:27:24

文章被收录于专栏：TA码字TA码字

上一篇文章中，我们主要介绍了在 kubernetes 中不同版本的资源是如何注册到 schema 对象之中，包括内部版本资源，所有外部版本资源。以及资源的 model 类型的注册，资源的初始化函数（即默认值函数）的注册，资源的 label 转换函数的注册，和内外部版本相互转换函数的注册。在本篇文章里，我们主要来介绍资源的数据访问层。

众所周知，kubernetes 所有资源都是存储在 etcd clsuter 中的，它是一个典型的分布式 kv 数据库，提供存储，查询，更新，监控对象变化的 watch 等操作。在数据访问层的设计上，也是秉承接口和实现的原则，定义接口功能，由相关的具体实现类(etcd3 实现)来实现功能。其相关的图解和源码如下：

// k8s.io/apiserver/pkg/storage/interfaces.go
type Interface interface {
  Versioner() Versioner

  Create(ctx context.Context, key string, obj, out runtime.Object, ttl uint64) error

  Delete(ctx context.Context, key string, out runtime.Object, preconditions *Preconditions,validateDeletion ValidateObjectFunc, cachedExistingObject runtime.Object) error

  Watch(ctx context.Context, key string, opts ListOptions) (watch.Interface, error)

  WatchList(ctx context.Context, key string, opts ListOptions) (watch.Interface, error)

  Get(ctx context.Context, key string, opts GetOptions, objPtr runtime.Object) error

  GetToList(ctx context.Context, key string, opts ListOptions, listObj runtime.Object) error

  List(ctx context.Context, key string, opts ListOptions, listObj runtime.Object) error

  GuaranteedUpdate(ctx context.Context, key string, ptrToType runtime.Object, ignoreNotFound bool, precondtions *Preconditions, tryUpdate UpdateFunc, cachedExistingObject runtime.Object) error

  Count(key string) (int64, error)
}
// k8s.io/apiserver/pkg/storage/etcd3/store.go
type store struct {
  client        *clientv3.Client
  codec         runtime.Codec
  versioner     storage.Versioner
  transformer   value.Transformer
  pathPrefix    string
  watcher       *watcher
  pagingEnabled bool
  leaseManager  *leaseManager
}
func (s *store) Versioner() Versioner{...}
func (s *store) Create(ctx context.Context, key string, obj, out runtime.Object, ttl uint64) error{...}
func (s *store) Delete(ctx context.Context, key string, out runtime.Object, preconditions *Preconditions,validateDeletion ValidateObjectFunc, cachedExistingObject runtime.Object) error{...}
func (s *store) Watch(ctx context.Context, key string, opts ListOptions) (watch.Interface, error){...}
func (s *store) WatchList(ctx context.Context, key string, opts ListOptions) (watch.Interface, error){...}
func (s *store) Get(ctx context.Context, key string, opts GetOptions, objPtr runtime.Object) error{...}
func (s *store) GetToList(ctx context.Context, key string, opts ListOptions, listObj runtime.Object) error{...}
func (s *store) List(ctx context.Context, key string, opts ListOptions, listObj runtime.Object) error{...}
func (s *store) GuaranteedUpdate(ctx context.Context, key string, ptrToType runtime.Object, ignoreNotFound bool, precondtions *Preconditions, tryUpdate UpdateFunc, cachedExistingObject runtime.Object) error{...}
func (s *store) Count(key string) (int64, error){...}

k8s.io/apiserver/pkg/storage/interfaces.go 中定义了资源数据访问层的接口，包括对资源的增删改查以及 watch 等操作。
etcd3.store 结构体实现了上述定义的接口功能，在该结构体中的实现中封装了对象 clientv3.Client 用来完成对 etcd3 集群的访问和对象操作。

另外 kubernetes 也支持 dry run 操作，即运行命令或者执行 API，但是不对资源存储造成影响，所以在以上的接口功能的基础上又定义了 DryRunnableStorage 结构体来实现此功能。其相关的图解和源码如下：

// k8s.io/apiserver/pkg/registry/generic/registry/dryrun.go
type DryRunnableStorage struct {
  Storage storage.Interface
  Codec   runtime.Codec
}

func (s *DryRunnableStorage) Create(ctx context.Context, key string, obj, out runtime.Object, ttl uint64, dryRun bool) error {
  if dryRun {
    if err := s.Storage.Get(ctx, key, storage.GetOptions{}, out); err == nil {
      return storage.NewKeyExistsError(key, 0)
    }
    return s.copyInto(obj, out)
  }
  return s.Storage.Create(ctx, key, obj, out, ttl)
}

func (s *DryRunnableStorage) copyInto(in, out runtime.Object) error {
  var data []byte

  data, err := runtime.Encode(s.Codec, in)
  if err != nil {
    return err
  }
  _, _, err = s.Codec.Decode(data, nil, out)
  if err != nil {
    return err
  }
  return nil
}
// ... other functions

该结构体包含上面介绍的数据访问接口类型的属性来完成基本数据访问操作。
该结构体包含 runtime.Codec 属性来完成必要的序列化和反序列化操作。
以创建操作的 dry run 做为例子，该操作判断资源是否存在，然后内部拷贝对象返回，并不会进行实际的创建操作。

对于数据访问对象，或者 DAO 对象的创建过程的图解和源码如下：

// k8s.io/apiserver/pkg/storage/storagebackend/factory/factory.go
func Create(c storagebackend.Config, newFunc func() runtime.Object) (storage.Interface, DestroyFunc, error) {
  switch c.Type {
  case "etcd2":
    return nil, nil, fmt.Errorf("%v is no longer a supported storage backend", c.Type)
  case storagebackend.StorageTypeUnset, storagebackend.StorageTypeETCD3:
    return newETCD3Storage(c, newFunc)
  default:
    return nil, nil, fmt.Errorf("unknown storage type: %s", c.Type)
  }
}