rollup 1.1.8 from v1.x to master (#225)

Guy Baron · vladshub · commit 7d2220c67061 · 2019-12-02T20:32:49.000+02:00
* 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) * 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) * 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 * Added sample application (#184) * fixing minor tech debt issues (#199) * fix(bug:200) logs are now being reported correctly Fixes issue #200 Also updated go.mod for newer versions of dependancies * fix wrongly set correlation ids when sending a command (#220) * adding the SagaCorrelationID header only for replies Fixing a bug where grabbit would set the message.CorrelationID and the message.SagaCorrelationID header even if the message was a command and not a reply. * fixing minor documentation issue * only sending headers over the wire if they contain a value (#222) * only sending headers over the wire if they contain a value fixing #221 * deleting file that was added by mistake * goodbye (#223) * goodbye * Fixing my last bug :) * Add saga latency metric (#224) records the execution time of the entire saga, from it's creation until deletion (in ms). having the "Service", "SagaType", "Initiator" as labels|
diff --git a/CORE_TEAM.md b/CORE_TEAM.md
@@ -0,0 +1,15 @@
+
+
+### Current members
+
+Vladislav Shub
+
+Avigail Berger
+
+Adi Yaskolka Weiss
+
+Daniel Witz
+
+### Past members
+
+Guy Baron
diff --git a/README.md b/README.md
@@ -10,7 +10,6 @@
 
 A lightweight transactional message bus on top of RabbitMQ supporting:
 
-
 1) Supported [Messaging Styles](https://github.com/wework/grabbit/blob/master/docs/MESSAGING.md) 
     - One Way (Fire and forget)
     - Publish/Subscribe
diff --git a/docs/LOGGING.md b/docs/LOGGING.md
@@ -54,7 +54,15 @@ func SomeHandler(invocation gbus.Invocation, message *gbus.BusMessage) error{
   }
 ```
 grabbit makes it easier handling these cases and reduce the repetitive task of logging
+<<<<<<< HEAD
+<<<<<<< HEAD
 these custom contextual attributes in cases of errors by integrating the [emperror errors package](https://github.com/emperror/errors).
+=======
+these custom contextual attributes in cases of errors by integrating the [emperror errors package] (https://github.com/emperror/errors).
+>>>>>>> Fix logging and added logging documentation (#176)
+=======
+these custom contextual attributes in cases of errors by integrating the [emperror errors package](https://github.com/emperror/errors).
+>>>>>>> added documentation on serialization support (#177)
 emperror drop-in replacement for the default errors package allows developers to add the needed contextual data on the error instance and have graabit log the error with all contextual attribute.
 
 ```go
diff --git a/docs/METRICS.md b/docs/METRICS.md
@@ -9,8 +9,9 @@ grabbit exposes and reports the following metrics to Prometheus
 | grabbit       | handlers      | result                            | records and counts each run of a handler, having the handler's name, message type and the result as labels|
 | 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                              |
+| grabbit       | saga          | timedout_sagas                    | counting the number of timedout saga instances                              |
+| grabbit       | saga          | latency                           | records the execution time of the entire saga, from it's creation until deletion (in ms). having the "Service", "SagaType", "Initiator" as labels|
+| 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                              |
 
diff --git a/examples/vacation_app/cmd/flights.go b/examples/vacation_app/cmd/flights.go
@@ -4,15 +4,14 @@ import (
 	"bufio"
 	"fmt"
 	"os"
-  
 	"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",
@@ -29,7 +28,6 @@ var runFlightsgServiceCmd = &cobra.Command{
 		gb := createBus(svcName)
 
 		gb.HandleMessage(messages.BookFlightsCmd{}, HandleBookFlightCommand)
-
 		gb.HandleMessage(messages.CancelFlightsCmd{}, HandleCancelFlightCommand)
 
 		gb.Start()
diff --git a/gbus/abstractions.go b/gbus/abstractions.go
@@ -18,6 +18,8 @@ const (
 	CMD Semantics = "cmd"
 	//EVT represenst a messge with event semantics in grabbit
 	EVT Semantics = "evt"
+	//REPLY represenst a messge with reply semantics in grabbit
+	REPLY Semantics = "reply"
 )
 
 //Bus interface provides the majority of functionality to Send, Reply and Publish messages to the Bus
diff --git a/gbus/builder/builder.go b/gbus/builder/builder.go
@@ -79,7 +79,8 @@ func (builder *defaultBuilder) Build(svcName string) gbus.Bus {
 		providerLogger := gb.Log().WithField("provider", "mysql")
 		mysqltx, err := mysql.NewTxProvider(builder.txConnStr)
 		if err != nil {
-			panic(err)
+			errMsg := fmt.Errorf("grabbit: transaction provider failed creating a transaction. %v", err)
+			panic(errMsg)
 		}
 		gb.TxProvider = mysqltx
 
@@ -91,7 +92,8 @@ func (builder *defaultBuilder) Build(svcName string) gbus.Bus {
 		if builder.purgeOnStartup {
 			err := sagaStore.Purge()
 			if err != nil {
-				panic(err)
+				errMsg := fmt.Errorf("grabbit: saga store faild to purge. error: %v", err)
+				panic(errMsg)
 			}
 		}
 		gb.Outbox = mysql.NewOutbox(gb.SvcName, mysqltx, builder.purgeOnStartup, builder.busCfg.OutboxCfg)
@@ -110,7 +112,8 @@ func (builder *defaultBuilder) Build(svcName string) gbus.Bus {
 	if builder.purgeOnStartup {
 		err := sagaStore.Purge()
 		if err != nil {
-			panic(err)
+			errMsg := fmt.Errorf("grabbit: saga store faild to purge. error: %v", err)
+			panic(errMsg)
 		}
 	}
 	glue := saga.NewGlue(gb, sagaStore, svcName, gb.TxProvider, gb.Log, timeoutManager)
diff --git a/gbus/bus.go b/gbus/bus.go
@@ -671,7 +671,7 @@ func (b *DefaultBus) publish(tx *sql.Tx, exchange, routingKey string, msg *amqp.
 	err := b.SafeWithRetries(publish, MaxRetryCount)
 
 	if err != nil {
-		b.Log().Printf("failed publishing message.\n error:%v", err)
+		b.Log().WithError(err).Error("failed publishing message")
 		return err
 	}
 	return err
diff --git a/gbus/messages.go b/gbus/messages.go
@@ -58,18 +58,30 @@ func GetMessageName(delivery amqp.Delivery) string {
 	return castToString(delivery.Headers["x-msg-name"])
 }
 
-//GetAMQPHeaders convert to AMQP headers Table everything but a payload
 func (bm *BusMessage) GetAMQPHeaders() (headers amqp.Table) {
-	headers = amqp.Table{}
-	headers["x-idempotency-key"] = bm.IdempotencyKey
-	headers["x-msg-saga-id"] = bm.SagaID
-	headers["x-msg-saga-correlation-id"] = bm.SagaCorrelationID
-	headers["x-grabbit-msg-rpc-id"] = bm.RPCID
-	headers["x-msg-name"] = bm.Payload.SchemaName()
+	headers = amqp.Table{
+		"x-msg-name":        bm.Payload.SchemaName(),
+		"x-idempotency-key": bm.IdempotencyKey,
+	}
+
+	/*
+	 only set the following headers if they contain a value
+	 https://github.com/wework/grabbit/issues/221
+	*/
+	setNonEmpty(headers, "x-msg-saga-id", bm.SagaID)
+	setNonEmpty(headers, "x-msg-saga-correlation-id", bm.SagaCorrelationID)
+	setNonEmpty(headers, "x-grabbit-msg-rpc-id", bm.RPCID)
 
 	return
 }
 
+func setNonEmpty(headers amqp.Table, headerName, headerValue string) {
+
+	if headerValue != "" {
+		headers[headerName] = headerValue
+	}
+}
+
 //SetFromAMQPHeaders convert from AMQP headers Table everything but a payload
 func (bm *BusMessage) SetFromAMQPHeaders(delivery amqp.Delivery) {
 	headers := delivery.Headers
diff --git a/gbus/metrics/saga_metrics.go b/gbus/metrics/saga_metrics.go
@@ -6,8 +6,22 @@ import (
 	io_prometheus_client "github.com/prometheus/client_model/go"
 )
 
-//SagaTimeoutCounter is the prometheus counter counting timed out saga instances
-var SagaTimeoutCounter = newSagaTimeoutCounter()
+var (
+	//SagaTimeoutCounter is the prometheus counter counting timed out saga instances
+	SagaTimeoutCounter = promauto.NewCounter(prometheus.CounterOpts{
+		Namespace: grabbitPrefix,
+		Subsystem: "saga",
+		Name:      "timedout_sagas",
+		Help:      "counting the number of timedout saga instances",
+	})
+	//SagaLatencySummary is the prometheus summary for the total duration of a saga
+	SagaLatencySummary = promauto.NewSummaryVec(prometheus.SummaryOpts{
+		Namespace: grabbitPrefix,
+		Subsystem: "saga",
+		Name:      "latency",
+		Help:      "The latency of the entire saga",
+	}, []string{"Service", "SagaType", "Initiator"})
+)
 
 //GetSagaTimeoutCounterValue gets the counter value of timed out sagas reported to prometheus
 func GetSagaTimeoutCounterValue() (float64, error) {
@@ -20,12 +34,3 @@ func GetSagaTimeoutCounterValue() (float64, error) {
 
 	return m.GetCounter().GetValue(), nil
 }
-
-func newSagaTimeoutCounter() prometheus.Counter {
-	return promauto.NewCounter(prometheus.CounterOpts{
-		Namespace: grabbitPrefix,
-		Subsystem: "saga",
-		Name:      "timedout_sagas",
-		Help:      "counting the number of timedout saga instances",
-	})
-}
diff --git a/gbus/saga/glue.go b/gbus/saga/glue.go
@@ -8,6 +8,7 @@ import (
 	"reflect"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/opentracing/opentracing-go"
 	slog "github.com/opentracing/opentracing-go/log"
@@ -256,6 +257,7 @@ func (imsm *Glue) completeOrUpdateSaga(tx *sql.Tx, instance *Instance) error {
 
 	if instance.isComplete() {
 		imsm.Log().WithField("saga_id", instance.ID).Info("saga has completed and will be deleted")
+		metrics.SagaLatencySummary.WithLabelValues(imsm.svcName, reflect.TypeOf(instance.UnderlyingInstance).String(), instance.StartedBy).Observe(float64(time.Since(instance.CreatedAt) / time.Millisecond))
 
 		deleteErr := imsm.sagaStore.DeleteSaga(tx, instance)
 		if deleteErr != nil {
diff --git a/gbus/saga/instance.go b/gbus/saga/instance.go
@@ -34,6 +34,9 @@ type Instance struct {
 	StartedByMessageID string
 	//StartedByRPCID the rpc id of the message that created the saga
 	StartedByRPCID string
+
+	//CreatedAt the time.Now() timestamp when the saga was created
+	CreatedAt time.Time
 }
 
 func (si *Instance) log() logrus.FieldLogger {
@@ -71,6 +74,7 @@ func (si *Instance) invoke(exchange, routingKey string, invocation *sagaInvocati
 		invocation.Log().Info("invoking saga instance")
 
 		span, sctx := opentracing.StartSpanFromContext(invocation.Ctx(), methodName)
+
 		// replace the original context with the conext built around the span so we ca
 		// trace the saga handler that is invoked
 		invocation.ctx = sctx
@@ -151,6 +155,7 @@ func NewInstance(sagaType reflect.Type, msgToMethodMap []*MsgToFuncPair) *Instan
 		ID:                 xid.New().String(),
 		UnderlyingInstance: newSaga,
 		MsgToMethodMap:     msgToMethodMap,
+		CreatedAt:          time.Now(),
 	}
 	return newInstance
 }
diff --git a/gbus/saga/invocation.go b/gbus/saga/invocation.go
@@ -34,25 +34,21 @@ type sagaInvocation struct {
 	startedByRPCID string
 }
 
-func (si *sagaInvocation) setCorrelationIDs(message *gbus.BusMessage, isEvent bool, targetService string) {
+func (si *sagaInvocation) setCorrelationIDs(message *gbus.BusMessage, targetService string, semantics gbus.Semantics) {
 
-	message.CorrelationID = si.inboundMsg.ID
 	message.SagaID = si.sagaID
 
-	if !isEvent {
-		//support saga-to-saga communication
-		if si.inboundMsg.SagaID != "" {
-			message.SagaCorrelationID = si.inboundMsg.SagaID
-		}
+	if semantics == gbus.REPLY {
+		message.CorrelationID = si.inboundMsg.ID
+		message.SagaCorrelationID = si.inboundMsg.SagaID
+
+	} else if semantics == gbus.CMD {
 		//if the saga is potentially invoking itself then set the SagaCorrelationID to reflect that
 		//https://github.com/wework/grabbit/issues/64
-
 		if targetService == si.hostingSvc {
 			message.SagaCorrelationID = message.SagaID
 		}
-
 	}
-
 }
 func (si *sagaInvocation) HostingSvc() string {
 	return si.hostingSvc
@@ -64,13 +60,13 @@ func (si *sagaInvocation) InvokingSvc() string {
 
 func (si *sagaInvocation) Reply(ctx context.Context, message *gbus.BusMessage) error {
 	_, targetService := si.decoratedInvocation.Routing()
-	si.setCorrelationIDs(message, false, targetService)
+	si.setCorrelationIDs(message, targetService, gbus.REPLY)
 	return si.decoratedInvocation.Reply(ctx, message)
 }
 
 func (si *sagaInvocation) ReplyToInitiator(ctx context.Context, message *gbus.BusMessage) error {
 
-	si.setCorrelationIDs(message, false, si.startedBy)
+	si.setCorrelationIDs(message, si.startedBy, gbus.REPLY)
 
 	//override the correlation ids to those of the message creating the saga
 	message.SagaCorrelationID = si.startedBySaga
@@ -93,13 +89,13 @@ func (si *sagaInvocation) Ctx() context.Context {
 
 func (si *sagaInvocation) Send(ctx context.Context, toService string,
 	command *gbus.BusMessage, policies ...gbus.MessagePolicy) error {
-	si.setCorrelationIDs(command, false, toService)
+	si.setCorrelationIDs(command, toService, gbus.CMD)
 	return si.decoratedBus.Send(ctx, toService, command, policies...)
 }
 
 func (si *sagaInvocation) Publish(ctx context.Context, exchange, topic string,
 	event *gbus.BusMessage, policies ...gbus.MessagePolicy) error {
-	si.setCorrelationIDs(event, true, "")
+	si.setCorrelationIDs(event, "", gbus.EVT)
 	return si.decoratedBus.Publish(ctx, exchange, topic, event, policies...)
 }
 
diff --git a/gbus/tx/mysql/migrations.go b/gbus/tx/mysql/migrations.go
@@ -63,6 +63,22 @@ func sagaStoreAddRPCIDDetails(svcName string) *migrator.Migration {
 	}
 }
 
+func sagaStoreAddCreatedAtDetails(svcName string) *migrator.Migration {
+	tblName := tx.GrabbitTableNameTemplate(svcName, "sagas")
+
+	addCreatorDetailsSQL := `ALTER TABLE ` + tblName + ` ADD COLUMN created_at DATETIME NOT NULL DEFAULT NOW() AFTER version`
+
+	return &migrator.Migration{
+		Name: "adding the created_at column to the saga table",
+		Func: func(tx *sql.Tx) error {
+			if _, err := tx.Exec(addCreatorDetailsSQL); err != nil {
+				return err
+			}
+			return nil
+		},
+	}
+}
+
 func outboxMigrations(svcName string) *migrator.Migration {
 
 	tblName := tx.GrabbitTableNameTemplate(svcName, "outbox")
@@ -158,6 +174,7 @@ func EnsureSchema(db *sql.DB, svcName string) {
 		outboxChangeColumnLength(svcName),
 		sagaStoreAddSagaCreatorDetails(svcName),
 		sagaStoreAddRPCIDDetails(svcName),
+		sagaStoreAddCreatedAtDetails(svcName),
 	))
 	if err != nil {
 		panic(err)
diff --git a/gbus/tx/sagastore.go b/gbus/tx/sagastore.go

Original file line number	Diff line number	Diff line change
`@@ -18,6 +18,8 @@ const (`
`18`	`18`	`CMD Semantics = "cmd"`
`19`	`19`	`//EVT represenst a messge with event semantics in grabbit`
`20`	`20`	`EVT Semantics = "evt"`
	`21`	`+ //REPLY represenst a messge with reply semantics in grabbit`
	`22`	`+ REPLY Semantics = "reply"`
`21`	`23`	`)`
`22`	`24`
`23`	`25`	`//Bus interface provides the majority of functionality to Send, Reply and Publish messages to the Bus`
Original file line number	Diff line number	Diff line change
`@@ -79,7 +79,8 @@ func (builder *defaultBuilder) Build(svcName string) gbus.Bus {`
`79`	`79`	`providerLogger := gb.Log().WithField("provider", "mysql")`
`80`	`80`	`mysqltx, err := mysql.NewTxProvider(builder.txConnStr)`
`81`	`81`	`if err != nil {`
`82`		`- panic(err)`
	`82`	`+ errMsg := fmt.Errorf("grabbit: transaction provider failed creating a transaction. %v", err)`
	`83`	`+ panic(errMsg)`
`83`	`84`	`}`
`84`	`85`	`gb.TxProvider = mysqltx`
`85`	`86`
`@@ -91,7 +92,8 @@ func (builder *defaultBuilder) Build(svcName string) gbus.Bus {`
`91`	`92`	`if builder.purgeOnStartup {`
`92`	`93`	`err := sagaStore.Purge()`
`93`	`94`	`if err != nil {`
`94`		`- panic(err)`
	`95`	`+ errMsg := fmt.Errorf("grabbit: saga store faild to purge. error: %v", err)`
	`96`	`+ panic(errMsg)`
`95`	`97`	`}`
`96`	`98`	`}`
`97`	`99`	`gb.Outbox = mysql.NewOutbox(gb.SvcName, mysqltx, builder.purgeOnStartup, builder.busCfg.OutboxCfg)`
`@@ -110,7 +112,8 @@ func (builder *defaultBuilder) Build(svcName string) gbus.Bus {`
`110`	`112`	`if builder.purgeOnStartup {`
`111`	`113`	`err := sagaStore.Purge()`
`112`	`114`	`if err != nil {`
`113`		`- panic(err)`
	`115`	`+ errMsg := fmt.Errorf("grabbit: saga store faild to purge. error: %v", err)`
	`116`	`+ panic(errMsg)`
`114`	`117`	`}`
`115`	`118`	`}`
`116`	`119`	`glue := saga.NewGlue(gb, sagaStore, svcName, gb.TxProvider, gb.Log, timeoutManager)`
Original file line number	Diff line number	Diff line change
`@@ -671,7 +671,7 @@ func (b DefaultBus) publish(tx sql.Tx, exchange, routingKey string, msg *amqp.`
`671`	`671`	`err := b.SafeWithRetries(publish, MaxRetryCount)`
`672`	`672`
`673`	`673`	`if err != nil {`
`674`		`- b.Log().Printf("failed publishing message.\n error:%v", err)`
	`674`	`+ b.Log().WithError(err).Error("failed publishing message")`
`675`	`675`	`return err`
`676`	`676`	`}`
`677`	`677`	`return err`