Fix #2479 - Check migration at server start (#2489)

* Validate migrations on start

* Output to logs

* Validate Flyway Migrations at Catalog start

* Add flyway validation test config

* Sonarcloud try stream

* Revert migration validation

catalog-rest-service/src/main/java/org/openmetadata/catalog/CatalogApplication.java

@@ -31,7 +31,9 @@ import io.federecio.dropwizard.swagger.SwaggerBundle;
 import io.federecio.dropwizard.swagger.SwaggerBundleConfiguration;
 import java.io.IOException;
 import java.lang.reflect.InvocationTargetException;
+import java.sql.SQLException;
 import java.time.temporal.ChronoUnit;
+import java.util.Optional;
 import javax.ws.rs.container.ContainerRequestFilter;
 import javax.ws.rs.container.ContainerResponseFilter;
 import javax.ws.rs.core.Response;
@@ -50,6 +52,8 @@ import org.openmetadata.catalog.events.EventPubSub;
 import org.openmetadata.catalog.exception.CatalogGenericExceptionMapper;
 import org.openmetadata.catalog.exception.ConstraintViolationExceptionMapper;
 import org.openmetadata.catalog.exception.JsonMappingExceptionMapper;
+import org.openmetadata.catalog.migration.Migration;
+import org.openmetadata.catalog.migration.MigrationConfiguration;
 import org.openmetadata.catalog.resources.CollectionRegistry;
 import org.openmetadata.catalog.resources.config.ConfigResource;
 import org.openmetadata.catalog.resources.search.SearchResource;
