clientset, err := kubernetes.NewForConfig(config)
// 获取node 列表 
nodes, err := clientset.CoreV1().Nodes().List(context.TODO(), metav1.ListOptions{})
// 更新
_, err = a.client.CoreV1().Nodes().Update(context.TODO(), newNode, metav1.UpdateOptions{})
// 发送patch 指令更新
patchTemplate := map[string]interface{}{
		"metadata": map[string]interface{}{
			"labels": map[string]interface{}{
				labelkey: labelvaule,
			},
		},
	}
patchdata, _ := json.Marshal(patchTemplate)
_, err := clientset.CoreV1().Nodes().Patch(ctx, Nodes[i].Name, types.StrategicMergePatchType, patchdata, metav1.PatchOptions{})

config,err := clientcmd.BuildConfigFromFlags("",kubeconfig)
clientset, err := kubernetes.NewForConfig(config)
pod,err := clientset
    .CoreV1()       // 选择APIGroupVersion 即 /api/v1
    .Pods("book")   // 命名空间
    .Get("example",metav1.GetOptions{}) // 访问 /api/v1/namespaces/book/pods/example

type CoreV1Interface interface {
	RESTClient() rest.Interface
	ConfigMapsGetter
	EventsGetter
	NamespacesGetter
	NodesGetter
	PersistentVolumesGetter
	PersistentVolumeClaimsGetter
	PodsGetter
	PodTemplatesGetter
	ReplicationControllersGetter
	SecretsGetter
	ServicesGetter
	...
}
type CoreV1Client struct {
	restClient rest.Interface   // 通用的REST 客户端
}

// k8s.io/client-go/deprecated/typed/core/v1/pod.go
type PodInterface interface {
	Create(*v1.Pod) (*v1.Pod, error)
	Update(*v1.Pod) (*v1.Pod, error)
	Delete(name string, options *metav1.DeleteOptions) error
	Get(name string, options metav1.GetOptions) (*v1.Pod, error)
	List(opts metav1.ListOptions) (*v1.PodList, error)
	Watch(opts metav1.ListOptions) (watch.Interface, error)
	Patch(name string, pt types.PatchType, data []byte, subresources ...string) (result *v1.Pod, err error)
	...
}
// pods implements PodInterface
type pods struct {
	client rest.Interface
	ns     string
}
// k8s.io/client-go/rest/client.go
type RESTClient struct {
    base *url.URL
    Client *http.Client
    ...
}
func (c *pods) Get(name string, options metav1.GetOptions) (result *v1.Pod, err error) {
	result = &v1.Pod{}
	err = c.client.Get().       // 新建Request 对象
		Namespace(c.ns).        // 设置Request.namespace
		Resource("pods").       // 设置Request.resource
		Name(name).             // 设置Request.resourceName
		VersionedParams(&options, scheme.ParameterCodec).
		Do(context.TODO()).     // 执行Request.request
		Into(result)
	return
}

// 通过informer 获取node 列表
factory := informers.NewSharedInformerFactory(clientset, 30*time.Second)
nodeInformer := factory.Core().V1().Nodes()
go nodeInformer.Informer().Run(stopCh)
if !cache.WaitForCacheSync(stopCh, nodeInformer.Informer().HasSynced) {
    runtime.HandleError(fmt.Errorf("Timed out waiting for caches to sync"))
    return
}
nodes, err := nodeInformer.Lister().List(labels.NewSelector())

