什么是registry

在Kubernetes中，registry可以对ETCD中的Kubernetes各类型进行增删查改操作，并对外提供了生成apiGroupInfo的方法。这里的registry包含了两层含义，Storage和Registry，如NodeStorage和NodeRegistry。Storage提供了通用的增删查改方法，Regsitry是对Storage的封装，提供了具体类型的增删查改方法。registry定义在/pkg/registry目录下，主要包含：

  • apps: 定义statefulsets相关的registry信息；
  • authentication: 定义tokenreview相关的registry信息；
  • authorization: 定义localsubjectaccessreview, selfsubjectaccessreview, subjectaccessreview相关的registry信息；
  • autoscaling: 定义horizontalpodautoscaler相关的registry信息；
  • batch: 定义scheduledjobs相关的registry信息；
  • certificates: 定义certificatesigningrequests相关的registry信息；
  • core: 定义componentstatus, configmap, controller, endpoint, event, limitrange, namespace, node, persistentvolume, persistentvolumeclaim, pod, podtemplate, rangeallocation, resourcequota, secret, service, serviceaccount的registry信息；
  • extensions: 定义replicationcontrollers, daemonset, deployment, ingress, networkpolicy, podsecuritypolicy, replicaset等相关的registry信息；
  • generic: 公共模块，对storage模块进行封装；
  • policy: 定义poddisruptionbudgets相关的registry信息；
  • rbac: 定义cluserrole, role等相关的registry信息；
  • storage: 定义storageclass相关的registry信息。

以上就是/pkg/registry目录下的主要目录，每个子目录都对应一个APIGroup，如core对应的是v1，每个子目录下都有个rest的目录，其中core目录实现了NewLegacyRESTStorage()方法生成对应的apiGroupInfo，其他目录实现了NewRESTStorage()方法生成对应的apiGroupInfo。关于apiGroupInfo将在以后分析，现在只需要知道apiGroupInfo是联系registry和apiserver的通道。

在NewLegacyRESTStorage()方法中，会调用NewStorage()生成storage，如nodeStorage等，进一步地生成restStorageMap，然后生成apiGroupInfo。

其他APIGroup使用NewRESTStorage()，会调用NewREST()生成storage，如rolesStorage。

我们以core包下的node为例来看下NewStorage()的实现，以rbac包下的role为例来看下NewREST()的实现。

nodeStorage

node定义在/pkg/registry/core/node目录下。我们先来看下/pkg/registry/core/rest/storage_core.go中的NewLegacyRESTStorage()是如何生成nodeStorage和NodeRegistry的，代码如下：

1
2
3
nodeStorage, err := nodeetcd.NewStorage(restOptionsGetter(api.Resource("nodes")), c.KubeletClientConfig, c.ProxyTransport)
......
restStorage.NodeRegistry = node.NewRegistry(nodeStorage.Node)

restOptionsGetter()

我们遇到的第一个问题就是restOptionsGetter()函数定义在哪。在NewLegacyRESTStorage()中，restOptionsGetter是以参数的形式传入的; NewLegacyRESTStorage()会被/pkg/master/master.go中的InstallLegacyAPI()方法调用，restOptionsGetter仍是以参数的形式传入；InstallLegacyAPI()会被/pkg/master/master.go中的New()调用，在New()上有restOptionsFactory的定义，可以看出，storageDecorator为registry.StorageWithCacher或generic.UndecoratedStorage。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
//***生成restOptionsFactory***//
restOptionsFactory := restOptionsFactory{
	deleteCollectionWorkers: c.DeleteCollectionWorkers,
	enableGarbageCollect ion: c.GenericConfig.EnableGarbageCollection,
	storageFactory:          c.StorageFactory,
}
//***赋值storageDecorator***//
if c.EnableWatchCache {
	restOptionsFactory.storageDecorator = registry.StorageWithCacher
} else {
	restOptionsFactory.storageDecorator = generic.UndecoratedStorage
}
......
	
m.InstallLegacyAPI(c.Config, restOptionsFactory.NewFor, legacyRESTStorageProvider)

所以restOptionsGetter就是restOptionsFactory的NewFor()方法，定义在/pkg/master/master.go中：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
func (f restOptionsFactory) NewFor(resource unversioned.GroupResource) generic.RESTOptions {
	//***获取storageConfig***//
	storageConfig, err := f.storageFactory.NewConfig(resource)
	if err != nil {
		glog.Fatalf("Unable to find storage destination for %v, due to %v", resource, err.Error())
	}
	return generic.RESTOptions{
		StorageConfig: storageConfig,
		Decorator:               f.storageDecorator,
		DeleteCollectionWorkers: f.deleteCollectionWorkers,
		EnableGarbageCollection: f.enableGarbageCollection,
		ResourcePrefix:          f.storageFactory.ResourcePrefix(resource),
	}
}

