【kubernetes/k8s源码分析】 rook operator agent源码分析

电玩女神 2022-02-05 07:47 224阅读 0赞

# 启动命令 #

rook ceph agent

### 启动默认参数 ###

--help=false, --log-flush-frequency=5s, --log-level=INFO

挂载路经：/usr/libexec/kubernetes/kubelet-plugins/volume/exec

> 每个 k8s node 中都启动 rook agent，接受 rook operator 管理，每个agent都配置一个Flexvolume插件，该插件与Kubernetes的卷控制器框架集成在一起。处理节点上所需的所有存储操作，例如附加网络存储设备，安装卷和格式化文件系统

所有代码一个模式，直接分析startAgent函数

# 1. 创建客户端，创建上下文 #

clientset, apiExtClientset, rookClientset, err := rook.GetClientset()
    if err != nil {
    	rook.TerminateFatal(fmt.Errorf("failed to get k8s client. %+v", err))
    }
    
    logger.Infof("starting rook ceph agent")
    context := &clusterd.Context{
    	Executor:              &exec.CommandExecutor{},
    	ConfigDir:             k8sutil.DataDir,
    	NetworkInfo:           clusterd.NetworkInfo{},
    	Clientset:             clientset,
    	APIExtensionClientset: apiExtClientset,
    	RookClientset:         rookClientset,
    }

###   1.1 attachement.New ###

创建CRD controller**管理CRD资源**，实现了Attachment接口

*  Create(volumeAttachment \*rookalpha.Volume) error
 *  Get(namespace, name string) (\*rookalpha.Volume, error)
 *  List(namespace string) (\*rookalpha.VolumeList, error) Update(volumeAttachment \*rookalpha.Volume) error
 *  Delete(namespace, name string) error

volumeAttachmentController, err := attachment.New(a.context)
    if err != nil {
    	return fmt.Errorf("failed to create volume attachment controller: %+v", err)
    }

###   1.2 NewVolumeManager ###

NewVolumeManager对于ceph卷创建attacher

// NewVolumeManager create attacher for ceph volumes
    func NewVolumeManager(context *clusterd.Context) (*VolumeManager, error) {
    	vm := &VolumeManager{
    		context:          context,
    		devicePathFinder: &devicePathFinder{},
    	}
    	err := vm.Init()
    	return vm, err
    }

**    1.2.1 Init函数**

使用系统调用cat /lib/modules/3.10.0-957.el7.x86\_64/modules.builtin | grep rbd，rbd没有加载内和模块

// check if the rbd is a builtin kernel module, if it is then we don't need to load it manually
    in, err := sys.IsBuiltinKernelModule(rbdKernelModuleName, vm.context.Executor)
    if err != nil {
    	return err
    }
    if in == true {
    	logger.Noticef("volume manager is a builtin kernel module, don't load it manually")
    	return nil
    }

**   1.2.2 执行系统调用命令 modinfo -F parm rbd**

\# modinfo -F parm rbd  
       single\_major:Use a single major number for all rbd devices (default: true) (bool)

func CheckKernelModuleParam(name, param string, executor pkgexec.Executor) (bool, error) {
    	out, err := executor.ExecuteCommandWithOutput(false, "check kmod param", "modinfo", "-F", "parm", name)
    	if err != nil {
    		return false, fmt.Errorf("failed to check for %s module %s param: %+v", name, param, err)
    	}
    
    	result := Grep(out, fmt.Sprintf("^%s", param))
    	return result != "", nil
    }

**   1.2.3 执行系统调用命令 odprobe rbd single\_major=Y **

func LoadKernelModule(name string, options []string, executor pkgexec.Executor) error {
    	if options == nil {
    		options = []string{}
    	}
    
    	args := append([]string{name}, options...)
    
    	if err := executor.ExecuteCommand(false, fmt.Sprintf("modprobe %s", name), "modprobe", args[:]...); err != nil {
    		return fmt.Errorf("failed to load kernel module %s: %+v", name, err)
    	}
    
    	return nil
    }

# 2. 创建controller来处理flexvolume driver的事件 #

**   Controller实现了 Attach Detach方法**，FlexvolumeServer 包裹了 Controller

flexvolumeController := flexvolume.NewController(a.context, volumeAttachmentController, volumeManager, mountSecurityMode)
    
    flexvolumeServer := flexvolume.NewFlexvolumeServer(
    	a.context,
    	flexvolumeController,
    )

# 3. FlexvolumeServer Start函数 #

注册rook，rook-ceph的driver name，监听socker与driver通信，路径为/flexmnt，宿主机路径/usr/libexec/kubernetes/kubelet-plugins/volume/exec

创建 socket

flexDriverVendors := []string{flexvolume.FlexvolumeVendor, flexvolume.FlexvolumeVendorLegacy}
    for i, vendor := range flexDriverVendors {
    	if i > 0 {
    		// Wait before the next driver is registered. In 1.11 and newer there is a timing issue if flex drivers are registered too quickly.
    		// See https://github.com/rook/rook/issues/1501 and https://github.com/kubernetes/kubernetes/issues/60694
    		time.Sleep(time.Second)
    	}
    
    	err = flexvolumeServer.Start(vendor, driverName)
    	if err != nil {
    		return fmt.Errorf("failed to start flex volume server %s/%s, %+v", vendor, driverName, err)
    	}
    
    	// Wait before the next driver is registered
    	time.Sleep(time.Second)
    
    	// Register drivers both with the name of the namespace and the name "rook"
    	// for the volume plugins not based on the namespace.
    	err = flexvolumeServer.Start(vendor, flexvolume.FlexDriverName)
    	if err != nil {
    		return fmt.Errorf("failed to start flex volume server %s/%s. %+v", vendor, flexvolume.FlexDriverName, err)
    	}
    }

# 4. NewClusterController函数 #

创建一个controller实例，watch集群事件，rook-ceph名下的CephCluster资源

// NewClusterController creates a new instance of a ClusterController
    func NewClusterController(context *clusterd.Context, flexvolumeController flexvolume.VolumeController,
    	volumeAttachment attachment.Attachment) *ClusterController {
    
    	return &ClusterController{
    		context:              context,
    		volumeAttachment:     volumeAttachment,
    		flexvolumeController: flexvolumeController,
    	}
    }