dongjoon-hyun commented on code in PR #58574:
URL: https://github.com/apache/spark/pull/58574#discussion_r3954617551
##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -540,6 +553,14 @@ class SparkContext(config: SparkConf) extends Logging {
// the bound port to the cluster manager properly
_ui.foreach(_.bind())
+ // OIDC credential propagation: selection phase. When enabled, discover
and initialize the
+ // credential provider(s) and apply their declared Spark properties (e.g.
the S3A credentials
+ // provider class) into _conf, so that the driver's Hadoop Configuration
below -- and other
+ // config-derived components -- pick them up. This performs no credential
resolution/IO;
+ // actual acquisition happens later in the scheduler backend
(UserCredentialManager). The
+ // returned loader is retained so that resolution phase reuses the
initialized providers.
+ _userCredentialProviderLoader =
UserCredentialManager.applyProviderProperties(_conf)
Review Comment:
Separately from the above: since this phase depends only on `SparkConf` (it
never touches `_env`), is there a reason not to run it at the existing "This
should be set as early as possible" slot next to
`enableMagicCommitterIfNeeded(_conf)` (line ~447)? Right now it sits after the
`spark.logConf` dump (so the logged configuration omits the auto-configured
key), after `DriverLogger`, and after `createSparkEnv` (where `SecurityManager`
builds its own Hadoop `Configuration`). Moving it up would also let us drop the
new caveat in the `CredentialProvider` javadoc about components created earlier
in `SparkContext` initialization.
##########
core/src/main/scala/org/apache/spark/deploy/security/UserCredentialManager.scala:
##########
@@ -495,10 +473,103 @@ private[spark] object UserCredentialManager {
}
val tokenIngestor = new FileTokenIngestor(Paths.get(tokenFile))
- Some(new UserCredentialManager(sparkConf, tokenIngestor,
onCredentialsUpdate))
+ Some(new UserCredentialManager(sparkConf, tokenIngestor,
onCredentialsUpdate, loader))
}
}
+ /**
+ * Selection phase of OIDC credential propagation, run early during
`SparkContext`
+ * initialization (before the driver's Hadoop `Configuration` is
materialized).
+ *
+ * When OIDC credential propagation is enabled, this discovers and
initializes the
+ * `CredentialProvider` for each unambiguously-resolvable scheme and applies
the provider's
+ * [[CredentialProvider#additionalSparkProperties]] declarations into
`sparkConf` (only for
+ * keys the user has not already set). This is the driver-side counterpart
to the executor
+ * path (where these properties travel via `SparkAppConfig` before the
executor's environment
+ * is built): applying them into `sparkConf` here -- before `SparkContext`
materializes the
+ * driver's Hadoop `Configuration` and other config-derived components --
lets driver-side
+ * access (e.g. output-path existence checks, the commit protocol) use the
propagated
+ * credentials rather than falling back to the default credential chain.
+ *
+ * This phase performs NO credential resolution and NO network I/O: provider
discovery,
Review Comment:
The "NO network I/O ... `init()` ... are all I/O-free" claim does not hold
for the one shipped provider. `AwsStsCredentialProvider.init` builds an
`StsClient`; when neither `spark.security.oidc.aws.region` nor the STS endpoint
is configured, `resolveRegion` returns `null` and AWS SDK v2 resolves the
region eagerly in `build()` via `DefaultAwsRegionProviderChain`, whose last
link (`InstanceProfileRegionProvider`) makes HTTP calls to the EC2 IMDS with 3
attempts. I verified against `bundle-2.35.4`: with no region configured,
`build()` blocks ~0.8s and throws `SdkClientException("Unable to load region
...")` on a non-EC2 host (longer where IMDS traffic is dropped).
So on a dev box / Minikube / on-prem K8s / CI, `new SparkContext` now blocks
inside the "I/O-free" selection phase, the exception is swallowed at DEBUG (see
below), and in cluster mode the same probe is paid a second time in `start()`.
Since `additionalSparkProperties()` is a constant map that does not need
`init()`, one option is to obtain the properties without initializing the
provider in the selection phase, or to build the `StsClient` lazily in
`resolve()`. At minimum the I/O-free wording here and in `SparkContext` should
go.
##########
core/src/test/scala/org/apache/spark/deploy/security/UserCredentialManagerSuite.scala:
##########
@@ -611,90 +614,70 @@ class UserCredentialManagerSuite extends SparkFunSuite {
"stop() should close providers after credential renewal exits")
}
- // ========== additionalSparkProperties application ==========
+ // ========== additionalSparkProperties application (selection phase)
==========
- test("start() applies additionalSparkProperties from active providers") {
+ test("applyProviderProperties applies additionalSparkProperties from
selected providers") {
val conf = createSparkConf()
conf.set("spark.security.oidc.provider.fake",
"org.apache.spark.security.FakeCredentialProvider")
- val ctx = createUserContext()
- val manager = new UserCredentialManager(
- conf, createIngestor(ctx), (_, _) => ())
+ UserCredentialManager.applyProviderProperties(conf)
- try {
- manager.start()
- assert(conf.get("spark.hadoop.fs.fake.credentials.provider") ===
- "org.apache.spark.security.FakeExecutorCredentialProvider")
- } finally {
- manager.stop()
- }
+ // spark.hadoop.* property is applied ...
+ assert(conf.get("spark.hadoop.fs.fake.credentials.provider") ===
+ "org.apache.spark.security.FakeExecutorCredentialProvider")
+ // ... and so is a non-Hadoop spark.* property (type-agnostic; this is
what lets
+ // arbitrary provider-declared driver-side wiring, not just Hadoop FS
config, take effect).
+ assert(conf.get("spark.fake.credentials.enabled") === "true")
}
- test("start() does not overwrite user-set properties") {
+ test("applyProviderProperties is a no-op when OIDC is disabled") {
+ val conf = new SparkConf(false)
+ .set(SECURITY_OIDC_ENABLED, false)
+ UserCredentialManager.applyProviderProperties(conf)
+ assert(!conf.contains("spark.hadoop.fs.fake.credentials.provider"))
+ assert(!conf.contains("spark.fake.credentials.enabled"))
+ }
+
+ test("applyProviderProperties does not overwrite user-set properties") {
val conf = createSparkConf()
conf.set("spark.security.oidc.provider.fake",
"org.apache.spark.security.FakeCredentialProvider")
- // User explicitly sets the property before start()
+ // User explicitly sets the property beforehand.
conf.set("spark.hadoop.fs.fake.credentials.provider", "user.Custom")
- val ctx = createUserContext()
- val manager = new UserCredentialManager(
- conf, createIngestor(ctx), (_, _) => ())
+ UserCredentialManager.applyProviderProperties(conf)
- try {
- manager.start()
- // User-set value must NOT be overwritten
- assert(conf.get("spark.hadoop.fs.fake.credentials.provider") ===
"user.Custom")
- } finally {
- manager.stop()
- }
+ // User-set value must NOT be overwritten; the unset one is still applied.
+ assert(conf.get("spark.hadoop.fs.fake.credentials.provider") ===
"user.Custom")
+ assert(conf.get("spark.fake.credentials.enabled") === "true")
}
- test("start() handles provider returning null from
additionalSparkProperties") {
- // AnotherFakeCredentialProvider uses default (empty map), not null.
- // This test verifies the defensive null check doesn't crash
- // with a provider that inherits the default empty map.
+ test("applyProviderProperties skips ambiguous schemes without failing") {
Review Comment:
This is the only test that takes the zero-config `discoverAllSchemes()`
branch (every other new test sets `spark.security.oidc.provider.fake`
explicitly), and it asserts nothing. It would be good to assert that `fake`
(the single-candidate scheme) was auto-selected, e.g.
`conf.get("spark.hadoop.fs.fake.credentials.provider") ===
"org.apache.spark.security.FakeExecutorCredentialProvider"`; the removed
`start()`-level test used to cover that.
Also, no core test constructs a `SparkContext` to check that
`sc.hadoopConfiguration.get("fs.fake.credentials.provider")` is populated or
that the backend reuses `sc.userCredentialProviderLoader`
(`FakeCredentialProvider.getInitCount` should stay at 1 across selection +
`start()`). A `LocalSparkContext`-based test would catch a reorder of the call
below `_hadoopConfiguration` or a backend passing a fresh loader; today only
the Minikube-gated E2E suite would.
##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -540,6 +553,14 @@ class SparkContext(config: SparkConf) extends Logging {
// the bound port to the cluster manager properly
_ui.foreach(_.bind())
+ // OIDC credential propagation: selection phase. When enabled, discover
and initialize the
+ // credential provider(s) and apply their declared Spark properties (e.g.
the S3A credentials
+ // provider class) into _conf, so that the driver's Hadoop Configuration
below -- and other
+ // config-derived components -- pick them up. This performs no credential
resolution/IO;
+ // actual acquisition happens later in the scheduler backend
(UserCredentialManager). The
+ // returned loader is retained so that resolution phase reuses the
initialized providers.
+ _userCredentialProviderLoader =
UserCredentialManager.applyProviderProperties(_conf)
Review Comment:
This now runs for every master, but `UserCredentialManager` is only created
and started by `CoarseGrainedSchedulerBackend`; `LocalSchedulerBackend` never
starts one, so in `local[*]` the driver's Hadoop conf gets
`fs.s3a.aws.credentials.provider=SparkOidcAwsCredentialsProvider` while
`SparkEnv.userCredentials` stays `null`. Every driver-side `s3a://` access then
throws `IllegalStateException("No credentials available in the executor
credential store")` from `SparkOidcAwsCredentialsProvider.resolveCredentials`,
whereas before this PR (when the auto-config loop lived in `start()`) local
mode used the default AWS chain. That contradicts "Does this PR introduce any
user-facing change? No."
The same gap exists inside the constructor even in cluster mode: between
this line and `_taskScheduler.start()` (~line 669, where the credential store
is populated), `addJar`/`addFile` for `spark.jars`/`spark.files` run against a
provider that cannot resolve. `spark.jars=s3a://...` (standalone cluster mode,
`local`, or K8s with `spark.kubernetes.jars.avoidDownloadSchemes=s3a`) is
silently dropped by `checkRemoteJarFile`'s `NonFatal` catch, and
`spark.files=s3a://...` makes `new SparkContext` throw because `addFile` has no
catch around `fs.getFileStatus`.
Could we either skip the selection phase when no resolution phase will
follow (e.g. `isLocal`), or have `SparkContext` own the manager lifecycle so
the store is populated before anything consumes the Hadoop conf?
##########
core/src/main/scala/org/apache/spark/deploy/security/UserCredentialManager.scala:
##########
@@ -495,10 +473,103 @@ private[spark] object UserCredentialManager {
}
val tokenIngestor = new FileTokenIngestor(Paths.get(tokenFile))
- Some(new UserCredentialManager(sparkConf, tokenIngestor,
onCredentialsUpdate))
+ Some(new UserCredentialManager(sparkConf, tokenIngestor,
onCredentialsUpdate, loader))
}
}
+ /**
+ * Selection phase of OIDC credential propagation, run early during
`SparkContext`
+ * initialization (before the driver's Hadoop `Configuration` is
materialized).
+ *
+ * When OIDC credential propagation is enabled, this discovers and
initializes the
+ * `CredentialProvider` for each unambiguously-resolvable scheme and applies
the provider's
+ * [[CredentialProvider#additionalSparkProperties]] declarations into
`sparkConf` (only for
+ * keys the user has not already set). This is the driver-side counterpart
to the executor
+ * path (where these properties travel via `SparkAppConfig` before the
executor's environment
+ * is built): applying them into `sparkConf` here -- before `SparkContext`
materializes the
+ * driver's Hadoop `Configuration` and other config-derived components --
lets driver-side
+ * access (e.g. output-path existence checks, the commit protocol) use the
propagated
+ * credentials rather than falling back to the default credential chain.
+ *
+ * This phase performs NO credential resolution and NO network I/O: provider
discovery,
+ * `init()`, and `additionalSparkProperties()` are all I/O-free. Actual
credential
+ * acquisition (and renewal) happens later in [[start]] on the scheduler
backend. This
+ * separation of provider SELECTION from credential RESOLUTION is
intentional.
+ *
+ * Scheme selection is limited to schemes for which a provider is
UNAMBIGUOUSLY selected:
+ * either an explicitly-configured scheme
(`spark.security.oidc.provider.<scheme>`) or a
+ * scheme with exactly one candidate provider on the classpath. Schemes with
multiple
+ * candidates and no explicit configuration are skipped here (there is no
basis to choose
+ * which provider's properties to apply); they are left to [[start]], where
`providerFor`
+ * raises a clear error prompting explicit configuration. This preserves the
established
+ * behavior that properties are only contributed by providers the job
actually uses.
+ *
+ * @param sparkConf The Spark configuration to apply properties into. Not
modified when OIDC
+ * credential propagation is disabled.
+ * @return the [[CredentialProviderLoader]] used, to be passed to [[create]]
so the resolution
+ * phase reuses the same (already-initialized) provider instances.
+ */
+ def applyProviderProperties(sparkConf: SparkConf): CredentialProviderLoader
= {
+ val loader = new CredentialProviderLoader()
+ if (!sparkConf.get(SECURITY_OIDC_ENABLED)) {
+ return loader
+ }
+
+ val confMap = sparkConf.getAll
+ .filter { case (k, _) => k.startsWith("spark.security.oidc.") }
+ .toMap.asJava
+
+ // Determine candidate schemes: explicitly-configured schemes take
precedence; otherwise
+ // fall back to all schemes discoverable on the classpath
(single-candidate schemes are the
+ // intended zero-config case).
+ val explicitSchemes = explicitSchemesFrom(confMap)
+ val schemes =
+ if (explicitSchemes.nonEmpty) explicitSchemes
+ else loader.discoverAllSchemes().asScala.toSet
+
+ for (scheme <- schemes) {
+ try {
+ val providerOpt = loader.providerFor(scheme, confMap)
Review Comment:
Behavior change worth calling out: before this PR, properties were applied
only for `activeProviders`, i.e. providers whose `resolve()` returned a
credential. Now they are applied for every *selected* provider before any
resolution. With two schemes configured, if the s3a STS exchange fails (or
`resolve()` returns `null`) while the other scheme resolves, `start()` still
succeeds (`resolveCredentials` only throws when the map is empty), and S3A on
driver and executors is now hard-wired to `SparkOidcAwsCredentialsProvider`,
which throws `No credential found for scheme 's3a'`. Previously the key was
left unset and S3A fell back to the default chain.
The new `CredentialProvider` javadoc says this is intentional ("independent
of whether a specific resolution has yet succeeded"), which is fine, but then
the scaladoc above at "This preserves the established behavior that properties
are only contributed by providers the job actually uses" is no longer accurate,
and the "no user-facing change" answer in the description should be updated.
##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -337,6 +344,12 @@ class SparkContext(config: SparkConf) extends Logging {
*/
def hadoopConfiguration: Configuration = _hadoopConfiguration
+ // The CredentialProviderLoader from the OIDC selection phase, reused by the
credential
+ // resolution phase so providers are initialized exactly once. Null if OIDC
is disabled or
Review Comment:
Nit: this comment says "Null if OIDC is disabled", but
`applyProviderProperties` allocates the loader before checking
`SECURITY_OIDC_ENABLED` and returns it either way, so the field is never null
after initialization; every `SparkContext` (including all tests) stores an
unused loader and `stop()` always runs `closeAll()` on it. Either allocate only
inside the enabled branch and return `null`/`Option`, or fix the comment.
Relatedly, the comment in `UserCredentialManager.stop()` ("This loader belongs
to this manager, so closing it cannot affect a later SparkContext") is no
longer true now that the loader is owned and closed by `SparkContext` as well;
it would be cleaner to have a single owner for `closeAll()`.
##########
core/src/main/scala/org/apache/spark/deploy/security/UserCredentialManager.scala:
##########
@@ -495,10 +473,103 @@ private[spark] object UserCredentialManager {
}
val tokenIngestor = new FileTokenIngestor(Paths.get(tokenFile))
- Some(new UserCredentialManager(sparkConf, tokenIngestor,
onCredentialsUpdate))
+ Some(new UserCredentialManager(sparkConf, tokenIngestor,
onCredentialsUpdate, loader))
}
}
+ /**
+ * Selection phase of OIDC credential propagation, run early during
`SparkContext`
+ * initialization (before the driver's Hadoop `Configuration` is
materialized).
+ *
+ * When OIDC credential propagation is enabled, this discovers and
initializes the
+ * `CredentialProvider` for each unambiguously-resolvable scheme and applies
the provider's
+ * [[CredentialProvider#additionalSparkProperties]] declarations into
`sparkConf` (only for
+ * keys the user has not already set). This is the driver-side counterpart
to the executor
+ * path (where these properties travel via `SparkAppConfig` before the
executor's environment
+ * is built): applying them into `sparkConf` here -- before `SparkContext`
materializes the
+ * driver's Hadoop `Configuration` and other config-derived components --
lets driver-side
+ * access (e.g. output-path existence checks, the commit protocol) use the
propagated
+ * credentials rather than falling back to the default credential chain.
+ *
+ * This phase performs NO credential resolution and NO network I/O: provider
discovery,
+ * `init()`, and `additionalSparkProperties()` are all I/O-free. Actual
credential
+ * acquisition (and renewal) happens later in [[start]] on the scheduler
backend. This
+ * separation of provider SELECTION from credential RESOLUTION is
intentional.
+ *
+ * Scheme selection is limited to schemes for which a provider is
UNAMBIGUOUSLY selected:
+ * either an explicitly-configured scheme
(`spark.security.oidc.provider.<scheme>`) or a
+ * scheme with exactly one candidate provider on the classpath. Schemes with
multiple
+ * candidates and no explicit configuration are skipped here (there is no
basis to choose
+ * which provider's properties to apply); they are left to [[start]], where
`providerFor`
+ * raises a clear error prompting explicit configuration. This preserves the
established
+ * behavior that properties are only contributed by providers the job
actually uses.
+ *
+ * @param sparkConf The Spark configuration to apply properties into. Not
modified when OIDC
+ * credential propagation is disabled.
+ * @return the [[CredentialProviderLoader]] used, to be passed to [[create]]
so the resolution
+ * phase reuses the same (already-initialized) provider instances.
+ */
+ def applyProviderProperties(sparkConf: SparkConf): CredentialProviderLoader
= {
+ val loader = new CredentialProviderLoader()
+ if (!sparkConf.get(SECURITY_OIDC_ENABLED)) {
+ return loader
+ }
+
+ val confMap = sparkConf.getAll
+ .filter { case (k, _) => k.startsWith("spark.security.oidc.") }
+ .toMap.asJava
+
+ // Determine candidate schemes: explicitly-configured schemes take
precedence; otherwise
+ // fall back to all schemes discoverable on the classpath
(single-candidate schemes are the
+ // intended zero-config case).
+ val explicitSchemes = explicitSchemesFrom(confMap)
+ val schemes =
+ if (explicitSchemes.nonEmpty) explicitSchemes
+ else loader.discoverAllSchemes().asScala.toSet
+
+ for (scheme <- schemes) {
+ try {
+ val providerOpt = loader.providerFor(scheme, confMap)
+ if (providerOpt.isPresent) {
+ val provider = providerOpt.get()
+ val props = provider.additionalSparkProperties()
+ if (props != null) {
+ props.forEach { (key, value) =>
+ if (!sparkConf.contains(key)) {
+ sparkConf.set(key, value)
+ logInfo(log"Auto-configured ${MDC(LogKeys.CONFIG, key)} from "
+
+ log"${MDC(LogKeys.CLASS_NAME, provider.getClass.getName)} " +
+ log"(scheme: ${MDC(LogKeys.URI, scheme)})")
+ }
+ }
+ }
+ }
+ } catch {
+ case NonFatal(e) =>
Review Comment:
This catch downgrades what used to be a WARN to DEBUG, and the message says
"deferred to credential resolution", but `start()` no longer applies properties
at all, so nothing is deferred: any `NonFatal` failure here (a throwing
`init()` or `additionalSparkProperties()`) means the property is never set for
the whole application, driver and executors alike.
Concrete case: on EKS with no explicit `spark.security.oidc.aws.region`,
`AwsStsCredentialProvider.init` -> `StsClient.build()` region resolution hits a
transient IMDS error here. `providerFor` does not record the provider as
initialized and `config` stays `null`, so `start()` re-runs `init()`
successfully, resolves credentials, and logs "Credential acquisition
successful", yet `fs.s3a.aws.credentials.provider` is never set and S3A
silently uses the default chain, with only a DEBUG line explaining why. Before
this PR the property was applied right after that successful resolve, so this
could not happen.
I would suggest logging at WARN here and keeping a fallback application in
`start()` for providers that resolved (idempotent under
`!sparkConf.contains(key)`).
--
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]
