Merge remote-tracking branch 'origin/develop' into eric/RW-11654

.github/workflows/leo-build-tag-publish-and-run-tests.yml

@@ -46,7 +46,7 @@ jobs:
       - name: Get inputs and/or set defaults
         id: prepare-outputs
         run: |
-          echo 'notify-failure-channel=ia-test-notification'           >> $GITHUB_OUTPUT
+          echo 'notify-failure-channel=ia-notification-test'          >> $GITHUB_OUTPUT
           echo 'delete-bee=true'                                       >> $GITHUB_OUTPUT
           echo 'log-results=true'                                      >> $GITHUB_OUTPUT
           if ${{ github.ref_name == 'develop' }}; then

Dockerfile

@@ -30,7 +30,7 @@ ENV TERRA_APP_VERSION 0.5.0
 ENV GALAXY_VERSION 2.8.1
 ENV NGINX_VERSION 4.3.0
 # If you update this here, make sure to also update reference.conf:
-ENV CROMWELL_CHART_VERSION 0.2.428
+ENV CROMWELL_CHART_VERSION 0.2.429
 ENV HAIL_BATCH_CHART_VERSION 0.1.9
 ENV RSTUDIO_CHART_VERSION 0.4.0
 ENV SAS_CHART_VERSION 0.5.0

core/src/main/scala/org/broadinstitute/dsde/workbench/leonardo/kubernetesModels.scala

@@ -536,6 +536,15 @@ object AppStatus {
 
   val monitoredStatuses: Set[AppStatus] =
     Set(Deleting, Provisioning)
+
+  implicit class EnrichedDiskStatus(status: AppStatus) {
+    def isDeletable: Boolean = deletableStatuses contains status
+
+    def isStoppable: Boolean = stoppableStatuses contains status
+
+    def isStartable: Boolean = startableStatuses contains status
+
+  }
 }
 
 final case class KubernetesService(id: ServiceId, config: ServiceConfig)

core/src/main/scala/org/broadinstitute/dsde/workbench/leonardo/models.scala

@@ -1,9 +1,9 @@
 package org.broadinstitute.dsde.workbench.leonardo
 
+import bio.terra.workspace.model.State
 import ca.mrvisser.sealerate
 import org.broadinstitute.dsde.workbench.azure.AzureCloudContext
 import org.broadinstitute.dsde.workbench.model.google.{GcsBucketName, GoogleProject}
-
 import java.util.UUID
 
 final case class WorkspaceId(value: UUID) extends AnyVal
@@ -57,3 +57,26 @@ object StagingBucket {
     override def asString: String = s"${storageContainerName.value}"
   }
 }
+
+/**
+ * Can't extend final enum State from WSM, so made a wrapper
+ * WSM state can be BROKEN, CREATING, DELETING, READY, UPDATING or None
+ * if None --> it is deletable and is deleted
+ * (already deleted in WSM, need to clean up leo resources)
+ */
+case class WsmState(state: Option[String]) {
+
+  val deletableStatuses: Set[String] = Set("BROKEN", "READY", "DELETED")
+  val possibleStatuses: Array[String] = State.values().map(_.toString) :+ "DELETED"
+
+//  def apply(): Unit =
+//    if (!possibleStatuses.contains(this.value)) {
+//      log.warn(s"Unrecognized WSM state $state, WSM resource may not be processed correctly")
+//    }
+  def value: String = state.getOrElse("DELETED").toUpperCase()
+
+  /** Any in-progress state cannot be deleted: CREATING, DELETING, UPDATING */
+  def isDeletable: Boolean = deletableStatuses contains this.value
+
+  def isDeleted: Boolean = this.value == "DELETED"
+}

http/src/main/resources/reference.conf

