Integrated Storage (Raft): Add Support for max_entry_size Config

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 CHANGES:
 
+* storage/raft: The storage configuration now accepts a new `kv_max_value_size` config that will limit
+  the total size in bytes of any entry commited via raft. It defaults to `"1572864"` -- three times the chunking size. [[GH-9027](https://github.com/hashicorp/vault/pull/9027)]
 * token: Token creation with custom token ID via `id` will no longer allow periods (`.`) as part of the input string. 
   The final generated token value may contain periods, such as the `s.` prefix for service token 
   indication. [[GH-8646](https://github.com/hashicorp/vault/pull/8646/files)]

physical/raft/raft.go

@@ -52,6 +52,8 @@ var (
 	snapshotsRetained = 2
 
 	restoreOpDelayDuration = 5 * time.Second
+
+	defaultKVMaxValueSize = uint64(3 * raftchunking.ChunkSize)
 )
 
 // RaftBackend implements the backend interfaces and uses the raft protocol to
@@ -107,6 +109,10 @@ type RaftBackend struct {
 
 	// permitPool is used to limit the number of concurrent storage calls.
 	permitPool *physical.PermitPool
+
+	// kvMaxValueSize imposes a limit on values in individual KV operations. It is
+	// suggested to use a value of 3x the Raft chunking size for optimal performance.
+	kvMaxValueSize uint64
 }
 
 // LeaderJoinInfo contains information required by a node to join itself as a
@@ -226,6 +232,7 @@ func NewRaftBackend(conf map[string]string, logger log.Logger) (physical.Backend
 	var log raft.LogStore
 	var stable raft.StableStore
 	var snap raft.SnapshotStore
+
 	var devMode bool
 	if devMode {
 		store := raft.NewInmemStore()
@@ -303,16 +310,27 @@ func NewRaftBackend(conf map[string]string, logger log.Logger) (physical.Backend
 		}
 	}
 
+	kvMaxValueSize := defaultKVMaxValueSize
+	if kvMaxValueSizeCfg := conf["kv_max_value_size"]; len(kvMaxValueSizeCfg) != 0 {
+		i, err := strconv.Atoi(kvMaxValueSizeCfg)
+		if err != nil {
+			return nil, fmt.Errorf("failed to parse 'kv_max_value_size': %w", err)
+		}
+
+		kvMaxValueSize = uint64(i)
+	}
+
 	return &RaftBackend{
-		logger:      logger,
-		fsm:         fsm,
-		conf:        conf,
-		logStore:    log,
-		stableStore: stable,
-		snapStore:   snap,
-		dataDir:     path,
-		localID:     localID,
-		permitPool:  physical.NewPermitPool(physical.DefaultParallelOperations),
+		logger:         logger,
+		fsm:            fsm,
+		conf:           conf,
+		logStore:       log,
+		stableStore:    stable,
+		snapStore:      snap,
+		dataDir:        path,
+		localID:        localID,
+		permitPool:     physical.NewPermitPool(physical.DefaultParallelOperations),
+		kvMaxValueSize: kvMaxValueSize,
 	}, nil
 }
 
@@ -925,12 +943,31 @@ func (b *RaftBackend) Get(ctx context.Context, path string) (*physical.Entry, er
 	b.permitPool.Acquire()
 	defer b.permitPool.Release()
 
-	return b.fsm.Get(ctx, path)
+	entry, err := b.fsm.Get(ctx, path)
+	if entry != nil {
+		valueLen := len(entry.Value)
+		if uint64(valueLen) > b.kvMaxValueSize {
+			b.logger.Warn(
+				"retrieved entry value is too large; see https://www.vaultproject.io/docs/configuration/storage/raft#raft-parameters",
+				"size", valueLen, "suggested", b.kvMaxValueSize,
+			)
+		}
+	}
+
+	return entry, err
 }
 
-// Put inserts an entry in the log for the put operation
+// Put inserts an entry in the log for the put operation. It will return an
+// error if the value exceeds the configured kv_max_value_size or if the call to
+// applyLog fails.
 func (b *RaftBackend) Put(ctx context.Context, entry *physical.Entry) error {
 	defer metrics.MeasureSince([]string{"raft-storage", "put"}, time.Now())
+
+	valueLen := len(entry.Value)
+	if uint64(valueLen) > b.kvMaxValueSize {
+		return fmt.Errorf("%s; got %d bytes, max: %d bytes", physical.ErrValueTooLarge, valueLen, b.kvMaxValueSize)
+	}
+
 	command := &LogData{
 		Operations: []*LogOperation{
 			&LogOperation{
@@ -1004,6 +1041,13 @@ func (b *RaftBackend) applyLog(ctx context.Context, command *LogData) error {
 		return errors.New("raft storage backend is not initialized")
 	}
 
+	for _, op := range command.Operations {
+		valueLen := len(op.Value)
+		if uint64(valueLen) > b.kvMaxValueSize {
+			return fmt.Errorf("%s; got %d bytes, max: %d bytes", physical.ErrValueTooLarge, valueLen, b.kvMaxValueSize)
+		}
+	}
+
 	commandBytes, err := proto.Marshal(command)
 	if err != nil {
 		return err

physical/raft/raft_test.go

@@ -7,6 +7,7 @@ import (
 	fmt "fmt"
 	"io"
 	"io/ioutil"
+	"math/rand"
 	"os"
 	"path/filepath"
 	"testing"
@@ -203,6 +204,61 @@ func TestRaft_Backend(t *testing.T) {
 	physical.ExerciseBackend(t, b)
 }
 
+func TestRaft_Backend_LargeValue(t *testing.T) {
+	b, dir := getRaft(t, true, true)
+	defer os.RemoveAll(dir)
+
+	t.Helper()
+
+	value := make([]byte, defaultKVMaxValueSize+1)
+	rand.Read(value)
+	entry := &physical.Entry{Key: "foo", Value: value}
+
+	if err := b.Put(context.Background(), entry); err == nil {
+		t.Fatal("expected error for put entry")
+	}
+
+	out, err := b.Get(context.Background(), entry.Key)
+	if err != nil {
+		t.Fatalf("unexpected error after failed put: %v", err)
+	}
+	if out != nil {
+		t.Fatal("expected response entry to be nil after a failed put")
+	}
+}
+
+func TestRaft_TransactionalBackend_LargeValue(t *testing.T) {
+	b, dir := getRaft(t, true, true)
+	defer os.RemoveAll(dir)
+
+	t.Helper()
+
+	value := make([]byte, defaultKVMaxValueSize+1)
+	rand.Read(value)
+
+	txns := []*physical.TxnEntry{
+		&physical.TxnEntry{
+			Operation: physical.PutOperation,
+			Entry: &physical.Entry{
+				Key:   "foo",
+				Value: value,
+			},
+		},
+	}
+
+	if err := b.Transaction(context.Background(), txns); err == nil {
+		t.Fatal("expected error for transactions")
+	}
+
+	out, err := b.Get(context.Background(), txns[0].Entry.Key)
+	if err != nil {
+		t.Fatalf("unexpected error after failed put: %v", err)
+	}
+	if out != nil {
+		t.Fatal("expected response entry to be nil after a failed put")
+	}
+}
+
 func TestRaft_Backend_ListPrefix(t *testing.T) {
 	b, dir := getRaft(t, true, true)
 	defer os.RemoveAll(dir)

website/pages/docs/configuration/storage/raft.mdx

@@ -94,6 +94,16 @@ time. To use Raft for HA coordination users must also use Raft for storage.
   still need to be unsealed manually. See the section below that describes the
   parameters accepted by the `retry_join` stanza.
 
+- `kv_max_value_size` `(integer: 1572864)` - This configures the maximum number of
+  bytes for a value in a Raft entry. It applies to both Put operations and transactions.
+  Any value size in an entry or transaction exceeding this configuration value
+  will cause the respective operation to fail. Raft has a suggested max size of
+  data in a raft log entry. This is based on current architecture, default timing,
+  etc. Integrated storage also uses a chunk size that is the threshold used for
+  breaking a large value into chunks. By default, the chunk size is the same as
+  raft's max size log entry. The default value for this configuration is 1572864
+  -- three times the chunking size.
+
 ### `retry_join` stanza
 
 - `leader_api_addr` `(string: "")` - Address of a possible leader node.