feat: add traces all along the migration path

cmd/buckets_upgrade.go

@@ -3,12 +3,13 @@
 import (
 	"github.com/formancehq/go-libs/v2/bun/bunconnect"
 	"github.com/formancehq/go-libs/v2/logging"
+	"github.com/formancehq/go-libs/v2/otlp"
+	"github.com/formancehq/go-libs/v2/otlp/otlptraces"
 	"github.com/formancehq/go-libs/v2/service"
-	"github.com/formancehq/ledger/internal/storage/bucket"
+	"github.com/formancehq/ledger/internal/storage"
 	"github.com/formancehq/ledger/internal/storage/driver"
-	"github.com/formancehq/ledger/internal/storage/ledger"
 	"github.com/spf13/cobra"
-	"github.com/uptrace/bun"
+	"go.uber.org/fx"
 )
 
 func NewBucketUpgrade() *cobra.Command {
@@ -17,22 +18,13 @@
 		Args:         cobra.ExactArgs(1),
 		SilenceUsage: true,
 		RunE: func(cmd *cobra.Command, args []string) error {
-			logger := logging.NewDefaultLogger(cmd.OutOrStdout(), service.IsDebug(cmd), false, false)
-			cmd.SetContext(logging.ContextWithLogger(cmd.Context(), logger))
+			return withStorageDriver(cmd, func(driver *driver.Driver) error {
+				if args[0] == "*" {
+					return driver.UpgradeAllBuckets(cmd.Context())
+				}
 
-			driver, db, err := getDriver(cmd)
-			if err != nil {
-				return err
-			}
-			defer func() {
-				_ = db.Close()
-			}()
-
-			if args[0] == "*" {
-				return driver.UpgradeAllBuckets(cmd.Context())
-			}
-
-			return driver.UpgradeBucket(cmd.Context(), args[0])
+				return driver.UpgradeBucket(cmd.Context(), args[0])
+			})
 		},
 	}
 
@@ -42,26 +34,32 @@
 	return cmd
 }
 
