rkrtechnology/litestream
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Log validation position

Browse Source
This commit is contained in:
Ben Johnson
2021-01-17 07:38:13 -07:00
parent 07a65cbac7
commit 4b65e6a88f
3 changed files with 16 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
db.go
View File

@@ -1319,7 +1319,7 @@ func (db *DB) Restore(ctx context.Context, opt RestoreOptions) error {
tmpPath := outputPath + ".tmp" tmpPath := outputPath + ".tmp"
// Copy snapshot to output path. // Copy snapshot to output path.
logger.Printf("restoring snapshot from replica %q, generation %q, index %08x to %s", r.Name(), generation, minWALIndex, tmpPath) log.Printf("%s(%s): restoring snapshot %s/%08x to %s", db.path, r.Name(), generation, minWALIndex, tmpPath)
if !opt.DryRun { if !opt.DryRun {
if err := db.restoreSnapshot(ctx, r, pos.Generation, pos.Index, tmpPath); err != nil { if err := db.restoreSnapshot(ctx, r, pos.Generation, pos.Index, tmpPath); err != nil {
return fmt.Errorf("cannot restore snapshot: %w", err) return fmt.Errorf("cannot restore snapshot: %w", err)
@@ -1328,13 +1328,13 @@ func (db *DB) Restore(ctx context.Context, opt RestoreOptions) error {
// Restore each WAL file until we reach our maximum index. // Restore each WAL file until we reach our maximum index.
for index := minWALIndex; index <= maxWALIndex; index++ { for index := minWALIndex; index <= maxWALIndex; index++ {
logger.Printf("restoring wal from replica %q, generation %q, index %08x to %s-wal", r.Name(), generation, index, tmpPath) log.Printf("%s(%s): restoring wal %s/%08x", db.path, r.Name(), generation, index)
if opt.DryRun { if opt.DryRun {
continue continue
} }
if err := db.restoreWAL(ctx, r, generation, index, tmpPath); os.IsNotExist(err) && index == minWALIndex && index == maxWALIndex { if err := db.restoreWAL(ctx, r, generation, index, tmpPath); os.IsNotExist(err) && index == minWALIndex && index == maxWALIndex {
logger.Printf("no wal available, snapshot only") log.Printf("%s(%s): no wal available, snapshot only", db.path, r.Name())
break // snapshot file only, ignore error break // snapshot file only, ignore error
} else if err != nil { } else if err != nil {
return fmt.Errorf("cannot restore wal: %w", err) return fmt.Errorf("cannot restore wal: %w", err)
@@ -1342,7 +1342,7 @@ func (db *DB) Restore(ctx context.Context, opt RestoreOptions) error {
} }
// Copy file to final location. // Copy file to final location.
logger.Printf("renaming database from temporary location") log.Printf("%s(%s): renaming database from temporary location", db.path, r.Name())
if !opt.DryRun { if !opt.DryRun {
if err := os.Rename(tmpPath, outputPath); err != nil { if err := os.Rename(tmpPath, outputPath); err != nil {
return err return err

8
replica.go
View File

@@ -461,6 +461,12 @@ func (r *FileReplica) retainer(ctx context.Context) {
// validator runs in a separate goroutine and handles periodic validation. // validator runs in a separate goroutine and handles periodic validation.
func (r *FileReplica) validator(ctx context.Context) { func (r *FileReplica) validator(ctx context.Context) {
// Initialize counters since validation occurs infrequently.
for _, status := range []string{"ok", "error"} {
internal.ReplicaValidationTotalCounterVec.WithLabelValues(r.db.Path(), r.Name(), status).Add(0)
}
// Exit validation if interval is not set.
if r.ValidationInterval <= 0 { if r.ValidationInterval <= 0 {
return return
} }
@@ -990,7 +996,7 @@ func ValidateReplica(ctx context.Context, r Replica) error {
if err != nil { if err != nil {
return fmt.Errorf("cannot compute checksum: %w", err) return fmt.Errorf("cannot compute checksum: %w", err)
} }
log.Printf("%s(%s): primary checksum computed: %08x", db.Path(), r.Name(), chksum0) log.Printf("%s(%s): primary checksum computed: %08x @ %s", db.Path(), r.Name(), chksum0, pos)
// Wait until replica catches up to position. // Wait until replica catches up to position.
log.Printf("%s(%s): waiting for replica", db.Path(), r.Name()) log.Printf("%s(%s): waiting for replica", db.Path(), r.Name())

5
s3/s3.go
View File

@@ -488,6 +488,11 @@ func (r *Replica) retainer(ctx context.Context) {
// validator runs in a separate goroutine and handles periodic validation. // validator runs in a separate goroutine and handles periodic validation.
func (r *Replica) validator(ctx context.Context) { func (r *Replica) validator(ctx context.Context) {
// Initialize counters since validation occurs infrequently.
for _, status := range []string{"ok", "error"} {
internal.ReplicaValidationTotalCounterVec.WithLabelValues(r.db.Path(), r.Name(), status).Add(0)
}
if r.ValidationInterval <= 0 { if r.ValidationInterval <= 0 {
return return
} }