fix(http): don't return 500 errors for partial writes

CHANGELOG.md

@@ -22,6 +22,7 @@ Replacement `tsi1` indexes will be automatically generated on startup for shards
 1. [20380](https://github.com/influxdata/influxdb/pull/20380): Remove duplication from task error messages.
 1. [20313](https://github.com/influxdata/influxdb/pull/20313): Automatically build `tsi1` indexes for shards that need it instead of falling back to `inmem`.
 1. [20313](https://github.com/influxdata/influxdb/pull/20313): Fix logging initialization for storage engine.
+1. [20442](https://github.com/influxdata/influxdb/pull/20442): Don't return 500 codes for partial write failures.
 
 ## v2.0.3 [2020-12-14]
 

http/legacy/write_handler.go

@@ -13,6 +13,7 @@ import (
 	"github.com/influxdata/influxdb/v2/kit/tracing"
 	kithttp "github.com/influxdata/influxdb/v2/kit/transport/http"
 	"github.com/influxdata/influxdb/v2/storage"
+	"github.com/influxdata/influxdb/v2/tsdb"
 	"go.uber.org/zap"
 )
 
@@ -141,6 +142,16 @@ func (h *WriteHandler) handleWrite(w http.ResponseWriter, r *http.Request) {
 	}
 
 	if err := h.PointsWriter.WritePoints(ctx, auth.OrgID, bucket.ID, parsed.Points); err != nil {
+		if partialErr, ok := err.(tsdb.PartialWriteError); ok {
+			h.HandleHTTPError(ctx, &influxdb.Error{
+				Code: influxdb.EUnprocessableEntity,
+				Op:   opWriteHandler,
+				Msg:  "failure writing points to database",
+				Err:  partialErr,
+			}, sw)
+			return
+		}
+
 		h.HandleHTTPError(ctx, &influxdb.Error{
 			Code: influxdb.EInternal,
 			Op:   opWriteHandler,

http/legacy/write_handler_test.go

@@ -19,6 +19,7 @@ import (
 	kithttp "github.com/influxdata/influxdb/v2/kit/transport/http"
 	"github.com/influxdata/influxdb/v2/models"
 	"github.com/influxdata/influxdb/v2/snowflake"
+	"github.com/influxdata/influxdb/v2/tsdb"
 	"github.com/stretchr/testify/assert"
 	"go.uber.org/zap/zaptest"
 )
@@ -187,6 +188,88 @@ func TestWriteHandler_BucketAndMappingExistsSpecificRP(t *testing.T) {
 	assert.Equal(t, "", w.Body.String())
 }
 
+func TestWriteHandler_PartialWrite(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	var (
+		// Mocked Services
+		eventRecorder  = mocks.NewMockEventRecorder(ctrl)
+		dbrpMappingSvc = mocks.NewMockDBRPMappingServiceV2(ctrl)
+		bucketService  = mocks.NewMockBucketService(ctrl)
+		pointsWriter   = mocks.NewMockPointsWriter(ctrl)
+
+		// Found Resources
+		orgID  = generator.ID()
+		bucket = &influxdb.Bucket{
+			ID:                  generator.ID(),
+			OrgID:               orgID,
+			Name:                "mydb/autogen",
+			RetentionPolicyName: "autogen",
+			RetentionPeriod:     72 * time.Hour,
+		}
+		mapping = &influxdb.DBRPMappingV2{
+			OrganizationID:  orgID,
+			BucketID:        bucket.ID,
+			Database:        "mydb",
+			RetentionPolicy: "autogen",
+			Default:         true,
+		}
+
+		lineProtocolBody = "m,t1=v1 f1=2 100"
+	)
+
+	findAutogenMapping := dbrpMappingSvc.
+		EXPECT().
+		FindMany(gomock.Any(), influxdb.DBRPMappingFilterV2{
+			OrgID:           &mapping.OrganizationID,
+			Database:        &mapping.Database,
+			RetentionPolicy: &mapping.RetentionPolicy,
+		}).Return([]*influxdb.DBRPMappingV2{mapping}, 1, nil)
+
+	findBucketByID := bucketService.
+		EXPECT().
+		FindBucketByID(gomock.Any(), bucket.ID).Return(bucket, nil)
+
+	points := parseLineProtocol(t, lineProtocolBody)
+	writePoints := pointsWriter.
+		EXPECT().
+		WritePoints(gomock.Any(), orgID, bucket.ID, pointsMatcher{points}).
+		Return(tsdb.PartialWriteError{Reason: "bad points", Dropped: 1})
+
+	recordWriteEvent := eventRecorder.EXPECT().
+		Record(gomock.Any(), gomock.Any())
+
+	gomock.InOrder(
+		findAutogenMapping,
+		findBucketByID,
+		writePoints,
+		recordWriteEvent,
+	)
+
+	perms := newPermissions(influxdb.WriteAction, influxdb.BucketsResourceType, &orgID, nil)
+	auth := newAuthorization(orgID, perms...)
+	ctx := pcontext.SetAuthorizer(context.Background(), auth)
+	r := newWriteRequest(ctx, lineProtocolBody)
+	params := r.URL.Query()
+	params.Set("db", "mydb")
+	params.Set("rp", "autogen")
+	r.URL.RawQuery = params.Encode()
+
+	handler := NewWriterHandler(&PointsWriterBackend{
+		HTTPErrorHandler:   DefaultErrorHandler,
+		Logger:             zaptest.NewLogger(t),
+		BucketService:      bucketService,
+		DBRPMappingService: dbrp.NewAuthorizedService(dbrpMappingSvc),
+		PointsWriter:       pointsWriter,
+		EventRecorder:      eventRecorder,
+	})
+	w := httptest.NewRecorder()
+	handler.ServeHTTP(w, r)
+	assert.Equal(t, http.StatusUnprocessableEntity, w.Code)
+	assert.Equal(t, `{"code":"unprocessable entity","message":"failure writing points to database: partial write: bad points dropped=1"}`, w.Body.String())
+}
+
 func TestWriteHandler_BucketAndMappingExistsNoPermissions(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()

http/write_handler.go

@@ -16,6 +16,7 @@ import (
 	kithttp "github.com/influxdata/influxdb/v2/kit/transport/http"
 	"github.com/influxdata/influxdb/v2/models"
 	"github.com/influxdata/influxdb/v2/storage"
+	"github.com/influxdata/influxdb/v2/tsdb"
 	"go.uber.org/zap"
 )
 
@@ -187,6 +188,16 @@ func (h *WriteHandler) handleWrite(w http.ResponseWriter, r *http.Request) {
 	requestBytes = parsed.RawSize
 
 	if err := h.PointsWriter.WritePoints(ctx, org.ID, bucket.ID, parsed.Points); err != nil {
+		if partialErr, ok := err.(tsdb.PartialWriteError); ok {
+			h.HandleHTTPError(ctx, &influxdb.Error{
+				Code: influxdb.EUnprocessableEntity,
+				Op:   opWriteHandler,
+				Msg:  "failure writing points to database",
+				Err:  partialErr,
+			}, sw)
+			return
+		}
+
 		h.HandleHTTPError(ctx, &influxdb.Error{
 			Code: influxdb.EInternal,
 			Op:   opWriteHandler,

http/write_handler_test.go

@@ -17,6 +17,7 @@ import (
 	kithttp "github.com/influxdata/influxdb/v2/kit/transport/http"
 	"github.com/influxdata/influxdb/v2/mock"
 	influxtesting "github.com/influxdata/influxdb/v2/testing"
+	"github.com/influxdata/influxdb/v2/tsdb"
 	"go.uber.org/zap/zaptest"
 )
 
@@ -124,6 +125,24 @@ func TestWriteHandler_handleWrite(t *testing.T) {
 				code: 204,
 			},
 		},
+		{
+			name: "partial write error is unprocessable",
+			request: request{
+				org:    "043e0780ee2b1000",
+				bucket: "04504b356e23b000",
+				body:   "m1,t1=v1 f1=1",
+				auth:   bucketWritePermission("043e0780ee2b1000", "04504b356e23b000"),
+			},
+			state: state{
+				org:      testOrg("043e0780ee2b1000"),
+				bucket:   testBucket("043e0780ee2b1000", "04504b356e23b000"),
+				writeErr: tsdb.PartialWriteError{Reason: "bad points", Dropped: 1},
+			},
+			wants: wants{
+				code: 422,
+				body: `{"code":"unprocessable entity","message":"failure writing points to database: partial write: bad points dropped=1"}`,
+			},
+		},
 		{
 			name: "points writer error is an internal error",
 			request: request{