3919

pending completion

Build # 3919

Build Type

Pull #373

travis-ci-com

Committed by

web-flow

Commit Message

<a href="https://github.com/broadinstitute/workbench-libs/commit/<a class=hub.com/broadinstitute/workbench-libs/commit/<a class="double-link" href="https://git"><a class=hub.com/broadinstitute/workbench-libs/commit/<a class="double-link" href="https://git"><a class=hub.com/broadinstitute/workbench-libs/commit/<a class="double-link" href="https://git"><a class=hub.com/broadinstitute/workbench-libs/commit/4a1201d573c4732a426e60ce991a89239b9c18a9">4a1201d57">&lt;a href=&quot;https://github.com/broadinstitute/workbench-libs/commit/</a><a class="double-link" href="https://github.com/broadinstitute/workbench-libs/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/broadinstitute/workbench-libs/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/broadinstitute/workbench-libs/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/broadinstitute/workbench-libs/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/broadinstitute/workbench-libs/commit/4a1201d573c4732a426e60ce991a89239b9c18a9">4a1201d57</a><a href="https://github.com/broadinstitute/workbench-libs/commit/4a1201d573c4732a426e60ce991a89239b9c18a9">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/broadinstitute/workbench-libs/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/broadinstitute/workbench-libs/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/broadinstitute/workbench-libs/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/broadinstitute/workbench-libs/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/broadinstitute/workbench-libs/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/broadinstitute/workbench-libs/commit/4a1201d573c4732a426e60ce991a89239b9c18a9&quot;&gt;4a1201d57&lt;/a&gt;&lt;a href=&quot;https://github.com/broadinstitute/workbench-libs/commit/4a1201d573c4732a426e60ce991a89239b9c18a9&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;htt... (continued)

Pull Request Pull Request #373: [IA-2350] [google2] Ability to resize and stop Dataproc clusters

Run Details

42 of 42 new or added lines in 2 files covered. (100.0%)

395 of 1793 relevant lines covered (22.03%)

0.22 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

10.31

/google2/src/main/scala/org/broadinstitute/dsde/workbench/google2/GoogleDataprocInterpreter.scala

package org.broadinstitute.dsde.workbench.google2

import cats.{Parallel, Show}
import cats.effect._
import cats.effect.concurrent.Semaphore
import cats.implicits._
import cats.mtl.Ask
import com.google.api.core.ApiFutures
import com.google.api.gax.rpc.StatusCode.Code
import com.google.cloud.dataproc.v1.{RegionName => _, _}
import com.google.common.util.concurrent.MoreExecutors
import com.google.protobuf.FieldMask
import io.chrisdavenport.log4cats.StructuredLogger
import org.broadinstitute.dsde.workbench.RetryConfig
import org.broadinstitute.dsde.workbench.google2.DataprocRole.Master
import org.broadinstitute.dsde.workbench.google2.util.RetryPredicates._
import org.broadinstitute.dsde.workbench.model.TraceId
import org.broadinstitute.dsde.workbench.model.google.GoogleProject
import org.broadinstitute.dsde.workbench.google2.GoogleDataprocInterpreter._

import scala.collection.JavaConverters._

