Return field type errors as client write errors

Fixes #2849

CHANGELOG.md

@@ -3,6 +3,8 @@
 ### Bugfixes
 
 - [#2869](https://github.com/influxdb/influxdb/issues/2869): Adding field to existing measurement causes panic
+- [#2849](https://github.com/influxdb/influxdb/issues/2849): RC32: Frequent write errors
+
 
 ## v0.9.0-rc33 [2015-06-09]
 

cluster/points_writer.go

@@ -2,6 +2,7 @@ package cluster
 
 import (
 	"errors"
+	"fmt"
 	"log"
 	"os"
 	"strings"
@@ -269,6 +270,7 @@ func (w *PointsWriter) writeToShard(shard *meta.ShardInfo, database, retentionPo
 
 	var wrote int
 	timeout := time.After(DefaultWriteTimeout)
+	var writeError error
 	for _, nodeID := range shard.OwnerIDs {
 		select {
 		case <-w.closing:
@@ -280,6 +282,11 @@ func (w *PointsWriter) writeToShard(shard *meta.ShardInfo, database, retentionPo
 			// If the write returned an error, continue to the next response
 			if err != nil {
 				w.Logger.Printf("write failed for shard %d on node %d: %v", shard.ID, nodeID, err)
+
+				// Keep track of the first error we see to return back to the client
+				if writeError == nil {
+					writeError = err
+				}
 				continue
 			}
 
@@ -296,5 +303,9 @@ func (w *PointsWriter) writeToShard(shard *meta.ShardInfo, database, retentionPo
 		return ErrPartialWrite
 	}
 
+	if writeError != nil {
+		return fmt.Errorf("write failed: %v", writeError)
+	}
+
 	return ErrWriteFailed
 }

cluster/points_writer_test.go

@@ -130,7 +130,7 @@ func TestPointsWriter_WritePoints(t *testing.T) {
 			name:        "write one error",
 			consistency: cluster.ConsistencyLevelOne,
 			err:         []error{fmt.Errorf("a failure"), fmt.Errorf("a failure"), fmt.Errorf("a failure")},
-			expErr:      cluster.ErrWriteFailed,
+			expErr:      fmt.Errorf("write failed: a failure"),
 		},
 
 		// Consistency any
@@ -185,7 +185,7 @@ func TestPointsWriter_WritePoints(t *testing.T) {
 			name:        "no writes succeed",
 			consistency: cluster.ConsistencyLevelOne,
 			err:         []error{fmt.Errorf("a failure"), fmt.Errorf("a failure"), fmt.Errorf("a failure")},
-			expErr:      cluster.ErrWriteFailed,
+			expErr:      fmt.Errorf("write failed: a failure"),
 		},
 
 		// Hinted handoff w/ ANY
@@ -257,7 +257,15 @@ func TestPointsWriter_WritePoints(t *testing.T) {
 		c.TSDBStore = store
 		c.HintedHandoff = hh
 
-		if err := c.WritePoints(pr); err != test.expErr {
+		err := c.WritePoints(pr)
+		if err == nil && test.expErr != nil {
+			t.Errorf("PointsWriter.WritePoints(): '%s' error: got %v, exp %v", test.name, err, test.expErr)
+		}
+
+		if err != nil && test.expErr == nil {
+			t.Errorf("PointsWriter.WritePoints(): '%s' error: got %v, exp %v", test.name, err, test.expErr)
+		}
+		if err != nil && test.expErr != nil && err.Error() != test.expErr.Error() {
 			t.Errorf("PointsWriter.WritePoints(): '%s' error: got %v, exp %v", test.name, err, test.expErr)
 		}
 	}

errors.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"runtime"
+	"strings"
 )
 
 var (
@@ -31,6 +32,10 @@ func Errorf(format string, a ...interface{}) (err error) {
 
 // IsClientError indicates whether an error is a known client error.
 func IsClientError(err error) bool {
+	if err == nil {
+		return false
+	}
+
 	if err == ErrFieldsRequired {
 		return true
 	}
@@ -38,6 +43,10 @@ func IsClientError(err error) bool {
 		return true
 	}
 
+	if strings.Contains(err.Error(), ErrFieldTypeConflict.Error()) {
+		return true
+	}
+
 	return false
 }
 

tsdb/shard.go

@@ -366,7 +366,7 @@ func (s *Shard) validateSeriesAndFields(points []Point) ([]*seriesCreate, []*fie
 			if f := mf.Fields[name]; f != nil {
 				// Field present in Metastore, make sure there is no type conflict.
 				if f.Type != influxql.InspectDataType(value) {
-					return nil, nil, fmt.Errorf("input field \"%s\" is type %T, already exists as type %s", name, value, f.Type)
+					return nil, nil, fmt.Errorf("field type conflict: input field \"%s\" is type %T, already exists as type %s", name, value, f.Type)
 				}
 
 				continue // Field is present, and it's of the same type. Nothing more to do.