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. |
||
| } | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,24 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one or more | ||
| * contributor license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright ownership. | ||
| * The ASF licenses this file to You under the Apache License, Version 2.0 | ||
| * (the "License"); you may not use this file except in compliance with | ||
| * the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
| package org.apache.spark.deploy.k8s.integrationtest | ||
|
|
||
| package object config { | ||
| val KUBERNETES_TEST_DOCKER_TAG_SYSTEM_PROPERTY = "spark.kubernetes.test.imageDockerTag" | ||
| val DRIVER_DOCKER_IMAGE = "spark.kubernetes.driver.docker.image" | ||
| val EXECUTOR_DOCKER_IMAGE = "spark.kubernetes.executor.docker.image" | ||
| val INIT_CONTAINER_DOCKER_IMAGE = "spark.kubernetes.initcontainer.docker.image" | ||
| } |
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.
Hmm. I remember @foxish just deleted this line recently so integration tests can run against GCE. Can you check with @foxish?
Uh oh!
There was an error while loading. Please reload this page.
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.
(Comment was for the wrong section)