Github user echarles commented on a diff in the pull request: https://github.com/apache/spark/pull/20451#discussion_r201781927 --- Diff: resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterManager.scala --- @@ -20,41 +20,65 @@ import java.io.File import java.util.concurrent.TimeUnit import com.google.common.cache.CacheBuilder -import io.fabric8.kubernetes.client.Config - -import org.apache.spark.{SparkContext, SparkException} -import org.apache.spark.deploy.k8s.{KubernetesUtils, SparkKubernetesClientFactory} +import io.fabric8.kubernetes.client.{Config, KubernetesClient} +import org.apache.spark.{SparkConf, SparkContext, SparkException} +import org.apache.spark.deploy.k8s.SparkKubernetesClientFactory import org.apache.spark.deploy.k8s.Config._ import org.apache.spark.deploy.k8s.Constants._ import org.apache.spark.internal.Logging import org.apache.spark.scheduler.{ExternalClusterManager, SchedulerBackend, TaskScheduler, TaskSchedulerImpl} import org.apache.spark.util.{SystemClock, ThreadUtils} -private[spark] class KubernetesClusterManager extends ExternalClusterManager with Logging { +trait ManagerSpecificHandlers { + def createKubernetesClient(sparkConf: SparkConf): KubernetesClient +} - override def canCreate(masterURL: String): Boolean = masterURL.startsWith("k8s") +private[spark] class KubernetesClusterManager extends ExternalClusterManager + with ManagerSpecificHandlers with Logging { - override def createTaskScheduler(sc: SparkContext, masterURL: String): TaskScheduler = { - if (masterURL.startsWith("k8s") && - sc.deployMode == "client" && - !sc.conf.get(KUBERNETES_DRIVER_SUBMIT_CHECK).getOrElse(false)) { - throw new SparkException("Client mode is currently not supported for Kubernetes.") + class InClusterHandlers extends ManagerSpecificHandlers { + override def createKubernetesClient(sparkConf: SparkConf): KubernetesClient = + SparkKubernetesClientFactory.createKubernetesClient( + KUBERNETES_MASTER_INTERNAL_URL, + Some(sparkConf.get(KUBERNETES_NAMESPACE)), + KUBERNETES_AUTH_DRIVER_MOUNTED_CONF_PREFIX, + sparkConf, + Some(new File(Config.KUBERNETES_SERVICE_ACCOUNT_TOKEN_PATH)), + Some(new File(Config.KUBERNETES_SERVICE_ACCOUNT_CA_CRT_PATH))) + } + + class OutClusterHandlers extends ManagerSpecificHandlers { + override def createKubernetesClient(sparkConf: SparkConf): KubernetesClient = + SparkKubernetesClientFactory.createKubernetesClient( + sparkConf.get("spark.master").replace("k8s://", ""), + Some(sparkConf.get(KUBERNETES_NAMESPACE)), + KUBERNETES_AUTH_DRIVER_MOUNTED_CONF_PREFIX, --- End diff -- the call to `createKubernetesClient` is not used in two different ways: + `KUBERNETES_AUTH_DRIVER_MOUNTED_CONF_PREFIX` is used in `KubernetesClusterManager` + `KUBERNETES_AUTH_SUBMISSION_CONF_PREFIX` is used in `KubernetesClientApplication` I would favor the second and remove the first. For the config place, I remember that the fabric8 k8s client does also some inspection to see if it is in or out cluster, and loads the config form the default place (depending the case), with possiblity to specify other places for the cert, token... (this is what we give as property to the end-user).
--- --------------------------------------------------------------------- To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org