WOR-1823 part 3: workspace deletion refactoring

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

@@ -17,19 +17,15 @@ import org.broadinstitute.dsde.rawls.entities.exceptions.{
 import org.broadinstitute.dsde.rawls.expressions.ExpressionEvaluator
 import org.broadinstitute.dsde.rawls.metrics.RawlsInstrumented
 import org.broadinstitute.dsde.rawls.model.AttributeUpdateOperations.{AttributeUpdateOperation, EntityUpdateDefinition}
-import org.broadinstitute.dsde.rawls.model.{
-  AttributeEntityReference,
-  Entity,
-  EntityCopyDefinition,
-  EntityQuery,
-  ErrorReport,
-  SamResourceTypeNames,
-  SamWorkspaceActions,
-  WorkspaceName,
-  _
+import org.broadinstitute.dsde.rawls.model._
+import org.broadinstitute.dsde.rawls.util.{
+  AttributeSupport,
+  AttributeUpdateOperationException,
+  EntitySupport,
+  JsonFilterUtils,
+  WorkspaceSupport
 }
-import org.broadinstitute.dsde.rawls.util.{AttributeSupport, EntitySupport, JsonFilterUtils, WorkspaceSupport}
-import org.broadinstitute.dsde.rawls.workspace.{AttributeUpdateOperationException, WorkspaceRepository}
+import org.broadinstitute.dsde.rawls.workspace.WorkspaceRepository
 import org.broadinstitute.dsde.rawls.{RawlsException, RawlsExceptionWithErrorReport}
 import slick.jdbc.{ResultSetConcurrency, ResultSetType, TransactionIsolation}
 

core/src/main/scala/org/broadinstitute/dsde/rawls/metrics/MetricsHelper.scala

@@ -40,11 +40,12 @@ object MetricsHelper {
   def incrementFastPassRevokedCounter(memberType: IamMemberType): IO[Unit] =
     incrementFastPassUpdatedCounter(memberType, "revoke")
 
-  def incrementCounter(name: String,
-                       count: Int = 1,
-                       labels: Map[String, String] = Map.empty,
-                       description: Option[String] = None
-  ): IO[Unit] = {
+  def incrementCounter(
+    name: String,
+    count: Int = 1,
+    labels: Map[String, String] = Map.empty,
+    description: Option[String] = None
+  ): Unit = {
     val metrics = meter
       .counterBuilder(s"$PREFIX/$name")
       .setDescription(description.getOrElse("none"))
@@ -53,7 +54,7 @@ object MetricsHelper {
     labels.foreach { case (k, v) =>
       labelBuilder.put(k, v)
     }
-    IO(metrics.add(count, labelBuilder.build()))
+    metrics.add(count, labelBuilder.build())
   }
 
   private def incrementFastPassUpdatedCounter(memberType: IamMemberType, action: String) =

core/src/main/scala/org/broadinstitute/dsde/rawls/submissions/SubmissionsRepository.scala

@@ -0,0 +1,53 @@
+package org.broadinstitute.dsde.rawls.submissions
+
+import org.broadinstitute.dsde.rawls.dataaccess.SlickDataSource
+import org.broadinstitute.dsde.rawls.dataaccess.slick.WorkflowRecord
+import org.broadinstitute.dsde.rawls.metrics.RawlsInstrumented
+import org.broadinstitute.dsde.rawls.model.{WorkflowStatuses, Workspace, WorkspaceName}
+
+import scala.concurrent.{ExecutionContext, Future}
+
+/**
+  * Data access for workflow submissions
+  *
+  * The intention of this class is to hide direct dependencies on Slick behind a relatively clean interface
+  * to ease testability of higher level business logic.
+  */
+class SubmissionsRepository(
+  dataSource: SlickDataSource,
+  trackDetailedSubmissionMetrics: Boolean = true,
+  override val workbenchMetricBaseName: String
+) extends RawlsInstrumented {
+
+  import dataSource.dataAccess.driver.api._
+
+  def getActiveWorkflowsAndSetStatusToAborted(
+    workspace: Workspace
+  )(implicit ex: ExecutionContext): Future[Seq[WorkflowRecord]] =
+    dataSource
+      .inTransaction { dataAccess =>
+        for {
+          // Gather any active workflows with external ids
+          workflowsToAbort <- dataAccess.workflowQuery.findActiveWorkflowsWithExternalIds(workspace)
+
+          // If a workflow is not done, automatically change its status to Aborted
+          _ <- dataAccess.workflowQuery.findWorkflowsByWorkspace(workspace).result.map { workflowRecords =>
+            workflowRecords
+              .filter(workflowRecord => !WorkflowStatuses.withName(workflowRecord.status).isDone)
+              .foreach { workflowRecord =>
+                dataAccess.workflowQuery.updateStatus(workflowRecord, WorkflowStatuses.Aborted) { status =>
+                  if (trackDetailedSubmissionMetrics)
+                    Option(
+                      workflowStatusCounter(
+                        workspaceSubmissionMetricBuilder(workspace.toWorkspaceName, workflowRecord.submissionId)
+                      )(
+                        status
+                      )
+                    )
+                  else None
+                }
+              }
+          }
+        } yield workflowsToAbort
+      }
+}

core/src/main/scala/org/broadinstitute/dsde/rawls/util/AttributeSupport.scala

@@ -1,7 +1,7 @@
 package org.broadinstitute.dsde.rawls.util
 
 import akka.http.scaladsl.model.StatusCodes
-import org.broadinstitute.dsde.rawls.RawlsExceptionWithErrorReport
+import org.broadinstitute.dsde.rawls.{RawlsException, RawlsExceptionWithErrorReport}
 import org.broadinstitute.dsde.rawls.model.Attributable.AttributeMap
 import org.broadinstitute.dsde.rawls.model.AttributeUpdateOperations.{
   AddListMember,
@@ -26,7 +26,6 @@ import org.broadinstitute.dsde.rawls.model.{
   ErrorReport,
   MethodConfiguration
 }
-import org.broadinstitute.dsde.rawls.workspace.{AttributeNotFoundException, AttributeUpdateOperationException}
 
 import scala.concurrent.Future
 
@@ -172,10 +171,13 @@ trait AttributeSupport {
    *
    * @param entity to update
    * @param operations sequence of operations
-   * @throws org.broadinstitute.dsde.rawls.workspace.AttributeNotFoundException when removing from a list attribute that does not exist
+   * @throws AttributeNotFoundException when removing from a list attribute that does not exist
    * @throws AttributeUpdateOperationException when adding or removing from an attribute that is not a list
    * @return the updated entity
    */
   def applyOperationsToEntity(entity: Entity, operations: Seq[AttributeUpdateOperation]): Entity =
     entity.copy(attributes = applyAttributeUpdateOperations(entity, operations))
 }
+
+class AttributeUpdateOperationException(message: String) extends RawlsException(message)
+class AttributeNotFoundException(message: String) extends AttributeUpdateOperationException(message)

core/src/main/scala/org/broadinstitute/dsde/rawls/util/TracingUtils.scala

@@ -96,6 +96,33 @@ object TracingUtils {
   def traceFuture[T](name: String)(f: RawlsTracingContext => Future[T])(implicit ec: ExecutionContext): Future[T] =
     traceFutureWithParent(name, RawlsTracingContext(otelContext = Option(Context.root())))(f)
 
+  /**
+    * Trace a sync operation with the RawlsRequestContext
+    *
+    * @param name The name that will be used in the trace
+    * @param parentContext the RawlsRequestContext to use for tracing, if `parentContext.otelContext` is defined
+    * @param f the operation to execute within the trace
+    * @return the result of the operation `f`
+    */
+  def trace[T](name: String, parentContext: RawlsRequestContext)(f: RawlsRequestContext => T): T =
+    parentContext.otelContext match {
+      case Some(otelContext) if instrumenter.shouldStart(otelContext, name) =>
+        val childContext = instrumenter.start(otelContext, name)
+        val result = Try(f(parentContext.copy(otelContext = Option(childContext))))
+        instrumenter.end(childContext, name, name, result.failed.toOption.orNull)
+        result.get
+      case _ =>
+        f(parentContext)
+    }
+
+  /**
+    * Trace a sync operation with the RawlsTracingContext
+    *
+    * @param name          The name that will be used in the trace
+    * @param parentContext the RawlsTracingContext to use for tracing, if `parentContext.otelContext` is defined
+    * @param f             the operation to execute within the trace
+    * @return the result of the operation `f`
+    */
   def traceNakedWithParent[T](name: String, parentContext: RawlsTracingContext)(
     f: RawlsTracingContext => T
   ): T =

core/src/main/scala/org/broadinstitute/dsde/rawls/workspace/WorkspaceRepository.scala

@@ -3,6 +3,7 @@ package org.broadinstitute.dsde.rawls.workspace
 import akka.http.scaladsl.model.StatusCodes
 import org.broadinstitute.dsde.rawls.RawlsExceptionWithErrorReport
 import org.broadinstitute.dsde.rawls.dataaccess.SlickDataSource
+import org.broadinstitute.dsde.rawls.dataaccess.slick.PendingBucketDeletionRecord
 import org.broadinstitute.dsde.rawls.model.Attributable.AttributeMap
 import org.broadinstitute.dsde.rawls.model.{
   ErrorReport,
@@ -18,7 +19,6 @@ import org.broadinstitute.dsde.rawls.model.{
 import org.broadinstitute.dsde.rawls.model.WorkspaceState.WorkspaceState
 import org.broadinstitute.dsde.rawls.util.TracingUtils.traceDBIOWithParent
 import org.joda.time.DateTime
-import slick.dbio.DBIO
 
 import java.util.UUID
 import scala.concurrent.{ExecutionContext, Future}
@@ -51,7 +51,9 @@ class WorkspaceRepository(dataSource: SlickDataSource) {
     _.workspaceQuery.getV2WorkspaceId(workspaceName)
   }
 
-  def listWorkspacesByIds(workspaceIds: Seq[UUID], attributeSpecs: Option[WorkspaceAttributeSpecs] = None): Future[Seq[Workspace]] = dataSource.inTransaction {
+  def listWorkspacesByIds(workspaceIds: Seq[UUID],
+                          attributeSpecs: Option[WorkspaceAttributeSpecs] = None
+  ): Future[Seq[Workspace]] = dataSource.inTransaction {
     _.workspaceQuery.listV2WorkspacesByIds(workspaceIds, attributeSpecs)
   }
 
@@ -77,6 +79,22 @@ class WorkspaceRepository(dataSource: SlickDataSource) {
       access.workspaceQuery.delete(workspaceName)
     }
 
+  def deleteRawlsWorkspace(workspace: Workspace)(implicit ex: ExecutionContext): Future[Unit] =
+    dataSource.inTransaction { dataAccess =>
+      for {
+        // Delete components of the workspace
+        _ <- dataAccess.submissionQuery.deleteFromDb(workspace.workspaceIdAsUUID)
+        _ <- dataAccess.methodConfigurationQuery.deleteFromDb(workspace.workspaceIdAsUUID)
+        _ <- dataAccess.entityQuery.deleteFromDb(workspace)
+
+        // Schedule bucket for deletion
+        _ <- dataAccess.pendingBucketDeletionQuery.save(PendingBucketDeletionRecord(workspace.bucketName))
+
+        // Delete the workspace
+        _ <- dataAccess.workspaceQuery.delete(workspace.toWorkspaceName)
+      } yield ()
+    }
+
   def createMCWorkspace(workspaceId: UUID,
                         workspaceName: WorkspaceName,
                         attributes: AttributeMap,
@@ -153,13 +171,13 @@ class WorkspaceRepository(dataSource: SlickDataSource) {
   def savePendingCloneWorkspaceFileTransfer(destWorkspace: UUID, sourceWorkspace: UUID, prefix: String): Future[Int] =
     dataSource.inTransaction(_.cloneWorkspaceFileTransferQuery.save(destWorkspace, sourceWorkspace, prefix))
 
-  def getSubmissionSummaryStats(workspaceId: UUID)(implicit ex: ExecutionContext): Future[Option[WorkspaceSubmissionStats]] =
+  def getSubmissionSummaryStats(workspaceId: UUID)(implicit
+    ex: ExecutionContext
+  ): Future[Option[WorkspaceSubmissionStats]] =
     dataSource.inTransaction(_.workspaceQuery.listSubmissionSummaryStats(Seq(workspaceId))).map(_.values.headOption)
 
   def listSubmissionSummaryStats(workspaceIds: Seq[UUID]): Future[Map[UUID, WorkspaceSubmissionStats]] =
-  dataSource.inTransaction(_.workspaceQuery.listSubmissionSummaryStats(workspaceIds))
-
-
+    dataSource.inTransaction(_.workspaceQuery.listSubmissionSummaryStats(workspaceIds))
 
   def getTags(workspaceIds: Seq[UUID], query: Option[String], limit: Option[Int] = None): Future[Seq[WorkspaceTag]] =
     dataSource.inTransaction(_.workspaceQuery.getTags(query, limit, Some(workspaceIds)))