1. 概述

用于动态生成informer/lister/client等

2. 类图

3. 具体代码解析

3.1 interface.go

// ResourceInterface的工厂模式，方法Resource可以获取对应gvr的ResourceInterface接口
type Interface interface {Resource(resource schema.GroupVersionResource) NamespaceableResourceInterface
}// 操作对应obj的接口
type ResourceInterface interface {Create(ctx context.Context, obj *unstructured.Unstructured, options metav1.CreateOptions, subresources ...string) (*unstructured.Unstructured, error)Update(ctx context.Context, obj *unstructured.Unstructured, options metav1.UpdateOptions, subresources ...string) (*unstructured.Unstructured, error)UpdateStatus(ctx context.Context, obj *unstructured.Unstructured, options metav1.UpdateOptions) (*unstructured.Unstructured, error)Delete(ctx context.Context, name string, options metav1.DeleteOptions, subresources ...string) errorDeleteCollection(ctx context.Context, options metav1.DeleteOptions, listOptions metav1.ListOptions) errorGet(ctx context.Context, name string, options metav1.GetOptions, subresources ...string) (*unstructured.Unstructured, error)List(ctx context.Context, opts metav1.ListOptions) (*unstructured.UnstructuredList, error)Watch(ctx context.Context, opts metav1.ListOptions) (watch.Interface, error)Patch(ctx context.Context, name string, pt types.PatchType, data []byte, options metav1.PatchOptions, subresources ...string) (*unstructured.Unstructured, error)Apply(ctx context.Context, name string, obj *unstructured.Unstructured, options metav1.ApplyOptions, subresources ...string) (*unstructured.Unstructured, error)ApplyStatus(ctx context.Context, name string, obj *unstructured.Unstructured, options metav1.ApplyOptions) (*unstructured.Unstructured, error)
}// 限定了namespace的ResourceInterface
type NamespaceableResourceInterface interface {Namespace(string) ResourceInterfaceResourceInterface
}

3.2 simple.go

相关函数：

// ConfigFor 返回提供的配置的副本，并设置了适当的动态客户端默认值。
func ConfigFor(inConfig *rest.Config) *rest.Config {config := rest.CopyConfig(inConfig)config.AcceptContentTypes = "application/json"config.ContentType = "application/json"config.NegotiatedSerializer = basicNegotiatedSerializer{} // this gets used for discovery and error handling typesif config.UserAgent == "" {config.UserAgent = rest.DefaultKubernetesUserAgent()}return config
}// // NewForConfig 创建新的动态客户端或返回错误。
func New(c rest.Interface) *DynamicClient {return &DynamicClient{client: c}
}//NewForConfigOrDie为给定的配置创建一个新的DynamicClient，如果配置中有错误，就会panic。
func NewForConfigOrDie(c *rest.Config) *DynamicClient {ret, err := NewForConfig(c)if err != nil {panic(err)}return ret
}// NewForConfig创建新的动态客户端或返回错误。
// NewForConfig相当于NewForConfigAndClient（c，httpClient），其中httpClient是使用rest.HTTPClientFor（c）生成的。
func NewForConfig(inConfig *rest.Config) (*DynamicClient, error) {config := ConfigFor(inConfig)httpClient, err := rest.HTTPClientFor(config)if err != nil {return nil, err}return NewForConfigAndClient(config, httpClient)
}// NewForConfigAndClient为给定的配置和http客户端创建一个新的动态客户
func NewForConfigAndClient(inConfig *rest.Config, h *http.Client) (*DynamicClient, error) {config := ConfigFor(inConfig)// for serializing the optionsconfig.GroupVersion = &schema.GroupVersion{}config.APIPath = "/if-you-see-this-search-for-the-break"restClient, err := rest.RESTClientForConfigAndClient(config, h)if err != nil {return nil, err}return &DynamicClient{client: restClient}, nil
}

dynamicResourceClient结构体定义及其方法分析：

