CORE-746 Quicksilver re-migration

core/src/main/resources/logback.xml

@@ -20,7 +20,7 @@
 
     <appender name="Console-Standard" class="ch.qos.logback.core.ConsoleAppender">
         <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
-            <Pattern>%date %-5level [%thread] %logger{36}: %message%n</Pattern>
+            <Pattern>%date %-5level %logger{36}: %message%n</Pattern>
         </encoder>
     </appender>
 
@@ -106,6 +106,27 @@
         <appender-ref ref="SYSLOG"/>
     </logger>
 
+    <logger name="org.broadinstitute.dsde.rawls.monitor.QuicksilverMigrationMonitor" level="info" additivity="false">
+        <appender-ref ref="file"/>
+        <appender-ref ref="${RAWLS_LOG_APPENDER}"/>
+        <appender-ref ref="SYSLOG"/>
+        <appender-ref ref="Sentry" />
+    </logger>
+
+    <logger name="akka.event.slf4j.Slf4jLogger" level="warn" additivity="false">
+        <appender-ref ref="file"/>
+        <appender-ref ref="${RAWLS_LOG_APPENDER}"/>
+        <appender-ref ref="SYSLOG"/>
+        <appender-ref ref="Sentry" />
+    </logger>
+
+    <logger name="akka.stream.Materializer" level="warn" additivity="false">
+        <appender-ref ref="file"/>
+        <appender-ref ref="${RAWLS_LOG_APPENDER}"/>
+        <appender-ref ref="SYSLOG"/>
+        <appender-ref ref="Sentry" />
+    </logger>
+
     <logger name="akka" level="info" additivity="false">
         <appender-ref ref="file"/>
         <appender-ref ref="${RAWLS_LOG_APPENDER}"/>

core/src/main/resources/swagger/api-docs.yaml

@@ -2178,6 +2178,22 @@ paths:
         204:
           description: No migration performed.
           content: { }
+  /api/workspaces/quicksilverValidation:
+    get:
+      tags:
+        - entities
+      summary: Validate previous migrations.
+      operationId: quicksilver_validation
+      parameters:
+        - name: hours
+          in: query
+          schema:
+            type: integer
+            default: 24
+      responses:
+        200:
+          description: Stuff.
+          content: { }
   /api/workspaces/{workspaceNamespace}/{workspaceName}/methodconfigs:
     get:
       tags:

core/src/main/scala/org/broadinstitute/dsde/rawls/Boot.scala

@@ -531,7 +531,7 @@ object Boot extends IOApp with LazyLogging {
         googleProjectRegistrationServiceConstructor
       )
 
