fix: upstream error propagation

api/http.go

@@ -1,21 +1,37 @@
 package api
 
 import (
+	"errors"
+	"fmt"
 	"net/http"
 
 	"github.com/flanksource/commons/logger"
 	"github.com/labstack/echo/v4"
 )
 
 type HTTPError struct {
-	Error   string `json:"error"`
+	Err     string `json:"error"`
 	Message string `json:"message,omitempty"`
 
 	// Data for machine-machine communication.
 	// usually contains a JSON data.
 	Data string `json:"data,omitempty"`
 }
 
+// Error implements the error interface. Not used by the application otherwise.
+func (e *HTTPError) Error() string {
+	return fmt.Sprintf("error: message=%s data=%s ", e.Message, e.Data)
+}
+
+func HTTPErrorFromErr(err error) *HTTPError {
+	var e *HTTPError
+	if errors.As(err, &e) {
+		return e
+	}
+
+	return nil
+}
+
 type HTTPSuccess struct {
 	Message string `json:"message"`
 	Payload any    `json:"payload,omitempty"`
@@ -28,7 +44,7 @@ func WriteError(c echo.Context, err error) error {
 		logger.WithValues("code", code, "error", message).Errorf(debugInfo)
 	}
 
-	return c.JSON(ErrorStatusCode(code), &HTTPError{Error: message, Data: data})
+	return c.JSON(ErrorStatusCode(code), &HTTPError{Err: message, Data: data})
 }
 
 // ErrorStatusCode returns the associated HTTP status code for an application error code.

upstream/client.go

@@ -109,9 +109,9 @@ func (t *UpstreamClient) push(ctx context.Context, method string, msg *PushData)
 		histogram.Label(StatusLabel, StatusError).Since(start)
 		respBody, _ := io.ReadAll(resp.Body)
 
-		var upstreamError api.Error
-		if json.Unmarshal(respBody, &upstreamError) == nil {
-			return &upstreamError
+		var httpErr api.HTTPError
+		if json.Unmarshal(respBody, &httpErr) == nil {
+			return &httpErr
 		}
 
 		return fmt.Errorf("upstream server returned error status[%d]: %s", resp.StatusCode, parseResponse(string(respBody)))

upstream/commands.go

@@ -258,7 +258,10 @@ func handleUpsertError(ctx context.Context, items []models.ExtendedDBTable, err
 		}
 	}
 
-	return api.Errorf(api.ECONFLICT, "foreign key error").WithData(PushFKError{IDs: lo.Uniq(conflicted)})
+	conflicted = lo.Uniq(conflicted)
+	return api.Errorf(api.ECONFLICT, "foreign key error").
+		WithData(PushFKError{IDs: conflicted}).
+		WithDebugInfo("foreign key error for %d items", len(conflicted))
 }
 
 func UpdateAgentLastSeen(ctx context.Context, id uuid.UUID) error {

upstream/controllers.go

@@ -38,7 +38,7 @@ func AgentAuthMiddleware(agentCache *cache.Cache) func(echo.HandlerFunc) echo.Ha
 			agentName := c.QueryParam(AgentNameQueryParam)
 			if agentName == "" {
 				histogram.Label(StatusLabel, StatusAgentError)
-				return c.JSON(http.StatusBadRequest, api.HTTPError{Error: "agent name is required"})
+				return c.JSON(http.StatusBadRequest, api.HTTPError{Err: "agent name is required"})
 			}
 
 			var agent *models.Agent
@@ -50,7 +50,7 @@ func AgentAuthMiddleware(agentCache *cache.Cache) func(echo.HandlerFunc) echo.Ha
 				if err != nil {
 					histogram.Label(StatusLabel, StatusAgentError)
 					return c.JSON(http.StatusBadRequest, api.HTTPError{
-						Error: fmt.Errorf("failed to create/fetch agent: %w", err).Error(),
+						Err: fmt.Errorf("failed to create/fetch agent: %w", err).Error(),
 					})
 				}
 
@@ -78,7 +78,7 @@ func PushHandler(c echo.Context) error {
 	err := json.NewDecoder(c.Request().Body).Decode(&req)
 	if err != nil {
 		histogram.Label(StatusLabel, StatusAgentError)
-		return c.JSON(http.StatusBadRequest, api.HTTPError{Error: err.Error(), Message: "invalid json request"})
+		return c.JSON(http.StatusBadRequest, api.HTTPError{Err: err.Error(), Message: "invalid json request"})
 	}
 
 	ctx.GetSpan().SetAttributes(attribute.Int("count", req.Count()))
@@ -112,7 +112,7 @@ func DeleteHandler(c echo.Context) error {
 	histogram := ctx.Histogram("push_queue_delete_handler", context.LatencyBuckets, StatusLabel, "", AgentLabel, "")
 	if err != nil {
 		histogram.Label(StatusLabel, StatusAgentError).Since(start)
-		return c.JSON(http.StatusBadRequest, api.HTTPError{Error: err.Error(), Message: "invalid json request"})
+		return c.JSON(http.StatusBadRequest, api.HTTPError{Err: err.Error(), Message: "invalid json request"})
 	}
 
 	ctx.GetSpan().SetAttributes(attribute.String("action", "delete"), attribute.Int("upstream.push.msg-count", req.Count()))
@@ -124,7 +124,7 @@ func DeleteHandler(c echo.Context) error {
 	ctx.Logger.V(3).Infof("Deleting push data %s", req.String())
 	if err := DeleteOnUpstream(ctx, &req); err != nil {
 		histogram.Label(StatusLabel, "error").Since(start)
-		return c.JSON(http.StatusInternalServerError, api.HTTPError{Error: err.Error(), Message: "failed to upsert upstream message"})
+		return c.JSON(http.StatusInternalServerError, api.HTTPError{Err: err.Error(), Message: "failed to upsert upstream message"})
 	}
 
 	histogram.Label(StatusLabel, StatusOK).Since(start)

upstream/jobs.go

@@ -102,13 +102,13 @@ func reconcileTable(ctx context.Context, config UpstreamConfig, table pushableTa
 		ctx.Tracef("pushing %s %d to upstream", table.TableName(), len(items))
 		pushError := client.Push(ctx, NewPushData(items))
 		if pushError != nil {
-			apiErr := api.FromError(pushError)
-			if apiErr == nil || apiErr.Data == "" {
+			httpError := api.HTTPErrorFromErr(pushError)
+			if httpError == nil || httpError.Data == "" {
 				return 0, fmt.Errorf("failed to push %s to upstream: %w", table.TableName(), pushError)
 			}
 
 			var foreignKeyErr PushFKError
-			if err := json.Unmarshal([]byte(apiErr.Data), &foreignKeyErr); err != nil {
+			if err := json.Unmarshal([]byte(httpError.Data), &foreignKeyErr); err != nil {
 				return 0, fmt.Errorf("failed to push %s to upstream (could not decode api error: %w): %w", table.TableName(), err, pushError)
 			}