diff --git a/core/src/main/scala/org/apache/spark/SparkContext.scala b/core/src/main/scala/org/apache/spark/SparkContext.scala index 5c8c30e2807c7..e1a1de5b865df 100644 --- a/core/src/main/scala/org/apache/spark/SparkContext.scala +++ b/core/src/main/scala/org/apache/spark/SparkContext.scala @@ -228,7 +228,8 @@ class SparkContext(config: SparkConf) extends Logging { // The CredentialProviderLoader created by the OIDC selection phase (applyProviderProperties), // retained so the later credential resolution phase (UserCredentialManager, started by the // scheduler backend) reuses the same loader. None when OIDC credential propagation is - // disabled or in local mode (the selection phase is skipped and allocates no loader). + // disabled (the selection phase is skipped and allocates no loader). The selection phase runs + // in local mode as well, since LocalSchedulerBackend now starts a resolution phase. // SparkContext is the single owner of this loader and is responsible for closing it in stop(). private var _userCredentialProviderLoader: Option[CredentialProviderLoader] = None private var _executorMemory: Int = _ @@ -348,7 +349,7 @@ class SparkContext(config: SparkConf) extends Logging { // The CredentialProviderLoader from the OIDC selection phase, reused by the credential // resolution phase so providers are initialized exactly once. `None` when OIDC credential - // propagation is disabled, in local mode, or before initialization. Internal. + // propagation is disabled or before initialization. Internal. private[spark] def userCredentialProviderLoader: Option[CredentialProviderLoader] = _userCredentialProviderLoader @@ -448,17 +449,18 @@ class SparkContext(config: SparkConf) extends Logging { // This should be set as early as possible. SparkContext.enableMagicCommitterIfNeeded(_conf) - // OIDC credential propagation: provider SELECTION phase. When enabled (and not in local - // mode), discover the credential provider(s) for the configured scheme(s) and apply their - // declared Spark properties (e.g. the S3A credentials provider class) into _conf, so that - // the driver's Hadoop Configuration built later -- and other config-derived components -- - // pick them up. This is done here, at the "as early as possible" slot, so the applied keys - // are visible to the spark.logConf dump and to any Hadoop Configuration built during - // createSparkEnv (e.g. by SecurityManager). It performs no credential resolution and no - // network I/O (providers are selected without init()); actual acquisition happens later in - // the scheduler backend (UserCredentialManager). Any returned loader is retained so the - // resolution phase reuses it, and SparkContext closes it in stop(). - _userCredentialProviderLoader = UserCredentialManager.applyProviderProperties(_conf, isLocal) + // OIDC credential propagation: provider SELECTION phase. When enabled, discover the + // credential provider(s) for the configured scheme(s) and apply their declared Spark + // properties (e.g. the S3A credentials provider class) into _conf, so that the driver's + // Hadoop Configuration built later -- and other config-derived components -- pick them up. + // This is done here, at the "as early as possible" slot, so the applied keys are visible to + // the spark.logConf dump and to any Hadoop Configuration built during createSparkEnv (e.g. + // by SecurityManager). It performs no credential resolution and no network I/O (providers + // are selected without init()); actual acquisition happens later in the scheduler backend + // (UserCredentialManager) -- including in local mode, where LocalSchedulerBackend now starts + // a resolution phase. Any returned loader is retained so the resolution phase reuses it, and + // SparkContext closes it in stop(). + _userCredentialProviderLoader = UserCredentialManager.applyProviderProperties(_conf) SparkContext.supplementJavaModuleOptions(_conf) SparkContext.supplementJavaIPv6Options(_conf) diff --git a/core/src/main/scala/org/apache/spark/deploy/security/UserCredentialManager.scala b/core/src/main/scala/org/apache/spark/deploy/security/UserCredentialManager.scala index 77ab05ac21d26..9d6a72a48884f 100644 --- a/core/src/main/scala/org/apache/spark/deploy/security/UserCredentialManager.scala +++ b/core/src/main/scala/org/apache/spark/deploy/security/UserCredentialManager.scala @@ -51,8 +51,9 @@ import org.apache.spark.util.{ThreadUtils, Utils} * safetyMargin` * 5. Retries with exponential backoff on failure * - * Intended to be started from `CoarseGrainedSchedulerBackend.start()` when - * `spark.security.oidc.enabled=true`, independently of + * Intended to be started from a scheduler backend -- `CoarseGrainedSchedulerBackend.start()` or + * `LocalSchedulerBackend.start()` (both via `SupportsDelegationToken.setupUserCredentialManager()`) + * -- when `spark.security.oidc.enabled=true`, independently of * `UserGroupInformation.isSecurityEnabled()`. * * Lifecycle: call `start()` exactly once, then `stop()` to shut down. @@ -482,11 +483,12 @@ private[spark] object UserCredentialManager extends Logging { * @param onCredentialsUpdate Callback to propagate credentials to executors * @param loader The [[CredentialProviderLoader]] from the selection phase * ([[applyProviderProperties]]), passed as an `Option`. When OIDC is enabled it - * must be `Some` (the selection phase, which runs earlier and is skipped only - * when OIDC is disabled or in local mode, produced it); reusing that same - * instance ensures providers are discovered and initialized exactly once, so the - * resolution phase reuses the already-selected providers. It is an error for the - * loader to be `None` while OIDC is enabled and a resolution phase is expected. + * must be `Some` (the selection phase, which runs earlier whenever OIDC is + * enabled -- in local mode as well -- and is skipped only when OIDC is disabled, + * produced it); reusing that same instance ensures providers are discovered and + * initialized exactly once, so the resolution phase reuses the already-selected + * providers. It is an error for the loader to be `None` while OIDC is enabled and + * a resolution phase is expected. * @return Some(manager) if enabled, None otherwise */ def create( @@ -503,7 +505,7 @@ private[spark] object UserCredentialManager extends Logging { "OIDC credential propagation is enabled but no CredentialProviderLoader was produced " + "by the selection phase. This indicates the selection phase " + "(UserCredentialManager.applyProviderProperties) did not run before the resolution " + - "phase, which should not happen outside local mode.") + "phase, which should not happen.") } val tokenFile = sparkConf.get(SECURITY_OIDC_IDENTITY_TOKEN_FILE).getOrElse { throw new IllegalArgumentException( @@ -538,13 +540,18 @@ private[spark] object UserCredentialManager extends Logging { * renewal) happen later in [[start]] on the scheduler backend. This separation of provider * SELECTION from credential RESOLUTION is intentional. * - * The phase is skipped entirely when `isLocal` is true: `LocalSchedulerBackend` does not start - * a [[UserCredentialManager]], so no resolution phase follows and no credentials are ever - * populated. Wiring a provider class into the driver's Hadoop `Configuration` in that case - * would make driver-side access fail (the provider would find no credentials) instead of - * falling back to the default chain. (Running credential resolution in local mode -- for - * parity with `HadoopDelegationTokenManager`, which does run in `LocalSchedulerBackend` -- is - * left to a follow-up.) + * This phase runs in local mode as well: `LocalSchedulerBackend` starts a + * [[UserCredentialManager]] (for parity with `HadoopDelegationTokenManager`, which also runs + * in `LocalSchedulerBackend`), so a resolution phase follows and populates the credentials the + * wiring points at. As in cluster mode, the resolution phase runs later than this selection + * phase (at scheduler-backend start), so there is a driver-side early-startup window: any + * driver-side access to a wired scheme during `SparkContext` construction -- for example + * fetching `spark.jars` / `spark.files` / `spark.archives`, or a `spark.checkpoint.dir` on such + * a scheme -- happens before the credentials exist and therefore cannot use them. This is the + * same driver-side window that exists in cluster mode; prefer `local://` for such resources. + * (Before SPARK-59296's follow-up, local mode had no resolution phase; the wiring applied here + * would never have been backed by resolved credentials, so this phase returned early in local + * mode.) * * Scheme selection is limited to schemes for which a provider is UNAMBIGUOUSLY selected: * either an explicitly-configured scheme (`spark.security.oidc.provider.`) or a @@ -554,17 +561,14 @@ private[spark] object UserCredentialManager extends Logging { * raises a clear error prompting explicit configuration. * * @param sparkConf The Spark configuration to apply properties into. Not modified when OIDC - * credential propagation is disabled or when `isLocal` is true. - * @param isLocal Whether the application runs in local mode (no scheduler backend that starts - * a resolution phase). + * credential propagation is disabled. * @return `Some(loader)` with the [[CredentialProviderLoader]] used, to be passed to * [[create]] so the resolution phase reuses the same loader; `None` when OIDC is - * disabled or when `isLocal` is true (no loader is allocated in those cases). + * disabled (no loader is allocated in that case). */ def applyProviderProperties( - sparkConf: SparkConf, - isLocal: Boolean): Option[CredentialProviderLoader] = { - if (!sparkConf.get(SECURITY_OIDC_ENABLED) || isLocal) { + sparkConf: SparkConf): Option[CredentialProviderLoader] = { + if (!sparkConf.get(SECURITY_OIDC_ENABLED)) { return None } diff --git a/core/src/main/scala/org/apache/spark/scheduler/SupportsDelegationToken.scala b/core/src/main/scala/org/apache/spark/scheduler/SupportsDelegationToken.scala index 6edcc7806e01c..27c00840afbe3 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/SupportsDelegationToken.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/SupportsDelegationToken.scala @@ -20,7 +20,7 @@ package org.apache.spark.scheduler import org.apache.hadoop.security.UserGroupInformation import org.apache.spark.deploy.SparkHadoopUtil -import org.apache.spark.deploy.security.HadoopDelegationTokenManager +import org.apache.spark.deploy.security.{HadoopDelegationTokenManager, UserCredentialManager} /** * A mix-in trait for SchedulerBackend that supports delegation tokens. @@ -30,6 +30,20 @@ private[spark] trait SupportsDelegationToken { // The token manager used to create security tokens. protected var delegationTokenManager: Option[HadoopDelegationTokenManager] = None + // The OIDC user-credential manager, a sibling of the Kerberos delegation token manager. + // Its lifecycle lives here (like delegationTokenManager) so that both + // CoarseGrainedSchedulerBackend and LocalSchedulerBackend share a single implementation; the + // two backends differ only in how they propagate credentials, expressed via + // propagateUserCredentials(). + protected var userCredentialManager: Option[UserCredentialManager] = None + + /** + * The task scheduler this backend belongs to. Implemented by the mixing-in backend (both + * CoarseGrainedSchedulerBackend and LocalSchedulerBackend already hold a `scheduler` field). + * Used to reach `scheduler.sc.conf` / `scheduler.sc.env` / the OIDC credential provider loader. + */ + protected def scheduler: TaskSchedulerImpl + /** * Create the delegation token manager to be used for the application. This method is called * once during the start of the scheduler backend (so after the object has already been @@ -42,6 +56,15 @@ private[spark] trait SupportsDelegationToken { */ protected def updateDelegationTokens(tokens: Array[Byte]): Unit + /** + * Propagate a freshly acquired set of OIDC user credentials. Called on the driver by the + * [[UserCredentialManager]] (initially from `start()`, then on each renewal). Implemented per + * backend: `CoarseGrainedSchedulerBackend` updates the driver store and broadcasts to executors + * via its `DriverEndpoint`; `LocalSchedulerBackend` (no remote executors) updates the shared + * credential store directly. + */ + protected def propagateUserCredentials(version: Long, credentials: Array[Byte]): Unit + /** * Whether the token manager should be started. The default implementation returns true when * Hadoop security is enabled. Backends that support direct credential providers override this @@ -75,4 +98,26 @@ private[spark] trait SupportsDelegationToken { protected def stopTokenManager(): Unit = { delegationTokenManager.foreach(_.stop()) } + + /** + * Start the [[UserCredentialManager]] if OIDC credential propagation is enabled. Called once + * during scheduler-backend start, independently of the Kerberos delegation token manager. + * + * Binds the manager to `scheduler.sc.conf` (the live SparkConf, not a clone) so the + * resolution-time fallback in `UserCredentialManager.start()` -- which applies provider-declared + * `spark.*` properties for any scheme the selection phase could not -- reaches the same conf on + * both backends. Reuses the [[org.apache.spark.security.CredentialProviderLoader]] produced by + * SparkContext's selection phase so providers are initialized exactly once; SparkContext remains + * the loader's single owner. `start()` invokes `propagateUserCredentials` synchronously for the + * initial credentials, so no separate initial store is needed here. + */ + protected def setupUserCredentialManager(): Unit = { + userCredentialManager = UserCredentialManager.create( + scheduler.sc.conf, propagateUserCredentials, scheduler.sc.userCredentialProviderLoader) + userCredentialManager.foreach(_.start()) + } + + protected def stopUserCredentialManager(): Unit = { + userCredentialManager.foreach(_.stop()) + } } diff --git a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala index d56da893f86fc..7bd3e49a09c8c 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala @@ -28,7 +28,6 @@ import com.google.common.cache.CacheBuilder import org.apache.spark.{ExecutorAllocationClient, SparkEnv, TaskState, VersionedCredentials} import org.apache.spark.deploy.SparkHadoopUtil -import org.apache.spark.deploy.security.UserCredentialManager import org.apache.spark.errors.SparkCoreErrors import org.apache.spark.executor.ExecutorLogUrlHandler import org.apache.spark.internal.{config, Logging} @@ -54,7 +53,7 @@ import org.apache.spark.util.ArrayImplicits._ * Spark's standalone deploy mode (spark.deploy.*). */ private[spark] -class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: RpcEnv) +class CoarseGrainedSchedulerBackend(protected val scheduler: TaskSchedulerImpl, val rpcEnv: RpcEnv) extends ExecutorAllocationClient with SchedulerBackend with SupportsDelegationToken with Logging { @@ -144,9 +143,6 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp // Current set of delegation tokens to send to executors. private val delegationTokens = new AtomicReference[Array[Byte]]() - // UserCredentialManager for OIDC credential propagation (if enabled). - private var userCredentialManager: Option[UserCredentialManager] = None - private val reviveThread = ThreadUtils.newDaemonSingleThreadScheduledExecutor("driver-revive-thread") @@ -1252,42 +1248,26 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp * Called from the DriverEndpoint receive loop (thread-safe access to executorDataMap). */ private def updateUserCredentials(version: Long, credentials: Array[Byte]): Unit = { - VersionedCredentials.updateIfNewer(SparkEnv.get.userCredentials, version, credentials) + VersionedCredentials.updateIfNewer(scheduler.sc.env.userCredentials, version, credentials) executorDataMap.values.foreach { ed => ed.executorEndpoint.send(UpdateUserCredentials(version, credentials)) } } /** - * Start the UserCredentialManager if OIDC credential propagation is enabled. - * Called from start(), independently of Kerberos/HadoopDelegationTokenManager. + * Propagate OIDC user credentials to executors. Called on the driver by the + * [[org.apache.spark.deploy.security.UserCredentialManager]] (initially and on each renewal). + * + * Updates the driver's own credential store synchronously so the credentials are available for + * `SparkAppConfig` (late-registering executors) and `TaskDescription` (task dispatch) with no + * null window, then broadcasts to registered executors via the `DriverEndpoint` (mirroring + * `HadoopDelegationTokenManager`'s `UpdateDelegationTokens` path) to ensure thread-safe access + * to `executorDataMap`. */ - private def setupUserCredentialManager(): Unit = { - // Reuse the loader from SparkContext's selection phase (Some when OIDC is enabled and not - // in local mode; None otherwise). Passing the Option straight through keeps SparkContext as - // the single owner of the loader: create() enforces that an enabled configuration has a - // loader, rather than silently allocating one here that no one would close. - userCredentialManager = UserCredentialManager.create(conf, { (version, credentials) => - // Send to DriverEndpoint to ensure thread-safe access to executorDataMap. - // This mirrors HadoopDelegationTokenManager's pattern of sending - // UpdateDelegationTokens via schedulerRef. - driverEndpoint.send(UpdateUserCredentials(version, credentials)) - }, scheduler.sc.userCredentialProviderLoader) - userCredentialManager.foreach { manager => - val (version, initialCredentials) = manager.start() - // Store initial credentials synchronously so they are available for SparkAppConfig - // (late-registering executors) and TaskDescription (task dispatch) immediately. - // Note: the onCredentialsUpdate callback above also triggers an async - // UpdateUserCredentials message that will redundantly call updateIfNewer. - // The synchronous set here ensures no null window before the async message - // is processed by DriverEndpoint. - VersionedCredentials.updateIfNewer( - SparkEnv.get.userCredentials, version, initialCredentials) - } - } - - private def stopUserCredentialManager(): Unit = { - userCredentialManager.foreach(_.stop()) + override protected def propagateUserCredentials( + version: Long, credentials: Array[Byte]): Unit = { + VersionedCredentials.updateIfNewer(scheduler.sc.env.userCredentials, version, credentials) + driverEndpoint.send(UpdateUserCredentials(version, credentials)) } /** diff --git a/core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala b/core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala index b6a90565aaff4..6d9bd5f5e6c78 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala @@ -21,7 +21,7 @@ import java.io.File import java.net.URL import java.nio.ByteBuffer -import org.apache.spark.{SparkConf, SparkContext, SparkEnv, TaskState} +import org.apache.spark.{SparkConf, SparkContext, SparkEnv, TaskState, VersionedCredentials} import org.apache.spark.TaskState.TaskState import org.apache.spark.deploy.SparkHadoopUtil import org.apache.spark.deploy.security.HadoopDelegationTokenManager @@ -112,12 +112,17 @@ private[spark] class LocalEndpoint( */ private[spark] class LocalSchedulerBackend( conf: SparkConf, - scheduler: TaskSchedulerImpl, + protected val scheduler: TaskSchedulerImpl, val totalCores: Int) extends SchedulerBackend with ExecutorBackend with SupportsDelegationToken with Logging { private val appId = conf.get("spark.test.appId", "local-" + System.currentTimeMillis) private var localEndpoint: RpcEndpointRef = null + // Set true by stop() so that a stop request arriving before start() (e.g. a launcher KILLED + // request; launcherBackend.connect() runs in the constructor and onStopRequest fires on its own + // thread) prevents start() from bringing up the executor endpoint and, in particular, the + // UserCredentialManager renewal thread (which performs network I/O) on an already-killed app. + @volatile private var stopped = false private val userClassPath = getUserClasspath(conf) private val listenerBus = scheduler.sc.listenerBus private val launcherBackend = new LauncherBackend() { @@ -137,6 +142,25 @@ private[spark] class LocalSchedulerBackend( SparkHadoopUtil.get.addDelegationTokens(tokens, conf) } + /** + * Propagate OIDC user credentials in local mode. There are no remote executors here: the driver + * and the single in-JVM executor share `scheduler.sc.env.userCredentials`, so this updates that + * store directly and subsequently dispatched tasks (and driver-side access) observe the new + * credentials. `UserCredentialManager.start()` calls this synchronously for the initial + * credentials, so they are in the store before any task is offered. + * + * The store is taken from `scheduler.sc.env` (this backend's own env), not the global + * `SparkEnv.get`: `UserCredentialManager.stop()` only waits a bounded time for the renewal + * thread, so a renewal that outlives this SparkContext must not write into a different + * SparkContext's store (e.g. a new context created in the same JVM by a notebook, test, or + * Spark Connect session). Binding to this env's store confines a late renewal to this + * application's (by then unused) store. + */ + override protected def propagateUserCredentials( + version: Long, credentials: Array[Byte]): Unit = { + VersionedCredentials.updateIfNewer(scheduler.sc.env.userCredentials, version, credentials) + } + /** * Returns a list of URLs representing the user classpath. * @@ -150,12 +174,19 @@ private[spark] class LocalSchedulerBackend( launcherBackend.connect() override def start(): Unit = { + // If a stop request already arrived (e.g. launcher KILLED before start()), do not bring up the + // executor endpoint or the token/credential managers on an app that is already stopping. + if (stopped) { + logInfo("Not starting LocalSchedulerBackend because it was already stopped") + return + } val rpcEnv = SparkEnv.get.rpcEnv val executorEndpoint = new LocalEndpoint(rpcEnv, userClassPath, scheduler, this, totalCores) localEndpoint = rpcEnv.setupEndpoint("LocalSchedulerBackendEndpoint", executorEndpoint) // call this after localEndpoint is assigned setupTokenManager() + setupUserCredentialManager() listenerBus.post(SparkListenerExecutorAdded( System.currentTimeMillis, @@ -197,8 +228,26 @@ private[spark] class LocalSchedulerBackend( } private def stop(finalState: SparkAppHandle.State): Unit = { - localEndpoint.ask(StopExecutor) - stopTokenManager() + // Mark stopped so a start() that has not run yet becomes a no-op (a KILLED request can race + // ahead of start()). The executor endpoint only needs stopping if start() already created it; + // guard on localEndpoint being non-null, mirroring CoarseGrainedSchedulerBackend's + // `if (driverEndpoint != null)`. The token and user-credential managers are always stopped + // (each isolated with tryLogNonFatalError so a failure in one does not skip the other or the + // launcher state update below); the UserCredentialManager renewal thread must be shut down + // before SparkContext.stop() closes the shared CredentialProviderLoader, otherwise a renewal + // task could race against an already-closed loader. + stopped = true + if (localEndpoint != null) { + Utils.tryLogNonFatalError { + localEndpoint.ask(StopExecutor) + } + } + Utils.tryLogNonFatalError { + stopTokenManager() + } + Utils.tryLogNonFatalError { + stopUserCredentialManager() + } try { launcherBackend.setState(finalState) } finally { diff --git a/core/src/test/scala/org/apache/spark/deploy/security/UserCredentialManagerSuite.scala b/core/src/test/scala/org/apache/spark/deploy/security/UserCredentialManagerSuite.scala index a0caf562fa1ad..e8ff807e7173c 100644 --- a/core/src/test/scala/org/apache/spark/deploy/security/UserCredentialManagerSuite.scala +++ b/core/src/test/scala/org/apache/spark/deploy/security/UserCredentialManagerSuite.scala @@ -24,6 +24,7 @@ import java.util.concurrent.{CountDownLatch, TimeUnit} import java.util.concurrent.atomic.{AtomicInteger, AtomicReference} import scala.concurrent.duration._ +import scala.jdk.CollectionConverters._ import org.scalatest.concurrent.Eventually.{eventually, timeout} @@ -731,8 +732,8 @@ class UserCredentialManagerSuite extends SparkFunSuite { conf.set("spark.security.oidc.provider.fake", "org.apache.spark.security.FakeCredentialProvider") - val loader = UserCredentialManager.applyProviderProperties(conf, isLocal = false) - assert(loader.isDefined, "a loader should be returned when OIDC is enabled and not local") + val loader = UserCredentialManager.applyProviderProperties(conf) + assert(loader.isDefined, "a loader should be returned when OIDC is enabled") // spark.hadoop.* property is applied ... assert(conf.get("spark.hadoop.fs.fake.credentials.provider") === @@ -742,13 +743,63 @@ class UserCredentialManagerSuite extends SparkFunSuite { assert(conf.get("spark.fake.credentials.enabled") === "true") } + test("selection then resolution reuse one loader and initialize the provider exactly once") { + // Ordering invariant behind the driver-side fix: the selection phase + // (applyProviderProperties) selects the provider WITHOUT init() and applies its declared + // properties, then the resolution phase (start()) reuses the SAME loader so the provider is + // initialized exactly once. This mirrors what SparkContext (selection) and the scheduler + // backend (resolution) do at runtime -- including LocalSchedulerBackend now that local mode + // runs a resolution phase. + val conf = createSparkConf() + conf.set("spark.security.oidc.provider.fake", + "org.apache.spark.security.FakeCredentialProvider") + + // Selection phase: applies properties and returns the loader to reuse. + val loaderOpt = UserCredentialManager.applyProviderProperties(conf) + assert(loaderOpt.isDefined) + val loader = loaderOpt.get + assert(conf.get("spark.hadoop.fs.fake.credentials.provider") === + "org.apache.spark.security.FakeExecutorCredentialProvider") + + // Observe the SAME provider instance the loader caches, WITHOUT initializing it, and assert + // the selection phase left it uninitialized. Do not call providerFor() here: that would + // initialize the provider before start(), making the post-start assertion below only prove + // loader idempotency rather than that start() reused this selection-phase loader. + val confMap = conf.getAll + .filter { case (k, _) => k.startsWith("spark.security.oidc.") } + .toMap.asJava + val provider = loader.selectProviderForProperties("fake", confMap).get() + .asInstanceOf[FakeCredentialProvider] + assert(provider.getInitCount === 0, + "the selection phase must not initialize the provider") + + // Resolution phase reuses the SAME loader (passed explicitly, as UserCredentialManager.create + // does with the loader from the selection phase). A mock ingestor avoids needing a token file. + // start() calls providerFor internally, which performs the single init() on this same cached + // instance -- so getInitCount goes 0 -> 1. If start() had regressed to a fresh loader, it would + // initialize a DIFFERENT FakeCredentialProvider and this instance's count would stay 0. + val manager = new UserCredentialManager( + conf, createIngestor(createUserContext()), (_: Long, _: Array[Byte]) => (), loader) + try { + val (version, bytes) = manager.start() + assert(version === 1L) + val creds = UserCredentialManager.deserializeUserCredentials(bytes) + assert(creds.forScheme("fake").isPresent) + assert(provider.getInitCount === 1, + "resolution must initialize the selection-phase provider exactly once") + } finally { + manager.stop() + loader.closeAll() + } + } + test("applyProviderProperties auto-selects a single-candidate scheme with no explicit config") { // Zero-config path: no spark.security.oidc.provider. is set, so the loader falls // back to discoverAllSchemes(). "fake" has exactly one candidate (FakeCredentialProvider), // so it must be auto-selected and its declared properties applied. val conf = createSparkConf() - val loader = UserCredentialManager.applyProviderProperties(conf, isLocal = false) + val loader = UserCredentialManager.applyProviderProperties(conf) assert(loader.isDefined) assert(conf.get("spark.hadoop.fs.fake.credentials.provider") === "org.apache.spark.security.FakeExecutorCredentialProvider") @@ -758,25 +809,12 @@ class UserCredentialManagerSuite extends SparkFunSuite { test("applyProviderProperties is a no-op and returns None when OIDC is disabled") { val conf = new SparkConf(false) .set(SECURITY_OIDC_ENABLED, false) - val loader = UserCredentialManager.applyProviderProperties(conf, isLocal = false) + val loader = UserCredentialManager.applyProviderProperties(conf) assert(loader.isEmpty, "no loader should be allocated when OIDC is disabled") assert(!conf.contains("spark.hadoop.fs.fake.credentials.provider")) assert(!conf.contains("spark.fake.credentials.enabled")) } - test("applyProviderProperties is a no-op and returns None in local mode") { - // Even with OIDC enabled, local mode has no scheduler backend that starts the resolution - // phase, so wiring a provider would leave driver-side access unable to resolve credentials. - // The selection phase must be skipped and no loader allocated. - val conf = createSparkConf() - conf.set("spark.security.oidc.provider.fake", - "org.apache.spark.security.FakeCredentialProvider") - val loader = UserCredentialManager.applyProviderProperties(conf, isLocal = true) - assert(loader.isEmpty, "no loader should be allocated in local mode") - 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", @@ -784,7 +822,7 @@ class UserCredentialManagerSuite extends SparkFunSuite { // User explicitly sets the property beforehand. conf.set("spark.hadoop.fs.fake.credentials.provider", "user.Custom") - UserCredentialManager.applyProviderProperties(conf, isLocal = false) + UserCredentialManager.applyProviderProperties(conf) // User-set value must NOT be overwritten; the unset one is still applied. assert(conf.get("spark.hadoop.fs.fake.credentials.provider") === "user.Custom") @@ -798,7 +836,7 @@ class UserCredentialManagerSuite extends SparkFunSuite { val conf = createSparkConf() // Do not set spark.security.oidc.provider.shared -> ambiguous for "shared". // No exception should escape. - UserCredentialManager.applyProviderProperties(conf, isLocal = false) + UserCredentialManager.applyProviderProperties(conf) // Nothing asserted about "shared"; the point is that the call returned normally. } @@ -813,7 +851,7 @@ class UserCredentialManagerSuite extends SparkFunSuite { try { // Must not fail even though AnotherFakeCredentialProvider throws; FakeCredentialProvider's // properties must still be applied (per-provider exception isolation). - UserCredentialManager.applyProviderProperties(conf, isLocal = false) + UserCredentialManager.applyProviderProperties(conf) assert(conf.get("spark.hadoop.fs.fake.credentials.provider") === "org.apache.spark.security.FakeExecutorCredentialProvider") } finally { diff --git a/core/src/test/scala/org/apache/spark/scheduler/local/LocalSchedulerBackendSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/local/LocalSchedulerBackendSuite.scala new file mode 100644 index 0000000000000..ab5a27b90a930 --- /dev/null +++ b/core/src/test/scala/org/apache/spark/scheduler/local/LocalSchedulerBackendSuite.scala @@ -0,0 +1,114 @@ +/* + * 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.spark.scheduler.local + +import java.io.File +import java.nio.charset.StandardCharsets +import java.nio.file.Files +import java.time.Instant +import java.util.Date + +import io.jsonwebtoken.Jwts + +import org.apache.spark.{LocalSparkContext, SparkConf, SparkContext, SparkEnv, SparkFunSuite} +import org.apache.spark.deploy.security.UserCredentialManager +import org.apache.spark.internal.config._ + +/** + * Tests that [[LocalSchedulerBackend]] starts a `UserCredentialManager` in local mode when OIDC + * credential propagation is enabled, for parity with `HadoopDelegationTokenManager` (which + * `LocalSchedulerBackend` already runs via `createTokenManager()`). This is the SPARK-59296 + * follow-up: before it, OIDC was a no-op in local mode. + */ +class LocalSchedulerBackendSuite extends SparkFunSuite with LocalSparkContext { + + /** Write an unsigned JWT (with the sub/iss/exp claims FileTokenIngestor requires) to `dir`. */ + private def writeTokenFile(dir: File): File = { + val jwt = Jwts.builder() + .subject("test-user") + .issuer("https://issuer.example.com") + .expiration(Date.from(Instant.now().plusSeconds(300))) + .compact() + val f = new File(dir, "oidc-token.jwt") + Files.write(f.toPath, jwt.getBytes(StandardCharsets.UTF_8)) + f + } + + private def oidcConf(enabled: Boolean, tokenFile: Option[File] = None): SparkConf = { + val conf = new SparkConf() + .setMaster("local[1]") + .setAppName("LocalSchedulerBackendSuite") + .set(SECURITY_OIDC_ENABLED, enabled) + // Short intervals so any renewal that fires during the test is cheap. + .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 5000L) + .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 1000L) + if (enabled) { + conf + .set(SECURITY_OIDC_IDENTITY_TOKEN_FILE, tokenFile.get.getAbsolutePath) + .set("spark.security.oidc.provider.fake", + "org.apache.spark.security.FakeCredentialProvider") + } + conf + } + + test("LocalSchedulerBackend runs OIDC selection and resolution in local mode") { + withTempDir { dir => + sc = new SparkContext(oidcConf(enabled = true, tokenFile = Some(writeTokenFile(dir)))) + + // The scheduler backend in local mode is a LocalSchedulerBackend. + assert(sc.schedulerBackend.isInstanceOf[LocalSchedulerBackend], + "local[1] should use LocalSchedulerBackend") + + // Selection phase ran on the driver (SparkContext) even in local mode: the provider's + // declared spark.hadoop.* property reached the driver's Hadoop Configuration (prefix + // stripped), and its non-Hadoop spark.* property reached SparkConf. + assert(sc.hadoopConfiguration.get("fs.fake.credentials.provider") === + "org.apache.spark.security.FakeExecutorCredentialProvider", + "selection phase should wire the provider into the driver's Hadoop Configuration") + assert(sc.getConf.get("spark.fake.credentials.enabled") === "true", + "selection phase should apply non-Hadoop provider properties too") + + // A loader was retained for reuse by the resolution phase. + assert(sc.userCredentialProviderLoader.isDefined, + "SparkContext should retain the selection-phase loader when OIDC is enabled") + + // Resolution phase ran in LocalSchedulerBackend: credentials were acquired and stored in + // the shared SparkEnv credential store (the same one tasks read from in local mode). + val stored = SparkEnv.get.userCredentials.get() + assert(stored != null, + "LocalSchedulerBackend should acquire and store OIDC credentials in local mode") + assert(stored.version >= 1L, "stored credentials should carry a version >= 1") + val creds = UserCredentialManager.deserializeUserCredentials(stored.bytes) + assert(creds.forScheme("fake").isPresent, + "stored credentials should contain the 'fake' scheme resolved by FakeCredentialProvider") + } + } + + test("LocalSchedulerBackend is a no-op for OIDC when disabled") { + sc = new SparkContext(oidcConf(enabled = false)) + + assert(sc.schedulerBackend.isInstanceOf[LocalSchedulerBackend]) + assert(sc.userCredentialProviderLoader.isEmpty, + "no loader should be allocated when OIDC is disabled") + assert(SparkEnv.get.userCredentials.get() == null, + "no OIDC credentials should be stored when OIDC is disabled") + // The provider's declared property must not have been applied. + assert(sc.hadoopConfiguration.get("fs.fake.credentials.provider") == null, + "no provider wiring should be applied when OIDC is disabled") + } +}