Merge pull request #48 from matt-gp/err-group

Switched From WaitGroup -> ErrorGroup

Author: matt-gp

CONTRIBUTING.md

@@ -295,8 +295,26 @@ if err != nil {
     logger.Error("Failed to get data", "id", id, "error", err)
     return nil, err
 }
+
+// Good: Check HTTP status codes and return errors for failures
+if resp.StatusCode >= http.StatusBadRequest {
+    logger.Error(
+        ctx,
+        logger,
+        fmt.Sprintf("received non-success status code: %d", resp.StatusCode),
+        log.String("status_code", strconv.Itoa(resp.StatusCode)),
+    )
+    return fmt.Errorf("received non-success status code: %d", resp.StatusCode)
+}
 ```
 
+**HTTP Error Handling:**
+- Treat all HTTP status codes >= 400 as errors
+- Log errors with relevant context (tenant, signal type, status code)
+- Record errors in distributed tracing spans
+- Return errors to enable retry logic in upstream systems
+- Always record metrics even for failed requests
+
 ### Logging
 
 Use structured logging:

README.md

@@ -301,7 +301,7 @@ func (h *Handlers) Logs(w http.ResponseWriter, r *http.Request) {
 **Processor Package (`internal/processor/`):**
 - `New[T ResourceData]()` - Create generic processor with signal-specific callbacks
 - `Partition(ctx, resources)` - Partition resources by tenant from resource attributes
-- `Dispatch(ctx, tenantMap)` - Concurrent forwarding to backend with tenant headers
+- `Dispatch(ctx, tenantMap)` - Concurrent forwarding to backend with tenant headers; returns error if any backend responds with status >= 400
 - `send(ctx, tenant, resources)` - HTTP client with protobuf marshaling and metrics
 
 **Handler Package (`internal/handler/`):**
@@ -564,6 +564,19 @@ When forwarding data to observability backends:
 5. Content-Type is set to `application/x-protobuf`
 6. Original protobuf format is preserved with proper headers via `addHeaders()`
 
+### Error Handling
+
+The proxy implements robust error handling for backend responses:
+
+- **Success Responses (< 400)**: Data is successfully forwarded and metrics are recorded with the response status code
+- **Error Responses (>= 400)**: The proxy treats all HTTP status codes of 400 or higher as errors:
+  - An error is logged with the status code, tenant, and signal type
+  - The request is marked as failed in distributed tracing
+  - An error is returned to the caller, which may trigger retry logic in upstream collectors
+  - Metrics are still recorded with the error status code for observability
+
+This ensures that client errors (4xx) and server errors (5xx) from the backend are properly surfaced and can be monitored through the proxy's own telemetry.
+
 ## Observability
 
 The service exposes metrics about its operation:
@@ -802,7 +815,7 @@ go tool cover -func=coverage.out
 **Processor Tests (`internal/processor/processor_test.go`):**
 - `TestNew` - Processor creation with various configurations
 - `TestPartition` - Tenant partitioning logic with primary/fallback labels and defaults
-- `TestDispatch` - Concurrent request dispatching to multiple tenants
+- `TestDispatch` - Concurrent request dispatching to multiple tenants with error handling for HTTP status >= 400
 - `TestSend` - Individual HTTP request handling with error scenarios
 
 **Handler Tests (`internal/handler/handlers_test.go`):**

go.mod

@@ -24,6 +24,8 @@ require (
 	go.uber.org/mock v0.6.0
 )
 
+require golang.org/x/sync v0.20.0
+
 require (
 	github.com/cenkalti/backoff/v5 v5.0.3 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect

go.sum

@@ -79,6 +79,8 @@ go.uber.org/mock v0.6.0 h1:hyF9dfmbgIX5EfOdasqLsWD6xqpNZlXblLB/Dbnwv3Y=
 go.uber.org/mock v0.6.0/go.mod h1:KiVJ4BqZJaMj4svdfmHM0AUx4NJYO8ZNpPnZn1Z+BBU=
 golang.org/x/net v0.51.0 h1:94R/GTO7mt3/4wIKpcR5gkGmRLOuE/2hNGeWq/GBIFo=
 golang.org/x/net v0.51.0/go.mod h1:aamm+2QF5ogm02fjy5Bb7CQ0WMt1/WVM7FtyaTLlA9Y=
+golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4=
+golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
 golang.org/x/sys v0.41.0 h1:Ivj+2Cp/ylzLiEU89QhWblYnOE9zerudt9Ftecq2C6k=
 golang.org/x/sys v0.41.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
 golang.org/x/text v0.34.0 h1:oL/Qq0Kdaqxa1KbNeMKwQq0reLCCaFtqu2eNuSeNHbk=

internal/processor/processor.go

@@ -9,7 +9,6 @@ import (
 	"net/http"
 	"slices"
 	"strconv"
-	"sync"
 	"time"
 
 	"github.com/matt-gp/otel-lgtm-proxy/internal/config"
@@ -26,6 +25,7 @@ import (
 	metricpb "go.opentelemetry.io/proto/otlp/metrics/v1"
 	resourcepb "go.opentelemetry.io/proto/otlp/resource/v1"
 	tracepb "go.opentelemetry.io/proto/otlp/trace/v1"
+	"golang.org/x/sync/errgroup"
 )
 
 // Client is an interface for making HTTP requests.
@@ -197,7 +197,7 @@ func (p *Processor[T]) Partition(ctx context.Context, resources []T) map[string]
 
 // Dispatch sends all the requests to the target.
 func (p *Processor[T]) Dispatch(ctx context.Context, tenantMap map[string][]T) error {
-	wg := sync.WaitGroup{}
+	errGroup, ctx := errgroup.WithContext(ctx)
 
 	for tenant, resources := range tenantMap {
 		ctx, span := p.tracer.Start(
@@ -210,13 +210,8 @@ func (p *Processor[T]) Dispatch(ctx context.Context, tenantMap map[string][]T) e
 		)
 		defer span.End()
 
-		wg.Add(1)
-
-		go func(tenant string, resources []T) {
-			defer wg.Done()
-
+		errGroup.Go(func() error {
 			tenantAttribute := attribute.String("signal.tenant", tenant)
-
 			resp, err := p.send(ctx, tenant, resources)
 			if err != nil {
 				p.proxyRecordsMetricAdd(ctx, tenant, int64(len(resources)), nil)
@@ -230,7 +225,7 @@ func (p *Processor[T]) Dispatch(ctx context.Context, tenantMap map[string][]T) e
 
 				span.RecordError(err)
 				span.SetStatus(codes.Error, "failed to send")
-				return
+				return err
 			}
 
 			signalResponseStatusCodeAttr := attribute.String(
@@ -251,6 +246,20 @@ func (p *Processor[T]) Dispatch(ctx context.Context, tenantMap map[string][]T) e
 				[]attribute.KeyValue{signalResponseStatusCodeAttr},
 			)
 
+			if resp.StatusCode >= http.StatusBadRequest {
+				logger.Error(
+					ctx,
+					p.logger,
+					fmt.Sprintf("received non-success status code: %d", resp.StatusCode),
+					p.signalTypeLogAttr(),
+					log.KeyValueFromAttribute(tenantAttribute),
+					log.KeyValueFromAttribute(signalResponseStatusCodeAttr),
+				)
+
+				span.SetStatus(codes.Error, fmt.Sprintf("received non-success status code: %d", resp.StatusCode))
+				return fmt.Errorf("received non-success status code: %d", resp.StatusCode)
+			}
+
 			logger.Debug(
 				ctx,
 				p.logger,
@@ -273,12 +282,11 @@ func (p *Processor[T]) Dispatch(ctx context.Context, tenantMap map[string][]T) e
 			)
 
 			span.SetStatus(codes.Ok, "sent successfully")
-		}(tenant, resources)
+			return nil
+		})
 	}
 
-	wg.Wait()
-
-	return nil
+	return errGroup.Wait()
 }
 
 // send sends an individual request to the target.

internal/processor/processor_test.go

@@ -580,6 +580,126 @@ func TestDispatch(t *testing.T) {
 			},
 			wantErr: false,
 		},
+		{
+			name: "http 400 bad request should return error",
+			tenantMap: map[string][]*logpb.ResourceLogs{
+				"tenant-a": {
+					{
+						Resource: &resourcepb.Resource{
+							Attributes: []*commonpb.KeyValue{
+								{Key: "tenant.id", Value: &commonpb.AnyValue{Value: &commonpb.AnyValue_StringValue{StringValue: "tenant-a"}}},
+							},
+						},
+					},
+				},
+			},
+			mockResponses: []struct {
+				statusCode int
+				body       string
+				err        error
+			}{
+				{statusCode: http.StatusBadRequest, body: "bad request", err: nil},
+			},
+			wantErr: true,
+		},
+		{
+			name: "http 404 not found should return error",
+			tenantMap: map[string][]*logpb.ResourceLogs{
+				"tenant-a": {
+					{
+						Resource: &resourcepb.Resource{
+							Attributes: []*commonpb.KeyValue{
+								{Key: "tenant.id", Value: &commonpb.AnyValue{Value: &commonpb.AnyValue_StringValue{StringValue: "tenant-a"}}},
+							},
+						},
+					},
+				},
+			},
+			mockResponses: []struct {
+				statusCode int
+				body       string
+				err        error
+			}{
+				{statusCode: http.StatusNotFound, body: "not found", err: nil},
+			},
+			wantErr: true,
+		},
+		{
+			name: "http 500 internal server error should return error",
+			tenantMap: map[string][]*logpb.ResourceLogs{
+				"tenant-a": {
+					{
+						Resource: &resourcepb.Resource{
+							Attributes: []*commonpb.KeyValue{
+								{Key: "tenant.id", Value: &commonpb.AnyValue{Value: &commonpb.AnyValue_StringValue{StringValue: "tenant-a"}}},
+							},
+						},
+					},
+				},
+			},
+			mockResponses: []struct {
+				statusCode int
+				body       string
+				err        error
+			}{
+				{statusCode: http.StatusInternalServerError, body: "server error", err: nil},
+			},
+			wantErr: true,
+		},
+		{
+			name: "http 503 service unavailable should return error",
+			tenantMap: map[string][]*logpb.ResourceLogs{
+				"tenant-a": {
+					{
+						Resource: &resourcepb.Resource{
+							Attributes: []*commonpb.KeyValue{
+								{Key: "tenant.id", Value: &commonpb.AnyValue{Value: &commonpb.AnyValue_StringValue{StringValue: "tenant-a"}}},
+							},
+						},
+					},
+				},
+			},
+			mockResponses: []struct {
+				statusCode int
+				body       string
+				err        error
+			}{
+				{statusCode: http.StatusServiceUnavailable, body: "service unavailable", err: nil},
+			},
+			wantErr: true,
+		},
+		{
+			name: "multiple tenants with one failing should return error",
+			tenantMap: map[string][]*logpb.ResourceLogs{
+				"tenant-a": {
+					{
+						Resource: &resourcepb.Resource{
+							Attributes: []*commonpb.KeyValue{
+								{Key: "tenant.id", Value: &commonpb.AnyValue{Value: &commonpb.AnyValue_StringValue{StringValue: "tenant-a"}}},
+							},
+						},
+					},
+				},
+				"tenant-b": {
+					{
+						Resource: &resourcepb.Resource{
+							Attributes: []*commonpb.KeyValue{
+								{Key: "tenant.id", Value: &commonpb.AnyValue{Value: &commonpb.AnyValue_StringValue{StringValue: "tenant-b"}}},
+							},
+						},
+					},
+				},
+			},
+			mockResponses: []struct {
+				statusCode int
+				body       string
+				err        error
+			}{
+				{statusCode: http.StatusOK, body: "ok", err: nil},
+				{statusCode: http.StatusBadRequest, body: "bad request", err: nil},
+			},
+			wantErr: true,
+		},
 	}
 
 	for _, tt := range tests {