Issue 733: If google marks cluster as Error but doesn't return error code then don't continue to monitor the cluster (#764)

* fix clustermonitoractor

Author: vkumra-broad

src/main/scala/org/broadinstitute/dsde/workbench/leonardo/monitor/ClusterMonitorActor.scala

@@ -70,6 +70,8 @@ class ClusterMonitorActor(val cluster: Cluster,
   // the Retry trait needs a reference to the ActorSystem
   override val system = context.system
 
+  private val internalError = "Internal Error"
+
   override def preStart(): Unit = {
     super.preStart()
     scheduleInitialMonitorPass
@@ -303,7 +305,7 @@ class ClusterMonitorActor(val cluster: Cluster,
         case Error if leoClusterStatus != Deleting && leoClusterStatus != Stopping =>
           gdDAO.getClusterErrorDetails(cluster.dataprocInfo.operationName).map {
             case Some(errorDetails) => FailedCluster(errorDetails, googleInstances)
-            case None => NotReadyCluster(ClusterStatus.Error, googleInstances)
+            case None => FailedCluster(ClusterErrorDetails(Code.INTERNAL.value, Some(internalError)), googleInstances)
           }
         // Take care we don't delete a Creating cluster if google hasn't updated their status yet
         case Deleted if leoClusterStatus == Creating =>

src/test/scala/org/broadinstitute/dsde/workbench/leonardo/monitor/ClusterMonitorSpec.scala

@@ -278,62 +278,73 @@ class ClusterMonitorSpec extends TestKit(ActorSystem("leonardotest")) with FlatS
 
   // Pre:
   // - cluster exists in the DB with status Creating
-  // - dataproc DAO returns status ERROR, but no error code
+  // - dataproc DAO returns status ERROR and error code CANCELLED
   // - compute DAO returns RUNNING
   // Post:
-  // - cluster is not changed in the DB
+  // - cluster status is set to Error in the DB
   // - instances are populated in the DB
-  // - monitor actor does not shut down
-  it should "keep monitoring in ERROR state with no error code" in isolatedDbTest {
+  // - monitor actor shuts down
+  it should "monitor until ERROR state with no restart and getClusterStatus returns Some(ErrorDetail) " in isolatedDbTest {
     val savedCreatingCluster = creatingCluster.save()
     creatingCluster shouldEqual savedCreatingCluster
-    
-    val dao = mock[GoogleDataprocDAO]
+
+    val gdDAO = mock[GoogleDataprocDAO]
     when {
-      dao.getClusterStatus(mockitoEq(creatingCluster.googleProject), mockitoEq(creatingCluster.clusterName))
+      gdDAO.getClusterStatus(mockitoEq(creatingCluster.googleProject), mockitoEq(creatingCluster.clusterName))
     } thenReturn Future.successful(ClusterStatus.Error)
 
     when {
-      dao.getClusterInstances(mockitoEq(creatingCluster.googleProject), mockitoEq(creatingCluster.clusterName))
-    } thenReturn Future.successful(clusterInstances)
+      gdDAO.getClusterErrorDetails(mockitoEq(creatingCluster.dataprocInfo.operationName))
+    } thenReturn Future.successful(Some(ClusterErrorDetails(Code.CANCELLED.value, Some("test message"))))
 
     when {
-      dao.getClusterErrorDetails(mockitoEq(creatingCluster.dataprocInfo.operationName))
-    } thenReturn Future.successful(None)
+      gdDAO.deleteCluster(mockitoEq(creatingCluster.googleProject), mockitoEq(creatingCluster.clusterName))
+    } thenReturn Future.successful(())
+
+    when {
+      gdDAO.getClusterInstances(mockitoEq(creatingCluster.googleProject), mockitoEq(creatingCluster.clusterName))
+    } thenReturn Future.successful(clusterInstances)
 
     val iamDAO = mock[GoogleIamDAO]
     when {
       iamDAO.removeIamRolesForUser(any[GoogleProject], any[WorkbenchEmail], mockitoEq(Set("roles/dataproc.worker")))
     } thenReturn Future.successful(())
 
+    when {
+      iamDAO.removeServiceAccountKey(any[GoogleProject], any[WorkbenchEmail], any[ServiceAccountKeyId])
+    } thenReturn Future.successful(())
+
     val computeDAO = stubComputeDAO(InstanceStatus.Running)
     val storageDAO = mock[GoogleStorageDAO]
     val authProvider = mock[LeoAuthProvider]
 
-    withClusterSupervisor(dao, computeDAO, iamDAO, storageDAO, authProvider, mockJupyterDAO, true) { actor =>
+    withClusterSupervisor(gdDAO, computeDAO, iamDAO, storageDAO, authProvider, mockJupyterDAO, false) { actor =>
 
       eventually {
         val updatedCluster = dbFutureValue {
           _.clusterQuery.getActiveClusterByName(creatingCluster.googleProject, creatingCluster.clusterName)
         }
         updatedCluster shouldBe 'defined
-        updatedCluster shouldBe Some(savedCreatingCluster.copy(instances = Set(masterInstance, workerInstance1, workerInstance2)))
+        updatedCluster.map(_.status) shouldBe Some(ClusterStatus.Error)
+        updatedCluster.flatMap(_.dataprocInfo.hostIp) shouldBe None
+        updatedCluster.map(_.instances) shouldBe Some(Set(masterInstance, workerInstance1, workerInstance2))
       }
       verify(storageDAO, never).deleteBucket(any[GcsBucketName], any[Boolean])
-      verify(iamDAO, never()).removeIamRolesForUser(any[GoogleProject], any[WorkbenchEmail], mockitoEq(Set("roles/dataproc.worker")))
-      verify(iamDAO, never()).removeServiceAccountKey(any[GoogleProject], any[WorkbenchEmail], any[ServiceAccountKeyId])
+      verify(iamDAO, if (clusterServiceAccount(creatingCluster.googleProject).isDefined) times(1) else never()).removeIamRolesForUser(any[GoogleProject], any[WorkbenchEmail], mockitoEq(Set("roles/dataproc.worker")))
+      verify(iamDAO, if (notebookServiceAccount(creatingCluster.googleProject).isDefined) times(1) else never()).removeServiceAccountKey(any[GoogleProject], any[WorkbenchEmail], any[ServiceAccountKeyId])
     }
   }
 
+
   // Pre:
   // - cluster exists in the DB with status Creating
-  // - dataproc DAO returns status ERROR and error code CANCELLED
+  // - dataproc DAO returns status ERROR and no ERROR CODE
   // - compute DAO returns RUNNING
   // Post:
   // - cluster status is set to Error in the DB
   // - instances are populated in the DB
   // - monitor actor shuts down
-  it should "monitor until ERROR state with no restart" in isolatedDbTest {
+  it should "monitor until ERROR state with no restart and getClusterErrorDetails returns None " in isolatedDbTest {
     val savedCreatingCluster = creatingCluster.save()
     creatingCluster shouldEqual savedCreatingCluster
 
@@ -344,7 +355,7 @@ class ClusterMonitorSpec extends TestKit(ActorSystem("leonardotest")) with FlatS
 
     when {
       gdDAO.getClusterErrorDetails(mockitoEq(creatingCluster.dataprocInfo.operationName))
-    } thenReturn Future.successful(Some(ClusterErrorDetails(Code.CANCELLED.value, Some("test message"))))
+    } thenReturn Future.successful(None)
 
     when {
       gdDAO.deleteCluster(mockitoEq(creatingCluster.googleProject), mockitoEq(creatingCluster.clusterName))