This is an automated email from the ASF dual-hosted git repository.
dgrove pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-openwhisk.git
The following commit(s) were added to refs/heads/master by this push:
new b0d3574 Delete pod when creating timeout (#4424)
b0d3574 is described below
commit b0d3574aa07501b433c5d18dc93b640082e67e70
Author: Lin <[email protected]>
AuthorDate: Thu Jul 25 04:38:03 2019 +0800
Delete pod when creating timeout (#4424)
---
.../kubernetes/KubernetesClient.scala | 52 +++++++++++-----------
.../kubernetes/KubernetesContainer.scala | 13 +++++-
.../kubernetes/test/KubernetesClientTests.scala | 4 ++
.../kubernetes/test/KubernetesContainerTests.scala | 5 +--
4 files changed, 45 insertions(+), 29 deletions(-)
diff --git
a/core/invoker/src/main/scala/org/apache/openwhisk/core/containerpool/kubernetes/KubernetesClient.scala
b/core/invoker/src/main/scala/org/apache/openwhisk/core/containerpool/kubernetes/KubernetesClient.scala
index 4dabc4f..777f069 100644
---
a/core/invoker/src/main/scala/org/apache/openwhisk/core/containerpool/kubernetes/KubernetesClient.scala
+++
b/core/invoker/src/main/scala/org/apache/openwhisk/core/containerpool/kubernetes/KubernetesClient.scala
@@ -19,47 +19,37 @@ package org.apache.openwhisk.core.containerpool.kubernetes
import java.io.IOException
import java.net.SocketTimeoutException
-import java.time.{Instant, ZoneId}
import java.time.format.DateTimeFormatterBuilder
+import java.time.{Instant, ZoneId}
import akka.actor.ActorSystem
import akka.http.scaladsl.model.Uri
-import akka.http.scaladsl.model.Uri.Path
-import akka.http.scaladsl.model.Uri.Query
-import akka.stream.{Attributes, Outlet, SourceShape}
-import akka.stream.ActorMaterializer
+import akka.http.scaladsl.model.Uri.{Path, Query}
import akka.stream.scaladsl.Source
import akka.stream.stage._
+import akka.stream.{ActorMaterializer, Attributes, Outlet, SourceShape}
import akka.util.ByteString
import io.fabric8.kubernetes.api.model._
-import pureconfig.loadConfigOrThrow
-import org.apache.openwhisk.common.Logging
-import org.apache.openwhisk.common.TransactionId
+import io.fabric8.kubernetes.client.{ConfigBuilder, DefaultKubernetesClient}
+import okhttp3.{Call, Callback, Request, Response}
+import okio.BufferedSource
+import org.apache.openwhisk.common.{Logging, TransactionId}
import org.apache.openwhisk.core.ConfigKeys
-import org.apache.openwhisk.core.containerpool.ContainerId
-import org.apache.openwhisk.core.containerpool.ContainerAddress
import org.apache.openwhisk.core.containerpool.docker.ProcessRunner
+import org.apache.openwhisk.core.containerpool.{ContainerAddress, ContainerId}
import org.apache.openwhisk.core.entity.ByteSize
import org.apache.openwhisk.core.entity.size._
-
-import scala.concurrent.duration._
-import scala.concurrent.ExecutionContext
-import scala.concurrent.Future
-import scala.concurrent.blocking
-import scala.util.Failure
-import scala.util.Success
-import scala.util.Try
-import spray.json._
+import pureconfig.loadConfigOrThrow
import spray.json.DefaultJsonProtocol._
-import collection.JavaConverters._
-import io.fabric8.kubernetes.client.ConfigBuilder
-import io.fabric8.kubernetes.client.DefaultKubernetesClient
-import okhttp3.{Call, Callback, Request, Response}
-import okio.BufferedSource
+import spray.json._
import scala.annotation.tailrec
+import scala.collection.JavaConverters._
import scala.collection.mutable
+import scala.concurrent.duration._
+import scala.concurrent.{blocking, ExecutionContext, Future}
import scala.util.control.NonFatal
+import scala.util.{Failure, Success, Try}
/**
* Configuration for kubernetes client command timeouts.
@@ -186,6 +176,17 @@ class KubernetesClient(
}
}.map(_ => ())
}
+ def rm(podName: String): Future[Unit] = {
+ Future {
+ blocking {
+ kubeRestClient
+ .inNamespace(kubeRestClient.getNamespace)
+ .pods()
+ .withName(podName)
+ .delete()
+ }
+ }.map(_ => ())
+ }
def rm(key: String, value: String, ensureUnpaused: Boolean = false)(implicit
transid: TransactionId): Future[Unit] = {
Future {
@@ -249,6 +250,7 @@ object KubernetesClient {
}
trait KubernetesApi {
+
def run(name: String,
image: String,
memory: ByteSize,
@@ -256,7 +258,7 @@ trait KubernetesApi {
labels: Map[String, String] = Map.empty)(implicit transid:
TransactionId): Future[KubernetesContainer]
def rm(container: KubernetesContainer)(implicit transid: TransactionId):
Future[Unit]
-
+ def rm(podName: String): Future[Unit]
def rm(key: String, value: String, ensureUnpaused: Boolean)(implicit
transid: TransactionId): Future[Unit]
def suspend(container: KubernetesContainer)(implicit transid:
TransactionId): Future[Unit]
diff --git
a/core/invoker/src/main/scala/org/apache/openwhisk/core/containerpool/kubernetes/KubernetesContainer.scala
b/core/invoker/src/main/scala/org/apache/openwhisk/core/containerpool/kubernetes/KubernetesContainer.scala
index 74fd292..d37139b 100644
---
a/core/invoker/src/main/scala/org/apache/openwhisk/core/containerpool/kubernetes/KubernetesContainer.scala
+++
b/core/invoker/src/main/scala/org/apache/openwhisk/core/containerpool/kubernetes/KubernetesContainer.scala
@@ -37,6 +37,8 @@ import org.apache.openwhisk.core.entity.ByteSize
import org.apache.openwhisk.core.entity.size._
import org.apache.openwhisk.http.Messages
+import scala.util.Failure
+
object KubernetesContainer {
/**
@@ -67,7 +69,16 @@ object KubernetesContainer {
for {
container <- kubernetes.run(podName, image, memory, environment,
labels).recoverWith {
- case _ => Future.failed(WhiskContainerStartupError(s"Failed to run
container with image '${image}'."))
+ case _ =>
+ kubernetes
+ .rm(podName)
+ .andThen {
+ case Failure(e) =>
+ log.error(this, s"Failed delete pod for '$name': ${e.getClass}
- ${e.getMessage}")
+ }
+ .transformWith { _ =>
+ Future.failed(WhiskContainerStartupError(s"Failed to run
container with image '${image}'."))
+ }
}
} yield container
}
diff --git
a/tests/src/test/scala/org/apache/openwhisk/core/containerpool/kubernetes/test/KubernetesClientTests.scala
b/tests/src/test/scala/org/apache/openwhisk/core/containerpool/kubernetes/test/KubernetesClientTests.scala
index 74accbe..0afaefa 100644
---
a/tests/src/test/scala/org/apache/openwhisk/core/containerpool/kubernetes/test/KubernetesClientTests.scala
+++
b/tests/src/test/scala/org/apache/openwhisk/core/containerpool/kubernetes/test/KubernetesClientTests.scala
@@ -212,6 +212,10 @@ object KubernetesClientTests {
Future.successful(())
}
+ override def rm(podName: String): Future[Unit] = {
+ rms += ContainerId(podName)
+ Future.successful(())
+ }
def rm(key: String, value: String, ensureUnpause: Boolean = false)(
implicit transid: TransactionId): Future[Unit] = {
rmByLabels += ((key, value))
diff --git
a/tests/src/test/scala/org/apache/openwhisk/core/containerpool/kubernetes/test/KubernetesContainerTests.scala
b/tests/src/test/scala/org/apache/openwhisk/core/containerpool/kubernetes/test/KubernetesContainerTests.scala
index 1b1352a..20a48d1 100644
---
a/tests/src/test/scala/org/apache/openwhisk/core/containerpool/kubernetes/test/KubernetesContainerTests.scala
+++
b/tests/src/test/scala/org/apache/openwhisk/core/containerpool/kubernetes/test/KubernetesContainerTests.scala
@@ -173,7 +173,6 @@ class KubernetesContainerTests
memory: ByteSize = 256.MB,
env: Map[String, String] = Map.empty,
labels: Map[String, String] = Map.empty)(implicit transid:
TransactionId): Future[KubernetesContainer] = {
- runs += ((name, image, env, labels))
Future.failed(ProcessUnsuccessfulException(ExitStatus(1), "", ""))
}
}
@@ -182,8 +181,8 @@ class KubernetesContainerTests
KubernetesContainer.create(transid = transid, name = "name", image =
"image", userProvidedImage = true)
a[WhiskContainerStartupError] should be thrownBy await(container)
- kubernetes.runs should have size 1
- kubernetes.rms should have size 0
+ kubernetes.runs should have size 0
+ kubernetes.rms should have size 1
}
/*
