fix: Moving custom migrations from cd to export service (#2196)

This moves the custom db migrations from the cd-service to the
manifest-repo-export-service.
The pure sql migrations are now run by both the cd-service and the
manifest-repo-export-service.

Many functions just had to be moved to another service, hence the big
git diff.

Now, the export-service takes care of the migrations during startup.
The cd-service does not run the migrations anymore during startup.
Instead, it calls (and waits for) the export-service, which then handles
the migrations.

This also adds a new helm parameter `cd.grpcMaxRecvMsgSize`.

Ref: SRX-V6RVYF

charts/kuberpult/templates/cd-service.yaml

@@ -147,6 +147,8 @@ spec:
           value: {{ required ".Values.git.url is required" .Values.git.url | quote }}
         - name: KUBERPULT_GIT_BRANCH
           value: {{ .Values.git.branch | quote }}
+        - name: KUBERPULT_VERSION
+          value: {{ $.Chart.AppVersion | quote}}
         - name: LOG_FORMAT
           value: {{ .Values.log.format | quote }}
         - name: LOG_LEVEL
@@ -280,6 +282,12 @@ spec:
         - name: KUBERPULT_MINOR_REGEXES
           value: {{ .Values.cd.minorRegexes | join "," | quote }}
 {{- end }}
+        - name: KUBERPULT_MIGRATION_SERVER
+          value: kuberpult-manifest-repo-export-service:8443
+        - name: KUBERPULT_MIGRATION_SERVER_SECURE
+          value: "false"
+        - name: KUBERPULT_GRPC_MAX_RECV_MSG_SIZE
+          value: "{{ .Values.cd.grpcMaxRecvMsgSize }}"
         volumeMounts:
 {{- if (eq .Values.db.dbOption "postgreSQL") }}
         - name: migrations

charts/kuberpult/templates/manifest-repo-export-service.yaml

@@ -185,6 +185,8 @@ spec:
 {{- end }}
         - name: KUBERPULT_DB_LOCATION
           value: "{{ .Values.db.location }}"
+        - name: KUBERPULT_DB_MIGRATIONS_LOCATION
+          value: {{ .Values.db.migrations }}
         - name: KUBERPULT_DB_NAME
           value: "{{ .Values.db.dbName }}"
         - name: KUBERPULT_DB_USER_NAME
@@ -207,8 +209,6 @@ spec:
           value: "{{ .Values.db.connections.manifestRepoExport.maxOpen }}"
         - name: KUBERPULT_DB_MAX_IDLE_CONNECTIONS
           value: "{{ .Values.db.connections.manifestRepoExport.maxIdle }}"
-        - name: KUBERPULT_ARGO_CD_GENERATE_FILES
-          value: {{ .Values.argocd.generateFiles | quote }}
 {{- if .Values.datadogTracing.enabled }}
         - name: DD_TRACE_DEBUG
           value: "{{ .Values.datadogTracing.debugging }}"
@@ -242,6 +242,9 @@ spec:
           # We mount the volume to the parent because kubernetes volumes belong to root.
           # This way the container can create /kp/repository itself without any permission issues.
           mountPath: /kp/
+        - name: migrations
+          mountPath: {{ .Values.db.migrations }}
+          readOnly: false
         - name: ssh
           mountPath: /etc/ssh
 {{- if .Values.datadogProfiling.enabled }}
@@ -273,6 +276,9 @@ spec:
         hostPath:
           path: {{ .Values.dogstatsdMetrics.hostSocketPath }}
 {{- end }}
+      - name: migrations
+        configMap:
+          name: kuberpult-migrations-cloudsql
 ---
 apiVersion: v1
 kind: Service

charts/kuberpult/tests/charts_test.go

@@ -181,6 +181,18 @@ ingress:
 					Name:  "KUBERPULT_DB_OPTION",
 					Value: "postgreSQL",
 				},
+				{
+					Name:  "KUBERPULT_MIGRATION_SERVER",
+					Value: "kuberpult-manifest-repo-export-service:8443",
+				},
+				{
+					Name:  "KUBERPULT_MIGRATION_SERVER_SECURE",
+					Value: "false",
+				},
+				{
+					Name:  "KUBERPULT_GRPC_MAX_RECV_MSG_SIZE",
+					Value: "4",
+				},
 			},
 			ExpectedMissing: []core.EnvVar{},
 		},

charts/kuberpult/values.yaml

@@ -105,6 +105,8 @@ cd:
 # With the queue, the cd-service processes only one request at a time, which is very much required when using git.
 # With the database, this is not required anymore.
     disableQueue: false
+  # The maximum message size in megabytes the client can receive.
+  grpcMaxRecvMsgSize: 4
   resources:
     limits:
       cpu: 2
