Merge branch 'master' into grzkv/improving_testing_and_bench_system

Author: grzkv

.github/workflows/ci.yml

@@ -13,7 +13,7 @@ jobs:
         with:
           go-version: ${{ matrix.go-version }}
       - name: setup linter
-        run: curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sudo sh -s -- -b /usr/local/bin v1.45.0
+        run: curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sudo sh -s -- -b /usr/local/bin v1.46.2
       - name: checkout code
         uses: actions/checkout@v2
       - name: lint

cmd/nanotube/process.go

@@ -73,7 +73,7 @@ func worker(wg *sync.WaitGroup, queue <-chan []byte, rules rules.Rules, rewrites
 
 func proc(s []byte, rules rules.Rules, rewrites rewrites.Rewrites, shouldNormalize bool, shouldLog bool, lg *zap.Logger,
 	metrics *metrics.Prom) { //nolint:golint,unparam
-	r, err := rec.ParseRecBytes(s, shouldNormalize, shouldLog, time.Now, lg)
+	r, err := rec.ParseRec(s, shouldNormalize, shouldLog, time.Now, lg)
 	if err != nil {
 		lg.Info("Error parsing incoming record", zap.String("record_str", string(s)),
 			zap.Binary("record_base64", s), zap.Error(err))
@@ -89,7 +89,7 @@ func proc(s []byte, rules rules.Rules, rewrites rewrites.Rewrites, shouldNormali
 	}
 
 	for _, rec := range recs {
-		rules.RouteRecBytes(rec, lg)
+		rules.RouteRec(rec, lg)
 	}
 }
 

pkg/conf/conf.go

@@ -215,7 +215,7 @@ func MakeDefault() Main {
 		MainQueueSize:        1000,
 		HostQueueSize:        1000,
 		MainQueueBatchSize:   1000,
-		BatchFlushPerdiodSec: 1,
+		BatchFlushPerdiodSec: 5,
 
 		WorkerPoolSize: 0,
 

pkg/in/batch.go

@@ -17,27 +17,36 @@ type BatchChan struct {
 	ms      *metrics.Prom
 	m       sync.Mutex
 	period  time.Duration
+	stop    chan bool
 }
 
 // NewBatchChan makes a new batched chan buffer.
-// It also starts a flushing goroutine in the background.
+// It also starts a flushing goroutine in the background if periodSec > 0.
 func NewBatchChan(q chan<- [][]byte, bufSize int, periodSec int, ms *metrics.Prom) *BatchChan {
 	qb := &BatchChan{
 		q:       q,
 		bufSize: bufSize,
 		ms:      ms,
 		period:  time.Second * time.Duration(periodSec),
+		stop:    make(chan bool),
 	}
 
-	go qb.periodicFlush()
+	if periodSec > 0 {
+		go qb.periodicFlush()
+	}
 
 	return qb
 }
 
 func (qb *BatchChan) periodicFlush() {
 	for {
-		time.Sleep(qb.period)
-		qb.Flush()
+		select {
+		case <-qb.stop:
+			return
+		default:
+			time.Sleep(qb.period)
+			qb.Flush()
+		}
 	}
 }
 
@@ -65,6 +74,12 @@ func (qb *BatchChan) Flush() {
 	qb.buf = [][]byte{}
 }
 
+// Close closes the batch channel. It does not close the target channel.
+// Must be called exactly once for every new instance.
+func (qb *BatchChan) Close() {
+	close(qb.stop)
+}
+
 func (qb *BatchChan) sendToMainQBuf() {
 	select {
 	case qb.q <- qb.buf:

pkg/in/tcp.go

@@ -57,6 +57,7 @@ loop:
 }
 
 // AcceptAndListenTCP listens for incoming TCP connections.
+// TODO: Remove after changing NK mechanics.
 func AcceptAndListenTCP(l net.Listener, queue chan<- []byte, term <-chan struct{},
 	cfg *conf.Main, connWG *sync.WaitGroup, ms *metrics.Prom, lg *zap.Logger) {
 	var wg sync.WaitGroup
@@ -101,6 +102,7 @@ loop:
 	connWG.Done()
 }
 
+// TODO: Remove after changing NK mechanics.
 func readFromConnectionTCP(wg *sync.WaitGroup, conn net.Conn, queue chan<- []byte, stop <-chan struct{}, cfg *conf.Main, ms *metrics.Prom, lg *zap.Logger) {
 	defer wg.Done() // executed after the connection is closed
 	defer func() {
@@ -145,6 +147,7 @@ func readFromConnectionTCPBuf(wg *sync.WaitGroup, conn net.Conn, queue chan<- []
 	scanForRecordsTCPBuf(conn, queue, cfg, ms, lg)
 }
 
+// TODO: Remove after changing NK mechanics.
 func scanForRecordsTCP(conn net.Conn, queue chan<- []byte, stop <-chan struct{}, cfg *conf.Main, ms *metrics.Prom, lg *zap.Logger) {
 	sc := bufio.NewScanner(conn)
 	in := make(chan []byte)
@@ -185,7 +188,11 @@ loop:
 func scanForRecordsTCPBuf(conn net.Conn, queue chan<- [][]byte, cfg *conf.Main, ms *metrics.Prom, lg *zap.Logger) {
 	sc := bufio.NewScanner(conn)
 
+	buf := make([]byte, 2048)
+	sc.Buffer(buf, bufio.MaxScanTokenSize)
+
 	qb := NewBatchChan(queue, int(cfg.MainQueueBatchSize), int(cfg.BatchFlushPerdiodSec), ms)
+	defer qb.Close()
 
 	for sc.Scan() {
 		rec := []byte{}
@@ -201,8 +208,6 @@ func scanForRecordsTCPBuf(conn net.Conn, queue chan<- [][]byte, cfg *conf.Main,
 		qb.Push(rec)
 	}
 
-	lg.Debug("finished scanning for records", zap.Stringer("addr", conn.RemoteAddr()))
-
 	qb.Flush()
 }
 

pkg/in/udp.go

@@ -66,6 +66,8 @@ func ListenUDPBuf(conn net.PacketConn, queue chan [][]byte, stop <-chan struct{}
 
 	buf := make([]byte, 64*1024) // 64k is the max UDP datagram size
 	qb := NewBatchChan(queue, int(cfg.MainQueueBatchSize), int(cfg.BatchFlushPerdiodSec), ms)
+	defer qb.Close()
+
 loop:
 	for {
 		select {

pkg/in/udp_test.go

@@ -1,20 +1,14 @@
 package in
 
 import (
-	"bufio"
 	"bytes"
 	"fmt"
-	"log"
 	"net"
-	"os"
-	"path/filepath"
 	"sync"
 	"testing"
 	"time"
 
-	"github.com/bookingcom/nanotube/pkg/conf"
-	"github.com/bookingcom/nanotube/pkg/metrics"
-	"go.uber.org/zap"
+	"github.com/bookingcom/nanotube/pkg/test"
 )
 
 type packetConnMock struct {
@@ -84,43 +78,12 @@ func (c *packetConnMock) SetWriteDeadline(t time.Time) error {
 	return nil
 }
 
-func setup(t *testing.T) (data [][]byte, ms *metrics.Prom, lg *zap.Logger) {
-	fixturesPath := "testdata/"
-
-	in, err := os.Open(filepath.Join(fixturesPath, "in"))
+func TestUdpStreaming(t *testing.T) {
+	data, ms, lg, err := test.Setup()
 	if err != nil {
-		t.Fatalf("error opening the in data file %v", err)
-	}
-	defer func() {
-		err := in.Close()
-		if err != nil {
-			t.Fatalf("error closing in data test file: %v", err)
-		}
-	}()
-
-	scanner := bufio.NewScanner(in)
-	for scanner.Scan() {
-		token := scanner.Bytes()
-		rec := make([]byte, len(token))
-		copy(rec, token)
-		data = append(data, rec)
-	}
-
-	if err := scanner.Err(); err != nil {
-		log.Fatalf("error while scan-reading the sample in metrics %v", err)
+		t.Fatalf("failed to setup the test: %v", err)
 	}
 
-	lg, _ = zap.NewProduction()
-
-	cfg := conf.MakeDefault()
-	ms = metrics.New(&cfg)
-
-	return
-}
-
-func TestUdpStreaming(t *testing.T) {
-	data, ms, lg := setup(t)
-
 	stop := make(chan struct{})
 	conn := &packetConnMock{
 		traffic: data,
@@ -147,7 +110,7 @@ func TestUdpStreaming(t *testing.T) {
 	wg.Wait()
 	close(q)
 
-	err := <-errCh
+	err = <-errCh
 	if err != nil {
 		t.Fatal(err)
 	}

pkg/k8s/container.go

@@ -21,8 +21,6 @@ import (
 	"go.uber.org/zap"
 )
 
-// TODO: Move debug logging to appropriate zap.Debug level.
-
 // Cont represents a container.
 type Cont struct {
 	ID           string

pkg/rec/rec_bytes.go

@@ -10,24 +10,24 @@ import (
 
 // RecBytes represents a single piece of data (a datapoint) that can be sent.
 type RecBytes struct { // nolint:revive
-	Path    []byte
-	RawVal  []byte // this is to avoid discrepancies in precision and formatting
-	RawTime []byte // to avoid differences when encoding, and save time
-	//	Raw  string // to avoid wasting time for serialization
+	Path []byte
+	Val  []byte
+	Time []byte
+
 	Received time.Time
 }
 
-// ParseRecBytes parses a single datapoint record from a string. Makes sure it's valid.
+// ParseRec parses a single datapoint record from a string. Makes sure it's valid.
 // Performs normalizations.
-func ParseRecBytes(s []byte, normalize bool, shouldLog bool, nowF func() time.Time, lg *zap.Logger) (*RecBytes, error) {
+func ParseRec(s []byte, normalize bool, shouldLog bool, nowF func() time.Time, lg *zap.Logger) (*RecBytes, error) {
 	pathStart, pathEnd, valStart, valEnd, timeStart, timeEnd, err := recFields(s)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to break record into fields")
 	}
 
 	var path []byte
 	if normalize {
-		path, err = normalizePathBytes(s[pathStart:pathEnd])
+		path, err = normalizePath(s[pathStart:pathEnd])
 		if err != nil {
 			return nil, errors.Wrap(err, "failed to normalize path")
 		}
@@ -39,8 +39,8 @@ func ParseRecBytes(s []byte, normalize bool, shouldLog bool, nowF func() time.Ti
 		Path:     path,
 		Received: nowF(),
 	}
-	res.RawVal = append(res.RawVal, s[valStart:valEnd]...)
-	res.RawTime = append(res.RawTime, s[timeStart:timeEnd]...)
+	res.Val = append(res.Val, s[valStart:valEnd]...)
+	res.Time = append(res.Time, s[timeStart:timeEnd]...)
 
 	return &res, nil
 }
@@ -100,20 +100,19 @@ func isWhitespace(c byte) bool {
 // Serialize makes record into a string ready to be sent via TCP w/ line protocol.
 func (r *RecBytes) Serialize() []byte {
 	// TODO (grzkv): Copy can be avoided if string was not changed
-	res := make([]byte, 0, len(r.Path)+len(r.RawTime)+len(r.RawVal)+3)
+	res := make([]byte, 0, len(r.Path)+len(r.Time)+len(r.Val)+3)
 	res = append(res, r.Path...)
 	res = append(res, ' ')
-	res = append(res, r.RawVal...)
+	res = append(res, r.Val...)
 	res = append(res, ' ')
-	res = append(res, r.RawTime...)
+	res = append(res, r.Time...)
 	res = append(res, '\n')
 
 	return res
 }
 
 // normalizePath does path normalization as described in the docs
-// returns: (updated path, was any normalization done)
-func normalizePathBytes(s []byte) ([]byte, error) {
+func normalizePath(s []byte) ([]byte, error) {
 	if len(s) == 0 {
 		return []byte{}, nil
 	}
@@ -181,21 +180,21 @@ func (r RecBytes) Copy() (*RecBytes, error) {
 	cpy := &RecBytes{
 		Received: r.Received,
 		Path:     make([]byte, len(r.Path)),
-		RawVal:   make([]byte, len(r.RawVal)),
-		RawTime:  make([]byte, len(r.RawTime)),
+		Val:      make([]byte, len(r.Val)),
+		Time:     make([]byte, len(r.Time)),
 	}
 
 	n := copy(cpy.Path, r.Path)
 	if n != len(r.Path) {
 		return nil, errors.Errorf("did not copy full path, expected %d bytes, copied %d bytes", len(r.Path), n)
 	}
-	n = copy(cpy.RawVal, r.RawVal)
-	if n != len(r.RawVal) {
-		return nil, errors.Errorf("did not copy full value, expected %d bytes, copied %d bytes", len(r.RawVal), n)
+	n = copy(cpy.Val, r.Val)
+	if n != len(r.Val) {
+		return nil, errors.Errorf("did not copy full value, expected %d bytes, copied %d bytes", len(r.Val), n)
 	}
-	n = copy(cpy.RawTime, r.RawTime)
-	if n != len(r.RawTime) {
-		return nil, errors.Errorf("did not copy full time, expected %d bytes, copied %d bytes", len(r.RawTime), n)
+	n = copy(cpy.Time, r.Time)
+	if n != len(r.Time) {
+		return nil, errors.Errorf("did not copy full time, expected %d bytes, copied %d bytes", len(r.Time), n)
 	}
 
 	return cpy, nil