Client Go四种交互模式之 DynamicClient实战案例详解

Client Go四种交互模式之 DynamicClient实战案例详解

引子

Kubernetes赢得了云原生平台之争，同时在绝大多数云原生场景中都凭借其高扩展性担任了重要角色。通过kube-apiserver提供的开放的模块，在不需要切分一个内外部的接口情况下，让我们具备了在同一应用前提下（Controller）与集群及其他系统交互的能力，甚至是自定义的资源描述我们特殊的操作，被称之为Operator Pattern。

尽管我们可以使用HTTP Client与API Server之间进行交互，但这其实并不简单。如果我们只考虑kubernetes的核心resources，那么这种交互方式恐怕不是最完美的，因为许多resources请求后响应回的数据结构和可能操作都是不一样的。作为加强，Kubernetes本身提供了一系列更加方便整合进kubernetes系统的方式，被称之为client-go。

这个项目中被最广泛使用客户端是ClientSet，它是交互客户端的类型之一。这意味着这个接口给kubernetes的每种资源都提供了可用的扩展方法和操作，这也就是为什么但凡要实现扩展，我们首先就应该考虑这种标准操作kubernetes内置接口的方式。

但是这也肯定不能满足全部场景，很多情况下我们要自己声明CRD(Custom Resources Definition)，ClientSet无法满足我们我们的需求了。这个时候，我们就需要DynamicClient，也就是k8s.io/client-go/dynamic.Interface接口，来让我们有能力入侵系统，并实现自定义的功能和操作。

这种方式的好处在于，首先，它避免了对kubernetes系统的强依赖，让我们可以操作自定义resources。如果你想通过其他操作作为构建块来建构自动化或者工作流，比如ExternalDNS,CertManager,Prometheus Operator等，通常就是把这些项目，使用Go语言数据类型，把它们注册到你的Client实例当中，作为依赖。操作的自由度越大，管控的难度显然就会越大，你需要管理这些扩展功能的迭代升级版本，同时又要保持你已经在集群上安装的版本与go.mod文件中声明的版本相一致。

第二，你可以协调多种或未知的resources。当你的operator实现了一个通用逻辑，它就可以与任何通用的Kubernetes resources(RBAC到Pods)交互，custom resources也不在话下，这个时候，dynamicClient就是你唯一的解决的方案。DynamicClient中有些例子对garbage collection controller是有重度依赖的，假如你打算对项目里任意的custom resources提供支持DynamicClient类型扩展也必不可少，比如KubeWatch。

因此，深入client-go项目的组成模块和弄清楚如何使用好它就是要具备的技能。

# 通过DynamicClient做基础操作

以下的代码都预设你在`kubernetes cluster`中进行使用。

许多DynamicClient相关的操作都与TypedClient类似，比如创建新实例，都可以通过给构造方法提供配置文件来完成。

代码语言：javascript复制

func newClient() (dynamic.Interface, error) {
	config, err := rest.InClusterConfig()
	if err != nil {
		return nil, err
	}

	dynClient, err := dynamic.NewForConfig(config)
	if err != nil {
		return nil, err
	}

	return dynClient, nil
}

尽管DynamicClient对于你通过resources声明的意图一无所知，它也没提供类似于CoreV1().Pod类似的方法。所以得告诉它怎么来解析和处理你提供的resources，这时候首先要做的，你需要提供一个schema.GroupVersionResource，它是一个提供了必要信息的golang数据结构，用它可以来构造针对于API Server的HTTP请求。

举例，如果你想实现一个从MongoDB Community Operator拉取到所有Mongodb resources列表的功能，你可以这么做：

代码语言：javascript复制

var monboDBResource = schema.GroupVersionResource{Group: "mongodbcommunity.mongodb.com", Version: "v1", Resource: "mongodbcommunity"}

func ListMongoDB(ctx context.Context, client dynamic.Interface, namespace string) ([]unstructured.Unstructured, error)  {
	list, err := client.Resource(monboDBResource).Namespace(namespace).List(ctx, metav1.ListOptions{})
	if err != nil {
		return nil, err
	}

	return list.Items, nil
}

这里要注意，如果你处理带有namespace的resources，那.Namespace(namespace)就是强制的，而使用空字符就是列出所有命名空间下的。

这段代码片段，我们可以看到DynamicClient的主要组成unstructured.Unstructured。这是一个封装任意JSON结构的特别类型，同时也符合标准的Kubernetes interface，比如runtime.Object，但是最为重要的是它提供了在unstructure package下一系列helpers用于操作这些数据。

扩充下我们上面的例子，如果我们想按一定比例来扩容MongoDB，我们可以像如下方式使用：

代码语言：javascript复制

