enable direct error handling for bundle plugin by returning downloade…

…r errors for bundles in manual trigger mode

Signed-off-by: Torsten Wunderlich <[email protected]>

cmd/exec_test.go

@@ -3,6 +3,7 @@ package cmd
 import (
 	"bytes"
 	"context"
+	"errors"
 	"fmt"
 	"os"
 	"path/filepath"
@@ -11,6 +12,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/open-policy-agent/opa/ast"
 	"github.com/open-policy-agent/opa/cmd/internal/exec"
 	"github.com/open-policy-agent/opa/internal/file/archive"
 	loggingtest "github.com/open-policy-agent/opa/logging/test"
@@ -321,15 +323,32 @@ main contains "hello" if {
 
 					ctx, cancel := context.WithCancel(context.Background())
 					defer cancel()
-					go func() {
+					go func(expectedErrors []string) {
 						err := runExecWithContext(ctx, params)
 						// Note(philipc): Catch the expected cancellation
 						// errors, allowing unexpected test failures through.
 						if err != context.Canceled {
-							t.Error(err)
-							return
+							var errs ast.Errors
+							if errors.As(err, &errs) {
+								for _, expErr := range expectedErrors {
+									found := false
+									for _, e := range errs {
+										if strings.Contains(e.Error(), expErr) {
+											found = true
+											break
+										}
+									}
+									if !found {
+										t.Errorf("Could not find expected error: %s in %v", expErr, errs)
+										return
+									}
+								}
+							} else {
+								t.Error(err)
+								return
+							}
 						}
-					}()
+					}(tc.expErrs)
 
 					if !test.Eventually(t, 5*time.Second, func() bool {
 						for _, expErr := range tc.expErrs {

plugins/bundle/errors.go

@@ -0,0 +1,34 @@
+package bundle
+
+import (
+	"errors"
+	"fmt"
+)
+
+// Errors represents a list of errors that occurred during a bundle load enriched by the bundle name.
+type Errors []Error
+
+func (e Errors) Unwrap() []error {
+	output := make([]error, len(e))
+	for i := range e {
+		output[i] = e[i]
+	}
+	return output
+}
+func (e Errors) Error() string {
+	err := errors.Join(e.Unwrap()...)
+	return err.Error()
+}
+
+type Error struct {
+	Name string
+	Err  error
+}
+
+func (e Error) Error() string {
+	return fmt.Sprintf("'%s': %v", e.Name, e.Err)
+}
+
+func (e Error) Unwrap() error {
+	return e.Err
+}

plugins/bundle/errors_test.go

@@ -0,0 +1,72 @@
+package bundle
+
+import (
+	"errors"
+	"fmt"
+	"testing"
+
+	"github.com/open-policy-agent/opa/download"
+)
+
+func TestErrors(t *testing.T) {
+	errs := Errors{
+		Error{Name: "foo", Err: fmt.Errorf("foo error")},
+		Error{Name: "bar", Err: fmt.Errorf("bar error")},
+	}
+
+	expected := "'foo': foo error\n'bar': bar error"
+	result := errs.Error()
+
+	if result != expected {
+		t.Errorf("Expected: %v \nbut got: %v", expected, result)
+	}
+}
+
+func TestUnwrapSlice(t *testing.T) {
+	fooErr := Error{Name: "foo", Err: fmt.Errorf("foo error")}
+	barErr := Error{Name: "bar", Err: fmt.Errorf("bar error")}
+
+	errs := Errors{fooErr, barErr}
+
+	result := errs.Unwrap()
+
+	if result[0].Error() != fooErr.Error() {
+		t.Fatalf("expected %v \nbut got: %v", fooErr, result[0])
+	}
+	if result[1].Error() != barErr.Error() {
+		t.Fatalf("expected %v \nbut got: %v", barErr, result[1])
+	}
+}
+
+func TestUnwrap(t *testing.T) {
+	fooErr := Error{Name: "foo", Err: download.HTTPError{StatusCode: 500}}
+	barErr := Error{Name: "bar", Err: download.HTTPError{StatusCode: 400}}
+
+	errs := Errors{fooErr, barErr}
+
+	// unwrap first bundle.Error
+	var bundleError Error
+	if !errors.As(errs, &bundleError) {
+		t.Fatal("failed to unwrap Error")
+	}
+	if bundleError.Error() != fooErr.Error() {
+		t.Fatalf("expected: %v \ngot: %v", fooErr, bundleError)
+	}
+
+	// unwrap first HTTPError
+	var httpError download.HTTPError
+	if !errors.As(errs, &httpError) {
+		t.Fatal("failed to unwrap Error")
+	}
+	if httpError.Error() != fooErr.Err.Error() {
+		t.Fatalf("expected: %v \ngot: %v", fooErr.Err, httpError)
+	}
+
+	// unwrap HTTPError from bundle.Error
+	if !errors.As(bundleError, &httpError) {
+		t.Fatal("failed to unwrap HTTPError")
+	}
+	if httpError.Error() != fooErr.Err.Error() {
+		t.Fatalf("expected: %v \nbgot: %v", fooErr.Err, httpError)
+	}
+}

plugins/bundle/plugin.go

@@ -257,16 +257,25 @@ func (p *Plugin) Loaders() map[string]Loader {
 
 // Trigger triggers a bundle download on all configured bundles.
 func (p *Plugin) Trigger(ctx context.Context) error {
+	var errs Errors
+
 	p.mtx.Lock()
 	downloaders := map[string]Loader{}
 	for name, dl := range p.downloaders {
 		downloaders[name] = dl
 	}
 	p.mtx.Unlock()
 
-	for _, d := range downloaders {
-		// plugin callback will log the trigger error and include it in the bundle status
-		_ = d.Trigger(ctx)
+	for name, d := range downloaders {
+		// plugin callback will also log the trigger error and include it in the bundle status
+		err := d.Trigger(ctx)
+		// only return errors for TriggerMode manual as periodic bundles will be retried
+		if err != nil && *p.config.Bundles[name].Trigger == plugins.TriggerManual {
+			errs = append(errs, Error{Name: name, Err: err})
+		}
+	}
+	if len(errs) > 0 {
+		return errs
 	}
 	return nil
 }

plugins/bundle/plugin_test.go

@@ -6204,6 +6204,68 @@ func TestPluginManualTriggerWithTimeout(t *testing.T) {
 	}
 }
 
+func TestPluginManualTriggerWithServerError(t *testing.T) {
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+
+	s := httptest.NewServer(http.HandlerFunc(func(resp http.ResponseWriter, _ *http.Request) {
+		resp.WriteHeader(500)
+	}))
+
+	// setup plugin pointing at fake server
+	manager := getTestManagerWithOpts([]byte(fmt.Sprintf(`{
+		"services": {
+			"default": {
+				"url": %q
+			}
+		}
+	}`, s.URL)))
+
+	var manual plugins.TriggerMode = "manual"
+	var periodic plugins.TriggerMode = "periodic"
+	var delay int64 = 10
+	polling := download.PollingConfig{MinDelaySeconds: &delay, MaxDelaySeconds: &delay}
+
+	plugin := New(&Config{
+		Bundles: map[string]*Source{
+			"manual": {
+				Service:        "default",
+				SizeLimitBytes: int64(bundle.DefaultSizeLimitBytes),
+				Config:         download.Config{Trigger: &manual},
+			},
+			"periodic": {
+				Service:        "default",
+				SizeLimitBytes: int64(bundle.DefaultSizeLimitBytes),
+				Config:         download.Config{Trigger: &periodic, Polling: polling},
+			},
+		},
+	}, manager)
+
+	err := plugin.Start(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	// manually trigger bundle download
+	err = plugin.Trigger(ctx)
+
+	plugin.Stop(ctx)
+
+	var bundleErrors Errors
+	if errors.As(err, &bundleErrors) {
+		if len(bundleErrors) != 1 {
+			t.Fatalf("expected exactly one error, got %d", len(bundleErrors))
+		}
+		for _, e := range bundleErrors {
+			if e.Name != "manual" {
+				t.Fatalf("expect error only for bundle with manual trigger mode")
+			}
+		}
+	} else {
+		t.Fatalf("expected type of error to be %s but got %s", reflect.TypeOf(bundleErrors), reflect.TypeOf(err))
+	}
+}
+
 func TestGetNormalizedBundleName(t *testing.T) {
 	cases := []struct {
 		input string