added metric report on saga timeout + bug fix in mysql timeout manager

gbus/metrics/saga_metrics.go

@@ -0,0 +1,29 @@
+package metrics
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+	io_prometheus_client "github.com/prometheus/client_model/go"
+)
+
+var SagaTimeoutCounter = newSagaTimeoutCounter()
+
+func GetSagaTimeoutCounterValue() (float64, error) {
+	m := &io_prometheus_client.Metric{}
+	err := SagaTimeoutCounter.Write(m)
+
+	if err != nil {
+		return 0, err
+	}
+
+	return m.GetCounter().GetValue(), nil
+}
+
+func newSagaTimeoutCounter() prometheus.Counter {
+	return promauto.NewCounter(prometheus.CounterOpts{
+		Namespace: grabbitPrefix,
+		Subsystem: "saga",
+		Name:      "timeedout_sagas",
+		Help:      "counting the number of timedout saga instances",
+	})
+}

gbus/saga/glue.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/sirupsen/logrus"
 	"github.com/wework/grabbit/gbus"
+	"github.com/wework/grabbit/gbus/metrics"
 )
 
 func fqnsFromMessages(objs []gbus.Message) []string {
@@ -247,6 +248,7 @@ func (imsm *Glue) registerEvent(exchange, topic string, event gbus.Message) erro
 func (imsm *Glue) TimeoutSaga(tx *sql.Tx, sagaID string) error {
 
 	saga, err := imsm.sagaStore.GetSagaByID(tx, sagaID)
+
 	//we are assuming that if the TimeoutSaga has been called but no instance returned from the store the saga
 	//has been completed already and
 	if err == ErrInstanceNotFound {
@@ -260,6 +262,8 @@ func (imsm *Glue) TimeoutSaga(tx *sql.Tx, sagaID string) error {
 		imsm.Log().WithError(timeoutErr).WithField("sagaID", sagaID).Error("failed to timeout saga")
 		return timeoutErr
 	}
+
+	metrics.SagaTimeoutCounter.Inc()
 	return imsm.completeOrUpdateSaga(tx, saga)
 }
 

gbus/tx/mysql/timeout.go

@@ -186,7 +186,7 @@ func (tm *TimeoutManager) RegisterTimeout(tx *sql.Tx, sagaID string, duration ti
 //ClearTimeout clears a timeout for a specific saga
 func (tm *TimeoutManager) ClearTimeout(tx *sql.Tx, sagaID string) error {
 
-	deleteSQL := `delete from ` + tm.timeoutsTableName + ` where saga_id_id = ?`
+	deleteSQL := `delete from ` + tm.timeoutsTableName + ` where saga_id = ?`
 	_, err := tx.Exec(deleteSQL, sagaID)
 	return err
 }

tests/saga_test.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/wework/grabbit/gbus"
+	"github.com/wework/grabbit/gbus/metrics"
 )
 
 /*
@@ -225,6 +226,10 @@ func TestSagaTimeout(t *testing.T) {
 	}
 
 	<-proceed
+	timeoutCounter, e := metrics.GetSagaTimeoutCounterValue()
+	if timeoutCounter != 1 || e != nil {
+		t.Errorf("saga timeout counter expected to be 1 actual %v", timeoutCounter)
+	}
 }
 
 func TestSagaSelfMessaging(t *testing.T) {