【kubeedge源码分析】edgecore源码分析之一初始化启动流程

水深无声 2023-06-13 11:24 44阅读 0赞

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob25nbGluemhhbmc_size_16_color_FFFFFF_t_70][]

在 edgecore 启动阶段注册模块由 `kubeedge/edge/cmd/app `包下的 `registerModules `函数完成，注册的模块包括：

*  Edged：一个运行在 edge 节点的 agent 程序，管理边缘的容器化应用程序
 *  EdgeHub：边缘的通信接口模块。这是一个 Web 套接字客户端，负责边缘计算与云服务的交互。包括同步云端资源到边缘端，以及报告边缘端 host 和 device 状态到云端 
 *  EdgeController：管理边缘节点。它是一个扩展的 Kubernetes 控制器，管理边缘节点和 pod 元数据，以便数据可以面向特定的边缘节点
 *  EventBus：使用 MQTT 处理内部边缘通信。MQTT 客户端与 MQTT 服务器（mosquitto）交互，为其他组件提供发布和订阅功能
 *  DeviceTwin：处理设备元数据的设备软件镜像。该模块有助于处理设备状态并将其同步到云上。它还为应用程序提供查询接口，它连接到一个轻量级数据库（SQLite）
 *  MetaManager：管理边缘节点上的元数据。这是 Edged 和 Edgehub 之间的消息处理器。负责在轻量级数据库（SQLite）中存储 / 检索元数据

>     // registerModules register all the modules started in edgecore
>     func registerModules() {
>        devicetwin.Register()
>        edged.Register()
>        edgehub.Register()
>        eventbus.Register()
>        edgemesh.Register()
>        metamanager.Register()
>        servicebus.Register()
>        test.Register()
>        dbm.InitDBManager()
>     }

# 1. main 函数 #

一看挺熟悉的格式，kubernetes 代码都是这么搞的，使用了 cobra 框架

func main() {
    	command := app.NewEdgeCoreCommand()
    	logs.InitLogs()
    	defer logs.FlushLogs()
    
    	if err := command.Execute(); err != nil {
    		os.Exit(1)
    	}
    }

###    1.1 NewEdgeCoreCommand 函数 ###

registerModules 注册模块 （第 2 章节讲解）

Run 函数真正的运行阶段，启动各个模块，以及设置优雅的终止程序 （第 3 章节讲解）

