Skip to content

Commit

Permalink
Plumb a custom logger through the core rather than only in Restore (#481
Browse files Browse the repository at this point in the history 
)
  • Loading branch information
@evanphx
evanphx authored May 20, 2023
1 parent ad3d653 commit 18760d2
Show file tree
Hide file tree
Showing 2 changed files with 23 additions and 14 deletions.
11 changes: 8 additions & 3 deletions db.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -94,6 +94,9 @@ type DB struct {
// List of replicas for the database.
// Must be set before calling Open().
Replicas []*Replica

// Where to send log messages, defaults to log.Default()
Logger *log.Logger
}

// NewDB returns a new instance of DB for a given path.
Expand All @@ -106,6 +109,8 @@ func NewDB(path string) *DB {
MaxCheckpointPageN: DefaultMaxCheckpointPageN,
CheckpointInterval: DefaultCheckpointInterval,
MonitorInterval: DefaultMonitorInterval,

Logger: log.Default(),
}

db.dbSizeGauge = dbSizeGaugeVec.WithLabelValues(db.path)
Expand Down Expand Up @@ -447,7 +452,7 @@ func (db *DB) init() (err error) {

// If we have an existing shadow WAL, ensure the headers match.
if err := db.verifyHeadersMatch(); err != nil {
log.Printf("%s: init: cannot determine last wal position, clearing generation; %s", db.path, err)
db.Logger.Printf("%s: init: cannot determine last wal position, clearing generation; %s", db.path, err)
if err := os.Remove(db.GenerationNamePath()); err != nil && !os.IsNotExist(err) {
return fmt.Errorf("remove generation name: %w", err)
}
Expand Down Expand Up @@ -726,7 +731,7 @@ func (db *DB) Sync(ctx context.Context) (err error) {
if info.generation, err = db.createGeneration(); err != nil {
return fmt.Errorf("create generation: %w", err)
}
log.Printf("%s: sync: new generation %q, %s", db.path, info.generation, info.reason)
db.Logger.Printf("%s: sync: new generation %q, %s", db.path, info.generation, info.reason)

// Clear shadow wal info.
info.shadowWALPath = db.ShadowWALPath(info.generation, 0)
Expand Down Expand Up @@ -1365,7 +1370,7 @@ func (db *DB) monitor() {

// Sync the database to the shadow WAL.
if err := db.Sync(db.ctx); err != nil && !errors.Is(err, context.Canceled) {
log.Printf("%s: sync error: %s", db.path, err)
db.Logger.Printf("%s: sync error: %s", db.path, err)
}
}
}
Expand Down
26 changes: 15 additions & 11 deletions replica.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -72,6 +72,9 @@ type Replica struct {
// Encryption identities and recipients
AgeIdentities []age.Identity
AgeRecipients []age.Recipient

// The logger to send logging messages to. Defaults to log.Default()
Logger *log.Logger
}

func NewReplica(db *DB, name string) *Replica {
Expand All @@ -84,6 +87,7 @@ func NewReplica(db *DB, name string) *Replica {
Retention: DefaultRetention,
RetentionCheckInterval: DefaultRetentionCheckInterval,
MonitorEnabled: true,
Logger: log.Default(),
}

return r
Expand Down Expand Up @@ -534,7 +538,7 @@ func (r *Replica) Snapshot(ctx context.Context) (info SnapshotInfo, err error) {
return info, err
}

log.Printf("%s(%s): snapshot written %s/%08x", r.db.Path(), r.Name(), pos.Generation, pos.Index)
r.Logger.Printf("%s(%s): snapshot written %s/%08x", r.db.Path(), r.Name(), pos.Generation, pos.Index)

return info, nil
}
Expand Down Expand Up @@ -602,7 +606,7 @@ func (r *Replica) deleteSnapshotsBeforeIndex(ctx context.Context, generation str
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)
}
log.Printf("%s(%s): snapshot deleted %s/%08x", r.db.Path(), r.Name(), generation, index)
r.Logger.Printf("%s(%s): snapshot deleted %s/%08x", r.db.Path(), r.Name(), generation, index)
}