NewFor()会生成一个generic.RESTOptions。这里还需要注意下，generic.RESTOptions中的Decorator一般情况下就是generic.UndercoratedStorage()方法。

NewStorage()

node的NewStorage()定义在/pkg/registry/core/node/etcd/etcd.go中，主要流程如下：

  1. 生成ETCD helper(现在只需知道ETCD helper是对ETCD client的封装即可)，使用opts.Decorator()生成，即registry.StorageWithCacher()或generic.UndercoratedStorage()方法，这些将在以后分析；
  2. 生成store，store为genericStore，store中包含ETCD helper；
  3. 生成nodeREST;
  4. 生成NodeStorage。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
// NewStorage returns a NodeStorage object that will work against nodes.
func NewStorage(opts generic.RESTOptions, kubeletClientConfig client.KubeletClientConfig, proxyTransport http.RoundTripper) (*NodeStorage, error) {
	prefix := "/" + opts.ResourcePrefix
	newListFunc := func() runtime.Object { return &api.NodeList{} }
	//***生成etcd client***//
	storageInterface, dFunc := opts.Decorator(
		opts.StorageConfig,
		cachesize.GetWatchCacheSizeByResource(cachesize.Nodes),
		&api.Node{},
		prefix,
		node.Strategy,
		newListFunc,
		node.NodeNameTriggerFunc)
	//***store为registry包中的store，定义在/pkg/registry/generic/registry/store.go中***//
	store := &registry.Store{
		NewFunc:     func() runtime.Object { return &api.Node{} },
		NewListFunc: newListFunc,
		KeyRootFunc: func(ctx api.Context) string {
			return prefix
		},
		//***KeyFunc***//
		KeyFunc: func(ctx api.Context, name string) (string, error) {
			return registry.NoNamespaceKeyFunc(ctx, prefix, name)
		},
		ObjectNameFunc: func(obj runtime.Object) (string, error) {
			return obj.(*api.Node).Name, nil
		},
		PredicateFunc:           node.MatchNode,
		QualifiedResource:       api.Resource("nodes"),
		EnableGarbageCollection: opts.EnableGarbageCollection,
		DeleteCollectionWorkers: opts.DeleteCollectionWorkers,
		CreateStrategy: node.Strategy,
		UpdateStrategy: node.Strategy,
		DeleteStrategy: node.Strategy,
		ExportStrategy: node.Strategy,
		Storage:     storageInterface,
		DestroyFunc: dFunc,
	}
	statusStore := *store
	statusStore.UpdateStrategy = node.StatusStrategy
	// Set up REST handlers
	//***生成nodeREST***//
	nodeREST := &REST{Store: store, proxyTransport: proxyTransport}
	//***生成statusREST***//
	statusREST := &StatusREST{store: &statusStore}
	//***生成proxyREST***//
	proxyREST := &noderest.ProxyREST{Store: store, ProxyTransport: proxyTransport}
	// Build a NodeGetter that looks up nodes using the REST handler
	nodeGetter := client.NodeGetterFunc(func(nodeName string) (*api.Node, error) {
		obj, err := nodeREST.Get(api.NewContext(), nodeName)
		if err != nil {
			return nil, err
		}
		node, ok := obj.(*api.Node)
		if !ok {
			return nil, fmt.Errorf("unexpected type %T", obj)
		}
		return node, nil
	})
	connectionInfoGetter, err := client.NewNodeConnectionInfoGetter(nodeGetter, kubeletClientConfig)
	if err != nil {
		return nil, err
	}
	nodeREST.connection = connectionInfoGetter
	proxyREST.Connection = connectionInfoGetter
	return &NodeStorage{
		Node:   nodeREST,
		Status: statusREST,
		Proxy:  proxyREST,
		KubeletConnectionInfo: connec   tionInfoGetter,
	}, nil
}

可以看出，NodeStorage本质是一个genericStore，关于Store，将在以后分析，现在只需知道Store中定义有Create(), Delete(), Get(), List(), Watch()等方法。

NewRegistry()

