ams 扩展字段bugfix (#429)

* 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

@@ -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 {

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
 		}
 

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