feat: logs exporter

.github/workflows/main.yml

@@ -77,8 +77,6 @@ jobs:
   GoReleaser:
     runs-on: "shipfox-4vcpu-ubuntu-2404"
     if: contains(github.event.pull_request.labels.*.name, 'build-images') || github.ref == 'refs/heads/main' || github.event_name == 'merge_group'
-    needs:
-      - Dirty
     steps:
       - name: Set up QEMU
         uses: docker/setup-qemu-action@v3
@@ -91,6 +89,7 @@ jobs:
       - uses: 'actions/checkout@v4'
         with:
           fetch-depth: 0
+          ref: ${{ github.event_name == 'pull_request' && github.event.pull_request.head.sha || github.sha }}
       - name: Setup Env
         uses: ./.github/actions/default
         with:

Earthfile

@@ -49,34 +49,4 @@ deploy:
     RUN kubectl patch Versions.formance.com default -p "{\"spec\":{\"ledger\": \"${tag}\"}}" --type=merge
 
 deploy-staging:
-    BUILD --pass-args core+deploy-staging
-
-export-database-schema:
-    FROM +sources
-    RUN go install github.com/roerohan/wait-for-it@latest
-    WITH DOCKER --load=postgres:15-alpine=+postgres --pull schemaspy/schemaspy:6.2.4
-        RUN bash -c '
-            echo "Creating PG server...";
-            postgresContainerID=$(docker run -d --rm -e POSTGRES_USER=root -e POSTGRES_PASSWORD=root -e POSTGRES_DB=formance --net=host postgres:15-alpine);
-            wait-for-it -w 127.0.0.1:5432;
-
-            echo "Creating bucket...";
-            go run main.go buckets upgrade _default --postgres-uri "postgres://root:root@127.0.0.1:5432/formance?sslmode=disable";
-
-            echo "Exporting schemas...";
-            docker run --rm -u root \
-              -v ./docs/database:/output \
-              --net=host \
-              schemaspy/schemaspy:6.2.4 -u root -db formance -t pgsql11 -host 127.0.0.1 -port 5432 -p root -schemas _system,_default;
-
-            docker kill "$postgresContainerID";
-        '
-    END
-    SAVE ARTIFACT docs/database/_system/diagrams AS LOCAL docs/database/_system/diagrams
-    SAVE ARTIFACT docs/database/_default/diagrams AS LOCAL docs/database/_default/diagrams
-
-openapi:
-    FROM core+base-image
-    WORKDIR /src
-    COPY openapi.yaml openapi.yaml
-    SAVE ARTIFACT ./openapi.yaml
+    BUILD --pass-args core+deploy-staging

Justfile

@@ -56,3 +56,8 @@ release-ci:
 
 release:
     @goreleaser release --clean
+
+generate-grpc-replication:
+    protoc --go_out=. --go_opt=paths=source_relative \
+        --go-grpc_out=. --go-grpc_opt=paths=source_relative \
+        ./internal/replication/grpc/replication_service.proto

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"`
+	ExperimentalExporters       bool `mapstructure:"experimental-exporters"`
+}
+
+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

