dongjoon-hyun commented on code in PR #58696:
URL: https://github.com/apache/spark/pull/58696#discussion_r4057909934
##########
core/src/main/scala/org/apache/spark/deploy/Client.scala:
##########
@@ -275,6 +277,24 @@ object Client {
// scalastyle:on println
new ClientApp().start(args, new SparkConf())
}
+
+ /**
+ * Environment variables to forward to the driver. Only variables whose name
starts with
+ * `SPARK_` are forwarded, excluding `SPARK_ENV_LOADED`, `SPARK_HOME`,
`SPARK_CONF_DIR`,
+ * `SPARK_LOCAL_IP` and `SPARK_LOCAL_HOSTNAME`. The rule is defined by
+ * `RestSubmissionClient.filterSystemEnvironment`, so this matches the REST
submission gateway.
+ * If `spark.standalone.submit.filterEnvironment` is disabled, the full
environment of the
+ * submitting process is forwarded instead.
+ */
+ private[deploy] def driverEnvironment(
Review Comment:
One flag is covering two unrelated concerns here: dropping non-Spark
variables (the point of this PR) and dropping the host-specific
`SPARK_LOCAL_IP` / `SPARK_LOCAL_HOSTNAME`.
With `spark.standalone.submit.filterEnvironment=false` the full `sys.env` is
forwarded again, which brings back exactly what SPARK-20025 fixed for the REST
path. Concretely: an operator on a bare-metal cluster without DNS sets
`SPARK_LOCAL_IP` on the submitting host, then flips this to `false` to get
`JAVA_HOME` / `HADOOP_CONF_DIR` back. The driver is then launched on a worker
with `SPARK_LOCAL_IP` pointing at the submitter's address and fails with
`BindException: Cannot assign requested address`; with `--supervise`, every
failover onto another node fails the same way.
Could we always drop `SPARK_LOCAL_(IP|HOSTNAME)` and let the flag control
only the `SPARK_` prefix filtering? That keeps the escape hatch for the new
behavior without reopening a known bug.
##########
core/src/main/scala/org/apache/spark/deploy/rest/StandaloneRestServer.scala:
##########
@@ -231,6 +231,10 @@ private[rest] class StandaloneSubmitRequestServlet(
_.replace(s":$masterRestPort", s":$masterPort")).getOrElse(masterUrl)
val appArgs = Option(request.appArgs).getOrElse(Array[String]())
// Filter SPARK_LOCAL_(IP|HOSTNAME) environment variables from being set
on the remote system.
+ // Since SPARK-59404, RestSubmissionClient.filterSystemEnvironment already
drops these on the
+ // client side, so this filter is duplicated. It is kept for backward
compatibility with
+ // older clients and with requests that do not come from
RestSubmissionClient.
+ // TODO(SPARK-59669): Remove this duplicated filter in Spark 5.x.
Review Comment:
I don't think this filter is duplicated, and I'd rather not schedule its
removal.
`RestSubmissionClient.filterSystemEnvironment` runs in the submitting
client's JVM. On the cluster side, neither `Master` nor `Worker` filters
`DriverDescription.command.environment` -- grepping for `SPARK_LOCAL` under
`core/src/main/scala/org/apache/spark/deploy/` returns only this one site. So
this is the only place the cluster sanitizes a client-supplied environment, not
a redundant copy of a client-side check.
If the TODO is acted on, any HTTP client posting to the Master REST port can
put `SPARK_LOCAL_IP` into `environmentVariables`, and the Worker passes it
straight into the driver's `ProcessBuilder`. An unmodified older `spark-submit`
reproduces it without any malicious intent. The second half of the comment
already says as much ("kept for backward compatibility with older clients"),
which reads as contradicting the TODO.
Suggest dropping the TODO line and describing this as the server-side check
instead.
##########
core/src/test/scala/org/apache/spark/deploy/ClientSuite.scala:
##########
@@ -48,4 +56,73 @@ class ClientSuite extends SparkFunSuite with Matchers {
// Invalid syntax.
ClientArguments.isValidJarUrl("hdfs:") should be (false)
}
+
+ /**
+ * Launches a driver through a [[ClientEndpoint]] wired to a fake master and
returns the
+ * [[Command]] carried by the [[RequestSubmitDriver]] message the client
sends.
+ */
+ private def submittedCommand(conf: SparkConf): Command = {
+ val env = RpcEnv.create("ClientSuite", "localhost", 0, conf, new
SecurityManager(conf))
+ try {
+ val submitted = Promise[RequestSubmitDriver]()
+ val master = env.setupEndpoint(Master.ENDPOINT_NAME, new RpcEndpoint {
+ override val rpcEnv: RpcEnv = env
+ // Record the submission without replying, so the client neither polls
the driver
+ // status nor exits the JVM.
+ override def receiveAndReply(context: RpcCallContext):
PartialFunction[Any, Unit] = {
+ case request: RequestSubmitDriver => submitted.success(request)
+ }
+ })
+ val args = new ClientArguments(
+ Array("launch", "spark://localhost:7077", "file:///path/to/app.jar",
"MainClass"))
+ env.setupEndpoint("client", new ClientEndpoint(env, args, Seq(master),
conf))
Review Comment:
This instantiates a real `ClientEndpoint`, whose `onError` calls
`System.exit(-1)`. `Inbox`'s `safelyCall` routes anything thrown from `onStart`
into `onError`, so a failure inside `onStart` -- say
`ResourceUtils.parseResourceRequirements` throwing because a
`spark.driver.resource.*` system property leaked in from an earlier suite in
the same forked JVM -- terminates the whole `core` test JVM instead of failing
this test, and surfaces as an opaque JVM exit with no pointer back here.
Using `new SparkConf(false)` removes the system-property path and makes
these two tests deterministic.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
