v1.1.6 rollup (#218)

* inceased outbox VARCHAR column length to 2048 (#155)

* added reply to initiator functionality to sagas (#157)

* added generic handler metrics with message type as the label (#144)

* added generic handler metrics with message type as the label

* add handler name label to the metrics

* adding new metrics to the read me

* Fix handle empty body (#156)

* set the correct Type and Content-Type headers on out going messages (#160)

* set the correct Type and Content-Type headers on out going messages

* refactoring

* fixing ReplyToInitiator not working  when initiator sends a message via the RPC interface (#163)


* fixing ReplyToInitiator not working  when initiator sends a message via the RPC interface

* Improved wording of saga documentation article (#164)

* better wording for documentation

* added golangcli lint configuration and fixed linting failures (#165)

* fixed logging issues (#167)

* allow getting the saga id of the current invoked saga (#168)

* setting the target saga id on the saga correlation id field (#171)

* support emperror (#174)

* setting the target saga id on the saga correlation id field

* added emperror support

* Fix logging and added logging documentation (#176)

* fixed logging issues and added documentation

logging via the invocation interface was broken and did not add
contextual data related to the invocation due to a bug in the way the Glogged structure is currently implemented.

Also added documentation on how logging should be done within a
handler including adding context to returned errors so that data gets logged

* added missing documentation file

* added documentation on serialization support (#177)

* fixed emperror url format

* added serialization documentation

* added documentation for messaging patterns, retries and transactional processing (#181)

* fixed emperror url format

* added serialization documentation

* added documentation for message semantics, retries and transactions

* Fix docmentation (#182)

* fixed emperror url format

* added serialization documentation

* added documentation for message semantics, retries and transactions

* fixing tx documentation page

* Added sample application (#184)

* Update README.md

* added ability to configure outbox (#186)

* fixing issues with invoking the GlobalrawMessageHandler (#189)

Fixing the following issues:

#187
#188

* Saga bug fixes (#198)

* fixing the way that the target service was resolved

closes #195

#195

* Fixing the value of the StartedBy field when creating a new saga
instance

closes #194
#194

* rolling back checking if replying to an event to maintain backward compatibility

* logging a Warn instead of rejecting the message when saga not found in store

closes #196
#196

* fixing minor tech debt issues (#199)

* v1.1.5 rollup to master (#185)

* fix(bug:200) logs are now being reported correctly

Fixes issue #200
Also updated go.mod for newer versions of dependancies

* added metrics for transactional outbox (#193)

* added metrics for transactional outbox

The follwoing metrics were added
outbox_total_records: reports the total amount of records currently in the outbox
outbox_pending_delivery: reports the total amount of records pending delivery currently in the outbox
outbox_pending_removal: reports the total amount of records that were sent and pending removal currently in the outbox

* reading status and count fields in the correct order from rows

* service name now gets added to log entries when a custom logger is set or in saga store (#204)

* service name now gets added to log entries when a custom logger is set

#200

* fixing issue that the saga store was not adding the service to its log
entires

#206

* Fix logging (#207)

* service name now gets added to log entries when a custom logger is set

#200

* fixing issue that the saga store was not adding the service to its log
entires

#206

* fixing minor issue with logging saga store initialization

* Add support for setting the idempotency key on a BusMessage (#208)

* Added x-idempotency-key header and the ability for client code to set it

#106

* Set the value of BusMessage.ID as the default value of BusMessage.IdempotencyKey

* fixing issue with txoutbox failing to deliver a message after 50 (#205)

failed attempts

#203

* Allow correctly replaying Events (#190)

* refactored returnDeadToQueue to allow correctly returning Events as well as Messages(commands)

* fixed lint issues

* code review fixes

* more review comments

* minor change to force coveralls to rebuild

* returnDeadToQueue - changed routing-key to always be that routing-key of the first death, not the latest death, also refactored naming and added comment for clarity

* moved getRoutingParamsFromDelivery to worker.go

* fixed type in migration name attribute (#213)

* adding all logging context data to worker and saga log entries (#215)

* adding all logging context data to worker and saga log entries

* added logging with context when command or reply received for saga def but no saga correlation id found

* removing minor discrepancies and updating documentation

docs/LOGGING.md

@@ -9,12 +9,16 @@ annotated with the following contextual data (added as logrus fields to the log
 allowing for a better debugging experience.
 
 - _service: the service name
+- correlation_id: the correlation id set for the message
+- exchange: the exchange the message was published to
 - handler_name: the name of the handler being invoked
+- idempotency_key: the idempotency key set for the message
 - message_id: the id of the processed message
 - message_name: the type of the message that is being processed
 - routing_key: the routing_key of the message
 - saga_id: the id of the saga instance being invoked
-- saga_def: the type of the saga that is being invoked 
+- saga_def: the type of the saga that is being invoked
+- worker: the worker identifier that is processing the message
 
 ```go
 

docs/METRICS.md

@@ -10,3 +10,7 @@ grabbit exposes and reports the following metrics to Prometheus
 | grabbit       | handlers      | latency                           | records the execution time of each run of a handler, having the handler's name, message type as labels|
 | grabbit       | messages      | rejected_messages                 | increments each time a message gets rejected                                |
 | grabbit       | saga          | timedout_sagas                   | counting the number of timedout saga instances                              |
+| grabbit       | outbox        | outbox_total_records                   | reports the total amount of records currently in the outbox                              |
+| grabbit       | outbox        | outbox_pending_delivery                   | reports the total amount of records pending delivery currently in the outbox                              |
+| grabbit       | outbox        | outbox_pending_removal                   | reports the total amount of records that were sent and pending removal currently in the outbox                              |
+

docs/SAGA.md

@@ -109,7 +109,7 @@ func (s *BookVacationSaga) HandleBookVacationCommand(invocation gbus.Invocation,
 	reply := gbus.NewBusMessage(BookVacationReply{
 		BookingId: s.BookingId})
 	//reply to the command so the caller can continue with his execution flow	
-	return invocation.Reply(noopTraceContext(), reply)
+	return invocation.Reply(context.Background(), reply)
 }
 ```
 

examples/vacation_app/cmd/client.go

@@ -14,7 +14,6 @@ import (
 	"github.com/wework/grabbit/gbus"
 )
 
-
 var runClientCmd = &cobra.Command{
 	Use:   "client",
 	Short: "Run the client app",
@@ -57,8 +56,7 @@ var runClientCmd = &cobra.Command{
 func HandleBookingComplete(invocation gbus.Invocation, message *gbus.BusMessage) error {
 	bookingComplete := message.Payload.(*messages.BookingComplete)
 	if bookingComplete.Success {
-		fmt.Printf("booking completed succesfully\n")
-
+		fmt.Printf("booking completed successfully\n")
 	} else {
 		fmt.Printf("failed to book vacation\n")
 	}

examples/vacation_app/cmd/flights.go

@@ -4,16 +4,15 @@ import (
 	"bufio"
 	"fmt"
 	"os"
-	"vacation_app/trace"
+  
 	"vacation_app/messages"
-
+	"vacation_app/trace"
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
 	"github.com/wework/grabbit/gbus"
 )
 
 
-
 var runFlightsgServiceCmd = &cobra.Command{
 	Use:   "flights",
 	Short: "Run the flights service",
@@ -30,7 +29,8 @@ var runFlightsgServiceCmd = &cobra.Command{
 		gb := createBus(svcName)
 
 		gb.HandleMessage(messages.BookFlightsCmd{}, HandleBookFlightCommand)
-	gb.HandleMessage(messages.CancelFlightsCmd{}, HandleCancelFlightCommand)
+
+		gb.HandleMessage(messages.CancelFlightsCmd{}, HandleCancelFlightCommand)
 
 		gb.Start()
 		defer gb.Shutdown()

gbus/abstractions.go

@@ -20,12 +20,6 @@ const (
 	EVT Semantics = "evt"
 )
 
-//BusConfiguration provides configuration passed to the bus builder
-type BusConfiguration struct {
-	MaxRetryCount     uint
-	BaseRetryDuration int
-}
-
 //Bus interface provides the majority of functionality to Send, Reply and Publish messages to the Bus
 type Bus interface {
 	HandlerRegister

gbus/builder/builder.go

@@ -29,13 +29,15 @@ type defaultBuilder struct {
 	dbPingTimeout    time.Duration
 	usingPingTimeout bool
 	logger           logrus.FieldLogger
+	busCfg           gbus.BusConfiguration
 }
 
 func (builder *defaultBuilder) Build(svcName string) gbus.Bus {
 
 	gb := &gbus.DefaultBus{
 		AmqpConnStr:   builder.connStr,
 		PrefetchCount: builder.PrefetchCount,
+		Glogged:       &gbus.Glogged{},
 
 		SvcName:              svcName,
 		PurgeOnStartup:       builder.purgeOnStartup,
@@ -53,11 +55,13 @@ func (builder *defaultBuilder) Build(svcName string) gbus.Bus {
 		Confirm:              builder.confirm,
 	}
 
+	var finalLogger logrus.FieldLogger
 	if builder.logger != nil {
-		gb.SetLogger(builder.logger)
+		finalLogger = builder.logger.WithField("_service", gb.SvcName)
 	} else {
-		gb.SetLogger(logrus.New())
+		finalLogger = logrus.WithField("_service", gb.SvcName)
 	}
+	gb.SetLogger(finalLogger)
 
 	if builder.workerNum < 1 {
 		gb.WorkerNum = 1
@@ -72,6 +76,7 @@ func (builder *defaultBuilder) Build(svcName string) gbus.Bus {
 	switch builder.txnlProvider {
 
 	case "mysql":
+		providerLogger := gb.Log().WithField("provider", "mysql")
 		mysqltx, err := mysql.NewTxProvider(builder.txConnStr)
 		if err != nil {
 			panic(err)
@@ -82,14 +87,15 @@ func (builder *defaultBuilder) Build(svcName string) gbus.Bus {
 
 		//TODO move purge logic into the NewSagaStore factory method
 		sagaStore = mysql.NewSagaStore(gb.SvcName, mysqltx)
+		sagaStore.SetLogger(providerLogger)
 		if builder.purgeOnStartup {
 			err := sagaStore.Purge()
 			if err != nil {
 				panic(err)
 			}
 		}
-		gb.Outbox = mysql.NewOutbox(gb.SvcName, mysqltx, builder.purgeOnStartup)
-		gb.Outbox.SetLogger(gb.Log())
+		gb.Outbox = mysql.NewOutbox(gb.SvcName, mysqltx, builder.purgeOnStartup, builder.busCfg.OutboxCfg)
+		gb.Outbox.SetLogger(providerLogger)
 		timeoutManager = mysql.NewTimeoutManager(gb, gb.TxProvider, gb.Log, svcName, builder.purgeOnStartup)
 
 	default:
@@ -109,6 +115,7 @@ func (builder *defaultBuilder) Build(svcName string) gbus.Bus {
 	}
 	glue := saga.NewGlue(gb, sagaStore, svcName, gb.TxProvider, gb.Log, timeoutManager)
 	glue.SetLogger(gb.Log())
+	sagaStore.SetLogger(glue.Log())
 	gb.Glue = glue
 	return gb
 }
@@ -182,6 +189,7 @@ func (builder *defaultBuilder) ConfigureHealthCheck(timeoutInSeconds time.Durati
 
 func (builder *defaultBuilder) WithConfiguration(config gbus.BusConfiguration) gbus.Builder {
 
+	builder.busCfg = config
 	gbus.MaxRetryCount = config.MaxRetryCount
 
 	if config.BaseRetryDuration > 0 {
@@ -207,6 +215,7 @@ type Nu struct {
 //Bus inits a new BusBuilder
 func (Nu) Bus(brokerConnStr string) gbus.Builder {
 	return &defaultBuilder{
+		busCfg:          gbus.BusConfiguration{},
 		PrefetchCount:   1,
 		connStr:         brokerConnStr,
 		serializer:      serialization.NewGobSerializer(),

gbus/bus.go

@@ -75,7 +75,9 @@ var (
 	//for a random retry time. Default is 10 but it is configurable.
 	BaseRetryDuration = 10 * time.Millisecond
 	//RPCHeaderName used to define the header in grabbit for RPC
-	RPCHeaderName = "x-grabbit-msg-rpc-id"
+	RPCHeaderName                  = "x-grabbit-msg-rpc-id"
+	ResurrectedHeaderName          = "x-resurrected-from-death"
+	FirstDeathRoutingKeyHeaderName = "x-first-death-routing-key"
 )
 
 func (b *DefaultBus) createRPCQueue() (amqp.Queue, error) {
@@ -488,23 +490,70 @@ func (b *DefaultBus) sendWithTx(ctx context.Context, ambientTx *sql.Tx, toServic
 
 func (b *DefaultBus) returnDeadToQueue(ctx context.Context, ambientTx *sql.Tx, publishing *amqp.Publishing) error {
 	if !b.started {
-		return errors.New("bus not strated or already shutdown, make sure you call bus.Start() before sending messages")
+		return errors.New("bus not started or already shutdown, make sure you call bus.Start() before sending messages")
+	}
+
+	targetQueue, ok := publishing.Headers["x-first-death-queue"].(string)
+	if !ok {
+		return fmt.Errorf("bad x-first-death-queue field - %v", publishing.Headers["x-first-death-queue"])
+	}
+	exchange, ok := publishing.Headers["x-first-death-exchange"].(string)
+	if !ok {
+		return fmt.Errorf("bad x-first-death-exchange field - %v", publishing.Headers["x-first-death-exchange"])
+	}
+	routingKey, err := extractFirstDeathRoutingKey(publishing.Headers)
+	if err != nil {
+		return err
 	}
-	//publishing.Headers.
-	exchange := fmt.Sprintf("%v", publishing.Headers["x-first-death-exchange"])
-	routingKey := fmt.Sprintf("%v", publishing.Headers["x-first-death-queue"])
+
+	publishing.Headers[FirstDeathRoutingKeyHeaderName] = routingKey // Set the original death routing key to be used later for replaying
+	publishing.Headers[ResurrectedHeaderName] = true                // mark message as resurrected
+	// publishing.Headers["x-first-death-exchange"] is not deleted and kept as is
 
 	delete(publishing.Headers, "x-death")
 	delete(publishing.Headers, "x-first-death-queue")
 	delete(publishing.Headers, "x-first-death-reason")
-	delete(publishing.Headers, "x-first-death-exchange")
+
+	b.Log().
+		WithField("message_id", publishing.MessageId).
+		WithField("target_queue", targetQueue).
+		WithField("first_death_routing_key", routingKey).
+		WithField("first_death_exchange", exchange).
+		Info("returning dead message to queue...")
 
 	send := func(tx *sql.Tx) error {
-		return b.publish(tx, exchange, routingKey, publishing)
+		// Publishing a "resurrected" message is done directly to the target queue using the default exchange
+		return b.publish(tx, "", targetQueue, publishing)
 	}
 	return b.withTx(send, ambientTx)
 }
 
+// Extracts the routing key of the first death of the message. "x-death" header contains a list of "deaths" that happened to this message, with
+// the most recent death always being first in the list, so fhe first death is the last one. More information: https://www.rabbitmq.com/dlx.html
+func extractFirstDeathRoutingKey(headers amqp.Table) (result string, err error) {
+	xDeathList, ok := headers["x-death"].([]interface{})
+	if !ok {
+		return "", fmt.Errorf("failed extracting routing-key from headers, bad 'x-death' field - %v", headers["x-death"])
+	}
+
+	xDeath, ok := xDeathList[0].(amqp.Table)
+	if !ok {
+		return "", fmt.Errorf("failed extracting routing-key from headers, bad 'x-death' field - %v", headers["x-death"])
+	}
+
+	routingKeys, ok := xDeath["routing-keys"].([]interface{})
+	if !ok {
+		return "", fmt.Errorf("failed extracting routing-key from headers, bad 'routing-keys' field - %v", xDeath["routing-keys"])
+	}
+
+	routingKey, ok := routingKeys[len(routingKeys)-1].(string)
+	if !ok {
+		return "", fmt.Errorf("failed extracting routing-key from headers, bad 'routing-keys' field - %v", xDeath["routing-keys"])
+	}
+
+	return routingKey, nil
+}
+
 //Publish implements GBus.Publish(topic, message)
 func (b *DefaultBus) Publish(ctx context.Context, exchange, topic string, message *BusMessage, policies ...MessagePolicy) error {
 	return b.publishWithTx(ctx, nil, exchange, topic, message, policies...)
@@ -717,13 +766,3 @@ type rpcPolicy struct {
 func (p rpcPolicy) Apply(publishing *amqp.Publishing) {
 	publishing.Headers[RPCHeaderName] = p.rpcID
 }
-
-//Log returns the default logrus.FieldLogger for the bus via the Glogged helper
-func (b *DefaultBus) Log() logrus.FieldLogger {
-	if b.Glogged == nil {
-		b.Glogged = &Glogged{
-			log: logrus.WithField("_service", b.SvcName),
-		}
-	}
-	return b.Glogged.Log()
-}

gbus/configuration.go

@@ -0,0 +1,32 @@
+package gbus
+
+import "time"
+
+//BusConfiguration provides configuration passed to the bus builder
+type BusConfiguration struct {
+	MaxRetryCount     uint
+	BaseRetryDuration int //TODO:Change type to uint
+	OutboxCfg         OutboxConfiguration
+}
+
+//OutboxConfiguration configures the transactional outbox
+type OutboxConfiguration struct {
+	/*
+		Ackers the number of goroutines configured to drain incoming ack/nack signals from the broker.
+		Increase this value if you are experiencing deadlocks.
+		Default is 10
+	*/
+	Ackers uint
+	//PageSize is the amount of pending messsage records the outbox selects from the database every iteration, the default is 500
+	PageSize uint
+	//MetricsInterval is the duration the outbox waits between each metrics report, default is 15 seconds
+	MetricsInterval time.Duration
+	//SendInterval is the duration the outbox waits before each iteration, default is 1 second
+	SendInterval time.Duration
+	/*
+		ScavengeInterval is the duration the outbox waits before attempting to re-send messages that
+		were already sent to the broker but were not yet confirmed.
+		Default is 60 seconds
+	*/
+	ScavengeInterval time.Duration
+}