K8s存储演进：从in-tree到CSI架构

# in-tree 插件代码位置示例（已废弃）# kubernetes/pkg/volume/aws_ebs/aws_ebs.go# kubernetes/pkg/volume/azure_disk/azure_disk.go# kubernetes/pkg/volume/gce_pd/gce_pd.go # K8s 社区已将 in-tree 插件标记为 deprecated# 迁移路径：in-tree -> CSI Driver (Migration Feature Gate)

# CSI gRPC 接口定义（简化）# Identity Serviceservice Identity {    rpc GetPluginInfo(GetPluginInfoRequest)        returns (GetPluginInfoResponse) {}    rpc GetPluginCapabilities(GetPluginCapabilitiesRequest)        returns (GetPluginCapabilitiesResponse) {}} # Controller Serviceservice Controller {    rpc CreateVolume(CreateVolumeRequest)        returns (CreateVolumeResponse) {}    rpc DeleteVolume(DeleteVolumeRequest)        returns (DeleteVolumeResponse) {}    rpc ControllerPublishVolume(ControllerPublishVolumeRequest)        returns (ControllerPublishVolumeResponse) {}} # Node Serviceservice Node {    rpc NodeStageVolume(NodeStageVolumeRequest)        returns (NodeStageVolumeResponse) {}    rpc NodePublishVolume(NodePublishVolumeRequest)        returns (NodePublishVolumeResponse) {}}

# CSI 卷生命周期状态流转# Provisioning -> Attaching -> Mounting -> InUse -> Unmounting -> Detaching -> Deleting volume_states = {    "PENDING": "PVC created, waiting for provisioner",    "CREATED": "CreateVolume called, PV exists",    "ATTACHED": "ControllerPublishVolume called",    "STAGED": "NodeStageVolume called",    "PUBLISHED": "NodePublishVolume called, Pod using volume",    "NODE_UNPUBLISHED": "NodeUnpublishVolume called",    "NODE_UNSTAGED": "NodeUnstageVolume called",    "DETACHED": "ControllerUnpublishVolume called",    "DELETED": "DeleteVolume called, PV removed"}

# StorageClass YAML 定义示例apiVersion: storage.k8s.io/v1kind: StorageClassmetadata:  name: fast-ssdprovisioner: ebs.csi.aws.com  # CSI 驱动名parameters:  type: gp3  iops: "3000"  throughput: "125"  encrypted: "true"reclaimPolicy: DeleteallowVolumeExpansion: truevolumeBindingMode: WaitForFirstConsumer

# StatefulSet volumeClaimTemplate 示例apiVersion: apps/v1kind: StatefulSetmetadata:  name: postgresspec:  serviceName: postgres  replicas: 3  template:    spec:      containers:      - name: postgres        image: postgres:15        volumeMounts:        - name: data          mountPath: /var/lib/postgresql/data  volumeClaimTemplates:  - metadata:      name: data    spec:      accessModes: ["ReadWriteOnce"]      storageClassName: fast-ssd      resources:        requests:          storage: 200Gi