// DynamicClient定义，实质上就是rest接口
type DynamicClient struct {client rest.Interface
}var _ Interface = &DynamicClient{}type dynamicResourceClient struct {client    *DynamicClientnamespace stringresource  schema.GroupVersionResource
}func (c *DynamicClient) Resource(resource schema.GroupVersionResource) NamespaceableResourceInterface {return &dynamicResourceClient{client: c, resource: resource}
}// 实现了NamespaceableResourceInterface接口的Namespace方法
func (c *dynamicResourceClient) Namespace(ns string) ResourceInterface {ret := *cret.namespace = nsreturn &ret
}func (c *dynamicResourceClient) Create(ctx context.Context, obj *unstructured.Unstructured, opts metav1.CreateOptions, subresources ...string) (*unstructured.Unstructured, error) {// 把Unstructured obj（内部其实是map类型）转化为字节数组outBytes, err := runtime.Encode(unstructured.UnstructuredJSONScheme, obj)if err != nil {return nil, err}name := ""// 如果subresources不为空，会校验object的nameif len(subresources) > 0 {accessor, err := meta.Accessor(obj)if err != nil {return nil, err}name = accessor.GetName()if len(name) == 0 {return nil, fmt.Errorf("name is required")}}if err := validateNamespaceWithOptionalName(c.namespace, name); err != nil {return nil, err}// 如果name不为空，再创建时则使用该name，否则随机生成。// （这里只是建立了tcp长链接和解码相应，具体的执行逻辑还是在k8s api-server中调用k8s api的etcd相关接口）result := c.client.client.Post().AbsPath(append(c.makeURLSegments(name), subresources...)...).SetHeader("Content-Type", runtime.ContentTypeJSON).Body(outBytes).SpecificallyVersionedParams(&opts, dynamicParameterCodec, versionV1).Do(ctx)if err := result.Error(); err != nil {return nil, err}// 返回原始结果（字节数组数据）retBytes, err := result.Raw()if err != nil {return nil, err}// 使用json编解码器  解码原始数据，得到一个Unstructured类型的数据uncastObj, err := runtime.Decode(unstructured.UnstructuredJSONScheme, retBytes)if err != nil {return nil, err}return uncastObj.(*unstructured.Unstructured), nil
}

3.3 scheme.go

scheme.go 实现了kubernetes/apimachinery的序列化和编解码

