cubefs

1
// Copyright 2022 The CubeFS Authors.
2
//
3
// Licensed under the Apache License, Version 2.0 (the "License");
4
// you may not use this file except in compliance with the License.
5
// You may obtain a copy of the License at
6
//
7
//     http://www.apache.org/licenses/LICENSE-2.0
8
//
9
// Unless required by applicable law or agreed to in writing, software
10
// distributed under the License is distributed on an "AS IS" BASIS,
11
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
12
// implied. See the License for the specific language governing
13
// permissions and limitations under the License.
14

15
package stat
16

17
import (
18
	"bufio"
19
	"errors"
20
	"fmt"
21
	"io/ioutil"
22
	"os"
23
	"path"
24
	"regexp"
25
	"sort"
26
	"strconv"
27
	"strings"
28
	"sync"
29
	"syscall"
30
	"time"
31

32
	"github.com/cubefs/cubefs/util/log"
33
)
34

35
const (
36
	Stat_Module        = "mem_stat"
37
	FileNameDateFormat = "20060102150405"
38
	ShiftedExtension   = ".old"
39
	PRO_MEM            = "/proc/%d/status"
40

41
	F_OK               = 0
42
	MaxTimeoutLevel    = 3
43
	DefaultStatLogSize = 200 * 1024 * 1024 // 200M
44
	DefaultHeadRoom    = 50 * 1024         // 50G
45
	MaxReservedDays    = 7 * 24 * time.Hour
46
)
47

48
var DefaultTimeOutUs = [MaxTimeoutLevel]uint32{100000, 500000, 1000000}
49

50
var DefaultStatInterval = 60 * time.Second // 60 seconds
51

52
var re = regexp.MustCompile(`\([0-9]*\)`)
53

54
type ShiftedFile []os.FileInfo
55

56
func (f ShiftedFile) Less(i, j int) bool {
57
	return f[i].ModTime().Before(f[j].ModTime())
58
}
59

60
func (f ShiftedFile) Len() int {
61
	return len(f)
62
}
63

64
func (f ShiftedFile) Swap(i, j int) {
65
	f[i], f[j] = f[j], f[i]
66
}
67

68
type typeInfo struct {
69
	typeName  string
70
	allCount  uint32
71
	failCount uint32
72
	maxTime   time.Duration
73
	minTime   time.Duration
74
	allTimeUs time.Duration
75
	timeOut   [MaxTimeoutLevel]uint32
76
}
77

78
type Statistic struct {
79
	logDir        string
80
	logMaxSize    int64
81
	logBaseName   string
82
	pid           int
83
	lastClearTime time.Time
84
	timeOutUs     [MaxTimeoutLevel]uint32
85
	typeInfoMap   map[string]*typeInfo
86
	closeStat     bool
87
	useMutex      bool
88
	sync.Mutex
89
}
90

91
var gSt *Statistic = nil
92

93
func NewStatistic(dir, logModule string, logMaxSize int64, timeOutUs [MaxTimeoutLevel]uint32, useMutex bool) (*Statistic, error) {
94
	dir = path.Join(dir, logModule)
95
	fi, err := os.Stat(dir)
96
	if err != nil {
97
		os.MkdirAll(dir, 0o755)
98
	} else {
99
		if !fi.IsDir() {
100
			return nil, errors.New(dir + " is not a directory")
101
		}
102
	}
103
	_ = os.Chmod(dir, 0o755)
104
	logName := path.Join(dir, Stat_Module)
105
	st := &Statistic{
106
		logDir:        dir,
107
		logMaxSize:    logMaxSize,
108
		logBaseName:   logName,
109
		pid:           os.Getpid(),
110
		lastClearTime: time.Time{},
111
		timeOutUs:     timeOutUs,
112
		typeInfoMap:   make(map[string]*typeInfo),
113
		closeStat:     false,
114
		useMutex:      useMutex,
115
		Mutex:         sync.Mutex{},
116
	}
117

118
	gSt = st
119
	go st.flushScheduler()
120
	return st, nil
121
}
122