// NewEdgeCoreCommand create edgecore cmd
    func NewEdgeCoreCommand() *cobra.Command {
    	opts := options.NewEdgeCoreOptions()
    	cmd := &cobra.Command{
    		Use: "edgecore",
    		......
    		Run: func(cmd *cobra.Command, args []string) {
    			verflag.PrintAndExitIfRequested()
    			flag.PrintFlags(cmd.Flags())
    
    			// To help debugging, immediately log version
    			klog.Infof("Version: %+v", version.Get())
    
    			registerModules()
    			// start all modules
    			core.Run()
    		},

# 2. registerModules  #

通过 registerModules 函数注册相应的模块，下面看看这些模块怎么注册的

// registerModules register all the modules started in edgecore
    func registerModules() {
    	devicetwin.Register()
    	edged.Register()
    	edgehub.Register()
    	eventbus.Register()
    	edgemesh.Register()
    	metamanager.Register()
    	servicebus.Register()
    	test.Register()
    	dbm.InitDBManager()
    }

###    2.1 Register 函数 ###

路径 github.com/kubeedge/beehive/pkg/core/module.go 引入了 beehive 包，使用全局 modules 和 disableModules 需要加载的与禁止的模块，各个模块需要所实现 Module 接口，需要实现下面的方法：

*  Name() string
 *  Group() string
 *  Start()
 *  Cleanup()

var (
    	// Modules map
    	modules         map[string]Module
    	disabledModules map[string]Module
    )
    
    func init() {
    	modules = make(map[string]Module)
    	disabledModules = make(map[string]Module)
    	config.AddConfigChangeCallback(moduleChangeCallback{})
    	eventListener := config.EventListener{Name: "eventListener1"}
    	config.CONFIG.RegisterListener(eventListener, "modules.enabled")
    }

看看其中一个模块的实现，devicetwin.go，路径 kubeege/edge/pkg/devicetwin/devicetwin.go

###    2.2 devicetwin ###

简单明了，调用 beehive 的 Register 函数注册模块

//DeviceTwin the module
    type DeviceTwin struct {
    	HeartBeatToModule map[string]chan interface{}
    	DTContexts        *dtcontext.DTContext
    	DTModules         map[string]dtmodule.DTModule
    	cancel            context.CancelFunc
    }
    
    // Register register devicetwin
    func Register() {
    	dtclient.InitDBTable()
    	dt := DeviceTwin{}
    	core.Register(&dt)
    }

**     2.2.1 DeviceTwin 方法 Name 和 Group**

可以得到 DeviceTwin 模块的 name 为 twin，group 为 twin

//Name get name of the module
    func (dt *DeviceTwin) Name() string {
    	return "twin"
    }
    
    //Group get group of the module
    func (dt *DeviceTwin) Group() string {
    	return modules.TwinGroup
    }

**     2.2.2 DeviceTwin 方法 Start **

deviceTwin 模块和轻量级数据库 sqlite 交互，SyncDeviceFromSqlite 函数从数据库同步数据，应该有表 device，device\_attr，device\_twin

//Start run the module
    func (dt *DeviceTwin) Start() {
    	var ctx context.Context
    	dtContexts, _ := dtcontext.InitDTContext()
    	dt.HeartBeatToModule = make(map[string]chan interface{})
    	dt.DTModules = make(map[string]dtmodule.DTModule)
    	dt.DTContexts = dtContexts
    	ctx, dt.cancel = context.WithCancel(context.Background())
    	err := SyncSqlite(dt.DTContexts)
    	if err != nil {
    		klog.Errorf("Start DeviceTwin Failed, Sync Sqlite error:%v", err)
    		return
    	}
    	dt.runDeviceTwin(ctx)
    }

# 3. StartModules 函数 #

// StartModules starts modules that are registered
    func StartModules() {
    	beehiveContext.InitContext(beehiveContext.MsgCtxTypeChannel)
    
    	modules := GetModules()
    	for name, module := range modules {
    		//Init the module
    		beehiveContext.AddModule(name)
    		//Assemble typeChannels for sendToGroup
    		beehiveContext.AddModuleGroup(name, module.Group())
    		go module.Start()
    		klog.Infof("Starting module %v", name)
    	}
    }

###    3.1 InitContext 函数 ###

beehive 采用 golang 的 channel 方式实现模块间通讯，未来可能有 unix socke t的通讯方式。kubernetes CSI 等模块间使用的时 socket 通讯方式，初始化全局上下文 context，context 包含两个对象 moduleContext，messageContext

// Context is global context object
    type Context struct {
       moduleContext  ModuleContext
       messageContext MessageContext
    }

// InitContext gets global context instance
    func InitContext(contextType string) {
    	once.Do(func() {
    		context = &Context{}
    		switch contextType {
    		case MsgCtxTypeChannel:
    			channelContext := NewChannelContext()
    			context.messageContext = channelContext
    			context.moduleContext = channelContext
    		default:
    			klog.Fatalf("Do not support context type:%s", contextType)
    		}
    	})
    }

**     3.1.1 moduleContext 模块管理**

*  `AddModule`为模块创建默认 buffer 大小为 1024 的channel，`channels`成员中将模块名字映射到该 channel，用于“单播” 
 *  `AddModuleGroup `将一个模块对应的 channel 添加到所属 group，也就是将 `ChannelContext `的 `typeChannels[group][module]`设置为模块对应channel，用于“组播”

//constants for channel context
    const (
       ChannelSizeDefault = 1024
    
       MessageTimeoutDefault = 30 * time.Second
    
       TickerTimeoutDefault = 20 * time.Millisecond
    )

3.1.1.1 结构体 ChannelContext

channel channels,typeChannels,anonChannels，定义的组如下：

const (
       // BusGroup group
       BusGroup = "bus"
       // HubGroup group
       HubGroup = "hub"
       // TwinGroup group
       TwinGroup = "twin"
       // MetaGroup group
       MetaGroup = "meta"
       //EdgedGroup group
       EdgedGroup = "edged"
       // UserGroup is ServiceBus group
       UserGroup = "user"
       // MeshGroup group
       MeshGroup = "mesh"
    )

// ChannelContext is object for Context channel
    type ChannelContext struct {
    	//ConfigFactory goarchaius.ConfigurationFactory
    	channels     map[string]chan model.Message
    	chsLock      sync.RWMutex
    	typeChannels map[string]map[string]chan model.Message
    	typeChsLock  sync.RWMutex
    	anonChannels map[string]chan model.Message
    	anonChsLock  sync.RWMutex
    }

//ModuleContext is interface for context module management
    type ModuleContext interface {
    	AddModule(module string)
    	AddModuleGroup(module, group string)
    	Cleanup(module string)
    }

**     3.1.2 messageContext 消息同步**

实现了模块间消息的同步与异步发送

//MessageContext is interface for message syncing
    type MessageContext interface {
    	// async mode
    	Send(module string, message model.Message)
    	Receive(module string) (model.Message, error)
    	// sync mode
    	SendSync(module string, message model.Message, timeout time.Duration) (model.Message, error)
    	SendResp(message model.Message)
    	// group broadcast
    	SendToGroup(moduleType string, message model.Message)
    	SendToGroupSync(moduleType string, message model.Message, timeout time.Duration) error
    }

### edgecore 启动流程总结： ###

*      调用  registerModules 函数主册各个模块
 *      启动各个模块调用 `StartModules`函数创建模块间通讯机制，并启动协程调用每个模块的`Start`函数

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob25nbGluemhhbmc_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20191117113350868.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob25nbGluemhhbmc=,size_16,color_FFFFFF,t_70