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
|
package driver
import (
"fmt"
"os"
"strings"
"sync"
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/util"
"github.com/container-storage-interface/spec/lib/go/csi"
"golang.org/x/net/context"
"google.golang.org/grpc"
"k8s.io/utils/mount"
)
func NewNodeServer(n *SeaweedFsDriver) *NodeServer {
return &NodeServer{
Driver: n,
volumeMutexes: NewKeyMutex(32),
}
}
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) {
notMnt, err := mount.New("").IsLikelyNotMountPoint(targetPath)
if err != nil {
if os.IsNotExist(err) {
if err = os.MkdirAll(targetPath, 0750); err != nil {
return false, err
}
notMnt = true
} else {
return false, err
}
}
return notMnt, nil
}
type KeyMutex struct {
mutexes []sync.RWMutex
size int32
}
func NewKeyMutex(size int32) *KeyMutex {
return &KeyMutex{
mutexes: make([]sync.RWMutex, size),
size: size,
}
}
func (km *KeyMutex) GetMutex(key string) *sync.RWMutex {
index := util.HashToInt32([]byte(key))
if index < 0 {
index = -index
}
return &km.mutexes[index%km.size]
}
|