var watchScheme = runtime.NewScheme(// 供basicNegotiatedSerializer来获取gvk/类型转化/创建obj
var basicScheme = runtime.NewScheme()// 为上面ResourceClient删除操作的Options创建删除CodecFactory（编解码器工厂），提供Scheme参数
var deleteScheme = runtime.NewScheme()// 为上面ResourceClient各种操作创建参数CodecFactory（编解码器工厂），提供Scheme参数
var parameterScheme = runtime.NewScheme()// 创建删除CodecFactory（编解码器工厂）,编解码DeleteOptions对象
var deleteOptionsCodec = serializer.NewCodecFactory(deleteScheme)// 创建参数CodecFactory（编解码器工厂）,
// 编解码ListOptions/GetOptions/CreateOptions/UpdateOptions/PatchOptions对象，作为请求参数
var dynamicParameterCodec = runtime.NewParameterCodec(parameterScheme)// 定义gv
var versionV1 = schema.GroupVersion{Version: "v1"}// 添加ListOptions/GetOptions/DeleteOptions/CreateOptions/UpdateOptions/PatchOptions，初始化各个schema
func init() {metav1.AddToGroupVersion(watchScheme, versionV1)metav1.AddToGroupVersion(basicScheme, versionV1)metav1.AddToGroupVersion(parameterScheme, versionV1)metav1.AddToGroupVersion(deleteScheme, versionV1)
}

结构体及其相关方法定义：

// 实现了NegotiatedSerializer接口,根据content判断，以何种方式编解码obj
type basicNegotiatedSerializer struct{}// 实现NegotiatedSerializer的SupportedMediaTypes方法,自定义可以处理的MediaType，并定义各种MediaType的Serializer
func (s basicNegotiatedSerializer) SupportedMediaTypes() []runtime.SerializerInfo {return []runtime.SerializerInfo{{MediaType:        "application/json",MediaTypeType:    "application",MediaTypeSubType: "json",EncodesAsText:    true,Serializer:       json.NewSerializer(json.DefaultMetaFactory, unstructuredCreater{basicScheme}, unstructuredTyper{basicScheme}, false),PrettySerializer: json.NewSerializer(json.DefaultMetaFactory, unstructuredCreater{basicScheme}, unstructuredTyper{basicScheme}, true),StreamSerializer: &runtime.StreamSerializerInfo{EncodesAsText: true,Serializer:    json.NewSerializer(json.DefaultMetaFactory, basicScheme, basicScheme, false),Framer:        json.Framer,},},}
}// 实现NegotiatedSerializer的EncoderForVersion方法,获取一个版本化的Encoder（编码为指定版本）
func (s basicNegotiatedSerializer) EncoderForVersion(encoder runtime.Encoder, gv runtime.GroupVersioner) runtime.Encoder {return runtime.WithVersionEncoder{Version:     gv,Encoder:     encoder,ObjectTyper: unstructuredTyper{basicScheme},}
}// 实现NegotiatedSerializer的DecoderToVersion方法, 获取原始的decoder
func (s basicNegotiatedSerializer) DecoderToVersion(decoder runtime.Decoder, gv runtime.GroupVersioner) runtime.Decoder {return decoder
}
// unstructuredCreater 包装了ObjectCreater（用于New一个gvk对应的obj），
// 如果有err，那么和ObjectCreater的区别是返回了空的obj，而err为空
type unstructuredCreater struct {nested runtime.ObjectCreater
}// 实现了ObjectCreater接口的New方法
func (c unstructuredCreater) New(kind schema.GroupVersionKind) (runtime.Object, error) {out, err := c.nested.New(kind)if err == nil {return out, nil}out = &unstructured.Unstructured{}out.GetObjectKind().SetGroupVersionKind(kind)return out, nil
}
// unstructuredTyper 包装了ObjectTyper（用于获取obj对应的gvk列表），
// 如果有err，那么和ObjectTyper的区别是会判断是否是Unstructured类型且obj对应的gvk是否为空type unstructuredTyper struct {nested runtime.ObjectTyper
}// 实现了ObjectTyper接口的ObjectKinds方法
func (t unstructuredTyper) ObjectKinds(obj runtime.Object) ([]schema.GroupVersionKind, bool, error) {kinds, unversioned, err := t.nested.ObjectKinds(obj)if err == nil {return kinds, unversioned, nil}// 判断obj是否是Unstructured类型且对应的gvk是否为空if _, ok := obj.(runtime.Unstructured); ok && !obj.GetObjectKind().GroupVersionKind().Empty() {return []schema.GroupVersionKind{obj.GetObjectKind().GroupVersionKind()}, false, nil}return nil, false, err
}func (t unstructuredTyper) Recognizes(gvk schema.GroupVersionKind) bool {return true
}

3.4 dynamiclister包

3.4.1 interface.go

interface.go 定义了获取（从indexer（缓存）中get/list）obj的接口

// Lister 获取资源和获取NamespaceLister。
type Lister interface {// List 列出索引器（缓存）中的所有资源。List(selector labels.Selector) (ret []*unstructured.Unstructured, err error)// Get 从索引器（缓存）中检索具有给定名称的资源Get(name string) (*unstructured.Unstructured, error)// Namespace 根据指定namespace返回一个对象，该对象可以列出和获取给定命名空间中的资源Namespace(namespace string) NamespaceLister
}// NamespaceLister 获取命名空间下的资源。类似于controller-runtime分析client包的Reader接口
type NamespaceLister interface {// List 列出索引器（缓存）中给定命名空间的所有资源List(selector labels.Selector) (ret []*unstructured.Unstructured, err error)// Get 从索引器（缓存）中检索给定命名空间和名称的资源。Get(name string) (*unstructured.Unstructured, error)
}

3.4.2 lister.go

