From 7a185b5054ced1f470371eadcfa533a757c9768d Mon Sep 17 00:00:00 2001
From: alick-liming <whandlm@163.com>
Date: Tue, 1 Dec 2020 19:43:23 +0800
Subject: [PATCH] =?UTF-8?q?ams=20=E6=89=A9=E5=B1=95=E5=AD=97=E6=AE=B5bugfi?=
 =?UTF-8?q?x=20(#429)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

* rdb资源增加volume

* rdb用户增加创建时间

* rdb用户添加时间

* rdb新增添加用户时间代码调整

* test

* 1.agent上报扩展字段 2.rdb标签批量修改

* 代码调整

* 代码调整

* ams扩展代码调整

* test

* test

* 测试

* 错误调整

Co-authored-by: alickliming <alickliming@didi.global.com>
---
 src/models/host.go                        | 12 +++---
 src/modules/ams/http/router_host.go       | 52 +++++++++++++++--------
 src/modules/rdb/rabbitmq/queue_consume.go |  9 +++-
 3 files changed, 49 insertions(+), 24 deletions(-)

diff --git a/src/models/host.go b/src/models/host.go
index 7fd346b3..39ceabb8 100644
--- a/src/models/host.go
+++ b/src/models/host.go
@@ -29,7 +29,7 @@ func (h *Host) Save() error {
 	return err
 }
 
-func HostNew(sn, ip, ident, name, cate string, fields map[string]interface{}) error {
+func HostNew(sn, ip, ident, name, cate string, fields map[string]interface{}) (*Host, error) {
 	host := new(Host)
 	host.SN = sn
 	host.IP = ip
@@ -42,22 +42,24 @@ func HostNew(sn, ip, ident, name, cate string, fields map[string]interface{}) er
 	defer session.Close()
 
 	if err := session.Begin(); err != nil {
-		return err
+		return nil, err
 	}
 
 	if _, err := session.Insert(host); err != nil {
 		session.Rollback()
-		return err
+		return nil, err
 	}
 
 	if len(fields) > 0 {
 		if _, err := session.Table(new(Host)).ID(host.Id).Update(fields); err != nil {
 			session.Rollback()
-			return err
+			return nil, err
 		}
 	}
 
-	return session.Commit()
+	err := session.Commit()
+
+	return host, err
 }
 
 func (h *Host) Update(fields map[string]interface{}) error {
diff --git a/src/modules/ams/http/router_host.go b/src/modules/ams/http/router_host.go
index fc9971b1..603b0e99 100644
--- a/src/modules/ams/http/router_host.go
+++ b/src/modules/ams/http/router_host.go
@@ -308,14 +308,41 @@ func v1HostRegister(c *gin.Context) {
 	}
 
 	if host == nil {
-		err = models.HostNew(f.SN, f.IP, f.Ident, f.Name, f.Cate, f.Fields)
+		var err error
+		host, err = models.HostNew(f.SN, f.IP, f.Ident, f.Name, f.Cate, f.Fields)
 		if err == nil {
 			cache.Set(cacheKey, f.Digest, cache.DEFAULT)
 		} else {
 			logger.Warning(err)
 		}
-		renderMessage(c, err)
-		return
+
+		if host == nil {
+			msg := "create host failed"
+			logger.Warningf("%s info:%v", msg, f)
+			renderMessage(c, msg)
+			return
+		}
+	}
+
+	if v, ok := oldFields["tenant"]; ok {
+		vStr := v.(string)
+		if vStr != "" {
+			err = models.HostUpdateTenant([]int64{host.Id}, vStr)
+			if err != nil {
+				logger.Error(err)
+				msg := "update host tenant err"
+				renderMessage(c, msg)
+				return
+			}
+
+			err = models.ResourceRegister([]models.Host{*host}, vStr)
+			if err != nil {
+				logger.Error(err)
+				msg := "register resource err"
+				renderMessage(c, msg)
+				return
+			}
+		}
 	}
 
 	if host.Tenant != "" {
@@ -346,26 +373,15 @@ func v1HostRegister(c *gin.Context) {
 	if err == nil {
 		cache.Set(cacheKey, f.Digest, cache.DEFAULT)
 	} else {
-		logger.Warning(err)
+		logger.Error(err)
+		msg := "update host err"
+		renderMessage(c, msg)
+		return
 	}
 
 	var objs []models.HostFieldValue
 	for k, v := range oldFields {
 		if k == "tenant" {
-			vStr := v.(string)
-			if vStr != "" {
-				err = models.HostUpdateTenant([]int64{host.Id}, vStr)
-				if err != nil {
-					logger.Warning(err)
-					continue
-				}
-
-				err = models.ResourceRegister([]models.Host{*host}, vStr)
-				if err != nil {
-					logger.Warning(err)
-					continue
-				}
-			}
 			continue
 		}
 
diff --git a/src/modules/rdb/rabbitmq/queue_consume.go b/src/modules/rdb/rabbitmq/queue_consume.go
index 48f50d5d..1a1dadc6 100644
--- a/src/modules/rdb/rabbitmq/queue_consume.go
+++ b/src/modules/rdb/rabbitmq/queue_consume.go
@@ -85,6 +85,8 @@ func consume(queueName string) bool {
 	}
 
 	size := 0
+
+	isSleep := false
 	for d := range msgs {
 		size++
 		logger.Infof("rabbitmq consume message: %s", d.Body)
@@ -93,10 +95,15 @@ func consume(queueName string) bool {
 			d.Ack(true)
 		} else {
 			// 底层代码认为不应该ack，说明处理的过程出现问题，可能是DB有问题之类的，sleep一下
-			return true
+			isSleep = true
+			continue
 		}
 	}
 
+	if isSleep {
+		return true
+	}
+
 	if size == 0 {
 		// MQ里没有消息，就sleep一下，否则上层代码一直在死循环空转，浪费算力
 		return true
-- 
GitLab