node-driver-registry
南风未起 2026-03-11 19:39:22 csi
# 1.简介
# 1.1.nfsdriver
csi-driver-nfs是nfs csi驱动程序，供kubernetes访问nfs-server，支持创建新子目录动态分配持久卷，配合PV/PVC管理volume。
注意
nfs-driver分为controller server和node server两部分，前者负责create/attach volume，后者负责mount volume
# 1.2.interface
CSI驱动需实现NodeServer、IdentityServer和ControllerServer接口，供csi controller和csi node调用实现卷管理。// for grpc type NodeServer interface { // 对于块设备的挂载与卸载 NodeStageVolume(context.Context, *NodeStageVolumeRequest) (*NodeStageVolumeResponse, error) NodeUnstageVolume(context.Context, *NodeUnstageVolumeRequest) (*NodeUnstageVolumeResponse, error) // 文件系统的挂载与卸载 NodePublishVolume(context.Context, *NodePublishVolumeRequest) (*NodePublishVolumeResponse, error) NodeUnpublishVolume(context.Context, *NodeUnpublishVolumeRequest) (*NodeUnpublishVolumeResponse, error) // 状态获取 NodeGetVolumeStats(context.Context, *NodeGetVolumeStatsRequest) (*NodeGetVolumeStatsResponse, error) // 扩容 NodeExpandVolume(context.Context, *NodeExpandVolumeRequest) (*NodeExpandVolumeResponse, error) // 容量获取 NodeGetCapabilities(context.Context, *NodeGetCapabilitiesRequest) (*NodeGetCapabilitiesResponse, error) // 基本信息 NodeGetInfo(context.Context, *NodeGetInfoRequest) (*NodeGetInfoResponse, error) } type ControllerServer interface { // 创建/删除卷 CreateVolume(context.Context, *CreateVolumeRequest) (*CreateVolumeResponse, error) DeleteVolume(context.Context, *DeleteVolumeRequest) (*DeleteVolumeResponse, error) // Attach/Detach ControllerPublishVolume(context.Context, *ControllerPublishRequest) (*ControllerPublishResponse,error) ControllerUnpublishVolume(context.Context, *ControllerUnpublishRequest) (*ControllerUnpublishResponse,error) // 检查卷合法性 ValidateVolumeCapabilities(context.Context, *ValidateRequest) (*ValidateResponse, error) // 卷列表 ListVolumes(context.Context, *ListVolumesRequest) (*ListVolumesResponse, error) // 获取容量 GetCapacity(context.Context, *GetCapacityRequest) (*GetCapacityResponse, error) // 获取支持的功能 ControllerGetCapabilities(context.Context, *ControllerGetRequest) (*ControllerGetResponse, error) // 创建/删除/查询快照 CreateSnapshot(context.Context, *CreateSnapshotRequest) (*CreateSnapshotResponse, error) DeleteSnapshot(context.Context, *DeleteSnapshotRequest) (*DeleteSnapshotResponse, error) ListSnapshots(context.Context, *ListSnapshotsRequest) (*ListSnapshotsResponse, error) // 扩容 ControllerExpandVolume(context.Context, *ControllerExpandRequest) (*ControllerExpandResponse, error) ControllerGetVolume(context.Context, *ControllerGetVolumeRequest) (*ControllerGetVolumeResponse, error) } type IdentityServer interface { // 插件信息 GetPluginInfo(context.Context, *GetPluginInfoRequest) (*GetPluginInfoResponse, error) // 获取插件功能 GetPluginCapabilities(context.Context, *GetCapabilitiesRequest) (*GetCapabilitiesResponse, error) // 探活 Probe(context.Context, *ProbeRequest) (*ProbeResponse, 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
注意
csi controller调用ControllerServer接口，csi node调用NodeServer接口
# 1.3.handler
main入口初始化driver对象，driver封装nfs相关能力及nodeServer，driver.Run()会启动nfs驱动服务。func main() { ... handle() } func handle() { driverOptions := nfs.DriverOptions{ NodeID: *nodeID, DriverName: *driverName, Endpoint: *endpoint, MountPermissions: *mountPermissions, WorkingMountDir: "/tmp", DefaultOnDeletePolicy: "delete", VolStatsCacheExpireInMinutes: 10, RemoveArchivedVolumePath: false, } // 实例化driver d := nfs.NewDriver(&driverOptions) // 启动服务 d.Run(false) }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24注意
driver会声明nfs相关能力，d.Run会一次启动三个接口实现的服务
# 2.入口
# 2.1.driver
driver的初始化相对简单，主要是创建driver及声明nfs支持的能力，以通知上游nfs驱动可以完成的动作及应该调用的接口。func NewDriver(options *DriverOptions) *Driver { // 实例化driver n := &Driver{ name: options.DriverName, version: driverVersion, nodeID: options.NodeID, endpoint: options.Endpoint, // unix socket连接地址 mountPermissions: options.MountPermissions, workingMountDir: "/tmp", volStatsCacheExpireInMinutes: 10, removeArchivedVolumePath: false, } // controller server支持的能力 n.AddControllerServiceCapabilities([]csi.ControllerServiceCapability_RPC_Type{ csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME, // volume创建删除 csi.ControllerServiceCapability_RPC_SINGLE_NODE_MULTI_WRITER, // attached volume支持多写 csi.ControllerServiceCapability_RPC_CLONE_VOLUME, // volume克隆 csi.ControllerServiceCapability_RPC_CREATE_DELETE_SNAPSHOT, // volume快照 }) // node server能力 n.AddNodeServiceCapabilities([]csi.NodeServiceCapability_RPC_Type{ csi.NodeServiceCapability_RPC_GET_VOLUME_STATS, // volume状态查询 csi.NodeServiceCapability_RPC_SINGLE_NODE_MULTI_WRITER, // volume支持mount到多个Pod csi.NodeServiceCapability_RPC_UNKNOWN, }) n.volumeLocks = NewVolumeLocks() // 修正缓存过期时间 if options.VolStatsCacheExpireInMinutes <= 0 { options.VolStatsCacheExpireInMinutes = 10 // default expire in 10 minutes } ... // getter是空实现,缓存需手动设置 getter := func(key string) (interface{}, error) { return nil, nil } // volume状态缓存(避免频繁统计) n.volStatsCache = azcache.NewTimedCache(10min, getter, false) ... // volume删除缓存 n.volDeletionCache = azcache.NewTimedCache(1min, getter, false) ... return n } // NewTimedCache creates a new azcache.Resource. func NewTimedCache(ttl time.Duration, getter GetFunc, disabled bool) (Resource, error) { ... // 对象获取器 provider := &ResourceProvider{ Getter: getter, } // 禁用 if disabled { return provider, nil } // 实例化TTL缓存 timedCache := &TimedCache{ Store: cache.NewStore(cacheKeyFunc), MutexLock: sync.RWMutex{}, TTL: ttl, resourceProvider: provider, } return timedCache, nil }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
注意
driver初始化构造的TTLCache不是用来自动加载资源，而是手动读取写入的缓存器
# 2.2.run
driver.Run()会启动nodeserver、indentifyserver和controllerserver，不同server组合服务于controller或者node。func (n *Driver) Run(testMode bool) { ... // 实例化mounter mounter := mount.New("") // Linux环境强制升级为force mounter if runtime.GOOS == "linux" { mounter = mounter.(mount.MounterForceUnmounter) } // 初始化nodeServer n.ns = NewNodeServer(n, mounter) // 初始化grpcServer s := NewNonBlockingGRPCServer() // 启动grpc s.Start(n.endpoint, NewDefaultIdentityServer(n), NewControllerServer(n), n.ns, false) s.Wait() } // start grpc server func (s *nonBlockingGRPCServer) Start(addr, ids IdentityServer, cs ControllerServer, ns NodeServer, mode bool) { s.wg.Add(1) go s.serve(endpoint, ids, cs, ns, testMode) } func (s *nonBlockingGRPCServer) serve(addr, ids IdentityServer, cs ControllerServer, ns NodeServer, mode bool) { // 解析地址 proto, addr, err := ParseEndpoint(endpoint) ... // unix socker协议 if proto == "unix" { // unix socket文件句柄 addr = "/" + addr // 移除旧的socket os.Remove(addr) ... } // 初始化conn listener listener, err := net.Listen(proto, addr) ... // grpc server server := grpc.NewServer(opts...) s.server = server // 注册identity server if ids != nil { csi.RegisterIdentityServer(server, ids) } // 注册controller server if cs != nil { csi.RegisterControllerServer(server, cs) } // 注册node server if ns != nil { csi.RegisterNodeServer(server, ns) } ... // 启动grpc server.Serve(listener) ... }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
注意
grpc server会注册indentify/node/controller实现，供上游grpc调用完成volume生命周期管理
# 2.3.identity
identity server提供插件信息获取、能力查询及探活接口，能力查询声明有没有controller，值不值得上游调用，探测用于检测服务健康状态。func (ids *IdentityServer) GetPluginInfo(_ context.Context, _ *csi.Request) (*csi.Response, error) { ... return &csi.GetPluginInfoResponse{ Name: ids.Driver.name, VendorVersion: ids.Driver.version, }, nil } // Probe check whether the plugin is running or not. This method does not need to return anything. func (ids *IdentityServer) Probe(_ context.Context, _ *csi.ProbeRequest) (*csi.ProbeResponse, error) { return &csi.ProbeResponse{Ready: &wrapperspb.BoolValue{Value: true}}, nil } func (ids *IdentityServer) GetPluginCapabilities(_ context.Context, _ *csi.Request) (*csi.Response, error) { return &csi.GetPluginCapabilitiesResponse{ Capabilities: []*csi.PluginCapability{ { Type: &csi.PluginCapability_Service_{ Service: &csi.PluginCapability_Service{ // 声明支持controller Type: csi.PluginCapability_Service_CONTROLLER_SERVICE, }, }, }, }, }, nil }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注意
identity server主要用于身份检查，接口实现相对简单
# 3.provision
# 3.1.createVolume
cs.CreateVolume()负责基于请求的volume属性创建存储，属于controller server侧的功能，NFS存储创建本质是创建一个新目录。func (cs *ControllerServer) CreateVolume(ctx Context, req *csi.CreateRequest) (*csi.CreateResponse, error) { // volume名称 name := req.GetName() ... // 检查支持的能力 isValidVolumeCapabilities(req.GetVolumeCapabilities()) ... // 设置volume属性 mountPermissions := cs.Driver.mountPermissions reqCapacity := req.GetCapacityRange().GetRequiredBytes() parameters := req.GetParameters() ... // 参数校验 for k, v := range parameters { switch strings.ToLower(k) { ... case mountPermissionsField: if v != "" { // 权限合法检查 mountPermissions = strconv.ParseUint(v, 8, 32) ... } ... } } // 申请锁 cs.Driver.volumeLocks.TryAcquire(name) ... defer cs.Driver.volumeLocks.Release(name) // 初始化volume对象 nfsVol := newNFSVolume(name, reqCapacity, parameters, cs.Driver.defaultOnDeletePolicy) ... // volume能力声明 if len(req.GetVolumeCapabilities()) > 0 { volCap = req.GetVolumeCapabilities()[0] } // mount basedir cs.internalMount(ctx, nfsVol, parameters, volCap) ... defer func() { // unmount basedir cs.internalUnmount(ctx, nfsVol) ... }() // 创建basedir/{subdir,name} internalVolumePath := getInternalVolumePath(cs.Driver.workingMountDir, nfsVol) os.MkdirAll(internalVolumePath, 0777) ... // 权限调整 if mountPermissions > 0 { // Reset directory permissions because of umask problems os.Chmod(internalVolumePath, os.FileMode(mountPermissions)) ... } // volume设置来源 if req.GetVolumeContentSource() != nil { // clone volume cs.copyVolume(ctx, req, nfsVol) ... } // 设置subdir位置 setKeyValueInMap(parameters, "subdir", nfsVol.subDir) return &csi.CreateVolumeResponse{ Volume: &csi.Volume{ VolumeId: nfsVol.id, // server-basedir-subdir-uuid-{ondelete} CapacityBytes: 0, // setting zero, provisioner will use requested size VolumeContext: parameters, ContentSource: req.GetVolumeContentSource(), // volume来源 }, }, nil }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
75
76
77
78
79
80
81
82
83
84
85
注意
createVolume本质是将basedir挂载到本地，创建subdir后unmount basedir
# 3.2.copyVolume
cs.copyVolume()负责拷贝volume内容，创建volume指定contentSource条件可实现volume clone，达到内容复制效果。func (cs *ControllerServer) copyVolume(ctx context.Context, req *csi.CreateRequest, vol *nfsVolume) error { vs := req.VolumeContentSource switch vs.Type.(type) { // snapshot来源 case *csi.VolumeContentSource_Snapshot: return cs.copyFromSnapshot(ctx, req, vol) // volume来源 case *csi.VolumeContentSource_Volume: return cs.copyFromVolume(ctx, req, vol) default: return status.Errorf(codes.InvalidArgument, "%v not a proper volume source", vs) } } func (cs *ControllerServer) copyFromSnapshot(...) error { // 基于请求构建snap属性 snap, err := getNfsSnapFromID(req.VolumeContentSource.GetSnapshot().GetSnapshotId()) ... // 构造snap volume对象 snapVol := volumeFromSnapshot(snap) ... // snap volume挂在至本地 cs.internalMount(ctx, snapVol, nil, volCap) ... defer func() { cs.internalUnmount(ctx, snapVol) ... }() // dst volume挂载至本地 cs.internalMount(ctx, dstVol, nil, volCap) ... defer func() { cs.internalUnmount(ctx, dstVol) ... }() // workdir/name/{subdir,name}/src.tar snapPath := filepath.Join(getInternalVolumePath(cs.Driver.workingMountDir, snapVol), snap.archiveName()) // workdir/name/{subdir,name} dstPath := getInternalVolumePath(cs.Driver.workingMountDir, dstVol) // snap数据解压至dst volume exec.Command("tar", "-xzvf", snapPath, "-C", dstPath).CombinedOutput() ... return nil } func (cs *ControllerServer) copyFromVolume(...) error { // 基于请求构建src volume对象 srcVol, err := getNfsVolFromID(req.GetVolumeContentSource().GetVolume().GetVolumeId()) ... // workdir/name/{subdir,name}/. srcPath := fmt.Sprintf("%v/.", getInternalVolumePath(cs.Driver.workingMountDir, srcVol)) // workdir/name/{subdir,name} dstPath := getInternalVolumePath(cs.Driver.workingMountDir, dstVol) ... // src volume挂载至本地 cs.internalMount(ctx, srcVol, nil, volCap) ... defer func() { cs.internalUnmount(ctx, srcVol) ... }() // dst volume挂载至本地 cs.internalMount(ctx, dstVol, nil, volCap) ... defer func() { cs.internalUnmount(ctx, dstVol) ... }() // clone数据至dst volume exec.Command("cp", "-a", srcPath, dstPath).CombinedOutput() ... return nil }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
75
76
77
78
79
80
81
82
83
84
85
86
87
88
注意
volume设置来源会触发内容拷贝，本质是由src volume复制内容至dst volume
# 3.3.internalMount
cs.internalMount()将basedir挂载至workingdir，后续基于workingdir新建subdir，完成后执行cs.internalUnmount解除挂载。// Mount nfs server at base-dir func (cs *ControllerServer) internalMount(...) error { ... // basedir作为共享目录 sharePath := filepath.Join(string(filepath.Separator) + vol.baseDir) // workdir/name作为挂载位置 targetPath := getInternalMountPath(cs.Driver.workingMountDir, vol) ... // 执行mount _, err := cs.Driver.ns.NodePublishVolume(ctx, &csi.NodePublishVolumeRequest{ TargetPath: targetPath, VolumeContext: volContext, VolumeCapability: volCap, VolumeId: vol.id, }) return err } // NodePublishVolume mount the volume func (ns *NodeServer) NodePublishVolume(...) (*csi.NodePublishVolumeResponse, error) { // volume属性 volCap := req.GetVolumeCapability() ... volumeID := req.GetVolumeId() ... // workdir/name targetPath := req.GetTargetPath() ... // 申请锁 lockKey := fmt.Sprintf("%s-%s", volumeID, targetPath) ns.Driver.volumeLocks.TryAcquire(lockKey) ... defer ns.Driver.volumeLocks.Release(lockKey) // mount属性 mountOptions := volCap.GetMount().GetMountFlags() if req.GetReadonly() { mountOptions = append(mountOptions, "ro") } ... // server:/basedir server = getServerFromSource(server) source := fmt.Sprintf("%s:%s", server, baseDir) // subdir被过滤,一定是空的 if subDir != "" { // replace pv/pvc name namespace metadata in subDir subDir = replaceWithMap(subDir, subDirReplaceMap) source = fmt.Sprintf("%s/%s", strings.TrimRight(source, "/"), subDir) } // 检测workdir/name挂载过没有 notMnt, err := ns.mounter.IsLikelyNotMountPoint(targetPath) if err != nil { // 目录不存在 if os.IsNotExist(err) { // 创建目录 os.MkdirAll(targetPath, os.FileMode(mountPermissions)) ... notMnt = true } ... } // 挂载过 if !notMnt { return &csi.NodePublishVolumeResponse{}, nil } // 执行mount server:/basedir——>workdir/name ns.mounter.Mount(source, targetPath, "nfs", mountOptions) ... // 权限调整 chmodIfPermissionMismatch(targetPath, os.FileMode(mountPermissions)) ... return &csi.NodePublishVolumeResponse{}, nil }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
75
76
77
78
79
80
81
82
83
84
85
86
注意
volume创建会先将mount basedir至本地，供上层新建subdir实现卷分配
# 3.4.internalUnmount
cs.internalUnmount()会进行mount point卸载，主要复制将mount至本地的basedir释放掉，避免无效挂载造成的泄漏及网络压力。// Unmount nfs server at base-dir func (cs *ControllerServer) internalUnmount(ctx context.Context, vol *nfsVolume) error { // workdir/name targetPath := getInternalMountPath(cs.Driver.workingMountDir, vol) _, err := cs.Driver.ns.NodeUnpublishVolume(ctx, &csi.NodeUnpublishVolumeRequest{ VolumeId: vol.id, TargetPath: targetPath, }) return err } // NodeUnpublishVolume unmount the volume func (ns *NodeServer) NodeUnpublishVolume(...) (*csi.NodeUnpublishVolumeResponse, error) { volumeID := req.GetVolumeId() ... targetPath := req.GetTargetPath() ... // 申请锁 lockKey := fmt.Sprintf("%s-%s", volumeID, targetPath) ns.Driver.volumeLocks.TryAcquire(lockKey) ... defer ns.Driver.volumeLocks.Release(lockKey) ... // unmount point+清理targetPath if forceUnmounter { err = mount.CleanupMountWithForce(targetPath, forceUnmounter, extensiveMountPointCheck, 30*time.Second) } else { err = mount.CleanupMountPoint(targetPath, ns.mounter, extensiveMountPointCheck) } ... return &csi.NodeUnpublishVolumeResponse{}, nil }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注意
卸载相对简单，执行
unmount {-f}和os.remove解除挂载及回收目录
# 3.5.deleteVolume
cs.DeleteVolume()负责基于请求的volume属性回收存储，属于controller server侧的功能，NFS存储回收本质是删除相应子目录。// DeleteVolume delete a volume func (cs *ControllerServer) DeleteVolume(...) (*csi.DeleteVolumeResponse, error) { volumeID := req.GetVolumeId() ... // 基于volID解析volume属性 nfsVol, err := getNfsVolFromID(volumeID) ... // 设置删除策略 if nfsVol.onDelete == "" { nfsVol.onDelete = cs.Driver.defaultOnDeletePolicy } // 申请锁 cs.Driver.volumeLocks.TryAcquire(volumeID) ... defer cs.Driver.volumeLocks.Release(volumeID) // 删除策略不是retain if !strings.EqualFold(nfsVol.onDelete, retain) { // 检查volumeID处理过没有 cache, err := cs.Driver.volDeletionCache.Get(volumeID, azcache.CacheReadTypeDefault) ... if cache != nil { return &csi.DeleteVolumeResponse{}, nil } // volume挂载至本地 cs.internalMount(ctx, nfsVol, nil, volCap) ... defer func() { cs.internalUnmount(ctx, nfsVol) ... }() // workdir/name/{subdir,name} internalVolumePath := getInternalVolumePath(cs.Driver.workingMountDir, nfsVol) // 归档策略 if strings.EqualFold(nfsVol.onDelete, archive) { // workdir/name/archived-{subdir,name} path := filepath.Join(getInternalMountPath(cs.Driver.workingDir, nfsVol), "archived-"+nfsVol.subDir) // subdir是嵌套路径 if strings.Contains(nfsVol.subDir, "/") { // 创建父目录 parentDir := filepath.Dir(path) os.MkdirAll(parentDir, 0777) ... } // 清理已有存档 if cs.Driver.removeArchivedVolumePath { os.RemoveAll(path) ... } // 重命名归档 os.Rename(internalVolumePath, archivedInternalVolumePath) ... // 500s触发一次检测,最长1min,检查volume目录已不存在 waitForPathNotExistWithTimeout(internalVolumePath, time.Minute) ... // 删除策略 } else { // 清理volume目录 os.RemoveAll(internalVolumePath) ... } } // 缓存已删除的volumeID cs.Driver.volDeletionCache.Set(volumeID, "") return &csi.DeleteVolumeResponse{}, nil }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
75
76
77
注意
nfs-driver不支持attach/detach，因此未实现对应接口
# 4.publish
# 4.1.publishVolume
ns.NodePublishVolume()负责实现volume mount，由node-driver-registrar注册至kubelet，供kubelet创建Pod过程驱动。// NodePublishVolume mount the volume func (ns *NodeServer) NodePublishVolume(req *csi.NodePublishRequest) (*csi.NodePublishResponse, error) { // volume属性获取 volCap := req.GetVolumeCapability() ... volumeID := req.GetVolumeId() ... // kubelet传入的挂载点 targetPath := req.GetTargetPath() ... // 申请锁 lockKey := fmt.Sprintf("%s-%s", volumeID, targetPath) ns.Driver.volumeLocks.TryAcquire(lockKey) ... defer ns.Driver.volumeLocks.Release(lockKey) ... // server:/basedir/subdir server = getServerFromSource(server) source := fmt.Sprintf("%s:%s", server, baseDir) if subDir != "" { // replace pv/pvc name namespace metadata in subDir subDir = replaceWithMap(subDir, subDirReplaceMap) source = strings.TrimRight(source, "/") source = fmt.Sprintf("%s/%s", source, subDir) } // targetPath挂载检查 notMnt, err := ns.mounter.IsLikelyNotMountPoint(targetPath) if err != nil { // 目录不存在新建 if os.IsNotExist(err) { os.MkdirAll(targetPath, os.FileMode(mountPermissions)) ... notMnt = true } else { return nil, status.Error(codes.Internal, err.Error()) } } // 已挂载(父子目录dev inode差异) if !notMnt { return &csi.NodePublishVolumeResponse{}, nil } // mount -t nfs <server>:<export_path> <targetPath> -o <options> ns.mounter.Mount(source, targetPath, "nfs", mountOptions) ... // 权限调整 if mountPermissions > 0 { chmodIfPermissionMismatch(targetPath, os.FileMode(mountPermissions)) ... } ... return &csi.NodePublishVolumeResponse{}, nil }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
注意
publishVolume本质是mount，目录会挂到/var/lib/kubelet/pods/<pod-uid>/volumes/kubernetes.io~csi/<volume-name>
# 4.2.unpublishVolume
ns.NodeUnpublishVolume()和publishVolume流程正好相反，会执行unmount卸载volume相关挂载点，释放挂载的存储资源。// NodeUnpublishVolume unmount the volume func (ns *NodeServer) NodeUnpublishVolume(req *csi.NodeUnpublishRequest) (*csi.NodeUnpublishResponse, error) { volumeID := req.GetVolumeId() ... // kubelet提供挂载点 targetPath := req.GetTargetPath() ... // 申请锁 lockKey := fmt.Sprintf("%s-%s", volumeID, targetPath) ns.Driver.volumeLocks.TryAcquire(lockKey) ... defer ns.Driver.volumeLocks.Release(lockKey) ... // 执行unmount if forceUnmounter { mount.CleanupMountWithForce(targetPath, forceUnmounter, extensiveMountPointCheck, 30*time.Second) } else { mount.CleanupMountPoint(targetPath, ns.mounter, extensiveMountPointCheck) } ... return &csi.NodeUnpublishVolumeResponse{}, nil }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注意
unpublish过程相对简单，这里不再具体分析