private[google2] class GoogleDataprocInterpreter[F[_]: StructuredLogger: Timer: Parallel: ContextShift](
  clusterControllerClient: ClusterControllerClient,
  googleComputeService: GoogleComputeService[F],
  blocker: Blocker,
  blockerBound: Semaphore[F],
  retryConfig: RetryConfig
)(implicit F: Async[F])
    extends GoogleDataprocService[F] {

  override def createCluster(
    project: GoogleProject,
    region: RegionName,
    clusterName: DataprocClusterName,
    createClusterConfig: Option[CreateClusterConfig]
  )(implicit ev: Ask[F, TraceId]): F[CreateClusterResponse] = {
    val config: ClusterConfig = createClusterConfig
      .map(
        config =>
          ClusterConfig.newBuilder
            .setGceClusterConfig(config.gceClusterConfig)
            .setInitializationActions(0, config.nodeInitializationAction)
            .setMasterConfig(config.instanceGroupConfig)
            .setConfigBucket(config.stagingBucket.value)
            .setSoftwareConfig(config.softwareConfig)
            .build()
      )
      .getOrElse(ClusterConfig.newBuilder.build())

    val cluster = Cluster
      .newBuilder()
      .setClusterName(clusterName.value)
      .setConfig(config)
      .build()

    val request = CreateClusterRequest
      .newBuilder()
      .setCluster(cluster)
      .setRegion(region.value)
      .setProjectId(project.value)
      .build()

    val createCluster = Async[F].async[ClusterOperationMetadata] { cb =>
      ApiFutures.addCallback(
        clusterControllerClient.createClusterAsync(request).getMetadata,
        callBack(cb),
        MoreExecutors.directExecutor()
      )
    }

    for {
      createCluster <- retryF(
        createCluster,
        s"com.google.cloud.dataproc.v1.ClusterControllerClient.createClusterAsync(${region}, ${clusterName}, ${createClusterConfig})"
      ).attempt

      result <- createCluster match {
        case Left(e: com.google.api.gax.rpc.ApiException) =>
          if (e.getStatusCode.getCode == Code.ALREADY_EXISTS)
            Async[F].pure(CreateClusterResponse.AlreadyExists: CreateClusterResponse)
          else
            Async[F].raiseError(e): F[CreateClusterResponse]
        case Left(e)  => Async[F].raiseError(e): F[CreateClusterResponse]
        case Right(v) => Async[F].pure(CreateClusterResponse.Success(v): CreateClusterResponse)
      }
    } yield result
  }

  /**
   * Strictly speaking, it is not possible to 'stop' a Dataproc cluster altogether.
   * Instead, we approximate by:
   *   1. removing pre-emptible instances (if any) by resizing the cluster, since they would not be possible to restart
   *   2. stopping underlying nodes individually
   */
  override def stopCluster(project: GoogleProject,
                           region: RegionName,
                           clusterName: DataprocClusterName,
                           instances: Set[DataprocInstance],
                           numPreemptibles: Option[Int],
                           metadata: Option[Map[String, String]])(
    implicit ev: Ask[F, TraceId]
  ): F[Option[ClusterOperationMetadata]] =
    for {
      // First, remove all its preemptible instances, if any
      _ <- if (numPreemptibles.exists(_ > 0))
        resizeCluster(project, region, clusterName, numWorkers = None, numPreemptibles = Some(0))
      else F.pure(none[ClusterOperationMetadata])

      // Then, stop each instance individually
      _ <- instances.toList.parTraverse { instance =>
        instance.dataprocRole match {
          case Master =>
            googleComputeService.addInstanceMetadata(
              instance.project,
              instance.zone,
              instance.name,
              metadata.get
            ) >> googleComputeService.stopInstance(instance.project, instance.zone, instance.name)
          case _ =>
            googleComputeService.stopInstance(instance.project, instance.zone, instance.name)
        }
      }
    } yield None

  override def resizeCluster(project: GoogleProject,
                             region: RegionName,
                             clusterName: DataprocClusterName,
                             numWorkers: Option[Int],
                             numPreemptibles: Option[Int])(
    implicit ev: Ask[F, TraceId]
  ): F[Option[ClusterOperationMetadata]] = {
    val workerMask = "config.worker_config.num_instances"
    val preemptibleMask = "config.secondary_worker_config.num_instances"

    val configAndMask = (numWorkers, numPreemptibles) match {
      case (Some(nw), Some(np)) =>
        val mask = FieldMask.newBuilder
          .addAllPaths(List(workerMask, preemptibleMask).asJava)
        val config = ClusterConfig.newBuilder
          .setWorkerConfig(InstanceGroupConfig.newBuilder.setNumInstances(nw))
          .setSecondaryWorkerConfig(InstanceGroupConfig.newBuilder.setNumInstances(np))
        Some((config, mask))

      case (Some(nw), None) =>
        val mask = FieldMask.newBuilder
          .addPaths(workerMask)
        val config = ClusterConfig.newBuilder
          .setWorkerConfig(InstanceGroupConfig.newBuilder.setNumInstances(nw))
        Some((config, mask))

      case (None, Some(np)) =>
        val mask = FieldMask.newBuilder
          .addPaths(preemptibleMask)
        val config = ClusterConfig.newBuilder
          .setSecondaryWorkerConfig(InstanceGroupConfig.newBuilder.setNumInstances(np))
        Some((config, mask))

      case (None, None) =>
        None
    }

    val updateClusterRequest = configAndMask.map {
      case (config, mask) =>
        val cluster = Cluster
          .newBuilder()
          .setClusterName(clusterName.value)
          .setConfig(config)
          .build()

        UpdateClusterRequest
          .newBuilder()
          .setClusterName(clusterName.value)
          .setCluster(cluster)
          .setRegion(region.value)
          .setProjectId(project.value)
          .setUpdateMask(mask)
          .build()
    }

    val updateCluster = updateClusterRequest
      .traverse { request =>
        Async[F]
          .async[ClusterOperationMetadata] { cb =>
            ApiFutures.addCallback(
              clusterControllerClient.updateClusterAsync(request).getMetadata,
              callBack(cb),
              MoreExecutors.directExecutor()
            )
          }
      }
      .handleErrorWith {
        case _: com.google.api.gax.rpc.NotFoundException => F.pure(none[ClusterOperationMetadata])
        case e                                           => F.raiseError[Option[ClusterOperationMetadata]](e)
      }

    for {
      traceId <- ev.ask
      res <- withLogging(
        updateCluster,
        Some(traceId),
        s"com.google.cloud.dataproc.v1.ClusterControllerClient.updateClusterAsync(${updateClusterRequest})"
      )
    } yield res
  }

  override def deleteCluster(project: GoogleProject, region: RegionName, clusterName: DataprocClusterName)(
    implicit ev: Ask[F, TraceId]
  ): F[Option[ClusterOperationMetadata]] = {
    val request = DeleteClusterRequest
      .newBuilder()
      .setRegion(region.value)
      .setProjectId(project.value)
      .setClusterName(clusterName.value)
      .build()

    val deleteCluster = Async[F]
      .async[ClusterOperationMetadata] { cb =>
        ApiFutures.addCallback(
          clusterControllerClient.deleteClusterAsync(request).getMetadata,
          callBack(cb),
          MoreExecutors.directExecutor()
        )
      }
      .map(Option(_))
      .handleErrorWith {
        case _: com.google.api.gax.rpc.NotFoundException => F.pure(none[ClusterOperationMetadata])
        case e                                           => F.raiseError[Option[ClusterOperationMetadata]](e)
      }

    for {
      traceId <- ev.ask
      res <- withLogging(
        deleteCluster,
        Some(traceId),
        s"com.google.cloud.dataproc.v1.ClusterControllerClient.deleteClusterAsync(${request})"
      )
    } yield res
  }

  override def getCluster(project: GoogleProject, region: RegionName, clusterName: DataprocClusterName)(
    implicit ev: Ask[F, TraceId]
  ): F[Option[Cluster]] = {
    val fa =
      F.delay(clusterControllerClient.getCluster(project.value, region.value, clusterName.value))
        .map(Option(_))
        .handleErrorWith {
          case _: com.google.api.gax.rpc.NotFoundException => F.pure(none[Cluster])
          case e                                           => F.raiseError[Option[Cluster]](e)
        }

    ev.ask
      .flatMap { traceId =>
        withLogging(
          fa,
          Some(traceId),
          s"com.google.cloud.dataproc.v1.ClusterControllerClient.getCluster(${project.value}, ${region.value}, ${clusterName.value})",
          Show.show[Option[Cluster]](c => s"${c.map(_.getStatus.toString).getOrElse("Not found")}")
        )
      }
  }

  override def getClusterInstances(project: GoogleProject, region: RegionName, clusterName: DataprocClusterName)(
    implicit ev: Ask[F, TraceId]
  ): F[Map[DataprocRole, Set[InstanceName]]] =
    for {
      cluster <- getCluster(project, region, clusterName)
    } yield cluster.map(c => getAllInstanceNames(c)).getOrElse(Map.empty)

  override def getClusterError(
    operationName: OperationName
  )(implicit ev: Ask[F, TraceId]): F[Option[ClusterError]] =
    for {
      error <- retryF(
        recoverF(
          Async[F].delay(clusterControllerClient.getOperationsClient().getOperation(operationName.value).getError()),
          whenStatusCode(404)
        ),
        s"com.google.cloud.dataproc.v1.ClusterControllerClient.getOperationsClient.getOperation(${operationName.value}).getError()"
      )
    } yield error.map(e => ClusterError(e.getCode, e.getMessage))

  private def retryF[A](fa: F[A], loggingMsg: String)(implicit ev: Ask[F, TraceId]): F[A] =
    tracedRetryGoogleF(retryConfig)(blockerBound.withPermit(blocker.blockOn(fa)), loggingMsg).compile.lastOrError
}