现在再来看来registry。registry是对storage的封装，可以更方便的增删查改ETCD中的对象。APIServer中的restHandler使用人是更通用的Storage中的方法对对象进行增删查改，感觉只有ApiServer中会使用registry，特别是在APIServer的controller(实现创建default和kube-system空间及default下kubernetes服务)中。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
type Registry interface {
	ListNodes(ctx api.Context, options *api.ListOptions) (*api.NodeList, error)
	CreateNode(ctx api.Context, node *api.Node) error
	UpdateNode(ctx api.Context, node *api.Node) error
	GetNode(ctx api.Context, nodeID string) (*api.Node, error)
	DeleteNode(ctx api.Context, nodeID string) error
	WatchNodes(ctx api.Context, options *api.ListOptions) (watch.Interface, error)
}
// storage puts strong typing around storage calls
type storage struct {
	rest.StandardStorage
}
// NewRegistry returns a new Registry interface for the given Storage. Any mismatched
// types will panic.
func NewRegistry(s rest.StandardStorage) Registry {
	return &storage{s}
}

NewRegistry()可以生成一个nodeRegistry。nodeRegistry就实现了node的增删查改的方法，如ListNodes(), CreateNode(), UpdateNode(), GetNode(), DeleteNode()和WatchNodes()等方法。

rolesStorage

role定义在/pkg/registry/rbac/role目录下，我们先来看/pkg/registry/rbac/rest/storage_rbac.go，storage_rabc.go中定义有NewRESTStorage()方法生成apiGroupInfo，NewRESTStorage()会调用v1alpha1Storage()方法，在v1alpha1Storage()中有：

1
rolesStorage := roleetcd.NewREST(restOptionsGetter(rbac.Resource("roles")))

restOptionsGetter已经分析过。

NewREST()

rolesStorage的NewREST()定义在/pkg/registry/rbac/role/etcd/etcd.go中：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
// NewREST returns a RESTStorage object that will work against Role objects.
func NewREST(opts generic.RESTOptions) *REST {
	prefix := "/" + opts.ResourcePrefix
	newListFunc := func() runtime.Object { return &rbac.RoleList{} }
	storageInterface, dFunc := opts.Decorator(
		opts.StorageConfig,
		cachesize.GetWatchCacheSizeByResource(cachesize.Roles),
		&rbac.Role{},
		prefix,
		role.Strategy,
		newListFunc,
		storage.NoTriggerPublisher,
	)
	store := &registry.Store{
		NewFunc:     func() runtime.Object { return &rbac.Role{} },
		NewListFunc: newListFunc,
		KeyRootFunc: func(ctx api.Context) string {
			return registry.NamespaceKeyRootFunc(ctx, prefix)
		},
		KeyFunc: func(ctx api.Context, id string) (string, error) {
			return registry.NamespaceKeyFunc(ctx, prefix, id)
		},
		ObjectNameFunc: func(obj runtime.Object) (string, error) {
			return obj.(*rbac.Role).Name, nil
		},
		PredicateFunc:           role.Matcher,
		QualifiedResource:       rbac.Resource("roles"),
		EnableGarbageCollection: opts.EnableGarbageCollection,
		DeleteCollectionWorkers: opts.DeleteCollectionWorkers,
		CreateStrategy: role.Strategy,
		UpdateStrategy: role.Strategy,
		DeleteStrategy: role.Strategy,
		Storage:     storageInterface,
		DestroyFunc: dFunc,
	}
	return &REST{store}
}

可以看出，roleStorage本质是一个genericStore。

NewRegistry()

NewRegistry()可以生成一个rolesRegistry，rolesRegistry定义有ListRoles(), CreateRole(), UpdateRole(), GetRole(), DeleteRole(), WatchRoles()等方法。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
// Registry is an interface for things that know how to store Roles.
type Registry interface {
	ListRoles(ctx api.Context, options *api.ListOptions) (*rbac.RoleList, error)
	CreateRole(ctx api.Context, role *rbac.Role) error
	UpdateRole(ctx api.Context, role *rbac.Role) error
	GetRole(ctx api.Context, name string) (*rbac.Role, error)
	DeleteRole(ctx api.Context, name string) error
	WatchRoles(ctx api.Context, options *api.ListOptions) (watch.Interface, error)
}
// storage puts strong typing around storage calls
type storage struct {
	rest.StandardStorage
}
// NewRegistry returns a new Registry interface for the given Storage. Any mismatched
// types will panic.
func NewRegistry(s rest.StandardStorage) Registry {
	return &storage{s}
}

总结

本次分析介绍了Storage和Registry，主要分析了nodeStorage和nodeRegistry，rolesStorage和rolesRegsitry。其中NodeStorage是对generic.Storage的封装，generic.Storage中定义了对ETCD中对象的通用操作；NodeRegistry是对NodeStorage的封装，NodeRegistry定义了关于node的增删查改操作；rolesStorage和rolesRegistry也一样，可以用来操作role。