在K8s中调度Dapr Wasm程序

用go语言实现一个containerd-shim，支持在k8s中调度wasm程序。通过为shim添加dapr 组件的host abi支持，使被调度的wasm程序可以获得访问dapr组件的能力。

代码开发

在项目wit-dapr中我们实现了dapr wasm abi的go SDK，使wasm程序能够自由使用dapr提供的各种组件能力。在此项目中我们借助于此sdk以及containerd的sdk，实现一个go的containerd shim，来调度dapr style wasm程序的运行。

在后文中的代码分析中我们可以看到，运行一个容器，首先会执行Create，然后会执行Start。在这里Create主要是一些准备工作，不涉及wasm程序的启动，我们主要来看下shim的Start流程，其主要代码如下：

func (p *process) Start(context.Context) (err error) {

  // 读取wasm镜像中的的wasm配置信息，其中包含了wasm程序所需要的dapr组件类型，包括http、state等
	c, err := LoadConfigFromFile(filepath.Join(p.rootfs, "config.yaml"))
	if err != nil {
		return errors.Wrapf(err, "unable to read config file")
	}
  // runtime是对一个wasm程序运行时的封装
	rt := NewRuntime(p.rootfs, c, moduleConfig)
	// 运行
	if err := rt.Run(); err != nil {
		p.mu.Unlock()
		return err
	}
	p.process = rt

	return nil
}

运行wasm程序代码


func (rt *Runtime) Run() error {
  // 根据配置进行wasm的加载，以及根据使用的组件进行导入Host函数
	if err := rt.initFromConfig(); err != nil {
		return err
	}
  // 根据使用的组件不同，http会启动一个server转发请求，其他组件会直接执行wasm程序
	err := rt.run()
	if err != nil {
		return err
	}

	return nil
}

解析配置加载wasm以及进行host 函数导入

func (rt *Runtime) initFromConfig() error {
  // 读取并加载wasm二进制文件
	wasmFile := rt.config.WASMPath
	wasmName := filepath.Base(wasmFile)
	wasmCode, err := os.ReadFile(filepath.Join(rt.rootfs, wasmFile))
	ctx := context.Background()
	runtime := wazero.NewRuntime(ctx)
	// todo detect wasm module valid
	wasmModule, err := runtime.CompileModule(ctx, wasmCode)
	if err != nil {
		return err
	}

	// 链接导入函数（Host提供的函数）
	wasi_snapshot_preview1.Instantiate(ctx, runtime)
	if rt.config.Trigger == "state" {
		host_state.Instantiate(ctx, runtime)
	}
	// ...
	return nil
}

根据组件类型（Trigger）的不同来执行不同的逻辑，其中http、input binding、pubsub的订阅（后两者还未支持）会等待事件触发再执行wasm程序；其他的会直接执行wasm程序。

func (rt *Runtime) run() error {
	if rt.config.Trigger == "http" {
		return rt.runServer()
	} else {
		return rt.runExecWasm()
	}
}

func (rt *Runtime) runServer() error {
	errch := make(chan error, 1)
	log.Printf("Listening on http://localhost:%d", rt.config.Port)
	go utils.StartServer(rt.config.Port, rt.appRouter(), true, false, rt.close, errch)
	go func() {
		err, ok := <-errch
		if ok {
			fmt.Println(err)
		}
		close(rt.finish)
	}()
	return nil
}

func (rt *Runtime) runExecWasm() error {
	if len(rt.config.ComponentPath) > 0 {
		comps, err := LoadComponents(filepath.Join(rt.rootfs, rt.config.ComponentPath))
		if err != nil {
			return err
		}
		host_state.AddComp(comps...)
	}

	ins, err := rt.runtime.InstantiateModule(context.TODO(), rt.module, rt.runtimeConfig)
	if err != nil {
		return err
	}
	ins.Close(context.Background())
	close(rt.finish)
	return nil
}

以上就是核心的代码修改，其全量代码可以在这里获得

测试运行

添加containerd-shim配置

拷贝containerd-shim-dapr-v1到/usr/local/bin/目录下

在/etc/containerd/config.toml中添加以下内容,并重启containerd：

[plugins.cri.containerd.runtimes.dapr]
  runtime_type = "io.containerd.dapr.v1"

containerd会根据runtime_type自动解析对应的二进制文件名，其中io.containerd.dapr.v1就会被解析为containerd-shim-dapr-v1。

重启containerd

sudo systemctl restart containerd

通过ctr运行wasm镜像

ctr run --rm --runtime=io.containerd.dapr.v1 docker.io/docker4zc/dwhttp:v0.0.3 httpwasm

