升级监控检查，支持dubbo ，多消息组通知

checkdubbo.sh

+#!/bin/sh
+echo $1 $2
+
+for res in `( sleep 1; echo 'live'; sleep 1 ) | telnet $1 $2  | grep true|wc -l` ; do
+  echo "result:${res}"
+done
+
+if test ${res} -eq 1
+then
+  echo "服务监控检查通过！"
+  exit 0
+else
+  echo "服务监控检查未通过！"
+  exit 1
+fi

docker/dockerfile

+FROM docker.io/consul:1.12.9
+MAINTAINER meng.cheng@163.com
+RUN  sed -i 's/dl-cdn.alpinelinux.org/mirrors.ustc.edu.cn/g' /etc/apk/repositories && apk update && apk add curl && apk add busybox-extras
\ No newline at end of file

main.go

@@ -81,7 +81,7 @@ func handlerKafkaMsg() {
 	}
 }
 
-//TODO 可优化成Raft算法，目前是固定节点的
+// TODO 可优化成Raft算法，目前是固定节点的
 func isMaster() bool {
 	//开发环境，但InfluxDb的Ip是生产环境ip的时候，不执行初始化操作
 	if global.IsDev() && strings.Contains(global.InfluxDbAddress, "172.20") {
@@ -132,10 +132,10 @@ func main() {
 	//创建server
 
 	//处理消息（阻塞）
-	handlerKafkaMsg()
+	//handlerKafkaMsg()
 
 	//阻塞main
-	//select {}
+	select {}
 }
 
 func init() {

pkg/global/global.go

@@ -2,13 +2,14 @@ package global
 
 import (
 	"encoding/json"
+	"git.quantgroup.cn/DevOps/enoch/pkg/registry"
 	"github.com/Shopify/sarama"
 	"github.com/valyala/fasthttp"
 	"github.com/vrg0/go-common/args"
 	"github.com/vrg0/go-common/conf"
 	"github.com/vrg0/go-common/kafka"
 	"github.com/vrg0/go-common/logger"
-	"github.com/vrg0/go-common/registry"
+	/*	"github.com/vrg0/go-common/registry"*/
 	"github.com/vrg0/go-common/util"
 	"go.uber.org/zap/zapcore"
 	"os"

pkg/node-check/node_check.go

@@ -5,21 +5,43 @@ import (
 	"fmt"
 	"git.quantgroup.cn/DevOps/enoch/pkg/global"
 	"git.quantgroup.cn/DevOps/enoch/pkg/glog"
+	"git.quantgroup.cn/DevOps/enoch/pkg/registry"
 	"github.com/vrg0/go-common/notify"
-	"github.com/vrg0/go-common/registry"
 	"strings"
 	"sync"
 	"time"
 )
 
 var (
-	notifyDingDing = newNotifyDingDing()
+	//notifyDingDing = newNotifyDingDing()
+	notifyDingDing = make(map[string]*notify.Notify)
 )
 
-func newNotifyDingDing() *notify.Notify {
-	cfgStr := global.Config.GetOrDefault(global.NamespaceApplication, "notify.dingding", `[]`)
+func init() {
+	newDingDings()
+}
+
+func newDingDings() {
+	cfgStr := global.Config.GetOrDefault(global.NamespaceApplication, "notify.dingding", `{}`)
+	glog.Info("dingding config :" + cfgStr)
+	dingdingconfig := make(map[string]string)
+	if err := json.Unmarshal([]byte(cfgStr), &dingdingconfig); err != nil {
+		errStr := fmt.Sprint("Json解析失败: ", string(cfgStr), err)
+		glog.Warn(errStr)
+		return
+	}
+
+	for key, value := range dingdingconfig {
+		notifyDingDing[key] = newNotifyDingDing(value)
+	}
+
+}
+
+func newNotifyDingDing(url string) *notify.Notify {
+	glog.Info("dingding  :" + url)
+	dingdingUrl := "[\"" + url + "\"]"
 	dstList := make([]string, 0)
-	_ = json.Unmarshal([]byte(cfgStr), &dstList)
+	_ = json.Unmarshal([]byte(dingdingUrl), &dstList)
 	return notify.New(dstList)
 }
 
@@ -31,7 +53,7 @@ func NodeHealthCheckAndNotify() {
 	}
 }
 
-//registry观察者，观察服务的状态，当节点挂掉时告警，当服务挂掉时告警
+// registry观察者，观察服务的状态，当节点挂掉时告警，当服务挂掉时告警
 type watch struct {
 	serviceMap     map[string]*registry.Service
 	serviceMapLock *sync.Mutex
@@ -101,12 +123,15 @@ func (w watch) statusCheckAndNotify(service *registry.Service) {
 		if !serviceUp {
 			sb := new(strings.Builder)
 			sb.WriteString(fmt.Sprintf("服务健康状态异常：%s", service.Name))
+			var currentNode *registry.Node
 			for _, node := range service.NodeMap {
 				sb.WriteString(fmt.Sprintf(" %s:%s", node.Id, node.Status))
+				currentNode = node
 			}
+
 			sb.WriteString("\n")
 			glog.Warn(sb.String())
-			notifyDingDing.SendText(sb.String())
+			getDingding(currentNode).SendText(sb.String())
 			return
 		}
 	}
@@ -131,8 +156,26 @@ func DelayedAlarm(node *registry.Node) {
 			if nowNode.Status == registry.Critical {
 				s := fmt.Sprintf("服务节点健康状态异常：%s %s:%s", node.ServiceName, node.Id, node.Status)
 				glog.Warn(s)
-				notifyDingDing.SendText(s)
+				getDingding(node).SendText(s)
 			}
 		}
 	}
 }
+
+func getDingding(node *registry.Node) *notify.Notify {
+	if node == nil {
+		return notifyDingDing["default"]
+	}
+	name := node.Meta["dep"]
+	glog.Info("node :" + node.ServiceName + " meta:" + name)
+	var dingding *notify.Notify
+	if name == "" {
+		dingding = notifyDingDing["default"]
+	} else {
+		dingding = notifyDingDing[name]
+		if dingding == nil {
+			dingding = notifyDingDing["default"]
+		}
+	}
+	return dingding
+}

pkg/node-check/node_check_test.go

@@ -2,8 +2,10 @@ package node_check
 
 import "testing"
 
-func TestNodeCheck(t *testing.T) {
-	NodeHealthCheckAndNotify()
+func TestNewNotifyDingDing(t *testing.T) {
 
-	select {}
+	url := "https://oapi.dingtalk.com/robot/send?access_token=f043707d09b391877e5f3bf1a03a04341cb036b3c3b89152d65f9d8f665a57be"
+
+	dingding := newNotifyDingDing(url)
+	dingding.SendText(" critical test")
 }

pkg/node-up-down/node_up_down.go

@@ -6,8 +6,8 @@ import (
 	"fmt"
 	"git.quantgroup.cn/DevOps/enoch/pkg/api-server"
 	"git.quantgroup.cn/DevOps/enoch/pkg/glog"
+	"git.quantgroup.cn/DevOps/enoch/pkg/registry"
 	"github.com/valyala/fasthttp"
-	"github.com/vrg0/go-common/registry"
 	"strconv"
 	"strings"
 )
@@ -16,7 +16,7 @@ func init() {
 	api_server.HandlerFunc(api_server.POST, "/upstreams/:nginx_upstream_name/targets", manage)
 }
 
-//函数改名
+// 函数改名
 func manage(ctx *fasthttp.RequestCtx) {
 	//参数获取
 	bodyParam := make(map[string]interface{})
@@ -31,7 +31,10 @@ func manage(ctx *fasthttp.RequestCtx) {
 
 	sysName, ok1 := ctx.UserValue("nginx_upstream_name").(string)
 	target, ok2 := bodyParam["target"].(string)
+
 	weight, ok3 := bodyParam["weight"].(float64)
+	dep, _ := bodyParam["dep"].(string)
+	monitorType, _ := bodyParam["type"].(string)
 	if !ok1 || !ok2 || !ok3 {
 		errStr := fmt.Sprint("服务上下线 参数读取失败失败：", string(body))
 		ctx.SetBodyString(errStr)
@@ -52,7 +55,7 @@ func manage(ctx *fasthttp.RequestCtx) {
 	} else {
 		glog.Info("服务下线开始: ", sysName, " ", target, " force=", force)
 	}
-	if err := doUpDown(target, sysName, weight, force); err != nil {
+	if err := doUpDown(target, sysName, weight, force, dep, monitorType); err != nil {
 		ctx.SetBodyString(err.Error())
 		ctx.SetStatusCode(400)
 		glog.Warn(err)
@@ -64,8 +67,8 @@ func manage(ctx *fasthttp.RequestCtx) {
 	}
 }
 
-//节点上下线
-func doUpDown(target string, sysName string, weight float64, force bool) error {
+// 节点上下线
+func doUpDown(target string, sysName string, weight float64, force bool, dep string, monitorType string) error {
 	endpointList := strings.Split(target, ",")
 	for _, endpoint := range endpointList {
 		ipPortStr := strings.Split(endpoint, ":")
@@ -78,7 +81,12 @@ func doUpDown(target string, sysName string, weight float64, force bool) error {
 		if err != nil {
 			return errors.New("failure of target resolution: " + portStr)
 		}
-		node := registry.NewNode(sysName, endpoint, ip, port)
+		var mate map[string]string
+		if dep != "" {
+			mate = make(map[string]string)
+			mate["dep"] = dep
+		}
+		node := registry.NewNodeTimeOut(sysName, endpoint, ip, port, 10, mate)
 
 		if weight > 0 {
 			//获取metadata（非必须，可获取失败）
@@ -91,10 +99,18 @@ func doUpDown(target string, sysName string, weight float64, force bool) error {
 						}*/
 
 			//上线
-			if err := registry.Register(node); err != nil {
-				glog.Info("服务上线失败: ", err.Error())
-				//return errors.New("服务上线失败:" + err.Error())
+			if monitorType == "dubbo" {
+				if err := registry.RegisterDubbo(node); err != nil {
+					glog.Info("服务上线失败: ", err.Error())
+
+				}
+			} else {
+				if err := registry.Register(node); err != nil {
+					glog.Info("服务上线失败: ", err.Error())
+					//return errors.New("服务上线失败:" + err.Error())
+				}
 			}
+
 		} else {
 			// 仅剩一个节点的时候不可以下线，强制下线忽略此步骤
 			if !force {

pkg/node-up-down/node_up_down_dubbo.go

+package node_up_down
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"git.quantgroup.cn/DevOps/enoch/pkg/api-server"
+	"git.quantgroup.cn/DevOps/enoch/pkg/glog"
+	"git.quantgroup.cn/DevOps/enoch/pkg/registry"
+	"github.com/valyala/fasthttp"
+	/*"github.com/vrg0/go-common/registry"*/
+	"strconv"
+	"strings"
+)
+
+func init() {
+	api_server.HandlerFunc(api_server.POST, "/upstreams_dubbo/:nginx_upstream_name/targets", manageDubbo)
+}
+
+// 函数改名
+func manageDubbo(ctx *fasthttp.RequestCtx) {
+	//参数获取
+	bodyParam := make(map[string]interface{})
+	body := ctx.Request.Body()
+	if err := json.Unmarshal(body, &bodyParam); err != nil {
+		errStr := fmt.Sprint("服务上下线 Json解析失败: ", string(body), err)
+		ctx.SetBodyString(errStr)
+		ctx.SetStatusCode(400)
+		glog.Warn(errStr)
+		return
+	}
+
+	sysName, ok1 := ctx.UserValue("nginx_upstream_name").(string)
+	target, ok2 := bodyParam["target"].(string)
+	weight, ok3 := bodyParam["weight"].(float64)
+	dep := bodyParam["dep"].(string)
+	if !ok1 || !ok2 || !ok3 {
+		errStr := fmt.Sprint("服务上下线 参数读取失败失败：", string(body))
+		ctx.SetBodyString(errStr)
+		ctx.SetStatusCode(400)
+		glog.Warn(errStr)
+		return
+	}
+
+	//强制上下线，默认关闭
+	force := false
+	if f, ok := bodyParam["force"].(bool); ok {
+		force = f
+	}
+
+	//上下线
+	if weight > 0 {
+		glog.Info("服务上线开始: ", sysName, " ", target, " force=", force)
+	} else {
+		glog.Info("服务下线开始: ", sysName, " ", target, " force=", force)
+	}
+	if err := doUpDownDubbo(target, sysName, weight, force, dep); err != nil {
+		ctx.SetBodyString(err.Error())
+		ctx.SetStatusCode(400)
+		glog.Warn(err)
+		return
+	} else if weight > 0 {
+		glog.Info("服务上线成功: ", sysName, " ", target, " force=", force)
+	} else {
+		glog.Info("服务下线成功: ", sysName, " ", target, " force=", force)
+	}
+}
+
+// 节点上下线
+func doUpDownDubbo(target string, sysName string, weight float64, force bool, dep string) error {
+	endpointList := strings.Split(target, ",")
+	for _, endpoint := range endpointList {
+		ipPortStr := strings.Split(endpoint, ":")
+		if len(ipPortStr) != 2 {
+			return errors.New("failure of target resolution: " + endpoint)
+		}
+		ip := ipPortStr[0]
+		portStr := ipPortStr[1]
+		port, err := strconv.Atoi(portStr)
+		if err != nil {
+			return errors.New("failure of target resolution: " + portStr)
+		}
+		var mate map[string]string
+		if dep != "" {
+			mate = make(map[string]string)
+			mate["dep"] = dep
+		}
+		node := registry.NewNodeTimeOut(sysName, endpoint, ip, port, 10, mate)
+
+		if weight > 0 {
+			//获取metadata（非必须，可获取失败）
+			/*			v, err := consul_kv.GetValue("node:" + sysName + ":" + endpoint)
+						if err == nil {
+							meta := make(map[string]string)
+							if e := json.Unmarshal([]byte(v), &meta); e == nil {
+								node.Meta = meta
+							}
+						}*/
+
+			//上线
+			if err := registry.RegisterDubbo(node); err != nil {
+				glog.Info("服务上线失败: ", err.Error())
+				//return errors.New("服务上线失败:" + err.Error())
+			}
+		} else {
+			// 仅剩一个节点的时候不可以下线，强制下线忽略此步骤
+			if !force {
+				if service, ok := registry.GetService(node.ServiceName); ok && len(service.NodeMap) == 1 {
+					if _, ok := service.NodeMap[node.Id]; ok {
+						return errors.New("服务下线失败:服务仅剩一个可用节点")
+					}
+				}
+			}
+
+			//下线
+			if err := registry.Deregister(node); err != nil {
+				glog.Info("服务下线失败: ", err.Error())
+				//return errors.New("服务下线失败:" + err.Error())
+			}
+		}
+	}
+
+	return nil
+}

pkg/registry/consul_test.go

+package registry
+
+import (
+	"fmt"
+	"io"
+	"net/http"
+	"testing"
+)
+
+var (
+	testConsulRegister Registry = nil
+)
+
+// 正常初始化
+func TestConsulRegistry_Init(t *testing.T) {
+	if testConsulRegister == nil {
+		dc := "dc1"
+		cluster := []string{"127.0.0.1:8500"}
+		r := newConsulRegistry()
+		if e := r.Init(map[string]interface{}{"dc": dc, "cluster": cluster}); e != nil {
+			t.Error(e)
+		} else {
+			testConsulRegister = r
+		}
+	}
+}
+
+// 默认初始化
+func TestConsulRegistry_Init2(t *testing.T) {
+	r := newConsulRegistry()
+	if e := r.Init(map[string]interface{}{}); e != nil {
+		t.Error(e)
+	}
+}
+
+// 服务注册，正常
+func TestConsulRegistry_Register(t *testing.T) {
+	//开启http-service
+	TestConsulRegistry_Init(t)
+	http.HandleFunc("/tech/health/check", func(w http.ResponseWriter, _ *http.Request) {
+		_, _ = io.WriteString(w, "Pong!\n")
+	})
+	go func() { _ = http.ListenAndServe("192.168.29.88:8989", nil) }()
+
+	//注册服务，预期会阻塞运行，直到 健康检查的状态为 passing。预期返回nil
+	node := NewNode("aaaa", "192.168.29.88:8989", "192.168.29.88", 8989)
+	if e := testConsulRegister.Register(node); e != nil {
+		t.Error(e)
+	}
+}
+
+// 服务注册，异常
+func TestConsulRegistry_Register2(t *testing.T) {
+	TestConsulRegistry_Init(t)
+
+	//健康检查的状态为 passing。 预期所以会阻塞10秒，返回 timeout
+	node := NewNode("aaaa", "192.168.29.88:9898", "192.168.29.88", 8989)
+	if e := testConsulRegister.Register(node); e != nil {
+		if e.Error() != "register timeout" {
+			t.Error(e)
+		}
+	}
+}
+
+// 服务注销
+func TestConsulRegistry_Deregister(t *testing.T) {
+	TestConsulRegistry_Register(t)
+
+	//会阻塞执行，直到服务节点被删除，或者整个服务被删除。预期返回nil
+	node := NewNode("aaaa", "192.168.29.88:8989", "192.168.29.88", 8989)
+	if e := testConsulRegister.Deregister(node); e != nil {
+		t.Error(e)
+	}
+}
+
+// 获取service
+func TestConsulRegistry_GetService(t *testing.T) {
+	//预期：获取到aaaa服务有一个节点，id为192.168.29.88:8989
+	TestConsulRegistry_Register(t)
+	if service, ok := testConsulRegister.GetService("aaaa"); !ok {
+		t.Error("get service fatal")
+	} else {
+		if len(service.NodeMap) != 1 {
+			t.Error("service.NodeMap is not eq 1")
+		}
+		if _, ok := service.NodeMap["192.168.29.88:8989"]; !ok {
+			t.Error("service.NodeMap is not exists")
+		}
+	}
+}
+
+// 获取 service Map
+func TestConsulRegistry_GetServiceMap(t *testing.T) {
+	//预期：aaaa服务存在
+	TestConsulRegistry_Register(t)
+	if serviceMap, ok := testConsulRegister.GetServiceMap(); !ok {
+		t.Error("get service map fatal")
+	} else {
+		if _, ok := serviceMap["aaaa"]; !ok {
+			t.Error("aaaa is not exists")
+		}
+
+		for _, v := range serviceMap {
+			for _, n := range v.NodeMap {
+				fmt.Println(n)
+			}
+			fmt.Println("------------")
+		}
+	}
+}
+
+// 观察者接口的实现
+type testObserver struct{}
+
+func (testObserver) DeleteService(serviceName string) {
+	fmt.Println(serviceName)
+}
+func (testObserver) UpdateNodes(service *Service) {
+	for _, node := range service.NodeMap {
+		fmt.Println(node)
+	}
+}
+
+// 设置观察者
+func TestConsulRegistry_SetObserver(t *testing.T) {
+	//预期：注册节点成功后，打印aaaa相关的信息
+	TestConsulRegistry_Register(t)
+	o := new(testObserver)
+	if e := testConsulRegister.SetObserver(o); e != nil {
+		t.Error(e)
+	}
+
+	node := NewNode("aaaa", "192.168.29.88:8989", "192.168.29.88", 8989)
+	if e := testConsulRegister.Deregister(node); e != nil {
+		t.Error(e)
+	}
+}

pkg/registry/registry.go

+package registry
+
+import (
+	"errors"
+)
+
+// 节点
+type Node struct {
+	ServiceName string            //服务名称
+	Id          string            //节点编号
+	Address     string            //地址
+	Port        int               //端口
+	Meta        map[string]string //元数据
+	Status      string            //状态
+	Timeout     int               //超时时间
+}
+
+// 服务
+type Service struct {
+	Name    string              //服务名称
+	NodeMap map[string]*Node    //节点映射
+	TagMap  map[string]struct{} //标签
+}
+
+// 服务映射
+type ServiceMap map[string]*Service
+
+// 观察者
+type Observer interface {
+	//删除节点事件
+	DeleteService(serviceName string)
+
+	//更新节点事件
+	UpdateNodes(service *Service)
+}
+
+// 注册器
+type Registry interface {
+	//初始化
+	Init(options map[string]interface{}) error
+
+	//服务注册
+	Register(node *Node) error
+
+	//服务注册
+	RegisterDubbo(node *Node) error
+
+	//服务注销
+	Deregister(node *Node) error
+
+	//获取服务映射，成功返回true，失败返回false
+	GetServiceMap() (ServiceMap, bool)
+
+	//获取服务，成功返回true，失败返回false
+	GetService(serviceName string) (*Service, bool)
+
+	//设置观察者
+	SetObserver(observer Observer) error
+}
+
+var (
+	//注册器映射
+	newRegistryMap = make(map[string]func() Registry, 0)
+
+	//默认注册器
+	defaultRegistry Registry = nil
+)
+
+const (
+	Unknown  = "unknown"
+	Passing  = "passing"
+	Critical = "critical"
+)
+
+// 新建注册器
+func New(name string) (Registry, error) {
+	if newRegistry, ok := newRegistryMap[name]; ok {
+		return newRegistry(), nil
+	} else {
+		return nil, errors.New("not found registry")
+	}
+}
+
+// 新建节点
+func NewNode(serviceName string, id string, address string, port int) *Node {
+	return &Node{
+		ServiceName: serviceName,
+		Id:          id,
+		Address:     address,
+		Port:        port,
+		Status:      Unknown,
+		Meta:        make(map[string]string, 0),
+		Timeout:     10,
+	}
+}
+
+// 新建节点
+func NewNodeTimeOut(serviceName string, id string, address string, port int, timeout int, meta map[string]string) *Node {
+	var meta1 map[string]string
+	if meta == nil {
+		meta1 = make(map[string]string, 0)
+	} else {
+		meta1 = meta
+	}
+	return &Node{
+		ServiceName: serviceName,
+		Id:          id,
+		Address:     address,
+		Port:        port,
+		Status:      Unknown,
+		Meta:        meta1,
+		Timeout:     timeout,
+	}
+}
+
+// 新建服务
+func NewService(name string) *Service {
+	return &Service{
+		Name:    name,
+		NodeMap: make(map[string]*Node, 0),
+		TagMap:  make(map[string]struct{}, 0),
+	}
+}
+
+// 深拷贝节点
+func DeepCopyNode(node *Node) *Node {
+	rtn := NewNode(node.ServiceName, node.Id, node.Address, node.Port)
+	rtn.Status = node.Status
+	for k, v := range node.Meta {
+		rtn.Meta[k] = v
+	}
+	return rtn
+}
+
+// 深拷贝服务
+func DeepCopyService(service *Service) *Service {
+	rtn := NewService(service.Name)
+	for _, node := range service.NodeMap {
+		rtn.NodeMap[node.Id] = DeepCopyNode(node)
+	}
+	for k, v := range service.TagMap {
+		rtn.TagMap[k] = v
+	}
+	return rtn
+}
+
+// 深拷贝服务列表
+func DeepCopyServiceMap(serviceMap ServiceMap) ServiceMap {
+	rtn := make(ServiceMap)
+	for name, service := range serviceMap {
+		rtn[name] = DeepCopyService(service)
+	}
+	return rtn
+}
+
+// 初始化
+func Init(name string, options map[string]interface{}) error {
+	//懒汉单例
+	if defaultRegistry != nil {
+		return nil
+	}
+
+	if newRegistry, e := New(name); e != nil {
+		return e
+	} else {
+		defaultRegistry = newRegistry
+	}
+
+	return defaultRegistry.Init(options)
+}
+
+// 服务注册
+func Register(node *Node) error {
+	return defaultRegistry.Register(node)
+}
+
+// 服务注册
+func RegisterDubbo(node *Node) error {
+	return defaultRegistry.RegisterDubbo(node)
+}
+
+// 服务注销
+func Deregister(node *Node) error {
+	return defaultRegistry.Deregister(node)
+}
+
+// 获取服务映射
+func GetServiceMap() (ServiceMap, bool) {
+	return defaultRegistry.GetServiceMap()
+}
+
+// 获取服务
+func GetService(serviceName string) (*Service, bool) {
+	return defaultRegistry.GetService(serviceName)
+}
+
+// 设置观察者
+func SetObserver(observer Observer) error {
+	return defaultRegistry.SetObserver(observer)
+}

pkg/registry/registry_test.go

+package registry
+
+import (
+	"testing"
+)
+
+type testRegistry struct{}
+
+func newTestRegistry() Registry {
+	return &testRegistry{}
+}
+
+func init() {
+	if _, ok := newRegistryMap["test"]; !ok {
+		newRegistryMap["test"] = newTestRegistry
+	}
+}
+
+func (testRegistry) Init(options map[string]interface{}) error {
+	return nil
+}
+
+func (testRegistry) Register(node *Node) error {
+	return nil
+}
+
+func (testRegistry) Deregister(node *Node) error {
+	return nil
+}
+
+func (testRegistry) GetServiceMap() (ServiceMap, bool) {
+	return nil, true
+}
+
+func (testRegistry) GetService(serviceName string) (*Service, bool) {
+	return nil, true
+}
+
+func (testRegistry) SetObserver(observer Observer) error {
+	return nil
+}
+
+func defaultRegistryClear() {
+	defaultRegistry = nil
+}
+
+func TestInit(t *testing.T) {
+	//registry不存在的情况，预期会报错
+	defaultRegistryClear()
+	if e := Init("testXXX", nil); e != nil {
+		if e.Error() != "not found registry" {
+			t.Error(e)
+		}
+	}
+
+	//registry存在的情况，预期不报错
+	defaultRegistryClear()
+	if e := Init("test", nil); e != nil {
+		t.Error(e)
+	}
+}
+
+func TestNew(t *testing.T) {
+	//registry存在的情况，预期不报错
+	if _, e := New("test"); e != nil {
+		t.Error(e)
+	}
+
+	//registry不存在的情况，预期报错
+	if _, e := New("testXXX"); e != nil {
+		if e.Error() != "not found registry" {
+			t.Error(e)
+		}
+	}
+}
+
+func TestDeepCopyNode(t *testing.T) {
+	oldNode := &Node{
+		ServiceName: "ttt",
+		Id:          "123",
+		Address:     "1.1.1.1",
+		Port:        12345,
+		Meta:        make(map[string]string),
+		Status:      Passing,
+	}
+
+	newNode := DeepCopyNode(oldNode)
+
+	if &oldNode.Meta == &newNode.Meta {
+		t.Error("node.Meta copy fatal")
+	}
+}
+
+func TestDeepCopyService(t *testing.T) {
+	oldNode := &Node{
+		ServiceName: "ttt",
+		Id:          "123",
+		Address:     "1.1.1.1",
+		Port:        12345,
+		Meta:        make(map[string]string),
+		Status:      Passing,
+	}
+
+	oldService := &Service{
+		Name:    oldNode.ServiceName,
+		NodeMap: map[string]*Node{oldNode.Id: oldNode},
+		TagMap:  map[string]struct{}{},
+	}
+
+	newService := DeepCopyService(oldService)
+
+	if &oldService.TagMap == &newService.TagMap {
+		t.Error("service.TagMap fatal")
+	}
+
+	if &oldService.NodeMap == &newService.NodeMap {
+		t.Error("service.NodeMap fatal")
+	}
+
+	if oldService.NodeMap[oldNode.Id] == newService.NodeMap[oldNode.Id] {
+		t.Error("service.NodeMap[] fatal")
+	}
+}
+
+func TestDeepCopyServiceMap(t *testing.T) {
+	oldNode := &Node{
+		ServiceName: "ttt",
+		Id:          "123",
+		Address:     "1.1.1.1",
+		Port:        12345,
+		Meta:        make(map[string]string),
+		Status:      Passing,
+	}
+
+	oldService := &Service{
+		Name:    oldNode.ServiceName,
+		NodeMap: map[string]*Node{oldNode.Id: oldNode},
+		TagMap:  map[string]struct{}{},
+	}
+
+	oldServiceMap := ServiceMap{oldService.Name: oldService}
+
+	newServiceMap := DeepCopyServiceMap(oldServiceMap)
+
+	if &oldServiceMap == &newServiceMap {
+		t.Error("ServiceMap fatal")
+	}
+
+	if oldServiceMap[oldService.Name] == newServiceMap[oldService.Name] {
+		t.Error("ServiceMap[] fatal")
+	}
+}
+
+func TestDeregister(t *testing.T) {
+	TestInit(t)
+	if e := Deregister(nil); e != nil {
+		t.Error(e)
+	}
+}
+
+func TestGetService(t *testing.T) {
+	TestInit(t)
+	if _, ok := GetService("ttt"); !ok {
+		t.Error("GetService")
+	}
+}
+
+func TestGetServiceMap(t *testing.T) {
+	TestInit(t)
+	if _, ok := GetServiceMap(); !ok {
+		t.Error("GetServiceMap")
+	}
+}
+
+func TestSetObserver(t *testing.T) {
+	TestInit(t)
+	if e := SetObserver(nil); e != nil {
+		t.Log(e)
+	}
+}
+
+func TestRegister(t *testing.T) {
+	TestInit(t)
+	if e := Register(nil); e != nil {
+		t.Error(e)
+	}
+}
+
+func TestNewNode(t *testing.T) {
+	node := NewNode("ttt", "1", "1.1.1.1", 1111)
+	if node == nil {
+		t.Error("node")
+	}
+	if node.ServiceName != "ttt" {
+		t.Error("node.ServiceName")
+	}
+	if node.Id != "1" {
+		t.Error("node.Id")
+	}
+	if node.Address != "1.1.1.1" {
+		t.Error("node.Address")
+	}
+	if node.Port != 1111 {
+		t.Error("node.Port")
+	}
+	if node.Status != Unknown {
+		t.Error("node.Status")
+	}
+	if node.Meta == nil || len(node.Meta) != 0 {
+		t.Error("node.Meta")
+	}
+}
+
+func TestNewService(t *testing.T) {
+	service := NewService("ttt")
+	if service == nil {
+		t.Error("service")
+	}
+	if service.Name != "ttt" {
+		t.Error("service.ServiceName")
+	}
+	if service.NodeMap == nil || len(service.NodeMap) != 0 {
+		t.Error("service.NodeMap")
+	}
+	if service.TagMap == nil || len(service.TagMap) != 0 {
+		t.Error("service.TagMap")
+	}
+}

script/checkdubbo.sh

+#!/bin/sh
+echo $1 $2
+
+for res in `( sleep 1; echo 'ready'; sleep 1 ) | telnet $1 $2  | grep true|wc -l` ; do
+  echo "result:${res}"
+done
+
+if test ${res} -eq 1
+then
+  echo "服务监控检查通过！"
+  exit 0
+else
+  echo "服务监控检查未通过！"
+  exit 1
+fi