Merge branch 'develop'

Author: Laisky

.docker/Dockerfile

@@ -1,5 +1,10 @@
 # docker build --build-arg http_proxy=http://172.16.4.26:17777 --build-arg https_proxy=http://172.16.4.26:17777
-FROM ppcelery/gobase:1.14.0-alpine3.11 AS gobuild
+FROM golang:1.14.3-buster AS gobuild
+
+# install dependencies
+RUN apt-get update \
+    && apt-get install -y --no-install-recommends g++ make gcc git build-essential ca-certificates curl \
+    && update-ca-certificates
 
 ENV GO111MODULE=on
 WORKDIR /go-fluentd
@@ -14,8 +19,9 @@ RUN go build -a -ldflags '-w -extldflags "-static"' entrypoints/main.go
 
 
 # copy executable file and certs to a pure container
-FROM alpine:3.11
+FROM debian:buster
 COPY --from=gobuild /etc/ssl/certs /etc/ssl/certs
 COPY --from=gobuild /go-fluentd/main go-fluentd
 
-CMD ["./go-fluentd", "--config=/etc/go-fluentd/settings"]
+ENTRYPOINT [ "./go-fluentd" ]
+CMD ["--config=/etc/go-fluentd/settings.yml"]

.docker/README.md

@@ -13,13 +13,6 @@ there are two kinds of runner container:
 build on machine that should installed docker & golang.
 
 ```sh
-go mod download
-go mod vendor
-
-# build base image
-docker build . -f ./.docker/gobase.Dockerfile -t ppcelery/gobase:1.14.0-alpine3.11
-docker push ppcelery/gobase:1.14.0-alpine3.11
-
 # build image
 docker build . -f ./.docker/Dockerfile -t ppcelery/go-fluentd:1.12.7
 docker push ppcelery/go-fluentd:1.12.7
@@ -29,7 +22,7 @@ docker run -it --rm \
     -v /opt/configs/go-fluentd:/etc/go-fluentd \
     -v /data/log/fluentd/go-concator:/data/log/fluentd/go-concator \
     ppcelery/go-fluentd:1.12.7 \
-    ./go-fluentd --config=/etc/go-fluentd --env=prod --addr=0.0.0.0:22800 --log-level=error
+    ./go-fluentd --config=/etc/go-fluentd/ --env=prod --addr=0.0.0.0:22800 --log-level=error
 ```
 
 ## go-fluentd-foward

.docker/gobase.Dockerfile

