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 exporter
16

17
import (
18
	"fmt"
19
	"net"
20
	"net/http"
21
	"os"
22
	"path/filepath"
23
	"strconv"
24
	"strings"
25
	"time"
26

27
	"github.com/cubefs/cubefs/proto"
28
	"github.com/cubefs/cubefs/util/config"
29
	"github.com/cubefs/cubefs/util/log"
30
	"github.com/gorilla/mux"
31
	"github.com/prometheus/client_golang/prometheus"
32
	"github.com/prometheus/client_golang/prometheus/promhttp"
33
	"github.com/prometheus/client_golang/prometheus/push"
34
)
35

36
const (
37
	PromHandlerPattern      = "/metrics"       // prometheus handler
38
	AppName                 = "cfs"            // app name
39
	ConfigKeyExporterEnable = "exporterEnable" // exporter enable
40
	ConfigKeyExporterPort   = "exporterPort"   // exporter port
41
	ConfigKeyConsulAddr     = "consulAddr"     // consul addr
42
	ConfigKeyConsulMeta     = "consulMeta"     // consul meta
43
	ConfigKeyIpFilter       = "ipFilter"       // add ip filter
44
	ConfigKeyEnablePid      = "enablePid"      // enable report partition id
45
	ConfigKeyPushAddr       = "pushAddr"       // enable push data to gateway
46
	ChSize                  = 1024 * 10        // collect chan size
47

48
	// monitor label name
49
	Vol    = "vol"
50
	Disk   = "disk"
51
	PartId = "partid"
52
	Op     = "op"
53
	Type   = "type"
54
	Err    = "err"
55
)
56

57
var (
58
	namespace         string
59
	clustername       string
60
	modulename        string
61
	pushAddr          string
62
	exporterPort      int64
63
	enabledPrometheus = false
64
	enablePush        = false
65
	EnablePid         = false
66
	replacer          = strings.NewReplacer("-", "_", ".", "_", " ", "_", ",", "_", ":", "_")
67
	registry          = prometheus.NewRegistry()
68
)
69

70
func metricsName(name string) string {
71
	return replacer.Replace(fmt.Sprintf("%s_%s", namespace, name))
72
}
73

74
// Init initializes the exporter.
75
func Init(role string, cfg *config.Config) {
76
	modulename = role
77
	if !cfg.GetBoolWithDefault(ConfigKeyExporterEnable, true) {
78
		log.LogInfof("%v exporter disabled", role)
79
		return
80
	}
81

82
	EnablePid = cfg.GetBoolWithDefault(ConfigKeyEnablePid, false)
83
	log.LogInfo("enable report partition id info? ", EnablePid)
84

85
	port := cfg.GetInt64(ConfigKeyExporterPort)
86

87
	if port < 0 {
88
		log.LogInfof("%v exporter port set random default", port)
89
	}
90

91
	exporterPort = port
92
	enabledPrometheus = true
93

94
	pushAddr = cfg.GetString(ConfigKeyPushAddr)
95
	log.LogInfof("pushAddr %v ", pushAddr)
96
	if pushAddr != "" {
97
		enablePush = true
98
	}
99

100
	http.Handle(PromHandlerPattern, promhttp.HandlerFor(prometheus.DefaultGatherer, promhttp.HandlerOpts{
101
		Timeout: 60 * time.Second,
102
	}))
103

104
	namespace = AppName + "_" + role
105
	addr := fmt.Sprintf(":%d", port)
106
	l, err := net.Listen("tcp", addr)
107
	if err != nil {
108
		log.LogError("exporter tcp listen error: ", err)
109
		return
110
	}
111

112
	exporterPort = int64(l.Addr().(*net.TCPAddr).Port)
113

114
	go func() {
115
		err = http.Serve(l, nil)
116
		if err != nil {
117
			log.LogError("exporter http serve error: ", err)
118
			return
119
		}
120
	}()
121

122
	collect()
123

124
	m := NewGauge("start_time")
125
	m.Set(float64(time.Now().Unix() * 1000))
126

127
	log.LogInfof("exporter Start: %v", exporterPort)
128
}
129

