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

Update wal segment naming

Browse Source
This commit is contained in:
Ben Johnson
2021-01-14 15:26:29 -07:00
parent 1e4e9633cc
commit e1c9e09161
5 changed files with 27 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
cmd/litestream/generations.go
View File

@@ -88,6 +88,7 @@ func (c *GenerationsCommand) Run(ctx context.Context, args []string) (err error)
stats.CreatedAt.Format(time.RFC3339), stats.CreatedAt.Format(time.RFC3339),
stats.UpdatedAt.Format(time.RFC3339), stats.UpdatedAt.Format(time.RFC3339),
) )
w.Flush()
} }
} }
w.Flush() w.Flush()

4
db.go
View File

@@ -1313,7 +1313,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 %s://%s/%016x to %s", r.Name(), generation, minWALIndex, tmpPath) logger.Printf("restoring snapshot from replica %q, generation %q, index %08x to %s", 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)
@@ -1322,7 +1322,7 @@ 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 %s://%s/%016x to %s-wal", r.Name(), generation, index, tmpPath) logger.Printf("restoring wal from replica %q, generation %q, index %08x to %s-wal", r.Name(), generation, index, tmpPath)
if opt.DryRun { if opt.DryRun {
continue continue
} }

8
litestream.go
View File

@@ -219,7 +219,7 @@ func ParseSnapshotPath(s string) (index int, ext string, err error) {
return int(i64), a[2], nil return int(i64), a[2], nil
} }
var snapshotPathRegex = regexp.MustCompile(`^([0-9a-f]{16})(.snapshot(?:.gz)?)$`) var snapshotPathRegex = regexp.MustCompile(`^([0-9a-f]{8})(.snapshot(?:.gz)?)$`)
// IsWALPath returns true if s is a path to a WAL file. // IsWALPath returns true if s is a path to a WAL file.
func IsWALPath(s string) bool { func IsWALPath(s string) bool {
@@ -245,7 +245,7 @@ func ParseWALPath(s string) (index int, offset, sz int64, ext string, err error)
// FormatWALPath formats a WAL filename with a given index. // FormatWALPath formats a WAL filename with a given index.
func FormatWALPath(index int) string { func FormatWALPath(index int) string {
assert(index >= 0, "wal index must be non-negative") assert(index >= 0, "wal index must be non-negative")
return fmt.Sprintf("%016x%s", index, WALExt) return fmt.Sprintf("%08x%s", index, WALExt)
} }
// FormatWALPathWithOffsetSize formats a WAL filename with a given index, offset & size. // FormatWALPathWithOffsetSize formats a WAL filename with a given index, offset & size.
@@ -253,10 +253,10 @@ func FormatWALPathWithOffsetSize(index int, offset, sz int64) string {
assert(index >= 0, "wal index must be non-negative") assert(index >= 0, "wal index must be non-negative")
assert(offset >= 0, "wal offset must be non-negative") assert(offset >= 0, "wal offset must be non-negative")
assert(sz >= 0, "wal size must be non-negative") assert(sz >= 0, "wal size must be non-negative")
return fmt.Sprintf("%016x_%016x_%016x%s", index, offset, sz, WALExt) return fmt.Sprintf("%08x_%08x_%08x%s", index, offset, sz, WALExt)
} }
var walPathRegex = regexp.MustCompile(`^([0-9a-f]{16})(?:_([0-9a-f]{16})_([0-9a-f]{16}))?(.wal(?:.gz)?)$`) var walPathRegex = regexp.MustCompile(`^([0-9a-f]{8})(?:_([0-9a-f]{8})_([0-9a-f]{8}))?(.wal(?:.gz)?)$`)
// isHexChar returns true if ch is a lowercase hex character. // isHexChar returns true if ch is a lowercase hex character.
func isHexChar(ch rune) bool { func isHexChar(ch rune) bool {

8
replica.go
View File

@@ -71,6 +71,7 @@ type GenerationStats struct {
// Default file replica settings. // Default file replica settings.
const ( const (
DefaultRetention = 24 * time.Hour DefaultRetention = 24 * time.Hour
DefaultRetentionCheckInterval = 1 * time.Hour
) )
var _ Replica = (*FileReplica)(nil) var _ Replica = (*FileReplica)(nil)
@@ -109,6 +110,7 @@ func NewFileReplica(db *DB, name, dst string) *FileReplica {
cancel: func() {}, cancel: func() {},
Retention: DefaultRetention, Retention: DefaultRetention,
RetentionCheckInterval: DefaultRetentionCheckInterval,
MonitorEnabled: true, MonitorEnabled: true,
} }
} }
@@ -145,7 +147,7 @@ func (r *FileReplica) SnapshotDir(generation string) string {
// SnapshotPath returns the path to a snapshot file. // SnapshotPath returns the path to a snapshot file.
func (r *FileReplica) SnapshotPath(generation string, index int) string { func (r *FileReplica) SnapshotPath(generation string, index int) string {
return filepath.Join(r.SnapshotDir(generation), fmt.Sprintf("%016x.snapshot.gz", index)) return filepath.Join(r.SnapshotDir(generation), fmt.Sprintf("%08x.snapshot.gz", index))
} }
// MaxSnapshotIndex returns the highest index for the snapshots. // MaxSnapshotIndex returns the highest index for the snapshots.
@@ -176,7 +178,7 @@ func (r *FileReplica) WALDir(generation string) string {
// WALPath returns the path to a WAL file. // WALPath returns the path to a WAL file.
func (r *FileReplica) WALPath(generation string, index int) string { func (r *FileReplica) WALPath(generation string, index int) string {
return filepath.Join(r.WALDir(generation), fmt.Sprintf("%016x.wal", index)) return filepath.Join(r.WALDir(generation), fmt.Sprintf("%08x.wal", index))
} }
// Generations returns a list of available generation names. // Generations returns a list of available generation names.
@@ -776,7 +778,7 @@ func (r *FileReplica) deleteGenerationSnapshotsBefore(ctx context.Context, gener
continue continue
} }
log.Printf("%s(%s): generation %q snapshot no longer retained, deleting %s", r.db.Path(), r.Name(), generation, fi.Name()) log.Printf("%s(%s): retention exceeded, deleting from generation %q: %s", r.db.Path(), r.Name(), generation, fi.Name())
if err := os.Remove(filepath.Join(dir, fi.Name())); err != nil { if err := os.Remove(filepath.Join(dir, fi.Name())); err != nil {
return err return err
} }

17
s3/s3.go
View File

@@ -121,7 +121,7 @@ func (r *Replica) SnapshotDir(generation string) string {
// SnapshotPath returns the path to a snapshot file. // SnapshotPath returns the path to a snapshot file.
func (r *Replica) SnapshotPath(generation string, index int) string { func (r *Replica) SnapshotPath(generation string, index int) string {
return path.Join(r.SnapshotDir(generation), fmt.Sprintf("%016x.snapshot.gz", index)) return path.Join(r.SnapshotDir(generation), fmt.Sprintf("%08x.snapshot.gz", index))
} }
// MaxSnapshotIndex returns the highest index for the snapshots. // MaxSnapshotIndex returns the highest index for the snapshots.
@@ -213,10 +213,9 @@ func (r *Replica) snapshotStats(ctx context.Context, generation string) (n int,
if err := r.s3.ListObjectsPagesWithContext(ctx, &s3.ListObjectsInput{ if err := r.s3.ListObjectsPagesWithContext(ctx, &s3.ListObjectsInput{
Bucket: aws.String(r.Bucket), Bucket: aws.String(r.Bucket),
Prefix: aws.String(r.SnapshotDir(generation) + "/"), Prefix: aws.String(r.SnapshotDir(generation) + "/"),
Delimiter: aws.String("/"),
}, func(page *s3.ListObjectsOutput, lastPage bool) bool { }, func(page *s3.ListObjectsOutput, lastPage bool) bool {
for _, obj := range page.Contents { for _, obj := range page.Contents {
if !litestream.IsSnapshotPath(*obj.Key) { if !litestream.IsSnapshotPath(path.Base(*obj.Key)) {
continue continue
} }
modTime := obj.LastModified.UTC() modTime := obj.LastModified.UTC()
@@ -240,10 +239,9 @@ func (r *Replica) walStats(ctx context.Context, generation string) (n int, min,
if err := r.s3.ListObjectsPagesWithContext(ctx, &s3.ListObjectsInput{ if err := r.s3.ListObjectsPagesWithContext(ctx, &s3.ListObjectsInput{
Bucket: aws.String(r.Bucket), Bucket: aws.String(r.Bucket),
Prefix: aws.String(r.WALDir(generation) + "/"), Prefix: aws.String(r.WALDir(generation) + "/"),
Delimiter: aws.String("/"),
}, func(page *s3.ListObjectsOutput, lastPage bool) bool { }, func(page *s3.ListObjectsOutput, lastPage bool) bool {
for _, obj := range page.Contents { for _, obj := range page.Contents {
if !litestream.IsWALPath(*obj.Key) { if !litestream.IsWALPath(path.Base(*obj.Key)) {
continue continue
} }
modTime := obj.LastModified.UTC() modTime := obj.LastModified.UTC()
@@ -637,6 +635,7 @@ func (r *Replica) syncWAL(ctx context.Context) (err error) {
defer rd.Close() defer rd.Close()
// Read to intermediate buffer to determine size. // Read to intermediate buffer to determine size.
pos := rd.Pos()
b, err := ioutil.ReadAll(rd) b, err := ioutil.ReadAll(rd)
if err != nil { if err != nil {
return err return err
@@ -652,7 +651,7 @@ func (r *Replica) syncWAL(ctx context.Context) (err error) {
// that files are contiguous without having to decompress. // that files are contiguous without having to decompress.
walPath := path.Join( walPath := path.Join(
r.WALDir(rd.Pos().Generation), r.WALDir(rd.Pos().Generation),
litestream.FormatWALPathWithOffsetSize(rd.Pos().Index, rd.Pos().Offset, int64(len(b)))+".gz", litestream.FormatWALPathWithOffsetSize(pos.Index, pos.Offset, int64(len(b)))+".gz",
) )
if _, err := r.uploader.UploadWithContext(ctx, &s3manager.UploadInput{ if _, err := r.uploader.UploadWithContext(ctx, &s3manager.UploadInput{
@@ -708,7 +707,7 @@ func (r *Replica) WALReader(ctx context.Context, generation string, index int) (
var innerErr error var innerErr error
if err := r.s3.ListObjectsPagesWithContext(ctx, &s3.ListObjectsInput{ if err := r.s3.ListObjectsPagesWithContext(ctx, &s3.ListObjectsInput{
Bucket: aws.String(r.Bucket), Bucket: aws.String(r.Bucket),
Prefix: aws.String(path.Join(r.WALDir(generation), fmt.Sprintf("%016x_", index))), Prefix: aws.String(path.Join(r.WALDir(generation), fmt.Sprintf("%08x_", index))),
}, func(page *s3.ListObjectsOutput, lastPage bool) bool { }, func(page *s3.ListObjectsOutput, lastPage bool) bool {
for _, obj := range page.Contents { for _, obj := range page.Contents {
// Read the offset & size from the filename. We need to check this // Read the offset & size from the filename. We need to check this
@@ -717,7 +716,7 @@ func (r *Replica) WALReader(ctx context.Context, generation string, index int) (
if err != nil { if err != nil {
continue continue
} else if off != offset { } else if off != offset {
innerErr = fmt.Errorf("out of sequence wal segments: %s/%016x", generation, index) innerErr = fmt.Errorf("out of sequence wal segments: %s/%08x", generation, index)
return false return false
} }
@@ -858,7 +857,7 @@ func (r *Replica) deleteGenerationBefore(ctx context.Context, generation string,
} }
for _, objID := range objIDs[i:j] { for _, objID := range objIDs[i:j] {
log.Printf("%s(%s): generation %q file no longer retained, deleting %s", r.db.Path(), r.Name(), generation, path.Base(*objID.Key)) log.Printf("%s(%s): retention exceeded, deleting from generation %q: %s", r.db.Path(), r.Name(), generation, path.Base(*objID.Key))
} }
if _, err := r.s3.DeleteObjectsWithContext(ctx, &s3.DeleteObjectsInput{ if _, err := r.s3.DeleteObjectsWithContext(ctx, &s3.DeleteObjectsInput{