3662

pending completion

Build # 3662

Build Type

Pull #346

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

Pull Request Pull Request #346: [IA-2072] Add KubernetesService.getServiceExternalIp method

Run Details

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

420 of 1795 relevant lines covered (23.4%)

0.23 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/KubernetesInterpreter.scala

package org.broadinstitute.dsde.workbench.google2

import java.io.ByteArrayInputStream
import java.util.UUID
import java.util.concurrent.TimeUnit

import cats.effect.concurrent.Semaphore
import cats.effect.implicits._
import cats.effect.{Async, Blocker, ContextShift, Effect, Timer}
import cats.implicits._
import cats.mtl.ApplicativeAsk
import com.google.auth.oauth2.{AccessToken, GoogleCredentials}
import com.google.common.cache.{CacheBuilder, CacheLoader}
import com.google.container.v1.Cluster
import io.chrisdavenport.log4cats.StructuredLogger
import io.kubernetes.client.ApiClient
import io.kubernetes.client.apis.{CoreV1Api, RbacAuthorizationV1Api}
import io.kubernetes.client.util.Config
import org.broadinstitute.dsde.workbench.google2.GKEModels.KubernetesClusterId
import org.broadinstitute.dsde.workbench.google2.JavaSerializableInstances._
import org.broadinstitute.dsde.workbench.google2.JavaSerializableSyntax._
import org.broadinstitute.dsde.workbench.google2.KubernetesModels._
import org.broadinstitute.dsde.workbench.google2.KubernetesSerializableName.{PodName, ServiceName}
import org.broadinstitute.dsde.workbench.model.{IP, TraceId}

import scala.collection.JavaConverters._

// This uses a kubernetes client library to make calls to the kubernetes API. The client library is autogenerated from the kubernetes API.
// It is highly recommended to use the kubernetes API docs here https://kubernetes.io/docs/reference/generated/kubernetes-api as opposed to the client library docs

