Merge branch 'avonbertoldi/pulp-push-retry' into 'main'

Retry pulp content push  commands on specific errors

See merge request https://gitlab.com/gitlab-org/gitlab-runner/-/merge_requests/6197

Merged-by: Vishal Tak <vtak@gitlab.com>
Approved-by: Balasankar 'Balu' C <balasankar@gitlab.com>
Approved-by: Vishal Tak <vtak@gitlab.com>
Reviewed-by: GitLab Duo <gitlab-duo@gitlab.com>
Co-authored-by: Axel von Bertoldi <avonbertoldi@gitlab.com>

Author: Vishal Tak

magefiles/pulp/push.go

@@ -12,7 +12,9 @@ import (
 	"path/filepath"
 	"regexp"
 	"strings"
+	"time"
 
+	"github.com/jpillora/backoff"
 	"github.com/magefile/mage/sh"
 	"github.com/samber/lo"
 	"github.com/sourcegraph/conc/pool"
@@ -105,15 +107,28 @@ type (
 )
 
 func (p *basePusher) runPulpCmd(args ...string) error {
-	fmt.Println("executing", "pulp", strings.Join(args, " "))
+	slog.Info("executing", "cmd", "pulp", "args", args)
 	if p.dryrun {
 		return nil
 	}
 	return p.run("pulp", args...)
 }
 
+var pulpRetryErrors = []*regexp.Regexp{
+	regexp.MustCompile(`Artifact with sha256 checksum of '.*' already exists`),
+}
+
+func (p *basePusher) retryPulpCmd(args []string, out io.Writer) error {
+	slog.Info("executing", "cmd", "pulp", "args", args)
+	if p.dryrun {
+		return nil
+	}
+
+	return newRetryCommand("pulp", args, pulpRetryErrors, out, p.exec).run()
+}
+
 func (p *basePusher) execCmd(out io.Writer, cmd string, args ...string) error {
-	fmt.Println("executing", cmd, strings.Join(args, " "))
+	slog.Info("executing", "cmd", cmd, "args", args)
 	if p.dryrun {
 		return nil
 	}
@@ -132,8 +147,7 @@ func (p *debPusher) Push(releases, pkgFiles []string) error {
 	for _, release := range releases {
 		for _, pkgFile := range pkgFiles {
 			pool.Go(func() error {
-				slog.Debug("Pushing", "package", pkgFile, "release", release)
-				return p.runPulpCmd(p.pushArgs(release, pkgFile)...)
+				return p.retryPulpCmd(p.pushArgs(release, pkgFile), io.Discard)
 			})
 		}
 	}
@@ -241,7 +255,7 @@ func (p *rpmPusher) doPush(pkgFile, repo string) (string, error) {
 	args := p.pushArgs(pkgFile, repo)
 
 	out := bytes.Buffer{}
-	if err := p.execCmd(&out, "pulp", args...); err != nil {
+	if err := p.retryPulpCmd(args, &out); err != nil {
 		return "", err
 	}
 
@@ -326,3 +340,59 @@ func parseRPMInfo(out io.Reader) (rpmInfo, error) {
 
 	return info, nil
 }
+
+type retryCommand struct {
+	cmd           string
+	args          []string
+	backoff       backoff.Backoff
+	out           io.Writer
+	retryableErrs []*regexp.Regexp
+	exec          shExec
+}
+
+func newRetryCommand(cmd string, args []string, retryableErrs []*regexp.Regexp, out io.Writer, exec shExec) *retryCommand {
+	return &retryCommand{
+		cmd:  cmd,
+		args: args,
+		backoff: backoff.Backoff{
+			Min: time.Second,
+			Max: 5 * time.Second,
+		},
+		out:           out,
+		retryableErrs: retryableErrs,
+		exec:          exec,
+	}
+}
+
+func (c *retryCommand) run() error {
+	for i := range 5 {
+		slog.Info("attempting to run command", "attempt", i+1, "command", c.cmd, "args", c.args)
+
+		outBuf, errBuf := bytes.Buffer{}, bytes.Buffer{}
+		stdout := io.MultiWriter(&outBuf, os.Stdout)
+		stderr := io.MultiWriter(&errBuf, os.Stderr)
+
+		_, err := c.exec(nil, stdout, stderr, c.cmd, c.args...)
+
+		if err == nil {
+			_, _ = io.Copy(c.out, &outBuf)
+			return nil
+		}
+		if c.isRetryable(errBuf.String()) {
+			time.Sleep(c.backoff.Duration())
+			continue
+		}
+		return fmt.Errorf("execution of command (%s %s) failed: %s", c.cmd, c.args, errBuf.String())
+	}
+
+	return fmt.Errorf("execution of command (%s %s) failed after 5 retries ", c.cmd, c.args)
+}
+
+func (c *retryCommand) isRetryable(stderr string) bool {
+	for _, re := range c.retryableErrs {
+		if re.MatchString(stderr) {
+			return true
+		}
+	}
+	return false
+}

magefiles/pulp/push_test.go

@@ -5,9 +5,12 @@ package pulp
 import (
 	"fmt"
 	"io"
+	"regexp"
 	"strings"
 	"testing"
+	"time"
 
+	"github.com/jpillora/backoff"
 	"github.com/stretchr/testify/require"
 )
 
@@ -193,6 +196,161 @@ Description : GitLab Runner
 	}
 }
 
+func TestRetryCommandRun(t *testing.T) {
+	tests := map[string]struct {
+		execBehavior    func(attempt int) (bool, string, error) // returns (success, stderr, error )
+		retryableErrs   []*regexp.Regexp
+		expectedError   bool
+		errorContains   string
+		expectedAttempt int
+	}{
+		"successful on first attempt": {
+			execBehavior: func(attempt int) (bool, string, error) {
+				return true, "", nil
+			},
+			retryableErrs:   []*regexp.Regexp{},
+			expectedError:   false,
+			expectedAttempt: 1,
+		},
+		"successful on second attempt with retryable error": {
+			execBehavior: func(attempt int) (bool, string, error) {
+				if attempt == 1 {
+					return false, "Artifact with checksum of 'abc123' already exists.", fmt.Errorf("artifact error")
+				}
+				return true, "", nil
+			},
+			retryableErrs: []*regexp.Regexp{
+				regexp.MustCompile(`Artifact with checksum of '.*' already exists\.`),
+			},
+			expectedError:   false,
+			expectedAttempt: 2,
+		},
+		"successful on third attempt with retryable error": {
+			execBehavior: func(attempt int) (bool, string, error) {
+				if attempt <= 2 {
+					return false, "Artifact with checksum of 'xyz789' already exists.", fmt.Errorf("artifact error")
+				}
+				return true, "", nil
+			},
+			retryableErrs: []*regexp.Regexp{
+				regexp.MustCompile(`Artifact with checksum of '.*' already exists\.`),
+			},
+			expectedError:   false,
+			expectedAttempt: 3,
+		},
+		"fails with non-retryable error on first attempt": {
+			execBehavior: func(attempt int) (bool, string, error) {
+				return false, "Permission denied: cannot access repository", fmt.Errorf("permission denied")
+			},
+			retryableErrs: []*regexp.Regexp{
+				regexp.MustCompile(`Artifact with checksum of '.*' already exists\.`),
+			},
+			expectedError:   true,
+			errorContains:   "Permission denied",
+			expectedAttempt: 1,
+		},
+		"fails after max retries with retryable error": {
+			execBehavior: func(attempt int) (bool, string, error) {
+				return false, "Artifact with checksum of 'def456' already exists.", fmt.Errorf("artifact error")
+			},
+			retryableErrs: []*regexp.Regexp{
+				regexp.MustCompile(`Artifact with checksum of '.*' already exists\.`),
+			},
+			expectedError:   true,
+			errorContains:   "failed after 5 retries",
+			expectedAttempt: 5,
+		},
+		"multiple retryable error patterns": {
+			execBehavior: func(attempt int) (bool, string, error) {
+				if attempt == 1 {
+					return false, "Connection timeout: server not responding", fmt.Errorf("timeout")
+				}
+				if attempt == 2 {
+					return false, "Artifact with checksum of 'ghi012' already exists.", fmt.Errorf("artifact error")
+				}
+				return true, "", nil
+			},
+			retryableErrs: []*regexp.Regexp{
+				regexp.MustCompile(`Artifact with checksum of '.*' already exists\.`),
+				regexp.MustCompile(`Connection timeout:.*`),
+			},
+			expectedError:   false,
+			expectedAttempt: 3,
+		},
+		"retryable error on last attempt succeeds": {
+			execBehavior: func(attempt int) (bool, string, error) {
+				if attempt < 5 {
+					return false, "Artifact with checksum of 'jkl345' already exists.", fmt.Errorf("artifact error")
+				}
+				return true, "", nil
+			},
+			retryableErrs: []*regexp.Regexp{
+				regexp.MustCompile(`Artifact with checksum of '.*' already exists\.`),
+			},
+			expectedError:   false,
+			expectedAttempt: 5,
+		},
+		"no retryable errors configured": {
+			execBehavior: func(attempt int) (bool, string, error) {
+				return false, "Some error message", fmt.Errorf("some error")
+			},
+			retryableErrs:   []*regexp.Regexp{},
+			expectedError:   true,
+			errorContains:   "Some error message",
+			expectedAttempt: 1,
+		},
+		"empty stderr with error": {
+			execBehavior: func(attempt int) (bool, string, error) {
+				return false, "", fmt.Errorf("command failed")
+			},
+			retryableErrs: []*regexp.Regexp{
+				regexp.MustCompile(`Artifact with checksum of '.*' already exists\.`),
+			},
+			expectedError:   true,
+			errorContains:   "execution of command",
+			expectedAttempt: 1,
+		},
+	}
+
+	for tn, tt := range tests {
+		t.Run(tn, func(t *testing.T) {
+			attempt := 0
+
+			// Create mock exec function that tracks attempts
+			execMock := func(env map[string]string, out io.Writer, stderr io.Writer, cmd string, args ...string) (bool, error) {
+				attempt++
+				success, stderrMsg, err := tt.execBehavior(attempt)
+
+				if stderrMsg != "" {
+					_, _ = io.WriteString(stderr, stderrMsg)
+				}
+
+				return success, err
+			}
+
+			// Create retryCommand with mocked exec
+			cmd := newRetryCommand("test-cmd", []string{"arg1", "arg2"}, tt.retryableErrs, io.Discard, execMock)
+			// make it a bit faster
+			cmd.backoff = backoff.Backoff{Min: 10 * time.Millisecond, Max: 50 * time.Millisecond}
+
+			// Run the command
+			err := cmd.run()
+
+			// Verify results
+			if tt.expectedError {
+				require.Error(t, err)
+				if tt.errorContains != "" {
+					require.Contains(t, err.Error(), tt.errorContains)
+				}
+			} else {
+				require.NoError(t, err)
+			}
+
+			require.Equal(t, tt.expectedAttempt, attempt, "expected %d attempts, got %d", tt.expectedAttempt, attempt)
+		})
+	}
+}
+
 func TestRpmPusherPush(t *testing.T) {
 	tests := map[string]struct {
 		releases      []string