golang设计模式-以kubernetes源码为例

// k8s.io/kubernetes/vendor/k8s.io/client-go/tools/cache/store.go
func NewStore(keyFunc KeyFunc) Store {
	return &cache{
		cacheStorage: NewThreadSafeStore(Indexers{}, Indices{}),
		keyFunc:      keyFunc,
	}
}

type cache struct {
	// cacheStorage bears the burden of thread safety for the cache
	cacheStorage ThreadSafeStore
	// keyFunc is used to make the key for objects stored in and retrieved from items, and
	// should be deterministic.
	keyFunc KeyFunc
}

type Store interface {
	Add(obj interface{}) error
	Update(obj interface{}) error
	Delete(obj interface{}) error
	List() []interface{}
	ListKeys() []string
	Get(obj interface{}) (item interface{}, exists bool, err error)
	GetByKey(key string) (item interface{}, exists bool, err error)

	// Replace will delete the contents of the store, using instead the
	// given list. Store takes ownership of the list, you should not reference
	// it after calling this function.
	Replace([]interface{}, string) error
	Resync() error
}

func GetInstance() *singleton {
    if instance == nil {     // <-- Not yet perfect. since it's not fully atomic
        mu.Lock()
        defer mu.Unlock()

        if instance == nil {
            instance = &singleton{}
        }
    }
    return instance
}

type singleton struct {
}

var instance *singleton
var once sync.Once

func GetInstance() *singleton {
    once.Do(func() {
        instance = &singleton{}
    })
    return instance
}

// k8s.io/kubernetes/vendor/k8s.io/apimachinery/pkg/runtime/serializer/codec_factory.go
func NewCodecFactory(scheme *runtime.Scheme) CodecFactory {
	serializers := newSerializersForScheme(scheme, json.DefaultMetaFactory)
	return newCodecFactory(scheme, serializers)
}

func (f CodecFactory) LegacyCodec(version ...schema.GroupVersion) runtime.Codec {
	return versioning.NewDefaultingCodecForScheme(f.scheme, f.legacySerializer, f.universal, schema.GroupVersions(version), runtime.InternalGroupVersioner)
}

//k8s.io/kubernetes/pkg/client/informers/informers_generated/internalversion/factory.go
func NewSharedInformerFactory(client internalclientset.Interface, defaultResync time.Duration) SharedInformerFactory {
	return &sharedInformerFactory{
		client:           client,
		defaultResync:    defaultResync,
		informers:        make(map[reflect.Type]cache.SharedIndexInformer),
		startedInformers: make(map[reflect.Type]bool),
	}
}

// SharedInformerFactory provides shared informers for resources in all known
// API group versions.
type SharedInformerFactory interface {
	internalinterfaces.SharedInformerFactory
	ForResource(resource schema.GroupVersionResource) (GenericInformer, error)
	WaitForCacheSync(stopCh <-chan struct{}) map[reflect.Type]bool

	Admissionregistration() admissionregistration.Interface
	Apps() apps.Interface
	Autoscaling() autoscaling.Interface
	Batch() batch.Interface
	Certificates() certificates.Interface
	Core() core.Interface
	Extensions() extensions.Interface
	Networking() networking.Interface
	Policy() policy.Interface
	Rbac() rbac.Interface
	Scheduling() scheduling.Interface
	Settings() settings.Interface
	Storage() storage.Interface
}

// sharedInformerFactory 是具体的stuct
func (f *sharedInformerFactory) Apps() apps.Interface {
	return apps.New(f)
}

// k8s.io/kubernetes/pkg/controller/client_builder.go

