add retries to csaf enricher

Author: dcaravel

scanner/enricher/csaf/fetcher.go

@@ -22,13 +22,17 @@ import (
 	"github.com/klauspost/compress/snappy"
 	"github.com/quay/claircore/libvuln/driver"
 	"github.com/quay/claircore/pkg/tmp"
+	"github.com/stackrox/rox/pkg/retry"
 	"github.com/stackrox/rox/pkg/utils"
 	"github.com/stackrox/rox/scanner/enricher/csaf/internal/zreader"
 )
 
 var (
 	// compressedFileTimeout matches Claircore's VEX https://github.com/quay/claircore/blob/v1.5.34/rhel/vex/fetcher.go.
 	compressedFileTimeout = 2 * time.Minute
+
+	advisoryFetchRetries        = 5
+	advisoryFetchRetryBaseDelay = 500 * time.Millisecond
 )
 
 // fingerprint is used to track the state of the changes.csv and deletions.csv endpoints.
@@ -339,49 +343,22 @@ func (e *Enricher) processChanges(ctx context.Context, w io.Writer, fp *fingerpr
 			continue
 		}
 
-		changed[path.Base(cvePath)] = true
-
 		advisoryURI, err := e.base.Parse(cvePath)
 		if err != nil {
 			return err
 		}