return itr.Close()
Expand Down Expand Up @@ -634,7 +638,7 @@ func (r *Replica) deleteWALSegmentsBeforeIndex(ctx context.Context, generation s
if err := r.Client.DeleteWALSegments(ctx, a); err != nil {
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))
r.Logger.Printf("%s(%s): wal segmented deleted before %s/%08x: n=%d", r.db.Path(), r.Name(), generation, index, len(a))

return nil
}
Expand Down Expand Up @@ -671,7 +675,7 @@ func (r *Replica) monitor(ctx context.Context) {

// Synchronize the shadow wal into the replication directory.
if err := r.Sync(ctx); err != nil {
log.Printf("%s(%s): monitor error: %s", r.db.Path(), r.Name(), err)
r.Logger.Printf("%s(%s): monitor error: %s", r.db.Path(), r.Name(), err)
continue
}
}
Expand Down Expand Up @@ -699,7 +703,7 @@ func (r *Replica) retainer(ctx context.Context) {
return
case <-ticker.C:
if err := r.EnforceRetention(ctx); err != nil {
log.Printf("%s(%s): retainer error: %s", r.db.Path(), r.Name(), err)
r.Logger.Printf("%s(%s): retainer error: %s", r.db.Path(), r.Name(), err)
continue
}
}
Expand All @@ -721,7 +725,7 @@ func (r *Replica) snapshotter(ctx context.Context) {
return
case <-ticker.C:
if _, err := r.Snapshot(ctx); err != nil && err != ErrNoGeneration {
log.Printf("%s(%s): snapshotter error: %s", r.db.Path(), r.Name(), err)
r.Logger.Printf("%s(%s): snapshotter error: %s", r.db.Path(), r.Name(), err)
continue
}
}
Expand Down Expand Up @@ -749,7 +753,7 @@ func (r *Replica) validator(ctx context.Context) {
return
case <-ticker.C:
if err := r.Validate(ctx); err != nil {
log.Printf("%s(%s): validation error: %s", r.db.Path(), r.Name(), err)
r.Logger.Printf("%s(%s): validation error: %s", r.db.Path(), r.Name(), err)
continue
}
}
Expand Down Expand Up @@ -786,7 +790,7 @@ func (r *Replica) Validate(ctx context.Context) error {
ReplicaName: r.Name(),
Generation: pos.Generation,
Index: pos.Index - 1,
Logger: log.New(os.Stderr, "", 0),
Logger: r.Logger,
}); err != nil {
return fmt.Errorf("cannot restore: %w", err)
}
Expand All @@ -811,7 +815,7 @@ func (r *Replica) Validate(ctx context.Context) error {
if mismatch {
status = "mismatch"
}
log.Printf("%s(%s): validator: status=%s db=%016x replica=%016x pos=%s", db.Path(), r.Name(), status, chksum0, chksum1, pos)
r.Logger.Printf("%s(%s): validator: status=%s db=%016x replica=%016x pos=%s", db.Path(), r.Name(), status, chksum0, chksum1, pos)

// Validate checksums match.
if mismatch {
Expand Down Expand Up @@ -853,7 +857,7 @@ func (r *Replica) waitForReplica(ctx context.Context, pos Pos) error {
// Obtain current position of replica, check if past target position.
curr := r.Pos()
if curr.IsZero() {
log.Printf("%s(%s): validator: no replica position available", db.Path(), r.Name())
r.Logger.Printf("%s(%s): validator: no replica position available", db.Path(), r.Name())
continue
}

Expand Down Expand Up @@ -1008,7 +1012,7 @@ func (r *Replica) Restore(ctx context.Context, opt RestoreOptions) (err error) {
// Ensure logger exists.
logger := opt.Logger
if logger == nil {
logger = log.New(ioutil.Discard, "", 0)
logger = r.Logger
}

logPrefix := r.Name()
Expand Down

0 comments on commit 18760d2

Please sign in to comment.