Merge branch 'v1.x' into metrics_on_dead_letter_handler

.github/stale.yml

@@ -0,0 +1,17 @@
+# Number of days of inactivity before an issue becomes stale
+daysUntilStale: 7
+# Number of days of inactivity before a stale issue is closed
+daysUntilClose: 7
+# Issues with these labels will never be considered stale
+  exemptLabels:
+    - bug
+    - enhancment
+# Label to use when marking an issue as stale
+staleLabel: wontfix
+# Comment to post when marking an issue as stale. Set to `false` to disable
+markComment: >
+  This issue has been automatically marked as stale because it has not had
+  recent activity. It will be closed if no further activity occurs. Thank you
+  for your contributions.
+# Comment to post when closing a stale issue. Set to `false` to disable
+closeComment: false

gbus/abstractions.go

@@ -127,7 +127,7 @@ type Saga interface {
 	New() Saga
 }
 
-//RegisterDeadletterHandler provides the ability to handle messages that were rejected as poison and arrive to the deadletter queue
+//Deadlettering provides the ability to handle messages that were rejected as poision and arrive to the deadletter queue
 type Deadlettering interface {
 	HandleDeadletter(handler DeadLetterMessageHandler)
 	ReturnDeadToQueue(ctx context.Context, publishing *amqp.Publishing) error

gbus/builder/builder.go

@@ -77,7 +77,9 @@ func (builder *defaultBuilder) Build(svcName string) gbus.Bus {
 			panic(err)
 		}
 		gb.TxProvider = mysqltx
+
 		mysql.EnsureSchema(mysqltx.Database, gb.SvcName)
+
 		//TODO move purge logic into the NewSagaStore factory method
 		sagaStore = mysql.NewSagaStore(gb.SvcName, mysqltx)
 		if builder.purgeOnStartup {

gbus/invocation.go

@@ -23,6 +23,7 @@ type defaultInvocationContext struct {
 	deliveryInfo DeliveryInfo
 }
 
+//DeliveryInfo provdes information as to the attempted deilvery of the invocation
 type DeliveryInfo struct {
 	Attempt       uint
 	MaxRetryCount uint

gbus/metrics/handler_metrics.go

@@ -26,6 +26,7 @@ type handlerMetrics struct {
 	latency prometheus.Summary
 }
 
+//AddHandlerMetrics adds a handlere to be tracked with metrics
 func AddHandlerMetrics(handlerName string) {
 	handlerMetrics := newHandlerMetrics(handlerName)
 	_, exists := handlerMetricsByHandlerName.LoadOrStore(handlerName, handlerMetrics)
@@ -35,6 +36,7 @@ func AddHandlerMetrics(handlerName string) {
 	}
 }
 
+//RunHandlerWithMetric runs a specific handler with metrics being collected and reported to prometheus
 func RunHandlerWithMetric(handleMessage func() error, handlerName string, logger logrus.FieldLogger) error {
 	handlerMetrics := GetHandlerMetrics(handlerName)
 	defer func() {
@@ -63,6 +65,7 @@ func RunHandlerWithMetric(handleMessage func() error, handlerName string, logger
 	return err
 }
 
+//GetHandlerMetrics gets the metrics handler associated with the handlerName
 func GetHandlerMetrics(handlerName string) *handlerMetrics {
 	entry, ok := handlerMetricsByHandlerName.Load(handlerName)
 	if ok {
@@ -99,14 +102,17 @@ func trackTime(functionToTrack func() error, observer prometheus.Observer) error
 	return functionToTrack()
 }
 
+//GetSuccessCount gets the value of the handlers success value
 func (hm *handlerMetrics) GetSuccessCount() (float64, error) {
 	return hm.getLabeledCounterValue(success)
 }
 
+//GetFailureCount gets the value of the handlers failure value
 func (hm *handlerMetrics) GetFailureCount() (float64, error) {
 	return hm.getLabeledCounterValue(failure)
 }
 
+//GetLatencySampleCount gets the value of the handlers latency value
 func (hm *handlerMetrics) GetLatencySampleCount() (*uint64, error) {
 	m := &io_prometheus_client.Metric{}
 	err := hm.latency.Write(m)

gbus/metrics/message_metrics.go

@@ -3,17 +3,19 @@ package metrics
 import (
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
-	"github.com/prometheus/client_model/go"
+	io_prometheus_client "github.com/prometheus/client_model/go"
 )
 
 var (
 	rejectedMessages = newRejectedMessagesCounter()
 )
 
+//ReportRejectedMessage reports a message being rejected to the metrics counter
 func ReportRejectedMessage() {
 	rejectedMessages.Inc()
 }
 
+//GetRejectedMessagesValue gets the value of the rejected message counter
 func GetRejectedMessagesValue() (float64, error) {
 	m := &io_prometheus_client.Metric{}
 	err := rejectedMessages.Write(m)

gbus/outbox.go

@@ -53,7 +53,6 @@ func (out *AMQPOutbox) init(amqp *amqp.Channel, confirm, resendOnNack bool) erro
 //Shutdown stops the outbox
 func (out *AMQPOutbox) Shutdown() {
 	close(out.stop)
-
 }
 
 //Post implements Outbox.Send

gbus/tx/mysql/migrations.go

@@ -2,8 +2,9 @@ package mysql
 
 import (
 	"database/sql"
+	"regexp"
+	"strings"
 
-	"fmt"
 	"github.com/lopezator/migrator"
 	"github.com/wework/grabbit/gbus/tx"
 )
@@ -86,7 +87,7 @@ func TimoutTableMigration(svcName string) *migrator.Migration {
 
 //EnsureSchema implements Grabbit's migrations strategy
 func EnsureSchema(db *sql.DB, svcName string) {
-	migrationsTable := fmt.Sprintf("grabbitMigrations_%s", svcName)
+	migrationsTable := sanitizedMigrationsTable(svcName)
 
 	migrate, err := migrator.New(migrator.TableName(migrationsTable), migrator.Migrations(
 		OutboxMigrations(svcName),
@@ -101,3 +102,10 @@ func EnsureSchema(db *sql.DB, svcName string) {
 		panic(err)
 	}
 }
+
+func sanitizedMigrationsTable(svcName string) string {
+	var re = regexp.MustCompile(`-|;|\\|`)
+	sanitized := re.ReplaceAllString(svcName, "")
+
+	return strings.ToLower("grabbitMigrations_" + sanitized)
+}

tests/bus_test.go

@@ -430,6 +430,17 @@ func TestHealthCheck(t *testing.T) {
 	}
 }
 
+func TestSanitizingSvcName(t *testing.T) {
+	svc4 := createNamedBusForTest(testSvc4)
+	err := svc4.Start()
+	if err != nil {
+		t.Error(err.Error())
+	}
+	defer svc4.Shutdown()
+
+	fmt.Println("succeeded sanitizing service name")
+}
+
 func noopTraceContext() context.Context {
 	return context.Background()
 	// tracer := opentracing.NoopTracer{}

tests/consts.go

@@ -12,12 +12,14 @@ var connStr string
 var testSvc1 string
 var testSvc2 string
 var testSvc3 string
+var testSvc4 string
 
 func init() {
 	connStr = "amqp://rabbitmq:rabbitmq@localhost"
 	testSvc1 = "testSvc1"
 	testSvc2 = "testSvc2"
 	testSvc3 = "testSvc3"
+	testSvc4 = "test-svc4"
 }
 
 func createBusWithConfig(svcName string, deadletter string, txnl, pos bool, conf gbus.BusConfiguration) gbus.Bus {