// dynamicLister 实现了 Lister 接口。
type dynamicLister struct {// 索引器（缓存）indexer cache.Indexer// 索引器对应的资源gvr，该索引器只存储该gvr对应的资源gvr     schema.GroupVersionResource
}// New 返回一个新的 Lister.
func New(indexer cache.Indexer, gvr schema.GroupVersionResource) Lister {return &dynamicLister{indexer: indexer, gvr: gvr}
}// List 列出索引器中的所有资源。
func (l *dynamicLister) List(selector labels.Selector) (ret []*unstructured.Unstructured, err error) {// 该方法到对应包在做具体分析，用来获取符合selector对应添加的itemerr = cache.ListAll(l.indexer, selector, func(m interface{}) {// 符合添加的item会追加到retret = append(ret, m.(*unstructured.Unstructured))})return ret, err
}// Get 从索引器中检索具有给定名称的资源
func (l *dynamicLister) Get(name string) (*unstructured.Unstructured, error) {obj, exists, err := l.indexer.GetByKey(name)if err != nil {return nil, err}if !exists {return nil, errors.NewNotFound(l.gvr.GroupResource(), name)}return obj.(*unstructured.Unstructured), nil
}// Namespace 返回一个对象，该对象可以从给定的命名空间中列出和获取资源.
func (l *dynamicLister) Namespace(namespace string) NamespaceLister {return &dynamicNamespaceLister{indexer: l.indexer, namespace: namespace, gvr: l.gvr}
}// dynamicNamespaceLister 实现了 NamespaceLister 接口。相比dynamicLister多了namespace属性，用来限定namespace
type dynamicNamespaceLister struct {indexer   cache.Indexernamespace stringgvr       schema.GroupVersionResource
}// List 列出索引器中给定命名空间的所有资源。
func (l *dynamicNamespaceLister) List(selector labels.Selector) (ret []*unstructured.Unstructured, err error) {err = cache.ListAllByNamespace(l.indexer, l.namespace, selector, func(m interface{}) {ret = append(ret, m.(*unstructured.Unstructured))})return ret, err
}// Get 从索引器中检索给定命名空间和名称的资源。
func (l *dynamicNamespaceLister) Get(name string) (*unstructured.Unstructured, error) {// 注意： 这里可以看到indexer中items的存放,当namespace不为空时，key是${namespace}/${name}obj, exists, err := l.indexer.GetByKey(l.namespace + "/" + name)if err != nil {return nil, err}if !exists {return nil, errors.NewNotFound(l.gvr.GroupResource(), name)}return obj.(*unstructured.Unstructured), nil
}

3.4.3 shim.go

// dynamicListerShim 实现了 cache.GenericLister
// 包装了Lister接口，只是List把返回slice中Unstructured对象变为object对象
type dynamicListerShim struct {lister Lister
}// NewRuntimeObjectShim 为 Lister 返回一个新的shim。
// 它包装 Lister 以便它实现 cache.GenericLister 接口func NewRuntimeObjectShim(lister Lister) cache.GenericLister { return &dynamicListerShim{lister: lister} }
func NewRuntimeObjectShim(lister Lister) cache.GenericLister {return &dynamicListerShim{lister: lister}
}// List 将返回跨命名空间的所有对象
func (s *dynamicListerShim) List(selector labels.Selector) (ret []runtime.Object, err error) {objs, err := s.lister.List(selector)if err != nil {return nil, err}ret = make([]runtime.Object, len(objs))// 返回slice中Unstructured对象变为Object对象for index, obj := range objs {ret[index] = obj}return ret, err
}// Get会假设name=key去尝试检索  
func (s *dynamicListerShim) Get(name string) (runtime.Object, error) {return s.lister.Get(name)
}// 获取限定命名空间的Lister
func (s *dynamicListerShim) ByNamespace(namespace string) cache.GenericNamespaceLister {return &dynamicNamespaceListerShim{namespaceLister: s.lister.Namespace(namespace),}
}// dynamicNamespaceListerShim 实现了 NamespaceLister 接口。
// 它包装了 NamespaceLister 以便它实现 cache.GenericNamespaceLister 接口
type dynamicNamespaceListerShim struct {namespaceLister NamespaceLister
}// List 将返回此命名空间中的所有对象
func (ns *dynamicNamespaceListerShim) List(selector labels.Selector) (ret []runtime.Object, err error) {objs, err := ns.namespaceLister.List(selector)if err != nil {return nil, err}ret = make([]runtime.Object, len(objs))for index, obj := range objs {ret[index] = obj}return ret, err
}// Get 将尝试按命名空间和名称检索
func (ns *dynamicNamespaceListerShim) Get(name string) (runtime.Object, error) {return ns.namespaceLister.Get(name)
}

