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

Add replica logging

Browse Source
This commit is contained in:
Ben Johnson
2021-06-06 08:02:48 -06:00
parent 8fb9c910f0
commit 8b70e3d8a8
3 changed files with 11 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
cmd/litestream/replicate.go
View File

@@ -185,7 +185,6 @@ func (c *ReplicateCommand) Close() (err error) {
} }
} }
} }
// TODO(windows): Clear DBs
return err return err
} }

18
replica.go
View File

@@ -455,10 +455,6 @@ func (r *Replica) Snapshot(ctx context.Context) (info SnapshotInfo, err error) {
return info, ErrNoGeneration return info, ErrNoGeneration
} }
// TODO: Check if snapshot already exists & skip.
// startTime := time.Now()
// Open db file descriptor, if not already open, & position at beginning. // Open db file descriptor, if not already open, & position at beginning.
if r.f == nil { if r.f == nil {
if r.f, err = os.Open(r.db.Path()); err != nil { if r.f, err = os.Open(r.db.Path()); err != nil {
@@ -491,9 +487,13 @@ func (r *Replica) Snapshot(ctx context.Context) (info SnapshotInfo, err error) {
// Delegate write to client & wait for writer goroutine to finish. // Delegate write to client & wait for writer goroutine to finish.
if info, err = r.Client.WriteSnapshot(ctx, pos.Generation, pos.Index, pr); err != nil { if info, err = r.Client.WriteSnapshot(ctx, pos.Generation, pos.Index, pr); err != nil {
return info, err return info, err
} else if err := g.Wait(); err != nil {
return info, err
} }
return info, g.Wait() log.Printf("%s(%s): snapshot written %s/%08x", r.db.Path(), r.Name(), pos.Generation, pos.Index)
return info, nil
} }
// EnforceRetention forces a new snapshot once the retention interval has passed. // EnforceRetention forces a new snapshot once the retention interval has passed.
@@ -559,9 +559,9 @@ func (r *Replica) deleteSnapshotsBeforeIndex(ctx context.Context, generation str
if err := r.Client.DeleteSnapshot(ctx, info.Generation, info.Index); err != nil { if err := r.Client.DeleteSnapshot(ctx, info.Generation, info.Index); err != nil {
return fmt.Errorf("delete snapshot %s/%08x: %w", info.Generation, info.Index, err) return fmt.Errorf("delete snapshot %s/%08x: %w", info.Generation, info.Index, err)
} }
log.Printf("%s(%s): snapshot deleted %s/%08x", r.db.Path(), r.Name(), generation, index)
} }
// log.Printf("%s(%s): retainer: deleting snapshots before %s/%08x; n=%d", r.db.Path(), r.Name(), generation, index, n)
return itr.Close() return itr.Close()
} }
@@ -584,11 +584,15 @@ func (r *Replica) deleteWALSegmentsBeforeIndex(ctx context.Context, generation s
return err return err
} }
if len(a) == 0 {
return nil
}
if err := r.Client.DeleteWALSegments(ctx, a); err != nil { if err := r.Client.DeleteWALSegments(ctx, a); err != nil {
return fmt.Errorf("delete wal segments: %w", err) return fmt.Errorf("delete wal segments: %w", err)
} }
log.Printf("%s(%s): wal segmented deleted before %s/%08x: n=%d", r.db.Path(), r.Name(), generation, index, len(a))
// log.Printf("%s(%s): retainer: deleting wal segment %s/%08x:%d", r.db.Path(), r.Name(), generation, index, offset)
return nil return nil
} }

2
sftp/replica_client.go
View File

@@ -18,8 +18,6 @@ import (
"golang.org/x/crypto/ssh" "golang.org/x/crypto/ssh"
) )
// TODO(sftp): Add public key support
// ReplicaClientType is the client type for this package. // ReplicaClientType is the client type for this package.
const ReplicaClientType = "sftp" const ReplicaClientType = "sftp"