-func getDriver(cmd *cobra.Command) (*driver.Driver, *bun.DB, error) {
+func withStorageDriver(cmd *cobra.Command, fn func(driver *driver.Driver) error) error {
 
-	connectionOptions, err := bunconnect.ConnectionOptionsFromFlags(cmd)
-	if err != nil {
-		return nil, nil, err
-	}
+	logger := logging.NewDefaultLogger(cmd.OutOrStdout(), service.IsDebug(cmd), false, false)
 
-	db, err := bunconnect.OpenSQLDB(cmd.Context(), *connectionOptions)
+	connectionOptions, err := bunconnect.ConnectionOptionsFromFlags(cmd)
 	if err != nil {
-		return nil, nil, err
+		return err
 	}
 
-	driver := driver.New(
-		db,
-		ledger.NewFactory(db),
-		bucket.NewDefaultFactory(),
+	var d *driver.Driver
+	app := fx.New(
+		fx.NopLogger,
+		otlp.FXModuleFromFlags(cmd, otlp.WithServiceVersion(Version)),
+		otlptraces.FXModuleFromFlags(cmd),
+		bunconnect.Module(*connectionOptions, service.IsDebug(cmd)),
+		storage.NewFXModule(storage.ModuleConfig{}),
+		fx.Supply(fx.Annotate(logger, fx.As(new(logging.Logger)))),
+		fx.Populate(&d),
 	)
-	if err := driver.Initialize(cmd.Context()); err != nil {
-		return nil, nil, err
+	err = app.Start(cmd.Context())
+	if err != nil {
+		return err
 	}
+	defer func() {
+		_ = app.Stop(cmd.Context())
+	}()
 
-	return driver, db, nil
+	return fn(d)
 }

cmd/root.go

@@ -2,11 +2,10 @@
 
 import (
 	"github.com/formancehq/go-libs/v2/bun/bunmigrate"
-	"github.com/formancehq/go-libs/v2/logging"
+	"github.com/formancehq/go-libs/v2/otlp"
+	"github.com/formancehq/go-libs/v2/otlp/otlptraces"
 	"github.com/formancehq/go-libs/v2/service"
-	"github.com/formancehq/ledger/internal/storage/bucket"
 	"github.com/formancehq/ledger/internal/storage/driver"
-	"github.com/formancehq/ledger/internal/storage/ledger"
 	"github.com/spf13/cobra"
 	"github.com/uptrace/bun"
 )
@@ -35,23 +34,30 @@
 	root.AddCommand(NewWorkerCommand())
 	root.AddCommand(NewDocsCommand())
 
-	root.AddCommand(bunmigrate.NewDefaultCommand(func(cmd *cobra.Command, _ []string, db *bun.DB) error {
-		logger := logging.NewDefaultLogger(cmd.OutOrStdout(), service.IsDebug(cmd), false, false)
-		cmd.SetContext(logging.ContextWithLogger(cmd.Context(), logger))
-
-		driver := driver.New(db, ledger.NewFactory(db), bucket.NewDefaultFactory())
-		if err := driver.Initialize(cmd.Context()); err != nil {
-			return err
-		}
-
-		return driver.UpgradeAllBuckets(cmd.Context())
-	}))
+	root.AddCommand(newMigrationCommand())
 	root.AddCommand(NewDocsCommand())
+
 	service.AddFlags(root.PersistentFlags())
 
 	return root
 }
 
+func newMigrationCommand() *cobra.Command {
+	ret := bunmigrate.NewDefaultCommand(func(cmd *cobra.Command, _ []string, db *bun.DB) error {
+		return withStorageDriver(cmd, func(driver *driver.Driver) error {
+			if err := driver.Initialize(cmd.Context()); err != nil {
+				return err
+			}
+
+			return driver.UpgradeAllBuckets(cmd.Context())
+		})
+	})
+	otlp.AddFlags(ret.Flags())
+	otlptraces.AddFlags(ret.Flags())
+
+	return ret
+}
+
 func Execute() {
 	service.Execute(NewRootCommand())
 }

cmd/serve.go

@@ -174,6 +174,7 @@ func NewServeCommand() *cobra.Command {
 
 	addWorkerFlags(cmd)
 	bunconnect.AddFlags(cmd.Flags())
+	otlp.AddFlags(cmd.Flags())
 	otlpmetrics.AddFlags(cmd.Flags())
 	otlptraces.AddFlags(cmd.Flags())
 	auth.AddFlags(cmd.Flags())

go.mod

@@ -14,7 +14,7 @@ require (
 	github.com/antlr/antlr4/runtime/Go/antlr v1.4.10
 	github.com/bluele/gcache v0.0.2
 	github.com/dop251/goja v0.0.0-20241009100908-5f46f2705ca3
-	github.com/formancehq/go-libs/v2 v2.2.3-0.20250328155139-11198164d016
+	github.com/formancehq/go-libs/v2 v2.2.3-0.20250401141012-7ef088564530
 	github.com/formancehq/ledger/pkg/client v0.0.0-00010101000000-000000000000
 	github.com/go-chi/chi/v5 v5.2.1
 	github.com/go-chi/cors v1.2.1

go.sum

@@ -104,8 +104,8 @@ github.com/fatih/color v1.18.0 h1:S8gINlzdQ840/4pfAwic/ZE0djQEH3wM94VfqLTZcOM=
 github.com/fatih/color v1.18.0/go.mod h1:4FelSpRwEGDpQ12mAdzqdOukCy4u8WUtOY6lkT/6HfU=
 github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg=
 github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U=
-github.com/formancehq/go-libs/v2 v2.2.3-0.20250328155139-11198164d016 h1:t2REX3SXrD4asj979f8pgRw/SfUPhYGxYZ7hkkP586w=
-github.com/formancehq/go-libs/v2 v2.2.3-0.20250328155139-11198164d016/go.mod h1:JvBjEDWNf7izCy2dq/eI3aMc9d28gChBe1rjw5yYlAs=
+github.com/formancehq/go-libs/v2 v2.2.3-0.20250401141012-7ef088564530 h1:c7loJTPm5e/AwayJIWgZZQAqMIBR/BrClfF5aldDlxE=
+github.com/formancehq/go-libs/v2 v2.2.3-0.20250401141012-7ef088564530/go.mod h1:JvBjEDWNf7izCy2dq/eI3aMc9d28gChBe1rjw5yYlAs=
 github.com/formancehq/numscript v0.0.11 h1:vZDfRfrhOkuInv5fLIXvWZU3ylK+fVgmR4la01dO5to=
 github.com/formancehq/numscript v0.0.11/go.mod h1:btuSv05cYwi9BvLRxVs5zrunU+O1vTgigG1T6UsawcY=
 github.com/fortytw2/leaktest v1.3.0 h1:u8491cBMTQ8ft8aeV+adlcytMZylmA5nnwwkRZjI8vw=

internal/api/bulking/bulker.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"github.com/alitto/pond"
 	"github.com/formancehq/go-libs/v2/logging"
+	"github.com/formancehq/go-libs/v2/otlp"
 	ledger "github.com/formancehq/ledger/internal"
 	ledgercontroller "github.com/formancehq/ledger/internal/controller/ledger"
 	"go.opentelemetry.io/otel/attribute"
@@ -56,7 +57,7 @@ func (b *Bulker) run(ctx context.Context, ctrl ledgercontroller.Controller, bulk
 				ret, logID, err := b.processElement(ctx, ctrl, element)
 				if err != nil {
 					hasError.Store(true)
-					span.RecordError(err)
+					otlp.RecordError(ctx, err)
 
 					result <- BulkElementResult{
 						Error: err,

internal/api/common/errors.go

@@ -4,8 +4,8 @@ import (
 	"errors"
 	"github.com/formancehq/go-libs/v2/api"
 	"github.com/formancehq/go-libs/v2/logging"
+	"github.com/formancehq/go-libs/v2/otlp"
 	"github.com/formancehq/go-libs/v2/platform/postgres"
-	"go.opentelemetry.io/otel/trace"
 	"net/http"
 )
 
@@ -38,10 +38,7 @@ func HandleCommonErrors(w http.ResponseWriter, r *http.Request, err error) {
 }
 
 func InternalServerError(w http.ResponseWriter, r *http.Request, err error) {
-	span := trace.SpanFromContext(r.Context())
-	if span != nil {
-		span.RecordError(err)
-	}
+	otlp.RecordError(r.Context(), err)
 	logging.FromContext(r.Context()).Error(err)
 	api.WriteErrorResponse(w, http.StatusInternalServerError, api.ErrorInternal, errors.New("Internal error. Consult logs/traces to have more details."))
 }

internal/api/router.go

@@ -4,6 +4,7 @@
 	"fmt"
 	"github.com/formancehq/go-libs/v2/api"
 	"github.com/formancehq/go-libs/v2/bun/bunpaginate"
+	"github.com/formancehq/go-libs/v2/otlp"
 	"github.com/formancehq/go-libs/v2/service"
 	"github.com/formancehq/ledger/internal/api/bulking"
 	"github.com/formancehq/ledger/internal/controller/system"
@@ -60,7 +61,7 @@
 							middleware.PrintPrettyStack(rvr)
 						}
 
-						trace.SpanFromContext(r.Context()).RecordError(fmt.Errorf("%s", rvr))
+						otlp.RecordError(r.Context(), fmt.Errorf("%s", rvr))
 
 						w.WriteHeader(http.StatusInternalServerError)
 					}

internal/storage/bucket/default_bucket.go

@@ -39,7 +39,7 @@ func (b *DefaultBucket) IsUpToDate(ctx context.Context) (bool, error) {
 }
 
 func (b *DefaultBucket) Migrate(ctx context.Context, options ...migrations.Option) error {
-	return runMigrate(ctx, b.tracer, b.db, b.name, options...)
+	return runMigrate(ctx, b.tracer, b.db, b.name, append(options, migrations.WithTracer(b.tracer))...)
 }
 
 func (b *DefaultBucket) HasMinimalVersion(ctx context.Context) (bool, error) {

internal/storage/bucket/migrations.go

@@ -7,6 +7,7 @@
 	"errors"
 	"fmt"
 	"github.com/formancehq/go-libs/v2/migrations"
+	"github.com/formancehq/go-libs/v2/otlp"
 	"github.com/uptrace/bun"
 	"go.opentelemetry.io/otel/trace"
 	"gopkg.in/yaml.v3"
@@ -55,6 +56,8 @@
 			if errors.Is(err, migrations.ErrAlreadyUpToDate) {
 				return nil
 			}
+			otlp.RecordError(ctx, err)
+
 			return err
 		}
 	}