aurora-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From s...@apache.org
Subject aurora git commit: Ensure Thermos is not overloaded by an unlimited number of lost processes
Date Thu, 22 Jun 2017 05:51:13 GMT
Repository: aurora
Updated Branches:
  refs/heads/master aae39a81e -> 13055df3b


Ensure Thermos is not overloaded by an unlimited number of lost processes

Included changes:

* Thermos may consider launched processes to be LOST. Instead of
  restarting those immediately, the restarts will now be at least
  `min_duration` seconds apart. Restarts will also be capped at the
  TOTAL_RUN_LIMIT of 100 restarts. This ensures neither Thermos nor the
  observer will be overloaded by checkpoints. The handling of the LOST
  state is now consistent with the handling of both FAILED and FINISHED.
* Mark the success_transition and failure_transition as private. They
  are only used within `TaskPlanner` itself.
* Fix documented default of `min_duration` (i.e 5s rather than 15s).

Reviewed at https://reviews.apache.org/r/60306/


Project: http://git-wip-us.apache.org/repos/asf/aurora/repo
Commit: http://git-wip-us.apache.org/repos/asf/aurora/commit/13055df3
Tree: http://git-wip-us.apache.org/repos/asf/aurora/tree/13055df3
Diff: http://git-wip-us.apache.org/repos/asf/aurora/diff/13055df3

Branch: refs/heads/master
Commit: 13055df3b93031fba859f61e5185611bf1ce5f4c
Parents: aae39a8
Author: Stephan Erb <serb@apache.org>
Authored: Thu Jun 22 00:49:40 2017 +0200
Committer: Stephan Erb <serb@apache.org>
Committed: Thu Jun 22 00:49:40 2017 +0200

----------------------------------------------------------------------
 docs/reference/configuration.md                 |  2 +-
 .../python/apache/thermos/common/planner.py     | 40 ++++++++++++++------
 .../apache/thermos/common/test_task_planner.py  | 14 +++++--
 3 files changed, 39 insertions(+), 17 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/aurora/blob/13055df3/docs/reference/configuration.md
----------------------------------------------------------------------
diff --git a/docs/reference/configuration.md b/docs/reference/configuration.md
index 6a9a3ff..bc7e098 100644
--- a/docs/reference/configuration.md
+++ b/docs/reference/configuration.md
@@ -44,7 +44,7 @@ behavior with its optional attributes. Remember, Processes are handled by
Thermo
    **max_failures**   | Integer     | Maximum process failures (Default: 1)
    **daemon**         | Boolean     | When True, this is a daemon process. (Default: False)
    **ephemeral**      | Boolean     | When True, this is an ephemeral process. (Default:
False)
-   **min_duration**   | Integer     | Minimum duration between process restarts in seconds.
(Default: 15)
+   **min_duration**   | Integer     | Minimum duration between process restarts in seconds.
(Default: 5)
    **final**          | Boolean     | When True, this process is a finalizing one that should
run last. (Default: False)
    **logger**         | Logger      | Struct defining the log behavior for the process. (Default:
Empty)
 

http://git-wip-us.apache.org/repos/asf/aurora/blob/13055df3/src/main/python/apache/thermos/common/planner.py
----------------------------------------------------------------------
diff --git a/src/main/python/apache/thermos/common/planner.py b/src/main/python/apache/thermos/common/planner.py
index da5120f..5c23d86 100644
--- a/src/main/python/apache/thermos/common/planner.py
+++ b/src/main/python/apache/thermos/common/planner.py
@@ -199,6 +199,7 @@ class TaskPlanner(object):
     self._last_terminal = {}  # process => timestamp of last terminal state
     self._failures = defaultdict(int)
     self._successes = defaultdict(int)
+    self._losses = defaultdict(int)
     self._attributes = {}
     self._ephemerals = set(process.name().get() for process in task.processes()
         if (self._filter is None or self._filter(process)) and process.ephemeral().get())
@@ -216,6 +217,10 @@ class TaskPlanner(object):
       return 0
     return self._attributes[process].min_duration - (now - self._last_terminal[process])
 
+  def _record_termination_time(self, process, timestamp=None):
+    timestamp = timestamp if timestamp is not None else self._clock.time()
+    self._last_terminal[process] = timestamp
+
   def is_ready(self, process, timestamp=None):
     return self.get_wait(process, timestamp) <= 0
 
@@ -253,15 +258,15 @@ class TaskPlanner(object):
     """Increment the failure count of a process, and reset it to runnable if maximum number
of
     failures has not been reached, or mark it as failed otherwise (ephemeral processes do
not
     count towards the success of a task, and are hence marked finished instead)"""
