K8s 实现 - Volume 实现

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

type Spec struct {
	Volume                          *v1.Volume
	PersistentVolume                *v1.PersistentVolume
	ReadOnly                        bool
	InlineVolumeSpecForCSIMigration bool
	Migrated                        bool
}

// 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 初始化 plugin
	Init(host VolumeHost) error
	// Name 返回 plugin name
	// 命名方式为 "example.com/volume"，"kubernetes.io" 由 kubernete 内置 Volume 预留使用
	GetPluginName() string
	// GetVolumeName 基于 spec 返回一个唯一的 volume name 或者 id
	// 对于 Attach 与 Detach 操作会传递该 name 来让 plugin 识别 volume
	GetVolumeName(spec *Spec) (string, error)
	// CanSupport 返回 plugin 是否支持该 spec
	CanSupport(spec *Spec) bool
	// RequiresRemount 表明 plugin 是否支持 volume 的重新挂载
	RequiresRemount(spec *Spec) bool
	// NewMounter 创建一个 Mounter
	NewMounter(spec *Spec, podRef *v1.Pod, opts VolumeOptions) (Mounter, error)
	// NewUnmounter 创建一个 UnMounter
	NewUnmounter(name string, podUID types.UID) (Unmounter, error)
	// ConstructVolumeSpec 由 volume name 得到一个 spec
	ConstructVolumeSpec(volumeName, volumePath string) (*Spec, error)
	// SupportsMountOption 表明是否支持挂载选项
	SupportsMountOption() bool
	// SupportsBulkVolumeVerification checks if volume plugin type is capable
	// of enabling bulk polling of all nodes. This can speed up verification of
	// attached volumes by quite a bit, but underlying pluging must support it.
	SupportsBulkVolumeVerification() bool
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

// DeletableVolumePlugin 表示支持 Delete 操作的 Volume Plugin
type DeletableVolumePlugin interface {
	VolumePlugin
	NewDeleter(spec *Spec) (Deleter, error)
}

// ProvisionableVolumePlugin 表示支持 Provisioner 操作的 Volume Plugin
type ProvisionableVolumePlugin interface {
	VolumePlugin
	NewProvisioner(options VolumeOptions) (Provisioner, error)
}

// AttachableVolumePlugin 表示支持 Attach/Detach 操作的 Volume Plugin
type AttachableVolumePlugin interface {
	DeviceMountableVolumePlugin
	NewAttacher() (Attacher, error)
	NewDetacher() (Detacher, error)
	CanAttach(spec *Spec) (bool, error)
}

