forked from d2iq-archive/marathon
-
Notifications
You must be signed in to change notification settings - Fork 6
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Fix fullscan index issue on tasks health (#64)
As it was very slow with several thousands of tasks/instances, this uses two indexes instead. Co-authored-by: Jean-Baptiste Catté <[email protected]>
- Loading branch information
Showing
4 changed files
with
270 additions
and
14 deletions.
There are no files selected for viewing
This file contains 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
103 changes: 103 additions & 0 deletions
103
src/main/scala/mesosphere/marathon/core/health/impl/HealthIndex.scala
This file contains 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 |
|---|---|---|
| @@ -0,0 +1,103 @@ | ||
| package mesosphere.marathon | ||
| package core.health.impl | ||
|
|
||
| import mesosphere.marathon.core.health.Health | ||
| import mesosphere.marathon.core.instance.Instance | ||
| import mesosphere.marathon.core.task.Task | ||
|
|
||
| import scala.collection.concurrent.TrieMap | ||
|
|
||
| trait HealthIndex { | ||
| def removeLeftOverHealthIfAny(key: Instance.Id): Option[Health] | ||
|
|
||
| def +=(kv: (Task.Id, Health)): this.type | ||
|
|
||
| def values: Iterable[Health] | ||
|
|
||
| def filterKeys(p: Task.Id => Boolean): collection.Map[Task.Id, Health] | ||
|
|
||
| def get(key: Instance.Id): Option[Health] | ||
|
|
||
| def getOrElse(key: Task.Id, default: => Health): Health | ||
|
|
||
| def retain(p: Task.Id => Boolean): this.type | ||
| } | ||
|
|
||
| /** | ||
| * Old implementation, search by Instance.Id is slow, kept for reference and for tests. | ||
| */ | ||
| case class HealthSingleTrieMap() extends HealthIndex { | ||
| private val healthByTaskId: TrieMap[Task.Id, Health] = TrieMap.empty[Task.Id, Health] | ||
|
|
||
| def removeLeftOverHealthIfAny(key: Instance.Id): Option[Health] = { | ||
| val healthOfInstanceId = healthByTaskId.find(_._1.instanceId == key) | ||
| if (healthOfInstanceId.isDefined) | ||
| healthByTaskId.remove(healthOfInstanceId.get._1) | ||
| else | ||
| None | ||
| } | ||
|
|
||
| /** Slow, full scan algorithm O(n) */ | ||
| def get(key: Instance.Id): Option[Health] = healthByTaskId.find(_._1.instanceId == key).map(_._2) | ||
|
|
||
| def getOrElse(key: Task.Id, default: => Health): Health = healthByTaskId.getOrElse(key, default) | ||
|
|
||
| def filterKeys(p: Task.Id => Boolean): collection.Map[Task.Id, Health] = { | ||
| healthByTaskId.filterKeys(p) | ||
| } | ||
|
|
||
| /** Slow */ | ||
| def retain(p: Task.Id => Boolean): this.type = { | ||
| healthByTaskId.retain((taskId, _) => p(taskId)) | ||
| this | ||
| } | ||
|
|
||
| def +=(kv: (Task.Id, Health)): this.type = { | ||
| healthByTaskId += kv | ||
| this | ||
| } | ||
|
|
||
| def values: Iterable[Health] = healthByTaskId.values | ||
| } | ||
|
|
||
| /** | ||
| * New implementation, search by Instance.Id is fast but we need two indexes. | ||
| */ | ||
| case class HealthDualTrieMap() extends HealthIndex { | ||
| private val healthByTaskId: TrieMap[Task.Id, Health] = TrieMap.empty[Task.Id, Health] | ||
| private val taskIdByInstanceId: TrieMap[Instance.Id, Task.Id] = TrieMap.empty[Instance.Id, Task.Id] | ||
|
|
||
| def removeLeftOverHealthIfAny(key: Instance.Id): Option[Health] = | ||
| taskIdByInstanceId.remove(key).flatMap(oldTaskId => healthByTaskId.remove(oldTaskId)) | ||
|
|
||
| /** Fast O(log(n)) */ | ||
| def get(key: Instance.Id): Option[Health] = taskIdByInstanceId.get(key).flatMap(t => healthByTaskId.get(t)) | ||
|
|
||
| def getOrElse(key: Task.Id, default: => Health): Health = healthByTaskId.getOrElse(key, default) | ||
|
|
||
| def filterKeys(p: Task.Id => Boolean): collection.Map[Task.Id, Health] = { | ||
| healthByTaskId.filterKeys(p) | ||
| } | ||
|
|
||
| /** Slow */ | ||
| def retain(p: Task.Id => Boolean): this.type = { | ||
| val oldSize = healthByTaskId.size | ||
| healthByTaskId.retain((taskId, _) => p(taskId)) | ||
| if (healthByTaskId.size < oldSize) { | ||
| val retainedInstanceIds = healthByTaskId.keySet.map(_.instanceId) | ||
| taskIdByInstanceId.retain((instanceId, _) => retainedInstanceIds.contains(instanceId)) | ||
| } | ||
| this | ||
| } | ||
|
|
||
| def +=(kv: (Task.Id, Health)): this.type = { | ||
| kv match { | ||
| case (taskId: Task.Id, _: Health) => | ||
| healthByTaskId += kv | ||
| taskIdByInstanceId += taskId.instanceId -> taskId | ||
| } | ||
| this | ||
| } | ||
|
|
||
| def values: Iterable[Health] = healthByTaskId.values | ||
| } |
This file contains 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
157 changes: 157 additions & 0 deletions
157
src/test/scala/mesosphere/marathon/core/health/impl/HealthIndexTest.scala
This file contains 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 |
|---|---|---|
| @@ -0,0 +1,157 @@ | ||
| package mesosphere.marathon | ||
| package core.health.impl | ||
|
|
||
| import mesosphere.UnitTest | ||
| import mesosphere.marathon.core.health.Health | ||
| import mesosphere.marathon.core.health.impl.HealthIndexTest._ | ||
| import mesosphere.marathon.core.instance.Instance | ||
| import mesosphere.marathon.core.task.Task | ||
| import mesosphere.marathon.state.{AbsolutePathId, Timestamp} | ||
|
|
||
| class HealthIndexTest extends UnitTest { | ||
| "HeathIndex" should { | ||
| "get items by Instance ID" in { | ||
| val indexSimple: HealthIndex = HealthSingleTrieMap() | ||
| val indexDual: HealthIndex = HealthDualTrieMap() | ||
|
|
||
| // Add | ||
| val entry1 = taskId1 -> health1 | ||
| indexSimple += entry1 | ||
| indexDual += entry1 | ||
|
|
||
| // Get by Instance ID | ||
| indexSimple.get(instanceId1) should equal(Some(health1)) | ||
| indexDual.get(instanceId1) should equal(Some(health1)) | ||
|
|
||
| indexSimple.get(instanceId2) should equal(None) | ||
| indexDual.get(instanceId2) should equal(None) | ||
| } | ||
|
|
||
| "get items by Task ID" in { | ||
| val indexSimple: HealthIndex = HealthSingleTrieMap() | ||
| val indexDual: HealthIndex = HealthDualTrieMap() | ||
|
|
||
| // Add | ||
| val entry1 = taskId1 -> health1 | ||
| indexSimple += entry1 | ||
| indexDual += entry1 | ||
|
|
||
| // Get by Task ID | ||
| indexSimple.getOrElse(taskId1, health2) should equal(health1) | ||
| indexDual.getOrElse(taskId1, health2) should equal(health1) | ||
|
|
||
| indexSimple.getOrElse(taskId3, health2) should equal(health2) | ||
| indexDual.getOrElse(taskId3, health2) should equal(health2) | ||
| } | ||
|
|
||
| "remove leftover items" in { | ||
| val indexSimple: HealthIndex = HealthSingleTrieMap() | ||
| val indexDual: HealthIndex = HealthDualTrieMap() | ||
|
|
||
| // Add | ||
| val entry1 = taskId1 -> health1 | ||
| indexSimple += entry1 | ||
| indexDual += entry1 | ||
|
|
||
| // Remove | ||
| indexSimple.removeLeftOverHealthIfAny(instanceId2) should equal(None) | ||
| indexSimple.removeLeftOverHealthIfAny(instanceId1) should equal(Some(health1)) | ||
| indexSimple.get(instanceId1) should equal(None) | ||
| indexSimple.removeLeftOverHealthIfAny(instanceId1) should equal(None) | ||
|
|
||
| indexDual.removeLeftOverHealthIfAny(instanceId2) should equal(None) | ||
| indexDual.removeLeftOverHealthIfAny(instanceId1) should equal(Some(health1)) | ||
| indexDual.get(instanceId1) should equal(None) | ||
| indexDual.removeLeftOverHealthIfAny(instanceId1) should equal(None) | ||
| } | ||
|
|
||
| "return items" in { | ||
| val indexSimple: HealthIndex = HealthSingleTrieMap() | ||
| val indexDual: HealthIndex = HealthDualTrieMap() | ||
|
|
||
| // Add | ||
| val entry1 = taskId1 -> health1 | ||
| indexSimple += entry1 | ||
| indexDual += entry1 | ||
|
|
||
| val entry2 = taskId2 -> health2 | ||
| indexSimple += entry2 | ||
| indexDual += entry2 | ||
|
|
||
| // Get values | ||
| val expected = Set(health1, health2) | ||
| indexSimple.values.toSet should equal(expected) | ||
| indexDual.values.toSet should equal(expected) | ||
| } | ||
|
|
||
| "filter items" in { | ||
| val indexSimple: HealthIndex = HealthSingleTrieMap() | ||
| val indexDual: HealthIndex = HealthDualTrieMap() | ||
|
|
||
| // Add | ||
| val entry1 = taskId1 -> health1 | ||
| indexSimple += entry1 | ||
| indexDual += entry1 | ||
|
|
||
| val entry2 = taskId2 -> health2 | ||
| indexSimple += entry2 | ||
| indexDual += entry2 | ||
|
|
||
| val entry3 = taskId3 -> health3 | ||
| indexSimple += entry3 | ||
| indexDual += entry3 | ||
|
|
||
| // Filter by task ID | ||
| indexSimple.filterKeys(taskId => taskId == taskId2) should equal(Map(entry2)) | ||
|
|
||
| indexDual.filterKeys(taskId => taskId == taskId2) should equal(Map(entry2)) | ||
| } | ||
|
|
||
| "retain items" in { | ||
| val indexSimple: HealthIndex = HealthSingleTrieMap() | ||
| val indexDual: HealthIndex = HealthDualTrieMap() | ||
|
|
||
| // Add | ||
| val entry1 = taskId1 -> health1 | ||
| indexSimple += entry1 | ||
| indexDual += entry1 | ||
|
|
||
| val entry2 = taskId2 -> health2 | ||
| indexSimple += entry2 | ||
| indexDual += entry2 | ||
|
|
||
| val entry3 = taskId3 -> health3 | ||
| indexSimple += entry3 | ||
| indexDual += entry3 | ||
|
|
||
| // Retain | ||
| indexSimple.retain(taskId => taskId == taskId2) | ||
| indexSimple.get(instanceId1) should equal(None) | ||
| indexSimple.get(instanceId2) should equal(Some(health2)) | ||
|
|
||
| indexDual.retain(taskId => taskId == taskId2) | ||
| indexDual.get(instanceId1) should equal(None) | ||
| indexDual.get(instanceId2) should equal(Some(health2)) | ||
| } | ||
|
|
||
| } | ||
|
|
||
| } | ||
|
|
||
| object HealthIndexTest { | ||
| val appId: AbsolutePathId = AbsolutePathId("/test") | ||
| val version: Timestamp = Timestamp(1) | ||
| val now: Timestamp = Timestamp(2) | ||
|
|
||
| val instanceId1: Instance.Id = Instance.Id.forRunSpec(appId) | ||
| val taskId1: Task.Id = Task.Id(instanceId1) | ||
| val health1: Health = Health(instanceId1) | ||
|
|
||
| val instanceId2: Instance.Id = Instance.Id.forRunSpec(appId) | ||
| val taskId2: Task.Id = Task.Id(instanceId2) | ||
| val health2: Health = Health(instanceId2) | ||
|
|
||
| val instanceId3: Instance.Id = Instance.Id.forRunSpec(appId) | ||
| val taskId3: Task.Id = Task.Id(instanceId3) | ||
| val health3: Health = Health(instanceId3) | ||
| } |