Merge pull request #14 from SenseUnit/cmd_output_retries

Snawoot · web-flow · commit fed00ae4ca37 · 2024-04-28T16:28:03.000+03:00
command output: retries option
diff --git a/README.md b/README.md
@@ -124,6 +124,7 @@ Configuration:
 * **`group`** (_uint64_) identifier of group.
 * **`command`** (_list of strings_) command and arguments.
 * **`timeout`** (_duration_) execution time limit for the command.
+* **`retries`** (_int_) attempts to retry failed command. Default is `1`.
 * **`wait_delay`** (_duration_) delay to wait for I/O to complete after process termination. Zero value disables I/O cancellation logic. Default is `100ms`.
 
 ### Configuration example
@@ -187,6 +188,7 @@ outputs:
         - "--group"
         - "1000"
       timeout: 5s
+      retries: 3
 
 ```
  
diff --git a/output/command.go b/output/command.go
@@ -19,6 +19,7 @@ type CommandConfig struct {
 	Group     *uint64
 	Command   []string
 	Timeout   time.Duration
+	Retries   *int
 	WaitDelay *time.Duration `yaml:"wait_delay"`
 }
 
@@ -27,6 +28,7 @@ type Command struct {
 	group     uint64
 	command   []string
 	timeout   time.Duration
+	retries   int
 	waitDelay time.Duration
 	syncQueue chan struct{}
 	shutdown  chan struct{}
@@ -49,11 +51,16 @@ func NewCommand(cfg *config.OutputConfig, bridge iface.GroupBridge) (*Command, e
 	if cc.WaitDelay != nil {
 		waitDelay = *cc.WaitDelay
 	}
+	retries := 1
+	if cc.Retries != nil && *cc.Retries > 1 {
+		retries = *cc.Retries
+	}
 	return &Command{
 		bridge:    bridge,
 		group:     *cc.Group,
 		command:   cc.Command,
 		timeout:   cc.Timeout,
+		retries:   retries,
 		waitDelay: waitDelay,
 		syncQueue: make(chan struct{}, 1),
 		shutdown:  make(chan struct{}),
@@ -117,6 +124,24 @@ func (o *Command) syncLoop() {
 }
 
 func (o *Command) runCommand() {
+	for i := 0; i < o.retries; i++ {
+		err := o.runCommandAttempt()
+		if err != nil {
+			var ee *exec.ExitError
+			if errors.As(err, &ee) {
+				log.Printf("command %v exited with code %d", o.command, ee.ExitCode())
+			} else {
+				log.Printf("command %v run error: %v", o.command, err)
+			}
+		} else {
+			log.Printf("command %v succeeded", o.command)
+			return
+		}
+	}
+	log.Printf("command %v all attempts failed!", o.command)
+}
+
+func (o *Command) runCommandAttempt() error {
 	ctx := context.Background()
 	if o.timeout > 0 {
 		ctx1, cancel := context.WithTimeout(ctx, o.timeout)
@@ -133,29 +158,16 @@ func (o *Command) runCommand() {
 	}
 	cmd.Stdin = &stdinBuf
 
-	err := func() error {
-		stdout := newOutputForwarder("stdout", o.command)
-		defer stdout.Close()
-		cmd.Stdout = stdout
-
-		stderr := newOutputForwarder("stderr", o.command)
-		defer stderr.Close()
-		cmd.Stderr = stderr
+	stdout := newOutputForwarder("stdout", o.command)
+	defer stdout.Close()
+	cmd.Stdout = stdout
 
-		log.Printf("starting sync command %v...", o.command)
-		return cmd.Run()
-	}()
+	stderr := newOutputForwarder("stderr", o.command)
+	defer stderr.Close()
+	cmd.Stderr = stderr
 
-	if err != nil {
-		var ee *exec.ExitError
-		if errors.As(err, &ee) {
-			log.Printf("command %v exited with code %d", o.command, ee.ExitCode())
-		} else {
-			log.Printf("command %v run error: %v", o.command, err)
-		}
-	} else {
-		log.Printf("command %v succeeded", o.command)
-	}
+	log.Printf("starting sync command %v...", o.command)
+	return cmd.Run()
 }
 
 type outputForwarder struct {
