3822

pending completion

Build # 3822

Build Type

Pull #360

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/a0c33eb9781de9aa89f232b550bc73b3be536418">a0c33eb97">&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/a0c33eb9781de9aa89f232b550bc73b3be536418">a0c33eb97</a><a href="https://github.com/broadinstitute/workbench-libs/commit/a0c33eb9781de9aa89f232b550bc73b3be536418">&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/a0c33eb9781de9aa89f232b550bc73b3be536418&quot;&gt;a0c33eb97&lt;/a&gt;&lt;a href=&quot;https://github.com/broadinstitute/workbench-libs/commit/a0c33eb9781de9aa89f232b550bc73b3be536418&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;htt... (continued)

Pull Request Pull Request #360: [IA-2224] Jc disk idempotency

Run Details

3 of 3 new or added lines in 1 file covered. (100.0%)

408 of 1828 relevant lines covered (22.32%)

0.22 hits per line

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

0.0

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

package org.broadinstitute.dsde.workbench.google2

import cats.effect.concurrent.Semaphore
import cats.effect.{Async, Blocker, ContextShift, Timer}
import cats.implicits._
import cats.mtl.ApplicativeAsk
import com.google.cloud.compute.v1.{
  Disk,
  DiskClient,
  DisksResizeRequest,
  Operation,
  ProjectZoneDiskName,
  ProjectZoneName
}
import fs2.Stream
import io.chrisdavenport.log4cats.StructuredLogger
import org.broadinstitute.dsde.workbench.RetryConfig
import org.broadinstitute.dsde.workbench.model.TraceId
import org.broadinstitute.dsde.workbench.model.google.GoogleProject
import org.broadinstitute.dsde.workbench.google2.util.RetryPredicates._

import scala.collection.JavaConverters._

private[google2] class GoogleDiskInterpreter[F[_]: StructuredLogger: Timer: ContextShift](
  diskClient: DiskClient,
  retryConfig: RetryConfig,
  blocker: Blocker,
  blockerBound: Semaphore[F]
)(implicit F: Async[F])
    extends GoogleDiskService[F] {

  override def createDisk(project: GoogleProject, zone: ZoneName, disk: Disk)(
    implicit ev: ApplicativeAsk[F, TraceId]
  ): F[Option[Operation]] = {
    val projectZone = ProjectZoneName.of(project.value, zone.value)
    retryF(
      recoverF(F.delay(diskClient.insertDisk(projectZone, disk)), whenStatusCode(409)),
      s"com.google.cloud.compute.v1DiskClient.insertDisk(${projectZone.toString}, ${disk.getName})"
    ).compile.lastOrError
  }

  def getDisk(project: GoogleProject, zone: ZoneName, diskName: DiskName)(
    implicit ev: ApplicativeAsk[F, TraceId]
  ): F[Option[Disk]] = {
    val projectZoneDiskName = ProjectZoneDiskName.of(diskName.value, project.value, zone.value)
    val fa = recoverF(F.delay(diskClient.getDisk(projectZoneDiskName)), whenStatusCode(404))

    ev.ask.flatMap(
      traceId =>
        withLogging(
          fa,
          Some(traceId),
          s"com.google.cloud.compute.v1.DiskClient.getDisk(${projectZoneDiskName.toString})"
        )
    )
  }

  override def deleteDisk(project: GoogleProject, zone: ZoneName, diskName: DiskName)(
    implicit ev: ApplicativeAsk[F, TraceId]
  ): F[Option[Operation]] = {
    val projectZoneDiskName = ProjectZoneDiskName.of(diskName.value, project.value, zone.value)
    val fa = recoverF(F.delay(diskClient.deleteDisk(projectZoneDiskName)), whenStatusCode(404))

    ev.ask.flatMap(
      traceId =>
        withLogging(
          fa,
          Some(traceId),
          s"com.google.cloud.compute.v1.DiskClient.deleteDisk(${projectZoneDiskName.toString})"
        )
    )
  }

  override def listDisks(project: GoogleProject, zone: ZoneName)(
    implicit ev: ApplicativeAsk[F, TraceId]
  ): Stream[F, Disk] = {
    val projectZone = ProjectZoneName.of(project.value, zone.value)
    for {
      pagedResults <- retryF(
        F.delay(diskClient.listDisks(projectZone)),
        s"com.google.cloud.compute.v1.DiskClient.listDisks(${projectZone.toString})"
      )

      res <- Stream.fromIterator[F](pagedResults.iterateAll().iterator().asScala)
    } yield res
  }

  override def resizeDisk(project: GoogleProject, zone: ZoneName, diskName: DiskName, newSizeGb: Int)(
    implicit ev: ApplicativeAsk[F, TraceId]
  ): F[Operation] = {
    val projectZoneDiskName = ProjectZoneDiskName.of(diskName.value, project.value, zone.value)
    val request = DisksResizeRequest.newBuilder().setSizeGb(newSizeGb.toString).build()
    retryF(
      F.delay(diskClient.resizeDisk(projectZoneDiskName, request)),
      s"com.google.cloud.compute.v1.DiskClient.resizeDisk(${projectZoneDiskName.toString}, $newSizeGb)"
    ).compile.lastOrError
  }

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

}

