miscellaneous cleanup (#110)

* miscellaneous cleanup

* tmp

* fix typo that inverts error-checking logic

Author: iurii-ssv

internal/analyzer/cmd.go

@@ -1,6 +1,7 @@
 package analyzer
 
 import (
+	"fmt"
 	"log/slog"
 	"maps"
 	"math"
@@ -9,6 +10,7 @@ import (
 	"path/filepath"
 	"slices"
 	"sort"
+	"strings"
 	"sync"
 	"time"
 
@@ -50,9 +52,15 @@ var CMD = &cobra.Command{
 		}
 
 		fileDirectory := configs.Values.Analyzer.LogFilesDirectory
-		files, err := os.ReadDir(fileDirectory)
+		filesAll, err := os.ReadDir(fileDirectory)
 		if err != nil {
-			return err
+			return fmt.Errorf("could not read logs directory: %w", err)
+		}
+		files := make([]os.DirEntry, 0, len(filesAll))
+		for _, file := range filesAll {
+			if strings.HasSuffix(file.Name(), ".log") {
+				files = append(files, file)
+			}
 		}
 
 		var (
@@ -102,8 +110,8 @@ var CMD = &cobra.Command{
 			case <-progressTicker.C:
 				slog.
 					With("count", len(files)).
-					With("filesSizeMB", math.Round(totalFileSizeMB)).
-					Info("⏳⏳⏳ processing file(s)...")
+					With("totalSizeMB", math.Round(totalFileSizeMB)).
+					Info("⏳⏳⏳ still processing file(s)...")
 			case peerRecord, isOpen := <-peerRecordsChan:
 				if isOpen {
 					peersReport.AddRecord(peerRecord)
@@ -283,7 +291,8 @@ func analyzeFile(
 	peerRecordChan chan<- report.PeerRecord,
 	clientRecordChan chan<- report.ClientRecord,
 	operatorRecordChan chan<- map[uint32]report.OperatorRecord,
-	errorChan chan<- error) {
+	errorChan chan<- error,
+) {
 	clientAnalyzer, err := client.New(filePath, time.Millisecond*800)
 	if err != nil {
 		errorChan <- err
@@ -319,19 +328,16 @@ func analyzeFile(
 		return
 	}
 
-	analyzerSvc, err := New(
+	analyzerSvc := New(
 		peersAnalyzer,
 		consensusAnalyzer,
 		operatorAnalyzer,
 		clientAnalyzer,
 		commitAnalyzer,
 		prepareAnalyzer,
 		configs.Values.Analyzer.Operators,
-		configs.Values.Analyzer.Cluster)
-	if err != nil {
-		errorChan <- err
-		return
-	}
+		configs.Values.Analyzer.Cluster,
+	)
 
 	result, err := analyzerSvc.Start()
 	if err != nil {

internal/analyzer/parser/client/service.go

@@ -4,6 +4,7 @@ import (
 	"bufio"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"log/slog"
 	"os"
 	"strconv"
@@ -15,7 +16,8 @@ import (
 )
 
 const (
-	attestationMsg    = "starting QBFT instance"
+	attestationMsg = "starting QBFT instance"
+
 	nodesUnhealthyMsg = "ethereum node(s) are either out of sync or down. Ensure the nodes are healthy to resume."
 	nodeUnhealthyMsg  = "node is not healthy"
 
@@ -66,19 +68,22 @@ func (s *Service) Analyze() (Stats, error) {
 		stats = Stats{
 			ConsensusClientResponseTimeDelayPercent: map[time.Duration]float32{
 				s.delay: 0,
-			}}
+			},
+		}
 
 		attestationEndpointResponseTimes   []time.Duration
 		attestationEndpointResponseTimeSum time.Duration
 		attestationEndpointTotalDelayed    = make(map[time.Duration]uint32)
 	)
 
+	lineNumber := 0
 	for scanner.Scan() {
 		var entry clientLogEntry
 		line := scanner.Text()
+		lineNumber++
 		err := json.Unmarshal([]byte(line), &entry)
 		if err != nil {
-			return stats, err
+			return stats, fmt.Errorf("unmarshal log line %d (file = `%s`): `%s`, err: %w", lineNumber, s.logFile.Name(), line, err)
 		}
 
 		if strings.Contains(entry.Message, attestationMsg) {
@@ -114,7 +119,7 @@ func (s *Service) Analyze() (Stats, error) {
 		logger := slog.
 			With("parserName", parserName).
 			With("fileName", s.logFile.Name())
-		if err == bufio.ErrTooLong {
+		if errors.Is(err, bufio.ErrTooLong) {
 			logger.Warn("the log line was too long, continue reading..")
 		} else {
 			logger.

internal/analyzer/parser/commit/service.go

@@ -4,6 +4,7 @@ import (
 	"bufio"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"log/slog"
 	"os"
 	"strings"
@@ -52,11 +53,13 @@ func (s *Service) Analyze() (map[parser.SignerID]Stats, error) {
 	proposeTime := make(map[parser.DutyID]time.Time)
 	commitTimes := make(map[parser.DutyID]map[parser.SignerID]time.Time)
 
+	lineNumber := 0
 	for scanner.Scan() {
 		var entry commitLogEntry
 		line := scanner.Text()
+		lineNumber++
 		if err := json.Unmarshal([]byte(line), &entry); err != nil {
-			return nil, err
+			return nil, fmt.Errorf("unmarshal log line %d (file = `%s`): `%s`, err: %w", lineNumber, s.logFile.Name(), line, err)
 		}
 
 		if strings.Contains(entry.Message, proposeMsg) {
@@ -79,7 +82,7 @@ func (s *Service) Analyze() (map[parser.SignerID]Stats, error) {
 		logger := slog.
 			With("parserName", parserName).
 			With("fileName", s.logFile.Name())
-		if err == bufio.ErrTooLong {
+		if errors.Is(err, bufio.ErrTooLong) {
 			logger.Warn("the log line was too long, continue reading..")
 		} else {
 			logger.

internal/analyzer/parser/consensus/service.go

@@ -4,6 +4,7 @@ import (
 	"bufio"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"log/slog"
 	"os"
 	"slices"
@@ -74,11 +75,13 @@ func (s *Service) Analyze() (Stats, error) {
 		blockRootSlots = make(map[parser.BlockRoot][]parser.Slot)
 	)
 
+	lineNumber := 0
 	for scanner.Scan() {
 		var entry logEntry
 		line := scanner.Text()
+		lineNumber++
 		if err := json.Unmarshal([]byte(line), &entry); err != nil {
-			return stats, err
+			return stats, fmt.Errorf("unmarshal log line %d (file = `%s`): `%s`, err: %w", lineNumber, s.logFile.Name(), line, err)
 		}
 
 		if strings.Contains(entry.Message, partialSignatureLogRecord) {
@@ -134,7 +137,7 @@ func (s *Service) Analyze() (Stats, error) {
 		logger := slog.
 			With("parserName", parserName).
 			With("fileName", s.logFile.Name())
-		if err == bufio.ErrTooLong {
+		if errors.Is(err, bufio.ErrTooLong) {
 			logger.Warn("the log line was too long, continue reading..")
 		} else {
 			logger.

internal/analyzer/parser/operator/service.go

@@ -53,11 +53,13 @@ func (s *Service) Analyze() (Stats, error) {
 		clusters [][]parser.SignerID
 	)
 
+	lineNumber := 0
 	for scanner.Scan() {
 		line := scanner.Text()
+		lineNumber++
 		var entry logEntry
 		if err := json.Unmarshal([]byte(line), &entry); err != nil {
-			return stats, err
+			return stats, fmt.Errorf("unmarshal log line %d (file = `%s`): `%s`, err: %w", lineNumber, s.logFile.Name(), line, err)
 		}
 
 		if strings.Contains(entry.Message, proposalMsg) {
@@ -100,7 +102,7 @@ func (s *Service) Analyze() (Stats, error) {
 		logger := slog.
 			With("parserName", parserName).
 			With("fileName", s.logFile.Name())
-		if err == bufio.ErrTooLong {
+		if errors.Is(err, bufio.ErrTooLong) {
 			logger.Warn("the log line was too long, continue reading..")
 		} else {
 			logger.

internal/analyzer/parser/peers/service.go

@@ -4,6 +4,7 @@ import (
 	"bufio"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"log/slog"
 	"os"
 	"regexp"
@@ -75,17 +76,14 @@ func (s *Service) Analyze() (Stats, error) {
 		nodeClientVersions    []string
 	)
 
+	lineNumber := 0
 	for scanner.Scan() {
 		var entry logEntry
 		line := scanner.Text()
-
+		lineNumber++
 		if strings.Contains(line, scoredPeersMsg) || strings.Contains(line, peerIdentityMsg) || strings.Contains(line, peerScoresMsg) {
 			if err := json.Unmarshal([]byte(line), &entry); err != nil {
-				slog.
-					With("err", err).
-					With("line", line).
-					Warn("failed to unmarshal line. Skipping")
-				continue
+				return stats, fmt.Errorf("unmarshal log line %d (file = `%s`): `%s`, err: %w", lineNumber, s.logFile.Name(), line, err)
 			}
 
 			if strings.EqualFold(entry.Message, scoredPeersMsg) {
@@ -113,7 +111,7 @@ func (s *Service) Analyze() (Stats, error) {
 		logger := slog.
 			With("parserName", parserName).
 			With("fileName", s.logFile.Name())
-		if err == bufio.ErrTooLong {
+		if errors.Is(err, bufio.ErrTooLong) {
 			logger.Warn("the log line was too long, continue reading..")
 		} else {
 			logger.

internal/analyzer/parser/prepare/service.go

@@ -4,6 +4,7 @@ import (
 	"bufio"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"log/slog"
 	"os"
 	"strings"
@@ -44,19 +45,21 @@ func New(logFilePath string, delay time.Duration) (*Service, error) {
 	}, nil
 }
 
-func (p *Service) Analyze() (map[parser.SignerID]Stats, error) {
-	defer p.logFile.Close()
+func (s *Service) Analyze() (map[parser.SignerID]Stats, error) {
+	defer s.logFile.Close()
 
-	scanner := parser.NewScanner(p.logFile)
+	scanner := parser.NewScanner(s.logFile)
 
 	leaderProposeTime := make(map[parser.DutyID]time.Time)
 	prepareSignerTimes := make(map[parser.DutyID]map[parser.SignerID]time.Time)
 
+	lineNumber := 0
 	for scanner.Scan() {
 		var entry prepareLogEntry
 		line := scanner.Text()
+		lineNumber++
 		if err := json.Unmarshal([]byte(line), &entry); err != nil {
-			return nil, err
+			return nil, fmt.Errorf("unmarshal log line %d (file = `%s`): `%s`, err: %w", lineNumber, s.logFile.Name(), line, err)
 		}
 
 		if strings.Contains(entry.Message, leaderProposeMsg) {
@@ -77,8 +80,8 @@ func (p *Service) Analyze() (map[parser.SignerID]Stats, error) {
 	if err := scanner.Err(); err != nil {
 		logger := slog.
 			With("parserName", parserName).
-			With("fileName", p.logFile.Name())
-		if err == bufio.ErrTooLong {
+			With("fileName", s.logFile.Name())
+		if errors.Is(err, bufio.ErrTooLong) {
 			logger.Warn("the log line was too long, continue reading..")
 		} else {
 			logger.
@@ -89,7 +92,7 @@ func (p *Service) Analyze() (map[parser.SignerID]Stats, error) {
 		}
 	}
 
-	stats := p.calcPrepareTimes(leaderProposeTime, prepareSignerTimes)
+	stats := s.calcPrepareTimes(leaderProposeTime, prepareSignerTimes)
 
 	return stats, nil
 }

internal/analyzer/report/client.go

@@ -21,7 +21,7 @@ var (
 	}
 
 	ssvClientHeaders = []string{
-		"Crashes(reason: unhealthy EL/CL): \n total(EL/CL)",
+		"SSV node terminates =\nEL-issue + CL-issue",
 	}
 )
 
@@ -47,7 +47,7 @@ func NewClient() *ClientReport {
 	t := table.New(os.Stdout)
 
 	headers := slices.Concat(consensusClientHeaders, ssvClientHeaders)
-	t.SetHeaders("Consensus Client Performance", "SSV Client Performance")
+	t.SetHeaders("Consensus Client Performance", "EL / CL issues")
 	t.AddHeaders(headers...)
 	t.SetAutoMerge(true)
 	t.SetHeaderColSpans(0, len(consensusClientHeaders), len(ssvClientHeaders))
@@ -125,7 +125,7 @@ func (c *ClientReport) Render() {
 			fmt.Sprint(record.consensusClientResponseTimeP50Total/time.Duration(record.consensusClientResponseTimeRecordCount)),
 			fmt.Sprint(record.consensusClientResponseTimeP90Total/time.Duration(record.consensusClientResponseTimeRecordCount)),
 			strings.Join(delayedResponses, "\n"),
-			fmt.Sprintf("%d(%d/%d)", record.ssvClientCrashesTotal, record.SSVClientELUnhealthyTotal, record.SSVClientCLUnhealthyTotal),
+			fmt.Sprintf("%d = %d + %d", record.ssvClientCrashesTotal, record.SSVClientELUnhealthyTotal, record.SSVClientCLUnhealthyTotal),
 		)
 	}