openEBS lvm_localpv

背景

一直想学习CSI相关内容，恰好自己对LVM比较熟悉，openEBS lvm_localpv 有实现CSI，那么就来看看其具体实现吧https://github.com/openebs/lvm-localpv

LVM背景

可以参考我之前写的博客

bookmark

CSI文档

https://github.com/container-storage-interface/spec/blob/master/spec.md

Identity Service

查询插件的元信息和能力，通常在初始化阶段由CO（container Orchestrator容器编排工具）调用

service Identity {
  rpc GetPluginInfo(GetPluginInfoRequest)
    returns (GetPluginInfoResponse) {}

  rpc GetPluginCapabilities(GetPluginCapabilitiesRequest)
    returns (GetPluginCapabilitiesResponse) {}

  rpc Probe (ProbeRequest)
    returns (ProbeResponse) {}
}

Controller Service 中心管控接口

负责处理卷的生命周期，用于控制平面操作，如创建、删除、扩展存储卷，负责对存储资源的协调和调度

service Controller {
  rpc CreateVolume (CreateVolumeRequest)
    returns (CreateVolumeResponse) {}

  rpc DeleteVolume (DeleteVolumeRequest)
    returns (DeleteVolumeResponse) {}

  rpc ControllerPublishVolume (ControllerPublishVolumeRequest)
    returns (ControllerPublishVolumeResponse) {}

  rpc ControllerUnpublishVolume (ControllerUnpublishVolumeRequest)
    returns (ControllerUnpublishVolumeResponse) {}
	// 验证controller信息，AccessMode、VolumeType等
  rpc ValidateVolumeCapabilities (ValidateVolumeCapabilitiesRequest)
    returns (ValidateVolumeCapabilitiesResponse) {}

  rpc ListVolumes (ListVolumesRequest)
    returns (ListVolumesResponse) {}

  rpc GetCapacity (GetCapacityRequest)
    returns (GetCapacityResponse) {}

  rpc ControllerGetCapabilities (ControllerGetCapabilitiesRequest)
    returns (ControllerGetCapabilitiesResponse) {}

  rpc CreateSnapshot (CreateSnapshotRequest)
    returns (CreateSnapshotResponse) {}

  rpc DeleteSnapshot (DeleteSnapshotRequest)
    returns (DeleteSnapshotResponse) {}

  rpc ListSnapshots (ListSnapshotsRequest)
    returns (ListSnapshotsResponse) {}

  rpc ControllerExpandVolume (ControllerExpandVolumeRequest)
    returns (ControllerExpandVolumeResponse) {}
}

Node Service 节点管控接口

负责管理实际的存储节点，处理对于节点上存储的访问、挂载、卸载等操作，将存储设备挂载到k8s节点上的容器中

service Node {
	
//在节点上初始化存储卷（格式化），并执行挂载到Global目录

  rpc NodeStageVolume (NodeStageVolumeRequest)
    returns (NodeStageVolumeResponse) {}
	
  rpc NodeUnstageVolume (NodeUnstageVolumeRequest)
    returns (NodeUnstageVolumeResponse) {}
	
//在节点上讲存储卷的 Global目录挂载到Pod挂载目录

  rpc NodePublishVolume (NodePublishVolumeRequest)
    returns (NodePublishVolumeResponse) {}

  rpc NodeUnpublishVolume (NodeUnpublishVolumeRequest)
    returns (NodeUnpublishVolumeResponse) {}
	
  rpc NodeGetVolumeStats (NodeGetVolumeStatsRequest)
    returns (NodeGetVolumeStatsResponse) {}

  rpc NodeExpandVolume(NodeExpandVolumeRequest)
    returns (NodeExpandVolumeResponse) {}

  rpc NodeGetCapabilities (NodeGetCapabilitiesRequest)
    returns (NodeGetCapabilitiesResponse) {}

  rpc NodeGetInfo (NodeGetInfoRequest)
    returns (NodeGetInfoResponse) {}
}

核心目录结构

非完整目录结构，只有**/pkg**下保留关键部分

