cubefs

cfs.go
242 строки · 5.9 Кб
Перенос по словам
1
package main
2

3
import (
4
	"fmt"
5
	"log"
6
	"os"
7
	"os/exec"
8
	"reflect"
9
	"strings"
10
	"time"
11

12
	"golang.org/x/text/cases"
13
	"golang.org/x/text/language"
14
	"k8s.io/mount-utils"
15
)
16

17
type cfsOption struct {
18
	AccessKey            string `json:"accessKey"`
19
	AttrValid            string `json:"attrValid"`
20
	Authenticate         string `json:"authenticate"`
21
	AutoInvalData        string `json:"autoInvalData"`
22
	BcacheBatchCnt       string `json:"bcacheBatchCnt"`
23
	BcacheCheckIntervalS string `json:"bcacheCheckIntervalS"`
24
	BcacheDir            string `json:"bcacheDir"`
25
	BcacheFilterFiles    string `json:"bcacheFilterFiles"`
26
	BuffersTotalLimit    string `json:"buffersTotalLimit"`
27
	C                    string `json:"c"`
28
	CacheAction          string `json:"cacheAction"`
29
	CertFile             string `json:"certFile"`
30

31
	ClientKey            string `json:"clientKey"`
32
	DisableDcache        string `json:"disableDcache"`
33
	EbsBlockSize         string `json:"ebsBlockSize"`
34
	EbsEndpoint          string `json:"ebsEndpoint"`
35
	EbsServerPath        string `json:"ebsServerPath"`
36
	EnablePosixACL       string `json:"enablePosixACL"`
37
	EnSyncWrite          string `json:"enSyncWrite"`
38
	EnableAudit          string `json:"enableAudit"`
39
	EnableHTTPS          string `json:"enableHTTPS"`
40
	EnableSummary        string `json:"enableSummary"`
41
	EnableUnixPermission string `json:"enableUnixPermission"`
42
	EnableXattr          string `json:"enableXattr"`
43
	FollowerRead         string `json:"followerRead"`
44
	FsyncOnClose         string `json:"fsyncOnClose"`
45
	IcacheTimeout        string `json:"icacheTimeout"`
46
	Keepcache            string `json:"keepcache"`
47
	LogDir               string `json:"logDir"`
48
	LogLevel             string `json:"logLevel"`
49
	LookupValid          string `json:"lookupValid"`
50
	MasterAddr           string `json:"masterAddr"`
51
	MaxStreamerLimit     string `json:"maxStreamerLimit"`
52
	Maxcpus              string `json:"maxcpus"`
53
	MetaSendTimeout      string `json:"metaSendTimeout"`
54
	MountPoint           string `json:"mountPoint"`
55
	N                    string `json:"n"`
56
	NearRead             string `json:"nearRead"`
57
	Owner                string `json:"owner"`
58
	P                    string `json:"p"`
59
	ProfPort             string `json:"profPort"`
60
	R                    string `json:"r"`
61
	Rdonly               string `json:"rdonly"`
62
	ReadRate             string `json:"readRate"`
63
	ReadThreads          string `json:"readThreads"`
64
	S                    string `json:"s"`
65
	SecretKey            string `json:"secretKey"`
66
	Subdir               string `json:"subdir"`
67
	TicketHost           string `json:"ticketHost"`
68
	VolName              string `json:"volName"`
69
	VolType              string `json:"volType"`
70
	WarnLogDir           string `json:"warnLogDir"`
71
	WriteRate            string `json:"writeRate"`
72
	WriteThreads         string `json:"writeThreads"`
73
	Writecache           string `json:"writecache"`
74
}
75

76
func (c *cfsOption) ConvertToCliOptions() (s []string) {
77
	tp := reflect.TypeOf(*c)
78
	el := reflect.ValueOf(c).Elem()
79

80
	// run foreground, since "daemon start failed" issue
81
	s = append(s, "-f")
82

83
	for i := 0; i < tp.NumField(); i++ {
84
		fi := tp.Field(i)
85

86
		js := fi.Tag.Get("json")
87
		if js == "" {
88
			continue
89
		}
90

91
		v := strings.Trim(el.FieldByName(fi.Name).String(), " ")
92
		if v != "" {
93
			if js == "n" || js == "r" {
94
				s = append(s, "-"+js)
95
			} else {
96
				s = append(s, "-"+js, v)
97
			}
98
		}
99

100
	}
101

102
	return s
103
}
104

