Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[SPARK-27023][K8S] Make k8s client timeouts configurable #23928

Closed
wants to merge 9 commits into from
Closed

[SPARK-27023][K8S] Make k8s client timeouts configurable #23928

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
2c28a06
configurable k8s client timeouts
onursatici Mar 1, 2019
0676f90
move default value declaration to k8s/Config
onursatici Mar 4, 2019
a0b88a3
remove prefix and suffix constants
onursatici Mar 4, 2019
7f83722
do not use formatted strings
onursatici Mar 5, 2019
ad17a02
remove client mode specific overrides, simplify config names
onursatici Mar 5, 2019
a316760
remove vars, update docs
onursatici Mar 5, 2019
e7fa51c
scoped import
onursatici Mar 5, 2019
280c176
inline
onursatici Mar 6, 2019
1902a13
remove redundant import
onursatici Mar 6, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 46 additions & 0 deletions docs/running-on-kubernetes.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -990,6 +990,52 @@ See the [configuration page](configuration.html) for information on Spark config
Specify whether executor pods should be deleted in case of failure or normal termination.
</td>
</tr>
<tr>
<td><code>spark.kubernetes.kubernetesClient.submission.connectionTimeout</code></td>
<td>10000</td>
onursatici marked this conversation as resolved.
Show resolved Hide resolved
dongjoon-hyun marked this conversation as resolved.
Show resolved Hide resolved
<td>
Connection timeout in milliseconds for the kubernetes client to use for starting the driver. In client mode, use
<code>spark.kubernetes.kubernetesClient.connectionTimeout</code> instead.
onursatici marked this conversation as resolved.
Show resolved Hide resolved
</td>
</tr>
<tr>
<td><code>spark.kubernetes.kubernetesClient.submission.requestTimeout</code></td>
<td>10000</td>
<td>
Request timeout in milliseconds for the kubernetes client to use for starting the driver. In client mode, use
<code>spark.kubernetes.kubernetesClient.submission.requestTimeout</code> instead.
</td>
</tr>
<tr>
<td><code>spark.kubernetes.kubernetesClient.driver.connectionTimeout</code></td>
<td>10000</td>
<td>
Connection timeout in milliseconds for the kubernetes client in driver to use when requesting executors.
In client mode, use <code>spark.kubernetes.kubernetesClient.connectionTimeout</code> instead.
onursatici marked this conversation as resolved.
Show resolved Hide resolved
</td>
</tr>
<tr>
<td><code>spark.kubernetes.kubernetesClient.driver.requestTimeout</code></td>
<td>10000</td>
<td>
Request timeout in milliseconds for the kubernetes client in driver to use when requesting executors.
In client mode, use <code>spark.kubernetes.kubernetesClient.connectionTimeout</code> instead.
</td>
</tr>
<tr>
<td><code>spark.kubernetes.kubernetesClient.connectionTimeout</code></td>
onursatici marked this conversation as resolved.
Show resolved Hide resolved
<td>10000</td>
<td>
In client mode, connection timeout in milliseconds for the kubernetes client in driver to use when requesting executors.
</td>
</tr>
<tr>
<td><code>spark.kubernetes.kubernetesClient.requestTimeout</code></td>
<td>10000</td>
<td>
In client mode, request timeout in milliseconds for the kubernetes client in driver to use when requesting executors.
</td>
</tr>
</table>

