fix(operator): replicas configuration (#1264)

formancehq · Feb 20, 2024 · ff20118 · ff20118
1 parent 0f3a7bc
commit ff20118
Show file tree

Hide file tree

Showing 12 changed files with 46 additions and 30 deletions.
diff --git a/components/operator/internal/resources/auths/deployment.go b/components/operator/internal/resources/auths/deployment.go
@@ -66,8 +66,9 @@ func createDeployment(ctx Context, stack *v1beta1.Stack, auth *v1beta1.Auth, dat
 		env = append(env, Env("CAOS_OIDC_DEV", "1"))
 	}
 
-	return deployments.CreateOrUpdate(ctx, stack, auth, "auth",
+	return deployments.CreateOrUpdate(ctx, auth, "auth",
 		deployments.WithMatchingLabels("auth"),
+		deployments.WithReplicasFromSettings(ctx, stack),
 		deployments.WithServiceAccountName(database.Status.URI.Query().Get("awsRole")),
 		func(t *appsv1.Deployment) error {
 			t.Spec.Template.Annotations = MergeMaps(t.Spec.Template.Annotations, map[string]string{

diff --git a/components/operator/internal/resources/benthos/controller.go b/components/operator/internal/resources/benthos/controller.go
@@ -281,7 +281,7 @@ func createDeployment(ctx Context, stack *v1beta1.Stack, b *v1beta1.Benthos) err
 		return streams[i].Name < streams[j].Name
 	})
 
-	_, err = deployments.CreateOrUpdate(ctx, stack, b, "benthos",
+	_, err = deployments.CreateOrUpdate(ctx, b, "benthos",
 		resourcereferences.Annotate[*appsv1.Deployment]("elasticsearch-secret-hash", resourceReference),
 		deployments.WithMatchingLabels("benthos"),
 		deployments.WithInitContainers(b.Spec.InitContainers...),

diff --git a/components/operator/internal/resources/deployments/deployments.go b/components/operator/internal/resources/deployments/deployments.go
@@ -45,6 +45,20 @@ func WithReplicas(replicas int32) func(t *appsv1.Deployment) error {
 	}
 }
 
+func WithReplicasFromSettings(ctx core.Context, stack *v1beta1.Stack) core.ObjectMutator[*appsv1.Deployment] {
+	return func(deployment *appsv1.Deployment) error {
+		replicas, err := settings.GetInt32(ctx, stack.Name, "deployments", deployment.Name, "replicas")
+		if err != nil {
+			return err
+		}
+		if replicas != nil {
+			deployment.Spec.Replicas = replicas
+		}
+
+		return nil
+	}
+}
+
 func WithContainers(containers ...corev1.Container) func(r *appsv1.Deployment) error {
 	return func(r *appsv1.Deployment) error {
 		r.Spec.Template.Spec.Containers = containers
@@ -69,7 +83,7 @@ func WithVolumes(volumes ...corev1.Volume) func(t *appsv1.Deployment) error {
 	}
 }
 
-func CreateOrUpdate(ctx core.Context, stack *v1beta1.Stack, owner interface {
+func CreateOrUpdate(ctx core.Context, owner interface {
 	client.Object
 	GetStack() string
 	SetCondition(condition v1beta1.Condition)
@@ -129,15 +143,6 @@ func CreateOrUpdate(ctx core.Context, stack *v1beta1.Stack, owner interface {
 			},
 		}
 
-		if deployment.Spec.Replicas == nil {
-			replicas, err := settings.GetInt32(ctx, stack.Name, "deployment", name, "replicas")
-			if err != nil {
-				return err
-			}
-
-			deployment.Spec.Replicas = replicas
-		}
-
 		for ind, container := range deployment.Spec.Template.Spec.InitContainers {
 			resourceRequirements, err := settings.GetResourceRequirements(ctx, owner.GetStack(),
 				"deployments", deployment.Name, "init-containers", container.Name, "resource-requirements")

diff --git a/components/operator/internal/resources/gateways/deployment.go b/components/operator/internal/resources/gateways/deployment.go
@@ -30,7 +30,8 @@ func createDeployment(ctx core.Context, stack *v1beta1.Stack,
 		return err
 	}
 
-	_, err = deployments.CreateOrUpdate(ctx, stack, gateway, "gateway",
+	_, err = deployments.CreateOrUpdate(ctx, gateway, "gateway",
+		deployments.WithReplicasFromSettings(ctx, stack),
 		settings.ConfigureCaddy(caddyfileConfigMap, image, env),
 		deployments.WithMatchingLabels("gateway"),
 	)

diff --git a/components/operator/internal/resources/ledgers/deployments.go b/components/operator/internal/resources/ledgers/deployments.go
@@ -67,7 +67,7 @@ func installLedgerSingleInstance(ctx core.Context, stack *v1beta1.Stack,
 		}
 	}
 
-	if err := createDeployment(ctx, stack, ledger, database, "ledger", *container, v2, deployments.WithReplicas(1)); err != nil {
+	if err := createDeployment(ctx, ledger, database, "ledger", *container, v2, deployments.WithReplicas(1)); err != nil {
 		return err
 	}
 
@@ -95,7 +95,7 @@ func installLedgerMonoWriterMultipleReader(ctx core.Context, stack *v1beta1.Stac
 			return err
 		}
 
-		if err := createDeployment(ctx, stack, ledger, database, name, container, v2, mutators...); err != nil {
+		if err := createDeployment(ctx, ledger, database, name, container, v2, mutators...); err != nil {
 			return err
 		}
 
@@ -117,7 +117,7 @@ func installLedgerMonoWriterMultipleReader(ctx core.Context, stack *v1beta1.Stac
 	}
 
 	container = createLedgerContainerReadOnly(v2)
-	if err := createDeployment("ledger-read", *container); err != nil {
+	if err := createDeployment("ledger-read", *container, deployments.WithReplicasFromSettings(ctx, stack)); err != nil {
 		return err
 	}
 
@@ -156,7 +156,7 @@ func uninstallLedgerMonoWriterMultipleReader(ctx core.Context, stack *v1beta1.St
 	return nil
 }
 
-func createDeployment(ctx core.Context, stack *v1beta1.Stack, ledger *v1beta1.Ledger, database *v1beta1.Database,
+func createDeployment(ctx core.Context, ledger *v1beta1.Ledger, database *v1beta1.Database,
 	name string, container corev1.Container, v2 bool, mutators ...core.ObjectMutator[*v1.Deployment]) error {
 	mutators = append([]core.ObjectMutator[*v1.Deployment]{
 		deployments.WithContainers(container),
@@ -175,7 +175,7 @@ func createDeployment(ctx core.Context, stack *v1beta1.Stack, ledger *v1beta1.Le
 		},
 	}, mutators...)
 
-	_, err := deployments.CreateOrUpdate(ctx, stack, ledger, name, mutators...)
+	_, err := deployments.CreateOrUpdate(ctx, ledger, name, mutators...)
 	return err
 }
 
@@ -292,8 +292,9 @@ func createGatewayDeployment(ctx core.Context, stack *v1beta1.Stack, ledger *v1b
 		return err
 	}
 
-	_, err = deployments.CreateOrUpdate(ctx, stack, ledger, "ledger-gateway",
+	_, err = deployments.CreateOrUpdate(ctx, ledger, "ledger-gateway",
 		settings.ConfigureCaddy(caddyfileConfigMap, caddyImage, env),
+		deployments.WithReplicasFromSettings(ctx, stack),
 		deployments.WithMatchingLabels("ledger"),
 	)
 	return err

diff --git a/components/operator/internal/resources/orchestrations/deployments.go b/components/operator/internal/resources/orchestrations/deployments.go
@@ -131,9 +131,10 @@ func createDeployment(ctx Context, stack *v1beta1.Stack, orchestration *v1beta1.
 	}
 	env = append(env, brokerEnvVars...)
 
-	_, err = deployments.CreateOrUpdate(ctx, stack, orchestration, "orchestration",
+	_, err = deployments.CreateOrUpdate(ctx, orchestration, "orchestration",
 		resourcereferences.Annotate[*appsv1.Deployment]("temporal-secret-hash", resourceReference),
 		deployments.WithServiceAccountName(database.Status.URI.Query().Get("awsRole")),
+		deployments.WithReplicasFromSettings(ctx, stack),
 		deployments.WithMatchingLabels("orchestration"),
 		deployments.WithContainers(v1.Container{
 			Name:          "api",

diff --git a/components/operator/internal/resources/payments/deployments.go b/components/operator/internal/resources/payments/deployments.go
@@ -78,7 +78,7 @@ func createFullDeployment(ctx core.Context, stack *v1beta1.Stack,
 		env = append(env, core.Env("PUBLISHER_TOPIC_MAPPING", "*:"+core.GetObjectName(stack.Name, "payments")))
 	}
 
-	_, err = deployments.CreateOrUpdate(ctx, stack, payments, "payments",
+	_, err = deployments.CreateOrUpdate(ctx, payments, "payments",
 		deployments.WithMatchingLabels("payments"),
 		deployments.WithServiceAccountName(database.Status.URI.Query().Get("awsRole")),
 		deployments.WithContainers(v1.Container{
@@ -112,8 +112,9 @@ func createReadDeployment(ctx core.Context, stack *v1beta1.Stack, payments *v1be
 	}
 	env = append(env, authEnvVars...)
 
-	_, err = deployments.CreateOrUpdate(ctx, stack, payments, "payments-read",
+	_, err = deployments.CreateOrUpdate(ctx, payments, "payments-read",
 		deployments.WithMatchingLabels("payments-read"),
+		deployments.WithReplicasFromSettings(ctx, stack),
 		deployments.WithServiceAccountName(database.Status.URI.Query().Get("awsRole")),
 		deployments.WithContainers(v1.Container{
 			Name:          "api",
@@ -160,7 +161,7 @@ func createConnectorsDeployment(ctx core.Context, stack *v1beta1.Stack, payments
 		env = append(env, core.Env("PUBLISHER_TOPIC_MAPPING", "*:"+core.GetObjectName(stack.Name, "payments")))
 	}
 
-	_, err = deployments.CreateOrUpdate(ctx, stack, payments, "payments-connectors",
+	_, err = deployments.CreateOrUpdate(ctx, payments, "payments-connectors",
 		deployments.WithMatchingLabels("payments-connectors"),
 		deployments.WithServiceAccountName(database.Status.URI.Query().Get("awsRole")),
 		deployments.WithContainers(v1.Container{
@@ -209,7 +210,8 @@ func createGateway(ctx core.Context, stack *v1beta1.Stack, p *v1beta1.Payments)
 		return err
 	}
 
-	_, err = deployments.CreateOrUpdate(ctx, stack, p, "payments",
+	_, err = deployments.CreateOrUpdate(ctx, p, "payments",
+		deployments.WithReplicasFromSettings(ctx, stack),
 		settings.ConfigureCaddy(caddyfileConfigMap, caddyImage, env),
 		deployments.WithMatchingLabels("payments"),
 		// notes(gfyrag): reset init containers in case of upgrading from v1 to v2

diff --git a/components/operator/internal/resources/reconciliations/deployments.go b/components/operator/internal/resources/reconciliations/deployments.go
@@ -38,7 +38,8 @@ func createDeployment(ctx core.Context, stack *v1beta1.Stack, reconciliation *v1
 	}
 	env = append(env, authEnvVars...)
 
-	_, err = deployments.CreateOrUpdate(ctx, stack, reconciliation, "reconciliation",
+	_, err = deployments.CreateOrUpdate(ctx, reconciliation, "reconciliation",
+		deployments.WithReplicasFromSettings(ctx, stack),
 		func(t *appsv1.Deployment) error {
 			t.Spec.Template.Spec.Containers = []v1.Container{{
 				Name:          "reconciliation",

diff --git a/components/operator/internal/resources/searches/controller.go b/components/operator/internal/resources/searches/controller.go
@@ -142,7 +142,8 @@ func Reconcile(ctx Context, stack *v1beta1.Stack, search *v1beta1.Search, versio
 		return err
 	}
 
-	_, err = deployments.CreateOrUpdate(ctx, stack, search, "search",
+	_, err = deployments.CreateOrUpdate(ctx, search, "search",
+		deployments.WithReplicasFromSettings(ctx, stack),
 		resourcereferences.Annotate[*appsv1.Deployment]("elasticsearch-secret-hash", resourceReference),
 		deployments.WithMatchingLabels("search"),
 		deployments.WithContainers(corev1.Container{

diff --git a/components/operator/internal/resources/stargates/deployment.go b/components/operator/internal/resources/stargates/deployment.go
@@ -40,7 +40,8 @@ func createDeployment(ctx core.Context, stack *v1beta1.Stack, stargate *v1beta1.
 		return err
 	}
 
-	_, err = deployments.CreateOrUpdate(ctx, stack, stargate, "stargate",
+	_, err = deployments.CreateOrUpdate(ctx, stargate, "stargate",
+		deployments.WithReplicasFromSettings(ctx, stack),
 		deployments.WithContainers(v1.Container{
 			Name:          "stargate",
 			Env:           env,

diff --git a/components/operator/internal/resources/wallets/deployment.go b/components/operator/internal/resources/wallets/deployment.go
@@ -42,7 +42,8 @@ func createDeployment(ctx core.Context, stack *v1beta1.Stack, wallets *v1beta1.W
 		return err
 	}
 
-	_, err = deployments.CreateOrUpdate(ctx, stack, wallets, "wallets",
+	_, err = deployments.CreateOrUpdate(ctx, wallets, "wallets",
+		deployments.WithReplicasFromSettings(ctx, stack),
 		deployments.WithContainers(v1.Container{
 			Name:          "wallets",
 			Args:          []string{"serve"},

diff --git a/components/operator/internal/resources/webhooks/deployment.go b/components/operator/internal/resources/webhooks/deployment.go
@@ -80,7 +80,8 @@ func createAPIDeployment(ctx core.Context, stack *v1beta1.Stack, webhooks *v1bet
 		}), " ")))
 	}
 
-	_, err = deployments.CreateOrUpdate(ctx, stack, webhooks, "webhooks",
+	_, err = deployments.CreateOrUpdate(ctx, webhooks, "webhooks",
+		deployments.WithReplicasFromSettings(ctx, stack),
 		deployments.WithMatchingLabels("webhooks"),
 		deployments.WithServiceAccountName(database.Status.URI.Query().Get("awsRole")),
 		deployments.WithContainers(v1.Container{
@@ -112,7 +113,7 @@ func createWorkerDeployment(ctx core.Context, stack *v1beta1.Stack, webhooks *v1
 		return fmt.Sprintf("%s-%s", stack.Name, from.Spec.Service)
 	}), " ")))
 
-	_, err = deployments.CreateOrUpdate(ctx, stack, webhooks, "webhooks-worker",
+	_, err = deployments.CreateOrUpdate(ctx, webhooks, "webhooks-worker",
 		deployments.WithMatchingLabels("webhooks-worker"),
 		deployments.WithServiceAccountName(database.Status.URI.Query().Get("awsRole")),
 		deployments.WithContainers(v1.Container{