修复节点配置

pkg/ruleEngine/nodes/influxdb_node.go

@@ -13,13 +13,13 @@ import (
 )
 
 type InfluxDBConfig struct {
-	URL           string            `json:"url"`         // 数据库地址
-	Token         string            `json:"token"`       // 连接Token
-	Org           string            `json:"org"`         // 组织名称
-	Bucket        string            `json:"bucket"`      // Bucket
-	Measurement   string            `json:"measurement"` // 度量名称
-	Tags          map[string]string `json:"tags"`        // 索引Tag
-	DataFieldName string            `json:"data_fields"` // 数据字段名称
+	URL           string      `json:"url"`         // 数据库地址
+	Token         string      `json:"token"`       // 连接Token
+	Org           string      `json:"org"`         // 组织名称
+	Bucket        string      `json:"bucket"`      // Bucket
+	Measurement   string      `json:"measurement"` // 度量名称
+	Tags          []*KeyValue `json:"tags"`        // 索引Tag
+	DataFieldName string      `json:"data_fields"` // 数据字段名称
 }
 
 // InfluxDBNode influxDB写入节点
@@ -40,7 +40,7 @@ func (i *InfluxDBNode) Init(ctx ruleEngine.Context, config string) error {
 		i.config = c
 	}
 	if i.config.Tags == nil {
-		i.config.Tags = make(map[string]string)
+		i.config.Tags = make([]*KeyValue, 0)
 	}
 	i.pool = grpool.New(10)
 	client := influxdb2.NewClient(i.config.URL, i.config.Token)
@@ -49,8 +49,12 @@ func (i *InfluxDBNode) Init(ctx ruleEngine.Context, config string) error {
 }
 
 func (i *InfluxDBNode) OnMessage(ctx ruleEngine.Context, message *protocol.Message) error {
+	var tags map[string]string
 	for k, v := range message.MetaData {
-		i.config.Tags[k] = v.(string)
+		tags[k] = v.(string)
+	}
+	for _, v := range i.config.Tags {
+		tags[v.Key] = v.Value
 	}
 	jsonData, err := gjson.DecodeToJson(message.Data)
 	if err != nil {
@@ -59,7 +63,7 @@ func (i *InfluxDBNode) OnMessage(ctx ruleEngine.Context, message *protocol.Messa
 	}
 	fields := jsonData.GetMap(i.config.DataFieldName)
 	if fields != nil {
-		point := influxdb2.NewPoint(i.config.Measurement, i.config.Tags, fields, time.Now())
+		point := influxdb2.NewPoint(i.config.Measurement, tags, fields, time.Now())
 		err = i.pool.Add(func() {
 			i.writeApi.WritePoint(point)
 		})

pkg/ruleEngine/nodes/rest_api_request_node.go

@@ -36,7 +36,7 @@ func (r *RestApiRequestNode) Init(ctx ruleEngine.Context, config string) error {
 	if config == "" {
 		r.config = &RestApiRequestNodeConfig{
 			Url:       "http://localhost:8899/test",
-			Headers:   make(map[string]string),
+			Headers:   make([]*KeyValue, 0),
 			Retry:     1,
 			Method:    "POST",
 			TimeOut:   5,
@@ -63,8 +63,8 @@ func (r *RestApiRequestNode) OnMessage(ctx ruleEngine.Context, message *protocol
 	for k, v := range message.MetaData {
 		headers[k] = v.(string)
 	}
-	for k, v := range r.config.Headers {
-		headers[k] = v
+	for _, v := range r.config.Headers {
+		headers[v.Key] = v.Value
 	}
 	req, err := utils.NewRequest(r.config.Method, r.config.Url, []byte(body))
 	if err != nil {
@@ -119,10 +119,15 @@ func (r *RestApiRequestNode) processResponse(ctx ruleEngine.Context, msg *protoc
 }
 
 type RestApiRequestNodeConfig struct {
-	Url       string            `json:"url"`        // 请求的地址
-	Headers   map[string]string `json:"headers"`    // 请求头
-	Retry     int               `json:"retry"`      // 重试次数
-	Method    string            `json:"method"`     // 请求方法
-	TimeOut   int               `json:"time_out"`   // 超时时间(秒)
-	RetryWait int               `json:"retry_wait"` // 重试等待时间
+	Url       string      `json:"url"`        // 请求的地址
+	Headers   []*KeyValue `json:"headers"`    // 请求头
+	Retry     int         `json:"retry"`      // 重试次数
+	Method    string      `json:"method"`     // 请求方法
+	TimeOut   int         `json:"time_out"`   // 超时时间(秒)
+	RetryWait int         `json:"retry_wait"` // 重试等待时间
+}
+
+type KeyValue struct {
+	Key   string `json:"key"`
+	Value string `json:"value"`
 }