update pandora go sdk

Author: 李红

mgr/dataflow.go

@@ -378,21 +378,18 @@ func checkSampleData(sampleData []string, logParser parser.Parser) ([]string, er
 }
 
 func getTransformerCreator(transformerConfig map[string]interface{}) (transforms.Creator, error) {
-	transformKeyType, ok := transformerConfig[KeyType]
+	transformKeyType, ok := transformerConfig[transforms.KeyType]
 	if !ok {
-		err := fmt.Errorf("missing param %s", KeyType)
-		return nil, err
+		return nil, fmt.Errorf("missing param %s", transforms.KeyType)
 	}
 	transformKeyTypeStr, ok := transformKeyType.(string)
 	if !ok {
-		err := fmt.Errorf("param %s must be of type string", KeyType)
-		return nil, err
+		return nil, fmt.Errorf("param %s must be of type string", transforms.KeyType)
 	}
 
 	create, ok := transforms.Transformers[transformKeyTypeStr]
 	if !ok {
-		err := fmt.Errorf("transformer of type %v not exist", transformKeyTypeStr)
-		return nil, err
+		return nil, fmt.Errorf("transformer of type %v not exist", transformKeyTypeStr)
 	}
 	return create, nil
 }

mgr/runner.go

@@ -28,6 +28,7 @@ import (
 	"github.com/qiniu/logkit/sender"
 	_ "github.com/qiniu/logkit/sender/builtin"
 	"github.com/qiniu/logkit/transforms"
+	"github.com/qiniu/logkit/transforms/ip"
 	. "github.com/qiniu/logkit/utils/models"
 )
 
@@ -286,9 +287,7 @@ func NewLogExportRunner(rc RunnerConfig, cleanChan chan<- cleaner.CleanSignal, r
 			//如果已经开启了，不要重复加
 			senderConfig[sender.KeyPandoraExtraInfo] = "false"
 		}
