forked from apache/spark
-
Notifications
You must be signed in to change notification settings - Fork 2
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Use a separate class to track components that need to be cleaned up (a…
…pache#122) * Refactor the cleaning up of Kubernetes components. Create a KubernetesComponentsCleaner which can register arbitrary pods, services, secrets, and ingresses. When an exception is thrown or the JVM shuts down, the cleaner automatically purges any of its registered components from Kubernetes. The components can be unregistered when the driver successfully begins running, so that the application persists beyond the lifetime of the spark-submit process. * Fix spacing * Address comments * Fix compiler error * Pull KubernetesComponentCleaner into instance variable * Remove a parameter * Remove redundant registerOrUpdateSecret for SSL * Remove Ingresses from component cleaner * Clear resources generically as opposed to specifying each type * Remove incorrect test assertion * Rename variable
- Loading branch information
Showing
2 changed files
with
152 additions
and
125 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
52 changes: 52 additions & 0 deletions
52
...es/core/src/main/scala/org/apache/spark/deploy/kubernetes/KubernetesResourceCleaner.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,52 @@ | ||
| /* | ||
| * 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.kubernetes | ||
|
|
||
| import io.fabric8.kubernetes.api.model.HasMetadata | ||
| import io.fabric8.kubernetes.client.KubernetesClient | ||
| import scala.collection.mutable | ||
|
|
||
| import org.apache.spark.internal.Logging | ||
| import org.apache.spark.util.Utils | ||
|
|
||
| private[spark] class KubernetesResourceCleaner | ||
| extends Logging { | ||
|
|
||
| private val resources = mutable.HashMap.empty[(String, String), HasMetadata] | ||
|
|
||
| // Synchronized because deleteAllRegisteredResourcesFromKubernetes may be called from a | ||
| // shutdown hook | ||
| def registerOrUpdateResource(resource: HasMetadata): Unit = synchronized { | ||
| resources.put((resource.getMetadata.getName, resource.getKind), resource) | ||
| } | ||
|
|
||
| def unregisterResource(resource: HasMetadata): Unit = synchronized { | ||
| resources.remove((resource.getMetadata.getName, resource.getKind)) | ||
| } | ||
|
|
||
| def deleteAllRegisteredResourcesFromKubernetes(kubernetesClient: KubernetesClient): Unit = { | ||
| synchronized { | ||
| logInfo(s"Deleting ${resources.size} registered Kubernetes resources:") | ||
| resources.values.foreach { resource => | ||
| Utils.tryLogNonFatalError { | ||
| kubernetesClient.resource(resource).delete() | ||
| } | ||
| } | ||
| resources.clear() | ||
| } | ||
| } | ||
| } |