105
// cfsParse Parse from string
106
func cfsParse(mountPoint, options string) (c *cfsOption) {
107
	c = &cfsOption{MountPoint: mountPoint}
108

109
	sli := strings.Split(options, ",")
110

111
	el := reflect.ValueOf(c).Elem()
112

113
	cas := cases.Title(language.Und, cases.NoLower)
114

115
	for _, v := range sli {
116
		oSli := strings.Split(v, "=")
117
		if len(oSli) == 0 {
118
			continue
119
		}
120

121
		field := el.FieldByName(cas.String(oSli[0]))
122
		if !field.CanSet() {
123
			log.Println("Warn: field can not set. field: ", oSli[0])
124
			continue
125
		}
126

127
		if len(oSli) == 1 {
128
			field.SetString("1")
129
		} else {
130
			field.SetString(strings.Join(oSli[1:], "="))
131
		}
132

133
	}
134

135
	return
136
}
137

138
func cfsIsMounted(mountPoint string) bool {
139
	mo := mount.New("")
140

141
	mps, err := mo.List()
142
	if err != nil {
143
		log.Fatalf("mountListErr, err: %s", err.Error())
144
	}
145

146
	for _, v := range mps {
147
		if v.Path == mountPoint {
148
			return true
149
		}
150
	}
151

152
	return false
153
}
154

155
// cfsMountPre exception handling
156
func cfsMountPre(mountPoint string) error {
157
	if _, err := os.Stat(mountPoint); os.IsNotExist(err) {
158
		return err
159
	}
160

161
	if cfsIsMounted(mountPoint) {
162
		return cfsUmount(mountPoint)
163
	}
164

165
	return nil
166
}
167

168
func cfsList() error {
169
	mo := mount.New("")
170

171
	mps, err := mo.List()
172
	if err != nil {
173
		log.Fatalf("mountListErr, err: %s", err.Error())
174
	}
175

176
	for _, v := range mps {
177
		if v.Type == "fuse" || v.Type == "fuse.cubefs" {
178
			fmt.Println(fmt.Sprintf("%s on %s type %s (%s)", v.Device, v.Path, v.Type, strings.Join(v.Opts, ",")))
179
		}
180
	}
181

182
	return nil
183
}
184

185
func cfsMount(mountPoint, options string) error {
186
	err := cfsMountPre(mountPoint)
187
	if err != nil {
188
		return err
189
	}
190

191
	c := cfsParse(mountPoint, options)
192

193
	co := c.ConvertToCliOptions()
194

195
	log.Println("cfsMount commands: ", getCfsClientPath(), strings.Join(co, " "))
196
	cmd := exec.Command(getCfsClientPath(), co...)
197
	cmd.Env = append(cmd.Env, "PATH=/usr/bin/:/bin/:$PATH")
198
	cmd.Stdout = logFile
199
	cmd.Stderr = logFile
200
	log.Println("[info] cfsMount, env: ", cmd.Env)
201

202
	err = cmd.Start()
203
	if err != nil {
204
		log.Println("cfsMountCmdStartErr, err: ", err.Error())
205
		return err
206
	}
207

208
	log.Println("[info] cfsMount started")
209

210
	err = cfsMountPost(mountPoint)
211
	if err != nil {
212
		return err
213
	}
214

215
	// err = cmd.Wait()
216
	// if err != nil {
217
	//     log.Println("[error] cfsMountWaitErr, err: ", err)
218
	//     return nil
219
	// }
220

221
	return nil
222
}
223

224
// cfsMountPost async mount check
225
func cfsMountPost(mountPoint string) error {
226
	log.Println("[info] cfsMountPost run")
227

228
	for i := 0; i < 20*5; i++ {
229
		if cfsIsMounted(mountPoint) {
230
			log.Println("[info] cfsMount mounted")
231
			break
232
		}
233
		time.Sleep(50 * time.Millisecond)
234
	}
235

236
	return nil
237
}
238

239
func cfsUmount(mountPoint string) error {
240
	mo := mount.New("")
241
	return mo.Unmount(mountPoint)
242
}
243
cubefs

Использование cookies