object GoogleDataprocInterpreter {
  // WARNING: Be very careful refactoring this function and make sure you test this out in console.
  // Incorrectness in this function can cause leonardo fail to stop all instances for a Dataproc cluster, which
  // incurs compute cost for users
  def getAllInstanceNames(cluster: Cluster): Map[DataprocRole, Set[InstanceName]] = {
    def getFromGroup(role: DataprocRole, group: InstanceGroupConfig): Map[DataprocRole, Set[InstanceName]] = {
      val instances = group.getInstanceNamesList
        .asByteStringList()
        .asScala
        .toList
        .map(byteString => InstanceName(byteString.toStringUtf8))
        .toSet

      if (instances.isEmpty) Map.empty else Map(role -> instances)
    }

    val res = Option(cluster.getConfig).map { config =>
      val master =
        Option(config.getMasterConfig).map(config => getFromGroup(DataprocRole.Master, config)).getOrElse(Map.empty)
      val workers =
        Option(config.getWorkerConfig).map(config => getFromGroup(DataprocRole.Worker, config)).getOrElse(Map.empty)
      val secondaryWorkers = Option(config.getSecondaryWorkerConfig)
        .map(config => getFromGroup(DataprocRole.SecondaryWorker, config))
        .getOrElse(Map.empty)

      master ++ workers ++ secondaryWorkers
    }

    res.getOrElse(Map.empty)
  }
}

