parthchandra commented on code in PR #57285:
URL: https://github.com/apache/spark/pull/57285#discussion_r3647293650
##########
core/src/main/scala/org/apache/spark/deploy/security/HadoopDelegationTokenManager.scala:
##########
@@ -199,16 +241,14 @@ 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.")
schedulerRef.send(UpdateDelegationTokens(tokens))
tokens
} catch {
case _: InterruptedException =>
- // Ignore, may happen if shutting down.
Review Comment:
added back
##########
core/src/main/scala/org/apache/spark/deploy/security/HadoopDelegationTokenManager.scala:
##########
@@ -226,24 +266,40 @@ 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 {
+ (new Credentials(), Long.MaxValue, 0)
Review Comment:
This is indeed valid. Throwing `IllegalStateException` is correct.
##########
core/src/main/scala/org/apache/spark/scheduler/SupportsDelegationToken.scala:
##########
@@ -21,19 +21,20 @@ import org.apache.hadoop.security.UserGroupInformation
import org.apache.spark.deploy.SparkHadoopUtil
import org.apache.spark.deploy.security.HadoopDelegationTokenManager
+import org.apache.spark.internal.Logging
/**
* A mix-in trait for SchedulerBackend that supports delegation tokens.
*/
-private[spark] trait SupportsDelegationToken {
+private[spark] trait SupportsDelegationToken extends Logging {
Review Comment:
Done
##########
core/src/main/scala/org/apache/spark/deploy/security/HadoopDelegationTokenManager.scala:
##########
@@ -76,28 +77,50 @@ 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.get(NETWORK_CRYPTO_ENABLED) ||
+ sparkConf.get(SASL_ENCRYPTION_ENABLED) ||
+ sparkConf.getBoolean("spark.ssl.rpc.enabled", false)
+ require(hasEncryption,
+ "RPC channel encryption (spark.network.crypto.enabled, " +
+ "spark.authenticate.enableSaslEncryption, or spark.ssl.rpc.enabled) must
be enabled when " +
+ "spark.security.credentials.directProviders.enabled is true. " +
+ "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)
Review Comment:
Fixed
##########
core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala:
##########
@@ -620,6 +620,10 @@ class CoarseGrainedSchedulerBackend(scheduler:
TaskSchedulerImpl, val rpcEnv: Rp
setupTokenManager()
}
+ override protected def tokenManagerRequired(): Boolean = {
+ super.tokenManagerRequired() ||
conf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED)
Review Comment:
Logging a warning now.
##########
core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala:
##########
@@ -27,6 +27,7 @@ import org.apache.spark.deploy.SparkHadoopUtil
import org.apache.spark.deploy.security.HadoopDelegationTokenManager
import org.apache.spark.executor.{Executor, ExecutorBackend}
import org.apache.spark.internal.{config, Logging, LogKeys}
+import org.apache.spark.internal.config._
Review Comment:
Removed
##########
core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala:
##########
@@ -129,6 +130,10 @@ private[spark] class LocalSchedulerBackend(
Some(new HadoopDelegationTokenManager(conf,
scheduler.sc.hadoopConfiguration, localEndpoint))
}
+ override protected def tokenManagerRequired(): Boolean = {
+ super.tokenManagerRequired() ||
conf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED)
Review Comment:
Done
##########
core/src/main/scala/org/apache/spark/deploy/security/HadoopDelegationTokenManager.scala:
##########
@@ -76,28 +77,50 @@ 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.get(NETWORK_CRYPTO_ENABLED) ||
+ sparkConf.get(SASL_ENCRYPTION_ENABLED) ||
+ sparkConf.getBoolean("spark.ssl.rpc.enabled", false)
+ require(hasEncryption,
+ "RPC channel encryption (spark.network.crypto.enabled, " +
+ "spark.authenticate.enableSaslEncryption, or spark.ssl.rpc.enabled) must
be enabled when " +
+ "spark.security.credentials.directProviders.enabled is true. " +
+ "Credential tokens must not be transmitted over unencrypted channels.")
Review Comment:
Good point. Changed
--
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]
