Merge branch 'master' of github.com:didi/nightingale

Dockerfile

@@ -1,22 +1,11 @@
-FROM golang AS builder
-# RUN apk add --no-cache git gcc
-WORKDIR /app
-
-COPY . .
-RUN ./control build docker
+FROM golang:1.13
 
-FROM buildpack-deps:buster-curl
-LABEL maintainer="llitfkitfk@gmail.com"
+LABEL maintainer="llitfkitfk@gmail.com,chenjiandongx@qq.com"
 
 WORKDIR /app
 
-COPY --from=builder /app/docker/scripts /app/scripts
-COPY --from=builder /app/etc /app/etc
-# Change default address (hard code) 
-RUN ./scripts/sed.sh
+RUN apt-get update && apt-get install net-tools -y
 
-COPY --from=builder /app/bin /usr/local/bin
-
-
-# ENTRYPOINT []
-# CMD []
+COPY . .
+RUN ./control build docker
+RUN mv /app/bin/* /usr/local/bin

README.md

@@ -19,13 +19,22 @@ cd nightingale
 ./control build
 ```
 
-## Quick Start (need install docker for [mac](https://docs.docker.com/docker-for-mac/install/)/[win](https://docs.docker.com/docker-for-windows/install/))
+## Quickstart with Docker
 
+We has offered a Docker demo for the users who want to give it a try. Before you get started, make sure you have installed **Docker** & **docker-compose** and there are some details you should know.
+
+* We highly recommend users prepare a new VM environment to use it.
+* All the core components will be installed on your OS according to the `docker-compose.yaml`.
+* Nightingale will use the following ports, `80`, `5800`, `5810`, `5811`, `5820`, `5821`, `5830`, `5831`, `5840`, `5841`, `6379`, `2058`, `3306`.
+
+Okay. Run it! Once the docker finish its jobs, visits http://your-env-ip in your broswer. Default username and password is `root:root`.
 ```bash
-docker-compose up -d
-# open http://localhost in web browser
+$ docker-compose up -d
 ```
 
+![dashboard](https://user-images.githubusercontent.com/19553554/78956965-8b9c6180-7b16-11ea-9747-6ed5e62b068d.png)
+
+
 ## Team
 
 [ulricqin](https://github.com/ulricqin) [710leo](https://github.com/710leo) [jsers](https://github.com/jsers) [hujter](https://github.com/hujter) [n4mine](https://github.com/n4mine) [heli567](https://github.com/heli567)

docker-compose.yml

@@ -1,81 +1,85 @@
 version: "3"
+
 volumes: 
   mysql-data:
+
 services:
   nginx:
     image: nginx:stable-alpine
+    network_mode: host
     ports: 
       - 80:80
     volumes: 
       - ./docker/nginx/nginx.conf:/etc/nginx/nginx.conf
       - ./docker/nginx/conf.d:/etc/nginx/conf.d
       - ./pub:/home/n9e/pub
-  api:
+
+  nightingale:
     build: .
-    image: api
+    image: nightingale
 
   monapi:
-    image: api
+    image: nightingale
+    network_mode: host
     restart: always
     command: n9e-monapi
     ports: 
       - 5800:5800
 
   transfer:
-    image: api
+    image: nightingale
+    network_mode: host
     restart: always
     command: n9e-transfer
     ports: 
       - 5810:5810
       - 5811:5811
 
   tsdb:
-    image: api
+    image: nightingale
+    network_mode: host
     restart: always
     command: n9e-tsdb
     ports: 
       - 5820:5820
       - 5821:5821
 
   index:
-    image: api
+    image: nightingale
+    network_mode: host
     restart: always
     command: n9e-index
     ports: 
       - 5830:5830
       - 5831:5831
 
   judge:
-    image: api
+    image: nightingale
+    network_mode: host
     restart: always
     command: n9e-judge
     ports: 
       - 5840:5840
       - 5841:5841
 
   collector:
-    image: api
+    image: nightingale
+    network_mode: host
     restart: always
     command: n9e-collector
     ports: 
       - 2058:2058
 
-  # web:
-  #   build:
-  #     context: web
-  #   restart: always
-  #   command: npm run dev
-  #   ports: 
-  #     - 8010:8010
-
   redis:
     image: redis
+    network_mode: host
     restart: always
     ports: 
       - 6379:6379
 
   mysql:
     image: mysql:5.7
+    network_mode: host
     restart: always
     environment:
       - MYSQL_ROOT_PASSWORD=1234
@@ -84,5 +88,3 @@ services:
     volumes:
       - ./sql:/docker-entrypoint-initdb.d
       - mysql-data:/var/lib/mysql
-
-

docker/nginx/nginx.conf

@@ -47,17 +47,17 @@ http {
     proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
 
     upstream n9e.monapi {
-        server monapi:5800;
+        server localhost:5800;
         keepalive 10;
     }
 
     upstream n9e.index {
-        server index:5830;
+        server localhost:5830;
         keepalive 10;
     }
 
     upstream n9e.transfer {
-        server transfer:5810;
+        server localhost:5810;
         keepalive 10;
     }
 

src/modules/collector/collector.go

@@ -73,6 +73,7 @@ func main() {
 
 	funcs.BuildMappers()
 	funcs.Collect()
+	funcs.InitRpcClients()
 
 	//插件采集
 	plugins.Detect()

src/modules/collector/sys/funcs/clients.go

@@ -0,0 +1,45 @@
+package funcs
+
+import (
+	"net/rpc"
+	"sync"
+)
+
+type RpcClientContainer struct {
+	M map[string]*rpc.Client
+	sync.RWMutex
+}
+
+var rpcClients *RpcClientContainer
+
+func InitRpcClients() {
+	rpcClients = &RpcClientContainer{
+		M: make(map[string]*rpc.Client),
+	}
+}
+
+func (rcc *RpcClientContainer) Get(addr string) *rpc.Client {
+	rcc.RLock()
+	defer rcc.RUnlock()
+
+	client, has := rcc.M[addr]
+	if !has {
+		return nil
+	}
+
+	return client
+}
+
+// Put 返回的bool表示affected，确实把自己塞进去了
+func (rcc *RpcClientContainer) Put(addr string, client *rpc.Client) bool {
+	rcc.Lock()
+	defer rcc.Unlock()
+
+	oc, has := rcc.M[addr]
+	if has && oc != nil {
+		return false
+	}
+
+	rcc.M[addr] = client
+	return true
+}

src/modules/collector/sys/funcs/push.go

@@ -44,45 +44,25 @@ func Push(metricItems []*dataobj.MetricValue) error {
 		items = append(items, item)
 	}
 
-	var mh codec.MsgpackHandle
-	mh.MapType = reflect.TypeOf(map[string]interface{}(nil))
-
 	addrs := address.GetRPCAddresses("transfer")
 	count := len(addrs)
 	retry := 0
 	for {
 		for _, i := range rand.Perm(count) {
 			addr := addrs[i]
-			var conn net.Conn
-			conn, err = net.DialTimeout("tcp", addr, time.Millisecond*3000)
-			if err != nil {
-				logger.Error("dial transfer err:", err)
-				continue
-			}
-
-			var bufconn = struct { // bufconn here is a buffered io.ReadWriteCloser
-				io.Closer
-				*bufio.Reader
-				*bufio.Writer
-			}{conn, bufio.NewReader(conn), bufio.NewWriter(conn)}
-
-			rpcCodec := codec.MsgpackSpecRpc.ClientCodec(bufconn, &mh)
-			client := rpc.NewClientWithCodec(rpcCodec)
-
-			var reply dataobj.TransferResp
-			err = client.Call("Transfer.Push", items, &reply)
-			client.Close()
+			reply, err := rpcCall(addr, items)
 			if err != nil {
 				logger.Error(err)
 				continue
 			} else {
 				if reply.Msg != "ok" {
-					err = fmt.Errorf("some item push err", reply)
+					err = fmt.Errorf("some item push err: %s", reply.Msg)
 					logger.Error(err)
 				}
 				return err
 			}
 		}
+
 		time.Sleep(time.Millisecond * 500)
 
 		retry += 1
@@ -94,6 +74,73 @@ func Push(metricItems []*dataobj.MetricValue) error {
 	return err
 }
 
+func rpcCall(addr string, items []*dataobj.MetricValue) (dataobj.TransferResp, error) {
+	var reply dataobj.TransferResp
+	var err error
+
+	client := rpcClients.Get(addr)
+	if client == nil {
+		client, err = rpcClient(addr)
+		if err != nil {
+			return reply, err
+		} else {
+			affected := rpcClients.Put(addr, client)
+			if !affected {
+				defer func() {
+					// 我尝试把自己这个client塞进map失败，说明已经有一个client塞进去了，那我自己用完了就关闭
+					client.Close()
+				}()
+			}
+		}
+	}
+
+	timeout := time.Duration(8) * time.Second
+	done := make(chan error, 1)
+
+	go func() {
+		err := client.Call("Transfer.Push", items, &reply)
+		done <- err
+	}()
+
+	select {
+	case <-time.After(timeout):
+		logger.Warningf("rpc call timeout, transfer addr: %s", addr)
+		rpcClients.Put(addr, nil)
+		client.Close()
+		return reply, fmt.Errorf("%s rpc call timeout", addr)
+	case err := <-done:
+		if err != nil {
+			rpcClients.Put(addr, nil)
+			client.Close()
+			return reply, fmt.Errorf("%s rpc call done, but fail: %v", addr, err)
+		}
+	}
+
+	return reply, nil
+}
+
+func rpcClient(addr string) (*rpc.Client, error) {
+	conn, err := net.DialTimeout("tcp", addr, time.Second*3)
+	if err != nil {
+		err = fmt.Errorf("dial transfer %s fail: %v", addr, err)
+		logger.Error(err)
+		return nil, err
+	}
+
+	var bufConn = struct {
+		io.Closer
+		*bufio.Reader
+		*bufio.Writer
+	}{conn, bufio.NewReader(conn), bufio.NewWriter(conn)}
+
+	var mh codec.MsgpackHandle
+	mh.MapType = reflect.TypeOf(map[string]interface{}(nil))
+
+	rpcCodec := codec.MsgpackSpecRpc.ClientCodec(bufConn, &mh)
+	client := rpc.NewClientWithCodec(rpcCodec)
+	return client, nil
+}
+
 func CounterToGauge(item *dataobj.MetricValue) error {
 	key := item.PK()