From 58681d3feb4b04ba835eefa265397c36b80de2e2 Mon Sep 17 00:00:00 2001 From: Piotr Tabor Date: Fri, 23 Dec 2022 14:09:21 +0100 Subject: [PATCH] Expose types of entries within the WAL log for access from the tools. Signed-off-by: Piotr Tabor --- server/storage/wal/decoder.go | 8 ++--- server/storage/wal/repair.go | 2 +- server/storage/wal/wal.go | 56 +++++++++++++++++----------------- server/storage/wal/wal_test.go | 6 ++-- 4 files changed, 36 insertions(+), 36 deletions(-) diff --git a/server/storage/wal/decoder.go b/server/storage/wal/decoder.go index cbe4b0821..1010fb074 100644 --- a/server/storage/wal/decoder.go +++ b/server/storage/wal/decoder.go @@ -121,8 +121,8 @@ func (d *decoder) decodeRecord(rec *walpb.Record) error { return err } - // skip crc checking if the record type is crcType - if rec.Type != crcType { + // skip crc checking if the record type is CrcType + if rec.Type != CrcType { _, err := d.crc.Write(rec.Data) if err != nil { return err @@ -204,13 +204,13 @@ func (d *decoder) LastCRC() uint32 { func (d *decoder) LastOffset() int64 { return d.lastValidOff } -func mustUnmarshalEntry(d []byte) raftpb.Entry { +func MustUnmarshalEntry(d []byte) raftpb.Entry { var e raftpb.Entry pbutil.MustUnmarshal(&e, d) return e } -func mustUnmarshalState(d []byte) raftpb.HardState { +func MustUnmarshalState(d []byte) raftpb.HardState { var s raftpb.HardState pbutil.MustUnmarshal(&s, d) return s diff --git a/server/storage/wal/repair.go b/server/storage/wal/repair.go index bca72f8ae..537340451 100644 --- a/server/storage/wal/repair.go +++ b/server/storage/wal/repair.go @@ -50,7 +50,7 @@ func Repair(lg *zap.Logger, dirpath string) bool { case err == nil: // update crc of the decoder when necessary switch rec.Type { - case crcType: + case CrcType: crc := decoder.LastCRC() // current crc of decoder must match the crc of the record. // do no need to match 0 crc, since the decoder is a new one at this case. diff --git a/server/storage/wal/wal.go b/server/storage/wal/wal.go index 5c15ebfa6..e913c76c1 100644 --- a/server/storage/wal/wal.go +++ b/server/storage/wal/wal.go @@ -36,11 +36,11 @@ import ( ) const ( - metadataType int64 = iota + 1 - entryType - stateType - crcType - snapshotType + MetadataType int64 = iota + 1 + EntryType + StateType + CrcType + SnapshotType // warnSyncDuration is the amount of time allotted to an fsync before // logging a warning @@ -56,7 +56,7 @@ var ( ErrMetadataConflict = errors.New("wal: conflicting metadata found") ErrFileNotFound = errors.New("wal: file not found") - ErrCRCMismatch = errors.New("wal: crc mismatch") + ErrCRCMismatch = walpb.ErrCRCMismatch ErrSnapshotMismatch = errors.New("wal: snapshot mismatch") ErrSnapshotNotFound = errors.New("wal: snapshot not found") ErrSliceOutOfRange = errors.New("wal: slice bounds out of range") @@ -166,7 +166,7 @@ func Create(lg *zap.Logger, dirpath string, metadata []byte) (*WAL, error) { if err = w.saveCrc(0); err != nil { return nil, err } - if err = w.encoder.encode(&walpb.Record{Type: metadataType, Data: metadata}); err != nil { + if err = w.encoder.encode(&walpb.Record{Type: MetadataType, Data: metadata}); err != nil { return nil, err } if err = w.SaveSnapshot(walpb.Snapshot{}); err != nil { @@ -454,8 +454,8 @@ func (w *WAL) ReadAll() (metadata []byte, state raftpb.HardState, ents []raftpb. var match bool for err = decoder.Decode(rec); err == nil; err = decoder.Decode(rec) { switch rec.Type { - case entryType: - e := mustUnmarshalEntry(rec.Data) + case EntryType: + e := MustUnmarshalEntry(rec.Data) // 0 <= e.Index-w.start.Index - 1 < len(ents) if e.Index > w.start.Index { // prevent "panic: runtime error: slice bounds out of range [:13038096702221461992] with capacity 0" @@ -469,17 +469,17 @@ func (w *WAL) ReadAll() (metadata []byte, state raftpb.HardState, ents []raftpb. } w.enti = e.Index - case stateType: - state = mustUnmarshalState(rec.Data) + case StateType: + state = MustUnmarshalState(rec.Data) - case metadataType: + case MetadataType: if metadata != nil && !bytes.Equal(metadata, rec.Data) { state.Reset() return nil, state, nil, ErrMetadataConflict } metadata = rec.Data - case crcType: + case CrcType: crc := decoder.LastCRC() // current crc of decoder must match the crc of the record. // do no need to match 0 crc, since the decoder is a new one at this case. @@ -489,7 +489,7 @@ func (w *WAL) ReadAll() (metadata []byte, state raftpb.HardState, ents []raftpb. } decoder.UpdateCRC(rec.Crc) - case snapshotType: + case SnapshotType: var snap walpb.Snapshot pbutil.MustUnmarshal(&snap, rec.Data) if snap.Index == w.start.Index { @@ -591,13 +591,13 @@ func ValidSnapshotEntries(lg *zap.Logger, walDir string) ([]walpb.Snapshot, erro for err = decoder.Decode(rec); err == nil; err = decoder.Decode(rec) { switch rec.Type { - case snapshotType: + case SnapshotType: var loadedSnap walpb.Snapshot pbutil.MustUnmarshal(&loadedSnap, rec.Data) snaps = append(snaps, loadedSnap) - case stateType: - state = mustUnmarshalState(rec.Data) - case crcType: + case StateType: + state = MustUnmarshalState(rec.Data) + case CrcType: crc := decoder.LastCRC() // current crc of decoder must match the crc of the record. // do no need to match 0 crc, since the decoder is a new one at this case. @@ -665,12 +665,12 @@ func Verify(lg *zap.Logger, walDir string, snap walpb.Snapshot) (*raftpb.HardSta for err = decoder.Decode(rec); err == nil; err = decoder.Decode(rec) { switch rec.Type { - case metadataType: + case MetadataType: if metadata != nil && !bytes.Equal(metadata, rec.Data) { return nil, ErrMetadataConflict } metadata = rec.Data - case crcType: + case CrcType: crc := decoder.LastCRC() // Current crc of decoder must match the crc of the record. // We need not match 0 crc, since the decoder is a new one at this point. @@ -678,7 +678,7 @@ func Verify(lg *zap.Logger, walDir string, snap walpb.Snapshot) (*raftpb.HardSta return nil, ErrCRCMismatch } decoder.UpdateCRC(rec.Crc) - case snapshotType: + case SnapshotType: var loadedSnap walpb.Snapshot pbutil.MustUnmarshal(&loadedSnap, rec.Data) if loadedSnap.Index == snap.Index { @@ -689,8 +689,8 @@ func Verify(lg *zap.Logger, walDir string, snap walpb.Snapshot) (*raftpb.HardSta } // We ignore all entry and state type records as these // are not necessary for validating the WAL contents - case entryType: - case stateType: + case EntryType: + case StateType: pbutil.MustUnmarshal(&state, rec.Data) default: return nil, fmt.Errorf("unexpected block type %d", rec.Type) @@ -748,7 +748,7 @@ func (w *WAL) cut() error { return err } - if err = w.encoder.encode(&walpb.Record{Type: metadataType, Data: w.metadata}); err != nil { + if err = w.encoder.encode(&walpb.Record{Type: MetadataType, Data: w.metadata}); err != nil { return err } @@ -905,7 +905,7 @@ func (w *WAL) Close() error { func (w *WAL) saveEntry(e *raftpb.Entry) error { // TODO: add MustMarshalTo to reduce one allocation. b := pbutil.MustMarshal(e) - rec := &walpb.Record{Type: entryType, Data: b} + rec := &walpb.Record{Type: EntryType, Data: b} if err := w.encoder.encode(rec); err != nil { return err } @@ -919,7 +919,7 @@ func (w *WAL) saveState(s *raftpb.HardState) error { } w.state = *s b := pbutil.MustMarshal(s) - rec := &walpb.Record{Type: stateType, Data: b} + rec := &walpb.Record{Type: StateType, Data: b} return w.encoder.encode(rec) } @@ -968,7 +968,7 @@ func (w *WAL) SaveSnapshot(e walpb.Snapshot) error { w.mu.Lock() defer w.mu.Unlock() - rec := &walpb.Record{Type: snapshotType, Data: b} + rec := &walpb.Record{Type: SnapshotType, Data: b} if err := w.encoder.encode(rec); err != nil { return err } @@ -980,7 +980,7 @@ func (w *WAL) SaveSnapshot(e walpb.Snapshot) error { } func (w *WAL) saveCrc(prevCrc uint32) error { - return w.encoder.encode(&walpb.Record{Type: crcType, Crc: prevCrc}) + return w.encoder.encode(&walpb.Record{Type: CrcType, Crc: prevCrc}) } func (w *WAL) tail() *fileutil.LockedFile { diff --git a/server/storage/wal/wal_test.go b/server/storage/wal/wal_test.go index 70883ba3b..1063ed438 100644 --- a/server/storage/wal/wal_test.go +++ b/server/storage/wal/wal_test.go @@ -74,16 +74,16 @@ func TestNew(t *testing.T) { var wb bytes.Buffer e := newEncoder(&wb, 0, 0) - err = e.encode(&walpb.Record{Type: crcType, Crc: 0}) + err = e.encode(&walpb.Record{Type: CrcType, Crc: 0}) if err != nil { t.Fatalf("err = %v, want nil", err) } - err = e.encode(&walpb.Record{Type: metadataType, Data: []byte("somedata")}) + err = e.encode(&walpb.Record{Type: MetadataType, Data: []byte("somedata")}) if err != nil { t.Fatalf("err = %v, want nil", err) } r := &walpb.Record{ - Type: snapshotType, + Type: SnapshotType, Data: pbutil.MustMarshal(&walpb.Snapshot{}), } if err = e.encode(r); err != nil {