123
// TODO: how to close?
124
func (st *Statistic) flushScheduler() {
125
	timer := time.NewTimer(DefaultStatInterval)
126
	defer timer.Stop()
127
	for {
128
		<-timer.C
129

130
		err := WriteStat()
131
		if err != nil {
132
			log.LogErrorf("WriteStat error: %v", err)
133
		}
134

135
		timer.Reset(DefaultStatInterval)
136

137
		fs := syscall.Statfs_t{}
138
		if err := syscall.Statfs(st.logDir, &fs); err != nil {
139
			log.LogErrorf("Get fs stat failed, err: %v", err)
140
			continue
141
		}
142
		diskSpaceLeft := int64(fs.Bavail * uint64(fs.Bsize))
143
		diskSpaceLeft -= DefaultHeadRoom * 1024 * 1024
144
		removeLogFile(diskSpaceLeft, Stat_Module)
145
	}
146
}
147

148
func removeLogFile(diskSpaceLeft int64, module string) {
149
	fInfos, err := ioutil.ReadDir(gSt.logDir)
150
	if err != nil {
151
		log.LogErrorf("ReadDir failed, logDir: %s, err: %v", gSt.logDir, err)
152
		return
153
	}
154
	var needDelFiles ShiftedFile
155
	for _, info := range fInfos {
156
		if deleteFileFilter(info, diskSpaceLeft, module) {
157
			needDelFiles = append(needDelFiles, info)
158
		}
159
	}
160
	sort.Sort(needDelFiles)
161
	for _, info := range needDelFiles {
162
		if err = os.Remove(path.Join(gSt.logDir, info.Name())); err != nil {
163
			log.LogErrorf("Remove log file failed, logFileName: %s, err: %v", info.Name(), err)
164
			continue
165
		}
166
		diskSpaceLeft += info.Size()
167
		if diskSpaceLeft > 0 && time.Since(info.ModTime()) < MaxReservedDays {
168
			break
169
		}
170
	}
171
}
172

173
func deleteFileFilter(info os.FileInfo, diskSpaceLeft int64, module string) bool {
174
	if diskSpaceLeft <= 0 {
175
		return info.Mode().IsRegular() && strings.HasSuffix(info.Name(), ShiftedExtension) && strings.HasPrefix(info.Name(), module)
176
	}
177
	return time.Since(info.ModTime()) > MaxReservedDays && strings.HasSuffix(info.Name(), ShiftedExtension) && strings.HasPrefix(info.Name(), module)
178
}
179

180
func CloseStat() {
181
	if gSt == nil {
182
		return
183
	}
184

185
	gSt.closeStat = true
186
}
187

188
func BeginStat() (bgTime *time.Time) {
189
	bg := time.Now()
190
	return &bg
191
}
192

193
func EndStat(typeName string, err error, bgTime *time.Time, statCount uint32) error {
194
	if gSt == nil {
195
		return nil
196
	}
197

198
	if gSt.closeStat {
199
		return nil
200
	}
201

202
	if gSt.useMutex {
203
		gSt.Lock()
204
		defer gSt.Unlock()
205
	}
206

207
	if err != nil {
208
		newErrStr := string(re.ReplaceAll([]byte(err.Error()), []byte("(xxx)")))
209
		baseLen := len(typeName) + 2
210
		if len(newErrStr)+baseLen > 41 {
211
			typeName = typeName + "[" + newErrStr[:41-baseLen] + "]"
212
		} else {
213
			typeName = typeName + "[" + newErrStr + "]"
214
		}
215
	}
216

217
	return addStat(typeName, err, bgTime, statCount)
218
}
219

