feat(metrics): add per-IP and DeFi transaction content metrics

Signed-off-by: Ales Verbic <verbotenj@blinklabs.io>

Author: verbotenj

go.mod

@@ -36,6 +36,7 @@ require (
 	github.com/jinzhu/copier v0.4.0 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
 	github.com/klauspost/cpuid/v2 v2.3.0 // indirect
+	github.com/kylelemons/godebug v1.1.0 // indirect
 	github.com/mailru/easyjson v0.7.7 // indirect
 	github.com/minio/sha256-simd v1.0.1 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect

internal/api/api.go

@@ -24,8 +24,10 @@ import (
 	"io"
 	"io/fs"
 	"mime"
+	"net"
 	"net/http"
 	"runtime/debug"
+	"strings"
 	"time"
 
 	ouroboros "github.com/blinklabs-io/gouroboros"
@@ -34,8 +36,8 @@ import (
 	_ "github.com/blinklabs-io/tx-submit-api/docs" // docs is generated by Swag CLI
 	"github.com/blinklabs-io/tx-submit-api/internal/config"
 	"github.com/blinklabs-io/tx-submit-api/internal/logging"
+	"github.com/blinklabs-io/tx-submit-api/internal/metrics"
 	"github.com/blinklabs-io/tx-submit-api/submit"
-	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	httpSwagger "github.com/swaggo/http-swagger"
 )
@@ -47,23 +49,6 @@ var staticFS embed.FS
 // submission. Cardano's protocol limit is ~16KB; 64KB gives ample overhead.
 const maxTxBodyBytes = 64 * 1024
 
-var (
-	// Gauge to match input-output-hk's metric type
-	txSubmitFailCount = prometheus.NewGauge(prometheus.GaugeOpts{
-		Name: "tx_submit_fail_count",
-		Help: "transactions failed",
-	})
-	// Gauge to match input-output-hk's metric type
-	txSubmitCount = prometheus.NewGauge(prometheus.GaugeOpts{
-		Name: "tx_submit_count",
-		Help: "transactions submitted",
-	})
-)
-
-func init() {
-	prometheus.MustRegister(txSubmitFailCount, txSubmitCount)
-}
-
 // corsMiddleware adds CORS headers allowing all origins.
 func corsMiddleware(next http.Handler) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -228,6 +213,8 @@ func Start(cfg *config.Config) error {
 	handler = recoveryMiddleware(handler)
 	handler = corsMiddleware(handler)
 
+	metrics.Register()
+
 	// Start metrics listener
 	go func() {
 		logger.Info("starting metrics listener",
@@ -352,16 +339,17 @@ func handleHasTx(w http.ResponseWriter, r *http.Request) {
 //	@Failure		500				{object}	string	"Server Error"
 //	@Router			/api/submit/tx [post]
 func handleSubmitTx(w http.ResponseWriter, r *http.Request) {
-	// First, initialize our configuration and loggers
 	cfg := config.GetConfig()
 	logger := logging.GetLogger()
+	clientIP := realClientIP(r)
 
-	// Check our headers for content-type
+	// Check our headers for content-type. Wrong content-type is rejected before
+	// reading the body, so no IP metric is recorded here.
 	mediaType, _, _ := mime.ParseMediaType(r.Header.Get("Content-Type"))
 	if mediaType != "application/cbor" {
 		logger.Error("invalid request body, should be application/cbor")
 		writeJSON(w, http.StatusUnsupportedMediaType, "invalid request body, should be application/cbor")
-		txSubmitFailCount.Inc()
+		metrics.IncTxSubmitFailCount()
 		return
 	}
 
@@ -376,9 +364,21 @@ func handleSubmitTx(w http.ResponseWriter, r *http.Request) {
 			logger.Error("failed to read request body", "err", err)
 			writeJSON(w, http.StatusInternalServerError, "failed to read request body")
 		}
-		txSubmitFailCount.Inc()
+		metrics.IncTxSubmitFailCount()
+		metrics.RecordTxRequest(clientIP, "error")
+		return
+	}
+
+	// Parse tx content signals before submitting so we can record them regardless
+	// of whether the node accepts or rejects the transaction.
+	txInfo, err := submit.ParseTxInfo(txRawBytes)
+	if err != nil {
+		writeJSON(w, http.StatusBadRequest, err.Error())
+		metrics.IncTxSubmitFailCount()
+		metrics.RecordTxRequest(clientIP, "error")
 		return
 	}
+
 	// Send TX
 	errorChan := make(chan error, 1)
 	submitConfig := &submit.Config{
@@ -391,9 +391,10 @@ func handleSubmitTx(w http.ResponseWriter, r *http.Request) {
 	}
 	txHash, err := submit.SubmitTx(submitConfig, txRawBytes)
 	if err != nil {
+		var txRejectErr *localtxsubmission.TransactionRejectedError
+		isRejected := errors.As(err, &txRejectErr)
 		if r.Header.Get("Accept") == "application/cbor" {
-			var txRejectErr *localtxsubmission.TransactionRejectedError
-			if errors.As(err, &txRejectErr) {
+			if isRejected && txRejectErr != nil {
 				w.Header().Set("Content-Type", "application/cbor")
 				w.WriteHeader(http.StatusBadRequest)
 				_, _ = w.Write(txRejectErr.ReasonCbor)
@@ -405,13 +406,21 @@ func handleSubmitTx(w http.ResponseWriter, r *http.Request) {
 		} else {
 			writeJSON(w, http.StatusBadRequest, err.Error())
 		}
-		txSubmitFailCount.Inc()
+		result := "error"
+		if isRejected {
+			result = "rejected"
+		}
+		metrics.IncTxSubmitFailCount()
+		metrics.RecordTxRequest(clientIP, result)
+		metrics.RecordTxContent(txInfo.ScriptType, txInfo.HasMinting, txInfo.HasReferenceInputs)
 		return
 	}
 
 	// Node confirmed the tx is in its mempool (AcceptTx received synchronously).
 	// Record success before responding so metrics always agree with the HTTP status.
-	txSubmitCount.Inc()
+	metrics.IncTxSubmitCount()
+	metrics.RecordTxRequest(clientIP, "accepted")
+	metrics.RecordTxContent(txInfo.ScriptType, txInfo.HasMinting, txInfo.HasReferenceInputs)
 	writeJSON(w, http.StatusAccepted, txHash)
 
 	// Drain errorChan in the background. Post-submission connection errors do not
@@ -427,3 +436,20 @@ func handleSubmitTx(w http.ResponseWriter, r *http.Request) {
 		}
 	}()
 }
+
+// realClientIP extracts the real client IP from the request, accounting for
+// reverse proxies. Priority: X-Real-IP > first entry of X-Forwarded-For > RemoteAddr.
+func realClientIP(r *http.Request) string {
+	if ip := strings.TrimSpace(r.Header.Get("X-Real-IP")); ip != "" {
+		return ip
+	}
+	if forwarded := r.Header.Get("X-Forwarded-For"); forwarded != "" {
+		first, _, _ := strings.Cut(forwarded, ",")
+		return strings.TrimSpace(first)
+	}
+	host, _, err := net.SplitHostPort(r.RemoteAddr)
+	if err != nil {
+		return r.RemoteAddr
+	}
+	return host
+}

internal/api/api_test.go

@@ -25,10 +25,13 @@ import (
 
 	"github.com/blinklabs-io/tx-submit-api/internal/config"
 	"github.com/blinklabs-io/tx-submit-api/internal/logging"
+	"github.com/blinklabs-io/tx-submit-api/internal/metrics"
+	"github.com/prometheus/client_golang/prometheus/testutil"
 )
 
 func TestMain(m *testing.M) {
 	logging.Setup(&config.LoggingConfig{Level: "error"})
+	metrics.RegisterForTesting()
 	os.Exit(m.Run())
 }
 
@@ -42,6 +45,71 @@ func newTestMux() http.Handler {
 	return handler
 }
 
+// --- realClientIP ---
+
+func TestRealClientIP(t *testing.T) {
+	t.Parallel()
+	tests := []struct {
+		name       string
+		xRealIP    string
+		xForwarded string
+		remoteAddr string
+		want       string
+	}{
+		{
+			name:       "X-Real-IP wins",
+			xRealIP:    "1.2.3.4",
+			xForwarded: "5.6.7.8",
+			remoteAddr: "9.10.11.12:9000",
+			want:       "1.2.3.4",
+		},
+		{
+			name:       "X-Forwarded-For single IP",
+			xForwarded: "10.0.0.1",
+			remoteAddr: "9.10.11.12:9000",
+			want:       "10.0.0.1",
+		},
+		{
+			name:       "X-Forwarded-For multiple IPs returns first",
+			xForwarded: "10.0.0.1, 172.16.0.1, 192.168.1.1",
+			remoteAddr: "9.10.11.12:9000",
+			want:       "10.0.0.1",
+		},
+		{
+			name:       "RemoteAddr with port strips port",
+			remoteAddr: "203.0.113.5:54321",
+			want:       "203.0.113.5",
+		},
+		{
+			name:       "RemoteAddr without port returned as-is",
+			remoteAddr: "203.0.113.5",
+			want:       "203.0.113.5",
+		},
+		{
+			name:       "IPv6 RemoteAddr strips port",
+			remoteAddr: "[::1]:8080",
+			want:       "::1",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			req := httptest.NewRequest(http.MethodGet, "/", nil)
+			req.RemoteAddr = tt.remoteAddr
+			if tt.xRealIP != "" {
+				req.Header.Set("X-Real-IP", tt.xRealIP)
+			}
+			if tt.xForwarded != "" {
+				req.Header.Set("X-Forwarded-For", tt.xForwarded)
+			}
+			if got := realClientIP(req); got != tt.want {
+				t.Errorf("want %q, got %q", tt.want, got)
+			}
+		})
+	}
+}
+
 // --- healthcheck ---
 
 func TestHealthcheck_OK(t *testing.T) {
@@ -182,3 +250,43 @@ func TestCORS(t *testing.T) {
 		})
 	}
 }
+
+// --- metrics ---
+
+func TestSubmitTx_RequestsTotal_InvalidCBOR(t *testing.T) {
+	// Not parallel: reads counter value which is package-global state.
+	before := testutil.ToFloat64(metrics.TxSubmitRequestsTotal().WithLabelValues("1.2.3.4", "error"))
+
+	rec := httptest.NewRecorder()
+	req := httptest.NewRequest(http.MethodPost, "/api/submit/tx", strings.NewReader("not-valid-cbor"))
+	req.Header.Set("Content-Type", "application/cbor")
+	req.Header.Set("X-Real-IP", "1.2.3.4")
+	newTestMux().ServeHTTP(rec, req)
+
+	if rec.Code != http.StatusBadRequest {
+		t.Fatalf("expected 400, got %d", rec.Code)
+	}
+	after := testutil.ToFloat64(metrics.TxSubmitRequestsTotal().WithLabelValues("1.2.3.4", "error"))
+	if after-before != 1 {
+		t.Errorf("requests_total{ip=1.2.3.4,result=error}: expected increment of 1, got %f", after-before)
+	}
+}
+
+func TestSubmitTx_RequestsTotal_NoIPMetricOnBadContentType(t *testing.T) {
+	// Not parallel: reads counter value which is package-global state.
+	before := testutil.ToFloat64(metrics.TxSubmitRequestsTotal().WithLabelValues("2.3.4.5", "error"))
+
+	rec := httptest.NewRecorder()
+	req := httptest.NewRequest(http.MethodPost, "/api/submit/tx", strings.NewReader("data"))
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("X-Real-IP", "2.3.4.5")
+	newTestMux().ServeHTTP(rec, req)
+
+	if rec.Code != http.StatusUnsupportedMediaType {
+		t.Fatalf("expected 415, got %d", rec.Code)
+	}
+	after := testutil.ToFloat64(metrics.TxSubmitRequestsTotal().WithLabelValues("2.3.4.5", "error"))
+	if after != before {
+		t.Errorf("requests_total should not increment on content-type rejection, got increment of %f", after-before)
+	}
+}