cubefs

1
// Copyright 2018 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 ump
16

17
import (
18
	"bytes"
19
	"encoding/json"
20
	"fmt"
21
	"net"
22
	"os"
23
	"strings"
24
)
25

26
type FunctionTp struct {
27
	Time         string `json:"time"`
28
	Key          string `json:"key"`
29
	HostName     string `json:"hostname"`
30
	ProcessState string `json:"processState"`
31
	ElapsedTime  string `json:"elapsedTime"`
32
}
33

34
type SystemAlive struct {
35
	Key      string `json:"key"`
36
	HostName string `json:"hostname"`
37
	Time     string `json:"time"`
38
}
39

40
type BusinessAlarm struct {
41
	Time         string `json:"time"`
42
	Key          string `json:"key"`
43
	HostName     string `json:"hostname"`
44
	BusinessType string `json:"type"`
45
	Value        string `json:"value"`
46
	Detail       string `json:"detail"`
47
}
48

49
const (
50
	FunctionTpSufixx    = "tp.log"
51
	SystemAliveSufixx   = "alive.log"
52
	BusinessAlarmSufixx = "business.log"
53
	LogFileOpt          = os.O_RDWR | os.O_CREATE | os.O_APPEND
54
	ChSize              = 102400
55
	BusinessAlarmType   = "BusinessAlarm"
56
	SystemAliveType     = "SystemAlive"
57
	FunctionTpType      = "FunctionTp"
58
	HostNameFile        = "/proc/sys/kernel/hostname"
59
	MaxLogSize          = 1024 * 1024 * 10
60
)
61

62
var (
63
	FunctionTpLogWrite    = &LogWrite{logCh: make(chan interface{}, ChSize)}
64
	SystemAliveLogWrite   = &LogWrite{logCh: make(chan interface{}, ChSize)}
65
	BusinessAlarmLogWrite = &LogWrite{logCh: make(chan interface{}, ChSize)}
66
	UmpDataDir            = "/export/home/tomcat/UMP-Monitor/logs/"
67
)
68

69
type LogWrite struct {
70
	logCh       chan interface{}
71
	logName     string
72
	logSize     int64
73
	seq         int
74
	logSufixx   string
75
	logFp       *os.File
76
	sigCh       chan bool
77
	bf          *bytes.Buffer
78
	jsonEncoder *json.Encoder
79
}
80

81
func (lw *LogWrite) initLogFp(sufixx string) (err error) {
82
	var fi os.FileInfo
83
	lw.seq = 0
84
	lw.sigCh = make(chan bool, 1)
85
	lw.logSufixx = sufixx
86
	lw.logName = fmt.Sprintf("%s%s%s", UmpDataDir, "ump_", lw.logSufixx)
87
	lw.bf = bytes.NewBuffer([]byte{})
88
	lw.jsonEncoder = json.NewEncoder(lw.bf)
89
	lw.jsonEncoder.SetEscapeHTML(false)
90
	if lw.logFp, err = os.OpenFile(lw.logName, LogFileOpt, 0o666); err != nil {
91
		return
92
	}
93
	if fi, err = lw.logFp.Stat(); err != nil {
94
		return
95
	}
96
	lw.logSize = fi.Size()
97

98
	return
99
}
100

101
func (lw *LogWrite) backGroundCheckFile() (err error) {
102
	if lw.logSize <= MaxLogSize {
103
		return
104
	}
105
	lw.logFp.Close()
106
	lw.seq++
107
	if lw.seq > 3 {
108
		lw.seq = 1
109
	}
110

111
	name := fmt.Sprintf("%s%s%s.%d", UmpDataDir, "ump_", lw.logSufixx, lw.seq)
112
	if _, err = os.Stat(name); err == nil {
113
		os.Remove(name)
114
	}
115
	os.Rename(lw.logName, name)
116

117
	if lw.logFp, err = os.OpenFile(lw.logName, LogFileOpt, 0o666); err != nil {
118
		lw.seq--
119
		return
120
	}
121
	if err = os.Truncate(lw.logName, 0); err != nil {
122
		lw.seq--
123
		return
124
	}
125
	lw.logSize = 0
126

127
	return
128
}
129

130
func (lw *LogWrite) backGroundWrite(umpType string) {
131
	for {
132
		var body []byte
133
		obj := <-lw.logCh
134
		switch umpType {
135
		case FunctionTpType:
136
			tp := obj.(*FunctionTp)
137
			lw.jsonEncoder.Encode(tp)
138
			body = append(body, lw.bf.Bytes()...)
139
			lw.bf.Reset()
140
			FunctionTpPool.Put(tp)
141
		case SystemAliveType:
142
			alive := obj.(*SystemAlive)
143
			lw.jsonEncoder.Encode(alive)
144
			body = append(body, lw.bf.Bytes()...)
145
			lw.bf.Reset()
146
			SystemAlivePool.Put(alive)
147
		case BusinessAlarmType:
148
			alarm := obj.(*BusinessAlarm)
149
			lw.jsonEncoder.Encode(alarm)
150
			body = append(body, lw.bf.Bytes()...)
151
			lw.bf.Reset()
152
			AlarmPool.Put(alarm)
153
		default:
154
			// do nothing
155
		}
156
		if lw.backGroundCheckFile() != nil {
157
			continue
158
		}
159
		body = append(body, []byte("\n")...)
160
		lw.logFp.Write(body)
161
		lw.logSize += (int64)(len(body))
162
	}
163
}
164

165
func initLogName(module, dataDir string) (err error) {
166
	if dataDir != "" {
167
		UmpDataDir = dataDir
168
		if !strings.HasSuffix(UmpDataDir, "/") {
169
			UmpDataDir += "/"
170
		}
171
	} else {
172
		return fmt.Errorf("warnLogDir dir not config")
173
	}
174
	if err = os.MkdirAll(UmpDataDir, 0o755); err != nil {
175
		return
176
	}
177

178
	if HostName, err = GetLocalIpAddr(); err != nil {
179
		return
180
	}
181

182
	if err = FunctionTpLogWrite.initLogFp(module + "_" + FunctionTpSufixx); err != nil {
183
		return
184
	}
185

186
	if err = SystemAliveLogWrite.initLogFp(module + "_" + SystemAliveSufixx); err != nil {
187
		return
188
	}
189

190
	if err = BusinessAlarmLogWrite.initLogFp(module + "_" + BusinessAlarmSufixx); err != nil {
191
		return
192
	}
193

194
	return
195
}
196

197
func GetLocalIpAddr() (localAddr string, err error) {
198
	addrs, err := net.InterfaceAddrs()
199
	if err != nil {
200
		return
201
	}
202
	for _, addr := range addrs {
203
		if ipNet, isIpNet := addr.(*net.IPNet); isIpNet && !ipNet.IP.IsLoopback() {
204
			if ipv4 := ipNet.IP.To4(); ipv4 != nil {
205
				localAddr = ipv4.String()
206
				return
207
			}
208
		}
209
	}
210
	err = fmt.Errorf("cannot get local ip")
211
	return
212
}
213

214
func backGroudWrite() {
215
	go FunctionTpLogWrite.backGroundWrite(FunctionTpType)
216
	go SystemAliveLogWrite.backGroundWrite(SystemAliveType)
217
	go BusinessAlarmLogWrite.backGroundWrite(BusinessAlarmType)
218
}
219