[SPARK-11714][Mesos] Make Spark on Mesos honor port restrictions on coarse grain mode #11157
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -47,6 +47,7 @@ private[mesos] trait MesosSchedulerUtils extends Logging { | |
|
|
||
|
||
| /** | ||
| * Creates a new MesosSchedulerDriver that communicates to the Mesos master. | ||
| * | ||
|
||
| * @param masterUrl The url to connect to Mesos master | ||
| * @param scheduler the scheduler class to receive scheduler callbacks | ||
| * @param sparkUser User to impersonate with when running tasks | ||
|
|
@@ -147,6 +148,20 @@ private[mesos] trait MesosSchedulerUtils extends Logging { | |
| res.asScala.filter(_.getName == name).map(_.getScalar.getValue).sum | ||
| } | ||
|
|
||
| /** | ||
| * Transforms a range resource to a list of ranges | ||
| * | ||
| * @param res the mesos resource list | ||
| * @param name the name of the resource | ||
| * @return the list of ranges returned | ||
| */ | ||
| protected def getRangeResource(res: JList[Resource], name: String): List[(Long, Long)] = { | ||
| // A resource can have multiple values in the offer since it can either be from | ||
| // a specific role or wildcard. | ||
| res.asScala.filter(_.getName == name).flatMap(_.getRanges.getRangeList.asScala | ||
| .map(r => (r.getBegin, r.getEnd)).toList).toList | ||
| } | ||
|
|
||
| /** | ||
| * Signal that the scheduler has registered with Mesos. | ||
| */ | ||
|
|
@@ -172,6 +187,7 @@ private[mesos] trait MesosSchedulerUtils extends Logging { | |
| /** | ||
| * Partition the existing set of resources into two groups, those remaining to be | ||
| * scheduled and those requested to be used for a new task. | ||
| * | ||
| * @param resources The full list of available resources | ||
| * @param resourceName The name of the resource to take from the available resources | ||
| * @param amountToUse The amount of resources to take from the available resources | ||
|
|
@@ -223,7 +239,8 @@ private[mesos] trait MesosSchedulerUtils extends Logging { | |
| /** | ||
| * Converts the attributes from the resource offer into a Map of name -> Attribute Value | ||
| * The attribute values are the mesos attribute types and they are | ||
| * | ||
| * @param offerAttributes the attributes offered | ||
| * @return | ||
| */ | ||
| protected def toAttributeMap(offerAttributes: JList[Attribute]): Map[String, GeneratedMessage] = { | ||
|
|
@@ -333,6 +350,7 @@ private[mesos] trait MesosSchedulerUtils extends Logging { | |
| /** | ||
| * Return the amount of memory to allocate to each executor, taking into account | ||
| * container overheads. | ||
| * | ||
| * @param sc SparkContext to use to get `spark.mesos.executor.memoryOverhead` value | ||
| * @return memory requirement as (0.1 * <memoryOverhead>) or MEMORY_OVERHEAD_MINIMUM | ||
| * (whichever is larger) | ||
|
|
@@ -357,6 +375,111 @@ private[mesos] trait MesosSchedulerUtils extends Logging { | |
| sc.conf.getTimeAsSeconds("spark.mesos.rejectOfferDurationForReachedMaxCores", "120s") | ||
| } | ||
|
|
||
| /** | ||
| * Checks executor ports if they are within some range of the offered list of ports ranges, | ||
| * | ||
| * @param conf the Spark Config | ||
| * @param ports the list of ports to check | ||
| * @return true if ports are within range false otherwise | ||
| */ | ||
| protected def checkPorts(conf: SparkConf, ports: List[(Long, Long)]): Boolean = { | ||
|
|
||
| def checkIfInRange(port: Long, ps: List[(Long, Long)]): Boolean = { | ||
| ps.exists{case (rangeStart, rangeEnd) => rangeStart <= port & rangeEnd >= port } | ||
| } | ||
|
|
||
| val portsToCheck = nonZeroPortValuesFromConfig(conf) | ||
| val withinRange = portsToCheck.forall(p => checkIfInRange(p, ports)) | ||
| // make sure we have enough ports to allocate per offer | ||
| val enoughPorts = | ||
| ports.map{case (rangeStart, rangeEnd) => rangeEnd - rangeStart + 1}.sum >= portsToCheck.size | ||
| enoughPorts && withinRange | ||
| } | ||
|
|
||
| /** | ||
| * Partitions port resources. | ||
| * | ||
| * @param requestedPorts non-zero ports to assign | ||
| * @param offeredResources the resources offered | ||
| * @return resources left, port resources to be used. | ||
| */ | ||
| def partitionPortResources(requestedPorts: List[Long], offeredResources: List[Resource]) | ||
| : (List[Resource], List[Resource]) = { | ||
| if (requestedPorts.isEmpty) { | ||
|
||
| (offeredResources, List[Resource]()) | ||
| } else { | ||
| // partition port offers | ||
| val (resourcesWithoutPorts, portResources) = filterPortResources(offeredResources) | ||
|
|
||
| val portsAndRoles = requestedPorts. | ||
| map(x => (x, findPortAndGetAssignedRangeRole(x, portResources))) | ||
|
|
||
| val assignedPortResources = createResourcesFromPorts(portsAndRoles) | ||
|
|
||
| // ignore non-assigned port resources, they will be declined implicitly by mesos | ||
| // no need for splitting port resources. | ||
| (resourcesWithoutPorts, assignedPortResources) | ||
| } | ||
| } | ||
|
|
||
| val managedPortNames = List("spark.executor.port", "spark.blockManager.port") | ||
|
|
||
| /** | ||
| * The values of the non-zero ports to be used by the executor process. | ||
| * @param conf the spark config to use | ||
| * @return the ono-zero values of the ports | ||
| */ | ||
| def nonZeroPortValuesFromConfig(conf: SparkConf): List[Long] = { | ||
| managedPortNames.map(conf.getLong(_, 0)).filter( _ != 0) | ||
| } | ||
|
|
||
| /** Creates a mesos resource for a specific port number. */ | ||
| private def createResourcesFromPorts(portsAndRoles: List[(Long, String)]) : List[Resource] = { | ||
| portsAndRoles.flatMap{ case (port, role) => | ||
| createMesosPortResource(List((port, port)), Some(role))} | ||
| } | ||
|
|
||
| /** Helper to create mesos resources for specific port ranges. */ | ||
| private def createMesosPortResource( | ||
| ranges: List[(Long, Long)], | ||
| role: Option[String] = None): List[Resource] = { | ||
| ranges.map { case (rangeStart, rangeEnd) => | ||
| val rangeValue = Value.Range.newBuilder() | ||
| .setBegin(rangeStart) | ||
| .setEnd(rangeEnd) | ||
| val builder = Resource.newBuilder() | ||
| .setName("ports") | ||
| .setType(Value.Type.RANGES) | ||
| .setRanges(Value.Ranges.newBuilder().addRange(rangeValue)) | ||
| role.foreach(r => builder.setRole(r)) | ||
| builder.build() | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Helper to assign a port to an offered range and get the latter's role | ||
| * info to use it later on. | ||
| */ | ||
| private def findPortAndGetAssignedRangeRole(port: Long, portResources: List[Resource]) | ||
| : String = { | ||
|
|
||
| val ranges = portResources. | ||
| map(resource => | ||
| (resource.getRole, resource.getRanges.getRangeList.asScala | ||
| .map(r => (r.getBegin, r.getEnd)).toList)) | ||
|
|
||
| val rangePortRole = ranges | ||
| .find { case (role, rangeList) => rangeList | ||
| .exists{ case (rangeStart, rangeEnd) => rangeStart <= port & rangeEnd >= port}} | ||
| // this is safe since we have previously checked about the ranges (see checkPorts method) | ||
| rangePortRole.map{ case (role, rangeList) => role}.get | ||
| } | ||
|
|
||
| /** Retrieves the port resources from a list of mesos offered resources */ | ||
| private def filterPortResources(resources: List[Resource]): (List[Resource], List[Resource]) = { | ||
| resources.partition { r => !(r.getType == Value.Type.RANGES && r.getName == "ports") } | ||
| } | ||
|
|
||
| /** | ||
| * spark.mesos.driver.frameworkId is set by the cluster dispatcher to correlate driver | ||
| * submissions with frameworkIDs. However, this causes issues when a driver process launches | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
newline here
Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I do not see exactly where i should place a new line. But i will separate vals anyway by adding comments on top of them. Vals should be grouped together according to the guidelines.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I just thought a newline separating the port handling would improve readability. I generally prefer to separate each logical block with newlines. Not a blocker.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
ok np