.
├── apis #volume、node、snapshot的CRD
├── builder
│   ├── nodebuilder
│   │   ├── kubernetes.go #CR对象在k8s中的构建代码
│   │   └── node.go #使用builder pattern 实现CR对象的初始化
│   ...
├── collector #CR metrics 收集
├── driver # CSIDriver的实现
│   ├── agent.go #实现node service的接口
│   ├── controller.go #实现controller service接口
│   ├── driver.go 
│   ├── grpc.go #对外提供rpc服务
│   ├── identity.go #实现Identity service
├── generated  # CRD对应的lister、clientset、informer
├── lvm
│   ├── iolimiter.go #实现磁盘io限制
│   ├── lvm_util.go #实现lvm管理
│   ├── mount.go #实现绑定、捆绑
│   └── volume.go #对接CR的 client，实现volume的CRUD
├── mgmt
│   ├── lvmnode
│   │   ├── builder.go #构造器
│   │   ├── lvmnode.go #控制器模式
│   │   └── start.go #整合builder和lvmnode代码，实现对外暴露
│   ...

创建PV的过程

若想创建PV，那么此时会调用CSI的CreateVolume

func (cs *controller) CreateVolume(
	ctx context.Context,
	req *csi.CreateVolumeRequest,
) (*csi.CreateVolumeResponse, error) {

	//1. 参数校验
	if err := cs.validateVolumeCreateReq(req); err != nil {
		return nil, status.Error(codes.InvalidArgument, err.Error())
	}
	//2. 构建请求参数
	params, err := NewVolumeParams(req.GetParameters())
	if err != nil {
		return nil, status.Errorf(codes.InvalidArgument,
			"failed to parse csi volume params: %v", err)
	}
	
	volName := strings.ToLower(req.GetName())
	size := getRoundedCapacity(req.GetCapacityRange().GetRequiredBytes())
	contentSource := req.GetVolumeContentSource()

	var vol *lvmapi.LVMVolume
	if contentSource != nil && contentSource.GetSnapshot() != nil {
		return nil, status.Error(codes.Unimplemented, "")
	} else if contentSource != nil && contentSource.GetVolume() != nil {
		return nil, status.Error(codes.Unimplemented, "")
	} else {
		// mark volume for leak protection if pvc gets deleted
		// before the creation of pv.
		var finishCreateVolume func()
		if finishCreateVolume, err = cs.leakProtection.BeginCreateVolume(volName,
			params.PVCNamespace, params.PVCName); err != nil {
			return nil, err
		}
		defer finishCreateVolume()
		//相当于数据入库
		//如etcd库
		//后续的操作，由volume controller完成
		vol, err = CreateLVMVolume(ctx, req, params)
	}

	if err != nil {
		return nil, err
	}
	sendEventOrIgnore(params.PVCName, volName,
		strconv.FormatInt(int64(size), 10),
		analytics.VolumeProvision)

	topology := map[string]string{lvm.LVMTopologyKey: vol.Spec.OwnerNodeID}
	cntx := map[string]string{lvm.VolGroupKey: vol.Spec.VolGroup, lvm.OpenEBSCasTypeKey: lvm.LVMCasTypeName}

	return csipayload.NewCreateVolumeResponseBuilder().
		WithName(volName).
		WithCapacity(size).
		WithTopology(topology).
		WithContext(cntx).
		WithContentSource(contentSource).
		Build(), nil
}

CreateVolume调用结束后，集群中LVMVolume CR创建成功,LVMVolume Controller监听到后，调用informer的AddFunc，将其加入WorkQueue中，

func (c *VolController) addVol(obj interface{}) {
	Vol, ok := c.getStructuredObject(obj)
	if !ok {
		runtime.HandleError(fmt.Errorf("Couldn't get Vol object %#v", obj))
		return
	}

	if lvm.NodeID != Vol.Spec.OwnerNodeID {
		return
	}
	c.enqueueVol(Vol)
}

而processNextWorkItem 作为消费者，从workQueue中读取 item