@@ -12,6 +12,11 @@ import (
 
 const (
 	ServiceName = "ledger"
+
+	NumscriptInterpreterFlag        = "experimental-numscript-interpreter"
+	NumscriptInterpreterFlagsToPass = "experimental-numscript-interpreter-flags"
+	ExperimentalFeaturesFlag = "experimental-features"
+	ExperimentalExporters    = "experimental-exporters"
 )
 
 var (
@@ -28,6 +33,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(ExperimentalExporters, false, "Enable exporters support")
+
 	root.AddCommand(NewServeCommand())
 	root.AddCommand(NewBucketsCommand())
 	root.AddCommand(NewVersionCommand())

cmd/serve.go

@@ -4,8 +4,13 @@ import (
 	"fmt"
 	"github.com/formancehq/go-libs/v3/logging"
 	"github.com/formancehq/ledger/internal/api/common"
+	"github.com/formancehq/ledger/internal/replication"
+	"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"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
 	"net/http"
 	"net/http/pprof"
 	"time"
@@ -37,21 +42,20 @@ import (
 	"go.uber.org/fx"
 )
 
-type ServeConfig struct {
+type ServeCommandConfig struct {
+	commonConfig        `mapstructure:",squash"`
 	WorkerConfiguration `mapstructure:",squash"`
 
-	Bind                        string   `mapstructure:"bind"`
-	BallastSizeInBytes          uint     `mapstructure:"ballast-size"`
-	NumscriptCacheMaxCount      uint     `mapstructure:"numscript-cache-max-count"`
-	AutoUpgrade                 bool     `mapstructure:"auto-upgrade"`
-	BulkMaxSize                 int      `mapstructure:"bulk-max-size"`
-	BulkParallel                int      `mapstructure:"bulk-parallel"`
-	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"`
+	Bind                   string `mapstructure:"bind"`
+	BallastSizeInBytes     uint   `mapstructure:"ballast-size"`
+	NumscriptCacheMaxCount uint   `mapstructure:"numscript-cache-max-count"`
+	AutoUpgrade            bool   `mapstructure:"auto-upgrade"`
+	BulkMaxSize            int    `mapstructure:"bulk-max-size"`
+	BulkParallel           int    `mapstructure:"bulk-parallel"`
+	DefaultPageSize        uint64 `mapstructure:"default-page-size"`
+	MaxPageSize            uint64 `mapstructure:"max-page-size"`
+	WorkerEnabled          bool   `mapstructure:"worker"`
+	WorkerAddress          string `mapstructure:"worker-grpc-address"`
 }
 
 const (
@@ -62,12 +66,9 @@ const (
 	BulkMaxSizeFlag            = "bulk-max-size"
 	BulkParallelFlag           = "bulk-parallel"
 
-	DefaultPageSizeFlag             = "default-page-size"
-	MaxPageSizeFlag                 = "max-page-size"
-	WorkerEnabledFlag               = "worker"
-	NumscriptInterpreterFlag        = "experimental-numscript-interpreter"
-	NumscriptInterpreterFlagsToPass = "experimental-numscript-interpreter-flags"
-	ExperimentalFeaturesFlag        = "experimental-features"
+	DefaultPageSizeFlag = "default-page-size"
+	MaxPageSizeFlag     = "max-page-size"
+	WorkerEnabledFlag   = "worker"
 )
 
 func NewServeCommand() *cobra.Command {
@@ -76,7 +77,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 +98,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,6 +125,7 @@ func NewServeCommand() *cobra.Command {
 						MaxPageSize:     cfg.MaxPageSize,
 						DefaultPageSize: cfg.DefaultPageSize,
 					},
+					Exporters: cfg.ExperimentalExporters,
 				}),
 				fx.Decorate(func(
 					params struct {
@@ -150,10 +154,18 @@ func NewServeCommand() *cobra.Command {
 			}
 
 			if cfg.WorkerEnabled {
-				options = append(options, worker.NewFXModule(worker.ModuleConfig{
-					Schedule:     cfg.HashLogsBlockCRONSpec,
-					MaxBlockSize: cfg.HashLogsBlockMaxSize,
-				}))
+				options = append(options,
+					newWorkerModule(cfg.WorkerConfiguration),
+					replication.NewFXEmbeddedClientModule(),
+				)
+			} else {
+				options = append(options,
+					worker.NewGRPCClientFxModule(
+						cfg.WorkerAddress,
+						grpc.WithTransportCredentials(insecure.NewCredentials()),
+					),
+					replication.NewFXGRPCClientModule(),
+				)
 			}
 
 			return service.New(cmd.OutOrStdout(), options...).Run(cmd)
@@ -171,6 +183,7 @@ func NewServeCommand() *cobra.Command {
 	cmd.Flags().Bool(ExperimentalFeaturesFlag, false, "Enable features configurability")
 	cmd.Flags().Bool(NumscriptInterpreterFlag, false, "Enable experimental numscript rewrite")
 	cmd.Flags().String(NumscriptInterpreterFlagsToPass, "", "Feature flags to pass to the experimental numscript interpreter")
+	cmd.Flags().String(WorkerGRPCAddressFlag, "localhost:8081", "GRPC address")
 
 	addWorkerFlags(cmd)
 	bunconnect.AddFlags(cmd.Flags())