Use a pre-installed Minikube instance -- porting over logic from PR 521 #14
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -19,13 +19,41 @@ package org.apache.spark.deploy.k8s.integrationtest | |
| import java.io.Closeable | ||
| import java.net.URI | ||
|
|
||
| import java.io.{IOException,InputStream,OutputStream} | ||
|
|
||
| object Utils extends Logging { | ||
|
|
||
| def tryWithResource[R <: Closeable, T](createResource: => R)(f: R => T): T = { | ||
| val resource = createResource | ||
| try f.apply(resource) finally resource.close() | ||
| } | ||
|
|
||
| def tryWithSafeFinally[T](block: => T)(finallyBlock: => Unit): T = { | ||
| var originalThrowable: Throwable = null | ||
| try { | ||
| block | ||
| } catch { | ||
| case t: Throwable => | ||
| // Purposefully not using NonFatal, because even fatal exceptions | ||
| // we don't want to have our finallyBlock suppress | ||
| originalThrowable = t | ||
| throw originalThrowable | ||
| } finally { | ||
| try { | ||
| finallyBlock | ||
| } catch { | ||
| case t: Throwable => | ||
| if (originalThrowable != null) { | ||
| originalThrowable.addSuppressed(t) | ||
| logWarning(s"Suppressing exception in finally: " + t.getMessage, t) | ||
| throw originalThrowable | ||
| } else { | ||
| throw t | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| def checkAndGetK8sMasterUrl(rawMasterURL: String): String = { | ||
| require(rawMasterURL.startsWith("k8s://"), | ||
| "Kubernetes master URL must start with k8s://.") | ||
|
|
@@ -57,4 +85,30 @@ object Utils extends Logging { | |
|
|
||
| s"k8s://$resolvedURL" | ||
| } | ||
|
|
||
| class RedirectThread( | ||
| in: InputStream, | ||
| out: OutputStream, | ||
| name: String, | ||
| propagateEof: Boolean = false) extends Thread(name) { | ||
| setDaemon(true) | ||
| override def run() { | ||
| scala.util.control.Exception.ignoring(classOf[IOException]) { | ||
| // FIXME: We copy the stream on the level of bytes to avoid encoding problems. | ||
| Utils.tryWithSafeFinally { | ||
| val buf = new Array[Byte](1024) | ||
|
There was a problem hiding this comment. Should this line 99-105 go into its own subroutine? There was a problem hiding this comment. As an alternative, do we have Guava available? Could we just use ByteStreams.copy() instead of the entire body here? There was a problem hiding this comment. These Utils are taken from spark core.... should I modify them? There was a problem hiding this comment. I'd argue that the cleanest modification (using Guava) is something we should do. There's no reason to replicate this code from the Spark core. There was a problem hiding this comment. I don't see this class being used anywhere anymore actually, so we can remove this. |
||
| var len = in.read(buf) | ||
| while (len != -1) { | ||
|
There was a problem hiding this comment. This input reading loop is typically done with a break command inside an infinite loop. (So there aren't two reads in the code) Not that it's too important to fix this, but have you considered using an approach like this so it can use breaks? https://alvinalexander.com/scala/break-continue-for-while-loops-in-scala-examples-how-to |
||
| out.write(buf, 0, len) | ||
| out.flush() | ||
| len = in.read(buf) | ||
| } | ||
| } { | ||
| if (propagateEof) { | ||
| out.close() | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -20,7 +20,7 @@ import io.fabric8.kubernetes.client.{ConfigBuilder, DefaultKubernetesClient} | |
|
|
||
| import org.apache.spark.deploy.k8s.integrationtest.Utils | ||
| import org.apache.spark.deploy.k8s.integrationtest.backend.IntegrationTestBackend | ||
| import org.apache.spark.deploy.k8s.integrationtest.config._ | ||
|
|
||
| private[spark] class GCETestBackend(val master: String) extends IntegrationTestBackend { | ||
| private var defaultClient: DefaultKubernetesClient = _ | ||
|
|
@@ -37,5 +37,7 @@ private[spark] class GCETestBackend(val master: String) extends IntegrationTestB | |
| defaultClient | ||
| } | ||
|
|
||
| override def dockerImageTag(): String = { | ||
| return System.getProperty(KUBERNETES_TEST_DOCKER_TAG_SYSTEM_PROPERTY, "latest") | ||
|
There was a problem hiding this comment. Why not generate a random ID like minikube backend code does? i.e. There was a problem hiding this comment. In the Minikube case we're building these images from scratch. In the GCE case, we don't create a Docker manager and hence are not building the images there. But this in itself seems to contradict this section of our readme:
which indicates that GCE-backed tests should be building images as well. Is this correct @foxish? There was a problem hiding this comment. That readme section is meant to highlight that we push the images to an image repository only in the cloud testing case, and don't have to in the minikube case since the images are built in the minikube VM's docker environment. That documentation pertains only to the use of the script, which avoids using maven for building images. There was a problem hiding this comment. The problem then with using a random ID tag here is that it's impossible for this tag to actually match anything. Using "latest" at least guarantees that we pick up some image in the default case. We can be more strict here and require the tag be explicitly specified. There was a problem hiding this comment. Looking a little closer I think the miscommunication is because the docker image manager isn't serving the image tag but is instead being handed the tag by the test backend. The responsibilities thus aren't clear and the coupling of the provision of a custom tag vs. a generated tag, and how that influences whether or not images are built or deleted, is unclear. I'm moving the generation of the tag vs. using the user-provided one into the docker manager. This should hopefully clarify the connection. |
||
| } | ||
| } | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
What does this line mean only with a URL?