func (c *VolController) processNextWorkItem() bool {
	obj, shutdown := c.workqueue.Get()

	if shutdown {
		return false
	}

	func(obj interface{}) error {
		defer c.workqueue.Done(obj)
		var key string
		var ok bool
		if key, ok = obj.(string); !ok {
			c.workqueue.Forget(obj)
			runtime.HandleError(fmt.Errorf("expected string in workqueue but got %#v", obj))
			return nil
		}
		if err := c.syncHandler(key); err != nil {
			c.workqueue.AddRateLimited(key)
			return fmt.Errorf("error syncing '%s': %s, requeuing", key, err.Error())
		}
		klog.Infof("Successfully synced '%s'", key)
		return nil
	}(obj)

	return true
}

后续调用链为processNextWorkItem → syncHandler → syncVol → CreateVolume → RunCommandSplit，在此执行lvcreate 指令

func RunCommandSplit(command string, args ...string) ([]byte, []byte, error) {
	var cmdStdout bytes.Buffer
	var cmdStderr bytes.Buffer

	cmd := exec.Command(command, args...)
	cmd.Stdout = &cmdStdout
	cmd.Stderr = &cmdStderr
	err := cmd.Run()

	output := cmdStdout.Bytes()
	error_output := cmdStderr.Bytes()

	if len(error_output) > 0 {
		klog.Warningf("lvm: said into stderr: %s", error_output)
	}

	return output, error_output, err
}

总结流程：

1. 用户创建 PV，触发 CSI 的 CreateVolume 请求。
2. CreateVolume 请求会解析参数并调用 CreateLVMVolume 来创建 LVM 逻辑卷。
3. LVMVolume CR 被创建并保存到 Kubernetes 中。
4. LVMVolume Controller 监听到 CR 的变化，将其加入工作队列。
5. 工作队列的消费者（processNextWorkItem）处理任务，调用 syncHandler。
6. syncHandler 调用 RunCommandSplit 执行 lvcreate 命令，在节点上创建实际的逻辑卷。
7. 逻辑卷创建完成后，LVMVolume CR 更新状态，表示卷已准备好。

Pod引用PVC的过程

kind: PersistentVolumeClaim
apiVersion: v1
metadata:
  name: csi-lvmpv
spec:
  accessModes:
    - ReadWriteOnce        ## Specify ReadWriteOnce(RWO) access modes
  storageClassName: openebs-lvm
  resources:
    requests:
      storage: 4Gi

对于使用PVC动态创建PV的场景，由storageClass对应的provisioner实现PV的创建，k8s调度器会根据节点的资源情况和Pod要求，将pod调度到合适节点，但在这个阶段会确认Pod最终在哪一个Node上运行，但不会立刻执行PVC的挂载

当POd被调度到Node上准备开始运行时，k8s会执行容器的启动流程

• 在容器启动过程中，Kubernetes 会通过 CSI 插件的 NodePublishVolume 操作将 PVC 挂载到容器内。这是实际的挂载操作，它会在节点上找到对应的存储卷（PV）并将其挂载到指定的目录（mountPath）

func (ns *node) NodePublishVolume(
	ctx context.Context,
	req *csi.NodePublishVolumeRequest,
) (*csi.NodePublishVolumeResponse, error) {

	var (
		err error
	)

	if err = ns.validateNodePublishReq(req); err != nil {
		return nil, err
	}

	vol, mountInfo, err := GetVolAndMountInfo(req)
	if err != nil {
		return nil, status.Error(codes.Internal, err.Error())
	}

	podLVinfo, err := getPodLVInfo(req)
	if err != nil {
		klog.Warningf("PodLVInfo could not be obtained for volume_id: %s, err = %v", req.VolumeId, err)
	}
	switch req.GetVolumeCapability().GetAccessType().(type) {
	case *csi.VolumeCapability_Block:
		// attempt block mount operation on the requested path
		err = lvm.MountBlock(vol, mountInfo, podLVinfo)
	case *csi.VolumeCapability_Mount:
		// attempt filesystem mount operation on the requested path
		err = lvm.MountFilesystem(vol, mountInfo, podLVinfo)
	}

	if err != nil {
		return nil, err
	}

	return &csi.NodePublishVolumeResponse{}, nil
}

