feat: Add unique ID per request (#386)

Taskfile.yaml

@@ -223,4 +223,4 @@ tasks:
     cmds:
       - docker compose rm -f
       - docker volume prune -f
-      - rm {{.BENCH_CPU_PROFILE}} {{.BENCH_MEM_PROFILE}}
+      - rm -f {{.BENCH_CPU_PROFILE}} {{.BENCH_MEM_PROFILE}}

pkg/api/apierrors/errors.go

@@ -26,10 +26,31 @@ const (
 	ErrScriptCompilationFailed = "COMPILATION_FAILED"
 	ErrScriptNoScript          = "NO_SCRIPT"
 	ErrScriptMetadataOverride  = "METADATA_OVERRIDE"
-
-	errorCodeKey = "_errorCode"
 )
 
+// TODO: update sharedapi.ErrorResponse with new details field
+type ErrorResponse struct {
+	ErrorCode    string `json:"error_code,omitempty"`
+	ErrorMessage string `json:"error_message,omitempty"`
+	Details      string `json:"details,omitempty"`
+}
+
+func ResponseError(c *gin.Context, err error) {
+	_ = c.Error(err)
+	status, code, details := coreErrorToErrorCode(c, err)
+
+	if status < 500 {
+		c.AbortWithStatusJSON(status,
+			ErrorResponse{
+				ErrorCode:    code,
+				ErrorMessage: err.Error(),
+				Details:      details,
+			})
+	} else {
+		c.AbortWithStatus(status)
+	}
+}
+
 func coreErrorToErrorCode(c *gin.Context, err error) (int, string, string) {
 	switch {
 	case ledger.IsConflictError(err):
@@ -51,7 +72,8 @@ func coreErrorToErrorCode(c *gin.Context, err error) (int, string, string) {
 	case storage.IsError(err):
 		return http.StatusServiceUnavailable, ErrStore, ""
 	default:
-		sharedlogging.GetLogger(c.Request.Context()).Errorf("internal errors: %s", err)
+		sharedlogging.GetLogger(c.Request.Context()).Errorf(
+			"unknown API response error: %s", err)
 		return http.StatusInternalServerError, ErrInternal, ""
 	}
 }
@@ -71,31 +93,3 @@ func EncodeLink(errStr string) string {
 	payloadB64 := base64.StdEncoding.EncodeToString(payload)
 	return fmt.Sprintf("https://play.numscript.org/?payload=%v", payloadB64)
 }
-
-func ErrorCode(c *gin.Context) string {
-	return c.GetString(errorCodeKey)
-}
-
-// TODO: update sharedapi.ErrorResponse with new details field
-type ErrorResponse struct {
-	ErrorCode    string `json:"error_code,omitempty"`
-	ErrorMessage string `json:"error_message,omitempty"`
-	Details      string `json:"details,omitempty"`
-}
-
-func ResponseError(c *gin.Context, err error) {
-	_ = c.Error(err)
-	status, code, details := coreErrorToErrorCode(c, err)
-	c.Set(errorCodeKey, code)
-
-	if status < 500 {
-		c.AbortWithStatusJSON(status,
-			ErrorResponse{
-				ErrorCode:    code,
-				ErrorMessage: err.Error(),
-				Details:      details,
-			})
-	} else {
-		c.AbortWithStatus(status)
-	}
-}

pkg/api/controllers/context_test.go

@@ -0,0 +1,62 @@
+package controllers_test
+
+import (
+	"context"
+	"net/url"
+	"testing"
+
+	"github.com/google/uuid"
+	"github.com/numary/ledger/pkg"
+	"github.com/numary/ledger/pkg/api"
+	"github.com/numary/ledger/pkg/api/controllers"
+	"github.com/numary/ledger/pkg/api/internal"
+	"github.com/numary/ledger/pkg/core"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/fx"
+)
+
+func TestContext(t *testing.T) {
+	internal.RunTest(t, fx.Invoke(func(lc fx.Lifecycle, api *api.API) {
+		lc.Append(fx.Hook{
+			OnStart: func(ctx context.Context) error {
+				t.Run("GET/stats", func(t *testing.T) {
+					rsp := internal.GetStats(api)
+					_, err := uuid.Parse(rsp.Header().Get(string(pkg.KeyContextID)))
+					require.NoError(t, err)
+				})
+				t.Run("GET/log", func(t *testing.T) {
+					rsp := internal.GetLogs(api, url.Values{})
+					_, err := uuid.Parse(rsp.Header().Get(string(pkg.KeyContextID)))
+					require.NoError(t, err)
+				})
+				t.Run("GET/accounts", func(t *testing.T) {
+					rsp := internal.GetAccounts(api, url.Values{})
+					_, err := uuid.Parse(rsp.Header().Get(string(pkg.KeyContextID)))
+					require.NoError(t, err)
+				})
+				t.Run("GET/transactions", func(t *testing.T) {
+					rsp := internal.GetTransactions(api, url.Values{})
+					_, err := uuid.Parse(rsp.Header().Get(string(pkg.KeyContextID)))
+					require.NoError(t, err)
+				})
+				t.Run("POST/transactions", func(t *testing.T) {
+					rsp := internal.PostTransaction(t, api, controllers.PostTransaction{}, true)
+					_, err := uuid.Parse(rsp.Header().Get(string(pkg.KeyContextID)))
+					require.NoError(t, err)
+				})
+				t.Run("POST/transactions/batch", func(t *testing.T) {
+					rsp := internal.PostTransactionBatch(t, api, core.Transactions{})
+					_, err := uuid.Parse(rsp.Header().Get(string(pkg.KeyContextID)))
+					require.NoError(t, err)
+				})
+				t.Run("GET/balances", func(t *testing.T) {
+					rsp := internal.GetBalances(api, url.Values{})
+					_, err := uuid.Parse(rsp.Header().Get(string(pkg.KeyContextID)))
+					require.NoError(t, err)
+				})
+
+				return nil
+			},
+		})
+	}))
+}

pkg/api/controllers/script_controller.go

@@ -47,7 +47,8 @@ func (ctl *ScriptController) PostScript(c *gin.Context) {
 			code = scriptError.Code
 			message = scriptError.Message
 		} else {
-			sharedlogging.GetLogger(c.Request.Context()).Errorf("internal errors executing script: %s", err)
+			sharedlogging.GetLogger(c.Request.Context()).Errorf(
+				"internal error executing script: %s", err)
 		}
 		res.ErrorResponse = sharedapi.ErrorResponse{
 			ErrorCode:    code,

pkg/api/internal/testing.go

@@ -313,9 +313,3 @@ func RunTest(t *testing.T, options ...fx.Option) {
 		}
 	}
 }
-
-func RunSubTest(t *testing.T, name string, opts ...fx.Option) {
-	t.Run(name, func(t *testing.T) {
-		RunTest(t, opts...)
-	})
-}

pkg/api/middlewares/ledger_middleware.go

@@ -1,9 +1,15 @@
 package middlewares
 
 import (
+	"context"
+	"fmt"
+
 	"github.com/formancehq/go-libs/sharedlogging"
 	"github.com/gin-gonic/gin"
+	"github.com/google/uuid"
+	"github.com/numary/ledger/pkg"
 	"github.com/numary/ledger/pkg/api/apierrors"
+	"github.com/numary/ledger/pkg/contextlogger"
 	"github.com/numary/ledger/pkg/ledger"
 	"github.com/numary/ledger/pkg/opentelemetry"
 )
@@ -12,9 +18,7 @@ type LedgerMiddleware struct {
 	resolver *ledger.Resolver
 }
 
-func NewLedgerMiddleware(
-	resolver *ledger.Resolver,
-) LedgerMiddleware {
+func NewLedgerMiddleware(resolver *ledger.Resolver) LedgerMiddleware {
 	return LedgerMiddleware{
 		resolver: resolver,
 	}
@@ -23,25 +27,34 @@ func NewLedgerMiddleware(
 func (m *LedgerMiddleware) LedgerMiddleware() gin.HandlerFunc {
 	return func(c *gin.Context) {
 		name := c.Param("ledger")
-
 		if name == "" {
 			return
 		}
 
 		ctx, span := opentelemetry.Start(c.Request.Context(), "Ledger access")
 		defer span.End()
 
+		contextKeyID := uuid.NewString()
+		id := span.SpanContext().SpanID()
+		if id == [8]byte{} {
+			sharedlogging.GetLogger(ctx).Debugf(
+				"ledger middleware SpanID is empty, new id generated %s", contextKeyID)
+		} else {
+			contextKeyID = fmt.Sprint(id)
+		}
+		ctx = context.WithValue(ctx, pkg.KeyContextID, contextKeyID)
+		c.Header(string(pkg.KeyContextID), contextKeyID)
+
+		loggerFactory := sharedlogging.StaticLoggerFactory(
+			contextlogger.New(ctx, sharedlogging.GetLogger(ctx)))
+		sharedlogging.SetFactory(loggerFactory)
+
 		l, err := m.resolver.GetLedger(ctx, name)
 		if err != nil {
 			apierrors.ResponseError(c, err)
 			return
 		}
-		defer func() {
-			err := l.Close(ctx)
-			if err != nil {
-				sharedlogging.GetLogger(ctx).Errorf("error closing ledger: %s", err)
-			}
-		}()
+		defer l.Close(ctx)
 		c.Set("ledger", l)
 
 		c.Request = c.Request.WithContext(ctx)

pkg/bus/monitor.go

@@ -77,7 +77,7 @@ func (l *ledgerMonitor) RevertedTransaction(ctx context.Context, ledger string,
 
 func (l *ledgerMonitor) publish(ctx context.Context, topic string, ev EventMessage) {
 	if err := l.publisher.Publish(ctx, topic, ev); err != nil {
-		sharedlogging.GetLogger(ctx).Errorf("Publishing message: %s", err)
+		sharedlogging.GetLogger(ctx).Errorf("publishing message: %s", err)
 		return
 	}
 }

pkg/context.go

@@ -0,0 +1,5 @@
+package pkg
+
+type ContextKeyIDType string
+
+var KeyContextID ContextKeyIDType = "contextID"

pkg/contextlogger/contextlogger.go

@@ -0,0 +1,70 @@
+package contextlogger
+
+import (
+	"context"
+
+	"github.com/formancehq/go-libs/sharedlogging"
+	"github.com/numary/ledger/pkg"
+)
+
+var _ sharedlogging.Logger = &ContextLogger{}
+
+type ContextLogger struct {
+	ctx              context.Context
+	underlyingLogger sharedlogging.Logger
+}
+
+func New(ctx context.Context, logger sharedlogging.Logger) *ContextLogger {
+	return &ContextLogger{
+		ctx:              ctx,
+		underlyingLogger: logger,
+	}
+}
+
+func (c ContextLogger) Debugf(format string, args ...any) {
+	id := c.ctx.Value(pkg.KeyContextID)
+	c.underlyingLogger.
+		WithFields(map[string]any{string(pkg.KeyContextID): id}).
+		Debugf(format, args...)
+}
+
+func (c ContextLogger) Infof(format string, args ...any) {
+	id := c.ctx.Value(pkg.KeyContextID)
+	c.underlyingLogger.
+		WithFields(map[string]any{string(pkg.KeyContextID): id}).
+		Infof(format, args...)
+}
+
+func (c ContextLogger) Errorf(format string, args ...any) {
+	id := c.ctx.Value(pkg.KeyContextID)
+	c.underlyingLogger.
+		WithFields(map[string]any{string(pkg.KeyContextID): id}).
+		Errorf(format, args...)
+}
+
+func (c ContextLogger) Debug(args ...any) {
+	c.underlyingLogger.Debug(args...)
+}
+
+func (c ContextLogger) Info(args ...any) {
+	c.underlyingLogger.Info(args...)
+}
+
+func (c ContextLogger) Error(args ...any) {
+	c.underlyingLogger.Error(args...)
+}
+
+func (c ContextLogger) WithFields(m map[string]any) sharedlogging.Logger {
+	m[string(pkg.KeyContextID)] = c.ctx.Value(pkg.KeyContextID)
+	return &ContextLogger{
+		ctx:              c.ctx,
+		underlyingLogger: c.underlyingLogger.WithFields(m),
+	}
+}
+
+func (c ContextLogger) WithContext(ctx context.Context) sharedlogging.Logger {
+	return &ContextLogger{
+		ctx:              ctx,
+		underlyingLogger: c.underlyingLogger,
+	}
+}

pkg/storage/sqlstorage/migrate.go

@@ -45,12 +45,9 @@ func (m Migrations) Swap(i, j int) {
 var _ sort.Interface = &Migrations{}
 
 func Migrate(ctx context.Context, schema Schema, migrations ...Migration) (bool, error) {
-
 	ctx, span := opentelemetry.Start(ctx, "Migrate")
 	defer span.End()
 
-	logger := sharedlogging.GetLogger(ctx)
-
 	q, args := sqlbuilder.
 		CreateTable(schema.Table("migrations")).
 		Define(`version varchar, date varchar, UNIQUE("version")`).
@@ -82,15 +79,15 @@ func Migrate(ctx context.Context, schema Schema, migrations ...Migration) (bool,
 		row := schema.QueryRowContext(ctx, sqlq, args...)
 		var v string
 		if err = row.Scan(&v); err != nil {
-			logger.Debugf("%s", err)
+			sharedlogging.GetLogger(ctx).Debugf("migration %s: %s", m.Number, err)
 		}
 		if v != "" {
-			logger.Debugf("version %s already up to date", m.Number)
+			sharedlogging.GetLogger(ctx).Debugf("migration %s: already up to date", m.Number)
 			continue
 		}
 		modified = true
 
-		logger.Debugf("running migrations %s", m.Number)
+		sharedlogging.GetLogger(ctx).Debugf("running migration %s", m.Number)
 
 		handlersForAnyEngine, ok := m.Handlers["any"]
 		if ok {
@@ -118,7 +115,7 @@ func Migrate(ctx context.Context, schema Schema, migrations ...Migration) (bool,
 		ib.Values(m.Number, time.Now().UTC().Format(time.RFC3339))
 		sqlq, args = ib.BuildWithFlavor(schema.Flavor())
 		if _, err = tx.ExecContext(ctx, sqlq, args...); err != nil {
-			logger.Errorf("failed to insert migration version %s: %s", m.Number, err)
+			sharedlogging.GetLogger(ctx).Errorf("failed to insert migration version %s: %s", m.Number, err)
 			return false, errorFromFlavor(Flavor(schema.Flavor()), err)
 		}
 	}

pkg/storage/sqlstorage/store_ledger_test.go

@@ -736,7 +736,9 @@ func testTooManyClient(t *testing.T, store *sqlstorage.Store) {
 
 func TestInitializeStore(t *testing.T) {
 	l := logrus.New()
-	l.Level = logrus.DebugLevel
+	if testing.Verbose() {
+		l.Level = logrus.DebugLevel
+	}
 	sharedlogging.SetFactory(sharedlogging.StaticLoggerFactory(sharedlogginglogrus.New(l)))
 
 	driver, stopFn, err := ledgertesting.StorageDriver()