-
Notifications
You must be signed in to change notification settings - Fork 29.3k
[SPARK-39006][K8S] Show a directional error message for executor PVC dynamic allocation failure #36374
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.
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
[SPARK-39006][K8S] Show a directional error message for executor PVC dynamic allocation failure #36374
Changes from 1 commit
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -23,6 +23,7 @@ import io.fabric8.kubernetes.api.model._ | |
|
|
||
| import org.apache.spark.deploy.k8s._ | ||
| import org.apache.spark.deploy.k8s.Constants.{ENV_EXECUTOR_ID, SPARK_APP_ID_LABEL} | ||
| import org.apache.spark.internal.config.EXECUTOR_INSTANCES | ||
|
|
||
| private[spark] class MountVolumesFeatureStep(conf: KubernetesConf) | ||
| extends KubernetesFeatureConfigStep { | ||
|
|
@@ -71,6 +72,7 @@ private[spark] class MountVolumesFeatureStep(conf: KubernetesConf) | |
| case KubernetesPVCVolumeConf(claimNameTemplate, storageClass, size) => | ||
| val claimName = conf match { | ||
| case c: KubernetesExecutorConf => | ||
| checkPVCClaimNameWhenMultiExecutors(claimNameTemplate) | ||
| claimNameTemplate | ||
| .replaceAll(PVC_ON_DEMAND, | ||
| s"${conf.resourceNamePrefix}-exec-${c.executorId}$PVC_POSTFIX-$i") | ||
|
|
@@ -120,6 +122,20 @@ private[spark] class MountVolumesFeatureStep(conf: KubernetesConf) | |
| override def getAdditionalKubernetesResources(): Seq[HasMetadata] = { | ||
| additionalResources.toSeq | ||
| } | ||
|
|
||
| private def checkPVCClaimNameWhenMultiExecutors(claimName: String): Unit = { | ||
| val invalidClaimName = | ||
| if (!claimName.contains(PVC_ON_DEMAND) && !claimName.contains(ENV_EXECUTOR_ID)) true | ||
| else false | ||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This is misleading because this is totally valid when there is only one executor.
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Done, check it when multiple executors |
||
|
|
||
| val executorInstances = conf.get(EXECUTOR_INSTANCES) | ||
| if (executorInstances.isEmpty) return | ||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Please try to avoid
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Yeah, remove |
||
| if (invalidClaimName && executorInstances.get > 1) { | ||
| throw new IllegalArgumentException("PVC ClaimName should contain " + | ||
| PVC_ON_DEMAND + " or " + ENV_EXECUTOR_ID + | ||
| " when multiple executors are required") | ||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. The error message looks informative but please include the current claim name in the error message too.
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Have added the current claim name |
||
| } | ||
| } | ||
| } | ||
|
|
||
| private[spark] object MountVolumesFeatureStep { | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -18,8 +18,9 @@ package org.apache.spark.deploy.k8s.features | |
|
|
||
| import scala.collection.JavaConverters._ | ||
|
|
||
| import org.apache.spark.SparkFunSuite | ||
| import org.apache.spark.{SparkConf, SparkFunSuite} | ||
| import org.apache.spark.deploy.k8s._ | ||
| import org.apache.spark.internal.config.EXECUTOR_INSTANCES | ||
|
|
||
| class MountVolumesFeatureStepSuite extends SparkFunSuite { | ||
| test("Mounts hostPath volumes") { | ||
|
|
@@ -148,6 +149,26 @@ class MountVolumesFeatureStepSuite extends SparkFunSuite { | |
| assert(executorPVC.getClaimName.endsWith("-exec-1-pvc-0")) | ||
| } | ||
|
|
||
| test("SPARK-39006 Show a directional error message for PVC Dynamic Allocation Failure") { | ||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Have two changes here:
|
||
| val volumeConf = KubernetesVolumeSpec( | ||
| "testVolume", | ||
| "/tmp", | ||
| "", | ||
| mountReadOnly = true, | ||
| KubernetesPVCVolumeConf("testClaimName") | ||
| ) | ||
| val conf = new SparkConf().set(EXECUTOR_INSTANCES, 2) | ||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Please add a positive test case where executor instance is 1.
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Fine, added a positive test here |
||
| val executorConf = | ||
| KubernetesTestConf.createExecutorConf(sparkConf = conf, volumes = Seq(volumeConf)) | ||
| val executorStep = new MountVolumesFeatureStep(executorConf) | ||
| assertThrows[IllegalArgumentException] { | ||
| executorStep.configurePod(SparkPod.initialPod()) | ||
| } | ||
| assert(intercept[IllegalArgumentException] { | ||
| executorStep.configurePod(SparkPod.initialPod()) | ||
| }.getMessage.contains("PVC ClaimName should contain OnDemand or SPARK_EXECUTOR_ID")) | ||
| } | ||
|
|
||
| test("Mounts emptyDir") { | ||
| val volumeConf = KubernetesVolumeSpec( | ||
| "testVolume", | ||
|
|
||
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.
checkPVCClaimNameWhenMultiExecutors->checkPVCClaimNamebecause this PR checks all cases instead ofWhenMultiExecutorsand doesn't raise exception for single instance case.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.
Change it as
checkPVCClaimName:)