class KubernetesInterpreter[F[_]: StructuredLogger: Effect: Timer: ContextShift](
  credentials: GoogleCredentials,
  gkeService: GKEService[F],
  blocker: Blocker,
  blockerBound: Semaphore[F]
)(implicit F: Async[F])
    extends KubernetesService[F] {

  //We cache a kubernetes client for each cluster
  val cache = CacheBuilder
    .newBuilder()
    // We expect calls to be batched, such as when a user's environment within a cluster is created/deleted/stopped.
    // TODO: Unhardcode expiration time
    .expireAfterWrite(2, TimeUnit.HOURS)
    .build(
      new CacheLoader[KubernetesClusterId, ApiClient] {
        def load(clusterId: KubernetesClusterId): ApiClient = {
          //we do not want to have to specify this at resource (class) creation time, so we create one on each load here
          implicit val traceId = ApplicativeAsk.const[F, TraceId](TraceId(UUID.randomUUID()))
          val res = for {
            _ <- StructuredLogger[F]
              .info(s"Determined that there is no cached client for kubernetes cluster ${clusterId}. Creating a client")
            clusterOpt <- gkeService.getCluster(clusterId)
            cluster <- F.fromEither(
              clusterOpt.toRight(
                KubernetesClusterNotFoundException(
                  s"Could not create client for cluster ${clusterId} because it does not exist in google"
                )
              )
            )
            token <- getToken
            client <- createClient(
              cluster,
              token
            )
          } yield client

          res.toIO.unsafeRunSync()
        }
      }
    )

  // https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.12/#podspec-v1-core
  override def createPod(clusterId: KubernetesClusterId, pod: KubernetesPod, namespace: KubernetesNamespace)(
    implicit ev: ApplicativeAsk[F, TraceId]
  ): F[Unit] =
    for {
      traceId <- ev.ask
      client <- blockingF(getClient(clusterId, new CoreV1Api(_)))
      call = blockingF(
        F.delay(
          client.createNamespacedPod(namespace.name.value, pod.getJavaSerialization, null, "true", null)
        )
      )
      _ <- withLogging(
        call,
        Some(traceId),
        s"io.kubernetes.client.apis.CoreV1Api.createNamespacedPod(${namespace.name.value}, ${pod.name.value}, null, true, null)"
      )
    } yield ()

  override def listPodStatus(clusterId: KubernetesClusterId, namespace: KubernetesNamespace)(
    implicit ev: ApplicativeAsk[F, TraceId]
  ): F[List[KubernetesPodStatus]] =
    for {
      traceId <- ev.ask
      client <- blockingF(getClient(clusterId, new CoreV1Api(_)))
      call = blockingF(
        F.delay(
          client.listNamespacedPod(namespace.name.value, null, "true", null, null, null, null, null, null, null)
        )
      )
      response <- withLogging(
        call,
        Some(traceId),
        s"io.kubernetes.client.apis.CoreV1Api.listNamespacedPod(${namespace.name.value}, null, true, null, null, null, null, null, null, null)"
      )

      listPodStatus = Option(response.getItems)
        .map { l =>
          l.asScala.toList.foldMap(
            v1Pod =>
              PodStatus.stringToPodStatus
                .get(v1Pod.getStatus.getPhase)
                .map(s => List(KubernetesPodStatus(PodName(v1Pod.getMetadata.getName), s)))
                .toRight(new RuntimeException(s"Unknown Google status ${v1Pod.getStatus.getPhase}"))
          )
        }
        .getOrElse(Nil.asRight[RuntimeException])

      res <- F.fromEither(listPodStatus)
    } yield res

  // Why we use a service over a deployment: https://matthewpalmer.net/kubernetes-app-developer/articles/service-kubernetes-example-tutorial.html
  // Services can be applied to pods/containers, while deployments are for pre-creating pods/containers.
  override def createService(clusterId: KubernetesClusterId,
                             service: KubernetesServiceKind,
                             namespace: KubernetesNamespace)(
    implicit ev: ApplicativeAsk[F, TraceId]
  ): F[Unit] =
    for {
      traceId <- ev.ask
      client <- blockingF(getClient(clusterId, new CoreV1Api(_)))
      call = blockingF(
        F.delay(
          client.createNamespacedService(namespace.name.value, service.getJavaSerialization, null, "true", null)
        )
      )
      _ <- withLogging(
        call,
        Some(traceId),
        s"io.kubernetes.client.apis.CoreV1Api.createNamespacedService(${namespace.name.value}, ${service.serviceName.value}, null, true, null)"
      )
    } yield ()

  override def getServiceExternalIp(clusterId: KubernetesClusterId,
                                    namespace: KubernetesNamespace,
                                    serviceName: ServiceName)(
    implicit ev: ApplicativeAsk[F, TraceId]
  ): F[Option[IP]] =
    for {
      traceId <- ev.ask
      client <- blockingF(getClient(clusterId, new CoreV1Api(_)))
      call = blockingF(
        F.delay(
          client.listNamespacedService(namespace.name.value, null, "true", null, null, null, null, null, null, null)
        )
      ).map(Option(_)).handleErrorWith {
        case e: io.kubernetes.client.ApiException if e.getCode == 404 => F.pure(None)
        case e: Throwable                                             => F.raiseError(e)
      }
      responseOpt <- withLogging(
        call,
        Some(traceId),
        s"io.kubernetes.client.apis.CoreV1Api.listNamespacedService(${namespace.name.value}, null, true, null, null, null, null, null, null, null)"
      )

      // Many of these fields can be null, so null-check everything
      ipOpt = for {
        response <- responseOpt
        items <- Option(response.getItems)
        item <- items.asScala.filter(_.getMetadata.getName == serviceName.value).headOption
        status <- Option(item.getStatus)
        lb <- Option(status.getLoadBalancer)
        ingresses <- Option(lb.getIngress)
        ingress <- ingresses.asScala.headOption
        ip <- Option(ingress.getIp)
        res = IP(ip)
      } yield res

    } yield ipOpt

  override def createNamespace(clusterId: KubernetesClusterId, namespace: KubernetesNamespace)(
    implicit ev: ApplicativeAsk[F, TraceId]
  ): F[Unit] =
    for {
      traceId <- ev.ask
      client <- blockingF(getClient(clusterId, new CoreV1Api(_)))
      call = blockingF(
        F.delay(
          client.createNamespace(namespace.getJavaSerialization, null, "true", null)
        )
      )
      _ <- withLogging(
        call,
        Some(traceId),
        s"io.kubernetes.client.apis.CoreV1Api.createNamespace(${namespace.getJavaSerialization}, null, true, null)"
      )
    } yield ()

  override def deleteNamespace(clusterId: KubernetesClusterId, namespace: KubernetesNamespace)(
    implicit ev: ApplicativeAsk[F, TraceId]
  ): F[Unit] = {
    val delete = for {
      traceId <- ev.ask
      client <- blockingF(getClient(clusterId, new CoreV1Api(_)))
      call = blockingF(
        F.delay(
          client.deleteNamespace(namespace.name.value, null, null, null, null, null, null)
        )
      )
      _ <- withLogging(
        call,
        Some(traceId),
        s"io.kubernetes.client.apis.CoreV1Api.deleteNamespace(${namespace.name.value}, null, null, null, null, null, null)"
      )
    } yield ()

    // There is a known bug with the client lib json decoding.  `com.google.gson.JsonSyntaxException` occurs every time.
    // See https://github.com/kubernetes-client/java/issues/86
    delete.handleErrorWith {
      case _: com.google.gson.JsonSyntaxException =>
        F.unit
      case e: Throwable => F.raiseError(e)
    }
  }

  override def createSecret(clusterId: KubernetesClusterId, namespace: KubernetesNamespace, secret: KubernetesSecret)(
    implicit ev: ApplicativeAsk[F, TraceId]
  ): F[Unit] =
    for {
      traceId <- ev.ask
      client <- blockingF(getClient(clusterId, new CoreV1Api(_)))
      call = blockingF(
        F.delay(
          client.createNamespacedSecret(namespace.name.value, secret.getJavaSerialization, null, "true", null)
        )
      )
      _ <- withLogging(
        call,
        Some(traceId),
        s"io.kubernetes.client.apis.CoreV1Api.createNamespacedSecret(${namespace.name.value}, ${secret.name.value}, null, true, null)"
      )
    } yield ()

  override def createServiceAccount(clusterId: KubernetesClusterId,
                                    serviceAccount: KubernetesServiceAccount,
                                    namespace: KubernetesNamespace)(
    implicit ev: ApplicativeAsk[F, TraceId]
  ): F[Unit] =
    for {
      traceId <- ev.ask
      client <- blockingF(getClient(clusterId, new CoreV1Api(_)))
      call = blockingF(
        F.delay(
          client.createNamespacedServiceAccount(namespace.name.value,
                                                serviceAccount.getJavaSerialization,
                                                null,
                                                "true",
                                                null)
        )
      )
      _ <- withLogging(
        call,
        Some(traceId),
        s"io.kubernetes.client.apis.CoreV1Api.createNamespacedServiceAccount(${namespace.name.value}, ${serviceAccount.name.value}, null, true, null)"
      )
    } yield ()

  override def createRole(clusterId: KubernetesClusterId, role: KubernetesRole, namespace: KubernetesNamespace)(
    implicit ev: ApplicativeAsk[F, TraceId]
  ): F[Unit] =
    for {
      traceId <- ev.ask
      client <- blockingF(getClient(clusterId, new RbacAuthorizationV1Api(_)))
      call = blockingF(
        F.delay(
          client.createNamespacedRole(namespace.name.value, role.getJavaSerialization, null, "true", null)
        )
      )
      _ <- withLogging(
        call,
        Some(traceId),
        s"io.kubernetes.client.apis.RbacAuthorizationV1Api.createNamespacedRole(${namespace.name.value}, ${role.name.value}, null, true, null)"
      )
    } yield ()

  override def createRoleBinding(clusterId: KubernetesClusterId,
                                 roleBinding: KubernetesRoleBinding,
                                 namespace: KubernetesNamespace)(
    implicit ev: ApplicativeAsk[F, TraceId]
  ): F[Unit] =
    for {
      traceId <- ev.ask
      client <- blockingF(getClient(clusterId, new RbacAuthorizationV1Api(_)))
      call = blockingF(
        F.delay(
          client.createNamespacedRoleBinding(namespace.name.value, roleBinding.getJavaSerialization, null, "true", null)
        )
      )
      _ <- withLogging(
        call,
        Some(traceId),
        s"io.kubernetes.client.apis.RbacAuthorizationV1Api.createNamespacedRoleBinding(${namespace.name.value}, ${roleBinding.name.value}, null, true, null)"
      )
    } yield ()

  //DO NOT QUERY THE CACHE DIRECTLY
  //There is a wrapper method that is necessary to ensure the token is refreshed
  //we never make the entry stale, because we always need to refresh the token (see comment above getToken)
  //if we did stale the entry we would have to unnecessarily re-do the google call
  private def getClient[A](clusterId: KubernetesClusterId, fa: ApiClient => A): F[A] =
    for {
      client <- F.delay(cache.get(clusterId))
      token <- getToken()
      _ <- F.delay(client.setApiKey(token.getTokenValue))
    } yield fa(client)

  //we always update the token, even for existing clients, so we don't have to maintain a reference to the last time each client was updated
  //unfortunately, the kubernetes client does not implement a gcp authenticator, so we must do this ourselves.
  //See this for details https://github.com/kubernetes-client/java/issues/290
  private def getToken(): F[AccessToken] =
    for {
      _ <- F.delay(credentials.refreshIfExpired())
    } yield credentials.getAccessToken

  //The underlying http client for ApiClient claims that it releases idle threads and that shutdown is not necessary
  //Here is a guide on how to proactively release resource if this proves to be problematic https://square.github.io/okhttp/4.x/okhttp/okhttp3/-ok-http-client/#shutdown-isnt-necessary
  private def createClient(cluster: Cluster, token: AccessToken): F[ApiClient] = {
    val endpoint = KubernetesApiServerIp(cluster.getEndpoint)
    val cert = KubernetesClusterCaCert(cluster.getMasterAuth.getClusterCaCertificate)

    for {
      cert <- F.fromEither(cert.base64Cert)
      certResource = autoClosableResourceF(new ByteArrayInputStream(cert))
      apiClient <- certResource.use { certStream =>
        F.delay(
          Config
            .fromToken(
              endpoint.url,
              token.getTokenValue
            )
            .setSslCaCert(certStream)
        )
      }
    } yield (apiClient) // appending here a .setDebugging(true) prints out useful API request/response info for development
  }

  // TODO: Retry once we know what Kubernetes error codes are applicable
  private def blockingF[A](fa: F[A]): F[A] = blockerBound.withPermit(blocker.blockOn(fa))
}

