feat: add log export

Author: gfyrag

Justfile

@@ -37,7 +37,7 @@ openapi:
   @yq eval-all '. as $item ireduce ({}; . * $item)' openapi/v1.yaml openapi/v2.yaml openapi/overlay.yaml > openapi.yaml
   @npx -y widdershins {{justfile_directory()}}/openapi/v2.yaml -o {{justfile_directory()}}/docs/api/README.md --search false --language_tabs 'http:HTTP' --summary --omitHeader
 
-generate-client:
+generate-client: openapi
   @speakeasy generate sdk -s openapi.yaml -o ./pkg/client -l go
 
 release-local:

cmd/serve.go

@@ -3,9 +3,10 @@ package cmd
 import (
 	"github.com/formancehq/go-libs/v2/logging"
 	"github.com/formancehq/ledger/internal/api/common"
-	replicationcontroller "github.com/formancehq/ledger/internal/replication/controller"
-	"github.com/formancehq/ledger/internal/replication/leadership"
 	"github.com/formancehq/ledger/internal/leadership"
+	"github.com/formancehq/ledger/internal/replication/drivers"
+	"github.com/formancehq/ledger/internal/replication/drivers/all"
+	"github.com/formancehq/ledger/internal/replication/runner"
 	systemstore "github.com/formancehq/ledger/internal/storage/system"
 	"net/http"
 	"net/http/pprof"
@@ -39,17 +40,20 @@ import (
 )
 
 const (
-	BindFlag                        = "bind"
-	BallastSizeInBytesFlag          = "ballast-size"
-	NumscriptCacheMaxCountFlag      = "numscript-cache-max-count"
-	AutoUpgradeFlag                 = "auto-upgrade"
-	ExperimentalFeaturesFlag        = "experimental-features"
-	BulkMaxSizeFlag                 = "bulk-max-size"
-	BulkParallelFlag                = "bulk-parallel"
-	NumscriptInterpreterFlag        = "experimental-numscript-interpreter"
+	BindFlag                     = "bind"
+	BallastSizeInBytesFlag       = "ballast-size"
+	NumscriptCacheMaxCountFlag   = "numscript-cache-max-count"
+	AutoUpgradeFlag              = "auto-upgrade"
+	ExperimentalFeaturesFlag     = "experimental-features"
+	BulkMaxSizeFlag              = "bulk-max-size"
+	BulkParallelFlag             = "bulk-parallel"
+	NumscriptInterpreterFlag     = "experimental-numscript-interpreter"
 	NumscriptInterpreterFlagsToPass = "numscript-interpreter-flags"
-	DefaultPageSizeFlag             = "default-page-size"
-	MaxPageSizeFlag                 = "max-page-size"
+	DefaultPageSizeFlag          = "default-page-size"
+	MaxPageSizeFlag              = "max-page-size"
+	PipelinesSyncPeriodFlag      = "pipelines-sync-period"
+	PipelinesPullIntervalFlag    = "pipelines-pull-interval"
+	PipelinesPushRetryPeriodFlag = "pipelines-push-retry-period"
 )
 
 func NewServeCommand() *cobra.Command {
@@ -100,6 +104,13 @@ func NewServeCommand() *cobra.Command {
 				auth.FXModuleFromFlags(cmd),
 				bunconnect.Module(*connectionOptions, service.IsDebug(cmd)),
 				storage.NewFXModule(serveConfiguration.autoUpgrade),
+				runner.NewFXModule(runner.ModuleConfig{
+					SyncPeriod:      serveConfiguration.pipelinesSyncPeriod,
+					PullInterval:    serveConfiguration.pipelinesPullInterval,
+					PushRetryPeriod: serveConfiguration.pipelinesPushRetryPeriod,
+				}),
+				drivers.NewFXModule(),
+				fx.Invoke(all.Register),
 				systemcontroller.NewFXModule(systemcontroller.ModuleConfiguration{
 					NumscriptInterpreter:      numscriptInterpreter,
 					NumscriptInterpreterFlags: numscriptInterpreterFlags,
@@ -114,7 +125,9 @@ func NewServeCommand() *cobra.Command {
 				}),
 				bus.NewFxModule(),
 				ballast.Module(serveConfiguration.ballastSize),
-				leadership.NewFXModule(),
+				leadership.NewFXModule(leadership.ModuleConfig{
+					LeadershipRetryPeriod: 5 * time.Second,
+				}),
 				api.Module(api.Config{
 					Version: Version,
 					Debug:   service.IsDebug(cmd),
@@ -148,8 +161,6 @@ func NewServeCommand() *cobra.Command {
 						params.Handler,
 					)
 				}),
-				replicationcontroller.NewFXModule(),
-				leadership.NewFXModule(),
 				fx.Invoke(func(lc fx.Lifecycle, h chi.Router) {
 					lc.Append(httpserver.NewHook(h, httpserver.WithAddress(serveConfiguration.bind)))
 				}),
@@ -169,6 +180,9 @@ func NewServeCommand() *cobra.Command {
 	cmd.Flags().String(NumscriptInterpreterFlagsToPass, "", "Feature flags to pass to the experimental numscript interpreter")
 	cmd.Flags().Uint64(MaxPageSizeFlag, 100, "Max page size")
 	cmd.Flags().Uint64(DefaultPageSizeFlag, 15, "Default page size")
+	cmd.Flags().Duration(PipelinesSyncPeriodFlag, 5*time.Second, "Pipelines sync period")
+	cmd.Flags().Duration(PipelinesPullIntervalFlag, runner.DefaultPullInterval, "Pipelines pull interval")
+	cmd.Flags().Duration(PipelinesPushRetryPeriodFlag, runner.DefaultPushRetryPeriod, "Pipelines push retry period")
 
 	service.AddFlags(cmd.Flags())
 	bunconnect.AddFlags(cmd.Flags())
@@ -184,10 +198,13 @@ func NewServeCommand() *cobra.Command {
 }
 
 type serveConfiguration struct {
-	ballastSize            uint
-	numscriptCacheMaxCount uint
-	autoUpgrade            bool
-	bind                   string
+	ballastSize              uint
+	numscriptCacheMaxCount   uint
+	autoUpgrade              bool
+	bind                     string
+	pipelinesSyncPeriod      time.Duration
+	pipelinesPullInterval    time.Duration
+	pipelinesPushRetryPeriod time.Duration
 }
 
 func discoverServeConfiguration(cmd *cobra.Command) serveConfiguration {
@@ -196,6 +213,9 @@ func discoverServeConfiguration(cmd *cobra.Command) serveConfiguration {
 	ret.numscriptCacheMaxCount, _ = cmd.Flags().GetUint(NumscriptCacheMaxCountFlag)
 	ret.autoUpgrade, _ = cmd.Flags().GetBool(AutoUpgradeFlag)
 	ret.bind, _ = cmd.Flags().GetString(BindFlag)
+	ret.pipelinesSyncPeriod, _ = cmd.Flags().GetDuration(PipelinesSyncPeriodFlag)
+	ret.pipelinesPullInterval, _ = cmd.Flags().GetDuration(PipelinesPullIntervalFlag)
+	ret.pipelinesPushRetryPeriod, _ = cmd.Flags().GetDuration(PipelinesPushRetryPeriodFlag)
 
 	return ret
 }

deployments/pulumi/pkg/component.go

@@ -7,6 +7,7 @@ import (
 	batchv1 "github.com/pulumi/pulumi-kubernetes/sdk/v4/go/kubernetes/batch/v1"
 	corev1 "github.com/pulumi/pulumi-kubernetes/sdk/v4/go/kubernetes/core/v1"
 	metav1 "github.com/pulumi/pulumi-kubernetes/sdk/v4/go/kubernetes/meta/v1"
+	networkingv1 "github.com/pulumi/pulumi-kubernetes/sdk/v4/go/kubernetes/networking/v1"
 	"github.com/pulumi/pulumi/sdk/v3/go/pulumi"
 	"github.com/pulumi/pulumi/sdk/v3/go/pulumix"
 	"slices"
@@ -30,6 +31,7 @@ type Component struct {
 	ServiceNamespace   pulumix.Output[string]
 	ServicePort        pulumix.Output[int]
 	ServiceInternalURL pulumix.Output[string]
+	Deployment         *appsv1.Deployment
 }
 
 type PostgresArgs struct {
@@ -62,6 +64,11 @@ type OtelMetricsArgs struct {
 	OtelMetricsExporterOTLPInsecure               pulumix.Input[bool]
 }
 
+type IngressArgs struct {
+	Host   pulumix.Input[string]
+	Secret pulumix.Input[*string]
+}
+
 type OtelArgs struct {
 	ResourceAttributes pulumix.Input[map[string]string]
 	ServiceName        pulumix.Input[string]
@@ -73,6 +80,7 @@ type OtelArgs struct {
 type ComponentArgs struct {
 	Postgres                      PostgresArgs
 	Otel                          *OtelArgs
+	Ingress                       *IngressArgs
 	Namespace                     pulumix.Input[string]
 	Timeout                       pulumix.Input[int]
 	Tag                           pulumix.Input[string]
@@ -442,7 +450,7 @@ func NewComponent(ctx *pulumi.Context, name string, args *ComponentArgs, opts ..
 		terminationGracePeriodSeconds = args.TerminationGracePeriodSeconds.ToOutput(ctx.Context()).Untyped().(pulumi.IntPtrOutput)
 	}
 
-	deployment, err := appsv1.NewDeployment(ctx, "ledger", &appsv1.DeploymentArgs{
+	cmp.Deployment, err = appsv1.NewDeployment(ctx, "ledger", &appsv1.DeploymentArgs{
 		Metadata: &metav1.ObjectMetaArgs{
 			Namespace: namespace.Untyped().(pulumi.StringOutput),
 			Labels: pulumi.StringMap{
@@ -560,7 +568,7 @@ func NewComponent(ctx *pulumi.Context, name string, args *ComponentArgs, opts ..
 			Namespace: namespace.Untyped().(pulumi.StringOutput),
 		},
 		Spec: &corev1.ServiceSpecArgs{
-			Selector: deployment.Spec.Selector().MatchLabels(),
+			Selector: cmp.Deployment.Spec.Selector().MatchLabels(),
 			Type:     pulumi.String("ClusterIP"),
 			Ports: corev1.ServicePortArray{
 				corev1.ServicePortArgs{
@@ -598,7 +606,61 @@ func NewComponent(ctx *pulumi.Context, name string, args *ComponentArgs, opts ..
 		return url
 	})
 
+	if args.Ingress != nil {
+		_, err = networkingv1.NewIngress(ctx, "ledger", &networkingv1.IngressArgs{
+			Metadata: &metav1.ObjectMetaArgs{
+				Namespace: namespace.Untyped().(pulumi.StringOutput),
+			},
+			Spec: &networkingv1.IngressSpecArgs{
+				Rules: networkingv1.IngressRuleArray{
+					networkingv1.IngressRuleArgs{
+						Host: args.Ingress.Host.ToOutput(ctx.Context()).Untyped().(pulumi.StringOutput),
+						Http: networkingv1.HTTPIngressRuleValueArgs{
+							Paths: networkingv1.HTTPIngressPathArray{
+								networkingv1.HTTPIngressPathArgs{
+									Backend: networkingv1.IngressBackendArgs{
+										Service: &networkingv1.IngressServiceBackendArgs{
+											Name: cmp.ServiceName.ToOutput(ctx.Context()).Untyped().(pulumi.StringOutput),
+											Port: networkingv1.ServiceBackendPortArgs{
+												Name: pulumi.String("http"),
+											},
+										},
+									},
+									Path:     pulumi.String("/"),
+									PathType: pulumi.String("Prefix"),
+								},
+							},
+						},
+					},
+				},
+				Tls: pulumix.Apply(args.Ingress.Secret, func(secret *string) networkingv1.IngressTLSArrayInput {
+					if secret == nil || *secret == "" {
+						return nil
+					}
+
+					return networkingv1.IngressTLSArray{
+						networkingv1.IngressTLSArgs{
+							Hosts: pulumi.StringArray{
+								args.Ingress.Host.ToOutput(ctx.Context()).Untyped().(pulumi.StringOutput),
+							},
+							SecretName: pulumi.String(*secret),
+						},
+					}
+				}).Untyped().(pulumi.AnyOutput).ApplyT(func(v any) networkingv1.IngressTLSArrayInput {
+					if v == nil {
+						return nil
+					}
+					return v.(networkingv1.IngressTLSArrayInput)
+				}).(networkingv1.IngressTLSArrayInput),
+			},
+		}, pulumi.Parent(cmp))
+		if err != nil {
+			return nil, err
+		}
+	}
+
 	if err := ctx.RegisterResourceOutputs(cmp, pulumi.Map{
+		"deployment-name":      cmp.Deployment.Metadata.Name(),
 		"service-name":         cmp.ServiceName,
 		"service-namespace":    cmp.ServiceNamespace,
 		"service-port":         cmp.ServicePort,