alexandrefimov commented on issue #7511:
URL: https://github.com/apache/kyuubi/issues/7511#issuecomment-5215521163
## Spark Connect in the Kyuubi Spark engine — design for the first slice
Spark Connect does not use `user_id` for anything except the session key: a
plan
is executed and authorized as the driver's own user. That single fact shapes
most
of what follows, so the evidence first.
Static checks on 4.0.3, 4.1.2 and 4.2.0: the `org.apache.spark.sql.connect`
package references neither `UserGroupInformation` nor `CurrentUserContext`,
and
`ExecuteThreadRunner` sets only `callSite.short` / `callSite.long` and job
tags.
Two runtime checks, both on 4.0.3: a driver started with `--proxy-user
testalice`
serving a client connected as `user_id=bob` returns `testalice` from
`current_user()` and `session_user()` — `SPARK_USER` was unset, so that value
comes from `UserGroupInformation.getCurrentUser` on the planning thread; and
with
`kyuubi-spark-authz` loaded into the same driver, a `CREATE TABLE` from the
`bob`
client is denied with `Permission denied: user [testalice] does not have
[create]
privilege on [default/poc_authz_probe]`.
At USER share level with `kyuubi.engine.doAs.enabled=true` (the default), the
driver already runs as the end user, so authorization identity matches the
Thrift
path — by deployment topology rather than by propagation. Three limitations
remain, and I would rather document them than paper over them:
- with `doAs` disabled, the Thrift path still authorizes as the session user
through the `kyuubi.session.user` local property set in
`SparkOperation.withLocalProperties` (`SparkOperation.scala:164`) and read
first by `AuthZUtils.getAuthzUgi`; Connect sets no such property, so it
would
authorize as the server user — a regression relative to Thrift at the same
share level;
- Connect operations bypass `withLocalProperties` entirely
(`SparkOperation.scala:158-194`), so the scheduler pool — tenant resource
policies — the statement id and the session-user signature are absent for
Connect queries;
- with `kyuubi.session.user.sign.enabled`, `AuthZUtils.getAuthzUgi` calls
`verifyKyuubiSessionUser` with a null user and throws
`Invalid user identifier [null]` (`AuthZUtils.scala:51-53`, `:123-133`). It
fails closed, but the feature is unusable in those deployments.
### Scope
As you outlined it. Engines are still provisioned through the existing
session
path; this slice starts Connect inside the engine and advertises the
endpoint.
Routing a Connect client to an engine is gateway-side work and out of scope,
together with YARN pool integration, share levels other than USER, and token
rotation.
### Starting Connect
`spark-connect_2.13-<version>.jar` ships in the standard Spark 4.x
distribution
and `org.apache.spark.sql.connect.SparkConnectPlugin` implements
`SparkPlugin`,
so starting the service is configuration only. `SparkProcessBuilder` appends
the
plugin class to `spark.plugins` — appends, since users set it for Delta and
others — along with the Connect settings, and no Spark Connect code or
dependency
enters `kyuubi-server`.
Reading state back — the bound port, the live sessions — does need the
Connect
classes, and that has to go through reflection: `SparkConnectService`'s Scala
signatures reference `io.grpc`, so a compile-time reference fails with
`Symbol 'term io.grpc' is missing from the classpath` unless grpc is added
to the
engine module's compile classpath. Reflection also keeps the default build
inert,
since the module compiles against the active Spark profile (`spark-3.5` by
default) and has no Spark-version-specific source directories.
The reflection targets are identical on 4.0.3, 4.1.2 and 4.2.0 —
`SparkConnectService$.localPort()` / `hostAddress()` / `started()`,
`sessionManager().listActiveSessions()`, `listActiveExecutions()`,
`stop(Option[Long], Option[TimeUnit])` — so I expect one implementation to
cover
the 4.x profiles in the tree. A capability check at startup fails the engine
with
an explicit message if they are missing, rather than degrading silently.
Configuration entry: `kyuubi.engine.spark.connect.enabled`, default `false`,
`version("1.12.0")`, immutable and admin-controlled. The engine key
(`EngineRef.engineSpace`, `EngineRef.scala:187-194`) does not include it, so
a
per-session value would let one user's sessions disagree about whether their
engine has Connect — the second session would silently reuse an engine with
no
endpoint. Immutable keeps the flag uniform per deployment; a client still
tells
the two apart by whether the endpoint attribute is present in discovery,
since
changing the flag does not retrofit a running engine. `settings.md` is
regenerated in the same PR as the entry.
### Discovery
`SparkTBinaryFrontendService.attributes`
(`externals/kyuubi-spark-sql-engine/.../SparkTBinaryFrontendService.scala:99-108`)
already publishes `kyuubi.engine.id` and `kyuubi.engine.url`. This slice
adds the
bound endpoint as `kyuubi.engine.connect.url` — a new `KyuubiReservedKeys`
constant next to `KYUUBI_ENGINE_URL` (`KyuubiReservedKeys.scala:34-36`) —
with
`host:port` as the value, read back after the plugin binds so that
`spark.connect.grpc.binding.port=0` works.
The value shape is not a style choice. Attributes are serialised into a
single
discovery node-name element, `k=v` joined with `;`
(`kyuubi-ha/.../ZookeeperDiscoveryClient.scala:361-366`, same in
`EtcdDiscoveryClient.scala:332-338`), so a value can contain neither `;` nor
`/`
— `=` is safe, since decoding splits with a limit of two
(`ZookeeperDiscoveryClient.scala:206-208`). That is why `kyuubi.engine.url`
is
already stored with its scheme stripped
(`SparkTBinaryFrontendService.scala:105-108`), and why the Connect endpoint
is
published without an `sc://` prefix. The same encoding means the endpoint
cannot
be deregistered independently of the Thrift node — removing one removes both,
which is what drain below relies on. Attributes also already surface through
`GET /api/v1/admin/engine` (`AdminResource.scala:377`), so operators see the
endpoint with no new REST work.
Ownership is already implied by the discovery path for USER share level
(`/serverSpace_version_USER_engineType/<user>[/<subdomain>]`), so this slice
publishes only the endpoint. Whether the attribute set should also carry an
explicit owner label is a question I would rather settle together with the
gateway-side placement work. No secrets are published here.
### Lifecycle
Engine termination is driven by `SessionManager.getActiveUserSessionCount`
(`kyuubi-common/.../session/SessionManager.scala:167`) through four call
sites:
the idle checker (`SessionManager.scala:346-373`), the max-lifetime checker
(`SparkSQLEngine.scala:165-214`), `gracefulStop`
(`SparkSQLEngine.scala:123-138`)
and `ServiceDiscovery.stopGracefully`
(`kyuubi-ha/.../ServiceDiscovery.scala:68-74`,
reached on discovery connection loss or node deletion — the `kyuubi-ctl`
path).
`startFastFailChecker` uses the same counter but only runs at CONNECTION
share
level (`SparkSQLEngine.scala:88-91`), so it is out of reach for this slice.
A Connect session is invisible to all of them. Connect exposes what is
needed:
`listActiveSessions()` returns per-session records carrying `userId` and
`lastAccessTimeMs`, and `listActiveExecutions()` returns either the active
executions or — when none is active — a `System.nanoTime()` reading of the
last
completion. That reading is monotonic and unrelated to wall clock, so it has
to
be converted against a nanoTime baseline before it can feed the idle timer,
which
compares against `System.currentTimeMillis()` (`SessionManager.scala:355`).
The plan:
`SparkSQLSessionManager` includes live Connect sessions in the active count,
which reaches all four call sites without touching `kyuubi-common`. The cost
is
that `getActiveUserSessionCount` acquires a second meaning: it no longer
agrees
with `allSessions()`, and its other consumers —
`ServiceDiscovery.stopGracefully`,
the "current opening sessions" log line, the engine UI — inherit that. The
alternative is a separate overridable predicate (`hasActiveWork` or similar)
consulted next to the counter, which keeps the counter honest at the cost of
touching every call site. See question 1.
Idle detection also needs a last-activity source. `latestLogoutTime` is
readable,
but it is advanced only inside `SessionManager.closeSession`
(`SessionManager.scala:124-131`) and at checker start (`:346-348`), so a
closing
Connect session leaves it untouched and the idle timer is already expired
when
the last one goes away.
The max-lifetime path needs an explicit answer rather than the counter alone.
After `maxLifetime + gracefulPeriod` it force-closes sessions drawn from
`allSessions()` (`SparkSQLEngine.scala:184-202`); Connect sessions are not in
`handleToSession`, so they would never be force-closed while the counter
never
reaches zero — an engine past its max lifetime with one idle-but-open Connect
session would never terminate. Either the graceful period force-closes
Connect
sessions through `SparkConnectService` as well, or the slice relies on
`spark.connect.session.manager.defaultSessionTimeout` and says so.
Drain: `gracefulStop` deregisters from discovery — which removes the Connect
endpoint with the node, as above — then waits for Connect sessions and
executions
to finish, and only then calls `SparkConnectService.stop(grace, unit)`. The
order
matters: that call starts with a gRPC `shutdown()` before awaiting
termination,
and afterwards a client is answered with `UNAVAILABLE`; since every Connect
statement is a separate RPC, stopping the service first would cut sessions
that
are still draining.
### Backend authentication
`spark.connect.authenticate.token` is present in the 4.0, 4.1 and 4.2 lines I
checked; on 4.0.3 a client connecting without it is rejected with
`UNAUTHENTICATED`. With Connect enabled the engine should refuse to start
unless
a token is configured — otherwise flipping the flag yields an open gRPC
endpoint
that accepts any `user_id` and executes with the engine owner's privileges,
which
is exactly the failure the section above describes. The check has to cover
the
runtime as well as the configuration: the key only exists from 4.0, while the
Connect plugin itself is present on 3.5, so a capability check that probes
only
the reflection targets would pass on a 3.5 runtime with the token silently
ignored. Gating on `KyuubiSparkUtil.SPARK_ENGINE_RUNTIME_VERSION >= "4.0"`
(`KyuubiSparkUtil.scala:114`) closes that.
On delivery: configuration reaching the engine goes through argv —
`SparkProcessBuilder.commands` expands
`conf.getEngineConf(EngineType.SPARK_SQL)`
into `--conf k=v` pairs (`SparkProcessBuilder.scala:133-157`) and
`Utils.redactCommandLineArgs` (`Utils.scala:329`) only touches the logged
copy —
so a SPARK-audience token key would be readable in `ps` by any user on the
host.
The precedent for avoiding that is already in the tree:
`FlinkProcessBuilder.generateTokenFile` (`FlinkProcessBuilder.scala:281-297`)
writes a generated credential to a file and passes the *path* through an env
var,
cleaning up in `close`. I would follow that shape, with the token placed into
`_sparkConf` before `createSpark()` (`SparkSQLEngine.scala:328-334`), since
the
plugin binds inside `getOrCreate`.
I am aware this argues against an existing pattern: engine credentials travel
through the launch conf today (`KyuubiSessionImpl.scala:137-139`, read back
at
`SparkSQLEngine.scala:328`). The difference I would draw is that those are
per-session Hadoop delegation tokens with their own expiry, while this one
is the
only thing standing between an unauthenticated client and an engine running
with
a specific user's privileges. Note also that the Spark UI and event logs are
covered by default — the engine's default `spark.redaction.regex` already
includes `token` (`SparkSQLEngine.scala:273-276`; an operator who overrides
the
pattern keeps that responsibility) — so the file buys exactly one thing,
keeping
the secret out of argv.
One token per engine rather than one per deployment, so that a leaked token
costs
a single engine rather than every engine sharing it. In cluster deployments
the
file still has to reach the driver host (staging dir, `spark.files`, or a
Secret);
which mechanism to use is open — see question 2.
### Tests and cost
The tests that exercise a live session — keep-alive, drain and the token
rejection — need a Spark Connect client on the test classpath. There is no
`spark-connect` dependency in the tree today, so that is a new test-scoped
dependency plus a runtime-version `assume` in the style of
`SparkOperationSuite.scala:200`. It is the largest hidden cost in the slice
and I
would rather name it up front than discover it in review.
With the flag off, behaviour is unchanged; with it on, the Connect port
binds and
the endpoint is advertised, and the node carries no secret. An engine with a
live
Connect session is not reaped by the idle timeout, and drain removes the
endpoint
before waiting for active work. A client connecting directly without the
token is
rejected.
### PRs
Tests ship with the feature in each PR, and `settings.md` is regenerated in
the
PR that adds the config entry.
1. start Connect: config entry, process builder, capability check,
`settings.md`;
2. advertise the endpoint through discovery;
3. lifecycle: keep-alive, max-lifetime handling and drain;
4. backend token at engine launch;
5. deployment documentation and the share-level follow-up note.
Happy to merge 1 and 2 if you would rather review one self-contained slice.
### Open questions
1. Connect activity in the lifecycle checkers: include Connect sessions in
`getActiveUserSessionCount` (reaches all four call sites, but the counter
stops agreeing with `allSessions()` for its other consumers), or add a
separate overridable predicate next to it and update the call sites?
2. Secret delivery: follow `FlinkProcessBuilder.generateTokenFile` —
generated
file plus path in an env var — and if so, which distribution mechanism for
cluster modes?
3. Fail at startup for non-USER share levels and for a missing token, as
proposed, or warn and continue?
4. The 4.0, 4.1 and 4.2 jobs already run the engine suites
(`.github/workflows/master.yml:64-81`), so I would gate the new tests with
`assume(SPARK_ENGINE_RUNTIME_VERSION >= "4.0")` and let them run in all
three — any reason to restrict that?
--
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]