@@ -70,7 +74,7 @@ public class CatalogApplication extends Application<CatalogApplicationConfig> {
   @Override
   public void run(CatalogApplicationConfig catalogConfig, Environment environment)
       throws ClassNotFoundException, IllegalAccessException, InstantiationException, NoSuchMethodException,
-          InvocationTargetException, IOException {
+          InvocationTargetException, IOException, SQLException {
 
     final JdbiFactory factory = new JdbiFactory();
     final Jdbi jdbi = factory.build(environment, catalogConfig.getDataSourceFactory(), "mysql3");
@@ -90,6 +94,9 @@ public class CatalogApplication extends Application<CatalogApplicationConfig> {
       jdbi.setSqlLogger(sqlLogger);
     }
 
+    // Validate flyway Migrations
+    validateMigrations(jdbi, catalogConfig.getMigrationConfiguration());
+
     // Register Authorizer
     registerAuthorizer(catalogConfig, environment, jdbi);
 
@@ -144,6 +151,27 @@ public class CatalogApplication extends Application<CatalogApplicationConfig> {
     super.initialize(bootstrap);
   }
 
+  private void validateMigrations(Jdbi jdbi, MigrationConfiguration conf) throws IOException {
+    LOG.info("Validating Flyway migrations");
+    Optional<String> lastMigrated = Migration.lastMigrated(jdbi);
+    String maxMigration = Migration.lastMigrationFile(conf);
+
+    if (lastMigrated.isEmpty()) {
+      System.out.println(
+          "Could not validate Flyway migrations in MySQL."
+              + " Make sure you have run `./bootstrap/bootstrap_storage.sh migrate-all` at least once.");
+      System.exit(1);
+    }
+    if (lastMigrated.get().compareTo(maxMigration) < 0) {
+      System.out.println(
+          "There are pending migrations to be run on MySQL."
+              + " Please backup your data and run `./bootstrap/bootstrap_storage.sh migrate-all`."
+              + " You can find more information on upgrading OpenMetadata at"
+              + " https://docs.open-metadata.org/install/upgrade-openmetadata");
+      System.exit(1);
+    }
+  }
+
   private void registerAuthorizer(CatalogApplicationConfig catalogConfig, Environment environment, Jdbi jdbi)
       throws NoSuchMethodException, ClassNotFoundException, IllegalAccessException, InvocationTargetException,
           InstantiationException, IOException {

catalog-rest-service/src/main/java/org/openmetadata/catalog/CatalogApplicationConfig.java

@@ -24,6 +24,7 @@ import javax.validation.constraints.NotNull;
 import org.openmetadata.catalog.airflow.AirflowConfiguration;
 import org.openmetadata.catalog.elasticsearch.ElasticSearchConfiguration;
 import org.openmetadata.catalog.events.EventHandlerConfiguration;
+import org.openmetadata.catalog.migration.MigrationConfiguration;
 import org.openmetadata.catalog.security.AuthenticationConfiguration;
 import org.openmetadata.catalog.security.AuthorizerConfiguration;
 import org.openmetadata.catalog.slack.SlackPublisherConfiguration;
@@ -56,6 +57,10 @@ public class CatalogApplicationConfig extends Configuration {
   @JsonProperty("slackEventPublishers")
   private List<SlackPublisherConfiguration> slackEventPublishers;
 
+  @NotNull
+  @JsonProperty("migrationConfiguration")
+  private MigrationConfiguration migrationConfiguration;
+
   public DataSourceFactory getDataSourceFactory() {
     return dataSourceFactory;
   }
@@ -108,6 +113,14 @@ public class CatalogApplicationConfig extends Configuration {
     return slackEventPublishers;
   }
 
+  public MigrationConfiguration getMigrationConfiguration() {
+    return migrationConfiguration;
+  }
+
+  public void setMigrationConfiguration(MigrationConfiguration migrationConfiguration) {
+    this.migrationConfiguration = migrationConfiguration;
+  }
+
   public void setSlackEventPublishers(List<SlackPublisherConfiguration> slackEventPublishers) {
     this.slackEventPublishers = slackEventPublishers;
   }

catalog-rest-service/src/main/java/org/openmetadata/catalog/jdbi3/MigrationDAO.java

@@ -0,0 +1,12 @@
+package org.openmetadata.catalog.jdbi3;
+
+import java.util.Optional;
+import org.jdbi.v3.core.statement.StatementException;
+import org.jdbi.v3.sqlobject.SingleValue;
+import org.jdbi.v3.sqlobject.statement.SqlQuery;
+
+public interface MigrationDAO {
+  @SqlQuery("SELECT MAX(version) FROM DATABASE_CHANGE_LOG")
+  @SingleValue
+  Optional<String> getMaxVersion() throws StatementException;
+}

catalog-rest-service/src/main/java/org/openmetadata/catalog/migration/Migration.java

@@ -0,0 +1,74 @@
+package org.openmetadata.catalog.migration;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import lombok.extern.slf4j.Slf4j;
+import org.jdbi.v3.core.Jdbi;
+import org.jdbi.v3.core.statement.StatementException;
+import org.openmetadata.catalog.jdbi3.MigrationDAO;
+
+@Slf4j
+public final class Migration {
+
+  /**
+   * Run a query to MySQL to retrieve the last migrated Flyway version. If the Flyway table DATABASE_CHANGE_LOG does not
+   * exist, we will stop the Catalog App and inform users how to run Flyway.
+   *
+   * @param jdbi JDBI connection
+   * @return Last migrated version, e.g., "003"
+   */
+  public static Optional<String> lastMigrated(Jdbi jdbi) {
+    try {
+      return jdbi.withExtension(MigrationDAO.class, MigrationDAO::getMaxVersion);
+    } catch (StatementException e) {
+      System.out.println(
+          "Exception encountered when trying to obtain last migrated Flyway version."
+              + " Make sure you have run `./bootstrap/bootstrap_storage.sh migrate-all` at least once.");
+      System.exit(1);
+    }
+    return Optional.empty();
+  }
+
+  public static String lastMigrationFile(MigrationConfiguration conf) throws IOException {
+    List<String> migrationFiles = getMigrationVersions(conf);
+    return Collections.max(migrationFiles);
+  }
+
+  /**
+   * Read the migrations path from the Catalog YAML config and return a list of all the files' versions.
+   *
+   * @param conf Catalog migration config
+   * @return List of migration files' versions
+   * @throws IOException If we cannot read the files
+   */
+  private static List<String> getMigrationVersions(MigrationConfiguration conf) throws IOException {
+    try (Stream<String> names =
+        Files.walk(Paths.get(conf.getPath()))
+            .filter(Files::isRegularFile)
+            .map(Path::toFile)
+            .map(File::getName)
+            .map(Migration::cleanName)) {
+
+      return names.collect(Collectors.toList());
+    }
+  }
+
+  /**
+   * Given a Flyway migration filename, e.g., v001__my_file.sql, return the version information "001".
+   *
+   * @param name Flyway migration filename
+   * @return File version
+   */
+  private static String cleanName(String name) {
+    return Arrays.asList(name.split("_")).get(0).replace("v", "");
+  }
+}

catalog-rest-service/src/main/java/org/openmetadata/catalog/migration/MigrationConfiguration.java

@@ -0,0 +1,15 @@
+package org.openmetadata.catalog.migration;
+
+import javax.validation.constraints.NotEmpty;
+
+public class MigrationConfiguration {
+  @NotEmpty private String path;
+
+  public String getPath() {
+    return path;
+  }
+
+  public void setPath(String path) {
+    this.path = path;
+  }
+}

catalog-rest-service/src/test/java/org/openmetadata/catalog/resources/EmbeddedMySqlSupport.java

@@ -50,6 +50,7 @@ public class EmbeddedMySqlSupport implements BeforeAllCallback, AfterAllCallback
               // TODO Remove hardcoding
               .dataSource(
                   "jdbc:mysql://localhost:3307/openmetadata_test_db?useSSL=false&serverTimezone=UTC", "test", "")
+              .table("DATABASE_CHANGE_LOG")
               .sqlMigrationPrefix("v")
               .load();
       flyway.clean();

catalog-rest-service/src/test/resources/openmetadata-secure-test.yaml

@@ -104,6 +104,9 @@ database:
   # the JDBC URL; the database is called washvalet
   url: jdbc:mysql://localhost:3307/openmetadata_test_db?useSSL=false&serverTimezone=UTC
 
+migrationConfiguration:
+  path: "../bootstrap/sql/mysql"
+
 elasticsearch:
   host: localhost
   port: 0

conf/openmetadata.yaml

@@ -110,6 +110,8 @@ database:
   # the JDBC URL; the database is called openmetadata_db
   url: jdbc:mysql://${MYSQL_HOST:-localhost}:${MYSQL_PORT:-3306}/${MYSQL_DATABASE:-openmetadata_db}?allowPublicKeyRetrieval=true&useSSL=false&serverTimezone=UTC
 
+migrationConfiguration:
+  path: "./bootstrap/sql/mysql"
 
 elasticsearch:
   host: ${ELASTICSEARCH_HOST:-localhost}