#### Pod template properties
Expand Down
9 changes: 9 additions & 0 deletions resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,6 +75,12 @@ private[spark] object Config extends Logging {
.toSequence
.createWithDefault(Nil)

val SUBMISSION_CLIENT_PREFIX =
"spark.kubernetes.kubernetesClient.submission"
val DRIVER_CLIENT_PREFIX =
"spark.kubernetes.kubernetesClient.driver"
val CLIENT_MODE_CLIENT_PREFIX =
"spark.kubernetes.kubernetesClient"
onursatici marked this conversation as resolved.
Show resolved Hide resolved
val KUBERNETES_AUTH_DRIVER_CONF_PREFIX =
"spark.kubernetes.authenticate.driver"
val KUBERNETES_AUTH_DRIVER_MOUNTED_CONF_PREFIX =
Expand All @@ -85,6 +91,9 @@ private[spark] object Config extends Logging {
val CLIENT_KEY_FILE_CONF_SUFFIX = "clientKeyFile"
val CLIENT_CERT_FILE_CONF_SUFFIX = "clientCertFile"
val CA_CERT_FILE_CONF_SUFFIX = "caCertFile"
val REQUEST_TIMEOUT_SUFFIX = "requestTimeout"
val CONNECTION_TIMEOUT_SUFFIX = "connectionTimeout"


val KUBERNETES_SERVICE_ACCOUNT_NAME =
ConfigBuilder(s"$KUBERNETES_AUTH_DRIVER_CONF_PREFIX.serviceAccountName")
Expand Down
7 changes: 7 additions & 0 deletions ...rnetes/core/src/main/scala/org/apache/spark/deploy/k8s/SparkKubernetesClientFactory.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@ private[spark] object SparkKubernetesClientFactory extends Logging {
master: String,
namespace: Option[String],
kubernetesAuthConfPrefix: String,
kubernetesClientPrefix: String,
sparkConf: SparkConf,
defaultServiceAccountToken: Option[File],
defaultServiceAccountCaCert: Option[File]): KubernetesClient = {
Expand All @@ -63,6 +64,10 @@ private[spark] object SparkKubernetesClientFactory extends Logging {
.getOption(s"$kubernetesAuthConfPrefix.$CLIENT_KEY_FILE_CONF_SUFFIX")
val clientCertFile = sparkConf
.getOption(s"$kubernetesAuthConfPrefix.$CLIENT_CERT_FILE_CONF_SUFFIX")
val requestTimeout =
sparkConf.getInt(s"$kubernetesClientPrefix.$REQUEST_TIMEOUT_SUFFIX", 10000)
onursatici marked this conversation as resolved.
Show resolved Hide resolved
val connectionTimeout =
sparkConf.getInt(s"$kubernetesClientPrefix.$CONNECTION_TIMEOUT_SUFFIX", 10000)
val dispatcher = new Dispatcher(
ThreadUtils.newDaemonCachedThreadPool("kubernetes-dispatcher"))

Expand All @@ -79,6 +84,8 @@ private[spark] object SparkKubernetesClientFactory extends Logging {
.withApiVersion("v1")
.withMasterUrl(master)
.withWebsocketPingInterval(0)
.withRequestTimeout(requestTimeout)
.withConnectionTimeout(connectionTimeout)
onursatici marked this conversation as resolved.
Show resolved Hide resolved
.withOption(oauthTokenValue) {
(token, configBuilder) => configBuilder.withOauthToken(token)
}.withOption(oauthTokenFile) {
Expand Down
1 change: 1 addition & 0 deletions .../core/src/main/scala/org/apache/spark/deploy/k8s/submit/KubernetesClientApplication.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -220,6 +220,7 @@ private[spark] class KubernetesClientApplication extends SparkApplication {
master,
Some(kubernetesConf.namespace),
KUBERNETES_AUTH_SUBMISSION_CONF_PREFIX,
SUBMISSION_CLIENT_PREFIX,
sparkConf,
None,
None)) { kubernetesClient =>
Expand Down
4 changes: 4 additions & 0 deletions ...core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,18 +44,21 @@ private[spark] class KubernetesClusterManager extends ExternalClusterManager wit
scheduler: TaskScheduler): SchedulerBackend = {
val wasSparkSubmittedInClusterMode = sc.conf.get(KUBERNETES_DRIVER_SUBMIT_CHECK)
val (authConfPrefix,
clientConfPrefix,
apiServerUri,
defaultServiceAccountToken,
defaultServiceAccountCaCrt) = if (wasSparkSubmittedInClusterMode) {
require(sc.conf.get(KUBERNETES_DRIVER_POD_NAME).isDefined,
"If the application is deployed using spark-submit in cluster mode, the driver pod name " +
"must be provided.")
(KUBERNETES_AUTH_DRIVER_MOUNTED_CONF_PREFIX,
DRIVER_CLIENT_PREFIX,
KUBERNETES_MASTER_INTERNAL_URL,
Some(new File(Config.KUBERNETES_SERVICE_ACCOUNT_TOKEN_PATH)),
Some(new File(Config.KUBERNETES_SERVICE_ACCOUNT_CA_CRT_PATH)))
} else {
(KUBERNETES_AUTH_CLIENT_MODE_PREFIX,
CLIENT_MODE_CLIENT_PREFIX,
KubernetesUtils.parseMasterUrl(masterURL),
None,
None)
Expand All @@ -65,6 +68,7 @@ private[spark] class KubernetesClusterManager extends ExternalClusterManager wit
apiServerUri,
Some(sc.conf.get(KUBERNETES_NAMESPACE)),
authConfPrefix,
clientConfPrefix,
sc.conf,
defaultServiceAccountToken,
defaultServiceAccountCaCrt)
Expand Down