add Permanent Errors

Author: ndakkoune

aws/logs_monitoring_go/internal/forwarding/client.go

@@ -24,7 +24,7 @@ var Client = newClient()
 func newClient() *http.Client {
 	transport := http.DefaultTransport.(*http.Transport).Clone()
 	transport.TLSHandshakeTimeout = tlsHandshakeTimeout
-	transport.MaxIdleConnsPerHost = maxConcurrency
+	transport.MaxIdleConnsPerHost = MaxConcurrency
 	transport.DialContext = (&net.Dialer{
 		Timeout:   dialContextTimeout,
 		KeepAlive: dialContextKeepAlive,

aws/logs_monitoring_go/internal/forwarding/forwarding.go

@@ -22,7 +22,7 @@ import (
 	"golang.org/x/sync/errgroup"
 )
 
-const maxConcurrency = 5
+const MaxConcurrency = 5
 
 type Forwarder struct {
 	cfg     *config.Config
@@ -39,7 +39,7 @@ func NewForwarder(cfg *config.Config, client *http.Client, storage string) Forwa
 }
 
 func (f Forwarder) Start(ctx context.Context, in <-chan model.LogEntry) error {
-	batches := make(chan []byte, maxConcurrency)
+	batches := make(chan []byte, MaxConcurrency)
 	batcher := batching.NewBatcher()
 
 	producerErrCh := make(chan error, 1)
@@ -49,7 +49,7 @@ func (f Forwarder) Start(ctx context.Context, in <-chan model.LogEntry) error {
 	}()
 
 	var eg errgroup.Group
-	eg.SetLimit(maxConcurrency)
+	eg.SetLimit(MaxConcurrency)
 
 	var errs []error
 	var mu sync.Mutex
@@ -66,7 +66,10 @@ func (f Forwarder) Start(ctx context.Context, in <-chan model.LogEntry) error {
 		}
 
 		eg.Go(func() error {
-			if err := f.send(ctx, body); err != nil {
+			if err := f.Send(ctx, body); err != nil {
+				if f.storage != nil {
+					err = f.storage.Put(ctx, body, f.storageTag)
+				}
 				mu.Lock()
 				errs = append(errs, err)
 				mu.Unlock()
@@ -79,7 +82,7 @@ func (f Forwarder) Start(ctx context.Context, in <-chan model.LogEntry) error {
 	return errors.Join(append(errs, <-producerErrCh)...)
 }
 
-func (f Forwarder) send(ctx context.Context, payload []byte) error {
+func (f Forwarder) Send(ctx context.Context, payload []byte) error {
 	ctx, cancel := context.WithTimeout(ctx, timeout)
 	defer cancel()
 

aws/logs_monitoring_go/internal/forwarding/middleware.go

@@ -87,6 +87,16 @@ func WithRetry(maxAttempts int, next http.RoundTripper) RoundTripperFunc {
 
 			if !isRetryable(resp.StatusCode) {
 				slog.LogAttrs(req.Context(), slog.LevelDebug, "request complete", attrs...)
+				if isPermanent(resp.StatusCode) {
+					body, err := io.ReadAll(resp.Body)
+					if err != nil {
+						slog.Warn("reading body response", slog.Any("error", err))
+					}
+					return nil, &PermanentError{
+						StatusCode: resp.StatusCode,
+						Reason:     string(body),
+					}
+				}
 				return resp, nil
 			}
 
@@ -102,6 +112,18 @@ func WithRetry(maxAttempts int, next http.RoundTripper) RoundTripperFunc {
 	}
 }
 
+func isPermanent(statusCode int) bool {
+	switch statusCode {
+	case http.StatusBadRequest,
+		http.StatusUnauthorized,
+		http.StatusForbidden,
+		http.StatusRequestEntityTooLarge:
+		return true
+	default:
+		return statusCode >= 400
+	}
+}
+
 func isRetryable(statusCode int) bool {
 	switch statusCode {
 	case http.StatusTooManyRequests,