Make db waiter to wait latest migration

Author: mustard-mh

components/gitpod-db/package.json

@@ -3,7 +3,7 @@
   "license": "AGPL-3.0",
   "version": "0.1.5",
   "scripts": {
-    "build": "yarn lint && npx tsc",
+    "build": "yarn lint && npx tsc && yarn generate-latest-migration",
     "build:clean": "yarn clean && yarn build",
     "lint": "yarn eslint src/*.ts src/**/*.ts",
     "lint:fix": "yarn eslint src/*.ts src/**/*.ts --fix",
@@ -18,7 +18,8 @@
     "migrate-migrations": "node ./lib/migrate-migrations.js",
     "clean": "yarn run rimraf lib",
     "clean:node": "yarn run rimraf node_modules",
-    "purge": "yarn clean && yarn clean:node && yarn run rimraf yarn.lock"
+    "purge": "yarn clean && yarn clean:node && yarn run rimraf yarn.lock",
+    "generate-latest-migration": "node lib/typeorm/generate-latest-migration.js"
   },
   "files": [
     "/lib"

components/gitpod-db/src/typeorm/generate-latest-migration.ts

@@ -0,0 +1,22 @@
+/**
+ * Copyright (c) 2023 Gitpod GmbH. All rights reserved.
+ * Licensed under the GNU Affero General Public License (AGPL).
+ * See License.AGPL.txt in the project root for license information.
+ */
+
+import { MigrationExecutor } from "typeorm";
+import { writeFile } from "node:fs/promises";
+import { Config } from "../config";
+import { TypeORM } from "./typeorm";
+import { join } from "node:path";
+
+async function generateLatestMigrationName() {
+    const config = new Config();
+    const typeorm = new TypeORM(config, {});
+    const conn = await typeorm.getConnection();
+    const t = await new MigrationExecutor(conn).getAllMigrations();
+    await writeFile(join(__dirname, "latest-migration.txt"), t.sort((m) => m.timestamp).pop()?.name ?? "");
+    process.exit(0);
+}
+
+generateLatestMigrationName();

components/service-waiter/BUILD.yaml

@@ -7,6 +7,10 @@ packages:
       - "go.sum"
     deps:
       - components/common-go:lib
+      - components/gitpod-db:lib
+    prep:
+      - ["sh", "-c", "mkdir -p cmd/resources"]
+      - ["sh", "-c", "cat _deps/components-gitpod-db--lib/package/lib/typeorm/latest-migration.txt > cmd/resources/latest-migration.txt"]
     env:
       - CGO_ENABLED=0
       - GOOS=linux

components/service-waiter/cmd/database.go

@@ -5,9 +5,12 @@
 package cmd
 
 import (
+	"context"
 	"crypto/tls"
 	"crypto/x509"
 	"database/sql"
+	_ "embed"
+	"errors"
 	"net"
 	"os"
 	"strings"
@@ -20,6 +23,11 @@ import (
 	"github.com/gitpod-io/gitpod/common-go/log"
 )
 
+const migrationTableName = "migrations"
+
+//go:embed resources/latest-migration.txt
+var latestMigrationName string
+
 // databaseCmd represents the database command
 var databaseCmd = &cobra.Command{
 	Use:   "database",
@@ -34,11 +42,19 @@ DB_CA_CERT and DB_USER(=gitpod)`,
 		}
 	},
 	Run: func(cmd *cobra.Command, args []string) {
+		timeout := getTimeout()
+		ctx, cancel := context.WithTimeout(cmd.Context(), timeout)
+		defer cancel()
+
 		cfg := mysql.NewConfig()
 		cfg.Addr = net.JoinHostPort(viper.GetString("host"), viper.GetString("port"))
 		cfg.Net = "tcp"
 		cfg.User = viper.GetString("username")
 		cfg.Passwd = viper.GetString("password")
+
+		// Must be "gitpod"
+		// Align to https://github.com/gitpod-io/gitpod/blob/884d922e8e33d8b936ec18d7fe3c8dcffde42b5a/components/gitpod-db/go/conn.go#L37
+		cfg.DBName = "gitpod"
 		cfg.Timeout = 1 * time.Second
 
 		dsn := cfg.FormatDSN()
@@ -65,46 +81,59 @@ DB_CA_CERT and DB_USER(=gitpod)`,
 			cfg.TLSConfig = tlsConfigName
 		}
 
-		timeout := getTimeout()
-		done := make(chan bool)
-		go func() {
-			log.WithField("timeout", timeout.String()).WithField("dsn", censoredDSN).Info("attempting to connect to DB")
-			for {
-				db, err := sql.Open("mysql", cfg.FormatDSN())
-				if err != nil {
-					continue
-				}
-				err = db.Ping()
-				if err != nil {
-					if strings.Contains(err.Error(), "Access denied") {
-						fail("Invalid credentials for the database. Check DB_USERNAME and DB_PASSWORD.")
-					}
-
-					log.WithError(err).Debug("retry")
-					<-time.After(time.Second)
-					continue
-				}
-
-				err = db.Close()
-				if err != nil {
-					log.WithError(err).Warn("cannot close DB connection used for probing")
-				}
+		migrationName := GetLatestMigrationName()
+		log.WithField("timeout", timeout.String()).WithField("dsn", censoredDSN).WithField("migration", migrationName).Info("waiting for database")
+		for ctx.Err() == nil {
+			log.Info("attempting to check if database is available")
+			if err := checkDbAvailable(ctx, cfg, migrationName); err != nil {
+				log.WithError(err).Debug("retry")
+				<-time.After(time.Second)
+			} else {
 				break
 			}
+		}
 
-			done <- true
-		}()
-
-		select {
-		case <-done:
+		if ctx.Err() != nil {
+			log.WithField("timeout", timeout.String()).WithError(ctx.Err()).Fatal("database did not become available in time")
+		} else {
 			log.Info("database became available")
-			return
-		case <-time.After(timeout):
-			log.WithField("timeout", timeout.String()).Fatal("database did not become available in time")
 		}
 	},
 }
 
+func GetLatestMigrationName() string {
+	return strings.TrimSpace(latestMigrationName)
+}
+
+// checkDbAvailable will connect and check if migrations table contains the latest migration
+func checkDbAvailable(ctx context.Context, cfg *mysql.Config, migration string) error {
+	db, err := sql.Open("mysql", cfg.FormatDSN())
+	if err != nil {
+		return err
+	}
+	// ignore error
+	defer db.Close()
+
+	// if migration name is not set, just ping the database
+	if migration == "" {
+		return db.PingContext(ctx)
+	}
+
+	log.Info("checking if database is migrated")
+	row := db.QueryRowContext(ctx, "SELECT name FROM "+migrationTableName+" WHERE name = ?", migration)
+	var name string
+	if err := row.Scan(&name); err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			log.Error("not yet migrated")
+			return err
+		}
+		log.WithError(err).Error("failed to query migrations")
+		return err
+	}
+
+	return nil
+}
+
 func init() {
 	rootCmd.AddCommand(databaseCmd)
 

components/service-waiter/cmd/database_test.go

@@ -0,0 +1,29 @@
+// Copyright (c) 2023 Gitpod GmbH. All rights reserved.
+// Licensed under the GNU Affero General Public License (AGPL).
+// See License.AGPL.txt in the project root for license information.
+
+package cmd
+
+import (
+	_ "embed"
+	"os"
+	"strings"
+	"testing"
+)
+
+func TestGetLatestMigrationName(t *testing.T) {
+	t.Run("should have latest migration name", func(t *testing.T) {
+		path, err := os.Getwd()
+		if err != nil {
+			t.Errorf("failed to get current work dir: %v", err)
+			return
+		}
+		if !strings.Contains(path, "components-service-waiter--app") {
+			t.Skip("skipping test; not running in leeway build")
+			return
+		}
+		if GetLatestMigrationName() == "" {
+			t.Errorf("migration name should not be empty")
+		}
+	})
+}

components/service-waiter/cmd/resources/latest-migration.txt

@@ -100,22 +100,6 @@ func getTimeout() time.Duration {
 	return timeout
 }
 
-// fail ends the waiting process propagating the message on its way out
-func fail(message string) {
-	terminationLog := "/dev/termination-log"
-
-	log.WithField("message", message).Warn("failed to wait for a service")
-
-	if _, err := os.Stat(terminationLog); !os.IsNotExist(err) {
-		err := os.WriteFile(terminationLog, []byte(message), 0600)
-		if err != nil {
-			log.WithError(err).Error("cannot write termination log")
-		}
-	}
-
-	os.Exit(1)
-}
-
 func envOrDefault(env, def string) (res string) {
 	res = os.Getenv(env)
 	if res == "" {