// ScaleMongoDB changes the number of members by the given proportion,
// which should be 0 =< proportion < 1.
func ScaleMongoDB(ctx context.Context, client dynamic.Interface, name string, namespace string, proportion uint) error {
	if proportion > 1 {
		return fmt.Errorf("proportion should be between 0 =< proportion < 1")
	}

	mongoDBClient := client.Resource(monboDBResource).Namespace(namespace)
	mdb, err := mongoDBClient.Get(ctx, name, metav1.GetOptions{})
	if err != nil {
		return err
	}

	members, found, err := unstructured.NestedInt64(mdb.UnstructuredContent(), "spec", "members")
	if err != nil {
		return err
	}

	if !found {
		return fmt.Errorf("members field not found on MongoDB spec")
	}

	scaled := int(members) * (1   int(proportion))

	patch := []interface{}{
		map[string]interface{}{
			"op":    "replace",
			"path":  "/spec/members",
			"value": scaled,
		},
	}

	payload, err := json.Marshal(patch)
	if err != nil {
		return err
	}

	_, err = mongoDBClient.Patch(ctx, name, types.JSONPatchType, payload, metav1.PatchOptions{})
	if err != nil {
		return err
	}

	return nil
}

这里我们利用unstructured.NestedInt64只访问我们要操作的字段，保证我们与MongoDB CRD间的耦合最小，同时又可以在安全类型之下来操作resource data。

unstructured package有很多像这种的helpers，不只用于读取，也可以在resource上的任意字段上进行写入。

在Kubernetes上执行所有常规操作(get,list,watch,create,patch,delete等)也遵循同样规则：即提供scheme.GroupVersionResource,同时处理unstructured.Unstructured结果。

# 带有DynamicClient的Controller

更高级和对Kubernetes client的使用是创建一个controller，它可对实际集群状态的变化进行响应，把它调整到期望状态。

通常，我们使用Informer，一个由k8s.io/client-go提供的模块，状态发生改变、创建就会在指定类型的client运行一个handler。幸运的是Dynamic package提供了一个我们可以使用的Informer模块。

举例，如果MongoDB被检测到清除了相关PersistentVolumeClaims，我们进行捕获，如下：

代码语言：javascript复制

package main

import (
	"fmt"
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/client-go/dynamic"
	"k8s.io/client-go/dynamic/dynamicinformer"
	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/util/workqueue"
	"time"
)

const maxRetries = 3

var monboDBResource = schema.GroupVersionResource{Group: "mongodbcommunity.mongodb.com", Version: "v1", Resource: "mongodbcommunity"}

type MongoDBController struct {
	informer cache.SharedIndexInformer
	stopper  chan struct{}
	queue    workqueue.RateLimitingInterface
}

func NewMongoDBController(client dynamic.Interface) (*MongoDBController, error) {
	dynInformer := dynamicinformer.NewDynamicSharedInformerFactory(client, 0)
	informer := dynInformer.ForResource(monboDBResource).Informer()
	stopper := make(chan struct{})

	queue := workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter())
	informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
		DeleteFunc: func(obj interface{}) {
			key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
			if err == nil {
				queue.Add(key)
			}
		},
	})

	return &MongoDBController{
		informer: informer,
		queue: queue,
		stopper: stopper,
	}, nil
}

func (m *MongoDBController) Stop() {
	close(m.stopper)
}

func (m *MongoDBController) Run() {
	defer utilruntime.HandleCrash()

	defer m.queue.ShutDown()

	go m.informer.Run(m.stopper)

	if !cache.WaitForCacheSync(m.stopper, m.informer.HasSynced) {
		utilruntime.HandleError(fmt.Errorf("timed out waiting for caches to sync"))
		return
	}

	wait.Until(m.runWorker, time.Second, m.stopper)
}

func (m *MongoDBController) runWorker() {
	for {
		key, quit := m.queue.Get()
		if quit {
			return
		}

		err := m.processItem(key.(string))

		if err == nil {
			m.queue.Forget(key)
		} else if m.queue.NumRequeues(key) < maxRetries {
			m.queue.AddRateLimited(key)
		} else {
			m.queue.Forget(key)
			utilruntime.HandleError(err)
		}

		m.queue.Done(key)
	}
}

func (m *MongoDBController) processItem(mongodb string) error {
	return nil
}

这段绝大多数代码都非常标准，例如使用特定化类型client controller的work queue,informer event handler,item processing。

此外，使用由DynamicClient所提供的解耦方式，在terms complexity上确实让人少了好多头痛事。

# 用DynamicClient进行测试

如果我们想用DynamicClient来进行伸缩，还要让其具备像TypedClient一样易于测试，这样才能保证整体扩展的灵活性和健壮性。

在上面给出的Controller例子中，dynamic package提供了一个等效的伪客户端，允许使用它做存根对象验证及断言行为执行。这里多说两句，如果对于单元测试了解不深入，可能对于我说的概念有点糊涂，模拟通过具有验证调用和交互的能力，而存根通常没有，存根通常用于提供固定数据或控制测试环境，而模拟则用于代码和外部环境间的交互。

代码语言：javascript复制

package main

import (
	"context"
	"k8s.io/apimachinery/pkg/runtime"
	dynamicfake "k8s.io/client-go/dynamic/fake"
)

