Skip to content

Commit

Permalink
kv: Use strict types for common fields
Browse files Browse the repository at this point in the history
This PR introduces 3 new typed fields to the kvpb package: RaftTerm,
RaftIndex and LeaseAppliedIndex. These fields were previously just
unit64 throughout the code and this made the code harder to read and
risked incorrect conversions.

It moves internal_raft.proto into the kvserver package to make this
possible to stay in the kv package.

Epic: none

Release note: None
  • Loading branch information
andrewbaptist committed Apr 4, 2023
1 parent ef357d1 commit 0d49b3a
Show file tree
Hide file tree
Showing 112 changed files with 772 additions and 602 deletions.
1 change: 1 addition & 0 deletions pkg/cli/BUILD.bazel
Original file line number Diff line number Diff line change
Expand Up @@ -128,6 +128,7 @@ go_library(
"//pkg/kv/kvpb",
"//pkg/kv/kvserver",
"//pkg/kv/kvserver/gc",
"//pkg/kv/kvserver/kvserverpb",
"//pkg/kv/kvserver/kvstorage",
"//pkg/kv/kvserver/liveness/livenesspb",
"//pkg/kv/kvserver/loqrecovery",
Expand Down
21 changes: 12 additions & 9 deletions pkg/cli/debug_check_store.go
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,8 @@ import (

"github.com/cockroachdb/cockroach/pkg/cli/clierrorplus"
"github.com/cockroachdb/cockroach/pkg/keys"
"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvserverpb"
"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvstorage"
"github.com/cockroachdb/cockroach/pkg/kv/kvserver/rditer"
"github.com/cockroachdb/cockroach/pkg/kv/kvserver/stateloader"
Expand Down Expand Up @@ -80,11 +82,11 @@ func runDebugCheckStoreCmd(cmd *cobra.Command, args []string) error {
}

type replicaCheckInfo struct {
truncatedIndex uint64
appliedIndex uint64
firstIndex uint64
lastIndex uint64
committedIndex uint64
truncatedIndex kvpb.RaftIndex
appliedIndex kvpb.RaftIndex
firstIndex kvpb.RaftIndex
lastIndex kvpb.RaftIndex
committedIndex kvpb.RaftIndex
}

type checkInput struct {
Expand Down Expand Up @@ -251,21 +253,22 @@ func checkStoreRaftState(
if err := kv.Value.GetProto(&hs); err != nil {
return err
}
getReplicaInfo(rangeID).committedIndex = hs.Commit
getReplicaInfo(rangeID).committedIndex = kvpb.RaftIndex(hs.Commit)
case bytes.Equal(suffix, keys.LocalRaftTruncatedStateSuffix):
var trunc roachpb.RaftTruncatedState
var trunc kvserverpb.RaftTruncatedState
if err := kv.Value.GetProto(&trunc); err != nil {
return err
}
getReplicaInfo(rangeID).truncatedIndex = trunc.Index
case bytes.Equal(suffix, keys.LocalRangeAppliedStateSuffix):
var state enginepb.RangeAppliedState
var state kvserverpb.RangeAppliedState
if err := kv.Value.GetProto(&state); err != nil {
return err
}
getReplicaInfo(rangeID).appliedIndex = state.RaftAppliedIndex
case bytes.Equal(suffix, keys.LocalRaftLogSuffix):
_, index, err := encoding.DecodeUint64Ascending(detail)
_, uIndex, err := encoding.DecodeUint64Ascending(detail)
index := kvpb.RaftIndex(uIndex)
if err != nil {
return err
}
Expand Down
12 changes: 6 additions & 6 deletions pkg/keys/keys.go
Original file line number Diff line number Diff line change
Expand Up @@ -356,21 +356,21 @@ func RaftLogPrefix(rangeID roachpb.RangeID) roachpb.Key {
}

// RaftLogKey returns a system-local key for a Raft log entry.
func RaftLogKey(rangeID roachpb.RangeID, logIndex uint64) roachpb.Key {
func RaftLogKey(rangeID roachpb.RangeID, logIndex kvpb.RaftIndex) roachpb.Key {
return MakeRangeIDPrefixBuf(rangeID).RaftLogKey(logIndex)
}

// RaftLogKeyFromPrefix returns a system-local key for a Raft log entry, using
// the provided Raft log prefix.
func RaftLogKeyFromPrefix(raftLogPrefix []byte, logIndex uint64) roachpb.Key {
return encoding.EncodeUint64Ascending(raftLogPrefix, logIndex)
func RaftLogKeyFromPrefix(raftLogPrefix []byte, logIndex kvpb.RaftIndex) roachpb.Key {
return encoding.EncodeUint64Ascending(raftLogPrefix, uint64(logIndex))
}

// DecodeRaftLogKeyFromSuffix parses the suffix of a system-local key for a Raft
// log entry and returns the entry's log index.
func DecodeRaftLogKeyFromSuffix(raftLogSuffix []byte) (uint64, error) {
func DecodeRaftLogKeyFromSuffix(raftLogSuffix []byte) (kvpb.RaftIndex, error) {
_, logIndex, err := encoding.DecodeUint64Ascending(raftLogSuffix)
return logIndex, err
return kvpb.RaftIndex(logIndex), err
}

// RaftReplicaIDKey returns a system-local key for a RaftReplicaID.
Expand Down Expand Up @@ -1092,7 +1092,7 @@ func (b RangeIDPrefixBuf) RaftLogPrefix() roachpb.Key {
}

// RaftLogKey returns a system-local key for a Raft log entry.
func (b RangeIDPrefixBuf) RaftLogKey(logIndex uint64) roachpb.Key {
func (b RangeIDPrefixBuf) RaftLogKey(logIndex kvpb.RaftIndex) roachpb.Key {
return RaftLogKeyFromPrefix(b.RaftLogPrefix(), logIndex)
}

Expand Down
3 changes: 2 additions & 1 deletion pkg/keys/printer.go
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@ import (
"strings"
"unicode/utf8"

"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
"github.com/cockroachdb/cockroach/pkg/roachpb"
"github.com/cockroachdb/cockroach/pkg/util/encoding"
"github.com/cockroachdb/cockroach/pkg/util/uuid"
Expand Down Expand Up @@ -305,7 +306,7 @@ func raftLogKeyParse(rangeID roachpb.RangeID, input string) (string, roachpb.Key
if err != nil {
panic(err)
}
return "", RaftLogKey(rangeID, index)
return "", RaftLogKey(rangeID, kvpb.RaftIndex(index))
}

func raftLogKeyPrint(buf *redact.StringBuilder, key roachpb.Key) {
Expand Down
3 changes: 2 additions & 1 deletion pkg/keys/printer_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ import (

"github.com/cockroachdb/apd/v3"
"github.com/cockroachdb/cockroach/pkg/keys"
"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
"github.com/cockroachdb/cockroach/pkg/roachpb"
"github.com/cockroachdb/cockroach/pkg/util/bitarray"
"github.com/cockroachdb/cockroach/pkg/util/duration"
Expand Down Expand Up @@ -250,7 +251,7 @@ func TestPrettyPrint(t *testing.T) {

{keys.RaftHardStateKey(roachpb.RangeID(1000001)), "/Local/RangeID/1000001/u/RaftHardState", revertSupportUnknown},
{keys.RangeTombstoneKey(roachpb.RangeID(1000001)), "/Local/RangeID/1000001/u/RangeTombstone", revertSupportUnknown},
{keys.RaftLogKey(roachpb.RangeID(1000001), uint64(200001)), "/Local/RangeID/1000001/u/RaftLog/logIndex:200001", revertSupportUnknown},
{keys.RaftLogKey(roachpb.RangeID(1000001), kvpb.RaftIndex(200001)), "/Local/RangeID/1000001/u/RaftLog/logIndex:200001", revertSupportUnknown},
{keys.RangeLastReplicaGCTimestampKey(roachpb.RangeID(1000001)), "/Local/RangeID/1000001/u/RangeLastReplicaGCTimestamp", revertSupportUnknown},

{keys.MakeRangeKeyPrefix(roachpb.RKey(tenSysCodec.TablePrefix(42))), `/Local/Range/Table/42`, revertSupportUnknown},
Expand Down
6 changes: 3 additions & 3 deletions pkg/kv/kvpb/api.proto
Original file line number Diff line number Diff line change
Expand Up @@ -1462,7 +1462,7 @@ message TruncateLogRequest {
RequestHeader header = 1 [(gogoproto.nullable) = false, (gogoproto.embed) = true];

// Log entries < this index are to be discarded.
uint64 index = 2;
uint64 index = 2 [(gogoproto.casttype) = "RaftIndex"];

// RangeID is used to double check that the correct range is being truncated.
// The header specifies a span, start and end keys, but not the range id
Expand All @@ -1482,7 +1482,7 @@ message TruncateLogRequest {
// discarded from the raft log to be performed once, at the leaseholder.
//
// Populated starting at cluster version LooselyCoupledRaftLogTruncation.
uint64 expected_first_index = 4;
uint64 expected_first_index = 4 [(gogoproto.casttype) = "RaftIndex"];
}

// TruncateLogResponse is the response to a TruncateLog() operation.
Expand Down Expand Up @@ -2099,7 +2099,7 @@ message SubsumeResponse {
// provide the sender of the Subsume request with an upper bound on the lease
// applied index of the CPut that left an intent on the local copy of the
// right-hand range descriptor.
uint64 lease_applied_index = 4;
uint64 lease_applied_index = 4 [(gogoproto.casttype) = "LeaseAppliedIndex"];

// FreezeStart is a timestamp that is guaranteed to be greater than the
// timestamps at which any requests were serviced by the responding replica
Expand Down
22 changes: 22 additions & 0 deletions pkg/kv/kvpb/data.go
Original file line number Diff line number Diff line change
Expand Up @@ -151,3 +151,25 @@ func TransactionRefreshTimestamp(pErr *Error) (bool, hlc.Timestamp) {
}
return true, timestamp
}

// LeaseAppliedIndex is attached to every Raft message and is used for replay
// protection.
type LeaseAppliedIndex uint64

func (s LeaseAppliedIndex) SafeValue() {}

// RaftTerm represents the term of a raft message. This corresponds to Term in
// HardState.Term in the Raft library. That type is a uint64, so it is necessary
// to cast to/from that type when dealing with the Raft library, however
// internally RaftTerm is used for all fields in CRDB.
type RaftTerm uint64

// RaftIndex represents the term of a raft message. This corresponds to Index in
// HardState.Index in the Raft library. That type is a uint64, so it is
// necessary to cast to/from that type when dealing with the Raft library,
// however internally RaftIndex is used for all fields in CRDB.
type RaftIndex uint64

// SafeValue implements the redact.SafeValue interface.
func (s RaftTerm) SafeValue() {}
func (s RaftIndex) SafeValue() {}
2 changes: 0 additions & 2 deletions pkg/kv/kvserver/BUILD.bazel
Original file line number Diff line number Diff line change
Expand Up @@ -125,7 +125,6 @@ go_library(
"//pkg/kv/kvserver/batcheval",
"//pkg/kv/kvserver/batcheval/result",
"//pkg/kv/kvserver/closedts",
"//pkg/kv/kvserver/closedts/ctpb",
"//pkg/kv/kvserver/closedts/sidetransport",
"//pkg/kv/kvserver/closedts/tracker",
"//pkg/kv/kvserver/concurrency",
Expand Down Expand Up @@ -364,7 +363,6 @@ go_test(
"//pkg/kv/kvserver/batcheval",
"//pkg/kv/kvserver/batcheval/result",
"//pkg/kv/kvserver/closedts",
"//pkg/kv/kvserver/closedts/ctpb",
"//pkg/kv/kvserver/closedts/tracker",
"//pkg/kv/kvserver/concurrency",
"//pkg/kv/kvserver/concurrency/isolation",
Expand Down
2 changes: 2 additions & 0 deletions pkg/kv/kvserver/allocator/allocatorimpl/BUILD.bazel
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ go_library(
visibility = ["//visibility:public"],
deps = [
"//pkg/gossip",
"//pkg/kv/kvpb",
"//pkg/kv/kvserver/allocator",
"//pkg/kv/kvserver/allocator/load",
"//pkg/kv/kvserver/allocator/storepool",
Expand Down Expand Up @@ -46,6 +47,7 @@ go_test(
"//pkg/config/zonepb",
"//pkg/gossip",
"//pkg/keys",
"//pkg/kv/kvpb",
"//pkg/kv/kvserver/allocator",
"//pkg/kv/kvserver/allocator/load",
"//pkg/kv/kvserver/allocator/storepool",
Expand Down
14 changes: 9 additions & 5 deletions pkg/kv/kvserver/allocator/allocatorimpl/allocator.go
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ import (
"strings"
"time"

"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
"github.com/cockroachdb/cockroach/pkg/kv/kvserver/allocator"
"github.com/cockroachdb/cockroach/pkg/kv/kvserver/allocator/load"
"github.com/cockroachdb/cockroach/pkg/kv/kvserver/allocator/storepool"
Expand Down Expand Up @@ -1957,7 +1958,7 @@ func (a *Allocator) ValidLeaseTargets(
leaseRepl interface {
StoreID() roachpb.StoreID
RaftStatus() *raft.Status
GetFirstIndex() uint64
GetFirstIndex() kvpb.RaftIndex
Desc() *roachpb.RangeDescriptor
},
opts allocator.TransferLeaseOptions,
Expand Down Expand Up @@ -2132,7 +2133,7 @@ func (a *Allocator) leaseholderShouldMoveDueToPreferences(
leaseRepl interface {
StoreID() roachpb.StoreID
RaftStatus() *raft.Status
GetFirstIndex() uint64
GetFirstIndex() kvpb.RaftIndex
},
allExistingReplicas []roachpb.ReplicaDescriptor,
) bool {
Expand Down Expand Up @@ -2217,7 +2218,7 @@ func (a *Allocator) TransferLeaseTarget(
StoreID() roachpb.StoreID
GetRangeID() roachpb.RangeID
RaftStatus() *raft.Status
GetFirstIndex() uint64
GetFirstIndex() kvpb.RaftIndex
Desc() *roachpb.RangeDescriptor
},
usageInfo allocator.RangeUsageInfo,
Expand Down Expand Up @@ -2485,7 +2486,7 @@ func (a *Allocator) ShouldTransferLease(
leaseRepl interface {
StoreID() roachpb.StoreID
RaftStatus() *raft.Status
GetFirstIndex() uint64
GetFirstIndex() kvpb.RaftIndex
Desc() *roachpb.RangeDescriptor
},
usageInfo allocator.RangeUsageInfo,
Expand Down Expand Up @@ -2852,7 +2853,10 @@ func FilterBehindReplicas(
// Other replicas may be filtered out if this function is called with the
// `raftStatus` of a non-raft leader replica.
func excludeReplicasInNeedOfSnapshots(
ctx context.Context, st *raft.Status, firstIndex uint64, replicas []roachpb.ReplicaDescriptor,
ctx context.Context,
st *raft.Status,
firstIndex kvpb.RaftIndex,
replicas []roachpb.ReplicaDescriptor,
) []roachpb.ReplicaDescriptor {
filled := 0
for _, repl := range replicas {
Expand Down
3 changes: 2 additions & 1 deletion pkg/kv/kvserver/allocator/allocatorimpl/allocator_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ import (
"github.com/cockroachdb/cockroach/pkg/config/zonepb"
"github.com/cockroachdb/cockroach/pkg/gossip"
"github.com/cockroachdb/cockroach/pkg/keys"
"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
"github.com/cockroachdb/cockroach/pkg/kv/kvserver/allocator"
"github.com/cockroachdb/cockroach/pkg/kv/kvserver/allocator/load"
"github.com/cockroachdb/cockroach/pkg/kv/kvserver/allocator/storepool"
Expand Down Expand Up @@ -1893,7 +1894,7 @@ func (r *mockRepl) RaftStatus() *raft.Status {
return raftStatus
}

func (r *mockRepl) GetFirstIndex() uint64 {
func (r *mockRepl) GetFirstIndex() kvpb.RaftIndex {
return 0
}

Expand Down
2 changes: 1 addition & 1 deletion pkg/kv/kvserver/api.proto
Original file line number Diff line number Diff line change
Expand Up @@ -55,7 +55,7 @@ message WaitForApplicationRequest {
StoreRequestHeader header = 1 [(gogoproto.nullable) = false, (gogoproto.embed) = true];
int64 range_id = 2 [(gogoproto.customname) = "RangeID",
(gogoproto.casttype) = "github.com/cockroachdb/cockroach/pkg/roachpb.RangeID"];
uint64 lease_index = 3;
uint64 lease_index = 3 [(gogoproto.casttype) = "github.com/cockroachdb/cockroach/pkg/kv/kvpb.LeaseAppliedIndex"];
}

message WaitForApplicationResponse {
Expand Down
3 changes: 2 additions & 1 deletion pkg/kv/kvserver/app_batch.go
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ package kvserver
import (
"context"

"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvserverbase"
"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvserverpb"
"github.com/cockroachdb/cockroach/pkg/kv/kvserver/logstore"
Expand Down Expand Up @@ -179,7 +180,7 @@ func (b *appBatch) runPostAddTriggers(
env.st,
env.eng,
env.sideloaded,
cmd.Term,
kvpb.RaftTerm(cmd.Term),
cmd.Index(),
*res.AddSSTable,
env.bulkLimiter,
Expand Down
2 changes: 2 additions & 0 deletions pkg/kv/kvserver/apply/BUILD.bazel
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ go_library(
importpath = "github.com/cockroachdb/cockroach/pkg/kv/kvserver/apply",
visibility = ["//visibility:public"],
deps = [
"//pkg/kv/kvpb",
"@com_github_cockroachdb_errors//:errors",
"@io_etcd_go_raft_v3//raftpb",
],
Expand All @@ -26,6 +27,7 @@ go_test(
args = ["-test.timeout=55s"],
deps = [
":apply",
"//pkg/kv/kvpb",
"@com_github_cockroachdb_errors//:errors",
"@com_github_stretchr_testify//require",
"@io_etcd_go_raft_v3//raftpb",
Expand Down
8 changes: 6 additions & 2 deletions pkg/kv/kvserver/apply/cmd.go
Original file line number Diff line number Diff line change
Expand Up @@ -10,14 +10,18 @@

package apply

import "context"
import (
"context"

"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
)

// Command is a command that has been successfully replicated through raft
// by being durably committed to the raft log of a quorum of peers in a raft
// group.
type Command interface {
// Index is the log index of the corresponding raft entry.
Index() uint64
Index() kvpb.RaftIndex
// IsTrivial returns whether the command can apply in a batch with other
// "trivial" commands. This is the case if the log entry represented by the
// Command is a simple write, as is the case for all user-issued mutations.
Expand Down
Loading

0 comments on commit 0d49b3a

Please sign in to comment.