-		if senderConfig[sender.KeySenderType] == sender.TypePandora {
-			senderConfig = setSenderConfig(senderConfig, serverConfigs)
-		}
+		senderConfig = setSenderConfig(senderConfig, serverConfigs)
 		s, err := sr.NewSender(senderConfig, meta.FtSaveLogPath())
 		if err != nil {
 			return nil, err
@@ -309,7 +308,7 @@ func createTransformers(rc RunnerConfig) ([]transforms.Transformer, error) {
 	transformers := make([]transforms.Transformer, 0)
 	for idx := range rc.Transforms {
 		tConf := rc.Transforms[idx]
-		tp := tConf[KeyType]
+		tp := tConf[transforms.KeyType]
 		if tp == nil {
 			return nil, fmt.Errorf("transformer config type is empty %v", tConf)
 		}
@@ -1292,18 +1291,19 @@ func MergeExtraInfoTags(meta *reader.Meta, tags map[string]interface{}) map[stri
 }
 
 func setSenderConfig(senderConfig conf.MapConf, serverConfigs []map[string]interface{}) conf.MapConf {
+	if senderConfig[sender.KeySenderType] != sender.TypePandora {
+		return senderConfig
+	}
+
 	for _, serverConfig := range serverConfigs {
-		keyType, ok := serverConfig[KeyType].(string)
-		if !ok || keyType != KeyIP {
-			continue
-		}
-		localEnable, ok := serverConfig[LocalEnable].(bool)
-		if !ok {
+		keyType, ok := serverConfig[transforms.KeyType].(string)
+		if !ok || keyType != ip.Name {
 			continue
 		}
 
 		autoCreate := senderConfig[sender.KeyPandoraAutoCreate]
-		if localEnable {
+		localEnable, ok := serverConfig[transforms.LocalEnable].(bool)
+		if ok && localEnable {
 			schema := fmt.Sprintf(",%v ip", KeyIP)
 			if autoCreate == fmt.Sprintf("%v ip", KeyIP) {
 				autoCreate = ""

mgr/runner_test.go

@@ -26,7 +26,9 @@ import (
 	_ "github.com/qiniu/logkit/sender/builtin"
 	"github.com/qiniu/logkit/sender/mock"
 	"github.com/qiniu/logkit/sender/pandora"
+	"github.com/qiniu/logkit/transforms"
 	_ "github.com/qiniu/logkit/transforms/builtin"
+	"github.com/qiniu/logkit/transforms/ip"
 	. "github.com/qiniu/logkit/utils/models"
 )
 
@@ -1896,12 +1898,42 @@ DONE:
 			break DONE
 		default:
 			dft++
+
 		}
 		time.Sleep(50 * time.Millisecond)
 		if dft > 60 {
 			break
 		}
 	}
 	assert.Equal(t, 1, ret)
+}
+
+func Test_setSenderConfig(t *testing.T) {
+	senderConfig := conf.MapConf{
+		sender.KeySenderType: sender.TypePandora,
+	}
+	serverConfigs := []map[string]interface{}{
+		{
+			transforms.KeyType: ip.Name,
+		},
+	}
+	actualConfig := setSenderConfig(senderConfig, serverConfigs)
+	assert.Equal(t, "ip ip", actualConfig[sender.KeyPandoraAutoCreate])
 
+	serverConfigs = []map[string]interface{}{
+		{
+			transforms.KeyType:     ip.Name,
+			transforms.LocalEnable: true,
+		},
+	}
+	actualConfig = setSenderConfig(senderConfig, serverConfigs)
+	assert.Equal(t, "", actualConfig[sender.KeyPandoraAutoCreate])
+
+	serverConfigs = []map[string]interface{}{
+		{
+			transforms.KeyType: "other",
+		},
+	}
+	actualConfig = setSenderConfig(senderConfig, serverConfigs)
+	assert.Equal(t, "", actualConfig[sender.KeyPandoraAutoCreate])
 }

sender/pandora/pandora.go

@@ -48,7 +48,6 @@ type Sender struct {
 	microsecondCounter uint64
 	extraInfo          map[string]string
 	sendType           string
-	EnbleServerIp      bool
 }
 
 // UserSchema was parsed pandora schema from user's raw schema
@@ -492,7 +491,6 @@ func newPandoraSender(opt *PandoraOption) (s *Sender, err error) {
 		log.Errorf("Runner[%v] Sender[%v]: auto create pandora repo error: %v, you can create on pandora portal, ignored...", opt.runnerName, opt.name, err)
 		err = nil
 	}
-	log.Infof("`````````````````````````````dsl: %v, schemas: %v, opt.autoCreate: %v", dsl, schemas, opt.autoCreate)
 	if initErr := s.client.InitOrUpdateWorkflow(&pipeline.InitOrUpdateWorkflowInput{
 		// 此处要的 schema 为 autoCreate 中用户指定的，所以 SchemaFree 要恒为 true
 		InitOptionChange: true,

transforms/ip/ip.go

@@ -232,7 +232,7 @@ func (_ *Transformer) SampleConfig() string {
 func (_ *Transformer) ConfigOptions() []Option {
 	return []Option{
 		{
-			KeyName:       LocalEnable,
+			KeyName:       transforms.LocalEnable,
 			Element:       Radio,
 			ChooseOnly:    true,
 			ChooseOptions: []interface{}{true, false},
@@ -252,7 +252,7 @@ func (_ *Transformer) ConfigOptions() []Option {
 			Description:   "要进行Transform变化的键(key)",
 			ToolTip:       "对该字段的值进行transform变换",
 			Type:          transforms.TransformTypeString,
-			AdvanceDepend: LocalEnable,
+			AdvanceDepend: transforms.LocalEnable,
 		},
 		{
 			KeyName:       "data_path",
@@ -263,7 +263,7 @@ func (_ *Transformer) ConfigOptions() []Option {
 			DefaultNoUse:  true,
 			Description:   "IP数据库路径(data_path)",
 			Type:          transforms.TransformTypeString,
-			AdvanceDepend: LocalEnable,
+			AdvanceDepend: transforms.LocalEnable,
 		},
 		{
 			KeyName:       "key_as_prefix",
@@ -275,7 +275,7 @@ func (_ *Transformer) ConfigOptions() []Option {
 			Element:       Checkbox,
 			Description:   "字段名称作为前缀(key_as_prefix)",
 			Type:          transforms.TransformTypeString,
-			AdvanceDepend: LocalEnable,
+			AdvanceDepend: transforms.LocalEnable,
 		},
 		{
 			KeyName:       "language",
@@ -287,7 +287,7 @@ func (_ *Transformer) ConfigOptions() []Option {
 			Description:   "mmdb格式库使用的语种",
 			Advance:       true,
 			Type:          transforms.TransformTypeString,
-			AdvanceDepend: LocalEnable,
+			AdvanceDepend: transforms.LocalEnable,
 		},
 	}
 }
@@ -314,8 +314,8 @@ func (t *Transformer) Close() error {
 
 func (t *Transformer) ServerConfig() map[string]interface{} {
 	config := make(map[string]interface{})
-	config[KeyType] = Name
-	config[LocalEnable] = t.LocalEnable
+	config[transforms.KeyType] = Name
+	config[transforms.LocalEnable] = t.LocalEnable
 	return config
 }
 

transforms/registry.go

@@ -4,6 +4,12 @@ import (
 	. "github.com/qiniu/logkit/utils/models"
 )
 
+const (
+	KeyType = "type"
+
+	LocalEnable = "local_enable"
+)
+
 const (
 	TransformTypeString  = "string"
 	TransformTypeLong    = "long"

utils/models/models.go

@@ -36,7 +36,6 @@ const (
 	KeyPandoraStash      = "pandora_stash"       // 当只有一条数据且 sendError 时候，将其转化为 raw 发送到 pandora_stash 这个字段
 	KeyPandoraSeparateId = "pandora_separate_id" // 当一条数据大于2M且 sendError 时候，将其切片，切片记录到 pandora_separate_id 这个字段
 	KeyIP                = "ip"                  // schema ip
-	KeyType              = "type"
 
 	SchemaFreeTokensPrefix = "schema_free_tokens_"
 	LogDBTokensPrefix      = "logdb_tokens_"
@@ -60,8 +59,6 @@ const (
 	Checkbox    = "checkbox"
 	Radio       = "radio"
 	InputNumber = "inputNumber"
-
-	LocalEnable = "local_enable"
 )
 
 type Option struct {