Merge #143863 #143864 #143937 #144040

143863: raft: lift max size from raftLog r=tbg a=pav-kv

There is no more built-in flow control in the `raftLog` type. This commit moves the final piece of it out. There is only one place where this limit is still used: `hasUnappliedConfChanges` check scans the log when campaigning.

Related to #143576

143864: logstore: rm MaybeInlineSideloadedRaftCommand alloc r=tbg a=pav-kv

This change makes `MaybeInlineSideloadedRaftCommand` use the single-entry `raftentry.Cache.Get()` instead of a scan, and eliminates unnecessary allocations in this path.

Epic: none
Release note: none

143937: raft: fix BenchmarkRawNode r=tbg a=pav-kv

Epic: none
Release note: none

144040: roachtest/tpce: openmetrics bytes emitting wrong new line character r=csgourav a=sambhav-jain-16

This commit addresses an issue in the TPC-E roachtest where the OpenMetrics bytes were emitting incorrect newline characters.

Epic: none

Release note: None

Co-authored-by: Pavel Kalinnikov <[email protected]>
Co-authored-by: Sambhav Jain <[email protected]>

Author: 3 people

pkg/cmd/roachtest/tests/tpce.go

@@ -400,7 +400,7 @@ func GetTpceOpenmetricsBytes(
 	buffer.WriteString("# HELP tpce_latency Latency metrics for TPC-E transactions\n")
 
 	latencyString := func(quantile, latency string) string {
-		return fmt.Sprintf(`tpce_latency{%s,unit="ms",is_higher_better="false",quantile="%s"} %s %d\n`, labelString, quantile, latency, now)
+		return fmt.Sprintf("tpce_latency{%s,unit=\"ms\",is_higher_better=\"false\",quantile=\"%s\"} %s %d\n", labelString, quantile, latency, now)
 	}
 
 	buffer.WriteString(latencyString("0.5", metrics.P50Latency))
@@ -411,7 +411,7 @@ func GetTpceOpenmetricsBytes(
 	buffer.WriteString(fmt.Sprintf("tpce_latency_sum{%s} %d %d\n", labelString, 0, now))
 	buffer.WriteString(fmt.Sprintf("tpce_latency_count{%s} %s %d\n", labelString, countOfLatencies, now))
 	buffer.WriteString("# TYPE tpce_avg_latency gauge\n")
-	buffer.WriteString(fmt.Sprintf(`tpce_avg_latency{%s,unit="ms",is_higher_better="false"} %s %d\n`, labelString, metrics.AvgLatency, now))
+	buffer.WriteString(fmt.Sprintf("tpce_avg_latency{%s,unit=\"ms\",is_higher_better=\"false\"} %s %d\n", labelString, metrics.AvgLatency, now))
 	buffer.WriteString("# EOF")
 
 	metricsBytes := buffer.Bytes()

pkg/kv/kvserver/logstore/logstore.go

@@ -675,20 +675,14 @@ func LoadEntries(
 		}
 		expectedIndex++
 
-		typ, _, err := raftlog.EncodingOf(ent)
-		if err != nil {
+		if typ, _, err := raftlog.EncodingOf(ent); err != nil {
 			return err
-		}
-		if typ.IsSideloaded() {
-			newEnt, err := MaybeInlineSideloadedRaftCommand(
+		} else if typ.IsSideloaded() {
+			if ent, err = MaybeInlineSideloadedRaftCommand(
 				ctx, rangeID, ent, sideloaded, eCache,
-			)
-			if err != nil {
+			); err != nil {
 				return err
 			}
-			if newEnt != nil {
-				ent = *newEnt
-			}
 		}
 
 		if sh.add(uint64(ent.Size())) {

pkg/kv/kvserver/logstore/sideload.go

@@ -149,8 +149,8 @@ func MaybeSideloadEntries(
 
 // MaybeInlineSideloadedRaftCommand takes an entry and inspects it. If its
 // command encoding version indicates a sideloaded entry, it uses the entryCache
-// or SideloadStorage to inline the payload, returning a new entry (which must
-// be treated as immutable by the caller) or nil (if inlining does not apply)
+// or SideloadStorage to inline the payload, and returns a new entry (which must
+// be treated as immutable by the caller).
 //
 // If a payload is missing, returns an error whose Cause() is
 // errSideloadedFileNotFound.
@@ -160,36 +160,24 @@ func MaybeInlineSideloadedRaftCommand(
 	ent raftpb.Entry,
 	sideloaded SideloadStorage,
 	entryCache *raftentry.Cache,
-) (*raftpb.Entry, error) {
+) (raftpb.Entry, error) {
 	typ, pri, err := raftlog.EncodingOf(ent)
-	if err != nil {
-		return nil, err
-	}
-	if !typ.IsSideloaded() {
-		return nil, nil
+	if err != nil || !typ.IsSideloaded() {
+		return ent, err
 	}
 	log.Event(ctx, "inlining sideloaded SSTable")
-	// We could unmarshal this yet again, but if it's committed we
-	// are very likely to have appended it recently, in which case
-	// we can save work.
-	cachedSingleton, _, _, _ := entryCache.Scan(
-		nil, rangeID, kvpb.RaftIndex(ent.Index), kvpb.RaftIndex(ent.Index+1), 1<<20,
-	)
-
-	if len(cachedSingleton) > 0 {
+	// We could unmarshal this yet again, but if it's committed we are very likely
+	// to have appended it recently, in which case we can save work.
+	if entry, hit := entryCache.Get(rangeID, kvpb.RaftIndex(ent.Index)); hit {
 		log.Event(ctx, "using cache hit")
-		return &cachedSingleton[0], nil
+		return entry, nil
 	}
 
-	// Make a shallow copy.
-	entCpy := ent
-	ent = entCpy
-
 	log.Event(ctx, "inlined entry not cached")
 	// (Bad) luck, for whatever reason the inlined proposal isn't in the cache.
 	e, err := raftlog.NewEntry(ent)
 	if err != nil {
-		return nil, err
+		return ent, err
 	}
 
 	if len(e.Cmd.ReplicatedEvalResult.AddSSTable.Data) > 0 {
@@ -201,12 +189,12 @@ func MaybeInlineSideloadedRaftCommand(
 		// be as a result of log entries that are very old, written
 		// when sending the log with snapshots was still possible).
 		log.Event(ctx, "entry already inlined")
-		return &ent, nil
+		return ent, nil
 	}
 
 	sideloadedData, err := sideloaded.Get(ctx, kvpb.RaftIndex(ent.Index), kvpb.RaftTerm(ent.Term))
 	if err != nil {
-		return nil, errors.Wrap(err, "loading sideloaded data")
+		return ent, errors.Wrap(err, "loading sideloaded data")
 	}
 	e.Cmd.ReplicatedEvalResult.AddSSTable.Data = sideloadedData
 	// TODO(tbg): there should be a helper that properly encodes a command, given
@@ -216,11 +204,11 @@ func MaybeInlineSideloadedRaftCommand(
 		raftlog.EncodeRaftCommandPrefix(data[:raftlog.RaftCommandPrefixLen], typ, e.ID, pri)
 		_, err := protoutil.MarshalToSizedBuffer(&e.Cmd, data[raftlog.RaftCommandPrefixLen:])
 		if err != nil {
-			return nil, err
+			return ent, err
 		}
 		ent.Data = data
 	}
-	return &ent, nil
+	return ent, nil
 }
 
 // AssertSideloadedRaftCommandInlined asserts that if the provided entry is a

pkg/kv/kvserver/logstore/sideload_test.go

@@ -13,7 +13,6 @@ import (
 	"os"
 	"path/filepath"
 	"reflect"
-	"regexp"
 	"sort"
 	"strconv"
 	"strings"
@@ -368,18 +367,6 @@ func TestRaftSSTableSideloadingInline(t *testing.T) {
 	v1, v2 := raftlog.EntryEncodingStandardWithAC, raftlog.EntryEncodingSideloadedWithAC
 	rangeID := roachpb.RangeID(1)
 
-	type testCase struct {
-		// Entry passed into maybeInlineSideloadedRaftCommand and the entry
-		// after having (perhaps) been modified.
-		thin, fat raftpb.Entry
-		// Populate the raft entry cache and sideload storage before running the test.
-		setup func(*raftentry.Cache, SideloadStorage)
-		// If nonempty, the error expected from maybeInlineSideloadedRaftCommand.
-		expErr string
-		// If nonempty, a regex that the recorded trace span must match.
-		expTrace string
-	}
-
 	sstFat := kvserverpb.ReplicatedEvalResult_AddSSTable{
 		Data:  []byte("foo"),
 		CRC32: 0, // not checked
@@ -388,91 +375,84 @@ func TestRaftSSTableSideloadingInline(t *testing.T) {
 		CRC32: 0, // not checked
 	}
 
-	putOnDisk := func(ec *raftentry.Cache, ss SideloadStorage) {
-		if err := ss.Put(context.Background(), 5, 6, sstFat.Data); err != nil {
-			t.Fatal(err)
-		}
+	putOnDisk := func(_ *raftentry.Cache, ss SideloadStorage) {
+		require.NoError(t, ss.Put(context.Background(), 5, 6, sstFat.Data))
 	}
 
-	testCases := map[string]testCase{
+	for _, test := range []struct {
+		name string
+		// Entry passed into maybeInlineSideloadedRaftCommand and the entry
+		// after having (perhaps) been modified.
+		thin, fat raftpb.Entry
+		// Populate the raft entry cache and sideload storage before running the test.
+		setup func(*raftentry.Cache, SideloadStorage)
+		// If nonempty, the error expected from maybeInlineSideloadedRaftCommand.
+		expErr string
+		// If nonempty, a regex that the recorded trace span must match.
+		expTrace string
+	}{
 		// Plain old v1 Raft command without payload. Don't touch.
-		"v1-no-payload": {thin: mkEnt(v1, 5, 6, &sstThin), fat: mkEnt(v1, 5, 6, &sstThin)},
+		{name: "v1-no-payload", thin: mkEnt(v1, 5, 6, &sstThin), fat: mkEnt(v1, 5, 6, &sstThin)},
 		// With payload, but command is v1. Don't touch. Note that the
 		// first of the two shouldn't happen in practice or we have a
 		// huge problem once we try to apply this entry.
-		"v1-slim-with-payload": {thin: mkEnt(v1, 5, 6, &sstThin), fat: mkEnt(v1, 5, 6, &sstThin)},
-		"v1-with-payload":      {thin: mkEnt(v1, 5, 6, &sstFat), fat: mkEnt(v1, 5, 6, &sstFat)},
+		{name: "v1-slim-with-payload", thin: mkEnt(v1, 5, 6, &sstThin), fat: mkEnt(v1, 5, 6, &sstThin)},
+		{name: "v1-with-payload", thin: mkEnt(v1, 5, 6, &sstFat), fat: mkEnt(v1, 5, 6, &sstFat)},
 		// v2 with payload, but payload is AWOL. This would be fatal in practice.
-		"v2-with-payload-missing-file": {
+		{
+			name: "v2-with-payload-missing-file",
 			thin: mkEnt(v2, 5, 6, &sstThin), fat: mkEnt(v2, 5, 6, &sstThin),
 			expErr: "not found",
 		},
 		// v2 with payload that's actually there. The request we'll see in
 		// practice.
-		"v2-with-payload-with-file-no-cache": {
+		{
+			name: "v2-with-payload-with-file-no-cache",
 			thin: mkEnt(v2, 5, 6, &sstThin), fat: mkEnt(v2, 5, 6, &sstFat),
 			setup: putOnDisk, expTrace: "inlined entry not cached",
 		},
-		"v2-with-payload-with-file-with-cache": {
+		{
+			name: "v2-with-payload-with-file-with-cache",
 			thin: mkEnt(v2, 5, 6, &sstThin), fat: mkEnt(v2, 5, 6, &sstFat),
 			setup: func(ec *raftentry.Cache, ss SideloadStorage) {
 				putOnDisk(ec, ss)
 				ec.Add(rangeID, []raftpb.Entry{mkEnt(v2, 5, 6, &sstFat)}, true)
 			}, expTrace: "using cache hit",
 		},
-		"v2-fat-without-file": {
+		{
+			name: "v2-fat-without-file",
 			thin: mkEnt(v2, 5, 6, &sstFat), fat: mkEnt(v2, 5, 6, &sstFat),
 			setup:    func(ec *raftentry.Cache, ss SideloadStorage) {},
 			expTrace: "already inlined",
 		},
-	}
-
-	runOne := func(k string, test testCase) {
-		ctx, getRecAndFinish := tracing.ContextWithRecordingSpan(
-			context.Background(), tracing.NewTracer(), "test-recording")
-		defer getRecAndFinish()
-
-		eng := storage.NewDefaultInMemForTesting()
-		defer eng.Close()
-		ss := newTestingSideloadStorage(eng)
-		ec := raftentry.NewCache(1024) // large enough
-		if test.setup != nil {
-			test.setup(ec, ss)
-		}
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			ctx, getRecAndFinish := tracing.ContextWithRecordingSpan(
+				context.Background(), tracing.NewTracer(), "test-recording")
+			defer getRecAndFinish()
 
-		thinCopy := *(protoutil.Clone(&test.thin).(*raftpb.Entry))
-		newEnt, err := MaybeInlineSideloadedRaftCommand(ctx, rangeID, thinCopy, ss, ec)
-		if err != nil {
-			if test.expErr == "" || !testutils.IsError(err, test.expErr) {
-				t.Fatalf("%s: %+v", k, err)
+			eng := storage.NewDefaultInMemForTesting()
+			defer eng.Close()
+			ss := newTestingSideloadStorage(eng)
+			ec := raftentry.NewCache(1024) // large enough
+			if test.setup != nil {
+				test.setup(ec, ss)
 			}
-		} else if test.expErr != "" {
-			t.Fatalf("%s: success, but expected error: %s", k, test.expErr)
-		} else {
-			mustEntryEq(t, thinCopy, test.thin)
-		}
-
-		if newEnt == nil {
-			newEnt = &thinCopy
-		}
-		mustEntryEq(t, *newEnt, test.fat)
 
-		if dump := getRecAndFinish().String(); test.expTrace != "" {
-			if ok, err := regexp.MatchString(test.expTrace, dump); err != nil {
-				t.Fatalf("%s: %+v", k, err)
-			} else if !ok {
-				t.Fatalf("%s: expected trace matching:\n%s\n\nbut got\n%s", k, test.expTrace, dump)
+			thinCopy := *(protoutil.Clone(&test.thin).(*raftpb.Entry))
+			newEnt, err := MaybeInlineSideloadedRaftCommand(ctx, rangeID, thinCopy, ss, ec)
+			if want := test.expErr; want != "" {
+				require.ErrorContains(t, err, want)
+			} else {
+				require.NoError(t, err)
+				mustEntryEq(t, thinCopy, test.thin)
 			}
-		}
-	}
+			mustEntryEq(t, newEnt, test.fat)
 
-	keys := make([]string, 0, len(testCases))
-	for k := range testCases {
-		keys = append(keys, k)
-	}
-	sort.Strings(keys)
-	for _, k := range keys {
-		runOne(k, testCases[k])
+			if want := test.expTrace; want != "" {
+				require.Contains(t, getRecAndFinish().String(), want)
+			}
+		})
 	}
 }
 

pkg/raft/log.go

@@ -83,36 +83,20 @@ type raftLog struct {
 	applied uint64
 
 	logger raftlogger.Logger
-
-	// maxApplyingEntsSize limits the outstanding byte size of the messages
-	// returned from calls to nextCommittedEnts that have not been acknowledged
-	// by a call to appliedTo.
-	maxApplyingEntsSize entryEncodingSize
 }
 
-// newLog returns log using the given storage and default options. It
-// recovers the log to the state that it just commits and applies the
-// latest snapshot.
+// newLog returns a raft log initialized to the state in the given storage.
 func newLog(storage Storage, logger raftlogger.Logger) *raftLog {
-	return newLogWithSize(storage, logger, noLimit)
-}
-
-// newLogWithSize returns a log using the given storage and max
-// message size.
-func newLogWithSize(
-	storage Storage, logger raftlogger.Logger, maxApplyingEntsSize entryEncodingSize,
-) *raftLog {
 	compacted, lastIndex := storage.Compacted(), storage.LastIndex()
 	lastTerm, err := storage.Term(lastIndex)
 	if err != nil {
 		panic(err) // TODO(pav-kv): the storage should always cache the last term.
 	}
 	last := entryID{term: lastTerm, index: lastIndex}
 	return &raftLog{
-		storage:             storage,
-		unstable:            newUnstable(last, logger),
-		termCache:           newTermCache(termCacheSize, last),
-		maxApplyingEntsSize: maxApplyingEntsSize,
+		storage:   storage,
+		unstable:  newUnstable(last, logger),
+		termCache: newTermCache(termCacheSize, last),
 
 		// Initialize our committed and applied pointers to the time of the last
 		// compaction.
@@ -292,7 +276,7 @@ func (l *raftLog) nextCommittedEnts(allowUnstable bool) (ents []pb.Entry) {
 	if span.Empty() {
 		return nil
 	}
-	ents, err := l.slice(uint64(span.After), uint64(span.Last), l.maxApplyingEntsSize)
+	ents, err := l.slice(uint64(span.After), uint64(span.Last), noLimit)
 	if err != nil {
 		l.logger.Panicf("unexpected error when getting unapplied entries (%v)", err)
 	}

pkg/raft/log_test.go

@@ -373,7 +373,6 @@ func TestNextCommittedEnts(t *testing.T) {
 }
 
 func TestAcceptApplying(t *testing.T) {
-	maxSize := entryEncodingSize(100)
 	snap := pb.Snapshot{
 		Metadata: pb.SnapshotMetadata{Term: 1, Index: 3},
 	}
@@ -398,7 +397,7 @@ func TestAcceptApplying(t *testing.T) {
 		t.Run("", func(t *testing.T) {
 			storage := NewMemoryStorage()
 			require.NoError(t, storage.ApplySnapshot(snap))
-			raftLog := newLogWithSize(storage, raftlogger.DiscardLogger, maxSize)
+			raftLog := newLog(storage, raftlogger.DiscardLogger)
 			require.True(t, raftLog.append(init))
 			require.NoError(t, storage.Append(init.sub(3, 4)))
 			raftLog.checkInvariants(t)