k8s源码篇k8s数据访问1之接口定义

参考

K8s源码分析(1-资源的数据访问层K8s源码分析(1-资源的服务层接口定义

资源的数据

资源的数据访问层

etcd访问

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

k8s.io/apiserver/pkg/storage/interfaces.go中定义了资源数据访问层的接口，包括对资源的增删改查以及watch等操作。etcdstore结构体实现了上述定义的接口功能，在该结构体中的实现中封装了对象clientvClient用来完成对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){...}

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

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

// k8s.io/apiserver/pkg/registry/generic/registry/dryrun.go
type DryRunnableStorage struct {
  Storage storage.Interface  // 上面的 etcd 访问接口
  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

数据访问对象

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

数据访问对象的创建是基于配置的，由storagebackenConfig决定。对于目前的kubernetes版本来说，只支持etcd3的数据访问层。对于etcd2或者其它类型存储的数据访问层是不支持的。

// 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)
  }
}

资源的服务层接口定义

在数据服务层的设计上，也是秉承接口和实现的原则，定义接口功能，由相关的具体实现类来实现功能。kubernetes在服务层上定义分为两大类:

一类是增删改查类接口，定义增删改查watch等操作另一类是操作的策略类型接口，用来定义资源在增删改查等不同操作中的逻辑。

增删改查接口

对于增删改查类接口定义，解和源码如下：

对于不同的增删改查以及watch操作均有不同类型的独立接口与之对应定义，例如Storage，Getter，Lister，Creater，Updater，Watcher等等。有聚合类型接口StandardStorage封装了一些独立接口功能。

// staging/src/k8s.io/apiserver/pkg/registry/rest/rest.go
type Storage interface {
    New() runtime.Object
}
type Scoper interface {
  NamespaceScoped() bool
}
type Getter interface {
  Get(ctx context.Context, name string, options *metav1.GetOptions) (runtime.Object, error)
}
type Lister interface {
  NewList() runtime.Object
  List(ctx context.Context, options *metainternalversion.ListOptions) (runtime.Object, error)
  TableConvertor
}
type Watcher interface {
  Watch(ctx context.Context, options *metainternalversion.ListOptions) (watch.Interface, error)
}
type Creater interface {
  New() runtime.Object
  Create(ctx context.Context, obj runtime.Object, createValidation ValidateObjectFunc, options *metav1.CreateOptions) (runtime.Object, error)
}
type Updater interface {
  New() runtime.Object
  Update(ctx context.Context, name string, objInfo UpdatedObjectInfo, createValidation ValidateObjectFunc, updateValidation ValidateObjectUpdateFunc, forceAllowCreate bool, options *metav1.UpdateOptions) (runtime.Object, bool, error)
}
type CreaterUpdater interface {
  Creater
  Update(ctx context.Context, name string, objInfo UpdatedObjectInfo, createValidation ValidateObjectFunc, updateValidation ValidateObjectUpdateFunc, forceAllowCreate bool, options *metav1.UpdateOptions) (runtime.Object, bool, error)
}
type GracefulDeleter interface {
  Delete(ctx context.Context, name string, deleteValidation ValidateObjectFunc, options *metav1.DeleteOptions) (runtime.Object, bool, error)
}
type CollectionDeleter interface {
  DeleteCollection(ctx context.Context, deleteValidation ValidateObjectFunc, options *metav1.DeleteOptions, listOptions *metainternalversion.ListOptions) (runtime.Object, error)
}
type StandardStorage interface {
  Getter
  Lister
  CreaterUpdater
  GracefulDeleter
  CollectionDeleter
  Watcher
}

操作策略接口

对于操作类策略接口定义，解和源码如下：