130
// Init initializes the exporter.
131
func InitWithRouter(role string, cfg *config.Config, router *mux.Router, exPort string) {
132
	modulename = role
133
	if !cfg.GetBoolWithDefault(ConfigKeyExporterEnable, true) {
134
		log.LogInfof("%v metrics exporter disabled", role)
135
		return
136
	}
137
	exporterPort, _ = strconv.ParseInt(exPort, 10, 64)
138
	enabledPrometheus = true
139
	router.NewRoute().Name("metrics").
140
		Methods(http.MethodGet).
141
		Path(PromHandlerPattern).
142
		Handler(promhttp.HandlerFor(prometheus.DefaultGatherer, promhttp.HandlerOpts{
143
			Timeout: 5 * time.Second,
144
		}))
145
	namespace = AppName + "_" + role
146

147
	collect()
148

149
	m := NewGauge("start_time")
150
	m.Set(float64(time.Now().Unix() * 1000))
151

152
	log.LogInfof("exporter Start: %v %v", exporterPort, m)
153
}
154

155
func RegistConsul(cluster string, role string, cfg *config.Config) {
156
	ipFilter := cfg.GetString(ConfigKeyIpFilter)
157
	host, err := GetLocalIpAddr(ipFilter)
158
	if err != nil {
159
		log.LogErrorf("get local ip error, %v", err.Error())
160
		return
161
	}
162

163
	rawmnt := cfg.GetString("subdir")
164
	if rawmnt == "" {
165
		rawmnt = "/"
166
	}
167
	mountPoint, _ := filepath.Abs(rawmnt)
168
	log.LogInfof("RegistConsul:%v", enablePush)
169
	if enablePush {
170
		log.LogWarnf("[RegisterConsul] use auto push data strategy, not register consul")
171
		autoPush(pushAddr, role, cluster, host, mountPoint)
172
		return
173
	}
174

175
	clustername = replacer.Replace(cluster)
176
	consulAddr := cfg.GetString(ConfigKeyConsulAddr)
177
	consulMeta := cfg.GetString(ConfigKeyConsulMeta)
178

179
	if exporterPort == int64(0) {
180
		exporterPort = cfg.GetInt64(ConfigKeyExporterPort)
181
	}
182

183
	if exporterPort == 0 {
184
		log.LogInfo("config export port is 0, use default 17510")
185
		exporterPort = 17510
186
	}
187

188
	if exporterPort != int64(0) && len(consulAddr) > 0 {
189
		if ok := strings.HasPrefix(consulAddr, "http"); !ok {
190
			consulAddr = "http://" + consulAddr
191
		}
192
		go DoConsulRegisterProc(consulAddr, AppName, role, cluster, consulMeta, host, exporterPort)
193
	}
194
}
195

196
func autoPush(pushAddr, role, cluster, ip, mountPoint string) {
197
	pid := os.Getpid()
198

199
	client := &http.Client{
200
		Timeout: time.Second * 10,
201
	}
202

203
	hostname, err := os.Hostname()
204
	if err != nil {
205
		log.LogWarnf("get host name failed %v", err)
206
	}
207

208
	pusher := push.New(pushAddr, "cbfs").
209
		Client(client).
210
		Gatherer(registry).
211
		Grouping("cip", ip).
212
		Grouping("role", role).
213
		Grouping("cluster", cluster).
214
		Grouping("pid", strconv.Itoa(pid)).
215
		Grouping("commit", proto.CommitID).
216
		Grouping("app", AppName).
217
		Grouping("mountPoint", mountPoint).
218
		Grouping("hostName", hostname)
219

220
	log.LogInfof("start push data, ip %s, addr %s, role %s, cluster %s, mountPoint %s, hostName %s",
221
		ip, pushAddr, role, cluster, mountPoint, hostname)
222

223
	ticker := time.NewTicker(time.Second * 15)
224
	go func() {
225
		for range ticker.C {
226
			if err := pusher.Push(); err != nil {
227
				log.LogWarnf("push monitor data to %s err, %s", pushAddr, err.Error())
228
			}
229
		}
230
	}()
231
}
232

233
func collect() {
234
	if !enabledPrometheus {
235
		return
236
	}
237
	go collectCounter()
238
	go collectGauge()
239
	go collectHistogram()
240
	go collectAlarm()
241
}
242