support pluggable components

[wenxuwan]

What would you like to be added:

Now dapr support pluggable component pluggable component，I think Layotto needs support too.

Why is this needed:

The ability of pluggable components allows users to use their favorite language and method to develop their own components without having to embed them in Layotto

[github-actions[bot]]

Hi @wenxuwan, Thanks for opening an issue! 🎉

[zhenjunMa]

Cool, this looks like a very valuable feature！

[cyb0225]

Cool! I'm extremely interested in this, and I will review the plugin design of Dapr in the next few days.

[wenxuwan]

Cool! I'm extremely interested in this, and I will review the plugin design of Dapr in the next few days.

Contributions welcome 😊

[github-actions[bot]]

This issue has been automatically marked as stale because it has not had recent activity in the last 30 days. It will be closed in the next 7 days unless it is tagged (pinned, good first issue or help wanted) or other activity occurs. Thank you for your contributions.

[github-actions[bot]]

This issue has been automatically closed because it has not had activity in the last 37 days. If this issue is still valid, please ping a maintainer and ask them to label it as pinned, good first issue or help wanted. Thank you for your contributions.

[wenxuwan]

/reopen

[zhenjunMa]

可以参考文档：https://docs.dapr.io/developing-applications/develop-components/pluggable-components/develop-pluggable/?_gl=1*hx2f1h*_ga*MTMyMjc1NzE4Ny4xNjMwNjM0NzAx*_ga_60C6Q1ETC1*MTY3ODAwOTgwMS44LjAuMTY3ODAwOTgwMS4wLjAuMA..

报名考核标准： 结合Layotto目前的运行逻辑，可以给出设计实现思路

[cyb0225]

@wenxuwan 展示一下最新进展 感觉发在社区里也比较合适, 其他项目的 member 也能看到, 可以讨论出更加适合 layotto 的方案. 这个不算正式的 propusoal, 讨论好后面我会重新整理一份 propusoal 配上简单的 代码demo 发到 ospp. 写的比较粗糙, 有不清楚的地方也可以在周会上讨论一下.

简单总结一下上一封邮件我对 dapr 源码实现的解读及几个 issue 中讨论的结果.

dapr pluggable component 源码实现

跨语言实现 dapr 使用 grpc proto 的特性支持跨语言实现接口. 组件注册 pluggable component 组件使用 uds 的方式开启 grpc 服务, 并将 sock 文件放到 dapr 指定目录里. 一个 grpc server 可以实现一个或多个 dapr 提供的 grpc server 接口. dapr 内部也提供了使用该 grpc conn 实现组件 interface 的封装.
dapr 启动, 读取该目录中的 sock 文件, 并建立 grpc 连接, 使用 proto 反射获取该连接实现了哪些 grpc server 接口, 与 dapr 支持的 grpc server 接口作判断, 如果包含在内, 则将该 grpc conn 对象封装成一个实现了 component interface 的对象, 与其他 build-in 组件一样注册到 factory 中, 组件的 type 是 sock 文件的名称. 最后读取配置文件中关于 component 的信息, 初始化对应的 pluggable 组件.

社区讨论

主要是看看他们为什么这么设计, 有没有其他的方案, 为什么淘汰了.

1. 没使用 wasm 支持夸语言注册. wasm 当前还不成熟, 网络等系统调用不支持. 但是后续 wasm 成熟了会切到 wasm, 无需在网络上多一跳, 性能更好. 相关 issue https://github.com/dapr/dapr/issues/3513#issuecomment-893620786
2. 目前 dapr pluuable component 不支持在 windows 平台开发, 原因是只有部分 windows 版本支持 uds. 相关 issue https://github.com/dapr/dapr/issues/6082.
3. dapr 不关心组件的启动. 且 dapr 的启动的依赖 pluggable component 先启动.
4. 为方便, dapr 并没有对 pluggable 组件新增配置文件类型, 而是与正常的 component 组件使用同一份. 相关 issue https://github.com/dapr/dapr/issues/5210#issuecomment-1253938711

下面是我近期更细致的看了 layotto 源码实现以及相关文档得出来的解决方案.

解决方案