// k8s.io/client-go/tools/cache/reflector.go
type Reflector struct {
  name string
  expectedTypeName string
  expectedType reflect.Type // 放到 Store 中的对象类型
  expectedGVK *schema.GroupVersionKind
  // 与 watch 源同步的目标 Store
  store Store
  // 用来执行 lists 和 watches 操作的 listerWatcher 接口（最重要的）
  listerWatcher ListerWatcher
  WatchListPageSize int64
  ...

// k8s.io/client-go/tools/cache/reflector.go
// Run 函数反复使用 ListAndWatch 函数来获取所有对象和后续的 deltas。
// 当 stopCh 被关闭的时候，Run函数才会退出。
func (r *Reflector) Run(stopCh <-chan struct{}) {
  wait.BackoffUntil(func() {
    if err := r.ListAndWatch(stopCh); err != nil {
      utilruntime.HandleError(err)
    }
  }, r.backoffManager, true, stopCh)
}
func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error {
	var resourceVersion string
	options := metav1.ListOptions{ResourceVersion: r.relistResourceVersion()}
	if err := func() error {
		var list runtime.Object
		listCh := make(chan struct{}, 1)
		go func() {
			pager := pager.New(...)
			pager.PageSize = xx
			list, paginatedResult, err = pager.List(context.Background(), options)
			close(listCh)   //close listCh后，下面的select 会解除阻塞
		}()
		select {
		case <-stopCh:
			return nil
		case r := <-panicCh:
			panic(r)
		case <-listCh:
		}
		...
		r.setLastSyncResourceVersion(resourceVersion)
		return nil
	}()
	// 处理resync 逻辑
	for {
		options = metav1.ListOptions{...}
		w, err := r.listerWatcher.Watch(options)
		if err := r.watchHandler(start, w, &resourceVersion, resyncerrc, stopCh); err != nil {...}
	}
}
func (r *Reflector) watchHandler(start time.Time, w watch.Interface, resourceVersion *string, errc chan error, stopCh <-chan struct{}) error {
loop:
	for {
		select {
		case <-stopCh:
			return errorStopRequested
		case err := <-errc:
			return err
		case event, ok := <-w.ResultChan():
			meta, err := meta.Accessor(event.Object)
			newResourceVersion := meta.GetResourceVersion()
			switch event.Type {
			case watch.Added:
				err := r.store.Add(event.Object)
			case watch.Modified:
				err := r.store.Update(event.Object)
			case watch.Deleted:
				err := r.store.Delete(event.Object)
			case watch.Bookmark:
				// A `Bookmark` means watch has synced here, just update the resourceVersion
			default:
				utilruntime.HandleError(fmt.Errorf("%s: unable to understand watch event %#v", r.name, event))
			}
			*resourceVersion = newResourceVersion
			r.setLastSyncResourceVersion(newResourceVersion)
			if rvu, ok := r.store.(ResourceVersionUpdater); ok {
				rvu.UpdateResourceVersion(newResourceVersion)
			}
		}
	}
	return nil
}

// k8s.io/client-go/tools/cache/delta_fifo.go
type Delta struct {
	Type   DeltaType
	Object interface{}
}
type DeltaFIFO struct {
	items map[string]Deltas //  存储key到元素对象的Map，提供Store能力
    queue []string      // key的队列，提供Queue能力
    ...
}

func (s *sharedIndexInformer) HandleDeltas(obj interface{}) error {
	// from oldest to newest
	for _, d := range obj.(Deltas) {
		switch d.Type {
		case Sync, Added, Updated:
			isSync := d.Type == Sync
			if old, exists, err := s.indexer.Get(d.Object); err == nil && exists {
				if err := s.indexer.Update(d.Object); err != nil {...}
				s.processor.distribute(updateNotification{oldObj: old, newObj: d.Object}, isSync)
			} else {
				if err := s.indexer.Add(d.Object); err != nil {...}
				s.processor.distribute(addNotification{newObj: d.Object}, isSync)
			}
		case Deleted:
			if err := s.indexer.Delete(d.Object); err != nil {...}
			s.processor.distribute(deleteNotification{oldObj: d.Object}, false)
		}
	}
	return nil
}

// k8s.io/client-go/tools/cache/shared_informer.go
type sharedProcessor struct {
	listenersStarted bool
	listenersLock    sync.RWMutex
	listeners        []*processorListener
	syncingListeners []*processorListener
	clock            clock.Clock
	wg               wait.Group
}
func (p *sharedProcessor) distribute(obj interface{}, sync bool) {
    for _, listener := range p.listeners {
        // 加入到processorListener 的addCh 中，随后进入pendingNotifications，因为这里不能阻塞
        listener.add(obj)    
    }
}
// k8s.io/client-go/tools/cache/shared_informer.go
type processorListener struct {
	nextCh chan interface{}
	addCh  chan interface{}
    handler ResourceEventHandler
    pendingNotifications buffer.RingGrowing
    ...
}
func (p *processorListener) add(notification interface{}) {
	p.addCh <- notification
}
func (p *sharedProcessor) run(stopCh <-chan struct{}) {
	func() {
		for _, listener := range p.listeners {
			p.wg.Start(listener.run)   // 消费nextCh     
			p.wg.Start(listener.pop)   // 消费addCh 经过 mq 转到 nextCh
		}
		p.listenersStarted = true
	}()
	...
}

func (p *processorListener) pop() {
	var nextCh chan<- interface{}
	var notification interface{}  // 用来做消息的中转，并在最开始的时候标记pendingNotifications 为空
	for {
        // select case channel 更多是事件驱动的感觉，哪个channel 来数据了或者可以 接收数据了就处理哪个 case 内逻辑
		select {
		case nextCh <- notification:
			// Notification dispatched
			notification, ok = p.pendingNotifications.ReadOne()
			if !ok { // Nothing to pop
				nextCh = nil // Disable this select case
			}
		case notificationToAdd, ok := <-p.addCh:
			if notification == nil { // No notification to pop (and pendingNotifications is empty)
				// Optimize the case - skip adding to pendingNotifications
				notification = notificationToAdd
				nextCh = p.nextCh
			} else { // There is already a notification waiting to be dispatched
				p.pendingNotifications.WriteOne(notificationToAdd)
			}
		}
	}
}
func (p *processorListener) run() {
	stopCh := make(chan struct{})
	wait.Until(func() {
		for next := range p.nextCh {
			switch notification := next.(type) {
			case updateNotification:
				p.handler.OnUpdate(notification.oldObj, notification.newObj)
			case addNotification:
				p.handler.OnAdd(notification.newObj)
			case deleteNotification:
				p.handler.OnDelete(notification.oldObj)
			default:
				utilruntime.HandleError(fmt.Errorf("unrecognized notification: %T", next))
			}
		}
		// the only way to get here is if the p.nextCh is empty and closed
		close(stopCh)
	}, 1*time.Second, stopCh)
}

type Interface interface{
    Stop()
    ResultChan() <- Event
}
type Event struct{
    Type EventType  // ADDED/MODIFIED/DELETED/ERROR
    Object runtime.Object
}

// k8s.io/apimachinery/pkg/runtime/scheme.go
type Scheme struct {
    // a Type is a particular Go struct，比如k8s.io/api/apps/v1.StatefulSet
	gvkToType map[schema.GroupVersionKind]reflect.Type
    typeToGVK map[reflect.Type][]schema.GroupVersionKind
    ...
}
func (s *Scheme) New(kind schema.GroupVersionKind) (Object, error) {
	if t, exists := s.gvkToType[kind]; exists {
		return reflect.New(t).Interface().(Object), nil
	}
    ...
	return nil, NewNotRegisteredErrForKind(s.schemeName, kind)
}

// k8s.io/client-go/dynamic/interface.go
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)
	Delete(ctx context.Context, name string, options metav1.DeleteOptions, subresources ...string) error
	Get(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)
    ...
}
// k8s.io/client-go/dynamic/simple.go
func (c *dynamicResourceClient) Get(ctx context.Context, name string, opts metav1.GetOptions, subresources ...string) (*unstructured.Unstructured, error) {
    // 这里直接拼接了 api resource 的请求路径
	result := c.client.client.Get().AbsPath(append(c.makeURLSegments(name), subresources...)...).SpecificallyVersionedParams(&opts, dynamicParameterCodec, versionV1).Do(ctx)
	retBytes, err := result.Raw()
	uncastObj, err := runtime.Decode(unstructured.UnstructuredJSONScheme, retBytes)
	return uncastObj.(*unstructured.Unstructured), nil
}

// k8s.io/client-go/rest/client.go
type Interface interface {
	GetRateLimiter() flowcontrol.RateLimiter
	Verb(verb string) *Request
	Post() *Request
	Put() *Request
	Patch(pt types.PatchType) *Request
	Get() *Request
	Delete() *Request
	APIVersion() schema.GroupVersion
}
// 你可以直接 根据 rest.Config 初始化RESTClient，从apiServer 中拿到数据
type RESTClient struct {
	// base is the root URL for all invocations of the client
	base *url.URL
	// versionedAPIPath is a path segment connecting the base URL to the resource root
	versionedAPIPath string
	// Set specific behavior of the client.  If not set http.DefaultClient will be used.
    Client *http.Client
    ...
}
// k8s.io/client-go/rest/request.go
func (r *Request) request(ctx context.Context, fn func(*http.Request, *http.Response)) error {
	client := r.c.Client    // http.Client
	// Right now we make about ten retry attempts if we get a Retry-After response.
	retries := 0
	for {
		url := r.URL().String()
		req, err := http.NewRequest(r.verb, url, r.body)
		req = req.WithContext(ctx)
		req.Header = r.headers
		resp, err := client.Do(req)
		done := func() bool {
			...
		}()
		if done {
			return nil
		}
	}
}