Skip to content

[Spark-21842][Mesos] Support Kerberos ticket renewal and creation in Mesos #19272

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
ArtRand wants to merge 23 commits into from
Closed

[Spark-21842][Mesos] Support Kerberos ticket renewal and creation in Mesos #19272

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
23 commits
Select commit Hold shift + click to select a range
44a6098
basic implementation of MesosCredentialRenewer
Sep 19, 2017
781c5a7
pull upstream
Sep 19, 2017
18d2c6c
working through comments
Sep 20, 2017
43ab547
cleanup
Sep 20, 2017
f136eaa
remove unused imports
Sep 20, 2017
488f72a
remove debugging code
Sep 20, 2017
f5925fd
use 75 percent of renewal time for the first renewal also
Sep 26, 2017
3f22efe
addressed comments, still pending testing
Oct 16, 2017
e522150
final comments
Oct 18, 2017
837157d
add comments
Oct 20, 2017
c95f80b
remove extra thread in MesosCredentialRenewer
Oct 21, 2017
e8bbc9e
refactor credential renewer, addressed comments
Oct 30, 2017
1596d80
merge upstream
Oct 30, 2017
864ab7e
resolve import problem
Oct 30, 2017
7e0590a
cleanup
Oct 31, 2017
f93c551
more cleanup
Oct 31, 2017
b2fbcf2
move hadoop delegation token management into Mesos renewer
Nov 6, 2017
4558cea
remove unused imports
Nov 6, 2017
5f254e5
no more conditionals, start renewer automatically
Nov 8, 2017
8df7e37
addressed comments
Nov 10, 2017
45b46ed
quick fixup, logging was in the wrong place
Nov 10, 2017
18d77ff
only renew tokens when using keytab
Nov 14, 2017
049e4b5
update tokens, make SparkHadoopUtil methods private
Nov 15, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions core/src/main/scala/org/apache/spark/deploy/SparkHadoopUtil.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -142,6 +142,13 @@ class SparkHadoopUtil extends Logging {
}
}

def addDelegationTokens(tokens: Array[Byte], sparkConf: SparkConf) {
Copy link
Contributor

@vanzin vanzin Oct 10, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Add a comment about what this method is doing and why it's needed. (YARN never sets the authentication method, so it'd be good to know why Mesos needs to do it.)

Copy link
Contributor

@vanzin vanzin Nov 15, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Always forget this class is public. Add private[spark].

val hadoopConf = newConfiguration(sparkConf)
hadoopConf.set("hadoop.security.authentication", "Token")
UserGroupInformation.setConfiguration(hadoopConf)
addCurrentUserCredentials(deserialize(tokens))
}