3.5 dynamicinformer包

3.5.1 interface.go

接口 定义了获取Informer的方法，等待缓存同步的方法，启动所有informer的方法

// Dynamic SharedInformerFactory 为动态客户端提供对共享informer和lister的访问
type DynamicSharedInformerFactory interface {// 启动所有informer的方法Start(stopCh <-chan struct{})// 获取Informer的方法ForResource(gvr schema.GroupVersionResource) informers.GenericInformer// 等待缓存同步的方法WaitForCacheSync(stopCh <-chan struct{}) map[schema.GroupVersionResource]bool
}// TweakListOptionsFunc 定义了一个辅助函数的签名，想要为 API 提供更多的列表选项
type TweakListOptionsFunc func(*metav1.ListOptions)

3.5.2 informer.go

公共函数：

// NewDynamicSharedInformerFactory 为所有命名空间构造一个 dynamicSharedInformerFactory 的新实例。
func NewDynamicSharedInformerFactory(client dynamic.Interface, defaultResync time.Duration) DynamicSharedInformerFactory {return NewFilteredDynamicSharedInformerFactory(client, defaultResync, metav1.NamespaceAll, nil)
}// NewFilteredDynamicSharedInformerFactory 构造了一个 dynamicSharedInformerFactory 的新实例。 
// 通过此工厂获得的lister将受到此处指定的相同过滤器的约束。
func NewFilteredDynamicSharedInformerFactory(client dynamic.Interface, defaultResync time.Duration, namespace string, tweakListOptions TweakListOptionsFunc) DynamicSharedInformerFactory {return &dynamicSharedInformerFactory{client:           client,defaultResync:    defaultResync,namespace:        namespace,informers:        map[schema.GroupVersionResource]informers.GenericInformer{},startedInformers: make(map[schema.GroupVersionResource]bool),tweakListOptions: tweakListOptions,}
}// NewFilteredDynamicInformer 为动态类型构造一个新的 Informer。
func NewFilteredDynamicInformer(client dynamic.Interface, gvr schema.GroupVersionResource, namespace string, resyncPeriod time.Duration, indexers cache.Indexers, tweakListOptions TweakListOptionsFunc) informers.GenericInformer {return &dynamicInformer{gvr: gvr,informer: cache.NewSharedIndexInformer(&cache.ListWatch{ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {if tweakListOptions != nil {tweakListOptions(&options)}return client.Resource(gvr).Namespace(namespace).List(context.TODO(), options)},WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {if tweakListOptions != nil {tweakListOptions(&options)}return client.Resource(gvr).Namespace(namespace).Watch(context.TODO(), options)},},&unstructured.Unstructured{},resyncPeriod,indexers,),}
}

结构体定义私有属性和方法：