-      if (appConfigManager.conf.getBooleanOption("backRawls").getOrElse(false)) {
+      if (true || appConfigManager.conf.getBooleanOption("backRawls").getOrElse(false)) {
         logger.info("This instance has been marked as BACK. Booting monitors...")
 
         BootMonitors.bootMonitors(
@@ -562,7 +562,9 @@ object Boot extends IOApp with LazyLogging {
           gcpBatchBackend,
           methodConfigResolver,
           bardService,
-          workspaceSettingRepository
+          workspaceSettingRepository,
+          entityManager,
+          appConfigManager.conf.getInt("QuicksilverMigrationMonitor.chunkSize")
         )
       } else
         logger.info(

core/src/main/scala/org/broadinstitute/dsde/rawls/dataaccess/slick/CompactEntityComponent.scala

@@ -44,6 +44,7 @@ trait CompactEntityComponent extends LazyLogging {
 
 class CompactEntityQuery(driverComponent: DriverComponent)
     extends CompactEntityMigration
+    with CompactEntityValidation
     with CompactEntityKeysCache
     with RawSqlQuery
     with CompactEntitySerialization {

core/src/main/scala/org/broadinstitute/dsde/rawls/dataaccess/slick/CompactEntityValidation.scala

@@ -0,0 +1,97 @@
+package org.broadinstitute.dsde.rawls.dataaccess.slick
+
+import org.broadinstitute.dsde.rawls.model.{AttributeName, MigratedEntity}
+import slick.jdbc.MySQLProfile.api._
+
+import java.sql.Timestamp
+import java.util.UUID
+
+/**
+ * README:
+ *    1. Restore a Rawls db backup to a temporary CloudSQL instance.
+ *    2. Change the passwords for rawls, root, and readonly users in your temporary CloudSQL
+ *        instance for safety.
+ *    3. Start a local cloud-sql-proxy connecting to the temporary CloudSQL instance.
+ *    4. Modify the `slick.db` section of rawls.conf to point at your local cloud-sql-proxy.
+ *        You'll probably need to use `host.docker.internal` for the hostname in the url,
+ *        and update the password to what you set in step 2.
+ *    5. Start Rawls locally via `config/docker-rsync-local-rawls.sh`
+ *    6. Call the `/api/workspaces/quicksilverValidation` API in your locally running Rawls
+ *    7. Watch your logs
+ *
+ */
+/*
+    -- ----- SQL to run prior to executing the re-migration ----- --
+
+    create table CURRENT_MIGRATION (
+        workspace_id binary(16)
+    ) comment="Holds state for the QuicksilverMigrationMonitor";
+
+    create table MIGRATION_WORKSPACES (
+      PRIMARY KEY (`id`),
+      id binary(16)
+    ) comment="All workspaces that need a re-migration";
+
+    insert into MIGRATION_WORKSPACES(id)
+    select distinct workspace_id from ENTITY where deleted = b'0' order by workspace_id asc;
+
+    create table ENTITY_CORRECTIONS (
+        `id` bigint unsigned NOT NULL AUTO_INCREMENT,
+        `workspace_id` binary(16) NOT NULL,
+        `entity_type` varchar(254) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin DEFAULT NULL,
+        `name` varchar(254) NOT NULL,
+        `attributes` json DEFAULT NULL,
+        `status` varchar(254) NOT NULL default 'OUTSTANDING',
+        PRIMARY KEY (`id`),
+        UNIQUE KEY `idx_corrections_entity_type_name` (`workspace_id`,`entity_type`,`name`),
+        KEY `idx_corrections_status` (`status`)
+    ) comment="Re-migrated entities. Contains only those active entities that had lists, and only their list attributes";
+ */
+trait CompactEntityValidation {
+  this: CompactEntityQuery =>
+
+  def mostRecentMigration: ReadAction[Timestamp] =
+    sql"""select max(LAST_UPDATED)
+         from WORKSPACE_SETTINGS
+         where SETTING_TYPE='CompactDataTables'""".as[Timestamp].head
+
+  def getRecentlyMigratedWorkspaces(maxDate: Timestamp, hours: Int): ReadAction[Seq[UUID]] =
+    sql"""select WORKSPACE_ID
+          from WORKSPACE_SETTINGS
+          where SETTING_TYPE='CompactDataTables'
+          and LAST_UPDATED > DATE_SUB($maxDate, INTERVAL $hours HOUR)""".as[UUID]
+
+  // ***** for tracking re-migration process
+  def getCurrentMigration: ReadAction[Option[UUID]] =
+    sql"""select workspace_id from CURRENT_MIGRATION""".as[UUID].headOption
+
+  def bootstrapCurrentMigration: ReadWriteAction[Int] =
+    sql"""insert into CURRENT_MIGRATION(workspace_id)
+          select id from MIGRATION_WORKSPACES order by id asc limit 1 """.asUpdate
+
+  def nextMigration(previousWorkspaceId: UUID): ReadAction[Option[UUID]] =
+    sql"""select id from MIGRATION_WORKSPACES where id > $previousWorkspaceId order by id asc limit 1"""
+      .as[UUID]
+      .headOption
+
+  def updateCurrentMigration(workspaceId: UUID): ReadWriteAction[Int] =
+    sql"""update CURRENT_MIGRATION set workspace_id = $workspaceId""".asUpdate
+
+  // ***** for persisting to the ENTITY_CORRECTIONS table
+  def saveMigratedEntities(migratedEntities: Seq[MigratedEntity]): ReadWriteAction[Int] = {
+    val startSql = sql"""insert into ENTITY_CORRECTIONS(workspace_id, entity_type, name, attributes)
+       values """
+
+    val paramSql = reduceSqlActionsWithDelim(
+      migratedEntities map { migratedEntity =>
+        sql"""(${migratedEntity.workspaceId}, ${migratedEntity.entityType}, ${migratedEntity.entityName}, ${migratedEntity.serializedAttributes})"""
+      },
+      sql","
+    )
+
+    concatSqlActions(startSql, paramSql).asUpdate
+  }
+
+  def restartWorkspace(workspaceId: UUID): ReadWriteAction[Int] =
+    sql"""delete from ENTITY_CORRECTIONS where workspace_id = $workspaceId""".asUpdate
+}

core/src/main/scala/org/broadinstitute/dsde/rawls/entities/EntityManager.scala

@@ -110,6 +110,45 @@ class EntityManager(providerBuilders: Set[EntityProviderBuilder[_ <: EntityProvi
       }
     }
   }
+
+  def getLocalProvider(requestArguments: EntityRequestArguments)(implicit
+    executionContext: ExecutionContext
+  ): Future[EntityProvider] = {
+    val targetTagFuture = Future(typeTag[LocalEntityProvider])
+    getSpecificProvider(targetTagFuture, requestArguments)
+  }
+
+  def getCompactProvider(requestArguments: EntityRequestArguments)(implicit
+    executionContext: ExecutionContext
+  ): Future[EntityProvider] = {
+    val targetTagFuture = Future(typeTag[CompactEntityProvider])
+    getSpecificProvider(targetTagFuture, requestArguments)
+  }
+
+  def getSpecificProvider(targetTagFuture: Future[TypeTag[_ <: EntityProvider]],
+                          requestArguments: EntityRequestArguments
+  )(implicit
+    executionContext: ExecutionContext
+  ): Future[EntityProvider] =
+    targetTagFuture map { targetTag =>
+      providerBuilders.find(_.builds == targetTag) match {
+        case None =>
+          throw new DataEntityException(
+            s"no entity provider available for ${requestArguments.workspace.toWorkspaceName}"
+          )
+        case Some(builder) =>
+          builder.build(requestArguments) match {
+            case Success(provider) =>
+              // Wrap the provider with AuditLoggingEntityProvider
+              new AuditLoggingEntityProvider(provider, requestArguments, metricsPrefix)
+            case Failure(regrets: DataEntityException) =>
+              throw new RawlsExceptionWithErrorReport(ErrorReport(regrets.code, regrets.getMessage))
+            case Failure(ex: Throwable) =>
+              throw new RawlsExceptionWithErrorReport(ErrorReport(StatusCodes.InternalServerError, ex.getMessage))
+          }
+      }
+    }
+
 }
 
 object EntityManager {