GH-85: fix buffering in json aggregator

core/aggregator/encoder.go

@@ -127,6 +127,7 @@ HandleLoop:
 			}
 		case _ = <-flushTick:
 			if previousFlushes == flushes {
+				a.Log.Debug("Flushing")
 				err = encoder.Flush()
 				if err != nil {
 					return

core/aggregator/jsonlines.go

@@ -6,6 +6,7 @@
 package aggregator
 
 import (
+	"bufio"
 	"io"
 
 	"github.com/json-iterator/go"
@@ -54,16 +55,26 @@ func NewJSONEncoder(w io.Writer, conf JSONLineEncoderConfig) SampleEncoder {
 	var apiConfig jsoniter.Config
 	config.Map(&apiConfig, conf.JSONIterConfig)
 	api := apiConfig.Froze()
-	stream := jsoniter.NewStream(api, w, conf.BufferSizeOrDefault())
-	return jsonEncoder{stream}
+	// NOTE(skipor): internal buffering is not working really. Don't know why
+	// OPTIMIZE(skipor): don't wrap into buffer, if already ioutil2.ByteWriter
+	buf := bufio.NewWriterSize(w, conf.BufferSizeOrDefault())
+	stream := jsoniter.NewStream(api, buf, conf.BufferSizeOrDefault())
+	return &jsonEncoder{stream, buf}
 }
 
 type jsonEncoder struct {
 	*jsoniter.Stream
+	buf *bufio.Writer
 }
 
-func (e jsonEncoder) Encode(s core.Sample) error {
+func (e *jsonEncoder) Encode(s core.Sample) error {
 	e.WriteVal(s)
 	e.WriteRaw("\n")
 	return e.Error
 }
+
+func (e *jsonEncoder) Flush() error {
+	err := e.Stream.Flush()
+	e.buf.Flush()
+	return err
+}

examples/custom_pandora/custom.yaml

@@ -13,7 +13,7 @@ pools:
 
   result:
     type: json
-    sink: stderr # Just for interactivity print result to stderr. Usually file user.
+    sink: stdout # Just for interactivity print result to stdout. Usually file used here.
 
   rps:
     - {duration: 2s, type: line,  from: 1, to: 5}