[SPARK-38954][CORE] Cloud Credential Refresh and Distribution Without Kerberos

core/src/main/scala/org/apache/spark/deploy/security/HadoopDelegationTokenManager.scala

@@ -34,6 +34,7 @@ import org.apache.spark.deploy.SparkHadoopUtil
 import org.apache.spark.internal.Logging
 import org.apache.spark.internal.LogKeys
 import org.apache.spark.internal.config._
+import org.apache.spark.internal.config.Network.NETWORK_CRYPTO_ENABLED
 import org.apache.spark.rpc.RpcEndpointRef
 import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages.UpdateDelegationTokens
 import org.apache.spark.security.HadoopDelegationTokenProvider
@@ -76,28 +77,52 @@ private[spark] class HadoopDelegationTokenManager(
   require((principal == null) == (keytab == null),
     "Both principal and keytab must be defined, or neither.")
 
+  if (sparkConf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED)) {
+    val hasEncryption = sparkConf.getBoolean("spark.ssl.rpc.enabled", false) ||
+      (sparkConf.get(NETWORK_AUTH_ENABLED) &&
+        (sparkConf.get(NETWORK_CRYPTO_ENABLED) || sparkConf.get(SASL_ENCRYPTION_ENABLED)))
+    require(hasEncryption,
+      "RPC channel encryption must be enabled when " +
+      "spark.security.credentials.directProviders.enabled is true: either " +
+      "spark.ssl.rpc.enabled=true, or spark.authenticate=true together with " +
+      "spark.network.crypto.enabled or spark.authenticate.enableSaslEncryption. " +
+      "Credential tokens must not be transmitted over unencrypted channels.")
+  }
+
   private val delegationTokenProviders = loadProviders()
+  if (sparkConf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED) && delegationTokenProviders.isEmpty) {
+    logWarning("spark.security.credentials.directProviders.enabled is true but " +
+      "no HadoopDelegationTokenProvider implementations were discovered. " +
+      "Ensure provider JARs are on the classpath with META-INF/services registration.")
+  }
   logDebug("Using the following builtin delegation token providers: " +
     s"${delegationTokenProviders.keys.mkString(", ")}.")
 
   private var renewalExecutor: ScheduledExecutorService = _
 
+  private def hasKerberosCredentials: Boolean =
+    sparkConf.get(KERBEROS_RENEWAL_CREDENTIALS) match {
+      case "keytab" => principal != null
+      case "ccache" => UserGroupInformation.getCurrentUser().hasKerberosCredentials()
+      case _ => false
+    }
+
   /** @return Whether delegation token renewal is enabled. */
-  def renewalEnabled: Boolean = sparkConf.get(KERBEROS_RENEWAL_CREDENTIALS) match {
-    case "keytab" => principal != null
-    case "ccache" => UserGroupInformation.getCurrentUser().hasKerberosCredentials()
-    case _ => false
+  def renewalEnabled: Boolean = {
+    hasKerberosCredentials ||
+      (sparkConf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED) && delegationTokenProviders.nonEmpty)
   }
 
   /**
-   * Start the token renewer. Requires a principal and keytab. Upon start, the renewer will
-   * obtain delegation tokens for all configured services and send them to the driver, and
-   * set up tasks to periodically get fresh tokens as needed.
+   * Start the token renewer. Upon start, the renewer will obtain delegation tokens for all
+   * configured services and send them to the driver, and set up tasks to periodically get
+   * fresh tokens as needed.
    *
-   * This method requires that a keytab has been provided to Spark, and will try to keep the
-   * logged in user's TGT valid while this manager is active.
+   * When Kerberos credentials are available, the manager keeps the TGT renewed and calls
+   * providers inside a privileged context. When only direct providers are configured, providers
+   * are called without Kerberos authentication.
    *
-   * @return New set of delegation tokens created for the configured principal.
+   * @return New set of delegation tokens created for the configured services.
    */
   def start(): Array[Byte] = {
     require(renewalEnabled, "Token renewal must be enabled to start the renewer.")
@@ -141,39 +166,58 @@ private[spark] class HadoopDelegationTokenManager(
     val hasKerberosCreds = principal != null ||
       Option(currentUser.getRealUser()).getOrElse(currentUser).hasKerberosCredentials()
 
-    // Delegation tokens can only be obtained if the real user has Kerberos credentials, so
-    // skip creation when those are not available.
     if (hasKerberosCreds) {
       val freshUGI = doLogin()
       freshUGI.doAs(new PrivilegedExceptionAction[Unit]() {
         override def run(): Unit = {
-          val (newTokens, _) = obtainDelegationTokens()
+          val (newTokens, _, _) = obtainDelegationTokens()
           creds.addAll(newTokens)
         }
       })
       if (!currentUser.equals(freshUGI)) {
         FileSystem.closeAllForUGI(freshUGI)
       }
+    } else if (sparkConf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED)) {
+      val (newTokens, _, _) = obtainDelegationTokens(isolateFailures = true)
+      creds.addAll(newTokens)
     }
   }
 
   /**
    * Fetch new delegation tokens for configured services.
    *
-   * @return 2-tuple (credentials with new tokens, time by which the tokens must be renewed)
+   * @param isolateFailures When true, catch per-provider exceptions and continue with remaining
+   *                        providers. When false, exceptions propagate to the caller.
+   * @return 3-tuple (credentials with new tokens, time by which the tokens must be renewed,
+   *         number of providers that failed). The failure count is always 0 when
+   *         isolateFailures is false (failures propagate instead).
    */