type dynamicSharedInformerFactory struct {// 构建ListWatch接口使用，为后来构建reflector，执行listWatch监控api resource提供clientclient        dynamic.Interface// 同步周期，informer同步deltaFIFO中数据到listener中的chan中defaultResync time.Duration// 命名空间namespace     stringlock      sync.Mutex// 缓存informer到map中informers map[schema.GroupVersionResource]informers.GenericInformer// startInformers 用于跟踪哪些 Informers 已启动。这允许安全地多次调用 Start()。startedInformers map[schema.GroupVersionResource]booltweakListOptions TweakListOptionsFunc
}// 实现DynamicSharedInformerFactory的ForResource方法
func (f *dynamicSharedInformerFactory) ForResource(gvr schema.GroupVersionResource) informers.GenericInformer {f.lock.Lock()defer f.lock.Unlock()key := gvr// 获取缓存map中的informerinformer, exists := f.informers[key]if exists {return informer}// 不存在就创建informer = NewFilteredDynamicInformer(f.client, gvr, f.namespace, f.defaultResync, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}, f.tweakListOptions)f.informers[key] = informerreturn informer
}// 实现SharedInformerFactory的Start方法，初始化所有请求的informers.
func (f *dynamicSharedInformerFactory) Start(stopCh <-chan struct{}) {f.lock.Lock()defer f.lock.Unlock()// 遍历所有informerfor informerType, informer := range f.informers {// 判断该informer是否已经启动if !f.startedInformers[informerType] {// 启动informergo informer.Informer().Run(stopCh)// 设置对应gvr的informer已经启动f.startedInformers[informerType] = true}}
}// 实现SharedInformerFactory的WaitForCacheSync方法，等待所有启动的informer的缓存同步。func (f *dynamicSharedInformerFactory) WaitForCacheSync(stopCh <-chan struct{}) map[schema.GroupVersionResource]bool {// 定义map，用于接收所有已经启动的informerinformers := func() map[schema.GroupVersionResource]cache.SharedIndexInformer {f.lock.Lock()defer f.lock.Unlock()informers := map[schema.GroupVersionResource]cache.SharedIndexInformer{}for informerType, informer := range f.informers {if f.startedInformers[informerType] {informers[informerType] = informer.Informer()}}return informers}()// 定义map，用于接收所有同步完成的informerres := map[schema.GroupVersionResource]bool{}// 遍历已经启动的所有informerfor informType, informer := range informers {// 执行同步方法// (1) 如果informer中controller为空，返回false，// (2) 如果informer.controller的queue还没有调用过Add/Update/Delete/AddIfNotPresent或者queue的initialPopulationCount != 0 (队列中还有数据)，返回falseres[informType] = cache.WaitForCacheSync(stopCh, informer.HasSynced)}return res
}// 动态Informer结构体，包装了SharedIndexInformer和gvr
type dynamicInformer struct {informer cache.SharedIndexInformergvr      schema.GroupVersionResource
}// 实现GenericInformer的Informer方法
func (d *dynamicInformer) Informer() cache.SharedIndexInformer {return d.informer
}// 实现GenericInformer的Lister方法，使用dynamicInformer的indexer和gvr构造以Lister
func (d *dynamicInformer) Lister() cache.GenericLister {return dynamiclister.NewRuntimeObjectShim(dynamiclister.New(d.informer.GetIndexer(), d.gvr))
}

4. client-go dynamic代码示例

package mainimport ("context""flag""fmt"apiv1 "k8s.io/api/core/v1"metav1 "k8s.io/apimachinery/pkg/apis/meta/v1""k8s.io/apimachinery/pkg/runtime""k8s.io/apimachinery/pkg/runtime/schema""k8s.io/client-go/dynamic""k8s.io/client-go/tools/clientcmd""k8s.io/client-go/util/homedir""log""path/filepath")func main() {var kubeconfig *stringif home := homedir.HomeDir(); home != "" {kubeconfig = flag.String("kubeconfig", filepath.Join(home,".kube", "config_local"), "(optional) absolute path to the kubeconfig file")} else {kubeconfig = flag.String("kubeconfig","","absolute path to the kubeconfig file")}flag.Parse()config, err := clientcmd.BuildConfigFromFlags("", *kubeconfig)if err != nil {log.Fatal(err)}dynamicClient, err := dynamic.NewForConfig(config)if err != nil {log.Fatal(err)}// dynamicClient的唯一关联方法所需的入参gvr := schema.GroupVersionResource{Version: "v1", Resource: "pods"}// 使用dynamicClient的查询列表方法，查询指定namespace下的所有pod，// 注意此方法返回的数据结构类型是UnstructuredListunstructObj, err := dynamicClient.Resource(gvr).Namespace("kube-system").List(context.TODO(),metav1.ListOptions{Limit: 100})if err != nil {log.Fatal(err)}podList := &apiv1.PodList{}err = runtime.DefaultUnstructuredConverter.FromUnstructured(unstructObj.UnstructuredContent(), podList)if err != nil {log.Fatal(err)}fmt.Println("namespace\t Status\t\t name")for _, item := range podList.Items {fmt.Printf("%v\t %v\t %v\n",item.Namespace,item.Status.Phase,item.Name,)}}