func NewForConfigOrDie(c *rest.Config) *Clientset {
	var cs Clientset
	cs.admissionregistrationV1alpha1 = admissionregistrationv1alpha1.NewForConfigOrDie(c)
	cs.appsV1beta1 = appsv1beta1.NewForConfigOrDie(c)
	cs.appsV1beta2 = appsv1beta2.NewForConfigOrDie(c)
	cs.appsV1 = appsv1.NewForConfigOrDie(c)
	cs.authenticationV1 = authenticationv1.NewForConfigOrDie(c)
	cs.authenticationV1beta1 = authenticationv1beta1.NewForConfigOrDie(c)
	cs.authorizationV1 = authorizationv1.NewForConfigOrDie(c)
	cs.authorizationV1beta1 = authorizationv1beta1.NewForConfigOrDie(c)
	cs.autoscalingV1 = autoscalingv1.NewForConfigOrDie(c)
	cs.autoscalingV2beta1 = autoscalingv2beta1.NewForConfigOrDie(c)
	cs.batchV1 = batchv1.NewForConfigOrDie(c)
	cs.batchV1beta1 = batchv1beta1.NewForConfigOrDie(c)
	cs.batchV2alpha1 = batchv2alpha1.NewForConfigOrDie(c)
	cs.certificatesV1beta1 = certificatesv1beta1.NewForConfigOrDie(c)
	cs.coreV1 = corev1.NewForConfigOrDie(c)
	cs.extensionsV1beta1 = extensionsv1beta1.NewForConfigOrDie(c)
	cs.networkingV1 = networkingv1.NewForConfigOrDie(c)
	cs.policyV1beta1 = policyv1beta1.NewForConfigOrDie(c)
	cs.rbacV1 = rbacv1.NewForConfigOrDie(c)
	cs.rbacV1beta1 = rbacv1beta1.NewForConfigOrDie(c)
	cs.rbacV1alpha1 = rbacv1alpha1.NewForConfigOrDie(c)
	cs.schedulingV1alpha1 = schedulingv1alpha1.NewForConfigOrDie(c)
	cs.settingsV1alpha1 = settingsv1alpha1.NewForConfigOrDie(c)
	cs.storageV1beta1 = storagev1beta1.NewForConfigOrDie(c)
	cs.storageV1 = storagev1.NewForConfigOrDie(c)

	cs.DiscoveryClient = discovery.NewDiscoveryClientForConfigOrDie(c)
	return &cs
}

// k8s.io/kubernetes/vendor/k8s.io/apiserver/pkg/apis/audit/zz_generated.deepcopy.go
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *GroupResources) DeepCopyInto(out *GroupResources) {
	*out = *in
	if in.Resources != nil {
		in, out := &in.Resources, &out.Resources
		*out = make([]string, len(*in))
		copy(*out, *in)
	}
	if in.ResourceNames != nil {
		in, out := &in.ResourceNames, &out.ResourceNames
		*out = make([]string, len(*in))
		copy(*out, *in)
	}
	return
}

// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new GroupResources.
func (in *GroupResources) DeepCopy() *GroupResources {
	if in == nil {
		return nil
	}
	out := new(GroupResources)
	in.DeepCopyInto(out)
	return out
}

// k8s.io/client-go/tools/cache/shared_informer.go
func (s *sharedIndexInformer) AddEventHandlerWithResyncPeriod(handler ResourceEventHandler, resyncPeriod time.Duration) {
    ...
	s.processor.addListener(listener)
    ...
}


// 分发
func (p *sharedProcessor) distribute(obj interface{}, sync bool) {
	p.listenersLock.RLock()
	defer p.listenersLock.RUnlock()

	if sync {
		for _, listener := range p.syncingListeners {
			listener.add(obj)
		}
	} else {
		for _, listener := range p.listeners {
			listener.add(obj)
		}
	}
}

// golang标准库 "go/token" 实现了Iterate方法，但是其实这里更像vistor模式...
func (s *FileSet) Iterate(f func(*File) bool)

// k8s每种对象都对应了一个strategy，决定了update,delete,get 的策略
// 比如 /k8s.io/kubernetes/pkg/registry/core/configmap/strategy.go

// strategy implements behavior for ConfigMap objects
type strategy struct {
	runtime.ObjectTyper
	names.NameGenerator
}

// Strategy is the default logic that applies when creating and updating ConfigMap
// objects via the REST API.
var Strategy = strategy{api.Scheme, names.SimpleNameGenerator}

// Strategy should implement rest.RESTCreateStrategy
var _ rest.RESTCreateStrategy = Strategy

// Strategy should implement rest.RESTUpdateStrategy
var _ rest.RESTUpdateStrategy = Strategy

func (strategy) NamespaceScoped() bool {
	return true
}

func (strategy) PrepareForCreate(ctx genericapirequest.Context, obj runtime.Object) {
	_ = obj.(*api.ConfigMap)
}

func (strategy) Validate(ctx genericapirequest.Context, obj runtime.Object) field.ErrorList {
	cfg := obj.(*api.ConfigMap)

	return validation.ValidateConfigMap(cfg)
}

// Canonicalize normalizes the object after validation.
func (strategy) Canonicalize(obj runtime.Object) {
}

func (strategy) AllowCreateOnUpdate() bool {
	return false
}

func (strategy) PrepareForUpdate(ctx genericapirequest.Context, newObj, oldObj runtime.Object) {
	_ = oldObj.(*api.ConfigMap)
	_ = newObj.(*api.ConfigMap)
}

func (strategy) AllowUnconditionalUpdate() bool {
	return true
}

