feature: dgraph_txn_aborts metric for prometheus

dgraph/cmd/alpha/metrics_test.go

@@ -19,12 +19,62 @@ package alpha
 import (
 	"net/http"
 	"regexp"
+	"strconv"
 	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/require"
 )
 
+func TestMetricTxnAborts(t *testing.T) {
+	mt := `
+    {
+	  set {
+		<0x71>  <name> "Bob" .
+	  }
+	}
+	`
+
+	// Create initial 'dgraph_txn_aborts' metric
+	mr1, err := mutationWithTs(mt, "application/rdf", false, false, 0)
+	require.NoError(t, err)
+	mr2, err := mutationWithTs(mt, "application/rdf", false, false, 0)
+	require.NoError(t, err)
+	require.NoError(t, commitWithTs(mr1.keys, mr1.preds, mr1.startTs))
+	require.Error(t, commitWithTs(mr2.keys, mr2.preds, mr2.startTs))
+
+	// Fetch Metrics
+	req, err := http.NewRequest("GET", addr+"/debug/prometheus_metrics", nil)
+	require.NoError(t, err)
+	_, body, err := runRequest(req)
+	require.NoError(t, err)
+	metricsMap, err := extractMetrics(string(body))
+	requiredMetric := "dgraph_txn_aborts"
+	txnAbort, ok := metricsMap[requiredMetric]
+	require.True(t, ok, "the required metric '%s' is not found", requiredMetric)
+	txnAbort1, _ := strconv.Atoi(txnAbort.(string))
+
+	// Create second 'dgraph_txn_aborts' metric
+	mr1, err = mutationWithTs(mt, "application/rdf", false, false, 0)
+	require.NoError(t, err)
+	mr2, err = mutationWithTs(mt, "application/rdf", false, false, 0)
+	require.NoError(t, err)
+	require.NoError(t, commitWithTs(mr1.keys, mr1.preds, mr1.startTs))
+	require.Error(t, commitWithTs(mr2.keys, mr2.preds, mr2.startTs))
+
+	// Fetch Updated Metrics
+	req, err = http.NewRequest("GET", addr+"/debug/prometheus_metrics", nil)
+	require.NoError(t, err)
+	_, body, err = runRequest(req)
+	require.NoError(t, err)
+	metricsMap, err = extractMetrics(string(body))
+	requiredMetric = "dgraph_txn_aborts"
+	txnAbort, ok = metricsMap["dgraph_txn_aborts"]
+	txnAbort2, _ := strconv.Atoi(txnAbort.(string))
+
+	require.Equal(t, txnAbort1+1, txnAbort2, "txnAbort was not incremented")
+}
+
 func TestMetrics(t *testing.T) {
 	req, err := http.NewRequest("GET", addr+"/debug/prometheus_metrics", nil)
 	require.NoError(t, err)
@@ -59,15 +109,16 @@ func TestMetrics(t *testing.T) {
 
 func extractMetrics(metrics string) (map[string]interface{}, error) {
 	lines := strings.Split(metrics, "\n")
-	metricRegex, err := regexp.Compile("(^[a-z0-9_]+)")
+	metricRegex, err := regexp.Compile("(^\\w+|\\d+$)")
+
 	if err != nil {
 		return nil, err
 	}
 	metricsMap := make(map[string]interface{})
 	for _, line := range lines {
-		matches := metricRegex.FindStringSubmatch(line)
+		matches := metricRegex.FindAllString(line, -1)
 		if len(matches) > 0 {
-			metricsMap[matches[0]] = struct{}{}
+			metricsMap[matches[0]] = matches[1]
 		}
 	}
 	return metricsMap, nil

worker/mutation.go

@@ -24,6 +24,8 @@ import (
 	"sync/atomic"
 	"time"
 
+	ostats "go.opencensus.io/stats"
+
 	"github.com/golang/glog"
 	"github.com/pkg/errors"
 	otrace "go.opencensus.io/trace"
@@ -783,6 +785,7 @@ func CommitOverNetwork(ctx context.Context, tc *api.TxnContext) (uint64, error)
 	span.Annotate(attributes, "")
 
 	if tctx.Aborted || tctx.CommitTs == 0 {
+		ostats.Record(context.Background(), x.TxnAborts.M(1))
 		return 0, dgo.ErrAborted
 	}
 	return tctx.CommitTs, nil

x/metrics.go

@@ -81,6 +81,8 @@ var (
 	// MaxAssignedTs records the latest max assigned timestamp.
 	MaxAssignedTs = stats.Int64("max_assigned_ts",
 		"Latest max assigned timestamp", stats.UnitDimensionless)
+    TxnAborts = stats.Int64("txn_aborts",
+    	"Number of transaction aborts",stats.UnitDimensionless)
 
 	// Conf holds the metrics config.
 	// TODO: Request statistics, latencies, 500, timeouts
@@ -147,6 +149,13 @@ var (
 			Aggregation: view.Count(),
 			TagKeys:     allTagKeys,
 		},
+		{
+			Name:        TxnAborts.Name(),
+			Measure:     TxnAborts,
+			Description: TxnAborts.Description(),
+			Aggregation: view.Count(),
+			TagKeys:     allTagKeys,
+		},
 
 		// Last value aggregations
 		{