1	package org.broadinstitute.dsde.workbench.google2
2
3	import cats.{Parallel, Show}
4	import cats.effect._
5	import cats.effect.concurrent.Semaphore
6	import cats.implicits._
7	import cats.mtl.Ask
8	import com.google.api.core.ApiFutures
9	import com.google.api.gax.rpc.StatusCode.Code
10	import com.google.cloud.dataproc.v1.{RegionName => _, _}
11	import com.google.common.util.concurrent.MoreExecutors
12	import com.google.protobuf.FieldMask
13	import io.chrisdavenport.log4cats.StructuredLogger
14	import org.broadinstitute.dsde.workbench.RetryConfig
15	import org.broadinstitute.dsde.workbench.google2.DataprocRole.Master
16	import org.broadinstitute.dsde.workbench.google2.util.RetryPredicates._
17	import org.broadinstitute.dsde.workbench.model.TraceId
18	import org.broadinstitute.dsde.workbench.model.google.GoogleProject
19	import org.broadinstitute.dsde.workbench.google2.GoogleDataprocInterpreter._
20
21	import scala.collection.JavaConverters._
22
23	private[google2] class GoogleDataprocInterpreter[F[_]: StructuredLogger: Timer: Parallel: ContextShift](
24	clusterControllerClient: ClusterControllerClient,
25	googleComputeService: GoogleComputeService[F],
26	blocker: Blocker,
27	blockerBound: Semaphore[F],
28	retryConfig: RetryConfig
29	)(implicit F: Async[F])
30	extends GoogleDataprocService[F] {
31
32	override def createCluster(
33	project: GoogleProject,
34	region: RegionName,
35	clusterName: DataprocClusterName,
36	createClusterConfig: Option[CreateClusterConfig]
37	)(implicit ev: Ask[F, TraceId]): F[CreateClusterResponse] = {
38	val config: ClusterConfig = createClusterConfig
39	.map(
40	config =>
41	ClusterConfig.newBuilder
42	.setGceClusterConfig(config.gceClusterConfig)
43	.setInitializationActions(0, config.nodeInitializationAction)
44	.setMasterConfig(config.instanceGroupConfig)
45	.setConfigBucket(config.stagingBucket.value)
46	.setSoftwareConfig(config.softwareConfig)
47	.build()	×
48	)
49	.getOrElse(ClusterConfig.newBuilder.build())	×
50
51	val cluster = Cluster
52	.newBuilder()
53	.setClusterName(clusterName.value)
54	.setConfig(config)
55	.build()	×
56
57	val request = CreateClusterRequest
58	.newBuilder()
59	.setCluster(cluster)
60	.setRegion(region.value)
61	.setProjectId(project.value)
62	.build()	×
63
64	val createCluster = Async[F].async[ClusterOperationMetadata] { cb =>	×
65	ApiFutures.addCallback(	×
66	clusterControllerClient.createClusterAsync(request).getMetadata,	×
67	callBack(cb),	×
68	MoreExecutors.directExecutor()	×
69	)
70	}
71
72	for {
73	createCluster <- retryF(	×
74	createCluster,
75	s"com.google.cloud.dataproc.v1.ClusterControllerClient.createClusterAsync(${region}, ${clusterName}, ${createClusterConfig})"
76	).attempt	×
77
78	result <- createCluster match {	×
79	case Left(e: com.google.api.gax.rpc.ApiException) =>
80	if (e.getStatusCode.getCode == Code.ALREADY_EXISTS)	×
81	Async[F].pure(CreateClusterResponse.AlreadyExists: CreateClusterResponse)	×
82	else
83	Async[F].raiseError(e): F[CreateClusterResponse]	×
84	case Left(e) => Async[F].raiseError(e): F[CreateClusterResponse]	×
85	case Right(v) => Async[F].pure(CreateClusterResponse.Success(v): CreateClusterResponse)	×
86	}
87	} yield result
88	}
89
90	/**
91	* Strictly speaking, it is not possible to 'stop' a Dataproc cluster altogether.
92	* Instead, we approximate by:
93	* 1. removing pre-emptible instances (if any) by resizing the cluster, since they would not be possible to restart
94	* 2. stopping underlying nodes individually
95	*/
96	override def stopCluster(project: GoogleProject,
97	region: RegionName,
98	clusterName: DataprocClusterName,
99	instances: Set[DataprocInstance],
100	numPreemptibles: Option[Int],
101	metadata: Option[Map[String, String]])(
102	implicit ev: Ask[F, TraceId]
103	): F[Option[ClusterOperationMetadata]] =
104	for {
105	// First, remove all its preemptible instances, if any
106	_ <- if (numPreemptibles.exists(_ > 0))	×
107	resizeCluster(project, region, clusterName, numWorkers = None, numPreemptibles = Some(0))	×
108	else F.pure(none[ClusterOperationMetadata])	×
109
110	// Then, stop each instance individually
111	_ <- instances.toList.parTraverse { instance =>	×
112	instance.dataprocRole match {	×
113	case Master =>
114	googleComputeService.addInstanceMetadata(	×
115	instance.project,
116	instance.zone,
117	instance.name,
118	metadata.get
119	) >> googleComputeService.stopInstance(instance.project, instance.zone, instance.name)	×
120	case _ =>
121	googleComputeService.stopInstance(instance.project, instance.zone, instance.name)	×
122	}
123	}
124	} yield None	×
125
126	override def resizeCluster(project: GoogleProject,
127	region: RegionName,
128	clusterName: DataprocClusterName,
129	numWorkers: Option[Int],
130	numPreemptibles: Option[Int])(
131	implicit ev: Ask[F, TraceId]
132	): F[Option[ClusterOperationMetadata]] = {
133	val workerMask = "config.worker_config.num_instances"	×
134	val preemptibleMask = "config.secondary_worker_config.num_instances"	×
135
136	val configAndMask = (numWorkers, numPreemptibles) match {
137	case (Some(nw), Some(np)) =>
138	val mask = FieldMask.newBuilder
139	.addAllPaths(List(workerMask, preemptibleMask).asJava)	×
140	val config = ClusterConfig.newBuilder
141	.setWorkerConfig(InstanceGroupConfig.newBuilder.setNumInstances(nw))	×
142	.setSecondaryWorkerConfig(InstanceGroupConfig.newBuilder.setNumInstances(np))	×
143	Some((config, mask))	×
144
145	case (Some(nw), None) =>
146	val mask = FieldMask.newBuilder
147	.addPaths(workerMask)	×
148	val config = ClusterConfig.newBuilder
149	.setWorkerConfig(InstanceGroupConfig.newBuilder.setNumInstances(nw))	×
150	Some((config, mask))	×
151
152	case (None, Some(np)) =>
153	val mask = FieldMask.newBuilder
154	.addPaths(preemptibleMask)	×
155	val config = ClusterConfig.newBuilder
156	.setSecondaryWorkerConfig(InstanceGroupConfig.newBuilder.setNumInstances(np))	×
157	Some((config, mask))	×
158
159	case (None, None) =>
160	None	×
161	}
162
163	val updateClusterRequest = configAndMask.map {	×
164	case (config, mask) =>
165	val cluster = Cluster
166	.newBuilder()
167	.setClusterName(clusterName.value)
168	.setConfig(config)
169	.build()	×
170
171	UpdateClusterRequest
172	.newBuilder()
173	.setClusterName(clusterName.value)
174	.setCluster(cluster)
175	.setRegion(region.value)
176	.setProjectId(project.value)
177	.setUpdateMask(mask)
178	.build()	×
179	}
180
181	val updateCluster = updateClusterRequest	×
182	.traverse { request =>	×
183	Async[F]	×
184	.async[ClusterOperationMetadata] { cb =>	×
185	ApiFutures.addCallback(	×
186	clusterControllerClient.updateClusterAsync(request).getMetadata,	×
187	callBack(cb),	×
188	MoreExecutors.directExecutor()	×
189	)
190	}
191	}
192	.handleErrorWith {	×
193	case _: com.google.api.gax.rpc.NotFoundException => F.pure(none[ClusterOperationMetadata])	×
194	case e => F.raiseError[Option[ClusterOperationMetadata]](e)	×
195	}
196
197	for {
198	traceId <- ev.ask	×
199	res <- withLogging(	×
200	updateCluster,
201	Some(traceId),	×
202	s"com.google.cloud.dataproc.v1.ClusterControllerClient.updateClusterAsync(${updateClusterRequest})"
203	)
204	} yield res
205	}
206
207	override def deleteCluster(project: GoogleProject, region: RegionName, clusterName: DataprocClusterName)(
208	implicit ev: Ask[F, TraceId]
209	): F[Option[ClusterOperationMetadata]] = {
210	val request = DeleteClusterRequest
211	.newBuilder()
212	.setRegion(region.value)
213	.setProjectId(project.value)
214	.setClusterName(clusterName.value)
215	.build()	×
216
217	val deleteCluster = Async[F]	×
218	.async[ClusterOperationMetadata] { cb =>	×
219	ApiFutures.addCallback(	×
220	clusterControllerClient.deleteClusterAsync(request).getMetadata,	×
221	callBack(cb),	×
222	MoreExecutors.directExecutor()	×
223	)
224	}
225	.map(Option(_))	×
226	.handleErrorWith {	×
227	case _: com.google.api.gax.rpc.NotFoundException => F.pure(none[ClusterOperationMetadata])	×
228	case e => F.raiseError[Option[ClusterOperationMetadata]](e)	×
229	}
230
231	for {
232	traceId <- ev.ask	×
233	res <- withLogging(	×
234	deleteCluster,
235	Some(traceId),	×
236	s"com.google.cloud.dataproc.v1.ClusterControllerClient.deleteClusterAsync(${request})"
237	)
238	} yield res
239	}
240
241	override def getCluster(project: GoogleProject, region: RegionName, clusterName: DataprocClusterName)(
242	implicit ev: Ask[F, TraceId]
243	): F[Option[Cluster]] = {
244	val fa =
245	F.delay(clusterControllerClient.getCluster(project.value, region.value, clusterName.value))	×
246	.map(Option(_))	×
247	.handleErrorWith {	×
248	case _: com.google.api.gax.rpc.NotFoundException => F.pure(none[Cluster])	×
249	case e => F.raiseError[Option[Cluster]](e)	×
250	}
251
252	ev.ask	×
253	.flatMap { traceId =>	×
254	withLogging(	×
255	fa,
256	Some(traceId),	×
257	s"com.google.cloud.dataproc.v1.ClusterControllerClient.getCluster(${project.value}, ${region.value}, ${clusterName.value})",
258	Show.show[Option[Cluster]](c => s"${c.map(_.getStatus.toString).getOrElse("Not found")}")	×
259	)
260	}
261	}
262
263	override def getClusterInstances(project: GoogleProject, region: RegionName, clusterName: DataprocClusterName)(
264	implicit ev: Ask[F, TraceId]
265	): F[Map[DataprocRole, Set[InstanceName]]] =
266	for {
267	cluster <- getCluster(project, region, clusterName)	×
268	} yield cluster.map(c => getAllInstanceNames(c)).getOrElse(Map.empty)	×
269
270	override def getClusterError(
271	operationName: OperationName
272	)(implicit ev: Ask[F, TraceId]): F[Option[ClusterError]] =
273	for {
274	error <- retryF(	×
275	recoverF(
276	Async[F].delay(clusterControllerClient.getOperationsClient().getOperation(operationName.value).getError()),
277	whenStatusCode(404)
278	),
279	s"com.google.cloud.dataproc.v1.ClusterControllerClient.getOperationsClient.getOperation(${operationName.value}).getError()"
280	)
281	} yield error.map(e => ClusterError(e.getCode, e.getMessage))	×
282
283	private def retryF[A](fa: F[A], loggingMsg: String)(implicit ev: Ask[F, TraceId]): F[A] =
284	tracedRetryGoogleF(retryConfig)(blockerBound.withPermit(blocker.blockOn(fa)), loggingMsg).compile.lastOrError	×
285	}
286
287	object GoogleDataprocInterpreter {
288	// WARNING: Be very careful refactoring this function and make sure you test this out in console.
289	// Incorrectness in this function can cause leonardo fail to stop all instances for a Dataproc cluster, which
290	// incurs compute cost for users
291	def getAllInstanceNames(cluster: Cluster): Map[DataprocRole, Set[InstanceName]] = {
292	def getFromGroup(role: DataprocRole, group: InstanceGroupConfig): Map[DataprocRole, Set[InstanceName]] = {
293	val instances = group.getInstanceNamesList
294	.asByteStringList()	1✔
295	.asScala
296	.toList
297	.map(byteString => InstanceName(byteString.toStringUtf8))	1✔
298	.toSet	1✔
299
300	if (instances.isEmpty) Map.empty else Map(role -> instances)	1✔
301	}
302
303	val res = Option(cluster.getConfig).map { config =>	1✔
304	val master =
305	Option(config.getMasterConfig).map(config => getFromGroup(DataprocRole.Master, config)).getOrElse(Map.empty)	1✔
306	val workers =
307	Option(config.getWorkerConfig).map(config => getFromGroup(DataprocRole.Worker, config)).getOrElse(Map.empty)	1✔
308	val secondaryWorkers = Option(config.getSecondaryWorkerConfig)	1✔
309	.map(config => getFromGroup(DataprocRole.SecondaryWorker, config))	1✔
310	.getOrElse(Map.empty)	×
311
312	master ++ workers ++ secondaryWorkers	1✔
313	}
314
315	res.getOrElse(Map.empty)	×
316	}
317	}

broadinstitute / workbench-libs / 3919

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous