diff --git a/docs/configuration/settings.md b/docs/configuration/settings.md index 2562f0ef94c..7d48895297f 100644 --- a/docs/configuration/settings.md +++ b/docs/configuration/settings.md @@ -191,6 +191,7 @@ You can configure the Kyuubi properties in `$KYUUBI_HOME/conf/kyuubi-defaults.co | kyuubi.engine.share.level.sub.domain | <undefined> | (deprecated) - Using kyuubi.engine.share.level.subdomain instead | string | 1.2.0 | | kyuubi.engine.share.level.subdomain | <undefined> | Allow end-users to create a subdomain for the share level of an engine. A subdomain is a case-insensitive string values that must be a valid zookeeper subpath. For example, for the `USER` share level, an end-user can share a certain engine within a subdomain, not for all of its clients. End-users are free to create multiple engines in the `USER` share level. When disable engine pool, use 'default' if absent. | string | 1.4.0 | | kyuubi.engine.single.spark.session | false | When set to true, this engine is running in a single session mode. All the JDBC/ODBC connections share the temporary views, function registries, SQL configuration and the current database. | boolean | 1.3.0 | +| kyuubi.engine.spark.connect.enabled | false | When true, the Spark SQL engine starts a Spark Connect service next to its Thrift frontend, so that Spark Connect clients can use the same engine. It requires a Spark distribution whose Spark Connect supports authentication, which is Spark 4.0 and later, a pre-shared `spark.connect.authenticate.token`, and the USER share level; the engine fails to start otherwise. | boolean | 1.13.0 | | kyuubi.engine.spark.event.loggers | SPARK | A comma-separated list of engine loggers, where engine/session/operation etc events go. | seq | 1.7.0 | | kyuubi.engine.spark.initialize.sql | SHOW DATABASES | The initialize sql for Spark engine. It fallback to `kyuubi.engine.initialize.sql`. | seq | 1.8.1 | | kyuubi.engine.spark.operation.incremental.collect | false | When true, the result will be sequentially calculated and returned to the Spark driver. Note that, kyuubi.operation.result.max.rows will be ignored on incremental collect mode. It fallback to `kyuubi.operation.incremental.collect` | boolean | 1.10.0 | diff --git a/docs/deployment/spark/index.rst b/docs/deployment/spark/index.rst index 1beb48d6406..1ecfe1cbf7c 100644 --- a/docs/deployment/spark/index.rst +++ b/docs/deployment/spark/index.rst @@ -30,3 +30,4 @@ Even if you don't use Kyuubi, as a simple Spark user, I'm sure you'll find the n dynamic_allocation aqe large_query_results + spark_connect diff --git a/docs/deployment/spark/spark_connect.md b/docs/deployment/spark/spark_connect.md new file mode 100644 index 00000000000..e6dc284f240 --- /dev/null +++ b/docs/deployment/spark/spark_connect.md @@ -0,0 +1,78 @@ + + +# Spark Connect in the Spark SQL Engine + +The Spark SQL engine can start a Spark Connect service next to its Thrift frontend, so that +Spark Connect clients reach a Kyuubi-managed engine: the same `SparkContext`, the same session +extensions and so the same authorization rules a JDBC client gets. Spark gives each Connect session +its own `SparkSession`, so temporary views, registered functions and SQL configs are not shared with +the JDBC sessions of that engine. + +The feature is off by default and is enabled per engine: + +```properties +kyuubi.engine.spark.connect.enabled=true +spark.connect.authenticate.token= +``` + +`kyuubi.engine.spark.connect.enabled` is immutable and the server decides it per engine, so a +session cannot turn Spark Connect on by itself — neither under that key nor under the +`spark.kyuubi.` prefixed copy that the engine would otherwise read. + +## Requirements + +| Requirement | Why | +|----------------------------------------------------------------------------------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| Spark 4.0 or later, with the `spark-connect` jar in the distribution | `spark.connect.authenticate.token` was added in Spark 4.0. Spark 3.5 serves Spark Connect without authentication, so the engine refuses to start rather than open an endpoint that authenticates no one | +| `spark.connect.authenticate.token`, or the `SPARK_CONNECT_AUTHENTICATE_TOKEN` environment variable | Every client that reaches the engine is authenticated by this pre-shared token | +| `kyuubi.engine.share.level=USER` and `kyuubi.engine.doAs.enabled=true` | A plan submitted over Spark Connect runs as the user the engine runs as, not as the `user_id` the client sends. Only these two together make that the session user; an engine started any other way serves Thrift only and says so in its launch log | + +A runtime that cannot authenticate Spark Connect clients, or a missing token, fails the engine with +a message naming what is missing: the deployment asked for Connect and cannot have it. The share +level is the session's own choice, so an engine that would not run as the session user starts +without Connect instead of failing. + +## What the engine configures + +Unless the deployment sets them, the engine adds: + +- `spark.plugins` gains `org.apache.spark.sql.connect.SparkConnectPlugin`, which is how Spark + starts Connect inside the driver; +- `spark.connect.grpc.binding.port=0`, so that engines of different users on the same host do not + compete for a fixed port. The port that was bound is logged as `sc://:`. + +Configs the deployment sets are kept, so a fixed port or extra plugins keep working. + +## Limitations + +- The Spark Connect endpoint is not advertised in engine discovery yet, so clients need the + address from the engine log. +- Engine idle timeout and graceful shutdown still count Thrift sessions only. An engine whose only + client is a Spark Connect session can be terminated as idle. +- The token is passed to the engine like any other Spark config, so it is visible in the driver's + command line. +- On a Spark distribution without [SPARK-58658](https://issues.apache.org/jira/browse/SPARK-58658), + an authenticated Spark Connect client reads the engine's configuration back through the Config + RPC, and the secrets the server hands the engine are in it — + `spark.kyuubi.ha.zookeeper.auth.digest` among them. The engine names them in + `spark.redaction.regex`, but only the patched handler consults that pattern. The fix is merged + on every Spark branch and released in none: 4.0.4, 4.1.3 and 4.2.0 all predate it, so it arrives + in the next patch release of each line. +- Share levels other than `USER`, and engines started with `kyuubi.engine.doAs.enabled=false`, do + not serve Spark Connect. + diff --git a/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/SparkSQLEngine.scala b/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/SparkSQLEngine.scala index 485a955a755..5f3f66471c5 100644 --- a/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/SparkSQLEngine.scala +++ b/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/SparkSQLEngine.scala @@ -38,6 +38,7 @@ import org.apache.kyuubi.config.KyuubiConf._ import org.apache.kyuubi.config.KyuubiReservedKeys.{KYUUBI_ENGINE_SUBMIT_TIME_KEY, KYUUBI_ENGINE_URL} import org.apache.kyuubi.engine.ShareLevel import org.apache.kyuubi.engine.spark.SparkSQLEngine.{countDownLatch, currentEngine} +import org.apache.kyuubi.engine.spark.connect.SparkConnectLauncher import org.apache.kyuubi.engine.spark.events.{EngineEvent, EngineEventsStore, SparkEventHandlerRegister} import org.apache.kyuubi.engine.spark.session.{SparkSessionImpl, SparkSQLSessionManager} import org.apache.kyuubi.events.EventBus @@ -326,6 +327,10 @@ object SparkSQLEngine extends Logging { } def createSpark(): SparkSession = { + if (SparkConnectLauncher.isEnabled(kyuubiConf)) { + SparkConnectLauncher.setup(kyuubiConf, _sparkConf) + } + val engineCredentials = kyuubiConf.getOption(KyuubiReservedKeys.KYUUBI_ENGINE_CREDENTIALS_KEY) kyuubiConf.unset(KyuubiReservedKeys.KYUUBI_ENGINE_CREDENTIALS_KEY) _sparkConf.set(s"spark.${KyuubiReservedKeys.KYUUBI_ENGINE_CREDENTIALS_KEY}", "") @@ -407,6 +412,12 @@ object SparkSQLEngine extends Logging { startInitTimeoutChecker(submitTime, initTimeout) spark = createSpark() sparkSessionCreated.set(true) + if (SparkConnectLauncher.isEnabled(kyuubiConf)) { + SparkConnectLauncher.boundEndpoint(spark.sparkContext.getConf).foreach { + case (host, port) => + info(s"Spark Connect service is listening on sc://$host:$port") + } + } try { startEngine(spark) // blocking main thread diff --git a/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/connect/SparkConnectLauncher.scala b/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/connect/SparkConnectLauncher.scala new file mode 100644 index 00000000000..da451b181c9 --- /dev/null +++ b/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/connect/SparkConnectLauncher.scala @@ -0,0 +1,163 @@ +/* + * 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.kyuubi.engine.spark.connect + +import scala.util.Try +import scala.util.control.NonFatal + +import org.apache.spark.SparkConf + +import org.apache.kyuubi.{KyuubiException, Logging} +import org.apache.kyuubi.config.KyuubiConf +import org.apache.kyuubi.config.KyuubiConf.{ENGINE_SHARE_LEVEL, ENGINE_SPARK_CONNECT_ENABLED} +import org.apache.kyuubi.engine.ShareLevel +import org.apache.kyuubi.util.reflect.{DynFields, DynMethods, ReflectUtils} + +/** + * Starts Spark Connect inside the Spark SQL engine, next to the Thrift frontend, by writing the + * Spark configs that make the driver load the Connect plugin. + * + * Spark Connect classes are never referenced at compile time: `io.grpc` is relocated to + * `org.apache.kyuubi.shade.io.grpc` in this module, and the Scala signatures of + * `SparkConnectService` carry `io.grpc` types, so a direct reference does not compile. That also + * keeps the engine building against the default `spark-3.5` profile, where the feature stays off. + */ +object SparkConnectLauncher extends Logging { + + final val PLUGINS_KEY = "spark.plugins" + final val BINDING_PORT_KEY = "spark.connect.grpc.binding.port" + final val AUTHENTICATE_TOKEN_KEY = "spark.connect.authenticate.token" + final val AUTHENTICATE_TOKEN_ENV = "SPARK_CONNECT_AUTHENTICATE_TOKEN" + final val DRIVER_HOST_KEY = "spark.driver.host" + + final private val CONNECT_PLUGIN_CLASS = "org.apache.spark.sql.connect.SparkConnectPlugin" + final private val CONNECT_CONFIG_CLASS = "org.apache.spark.sql.connect.config.Connect$" + final private val CONNECT_SERVICE_CLASS = + "org.apache.spark.sql.connect.service.SparkConnectService$" + + def isEnabled(kyuubiConf: KyuubiConf): Boolean = kyuubiConf.get(ENGINE_SPARK_CONNECT_ENABLED) + + /** + * Only the USER share level is supported for now: a Spark Connect client authenticates with a + * token that is shared by everyone reaching the engine, and the plan runs as the engine's own + * user, so a shared engine would execute one user's plans under another user's identity. + * + * `SparkProcessBuilder` decides whether an engine serves Connect at all, where the share level + * and doAs are the server's own values; this is the engine refusing a combination that should + * never have been launched. + */ + def checkShareLevel(kyuubiConf: KyuubiConf): Unit = { + val shareLevel = kyuubiConf.get(ENGINE_SHARE_LEVEL) + if (shareLevel != ShareLevel.USER.toString) { + throw new KyuubiException( + s"${ENGINE_SPARK_CONNECT_ENABLED.key} requires ${ENGINE_SHARE_LEVEL.key}=" + + s"${ShareLevel.USER}, but it is $shareLevel. Other share levels are not supported yet.") + } + } + + /** + * Spark 3.5 ships Spark Connect without any authentication - `spark.connect.authenticate.token` + * arrived in 4.0 - so enabling it there would open an unauthenticated endpoint on the engine. + * The capability is probed instead of comparing versions. + */ + def checkRuntimeSupport( + pluginClass: String = CONNECT_PLUGIN_CLASS, + configClass: String = CONNECT_CONFIG_CLASS): Unit = { + if (!ReflectUtils.isClassLoadable(pluginClass)) { + throw new KyuubiException( + s"${ENGINE_SPARK_CONNECT_ENABLED.key} is set, but $pluginClass is not on the" + + " engine classpath. Spark Connect ships in the spark-connect jar of a Spark 4.0 or" + + " later distribution.") + } + if (!authenticationSupported(configClass)) { + throw new KyuubiException( + s"${ENGINE_SPARK_CONNECT_ENABLED.key} is set, but this Spark runtime has a Spark Connect" + + s" without authentication support - $AUTHENTICATE_TOKEN_KEY was introduced in Spark" + + " 4.0. Enabling it here would expose an unauthenticated endpoint.") + } + } + + /** + * The engine refuses to open the endpoint without a pre-shared token. Generating one at engine + * launch is a separate change; today the deployment supplies it. + * + * The config wins over the environment even when it is empty, because that is the order + * `Connect.getAuthenticateToken` reads them in: an empty config would install an interceptor + * with an empty key rather than fall back to the environment. + */ + def checkAuthenticateToken(sparkConf: SparkConf): Unit = { + val token = sparkConf.getOption(AUTHENTICATE_TOKEN_KEY) + .orElse(sys.env.get(AUTHENTICATE_TOKEN_ENV)) + if (!token.exists(_.nonEmpty)) { + throw new KyuubiException( + s"${ENGINE_SPARK_CONNECT_ENABLED.key} is set, but no Spark Connect token is configured." + + s" Set $AUTHENTICATE_TOKEN_KEY or the $AUTHENTICATE_TOKEN_ENV environment variable so" + + " that clients are authenticated.") + } + } + + /** + * The Spark configs to add so that the driver starts Connect. Configs the deployment set are + * kept as they are. + */ + def sparkConfOverrides(sparkConf: SparkConf): Seq[(String, String)] = { + val plugins = sparkConf.get(PLUGINS_KEY, "").split(",").map(_.trim).filter(_.nonEmpty) + val pluginOverride = + if (plugins.contains(CONNECT_PLUGIN_CLASS)) Nil + else Seq(PLUGINS_KEY -> (plugins :+ CONNECT_PLUGIN_CLASS).mkString(",")) + // Kyuubi runs one engine per user on a host, so bind an ephemeral port unless the deployment + // picked one; the port that was actually bound is read back afterwards. + val portOverride = + if (sparkConf.contains(BINDING_PORT_KEY)) Nil else Seq(BINDING_PORT_KEY -> "0") + pluginOverride ++ portOverride + } + + def setup(kyuubiConf: KyuubiConf, sparkConf: SparkConf): Unit = { + checkShareLevel(kyuubiConf) + checkRuntimeSupport() + checkAuthenticateToken(sparkConf) + sparkConfOverrides(sparkConf).foreach { case (key, value) => sparkConf.set(key, value) } + } + + /** + * The endpoint a Spark Connect client reaches, once the plugin has started the service. + * + * The host comes from `spark.driver.host` rather than `SparkConnectService.hostAddress`: the + * latter is the local canonical host name, which on Kubernetes is the pod name and does not + * resolve outside the pod. + */ + def boundEndpoint( + sparkConf: SparkConf, + serviceClass: String = CONNECT_SERVICE_CLASS): Option[(String, Int)] = { + try { + val service = DynFields.builder() + .impl(serviceClass, "MODULE$") + .buildChecked[Object]() + .get(null) + val port = ReflectUtils.invokeAs[Int](service, "localPort") + sparkConf.getOption(DRIVER_HOST_KEY).map(_ -> port) + } catch { + case NonFatal(e) => + warn("Failed to read the Spark Connect endpoint back from the driver", e) + None + } + } + + private[connect] def authenticationSupported(configClass: String): Boolean = + Try(DynMethods.builder("getAuthenticateToken").impl(configClass).buildChecked()).isSuccess +} diff --git a/externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/connect/FakeConnectClasses.scala b/externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/connect/FakeConnectClasses.scala new file mode 100644 index 00000000000..8e8864eec15 --- /dev/null +++ b/externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/connect/FakeConnectClasses.scala @@ -0,0 +1,35 @@ +/* + * 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.kyuubi.engine.spark.connect + +/** + * Stand-ins for the Spark Connect classes the engine reaches by reflection, so that the reflection + * itself is exercised. The engine module does not depend on `spark-connect`, and adding the real + * jar to the test classpath is a dependency problem of its own. + */ +object FakeConnectPlugin + +/** Shaped like `org.apache.spark.sql.connect.config.Connect` on a Spark that authenticates. */ +object FakeConnectConfig { + def getAuthenticateToken: Option[String] = Some("a-token") +} + +/** Shaped like `org.apache.spark.sql.connect.service.SparkConnectService` with a bound port. */ +object FakeConnectService { + def localPort: Int = 15002 +} diff --git a/externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/connect/SparkConnectLauncherSuite.scala b/externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/connect/SparkConnectLauncherSuite.scala new file mode 100644 index 00000000000..05998f38fa1 --- /dev/null +++ b/externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/connect/SparkConnectLauncherSuite.scala @@ -0,0 +1,112 @@ +/* + * 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.kyuubi.engine.spark.connect + +import org.apache.spark.SparkConf + +import org.apache.kyuubi.{KyuubiException, KyuubiFunSuite} +import org.apache.kyuubi.config.KyuubiConf +import org.apache.kyuubi.config.KyuubiConf.{ENGINE_SHARE_LEVEL, ENGINE_SPARK_CONNECT_ENABLED} +import org.apache.kyuubi.engine.ShareLevel +import org.apache.kyuubi.engine.spark.connect.SparkConnectLauncher._ + +class SparkConnectLauncherSuite extends KyuubiFunSuite { + + private val connectPlugin = "org.apache.spark.sql.connect.SparkConnectPlugin" + private val fakePluginClass = FakeConnectPlugin.getClass.getName + private val fakeConfigClass = FakeConnectConfig.getClass.getName + private val fakeServiceClass = FakeConnectService.getClass.getName + + test("disabled unless the engine is configured for it") { + assert(!isEnabled(KyuubiConf())) + assert(isEnabled(KyuubiConf().set(ENGINE_SPARK_CONNECT_ENABLED, true))) + } + + test("add the Connect plugin and bind an ephemeral port") { + val overrides = sparkConfOverrides(new SparkConf(false)).toMap + assert(overrides(PLUGINS_KEY) === connectPlugin) + assert(overrides(BINDING_PORT_KEY) === "0") + } + + test("keep the plugins and the port the deployment configured") { + val sparkConf = new SparkConf(false) + .set(PLUGINS_KEY, "org.apache.spark.custom.Plugin") + .set(BINDING_PORT_KEY, "15002") + val overrides = sparkConfOverrides(sparkConf).toMap + assert(overrides(PLUGINS_KEY) === s"org.apache.spark.custom.Plugin,$connectPlugin") + assert(!overrides.contains(BINDING_PORT_KEY)) + } + + test("do not add the Connect plugin twice") { + val sparkConf = new SparkConf(false).set(PLUGINS_KEY, connectPlugin) + assert(!sparkConfOverrides(sparkConf).toMap.contains(PLUGINS_KEY)) + } + + test("only the USER share level is supported") { + checkShareLevel(KyuubiConf().set(ENGINE_SHARE_LEVEL, ShareLevel.USER.toString)) + Seq(ShareLevel.CONNECTION, ShareLevel.GROUP, ShareLevel.SERVER).foreach { shareLevel => + val e = intercept[KyuubiException] { + checkShareLevel(KyuubiConf().set(ENGINE_SHARE_LEVEL, shareLevel.toString)) + } + assert(e.getMessage.contains(s"requires ${ENGINE_SHARE_LEVEL.key}=${ShareLevel.USER}")) + } + } + + test("refuse to open the endpoint without a token") { + val e = intercept[KyuubiException] { + checkAuthenticateToken(new SparkConf(false)) + } + assert(e.getMessage.contains(AUTHENTICATE_TOKEN_KEY)) + checkAuthenticateToken(new SparkConf(false).set(AUTHENTICATE_TOKEN_KEY, "a-token")) + } + + test("an empty token is no token, the way Spark reads it") { + // Connect.getAuthenticateToken takes the config first and only falls back to the environment + // when it is absent, so an empty config value authenticates nobody + intercept[KyuubiException] { + checkAuthenticateToken(new SparkConf(false).set(AUTHENTICATE_TOKEN_KEY, "")) + } + } + + test("probe the runtime for Spark Connect authentication support") { + assert(authenticationSupported(fakeConfigClass)) + assert(!authenticationSupported(fakeServiceClass)) + assert(!authenticationSupported("org.apache.kyuubi.engine.spark.connect.NoSuchClass$")) + } + + test("accept a runtime whose Spark Connect authenticates, refuse one that does not") { + checkRuntimeSupport(fakePluginClass, fakeConfigClass) + val e = intercept[KyuubiException](checkRuntimeSupport(fakePluginClass, fakeServiceClass)) + assert(e.getMessage.contains("without authentication support")) + } + + test("read the bound endpoint back from the Connect service") { + val sparkConf = new SparkConf(false).set(DRIVER_HOST_KEY, "10.0.0.1") + assert(boundEndpoint(sparkConf, fakeServiceClass) === Some("10.0.0.1" -> 15002)) + // nothing to advertise when the driver host is unknown + assert(boundEndpoint(new SparkConf(false), fakeServiceClass).isEmpty) + // nor when the service is not there at all + assert(boundEndpoint(sparkConf).isEmpty) + } + + test("refuse to start when the Spark distribution has no Spark Connect") { + // the engine module does not depend on spark-connect, so this is the 'no Connect' runtime + val e = intercept[KyuubiException](checkRuntimeSupport()) + assert(e.getMessage.contains(connectPlugin)) + } +} diff --git a/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala b/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala index 9fff9ee32f4..2ed228120bf 100644 --- a/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala +++ b/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala @@ -4151,6 +4151,18 @@ object KyuubiConf { .toSequence() .createWithDefault(Seq("spark.driver.memory", "spark.executor.memory")) + val ENGINE_SPARK_CONNECT_ENABLED: ConfigEntry[Boolean] = + buildConf("kyuubi.engine.spark.connect.enabled") + .doc("When true, the Spark SQL engine starts a Spark Connect service next to its Thrift" + + " frontend, so that Spark Connect clients can use the same engine. It requires a Spark" + + " distribution whose Spark Connect supports authentication, which is Spark 4.0 and" + + " later, a pre-shared `spark.connect.authenticate.token`, and the USER share level;" + + " the engine fails to start otherwise.") + .version("1.13.0") + .immutable + .booleanConf + .createWithDefault(false) + val ENGINE_SPARK_INITIALIZE_SQL: ConfigEntry[Seq[String]] = buildConf("kyuubi.engine.spark.initialize.sql") .doc("The initialize sql for Spark engine. It fallback to `kyuubi.engine.initialize.sql`.") diff --git a/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilder.scala b/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilder.scala index 560e659d917..c1372e12ce5 100644 --- a/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilder.scala +++ b/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilder.scala @@ -34,7 +34,7 @@ import org.apache.hadoop.security.UserGroupInformation import org.apache.kyuubi._ import org.apache.kyuubi.config.KyuubiConf import org.apache.kyuubi.config.KyuubiConf._ -import org.apache.kyuubi.engine.{ApplicationManagerInfo, EngineType, KyuubiApplicationManager, ProcBuilder} +import org.apache.kyuubi.engine.{ApplicationManagerInfo, EngineType, KyuubiApplicationManager, ProcBuilder, ShareLevel} import org.apache.kyuubi.engine.KubernetesApplicationOperation.{KUBERNETES_SERVICE_HOST, KUBERNETES_SERVICE_PORT} import org.apache.kyuubi.engine.ProcBuilder.KYUUBI_ENGINE_LOG_PATH_KEY import org.apache.kyuubi.ha.HighAvailabilityConf @@ -146,6 +146,13 @@ class SparkProcessBuilder( if (AuthTypes.withName(conf.get(HA_ZK_ENGINE_AUTH_TYPE)) == AuthTypes.KERBEROS) { allConf = allConf ++ zkAuthKeytabFileConf(allConf) } + // The engine rebuilds its own KyuubiConf from every `spark.kyuubi.*` entry it is handed, so + // this prefixed copy would turn Spark Connect on inside the engine without the server ever + // agreeing - `kyuubi.engine.spark.connect.enabled` is immutable under its own name only. + allConf = allConf - convertConfigKey(ENGINE_SPARK_CONNECT_ENABLED.key) + if (!sparkConnectEnabled) { + allConf = allConf - ENGINE_SPARK_CONNECT_ENABLED.key + } // pass spark engine log path to spark conf (allConf ++ engineLogPathConf ++ @@ -163,6 +170,25 @@ class SparkProcessBuilder( buffer } + /** + * Whether this engine serves Spark Connect. The server decides, because a Spark Connect client + * is authorized as the user the engine runs as - its `user_id` never reaches execution - and + * only a `USER` engine with doAs runs as the session user. Other engines start without Connect + * rather than failing, since the share level is the session's own choice. + */ + private[kyuubi] lazy val sparkConnectEnabled: Boolean = { + conf.get(ENGINE_SPARK_CONNECT_ENABLED) && { + val shareLevel = conf.get(ENGINE_SHARE_LEVEL) + val runsAsSessionUser = shareLevel == ShareLevel.USER.toString && doAsEnabled + if (!runsAsSessionUser) { + warn(s"Not starting Spark Connect for this engine: it serves the session user only with" + + s" ${ENGINE_SHARE_LEVEL.key}=${ShareLevel.USER} and ${ENGINE_DO_AS_ENABLED.key}=true," + + s" which are $shareLevel and $doAsEnabled here") + } + runsAsSessionUser + } + } + override protected def module: String = "kyuubi-spark-sql-engine" protected def setupKerberos(buffer: mutable.Buffer[String]): Unit = { diff --git a/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilderSuite.scala b/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilderSuite.scala index be24c53c20c..6216d886791 100644 --- a/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilderSuite.scala +++ b/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilderSuite.scala @@ -31,6 +31,7 @@ import org.apache.kyuubi._ import org.apache.kyuubi.config.KyuubiConf import org.apache.kyuubi.config.KyuubiConf._ import org.apache.kyuubi.engine.ProcBuilder.KYUUBI_ENGINE_LOG_PATH_KEY +import org.apache.kyuubi.engine.ShareLevel import org.apache.kyuubi.engine.spark.SparkProcessBuilder._ import org.apache.kyuubi.ha.HighAvailabilityConf import org.apache.kyuubi.ha.client.AuthTypes @@ -59,6 +60,42 @@ class SparkProcessBuilderSuite extends KerberizedTestHelper with MockitoSugar { process.destroyForcibly() } + test("the server decides that an engine serves Spark Connect") { + val builder = new SparkProcessBuilder( + "kentyao", + true, + conf.set(ENGINE_SPARK_CONNECT_ENABLED, true) + .set(ENGINE_SHARE_LEVEL, ShareLevel.USER.toString)) + assert(builder.sparkConnectEnabled) + assert(builder.commands.exists(_ == s"spark.${ENGINE_SPARK_CONNECT_ENABLED.key}=true")) + } + + test("a session cannot turn Spark Connect on through the spark.kyuubi prefix") { + // the engine rebuilds its KyuubiConf from `spark.kyuubi.*`, where the key is not immutable + val builder = new SparkProcessBuilder( + "kentyao", + true, + conf.set(s"spark.${ENGINE_SPARK_CONNECT_ENABLED.key}", "true")) + assert(!builder.sparkConnectEnabled) + assert(!builder.commands.exists(_.contains(ENGINE_SPARK_CONNECT_ENABLED.key))) + } + + test("an engine that does not run as the session user serves Thrift only") { + val sharedEngine = new SparkProcessBuilder( + "kentyao", + true, + conf.set(ENGINE_SPARK_CONNECT_ENABLED, true) + .set(ENGINE_SHARE_LEVEL, ShareLevel.CONNECTION.toString)) + assert(!sharedEngine.sparkConnectEnabled) + assert(!sharedEngine.commands.exists(_.contains(ENGINE_SPARK_CONNECT_ENABLED.key))) + + val withoutDoAs = new SparkProcessBuilder( + "kentyao", + false, + conf.set(ENGINE_SPARK_CONNECT_ENABLED, true)) + assert(!withoutDoAs.sparkConnectEnabled) + } + test("capture error from spark process builder") { val processBuilder = new SparkProcessBuilder("kentyao", true, conf.set("spark.ui.port", "abc")) processBuilder.start