Aggregate errors with timestamps (#105)

pkg/sbproxy/notifications/handlers/broker_handler.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+
 	"github.com/Peripli/service-manager/pkg/util/slice"
 
 	"github.com/pkg/errors"

pkg/sbproxy/reconcile/reconcile_visibilities.go

@@ -198,7 +198,7 @@ func (r *resyncJob) reconcileServiceVisibilities(ctx context.Context, platformVi
 type visibilityProcessingState struct {
 	Ctx           context.Context
 	Mutex         sync.Mutex
-	ErrorOccurred error
+	ErrorOccurred *CompositeError
 
 	WaitGroupLimit chan struct{}
 	WaitGroup      sync.WaitGroup
@@ -207,6 +207,7 @@ type visibilityProcessingState struct {
 func (r *resyncJob) newVisibilityProcessingState(ctx context.Context) *visibilityProcessingState {
 	return &visibilityProcessingState{
 		Ctx:            ctx,
+		ErrorOccurred:  &CompositeError{},
 		WaitGroupLimit: make(chan struct{}, r.options.MaxParallelRequests),
 	}
 }
@@ -248,13 +249,10 @@ func execAsync(state *visibilityProcessingState, visibility *platform.Visibility
 			state.WaitGroup.Done()
 		}()
 
-		err := f(state.Ctx, visibility)
-		if err != nil {
+		if err := f(state.Ctx, visibility); err != nil {
 			state.Mutex.Lock()
 			defer state.Mutex.Unlock()
-			if state.ErrorOccurred == nil {
-				state.ErrorOccurred = err
-			}
+			state.ErrorOccurred.Add(err)
 		}
 	}()
 
@@ -263,7 +261,11 @@ func execAsync(state *visibilityProcessingState, visibility *platform.Visibility
 
 func await(state *visibilityProcessingState) error {
 	state.WaitGroup.Wait()
-	return state.ErrorOccurred
+	if state.ErrorOccurred.Len() != 0 {
+		return state.ErrorOccurred
+	}
+
+	return nil
 }
 
 // getVisibilityKey maps a generic visibility to a specific string. The string contains catalogID and scope for non-public plans

pkg/sbproxy/reconcile/reconciler.go

@@ -18,7 +18,10 @@ package reconcile
 
 import (
 	"context"
+	"fmt"
+	"strings"
 	"sync"
+	"time"
 
 	"github.com/Peripli/service-manager/pkg/types"
 
@@ -36,6 +39,44 @@ type Resyncer interface {
 	Resync(ctx context.Context)
 }
 
+// TimestampedError contains an error and a timestamp in time.RFC3339Nano format
+type TimestampedError struct {
+	Cause     error
+	Timestamp string
+}
+
+func (te TimestampedError) Error() string {
+	return fmt.Sprintf("%s-%s", te.Timestamp, te.Cause)
+}
+
+// CompositeError consists of multiple errors and attaches timestamps to them
+type CompositeError []error
+
+// Error implements the error interface and returns a string representation of the composite error
+func (ce *CompositeError) Error() string {
+	errs := make([]string, 0, len(*ce))
+	for i := range *ce {
+		if (*ce)[i] != nil {
+			errs = append(errs, (*ce)[i].Error())
+		}
+	}
+
+	return fmt.Sprintf("composite error: %v", strings.Join(errs, "; "))
+}
+
+// Add allows appending errors
+func (ce *CompositeError) Add(e error) {
+	*ce = append(*ce, TimestampedError{Cause: e, Timestamp: time.Now().Format(time.RFC3339Nano)})
+}
+
+// Len returns the number of errors present in the composite error. If the composite error is nil, Len returns 0.
+func (ce *CompositeError) Len() int {
+	if ce == nil {
+		return 0
+	}
+	return len(*ce)
+}
+
 // Reconciler takes care of propagating broker and visibility changes to the platform.
 // TODO if the reg credentials are changed (the ones under cf.reg) we need to update the already registered brokers
 type Reconciler struct {

pkg/sm/client.go

@@ -181,7 +181,7 @@ func (c *ServiceManagerClient) call(ctx context.Context, smURL string, params ma
 	}
 	q := fullURL.Query()
 	for k, v := range params {
-		q.Add(k,v)
+		q.Add(k, v)
 	}
 	fullURL.RawQuery = q.Encode()
 	return util.ListAll(ctx, c.httpClient.Do, fullURL.String(), list)