1	package org.broadinstitute.dsde.workbench.google2
2
3	import cats.effect.concurrent.Semaphore
4	import cats.effect.{Async, Blocker, ContextShift, Timer}
5	import cats.implicits._
6	import cats.mtl.ApplicativeAsk
7	import com.google.cloud.compute.v1.{
8	Disk,
9	DiskClient,
10	DisksResizeRequest,
11	Operation,
12	ProjectZoneDiskName,
13	ProjectZoneName
14	}
15	import fs2.Stream
16	import io.chrisdavenport.log4cats.StructuredLogger
17	import org.broadinstitute.dsde.workbench.RetryConfig
18	import org.broadinstitute.dsde.workbench.model.TraceId
19	import org.broadinstitute.dsde.workbench.model.google.GoogleProject
20	import org.broadinstitute.dsde.workbench.google2.util.RetryPredicates._
21
22	import scala.collection.JavaConverters._
23
24	private[google2] class GoogleDiskInterpreter[F[_]: StructuredLogger: Timer: ContextShift](
25	diskClient: DiskClient,
26	retryConfig: RetryConfig,
27	blocker: Blocker,
28	blockerBound: Semaphore[F]
29	)(implicit F: Async[F])
30	extends GoogleDiskService[F] {
31
32	override def createDisk(project: GoogleProject, zone: ZoneName, disk: Disk)(
33	implicit ev: ApplicativeAsk[F, TraceId]
34	): F[Option[Operation]] = {
35	val projectZone = ProjectZoneName.of(project.value, zone.value)	×
36	retryF(
37	recoverF(F.delay(diskClient.insertDisk(projectZone, disk)), whenStatusCode(409)),	×
38	s"com.google.cloud.compute.v1DiskClient.insertDisk(${projectZone.toString}, ${disk.getName})"	×
39	).compile.lastOrError	×
40	}
41
42	def getDisk(project: GoogleProject, zone: ZoneName, diskName: DiskName)(
43	implicit ev: ApplicativeAsk[F, TraceId]
44	): F[Option[Disk]] = {
45	val projectZoneDiskName = ProjectZoneDiskName.of(diskName.value, project.value, zone.value)	×
46	val fa = recoverF(F.delay(diskClient.getDisk(projectZoneDiskName)), whenStatusCode(404))	×
47
48	ev.ask.flatMap(	×
49	traceId =>
50	withLogging(	×
51	fa,
52	Some(traceId),	×
53	s"com.google.cloud.compute.v1.DiskClient.getDisk(${projectZoneDiskName.toString})"	×
54	)
55	)
56	}
57
58	override def deleteDisk(project: GoogleProject, zone: ZoneName, diskName: DiskName)(
59	implicit ev: ApplicativeAsk[F, TraceId]
60	): F[Option[Operation]] = {
61	val projectZoneDiskName = ProjectZoneDiskName.of(diskName.value, project.value, zone.value)	×
62	val fa = recoverF(F.delay(diskClient.deleteDisk(projectZoneDiskName)), whenStatusCode(404))	×
63
64	ev.ask.flatMap(	×
65	traceId =>
66	withLogging(	×
67	fa,
68	Some(traceId),	×
69	s"com.google.cloud.compute.v1.DiskClient.deleteDisk(${projectZoneDiskName.toString})"	×
70	)
71	)
72	}
73
74	override def listDisks(project: GoogleProject, zone: ZoneName)(
75	implicit ev: ApplicativeAsk[F, TraceId]
76	): Stream[F, Disk] = {
77	val projectZone = ProjectZoneName.of(project.value, zone.value)	×
78	for {
79	pagedResults <- retryF(	×
80	F.delay(diskClient.listDisks(projectZone)),	×
81	s"com.google.cloud.compute.v1.DiskClient.listDisks(${projectZone.toString})"	×
82	)
83
84	res <- Stream.fromIterator[F](pagedResults.iterateAll().iterator().asScala)	×
85	} yield res
86	}
87
88	override def resizeDisk(project: GoogleProject, zone: ZoneName, diskName: DiskName, newSizeGb: Int)(
89	implicit ev: ApplicativeAsk[F, TraceId]
90	): F[Operation] = {
91	val projectZoneDiskName = ProjectZoneDiskName.of(diskName.value, project.value, zone.value)	×
92	val request = DisksResizeRequest.newBuilder().setSizeGb(newSizeGb.toString).build()	×
93	retryF(
94	F.delay(diskClient.resizeDisk(projectZoneDiskName, request)),	×
95	s"com.google.cloud.compute.v1.DiskClient.resizeDisk(${projectZoneDiskName.toString}, $newSizeGb)"	×
96	).compile.lastOrError	×
97	}
98
99	private def retryF[A](fa: F[A], loggingMsg: String)(implicit ev: ApplicativeAsk[F, TraceId]): Stream[F, A] =
100	tracedRetryGoogleF(retryConfig)(blockerBound.withPermit(blocker.blockOn(fa)), loggingMsg)	×
101
102	}

broadinstitute / workbench-libs / 3822

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