@@ -212,7 +214,7 @@ frontend:
   batchClient:
     # This value needs to be higher than the network timeout for git
     timeout: 2m
-  # The maximum message size in mega bytes the client can receive.
+  # The maximum message size in megabytes the client can receive.
   grpcMaxRecvMsgSize: 4
 rollout:
   enabled: false
@@ -228,7 +230,7 @@ rollout:
     requests:
       cpu: 500m
       memory: 250Mi
-  # The maximum message size in mega bytes the client can receive.
+  # The maximum message size in megabytes the client can receive.
   grpcMaxRecvMsgSize: 4
 
 ingress:

database/migrations/sqlite/1736855898709581_migrationcutoff.up.sql

@@ -0,0 +1,5 @@
+CREATE TABLE IF NOT EXISTS custom_migration_cutoff
+(
+    migration_done_at TIMESTAMP NOT NULL,
+    kuberpult_version varchar(100) PRIMARY KEY -- the version as it appears on GitHub, e.g. "1.2.3"
+);

docker-compose.yml

@@ -40,6 +40,10 @@ services:
       - KUBERPULT_DB_MAX_OPEN_CONNECTIONS=10
       - KUBERPULT_DB_MAX_IDLE_CONNECTIONS=5
       - KUBERPULT_ALLOWED_DOMAINS=localhost
+      - KUBERPULT_MIGRATION_SERVER=manifest-repo-export-service:8443
+      - KUBERPULT_GRPC_MAX_RECV_MSG_SIZE=4
+      - KUBERPULT_MIGRATION_SERVER_SECURE=false
+      - KUBERPULT_VERSION=v0.1.2
     ports:
       - "8080:8080"
       - "8443:8443"
@@ -48,6 +52,8 @@ services:
       - ./database:/kp/database
     stop_grace_period: 0.5s
     depends_on:
+      manifest-repo-export-service:
+        condition: service_healthy
       postgres:
         condition: service_healthy
   manifest-repo-export-service:
@@ -61,6 +67,7 @@ services:
       - KUBERPULT_DB_NAME=kuberpult
       - KUBERPULT_DB_USER_NAME=postgres
       - KUBERPULT_DB_USER_PASSWORD=mypassword
+      - KUBERPULT_DB_MIGRATIONS_LOCATION=/kp/database/migrations/postgres/
       - KUBERPULT_DB_AUTH_PROXY_PORT=5432
       - KUBERPULT_GIT_URL=/kp/kuberpult/repository_remote
       - KUBERPULT_GIT_BRANCH=master
@@ -83,9 +90,12 @@ services:
       - "8090:8080"
       - "8444:8443"
     stop_grace_period: 0.5s
+    healthcheck:
+      test: ["CMD-SHELL", "wget --no-verbose --tries=1 --spider localhost:8080/healthz"]
+      interval: 1s
+      timeout: 5s
+      retries: 5
     depends_on:
-      cd-service:
-        condition: service_started
       postgres:
         condition: service_healthy
   frontend-service:

pkg/db/db.go

@@ -1216,43 +1216,6 @@ func (h *DBHandler) needsCommitEventsMigrations(ctx context.Context, transaction
 	return true, nil
 }
 