// DeviceMountableVolumePlugin 表示支持 Mount/Unmount 操作的 Volume Plugin
type DeviceMountableVolumePlugin interface {
	VolumePlugin
	NewDeviceMounter() (DeviceMounter, error)
	NewDeviceUnmounter() (DeviceUnmounter, error)
	GetDeviceMountRefs(deviceMountPath string) ([]string, error)
	CanDeviceMount(spec *Spec) (bool, error)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74

// Volume represents a directory used by pods or hosts on a node. All method
// implementations of methods in the volume interface must be idempotent.
type Volume interface {
	// GetPath 返回 volume 需要的挂载路径
	GetPath() string
	// MetricsProvider 提供统计接口
	MetricsProvider
}

// Mounter 接口为 Pod 提供 Volume 对应的目录
type Mounter interface {
	Volume
	// CanMount 优先于 Setup 调用，用于检查当前环境是否包含能够 Mount 的组件
	CanMount() error
	// SetUp 为 Pod 提供 Volume 的目录，目录路径由自己决定
	SetUp(mounterArgs MounterArgs) error
	// SetUpAt  为 Pod 提供 Volume 的目录，目录路径由 dir 指定
	SetUpAt(dir string, mounterArgs MounterArgs) error
	// GetAttributes 返回 Volume 的属性
	GetAttributes() Attributes
}

// Unmounter interface provides methods to cleanup/unmount the volumes.
type Unmounter interface {
	Volume
	// TearDown 取消为 Pod 提供的目录
	TearDown() error
	// TearDown 取消为 Pod 提供的指定目录
	TearDownAt(dir string) error
}

// BlockVolumeMapper interface is a mapper interface for block volume.
type BlockVolumeMapper interface {
	BlockVolume
}

type Provisioner interface {
	// Provision 创建一个 Volume
	Provision(selectedNode *v1.Node, allowedTopologies []v1.TopologySelectorTerm) (*v1.PersistentVolume, error)
}

type Deleter interface {
	Volume
	// Deleter 移除一个 Volume
	Delete() error
}

type Attacher interface {
	DeviceMounter
	// Attach Volume 到指定的 Node 上
	Attach(spec *Spec, nodeName types.NodeName) (string, error)
	// VolumesAreAttached 查询 Node 上的 attached Volume
	VolumesAreAttached(specs []*Spec, nodeName types.NodeName) (map[*Spec]bool, error)
	// WaitForAttach 等待 attach 结束
	WaitForAttach(spec *Spec, devicePath string, pod *v1.Pod, timeout time.Duration) (string, error)
}

type DeviceMounter interface {
	// GetDeviceMountPath 返回 Node 上 Volume 对应的块设备路径
	GetDeviceMountPath(spec *Spec) (string, error)
	// MountDevice 将 Volume 对应的块设备挂载到一个全局的目录（不提供给 Pod）
	MountDevice(spec *Spec, devicePath string, deviceMountPath string, deviceMounterArgs DeviceMounterArgs) error
}

type Detacher interface {
	DeviceUnmounter
	// Detach Volume 在指定的 Node
	Detach(volumeName string, nodeName types.NodeName) error
}

type DeviceUnmounter interface {
	// UnmountDevice 取消 Volume 对应块设备的挂载
	UnmountDevice(deviceMountPath string) error
}

1
2
3
4
5
6
7
8

type VolumePluginMgr struct {
	mutex                     sync.RWMutex
	plugins                   map[string]VolumePlugin  // 静态注册的 VolumePlugin
	prober                    DynamicPluginProber
	probedPlugins             map[string]VolumePlugin // probe 探测出的 VolumePlugin
	loggedDeprecationWarnings sets.String
	Host                      VolumeHost
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

// FindPluginBySpec 查找能够支持 spec 的 VolumePlugin
func (pm *VolumePluginMgr) FindPluginBySpec(spec *Spec) (VolumePlugin, error) {
	pm.mutex.RLock()
	defer pm.mutex.RUnlock()

	if spec == nil {
		return nil, fmt.Errorf("could not find plugin because volume spec is nil")
	}

	// 调用静态注册的 VolumePlugin.CanSupport() 函数筛选
	matches := []VolumePlugin{}
	for _, v := range pm.plugins {
		if v.CanSupport(spec) {
			matches = append(matches, v)
		}
	}

	// 进行一次 Probe Plugin
	pm.refreshProbedPlugins()

	// 调用动态注册的 VolumePlugin.CanSupport() 函数筛选
	for _, plugin := range pm.probedPlugins {
		if plugin.CanSupport(spec) {
			matches = append(matches, plugin)
		}
	}

	// 没有任何 Volume Plugin 能够处理
	if len(matches) == 0 {
		return nil, fmt.Errorf("no volume plugin matched")
	}

	// 多个 Volume Plugin 能够处理
	if len(matches) > 1 {
		matchedPluginNames := []string{}
		for _, plugin := range matches {
			matchedPluginNames = append(matchedPluginNames, plugin.GetPluginName())
		}
		return nil, fmt.Errorf("multiple volume plugins matched: %s", strings.Join(matchedPluginNames, ","))
	}

	// 返回唯一的 Volume Plugin
	// Issue warning if the matched provider is deprecated
	pm.logDeprecation(matches[0].GetPluginName())
	return matches[0], nil
}

// FindPluginByName 通过 plugin name 查找 Volume Plugin
func (pm *VolumePluginMgr) FindPluginByName(name string) (VolumePlugin, error) {
	// ...
}

func (pm *VolumePluginMgr) FindPersistentPluginBySpec(spec *Spec) (PersistentVolumePlugin, error) {
	// ...
}

// 其他各个类型的 VolumePlugin ...

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

type ProbeEvent struct {
	Plugin     VolumePlugin // VolumePlugin that was added/updated/removed. if ProbeEvent.Op is 'ProbeRemove', Plugin should be nil
	PluginName string
	Op         ProbeOperation // The operation to the plugin
}

type DynamicPluginProber interface {
	Init() error

	// If an error occurs, events are undefined.
	Probe() (events []ProbeEvent, err error)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

$ ./uds
Usage:
  flexvoldrv [command]

Available Commands:
  help        Help about any command
  init        Flex volume init command.
  mount       Flex volume unmount command.
  unmount     Flex volume unmount command.
  version     Print version

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

func (ctrl *PersistentVolumeController) updateVolume(volume *v1.PersistentVolume) {
    // 更新 Store
    new, err := ctrl.storeVolumeUpdate(volume)
	if err != nil {
	}
	if !new {
		return
	}

    // 执行 sync
	err = ctrl.syncVolume(volume)
	if err != nil {
		// log
	}
}

func (ctrl *PersistentVolumeController) updateClaim(claim *v1.PersistentVolumeClaim) {
    // 更新 Store
	new, err := ctrl.storeClaimUpdate(claim)
	if err != nil {
	}
	if !new {
		return
	}
    // 执行 sync
	err = ctrl.syncClaim(claim)
	if err != nil {
		// log
	}
}

1
2
3
4

func ProcessPodVolumes(pod *v1.Pod, addVolumes bool,
	desiredStateOfWorld cache.DesiredStateOfWorld, volumePluginMgr *volume.VolumePluginMgr,
	pvcLister corelisters.PersistentVolumeClaimLister, pvLister corelisters.PersistentVolumeLister,
	csiMigratedPluginManager csimigration.PluginManager, csiTranslator csimigration.InTreeToCSITranslator)

1
2

func (adc *attachDetachController) processVolumesInUse(
	nodeName types.NodeName, volumesInUse []v1.UniqueVolumeName)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

func (rc *reconciler) reconcile() {
	// Unmounts are triggered before mounts so that a volume that was
	// referenced by a pod that was deleted and is now referenced by another
	// pod is unmounted from the first pod before being mounted to the new
	// pod.
	rc.unmountVolumes()

	// Next we mount required volumes. This function could also trigger
	// attach if kubelet is responsible for attaching volumes.
	// If underlying PVC was resized while in-use then this function also handles volume
	// resizing.
	rc.mountAttachVolumes()

	// Ensure devices that should be detached/unmounted are detached/unmounted.
	rc.unmountDetachDevices()
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

func (rc *reconciler) unmountVolumes() {
	// 遍历当前节点所有的 mounted volume
	for _, mountedVolume := range rc.actualStateOfWorld.GetAllMountedVolumes() {
		// 检查该 volume 与 pod 是否存在于 desiredStateOfWorld
		if !rc.desiredStateOfWorld.PodExistsInVolume(mountedVolume.PodName, mountedVolume.VolumeName) {
			// 进行 unmount volume 操作（底层调用的是 VolumePlugin.TearDown）
			err := rc.operationExecutor.UnmountVolume(
				mountedVolume.MountedVolume, rc.actualStateOfWorld, rc.kubeletPodsDir)
			if err != nil &&
				!nestedpendingoperations.IsAlreadyExists(err) &&
				!exponentialbackoff.IsExponentialBackoff(err) {
				// Ignore nestedpendingoperations.IsAlreadyExists and exponentialbackoff.IsExponentialBackoff errors, they are expected.
				// Log all other errors.
				klog.ErrorS(err, mountedVolume.GenerateErrorDetailed(fmt.Sprintf("operationExecutor.UnmountVolume failed (controllerAttachDetachEnabled %v)", rc.controllerAttachDetachEnabled), err).Error())
			}
			if err == nil {
				klog.InfoS(mountedVolume.GenerateMsgDetailed("operationExecutor.UnmountVolume started", ""))
			}
		}
	}
}