220
func WriteStat() error {
221
	if gSt == nil {
222
		return nil
223
	}
224

225
	if gSt.useMutex {
226
		gSt.Lock()
227
		defer gSt.Unlock()
228
	}
229

230
	logFileName := gSt.logBaseName + ".log"
231
	statFile, err := os.OpenFile(logFileName, os.O_RDWR|os.O_APPEND|os.O_CREATE, 0o666)
232
	if err != nil {
233
		log.LogErrorf("OpenLogFile failed, logFileName: %s, err: %v\n", logFileName, err)
234
		return fmt.Errorf("OpenLogFile failed, logFileName %s\n", logFileName)
235
	}
236
	defer statFile.Close()
237

238
	statSpan := time.Since(gSt.lastClearTime) / 1e9
239
	ioStream := bufio.NewWriter(statFile)
240
	defer ioStream.Flush()
241

242
	fmt.Fprintf(ioStream, "\n===============  Statistic in %ds, %s  =====================\n",
243
		statSpan, time.Now().Format("2006-01-02 15:04:05"))
244

245
	if virt, res, err := GetProcessMemory(gSt.pid); err != nil {
246
		log.LogErrorf("Get process memory failed, err: %v", err)
247
		fmt.Fprintf(ioStream, "Get Mem Failed.\n")
248
	} else {
249
		fmt.Fprintf(ioStream, "Mem Allocated(kB): VIRT %-10d   RES %-10d\n", virt, res)
250
	}
251

252
	fmt.Fprintf(ioStream, "%-42s|%10s|%8s|%8s|%8s|%8s|%8s|%8s|%8s|\n",
253
		"", "TOTAL", "FAILED", "AVG(ms)", "MAX(ms)", "MIN(ms)",
254
		">"+strconv.Itoa(int(gSt.timeOutUs[0])/1000)+"ms",
255
		">"+strconv.Itoa(int(gSt.timeOutUs[1])/1000)+"ms",
256
		">"+strconv.Itoa(int(gSt.timeOutUs[2])/1000)+"ms")
257

258
	typeNames := make([]string, 0)
259
	for typeName := range gSt.typeInfoMap {
260
		typeNames = append(typeNames, typeName)
261
	}
262

263
	sort.Strings(typeNames)
264
	for _, typeName := range typeNames {
265
		typeInfo := gSt.typeInfoMap[typeName]
266
		avgUs := int32(0)
267
		if typeInfo.allCount > 0 {
268
			avgUs = int32(typeInfo.allTimeUs / time.Duration(typeInfo.allCount))
269
		}
270

271
		fmt.Fprintf(ioStream, "%-42s|%10d|%8d|%8.2f|%8.2f|%8.2f|%8d|%8d|%8d|\n",
272
			typeInfo.typeName, typeInfo.allCount, typeInfo.failCount,
273
			float32(avgUs)/1000, float32(typeInfo.maxTime)/1000, float32(typeInfo.minTime)/1000,
274
			typeInfo.timeOut[0], typeInfo.timeOut[1], typeInfo.timeOut[2])
275
	}
276

277
	fmt.Fprintf(ioStream, "-------------------------------------------------------------------"+
278
		"--------------------------------------------------\n")
279

280
	// clear stat
281
	gSt.lastClearTime = time.Now()
282
	gSt.typeInfoMap = make(map[string]*typeInfo)
283

284
	shiftFiles()
285

286
	return nil
287
}
288

289
func ClearStat() {
290
	if gSt == nil {
291
		return
292
	}
293

294
	if gSt.useMutex {
295
		gSt.Lock()
296
		defer gSt.Unlock()
297
	}
298

299
	gSt.lastClearTime = time.Now()
300
	gSt.typeInfoMap = make(map[string]*typeInfo)
301
}
302

303
func addStat(typeName string, err error, bgTime *time.Time, statCount uint32) error {
304
	if gSt == nil {
305
		return nil
306
	}
307

308
	if len(typeName) == 0 {
309
		return fmt.Errorf("AddStat fail, typeName %s\n", typeName)
310
	}
311

312
	if typeInfo, ok := gSt.typeInfoMap[typeName]; ok {
313
		typeInfo.allCount += statCount
314
		if err != nil {
315
			typeInfo.failCount += statCount
316
		}
317
		addTime(typeInfo, bgTime)
318
		return nil
319
	}
320

321
	typeInfo := &typeInfo{
322
		typeName:  typeName,
323
		allCount:  statCount,
324
		failCount: 0,
325
		maxTime:   0,
326
		minTime:   0,
327
		allTimeUs: 0,
328
		timeOut:   [3]uint32{},
329
	}
330

331
	if err != nil {
332
		typeInfo.failCount = statCount
333
	}
334

335
	gSt.typeInfoMap[typeName] = typeInfo
336
	addTime(typeInfo, bgTime)
337

338
	return nil
339
}
340