-// NeedsMigrations checks if we need migrations for any table.
-func (h *DBHandler) NeedsMigrations(ctx context.Context) (bool, error) {
-	span, ctx := tracer.StartSpanFromContext(ctx, "NeedsMigrations")
-	defer span.Finish()
-	var needsMigration bool = false
-	txError := h.WithTransaction(ctx, true, func(ctx context.Context, transaction *sql.Tx) error {
-		var checkFunctions = []CheckFun{
-			(*DBHandler).NeedsEventSourcingLightMigrations,
-			(*DBHandler).needsAppsMigrations,
-			(*DBHandler).needsDeploymentsMigrations,
-			(*DBHandler).needsReleasesMigrations,
-			(*DBHandler).needsEnvLocksMigrations,
-			(*DBHandler).needsAppLocksMigrations,
-			(*DBHandler).needsTeamLocksMigrations,
-			(*DBHandler).needsCommitEventsMigrations,
-			(*DBHandler).needsEnvironmentsMigrations,
-		}
-		for i := range checkFunctions {
-			f := checkFunctions[i]
-			needs, err := f(h, ctx, transaction)
-			if err != nil {
-				return err
-			}
-			if !needs {
-				logger.FromContext(ctx).Sugar().Warnf("migration skipped: %v", i)
-			}
-			if needs {
-				logger.FromContext(ctx).Sugar().Warnf("migration required: %v", i)
-				needsMigration = true
-				return nil
-			}
-		}
-		return nil
-	})
-	return needsMigration, txError
-}
-
 // For commit_events migrations, we need some transformer to be on the database before we run their migrations.
 func (h *DBHandler) RunCustomMigrationsEventSourcingLight(ctx context.Context) error {
 	return h.WithTransaction(ctx, false, func(ctx context.Context, transaction *sql.Tx) error {
@@ -3680,7 +3643,7 @@ func (h *DBHandler) needsEnvironmentsMigrations(ctx context.Context, transaction
 		return true, err
 	}
 	if arbitraryAllEnvsRow != nil {
-		log.Infof("custom migration for environments already ran because row %v was found, skipping custom migration", arbitraryAllEnvsRow)
+		log.Infof("custom migration for environments already ran because row (%v) was found, skipping custom migration", arbitraryAllEnvsRow)
 		return false, nil
 	}
 	return true, nil

services/manifest-repo-export-service/pkg/migrations/types.go → pkg/migrations/types.go

@@ -31,10 +31,13 @@ func ParseKuberpultVersion(version string) (*api.KuberpultVersion, error) {
 		// See the top-level Makefile and how the version is supplied to earthly in the line
 		// "earthly -P +integration-test --kuberpult_version=$(IMAGE_TAG_KUBERPULT)"
 		version = splitDash[1]
+	} else if len(splitDash) == 3 {
+		// this is what happens when we us run-kind.sh
+		version = splitDash[0]
 	} else if len(splitDash) == 1 {
 		// this is the normal case, we don't have to remove parts
 	} else {
-		return nil, fmt.Errorf("invalid version, expected 0 or 3 dashes")
+		return nil, fmt.Errorf("invalid version, expected 0, 2, or 3 dashes, but got %s", version)
 	}
 
 	version = strings.TrimPrefix(version, "v")
@@ -74,5 +77,20 @@ func CreateKuberpultVersion(major, minor, patch int) *api.KuberpultVersion {
 }
 
 func FormatKuberpultVersion(version *api.KuberpultVersion) string {
+	if version == nil {
+		return ""
+	}
 	return fmt.Sprintf("%d.%d.%d", version.Major, version.Minor, version.Patch)
 }
+
+func IsKuberpultVersionEqual(a, b *api.KuberpultVersion) bool {
+	if a == nil && b == nil {
+		return true
+	}
+	if a == nil || b == nil {
+		return false
+	}
+	return a.Major == b.Major &&
+		a.Minor == b.Minor &&
+		a.Patch == b.Patch
+}

services/manifest-repo-export-service/pkg/migrations/types_test.go → pkg/migrations/types_test.go

@@ -59,11 +59,23 @@ func TestParseKuberpultVersion(t *testing.T) {
 			expectedVersion:       CreateKuberpultVersion(12, 1, 2),
 			expectedError:         nil,
 		},
+		{
+			name:                  "should read first part for kind version",
+			kuberpultVersionInput: "v11.10.0-7-g1a2fd8d0",
+			expectedVersion:       CreateKuberpultVersion(11, 10, 0),
+			expectedError:         nil,
+		},
+		{
+			name:                  "just another kind version",
+			kuberpultVersionInput: "v1.13.2-8-g0a1fd1d1",
+			expectedVersion:       CreateKuberpultVersion(1, 13, 2),
+			expectedError:         nil,
+		},
 		{
 			name:                  "invalid number of dashes",
 			kuberpultVersionInput: "main-main-v12.1.2-7-g08f811e8",
 			expectedVersion:       nil,
-			expectedError:         errMatcher{msg: "invalid version, expected 0 or 3 dashes"},
+			expectedError:         errMatcher{msg: "invalid version, expected 0, 2, or 3 dashes, but got main-main-v12.1.2-7-g08f811e8"},
 		},
 		{
 			name:                  "0 dashes also works",

services/cd-service/pkg/cmd/server.go

@@ -18,8 +18,13 @@ package cmd
 
 import (
 	"context"
+	"crypto/tls"
+	"crypto/x509"
 	"database/sql"
 	"fmt"
+	"github.com/freiheit-com/kuberpult/pkg/migrations"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/insecure"
 	"net/http"
 	"os"
 	"regexp"
@@ -56,6 +61,8 @@ const (
 	datadogNameCd           = "kuberpult-cd-service"
 	minReleaseVersionsLimit = 5
 	maxReleaseVersionsLimit = 30
+
+	megaBytes int = 1024 * 1024
 )
 
 type Config struct {
@@ -111,6 +118,13 @@ type Config struct {
 	CacheTtlHours        uint     `default:"24" split_words:"true"`
 
 	DisableQueue bool `required:"true" split_words:"true"`
+
+	// the cd-service calls the manifest-export on startup, to run custom migrations:
+	MigrationServer       string `required:"true" split_words:"true"`
+	MigrationServerSecure bool   `required:"true" split_words:"true"`
+	GrpcMaxRecvMsgSize    int    `required:"true" split_words:"true"`
+
+	Version string `required:"true" split_words:"true"`
 }
 
 func (c *Config) storageBackend() repository.StorageBackend {
@@ -354,45 +368,54 @@ func RunServer() {
 			}
 			//Check for migrations -> for pulling
 			logger.FromContext(ctx).Sugar().Warnf("checking if migrations are required...")
-			if needsMigration, err := dbHandler.NeedsMigrations(ctx); err == nil && needsMigration {
-				logger.FromContext(ctx).Sugar().Warnf("starting to pull the repo")
-				err := repo.Pull(ctx)
+
+			var migrationClient api.MigrationServiceClient = nil
+			if c.MigrationServer == "" {
+				logger.FromContext(ctx).Fatal("MigrationServer required")
+			}
+			var cred credentials.TransportCredentials = insecure.NewCredentials()
+			if c.MigrationServerSecure {
+				systemRoots, err := x509.SystemCertPool()
 				if err != nil {
-					logger.FromContext(ctx).Fatal("Could not pull repository to perform custom migrations", zap.Error(err))
-				}
-				logger.FromContext(ctx).Sugar().Warnf("running custom migrations, because KUBERPULT_DB_WRITE_ESL_TABLE_ONLY=false")
-
-				migErr := dbHandler.RunCustomMigrations(
-					ctx,
-					repo.State().GetAppsAndTeams,
-					repo.State().WriteCurrentlyDeployed,
-					repo.State().WriteAllReleases,
-					repo.State().WriteCurrentEnvironmentLocks,
-					repo.State().WriteCurrentApplicationLocks,
-					repo.State().WriteCurrentTeamLocks,
-					repo.State().GetAllEnvironments,
-					repo.State().WriteAllQueuedAppVersions,
-					repo.State().WriteAllCommitEvents,
-				)
-				if migErr != nil {
-					logger.FromContext(ctx).Fatal("Error running custom database migrations", zap.Error(migErr))
-				} else {
-					logger.FromContext(ctx).Sugar().Warnf("finished running custom migrations")
+					msg := "failed to read CA certificates"
+					return fmt.Errorf(msg)
 				}
-			} else if err != nil {
-				logger.FromContext(ctx).Fatal("Error running custom database migrations", zap.Error(err))
+				//exhaustruct:ignore
+				cred = credentials.NewTLS(&tls.Config{
+					RootCAs: systemRoots,
+				})
 			}
-			logger.FromContext(ctx).Sugar().Warnf("Skipping git-related custom migrations, because all tables contain data.")
-			err = dbHandler.RunCustomMigrationReleaseEnvironments(ctx)
+			grpcClientOpts := []grpc.DialOption{
+				grpc.WithTransportCredentials(cred),
+				grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(c.GrpcMaxRecvMsgSize * megaBytes)),
+			}
+
+			rolloutCon, err := grpc.Dial(c.MigrationServer, grpcClientOpts...)
 			if err != nil {
-				return err
+				logger.FromContext(ctx).Fatal("grpc.dial.error", zap.Error(err), zap.String("addr", c.MigrationServer))
 			}
-			logger.FromContext(ctx).Sugar().Warnf("Applied custom migration for release environments")
-			err = dbHandler.RunCustomMigrationEnvironmentApplications(ctx)
+			migrationClient = api.NewMigrationServiceClient(rolloutCon)
+
+			kuberpultVersion, err := migrations.ParseKuberpultVersion(c.Version)
 			if err != nil {
-				return err
+				logger.FromContext(ctx).Fatal("env.parse.error", zap.Error(err), zap.String("version", c.Version))
+			}
+
+			response, migErr := migrationClient.EnsureCustomMigrationApplied(ctx, &api.EnsureCustomMigrationAppliedRequest{
+				Version: kuberpultVersion,
+			})
+
+			if migErr != nil {
+				logger.FromContext(ctx).Fatal("Error ensuring custom migrations are applied", zap.Error(migErr))
+			}
+			if response == nil {
+				logger.FromContext(ctx).Sugar().Fatal("Custom database migrations returned nil response")
 			}
-			logger.FromContext(ctx).Sugar().Warnf("Applied custom migrations for environment applications")
+			if !response.MigrationsApplied {
+				logger.FromContext(ctx).Sugar().Fatalf("Custom database migrations where not applied: %v", response)
+			}
+
+			logger.FromContext(ctx).Sugar().Warnf("finished running custom migrations")
 		} else {
 			logger.FromContext(ctx).Sugar().Warnf("Skipping custom migrations, because KUBERPULT_DB_WRITE_ESL_TABLE_ONLY=false")
 		}