Merge pull request #7 from chainstack/starter_errortrack

starter with errortrack flag

Author: xdfdm

Dockerfile

@@ -81,6 +81,7 @@ COPY --from=builder /app/build/bin/state /usr/local/bin/state
 COPY --from=builder /app/build/bin/txpool /usr/local/bin/txpool
 COPY --from=builder /app/build/bin/verkle /usr/local/bin/verkle
 COPY --from=builder /app/build/bin/caplin-phase1 /usr/local/bin/caplin-phase1
+COPY --from=builder /app/build/bin/starter /usr/local/bin/starter
 
 
 
@@ -109,4 +110,3 @@ LABEL org.label-schema.build-date=$BUILD_DATE \
       org.label-schema.vendor="Torquem" \
       org.label-schema.version=$VERSION
 
-ENTRYPOINT ["erigon"]

Makefile

@@ -105,6 +105,7 @@ geth: erigon
 erigon: go-version erigon.cmd
 	@rm -f $(GOBIN)/tg # Remove old binary to prevent confusion where users still use it because of the scripts
 
+COMMANDS += starter
 COMMANDS += devnet
 COMMANDS += erigon-el-mock
 COMMANDS += downloader

cmd/starter/README.md

@@ -0,0 +1,32 @@
+# Starter
+
+Starter is an additional utility for tracking errors in the logs of the main process. If the given substring is found in the logs more than N times in the given period of time, then the starter will restart the process.
+
+If starter receives a signal to stop, it will stop the main process and exit.
+
+## Build
+
+```go build -o /build/bin/starter cmd/starter/main.go```
+
+or
+
+```make starter```
+
+## Usage
+Each `--errortrack` flag requires 3 arguments after it, the rest of arguments is a command for child process.
+
+```starter [--errortrack error_substring error_limit time_window] <command> [args...]```
+
+- `error_substring` - substring to search in the logs
+- `error_limit` - maximum number of occurrences of the substring in the given period
+- `time_window` - time window for error searching (e.g. 60s, 1m, 1h, 1d)
+- `command` - command to run
+- `[args...]` - arguments for the command
+## Example
+This command will restart erigon if the substring 'No block bodies' is found in the logs more than 50 times in 5 minutes:
+
+```starter --errortrack 'No block bodies' 50 5m erigon --datadir /data --private.api.addr```
+
+It is possible to track multiple errors by using `--errortrack` flag multiple times.
+
+```starter --errortrack 'No block bodies' 50 5m --errortrack 'No peers' 10 60s erigon --datadir /data --private.api.addr```

cmd/starter/main.go

