consumer: bound the size of an error stored as Etcd shard status

consumer/shard.go

@@ -225,9 +225,14 @@ func servePrimary(s *shard) (err error) {
 		if err != nil && s.ctx.Err() == nil {
 			log.WithFields(log.Fields{"err": err, "shard": s.FQN()}).Error("servePrimary failed")
 
+			var statusErr = err.Error()
+			if len(statusErr) > MAX_ETCD_ERR_LEN {
+				statusErr = statusErr[:MAX_ETCD_ERR_LEN] + " ...[truncated]"
+			}
+
 			updateStatusWithRetry(s, pc.ReplicaStatus{
 				Code:   pc.ReplicaStatus_FAILED,
-				Errors: []string{err.Error()},
+				Errors: []string{statusErr},
 			})
 		}
 		s.wg.Done()
@@ -422,3 +427,7 @@ func backoff(attempt int) time.Duration {
 		return 5 * time.Second
 	}
 }
+
+// Etcd values should generally be small.
+// We don't want to push arbitrary length error strings.
+const MAX_ETCD_ERR_LEN = 2048

consumer/shard_test.go

@@ -3,6 +3,8 @@ package consumer
 import (
 	"context"
 	"errors"
+	"fmt"
+	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -222,3 +224,18 @@ func TestShardStoreStartCommitFails(t *testing.T) {
 
 	tf.allocateShard(spec) // Cleanup.
 }
+
+func TestShardErrorsAreTruncatedIfLarge(t *testing.T) {
+	var tf, cleanup = newTestFixture(t)
+	defer cleanup()
+
+	tf.app.newStoreErr = errors.New(strings.Repeat("e", MAX_ETCD_ERR_LEN+1))
+	tf.allocateShard(makeShard(shardA), localID)
+
+	var expect = fmt.Sprintf("completeRecovery: app.NewStore: %s", strings.Repeat("e", MAX_ETCD_ERR_LEN))
+	expect = fmt.Sprintf("%s ...[truncated]", expect[:MAX_ETCD_ERR_LEN])
+
+	require.Equal(t, expect, expectStatusCode(t, tf.state, pc.ReplicaStatus_FAILED).Errors[0])
+
+	tf.allocateShard(makeShard(shardA)) // Cleanup.
+}