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

src/models/resource.go

@@ -292,6 +292,17 @@ func ResourceRegister(hosts []Host, tenant string) error {
 			return err
 		}
 
+		// ident agent修改ident带来重复问题
+		if res == nil {
+			ident := hosts[i].Ident
+			if ident != "" {
+				res, err = ResourceGet("ident=?", ident)
+				if err != nil {
+					return err
+				}
+			}
+		}
+
 		if res == nil {
 			res = &Resource{
 				UUID:   uuid,
@@ -326,6 +337,7 @@ func ResourceRegister(hosts []Host, tenant string) error {
 					return err
 				}
 			}
+			res.UUID = uuid
 			res.Ident = hosts[i].Ident
 			res.Name = hosts[i].Name
 			res.Cate = hosts[i].Cate
@@ -343,7 +355,7 @@ func ResourceRegister(hosts []Host, tenant string) error {
 			}
 
 			res.Extend = string(js)
-			err = res.Update("ident", "name", "cate", "extend", "tenant")
+			err = res.Update("uuid", "ident", "name", "cate", "extend", "tenant")
 			if err != nil {
 				return err
 			}

src/modules/ams/http/router.go

@@ -14,6 +14,7 @@ func Config(r *gin.Engine) {
 		userLogin.POST("/hosts", hostPost)
 		userLogin.GET("/host/:id", hostGet)
 		userLogin.PUT("/hosts/tenant", hostTenantPut)
+		userLogin.PUT("/hosts/node", hostNodePut)
 		userLogin.PUT("/hosts/back", hostBackPut)
 		userLogin.PUT("/hosts/note", hostNotePut)
 		userLogin.PUT("/hosts/cate", hostCatePut)

src/modules/ams/http/router_host.go

@@ -146,6 +146,78 @@ func hostTenantPut(c *gin.Context) {
 	renderMessage(c, err)
 }
 
+type hostNodeForm struct {
+	Ids    []int64 `json:"ids"`
+	NodeId int64   `json:"nodeid"`
+}
+
+func (f *hostNodeForm) Validate() {
+	if len(f.Ids) == 0 {
+		bomb("ids is empty")
+	}
+
+	if f.NodeId == 0 {
+		bomb("nodeid is blank")
+	}
+
+	if f.NodeId < 0 {
+		bomb("nodeid is illegal")
+	}
+}
+
+// 管理员修改主机设备的节点，相当于挂载设备到节点
+func hostNodePut(c *gin.Context) {
+	var f hostNodeForm
+	bind(c, &f)
+	f.Validate()
+
+	loginUser(c).CheckPermGlobal("ams_host_modify")
+	node, err := models.NodeGet("id=?", f.NodeId)
+	dangerous(err)
+	if node == nil {
+		bomb("node is nil")
+	}
+
+	if node.Leaf != 1 {
+		bomb("node is not leaf")
+	}
+
+	hosts, err := models.HostByIds(f.Ids)
+	dangerous(err)
+	if len(hosts) == 0 {
+		bomb("hosts is empty")
+	}
+
+	for _, h := range hosts {
+		if h.Tenant != "" {
+			bomb("%s already belongs to %s", h.Name, h.Tenant)
+		}
+	}
+
+	// 绑定租户
+	tenant := node.Tenant()
+	err = models.HostUpdateTenant(f.Ids, tenant)
+	dangerous(err)
+	dangerous(models.ResourceRegister(hosts, tenant))
+
+	// 绑定到节点
+	var resUuids []string
+	for _, id := range f.Ids {
+		idStr := fmt.Sprintf("host-%d", id)
+		resUuids = append(resUuids, idStr)
+	}
+	if len(resUuids) == 0 {
+		bomb("res is empty")
+	}
+	resIds, err := models.ResourceIdsByUUIDs(resUuids)
+	dangerous(err)
+	if len(resIds) == 0 {
+		bomb("res ids is empty")
+	}
+
+	renderMessage(c, node.Bind(resIds))
+}
+
 type hostNoteForm struct {
 	Ids  []int64 `json:"ids"`
 	Note string  `json:"note"`

src/modules/judge/judge.go

@@ -117,7 +117,7 @@ func pconf() {
 
 func start() {
 	runner.Init()
-	fmt.Println("transfer start, use configuration file:", *conf)
+	fmt.Println("judge start, use configuration file:", *conf)
 	fmt.Println("runner.Cwd:", runner.Cwd)
 	fmt.Println("runner.Hostname:", runner.Hostname)
 }

src/modules/monapi/config/yaml.go

@@ -29,6 +29,12 @@ type ConfYaml struct {
 	Link          linkSection         `yaml:"link"`
 	IndexMod      string              `yaml:"indexMod"`
 	I18n          i18n.I18nSection    `yaml:"i18n"`
+	Tpl           tplSection          `yaml:"tpl"`
+}
+
+type tplSection struct {
+	AlertPath  string `yaml:"alertPath"`
+	ScreenPath string `yaml:"screenPath"`
 }
 
 type mergeSection struct {
@@ -175,6 +181,11 @@ func Parse(ymlfile string) error {
 		"converge": true, // 历史告警的数据库表，对于已收敛的告警，默认删掉，不保留，省得告警太多
 	})
 
+	viper.SetDefault("tpl", map[string]string{
+		"alertPath":  "./etc/alert",
+		"screenPath": "./etc/screen",
+	})
+
 	err = viper.Unmarshal(&yaml)
 	if err != nil {
 		return fmt.Errorf("Unmarshal %v", err)

src/modules/monapi/http/router.go

@@ -144,6 +144,12 @@ func Config(r *gin.Engine) {
 		aggr.GET("/:id", aggrCalcGet)
 	}
 
+	tpl := r.Group("/api/mon/tpl")
+	{
+		tpl.GET("", tplNameGets)
+		tpl.GET("/content", tplGet)
+	}
+
 	aggrs := r.Group("/api/mon/aggrs").Use()
 	{
 		aggrs.GET("", aggrCalcsWithEndpointGet)

src/modules/monapi/http/router_tpl.go

+package http
+
+import (
+	"github.com/didi/nightingale/src/modules/monapi/config"
+
+	"github.com/gin-gonic/gin"
+	"github.com/toolkits/pkg/file"
+)
+
+func tplNameGets(c *gin.Context) {
+	tplType := mustQueryStr(c, "tplType")
+
+	var files []string
+	var err error
+	switch tplType {
+	case "alert":
+		files, err = file.FilesUnder(config.Get().Tpl.AlertPath)
+		dangerous(err)
+	case "screen":
+		files, err = file.FilesUnder(config.Get().Tpl.ScreenPath)
+		dangerous(err)
+	default:
+		bomb("tpl type not found")
+	}
+
+	renderData(c, files, err)
+}
+
+func tplGet(c *gin.Context) {
+	tplName := mustQueryStr(c, "tplName")
+	tplType := mustQueryStr(c, "tplType")
+
+	var filePath string
+	switch tplType {
+	case "alert":
+		filePath = config.Get().Tpl.AlertPath + "/" + tplName
+	case "screen":
+		filePath = config.Get().Tpl.ScreenPath + "/" + tplName
+	default:
+		bomb("tpl type not found")
+	}
+
+	if !file.IsExist(filePath) {
+		bomb("tpl not found")
+	}
+
+	content, err := file.ToString(filePath)
+	renderData(c, content, err)
+}

src/modules/transfer/backend/m3db/m3db.go

@@ -3,6 +3,7 @@ package m3db
 import (
 	"fmt"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/didi/nightingale/src/common/dataobj"
@@ -89,23 +90,31 @@ func (p *Client) Push2Queue(items []*dataobj.MetricValue) {
 		logger.Errorf("unable to get m3db session: %s", err)
 		return
 	}
-
-	errCnt := 0
+	var errCnt int32
+	var (
+		wg sync.WaitGroup
+	)
 	for _, item := range items {
-		if err := session.WriteTagged(
-			p.namespaceID,
-			mvID(item),
-			ident.NewTagsIterator(mvTags(item)),
-			time.Unix(item.Timestamp, 0),
-			item.Value,
-			xtime.Second,
-			nil,
-		); err != nil {
-			logger.Errorf("unable to writeTagged: %s", err)
-			errCnt++
-		}
+		wg.Add(1)
+		go func(dm *dataobj.MetricValue) {
+			err := session.WriteTagged(
+				p.namespaceID,
+				mvID(dm),
+				ident.NewTagsIterator(mvTags(dm)),
+				time.Unix(dm.Timestamp, 0),
+				dm.Value,
+				xtime.Second,
+				nil)
+			if err != nil {
+				logger.Errorf("unable to writeTagged: %s", err)
+				atomic.AddInt32(&errCnt, 1)
+			}
+			wg.Done()
+		}(item)
+
 	}
-	stats.Counter.Set("m3db.queue.err", errCnt)
+	wg.Wait()
+	stats.Counter.Set("m3db.queue.err", int(errCnt))
 }
 
 // QueryData: || (|| endpoints...) (&& tags...)