Updated logrus to 1.0, simplified log messages

Author: vgarvardt

.travis.yml

@@ -11,7 +11,7 @@ go:
 
 install:
   - go get -v -d
-  - go get github.com/onsi/ginkgo/ginkgo
-  - go get github.com/onsi/gomega
-  
+  - go get -u github.com/onsi/ginkgo/ginkgo
+  - go get -u github.com/onsi/gomega
+
 script: ginkgo -r

aggregate_repository.go

@@ -2,7 +2,8 @@ package goengine
 
 import (
 	"fmt"
-	log "github.com/Sirupsen/logrus"
+
+	log "github.com/sirupsen/logrus"
 )
 
 type AggregateRepository interface {
@@ -21,7 +22,7 @@ func NewPublisherRepository(eventStore EventStore, eventBus VersionedEventPublis
 }
 
 func (r *PublisherRepository) Load(id string, streamName StreamName) (*EventStream, error) {
-	log.Debugf("Loading events from %s stream for aggregate %s", streamName, id)
+	log.WithFields(log.Fields{"stream": streamName, "id": id}).Debug("Loading events from stream for aggregate")
 	stream, err := r.EventStore.GetEventsFor(streamName, id)
 	if nil != err {
 		return nil, err
@@ -33,14 +34,14 @@ func (r *PublisherRepository) Load(id string, streamName StreamName) (*EventStre
 func (r *PublisherRepository) Save(aggregateRoot AggregateRoot, streamName StreamName) error {
 	events := aggregateRoot.GetUncommittedEvents()
 	eventStream := NewEventStream(streamName, events)
-	log.Debugf("Saving %d events to %s stream", len(events), streamName)
+	log.WithFields(log.Fields{"count": len(events), "stream": streamName}).Debug("Saving events to stream")
 	err := r.EventStore.Append(eventStream)
 	if nil != err {
 		return err
 	}
 
 	if nil == r.EventBus {
-		log.Debug("Event bus not detected, skiping publishing events")
+		log.Debug("Event bus not detected, skipping publishing events")
 		return nil
 	}
 
@@ -52,7 +53,7 @@ func (r *PublisherRepository) Save(aggregateRoot AggregateRoot, streamName Strea
 }
 
 func (r *PublisherRepository) Reconstitute(id string, source AggregateRoot, streamName StreamName) error {
-	log.Debugf("Reconstituting aggregate %s from %s stream", id, streamName)
+	log.WithFields(log.Fields{"stream": streamName, "id": id}).Debug("Reconstituting aggregate from stream")
 	stream, err := r.Load(id, streamName)
 	if nil != err {
 		return err
@@ -68,6 +69,6 @@ func (r *PublisherRepository) Reconstitute(id string, source AggregateRoot, stre
 	}
 
 	source.SetVersion(events[len(events)-1].Version)
-	log.Debugf("Aggregate %s reconstituted", id)
+	log.WithField("id", id).Debug("Aggregate reconstituted")
 	return nil
 }

aggregate_root.go

@@ -3,9 +3,9 @@ package goengine
 import (
 	"fmt"
 
-	log "github.com/Sirupsen/logrus"
 	"github.com/hellofresh/goengine/reflection"
 	"github.com/pborman/uuid"
+	log "github.com/sirupsen/logrus"
 )
 
 type AggregateRoot interface {
@@ -17,10 +17,10 @@ type AggregateRoot interface {
 }
 
 type AggregateRootBased struct {
-	ID               string
-	version          int
-	source           interface{}
-	uncommitedEvents []*DomainMessage
+	ID                string
+	version           int
+	source            interface{}
+	uncommittedEvents []*DomainMessage
 }
 
 // NewAggregateRootBased constructor
@@ -46,18 +46,21 @@ func (r *AggregateRootBased) SetVersion(version int) {
 }
 
 func (r *AggregateRootBased) GetUncommittedEvents() []*DomainMessage {
-	stream := r.uncommitedEvents
-	r.uncommitedEvents = nil
-	log.Debugf("%d Uncommited events cleaned", len(stream))
+	stream := r.uncommittedEvents
+	r.uncommittedEvents = nil
+	log.WithField("count", len(stream)).Debug("Uncommitted events cleaned")
 
 	return stream
 }
 
 func (r *AggregateRootBased) Apply(event DomainEvent) {
 	t := reflection.TypeOf(event)
-	log.Debugf("source: %+v; MethodName: %+v; event: %+v", r.source, fmt.Sprintf("When%s", t.Name()), event)
-	reflection.CallMethod(r.source, fmt.Sprintf("When%s", t.Name()), event)
-	log.Debugf("Event %s applied", t.Name())
+	methodName := fmt.Sprintf("When%s", t.Name())
+
+	entry := log.WithFields(log.Fields{"source": fmt.Sprintf("%+v", r.source), "method": methodName, "event": fmt.Sprintf("%+v", event)})
+	entry.Debug("Applying event")
+	reflection.CallMethod(r.source, methodName, event)
+	entry.Debug("Event applied")
 }
 
 func (r *AggregateRootBased) RecordThat(event DomainEvent) {
@@ -68,6 +71,6 @@ func (r *AggregateRootBased) RecordThat(event DomainEvent) {
 
 func (r *AggregateRootBased) Record(event DomainEvent) {
 	message := RecordNow(r.ID, r.version, event)
-	r.uncommitedEvents = append(r.uncommitedEvents, message)
+	r.uncommittedEvents = append(r.uncommittedEvents, message)
 	log.Debug("Event recorded")
 }

cmd/goengine/main.go

@@ -3,11 +3,10 @@ package main
 import (
 	"os"
 
-	log "github.com/Sirupsen/logrus"
 	"github.com/hellofresh/goengine"
 	"github.com/hellofresh/goengine/mongodb"
 	"github.com/hellofresh/goengine/rabbit"
-
+	log "github.com/sirupsen/logrus"
 	"gopkg.in/mgo.v2"
 )
 
@@ -16,7 +15,7 @@ func main() {
 	var streamName goengine.StreamName = "test"
 
 	mongoDSN := os.Getenv("STORAGE_DSN")
-	log.Infof("Connecting to the database %s", mongoDSN)
+	log.WithField("dsn", mongoDSN).Debug("Connecting to the database")
 	session, err := mgo.Dial(mongoDSN)
 	if err != nil {
 		log.Panic(err)

domain_message.go

@@ -1,6 +1,9 @@
 package goengine
 
-import "time"
+import (
+	"fmt"
+	"time"
+)
 
 type DomainEvent interface {
 	OccurredOn() time.Time
@@ -13,6 +16,10 @@ type DomainMessage struct {
 	RecordedOn time.Time   `json:"recorded_on"`
 }
 
+func (dm *DomainMessage) String() string {
+	return fmt.Sprintf("DomainMessage{ ID: %s, Version: %d }", dm.ID, dm.Version)
+}
+
 func NewDomainMessage(id string, version int, payload DomainEvent, recordedOn time.Time) *DomainMessage {
 	return &DomainMessage{id, version, payload, recordedOn}
 }

event_dispatcher_manager.go

@@ -1,7 +1,7 @@
 package goengine
 
 import (
-	log "github.com/Sirupsen/logrus"
+	log "github.com/sirupsen/logrus"
 )
 
 // VersionedEventDispatchManager is responsible for coordinating receiving messages
@@ -52,22 +52,25 @@ func (m *VersionedEventDispatchManager) Listen(stop <-chan bool, exclusive bool)
 		// signifying successful processing of the message.
 		// This should eventually call an event handler. See NewVersionedEventDispatcher()
 		case event := <-receiveEventChannel:
-			log.Debugf("EventDispatchManager.DispatchEvent: %s", event.Event)
+			entry := log.WithField("event", event.Event)
+			entry.Debugf("EventDispatchManager.DispatchEvent")
 			if err := m.versionedEventDispatcher.DispatchEvent(event.Event); err != nil {
-				log.Println("Error dispatching event: ", err)
+				entry.WithError(err).Error("Error dispatching event")
 			}
 
 			event.ProcessedSuccessfully <- true
-			log.Debug("EventDispatchManager.DispatchSuccessful")
+			entry.Debug("EventDispatchManager.DispatchSuccessful")
+
 		case <-stop:
 			log.Debug("EventDispatchManager.Stopping")
 			closeSignal := make(chan error)
 			closeChannel <- closeSignal
-			defer log.Debug("EventDispatchManager.Stopped")
+			log.Debug("EventDispatchManager.Stopped")
 			return <-closeSignal
-		// Receiving on this channel signifys an error has occured worker processor side
+
+		// Receiving on this channel signifys an error has occurred worker processor side
 		case err := <-errorChannel:
-			log.Debugf("EventDispatchManager.ErrorReceived: %s", err)
+			log.WithError(err).Debugf("EventDispatchManager.ErrorReceived")
 			return err
 		}
 	}

rabbit/eventbus.go

@@ -6,9 +6,9 @@ import (
 	"fmt"
 	"time"
 
-	log "github.com/Sirupsen/logrus"
 	"github.com/hellofresh/goengine"
 	"github.com/hellofresh/goengine/reflection"
+	log "github.com/sirupsen/logrus"
 	"github.com/streadway/amqp"
 )
 
@@ -136,7 +136,7 @@ func (bus *EventBus) ReceiveEvents(options goengine.VersionedEventReceiverOption
 								conn, c, events, err = connR, cR, eventsR, errR
 							}
 
-							log.Debug(err)
+							log.WithError(err).Debug("Failed to reconnect to RabbitMQ")
 
 							return errR
 						}, 5)

rabbit/retry.go

@@ -4,26 +4,29 @@ import (
 	"math"
 	"time"
 
-	log "github.com/Sirupsen/logrus"
+	log "github.com/sirupsen/logrus"
 )
 
 type function func() error
 
 func exponential(operation function, maxRetries int) error {
 	var err error
-	var sleepTime int
+	var sleepMs int
 	for i := 0; i < maxRetries; i++ {
 		err = operation()
 		if err == nil {
 			return nil
 		}
+
 		if i == 0 {
-			sleepTime = 1
+			sleepMs = 1
 		} else {
-			sleepTime = int(math.Exp2(float64(i)) * 100)
+			sleepMs = int(math.Exp2(float64(i)) * 100)
 		}
-		time.Sleep(time.Duration(sleepTime) * time.Millisecond)
-		log.Debugf("Retry exponential: Attempt %d, sleep %d", i, sleepTime)
+
+		sleepTime := time.Duration(sleepMs) * time.Millisecond
+		time.Sleep(sleepTime)
+		log.WithFields(log.Fields{"attempt": i, "sleep": sleepTime.String()}).Debug("Retry exponential")
 	}
 
 	return err

type_registry.go

@@ -3,9 +3,9 @@ package goengine
 import (
 	"reflect"
 
-	log "github.com/Sirupsen/logrus"
 	"github.com/hellofresh/goengine/errors"
 	"github.com/hellofresh/goengine/reflection"
+	log "github.com/sirupsen/logrus"
 )
 
 // TypeRegistry is a registry for go types
@@ -35,15 +35,16 @@ func NewInMemmoryTypeRegistry() *InMemoryTypeRegistry {
 func (r *InMemoryTypeRegistry) RegisterType(i interface{}) {
 	rawType := reflection.TypeOf(i)
 	r.types[rawType.String()] = rawType
-	log.Debugf("Type %s was registered", rawType.String())
+	log.WithField("type", rawType.String()).Debug("Type was registered")
 }
 
 func (r *InMemoryTypeRegistry) RegisterAggregate(aggregate AggregateRoot, events ...interface{}) {
 	r.RegisterType(aggregate)
-	log.Debugf("Aggregate %s was registered", aggregate.GetID())
+	entry := log.WithField("aggregate", aggregate.GetID())
+	entry.Debug("Aggregate was registered")
 
 	r.RegisterEvents(events)
-	log.Debugf("%s events were registered for aggregate %s", len(events), aggregate.GetID())
+	entry.WithField("count", len(events)).Debug("Events were registered for aggregate")
 }
 
 func (r *InMemoryTypeRegistry) RegisterEvents(events ...interface{}) {
@@ -66,6 +67,6 @@ func (r *InMemoryTypeRegistry) Get(name string) (interface{}, error) {
 		return reflect.New(typ).Interface(), nil
 	}
 
-	log.Debugf("Type %s not found", name)
+	log.WithField("type", name).Debug("Type not found")
 	return nil, errors.ErrorTypeNotFound
 }