HomeMesosphereNo notifications. 4 unresolved issues.
  • Search
Differential D1003

Consider Reserved instances for validating constraints.
ClosedAll Users
Actions

Authored by meichstedt on Aug 18 2017, 7:10 PM.

Details

Reviewers
jeschkies
kensipe
timcharper
jenkins
Commits
rMARATHONe835e1c3f182: Consider Reserved instances for validating constraints.
JIRA Issues
JIRA COPS-598 Hostname:unique constraint violated with persistent volumes
JIRA MARATHON-1703 Multiple instances of a Task with persistent volume do not dispatch to different nodes when unique hostname constraint is set
Summary

When matching offers for resident tasks, Marathon would not consider instances in state Reserved when validating constraints. This potentially resulted in multiple reservations on the same agent, even if a hostname: unique constraint was applied to the run spec. The problem boils down to the fact that the ResourceMatcher filtered for instance.isLaunched and only considered those instances. This was accidentally introduced via the fact that previously, instances which weren't launched had no runSpecVersion and therefore could not be compared to the last config change.

Compare 3bf03f86, which needed to access _.launched to compare the version. Deriving the need to check for isLaunched later was a wrong assumption.

While at it, StrictLogging has been introduced and unnecessary if(xyzEnabled) statements have been removed.

Added two unit tests for offer matching with unique constraints

Test Plan

sbt test

Diff Detail

Repository
rMARATHON marathon
Lint
Automatic diff as part of commit; lint not applicable.
Unit
Automatic diff as part of commit; unit tests not applicable.
meichstedt created this revision.Aug 18 2017, 7:10 PM
jenkins requested changes to this revision.Aug 18 2017, 7:14 PM
This revision now requires changes to proceed.Aug 18 2017, 7:14 PM
meichstedt edited the summary of this revision. (Show Details)Aug 18 2017, 7:17 PM
kensipe accepted this revision.Aug 18 2017, 7:21 PM

wow... if we take away the logger updates and the variable name change... It looks like this is a 1 liner. nice work!
Waiting for tests... code lgtm

meichstedt updated this revision to Diff 4073.Aug 18 2017, 7:24 PM

clarifying test names

jenkins requested changes to this revision.Aug 18 2017, 7:25 PM
This revision now requires changes to proceed.Aug 18 2017, 7:25 PM
jenkins added a comment.Aug 18 2017, 7:25 PM
✗ Build of 4072 failed jenkins-public-marathon-phabricator-813.

Error message:

Stage Compile and Test failed.

(๑′°︿°๑)

meichstedt updated this revision to Diff 4074.Aug 18 2017, 7:26 PM

clarifying test names

jenkins requested changes to this revision.Aug 18 2017, 7:30 PM
This revision now requires changes to proceed.Aug 18 2017, 7:30 PM
jenkins added a comment.Aug 18 2017, 7:43 PM
✗ Build of 4074 failed jenkins-public-marathon-phabricator-815.

Error message:

Stage Compile and Test failed.

(๑′°︿°๑)

jenkins added a comment.Aug 18 2017, 7:46 PM
✗ Build of 4073 failed jenkins-public-marathon-phabricator-814.

Error message:

Stage Compile and Test failed.

(๑′°︿°๑)

jenkins added a comment.Aug 18 2017, 9:10 PM
✗ Build of 4074 failed jenkins-public-marathon-phabricator-817.

Error message:

Stage Compile and Test failed.

(๑′°︿°๑)

meichstedt added a comment.Aug 18 2017, 9:36 PM

mesosphere.marathon.integration.ResidentTaskIntegrationTest.ResidentTaskIntegrationTest should resident task can be deployed along with constraints

might actually be a regression, need to investigate.

jenkins added a comment.Aug 18 2017, 9:47 PM
✗ Build of 4074 failed jenkins-public-marathon-phabricator-818.

Error message:

Stage Compile and Test failed.

(๑′°︿°๑)

meichstedt updated this revision to Diff 4076.Aug 18 2017, 10:36 PM

fixed an embarrassing bug that has been there for ages.

jenkins requested changes to this revision.Aug 18 2017, 10:44 PM
This revision now requires changes to proceed.Aug 18 2017, 10:44 PM
jenkins added a comment.Aug 18 2017, 10:48 PM
✗ Build of 4074 failed jenkins-public-marathon-phabricator-819.

Error message:

Stage Compile and Test failed.

(๑′°︿°๑)

jenkins added a comment.Aug 18 2017, 10:59 PM
✗ Build of 4076 failed jenkins-public-marathon-phabricator-820.

Error message:

Stage Compile and Test failed.

(๑′°︿°๑)

jenkins accepted this revision.Aug 19 2017, 1:27 AM
This revision is now accepted and ready to land.Aug 19 2017, 1:27 AM
This revision is now accepted and ready to land.Aug 19 2017, 1:27 AM
jenkins added a comment.Aug 19 2017, 1:27 AM
✔ Build of 4076 completed jenkins-public-marathon-phabricator-821.

You can create a DC/OS with your patched Marathon by creating a new pull
request with the following changes in buildinfo.json:

＼\ ٩( ᐛ )و /／

timcharper accepted this revision.Aug 21 2017, 10:32 AM
timcharper added inline comments.Aug 21 2017, 10:34 AM
src/main/scala/mesosphere/mesos/ResourceMatcher.scala
179

wondering why did we have the .isLaunched filter here before?

Closed by commit rMARATHONe835e1c3f182: Consider Reserved instances for validating constraints. (authored by meichstedt, committed by timcharper). · Explain WhyAug 21 2017, 12:48 PM
This revision was automatically updated to reflect the committed changes.
timcharper added a comment.Aug 21 2017, 12:48 PM

I've confirmed locally that this fixes the bug

Revision Contents

Diff 4078

View Options

src/main/scala/mesosphere/marathon/core/launcher/impl/InstanceOpFactoryImpl.scala