-  private def obtainDelegationTokens(): (Credentials, Long) = {
+  private def obtainDelegationTokens(
+      isolateFailures: Boolean = false): (Credentials, Long, Int) = {
     val creds = new Credentials()
+    var failureCount = 0
     val nextRenewal = delegationTokenProviders.values.flatMap { provider =>
       if (provider.delegationTokensRequired(sparkConf, hadoopConf)) {
-        provider.obtainDelegationTokens(hadoopConf, sparkConf, creds)
+        if (isolateFailures) {
+          try {
+            provider.obtainDelegationTokens(hadoopConf, sparkConf, creds)
+          } catch {
+            case e: Exception =>
+              logWarning(log"Failed to obtain credentials from " +
+                log"${MDC(LogKeys.SERVICE_NAME, provider.serviceName)}.", e)
+              failureCount += 1
+              None
+          }
+        } else {
+          provider.obtainDelegationTokens(hadoopConf, sparkConf, creds)
+        }
       } else {
         logDebug(s"Service ${provider.serviceName} does not require a token." +
           s" Check your configuration to see if security is disabled or not.")
         None
       }
     }.foldLeft(Long.MaxValue)(math.min)
-    (creds, nextRenewal)
+    (creds, nextRenewal, failureCount)
   }
 
   // Visible for testing.
@@ -199,8 +243,7 @@ private[spark] class HadoopDelegationTokenManager(
    */
   private def updateTokensTask(): Array[Byte] = {
     try {
-      val freshUGI = doLogin()
-      val creds = obtainTokensAndScheduleRenewal(freshUGI)
+      val creds = obtainTokensAndScheduleRenewal()
       val tokens = SparkHadoopUtil.get.serialize(creds)
 
       logInfo("Updating delegation tokens.")
@@ -226,24 +269,44 @@ private[spark] class HadoopDelegationTokenManager(
    *
    * @return Credentials containing the new tokens.
    */
-  private def obtainTokensAndScheduleRenewal(ugi: UserGroupInformation): Credentials = {
-    ugi.doAs(new PrivilegedExceptionAction[Credentials]() {
-      override def run(): Credentials = {
-        val (creds, nextRenewal) = obtainDelegationTokens()
-
-        // Calculate the time when new credentials should be created, based on the configured
-        // ratio.
-        val now = System.currentTimeMillis
-        val ratio = sparkConf.get(CREDENTIALS_RENEWAL_INTERVAL_RATIO)
-        val delay = (ratio * (nextRenewal - now)).toLong
-        logInfo(log"Calculated delay on renewal is ${MDC(LogKeys.DELAY, delay)}," +
-          log" based on next renewal ${MDC(LogKeys.NEXT_RENEWAL_TIME, nextRenewal)}" +
-          log" and the ratio ${MDC(LogKeys.CREDENTIALS_RENEWAL_INTERVAL_RATIO, ratio)}," +
-          log" and current time ${MDC(LogKeys.CURRENT_TIME, now)}")
-        scheduleRenewal(delay)
-        creds
+  private def obtainTokensAndScheduleRenewal(): Credentials = {
+    val (creds, nextRenewal, failureCount) = if (hasKerberosCredentials) {
+      val freshUGI = doLogin()
+      val currentUser = UserGroupInformation.getCurrentUser()
+      val result = freshUGI.doAs(
+        new PrivilegedExceptionAction[(Credentials, Long, Int)]() {
+          override def run(): (Credentials, Long, Int) = {
+            obtainDelegationTokens()
+          }
+        })
+      if (!currentUser.equals(freshUGI)) {
+        FileSystem.closeAllForUGI(freshUGI)
       }
-    })
+      result
+    } else if (sparkConf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED)) {
+      obtainDelegationTokens(isolateFailures = true)
+    } else {
+      // Reachable when ccache-based Kerberos credentials expired after start().
+      // Throw so that updateTokensTask() reschedules a retry instead of silently
+      // distributing empty credentials and never renewing again.
+      throw new IllegalStateException("Cannot obtain delegation tokens: no Kerberos " +
+        "credentials are available and direct credential providers are not enabled.")
+    }
+
+    // Scheduling does not require a privileged context (only token acquisition does).
+    val now = System.currentTimeMillis
+    val ratio = sparkConf.get(CREDENTIALS_RENEWAL_INTERVAL_RATIO)
+    val delay = if (failureCount > 0 && nextRenewal == Long.MaxValue) {
+      TimeUnit.SECONDS.toMillis(sparkConf.get(CREDENTIALS_RENEWAL_RETRY_WAIT))
+    } else {
+      (ratio * (nextRenewal - now)).toLong
+    }
+    logInfo(log"Calculated delay on renewal is ${MDC(LogKeys.DELAY, delay)}," +
+      log" based on next renewal ${MDC(LogKeys.NEXT_RENEWAL_TIME, nextRenewal)}" +
+      log" and the ratio ${MDC(LogKeys.CREDENTIALS_RENEWAL_INTERVAL_RATIO, ratio)}," +
+      log" and current time ${MDC(LogKeys.CURRENT_TIME, now)}")
+    scheduleRenewal(delay)
+    creds
   }
 
   private def doLogin(): UserGroupInformation = {

core/src/main/scala/org/apache/spark/internal/config/package.scala

@@ -1690,6 +1690,18 @@ package object config {
       .timeConf(TimeUnit.SECONDS)
       .createWithDefaultString("1h")
 
+  private[spark] val CREDENTIALS_DIRECT_PROVIDERS_ENABLED =
+    ConfigBuilder("spark.security.credentials.directProviders.enabled")
+      .doc(
+        "When true, enables delegation token collection and renewal without Kerberos. " +
+        "Providers are called directly (without doLogin/doAs) and participate in the " +
+        "same renewal and distribution lifecycle as Kerberos delegation token providers. " +
+        "Providers that require Kerberos self-gate via delegationTokensRequired.")
+      .version("4.3.0")
+      .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+      .booleanConf
+      .createWithDefault(false)
+
   private[spark] val SHUFFLE_SORT_INIT_BUFFER_SIZE =
     ConfigBuilder("spark.shuffle.sort.initialBufferSize")
       .internal()

core/src/main/scala/org/apache/spark/scheduler/SupportsDelegationToken.scala

@@ -33,7 +33,7 @@ private[spark] trait SupportsDelegationToken {
   /**
    * 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
-   * fully constructed), only if security is enabled in the Hadoop configuration.
+   * fully constructed), when security is enabled or direct credential providers are configured.
    */
   protected def createTokenManager(): Option[HadoopDelegationTokenManager] = None
 
@@ -42,8 +42,14 @@ private[spark] trait SupportsDelegationToken {
    */
   protected def updateDelegationTokens(tokens: Array[Byte]): Unit
 
+  /**
+   * Whether the token manager should be started. Returns true if Hadoop security is enabled
+   * or if direct credential providers are configured.
+   */
+  protected def tokenManagerRequired(): Boolean = UserGroupInformation.isSecurityEnabled
+
   protected def setupTokenManager(): Unit = {
-    if (UserGroupInformation.isSecurityEnabled) {
+    if (tokenManagerRequired()) {
       delegationTokenManager = createTokenManager()
       delegationTokenManager.foreach { dtm =>
         val ugi = UserGroupInformation.getCurrentUser

core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala

@@ -618,6 +618,15 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
 
   override def start(): Unit = {
     setupTokenManager()
+    if (conf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED) && delegationTokenManager.isEmpty) {
+      logWarning("spark.security.credentials.directProviders.enabled is set but " +
+        "this cluster manager does not support credential distribution. " +
+        "No tokens will be collected or renewed.")
+    }
+  }
+
+  override protected def tokenManagerRequired(): Boolean = {
+    super.tokenManagerRequired() || conf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED)
   }
 
   protected def createDriverEndpoint(): DriverEndpoint = new DriverEndpoint()

core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala

@@ -129,6 +129,10 @@ private[spark] class LocalSchedulerBackend(
     Some(new HadoopDelegationTokenManager(conf, scheduler.sc.hadoopConfiguration, localEndpoint))
   }
 
+  override protected def tokenManagerRequired(): Boolean = {
+    super.tokenManagerRequired() || conf.get(config.CREDENTIALS_DIRECT_PROVIDERS_ENABLED)
+  }
+
   override protected def updateDelegationTokens(tokens: Array[Byte]): Unit = {
     SparkHadoopUtil.get.addDelegationTokens(tokens, conf)
   }

core/src/test/resources/META-INF/services/org.apache.spark.security.HadoopDelegationTokenProvider

@@ -16,3 +16,6 @@
 #
 
 org.apache.spark.deploy.security.ExceptionThrowingDelegationTokenProvider
+org.apache.spark.deploy.security.TestNonKerberosTokenProvider
+org.apache.spark.deploy.security.TestDisabledProvider
+org.apache.spark.deploy.security.TestFailingProvider