/**
* Returns a function that can be called to find Hadoop FileSystem bytes read. If
* getFSBytesReadOnThreadCallback is called from thread r at time t, the returned callback will
Expand Down
17 changes: 4 additions & 13 deletions core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,8 +26,6 @@ import scala.collection.mutable
import scala.util.{Failure, Success}
import scala.util.control.NonFatal

import org.apache.hadoop.security.UserGroupInformation

import org.apache.spark._
import org.apache.spark.TaskState.TaskState
import org.apache.spark.deploy.SparkHadoopUtil
Expand Down Expand Up @@ -125,8 +123,11 @@ private[spark] class CoarseGrainedExecutorBackend(
executor.stop()
}
}.start()

// This message is only sent by Mesos Drivers, and is not expected from other
Copy link
Contributor

@vanzin vanzin Oct 10, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No need to add this comment.

// SchedulerBackends at this time
case UpdateDelegationTokens(tokens) =>
Copy link
Contributor

@skonto skonto Sep 19, 2017 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Let's add a comment that this is received only in mesos case, since CoarseGrainedExecutorBackend is used by both yarn and standalone.

CoarseGrainedExecutorBackend.addDelegationTokens(tokens, env.conf)
SparkHadoopUtil.get.addDelegationTokens(tokens, env.conf)
}

override def onDisconnected(remoteAddress: RpcAddress): Unit = {
Expand Down Expand Up @@ -178,16 +179,6 @@ private[spark] class CoarseGrainedExecutorBackend(

private[spark] object CoarseGrainedExecutorBackend extends Logging {

private def addDelegationTokens(tokens: Array[Byte], sparkConf: SparkConf) {
logInfo(s"Found delegation tokens of ${tokens.length} bytes")
val hadoopConf = SparkHadoopUtil.get.newConfiguration(sparkConf)
hadoopConf.set("hadoop.security.authentication", "Token")
UserGroupInformation.setConfiguration(hadoopConf)
val creds = SparkHadoopUtil.get.deserialize(tokens)
// decode tokens and add them to the credentials
UserGroupInformation.getCurrentUser.addCredentials(SparkHadoopUtil.get.deserialize(tokens))
}

private def run(
driverUrl: String,
executorId: String,
Expand Down
6 changes: 2 additions & 4 deletions core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -161,11 +161,9 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
}

case UpdateDelegationTokens(tokens) =>
logDebug("Asking each executor to update HDFS delegation tokens")
for ((x, executorData) <- executorDataMap) {
executorData.executorEndpoint.send(UpdateDelegationTokens(tokens))
executorDataMap.values.foreach {
ed => ed.executorEndpoint.send(UpdateDelegationTokens(tokens))
Copy link
Contributor

@vanzin vanzin Oct 10, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ed => goes in previous line. (The whole thing might fit in one line, too.)

}

}

override def receiveAndReply(context: RpcCallContext): PartialFunction[Any, Unit] = {
Expand Down
2 changes: 1 addition & 1 deletion ...s/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@

package org.apache.spark.scheduler.cluster.mesos

import org.apache.spark.{SparkContext, SparkException}
import org.apache.spark.SparkContext
Copy link
Contributor

@vanzin vanzin Oct 24, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Change not needed?

Copy link
Author

@ArtRand ArtRand Oct 30, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

SparkException is unused, not sure why it was there in the first place

import org.apache.spark.internal.config._
import org.apache.spark.scheduler.{ExternalClusterManager, SchedulerBackend, TaskScheduler, TaskSchedulerImpl}

Expand Down
7 changes: 5 additions & 2 deletions ...n/scala/org/apache/spark/scheduler/cluster/mesos/MesosCoarseGrainedSchedulerBackend.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -198,16 +198,19 @@ private[spark] class MesosCoarseGrainedSchedulerBackend(
sc.conf.getOption("spark.mesos.driver.frameworkId").map(_ + suffix)
)

if (principal != null) {
// check that the credentials are defined, even though it's likely that auth would have failed
// already if you've made it this far
if (principal != null && hadoopDelegationCreds.isDefined) {
logDebug(s"Principal found ($principal) starting token renewer")
val credentialRenewerThread = new Thread {
setName("MesosCredentialRenewer")
override def run(): Unit = {
val dummy: Option[Array[Byte]] = None
Copy link
Contributor

@susanxhuynh susanxhuynh Sep 20, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What is this for?

Copy link
Author

@ArtRand ArtRand Sep 20, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

whoops!

val credentialRenewer =
new MesosCredentialRenewer(
conf,
hadoopDelegationTokenManager.get,
MesosCredentialRenewer.getTokenRenewalInterval(hadoopDelegationCreds.get, conf),
MesosCredentialRenewer.getTokenRenewalTime(hadoopDelegationCreds.get, conf),
Copy link

@kalvinnchau kalvinnchau Sep 25, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This sets the first renewal time to be the expiration time of the token.

It should be similar to the way next renewal time in the MesosCredentialRenewer class is calculated so that it renews the first token after 75% of expiration time has passed:

val currTime = System.currentTimeMillis()
val renewTime = MesosCredentialRenewer.getTokenRenewalTime(hadoopDelegationCreds.get, conf)
val rt = 0.75 * (renewTime - currTime)

val credentialRenewer =
   new MesosCredentialRenewer(
     conf,
     hadoopDelegationTokenManager.get,
     (currTime + rt).toLong,
     driverEndpoint)
credentialRenewer.scheduleTokenRenewal()

driverEndpoint)
credentialRenewer.scheduleTokenRenewal()
}
Expand Down
13 changes: 7 additions & 6 deletions ...esos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosCredentialRenewer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,7 +63,8 @@ class MesosCredentialRenewer(

def scheduleTokenRenewal(): Unit = {
def scheduleRenewal(runnable: Runnable): Unit = {
val remainingTime = timeOfNextRenewal - System.currentTimeMillis()
// val remainingTime = timeOfNextRenewal - System.currentTimeMillis()
val remainingTime = 5000
Copy link
Contributor

@susanxhuynh susanxhuynh Sep 20, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why 5000?

Copy link
Author

@ArtRand ArtRand Sep 20, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

well that's embarrassing, just a debugging tool that I forgot to remove.

if (remainingTime <= 0) {
logInfo("Credentials have expired, creating new ones now.")
runnable.run()
Expand All @@ -82,8 +83,8 @@ class MesosCredentialRenewer(
} catch {
case e: Exception =>
// Log the error and try to write new tokens back in an hour
Copy link
Contributor

@susanxhuynh susanxhuynh Sep 19, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Comment says "an hour" but code has 20 seconds.

Copy link
Author

@ArtRand ArtRand Sep 20, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

good catch, I changed the code to match the YARN equivalent.

logWarning("Couldn't broadcast tokens, trying agin in 20 seconds", e)
credentialRenewerThread.schedule(this, 20, TimeUnit.SECONDS)
logWarning("Couldn't broadcast tokens, trying again in an hour", e)
credentialRenewerThread.schedule(this, 1, TimeUnit.HOURS)
return
}
scheduleRenewal(this)
Expand Down Expand Up @@ -136,15 +137,15 @@ class MesosCredentialRenewer(
}

object MesosCredentialRenewer extends Logging {
def getTokenRenewalInterval(bytes: Array[Byte], conf: SparkConf): Long = {
def getTokenRenewalTime(bytes: Array[Byte], conf: SparkConf): Long = {
val hadoopConf = SparkHadoopUtil.get.newConfiguration(conf)
val creds = SparkHadoopUtil.get.deserialize(bytes)
val intervals = creds.getAllTokens.asScala.flatMap { t =>
val renewalTimes = creds.getAllTokens.asScala.flatMap { t =>
Try {
Copy link
Contributor

@skonto skonto Sep 19, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

t -> token
This method does not return an interval, it just returns the new expiration time.
Compare with:

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

Line 102 in b9ab791

private def getTokenRenewalInterval(

t.renew(hadoopConf)
}.toOption
}
if (intervals.isEmpty) Long.MaxValue else intervals.min
if (renewalTimes.isEmpty) Long.MaxValue else renewalTimes.min
}
}