11package mesosphere.marathon
22package core.launcher.impl
33
44import java.time.Clock
55
6import com.typesafe.scalalogging.StrictLogging
67import mesosphere.marathon.core.condition.Condition
78import mesosphere.marathon.core.instance.Instance.{ AgentInfo, InstanceState }
89import mesosphere.marathon.core.instance.update.InstanceUpdateOperation
910import mesosphere.marathon.core.instance.{ Instance, LegacyAppInstance }
1011import mesosphere.marathon.core.launcher.{ InstanceOp, InstanceOpFactory, OfferMatchResult }
1112import mesosphere.marathon.core.plugin.PluginManager
1213import mesosphere.marathon.core.pod.PodDefinition
1314import mesosphere.marathon.core.task.Task
1415import mesosphere.marathon.core.task.state.NetworkInfo
1516import mesosphere.marathon.plugin.task.RunSpecTaskProcessor
1617import mesosphere.marathon.plugin.{ ApplicationSpec, PodSpec }
1718import mesosphere.marathon.state._
1819import mesosphere.marathon.stream.Implicits._
1920import mesosphere.mesos.ResourceMatcher.ResourceSelector
2021import mesosphere.mesos.{ NoOfferMatchReason, PersistentVolumeMatcher, ResourceMatchResponse, ResourceMatcher, RunSpecOfferMatcher, TaskBuilder, TaskGroupBuilder }
2122import mesosphere.util.state.FrameworkId
2223import org.apache.mesos.Protos.{ ExecutorInfo, TaskGroupInfo, TaskInfo }
2324import org.apache.mesos.{ Protos => Mesos }
24import org.slf4j.LoggerFactory
2525
2626import scala.concurrent.duration._
2727
2828class InstanceOpFactoryImpl(
2929 config: MarathonConf,
3030 pluginManager: PluginManager = PluginManager.None)(implicit clock: Clock)
31 extends InstanceOpFactory {
31 extends InstanceOpFactory with StrictLogging {
3232
3333 import InstanceOpFactoryImpl._
3434
35 private[this] val log = LoggerFactory.getLogger(getClass)
3635 private[this] val taskOperationFactory = {
3736 val principalOpt = config.mesosAuthenticationPrincipal.get
3837 val roleOpt = config.mesosRole.get
3938
4039 new InstanceOpFactoryHelper(principalOpt, roleOpt)
4140 }
4241
4342 private[this] lazy val runSpecTaskProc: RunSpecTaskProcessor = combine(
4443 pluginManager.plugins[RunSpecTaskProcessor].toIndexedSeq)
4544
4645 override def matchOfferRequest(request: InstanceOpFactory.Request): OfferMatchResult = {
47 log.debug("matchOfferRequest")
46 logger.debug("matchOfferRequest")
4847
4948 request.runSpec match {
5049 case app: AppDefinition =>
5150 if (request.isForResidentRunSpec) {
5251 inferForResidents(app, request)
5352 } else {
5453 inferNormalTaskOp(app, request)
5554 }
▲ Show 20 Lines • Show All 80 Lines • ▼ Show 20 Line(s)
136135 * - schedule a ReserveAndCreate TaskOp
137136 */
138137
139138 def maybeLaunchOnReservation: Option[OfferMatchResult] = if (needToLaunch) {
140139 val maybeVolumeMatch = PersistentVolumeMatcher.matchVolumes(offer, request.reserved)
141140
142141 maybeVolumeMatch.map { volumeMatch =>
143142
144 // we must not consider the volumeMatch's Reserved instance because that would lead to a violation of constraints
145 // by the Reserved instance that we actually want to launch
143 // The volumeMatch identified a specific instance that matches the volume's reservation labels.
144 // This is the instance we want to launch. However, when validating constraints, we need to exclude that one
145 // instance: it would be considered as an instance on that agent, and would violate e.g. a hostname:unique
146 // constraint although it is just a placeholder for the instance that will be launched.
146147 val instancesToConsiderForConstraints: Stream[Instance] =
147 instances.valuesIterator.toStream.filterNotAs(_.instanceId != volumeMatch.instance.instanceId)
148 instances.valuesIterator.toStream.filterAs(_.instanceId != volumeMatch.instance.instanceId)
148149
149150 // resources are reserved for this role, so we only consider those resources
150151 val rolesToConsider = config.mesosRole.get.toSet
151152 val reservationLabels = TaskLabels.labelsForTask(request.frameworkId, volumeMatch.instance.appTask.taskId).labels
152153 val resourceMatchResponse =
153154 ResourceMatcher.matchResources(
154155 offer, runSpec, instancesToConsiderForConstraints,
155156 ResourceSelector.reservedWithLabels(rolesToConsider, reservationLabels)
Show All 13 Lines
169170 val configuredRoles = if (runSpec.acceptedResourceRoles.isEmpty) {
170171 config.defaultAcceptedResourceRolesSet
171172 } else {
172173 runSpec.acceptedResourceRoles
173174 }
174175 // We can only reserve unreserved resources
175176 val rolesToConsider = Set(ResourceRole.Unreserved).intersect(configuredRoles)
176177 if (rolesToConsider.isEmpty) {
177 log.warn(s"Will never match for ${runSpec.id}. The runSpec is not configured to accept unreserved resources.")
178 logger.warn(s"Will never match for ${runSpec.id}. The runSpec is not configured to accept unreserved resources.")
178179 }
179180
180181 val resourceMatchResponse =
181182 ResourceMatcher.matchResources(offer, runSpec, instances.valuesIterator.toStream, ResourceSelector.reservable)
182183 resourceMatchResponse match {
183184 case matches: ResourceMatchResponse.Match =>
184185 val instanceOp = reserveAndCreateVolumes(request.frameworkId, runSpec, offer, matches.resourceMatch)
185186 Some(OfferMatchResult.Match(app, request.offer, instanceOp, clock.now()))
186187 case matchesNot: ResourceMatchResponse.NoMatch =>
187188 Some(OfferMatchResult.NoMatch(app, request.offer, matchesNot.reasons, clock.now()))
188189 }
189190 } else None
190191
191192 maybeLaunchOnReservation
192193 .orElse(maybeReserveAndCreateVolumes)
193194 .getOrElse {
194 log.warn("No need to reserve or launch and offer request isForResidentRunSpec")
195 logger.warn("No need to reserve or launch and offer request isForResidentRunSpec")
195196 OfferMatchResult.NoMatch(app, request.offer,
196197 Seq(NoOfferMatchReason.NoCorrespondingReservationFound), clock.now())
197198 }
198199 }
199200
200201 private[this] def launchOnReservation(
201202 spec: AppDefinition,
202203 offer: Mesos.Offer,
▲ Show 20 Lines • Show All 151 Lines • Show Last 20 Lines
View Options

src/main/scala/mesosphere/mesos/ResourceMatcher.scala

11package mesosphere.mesos
22
3import com.typesafe.scalalogging.StrictLogging
34import mesosphere.marathon.core.instance.Instance
45import mesosphere.marathon.core.launcher.impl.TaskLabels
5import mesosphere.marathon.state.{ PersistentVolume, ResourceRole, RunSpec, DiskType, DiskSource }
6import mesosphere.marathon.state.{ DiskSource, DiskType, PersistentVolume, ResourceRole, RunSpec }
67import mesosphere.marathon.stream.Implicits._
78import mesosphere.marathon.tasks.{ PortsMatch, PortsMatcher, ResourceUtil }
89import mesosphere.mesos.protos.Resource
910import org.apache.mesos.Protos
1011import org.apache.mesos.Protos.Offer
1112import org.apache.mesos.Protos.Resource.DiskInfo.Source
12import org.slf4j.LoggerFactory
1313
1414import scala.annotation.tailrec
1515import scala.collection.immutable.Seq
1616
17object ResourceMatcher {
17object ResourceMatcher extends StrictLogging {
1818 import ResourceUtil.RichResource
1919 type Role = String
2020
21 private[this] val log = LoggerFactory.getLogger(getClass)
22
2321 /**
2422 * A successful match result of the [[ResourceMatcher]].matchResources method.
2523 */
2624 case class ResourceMatch(scalarMatches: Seq[ScalarMatch], portsMatch: PortsMatch) {
2725 lazy val hostPorts: Seq[Option[Int]] = portsMatch.hostPorts
2826
2927 def scalarMatch(name: String): Option[ScalarMatch] = scalarMatches.find(_.resourceName == name)
3028
▲ Show 20 Lines • Show All 94 Lines • ▼ Show 20 Line(s)
125123 * If a task uses local volumes, this method is typically called twice for every launch. Once
126124 * for the reservation on UNRESERVED resources and once for every (re-)launch on RESERVED resources.
127125 *
128126 * If matching on RESERVED resources as specified by the ResourceSelector, resources for volumes
129127 * have to be matched separately (e.g. by the [[PersistentVolumeMatcher]]). If matching on UNRESERVED
130128 * resources, the disk resources for the local volumes are included since they must become part of
131129 * the reservation.
132130 */
133 def matchResources(offer: Offer, runSpec: RunSpec, runningInstances: => Seq[Instance],
131 def matchResources(offer: Offer, runSpec: RunSpec, knownInstances: => Seq[Instance],
134132 selector: ResourceSelector): ResourceMatchResponse = {
135133
136134 val groupedResources: Map[Role, Seq[Protos.Resource]] = offer.getResourcesList.groupBy(_.getName).map { case (k, v) => k -> v.to[Seq] }
137135
138136 val scalarResourceMatch = matchScalarResource(groupedResources, selector) _
139137 val diskResourceMatch = matchDiskResource(groupedResources, selector) _
140138
141139 // Local volumes only need to be matched if we are making a reservation for resident tasks --
Show All 28 Lines
170168 noOfferMatchReasons += NoOfferMatchReason.UnfulfilledRole
171169 }
172170
173171 logUnsatisfiedResources(offer, selector, scalarMatchResults)
174172
175173 def portsMatchOpt: Option[PortsMatch] = PortsMatcher(runSpec, offer, selector).portsMatch
176174
177175 val meetsAllConstraints: Boolean = {
178 lazy val instances = runningInstances.filter { inst =>
179 inst.isLaunched && inst.runSpecVersion >= runSpec.versionInfo.lastConfigChangeVersion
timcharperUnsubmitted
Not Done

wondering why did we have the .isLaunched filter here before?

176 lazy val instances = knownInstances.filter { inst =>
177 // we ignore instances of older configurations: this way we can place a new instance on an agent that already
178 // hosts an old instance that will be killed once a new one is running/healthy/ready
179 inst.runSpecVersion >= runSpec.versionInfo.lastConfigChangeVersion
180180 }
181181 val badConstraints = runSpec.constraints.filterNot { constraint =>
182182 Constraints.meetsConstraint(instances, offer, constraint)
183183 }
184184
185185 if (badConstraints.nonEmpty) {
186186 // Add constraints to noOfferMatchReasons
187187 noOfferMatchReasons += NoOfferMatchReason.UnfulfilledConstraint
188 if (log.isInfoEnabled) {
189 log.info(
188 logger.info(
190189 s"Offer [${offer.getId.getValue}]. Constraints for run spec [${runSpec.id}] not satisfied.\n" +
191190 s"The conflicting constraints are: [${badConstraints.mkString(", ")}]"
192191 )
193192 }
194 }
195193
196194 badConstraints.isEmpty
197195 }
198196
199197 val resourceMatchOpt = if (scalarMatchResults.forall(_.matches) && meetsAllConstraints) {
200198 portsMatchOpt match {
201199 case Some(portsMatch) =>
202200 Some(ResourceMatch(scalarMatchResults.collect { case m: ScalarMatch => m }, portsMatch))
▲ Show 20 Lines • Show All 259 Lines • ▼ Show 20 Line(s)
462460 }
463461 }
464462 }
465463
466464 private[this] def logUnsatisfiedResources(
467465 offer: Offer,
468466 selector: ResourceSelector,
469467 scalarMatchResults: Seq[ScalarMatchResult]): Unit = {
470 if (log.isInfoEnabled && scalarMatchResults.exists(!_.matches)) {
468 if (scalarMatchResults.exists(!_.matches)) {
471469 val basicResourceString = scalarMatchResults.mkString(", ")
472 log.info(
470 logger.info(
473471 s"Offer [${offer.getId.getValue}]. " +
474472 s"$selector. " +
475473 s"Not all basic resources satisfied: $basicResourceString")
476474 }
477475 }
478476}
View Options

src/main/scala/mesosphere/mesos/RunSpecOfferMatcher.scala

11package mesosphere.mesos
22
33import com.google.protobuf.TextFormat
4import com.typesafe.scalalogging.StrictLogging
45import mesosphere.marathon.core.instance.Instance
56import mesosphere.marathon.core.pod.PodDefinition
67import mesosphere.marathon.state.{ AppDefinition, RunSpec }
78import mesosphere.mesos.ResourceMatcher.ResourceSelector
89import org.apache.mesos.Protos.Offer
9import org.slf4j.LoggerFactory
1010
1111import scala.collection.immutable.Seq
1212
13object RunSpecOfferMatcher {
13object RunSpecOfferMatcher extends StrictLogging {
1414
15 val log = LoggerFactory.getLogger(getClass)
16
17 def matchOffer(runSpec: RunSpec, offer: Offer, otherInstances: => Seq[Instance], givenAcceptedResourceRoles: Set[String]): ResourceMatchResponse = {
15 /**
16 * @param runSpec The runSpec for which the given offer shall be matched
17 * @param offer The Mesos offer that shall be matched
18 * @param knownInstances All instances associated with the given runSpec, needed to validate constraints
19 * @param givenAcceptedResourceRoles The resource roles for which to look.
20 */
21 def matchOffer(runSpec: RunSpec, offer: Offer, knownInstances: => Seq[Instance], givenAcceptedResourceRoles: Set[String]): ResourceMatchResponse = {
1822 val acceptedResourceRoles: Set[String] = {
1923 val roles = if (runSpec.acceptedResourceRoles.isEmpty) {
2024 givenAcceptedResourceRoles
2125 } else {
2226 runSpec.acceptedResourceRoles
2327 }
24 if (log.isDebugEnabled) log.debug(s"acceptedResourceRoles $roles")
28 logger.debug(s"acceptedResourceRoles $roles")
2529 roles
2630 }
2731
2832 val resourceMatchResponse =
29 ResourceMatcher.matchResources(offer, runSpec, otherInstances, ResourceSelector.any(acceptedResourceRoles))
33 ResourceMatcher.matchResources(offer, runSpec, knownInstances, ResourceSelector.any(acceptedResourceRoles))
3034
3135 def logInsufficientResources(): Unit = {
3236 val runSpecHostPorts = runSpec match {
3337 case app: AppDefinition => if (app.requirePorts) app.portNumbers else app.portNumbers.map(_ => 0)
3438 case pod: PodDefinition => pod.containers.flatMap(container => container.endpoints.flatMap(_.hostPort))
3539 }
3640 val hostPorts = runSpec.container.withFilter(_.portMappings.nonEmpty).map(_.hostPorts).getOrElse(runSpecHostPorts.map(Some(_)))
3741 val staticHostPorts = hostPorts.filter(!_.contains(0))
Show All 10 Lines
4852 } else {
4953 None
5054 }
5155
5256 val portStrings = Seq(maybeStatic, maybeDynamic).flatten.mkString(" + ")
5357
5458 val portsString = s"ports=($portStrings)"
5559
56 log.debug(
60 logger.debug(
5761 s"Offer [${offer.getId.getValue}]. Insufficient resources for [${runSpec.id}] " +
5862 s"(need cpus=${runSpec.resources.cpus}, mem=${runSpec.resources.mem}, disk=${runSpec.resources.disk}, " +
5963 s"gpus=${runSpec.resources.gpus}, $portsString, available in offer: " +
6064 s"[${TextFormat.shortDebugString(offer)}]"
6165 )
6266 }
6367
6468 resourceMatchResponse match {
6569 case matches: ResourceMatchResponse.Match =>
66 log.debug(s"Offer [${offer.getId.getValue}] matches resources for [${runSpec.id}].")
70 logger.debug(s"Offer [${offer.getId.getValue}] matches resources for [${runSpec.id}].")
6771 matches
6872 case matchesNot: ResourceMatchResponse.NoMatch =>
69 if (log.isDebugEnabled) logInsufficientResources()
73 logInsufficientResources()
7074 matchesNot
7175 }
7276 }
7377}
View Options

src/test/scala/mesosphere/marathon/integration/ResidentTaskIntegrationTest.scala

Show First 20 Lines • Show All 221 Lines • ▼ Show 20 Line(s)
222222 all should have size 5
223223
224224 And("exactly 5 instances are running")
225225 all.filter(_.launched) should have size 5
226226
227227 And("all 5 tasks are of the new version")
228228 all.map(_.version).forall(_.contains(newVersion)) shouldBe true
229229 }
230
231 /**
232 * FIXME (3043): implement the following tests. TASK_LOST can be induced when launching a task with permission:
233 *
234 * When a framework launches a task, “run_tasks” ACLs are checked to see if the framework
235 * (FrameworkInfo.principal) is authorized to run the task/executor as the given user. If not authorized,
236 * the launch is rejected and the framework gets a TASK_LOST.
237 *
238 * (From http://mesos.apache.org/documentation/latest/authorization/)
239 */
240
241 "taskLostBehavior = RELAUNCH_AFTER_TIMEOUT, timeout = 10s" in {
242 pending
243 Given("A resident app with 1 instance")
244 When("The task is lost")
245 Then("The task is not relaunched within the timeout")
246 And("The task is relaunched with a new Id after the timeout")
247 }
248
249 "taskLostBehavior = WAIT_FOREVER" in {
250 pending
251 Given("A resident app with 1 instance")
252 When("The task is lost")
253 Then("No timeout is scheduled") // can we easily verify this?
254 And("The task is not relaunched") // can we verify this without waiting?
255 }
256
257 "relaunchEscalationTimeoutSeconds = 5s" in {
258 pending
259 Given("A resident app with 1 instance")
260 When("The task terminates")
261 And("We don't get an offer within the timeout")
262 Then("We launch a new task on any matching offer")
263 }
264
265230 }
266231
267232 class Fixture {
268233
269234 val cpus: Double = 0.001
270235 val mem: Double = 1.0
271236 val disk: Double = 1.0
272237 val gpus: Double = 0.0
▲ Show 20 Lines • Show All 115 Lines • Show Last 20 Lines
View Options

src/test/scala/mesosphere/mesos/ResourceMatcherTest.scala

11package mesosphere.mesos
22
33import mesosphere.UnitTest
44import mesosphere.marathon.Protos.Constraint
55import mesosphere.marathon.Protos.Constraint.Operator
66import mesosphere.marathon._
77import mesosphere.marathon.core.instance.{ Instance, TestInstanceBuilder }
88import mesosphere.marathon.core.launcher.impl.TaskLabels
99import mesosphere.marathon.core.pod.{ BridgeNetwork, ContainerNetwork }
1010import mesosphere.marathon.core.task.Task
11import mesosphere.marathon.core.task.state.NetworkInfoTestDefaults
1112import mesosphere.marathon.raml.Resources
1213import mesosphere.marathon.state.PathId._
1314import mesosphere.marathon.state.VersionInfo._
1415import mesosphere.marathon.state._
1516import mesosphere.marathon.stream.Implicits._
1617import mesosphere.marathon.tasks.PortsMatcher
1718import mesosphere.marathon.test.MarathonTestHelper
1819import mesosphere.mesos.ResourceMatcher.ResourceSelector
Show All 15 Lines
3435 offer.getResourcesList.find(_.getName == "disk") should be('empty)
3536
3637 val app = AppDefinition(
3738 id = "/test".toRootPath,
3839 resources = Resources(cpus = 1.0, mem = 128.0, disk = 0.0),
3940 portDefinitions = PortDefinitions(0, 0)
4041 )
4142
42 val resourceMatchResponse = ResourceMatcher.matchResources(offer, app, runningInstances = Seq.empty, unreservedResourceSelector)
43 val resourceMatchResponse = ResourceMatcher.matchResources(offer, app, knownInstances = Seq.empty, unreservedResourceSelector)
4344
4445 resourceMatchResponse shouldBe a[ResourceMatchResponse.Match]
4546 val res = resourceMatchResponse.asInstanceOf[ResourceMatchResponse.Match].resourceMatch
4647
4748 res.scalarMatch(Resource.CPUS).get.roles should be(Seq(ResourceRole.Unreserved))
4849 res.scalarMatch(Resource.MEM).get.roles should be(Seq(ResourceRole.Unreserved))
4950 res.scalarMatch(Resource.DISK) should be(empty)
5051
5152 res.hostPorts should have size 2
5253 }
5354
5455 "match resources success" in {
5556 val offer = MarathonTestHelper.makeBasicOffer().build()
5657 val app = AppDefinition(
5758 id = "/test".toRootPath,
5859 resources = Resources(cpus = 1.0, mem = 128.0, disk = 0.0),
5960 portDefinitions = PortDefinitions(0, 0)
6061 )
6162
62 val resourceMatchResponse = ResourceMatcher.matchResources(offer, app, runningInstances = Seq.empty, unreservedResourceSelector)
63 val resourceMatchResponse = ResourceMatcher.matchResources(offer, app, knownInstances = Seq.empty, unreservedResourceSelector)
6364
6465 resourceMatchResponse shouldBe a[ResourceMatchResponse.Match]
6566 val res = resourceMatchResponse.asInstanceOf[ResourceMatchResponse.Match].resourceMatch
6667
6768 res.scalarMatch(Resource.CPUS).get.roles should be(Seq(ResourceRole.Unreserved))
6869 res.scalarMatch(Resource.MEM).get.roles should be(Seq(ResourceRole.Unreserved))
6970 res.scalarMatch(Resource.DISK) should be(empty)
7071
Show All 11 Lines
8283
8384 portMappings = Seq(
8485 Container.PortMapping(31001, Some(0), 0, "tcp", Some("qax")),
8586 Container.PortMapping(31002, Some(0), 0, "tcp", Some("qab"))
8687 )
8788 ))
8889 )
8990
90 val resourceMatchResponse = ResourceMatcher.matchResources(offer, app, runningInstances = Seq.empty, unreservedResourceSelector)
91 val resourceMatchResponse = ResourceMatcher.matchResources(offer, app, knownInstances = Seq.empty, unreservedResourceSelector)
9192
9293 resourceMatchResponse shouldBe a[ResourceMatchResponse.Match]
9394 val res = resourceMatchResponse.asInstanceOf[ResourceMatchResponse.Match].resourceMatch
9495
9596 res.scalarMatch(Resource.CPUS).get.roles should be(Seq(ResourceRole.Unreserved))
9697 res.scalarMatch(Resource.MEM).get.roles should be(Seq(ResourceRole.Unreserved))
9798 res.scalarMatch(Resource.DISK) should be(empty)
9899
Show All 12 Lines
111112 portMappings = Seq(
112113 Container.PortMapping(0, Some(0), 0, "tcp", Some("yas")),
113114 Container.PortMapping(31001, None, 0, "tcp", Some("qax")),
114115 Container.PortMapping(31002, Some(0), 0, "tcp", Some("qab"))
115116 )
116117 ))
117118 )
118119
119 val resourceMatchResponse = ResourceMatcher.matchResources(offer, app, runningInstances = Seq.empty, unreservedResourceSelector)
120 val resourceMatchResponse = ResourceMatcher.matchResources(offer, app, knownInstances = Seq.empty, unreservedResourceSelector)
120121
121122 resourceMatchResponse shouldBe a[ResourceMatchResponse.Match]
122123 val res = resourceMatchResponse.asInstanceOf[ResourceMatchResponse.Match].resourceMatch
123124
124125 res.scalarMatch(Resource.CPUS).get.roles should be(Seq(ResourceRole.Unreserved))
125126 res.scalarMatch(Resource.MEM).get.roles should be(Seq(ResourceRole.Unreserved))
126127 res.scalarMatch(Resource.DISK) should be(empty)
127128
Show All 22 Lines
150151 val app = AppDefinition(
151152 id = "/test".toRootPath,
152153 resources = Resources(cpus = 2.0, mem = 128.0, disk = 2.0),
153154 portDefinitions = PortDefinitions(0)
154155 )
155156
156157 val resourceMatchResponse = ResourceMatcher.matchResources(
157158 offer, app,
158 runningInstances = Seq(), ResourceSelector.reservedWithLabels(Set(ResourceRole.Unreserved, "marathon"), labels))
159 knownInstances = Seq(), ResourceSelector.reservedWithLabels(Set(ResourceRole.Unreserved, "marathon"), labels))
159160
160161 resourceMatchResponse shouldBe a[ResourceMatchResponse.Match]
161162 val res = resourceMatchResponse.asInstanceOf[ResourceMatchResponse.Match].resourceMatch
162163
163164 res.scalarMatches should have size 3
164165 res.scalarMatch(Resource.CPUS).get.consumed.toSet should be(
165166 Set(
166167 GeneralScalarMatch.Consumption(1.0, "marathon", reservation = Some(cpuReservation)),
Show All 14 Lines
181182
182183 res.portsMatch.hostPortsWithRole.toSet should be(
183184 Set(Some(PortsMatcher.PortWithRole(ResourceRole.Unreserved, 80, reservation = Some(portsReservation))))
184185 )
185186
186187 // reserved resources with labels should not be matched by selector if don't match for reservation with labels
187188 ResourceMatcher.matchResources(
188189 offer, app,
189 runningInstances = Seq(), ResourceSelector.any(Set(ResourceRole.Unreserved, "marathon"))) shouldBe a[ResourceMatchResponse.NoMatch]
190 knownInstances = Seq(), ResourceSelector.any(Set(ResourceRole.Unreserved, "marathon"))) shouldBe a[ResourceMatchResponse.NoMatch]
190191 }
191192
192193 "dynamically reserved resources are matched if they have no labels" in {
193194 val cpuReservation = MarathonTestHelper.reservation(principal = "cpuPrincipal")
194195 val cpuReservation2 = MarathonTestHelper.reservation(principal = "cpuPrincipal")
195196 val memReservation = MarathonTestHelper.reservation(principal = "memPrincipal")
196197 val diskReservation = MarathonTestHelper.reservation(principal = "memPrincipal")
197198 val portsReservation = MarathonTestHelper.reservation(principal = "portPrincipal")
Show All 11 Lines
209210 val app = AppDefinition(
210211 id = "/test".toRootPath,
211212 resources = Resources(cpus = 2.0, mem = 128.0, disk = 2.0),
212213 portDefinitions = PortDefinitions(0)
213214 )
214215
215216 val resourceMatchResponse = ResourceMatcher.matchResources(
216217 offer, app,
217 runningInstances = Seq(), ResourceSelector.any(Set(ResourceRole.Unreserved, "marathon")))
218 knownInstances = Seq(), ResourceSelector.any(Set(ResourceRole.Unreserved, "marathon")))
218219
219220 resourceMatchResponse shouldBe a[ResourceMatchResponse.Match]
220221 val res = resourceMatchResponse.asInstanceOf[ResourceMatchResponse.Match].resourceMatch
221222
222223 res.scalarMatches should have size 3
223224 res.scalarMatch(Resource.CPUS).get.consumed.toSet should be(
224225 Set(
225226 GeneralScalarMatch.Consumption(1.0, "marathon", reservation = Some(cpuReservation)),
Show All 38 Lines
264265 val app = AppDefinition(
265266 id = "/test".toRootPath,
266267 resources = Resources(cpus = 2.0, mem = 128.0, disk = 2.0),
267268 portDefinitions = PortDefinitions(0)
268269 )
269270
270271 val resourceMatchResponse = ResourceMatcher.matchResources(
271272 offer, app,
272 runningInstances = Seq(), ResourceSelector.any(Set(ResourceRole.Unreserved, "marathon")))
273 knownInstances = Seq(), ResourceSelector.any(Set(ResourceRole.Unreserved, "marathon")))
273274
274275 resourceMatchResponse shouldBe a[ResourceMatchResponse.NoMatch]
275276 }
276277
277278 "ResourceSelector.reservedWithLabels should not match disk resource without label" in {
278279 val cpuReservation = MarathonTestHelper.reservation(principal = "cpuPrincipal", labels = Map("some" -> "label"))
279280 val memReservation = MarathonTestHelper.reservation(principal = "memPrincipal", labels = Map("some" -> "label"))
280281
281282 val offer =
282283 MarathonTestHelper.makeBasicOffer(role = "marathon")
283284 .clearResources()
284285 .addResources(MarathonTestHelper.scalarResource("cpus", 1.0, role = "marathon", reservation = Some(cpuReservation)))
285286 .addResources(MarathonTestHelper.scalarResource("mem", 128.0, reservation = Some(memReservation)))
286287 .addResources(MarathonTestHelper.reservedDisk(id = "disk", size = 1024.0))
287288 .build()
288289
289290 val app = AppDefinition(
290291 id = "/test".toRootPath,
291292 resources = Resources(cpus = 1.0, mem = 128.0, disk = 2.0),
292293 portDefinitions = PortDefinitions()
293294 )
294295
295296 val resourceMatchResponse = ResourceMatcher.matchResources(
296297 offer, app,
297 runningInstances = Seq(), ResourceSelector.reservedWithLabels(Set(ResourceRole.Unreserved, "marathon"), Map("some" -> "label"))
298 knownInstances = Seq(), ResourceSelector.reservedWithLabels(Set(ResourceRole.Unreserved, "marathon"), Map("some" -> "label"))
298299 )
299300
300301 resourceMatchResponse shouldBe a[ResourceMatchResponse.NoMatch]
301302 }
302303
303304 "match resources success with preserved roles" in {
304305 val offer = MarathonTestHelper.makeBasicOffer(role = "marathon").build()
305306 val app = AppDefinition(
306307 id = "/test".toRootPath,
307308 resources = Resources(cpus = 1.0, mem = 128.0, disk = 0.0),
308309 portDefinitions = PortDefinitions(0, 0)
309310 )
310311
311312 val resourceMatchResponse = ResourceMatcher.matchResources(
312313 offer, app,
313 runningInstances = Seq(), ResourceSelector.any(Set("marathon")))
314 knownInstances = Seq(), ResourceSelector.any(Set("marathon")))
314315
315316 resourceMatchResponse shouldBe a[ResourceMatchResponse.Match]
316317 val res = resourceMatchResponse.asInstanceOf[ResourceMatchResponse.Match].resourceMatch
317318
318319 res.scalarMatch(Resource.CPUS).get.roles should be(Seq("marathon"))
319320 res.scalarMatch(Resource.MEM).get.roles should be(Seq("marathon"))
320321 res.scalarMatch(Resource.DISK) should be(empty)
321322 }
322323
323324 "match resources failure because of incorrect roles" in {
324325 val offer = MarathonTestHelper.makeBasicOffer(role = "marathon").build()
325326 val app = AppDefinition(
326327 id = "/test".toRootPath,
327328 resources = Resources(cpus = 1.0, mem = 128.0, disk = 0.0),
328329 portDefinitions = PortDefinitions(0, 0)
329330 )
330331
331332 val resourceMatchResponse = ResourceMatcher.matchResources(
332333 offer, app,
333 runningInstances = Seq(), unreservedResourceSelector)
334 knownInstances = Seq(), unreservedResourceSelector)
334335
335336 resourceMatchResponse shouldBe a[ResourceMatchResponse.NoMatch]
336337 }
337338
338339 "match resources success with constraints" in {
339340 val offer = MarathonTestHelper.makeBasicOffer(beginPort = 0, endPort = 0).setHostname("host1").build()
340341 val app = AppDefinition(
341342 id = "/test".toRootPath,
342343 resources = Resources(cpus = 1.0, mem = 128.0, disk = 0.0),
343344 constraints = Set(
344345 Constraint.newBuilder
345346 .setField("hostname")
346347 .setOperator(Operator.LIKE)
347348 .setValue("host1")
348349 .build()
349350 )
350351 )
351352
352 val resourceMatchResponse = ResourceMatcher.matchResources(offer, app, runningInstances = Seq.empty, unreservedResourceSelector)
353 val resourceMatchResponse = ResourceMatcher.matchResources(offer, app, knownInstances = Seq.empty, unreservedResourceSelector)
353354
354355 resourceMatchResponse should not be a[ResourceMatchResponse.NoMatch]
355356 }
356357
357358 "match resources fails on constraints" in {
358359 val offer = MarathonTestHelper.makeBasicOffer(beginPort = 0, endPort = 0).setHostname("host1").build()
359360 val app = AppDefinition(
360361 id = "/test".toRootPath,
361362 resources = Resources(cpus = 1.0, mem = 128.0, disk = 0.0),
362363 constraints = Set(
363364 Constraint.newBuilder
364365 .setField("hostname")
365366 .setOperator(Operator.LIKE)
366367 .setValue("host2")
367368 .build()
368369 )
369370 )
370371
371 val resourceMatchResponse = ResourceMatcher.matchResources(offer, app, runningInstances = Seq.empty, unreservedResourceSelector)
372 val resourceMatchResponse = ResourceMatcher.matchResources(offer, app, knownInstances = Seq.empty, unreservedResourceSelector)
372373
373374 resourceMatchResponse shouldBe a[ResourceMatchResponse.NoMatch]
374375 }
375376
376377 "match resources fail on cpu" in {
377378 val offer = MarathonTestHelper.makeBasicOffer(cpus = 0.1).build()
378379 val app = AppDefinition(
379380 id = "/test".toRootPath,
380381 resources = Resources(cpus = 1.0, mem = 128.0, disk = 0.0),
381382 portDefinitions = PortDefinitions(0, 0)
382383 )
383384
384 val resourceMatchResponse = ResourceMatcher.matchResources(offer, app, runningInstances = Seq.empty, unreservedResourceSelector)
385 val resourceMatchResponse = ResourceMatcher.matchResources(offer, app, knownInstances = Seq.empty, unreservedResourceSelector)
385386
386387 resourceMatchResponse shouldBe a[ResourceMatchResponse.NoMatch]
387388 }
388389
389390 "match resources fail on mem" in {
390391 val offer = MarathonTestHelper.makeBasicOffer(mem = 0.1).build()
391392 val app = AppDefinition(
392393 id = "/test".toRootPath,
393394 resources = Resources(cpus = 1.0, mem = 128.0, disk = 0.0),
394395 portDefinitions = PortDefinitions(0, 0)
395396 )
396397
397 val resourceMatchResponse = ResourceMatcher.matchResources(offer, app, runningInstances = Seq.empty, unreservedResourceSelector)
398 val resourceMatchResponse = ResourceMatcher.matchResources(offer, app, knownInstances = Seq.empty, unreservedResourceSelector)
398399
399400 resourceMatchResponse shouldBe a[ResourceMatchResponse.NoMatch]
400401 }
401402
402403 "match resources should always match constraints and therefore return NoOfferMatchReason.UnfulfilledConstraint in case of no match" in {
403404 val offer = MarathonTestHelper.makeBasicOffer(cpus = 0.5).build()
404405 val app = AppDefinition(
405406 id = "/test".toRootPath,
406407 resources = Resources(cpus = 1.0, mem = 128.0, disk = 0.0), // cpu does not match
407408 constraints = Set(
408409 Constraint.newBuilder.setField("test") // and constraint does not match
409410 .setOperator(Operator.LIKE)
410411 .setValue("test")
411412 .build()
412413 )
413414 )
414415
415 val resourceMatchResponse = ResourceMatcher.matchResources(offer, app, runningInstances = Seq.empty, unreservedResourceSelector)
416 val resourceMatchResponse = ResourceMatcher.matchResources(offer, app, knownInstances = Seq.empty, unreservedResourceSelector)
416417
417418 resourceMatchResponse shouldBe a[ResourceMatchResponse.NoMatch]
418419 val noMatch = resourceMatchResponse.asInstanceOf[ResourceMatchResponse.NoMatch]
419420
420421 noMatch.reasons should contain(NoOfferMatchReason.UnfulfilledConstraint)
421422 noMatch.reasons should contain(NoOfferMatchReason.InsufficientCpus)
422423 }
423424
424425 "match resources fail on disk" in {
425426 val offer = MarathonTestHelper.makeBasicOffer(disk = 0.1).build()
426427 val app = AppDefinition(
427428 id = "/test".toRootPath,
428429 resources = Resources(cpus = 1.0, mem = 128.0, disk = 1.0),
429430 portDefinitions = PortDefinitions(0, 0)
430431 )
431432
432 val resourceMatchResponse = ResourceMatcher.matchResources(offer, app, runningInstances = Seq.empty, unreservedResourceSelector)
433 val resourceMatchResponse = ResourceMatcher.matchResources(offer, app, knownInstances = Seq.empty, unreservedResourceSelector)
433434
434435 resourceMatchResponse shouldBe a[ResourceMatchResponse.NoMatch]
435436 }
436437
437438 "match resources fail on ports" in {
438439 val offer = MarathonTestHelper.makeBasicOffer(beginPort = 0, endPort = 0).build()
439440 val app = AppDefinition(
440441 id = "/test".toRootPath,
441442 resources = Resources(cpus = 1.0, mem = 128.0, disk = 0.0),
442443 portDefinitions = PortDefinitions(1, 2)
443444 )
444445
445 val resourceMatchResponse = ResourceMatcher.matchResources(offer, app, runningInstances = Seq.empty, unreservedResourceSelector)
446 val resourceMatchResponse = ResourceMatcher.matchResources(offer, app, knownInstances = Seq.empty, unreservedResourceSelector)
446447
447448 resourceMatchResponse shouldBe a[ResourceMatchResponse.NoMatch]
448449 }
449450
450451 "resource matcher should not respond with NoOfferMatchReason.UnfulfilledRole if role matches" in {
451452 val offer = MarathonTestHelper.makeBasicOffer(cpus = 0.5, role = "A").build()
452453 val app = AppDefinition(
453454 id = "/test".toRootPath,
454455 resources = Resources(cpus = 1.0, mem = 128.0, disk = 0.0), // make sure it mismatches
455456 acceptedResourceRoles = Set("A", "B")
456457 )
457458
458 val resourceMatchResponse = ResourceMatcher.matchResources(offer, app, runningInstances = Seq.empty, ResourceSelector.any(Set("A", "B")))
459 val resourceMatchResponse = ResourceMatcher.matchResources(offer, app, knownInstances = Seq.empty, ResourceSelector.any(Set("A", "B")))
459460
460461 resourceMatchResponse shouldBe a[ResourceMatchResponse.NoMatch]
461462 val noMatch = resourceMatchResponse.asInstanceOf[ResourceMatchResponse.NoMatch]
462463
463464 noMatch.reasons should not contain NoOfferMatchReason.UnfulfilledRole
464465 }
465466
466467 "resource matcher should respond with NoOfferMatchReason.UnfulfilledRole if runSpec requires unreserved Role but resources are reserved" in {
467468 val offer = MarathonTestHelper.makeBasicOffer(cpus = 0.5, role = "A").build()
468469 val app = AppDefinition(
469470 id = "/test".toRootPath,
470471 resources = Resources(cpus = 1.0, mem = 128.0, disk = 0.0), // make sure it mismatches
471472 acceptedResourceRoles = Set(ResourceRole.Unreserved)
472473 )
473474
474 val resourceMatchResponse = ResourceMatcher.matchResources(offer, app, runningInstances = Seq.empty, unreservedResourceSelector)
475 val resourceMatchResponse = ResourceMatcher.matchResources(offer, app, knownInstances = Seq.empty, unreservedResourceSelector)
475476
476477 resourceMatchResponse shouldBe a[ResourceMatchResponse.NoMatch]
477478 val noMatch = resourceMatchResponse.asInstanceOf[ResourceMatchResponse.NoMatch]
478479
479480 noMatch.reasons should contain(NoOfferMatchReason.UnfulfilledRole)
480481 }
481482
482483 "resource matcher should respond with NoOfferMatchReason.UnfulfilledRole if runSpec has no role defined" in {
483484 val offer = MarathonTestHelper.makeBasicOffer(cpus = 0.5, role = "A").build()
484485 val app = AppDefinition(
485486 id = "/test".toRootPath,
486487 resources = Resources(cpus = 1.0, mem = 128.0, disk = 0.0) // make sure it mismatches
487488 )
488489
489 val resourceMatchResponse = ResourceMatcher.matchResources(offer, app, runningInstances = Seq.empty, unreservedResourceSelector)
490 val resourceMatchResponse = ResourceMatcher.matchResources(offer, app, knownInstances = Seq.empty, unreservedResourceSelector)
490491
491492 resourceMatchResponse shouldBe a[ResourceMatchResponse.NoMatch]
492493 val noMatch = resourceMatchResponse.asInstanceOf[ResourceMatchResponse.NoMatch]
493494
494495 noMatch.reasons should contain(NoOfferMatchReason.UnfulfilledRole)
495496 }
496497
497498 "resource matcher should respond with NoOfferMatchReason.UnfulfilledRole if role mismatches and offer contains other role" in {
498499 val offer = MarathonTestHelper.makeBasicOffer(cpus = 0.5, role = "C").build()
499500 val app = AppDefinition(
500501 id = "/test".toRootPath,
501502 resources = Resources(cpus = 1.0, mem = 128.0, disk = 0.0), // make sure it mismatches
502503 acceptedResourceRoles = Set("A", "B")
503504 )
504505
505 val resourceMatchResponse = ResourceMatcher.matchResources(offer, app, runningInstances = Seq.empty, unreservedResourceSelector)
506 val resourceMatchResponse = ResourceMatcher.matchResources(offer, app, knownInstances = Seq.empty, unreservedResourceSelector)
506507
507508 resourceMatchResponse shouldBe a[ResourceMatchResponse.NoMatch]
508509 val noMatch = resourceMatchResponse.asInstanceOf[ResourceMatchResponse.NoMatch]
509510
510511 noMatch.reasons should contain(NoOfferMatchReason.UnfulfilledRole)
511512 }
512513
513514 "resource matcher should respond with all NoOfferMatchReason.Insufficient{Cpus, Memory, Gpus, Disk} if mismatches" in {
514515 val offer = MarathonTestHelper.makeBasicOffer(cpus = 1, mem = 1, disk = 1, gpus = 1).build()
515516 val app = AppDefinition(
516517 id = "/test".toRootPath,
517518 resources = Resources(cpus = 2, mem = 2, disk = 2, gpus = 2) // make sure it mismatches
518519 )
519520
520 val resourceMatchResponse = ResourceMatcher.matchResources(offer, app, runningInstances = Seq.empty, unreservedResourceSelector)
521 val resourceMatchResponse = ResourceMatcher.matchResources(offer, app, knownInstances = Seq.empty, unreservedResourceSelector)
521522
522523 resourceMatchResponse shouldBe a[ResourceMatchResponse.NoMatch]
523524 val noMatch = resourceMatchResponse.asInstanceOf[ResourceMatchResponse.NoMatch]
524525
525526 noMatch.reasons should contain allOf (NoOfferMatchReason.InsufficientCpus, NoOfferMatchReason.InsufficientMemory,
526527 NoOfferMatchReason.InsufficientGpus, NoOfferMatchReason.InsufficientDisk)
527528 }
528529
529530 "resource matcher should respond with NoOfferMatchReason.InsufficientPorts if ports mismatch and other requirements matches" in {
530531 val offer = MarathonTestHelper.makeBasicOffer(cpus = 1, mem = 1, disk = 1, beginPort = 0, endPort = 0).build()
531532 val app = AppDefinition(
532533 id = "/test".toRootPath,
533534 resources = Resources(cpus = 1, mem = 1, disk = 1),
534535 portDefinitions = PortDefinitions(1, 2) // this match fails
535536 )
536537
537 val resourceMatchResponse = ResourceMatcher.matchResources(offer, app, runningInstances = Seq.empty, unreservedResourceSelector)
538 val resourceMatchResponse = ResourceMatcher.matchResources(offer, app, knownInstances = Seq.empty, unreservedResourceSelector)
538539
539540 resourceMatchResponse shouldBe a[ResourceMatchResponse.NoMatch]
540541 val noMatch = resourceMatchResponse.asInstanceOf[ResourceMatchResponse.NoMatch]
541542
542543 noMatch.reasons should be(Seq(NoOfferMatchReason.InsufficientPorts))
543544 }
544545
545546 "resource matcher should not respond with NoOfferMatchReason.InsufficientPorts other requirements mismatches, even if port requirements mismatch" in {
546547 // NoOfferMatchReason.InsufficientPorts is calculated lazy and should only be calculated if all other requirements matches
547548 val offer = MarathonTestHelper.makeBasicOffer(cpus = 1, mem = 1, disk = 1, beginPort = 0, endPort = 0).build()
548549 val app = AppDefinition(
549550 id = "/test".toRootPath,
550551 resources = Resources(cpus = 2, mem = 1, disk = 1), // this match fails
551552 portDefinitions = PortDefinitions(1, 2) // this would fail as well, but is not evaluated of the resource matcher
552553 )
553554
554 val resourceMatchResponse = ResourceMatcher.matchResources(offer, app, runningInstances = Seq.empty, unreservedResourceSelector)
555 val resourceMatchResponse = ResourceMatcher.matchResources(offer, app, knownInstances = Seq.empty, unreservedResourceSelector)
555556
556557 resourceMatchResponse shouldBe a[ResourceMatchResponse.NoMatch]
557558 val noMatch = resourceMatchResponse.asInstanceOf[ResourceMatchResponse.NoMatch]
558559
559560 noMatch.reasons should not contain NoOfferMatchReason.InsufficientPorts
560561 }
561562
562563 "match resources success with constraints and old tasks in previous version" in {
▲ Show 20 Lines • Show All 114 Lines • ▼ Show 20 Line(s)
677678 disk = 0.0
678679 ),
679680 container = Some(Container.Mesos(
680681 volumes = List(volume))),
681682 versionInfo = OnlyVersion(Timestamp(2)))
682683
683684 ResourceMatcher.matchResources(
684685 offerDisksTooSmall, app,
685 runningInstances = Seq(),
686 knownInstances = Seq(),
686687 ResourceSelector.reservable) shouldBe a[ResourceMatchResponse.NoMatch]
687688
688689 val resourceMatchResponse = ResourceMatcher.matchResources(
689690 offerSufficeWithMultOffers, app,
690 runningInstances = Seq(),
691 knownInstances = Seq(),
691692 ResourceSelector.reservable)
692693
693694 resourceMatchResponse shouldBe a[ResourceMatchResponse.Match]
694695 resourceMatchResponse.asInstanceOf[ResourceMatchResponse.Match].resourceMatch.scalarMatch("disk").get.consumed.toSet shouldBe Set(
695696 DiskResourceMatch.Consumption(1024.0, "*", None, DiskSource(DiskType.Path, Some("/path2")), Some(volume)),
696697 DiskResourceMatch.Consumption(476.0, "*", None, DiskSource(DiskType.Path, Some("/path2")), Some(volume)))
697698 }
698699
Show All 21 Lines
720721 disk = 0.0
721722 ),
722723 container = Some(Container.Mesos(
723724 volumes = List(volume))),
724725 versionInfo = OnlyVersion(Timestamp(2)))
725726
726727 ResourceMatcher.matchResources(
727728 offers("/mnt/disk-a"), app,
728 runningInstances = Seq(),
729 knownInstances = Seq(),
729730 ResourceSelector.reservable) shouldBe a[ResourceMatchResponse.NoMatch]
730731
731732 ResourceMatcher.matchResources(
732733 offers("/mnt/disk-b"), app,
733 runningInstances = Seq(),
734 knownInstances = Seq(),
734735 ResourceSelector.reservable) shouldBe a[ResourceMatchResponse.Match]
735736 }
736737
737738 "mount disk enforces maxSize constraints" in {
738739 val offer =
739740 MarathonTestHelper.makeBasicOffer().
740741 addResources(
741742 MarathonTestHelper.scalarResource("disk", 1024.0,
Show All 19 Lines
761762 container = Some(Container.Mesos(
762763 volumes = List(volume))),
763764 versionInfo = OnlyVersion(Timestamp(2)))
764765 app
765766 }
766767
767768 inside(ResourceMatcher.matchResources(
768769 offer, mountRequest(500, None),
769 runningInstances = Seq(),
770 knownInstances = Seq(),
770771 ResourceSelector.reservable)) {
771772 case matches: ResourceMatchResponse.Match =>
772773 matches.resourceMatch.scalarMatches.collectFirst {
773774 case m: DiskResourceMatch =>
774775 (m.consumedValue, m.consumed.head.persistentVolume.get.persistent.size)
775776 } shouldBe Some((1024, 1024))
776777 }
777778
778779 ResourceMatcher.matchResources(
779780 offer, mountRequest(500, Some(750)),
780 runningInstances = Seq(),
781 knownInstances = Seq(),
781782 ResourceSelector.reservable) shouldBe a[ResourceMatchResponse.NoMatch]
782783
783784 ResourceMatcher.matchResources(
784785 offer, mountRequest(500, Some(1024)),
785 runningInstances = Seq(),
786 knownInstances = Seq(),
786787 ResourceSelector.reservable) shouldBe a[ResourceMatchResponse.Match]
787788 }
789
790 "a Reserved instance prevents creation of another reservation when hostname constraint is set" in {
791 val offer = MarathonTestHelper.makeBasicOffer().
792 addResources(MarathonTestHelper.scalarResource("disk", 1024.0)).
793 setHostname(NetworkInfoTestDefaults.defaultHostName).
794 build()
795
796 val volume = PersistentVolume(
797 containerPath = "/var/data",
798 mode = Mesos.Volume.Mode.RW,
799 persistent = PersistentVolumeInfo(
800 size = 500, `type` = DiskType.Root))
801
802 val app = AppDefinition(
803 id = "/test-persistent-volumes-with-unique-constraint".toRootPath,
804 instances = 3,
805 resources = Resources(cpus = 0.1, mem = 32.0, disk = 0.0),
806 constraints = Set(Constraint.newBuilder.setField("hostname").
807 setOperator(Constraint.Operator.UNIQUE).build),
808 container = Some(Container.Mesos(
809 volumes = List(volume))))
810
811 // Since offer matcher checks the instance version it's should be >= app.version
812 val instance = TestInstanceBuilder.newBuilder(app.id, version = app.version).addTaskReserved(
813 Task.LocalVolumeId(app.id, volume))
814 .getInstance()
815
816 val response = ResourceMatcher.matchResources(
817 offer, app, knownInstances = Seq(instance), ResourceSelector.reservable)
818
819 response shouldBe a[ResourceMatchResponse.NoMatch]
788820 }
821
822 "a Reserved instance DOES NOT prevent creation of another reservation when NO hostname constraint is set" in {
823 val offer = MarathonTestHelper.makeBasicOffer().
824 addResources(MarathonTestHelper.scalarResource("disk", 1024.0)).
825 setHostname(NetworkInfoTestDefaults.defaultHostName).
826 build()
827
828 val volume = PersistentVolume(
829 containerPath = "/var/data",
830 mode = Mesos.Volume.Mode.RW,
831 persistent = PersistentVolumeInfo(
832 size = 500, `type` = DiskType.Root))
833
834 val app = AppDefinition(
835 id = "/test-persistent-volumes-without-unique-constraint".toRootPath,
836 instances = 3,
837 resources = Resources(cpus = 0.1, mem = 32.0, disk = 0.0),
838 container = Some(Container.Mesos(
839 volumes = List(volume))))
840
841 // Since offer matcher checks the instance version it's should be >= app.version
842 val instance = TestInstanceBuilder.newBuilder(app.id, version = app.version).addTaskReserved(
843 Task.LocalVolumeId(app.id, volume))
844 .getInstance()
845
846 val response = ResourceMatcher.matchResources(
847 offer, app, knownInstances = Seq(instance), ResourceSelector.reservable)
848
849 response shouldBe a[ResourceMatchResponse.Match]
850 }
851 }
852
789853 val appId = PathId("/test")
790854 def instance(id: String, version: Timestamp, attrs: Map[String, String]): Instance = { // linter:ignore:UnusedParameter
791855 val attributes: Seq[Attribute] = attrs.map {
792856 case (name, v) => TextAttribute(name, v): Attribute
793857 }(collection.breakOut)
794858 TestInstanceBuilder.newBuilder(appId, version = version).addTaskWithBuilder().taskStaged()
795859 .build()
796860 .withAgentInfo(attributes = Some(attributes))
797861 .getInstance()
798862 }
799863
800864 lazy val unreservedResourceSelector = ResourceSelector.any(Set(ResourceRole.Unreserved))
801865}