func (strategy) ValidateUpdate(ctx genericapirequest.Context, newObj, oldObj runtime.Object) field.ErrorList {
	oldCfg, newCfg := oldObj.(*api.ConfigMap), newObj.(*api.ConfigMap)

	return validation.ValidateConfigMapUpdate(newCfg, oldCfg)
}



// k8s.io/kubernetes/pkg/registry/core/configmap/storage/storage.go
// NewREST returns a RESTStorage object that will work with ConfigMap objects.
func NewREST(optsGetter generic.RESTOptionsGetter) *REST {
	store := &genericregistry.Store{
		NewFunc:                  func() runtime.Object { return &api.ConfigMap{} },
		NewListFunc:              func() runtime.Object { return &api.ConfigMapList{} },
		DefaultQualifiedResource: api.Resource("configmaps"),

		CreateStrategy: configmap.Strategy,
		UpdateStrategy: configmap.Strategy,
		DeleteStrategy: configmap.Strategy,
	}
	options := &generic.StoreOptions{RESTOptions: optsGetter}
	if err := store.CompleteWithOptions(options); err != nil {
		panic(err) // TODO: Propagate error up
	}
	return &REST{store}
}

// k8s.io/kubernetes/vendor/github.com/coreos/etcd/raft/node.go
// Node represents a node in a raft cluster.
type Node interface {
	....
	Step(ctx context.Context, msg pb.Message) error

    ....

	// Status returns the current status of the raft state machine.
	Status() Status
    ....
}

// k8s.io/kubernetes/pkg/registry/core/service/portallocator/allocator.go
// NewFromSnapshot allocates a PortAllocator and initializes it from a snapshot.
func NewFromSnapshot(snap *api.RangeAllocation) (*PortAllocator, error) {
	pr, err := net.ParsePortRange(snap.Range)
	if err != nil {
		return nil, err
	}
	r := NewPortAllocator(*pr)
	if err := r.Restore(*pr, snap.Data); err != nil {
		return nil, err
	}
	return r, nil
}

func (r *PortAllocator) Snapshot(dst *api.RangeAllocation) error {
	snapshottable, ok := r.alloc.(allocator.Snapshottable)
	if !ok {
		return fmt.Errorf("not a snapshottable allocator")
	}
	rangeString, data := snapshottable.Snapshot()
	dst.Range = rangeString
	dst.Data = data
	return nil
}

// k8s.io/kubernetes/staging/src/k8s.io/apiserver/pkg/storage/cacher.go

var timerPool sync.Pool

func (c *cacheWatcher) add(event *watchCacheEvent, budget *timeBudget) {
	...

	t, ok := timerPool.Get().(*time.Timer)
	if ok {
		t.Reset(timeout)
	} else {
		t = time.NewTimer(timeout)
	}
	defer timerPool.Put(t)

	....
}

// 这种wrapper 把函数处理交给childrens
// k8s.io/test-infra/velodrome/transform/plugins/multiplexer_wrapper.go
func NewMultiplexerPluginWrapper(plugins ...Plugin) *MultiplexerPluginWrapper {
	return &MultiplexerPluginWrapper{
		plugins: plugins,
	}
}

// 这种warpper 自己处理完了以后再交给clildrens
// k8s.io/test-infra/velodrome/transform/plugins/author_logger_wrapper.go
func NewAuthorLoggerPluginWrapper(plugin Plugin) *AuthorLoggerPluginWrapper {
	return &AuthorLoggerPluginWrapper{
		plugin: plugin,
	}
}

// 自己不处理，交给clildrens
...

func(a *A)Visit(Vistor func(*A) error){
    Vistor(a)
}

// k8s.io/kubernetes/pkg/kubectl/cmd/util/openapi/openapi.go
type SchemaVisitor interface {
	VisitArray(*Array)
	VisitMap(*Map)
	VisitPrimitive(*Primitive)
	VisitKind(*Kind)
	VisitReference(Reference)
}

// Schema is the base definition of an openapi type.
type Schema interface {
	// Giving a visitor here will let you visit the actual type.
	Accept(SchemaVisitor)

	// Pretty print the name of the type.
	GetName() string
	// Describes how to access this field.
	GetPath() *Path
	// Describes the field.
	GetDescription() string
	// Returns type extensions.
	GetExtensions() map[string]interface{}
}

// /k8s.io/kubernetes/pkg/kubectl/resource/builder.go
func (b *Builder) visitByName() *Result {
	...

	visitors := []Visitor{}
	for _, name := range b.names {
		info := NewInfo(client, mapping, selectorNamespace, name, b.export)
		visitors = append(visitors, info)
	}
	result.visitor = VisitorList(visitors)
	result.sources = visitors
	return result
}