Merge pull request #8 from red-bashmak/feature/gatling-binary-log

Feature/gatling binary log

Author: DanielPerfana

gatlingparser/decoders.go

@@ -3,9 +3,13 @@ package gatlingparser
 import (
 	"bufio"
 	"encoding/binary"
+	"encoding/hex"
 	"fmt"
+	"io"
 	"math"
 	"strings"
+
+	l "github.com/perfana/x2i/logger"
 )
 
 const (
@@ -366,7 +370,15 @@ func ReadErrorRecord(reader *bufio.Reader, runStartTimestamp int64) (ErrorRecord
 }
 
 func ReadNotHeaderRecord(reader *bufio.Reader, runStartTimestapm int64, scenarios []string) (interface{}, error) {
-	recordType, err := reader.ReadByte()
+	headBytes, err := reader.Peek(8)
+	if err == io.EOF {
+		return nil, err
+	}
+	if err != nil {
+		return nil, fmt.Errorf("could not read bytes from log file - %v", err)
+	}
+	var recordType byte
+	recordType, err = reader.ReadByte()
 	if err != nil {
 		return nil, err
 	}
@@ -381,6 +393,7 @@ func ReadNotHeaderRecord(reader *bufio.Reader, runStartTimestapm int64, scenario
 	case ErrorRecordType:
 		return ReadErrorRecord(reader, runStartTimestapm)
 	default:
+		l.Debugf("Unknown record start fragment: %s\n", hex.EncodeToString(headBytes))
 		return nil, fmt.Errorf("unknown record type: %d", recordType)
 	}
 }

gatlingparser/gatlingparser.go

@@ -44,6 +44,7 @@ import (
 	"github.com/perfana/x2i/influx"
 	l "github.com/perfana/x2i/logger"
 	"github.com/spf13/cobra"
+	"golang.org/x/mod/semver"
 )
 
 const (
@@ -79,8 +80,6 @@ var (
 	groupLine   = regexp.MustCompile(`GROUP\s`)
 	runLine     = regexp.MustCompile(`^RUN\s`)
 	errorLine   = regexp.MustCompile(`^ERROR\s`)
-	// for checking gatling version
-	gatlingVersion313x = regexp.MustCompile(`3\.13\..*`)
 
 	parserStopped = make(chan struct{})
 )
@@ -596,8 +595,12 @@ func processRemainingRecords(
 	scenarios []string,
 	records chan<- interface{},
 ) {
-	defer close(records)
-	defer wg.Done()
+	defer func() {
+		close(records)
+		wg.Done()
+	}()
+
+	latestReadTime := time.Now()
 
 	for {
 		select {
@@ -608,19 +611,31 @@ func processRemainingRecords(
 			record, err := ReadNotHeaderRecord(reader, runMessage.Start, scenarios)
 			if err != nil {
 				if err == io.EOF {
-					time.Sleep(time.Duration(waitTime) * time.Second) // Wait if end of file
+					// If no new data read for more than value provided by 'stop-timeout' key then processing is stopped
+					if time.Now().After(latestReadTime.Add(time.Duration(waitTime) * time.Second)) {
+						l.Infof("No new entries found for %d seconds. Stopping application...", waitTime)
+						return
+					}
+					time.Sleep(time.Second) // Wait if end of file
 					continue
 				}
 				l.Errorf("Reading error: %v", err)
 				continue
 			}
 
 			records <- record
+			latestReadTime = time.Now()
 		}
 	}
 }
 
-func writeRecords(wg *sync.WaitGroup, records <-chan interface{}) {
+type RecordsWriter interface {
+	writeAll(wg *sync.WaitGroup, records <-chan interface{})
+}
+
+type InfluxRecordsWriter struct{}
+
+func (w *InfluxRecordsWriter) writeAll(wg *sync.WaitGroup, records <-chan interface{}) {
 	defer wg.Done()
 	for record := range records {
 		switch r := record.(type) {
@@ -665,7 +680,45 @@ func writeRecords(wg *sync.WaitGroup, records <-chan interface{}) {
 	}
 }
 
-func fileProcessorBinary(ctx context.Context, file *os.File) {
+type SumRecordsWriter struct{}
+
+func (w *SumRecordsWriter) writeAll(wg *sync.WaitGroup, records <-chan interface{}) {
+	defer wg.Done()
+	var (
+		users       = 0
+		reqs        = 0
+		rumMessages = 0
+		errors      = 0
+		groups      = 0
+	)
+
+	for record := range records {
+		// l.Infoln(record)
+		switch r := record.(type) {
+		case RunMessage:
+			rumMessages += 1
+
+		case RequestRecord:
+			reqs += 1
+
+		case GroupRecord:
+			groups += 1
+
+		case UserRecord:
+			users += 1
+
+		case ErrorRecord:
+			errors += 1
+
+		default:
+			l.Errorf("Unknown record type: %T", r)
+		}
+
+	}
+	l.Debugf("msg = %d, users = %d, reqs = %d, groups = %d, errors = %d\n", rumMessages, users, reqs, groups, errors)
+}
+
+func fileProcessorBinary(ctx context.Context, file *os.File, recordsWriter  RecordsWriter) {
 	defer func() { parserStopped <- struct{}{} }()
 	reader := bufio.NewReader(file)
 	runMessage, scenarios, err := processLogHeader(reader)
@@ -675,14 +728,13 @@ func fileProcessorBinary(ctx context.Context, file *os.File) {
 	}
 
 	wg := &sync.WaitGroup{}
-	records := make(chan interface{}, 10)
+	records := make(chan interface{}, 100)
 	records <- *runMessage
 
 	wg.Add(2)
 	go processRemainingRecords(ctx, wg, reader, *runMessage, scenarios, records)
-	go writeRecords(wg, records)
+	go recordsWriter.writeAll(wg, records)
 	wg.Wait()
-
 }
 
 func parseStart(ctx context.Context, wg *sync.WaitGroup) {
@@ -703,8 +755,12 @@ func parseStart(ctx context.Context, wg *sync.WaitGroup) {
 	if err != nil {
 		l.Errorf("Failed to read %s file: %v\n", simulationLogFileName, err)
 	}
-	if gatlingVersion313x.MatchString(ver) {
-		fileProcessorBinary(ctx, file)
+	if !semver.IsValid(ver) {
+		ver = "v" + ver
+	}
+	if semver.Compare(ver, "v3.12.1") >= 0 {
+		writer := InfluxRecordsWriter{}
+		fileProcessorBinary(ctx, file, &writer)
 	} else {
 		fileProcessor(ctx, file)
 	}

gatlingparser/gatlingparser_test.go

@@ -0,0 +1,26 @@
+package gatlingparser
+
+import (
+	"context"
+	"os"
+	"testing"
+	l "github.com/perfana/x2i/logger"
+)
+
+func TestFileProcessorBinary(t *testing.T) {
+	l.InitLogger("testlog.log")
+	waitTime = 10
+	ctx := context.Background()
+	
+	file, err := os.Open("../test/data/simulation.log")
+	if err != nil {
+		t.Errorf("os.Open(%v) returned error: %v", ctx, err)
+	}
+	go func ()  {
+		 <- parserStopped
+	}()
+
+	writer := SumRecordsWriter{}
+	
+	fileProcessorBinary(ctx, file, &writer)
+}

go.mod

@@ -10,4 +10,5 @@ require (
 require (
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
+	golang.org/x/mod v0.23.0
 )

go.sum

@@ -8,5 +8,7 @@ github.com/spf13/cobra v1.8.1 h1:e5/vxKd/rZsfSJMUX1agtjeTDf+qv1/JdBF8gg5k9ZM=
 github.com/spf13/cobra v1.8.1/go.mod h1:wHxEcudfqmLYa8iTfL+OuZPbBZkmvliBWKIezN3kD9Y=
 github.com/spf13/pflag v1.0.5 h1:iy+VFUOCP1a+8yFto/drg2CJ5u0yRoB7fZw3DKv/JXA=
 github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
+golang.org/x/mod v0.23.0 h1:Zb7khfcRGKk+kqfxFaP5tZqCnDZMjC5VtUBs87Hr6QM=
+golang.org/x/mod v0.23.0/go.mod h1:6SkKJ3Xj0I0BrPOZoBy3bdMptDDU9oJrpohJ3eWZ1fY=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=