add kafka username password auth

sundy-li · Mar 15, 2018 · ef044cc · ef044cc
1 parent e46bc72
commit ef044cc
Show file tree

Hide file tree

Showing 3 changed files with 31 additions and 0 deletions.
diff --git a/config/config.go b/config/config.go
@@ -12,6 +12,8 @@ type Config struct {
 		GroupBlacklist string `json:"groupBlacklist"`
 		Logconfig      string `json:"logconfig"`
 		Pidfile        string `json:"pidfile"`
+
+		TopicFilter string `json:"topicFilter"`
 	} `json:"general"`
 
 	Influxdb struct {
@@ -28,6 +30,11 @@ type Config struct {
 		OffsetTopic   string `json:"offsetTopic"`
 		ClientProfile string `json:"ClientProfile"`
 		OffsetsTopic  string `gcfg:"offsetsTopic"`
+
+		Sasl struct {
+			Username string
+			Password string
+		}
 	} `json:"kafka"`
 
 	Zookeeper struct {

diff --git a/config/server.json.sample b/config/server.json.sample
@@ -5,6 +5,8 @@
     "clientId": "burrowx-lagchecker",
     "groupBlacklist": "^(console-consumer-|python-kafka-consumer-).*$",
 
+    "topicFilter" : "*",
+
     "clientProfile" : {
         "some_key" : {
           "clientId" : "xxxx",

diff --git a/monitor/client.go b/monitor/client.go
@@ -8,6 +8,7 @@ import (
 	"errors"
 	"fmt"
 	"io/ioutil"
+	"regexp"
 	"strings"
 	"sync"
 	"time"
@@ -38,6 +39,8 @@ type KafkaClient struct {
 	topicOffset        map[string]*TopicFullOffset //topic => offset
 
 	importer *Importer
+
+	topicFilterRegexps []*regexp.Regexp
 }
 
 type BrokerTopicRequest struct {
@@ -72,6 +75,11 @@ func NewKafkaClient(cfg *config.Config, cluster string) (*KafkaClient, error) {
 	}
 	clientConfig.Net.TLS.Config.InsecureSkipVerify = profile.TLSNoVerify
 
+	if cfg.Kafka[cluster].Sasl.Username != "" {
+		clientConfig.Net.SASL.Enable = true
+		clientConfig.Net.SASL.User = cfg.Kafka[cluster].Sasl.Username
+		clientConfig.Net.SASL.Password = cfg.Kafka[cluster].Sasl.Password
+	}
 	sclient, err := sarama.NewClient(strings.Split(cfg.Kafka[cluster].Brokers, ","), clientConfig)
 	if err != nil {
 		return nil, err
@@ -111,6 +119,14 @@ func NewKafkaClient(cfg *config.Config, cluster string) (*KafkaClient, error) {
 		importer: importer,
 	}
 
+	if cfg.General.TopicFilter != "" {
+		patterns := strings.Split(cfg.General.TopicFilter, ",")
+		client.topicFilterRegexps = make([]*regexp.Regexp, 0, 10)
+		for _, p := range patterns {
+			client.topicFilterRegexps = append(client.topicFilterRegexps, regexp.MustCompile(p))
+		}
+	}
+
 	return client, nil
 }
 
@@ -348,7 +364,13 @@ func (client *KafkaClient) CombineTopicAndConsumer() {
 func (client *KafkaClient) RefreshTopicMap() {
 	client.topicMapLock.Lock()
 	topics, _ := client.client.Topics()
+	//filter topic by topicFilter
 	for _, topic := range topics {
+		for _, reg := range client.topicFilterRegexps {
+			if !reg.MatchString(topic) {
+				continue
+			}
+		}
 		partitions, _ := client.client.Partitions(topic)
 		client.topicMap[topic] = len(partitions)
 	}