ci: release v1.10.0

Author: Laisky

CHANGELOG.md

@@ -2,6 +2,18 @@
 *CURRENT*
 ---
 
+- 2019-08-20 (Laisky) ci: upgrade go-utils
+- 2019-08-20 (Laisky) ci: upgrade go-utils
+- 2019-08-19 (Laisky) docs: update quickstart
+- 2019-08-19 (Laisky) feat: enable gz in journal
+- 2019-08-19 (Laisky) style: more log
+
+*v1.9.3*
+---
+
+- 2019-08-15 (Laisky) build: upgrade go-utils to v1.5.4
+- 2019-08-15 (Laisky) fix: double default postfilter
+- 2019-08-14 (Laisky) docs: update changelog
 - 2019-08-14 (Laisky) feat: support `@RANDOM_STRING`
 - 2019-08-14 (Laisky) build: upgrade to go 1.12.7
 - 2019-08-14 (Laisky) feat(paas-390): add wuling mapping

docs/quickstart.md

@@ -2,12 +2,10 @@
 
 Running a minimal example with app, go-fluentd and fluentd.
 
-App ---> Go-Fluentd ---> Fluentd
+App ---> Go-Fluentd(with stdout sender plugin)
 
 - App: generate and emit logs
 - Go-Fluentd: collect and parse log
-- Fluentd: just a example for backend, it also could be ElasticSearch or something else
-
 
 
 ## Prepare
@@ -37,7 +35,6 @@ $ sudo docker-compose ps
 
          Name                        Command               State                          Ports
 -----------------------------------------------------------------------------------------------------------------------
-example_fluentd_1         /bin/entrypoint.sh /bin/sh ...   Up      24224/tcp, 5140/tcp
 example_go-fluentd_1      ./go-fluentd --config=/etc ...   Up      127.0.0.1:24225->24225/tcp, 127.0.0.1:8080->8080/tcp
 example_log-generator_1   python /app.py                   Up
 ```
@@ -53,12 +50,6 @@ You can check the logs that parserd by go-fluentd:
 ```sh
 $ sudo docker logs example_fluentd_1
 
