Merge pull request #182 from bookingcom/grzkv/fuzz_and_refactoring

Misc improvements

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/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,6 +188,9 @@ 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()
 

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