@@ -0,0 +1,210 @@
+package main
+
+import (
+	"bufio"
+	"fmt"
+	"io"
+	"os"
+	"os/exec"
+	"os/signal"
+	"sort"
+	"strconv"
+	"strings"
+	"sync"
+	"syscall"
+	"time"
+)
+
+type ErrorTracker struct {
+	Substring string
+	Limit     int
+	Window    time.Duration
+	Errors    []time.Time
+}
+
+func (et *ErrorTracker) AddError(t time.Time) {
+	et.Errors = append(et.Errors, t)
+	if len(et.Errors) > et.Limit {
+		et.Errors = et.Errors[1:]
+	}
+}
+
+func (et *ErrorTracker) ShouldRestartProcess() bool {
+	if len(et.Errors) < et.Limit {
+		return false
+	}
+
+	firstError := et.Errors[0]
+	lastError := et.Errors[len(et.Errors)-1]
+	return lastError.Sub(firstError) <= et.Window
+}
+
+func parseErrorTrackers() ([]ErrorTracker, []string) {
+	trackers := make([]ErrorTracker, 0)
+	args := os.Args[1:]
+	maxArg := -1
+
+	for i, arg := range args {
+		if arg == "--errortrack" {
+			if i+3 >= len(args) {
+				log("Invalid --errortrack parameters")
+				os.Exit(1)
+			}
+
+			substring := args[i+1]
+			limit, err := strconv.Atoi(args[i+2])
+			if err != nil {
+				log("Error parsing limit: %v", err)
+				os.Exit(1)
+			}
+
+			window, err := time.ParseDuration(args[i+3])
+			if err != nil {
+				log("Error parsing time window: %v", err)
+				os.Exit(1)
+			}
+
+			trackers = append(trackers, ErrorTracker{Substring: substring, Limit: limit, Window: window})
+			if i+3 > maxArg {
+				maxArg = i + 3
+			}
+		}
+	}
+
+	if maxArg >= len(args)-1 {
+		log("No command provided")
+		os.Exit(1)
+	}
+
+	cmdArgs := args[maxArg+1:]
+	return trackers, cmdArgs
+}
+
+func log(format string, a ...interface{}) {
+	fmt.Printf(fmt.Sprintf("[starter] %s ", time.Now().Format("[01-02|15:04:05.000]"))+format+"\n", a...)
+}
+
+func main() {
+	trackers, cmdArgs := parseErrorTrackers()
+	if len(cmdArgs) == 0 {
+		log("No command provided")
+		os.Exit(1)
+	}
+
+	cmdName := cmdArgs[0]
+	cmdArgs = cmdArgs[1:]
+
+	// Set up signal handling
+	sigChan := make(chan os.Signal, 1)
+	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
+	restartChan := make(chan struct{}, 2)
+	exitChan := make(chan struct{}, 2)
+	var cmd *exec.Cmd
+	var stdoutPipe, stderrPipe io.ReadCloser
+	var wg sync.WaitGroup
+	restartInit := false
+
+	scanAndTrackErrors := func(scanner *bufio.Scanner, trackers []ErrorTracker, lastwriter io.Writer, restartChan, exitChan chan struct{}) {
+
+		for scanner.Scan() {
+
+			line := scanner.Text()
+
+			fmt.Fprintln(lastwriter, line)
+
+			if restartInit {
+				continue
+			}
+
+			for i := range trackers {
+				if strings.Contains(line, trackers[i].Substring) {
+					trackers[i].AddError(time.Now())
+
+					if len(trackers[i].Errors) == 1 {
+						log("Find error substring %q", trackers[i].Substring)
+					} else if len(trackers[i].Errors) < trackers[i].Limit {
+						log("Find error substring %q, %d errors in %s", trackers[i].Substring, len(trackers[i].Errors), trackers[i].Errors[len(trackers[i].Errors)-1].Sub(trackers[i].Errors[0]).Round(time.Second))
+					} else {
+						// find the number of errors in the last window (last is the most recent)
+						n := sort.Search(len(trackers[i].Errors), func(j int) bool {
+							return trackers[i].Errors[len(trackers[i].Errors)-1].Sub(trackers[i].Errors[j]) <= trackers[i].Window
+						})
+						log("Find error substring %q, %d errors in %s (%d errors in last %s)", trackers[i].Substring, len(trackers[i].Errors),
+							trackers[i].Errors[len(trackers[i].Errors)-1].Sub(trackers[i].Errors[0]).Round(time.Second),
+							len(trackers[i].Errors)-n, trackers[i].Window.Round(time.Second),
+						)
+					}
+
+					if trackers[i].ShouldRestartProcess() && !restartInit {
+						log("Restarting process due to error limit for substring %q", trackers[i].Substring)
+						restartInit = true
+						restartChan <- struct{}{}
+					}
+				}
+			}
+		}
+		if !restartInit {
+			exitChan <- struct{}{}
+		}
+	}
+
+	runCmd := func() {
+		cmd = exec.Command(cmdName, cmdArgs...)
+		stdoutPipe, _ = cmd.StdoutPipe()
+		stderrPipe, _ = cmd.StderrPipe()
+		restartInit = false
+
+		wg.Add(2)
+		go func() {
+			defer wg.Done()
+			scanner := bufio.NewScanner(stdoutPipe)
+			scanAndTrackErrors(scanner, trackers, os.Stdout, restartChan, exitChan)
+		}()
+
+		go func() {
+			defer wg.Done()
+			scanner := bufio.NewScanner(stderrPipe)
+			scanAndTrackErrors(scanner, trackers, os.Stderr, restartChan, exitChan)
+		}()
+
+		log("Starting process %q", cmdName)
+		cmd.Start()
+
+	}
+
+	runCmd()
+
+	for {
+		select {
+		case sig := <-sigChan:
+			log("Received signal %v", sig)
+			switch sig {
+			case syscall.SIGINT, syscall.SIGTERM:
+				log("Stopping %q process", cmdName)
+				cmd.Process.Signal(sig)
+				cmd.Wait()
+				wg.Wait()
+				time.Sleep(100 * time.Millisecond)
+				os.Exit(0)
+			}
+		case <-restartChan:
+			log("Too many errors, restarting process")
+			if err := cmd.Process.Signal(syscall.SIGINT); err != nil {
+				log("Error sending signal to process: %v", err)
+			}
+			cmd.Wait()
+			wg.Wait()
+			log("%q exited, restarting", cmdName)
+			for i := range trackers {
+				trackers[i].Errors = []time.Time{}
+			}
+			runCmd()
+		case <-exitChan:
+			log("%q exited, exiting", cmdName)
+			cmd.Wait()
+			wg.Wait()
+			time.Sleep(100 * time.Millisecond)
+			os.Exit(0)
+		}
+	}
+}