layotto 启动流程主要集中在 main.go NewRuntimeGrpcServer函数里. 主要流程是解析配置, 初始化 runtime 实例, 启动 runtime 实例, 启动实例的时候, 注册了 grpc api 和一堆 build-in 组件, 还有 dapr 实现的组件. 在 runtime Run 函数里, 先使用 option 模式, 将各种 main 函数传入的类型整合到一个对象里. 然后, 在 initRuntime中调用实现组件的初始化,

按照 dapr 的方式注册组件的话, 我们需要在 initRuntime 调用DefaultInitRuntimeStage前先读取执行一段注册 pluggable 组件的逻辑, 然在 initRuntime 里面读取配置文件, 根据 type 获取对应组件的实例并初始化.

func (m *MosnRuntime) Run(opts ...Option) (mgrpc.RegisteredServer, error) {
    // 0. mark already started
    m.started = true
    // 1. init runtime stage
    // prepare runtimeOptions
    o := newRuntimeOptions()
    for _, opt := range opts {
        opt(o)
    }
    // set ErrInterceptor
    if o.errInt != nil {
        m.errInt = o.errInt
    } else {
        m.errInt = func(err error, format string, args ...interface{}) {
            log.DefaultLogger.Errorf("[runtime] occurs an error: "+err.Error()+", "+format, args...)
        }
    }

    // ==============================================
    // 我觉得在这里加一段逻辑是最合适的.
    // 这里将 option 对象也传入, pluggable component 组件注册好后可以直接 append 到 o.service 的各个组件中,
    // 这样也可以保证后面 initRuntime 的代码是不用动的.
    if err := m.initPluggableComponent(o); err != nil {
        return nil, err
    }    
    // ==============================================

    // init runtime with runtimeOptions
    if err := m.initRuntime(o); err != nil {
        return nil, err
    }

在 initPluggableComponent函数里的逻辑就跟 dapr 差不多

const sockDir = "/tmp/layotto/pluggable-component" // socket 文件默认存储路径, 也可以通过环境变量改

var pluggableCompoenntMap = make(map[string]func(*grpc.Conn)) 

func (m *MosnRuntime)initPluggableComponent(o *runtimeOptions) error {
    // 1. 读取 sockDir 下的文件.
    files := readDir(sockDir) 

    for _, v := range files {
        // 2. 建立 grcp conn 连接, 这里只是将组件注册到对应的类型中, 并不会
        conn := grpc.Dial(v)
        defer conn.Close()

        // 3. 使用 grpcreflect 获取反射接口, 获取 service list.
        client := grpcreflect.New(conn)
        services := client.List()

        // 4. 有一个 map 类型叫 pluggableCompoenntMap, 判断这个 service 是不是在这个 map 里面存在, 如果存在则说明也这个接口
        for _, v := range services {
            // 这里返回的 callback 是一个回调函数作用就是将对应的 conn 注册到 o.service 的 Factory 里面
            callback, ok := pluggableCompoenntMap[v]
            if !ok {
                continue
            }
            // 5. 如果这个 conn 连接的对象是在 pluggableCompoenntMap 里面的, 我们就将他注册到 o.service 的对象组件 Factory 里面
            // 组件的 type 就是 sock 文件的名称.

            // 字符串分割获得 compoennt type 类型
            fileName := v.Name()
            componentType := strings.Split(fileName, ".")[0]

            // 获取这个 grpc server 的连接类型,因为我们这里只是注册组件, 而不是真正建立连接, 
            // 所以不能用 conn, 只是传入一个创建连接的函数. 真正的连接在 config 读取到对应 
            // type 的组件才执行这个函数进行 init 连接.
            dialer := socketDialer(v) // 这个函数就是根据 sock 文件名称, 返回一个创建 grpc 连接的回调函数
            callback(o, componentType, dialer)
        }        
    } 

    return nil
}

其他组件要先根据 proto client 实现
// 在各个 compoennt 组件目录下, 下面拿 hello 接口就是 components/hello/grpc.go, 这个文件也可以放在 pkg 中单开一个目录
// 里面有一个结构体, 根据 grpc client conn 实现了 hello 接口
type HelloGRPC struct {
    conn *grpc.Conn
    dialer func()*grpc.Conn
}

var _ HelloService = &HelloGRPC{} 
func (hw *HelloGRPC) Hello(ctx context.Context, req *hello.HelloRequest) (*hello.HelloReponse, error) {
    // 调用 grpc client 接口, 实现 Hello 接口
    hw.conn.SayHello(req)
}

// 根据这里的 dialer 不是直接是一个 grpc conn 连接, 是一个创建 conn 连接的函数, 真正的连接在 Init 函数执行.
func NewHelloGRPC(dialer func() *grpc.Conn) func(l logger.Logger) HelloService {}
func (hw *HelloGRPC) Init(config *hello.HelloConfig) error {
    // 解析配置....
    // 创建 conn 连接
    hw.conn = dialer()
    // 初始化 pluggable component
    hw.conn.Init(config)
}

// 使用 init 函数将该 component 的初始化方式提前注册到 map 中
func init() {
    protoRef = .... // 这个对象是 proto 反射获得的字符串, 与上面注册 pluggable component 反射得到的 proto 应该一致.  
    // 注册回调接口, 每个函数应该都有
    pluggableCompoenntMap[protoRef] = func(o *runtimeOptions, comPonentType string, dialer func()*grpc.Conn) {
        // 封装实现接口
        component := NewHelloGRPC(dialer)
        o.service.hellos = append(o.service.hellos, component)
    }
}

上面的代码只是简单的思路, 可能有很多设计上的缺陷, 比如有些地方没用函数封装, 直接操作 map, 读取部分全局对象没有用 atomic.Value 类型造成并发安全问题, 以及我在注册回调接口为了方便是直接拿 option 对象里的 service Factory 直接 append 组件进去, option 放在这里可能不太合适, 但是是当下最快的解法. 真正实践的话, 可能需要一个中间类型去解耦.

关于 proto 文件 layotto 目前有几个 component 接口定义是采用 dapr 的, dapr 提供了相应的 proto 文件, 且 api 接口也是直接复用 dapr 的, 所以我觉得可以直接复用他的 proto 文件. 我们只需要在相应的 component 文件里添加是上述 hello 组件的 init 函数, 将组件的 proto 类型和回调函数注册到 pluggableCompoenntMap 中即可.

不过 layotto 引用了dapr 的 pubsub, bindings, state 和 secretstores, dapr 只支持前三个的 pluggable component, 也就是说 secretstores 从 grpc 实现 api 接口是需要我们实现的.

layotto 其他类型的接口貌似都是直接用 go interface 定义的, 没有相关的 proto 文件, 我觉得可以参考 dapr 相关 grpc 的定义 https://github.com/dapr/dapr/tree/master/dapr/proto/components/v1, 来设计一份 proto 文件, 再通过 grpc client 去实现封装 component interface, 如上面 hello 组件的伪代码一样.

其他扩展

我不确定这些功能是否有必要添加, 需要一起讨论一下.

1. dapr 在注册组件, 是在配置文件添加一个字段 ignoreError, 为 true 时, 如果该组件初始化失败会忽视错误, dapr 正常启动. 我看 layotto 代码里的实现是遇到错误直接返回的.

[cyb0225]

接上文中的其他扩展, 我们需要 pluggable component 支持一些 build-in 组件支持的功能. 日志: build-in 组件可以直接使用 pkg 里面的 log, 但是由于不同语言之间的打印日志方式是不同的, 所以可能需要做一个打印日志的规范, 然后再做到统一的日志收集. 二种方式可以考虑使用 grpc 调度, 让 layotto 内置的程序来负责日志打印. 但是性能上会有损耗.

Actuator 健康检查: build-in 组件是可以直接使用 pkg 中的 actuator 将组件注册到健康检查中的, 其他语言实现的pluggable component 无法使用 pkg 中的 actuator. 由于 pluggable component 是通过 grpc 的方式与 layotto 进行通信的, 所以可能需要使用类似心跳包的机制来确保 component 是存活的. 并且我认为可以在 proto 层面支持 pluggable component 支持接入 layotto 的 actuator.

[github-actions[bot]]

This issue has been automatically marked as stale because it has not had recent activity in the last 30 days. It will be closed in the next 7 days unless it is tagged (pinned, good first issue or help wanted) or other activity occurs. Thank you for your contributions.

[github-actions[bot]]

This issue has been automatically closed because it has not had activity in the last 37 days. If this issue is still valid, please ping a maintainer and ask them to label it as pinned, good first issue or help wanted. Thank you for your contributions.