feat: add clickhouse on pulumi

Author: gfyrag

deployments/pulumi/main.go

@@ -4,6 +4,8 @@ import (
 	"github.com/formancehq/ledger/deployments/pulumi/pkg"
 	"github.com/formancehq/ledger/deployments/pulumi/pkg/config"
 	"github.com/pulumi/pulumi/sdk/v3/go/pulumi"
+
+	_ "github.com/formancehq/ledger/deployments/pulumi/pkg/connectors/clickhouse"
 )
 
 func main() {

deployments/pulumi/pkg/component.go

@@ -3,7 +3,9 @@ package ledger
 import (
 	"fmt"
 	"github.com/formancehq/ledger/deployments/pulumi/pkg/api"
+	"github.com/formancehq/ledger/deployments/pulumi/pkg/connectors"
 	"github.com/formancehq/ledger/deployments/pulumi/pkg/devbox"
+	"github.com/formancehq/ledger/deployments/pulumi/pkg/initialize"
 	"github.com/formancehq/ledger/deployments/pulumi/pkg/storage"
 	"github.com/formancehq/ledger/deployments/pulumi/pkg/utils"
 	"github.com/formancehq/ledger/deployments/pulumi/pkg/worker"
@@ -20,6 +22,7 @@ type ComponentArgs struct {
 	InstallDevBox pulumix.Input[bool]
 	Database      storage.DatabaseArgs
 	API           api.Args
+	Connectors    connectors.Args
 	Worker        worker.Args
 	Ingress       *api.IngressArgs
 }
@@ -34,11 +37,13 @@ func (args *ComponentArgs) SetDefaults() {
 type Component struct {
 	pulumi.ResourceState
 
-	API       *api.Component
-	Worker    *worker.Component
-	Storage   *storage.Component
-	Namespace *corev1.Namespace
-	Devbox    *devbox.Component
+	API        *api.Component
+	Worker     *worker.Component
+	Storage    *storage.Component
+	Namespace  *corev1.Namespace
+	Devbox     *devbox.Component
+	Connectors *connectors.Component
+	Initialize *initialize.Component
 }
 
 func NewComponent(ctx *pulumi.Context, name string, args ComponentArgs, opts ...pulumi.ResourceOption) (*Component, error) {
@@ -96,6 +101,14 @@ func NewComponent(ctx *pulumi.Context, name string, args ComponentArgs, opts ...
 		return nil, err
 	}
 
+	cmp.Connectors, err = connectors.NewComponent(ctx, "connectors", connectors.ComponentArgs{
+		CommonArgs: args.CommonArgs,
+		Args:       args.Connectors,
+	})
+	if err != nil {
+		return nil, err
+	}
+
 	cmp.Worker, err = worker.NewComponent(ctx, "worker", worker.ComponentArgs{
 		CommonArgs: args.CommonArgs,
 		Args:       args.Worker,
@@ -106,6 +119,12 @@ func NewComponent(ctx *pulumi.Context, name string, args ComponentArgs, opts ...
 		return nil, err
 	}
 
+	cmp.Initialize, err = initialize.NewComponent(ctx, "initialize", initialize.ComponentArgs{
+		CommonArgs: args.CommonArgs,
+		Connectors: cmp.Connectors,
+		API:        cmp.API,
+	}, options...)
+
 	installDevBox, err := internals.UnsafeAwaitOutput(ctx.Context(), args.InstallDevBox.ToOutput(ctx.Context()))
 	if err != nil {
 		return nil, err

deployments/pulumi/pkg/config/config.go

@@ -1,17 +1,21 @@
 package config
 
 import (
+	"encoding/json"
 	"errors"
 	"fmt"
+	. "github.com/formancehq/go-libs/v2/collectionutils"
 	pulumi_ledger "github.com/formancehq/ledger/deployments/pulumi/pkg"
 	"github.com/formancehq/ledger/deployments/pulumi/pkg/api"
+	"github.com/formancehq/ledger/deployments/pulumi/pkg/connectors"
 	"github.com/formancehq/ledger/deployments/pulumi/pkg/storage"
 	"github.com/formancehq/ledger/deployments/pulumi/pkg/utils"
 	"github.com/formancehq/ledger/deployments/pulumi/pkg/worker"
 	"github.com/pulumi/pulumi-aws/sdk/v6/go/aws/rds"
 	"github.com/pulumi/pulumi/sdk/v3/go/pulumi"
 	"github.com/pulumi/pulumi/sdk/v3/go/pulumi/config"
 	"github.com/pulumi/pulumi/sdk/v3/go/pulumix"
+	"reflect"
 	"time"
 )
 
@@ -248,25 +252,66 @@ func (d API) toInput() api.Args {
 	}
 }
 
-type Worker struct {
-	Connectors map[string]map[string]any `json:"connectors" yaml:"connectors"`
+type Connector struct {
+	Driver string `json:"driver" yaml:"driver"`
+	Config any    `json:"config" yaml:"config"`
 }
 
-func (w Worker) toInput() worker.Args {
-	connectors := make(map[string]pulumi.Map)
-	for name, rawConfig := range w.Connectors {
-		convertedConfig := pulumi.Map{}
-		for k, v := range rawConfig {
-			convertedConfig[k] = pulumi.Any(v)
+func (c Connector) toInput() connectors.ConnectorArgs {
+	return connectors.ConnectorArgs{
+		Driver: c.Driver,
+		Config: c.Config,
+	}
+}
+
+type Connectors map[string]Connector
+
+func (c *Connectors) UnmarshalJSON(data []byte) error {
+	asMap := make(map[string]json.RawMessage, 0)
+	if err := json.Unmarshal(data, &asMap); err != nil {
+		return fmt.Errorf("error unmarshalling connectors into an array: %w", err)
+	}
+
+	*c = make(map[string]Connector)
+	for id, elem := range asMap {
+		type def struct {
+			Driver string `json:"driver" yaml:"driver"`
+		}
+		d := def{}
+		if err := json.Unmarshal(elem, &d); err != nil {
+			return fmt.Errorf("error unmarshalling connector definition %s: %w", id, err)
+		}
+
+		cfg, err := connectors.GetConnectorConfig(d.Driver)
+		if err != nil {
+			return err
+		}
+
+		if err := json.Unmarshal(elem, cfg); err != nil {
+			return fmt.Errorf("error unmarshalling connector config %s: %w", id, err)
+		}
+
+		(*c)[id] = Connector{
+			Driver: d.Driver,
+			Config: reflect.ValueOf(cfg).Elem().Interface(),
 		}
-		connectors[name] = convertedConfig
 	}
 
-	return worker.Args{
-		Connectors: connectors,
+	return nil
+}
+
+func (c *Connectors) toInput() connectors.Args {
+	return connectors.Args{
+		Connectors: ConvertMap(*c, Connector.toInput),
 	}
 }
 
+type Worker struct{}
+
+func (w Worker) toInput() worker.Args {
+	return worker.Args{}
+}
+
 type Monitoring struct {
 	// ResourceAttributes is the resource attributes for OpenTelemetry
 	ResourceAttributes map[string]string `json:"resource-attributes" yaml:"resource-attributes"`
@@ -431,6 +476,9 @@ type Config struct {
 	// Worker is the worker configuration for the ledger
 	Worker *Worker `json:"worker" yaml:"worker"`
 
+	// Connectors is the connectors configuration for the ledger
+	Connectors Connectors `json:"connectors" yaml:"connectors"`
+
 	// Ingress is the ingress configuration for the ledger
 	Ingress *Ingress `json:"ingress" yaml:"ingress"`
 
@@ -446,10 +494,11 @@ func (cfg Config) ToInput() pulumi_ledger.ComponentArgs {
 		CommonArgs:    cfg.Common.toInput(),
 		Database:      cfg.Storage.toInput(),
 		API:           cfg.API.toInput(),
-		Worker: cfg.Worker.toInput(),
+		Worker:        cfg.Worker.toInput(),
 		Timeout:       pulumix.Val(cfg.Timeout),
 		Ingress:       cfg.Ingress.toInput(),
 		InstallDevBox: pulumix.Val(cfg.InstallDevBox),
+		Connectors:    cfg.Connectors.toInput(),
 	}
 }
 
@@ -494,6 +543,13 @@ func Load(ctx *pulumi.Context) (*Config, error) {
 		}
 	}
 
+	connectors := Connectors{}
+	if err := config.GetObject(ctx, "connectors", &connectors); err != nil {
+		if !errors.Is(err, config.ErrMissingVar) {
+			return nil, err
+		}
+	}
+
 	otel := &Monitoring{}
 	if err := config.GetObject(ctx, "monitoring", otel); err != nil {
 		if !errors.Is(err, config.ErrMissingVar) {
@@ -512,7 +568,8 @@ func Load(ctx *pulumi.Context) (*Config, error) {
 		InstallDevBox: config.GetBool(ctx, "install-dev-box"),
 		Storage:       storage,
 		API:           api,
-		Worker: worker,
+		Worker:        worker,
 		Ingress:       ingress,
+		Connectors:    connectors,
 	}, nil
 }

deployments/pulumi/pkg/connectors/clickhouse/component_external.go

@@ -0,0 +1,42 @@
+package clickhouse
+
+import (
+	"fmt"
+	"github.com/formancehq/ledger/deployments/pulumi/pkg/connectors"
+	"github.com/pulumi/pulumi/sdk/v3/go/pulumi"
+	"github.com/pulumi/pulumi/sdk/v3/go/pulumix"
+)
+
+type externalComponent struct {
+	pulumi.ResourceState
+
+	DSN pulumix.Output[string]
+}
+
+func (e *externalComponent) GetConfig() pulumi.AnyOutput {
+	return pulumi.Any(map[string]any{
+		"dsn": e.DSN,
+	})
+}
+
+type externalComponentArgs struct {
+	DSN pulumi.String
+}
+
+func newExternalComponent(ctx *pulumi.Context, name string, args externalComponentArgs, opts ...pulumi.ResourceOption) (*externalComponent, error) {
+	cmp := &externalComponent{}
+	err := ctx.RegisterComponentResource("Formance:Ledger:Clickhouse:External", name, cmp, opts...)
+	if err != nil {
+		return nil, err
+	}
+
+	cmp.DSN = args.DSN.ToOutput(ctx.Context())
+
+	if err := ctx.RegisterResourceOutputs(cmp, pulumi.Map{}); err != nil {
+		return nil, fmt.Errorf("registering outputs: %w", err)
+	}
+
+	return cmp, nil
+}
+
+var _ connectors.ConnectorComponent = (*externalComponent)(nil)

deployments/pulumi/pkg/connectors/clickhouse/component_internal.go

@@ -0,0 +1,100 @@
+package clickhouse
+
+import (
+	"errors"
+	"fmt"
+	"github.com/formancehq/ledger/deployments/pulumi/pkg/connectors"
+	"github.com/formancehq/ledger/deployments/pulumi/pkg/utils"
+	corev1 "github.com/pulumi/pulumi-kubernetes/sdk/v4/go/kubernetes/core/v1"
+	helm "github.com/pulumi/pulumi-kubernetes/sdk/v4/go/kubernetes/helm/v4"
+	"github.com/pulumi/pulumi/sdk/v3/go/pulumi"
+	"github.com/pulumi/pulumi/sdk/v3/go/pulumi/internals"
+	"github.com/pulumi/pulumi/sdk/v3/go/pulumix"
+)
+
+type internalComponent struct {
+	pulumi.ResourceState
+
+	DSN     pulumix.Output[string]
+	Chart   *helm.Chart
+	Service *corev1.Service
+}
+
+func (e *internalComponent) GetConfig() pulumi.AnyOutput {
+	return pulumi.Any(map[string]any{
+		"dsn": pulumi.Sprintf(
+			"clickhouse://%s.%s.svc.cluster.local:%d",
+			e.Service.Metadata.Name().Elem(),
+			e.Service.Metadata.Namespace().Elem(),
+			9000,
+		),
+	})
+}
+
+type internalComponentArgs struct {
+	utils.CommonArgs
+}
+
+func newInternalComponent(ctx *pulumi.Context, name string, args internalComponentArgs, opts ...pulumi.ResourceOption) (*internalComponent, error) {
+	cmp := &internalComponent{}
+	err := ctx.RegisterComponentResource("Formance:Ledger:Clickhouse:Internal", name, cmp, opts...)
+	if err != nil {
+		return nil, err
+	}
+
+	cmp.Chart, err = helm.NewChart(ctx, "clickhouse", &helm.ChartArgs{
+		Chart:     pulumi.String("oci://registry-1.docker.io/bitnamicharts/clickhouse"),
+		Version:   pulumi.String("8.0.5"),
+		Name:      pulumi.String("clickhouse"),
+		Namespace: args.Namespace.ToOutput(ctx.Context()).Untyped().(pulumi.StringOutput),
+		Values: pulumi.Map{
+			"replicaCount": pulumi.Int(1),
+			"shards":       pulumi.Int(1),
+			"zookeeper": pulumi.Map{
+				"enabled": pulumi.Bool(false),
+			},
+		},
+	}, pulumi.Parent(cmp))
+	if err != nil {
+		return nil, err
+	}
+
+	ret, err := internals.UnsafeAwaitOutput(ctx.Context(), pulumix.ApplyErr(cmp.Chart.Resources, func(resources []any) (*corev1.Service, error) {
+		for _, resource := range resources {
+			service, ok := resource.(*corev1.Service)
+			if !ok {
+				continue
+			}
+			ret, err := internals.UnsafeAwaitOutput(ctx.Context(), pulumix.Apply2(
+				service.Spec.Type().Elem(),
+				service.Spec.ClusterIP().Elem(),
+				func(serviceType, clusterIP string) *corev1.Service {
+					// select not headless service
+					if serviceType != "ClusterIP" || clusterIP == "None" {
+						return nil
+					}
+					return service
+				},
+			))
+			if err != nil {
+				return nil, err
+			}
+			if ret.Value != nil {
+				return ret.Value.(*corev1.Service), nil
+			}
+		}
+		return nil, errors.New("not found")
+	}))
+	if err != nil {
+		return nil, err
+	}
+	cmp.Service = ret.Value.(*corev1.Service)
+
+	if err := ctx.RegisterResourceOutputs(cmp, pulumi.Map{}); err != nil {
+		return nil, fmt.Errorf("registering outputs: %w", err)
+	}
+
+	return cmp, nil
+}
+
+var _ connectors.ConnectorComponent = (*internalComponent)(nil)