import jsonfrom dataclasses import dataclass, fieldfrom typing import Dict, List, Optionalfrom enum import Enum  class VolumeState(Enum):    """CSI 卷生命周期状态。"""    PENDING = "PENDING"    CREATED = "CREATED"    ATTACHED = "ATTACHED"    STAGED = "STAGED"    PUBLISHED = "PUBLISHED"    UNPUBLISHED = "NODE_UNPUBLISHED"    UNSTAGED = "NODE_UNSTAGED"    DETACHED = "DETACHED"    DELETED = "DELETED"  @dataclassclass Topology:    """存储拓扑信息。"""    segments: Dict[str, str] = field(default_factory=dict)     def matches_node(self, node_labels: Dict[str, str]) -> bool:        """检查节点是否在拓扑域内。"""        for key, value in self.segments.items():            if node_labels.get(key) != value:                return False        return True  @dataclassclass CSIVolume:    """CSI 卷对象。"""    volume_id: str    capacity_bytes: int    accessible_topology: List[Topology] = field(default_factory=list)    state: VolumeState = VolumeState.PENDING    node_id: Optional[str] = None    staging_path: Optional[str] = None    publish_path: Optional[str] = None  class CSIDriverSimulator:    """CSI 驱动模拟器：模拟卷生命周期管理。"""     def __init__(self, driver_name: str, version: str):        self.driver_name = driver_name        self.version = version        self.volumes: Dict[str, CSIVolume] = {}        self._counter = 0     def get_plugin_info(self) -> Dict:        """Identity Service: 返回驱动信息。"""        return {            "name": self.driver_name,            "vendor_version": self.version,            "manifest": {"description": "CSI Driver Simulator"}        }     def create_volume(self, name: str, capacity_gb: int,                      topology: Optional[Dict] = None) -> CSIVolume:        """Controller Service: 创建底层存储卷。"""        self._counter += 1        volume_id = f"vol-{self._counter:04d}"        topo = Topology(segments=topology or {})        volume = CSIVolume(            volume_id=volume_id,            capacity_bytes=capacity_gb * 1024**3,            accessible_topology=[topo] if topology else [],            state=VolumeState.CREATED        )        self.volumes[volume_id] = volume        print(f"[CreateVolume] 创建卷 {volume_id}, 容量 {capacity_gb}GB, 拓扑 {topology}")        return volume     def controller_publish_volume(self, volume_id: str, node_id: str) -> bool:        """Controller Service: 将卷附加到节点。"""        vol = self.volumes.get(volume_id)        if not vol:            print(f"[ControllerPublish] 错误: 卷 {volume_id} 不存在")            return False        vol.node_id = node_id        vol.state = VolumeState.ATTACHED        print(f"[ControllerPublish] 卷 {volume_id} 附加到节点 {node_id}")        return True     def node_stage_volume(self, volume_id: str, staging_path: str,                          fs_type: str = "ext4") -> bool:        """Node Service: 格式化并挂载到全局目录。"""        vol = self.volumes.get(volume_id)        if not vol or vol.state != VolumeState.ATTACHED:            print(f"[NodeStage] 错误: 卷 {volume_id} 未附加")            return False        vol.staging_path = staging_path        vol.state = VolumeState.STAGED        print(f"[NodeStage] 卷 {volume_id} 格式化为 {fs_type} 并挂载到 {staging_path}")        return True     def node_publish_volume(self, volume_id: str, target_path: str,                            access_mode: str = "ReadWriteOnce") -> bool:        """Node Service: bind-mount 到 Pod 目录。"""        vol = self.volumes.get(volume_id)        if not vol or vol.state != VolumeState.STAGED:            print(f"[NodePublish] 错误: 卷 {volume_id} 未暂存")            return False        vol.publish_path = target_path        vol.state = VolumeState.PUBLISHED        print(f"[NodePublish] 卷 {volume_id} 挂载到 {target_path} ({access_mode})")        return True     def create_snapshot(self, volume_id: str, snapshot_name: str) -> Dict:        """创建卷快照。"""        vol = self.volumes.get(volume_id)        if not vol:            return {"error": "volume not found"}        snapshot_id = f"snap-{volume_id}-{snapshot_name}"        print(f"[CreateSnapshot] 为卷 {volume_id} 创建快照 {snapshot_id}")        return {"snapshot_id": snapshot_id, "source_volume_id": volume_id}     def delete_volume(self, volume_id: str) -> bool:        """Controller Service: 删除底层存储卷。"""        vol = self.volumes.get(volume_id)        if not vol:            return False        vol.state = VolumeState.DELETED        print(f"[DeleteVolume] 删除卷 {volume_id}")        return True     def get_volume_status(self, volume_id: str) -> Dict:        """获取卷状态。"""        vol = self.volumes.get(volume_id)        if not vol:            return {"error": "not found"}        return {            "volume_id": vol.volume_id,            "state": vol.state.value,            "capacity_gb": vol.capacity_bytes // 1024**3,            "node_id": vol.node_id,            "topology": [t.segments for t in vol.accessible_topology]        }  if __name__ == "__main__":    driver = CSIDriverSimulator("ebs.csi.aws.com", "v1.30.0")     print("=== 驱动信息 ===")    info = driver.get_plugin_info()    print(json.dumps(info, indent=2))     print("\n=== 动态供给卷（拓扑感知）===")    vol = driver.create_volume(        "pg-data-postgres-0",        capacity_gb=200,        topology={"topology.kubernetes.io/zone": "us-east-1a"}    )     print("\n=== 卷附加 ===")    driver.controller_publish_volume(vol.volume_id, "i-0abc123")     print("\n=== 节点暂存 ===")    driver.node_stage_volume(vol.volume_id,                             "/var/lib/kubelet/plugins/ebs.csi.aws.com/staging")     print("\n=== 发布到 Pod ===")    driver.node_publish_volume(vol.volume_id,                               "/var/lib/kubelet/pods/pg-0/volumes/data")     print("\n=== 创建快照 ===")    snap = driver.create_snapshot(vol.volume_id, "daily-backup")    print(json.dumps(snap, indent=2))     print("\n=== 卷状态 ===")    status = driver.get_volume_status(vol.volume_id)    print(json.dumps(status, indent=2, ensure_ascii=False))