【kubeedge源码分析】edgecore源码分析之二edged模块

我会带着你远行 2023-06-14 05:25 19阅读 0赞

# 描述 #

*  Edged：一个运行在 edge 节点的 agent 程序，管理边缘的容器化应用程序，目前支持 docker 运行时，将来会支持更多的运行时，比如 containerd等

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob25nbGluemhhbmc_size_16_color_FFFFFF_t_70][]

*Fig 1: EdgeD Functionalities*

# 1. Register 函数 #

根据 registerModules 中模块注册，将调用 edged 模块，路径 kubeedge/edge/pkg/edged/edged.go，注册 edged 模块

// Register register edged
    func Register() {
    	edged, err := newEdged()
    	if err != nil {
    		klog.Errorf("init new edged error, %v", err)
    		return
    	}
    	core.Register(edged)
    }

###    1.1 newEged 函数 ###

newEged 实例化 eged 对象，并进行初始化，getConfig 函数主要是初始化 kubelet 启动参数，包括 node 名字，namespace imageGC docker endpoint CRI 接口类型，dns，内存等等

包括 pod manager 初始化，imageGC 出发策略等，实例化 eged 并初始化 kubelet 启动参数，根目录默认为 /var/lib/edged

//newEdged creates new edged object and initialises it
    func newEdged() (*edged, error) {
    	conf := getConfig()
    	backoff := flowcontrol.NewBackOff(backOffPeriod, MaxContainerBackOff)
    
    	podManager := podmanager.NewPodManager()
    	policy := images.ImageGCPolicy{
    		HighThresholdPercent: conf.imageGCHighThreshold,
    		LowThresholdPercent:  conf.imageGCLowThreshold,
    		MinAge:               minAge,
    	}
    	// build new object to match interface
    	recorder := record.NewEventRecorder()

**     1.1.1 如果未设置运行时 endpoint，则设置默认未 unix:///var/run/dockershim.sock**

if conf.remoteRuntimeEndpoint != "" {
    	// remoteImageEndpoint is same as remoteRuntimeEndpoint if not explicitly specified
    	if conf.remoteImageEndpoint == "" {
    		conf.remoteImageEndpoint = conf.remoteRuntimeEndpoint
    	}
    }

**     1.1.2 如果运行时 endpoint 为 docker 的情况**

目前支持的 runtime dockershim，初始化 docker 服务，启动 dockershim GRPC 服务，并注册 runtime 和 image服务

//create and start the docker shim running as a grpc server
    if conf.remoteRuntimeEndpoint == DockerShimEndpoint || conf.remoteRuntimeEndpoint == DockerShimEndpointDeprecated {
    	streamingConfig := &streaming.Config{}
    	......
    	if err := server.Start(); err != nil {
    		return nil, err
    	}
    
    }

**     1.1.3 使用 socket 建立 GRPC 连接 runtime 和 image 服务**

containerRefManager := kubecontainer.NewRefManager()
    httpClient := &http.Client{}
    runtimeService, imageService, err := getRuntimeAndImageServices(conf.remoteRuntimeEndpoint, conf.remoteRuntimeEndpoint, conf.RuntimeRequestTimeout)
    if err != nil {
    	return nil, err
    }
    if ed.os == nil {
    	ed.os = kubecontainer.RealOS{}
    }

**     1.1.4 容器运行时 manager 实例化**

复用了 kubelet 的 NewKubeGerericRuntimeManager 函数，实现在 kubernetes/pkg/kubelet/kuberuntime/kuberuntime\_manager.go

var machineInfo cadvisorapi.MachineInfo
    machineInfo.MemoryCapacity = uint64(conf.memoryCapacity)
    containerRuntime, err := kuberuntime.NewKubeGenericRuntimeManager(
    	recorder,
    	ed.livenessManager,
    	"",
    	containerRefManager,
    	&machineInfo,
    	ed,
    	ed.os,
    	ed,
    	httpClient,
    	backoff,
    	false,
    	0,
    	0,
    	false,
    	metav1.Duration{Duration: 100 * time.Millisecond},
    	runtimeService,
    	imageService,
    	ed.clcm.InternalContainerLifecycle(),
    	nil,
    	nil,
    )

**     1.1.5 实例化 cadvisor manager**

复用了 kubelet 使用 cadvisor 模块，实现在 kubernetes/pkg/kubelet/cm/container\_manager\_linux.go

cadvisorInterface, err := cadvisor.New("")
    containerManager, err := cm.NewContainerManager(mount.New(""),
    	cadvisorInterface,
    	cm.NodeConfig{
    		CgroupDriver:       conf.cgroupDriver,
    		SystemCgroupsName:  conf.cgroupDriver,
    		KubeletCgroupsName: conf.cgroupDriver,
    		ContainerRuntime:   conf.runtimeType,
    		KubeletRootDir:     DefaultRootDir,
    	},
    	false,
    	conf.devicePluginEnabled,
    	recorder)

**     1.1.6 实例化 imageGC containerGC manager**

同样使用了 kubelet 中该代码，不再说明，可以参看 imageGC containerGC 文章

[https://blog.csdn.net/zhonglinzhang/article/details/92796057][https_blog.csdn.net_zhonglinzhang_article_details_92796057]

[https://blog.csdn.net/zhonglinzhang/article/details/92771215][https_blog.csdn.net_zhonglinzhang_article_details_92771215]

imageGCManager, err := images.NewImageGCManager(ed.containerRuntime, statsProvider, recorder, nodeRef, policy, conf.PodSandboxImage)
    if err != nil {
    	return nil, fmt.Errorf("failed to initialize image manager: %v", err)
    }
    ed.imageGCManager = imageGCManager
    
    containerGCManager, err := kubecontainer.NewContainerGC(containerRuntime, containerGCPolicy, &containers.KubeSourcesReady{})
    if err != nil {
    	return nil, fmt.Errorf("init Container GC Manager failed with error %s", err.Error())
    }

**     1.1.7 实例化 volume plugin manager**

包裹了 kubernetes/pkg/volume/plugins.go，初始化 plugin

可以参考：[https://blog.csdn.net/zhonglinzhang/article/details/82800287][https_blog.csdn.net_zhonglinzhang_article_details_82800287]

// NewInitializedVolumePluginMgr returns a new instance of volume.VolumePluginMgr
    func NewInitializedVolumePluginMgr(
    	edge *edged,
    	plugins []volume.VolumePlugin) (*volume.VolumePluginMgr, error) {
    	evh := &edgedVolumeHost{
    		edge:            edge,
    		volumePluginMgr: volume.VolumePluginMgr{},
    	}
    
    	if err := evh.volumePluginMgr.InitPlugins(plugins, nil, evh); err != nil {
    		return nil, fmt.Errorf(
    			"Could not initialize volume plugins for KubeletVolumePluginMgr: %v",
    			err)
    	}
    
    	return &evh.volumePluginMgr, nil
    }

###    1.2 注册 edged 模块加入到全局变量 modules map中 ###

// Register register module
    func Register(m Module) {
    	if isModuleEnabled(m.Name()) {
    		modules[m.Name()] = m
    		klog.Infof("Module %v registered", m.Name())
    	} else {
    		disabledModules[m.Name()] = m
    		klog.Warningf("Module %v is not register, please check modules.yaml", m.Name())
    	}
    }

edged 同样实现了  Module 接口

>     // Module interface
>     type Module interface {
>        Name() string
>        Group() string
>        Start()
>        Cleanup()
>     }

# 2. edged 的 name 设置为 edged，group 设置为 edged #

func (e *edged) Name() string {
    	return modules.EdgedModuleName
    }
    
    func (e *edged) Group() string {
    	return modules.EdgedGroup
    }

启动模块分别调用不同的 Start() 方法

# 3. edged 的 Start 函数 #

func (e *edged) Start() {
    	e.metaClient = client.New()
    	var ctx context.Context
    
    	ctx, e.cancel = context.WithCancel(context.Background())
    	// use self defined client to replace fake kube client
    	e.kubeClient = fakekube.NewSimpleClientset(e.metaClient)

###    3.1 NewManager 函数包裹了 kubelet 的 status manager ###

参考 status manager：[https://blog.csdn.net/zhonglinzhang/article/details/75102193][https_blog.csdn.net_zhonglinzhang_article_details_75102193]

//NewManager creates and returns a new manager object
    func NewManager(kubeClient clientset.Interface, podManager podmanager.Manager, podDeletionSafety status.PodDeletionSafetyProvider, metaClient client.CoreInterface) status.Manager {
    	kubeManager := status.NewManager(kubeClient, podManager, podDeletionSafety)
    	return &manager{
    		Manager:           kubeManager,
    		metaClient:        metaClient,
    		podManager:        podManager,
    		apiStatusVersions: make(map[types.UID]*v1.PodStatus),
    	}
    }

###    3.2 initializeModules 函数调用了 kubelet 中 container manager 模块启动服务 ###

func (e *edged) initializeModules() error {
    	node, _ := e.initialNode()
    	if err := e.containerManager.Start(node, e.GetActivePods, edgedutil.NewSourcesReady(), e.statusManager, e.runtimeService); err != nil {
    		klog.Errorf("Failed to start device plugin manager %v", err)
    		return err
    	}
    	return nil
    }

###    3.3 volume manager  ###

不用看了，使用了 kubernetes 的 volume manager 并启动，还有下面的 probe manager

e.volumeManager = volumemanager.NewVolumeManager(
    	true,
    	types.NodeName(e.nodeName),
    	e.podManager,
    	e.statusManager,
    	e.kubeClient,
    	e.volumePluginMgr,
    	e.containerRuntime,
    	e.mounter,
    	e.getPodsDir(),
    	record.NewEventRecorder(),
    	false,
    	false,
    )
    go e.volumeManager.Run(edgedutil.NewSourcesReady(), utilwait.NeverStop)

###    3.4 podAddWorkerRun ###

并发 5 个 worker 处理新增 pod 的请求，调用 consumePodAddition 函数创建 pod，创建 pod 的工作目录

func (e *edged) consumePodAddition(namespacedName *types.NamespacedName) error {
    	podName := namespacedName.Name
    	klog.Infof("start to consume added pod [%s]", podName)
    	pod, ok := e.podManager.GetPodByName(namespacedName.Namespace, podName)
    	if !ok || pod.DeletionTimestamp != nil {
    		return apis.ErrPodNotFound
    	}
    
    	if err := e.makePodDataDirs(pod); err != nil {
    		klog.Errorf("Unable to make pod data directories for pod %q: %v", format.Pod(pod), err)
    		return err
    	}

### 总结： ###

edged 模块的 Register 函数 newEdged 初始化如 kubelet 启动的参数

### *参考：* ###

*    * [https://docs.kubeedge.io/en/latest/modules/edge/edged.html][https_docs.kubeedge.io_en_latest_modules_edge_edged.html]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob25nbGluemhhbmc_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/2019111819405887.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob25nbGluemhhbmc=,size_16,color_FFFFFF,t_70
[https_blog.csdn.net_zhonglinzhang_article_details_92796057]: https://blog.csdn.net/zhonglinzhang/article/details/92796057
[https_blog.csdn.net_zhonglinzhang_article_details_92771215]: https://blog.csdn.net/zhonglinzhang/article/details/92771215
[https_blog.csdn.net_zhonglinzhang_article_details_82800287]: https://blog.csdn.net/zhonglinzhang/article/details/82800287
[https_blog.csdn.net_zhonglinzhang_article_details_75102193]: https://blog.csdn.net/zhonglinzhang/article/details/75102193
[https_docs.kubeedge.io_en_latest_modules_edge_edged.html]: https://docs.kubeedge.io/en/latest/modules/edge/edged.html