1	package org.broadinstitute.dsde.workbench.google2
2
3	import java.io.ByteArrayInputStream
4	import java.util.UUID
5	import java.util.concurrent.TimeUnit
6
7	import cats.effect.concurrent.Semaphore
8	import cats.effect.implicits._
9	import cats.effect.{Async, Blocker, ContextShift, Effect, Timer}
10	import cats.implicits._
11	import cats.mtl.ApplicativeAsk
12	import com.google.auth.oauth2.{AccessToken, GoogleCredentials}
13	import com.google.common.cache.{CacheBuilder, CacheLoader}
14	import com.google.container.v1.Cluster
15	import io.chrisdavenport.log4cats.StructuredLogger
16	import io.kubernetes.client.ApiClient
17	import io.kubernetes.client.apis.{CoreV1Api, RbacAuthorizationV1Api}
18	import io.kubernetes.client.util.Config
19	import org.broadinstitute.dsde.workbench.google2.GKEModels.KubernetesClusterId
20	import org.broadinstitute.dsde.workbench.google2.JavaSerializableInstances._
21	import org.broadinstitute.dsde.workbench.google2.JavaSerializableSyntax._
22	import org.broadinstitute.dsde.workbench.google2.KubernetesModels._
23	import org.broadinstitute.dsde.workbench.google2.KubernetesSerializableName.{PodName, ServiceName}
24	import org.broadinstitute.dsde.workbench.model.{IP, TraceId}
25
26	import scala.collection.JavaConverters._
27
28	// This uses a kubernetes client library to make calls to the kubernetes API. The client library is autogenerated from the kubernetes API.
29	// It is highly recommended to use the kubernetes API docs here https://kubernetes.io/docs/reference/generated/kubernetes-api as opposed to the client library docs
30
31	class KubernetesInterpreter[F[_]: StructuredLogger: Effect: Timer: ContextShift](
32	credentials: GoogleCredentials,
33	gkeService: GKEService[F],
34	blocker: Blocker,
35	blockerBound: Semaphore[F]
36	)(implicit F: Async[F])
37	extends KubernetesService[F] {
38
39	//We cache a kubernetes client for each cluster
40	val cache = CacheBuilder
41	.newBuilder()
42	// We expect calls to be batched, such as when a user's environment within a cluster is created/deleted/stopped.
43	// TODO: Unhardcode expiration time
44	.expireAfterWrite(2, TimeUnit.HOURS)	×
45	.build(	×
46	new CacheLoader[KubernetesClusterId, ApiClient] {	×
47	def load(clusterId: KubernetesClusterId): ApiClient = {
48	//we do not want to have to specify this at resource (class) creation time, so we create one on each load here
49	implicit val traceId = ApplicativeAsk.const[F, TraceId](TraceId(UUID.randomUUID()))	×
50	val res = for {
51	_ <- StructuredLogger[F]	×
52	.info(s"Determined that there is no cached client for kubernetes cluster ${clusterId}. Creating a client")	×
53	clusterOpt <- gkeService.getCluster(clusterId)	×
54	cluster <- F.fromEither(	×
55	clusterOpt.toRight(	×
56	KubernetesClusterNotFoundException(	×
57	s"Could not create client for cluster ${clusterId} because it does not exist in google"	×
58	)
59	)
60	)
61	token <- getToken	×
62	client <- createClient(	×
63	cluster,
64	token
65	)
66	} yield client
67
68	res.toIO.unsafeRunSync()	×
69	}
70	}
71	)
72
73	// https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.12/#podspec-v1-core
74	override def createPod(clusterId: KubernetesClusterId, pod: KubernetesPod, namespace: KubernetesNamespace)(
75	implicit ev: ApplicativeAsk[F, TraceId]
76	): F[Unit] =
77	for {
78	traceId <- ev.ask	×
79	client <- blockingF(getClient(clusterId, new CoreV1Api(_)))	×
80	call = blockingF(	×
81	F.delay(	×
82	client.createNamespacedPod(namespace.name.value, pod.getJavaSerialization, null, "true", null)	×
83	)
84	)
85	_ <- withLogging(	×
86	call,
87	Some(traceId),	×
88	s"io.kubernetes.client.apis.CoreV1Api.createNamespacedPod(${namespace.name.value}, ${pod.name.value}, null, true, null)"	×
89	)
90	} yield ()	×
91
92	override def listPodStatus(clusterId: KubernetesClusterId, namespace: KubernetesNamespace)(
93	implicit ev: ApplicativeAsk[F, TraceId]
94	): F[List[KubernetesPodStatus]] =
95	for {
96	traceId <- ev.ask	×
97	client <- blockingF(getClient(clusterId, new CoreV1Api(_)))	×
98	call = blockingF(	×
99	F.delay(	×
100	client.listNamespacedPod(namespace.name.value, null, "true", null, null, null, null, null, null, null)	×
101	)
102	)
103	response <- withLogging(	×
104	call,
105	Some(traceId),	×
106	s"io.kubernetes.client.apis.CoreV1Api.listNamespacedPod(${namespace.name.value}, null, true, null, null, null, null, null, null, null)"	×
107	)
108
109	listPodStatus = Option(response.getItems)	×
110	.map { l =>
111	l.asScala.toList.foldMap(	×
112	v1Pod =>
113	PodStatus.stringToPodStatus
114	.get(v1Pod.getStatus.getPhase)	×
115	.map(s => List(KubernetesPodStatus(PodName(v1Pod.getMetadata.getName), s)))	×
116	.toRight(new RuntimeException(s"Unknown Google status ${v1Pod.getStatus.getPhase}"))	×
117	)
118	}
119	.getOrElse(Nil.asRight[RuntimeException])	×
120
121	res <- F.fromEither(listPodStatus)	×
122	} yield res
123
124	// Why we use a service over a deployment: https://matthewpalmer.net/kubernetes-app-developer/articles/service-kubernetes-example-tutorial.html
125	// Services can be applied to pods/containers, while deployments are for pre-creating pods/containers.
126	override def createService(clusterId: KubernetesClusterId,
127	service: KubernetesServiceKind,
128	namespace: KubernetesNamespace)(
129	implicit ev: ApplicativeAsk[F, TraceId]
130	): F[Unit] =
131	for {
132	traceId <- ev.ask	×
133	client <- blockingF(getClient(clusterId, new CoreV1Api(_)))	×
134	call = blockingF(	×
135	F.delay(	×
136	client.createNamespacedService(namespace.name.value, service.getJavaSerialization, null, "true", null)	×
137	)
138	)
139	_ <- withLogging(	×
140	call,
141	Some(traceId),	×
142	s"io.kubernetes.client.apis.CoreV1Api.createNamespacedService(${namespace.name.value}, ${service.serviceName.value}, null, true, null)"	×
143	)
144	} yield ()	×
145
146	override def getServiceExternalIp(clusterId: KubernetesClusterId,
147	namespace: KubernetesNamespace,
148	serviceName: ServiceName)(
149	implicit ev: ApplicativeAsk[F, TraceId]
150	): F[Option[IP]] =
151	for {
152	traceId <- ev.ask	×
153	client <- blockingF(getClient(clusterId, new CoreV1Api(_)))	×
154	call = blockingF(	×
155	F.delay(	×
156	client.listNamespacedService(namespace.name.value, null, "true", null, null, null, null, null, null, null)	×
157	)
158	).map(Option(_)).handleErrorWith {	×
159	case e: io.kubernetes.client.ApiException if e.getCode == 404 => F.pure(None)	×
160	case e: Throwable => F.raiseError(e)	×
161	}
162	responseOpt <- withLogging(	×
163	call,
164	Some(traceId),	×
165	s"io.kubernetes.client.apis.CoreV1Api.listNamespacedService(${namespace.name.value}, null, true, null, null, null, null, null, null, null)"	×
166	)
167
168	// Many of these fields can be null, so null-check everything
169	ipOpt = for {
170	response <- responseOpt	×
171	items <- Option(response.getItems)	×
172	item <- items.asScala.filter(_.getMetadata.getName == serviceName.value).headOption	×
173	status <- Option(item.getStatus)	×
174	lb <- Option(status.getLoadBalancer)	×
175	ingresses <- Option(lb.getIngress)	×
176	ingress <- ingresses.asScala.headOption	×
177	ip <- Option(ingress.getIp)	×
178	res = IP(ip)	×
179	} yield res
180
181	} yield ipOpt
182
183	override def createNamespace(clusterId: KubernetesClusterId, namespace: KubernetesNamespace)(
184	implicit ev: ApplicativeAsk[F, TraceId]
185	): F[Unit] =
186	for {
187	traceId <- ev.ask	×
188	client <- blockingF(getClient(clusterId, new CoreV1Api(_)))	×
189	call = blockingF(	×
190	F.delay(	×
191	client.createNamespace(namespace.getJavaSerialization, null, "true", null)	×
192	)
193	)
194	_ <- withLogging(	×
195	call,
196	Some(traceId),	×
197	s"io.kubernetes.client.apis.CoreV1Api.createNamespace(${namespace.getJavaSerialization}, null, true, null)"	×
198	)
199	} yield ()	×
200
201	override def deleteNamespace(clusterId: KubernetesClusterId, namespace: KubernetesNamespace)(
202	implicit ev: ApplicativeAsk[F, TraceId]
203	): F[Unit] = {
204	val delete = for {
205	traceId <- ev.ask	×
206	client <- blockingF(getClient(clusterId, new CoreV1Api(_)))	×
207	call = blockingF(	×
208	F.delay(	×
209	client.deleteNamespace(namespace.name.value, null, null, null, null, null, null)	×
210	)
211	)
212	_ <- withLogging(	×
213	call,
214	Some(traceId),	×
215	s"io.kubernetes.client.apis.CoreV1Api.deleteNamespace(${namespace.name.value}, null, null, null, null, null, null)"	×
216	)
217	} yield ()	×
218
219	// There is a known bug with the client lib json decoding. `com.google.gson.JsonSyntaxException` occurs every time.
220	// See https://github.com/kubernetes-client/java/issues/86
221	delete.handleErrorWith {	×
222	case _: com.google.gson.JsonSyntaxException =>
223	F.unit	×
224	case e: Throwable => F.raiseError(e)	×
225	}
226	}
227
228	override def createSecret(clusterId: KubernetesClusterId, namespace: KubernetesNamespace, secret: KubernetesSecret)(
229	implicit ev: ApplicativeAsk[F, TraceId]
230	): F[Unit] =
231	for {
232	traceId <- ev.ask	×
233	client <- blockingF(getClient(clusterId, new CoreV1Api(_)))	×
234	call = blockingF(	×
235	F.delay(	×
236	client.createNamespacedSecret(namespace.name.value, secret.getJavaSerialization, null, "true", null)	×
237	)
238	)
239	_ <- withLogging(	×
240	call,
241	Some(traceId),	×
242	s"io.kubernetes.client.apis.CoreV1Api.createNamespacedSecret(${namespace.name.value}, ${secret.name.value}, null, true, null)"	×
243	)
244	} yield ()	×
245
246	override def createServiceAccount(clusterId: KubernetesClusterId,
247	serviceAccount: KubernetesServiceAccount,
248	namespace: KubernetesNamespace)(
249	implicit ev: ApplicativeAsk[F, TraceId]
250	): F[Unit] =
251	for {
252	traceId <- ev.ask	×
253	client <- blockingF(getClient(clusterId, new CoreV1Api(_)))	×
254	call = blockingF(	×
255	F.delay(	×
256	client.createNamespacedServiceAccount(namespace.name.value,	×
257	serviceAccount.getJavaSerialization,	×
258	null,	×
259	"true",	×
260	null)	×
261	)
262	)
263	_ <- withLogging(	×
264	call,
265	Some(traceId),	×
266	s"io.kubernetes.client.apis.CoreV1Api.createNamespacedServiceAccount(${namespace.name.value}, ${serviceAccount.name.value}, null, true, null)"	×
267	)
268	} yield ()	×
269
270	override def createRole(clusterId: KubernetesClusterId, role: KubernetesRole, namespace: KubernetesNamespace)(
271	implicit ev: ApplicativeAsk[F, TraceId]
272	): F[Unit] =
273	for {
274	traceId <- ev.ask	×
275	client <- blockingF(getClient(clusterId, new RbacAuthorizationV1Api(_)))	×
276	call = blockingF(	×
277	F.delay(	×
278	client.createNamespacedRole(namespace.name.value, role.getJavaSerialization, null, "true", null)	×
279	)
280	)
281	_ <- withLogging(	×
282	call,
283	Some(traceId),	×
284	s"io.kubernetes.client.apis.RbacAuthorizationV1Api.createNamespacedRole(${namespace.name.value}, ${role.name.value}, null, true, null)"	×
285	)
286	} yield ()	×
287
288	override def createRoleBinding(clusterId: KubernetesClusterId,
289	roleBinding: KubernetesRoleBinding,
290	namespace: KubernetesNamespace)(
291	implicit ev: ApplicativeAsk[F, TraceId]
292	): F[Unit] =
293	for {
294	traceId <- ev.ask	×
295	client <- blockingF(getClient(clusterId, new RbacAuthorizationV1Api(_)))	×
296	call = blockingF(	×
297	F.delay(	×
298	client.createNamespacedRoleBinding(namespace.name.value, roleBinding.getJavaSerialization, null, "true", null)	×
299	)
300	)
301	_ <- withLogging(	×
302	call,
303	Some(traceId),	×
304	s"io.kubernetes.client.apis.RbacAuthorizationV1Api.createNamespacedRoleBinding(${namespace.name.value}, ${roleBinding.name.value}, null, true, null)"	×
305	)
306	} yield ()	×
307
308	//DO NOT QUERY THE CACHE DIRECTLY
309	//There is a wrapper method that is necessary to ensure the token is refreshed
310	//we never make the entry stale, because we always need to refresh the token (see comment above getToken)
311	//if we did stale the entry we would have to unnecessarily re-do the google call
312	private def getClient[A](clusterId: KubernetesClusterId, fa: ApiClient => A): F[A] =
313	for {
314	client <- F.delay(cache.get(clusterId))	×
315	token <- getToken()	×
316	_ <- F.delay(client.setApiKey(token.getTokenValue))	×
317	} yield fa(client)	×
318
319	//we always update the token, even for existing clients, so we don't have to maintain a reference to the last time each client was updated
320	//unfortunately, the kubernetes client does not implement a gcp authenticator, so we must do this ourselves.
321	//See this for details https://github.com/kubernetes-client/java/issues/290
322	private def getToken(): F[AccessToken] =
323	for {
324	_ <- F.delay(credentials.refreshIfExpired())	×
325	} yield credentials.getAccessToken	×
326
327	//The underlying http client for ApiClient claims that it releases idle threads and that shutdown is not necessary
328	//Here is a guide on how to proactively release resource if this proves to be problematic https://square.github.io/okhttp/4.x/okhttp/okhttp3/-ok-http-client/#shutdown-isnt-necessary
329	private def createClient(cluster: Cluster, token: AccessToken): F[ApiClient] = {
330	val endpoint = KubernetesApiServerIp(cluster.getEndpoint)	×
331	val cert = KubernetesClusterCaCert(cluster.getMasterAuth.getClusterCaCertificate)	×
332
333	for {
334	cert <- F.fromEither(cert.base64Cert)	×
335	certResource = autoClosableResourceF(new ByteArrayInputStream(cert))	×
336	apiClient <- certResource.use { certStream =>	×
337	F.delay(	×
338	Config
339	.fromToken(
340	endpoint.url,
341	token.getTokenValue
342	)
343	.setSslCaCert(certStream)	×
344	)
345	}
346	} yield (apiClient) // appending here a .setDebugging(true) prints out useful API request/response info for development
347	}
348
349	// TODO: Retry once we know what Kubernetes error codes are applicable
350	private def blockingF[A](fa: F[A]): F[A] = blockerBound.withPermit(blocker.blockOn(fa))	×
351	}

broadinstitute / workbench-libs / 3662

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