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
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
|
package driver
import (
"fmt"
"os"
"path/filepath"
"strings"
"sync"
"github.com/container-storage-interface/spec/lib/go/csi"
"github.com/seaweedfs/seaweedfs/weed/glog"
"golang.org/x/net/context"
"google.golang.org/grpc"
"k8s.io/utils/mount"
)
func NewNodeServer(n *SeaweedFsDriver) *NodeServer {
if err := removeDirContent(n.CacheDir); err != nil {
glog.Warning("error cleaning up cache dir")
}
return &NodeServer{
Driver: n,
volumeMutexes: NewKeyMutex(),
}
}
func NewIdentityServer(d *SeaweedFsDriver) *IdentityServer {
return &IdentityServer{
Driver: d,
}
}
func NewControllerServer(d *SeaweedFsDriver) *ControllerServer {
return &ControllerServer{
Driver: d,
}
}
func NewControllerServiceCapability(cap csi.ControllerServiceCapability_RPC_Type) *csi.ControllerServiceCapability {
return &csi.ControllerServiceCapability{
Type: &csi.ControllerServiceCapability_Rpc{
Rpc: &csi.ControllerServiceCapability_RPC{
Type: cap,
},
},
}
}
func ParseEndpoint(ep string) (string, string, error) {
if strings.HasPrefix(strings.ToLower(ep), "unix://") || strings.HasPrefix(strings.ToLower(ep), "tcp://") {
s := strings.SplitN(ep, "://", 2)
if s[1] != "" {
return s[0], s[1], nil
}
}
return "", "", fmt.Errorf("Invalid endpoint: %v", ep)
}
func logGRPC(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
glog.V(3).Infof("GRPC %s request %+v", info.FullMethod, req)
resp, err := handler(ctx, req)
if err != nil {
glog.Errorf("GRPC error: %v", err)
}
glog.V(3).Infof("GRPC %s response %+v", info.FullMethod, resp)
return resp, err
}
func checkMount(targetPath string) (bool, error) {
mounter := mount.New("")
notMnt, err := mount.IsNotMountPoint(mounter, targetPath)
if err != nil {
if os.IsNotExist(err) {
if err = os.MkdirAll(targetPath, 0750); err != nil {
return false, err
}
notMnt = true
} else if mount.IsCorruptedMnt(err) {
if err := mounter.Unmount(targetPath); err != nil {
return false, err
}
notMnt, err = mount.IsNotMountPoint(mounter, targetPath)
} else {
return false, err
}
}
return notMnt, nil
}
func removeDirContent(path string) error {
files, err := filepath.Glob(filepath.Join(path, "*"))
if err != nil {
return err
}
for _, file := range files {
err = os.RemoveAll(file)
if err != nil {
return err
}
}
return nil
}
type KeyMutex struct {
mutexes sync.Map
}
func NewKeyMutex() *KeyMutex {
return &KeyMutex{}
}
func (km *KeyMutex) GetMutex(key string) *sync.Mutex {
m, _ := km.mutexes.LoadOrStore(key, &sync.Mutex{})
return m.(*sync.Mutex)
}
func (km *KeyMutex) RemoveMutex(key string) {
km.mutexes.Delete(key)
}
|