-		req, err := http.NewRequestWithContext(ctx, http.MethodGet, advisoryURI.String(), nil)
-		if err != nil {
-			return fmt.Errorf("error creating advisory request %w", err)
-		}
-
-		// Use a func here as we're in a loop and want to make sure the
-		// body is closed in all events.
-		err = func() error {
-			res, err := e.c.Do(req)
-			if err != nil {
-				return fmt.Errorf("error making advisory request %w", err)
-			}
-			defer utils.IgnoreError(res.Body.Close)
-			err = checkResponse(res, http.StatusOK)
-			if err != nil {
-				return fmt.Errorf("unexpected response: %w", err)
-			}
 
-			// Add compacted JSON to buffer.
-			_, err = buf.ReadFrom(res.Body)
-			if err != nil {
-				return fmt.Errorf("error reading from buffer: %w", err)
-			}
-			slog.DebugContext(ctx, "copying body to file", "url", advisoryURI.String())
-			err = json.Compact(&bc, buf.Bytes())
-			if err != nil {
-				return fmt.Errorf("error compressing JSON: %w", err)
-			}
-
-			bc.WriteByte('\n')
-			_, _ = w.Write(bc.Bytes())
-			l++
-			return nil
-		}()
-		if !errors.Is(err, nil) {
-			return err
+		// Fetch individual advisory with retries. If the advisory is
+		// unavailable after all attempts, skip it — the compressed
+		// archive (downloaded next) will likely container a version of it.
+		advisoryErr := e.fetchAdvisory(ctx, advisoryURI, &buf, &bc, w)
+		if advisoryErr != nil {
+			slog.WarnContext(ctx, "skipping advisory due to fetch failure; will fall back to archive",
+				"advisory", cvePath, "reason", advisoryErr)
+			continue
 		}
+		changed[path.Base(cvePath)] = true
+		l++
 	}
 
 	if !errors.Is(err, io.EOF) {
@@ -390,6 +367,46 @@ func (e *Enricher) processChanges(ctx context.Context, w io.Writer, fp *fingerpr
 	return nil
 }
 
+func (e *Enricher) fetchAdvisory(ctx context.Context, advisoryURI *url.URL, buf, bc *bytes.Buffer, w io.Writer) error {
+	fetch := func() error {
+		buf.Reset()
+		bc.Reset()
+		req, err := http.NewRequestWithContext(ctx, http.MethodGet, advisoryURI.String(), nil)
+		if err != nil {
+			return err
+		}
+		res, err := e.c.Do(req)
+		if err != nil {
+			return err
+		}
+		defer utils.IgnoreError(res.Body.Close)
+		if err := checkResponse(res, http.StatusOK); err != nil {
+			return err
+		}
+		if _, err := buf.ReadFrom(res.Body); err != nil {
+			return fmt.Errorf("error reading response body: %w", err)
+		}
+		if err := json.Compact(bc, buf.Bytes()); err != nil {
+			return fmt.Errorf("error compacting JSON: %w", err)
+		}
+		bc.WriteByte('\n')
+		_, _ = w.Write(bc.Bytes())
+		return nil
+	}
+
+	return retry.WithRetry(fetch,
+		retry.WithContext(ctx),
+		retry.Tries(advisoryFetchRetries),
+		retry.BetweenAttempts(func(attempt int) {
+			time.Sleep(advisoryFetchRetryBaseDelay * time.Duration(1<<attempt))
+		}),
+		retry.OnFailedAttempts(func(err error) {
+			slog.WarnContext(ctx, "advisory fetch attempt failed",
+				"advisory", advisoryURI, "reason", err)
+		}),
+	)
+}
+
 // checkResponse takes a http.Response and a variadic of ints representing
 // acceptable http status codes. The error returned will attempt to include
 // some content from the server's response.

scanner/enricher/csaf/fetcher_test.go

@@ -3,12 +3,21 @@ package csaf
 import (
 	"bufio"
 	"bytes"
+	"context"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/klauspost/compress/snappy"
 	"github.com/quay/claircore/test"
 	testvex "github.com/quay/claircore/test/vex"
 	"github.com/quay/claircore/toolkit/types/csaf"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestFetchEnrichment(t *testing.T) {
@@ -57,3 +66,94 @@ func TestFetchEnrichment(t *testing.T) {
 		t.Errorf("got %d entries but expected %d", lnCt, expectedLnCt)
 	}
 }
+
+func TestFetchAdvisory_RetriesOnFailure(t *testing.T) {
+	origRetries := advisoryFetchRetries
+	origBaseDelay := advisoryFetchRetryBaseDelay
+	advisoryFetchRetries = 3
+	advisoryFetchRetryBaseDelay = 1 * time.Millisecond
+	t.Cleanup(func() {
+		advisoryFetchRetries = origRetries
+		advisoryFetchRetryBaseDelay = origBaseDelay
+	})
+
+	validJSON := `{"document":{"tracking":{"id":"RHSA-2024:0001"}}}`
+
+	t.Run("succeeds after transient failure", func(t *testing.T) {
+		var attempts atomic.Int32
+		srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+			if attempts.Add(1) < 3 {
+				w.WriteHeader(http.StatusInternalServerError)
+				return
+			}
+			fmt.Fprint(w, validJSON)
+		}))
+		t.Cleanup(srv.Close)
+
+		e := &Enricher{c: srv.Client()}
+		u, _ := url.Parse(srv.URL + "/advisory.json")
+		var buf, bc bytes.Buffer
+		var w bytes.Buffer
+		err := e.fetchAdvisory(context.Background(), u, &buf, &bc, &w)
+		require.NoError(t, err)
+		assert.Equal(t, int32(3), attempts.Load())
+		assert.Contains(t, w.String(), "RHSA-2024:0001")
+	})
+
+	t.Run("returns error after all retries exhausted", func(t *testing.T) {
+		srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+			w.WriteHeader(http.StatusNotFound)
+			fmt.Fprint(w, "not found")
+		}))
+		t.Cleanup(srv.Close)
+
+		e := &Enricher{c: srv.Client()}
+		u, _ := url.Parse(srv.URL + "/advisory.json")
+		var buf, bc bytes.Buffer
+		var w bytes.Buffer
+		err := e.fetchAdvisory(context.Background(), u, &buf, &bc, &w)
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "404")
+		assert.Empty(t, w.String())
+	})
+}
+
+func TestProcessChanges_SkipsUnavailableAdvisory(t *testing.T) {
+	origRetries := advisoryFetchRetries
+	advisoryFetchRetries = 1
+	t.Cleanup(func() {
+		advisoryFetchRetries = origRetries
+	})
+
+	validJSON := `{"document":{"tracking":{"id":"RHSA-2024:0001"}}}`
+	mux := http.NewServeMux()
+	mux.HandleFunc("/changes.csv", func(w http.ResponseWriter, _ *http.Request) {
+		w.Header().Set("Etag", "test-etag")
+		// Two advisories: one available, one returning 404.
+		fmt.Fprint(w,
+			`"2024/rhsa-2024_0001.json","2025-01-10T18:37:32+00:00"`+"\n"+
+				`"2024/rhsa-2024_0002.json","2025-01-10T18:37:32+00:00"`)
+	})
+	mux.HandleFunc("/2024/rhsa-2024_0001.json", func(w http.ResponseWriter, _ *http.Request) {
+		fmt.Fprint(w, validJSON)
+	})
+	mux.HandleFunc("/2024/rhsa-2024_0002.json", func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusNotFound)
+		fmt.Fprint(w, "<!DOCTYPE html><html>not found</html>")
+	})
+	srv := httptest.NewServer(mux)
+	t.Cleanup(srv.Close)
+
+	u, _ := url.Parse(srv.URL + "/")
+	e := &Enricher{c: srv.Client(), base: u}
+	fp := &fingerprint{}
+	changed := map[string]bool{}
+	var w bytes.Buffer
+
+	err := e.processChanges(context.Background(), &w, fp, changed)
+	require.NoError(t, err)
+	assert.True(t, changed["rhsa-2024_0001.json"], "available advisory should be marked as changed")
+	assert.False(t, changed["rhsa-2024_0002.json"], "404'd advisory should not be marked as changed")
+	assert.Contains(t, w.String(), "RHSA-2024:0001")
+	assert.NotContains(t, w.String(), "RHSA-2024:0002")
+}