@@ -28,30 +28,53 @@ type Acceptor struct {
 
 // NewAcceptor create new Acceptor
 func NewAcceptor(cfg *AcceptorCfg, recvs ...recvs.AcceptorRecvItf) *Acceptor {
-	utils.Logger.Info("create Acceptor")
-
-	if cfg.MaxRotateID < 100 {
-		utils.Logger.Error("MaxRotateID should not too small", zap.Int64("rotate", cfg.MaxRotateID))
-	} else if cfg.MaxRotateID < 1000000 {
-		utils.Logger.Warn("MaxRotateID should not too small", zap.Int64("rotate", cfg.MaxRotateID))
-	}
-
-	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,
 // you can use `acceptor.MessageChan()` to load messages`
 func (a *Acceptor) Run(ctx context.Context) {
 	// got exists max id from legacy
-	utils.Logger.Info("process legacy data...")
+	libs.Logger.Info("process legacy data...")
 	maxID, err := a.Journal.LoadMaxID()
 	if err != nil {
-		utils.Logger.Panic("try to process legacy messages got error", zap.Error(err))
+		libs.Logger.Panic("try to process legacy messages got error", zap.Error(err))
 	}
 
 	couter, err := utils.NewParallelCounterFromN((maxID+1)%a.MaxRotateID, 10000, a.MaxRotateID)
@@ -60,7 +83,7 @@ func (a *Acceptor) Run(ctx context.Context) {
 	}
 
 	for _, recv := range a.recvs {
-		utils.Logger.Info("enable recv", zap.String("name", recv.GetName()))
+		libs.Logger.Info("enable recv", zap.String("name", recv.GetName()))
 		recv.SetAsyncOutChan(a.asyncOutChan)
 		recv.SetSyncOutChan(a.syncOutChan)
 		recv.SetMsgPool(a.MsgPool)

acceptor.go

@@ -2,7 +2,6 @@ package acceptorFilters
 
 import (
 	"github.com/Laisky/go-fluentd/libs"
-	utils "github.com/Laisky/go-utils"
 	"github.com/Laisky/zap"
 )
 
@@ -19,7 +18,7 @@ type DefaultFilter struct {
 }
 
 func NewDefaultFilter(cfg *DefaultFilterCfg) *DefaultFilter {
-	utils.Logger.Info("NewDefaultFilter")
+	libs.Logger.Info("NewDefaultFilter")
 
 	f := &DefaultFilter{
 		BaseFilter:       &BaseFilter{},
@@ -45,13 +44,13 @@ func (f *DefaultFilter) IsTagInConfigs(tag string) (ok bool) {
 
 func (f *DefaultFilter) Filter(msg *libs.FluentMsg) *libs.FluentMsg {
 	if f.RemoveEmptyTag && msg.Tag == "" {
-		utils.Logger.Warn("discard log since empty tag", zap.String("tag", msg.Tag))
+		libs.Logger.Warn("discard log since empty tag", zap.String("tag", msg.Tag))
 		f.DiscardMsg(msg)
 		return nil
 	}
 
 	if f.RemoveUnsupportTag && !f.IsTagInConfigs(msg.Tag) {
-		utils.Logger.Warn("discard log since unsupported tag", zap.String("tag", msg.Tag))
+		libs.Logger.Warn("discard log since unsupported tag", zap.String("tag", msg.Tag))
 		f.DiscardMsg(msg)
 		return nil
 	}

acceptorFilters/default_f.go

@@ -28,25 +28,24 @@ type AcceptorPipeline struct {
 }
 
 func NewAcceptorPipeline(ctx context.Context, cfg *AcceptorPipelineCfg, filters ...AcceptorFilterItf) (a *AcceptorPipeline, err error) {
-	utils.Logger.Info("NewAcceptorPipeline")
-	if cfg.NFork < 1 {
-		panic(fmt.Errorf("NFork should greater than 1, got: %v", cfg.NFork))
-	}
-
 	a = &AcceptorPipeline{
 		AcceptorPipelineCfg: cfg,
 		filters:             filters,
 		reEnterChan:         make(chan *libs.FluentMsg, cfg.ReEnterChanSize),
 		counter:             utils.NewCounter(),
 	}
+	if err := a.valid(); err != nil {
+		libs.Logger.Panic("invalid cfg for acceptor pipeline")
+	}
+
 	a.registerMonitor()
 	for _, filter := range a.filters {
 		filter.SetUpstream(a.reEnterChan)
 		filter.SetMsgPool(a.MsgPool)
 	}
 
 	if a.IsThrottle {
-		utils.Logger.Info("enable acceptor throttle",
+		libs.Logger.Info("enable acceptor throttle",
 			zap.Int("max", a.ThrottleMax),
 			zap.Int("n_perf_sec", a.ThrottleNPerSec))
 		if a.throttle, err = utils.NewThrottleWithCtx(
@@ -59,9 +58,53 @@ func NewAcceptorPipeline(ctx context.Context, cfg *AcceptorPipelineCfg, filters
 		}
 	}
 
+	libs.Logger.Info("new acceptor pipeline",
+		zap.Int("n_fork", a.NFork),
+		zap.Int("out_buf_len", a.OutChanSize),
+		zap.Int("reenter_chan_len", a.ReEnterChanSize),
+		zap.Int("throttle_max", a.ThrottleMax),
+		zap.Int("throttle_per_sec", a.ThrottleNPerSec),
+		zap.Bool("is_throttle", a.IsThrottle),
+	)
 	return a, nil
 }
 
+func (f *AcceptorPipeline) valid() error {
+	if f.NFork <= 0 {
+		f.NFork = 4
+		libs.Logger.Info("reset n_fork", zap.Int("n_fork", f.NFork))
+	}
+
+	if f.OutChanSize <= 0 {
+		f.OutChanSize = 1000
+		libs.Logger.Info("reset out_buf_len", zap.Int("out_buf_len", f.OutChanSize))
+	}
+
+	if f.ReEnterChanSize <= 0 {
+		f.ReEnterChanSize = 1000
+		libs.Logger.Info("reset reenter_chan_len", zap.Int("reenter_chan_len", f.ReEnterChanSize))
+	}
+
+	if f.NFork <= 0 {
+		f.NFork = 4
+		libs.Logger.Info("reset n_fork", zap.Int("n_fork", f.NFork))
+	}
+
+	if f.IsThrottle {
+		if f.ThrottleMax <= 0 {
+			f.ThrottleMax = 10000
+			libs.Logger.Info("reset throttle_max", zap.Int("throttle_max", f.ThrottleMax))
+		}
+
+		if f.ThrottleNPerSec <= 0 {
+			f.ThrottleNPerSec = 1000
+			libs.Logger.Info("reset throttle_per_sec", zap.Int("throttle_per_sec", f.ThrottleNPerSec))
+		}
+	}
+
+	return nil
+}
+
 func (f *AcceptorPipeline) registerMonitor() {
 	monitor.AddMetric("acceptorPipeline", func() map[string]interface{} {
 		metrics := map[string]interface{}{
@@ -88,7 +131,7 @@ func (f *AcceptorPipeline) Wrap(ctx context.Context, asyncInChan, syncInChan cha
 				msg    *libs.FluentMsg
 				ok     bool
 			)
-			defer utils.Logger.Info("quit acceptorPipeline asyncChan", zap.String("last_msg", fmt.Sprint(msg)))
+			defer libs.Logger.Info("quit acceptorPipeline asyncChan", zap.String("last_msg", fmt.Sprint(msg)))
 
 		NEXT_ASYNC_MSG:
 			for {
@@ -97,21 +140,21 @@ func (f *AcceptorPipeline) Wrap(ctx context.Context, asyncInChan, syncInChan cha
 					return
 				case msg, ok = <-f.reEnterChan: // CAUTION: do not put msg into reEnterChan forever
 					if !ok {
-						utils.Logger.Info("reEnterChan closed")
+						libs.Logger.Info("reEnterChan closed")
 						return
 					}
 				case msg, ok = <-asyncInChan:
 					if !ok {
-						utils.Logger.Info("asyncInChan closed")
+						libs.Logger.Info("asyncInChan closed")
 						return
 					}
 				}
 				f.counter.Count()
 
-				// utils.Logger.Debug("AcceptorPipeline got msg")
+				// libs.Logger.Debug("AcceptorPipeline got msg")
 
 				if f.IsThrottle && !f.throttle.Allow() {
-					utils.Logger.Warn("discard msg by throttle", zap.String("tag", msg.Tag))
+					libs.Logger.Warn("discard msg by throttle", zap.String("tag", msg.Tag))
 					f.DiscardMsg(msg)
 					continue
 				}
@@ -129,7 +172,7 @@ func (f *AcceptorPipeline) Wrap(ctx context.Context, asyncInChan, syncInChan cha
 					case outChan <- msg:
 					case skipDumpChan <- msg: // baidu has low disk performance
 					default:
-						utils.Logger.Error("discard msg since disk & downstream are busy", zap.String("tag", msg.Tag))
+						libs.Logger.Error("discard msg since disk & downstream are busy", zap.String("tag", msg.Tag))
 						f.MsgPool.Put(msg)
 					}
 				}
@@ -143,7 +186,7 @@ func (f *AcceptorPipeline) Wrap(ctx context.Context, asyncInChan, syncInChan cha
 				msg    *libs.FluentMsg
 				ok     bool
 			)
-			defer utils.Logger.Info("quit acceptorPipeline syncChan", zap.String("last_msg", fmt.Sprint(msg)))
+			defer libs.Logger.Info("quit acceptorPipeline syncChan", zap.String("last_msg", fmt.Sprint(msg)))
 
 		NEXT_SYNC_MSG:
 			for {
@@ -152,15 +195,15 @@ func (f *AcceptorPipeline) Wrap(ctx context.Context, asyncInChan, syncInChan cha
 					return
 				case msg, ok = <-syncInChan:
 					if !ok {
-						utils.Logger.Info("syncInChan closed")
+						libs.Logger.Info("syncInChan closed")
 						return
 					}
 				}
-				// utils.Logger.Debug("AcceptorPipeline got blockable msg")
+				// libs.Logger.Debug("AcceptorPipeline got blockable msg")
 				f.counter.Count()
 
 				if f.IsThrottle && !f.throttle.Allow() {
-					utils.Logger.Warn("discard msg by throttle", zap.String("tag", msg.Tag))
+					libs.Logger.Warn("discard msg by throttle", zap.String("tag", msg.Tag))
 					f.DiscardMsg(msg)
 					continue
 				}

acceptorFilters/pipeline.go

@@ -5,7 +5,6 @@ import (
 	"regexp"
 
 	"github.com/Laisky/go-fluentd/libs"
-	"github.com/Laisky/go-utils"
 	"github.com/Laisky/zap"
 )
 
@@ -22,7 +21,7 @@ type SparkFilter struct {
 }
 
 func NewSparkFilter(cfg *SparkFilterCfg) *SparkFilter {
-	utils.Logger.Info("NewSparkFilter",
+	libs.Logger.Info("NewSparkFilter",
 		zap.String("regex", cfg.IgnoreRegex.String()),
 		zap.String("tag", cfg.Tag))
 
@@ -52,7 +51,7 @@ func (f *SparkFilter) Filter(msg *libs.FluentMsg) *libs.FluentMsg {
 	}
 
 	// discard some format
-	// utils.Logger.Debug("ignore spark log",
+	// libs.Logger.Debug("ignore spark log",
 	// 	zap.String("tag", f.Tag),
 	// 	zap.ByteString("log", msg.Message[f.MsgKey].([]byte)))
 	if f.IgnoreRegex.Match(msg.Message[f.MsgKey].([]byte)) {