7x31
/
sniffer-agent
mirror of https://github.com/zr-hebo/sniffer-agent.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'buffer-pool' into kafka-compress

This commit is contained in:
hebo 2019-12-13 11:40:07 +08:00
parent c827789a83 03c752a209
commit c95ded78e9
1 changed files with 21 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
exporter/kafka.go
View File

@ -15,6 +15,8 @@ var (
kafkaGroupID string kafkaGroupID string
asyncTopic string asyncTopic string
syncTopic string syncTopic string
compress string
compressType sarama.CompressionCodec
) )
func init() { func init() {
@ -27,8 +29,11 @@ func init() {
&asyncTopic, &asyncTopic,
"kafka-async-topic", "", "kafka async send topic. No default value") "kafka-async-topic", "", "kafka async send topic. No default value")
flag.StringVar( flag.StringVar(
&syncTopic, &syncTopic,
"kafka-sync-topic", "", "kafka sync send topic. No default value") "kafka-sync-topic", "", "kafka sync send topic. No default value")
flag.StringVar(
&compress,
"compress-type", "", "kafka message compress type. Default value is no compress")
} }
type kafkaExporter struct { type kafkaExporter struct {
@ -39,6 +44,20 @@ type kafkaExporter struct {
} }
func checkParams() { func checkParams() {
switch compress {
case "":
compressType = sarama.CompressionNone
case "gzip":
compressType = sarama.CompressionGZIP
case "snappy":
compressType = sarama.CompressionSnappy
case "lz4":
compressType = sarama.CompressionLZ4
default:
panic(fmt.Sprintf("cannot support compress type: %s", compress))
}
fmt.Printf("kafka message compress type: %s", compress)
params := make(map[string]string) params := make(map[string]string)
params["kafka-server"] = kafkaServer params["kafka-server"] = kafkaServer
params["kafka-group-id"] = kafkaGroupID params["kafka-group-id"] = kafkaGroupID
@ -57,6 +76,7 @@ func NewKafkaExporter() (ke *kafkaExporter) {
conf := sarama.NewConfig() conf := sarama.NewConfig()
conf.Producer.Return.Successes = true conf.Producer.Return.Successes = true
conf.ClientID = kafkaGroupID conf.ClientID = kafkaGroupID
conf.Producer.Compression = compressType
addrs := strings.Split(kafkaServer, ",") addrs := strings.Split(kafkaServer, ",")
syncProducer, err := sarama.NewSyncProducer(addrs, conf) syncProducer, err := sarama.NewSyncProducer(addrs, conf)
if err != nil { if err != nil {