-    timestamp = timestamp if timestamp is not None else self._clock.time()
-    self._last_terminal[process] = timestamp
+    self._record_termination_time(process, timestamp)
     self._failures[process] += 1
-    self.failure_transition(process)
+    self._failure_transition(process)
 
   def has_reached_run_limit(self, process):
-    return (self._successes[process] + self._failures[process]) >= self.TOTAL_RUN_LIMIT
+    runs = self._successes[process] + self._failures[process] + self._losses[process]
+    return runs >= self.TOTAL_RUN_LIMIT
 
-  def failure_transition(self, process):
+  def _failure_transition(self, process):
     if self.has_reached_run_limit(process):
       self._planner.set_failed(process)
       return
@@ -276,12 +281,11 @@ class TaskPlanner(object):
 
   def add_success(self, process, timestamp=None):
     """Reset a process to runnable if it is a daemon, or mark it as finished otherwise."""
-    timestamp = timestamp if timestamp is not None else self._clock.time()
-    self._last_terminal[process] = timestamp
+    self._record_termination_time(process, timestamp)
     self._successes[process] += 1
-    self.success_transition(process)
+    self._success_transition(process)
 
-  def success_transition(self, process):
+  def _success_transition(self, process):
     if self.has_reached_run_limit(process):
       self._planner.set_failed(process)
       return
@@ -295,9 +299,21 @@ class TaskPlanner(object):
     """Force a process to be in failed state.  E.g. kill -9 and you want it pinned failed."""
     self._planner.set_failed(process)
 
-  def lost(self, process):
-    """Mark a process as lost.  This sets its runnable state back to the previous runnable
-       state and does not increment its failure count."""
+  def lost(self, process, timestamp=None):
+    """Mark a process as lost. This sets its runnable state back to the previous runnable
+       state and does not increment the failure count of the process.
+
+       In order to prevent Thermos from overloading itself, even restarts of lost processes
will be
+       throttled by the `min_duration` uptime and capped at TOTAL_RUN_LIMIT."""
+    self._record_termination_time(process, timestamp)
+    self._losses[process] += 1
+    self._lost_transition(process)
+
+  def _lost_transition(self, process):
+    if self.has_reached_run_limit(process):
+      self._planner.set_failed(process)
+      return
+
     self._planner.reset(process)
 
   def is_complete(self):

http://git-wip-us.apache.org/repos/asf/aurora/blob/13055df3/src/test/python/apache/thermos/common/test_task_planner.py
----------------------------------------------------------------------
diff --git a/src/test/python/apache/thermos/common/test_task_planner.py b/src/test/python/apache/thermos/common/test_task_planner.py
index 132c1ec..b56ef84 100644
--- a/src/test/python/apache/thermos/common/test_task_planner.py
+++ b/src/test/python/apache/thermos/common/test_task_planner.py
@@ -287,8 +287,8 @@ def test_task_lost():
   p.add_failure('d1', timestamp=1)
   assert approx_equal(p.min_wait(timestamp=1), 1)
   p.set_running('d1')
-  p.lost('d1')
-  assert approx_equal(p.min_wait(timestamp=1), 1)
+  p.lost('d1', timestamp=2)
+  assert approx_equal(p.min_wait(timestamp=2), 1)
   p.set_running('d1')
   p.add_failure('d1', timestamp=3)
   assert p.min_wait(timestamp=3) == TaskPlanner.INFINITY
@@ -316,7 +316,7 @@ def test_task_filters():
 
 def test_task_max_runs():
   class CappedTaskPlanner(TaskPlanner):
-    TOTAL_RUN_LIMIT = 2
+    TOTAL_RUN_LIMIT = 3
   dt = p1(daemon=True, max_failures=0)
 
   p = CappedTaskPlanner(empty_task(processes=[dt(name='d1', max_failures=100, daemon=False)]))
@@ -324,7 +324,10 @@ def test_task_max_runs():
   p.add_failure('d1', timestamp=1)
   assert 'd1' in p.runnable
   p.set_running('d1')
-  p.add_failure('d1', timestamp=2)
+  p.lost('d1', timestamp=2)
+  assert 'd1' in p.runnable
+  p.set_running('d1')
+  p.add_failure('d1', timestamp=3)
   assert 'd1' not in p.runnable
 
   p = CappedTaskPlanner(empty_task(processes=[dt(name='d1', max_failures=100)]))
@@ -333,4 +336,7 @@ def test_task_max_runs():
   assert 'd1' in p.runnable
   p.set_running('d1')
   p.add_success('d1', timestamp=2)
+  assert 'd1' in p.runnable
+  p.set_running('d1')
+  p.lost('d1', timestamp=3)
   assert 'd1' not in p.runnable


Mime
View raw message