@@ -229,7 +229,7 @@ azure {
    coa-app-config {
      instrumentation-enabled = false
      chart-name = "cromwell-helm/cromwell-on-azure"
-     chart-version = "0.2.428"
+     chart-version = "0.2.429"
      release-name-suffix = "coa-rls"
      namespace-name-suffix = "coa-ns"
      ksa-name = "coa-ksa"
@@ -765,7 +765,7 @@ gke {
   cromwellApp {
     # If you update the chart name or version here, make sure to also update it in the dockerfile:
     chartName = "/leonardo/cromwell"
-    chartVersion = "0.2.428"
+    chartVersion = "0.2.429"
     services = [
       {
         name = "cromwell-service"

http/src/main/scala/org/broadinstitute/dsde/workbench/leonardo/dao/WsmApiClientProvider.scala

@@ -3,14 +3,16 @@ package org.broadinstitute.dsde.workbench.leonardo.dao
 import bio.terra.common.tracing.JerseyTracingFilter
 import bio.terra.workspace.api.{ControlledAzureResourceApi, ResourceApi}
 import bio.terra.workspace.client.ApiClient
+import bio.terra.workspace.model.State
 import cats.effect.Async
 import cats.mtl.Ask
 import cats.syntax.all._
 import io.opencensus.trace.Tracing
-import org.broadinstitute.dsde.workbench.leonardo.AppContext
+import org.broadinstitute.dsde.workbench.leonardo.{AppContext, WorkspaceId, WsmControlledResourceId, WsmState}
 import org.broadinstitute.dsde.workbench.leonardo.util.WithSpanFilter
 import org.glassfish.jersey.client.ClientConfig
 import org.http4s.Uri
+import org.typelevel.log4cats.StructuredLogger
 
 /**
  * Represents a way to get a client for interacting with workspace manager controlled resources.
@@ -20,8 +22,36 @@ import org.http4s.Uri
  *
  */
 trait WsmApiClientProvider[F[_]] {
+
+  val possibleStatuses: Array[WsmState] =
+    State.values().map(_.toString).map(Some(_)).map(WsmState(_)) :+ WsmState(Some("DELETED"))
+
   def getControlledAzureResourceApi(token: String)(implicit ev: Ask[F, AppContext]): F[ControlledAzureResourceApi]
   def getResourceApi(token: String)(implicit ev: Ask[F, AppContext]): F[ResourceApi]
+  def getDiskState(token: String, workspaceId: WorkspaceId, wsmResourceId: WsmControlledResourceId)(implicit
+    ev: Ask[F, AppContext],
+    log: StructuredLogger[F]
+  ): F[WsmState]
+
+  def getVmState(token: String, workspaceId: WorkspaceId, wsmResourceId: WsmControlledResourceId)(implicit
+    ev: Ask[F, AppContext],
+    log: StructuredLogger[F]
+  ): F[WsmState]
+
+  def getDatabaseState(token: String, workspaceId: WorkspaceId, wsmResourceId: WsmControlledResourceId)(implicit
+    ev: Ask[F, AppContext],
+    log: StructuredLogger[F]
+  ): F[WsmState]
+
+  def getNamespaceState(token: String, workspaceId: WorkspaceId, wsmResourceId: WsmControlledResourceId)(implicit
+    ev: Ask[F, AppContext],
+    log: StructuredLogger[F]
+  ): F[WsmState]
+
+  def getIdentityState(token: String, workspaceId: WorkspaceId, wsmResourceId: WsmControlledResourceId)(implicit
+    ev: Ask[F, AppContext],
+    log: StructuredLogger[F]
+  ): F[WsmState]
 }
 
 class HttpWsmClientProvider[F[_]](baseWorkspaceManagerUrl: Uri)(implicit F: Async[F]) extends WsmApiClientProvider[F] {
@@ -41,11 +71,82 @@ class HttpWsmClientProvider[F[_]](baseWorkspaceManagerUrl: Uri)(implicit F: Asyn
       _ = client.setAccessToken(token)
     } yield client
 
+  private def toWsmStatus(
+    state: Option[String]
+  )(implicit logger: StructuredLogger[F]): WsmState = {
+    val wsmState = WsmState(state)
+    if (!possibleStatuses.contains(wsmState)) logger.warn("Invalid Wsm status")
+    wsmState
+  }
+  override def getResourceApi(token: String)(implicit ev: Ask[F, AppContext]): F[ResourceApi] =
+    getApiClient(token).map(apiClient => new ResourceApi(apiClient))
+
   override def getControlledAzureResourceApi(token: String)(implicit
     ev: Ask[F, AppContext]
   ): F[ControlledAzureResourceApi] =
     getApiClient(token).map(apiClient => new ControlledAzureResourceApi(apiClient))
 
-  override def getResourceApi(token: String)(implicit ev: Ask[F, AppContext]): F[ResourceApi] =
-    getApiClient(token).map(apiClient => new ResourceApi(apiClient))
+  override def getVmState(token: String, workspaceId: WorkspaceId, wsmResourceId: WsmControlledResourceId)(implicit
+    ev: Ask[F, AppContext],
+    log: StructuredLogger[F]
+  ): F[WsmState] = for {
+    wsmApi <- getControlledAzureResourceApi(token)
+    attempt <- F.delay(wsmApi.getAzureVm(workspaceId.value, wsmResourceId.value)).attempt
+    state = attempt match {
+      case Right(result) => Some(result.getMetadata.getState.getValue)
+      case Left(_)       => None
+    }
+  } yield toWsmStatus(state)
+
+  override def getDiskState(token: String, workspaceId: WorkspaceId, wsmResourceId: WsmControlledResourceId)(implicit
+    ev: Ask[F, AppContext],
+    log: StructuredLogger[F]
+  ): F[WsmState] = for {
+    wsmApi <- getControlledAzureResourceApi(token)
+    attempt <- F.delay(wsmApi.getAzureDisk(workspaceId.value, wsmResourceId.value)).attempt
+    state = attempt match {
+      case Right(result) => Some(result.getMetadata.getState.getValue)
+      case Left(_)       => None
+    }
+  } yield toWsmStatus(state)
+
+  override def getDatabaseState(token: String, workspaceId: WorkspaceId, wsmResourceId: WsmControlledResourceId)(
+    implicit
+    ev: Ask[F, AppContext],
+    log: StructuredLogger[F]
+  ): F[WsmState] = for {
+    wsmApi <- getControlledAzureResourceApi(token)
+    attempt <- F.delay(wsmApi.getAzureDatabase(workspaceId.value, wsmResourceId.value)).attempt
+    state = attempt match {
+      case Right(result) => Some(result.getMetadata.getState.getValue)
+      case Left(_)       => None
+    }
+  } yield toWsmStatus(state)
+
+  override def getNamespaceState(token: String, workspaceId: WorkspaceId, wsmResourceId: WsmControlledResourceId)(
+    implicit
+    ev: Ask[F, AppContext],
+    log: StructuredLogger[F]
+  ): F[WsmState] = for {
+    wsmApi <- getControlledAzureResourceApi(token)
+    attempt <- F.delay(wsmApi.getAzureKubernetesNamespace(workspaceId.value, wsmResourceId.value)).attempt
+    state = attempt match {
+      case Right(result) => Some(result.getMetadata.getState.getValue)
+      case Left(_)       => None
+    }
+  } yield toWsmStatus(state)
+
+  override def getIdentityState(token: String, workspaceId: WorkspaceId, wsmResourceId: WsmControlledResourceId)(
+    implicit
+    ev: Ask[F, AppContext],
+    log: StructuredLogger[F]
+  ): F[WsmState] = for {
+    wsmApi <- getControlledAzureResourceApi(token)
+    attempt <- F.delay(wsmApi.getAzureManagedIdentity(workspaceId.value, wsmResourceId.value)).attempt
+    state = attempt match {
+      case Right(result) => Some(result.getMetadata.getState.getValue)
+      case Left(_)       => None
+    }
+  } yield toWsmStatus(state)
+
 }

http/src/main/scala/org/broadinstitute/dsde/workbench/leonardo/db/ClusterComponent.scala

@@ -703,7 +703,7 @@ object clusterQuery extends TableQuery(new ClusterTable(_)) {
       _ <- clusterImageQuery.upsert(id, welderImage)
     } yield ()
 
-  def updateDiskStatus(runtimeId: Long, dateAccessed: Instant)(implicit
+  def setDiskDeleted(runtimeId: Long, dateAccessed: Instant)(implicit
     ec: ExecutionContext
   ): DBIO[Unit] =
     for {
@@ -712,9 +712,20 @@ object clusterQuery extends TableQuery(new ClusterTable(_)) {
         .result
         .headOption
       diskIdOpt <- runtimeConfigId.flatTraverse(rid => RuntimeConfigQueries.getDiskId(rid))
-      _ <- diskIdOpt.traverse(diskId => persistentDiskQuery.updateStatus(diskId, DiskStatus.Deleted, dateAccessed))
+      _ <- diskIdOpt.traverse(diskId => persistentDiskQuery.delete(diskId, dateAccessed))
     } yield ()
 
+  def getDiskId(runtimeId: Long)(implicit
+    ec: ExecutionContext
+  ): DBIO[Option[DiskId]] =
+    for {
+      runtimeConfigId <- findByIdQuery(runtimeId)
+        .map(_.runtimeConfigId)
+        .result
+        .headOption
+      diskIdOpt <- runtimeConfigId.flatTraverse(rid => RuntimeConfigQueries.getDiskId(rid))
+    } yield diskIdOpt
+
   def setToRunning(id: Long, hostIp: IP, dateAccessed: Instant): DBIO[Int] =
     updateClusterStatusAndHostIp(id, RuntimeStatus.Running, Some(hostIp), dateAccessed)
 

http/src/main/scala/org/broadinstitute/dsde/workbench/leonardo/db/PersistentDiskComponent.scala

@@ -254,7 +254,7 @@ object persistentDiskQuery {
 
   def nullifyDiskIds = persistentDiskQuery.tableQuery.map(x => x.lastUsedBy).update(None)
 
-  def delete(id: DiskId, destroyedDate: Instant) =
+  def delete(id: DiskId, destroyedDate: Instant): DBIO[Int] =
     findByIdQuery(id)
       .map(d => (d.status, d.destroyedDate, d.dateAccessed))
       .update((DiskStatus.Deleted, destroyedDate, destroyedDate))

http/src/main/scala/org/broadinstitute/dsde/workbench/leonardo/http/Boot.scala

@@ -185,7 +185,8 @@ object Boot extends IOApp {
         appDependencies.authProvider,
         appDependencies.wsmDAO,
         appDependencies.samDAO,
-        appDependencies.publisherQueue
+        appDependencies.publisherQueue,
+        appDependencies.wsmClientProvider
       )
 
       val leoKubernetesService: LeoAppServiceInterp[IO] =
@@ -197,7 +198,8 @@ object Boot extends IOApp {
           appDependencies.googleDependencies.googleComputeService,
           googleDependencies.googleResourceService,
           gkeCustomAppConfig,
-          appDependencies.wsmDAO
+          appDependencies.wsmDAO,
+          appDependencies.wsmClientProvider
         )
 
       val azureService = new RuntimeV2ServiceInterp[IO](

http/src/main/scala/org/broadinstitute/dsde/workbench/leonardo/http/service/DiskServiceInterp.scala

@@ -450,7 +450,14 @@ case class PersistentDiskAlreadyExistsException(googleProject: GoogleProject,
 
 case class DiskCannotBeDeletedException(id: DiskId, status: DiskStatus, cloudContext: CloudContext, traceId: TraceId)
     extends LeoException(
-      s"Persistent disk ${id.value} cannot be deleted in ${status} status. CloudContext: ${cloudContext.asStringWithProvider}",
+      s"Persistent disk ${id.value} cannot be deleted in $status status. CloudContext: ${cloudContext.asStringWithProvider}",
+      StatusCodes.Conflict,
+      traceId = Some(traceId)
+    )
+
+case class DiskCannotBeDeletedWsmException(id: DiskId, status: WsmState, cloudContext: CloudContext, traceId: TraceId)
+    extends LeoException(
+      s"Persistent disk ${id.value} cannot be deleted in ${status.value} status, please wait and try again. CloudContext: ${cloudContext.asStringWithProvider}",
       StatusCodes.Conflict,
       traceId = Some(traceId)
     )