added tracing

gbus/bus.go

@@ -5,11 +5,13 @@ import (
 	"database/sql"
 	"errors"
 	"fmt"
-
 	"runtime/debug"
 	"sync"
 	"time"
 
+	"github.com/opentracing-contrib/go-amqp/amqptracer"
+	"github.com/opentracing/opentracing-go"
+	slog "github.com/opentracing/opentracing-go/log"
 	"github.com/rs/xid"
 	log "github.com/sirupsen/logrus"
 	"github.com/streadway/amqp"
@@ -550,16 +552,21 @@ func (b *DefaultBus) monitorAMQPErrors() {
 func (b *DefaultBus) sendImpl(ctx context.Context, tx *sql.Tx, toService, replyTo, exchange, topic string, message *BusMessage, policies ...MessagePolicy) (er error) {
 	b.SenderLock.Lock()
 	defer b.SenderLock.Unlock()
-
+	span, ctx := opentracing.StartSpanFromContext(ctx, "sendImpl")
 	defer func() {
 		if err := recover(); err != nil {
-
 			errMsg := fmt.Sprintf("panic recovered panicking err:\n%v\n%s", err, debug.Stack())
 			er = errors.New(errMsg)
+			span.LogFields(slog.Error(er))
 		}
+		span.Finish()
 	}()
 
 	headers := message.GetAMQPHeaders()
+	err := amqptracer.Inject(span, headers)
+	if err != nil {
+		b.log().WithError(err).Error("could not inject headers")
+	}
 
 	buffer, err := b.Serializer.Encode(message.Payload)
 	if err != nil {
@@ -575,17 +582,8 @@ func (b *DefaultBus) sendImpl(ctx context.Context, tx *sql.Tx, toService, replyT
 		ContentEncoding: b.Serializer.Name(),
 		Headers:         headers,
 	}
+	span.LogFields(message.GetTraceLog()...)
 
-	/* TODO:FIX Opentracing context
-	sp := opentracing.SpanFromContext(ctx)
-	if sp != nil {
-		defer sp.Finish()
-	}
-	// Inject the span context into the AMQP header.
-	if err := amqptracer.Inject(sp, msg.Headers); err != nil {
-		return err
-	}
-	*/
 	for _, defaultPolicy := range b.DefaultPolicies {
 		defaultPolicy.Apply(&msg)
 	}
@@ -609,7 +607,7 @@ func (b *DefaultBus) sendImpl(ctx context.Context, tx *sql.Tx, toService, replyT
 			b.log().WithField("message_id", msg.MessageId).Info("sending message to outbox")
 			saveErr := b.Outbox.Save(tx, exchange, key, msg)
 			if saveErr != nil {
-				log.Printf("fialed to save to transactional outbox\n%v", saveErr)
+				log.WithError(saveErr).Error("failed to save to transactional outbox")
 			}
 			return saveErr
 		}

gbus/messages.go

@@ -1,6 +1,7 @@
 package gbus
 
 import (
+	"github.com/opentracing/opentracing-go/log"
 	"github.com/rs/xid"
 	"github.com/streadway/amqp"
 )
@@ -60,6 +61,18 @@ func (bm *BusMessage) SetPayload(payload Message) {
 	bm.Payload = payload
 }
 
+func (bm *BusMessage) GetTraceLog() (fields []log.Field) {
+	return []log.Field{
+		log.String("message", bm.PayloadFQN),
+		log.String("ID", bm.ID),
+		log.String("SagaID", bm.SagaID),
+		log.String("CorrelationID", bm.CorrelationID),
+		log.String("SagaCorrelationID", bm.SagaCorrelationID),
+		log.String("Semantics", bm.Semantics),
+		log.String("RPCID", bm.RPCID),
+	}
+}
+
 func castToString(i interface{}) string {
 	v, ok := i.(string)
 	if !ok {

gbus/saga/glue.go

@@ -3,7 +3,6 @@ package saga
 import (
 	"database/sql"
 	"fmt"
-
 	"reflect"
 	"strings"
 	"sync"

gbus/saga/timeout.go

@@ -5,7 +5,6 @@ import (
 	"time"
 
 	"github.com/sirupsen/logrus"
-
 	"github.com/wework/grabbit/gbus"
 )
 
@@ -25,12 +24,6 @@ func (tm *TimeoutManager) RequestTimeout(svcName, sagaID string, duration time.D
 			SagaID: sagaID}
 		msg := gbus.NewBusMessage(reuqestTimeout)
 		msg.SagaCorrelationID = sagaID
-		/* TODO:FIX Opentracing
-		span := opentracing.GlobalTracer().StartSpan("timeout")
-		if span != nil {
-			defer span.Finish()
-		}
-		*/
 		if err := tm.bus.Send(context.Background(), svcName, msg); err != nil {
 			//TODO: add logger
 			logrus.WithError(err).Error("could not send timeout to bus")

gbus/serialization/proto.go

@@ -10,7 +10,6 @@ import (
 
 	"github.com/golang/protobuf/proto"
 	"github.com/sirupsen/logrus"
-
 	"github.com/wework/grabbit/gbus"
 )
 

gbus/tx/mysql/sagastore.go

@@ -5,9 +5,7 @@ import (
 	"fmt"
 
 	log "github.com/sirupsen/logrus"
-
 	"github.com/wework/grabbit/gbus"
-
 	"github.com/wework/grabbit/gbus/saga"
 	"github.com/wework/grabbit/gbus/tx"
 )

gbus/tx/mysql/txoutbox.go

@@ -9,9 +9,8 @@ import (
 	"strings"
 	"time"
 
-	log "github.com/sirupsen/logrus"
-
 	"github.com/rs/xid"
+	log "github.com/sirupsen/logrus"
 	"github.com/streadway/amqp"
 	"github.com/wework/grabbit/gbus"
 )