feat: add logs exporter

Author: gfyrag

.github/workflows/main.yml

@@ -74,8 +74,6 @@ jobs:
   GoReleaser:
     runs-on: "formance-runner"
     if: contains(github.event.pull_request.labels.*.name, 'build-images') || github.ref == 'refs/heads/main' || github.event_name == 'merge_group'
-    needs:
-      - Dirty
     steps:
       - uses: earthly/actions-setup@v1
         with:

cmd/config.go

@@ -2,18 +2,50 @@ package cmd
 
 import (
 	"fmt"
+	"github.com/mitchellh/mapstructure"
+	"github.com/robfig/cron/v3"
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
+	"reflect"
 )
 
-func LoadConfig[V any](cmd *cobra.Command) (*V, error){
+type commonConfig struct {
+	NumscriptInterpreter        bool     `mapstructure:"experimental-numscript-interpreter"`
+	NumscriptInterpreterFlags   []string `mapstructure:"experimental-numscript-interpreter-flags"`
+	ExperimentalFeaturesEnabled bool     `mapstructure:"experimental-features"`
+	ExperimentalConnectors      bool     `mapstructure:"experimental-connectors"`
+}
+
+func decodeCronSchedule(sourceType, destType reflect.Type, value any) (any, error) {
+	if sourceType.Kind() != reflect.String {
+		return value, nil
+	}
+	if destType != reflect.TypeOf((*cron.Schedule)(nil)).Elem() {
+		return value, nil
+	}
+
+	parser := cron.NewParser(cron.Second | cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow | cron.Descriptor)
+	schedule, err := parser.Parse(value.(string))
+	if err != nil {
+		return nil, fmt.Errorf("parsing cron schedule: %w", err)
+	}
+
+	return schedule, nil
+}
+
+func LoadConfig[V any](cmd *cobra.Command) (*V, error) {
 	v := viper.New()
 	if err := v.BindPFlags(cmd.Flags()); err != nil {
 		return nil, fmt.Errorf("binding flags: %w", err)
 	}
 
 	var cfg V
-	if err := v.Unmarshal(&cfg); err != nil {
+	if err := v.Unmarshal(&cfg,
+		viper.DecodeHook(mapstructure.ComposeDecodeHookFunc(
+			decodeCronSchedule,
+			mapstructure.StringToTimeDurationHookFunc(),
+		)),
+	); err != nil {
 		return nil, fmt.Errorf("unmarshalling config: %w", err)
 	}
 

cmd/docs_events.go

@@ -3,7 +3,7 @@ package cmd
 import (
 	"encoding/json"
 	"fmt"
-	"github.com/formancehq/ledger/internal/bus"
+	"github.com/formancehq/ledger/pkg/events"
 	"github.com/invopop/jsonschema"
 	"github.com/spf13/cobra"
 	"os"
@@ -30,10 +30,10 @@ func NewDocEventsCommand() *cobra.Command {
 			}
 
 			for _, o := range []any{
-				bus.CommittedTransactions{},
-				bus.DeletedMetadata{},
-				bus.SavedMetadata{},
-				bus.RevertedTransaction{},
+				events.CommittedTransactions{},
+				events.DeletedMetadata{},
+				events.SavedMetadata{},
+				events.RevertedTransaction{},
 			} {
 				schema := jsonschema.Reflect(o)
 				data, err := json.MarshalIndent(schema, "", "  ")

cmd/root.go

@@ -14,6 +14,11 @@ import (
 
 const (
 	ServiceName = "ledger"
+
+	NumscriptInterpreterFlag        = "experimental-numscript-interpreter"
+	NumscriptInterpreterFlagsToPass = "experimental-numscript-interpreter-flags"
+	ExperimentalFeaturesFlag        = "experimental-features"
+	ExperimentalConnectors          = "experimental-connectors"
 )
 
 var (
@@ -30,6 +35,11 @@ func NewRootCommand() *cobra.Command {
 		Version:           Version,
 	}
 
+	root.PersistentFlags().Bool(ExperimentalFeaturesFlag, false, "Enable features configurability")
+	root.PersistentFlags().Bool(NumscriptInterpreterFlag, false, "Enable experimental numscript rewrite")
+	root.PersistentFlags().String(NumscriptInterpreterFlagsToPass, "", "Feature flags to pass to the experimental numscript interpreter")
+	root.PersistentFlags().Bool(ExperimentalConnectors, false, "Enable connectors support")
+
 	root.AddCommand(NewServeCommand())
 	root.AddCommand(NewBucketsCommand())
 	root.AddCommand(NewVersionCommand())

cmd/serve.go

@@ -4,8 +4,9 @@ import (
 	"fmt"
 	"github.com/formancehq/go-libs/v2/logging"
 	"github.com/formancehq/ledger/internal/api/common"
+	"github.com/formancehq/ledger/internal/replication/drivers"
+	"github.com/formancehq/ledger/internal/replication/drivers/all"
 	systemstore "github.com/formancehq/ledger/internal/storage/system"
-	"github.com/formancehq/ledger/internal/worker"
 	"net/http"
 	"net/http/pprof"
 	"time"
@@ -37,7 +38,8 @@ import (
 	"go.uber.org/fx"
 )
 
-type ServeConfig struct {
+type ServeCommandConfig struct {
+	commonConfig        `mapstructure:",squash"`
 	WorkerConfiguration `mapstructure:",squash"`
 
 	Bind                        string   `mapstructure:"bind"`
@@ -49,9 +51,6 @@ type ServeConfig struct {
 	DefaultPageSize             uint64   `mapstructure:"default-page-size"`
 	MaxPageSize                 uint64   `mapstructure:"max-page-size"`
 	WorkerEnabled               bool     `mapstructure:"worker"`
-	NumscriptInterpreter        bool     `mapstructure:"experimental-numscript-interpreter"`
-	NumscriptInterpreterFlags   []string `mapstructure:"experimental-numscript-interpreter-flags"`
-	ExperimentalFeaturesEnabled bool     `mapstructure:"experimental-features"`
 }
 
 const (
@@ -65,9 +64,6 @@ const (
 	DefaultPageSizeFlag             = "default-page-size"
 	MaxPageSizeFlag                 = "max-page-size"
 	WorkerEnabledFlag               = "worker"
-	NumscriptInterpreterFlag        = "experimental-numscript-interpreter"
-	NumscriptInterpreterFlagsToPass = "experimental-numscript-interpreter-flags"
-	ExperimentalFeaturesFlag        = "experimental-features"
 )
 
 func NewServeCommand() *cobra.Command {
@@ -76,7 +72,7 @@ func NewServeCommand() *cobra.Command {
 		SilenceUsage: true,
 		RunE: func(cmd *cobra.Command, _ []string) error {
 
-			cfg, err := LoadConfig[ServeConfig](cmd)
+			cfg, err := LoadConfig[ServeCommandConfig](cmd)
 			if err != nil {
 				return fmt.Errorf("loading config: %w", err)
 			}
@@ -97,6 +93,8 @@ func NewServeCommand() *cobra.Command {
 				storage.NewFXModule(storage.ModuleConfig{
 					AutoUpgrade: cfg.AutoUpgrade,
 				}),
+				drivers.NewFXModule(),
+				fx.Invoke(all.Register),
 				systemcontroller.NewFXModule(systemcontroller.ModuleConfiguration{
 					NumscriptInterpreter:      cfg.NumscriptInterpreter,
 					NumscriptInterpreterFlags: cfg.NumscriptInterpreterFlags,
@@ -122,18 +120,19 @@ func NewServeCommand() *cobra.Command {
 						MaxPageSize:     cfg.MaxPageSize,
 						DefaultPageSize: cfg.DefaultPageSize,
 					},
+					Connectors: cfg.ExperimentalConnectors,
 				}),
 				fx.Decorate(func(
 					params struct {
-					fx.In
+						fx.In
 
-					Handler          chi.Router
-					HealthController *health.HealthController
-					Logger           logging.Logger
+						Handler          chi.Router
+						HealthController *health.HealthController
+						Logger           logging.Logger
 
-					MeterProvider *metric.MeterProvider         `optional:"true"`
-					Exporter      *otlpmetrics.InMemoryExporter `optional:"true"`
-				},
+						MeterProvider *metric.MeterProvider         `optional:"true"`
+						Exporter      *otlpmetrics.InMemoryExporter `optional:"true"`
+					},
 				) chi.Router {
 					return assembleFinalRouter(
 						service.IsDebug(cmd),
@@ -150,10 +149,7 @@ func NewServeCommand() *cobra.Command {
 			}
 
 			if cfg.WorkerEnabled {
-				options = append(options, worker.NewFXModule(worker.ModuleConfig{
-					Schedule:     cfg.WorkerConfiguration.HashLogsBlockCRONSpec,
-					MaxBlockSize: cfg.WorkerConfiguration.HashLogsBlockMaxSize,
-				}))
+				options = append(options, newWorkerModule(cfg.WorkerConfiguration))
 			}
 
 			return service.New(cmd.OutOrStdout(), options...).Run(cmd)

cmd/worker.go

@@ -7,25 +7,46 @@ import (
 	"github.com/formancehq/go-libs/v2/otlp/otlpmetrics"
 	"github.com/formancehq/go-libs/v2/otlp/otlptraces"
 	"github.com/formancehq/go-libs/v2/service"
+	"github.com/formancehq/ledger/internal/replication"
+	"github.com/formancehq/ledger/internal/replication/drivers"
+	"github.com/formancehq/ledger/internal/replication/drivers/all"
 	"github.com/formancehq/ledger/internal/storage"
 	"github.com/formancehq/ledger/internal/worker"
+	"github.com/robfig/cron/v3"
 	"github.com/spf13/cobra"
 	"go.uber.org/fx"
+	"time"
 )
 
 const (
+	WorkerPipelinesSyncPeriodFlag      = "worker-pipelines-sync-period"
+	WorkerPipelinesPullIntervalFlag    = "worker-pipelines-pull-interval"
+	WorkerPipelinesPushRetryPeriodFlag = "worker-pipelines-push-retry-period"
+
 	WorkerAsyncBlockHasherMaxBlockSizeFlag = "worker-async-block-hasher-max-block-size"
 	WorkerAsyncBlockHasherScheduleFlag     = "worker-async-block-hasher-schedule"
 )
 
 type WorkerConfiguration struct {
-	HashLogsBlockMaxSize  int    `mapstructure:"worker-async-block-hasher-max-block-size"`
-	HashLogsBlockCRONSpec string `mapstructure:"worker-async-block-hasher-schedule"`
+	HashLogsBlockMaxSize  int           `mapstructure:"worker-async-block-hasher-max-block-size"`
+	HashLogsBlockCRONSpec cron.Schedule `mapstructure:"worker-async-block-hasher-schedule"`
+
+	SyncPeriod      time.Duration `mapstructure:"worker-pipelines-sync-period"`
+	PushRetryPeriod time.Duration `mapstructure:"worker-pipelines-push-retry-period"`
+	PullInterval    time.Duration `mapstructure:"worker-pipelines-pull-interval"`
+}
+
+type WorkerCommandConfiguration struct {
+	WorkerConfiguration `mapstructure:",squash"`
+	commonConfig        `mapstructure:",squash"`
 }
 
 func addWorkerFlags(cmd *cobra.Command) {
 	cmd.Flags().Int(WorkerAsyncBlockHasherMaxBlockSizeFlag, 1000, "Max block size")
 	cmd.Flags().String(WorkerAsyncBlockHasherScheduleFlag, "0 * * * * *", "Schedule")
+	cmd.Flags().Duration(WorkerPipelinesSyncPeriodFlag, 5*time.Second, "Pipelines sync period")
+	cmd.Flags().Duration(WorkerPipelinesPullIntervalFlag, 5*time.Second, "Pipelines pull interval")
+	cmd.Flags().Duration(WorkerPipelinesPushRetryPeriodFlag, 10*time.Second, "Pipelines push retry period")
 }
 
 func NewWorkerCommand() *cobra.Command {
@@ -38,7 +59,7 @@ func NewWorkerCommand() *cobra.Command {
 				return err
 			}
 
-			cfg, err := LoadConfig[WorkerConfiguration](cmd)
+			cfg, err := LoadConfig[WorkerCommandConfiguration](cmd)
 			if err != nil {
 				return fmt.Errorf("loading config: %w", err)
 			}
@@ -50,10 +71,9 @@ func NewWorkerCommand() *cobra.Command {
 				otlpmetrics.FXModuleFromFlags(cmd),
 				bunconnect.Module(*connectionOptions, service.IsDebug(cmd)),
 				storage.NewFXModule(storage.ModuleConfig{}),
-				worker.NewFXModule(worker.ModuleConfig{
-					MaxBlockSize: cfg.HashLogsBlockMaxSize,
-					Schedule:     cfg.HashLogsBlockCRONSpec,
-				}),
+				drivers.NewFXModule(),
+				fx.Invoke(all.Register),
+				newWorkerModule(cfg.WorkerConfiguration),
 			).Run(cmd)
 		},
 	}
@@ -66,3 +86,17 @@ func NewWorkerCommand() *cobra.Command {
 
 	return cmd
 }
+
+func newWorkerModule(configuration WorkerConfiguration) fx.Option {
+	return worker.NewFXModule(worker.ModuleConfig{
+		AsyncBlockRunnerConfig: storage.AsyncBlockRunnerConfig{
+			MaxBlockSize: configuration.HashLogsBlockMaxSize,
+			Schedule:     configuration.HashLogsBlockCRONSpec,
+		},
+		ReplicationConfig: runner.ModuleConfig{
+			SyncPeriod:      configuration.SyncPeriod,
+			PushRetryPeriod: configuration.PushRetryPeriod,
+			PullInterval:    configuration.PullInterval,
+		},
+	})
+}

deployments/pulumi/.gitignore

@@ -1 +1,2 @@
 Pulumi.*.yaml
+/examples/