feat(operator-utils): add operator utils (#1243)

Earthfile

@@ -94,6 +94,9 @@ goreleaser:
 
 all-ci-goreleaser:
     LOCALLY
+    FOR component IN $(cd ./tools && ls -d */)
+        BUILD --pass-args +goreleaser --type=components --components=$component --mode=ci
+    END
     FOR component IN $(cd ./components && ls -d */)
         BUILD --pass-args +goreleaser --type=components --components=$component --mode=ci
     END
@@ -103,6 +106,9 @@ all-ci-goreleaser:
 
 build-all:
     LOCALLY
+    FOR component IN $(cd ./tools && ls -d */)
+        BUILD --pass-args ./tools/${component}+build-image
+    END
     FOR component IN $(cd ./components && ls -d */)
         BUILD --pass-args ./components/${component}+build-image
     END
@@ -115,6 +121,9 @@ deploy-all:
     WAIT
         BUILD --pass-args ./components/+deploy --components=operator
     END
+    FOR component IN $(cd ./tools && ls -d */)
+        BUILD --pass-args ./tools/+deploy --components=$component
+    END
     FOR component IN $(cd ./components && ls -d */)
         IF [ "$component" != "operator" ]
             BUILD --pass-args ./components/+deploy --components=$component
@@ -140,6 +149,9 @@ tests-integration:
 pre-commit: # Generate the final spec and run all the pre-commit hooks
     LOCALLY
     BUILD --pass-args ./releases+sdk-generate
+    FOR component IN $(cd ./tools && ls -d */)
+        BUILD --pass-args ./tools/${component}+pre-commit
+    END
     FOR component IN $(cd ./components && ls -d */)
         BUILD --pass-args ./components/${component}+pre-commit
     END
@@ -150,6 +162,9 @@ pre-commit: # Generate the final spec and run all the pre-commit hooks
 
 tidy: # Run tidy on all the components
     LOCALLY
+    FOR component IN $(cd ./tools && ls -d */)
+            BUILD --pass-args ./tools/${component}+tidy
+        END
     FOR component IN $(cd ./components && ls -d */)
         BUILD --pass-args ./components/${component}+tidy
     END

components/ledger/libs/bun/bunmigrate/run.go

@@ -17,12 +17,29 @@ import (
 	"io"
 )
 
-func ensureDatabaseExists(ctx context.Context, connectionOptions bunconnect.ConnectionOptions) error {
+func isDatabaseExists(ctx context.Context, db *bun.DB, name string) (bool, error) {
+	row := db.QueryRowContext(ctx, `SELECT datname FROM pg_database WHERE datname = ?`, name)
+	if row.Err() != nil {
+		return false, errors.Wrap(row.Err(), "checking database list")
+	}
+
+	if err := row.Scan(pointer.For("")); err != nil {
+		if !errors.Is(err, sql.ErrNoRows) {
+			return false, errors.Wrap(err, "scanning database row")
+		}
+
+		return false, nil
+	}
+
+	return true, nil
+}
+
+func onPostgresDB(ctx context.Context, connectionOptions bunconnect.ConnectionOptions, callback func(db *bun.DB) error) error {
 	url, err := dburl.Parse(connectionOptions.DatabaseSourceName)
 	if err != nil {
-		return err
+		return errors.Wrapf(err, "parsing dsn: %s", connectionOptions.DatabaseSourceName)
 	}
-	originalPath := url.Path
+
 	url.Path = "postgres" // notes(gfyrag): default "postgres" database (most of the time?)
 	connectionOptions.DatabaseSourceName = url.String()
 
@@ -37,29 +54,61 @@ func ensureDatabaseExists(ctx context.Context, connectionOptions bunconnect.Conn
 		}
 	}()
 
-	row := db.QueryRowContext(ctx, `SELECT datname FROM pg_database WHERE datname = ?`, originalPath[1:])
-	if row.Err() != nil {
-		return row.Err()
-	}
+	return callback(db)
+}
 
-	if err := row.Scan(pointer.For("")); err != nil {
-		if !errors.Is(err, sql.ErrNoRows) {
-			return err
+func EnsureDatabaseNotExists(ctx context.Context, connectionOptions bunconnect.ConnectionOptions) error {
+	return onPostgresDB(ctx, connectionOptions, func(db *bun.DB) error {
+
+		url, err := dburl.Parse(connectionOptions.DatabaseSourceName)
+		if err != nil {
+			return errors.Wrapf(err, "parsing dsn: %s", connectionOptions.DatabaseSourceName)
 		}
 
-		_, err = db.ExecContext(ctx, fmt.Sprintf(`CREATE DATABASE "%s"`, originalPath[1:]))
+		databaseExists, err := isDatabaseExists(ctx, db, url.Path[1:])
 		if err != nil {
-			return err
+			return errors.Wrap(err, "checking if database exists")
 		}
-	}
 
-	return nil
+		if databaseExists {
+			_, err = db.ExecContext(ctx, fmt.Sprintf(`DROP DATABASE "%s"`, url.Path[1:]))
+			if err != nil {
+				return errors.Wrap(err, "dropping database")
+			}
+		}
+
+		return nil
+	})
+}
+
+func EnsureDatabaseExists(ctx context.Context, connectionOptions bunconnect.ConnectionOptions) error {
+	return onPostgresDB(ctx, connectionOptions, func(db *bun.DB) error {
+
+		url, err := dburl.Parse(connectionOptions.DatabaseSourceName)
+		if err != nil {
+			return errors.Wrapf(err, "parsing dsn: %s", connectionOptions.DatabaseSourceName)
+		}
+
+		databaseExists, err := isDatabaseExists(ctx, db, url.Path[1:])
+		if err != nil {
+			return errors.Wrap(err, "checking if database exists")
+		}
+
+		if !databaseExists {
+			_, err = db.ExecContext(ctx, fmt.Sprintf(`CREATE DATABASE "%s"`, url.Path[1:]))
+			if err != nil {
+				return errors.Wrap(err, "creating database")
+			}
+		}
+
+		return nil
+	})
 }
 
 func run(ctx context.Context, output io.Writer, args []string, connectionOptions *bunconnect.ConnectionOptions,
 	executor func(args []string, db *bun.DB) error) error {
 
-	if err := ensureDatabaseExists(ctx, *connectionOptions); err != nil {
+	if err := EnsureDatabaseExists(ctx, *connectionOptions); err != nil {
 		return err
 	}
 

components/operator/Earthfile

@@ -112,7 +112,7 @@ tests:
     ENV ENVTEST_VERSION 1.28.0
     RUN setup-envtest use $ENVTEST_VERSION -p path
     ENV KUBEBUILDER_ASSETS /root/.local/share/kubebuilder-envtest/k8s/$ENVTEST_VERSION-linux-$(go env GOHOSTARCH)
-    DO --pass-args core+GO_INSTALL --package=github.com/onsi/ginkgo/v2/ginkgo@v2.13.2
+    DO --pass-args core+GO_INSTALL --package=github.com/onsi/ginkgo/v2/ginkgo@v2.14.0
     COPY (+sources/*) /src
     COPY --pass-args (+manifests/config) /src/components/operator/config
     COPY --pass-args (+generate/internal) /src/components/operator/internal

components/operator/internal/core/stacks.go

@@ -1,6 +1,8 @@
 package core
 
 import (
+	"github.com/formancehq/operator/api/formance.com/v1beta1"
+	"k8s.io/apimachinery/pkg/types"
 	"reflect"
 
 	"github.com/pkg/errors"
@@ -89,3 +91,24 @@ func GetIfExists(ctx Context, stackName string, to client.Object) (bool, error)
 	}
 	return true, nil
 }
+
+func GetImageVersionForStack(ctx Context, stack *v1beta1.Stack, name string) (string, error) {
+	if stack.Spec.Version != "" {
+		return stack.Spec.Version, nil
+	}
+	if stack.Spec.VersionsFromFile == "" {
+		return "latest", nil
+	}
+	versions := &v1beta1.Versions{}
+	if err := ctx.GetClient().Get(ctx, types.NamespacedName{
+		Name: stack.Spec.VersionsFromFile,
+	}, versions); err != nil {
+		return "", nil
+	}
+
+	version := versions.Spec[name]
+	if version == "" {
+		return "latest", nil
+	}
+	return version, nil
+}

components/operator/internal/resources/databases/controller.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"github.com/formancehq/operator/api/formance.com/v1beta1"
 	"github.com/formancehq/operator/internal/core"
+	"github.com/formancehq/operator/internal/resources/registries"
 	"github.com/formancehq/operator/internal/resources/secretreferences"
 	"github.com/formancehq/operator/internal/resources/settings"
 	"github.com/formancehq/stack/libs/go-libs/pointer"
@@ -74,7 +75,7 @@ func Reconcile(ctx core.Context, stack *v1beta1.Stack, database *v1beta1.Databas
 		database.Status.URI = databaseURL
 		database.Status.Database = dbName
 
-		if err := createDatabase(ctx, database, secretReference); err != nil {
+		if err := createDatabase(ctx, stack, database, secretReference); err != nil {
 			return err
 		}
 	case database.Status.URI.String() != databaseURL.String():
@@ -84,36 +85,40 @@ func Reconcile(ctx core.Context, stack *v1beta1.Stack, database *v1beta1.Databas
 	return nil
 }
 
-func createDatabase(ctx core.Context, database *v1beta1.Database, secretReference *v1beta1.SecretReference) error {
+func createDatabase(ctx core.Context, stack *v1beta1.Stack, database *v1beta1.Database, secretReference *v1beta1.SecretReference) error {
 	log := log.FromContext(ctx)
 	log = log.WithValues("name", database.Name)
 	annotations := make(map[string]string)
 	if secretReference != nil {
 		annotations["secret-hash"] = secretReference.Status.Hash
 	}
 
+	operatorUtilsImageVersion, err := core.GetImageVersionForStack(ctx, stack, "operator-utils")
+	if err != nil {
+		return err
+	}
+
+	operatorUtilsImage, err := registries.GetImage(ctx, stack, "operator-utils", operatorUtilsImageVersion)
+	if err != nil {
+		return err
+	}
+
 	jobName := fmt.Sprintf("%s-create-database", database.Spec.Service)
 	job, result, err := core.CreateOrUpdate[*batchv1.Job](ctx, types.NamespacedName{
 		Namespace: database.Spec.Stack,
 		Name:      jobName,
 	},
 		func(t *batchv1.Job) error {
-			// PG does not support 'CREATE IF NOT EXISTS ' construct, emulate it with the above query
-			createDBCommand := `echo SELECT \'CREATE DATABASE \"${POSTGRES_DATABASE}\"\' WHERE NOT EXISTS \(SELECT FROM pg_database WHERE datname = \'${POSTGRES_DATABASE}\'\)\\gexec | psql -h ${POSTGRES_HOST} -p ${POSTGRES_PORT} -U ${POSTGRES_USERNAME}`
-			if isDisabledSSLMode(database.Status.URI) {
-				createDBCommand += ` "sslmode=disable"`
-			}
-
 			t.Spec.BackoffLimit = pointer.For(int32(10000))
 			t.Spec.Template.Spec.RestartPolicy = v1.RestartPolicyOnFailure
 			t.Spec.TTLSecondsAfterFinished = pointer.For(int32(30))
 			if len(t.Spec.Template.Spec.Containers) == 0 {
 				t.Spec.Template.Spec.Containers = []v1.Container{{}}
 			}
 			t.Spec.Template.Spec.Containers[0].Name = "create-database"
-			t.Spec.Template.Spec.Containers[0].Image = "postgres:15-alpine"
-			t.Spec.Template.Spec.Containers[0].Args = []string{"sh", "-c", createDBCommand}
-			t.Spec.Template.Spec.Containers[0].Env = psqlEnvVars(database)
+			t.Spec.Template.Spec.Containers[0].Image = operatorUtilsImage
+			t.Spec.Template.Spec.Containers[0].Args = []string{"db", "create"}
+			t.Spec.Template.Spec.Containers[0].Env = GetPostgresEnvVars(database)
 
 			return nil
 		},
@@ -163,26 +168,38 @@ func Delete(ctx core.Context, database *v1beta1.Database) error {
 		annotations["secret-hash"] = secretReference.Status.Hash
 	}
 
+	stack := &v1beta1.Stack{}
+	if err := ctx.GetClient().Get(ctx, types.NamespacedName{
+		Name: database.Spec.Stack,
+	}, stack); err != nil {
+		return err
+	}
+
+	operatorUtilsImageVersion, err := core.GetImageVersionForStack(ctx, stack, "operator-utils")
+	if err != nil {
+		return err
+	}
+
+	operatorUtilsImage, err := registries.GetImage(ctx, stack, "operator-utils", operatorUtilsImageVersion)
+	if err != nil {
+		return err
+	}
+
 	job, _, err := core.CreateOrUpdate[*batchv1.Job](ctx, types.NamespacedName{
 		Namespace: database.Spec.Stack,
 		Name:      fmt.Sprintf("%s-drop-database", database.Spec.Service),
 	},
 		func(t *batchv1.Job) error {
-			dropDBCommand := `psql -h ${POSTGRES_HOST} -p ${POSTGRES_PORT} -U ${POSTGRES_USERNAME} -c "DROP DATABASE \"${POSTGRES_DATABASE}\""`
-			if isDisabledSSLMode(database.Status.URI) {
-				dropDBCommand += ` "sslmode=disable"`
-			}
-
 			t.Spec.BackoffLimit = pointer.For(int32(10000))
 			t.Spec.TTLSecondsAfterFinished = pointer.For(int32(30))
 			t.Spec.Template.Spec.RestartPolicy = v1.RestartPolicyOnFailure
 			if len(t.Spec.Template.Spec.Containers) == 0 {
 				t.Spec.Template.Spec.Containers = []v1.Container{{}}
 			}
 			t.Spec.Template.Spec.Containers[0].Name = "drop-database"
-			t.Spec.Template.Spec.Containers[0].Image = "postgres:15-alpine"
-			t.Spec.Template.Spec.Containers[0].Args = []string{"sh", "-c", dropDBCommand}
-			t.Spec.Template.Spec.Containers[0].Env = psqlEnvVars(database)
+			t.Spec.Template.Spec.Containers[0].Image = operatorUtilsImage
+			t.Spec.Template.Spec.Containers[0].Args = []string{"db", "drop"}
+			t.Spec.Template.Spec.Containers[0].Env = GetPostgresEnvVars(database)
 
 			return nil
 		},
@@ -202,13 +219,6 @@ func Delete(ctx core.Context, database *v1beta1.Database) error {
 	return nil
 }
 
-func psqlEnvVars(database *v1beta1.Database) []v1.EnvVar {
-	return append(GetPostgresEnvVars(database),
-		// psql use PGPASSWORD env var
-		core.Env("PGPASSWORD", "$(POSTGRES_PASSWORD)"),
-	)
-}
-
 func init() {
 	core.Init(
 		core.WithResourceReconciler(Reconcile,

components/operator/internal/resources/registries/image.go

@@ -2,7 +2,6 @@ package registries
 
 import (
 	"fmt"
-
 	"github.com/formancehq/operator/api/formance.com/v1beta1"
 	"github.com/formancehq/operator/internal/core"
 )

ee/agent/Earthfile

@@ -95,7 +95,7 @@ tests:
     ENV ENVTEST_VERSION 1.28.0
     RUN setup-envtest use $ENVTEST_VERSION -p path
     ENV KUBEBUILDER_ASSETS /root/.local/share/kubebuilder-envtest/k8s/$ENVTEST_VERSION-linux-$(go env GOHOSTARCH)
-    DO --pass-args core+GO_INSTALL --package=github.com/onsi/ginkgo/v2/ginkgo@v2.13.2
+    DO --pass-args core+GO_INSTALL --package=github.com/onsi/ginkgo/v2/ginkgo@v2.14.0
     COPY --pass-args +sources/* /src
     COPY --pass-args ../../components/operator+manifests/config /src/components/operator/config
     WORKDIR /src/ee/agent

ee/stargate/go.mod

@@ -15,7 +15,6 @@ require (
 	github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.0.0-rc.5
 	github.com/hashicorp/go-retryablehttp v0.7.2
 	github.com/lestrrat-go/jwx v1.2.26
-	github.com/nats-io/nats-server v1.4.1
 	github.com/nats-io/nats-server/v2 v2.9.23
 	github.com/nats-io/nats.go v1.31.0
 	github.com/pkg/errors v0.9.1
@@ -86,8 +85,6 @@ require (
 	github.com/magiconair/properties v1.8.7 // indirect
 	github.com/minio/highwayhash v1.0.2 // indirect
 	github.com/mitchellh/mapstructure v1.5.0 // indirect
-	github.com/nats-io/gnatsd v1.4.1 // indirect
-	github.com/nats-io/go-nats v1.7.2 // indirect
 	github.com/nats-io/jwt/v2 v2.5.0 // indirect
 	github.com/nats-io/nkeys v0.4.6 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect

ee/stargate/go.sum

@@ -364,14 +364,8 @@ github.com/modern-go/reflect2 v1.0.1/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3Rllmb
 github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk=
 github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
 github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
-github.com/nats-io/gnatsd v1.4.1 h1:RconcfDeWpKCD6QIIwiVFcvForlXpWeJP7i5/lDLy44=
-github.com/nats-io/gnatsd v1.4.1/go.mod h1:nqco77VO78hLCJpIcVfygDP2rPGfsEHkGTUk94uh5DQ=
-github.com/nats-io/go-nats v1.7.2 h1:cJujlwCYR8iMz5ofZSD/p2WLW8FabhkQ2lIEVbSvNSA=
-github.com/nats-io/go-nats v1.7.2/go.mod h1:+t7RHT5ApZebkrQdnn6AhQJmhJJiKAvJUio1PiiCtj0=
 github.com/nats-io/jwt/v2 v2.5.0 h1:WQQ40AAlqqfx+f6ku+i0pOVm+ASirD4fUh+oQsiE9Ak=
 github.com/nats-io/jwt/v2 v2.5.0/go.mod h1:24BeQtRwxRV8ruvC4CojXlx/WQ/VjuwlYiH+vu/+ibI=
-github.com/nats-io/nats-server v1.4.1 h1:Ul1oSOGNV/L8kjr4v6l2f9Yet6WY+LevH1/7cRZ/qyA=
-github.com/nats-io/nats-server v1.4.1/go.mod h1:c8f/fHd2B6Hgms3LtCaI7y6pC4WD1f4SUxcCud5vhBc=
 github.com/nats-io/nats-server/v2 v2.9.23 h1:6Wj6H6QpP9FMlpCyWUaNu2yeZ/qGj+mdRkZ1wbikExU=
 github.com/nats-io/nats-server/v2 v2.9.23/go.mod h1:wEjrEy9vnqIGE4Pqz4/c75v9Pmaq7My2IgFmnykc4C0=
 github.com/nats-io/nats.go v1.31.0 h1:/WFBHEc/dOKBF6qf1TZhrdEfTmOZ5JzdJ+Y3m6Y/p7E=