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

Reduce s3 LIST operations

Browse Source
This commit is contained in:
Ben Johnson
2021-01-15 13:31:04 -07:00
parent b94ee366e5
commit 290e06e60d
1 changed files with 32 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

62
s3/s3.go
View File

@@ -651,47 +651,48 @@ func (r *Replica) Sync(ctx context.Context) (err error) {
return err return err
} }
// Ensure sync & retainer do not calculate position or snapshot at the same time. // Find current position of database.
if err := func() error { dpos, err := r.db.Pos()
r.snapshotMu.Lock() if err != nil {
defer r.snapshotMu.Unlock() return fmt.Errorf("cannot determine current generation: %w", err)
} else if dpos.IsZero() {
return fmt.Errorf("no generation, waiting for data")
}
generation := dpos.Generation
// Find current position of database. // Calculate position if we don't have a previous position or if the generation changes.
dpos, err := r.db.Pos() // Ensure sync & retainer do not snapshot at the same time.
if err != nil { if lastPos := r.LastPos(); lastPos.IsZero() || lastPos.Generation != generation {
return fmt.Errorf("cannot determine current generation: %w", err) if err := func() error {
} else if dpos.IsZero() { r.snapshotMu.Lock()
return fmt.Errorf("no generation, waiting for data") defer r.snapshotMu.Unlock()
}
generation := dpos.Generation
// Create snapshot if no snapshots exist for generation. // Create snapshot if no snapshots exist for generation.
if n, err := r.snapshotN(generation); err != nil { if n, err := r.snapshotN(generation); err != nil {
return err
} else if n == 0 {
if err := r.snapshot(ctx, generation, dpos.Index); err != nil {
return err return err
} else if n == 0 {
if err := r.snapshot(ctx, generation, dpos.Index); err != nil {
return err
}
r.snapshotTotalGauge.Set(1.0)
} else {
r.snapshotTotalGauge.Set(float64(n))
} }
r.snapshotTotalGauge.Set(1.0)
} else {
r.snapshotTotalGauge.Set(float64(n))
}
// Determine position, if necessary. // Determine position, if necessary.
if r.LastPos().IsZero() {
pos, err := r.CalcPos(ctx, generation) pos, err := r.CalcPos(ctx, generation)
if err != nil { if err != nil {
return fmt.Errorf("cannot determine replica position: %s", err) return fmt.Errorf("cannot determine replica position: %s", err)
} }
r.mu.Lock() r.mu.Lock()
defer r.mu.Unlock()
r.pos = pos r.pos = pos
r.mu.Unlock()
}
return nil return nil
}(); err != nil { }(); err != nil {
return err return err
}
} }
// Read all WAL files since the last position. // Read all WAL files since the last position.
@@ -724,7 +725,8 @@ func (r *Replica) syncWAL(ctx context.Context) (err error) {
var buf bytes.Buffer var buf bytes.Buffer
gw := gzip.NewWriter(&buf) gw := gzip.NewWriter(&buf)
if _, err := gw.Write(b); err != nil { n, err := gw.Write(b)
if err != nil {
return err return err
} else if err := gw.Close(); err != nil { } else if err := gw.Close(); err != nil {
return err return err
@@ -745,7 +747,7 @@ func (r *Replica) syncWAL(ctx context.Context) (err error) {
return err return err
} }
r.putOperationTotalCounter.Inc() r.putOperationTotalCounter.Inc()
r.putOperationBytesCounter.Add(float64(buf.Len())) // compressed bytes r.putOperationBytesCounter.Add(float64(n)) // compressed bytes
// Save last replicated position. // Save last replicated position.
r.mu.Lock() r.mu.Lock()