func MountFilesystem(vol *apis.LVMVolume, mount *MountInfo, podinfo *PodLVInfo) error {
	if err := os.MkdirAll(mount.MountPath, 0755); err != nil {
		return status.Errorf(codes.Internal, "Could not create dir {%q}, err: %v", mount.MountPath, err)
	}

	return MountVolume(vol, mount, podinfo)
}

func MountVolume(vol *apis.LVMVolume, mount *MountInfo, podLVInfo *PodLVInfo) error {
	volume := vol.Spec.VolGroup + "/" + vol.Name
	mounted, err := verifyMountRequest(vol, mount.MountPath)
	if err != nil {
		return err
	}

	if mounted {
		klog.Infof("lvm : already mounted %s => %s", volume, mount.MountPath)
		return nil
	}

	devicePath := DevPath + volume

	err = FormatAndMountVol(devicePath, mount)
	if err != nil {
		return status.Errorf(
			codes.Internal,
			"failed to format and mount the volume error: %s",
			err.Error(),
		)
	}

	klog.Infof("lvm: volume %v mounted %v fs %v", volume, mount.MountPath, mount.FSType)

	if ioLimitsEnabled && podLVInfo != nil {
		if err := setIOLimits(vol, podLVInfo, devicePath); err != nil {
			klog.Warningf("lvm: error setting io limits: podUid %s, device %s, err=%v", podLVInfo.UID, devicePath, err)
		} else {
			klog.Infof("lvm: io limits set for podUid %v, device %s", podLVInfo.UID, devicePath)
		}
	}

	return nil
}

总结流程：

1. PVC 创建：由 Provisioner 动态创建 PV。
2. Pod 调度：调度器根据资源情况将 Pod 调度到合适的节点。
3. NodePublishVolume 挂载：在 Pod 容器启动时，执行实际的卷挂载操作。
4. 挂载类型：根据请求类型，执行 块设备挂载 或 文件系统挂载，并处理相关的 IO 限制

监控指标

对于磁盘IO，openEBS lvm_localPV会关注这些指标

Capacity-based Metrics

• 节点上的总预配容量：节点上所有卷组（Volume Groups，VG）的总容量。使用命令 vgs -o vg_size <vg_name> 可以获取特定卷组（VG）的总容量（vg_size）。如果不指定 <vg_name>，则会列出所有卷组的总容量。
• 节点上的总空闲容量：节点上所有卷组（VG）的总空闲容量。使用命令 vgs -o vg_free <vg_name> 可以获取特定卷组（VG）的空闲容量（vg_free）。如果不指定 <vg_name>，则会列出所有卷组的空闲容量。
• 节点上的总已用容量：节点上所有卷组（VG）的总已用容量。通过 vg_size 减去 vg_free 得到已用容量（vg_used）值。
• 节点上的总已分配容量：节点上所有逻辑卷（LV）的总容量。使用命令 lvs -o lv_size <lv_full_name> 可以获取特定逻辑卷（LV）的容量（lv_size）。如果不指定 <lv_full_name>，则会列出所有逻辑卷的容量。
• 节点上所有 PVC 的总已用容量：节点上所有逻辑卷（LV）对应 PVC 的总已用容量。使用命令 lvs -o lv_size,data_percent,snap_percent,metadata_percent <lv_full_name> 可以获取特定逻辑卷（LV）的已用容量（lv_used）。如果不指定 <lv_full_name>，则会列出所有逻辑卷的已用容量。

usage-based Metrics

• IOPs
  • 读取 IOPs：每秒从逻辑卷（LV）完成的读取请求数量。
  • 写入 IOPs：每秒向逻辑卷（LV）完成的写入请求数量。
• Throughput
  • 读取吞吐量：每秒从逻辑卷（LV）读取的字节数。
  • 写入吞吐量：每秒向逻辑卷（LV）写入的字节数。
• Latency
  • 读取延迟：每个读取请求从发起到响应的平均时间（单位：毫秒）。
  • 写入延迟：每个写入请求从发起到响应的平均时间（单位：毫秒）。
• 等待中的 I/O：排队等待处理的读取和写入请求的数量，尚未被服务。
• 状态：逻辑卷（LV）的状态，指示其是“活动”还是“不可用”。