查看containerd-shim日志

如果要查看containerd-shim的日志，可以在 /etc/containerd/config.toml 中设置启用 shim_debug ，containerd会将shim日志转发到自己的日志。您还可以设置 level = "debug" 以启用调试日志。要查看日志，请运行 sudo journalctl -u containerd 。请注意，启用调试日志可能会生成大量的日志数据，因此在生产环境中使用时请谨慎操作，以免对系统性能产生不利影响。下面是启用这两个选项的containerd配置文件：

[debug]
  level = "debug"
[plugins.cri.containerd.runtimes.dapr]
  shim_debug = true
  runtime_type = "io.containerd.dapr.v1"

查看日志

journalctl -u containerd

调试containerd-shim

代码修改

添加以下代码，并在Create函数开始时调用此函数。

// setupDebuggerEvent listens for an user1 signal to allow a debugger such as delve
// to attach for advanced debugging. It's called when handling a ContainerCreate
func setupDebuggerEvent() {
	logrus.Infof("enter setupDebuggerEvent with CONTAINERD_SHIM_WASM_V1_WAIT_DEBUGGER=%s", os.Getenv("CONTAINERD_SHIM_WASM_V1_WAIT_DEBUGGER"))
	logrus.Infof("enter setupDebuggerEvent print enviroment %v", os.Environ())

	if os.Getenv("CONTAINERD_SHIM_WASM_V1_WAIT_DEBUGGER") == "" {
		return
	}

	// wait user1 signal
	c := make(chan os.Signal, 1)
	signal.Notify(c, syscall.SIGUSR1)
	<-c
	logrus.Infof("received SIGUSR1, continue...")
}

修改containerd配置，添加以下内容可以在启动shim时附带此环境变量。可以通过systemctl edit containerd or vim /etc/systemd/system/containerd.service.d/override.conf来修改：

[Service]
Environment="CONTAINERD_SHIM_WASM_V1_WAIT_DEBUGGER=true"

修改完成后记得重启containerd

systemctl daemon-reload
systemctl restart containerd

这样在shim的创建接口被调用的时候，会停下来等我们发送信号，我们可以在这个时候进行调试连接，连上了之后发信号继续执行.

首先通过ps 查看进程。然后dlv attach 到进程，IDE进行remote连接

$ ps -ef | grep shim
root     2939360       1  0 11:24 ?        00:00:00 /usr/local/bin/containerd-shim-dapr-v1 -namespace default -id testwasm -address /run/containerd/containerd.sock
root     2941376 2934065  0 11:35 pts/1    00:00:00 grep --color=auto shim
$ dlv --listen=:40001  --headless=true --log --api-version=2 --check-go-version=false --only-same-user=false attach 2939360

然后我们可以另开一个窗口，向shim进程发送信号，继续执行Create逻辑：

kill -SIGUSR1 2939360

运行wasm container

通过ctr来运行wasm容器，其中docker.io/docker4zc/dwstate:v0.0.3和docker.io/docker4zc/dwhttp:v0.0.3是两个我已经构建好的镜像。其源码分别在state和http

ctr run --rm --runtime=io.containerd.dapr.v1 docker.io/docker4zc/dwstate:v0.0.3 statewasm

对应源码为，进行向state存储中写入key为"z"值为"value"的操作。

func main() {
	state.DaprStateStateInterfaceSet("state", state.DaprStateStateTypesSetRequest{Key: "z", Value: []byte("value")})
	res := state.DaprStateStateInterfaceGet("state", state.DaprStateStateTypesGetRequest{Key: "z"})
	if res.IsOk() {
		fmt.Println(string(res.Val.Data))
	} else {
		fmt.Println(res.Err)
	}
}

由于使用redis的state存储，我们可以查看redis中确实数据写入成功。

运行http wasm容器

ctr run --rm --runtime=io.containerd.dapr.v1 docker.io/docker4zc/dwhttp:v0.0.3 httpwasm

其源码为接受网络请求，并返回Hello from WASM!.


func init() {
	a := &HostImpl{}
	http.SetExportsDaprHttpHttpHandler(a)
}

type HostImpl struct {
}

func (h *HostImpl) HandleHttpRequest(req http.DaprHttpHttpTypesRequest) http.DaprHttpHttpTypesResponse {
	headers := []http.DaprHttpHttpTypesTuple2StringStringT{http.DaprHttpHttpTypesTuple2StringStringT{"Content-Type", "text/plain"}}
	return http.DaprHttpHttpTypesResponse{
		Status:  200,
		Body:    http.Some([]byte("Hello from WASM!")),
		Headers: http.Some(headers),
	}
}