341
func addTime(typeInfo *typeInfo, bgTime *time.Time) {
342
	if bgTime == nil {
343
		return
344
	}
345

346
	timeCostUs := time.Since(*bgTime) / 1e3
347
	if timeCostUs == 0 {
348
		return
349
	}
350

351
	if timeCostUs >= time.Duration(gSt.timeOutUs[0]) && timeCostUs < time.Duration(gSt.timeOutUs[1]) {
352
		typeInfo.timeOut[0]++
353
	} else if timeCostUs >= time.Duration(gSt.timeOutUs[1]) && timeCostUs < time.Duration(gSt.timeOutUs[2]) {
354
		typeInfo.timeOut[1]++
355
	} else if timeCostUs > time.Duration(gSt.timeOutUs[2]) {
356
		typeInfo.timeOut[2]++
357
	}
358

359
	if timeCostUs > typeInfo.maxTime {
360
		typeInfo.maxTime = timeCostUs
361
	}
362
	if typeInfo.minTime == 0 || timeCostUs < typeInfo.minTime {
363
		typeInfo.minTime = timeCostUs
364
	}
365

366
	typeInfo.allTimeUs += timeCostUs
367
}
368

369
func shiftFiles() error {
370
	logFileName := gSt.logBaseName + ".log"
371
	fileInfo, err := os.Stat(logFileName)
372
	if err != nil {
373
		return err
374
	}
375

376
	if fileInfo.Size() < gSt.logMaxSize {
377
		return nil
378
	}
379

380
	if syscall.Access(logFileName, F_OK) == nil {
381
		logNewFileName := logFileName + "." + time.Now().Format(
382
			FileNameDateFormat) + ShiftedExtension
383
		if syscall.Rename(logFileName, logNewFileName) != nil {
384
			log.LogErrorf("RenameFile failed, logFileName: %s, logNewFileName: %s, err: %v\n",
385
				logFileName, logNewFileName, err)
386
			return fmt.Errorf("RenameFile failed, logFileName %s, logNewFileName %s\n",
387
				logFileName, logNewFileName)
388
		}
389
	}
390

391
	return nil
392
}
393

394
func StatBandWidth(typeName string, Size uint32) {
395
	EndStat(typeName+"[FLOW_KB]", nil, nil, Size/1024)
396
}
397

398
func GetMememory() (Virt, Res uint64, err error) {
399
	return GetProcessMemory(gSt.pid)
400
}
401

402
func GetProcessMemory(pid int) (Virt, Res uint64, err error) {
403
	proFileName := fmt.Sprintf(PRO_MEM, pid)
404
	fp, err := os.Open(proFileName)
405
	if err != nil {
406
		return
407
	}
408
	defer fp.Close()
409
	scan := bufio.NewScanner(fp)
410
	for scan.Scan() {
411
		line := scan.Text()
412
		fields := strings.Split(line, ":")
413
		key := fields[0]
414
		if key == "VmRSS" {
415
			value := strings.TrimSpace(fields[1])
416
			value = strings.Replace(value, " kB", "", -1)
417
			Res, err = strconv.ParseUint(value, 10, 64)
418
			if err != nil {
419
				return
420
			}
421
		} else if key == "VmSize" {
422
			value := strings.TrimSpace(fields[1])
423
			value = strings.Replace(value, " kB", "", -1)
424
			Virt, err = strconv.ParseUint(value, 10, 64)
425
			if err != nil {
426
				return
427
			}
428
		} else {
429
			continue
430
		}
431
	}
432
	return
433
}
434