support pluggable components #888
Comments
|
Hi @wenxuwan, |
|
Cool, this looks like a very valuable feature！ |
|
Cool! I'm extremely interested in this, and I will review the plugin design of Dapr in the next few days. |
Contributions welcome 😊 |
|
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. |
|
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. |
|
/reopen |
|
@wenxuwan 简单总结一下上一封邮件我对 dapr 源码实现的解读及几个 issue 中讨论的结果. dapr pluggable component 源码实现跨语言实现 社区讨论
下面是我近期更细致的看了 layotto 源码实现以及相关文档得出来的解决方案. 解决方案layotto 启动流程主要集中在 main.go NewRuntimeGrpcServer函数里. 主要流程是解析配置, 初始化 runtime 实例, 启动 runtime 实例, 启动实例的时候, 注册了 grpc api 和一堆 build-in 组件, 还有 dapr 实现的组件. 按照 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)
}
}
关于 proto 文件
layotto 其他类型的接口貌似都是直接用 go interface 定义的, 没有相关的 proto 文件, 我觉得可以参考 dapr 相关 grpc 的定义 https://github.com/dapr/dapr/tree/master/dapr/proto/components/v1, 来设计一份 proto 文件, 再通过 grpc client 去实现封装 component interface, 如上面 hello 组件的伪代码一样. 其他扩展
|
|
接上文中的其他扩展, 我们需要 pluggable component 支持一些 build-in 组件支持的功能. Actuator 健康检查: build-in 组件是可以直接使用 pkg 中的 actuator 将组件注册到健康检查中的, 其他语言实现的pluggable component 无法使用 pkg 中的 actuator. 由于 pluggable component 是通过 grpc 的方式与 layotto 进行通信的, 所以可能需要使用类似心跳包的机制来确保 component 是存活的. 并且我认为可以在 proto 层面支持 pluggable component 支持接入 layotto 的 actuator. |
|
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. |
|
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. |
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
The text was updated successfully, but these errors were encountered: