Merge branch 'feature/test' into develop

Author: Laisky

acceptor.go

@@ -28,31 +28,43 @@ type Acceptor struct {
 
 // NewAcceptor create new Acceptor
 func NewAcceptor(cfg *AcceptorCfg, recvs ...recvs.AcceptorRecvItf) *Acceptor {
-	libs.Logger.Info("create Acceptor")
-
-	if cfg.MaxRotateID == 0 {
-		cfg.MaxRotateID = 372036854775807
-		libs.Logger.Info("reset MaxRotateID", zap.Int("MaxRotateID", 372036854775807))
-	} else if cfg.MaxRotateID < 100 {
-		libs.Logger.Error("MaxRotateID should not too small", zap.Int64("rotate", cfg.MaxRotateID))
-	} else if cfg.MaxRotateID < 1000000 {
-		libs.Logger.Warn("MaxRotateID should not too small", zap.Int64("rotate", cfg.MaxRotateID))
-	}
-	if cfg.SyncOutChanSize == 0 {
-		cfg.SyncOutChanSize = 10000
-		libs.Logger.Info("reset SyncOutChanSize", zap.Int("SyncOutChanSize", 10000))
-	}
-	if cfg.AsyncOutChanSize == 0 {
-		cfg.AsyncOutChanSize = 10000
-		libs.Logger.Info("reset AsyncOutChanSize", zap.Int("AsyncOutChanSize", 10000))
-	}
-
-	return &Acceptor{
+	a := &Acceptor{
 		AcceptorCfg:  cfg,
 		syncOutChan:  make(chan *libs.FluentMsg, cfg.SyncOutChanSize),
 		asyncOutChan: make(chan *libs.FluentMsg, cfg.AsyncOutChanSize),
 		recvs:        recvs,
 	}
+	if err := a.valid(); err != nil {
+		libs.Logger.Panic("new acceptor", zap.Error(err))
+	}
+
+	libs.Logger.Info("create acceptor",
+		zap.Int64("max_rotate_id", a.MaxRotateID),
+		zap.Int("sync_out_chan_size", a.SyncOutChanSize),
+		zap.Int("async_out_chan_size", a.AsyncOutChanSize),
+	)
+	return a
+}
+
+func (a *Acceptor) valid() error {
+	if a.MaxRotateID == 0 {
+		a.MaxRotateID = 372036854775807
+		libs.Logger.Info("reset max_rotate_id", zap.Int64("max_rotate_id", a.MaxRotateID))
+	} else if a.MaxRotateID < 1000000 {
+		libs.Logger.Warn("max_rotate_id should not too small", zap.Int64("max_rotate_id", a.MaxRotateID))
+	}
+
+	if a.SyncOutChanSize == 0 {
+		a.SyncOutChanSize = 10000
+		libs.Logger.Info("reset sync_out_chan_size", zap.Int("sync_out_chan_size", a.SyncOutChanSize))
+	}
+
+	if a.AsyncOutChanSize == 0 {
+		a.AsyncOutChanSize = 10000
+		libs.Logger.Info("reset async_out_chan_size", zap.Int("async_out_chan_size", a.AsyncOutChanSize))
+	}
+
+	return nil
 }
 
 // Run starting acceptor to listening and receive messages,

dispacher.go

@@ -30,8 +30,6 @@ type Dispatcher struct {
 
 // NewDispatcher create new Dispatcher
 func NewDispatcher(cfg *DispatcherCfg) *Dispatcher {
-	libs.Logger.Info("create Dispatcher")
-
 	d := &Dispatcher{
 		DispatcherCfg: cfg,
 		outChan:       make(chan *libs.FluentMsg, cfg.OutChanSize),
@@ -44,6 +42,10 @@ func NewDispatcher(cfg *DispatcherCfg) *Dispatcher {
 		libs.Logger.Panic("config invalid", zap.Error(err))
 	}
 
+	libs.Logger.Info("create Dispatcher",
+		zap.Int("n_fork", d.NFork),
+		zap.Int("out_chan_size", d.OutChanSize),
+	)
 	return d
 }
 
@@ -152,6 +154,10 @@ func (d *Dispatcher) registerMonitor() {
 		metrics := map[string]interface{}{
 			"msgPerSec": d.counter.GetSpeed(),
 			"msgTotal":  d.counter.Get(),
+			"config": map[string]interface{}{
+				"n_fork":        d.NFork,
+				"out_chan_size": d.OutChanSize,
+			},
 		}
 		d.tag2Counter.Range(func(tagi interface{}, ci interface{}) bool {
 			metrics[tagi.(string)+".MsgPerSec"] = ci.(*utils.Counter).GetSpeed()

docs/example/app/app.py

@@ -44,7 +44,7 @@ def main():
     while 1:
         cnt += 1
         print(produce_log())
-        time.sleep(0.01)
+        time.sleep(0.1)
         if time.time() - t > 10:
             t = time.time()
             print(produce_log() + f"speed: {cnt/10}/s")

docs/example/docker-compose.yml

@@ -10,7 +10,7 @@ services:
       driver: "fluentd"
       options:
         tag: test.sit
-        fluentd-address: "127.0.0.1:24225"
+        fluentd-address: "127.0.0.1:25225"
         fluentd-async-connect: "true"
         mode: non-blocking
         max-buffer-size: 1m
@@ -23,10 +23,11 @@ services:
     image: ppcelery/go-fluentd:test
     restart: "on-failure"
     ports:
-      - 8080:8080
-      - 24225:24225
+      - 28080:8080
+      - 25225:24225
     volumes:
       - ./go-fluentd/settings.yml:/etc/go-fluentd/settings.yml:ro
+      - ./data:/data
     command: --config=/etc/go-fluentd/settings.yml
       --env=sit
       --addr=0.0.0.0:8080

docs/example/go-fluentd/settings.yml

@@ -12,6 +12,7 @@ consts:
 settings:
   acceptor:
     recvs:
+      plugins:
         # 监听 DC/OS fluentd log-driver 的日志
         fluentd:
           type: fluentd
@@ -31,7 +32,7 @@ settings:
         log_level: info
         is_commit: true
   journal:
-    buf_dir_path: ./data
+    buf_dir_path: /data
     is_compress: true
   acceptor_filters:
     plugins:
@@ -54,17 +55,12 @@ settings:
         type: parser
         tags:
           - test
-        msg_key: log
         # 2018-03-06 16:56:22.514 | mscparea | INFO  | http-nio-8080-exec-1 | com.google.cloud.cp.core.service.impl.CPBusiness.reflectAdapterRequest | 84: test
         pattern: (?ms)^(?P<time>.{23}) {0,}\| {0,}(?P<app>[^\|]+) {0,}\| {0,}(?P<level>[^\|]+) {0,}\| {0,}(?P<thread>[^\|]+) {0,}\| {0,}(?P<class>[^ ]+) {0,}\| {0,}(?P<line>\d+) {0,}([\|:] {0,}(?P<args>\{.*\}))?([\|:] {0,}(?P<message>.*))?
         is_remove_orig_log: true
         must_include: app
         # ⬇⬇ time
-        time_key: time
         time_format: "2006-01-02 15:04:05.000 -0700"
-        new_time_format: "2006-01-02T15:04:05.000000Z"
-        reserved_time_key: false
-        new_time_key: "@timestamp"
         append_time_zone:
           sit: "+0800"
           perf: "+0800"

docs/settings/tiny_settings.yml

@@ -12,6 +12,7 @@ consts:
 settings:
   acceptor:
     recvs:
+      plugins:
         # 监听 DC/OS fluentd log-driver 的日志
         fluentd:
           type: fluentd
@@ -54,17 +55,12 @@ settings:
         type: parser
         tags:
           - test
-        msg_key: log
         # 2018-03-06 16:56:22.514 | mscparea | INFO  | http-nio-8080-exec-1 | com.google.cloud.cp.core.service.impl.CPBusiness.reflectAdapterRequest | 84: test
         pattern: (?ms)^(?P<time>.{23}) {0,}\| {0,}(?P<app>[^\|]+) {0,}\| {0,}(?P<level>[^\|]+) {0,}\| {0,}(?P<thread>[^\|]+) {0,}\| {0,}(?P<class>[^ ]+) {0,}\| {0,}(?P<line>\d+) {0,}([\|:] {0,}(?P<args>\{.*\}))?([\|:] {0,}(?P<message>.*))?
         is_remove_orig_log: true
         must_include: app
         # ⬇⬇ time
-        time_key: time
         time_format: "2006-01-02 15:04:05.000 -0700"
-        new_time_format: "2006-01-02T15:04:05.000000Z"
-        reserved_time_key: false
-        new_time_key: "@timestamp"
         append_time_zone:
           sit: "+0800"
           perf: "+0800"

go.mod

@@ -4,12 +4,12 @@ go 1.12
 
 require (
 	github.com/Laisky/gin-middlewares v1.1.1
-	github.com/Laisky/go-journal v1.1.3
+	github.com/Laisky/go-journal v1.1.5
 	github.com/Laisky/go-kafka v1.0.1
 	github.com/Laisky/go-syslog v2.3.3+incompatible
 	github.com/Laisky/go-utils v1.12.9
 	github.com/Laisky/zap v1.12.2
-	github.com/Shopify/sarama v1.26.3
+	github.com/Shopify/sarama v1.26.4
 	github.com/cespare/xxhash v1.1.0
 	github.com/gin-contrib/pprof v1.3.0
 	github.com/gin-gonic/gin v1.6.3

go.sum

@@ -5,8 +5,8 @@ github.com/Laisky/gin-middlewares v1.1.1 h1:EVFwX94bMVSeJ4a6TXgT6BXvhqyn6j0kKQhE
 github.com/Laisky/gin-middlewares v1.1.1/go.mod h1:IJxjhBGdN+o6BfGME2qbxZl54i5jQmcnwMh6QUd89T8=
 github.com/Laisky/go-chaining v0.0.0-20180507092046-43dcdc5a21be h1:7Rxhm6IjOtDAyj8eScOFntevwzkWhx94zi48lxo4m4w=
 github.com/Laisky/go-chaining v0.0.0-20180507092046-43dcdc5a21be/go.mod h1:1mdzaETo0kjvCQICPSePsoaatJN4l7JvEA1200lyevo=
-github.com/Laisky/go-journal v1.1.3 h1:6AMsDuhrWcit6oHkVhcvyJmitt4fFicP4UpvcT8ALlI=
-github.com/Laisky/go-journal v1.1.3/go.mod h1:6302T+Uo0+xYp5O9Z+GalHnl+R3hmRyw0aa11IqNX90=
+github.com/Laisky/go-journal v1.1.5 h1:6IiDPBrUSiM7TZAHRa7tIoHVBnkPGW2dFYj3q3XryZk=
+github.com/Laisky/go-journal v1.1.5/go.mod h1:6302T+Uo0+xYp5O9Z+GalHnl+R3hmRyw0aa11IqNX90=
 github.com/Laisky/go-kafka v1.0.1 h1:DNMBLshSMUUedH9nmBhGrpEf0FLGIAQfPlbQBkFcCfw=
 github.com/Laisky/go-kafka v1.0.1/go.mod h1:AksqzI2+yDSX6MtQikP6aTg9om52pSdropjst0oqeZE=
 github.com/Laisky/go-syslog v2.3.3+incompatible h1:TSHhP3iadAPDzC5efyYLPnGkv2pvUtuUInm7poVRkFA=
@@ -25,8 +25,8 @@ github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAE
 github.com/RoaringBitmap/roaring v0.4.23 h1:gpyfd12QohbqhFO4NVDUdoPOCXsyahYRQhINmlHxKeo=
 github.com/RoaringBitmap/roaring v0.4.23/go.mod h1:D0gp8kJQgE1A4LQ5wFLggQEyvDi06Mq5mKs52e1TwOo=
 github.com/Shopify/sarama v1.26.1/go.mod h1:NbSGBSSndYaIhRcBtY9V0U7AyH+x71bG668AuWys/yU=
-github.com/Shopify/sarama v1.26.3 h1:wSN3FpDXLe3e2z47OzGii5VAK693oVkyHFwh240jWjg=
-github.com/Shopify/sarama v1.26.3/go.mod h1:NbSGBSSndYaIhRcBtY9V0U7AyH+x71bG668AuWys/yU=
+github.com/Shopify/sarama v1.26.4 h1:+17TxUq/PJEAfZAll0T7XJjSgQWCpaQSoki/x5yN8o8=
+github.com/Shopify/sarama v1.26.4/go.mod h1:NbSGBSSndYaIhRcBtY9V0U7AyH+x71bG668AuWys/yU=
 github.com/Shopify/toxiproxy v2.1.4+incompatible h1:TKdv8HiTLgE5wdJuEML90aBgNWsokNbMijUGhmcoBJc=
 github.com/Shopify/toxiproxy v2.1.4+incompatible/go.mod h1:OXgGpZ6Cli1/URJOF1DMxUHB2q5Ap20/P/eIdh4G0pI=
 github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=

journal.go

@@ -571,14 +571,17 @@ func (j *Journal) DumpMsgFlow(ctx context.Context, msgPool *sync.Pool, dumpChan,
 			default:
 				select {
 				case jji.(chan *libs.FluentMsg) <- msg:
-				case j.outChan <- msg:
-					libs.Logger.Warn("skip dump since journal is busy", zap.String("tag", msg.Tag))
 				default:
-					libs.Logger.Error("discard log since of journal & downstream busy",
-						zap.String("tag", msg.Tag),
-						zap.String("msg", fmt.Sprint(msg)),
-					)
-					j.MsgPool.Put(msg)
+					select {
+					case j.outChan <- msg:
+						libs.Logger.Warn("skip dump since journal is busy", zap.String("tag", msg.Tag))
+					default:
+						libs.Logger.Error("discard log since of journal & downstream busy",
+							zap.String("tag", msg.Tag),
+							zap.String("msg", fmt.Sprint(msg)),
+						)
+						j.MsgPool.Put(msg)
+					}
 				}
 			}
 		}
@@ -641,7 +644,18 @@ func (j *Journal) startCommitRunner(ctx context.Context) {
 
 func (j *Journal) registerMonitor() {
 	monitor.AddMetric("journal", func() map[string]interface{} {
-		result := map[string]interface{}{}
+		result := map[string]interface{}{
+			"config": map[string]interface{}{
+				"compress":             j.IsCompress,
+				"buf_dir_path":         j.BufDirPath,
+				"buf_file_bytes":       j.BufSizeBytes,
+				"gc_inteval_sec":       j.GCIntervalSec / time.Second,
+				"journal_out_chan_len": j.JournalOutChanLen,
+				"commit_id_chan_len":   j.CommitIDChanLen,
+				"child_data_chan_len":  j.ChildJournalDataInchanLen,
+				"child_id_chan_len":    j.ChildJournalIDInchanLen,
+			},
+		}
 		j.tag2JMap.Range(func(k, v interface{}) bool {
 			result[k.(string)+".journal"] = v.(*journal.Journal).GetMetric()
 			return true

producer.go

@@ -94,15 +94,15 @@ func NewProducer(cfg *ProducerCfg, senders ...senders.SenderItf) (*Producer, err
 	}
 
 	libs.Logger.Info("new producer",
-		zap.Int("nfork", 1),
+		zap.Int("nfork", p.NFork),
 		zap.Int("discard_chan_size", p.DiscardChanSize),
 	)
 	return p, nil
 }
 
 func (p *Producer) valid() error {
 	if p.NFork <= 0 {
-		p.NFork = 1
+		p.NFork = 4
 		libs.Logger.Info("reset nfork", zap.Int("nfork", 1))
 	}
 
@@ -118,6 +118,10 @@ func (p *Producer) valid() error {
 func (p *Producer) registerMonitor() {
 	monitor.AddMetric("producer", func() map[string]interface{} {
 		metrics := map[string]interface{}{
+			"config": map[string]interface{}{
+				"nfork":             p.NFork,
+				"discard_chan_size": p.DiscardChanSize,
+			},
 			"msgPerSec": p.counter.GetSpeed(),
 			"msgTotal":  p.counter.Get(),
 		}

tagFilters/parser_f.go

@@ -220,27 +220,65 @@ type ParserFact struct {
 }
 
 func NewParserFact(cfg *ParserFactCfg) *ParserFact {
-	libs.Logger.Info("create new connectorfactory")
-
-	if cfg.NFork < 1 {
-		libs.Logger.Warn("nfork should bigger than 1")
-		cfg.NFork = 1
-	}
-
 	cf := &ParserFact{
 		BaseTagFilterFactory: &BaseTagFilterFactory{},
 		ParserFactCfg:        cfg,
 	}
+	if err := cf.valid(); err != nil {
+		libs.Logger.Panic("new parser", zap.Error(err))
+	}
 
 	cf.tagsset = map[string]struct{}{}
 	for _, tag := range cf.Tags {
 		libs.Logger.Info("Parser factory add tag", zap.String("tag", tag+"."+cf.Env))
 		cf.tagsset[tag+"."+cf.Env] = struct{}{}
 	}
 
+	libs.Logger.Info("new parser",
+		zap.Int("n_fork", cf.NFork),
+		zap.String("msg_key", cf.MsgKey),
+		zap.String("time_key", cf.TimeKey),
+		zap.String("new_time_format", cf.NewTimeFormat),
+		zap.String("new_time_key", cf.NewTimeKey),
+		zap.String("msg_key", cf.MsgKey),
+	)
 	return cf
 }
 
+func (cf *ParserFact) valid() error {
+	if cf.NFork < 1 {
+		cf.NFork = 4
+		libs.Logger.Info("reset n_fork", zap.Int("n_fork", cf.NFork))
+	}
+
+	if cf.MsgKey == "" {
+		cf.MsgKey = "log"
+		libs.Logger.Info("reset msg_key", zap.String("msg_key", cf.MsgKey))
+	}
+
+	if cf.TimeKey == "" {
+		cf.TimeKey = "time"
+		libs.Logger.Info("reset time_key", zap.String("time_key", cf.TimeKey))
+	}
+
+	if cf.NewTimeFormat == "" {
+		cf.NewTimeFormat = "2006-01-02T15:04:05.000000Z"
+		libs.Logger.Info("reset new_time_format", zap.String("new_time_format", cf.NewTimeFormat))
+	}
+
+	if cf.NewTimeKey == "" {
+		cf.NewTimeKey = "@timestamp"
+		libs.Logger.Info("reset new_time_key", zap.String("new_time_key", cf.NewTimeKey))
+	}
+
+	if cf.MsgKey == "" {
+		cf.MsgKey = "log"
+		libs.Logger.Info("reset msg_key", zap.String("msg_key", cf.MsgKey))
+	}
+
+	return nil
+}
+
 func (cf *ParserFact) GetName() string {
 	return cf.Name + "-parser"
 }