func TestDynamicClient(t *testing.T) {
	mdb := &unstructured.Unstructured{}
	mdb.SetUnstructuredContent(map[string]interface{}{
		"apiVersion": "mongodbcommunity.mongodb.com/v1",
		"kind": "MongoDBCommunity",
		"metadata": map[string]interface{} {
			"name":      "mongodb-test",
			"namespace": "default",
		},
		"spec": map[string]interface{}{
			"members": 3,
		},
	})

	dynamicClient := dynamicfake.NewSimpleDynamicClient(runtime.NewScheme(), mdb)
  NotifyMongoDBs(context.Background(), dynamicClient)

	AssertActions(t,  dynamicClient.Actions(), []ExpectedAction{
		{
			Verb: "list",
			Namespace: "default",
			Resource: "mongodbcommunity",
		},
	})

}

使用unstructured.Unstructured类型，我们可以像使用YAML文件中相同标签一样来创建Kubernetes对象，但要带有maps。

在执行测试逻辑之后，我们可以使用dynamicClient.Actions()来查看我们代码里执行的所有操作。然而，在每个测试上面手动断言这些actions，通常会导致代码可读性下降和不够完善的断言。

与此同时，我经常使用特定的断言功能AssertActions来校验是不是每个预期的action都可以在可执行actions中被查找到。关键点在于，此功能无法提供一个准确的列表匹配，举例，如果正在使用的client执行了一个删除操作，测试无法进行中断，这里仅有的用于失败的AssertAction条件是列表提供的operation在预期的列表中查询不到。可以在，仅当预期的actions被执行到时，来改变断言功能或者创建一个sibling function来做验证。

尽管当前的实现比较麻烦，但是此功能与DynamicClient或TypedClient均可协调使用。

代码语言：javascript复制

type ExpectedAction struct {
	Verb string
	Name string
	Namespace string
	Resource string

	// Patch action
	PatchType types.PatchType
	PatchPayload []map[string]interface{}
}

func AssertActions(t *testing.T, got []kubetesting.Action, expected []ExpectedAction) {
	if len(expected) > len(got) {
		t.Fatalf("executed actions too short, expected %d, got %d", len(expected), len(got))
		return
	}

	for i, expectedAction := range expected {
		if !AssertExpectedAction(got, expectedAction) {
			t.Fatalf("action %d does not match any of the got actions", i)
		}
	}
}

func AssertExpectedAction(got []kubetesting.Action, expectedAction ExpectedAction) bool {
	for _, gotAction := range got {
		switch expectedAction.Verb {
		case "get":
			getAction, ok := gotAction.(kubetesting.GetAction)
			if !ok {
				continue
			}

			if getAction.GetName() != expectedAction.Name {
				continue
			}

			if !validateNamespaceAndResource(getAction, expectedAction) {
				continue
			}

			return true
		case "list":
			listAction, ok := gotAction.(kubetesting.ListAction)
			if !ok {
				continue
			}

			if !validateNamespaceAndResource(listAction, expectedAction) {
				continue
			}

			return true
		case "watch":
			watchAction, ok := gotAction.(kubetesting.WatchAction)
			if !ok {
				continue
			}

			if !validateNamespaceAndResource(watchAction, expectedAction) {
				continue
			}

			return true
		case "create":
			createAction, ok := gotAction.(kubetesting.CreateAction)
			if !ok {
				continue
			}

			if !validateNamespaceAndResource(createAction, expectedAction) {
				continue
			}

			return true
		case "update":
			updateAction, ok := gotAction.(kubetesting.UpdateAction)
			if !ok {
				continue
			}

			if !validateNamespaceAndResource(updateAction, expectedAction) {
				continue
			}

			return true
		case "delete":
			deleteAction, ok := gotAction.(kubetesting.DeleteAction)
			if !ok {
				continue
			}

			if deleteAction.GetName() != expectedAction.Name {
				continue
			}

			if !validateNamespaceAndResource(deleteAction, expectedAction) {
				continue
			}

			return true
		case "patch":
			patchAction, ok := gotAction.(kubetesting.PatchAction)
			if !ok {
				continue
			}

			if patchAction.GetName() != expectedAction.Name {
				continue
			}

			if !validateNamespaceAndResource(patchAction, expectedAction) {
				continue
			}

			if patchAction.GetPatchType() != expectedAction.PatchType {
				continue
			}

			patchBytes, err := json.Marshal(expectedAction.PatchPayload)
			if err != nil {
				continue
			}

			if !bytes.Equal(patchAction.GetPatch(), patchBytes) {
				continue
			}

			return true
		}
	}

	return false
}

func validateNamespaceAndResource(action kubetesting.Action, expectedAction ExpectedAction) bool {
	return action.GetNamespace() == expectedAction.Namespace && action.GetResource().Resource == expectedAction.Resource
}

这个断言功能允许你随便往里加判断条件，比如验证list/watch限制以及create/update主体。

总结

Kubernetes的生态系统非常丰富，我们时不时会发现这样的宝藏。我强烈大家阅读k8s.io/client-go源码，当然仅仅读它也不够，sigs.k8s.io/controller-runtime项目和Kubernetes Reference API都是极好的。

接口 系统 go client 测试

0 人点赞