Merge pull request #33 from hellofresh/patch/logger-update

Update logger interface

Author: boekkooi-fresh

Gopkg.lock

@@ -84,10 +84,10 @@ func (m *MetadataMatcher) Matches(metadata metadata.Metadata) bool {
 		valid, err := c.Matches(metadata.Value(c.field))
 		if err != nil {
 			if m.logger != nil {
-				m.logger.
-					WithError(err).
-					WithField("field", c.field).
-					Warn("metadata constraint failed with error")
+				m.logger.Warn("metadata constraint failed with error", func(e goengine.LoggerEntry) {
+					e.Error(err)
+					e.String("field", c.field)
+				})
 			}
 			return false
 		}

driver/inmemory/matcher.go

@@ -124,10 +124,10 @@ func (b *BackgroundProcessor) startProcessor(ctx context.Context, handler Proces
 		case notification := <-b.queue:
 			// Execute the notification
 			if err := handler(ctx, notification, b.Queue); err != nil {
-				b.logger.
-					WithError(err).
-					WithField("notification", notification).
-					Error("the ProcessHandler produced an error")
+				b.logger.Error("the ProcessHandler produced an error", func(e goengine.LoggerEntry) {
+					e.Error(err)
+					e.Any("notification", notification)
+				})
 			}
 		}
 	}

driver/sql/internal/background_processor.go

@@ -85,7 +85,9 @@ func (s *NotificationProjector) Execute(ctx context.Context, notification *drive
 	}
 	defer func() {
 		if err := projectConn.Close(); err != nil {
-			s.logger.WithError(err).Warn("failed to db close project connection")
+			s.logger.Warn("failed to db close project connection", func(e goengine.LoggerEntry) {
+				e.Error(err)
+			})
 		}
 	}()
 
@@ -95,7 +97,9 @@ func (s *NotificationProjector) Execute(ctx context.Context, notification *drive
 	}
 	defer func() {
 		if err := streamConn.Close(); err != nil {
-			s.logger.WithError(err).Warn("failed to db close stream connection")
+			s.logger.Warn("failed to db close stream connection", func(e goengine.LoggerEntry) {
+				e.Error(err)
+			})
 		}
 	}()
 
