【kubernetes/k8s源码分析】kubernetes volume plugin源码分析

系统管理员 2022-03-25 01:56 356阅读 0赞

# plugin调用函数： #

文章[https://blog.csdn.net/zhonglinzhang/article/details/82800287][https_blog.csdn.net_zhonglinzhang_article_details_82800287] 分析volume manager调用

*  reconcile -> 
 *  MountVolume ->
 *  **NewMouter根据插件调用 ->**
 *  执行plugin的SetUp

# volumePlugin接口 #

>  *  存储提供的扩展接口, 包含了各类存储提供者的plugin实现
>  *  实现自定义的Plugins 可以通过FlexVolume(K8s 1.8版本)
>  *  容器存储接口 CSI（Container Storage Interface）
>  *  支持这套标准以后，K8S和存储提供者之间将彻底解耦，这一功能让新的卷插件的安装像部署Pod 一样简单，第三方存储开发的代码也不需要加入核心的Kubernetes代码中。

// VolumePlugin is an interface to volume plugins that can be used on a
    // kubernetes node (e.g. by kubelet) to instantiate and manage volumes.
    type VolumePlugin interface {
    	
    	Init(host VolumeHost) error
    
    	GetPluginName() string
    
    	GetVolumeName(spec *Spec) (string, error)
    
    	CanSupport(spec *Spec) bool
    
    
    	RequiresRemount() bool
    
    
    	NewMounter(spec *Spec, podRef *v1.Pod, opts VolumeOptions) (Mounter, error)
    
    	NewUnmounter(name string, podUID types.UID) (Unmounter, error)
    
    	ConstructVolumeSpec(volumeName, mountPath string) (*Spec, error)
    
    	SupportsMountOption() bool
    
    	SupportsBulkVolumeVerification() bool
    }

# 1. kubelet 启动注册volume plugins #

*       cmd/kubelet/app/server.go中run函数调用UnsecuredDependencies函数注册volume plugins，主要函数为ProbeVolumePlugins()

func UnsecuredDependencies(s *options.KubeletServer) (*kubelet.Dependencies, error) {
    
    
    	return &kubelet.Dependencies{
    		
    		Mounter:             mounter,
    		NetworkPlugins:      ProbeNetworkPlugins(s.CNIConfDir, s.CNIBinDir),
    		OOMAdjuster:         oom.NewOOMAdjuster(),
    		OSInterface:         kubecontainer.RealOS{},
    		Writer:              writer,
    		VolumePlugins:       ProbeVolumePlugins(),
    
    }

###    1.1 ProbeVolumePlugins函数 ###

*      注册到结构体为volume.volumePlugin列表，接口如VolumePlugin所示，主要讲解host\_path，以及nfs

func ProbeVolumePlugins() []volume.VolumePlugin {
    	allPlugins := []volume.VolumePlugin{}
    
    	// The list of plugins to probe is decided by the kubelet binary, not
    	// by dynamic linking or other "magic".  Plugins will be analyzed and
    	// initialized later.
    	//
    	// Kubelet does not currently need to configure volume plugins.
    	// If/when it does, see kube-controller-manager/app/plugins.go for example of using volume.VolumeConfig
    	allPlugins = append(allPlugins, aws_ebs.ProbeVolumePlugins()...)
    	allPlugins = append(allPlugins, empty_dir.ProbeVolumePlugins()...)
    	allPlugins = append(allPlugins, gce_pd.ProbeVolumePlugins()...)
    	allPlugins = append(allPlugins, git_repo.ProbeVolumePlugins()...)
    	allPlugins = append(allPlugins, host_path.ProbeVolumePlugins(volume.VolumeConfig{})...)
    	allPlugins = append(allPlugins, nfs.ProbeVolumePlugins(volume.VolumeConfig{})...)
    	
    	if utilfeature.DefaultFeatureGate.Enabled(features.CSIPersistentVolume) {
    		allPlugins = append(allPlugins, csi.ProbeVolumePlugins()...)
    	}
    	return allPlugins
    }

## ##

# 2. kubelet 初始化volume plugins #

*  pkg/kubelet/kubelet.go中NewMainKubelet函数中初始化volume plugin
 *  mountpod.NewMnager函数中kubelet的root  dir为 /var/lib/kubelet，目录下面有文件plugin-containers，plugins，pods
 *  **InitPlugins初始化注册的所有volune plugin(2**.1讲解)

func NewInitializedVolumePluginMgr(
    	kubelet *Kubelet,
    	secretManager secret.Manager,
    	configMapManager configmap.Manager,
    	plugins []volume.VolumePlugin,
    	prober volume.DynamicPluginProber) (*volume.VolumePluginMgr, error) {
    
    	mountPodManager, err := mountpod.NewManager(kubelet.getRootDir(), kubelet.podManager)
    	if err != nil {
    		return nil, err
    	}
    	kvh := &kubeletVolumeHost{
    		kubelet:          kubelet,
    		volumePluginMgr:  volume.VolumePluginMgr{},
    		secretManager:    secretManager,
    		configMapManager: configMapManager,
    		mountPodManager:  mountPodManager,
    	}
    
    	if err := kvh.volumePluginMgr.InitPlugins(plugins, prober, kvh); err != nil {
    		return nil, fmt.Errorf(
    			"Could not initialize volume plugins for KubeletVolumePluginMgr: %v",
    			err)
    	}
    
    	return &kvh.volumePluginMgr, nil
    }

###    2.1 InitPlugins函数 ###

*  路径pkg/volume/plugins.go
 *  调用各个plugin的init初始化

func (pm *VolumePluginMgr) InitPlugins(plugins []VolumePlugin, prober DynamicPluginProber, host VolumeHost) error {
    
    	if pm.plugins == nil {
    		pm.plugins = map[string]VolumePlugin{}
    	}
    
    	allErrs := []error{}
    	for _, plugin := range plugins {
    		name := plugin.GetPluginName()
    		if errs := validation.IsQualifiedName(name); len(errs) != 0 {
    			allErrs = append(allErrs, fmt.Errorf("volume plugin has invalid name: %q: %s", name, strings.Join(errs, ";")))
    			continue
    		}
    
    		if _, found := pm.plugins[name]; found {
    			allErrs = append(allErrs, fmt.Errorf("volume plugin %q was registered more than once", name))
    			continue
    		}
    		err := plugin.Init(host)
    		if err != nil {
    			glog.Errorf("Failed to load volume plugin %s, error: %s", name, err.Error())
    			allErrs = append(allErrs, err)
    			continue
    		}
    		pm.plugins[name] = plugin
    		glog.V(1).Infof("Loaded volume plugin %q", name)
    	}
    	return utilerrors.NewAggregate(allErrs)
    }

# VolumeHost 接口 #

路径pkg/volume/plugins.go

// VolumeHost is an interface that plugins can use to access the kubelet.
    type VolumeHost interface {
    	// GetPluginDir returns the absolute path to a directory under which
    	// a given plugin may store data.  This directory might not actually
    	// exist on disk yet.  For plugin data that is per-pod, see
    	// GetPodPluginDir().
    	GetPluginDir(pluginName string) string
    
    	// GetVolumeDevicePluginDir returns the absolute path to a directory
    	// under which a given plugin may store data.
    	// ex. plugins/kubernetes.io/{PluginName}/{DefaultKubeletVolumeDevicesDirName}/{volumePluginDependentPath}/
    	GetVolumeDevicePluginDir(pluginName string) string
    
    	// GetPodsDir returns the absolute path to a directory where all the pods
    	// information is stored
    	GetPodsDir() string
    
    	// GetPodVolumeDir returns the absolute path a directory which
    	// represents the named volume under the named plugin for the given
    	// pod.  If the specified pod does not exist, the result of this call
    	// might not exist.
    	GetPodVolumeDir(podUID types.UID, pluginName string, volumeName string) string
    
    	// GetPodPluginDir returns the absolute path to a directory under which
    	// a given plugin may store data for a given pod.  If the specified pod
    	// does not exist, the result of this call might not exist.  This
    	// directory might not actually exist on disk yet.
    	GetPodPluginDir(podUID types.UID, pluginName string) string

# VolumeConfig结构体 #

type VolumeConfig struct {
    	// RecyclerPodTemplate is pod template that understands how to scrub clean
    	// a persistent volume after its release. The template is used by plugins
    	// which override specific properties of the pod in accordance with that
    	// plugin. See NewPersistentVolumeRecyclerPodTemplate for the properties
    	// that are expected to be overridden.
    	RecyclerPodTemplate *v1.Pod
    
    	// RecyclerMinimumTimeout is the minimum amount of time in seconds for the
    	// recycler pod's ActiveDeadlineSeconds attribute. Added to the minimum
    	// timeout is the increment per Gi of capacity.
    	RecyclerMinimumTimeout int
    
    	// RecyclerTimeoutIncrement is the number of seconds added to the recycler
    	// pod's ActiveDeadlineSeconds for each Gi of capacity in the persistent
    	// volume. Example: 5Gi volume x 30s increment = 150s + 30s minimum = 180s
    	// ActiveDeadlineSeconds for recycler pod
    	RecyclerTimeoutIncrement int
    
    	// PVName is name of the PersistentVolume instance that is being recycled.
    	// It is used to generate unique recycler pod name.
    	PVName string
    
    	// OtherAttributes stores config as strings.  These strings are opaque to
    	// the system and only understood by the binary hosting the plugin and the
    	// plugin itself.
    	OtherAttributes map[string]string
    
    	// ProvisioningEnabled configures whether provisioning of this plugin is
    	// enabled or not. Currently used only in host_path plugin.
    	ProvisioningEnabled bool
    }

# 2. host\_path plugin #

###     hostPathPlugin 结构体 ###

type hostPathPlugin struct {
    	host   volume.VolumeHost
    	config volume.VolumeConfig
    }

###   HostPathVolumeSource结构体   ###

// Represents a host path mapped into a pod.
    // Host path volumes do not support ownership management or SELinux relabeling.
    type HostPathVolumeSource struct {
    	// Path of the directory on the host.
    	// If the path is a symlink, it will follow the link to the real path.
    	// More info: https://kubernetes.io/docs/concepts/storage/volumes#hostpath
    	Path string `json:"path" protobuf:"bytes,1,opt,name=path"`
    	// Type for HostPath Volume
    	// Defaults to ""
    	// More info: https://kubernetes.io/docs/concepts/storage/volumes#hostpath
    	// +optional
    	Type *HostPathType `json:"type,omitempty" protobuf:"bytes,2,opt,name=type"`
    }

type 字段支持以下值：

<table> 
 <tbody> 
 <tr> 
 <td> 值 </td> 
 <td> 行为 </td> 
 </tr> 
 <tr> 
 <td>&nbsp;</td> 
 <td> 空字符串（默认）用于向后兼容，这意味着在挂载 hostPath 卷之前不会执行任何检查。 </td> 
 </tr> 
 <tr> 
 <td> DirectoryOrCreate </td> 
 <td> 如果在给定的路径上没有任何东西存在，那么将根据需要在那里创建一个空目录，权限设置为 0755，与 Kubelet 具有相同的组和所有权。 </td> 
 </tr> 
 <tr> 
 <td> Directory </td> 
 <td> 路径下必须存在目录 </td> 
 </tr> 
 <tr> 
 <td> FileOrCreate </td> 
 <td> 如果在给定的路径上没有任何东西存在，那么会根据需要创建一个空文件，权限设置为 0644，与 Kubelet 具有相同的组和所有权。 </td> 
 </tr> 
 <tr> 
 <td> File </td> 
 <td> 给定的路径下必须存在文件 </td> 
 </tr> 
 <tr> 
 <td> Socket </td> 
 <td> 路径下必须存在 UNIX 套接字 </td> 
 </tr> 
 <tr> 
 <td> CharDevice </td> 
 <td> 路径下必须存在字符设备 </td> 
 </tr> 
 <tr> 
 <td> BlockDevice </td> 
 <td> 路径下必须存在块设备 </td> 
 </tr> 
 </tbody> 
</table>

// HostPath volumes represent a bare host file or directory mount.
    // The direct at the specified path will be directly exposed to the container.
    type hostPath struct {
    	path     string
    	pathType *v1.HostPathType
    	volume.MetricsNil
    }
    
    func (hp *hostPath) GetPath() string {
    	return hp.path
    }
    
    type hostPathMounter struct {
    	*hostPath
    	readOnly bool
    	mounter  mount.Interface
    }

###    2.1 NewMounter函数 ###

调用getVolumeSource取得path pathType，填充hostPathMounter实例

func (plugin *hostPathPlugin) NewMounter(spec *volume.Spec, pod *v1.Pod, opts volume.VolumeOptions) (volume.Mounter, error) {
    	hostPathVolumeSource, readOnly, err := getVolumeSource(spec)
    	if err != nil {
    		return nil, err
    	}
    
    	path := hostPathVolumeSource.Path
    	pathType := new(v1.HostPathType)
    	if hostPathVolumeSource.Type == nil {
    		*pathType = v1.HostPathUnset
    	} else {
    		pathType = hostPathVolumeSource.Type
    	}
    	return &hostPathMounter{
    		hostPath: &hostPath{path: path, pathType: pathType},
    		readOnly: readOnly,
    		mounter:  plugin.host.GetMounter(plugin.GetPluginName()),
    	}, nil
    }

###   2.2 SetUp函数 ###

对于hostpath插件SetUp does nothing

// SetUp does nothing.
    func (b *hostPathMounter) SetUp(fsGroup *int64) error {
    	err := validation.ValidatePathNoBacksteps(b.GetPath())
    	if err != nil {
    		return fmt.Errorf("invalid HostPath `%s`: %v", b.GetPath(), err)
    	}
    
    	if *b.pathType == v1.HostPathUnset {
    		return nil
    	}
    	return checkType(b.GetPath(), b.pathType, b.mounter)
    }

# 3. NFS #

路径 pkg/volume/nfs/nfs.go

###    nfsPlugin结构体 ###

// This is the primary entrypoint for volume plugins.
    // The volumeConfig arg provides the ability to configure recycler behavior.  It is implemented as a pointer to allow nils.
    // The nfsPlugin is used to store the volumeConfig and give it, when needed, to the func that creates NFS Recyclers.
    // Tests that exercise recycling should not use this func but instead use ProbeRecyclablePlugins() to override default behavior.
    func ProbeVolumePlugins(volumeConfig volume.VolumeConfig) []volume.VolumePlugin {
    	return []volume.VolumePlugin{
    		&nfsPlugin{
    			host:   nil,
    			config: volumeConfig,
    		},
    	}
    }
    
    type nfsPlugin struct {
    	host   volume.VolumeHost
    	config volume.VolumeConfig
    }

###    3.1 Init函数 ###

func (plugin *nfsPlugin) Init(host volume.VolumeHost) error {
    	plugin.host = host
    	return nil
    }

###   3.2 NewMounter函数 ###

func (plugin *nfsPlugin) NewMounter(spec *volume.Spec, pod *v1.Pod, _ volume.VolumeOptions) (volume.Mounter, error) {
    	return plugin.newMounterInternal(spec, pod, plugin.host.GetMounter(plugin.GetPluginName()))
    }
    
    func (plugin *nfsPlugin) newMounterInternal(spec *volume.Spec, pod *v1.Pod, mounter mount.Interface) (volume.Mounter, error) {
    	source, readOnly, err := getVolumeSource(spec)
    	if err != nil {
    		return nil, err
    	}
    
    	return &nfsMounter{
    		nfs: &nfs{
    			volName: spec.Name(),
    			mounter: mounter,
    			pod:     pod,
    			plugin:  plugin,
    		},
    		server:       source.Server,
    		exportPath:   source.Path,
    		readOnly:     readOnly,
    		mountOptions: util.MountOptionFromSpec(spec),
    	}, nil
    }

###    3.3 SetUp函数 ###

Mount函数根据初始化kubeletVolumeHost实现了Mounter接口（pkg/kubelet/volume\_host.go）

// SetUp attaches the disk and bind mounts to the volume path.
    func (b *nfsMounter) SetUp(fsGroup *int64) error {
    	return b.SetUpAt(b.GetPath(), fsGroup)
    }
    
    func (b *nfsMounter) SetUpAt(dir string, fsGroup *int64) error {
    	notMnt, err := b.mounter.IsNotMountPoint(dir)
    	glog.V(4).Infof("NFS mount set up: %s %v %v", dir, !notMnt, err)
    	if err != nil && !os.IsNotExist(err) {
    		return err
    	}
    	if !notMnt {
    		return nil
    	}
    	if err := os.MkdirAll(dir, 0750); err != nil {
    		return err
    	}
    	source := fmt.Sprintf("%s:%s", b.server, b.exportPath)
    	options := []string{}
    	if b.readOnly {
    		options = append(options, "ro")
    	}
    	mountOptions := util.JoinMountOptions(b.mountOptions, options)
    	err = b.mounter.Mount(source, dir, "nfs", mountOptions)
    	。。。。
    	return nil
    }

3.3.1 追踪到使用了mount命令执行

// Mount runs mount(8) using given exec interface.
 func (m *execMounter) Mount(source string, target string, fstype string, options []string) error {
 	bind, bindRemountOpts := isBind(options)
 
 	if bind {
 		err := m.doExecMount(source, target, fstype, []string{"bind"})
 		if err != nil {
 			return err
 		}
 		return m.doExecMount(source, target, fstype, bindRemountOpts)
 	}
 
 	return m.doExecMount(source, target, fstype, options)
 }
 
 // doExecMount calls exec(mount <what> <where>) using given exec interface.
 func (m *execMounter) doExecMount(source, target, fstype string, options []string) error {
 	glog.V(5).Infof("Exec Mounting %s %s %s %v", source, target, fstype, options)
 	mountArgs := makeMountArgs(source, target, fstype, options)
 	output, err := m.exec.Run("mount", mountArgs...)
 	glog.V(5).Infof("Exec mounted %v: %v: %s", mountArgs, err, string(output))
 	if err != nil {
 		return fmt.Errorf("mount failed: %v\nMounting command: %s\nMounting arguments: %s %s %s %v\nOutput: %s\n",
 			err, "mount", source, target, fstype, options, string(output))
 	}
 
 	return err
 }

[https_blog.csdn.net_zhonglinzhang_article_details_82800287]: https://blog.csdn.net/zhonglinzhang/article/details/82800287