临时

go.mod

@@ -5,11 +5,14 @@ go 1.12
 require (
 	github.com/Shopify/sarama v1.23.1
 	github.com/buaazp/fasthttprouter v0.1.1
+	github.com/google/go-cmp v0.3.1 // indirect
+	github.com/gorilla/websocket v1.4.1 // indirect
 	github.com/influxdata/influxdb v1.7.9
 	github.com/mkevac/debugcharts v0.0.0-20180124214838-d3203a8fa926
+	github.com/shirou/gopsutil v2.19.11+incompatible // indirect
 	github.com/shirou/w32 v0.0.0-20160930032740-bb4de0191aa4 // indirect
 	github.com/valyala/fasthttp v1.6.0
-	github.com/vrg0/go-common v0.0.0-20191111075058-891b371eb18a
+	github.com/vrg0/go-common v0.0.0-20191213082238-e4e6080702f1
 	go.uber.org/zap v1.13.0
 	gopkg.in/alexcesaro/quotedprintable.v3 v3.0.0-20150716171945-2caba252f4dc // indirect
 	gopkg.in/gomail.v2 v2.0.0-20160411212932-81ebce5c23df

pkg/node-check/node_check.go

+package node_check
+
+/*
+import (
+	"git.quantgroup.cn/DevOps/enoch/pkg/glog"
+	"github.com/vrg0/go-common/registry"
+	"sync"
+)
+
+const (
+	Passing  = "passing"
+	Critical = "critical"
+)
+
+//节点健康状态检查和告警
+func NodeHealthCheckAndNotify() {
+	//设置观察者节点
+	if err := registry.SetObserver(&watch{}); err != nil {
+		glog.Info("设置观察者节点失败", err)
+		return
+	}
+}
+
+type watch struct {
+	serviceMap     map[string]*registry.Service
+	serviceMapLock *sync.Mutex
+}
+
+func (w watch) DeleteService(serviceName string) {
+	w.serviceMapLock.Lock()
+	defer w.serviceMapLock.Unlock()
+
+	glog.Info("服务下线：", serviceName)
+
+	//删除服务
+	delete(w.serviceMap, serviceName)
+}
+
+func (w watch) UpdateNodes(service *registry.Service) {
+	w.serviceMapLock.Lock()
+	defer w.serviceMapLock.Unlock()
+
+	if s, ok := w.serviceMap[service.Name]; !ok {
+		glog.Info("服务上线:", service.Name)
+	} else {
+		for id, node := range service.NodeMap {
+			if node.Status == Critical {
+			}
+		}
+	}
+
+
+	//更新服务状态
+	w.serviceMap[service.Name] = service
+}
+
+ */
+/*
+type watch struct{}
+func (w watch) DeleteService(serviceName string) {
+	//pass
+	servicesStatusLock.Lock()
+	defer servicesStatusLock.Unlock()
+
+	delete(servicesStatus, serviceName)
+}
+func (w watch) UpdateNodes(service *registry.Service) {
+	servicesStatusLock.Lock()
+	defer servicesStatusLock.Unlock()
+
+	//单个节点挂了告警
+	if oldService, ok := servicesStatus[service.Name]; ok {
+		for _, node := range service.NodeMap {
+			if oldNode, ok := oldService.NodeMap[node.Id]; ok {
+				if oldNode.Status == Passing && node.Status == Critical {
+					logger.Warning.Print(service.Name, " ", node.Id, "---!!!node critical!!!---")
+					if _, ok := IgnoreServiceMap[service.Name]; !ok {
+						_ = dingding.SenderDingDing(service.Name+" "+node.Id+" "+"---!!!node critical!!!---", dingding.DefaultDingURL)
+					}
+				}
+			}
+		}
+	}
+
+	//整个服务挂了告警
+	//如果 服务存在，并且服务的old状态为passing，并且服务的now状态为critical，则报警，否贼记录服务状态
+	serviceString := serviceStr(service)
+	if oldService, ok := servicesStatus[service.Name]; ok && serviceStatus(oldService) && !serviceStatus(service) {
+		logger.Warning.Print(serviceString, "---!!!service critical!!!---")
+		if _, ok := IgnoreServiceMap[service.Name]; !ok {
+			_ = dingding.SenderDingDing(serviceString+"---!!!service critical!!!---", dingding.DefaultDingURL)
+		}
+		handler(service.Name)
+	} else {
+		logger.Info.Print(serviceString)
+	}
+
+	//更新服务状态
+	//深拷贝对象
+	newService := registry.NewService(service.Name)
+	for kk, vv := range service.NodeMap {
+		newNode := registry.Node{
+			ServiceName:vv.ServiceName,
+			Id:vv.Id,
+			Port:vv.Port,
+			Address:vv.Address,
+			Status:vv.Status,
+		}
+		for x, y := range vv.Meta {
+			newNode.Meta[x] = y
+		}
+		newService.NodeMap[kk] = &newNode
+	}
+
+	servicesStatus[service.Name] = newService
+}
+*/
+
+/*
+//节点的健康状态告警
+func NodeCheck() {
+	//服务状态初始化
+	InitServiceStatus()
+
+	//设置观察者节点
+	if e := registry.SetObserver(&watch{}); e != nil {
+		glog.Info("设置观察者节点失败", e)
+		return
+	}
+}
+
+func InitServiceStatus() {
+	if s, ok := registry.GetServiceMap(); ok {
+		servicesStatus = s
+	}
+}
+*/
+
+/*
+
+var HandlerMap = new(sync.Map)
+var HttpGetRetryCount = 3
+var HttpTimeOut = time.Second * 10
+var IgnoreServiceMap = make(map[string]struct{})
+
+func init() {
+	//TODO 灵活配置
+	HandlerMap.Store("aaaa", "http://www.baidasdfasdfasdfasdfu.com/")
+	HandlerMap.Store("heimdallr", "http://172.20.6.33:8989/service-down")
+	//	HandlerMap.Store("aaaa", "http://172.20.6.33:8989/service-down")
+
+	//TODO 灵活配置
+	IgnoreServiceMap["vcc-talos"] = struct{}{}
+	IgnoreServiceMap["aaaa"] = struct{}{}
+}
+
+func httpGet(url string, timeout time.Duration) (*http.Response, error) {
+	ctx, cancel := context.WithCancel(context.TODO())
+
+	req, e := http.NewRequest("GET", url, nil)
+	if e != nil {
+		return nil, e
+	} else {
+		req = req.WithContext(ctx)
+	}
+
+	_ = time.AfterFunc(timeout, func() {
+		cancel()
+	})
+
+	return http.DefaultClient.Do(req)
+}
+
+func handler(serviceName string) {
+	if url, ok := HandlerMap.Load(serviceName); ok {
+		for i := 0; i < HttpGetRetryCount; i++ {
+			if resp, e := httpGet(url.(string), HttpTimeOut); e != nil {
+				logger.Error.Print(" handler service: ", serviceName, " ", e)
+			} else {
+				logger.Info.Print(" handler service: ", serviceName, " ", resp.StatusCode)
+				break
+			}
+		}
+	} else {
+		logger.Info.Print(" handler service: ", serviceName, " ", "not found handler hook api")
+	}
+}
+
+
+
+func serviceStatus(service *registry.Service) bool {
+	for _, node := range service.NodeMap {
+		if node.Status == Passing {
+			return true
+		}
+	}
+
+	return false
+}
+
+func serviceStr(service *registry.Service) string {
+	rtn := service.Name + " "
+
+	for _, node := range service.NodeMap {
+		rtn += node.Id + ":" + node.Status + " "
+	}
+
+	return rtn
+}
+
+
+func (w watch) AddNode(node *registry.Node) {
+	//pass
+}
+
+func (w watch) DelNode(node *registry.Node) {
+	//pass
+}
+
+var (
+	servicesStatus = make(map[string]*registry.Service)
+	servicesStatusLock = new(sync.Mutex)
+)
+
+func InitServiceStatus() {
+	servicesStatus = registry.GetServiceMap()
+}
+
+func NodeCheck() {
+	defer func() {
+		if e := recover(); e != nil {
+			logger.Info.Print("node check panic: ", e)
+			_ = dingding.SenderDingDing("node check panic!", dingding.DefaultDingURL)
+			time.Sleep(time.Second * 1)
+			NodeCheck()
+		}
+	}()
+
+	//注册器初始化
+	dc := "3c"
+	cluster := []string{"172.30.12.2:8500", "172.30.12.3:8500", "172.30.12.4:8500"}
+	if e := registry.Init("consul", map[string]interface{}{"dc": dc, "cluster": cluster}); e != nil {
+		logger.Info.Print("registry init error:", e)
+		os.Exit(-1)
+	}
+	time.Sleep(time.Second * 1)
+
+	//服务状态初始化
+	InitServiceStatus()
+
+	//设置观察者
+	if e := registry.SetObserver("watch", &watch{}); e != nil {
+		logger.Info.Print("set observer error:", e)
+		os.Exit(-1)
+	}
+
+	select {}
+}
+*/

pkg/node-check/node_check_test.go

+package node_check
+
+/*
+import "testing"
+
+func TestNodeCheck(t *testing.T) {
+	go NodeCheck()
+
+	select{}
+}
+ */