logstore: add EntryStats

Epic: none
Release note: none

Author: pav-kv

pkg/kv/kvserver/logstore/logstore.go

@@ -84,15 +84,28 @@ type RaftState struct {
 	ByteSize  int64
 }
 
+// EntryStats contains stats about the appended log slice.
+type EntryStats struct {
+	RegularEntries    int
+	RegularBytes      int64
+	SideloadedEntries int
+	SideloadedBytes   int64
+}
+
+// Add increments the stats with the given delta.
+func (e *EntryStats) Add(delta EntryStats) {
+	e.RegularEntries += delta.RegularEntries
+	e.RegularBytes += delta.RegularBytes
+	e.SideloadedEntries += delta.SideloadedEntries
+	e.SideloadedBytes += delta.SideloadedBytes
+}
+
 // AppendStats describes a completed log storage append operation.
 type AppendStats struct {
 	Begin crtime.Mono
 	End   crtime.Mono
 
-	RegularEntries    int
-	RegularBytes      int64
-	SideloadedEntries int
-	SideloadedBytes   int64
+	EntryStats
 
 	PebbleBegin crtime.Mono
 	PebbleEnd   crtime.Mono
@@ -186,22 +199,19 @@ func (s *LogStore) storeEntriesAndCommitBatch(
 		stats.Begin = crtime.NowMono()
 		// All of the entries are appended to distinct keys, returning a new
 		// last index.
-		thinEntries, numSideloaded, sideLoadedEntriesSize, otherEntriesSize, err := MaybeSideloadEntries(ctx, m.Entries, s.Sideload)
+		thinEntries, entryStats, err := MaybeSideloadEntries(ctx, m.Entries, s.Sideload)
 		if err != nil {
 			const expl = "during sideloading"
 			return RaftState{}, errors.Wrap(err, expl)
 		}
-		state.ByteSize += sideLoadedEntriesSize
+		stats.EntryStats.Add(entryStats) // TODO(pav-kv): just return the stats.
+		state.ByteSize += entryStats.SideloadedBytes
 		if state, err = logAppend(
 			ctx, s.StateLoader.RaftLogPrefix(), batch, state, thinEntries,
 		); err != nil {
 			const expl = "during append"
 			return RaftState{}, errors.Wrap(err, expl)
 		}
-		stats.RegularEntries += len(thinEntries) - numSideloaded
-		stats.RegularBytes += otherEntriesSize
-		stats.SideloadedEntries += numSideloaded
-		stats.SideloadedBytes += sideLoadedEntriesSize
 		stats.End = crtime.NowMono()
 	}
 

pkg/kv/kvserver/logstore/sideload.go

@@ -69,21 +69,17 @@ type SideloadStorage interface {
 // in parts or entirely by the same memory.
 func MaybeSideloadEntries(
 	ctx context.Context, input []raftpb.Entry, sideloaded SideloadStorage,
-) (
-	_ []raftpb.Entry,
-	numSideloaded int,
-	sideloadedEntriesSize int64,
-	otherEntriesSize int64,
-	_ error,
-) {
+) ([]raftpb.Entry, EntryStats, error) {
+	var stats EntryStats
 	var output []raftpb.Entry
 	for i := range input {
 		typ, pri, err := raftlog.EncodingOf(input[i])
 		if err != nil {
-			return nil, 0, 0, 0, err
+			return nil, EntryStats{}, err
 		}
 		if !typ.IsSideloaded() {
-			otherEntriesSize += int64(len(input[i].Data))
+			stats.RegularEntries++
+			stats.RegularBytes += int64(len(input[i].Data))
 			continue
 		}
 
@@ -100,7 +96,7 @@ func MaybeSideloadEntries(
 		// Unmarshal the command into an object that we can mutate.
 		e, err := raftlog.NewEntry(input[i])
 		if err != nil {
-			return nil, 0, 0, 0, err
+			return nil, EntryStats{}, err
 		}
 		if e.Cmd.ReplicatedEvalResult.AddSSTable == nil {
 			// Still no AddSSTable; someone must've proposed a v2 command
@@ -109,7 +105,6 @@ func MaybeSideloadEntries(
 			log.Warning(ctx, "encountered sideloaded Raft command without inlined payload")
 			continue
 		}
-		numSideloaded++
 
 		// Actually strip the command.
 		dataToSideload := e.Cmd.ReplicatedEvalResult.AddSSTable.Data
@@ -123,28 +118,30 @@ func MaybeSideloadEntries(
 			raftlog.EncodeRaftCommandPrefix(data[:raftlog.RaftCommandPrefixLen], typ, e.ID, pri)
 			_, err := protoutil.MarshalToSizedBuffer(&e.Cmd, data[raftlog.RaftCommandPrefixLen:])
 			if err != nil {
-				return nil, 0, 0, 0, errors.Wrap(err, "while marshaling stripped sideloaded command")
+				return nil, EntryStats{}, errors.Wrap(err, "while marshaling stripped sideloaded command")
 			}
 			outputEnt.Data = data
 		}
 
 		log.Eventf(ctx, "writing payload at index=%d term=%d", outputEnt.Index, outputEnt.Term)
 		if err := sideloaded.Put(ctx, kvpb.RaftIndex(outputEnt.Index), kvpb.RaftTerm(outputEnt.Term), dataToSideload); err != nil { // TODO could verify checksum here
-			return nil, 0, 0, 0, err
+			return nil, EntryStats{}, err
 		}
-		sideloadedEntriesSize += int64(len(dataToSideload))
+
+		stats.SideloadedEntries++
+		stats.SideloadedBytes += int64(len(dataToSideload))
 	}
 
 	if output != nil { // there is at least one sideloaded command
 		// Sync the sideloaded storage directory so that the commands are durable.
 		if err := sideloaded.Sync(); err != nil {
-			return nil, 0, 0, 0, err
+			return nil, EntryStats{}, err
 		}
 	} else { // we never saw a sideloaded command
 		output = input
 	}
 
-	return output, numSideloaded, sideloadedEntriesSize, otherEntriesSize, nil
+	return output, stats, nil
 }
 
 // MaybeInlineSideloadedRaftCommand takes an entry and inspects it. If its

pkg/kv/kvserver/logstore/sideload_test.go

@@ -520,22 +520,22 @@ func TestRaftSSTableSideloadingSideload(t *testing.T) {
 			eng := storage.NewDefaultInMemForTesting()
 			defer eng.Close()
 			sideloaded := newTestingSideloadStorage(eng)
-			postEnts, numSideloaded, size, nonSideloadedSize, err := MaybeSideloadEntries(ctx, test.preEnts, sideloaded)
+			postEnts, stats, err := MaybeSideloadEntries(ctx, test.preEnts, sideloaded)
 			if err != nil {
 				t.Fatal(err)
 			}
 			if len(addSST.Data) == 0 {
 				t.Fatal("invocation mutated original AddSSTable struct in memory")
 			}
-			require.Equal(t, test.nonSideloadedSize, nonSideloadedSize)
+			require.Equal(t, test.nonSideloadedSize, stats.RegularBytes)
 			var expNumSideloaded int
 			if test.size > 0 {
 				expNumSideloaded = 1
 			}
-			require.Equal(t, expNumSideloaded, numSideloaded)
+			require.Equal(t, expNumSideloaded, stats.SideloadedEntries)
 			require.Equal(t, test.postEnts, postEnts)
-			if test.size != size {
-				t.Fatalf("expected %d sideloadedSize, but found %d", test.size, size)
+			if test.size != stats.SideloadedBytes {
+				t.Fatalf("expected %d sideloadedSize, but found %d", test.size, stats.SideloadedBytes)
 			}
 			actKeys, err := sideloaded.eng.Env().List(sideloaded.Dir())
 			if oserror.IsNotExist(err) {

pkg/kv/kvserver/replica_raft_test.go

@@ -107,15 +107,17 @@ func Test_handleRaftReadyStats_SafeFormat(t *testing.T) {
 			numConfChangeEntries: 6,
 		},
 		append: logstore.AppendStats{
-			Begin:             ts(2),
-			End:               ts(3),
-			RegularEntries:    7,
-			RegularBytes:      1024,
-			SideloadedEntries: 3,
-			SideloadedBytes:   5 * (1 << 20),
-			PebbleBegin:       ts(3),
-			PebbleEnd:         ts(4),
-			PebbleBytes:       1024 * 5,
+			Begin: ts(2),
+			End:   ts(3),
+			EntryStats: logstore.EntryStats{
+				RegularEntries:    7,
+				RegularBytes:      1024,
+				SideloadedEntries: 3,
+				SideloadedBytes:   5 * (1 << 20),
+			},
+			PebbleBegin: ts(3),
+			PebbleEnd:   ts(4),
+			PebbleBytes: 1024 * 5,
 			PebbleCommitStats: storage.BatchCommitStats{
 				BatchCommitStats: pebble.BatchCommitStats{
 					TotalDuration:               100 * time.Millisecond,