Merge pull request #25 from lidaboo/lidabao

add the optional config username and password for kafka
kevwan · Oct 22, 2022 · 75072e0 · 75072e0
2 parents f418100 + c6b51f5
commit 75072e0
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 3 deletions.
diff --git a/stash/config/config.go b/stash/config/config.go
@@ -21,8 +21,8 @@ type (
 		TimeZone      string `json:",optional"`
 		MaxChunkBytes int    `json:",default=15728640"` // default 15M
 		Compress      bool   `json:",default=false"`
-		Username	  string `json:",optional"`
-		Password	  string `json:",optional"`
+		Username      string `json:",optional"`
+		Password      string `json:",optional"`
 	}
 
 	Filter struct {
@@ -44,6 +44,8 @@ type (
 		Processors int    `json:",default=8"`
 		MinBytes   int    `json:",default=10240"`    // 10K
 		MaxBytes   int    `json:",default=10485760"` // 10M
+		Username   string `json:",optional"`
+		Password   string `json:",optional"`
 	}
 
 	Cluster struct {

diff --git a/stash/stash.go b/stash/stash.go
@@ -33,6 +33,8 @@ func toKqConf(c config.KafkaConf) []kq.KqConf {
 			Processors:  c.Processors,
 			MinBytes:    c.MinBytes,
 			MaxBytes:    c.MaxBytes,
+			Username:    c.Username,
+			Password:    c.Password,
 		})
 	}
 
@@ -53,7 +55,7 @@ func main() {
 		client, err := elastic.NewClient(
 			elastic.SetSniff(false),
 			elastic.SetURL(processor.Output.ElasticSearch.Hosts...),
-			elastic.SetBasicAuth(processor.Output.ElasticSearch.Username,processor.Output.ElasticSearch.Password),
+			elastic.SetBasicAuth(processor.Output.ElasticSearch.Username, processor.Output.ElasticSearch.Password),
 		)
 		logx.Must(err)