@@ -110,8 +114,6 @@ func (s *NotificationProjector) project(
 	streamConn *sql.Conn,
 	notification *driverSQL.ProjectionNotification,
 ) error {
-	logger := s.logger.WithField("notification", notification)
-
 	// Acquire the projection
 	releaseLock, rawState, err := s.storage.Acquire(ctx, conn, notification)
 	if err != nil {
@@ -126,7 +128,10 @@ func (s *NotificationProjector) project(
 	}
 	defer func() {
 		if err := eventStream.Close(); err != nil {
-			logger.WithError(err).Warn("failed to close the event stream")
+			s.logger.Warn("failed to close the event stream", func(e goengine.LoggerEntry) {
+				e.Any("notification", notification)
+				e.Error(err)
+			})
 		}
 	}()
 
@@ -167,9 +172,10 @@ func (s *NotificationProjector) projectStream(
 		// Resolve the payload event name
 		eventName, err := s.resolver.ResolveName(msg.Payload())
 		if err != nil {
-			s.logger.
-				WithField("payload", msg.Payload()).
-				Warn("skipping event: unable to resolve payload name")
+			s.logger.Warn("skipping event: unable to resolve payload name", func(e goengine.LoggerEntry) {
+				e.Error(err)
+				e.Any("payload", msg.Payload())
+			})
 			continue
 		}
 

driver/sql/internal/notification_projector.go

@@ -42,9 +42,9 @@ func (e *ConjoinedEventStore) AppendTo(ctx context.Context, streamName goengine.
 	}
 	defer func() {
 		if err := tx.Rollback(); err != nil && err != sql.ErrTxDone {
-			e.logger.
-				WithError(err).
-				Error("could not rollback transaction")
+			e.logger.Error("could not rollback transaction", func(e goengine.LoggerEntry) {
+				e.Error(err)
+			})
 		}
 	}()
 
@@ -58,9 +58,10 @@ func (e *ConjoinedEventStore) AppendTo(ctx context.Context, streamName goengine.
 		// Resolve the payload event name
 		eventName, err := e.resolver.ResolveName(msg.Payload())
 		if err != nil {
-			e.logger.
-				WithField("payload", msg.Payload()).
-				Warn("skipping event: unable to resolve payload name")
+			e.logger.Warn("skipping event: unable to resolve payload name", func(e goengine.LoggerEntry) {
+				e.Error(err)
+				e.Any("payload", msg.Payload())
+			})
 			continue
 		}
 

driver/sql/postgres/conjoined_eventstore.go

@@ -97,10 +97,10 @@ func (e *EventStore) Create(ctx context.Context, streamName goengine.StreamName)
 		}
 
 		if errRollback := tx.Rollback(); errRollback != nil {
-			e.logger.
-				WithError(errRollback).
-				WithField("query", q).
-				Error("could not rollback transaction")
+			e.logger.Error("could not rollback transaction", func(e goengine.LoggerEntry) {
+				e.Error(errRollback)
+				e.String("query", q)
+			})
 		}
 
 		return err
@@ -217,24 +217,21 @@ func (e *EventStore) AppendToWithExecer(ctx context.Context, conn driverSQL.Exec
 		data...,
 	)
 	if err != nil {
-		e.logger.
-			WithError(err).
-			WithFields(goengine.Fields{
-				"streamName":   streamName,
-				"streamEvents": streamEvents,
-			}).
-			Warn("failed to insert messages into the event stream")
+		e.logger.Warn("failed to insert messages into the event stream", func(e goengine.LoggerEntry) {
+			e.Error(err)
+			e.String("streamName", string(streamName))
+			e.Any("streamEvents", streamEvents)
+		})
 
 		return err
 	}
 
-	e.logger.
-		WithFields(goengine.Fields{
-			"streamName":   streamName,
-			"streamEvents": streamEvents,
-			"result":       result,
-		}).
-		Debug("inserted messages into the event stream")
+	e.logger.Debug("inserted messages into the event stream", func(e goengine.LoggerEntry) {
+		e.Error(err)
+		e.String("streamName", string(streamName))
+		e.Any("streamEvents", streamEvents)
+		e.Any("result", result)
+	})
 
 	return nil
 }
@@ -306,10 +303,10 @@ func (e *EventStore) tableExists(ctx context.Context, tableName string) bool {
 	).Scan(&exists)
 
 	if err != nil {
-		e.logger.
-			WithError(err).
-			WithField("table", tableName).
-			Warn("error on reading from information_schema")
+		e.logger.Warn("error on reading from information_schema", func(e goengine.LoggerEntry) {
+			e.Error(err)
+			e.String("table", tableName)
+		})
 
 		return false
 	}

driver/sql/postgres/eventstore.go

@@ -60,7 +60,9 @@ func NewAggregateProjector(
 	if logger == nil {
 		logger = goengine.NopLogger
 	}
-	logger = logger.WithField("projection", projection)
+	logger = logger.WithFields(func(e goengine.LoggerEntry) {
+		e.String("projection", projection.Name())
+	})
 
 	processor, err := internal.NewBackgroundProcessor(10, 32, logger)
 	if err != nil {
@@ -158,20 +160,24 @@ func (a *AggregateProjector) processNotification(
 	}
 
 	// Resolve the action to take based on the error that occurred
-	logger := a.logger.WithError(err).WithField("notification", notification)
+	logFields := func(e goengine.LoggerEntry) {
+		e.Error(err)
+		e.Int64("notification.no", notification.No)
+		e.String("notification.aggregate_id", notification.AggregateID)
+	}
 	switch resolveErrorAction(a.projectionErrorHandler, notification, err) {
 	case errorFail:
-		logger.Debug("ProcessHandler->ErrorHandler: marking projection as failed")
+		a.logger.Debug("ProcessHandler->ErrorHandler: marking projection as failed", logFields)
 		return a.markProjectionAsFailed(ctx, notification)
 	case errorIgnore:
-		logger.Debug("ProcessHandler->ErrorHandler: ignoring error")
+		a.logger.Debug("ProcessHandler->ErrorHandler: ignoring error", logFields)
 		return nil
 	case errorRetry:
-		logger.Debug("ProcessHandler->ErrorHandler: re-queueing notification")
+		a.logger.Debug("ProcessHandler->ErrorHandler: re-queueing notification", logFields)
 		return queue(ctx, notification)
 	}
 
-	logger.Debug("ProcessHandler->ErrorHandler: error fallthrough")
+	a.logger.Debug("ProcessHandler->ErrorHandler: error fallthrough", logFields)
 	return err
 }
 
@@ -183,7 +189,9 @@ func (a *AggregateProjector) triggerOutOfSyncProjections(ctx context.Context, qu
 	}
 	defer func() {
 		if err := conn.Close(); err != nil {
-			a.logger.WithError(err).Warn("failed to db close LoadOutOfSync connection")
+			a.logger.Warn("failed to db close LoadOutOfSync connection", func(e goengine.LoggerEntry) {
+				e.Error(err)
+			})
 		}
 	}()
 
@@ -193,7 +201,9 @@ func (a *AggregateProjector) triggerOutOfSyncProjections(ctx context.Context, qu
 	}
 	defer func() {
 		if err := rows.Close(); err != nil {
-			a.logger.WithError(err).Error("failed to close LoadOutOfSync rows")
+			a.logger.Error("failed to close LoadOutOfSync rows", func(e goengine.LoggerEntry) {
+				e.Error(err)
+			})
 		}
 	}()
 
@@ -219,13 +229,19 @@ func (a *AggregateProjector) triggerOutOfSyncProjections(ctx context.Context, qu
 			AggregateID: aggregateID,
 		}
 
-		logger := a.logger.WithField("notification", notification)
 		if err := queue(ctx, notification); err != nil {
-			logger.WithError(err).Error("failed to queue notification")
+			a.logger.Error("failed to queue notification", func(e goengine.LoggerEntry) {
+				e.Error(err)
+				e.Int64("notification.no", notification.No)
+				e.String("notification.aggregate_id", notification.AggregateID)
+			})
 			return err
 		}
 
-		a.logger.Debug("send catchup")
+		a.logger.Debug("send catchup", func(e goengine.LoggerEntry) {
+			e.Int64("notification.no", notification.No)
+			e.String("notification.aggregate_id", notification.AggregateID)
+		})
 	}
 
 	return rows.Close()
@@ -239,7 +255,9 @@ func (a *AggregateProjector) markProjectionAsFailed(ctx context.Context, notific
 
 	defer func() {
 		if err := conn.Close(); err != nil {
-			a.logger.WithError(err).Warn("failed to db close failure connection")
+			a.logger.Warn("failed to db close failure connection", func(e goengine.LoggerEntry) {
+				e.Error(err)
+			})
 		}
 	}()
 

driver/sql/postgres/projector_aggregate.go

@@ -129,10 +129,11 @@ func (a *aggregateProjectionStorage) PersistState(conn driverSQL.Execer, notific
 		return err
 	}
 
-	a.logger.WithFields(goengine.Fields{
-		"notification": notification,
-		"state":        state,
-	}).Debug("updated projection state")
+	a.logger.Debug("updated projection state", func(e goengine.LoggerEntry) {
+		e.Int64("notification.no", notification.No)
+		e.String("notification.aggregate_id", notification.AggregateID)
+		e.Any("state", state)
+	})
 	return nil
 }
 
@@ -149,7 +150,10 @@ func (a *aggregateProjectionStorage) Acquire(
 	conn *sql.Conn,
 	notification *driverSQL.ProjectionNotification,
 ) (func(), *driverSQL.ProjectionRawState, error) {
-	logger := a.logger.WithField("notification", notification)
+	logFields := func(e goengine.LoggerEntry) {
+		e.Int64("notification.no", notification.No)
+		e.String("notification.aggregate_id", notification.AggregateID)
+	}
 	aggregateID := notification.AggregateID
 
 	res := conn.QueryRowContext(ctx, a.queryAcquireLock, aggregateID, notification.No)
@@ -178,9 +182,12 @@ func (a *aggregateProjectionStorage) Acquire(
 		// The projection was locked by another process that died and for this reason not unlocked
 		// In this case a application needs to decide what to do to avoid invalid projection states
 		if err := a.releaseProjectionConnectionLock(conn, aggregateID); err != nil {
-			logger.WithError(err).Error("failed to release lock for a projection with a locked row")
+			a.logger.Error("failed to release lock for a projection with a locked row", func(e goengine.LoggerEntry) {
+				logFields(e)
+				e.Error(err)
+			})
 		} else {
-			logger.Debug("released connection lock for a locked projection")
+			a.logger.Debug("released connection lock for a locked projection", logFields)
 		}
 
 		return nil, nil, driverSQL.ErrProjectionPreviouslyLocked
@@ -190,20 +197,26 @@ func (a *aggregateProjectionStorage) Acquire(
 	_, err := conn.ExecContext(ctx, a.querySetRowLocked, aggregateID, true)
 	if err != nil {
 		if releaseErr := a.releaseProjection(conn, aggregateID); releaseErr != nil {
-			logger.WithError(releaseErr).Error("failed to release lock while setting projection rows as locked")
+			a.logger.Error("failed to release lock while setting projection rows as locked", func(e goengine.LoggerEntry) {
+				logFields(e)
+				e.Error(releaseErr)
+			})
 		} else {
-			logger.Debug("failed to set projection as locked")
+			a.logger.Debug("failed to set projection as locked", logFields)
 		}
 
 		return nil, nil, err
 	}
-	logger.Debug("acquired projection lock")
+	a.logger.Debug("acquired projection lock", logFields)
 
 	return func() {
 		if err := a.releaseProjection(conn, aggregateID); err != nil {
-			logger.WithError(err).Error("failed to release projection lock")
+			a.logger.Error("failed to release projection lock", func(e goengine.LoggerEntry) {
+				logFields(e)
+				e.Error(err)
+			})
 		} else {
-			logger.Debug("released projection lock")
+			a.logger.Debug("released projection lock", logFields)
 		}
 	}, &driverSQL.ProjectionRawState{ProjectionState: rawState, Position: position}, nil
 }