-2019-02-28 08:41:21.534852412 +0000 test.sit: {"tag":"test.sit","app":"app","thread":"thread","class":"class","message":"0.8336017742577866\n0.059360002847527626\n0.9471091772460405","msgid":1377,"container_name":"/example_log-generator_1","source":"stdout","level":"INFO","line":"64","datasource":"test","@timestamp":"2019-02-21T07:41:17.871000Z","container_id":"24d6069f241ad94719ac1eee15dce43e29a3a32af67c478ded9c474066389260"}
-2019-02-28 08:41:21.539087614 +0000 test.sit: {"container_name":"/example_log-generator_1","msgid":1026,"line":"64","message":"0.7159115118036709","datasource":"test","level":"INFO","thread":"thread","class":"class","@timestamp":"2019-02-03T17:41:13.813000Z","container_id":"24d6069f241ad94719ac1eee15dce43e29a3a32af67c478ded9c474066389260","source":"stdout","tag":"test.sit","app":"app"}
-```
-
-With format:
-
 ```js
 {
     "tag": "test.sit",
@@ -106,49 +97,52 @@ You can load monitor metrics by <http://localhost:8080/monitor>
 // http://localhost:8080/monitor
 
 {
-    "acceptorPipeline": {
-        "msgPerSec": 81.1
-    },
-    "controllor": {
-        "goroutine": 37,
-        "skipDumpChanCap": 150000,
-        "skipDumpChanLen": 0,
-        "waitAccepPipelineAsyncChanCap": 100000,
-        "waitAccepPipelineAsyncChanLen": 0,
-        "waitAccepPipelineSyncChanCap": 10000,
-        "waitAccepPipelineSyncChanLen": 0,
-        "waitCommitChanCap": 500000,
-        "waitCommitChanLen": 0,
-        "waitDispatchChanCap": 100000,
-        "waitDispatchChanLen": 0,
-        "waitDumpChanCap": 150000,
-        "waitDumpChanLen": 0,
-        "waitPostPipelineChanCap": 10000,
-        "waitPostPipelineChanLen": 0,
-        "waitProduceChanCap": 10000,
-        "waitProduceChanLen": 0
-    },
-    "dispatcher": {
-        "msgPerSec": 81.1,
-        "test.sit.ChanCap": 10000,
-        "test.sit.ChanLen": 0,
-        "test.sit.MsgPerSec": 81.1
-    },
-    "producer": {
-        "discardChanCap": 50000,
-        "discardChanLen": 0,
-        "msgPerSec": 36.7,
-        "test.sit.fluentd.ChanCap": 50000,
-        "test.sit.fluentd.ChanLen": 0,
-        "waitToDiscardMsgNum": 0
-    },
-    "tagpipeline": {
-        "test.sit.concator_tagfilter.ChanCap": 10000,
-        "test.sit.concator_tagfilter.ChanLen": 0,
-        "test.sit.spring.ChanCap": 10000,
-        "test.sit.spring.ChanLen": 0
-    },
-    "ts": "2019-02-28T08:32:21.560776792Z"
+  "acceptorPipeline": {
+    "msgPerSec": 252160.4
+  },
+  "controllor": {
+    "goroutine": 64,
+    "skipDumpChanCap": 150000,
+    "skipDumpChanLen": 5299,
+    "waitAccepPipelineAsyncChanCap": 100000,
+    "waitAccepPipelineAsyncChanLen": 17174,
+    "waitAccepPipelineSyncChanCap": 10000,
+    "waitAccepPipelineSyncChanLen": 0,
+    "waitCommitChanCap": 500000,
+    "waitCommitChanLen": 500000,
+    "waitDispatchChanCap": 100000,
+    "waitDispatchChanLen": 99,
+    "waitDumpChanCap": 150000,
+    "waitDumpChanLen": 150000,
+    "waitPostPipelineChanCap": 10000,
+    "waitPostPipelineChanLen": 9777,
+    "waitProduceChanCap": 50000,
+    "waitProduceChanLen": 49946
+  },
+  "dispatcher": {
+    "app.spring.perf.ChanCap": 10000,
+    "app.spring.perf.ChanLen": 0,
+    "app.spring.perf.MsgPerSec": 123304.8,
+    "msgPerSec": 123306.7
+  },
+  "journal": {
+    "idsSetLen": 644303
+  },
+  "producer": {
+    "app.spring.perf.localtest.ChanCap": 50000,
+    "app.spring.perf.localtest.ChanLen": 50000,
+    "discardChanCap": 50000,
+    "discardChanLen": 50000,
+    "msgPerSec": 19111.5,
+    "waitToDiscardMsgNum": 0
+  },
+  "tagpipeline": {
+    "app.spring.perf.concator.ChanCap": 10000,
+    "app.spring.perf.concator.ChanLen": 0,
+    "app.spring.perf.spring-parser.ChanCap": 10000,
+    "app.spring.perf.spring-parser.ChanLen": 4364
+  },
+  "ts": "2019-08-20T01:06:43.934658174Z"
 }
 ```
 </p>

example/docker-compose.yml

@@ -21,7 +21,7 @@ services:
     command: python /app.py
 
   go-fluentd:
-    image: ppcelery/go-fluentd:1.7.2
+    image: ppcelery/go-fluentd:1.9.3
     restart: "on-failure"
     networks:
       - fluentd-test
@@ -40,18 +40,6 @@ services:
       - ./go-fluentd/settings.yml:/etc/go-fluentd/settings.yml:ro
     command: ./go-fluentd --config=/etc/go-fluentd --env=sit  --addr=0.0.0.0:8080 --log-level=info
 
-  fluentd:
-    image: fluent/fluentd:v1.1.0
-    restart: "on-failure"
-    networks:
-      - fluentd-test
-    logging:
-      driver: "json-file"
-      options:
-        max-size: 100m
-    volumes:
-      - ./fluentd/fluent.conf:/fluentd/etc/fluent.conf:ro
-
 
 networks:
   fluentd-test:

example/go-fluentd/settings.yml

@@ -25,18 +25,14 @@ settings:
     discard_chan_size: 50000
     sender_inchan_size: 50000
     plugins:
-      fluentd:
-        type: fluentd
-        active_env:
-          - sit
-        tags:
-          - test.sit
-        forks: 3
-        addr: fluentd:24225
-        msg_batch_size: 10000
-        max_wait_sec: 5
-        retry_chan_len: 50000
-        is_discard_when_blocked: true
+      localtest:
+        active_env: *all-env
+        tags: *all-tags
+        type: "null"
+        forks: 5
+        log_level: info
+        is_commit: true
+        is_discard_when_blocked: false
   journal:
     buf_dir_path: /data/log/fluentd/go-concator
     buf_file_bytes: 209315200  # 200 MB

go.mod

@@ -5,7 +5,7 @@ go 1.12
 require (
 	github.com/Depado/ginprom v1.1.2
 	github.com/Laisky/go-syslog v2.3.3+incompatible
-	github.com/Laisky/go-utils v1.5.4
+	github.com/Laisky/go-utils v1.6.1
 	github.com/Laisky/zap v1.9.2
 	github.com/OneOfOne/xxhash v1.2.5 // indirect
 	github.com/Shopify/sarama v1.22.1

go.sum

@@ -16,8 +16,8 @@ github.com/Laisky/go-chaining v0.0.0-20180507092046-43dcdc5a21be h1:7Rxhm6IjOtDA
 github.com/Laisky/go-chaining v0.0.0-20180507092046-43dcdc5a21be/go.mod h1:1mdzaETo0kjvCQICPSePsoaatJN4l7JvEA1200lyevo=
 github.com/Laisky/go-syslog v2.3.3+incompatible h1:TSHhP3iadAPDzC5efyYLPnGkv2pvUtuUInm7poVRkFA=
 github.com/Laisky/go-syslog v2.3.3+incompatible/go.mod h1:PPmESkLU3DEbJ3fRXam2hqJTNQVFMggsDXBnOtu2ITk=
-github.com/Laisky/go-utils v1.5.4 h1:A8AF990tuyT29QW+l5rj4omZBju+uihrDdqH+X/qv9s=
-github.com/Laisky/go-utils v1.5.4/go.mod h1:MpydOPP7RWH/cINQIgI1exzT8/9wumQ49iw91Y/lDOk=
+github.com/Laisky/go-utils v1.6.1 h1:wfzkGpiFqBDX4TNtPFerElxO9kamc7qk5r74bcz88zo=
+github.com/Laisky/go-utils v1.6.1/go.mod h1:MpydOPP7RWH/cINQIgI1exzT8/9wumQ49iw91Y/lDOk=
 github.com/Laisky/zap v1.9.2 h1:7dTtABboHk8DnT0d6Dc8A9Opu2cyIEaMlL9JO11zvag=
 github.com/Laisky/zap v1.9.2/go.mod h1:CQdLb2wEfqBvoNLmfOp7wnKTOMvhc4DQRc3xfshL4EQ=
 github.com/OneOfOne/xxhash v1.2.2 h1:KMrpdQIwFcEqXDklaen+P1axHaj9BSKzvpUUfnHldSE=

senders/elasticsearch.go

@@ -119,7 +119,7 @@ func (s *ElasticSearchSender) SendBulkMsgs(ctx *bulkOpCtx, msgs []*libs.FluentMs
 		return errors.Wrap(err, "try to compress messages got error")
 	}
 
-	ctx.gzWriter.Flush()
+	ctx.gzWriter.Close()
 	req, err := http.NewRequest("POST", s.Addr, ctx.buf)
 	if err != nil {
 		return errors.Wrap(err, "try to init es request got error")
@@ -141,8 +141,8 @@ func (s *ElasticSearchSender) SendBulkMsgs(ctx *bulkOpCtx, msgs []*libs.FluentMs
 }
 
 type ESResp struct {
-	Errors bool        `json:"errors"`
-	Items  []*ESOpResp `json:"items"`
+	Errors bool `json:"errors"`
+	// Items  []*ESOpResp `json:"items"`
 }
 
 type ESOpResp struct {
@@ -161,15 +161,18 @@ func isStatusCodeOk(s int) bool {
 
 func (s *ElasticSearchSender) checkResp(resp *http.Response) (err error) {
 	if !isStatusCodeOk(resp.StatusCode) {
-		return fmt.Errorf("server return error code %v", resp.StatusCode)
+		err = fmt.Errorf("server return error code %v", resp.StatusCode)
 	}
 
 	ret := &ESResp{}
-	bb, err := ioutil.ReadAll(resp.Body)
+	bb, err2 := ioutil.ReadAll(resp.Body)
+	if err2 != nil {
+		utils.Logger.Error("try to read es resp body got error", zap.Error(err2))
+	}
 	if err != nil {
-		utils.Logger.Error("try to read es resp body got error", zap.Error(err))
-		return nil
+		return errors.Wrap(err, string(bb))
 	}
+	utils.Logger.Debug("got es response", zap.ByteString("resp", bb))
 
 	if err = json.Unmarshal(bb, ret); err != nil {
 		utils.Logger.Error("try to unmarshal body got error, body",
@@ -178,15 +181,8 @@ func (s *ElasticSearchSender) checkResp(resp *http.Response) (err error) {
 		return nil
 	}
 
-	for _, v := range ret.Items {
-		if !isStatusCodeOk(resp.StatusCode) {
-			// do not retry if there is part of msgs got error
-			utils.Logger.Warn("bulk got error for idx",
-				zap.ByteString("body", bb),
-				zap.String("idx", v.Index.Index),
-				zap.Int("status", v.Index.Status))
-			return nil
-		}
+	if ret.Errors {
+		utils.Logger.Error("es return error", zap.ByteString("resp", bb))
 	}
 
 	return nil