You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@aurora.apache.org by wi...@apache.org on 2014/10/13 23:31:34 UTC
git commit: Add exit status contract between thermos runner and
aurora executor.
Repository: incubator-aurora
Updated Branches:
refs/heads/master add3d4d65 -> 7c50a35d3
Add exit status contract between thermos runner and aurora executor.
Adds an exit status contract between the ThermosTaskRunner and the
thermos_runner.pex so that we can discriminate between losses / failures.
Testing Done:
build-support/jenkins/build.sh
Bugs closed: AURORA-175
Reviewed at https://reviews.apache.org/r/26333/
Project: http://git-wip-us.apache.org/repos/asf/incubator-aurora/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-aurora/commit/7c50a35d
Tree: http://git-wip-us.apache.org/repos/asf/incubator-aurora/tree/7c50a35d
Diff: http://git-wip-us.apache.org/repos/asf/incubator-aurora/diff/7c50a35d
Branch: refs/heads/master
Commit: 7c50a35d3143f64548239b99e8b348fad12fb2ff
Parents: add3d4d
Author: Brian Wickman <wi...@apache.org>
Authored: Mon Oct 13 14:31:28 2014 -0700
Committer: Brian Wickman <wi...@apache.org>
Committed: Mon Oct 13 14:31:28 2014 -0700
----------------------------------------------------------------------
src/main/python/apache/aurora/executor/BUILD | 7 ++
.../apache/aurora/executor/thermos_runner.py | 46 ++++++-
.../apache/aurora/executor/thermos_statuses.py | 5 +
.../aurora/executor/thermos_task_runner.py | 73 +++++++++--
.../aurora/executor/test_thermos_executor.py | 14 +--
.../aurora/executor/test_thermos_task_runner.py | 122 +++++++++++++++++--
6 files changed, 230 insertions(+), 37 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-aurora/blob/7c50a35d/src/main/python/apache/aurora/executor/BUILD
----------------------------------------------------------------------
diff --git a/src/main/python/apache/aurora/executor/BUILD b/src/main/python/apache/aurora/executor/BUILD
index 5f0ef7c..ca4193d 100644
--- a/src/main/python/apache/aurora/executor/BUILD
+++ b/src/main/python/apache/aurora/executor/BUILD
@@ -15,6 +15,11 @@
import os
python_library(
+ name = 'thermos_statuses',
+ sources = ['thermos_statuses.py'],
+)
+
+python_library(
name = 'thermos_task_runner',
sources = ['thermos_task_runner.py'],
dependencies = [
@@ -30,6 +35,7 @@ python_library(
'src/main/python/apache/aurora/executor/common:status_checker',
'src/main/python/apache/aurora/executor/common:task_info',
'src/main/python/apache/aurora/executor/common:task_runner',
+ ':thermos_statuses',
]
)
@@ -102,6 +108,7 @@ python_library(
'src/main/python/apache/thermos/common:planner',
'src/main/python/apache/thermos/config:schema',
'src/main/python/apache/thermos/core',
+ ':thermos_statuses',
],
)
http://git-wip-us.apache.org/repos/asf/incubator-aurora/blob/7c50a35d/src/main/python/apache/aurora/executor/thermos_runner.py
----------------------------------------------------------------------
diff --git a/src/main/python/apache/aurora/executor/thermos_runner.py b/src/main/python/apache/aurora/executor/thermos_runner.py
index 396753e..e10f438 100644
--- a/src/main/python/apache/aurora/executor/thermos_runner.py
+++ b/src/main/python/apache/aurora/executor/thermos_runner.py
@@ -13,7 +13,9 @@
#
import functools
+import getpass
import os
+import pwd
import signal
import sys
import traceback
@@ -23,8 +25,17 @@ from twitter.common import app, log
from apache.thermos.common.options import add_port_to
from apache.thermos.common.planner import TaskPlanner
from apache.thermos.config.loader import ThermosConfigLoader
+from apache.thermos.core.process import Process
from apache.thermos.core.runner import TaskRunner
+from .thermos_statuses import (
+ INTERNAL_ERROR,
+ INVALID_TASK,
+ TERMINAL_TASK,
+ UNKNOWN_ERROR,
+ UNKNOWN_USER
+)
+
app.add_option(
"--thermos_json",
dest="thermos_json",
@@ -124,7 +135,20 @@ def proxy_main(args, opts):
missing_ports = set(thermos_task.ports()) - set(prebound_ports)
if missing_ports:
- app.error('ERROR! Unbound ports: %s' % ' '.join(port for port in missing_ports))
+ log.error('ERROR! Unbound ports: %s' % ' '.join(port for port in missing_ports))
+ sys.exit(INTERNAL_ERROR)
+
+ if opts.setuid:
+ user = opts.setuid
+ else:
+ user = getpass.getuser()
+
+ # if we cannot get the uid, this is an unknown user and we should fail
+ try:
+ pwd.getpwnam(user).pw_uid
+ except KeyError:
+ log.error('Unknown user: %s' % user)
+ sys.exit(UNKNOWN_USER)
task_runner = TaskRunner(
thermos_task.task,
@@ -143,10 +167,22 @@ def proxy_main(args, opts):
try:
task_runner.run()
except TaskRunner.InternalError as err:
- app.error('Internal error: %s' % err)
+ log.error('Internal error: %s' % err)
+ sys.exit(INTERNAL_ERROR)
except TaskRunner.InvalidTask as err:
- app.error(str(err))
- except TaskRunner.StateError:
- app.error('Task appears to already be in a terminal state.')
+ log.error('Invalid task: %s' % err)
+ sys.exit(INVALID_TASK)
+ except TaskRunner.StateError as err:
+ log.error('Checkpoint error: %s' % err)
+ sys.exit(TERMINAL_TASK)
+ except Process.UnknownUserError as err:
+ log.error('User ceased to exist: %s' % err)
+ sys.exit(UNKNOWN_USER)
except KeyboardInterrupt:
+ log.info('Caught ^C, tearing down runner.')
runner_teardown(task_runner)
+ except Exception as e:
+ log.error('Unknown exception: %s' % e)
+ for line in traceback.format_exc().splitlines():
+ log.error(line)
+ sys.exit(UNKNOWN_ERROR)
http://git-wip-us.apache.org/repos/asf/incubator-aurora/blob/7c50a35d/src/main/python/apache/aurora/executor/thermos_statuses.py
----------------------------------------------------------------------
diff --git a/src/main/python/apache/aurora/executor/thermos_statuses.py b/src/main/python/apache/aurora/executor/thermos_statuses.py
new file mode 100644
index 0000000..634185a
--- /dev/null
+++ b/src/main/python/apache/aurora/executor/thermos_statuses.py
@@ -0,0 +1,5 @@
+INTERNAL_ERROR = 101
+INVALID_TASK = 102
+UNKNOWN_ERROR = 103
+UNKNOWN_USER = 104
+TERMINAL_TASK = 105
http://git-wip-us.apache.org/repos/asf/incubator-aurora/blob/7c50a35d/src/main/python/apache/aurora/executor/thermos_task_runner.py
----------------------------------------------------------------------
diff --git a/src/main/python/apache/aurora/executor/thermos_task_runner.py b/src/main/python/apache/aurora/executor/thermos_task_runner.py
index 7b346e2..4e68844 100644
--- a/src/main/python/apache/aurora/executor/thermos_task_runner.py
+++ b/src/main/python/apache/aurora/executor/thermos_task_runner.py
@@ -16,6 +16,7 @@ import errno
import getpass
import os
import signal
+import struct
import subprocess
import threading
import time
@@ -35,6 +36,13 @@ from apache.thermos.monitoring.monitor import TaskMonitor
from .common.status_checker import StatusResult
from .common.task_info import mesos_task_instance_from_assigned_task, resolve_ports
from .common.task_runner import TaskError, TaskRunner, TaskRunnerProvider
+from .thermos_statuses import (
+ INTERNAL_ERROR,
+ INVALID_TASK,
+ TERMINAL_TASK,
+ UNKNOWN_ERROR,
+ UNKNOWN_USER
+)
from gen.apache.thermos.ttypes import TaskState
@@ -77,7 +85,7 @@ class ThermosTaskRunner(TaskRunner):
self._runner_pex = runner_pex
self._task_id = task_id
self._task = task
- self._popen = None
+ self._popen, self._popen_signal, self._popen_rc = None, None, None
self._monitor = None
self._status = None
self._ports = portmap
@@ -126,6 +134,16 @@ class ThermosTaskRunner(TaskRunner):
def task_state(self):
return self._monitor.task_state() if self._monitor else None
+ @classmethod
+ def _decode_status(cls, status):
+ # Per os.waitpid documentation, status is:
+ # a 16-bit number, whose low byte is the signal number that killed the
+ # process, and whose high byte is the exit status (if the signal
+ # number is zero); the high bit of the low byte is set if a core file
+ # was produced.
+ exit_signal, exit_status = struct.unpack('bb', struct.pack('H', status))
+ return exit_signal & 0x7F, exit_status # strip the signal high bit
+
@property
def is_alive(self):
"""
@@ -133,6 +151,7 @@ class ThermosTaskRunner(TaskRunner):
"""
if not self._popen:
return False
+
if self._dead.is_set():
return False
@@ -142,11 +161,13 @@ class ThermosTaskRunner(TaskRunner):
# should not mix a Thermos task runner in the same process as this
# thread.
try:
- pid, _ = os.waitpid(self._popen.pid, os.WNOHANG)
+ pid, status = os.waitpid(self._popen.pid, os.WNOHANG)
if pid == 0:
return True
else:
- log.info('Detected runner termination: pid=%s' % pid)
+ self._popen_signal, self._popen_rc = self._decode_status(status)
+ log.info('Detected runner termination: pid=%s, signal=%s, rc=%s' % (
+ pid, self._popen_signal, self._popen_rc))
except OSError as e:
log.error('is_alive got OSError: %s' % e)
if e.errno != errno.ECHILD:
@@ -158,10 +179,26 @@ class ThermosTaskRunner(TaskRunner):
def compute_status(self):
if self.is_alive:
return None
- exit_state = self.EXIT_STATE_MAP.get(self.task_state())
- if exit_state is None:
- log.error('Received unexpected exit state from TaskMonitor.')
- return exit_state
+ if self._popen_signal != 0:
+ return StatusResult('Task killed by signal %s.' % self._popen_signal, mesos_pb2.TASK_KILLED)
+ if self._popen_rc == 0 or self._popen_rc == TERMINAL_TASK:
+ exit_state = self.EXIT_STATE_MAP.get(self.task_state())
+ if exit_state is None:
+ log.error('Received unexpected exit state from TaskMonitor.')
+ return StatusResult('Task checkpoint could not be read.', mesos_pb2.TASK_LOST)
+ else:
+ return exit_state
+ elif self._popen_rc == UNKNOWN_USER:
+ return StatusResult('Task started with unknown user.', mesos_pb2.TASK_FAILED)
+ elif self._popen_rc == INTERNAL_ERROR:
+ return StatusResult('Thermos failed with internal error.', mesos_pb2.TASK_LOST)
+ elif self._popen_rc == INVALID_TASK:
+ return StatusResult('Thermos received an invalid task.', mesos_pb2.TASK_FAILED)
+ elif self._popen_rc == UNKNOWN_ERROR:
+ return StatusResult('Thermos failed with an unknown error.', mesos_pb2.TASK_LOST)
+ else:
+ return StatusResult('Thermos exited for unknown reason (exit status: %s)' % self._popen_rc,
+ mesos_pb2.TASK_LOST)
def terminate_runner(self, as_loss=False):
"""
@@ -242,16 +279,30 @@ class ThermosTaskRunner(TaskRunner):
self.forked.set()
+ self.wait_start(timeout=timeout)
+
+ def wait_start(self, timeout=MAX_WAIT):
log.debug('Waiting for task to start.')
def is_started():
return self._monitor and (self._monitor.active or self._monitor.finished)
waited = Amount(0, Time.SECONDS)
- while not is_started() and waited < timeout:
- log.debug(' - sleeping...')
- self._clock.sleep(self.POLL_INTERVAL.as_(Time.SECONDS))
- waited += self.POLL_INTERVAL
+
+ while waited < timeout:
+ if not is_started():
+ log.debug(' - sleeping...')
+ self._clock.sleep(self.POLL_INTERVAL.as_(Time.SECONDS))
+ waited += self.POLL_INTERVAL
+ else:
+ break
+
+ if not self.is_alive:
+ if self._popen_rc != 0:
+ raise TaskError('Task failed: %s' % self._popen_reason())
+ else:
+ log.info('Task runner exited: %s' % self._popen_reason())
+ break
if not is_started():
log.error('Task did not start with in deadline, forcing loss.')
http://git-wip-us.apache.org/repos/asf/incubator-aurora/blob/7c50a35d/src/test/python/apache/aurora/executor/test_thermos_executor.py
----------------------------------------------------------------------
diff --git a/src/test/python/apache/aurora/executor/test_thermos_executor.py b/src/test/python/apache/aurora/executor/test_thermos_executor.py
index f6ca4df..65e8cce 100644
--- a/src/test/python/apache/aurora/executor/test_thermos_executor.py
+++ b/src/test/python/apache/aurora/executor/test_thermos_executor.py
@@ -339,7 +339,7 @@ class TestThermosExecutor(object):
updates = proxy_driver.method_calls['sendStatusUpdate']
assert len(updates) == 3
- assert updates[-1][0][0].state == mesos_pb2.TASK_LOST
+ assert updates[-1][0][0].state == mesos_pb2.TASK_KILLED
def test_task_killed(self):
proxy_driver = ProxyDriver()
@@ -379,18 +379,6 @@ class TestThermosExecutor(object):
assert len(updates) == 3
assert updates[-1][0][0].state == mesos_pb2.TASK_KILLED
- def test_task_lost(self):
- proxy_driver = ProxyDriver()
-
- with temporary_dir() as checkpoint_root:
- runner, executor = make_executor(proxy_driver, checkpoint_root, SLEEP60_MTI)
- runner.lose(force=True)
- executor.terminated.wait()
-
- updates = proxy_driver.method_calls['sendStatusUpdate']
- assert len(updates) == 3
- assert updates[-1][0][0].state == mesos_pb2.TASK_LOST
-
def test_task_health_failed(self):
proxy_driver = ProxyDriver()
with SignalServer(UnhealthyHandler) as port:
http://git-wip-us.apache.org/repos/asf/incubator-aurora/blob/7c50a35d/src/test/python/apache/aurora/executor/test_thermos_task_runner.py
----------------------------------------------------------------------
diff --git a/src/test/python/apache/aurora/executor/test_thermos_task_runner.py b/src/test/python/apache/aurora/executor/test_thermos_task_runner.py
index 8e4042b..4fc9d4b 100644
--- a/src/test/python/apache/aurora/executor/test_thermos_task_runner.py
+++ b/src/test/python/apache/aurora/executor/test_thermos_task_runner.py
@@ -15,12 +15,13 @@
import contextlib
import getpass
import os
+import signal
import subprocess
import sys
import tempfile
import time
-from mesos.interface.mesos_pb2 import TaskState
+from mesos.interface import mesos_pb2
from twitter.common import log
from twitter.common.contextutil import temporary_dir
from twitter.common.dirutil import safe_rmtree
@@ -29,8 +30,17 @@ from twitter.common.quantity import Amount, Time
from apache.aurora.config.schema.base import MB, MesosTaskInstance, Process, Resources, Task
from apache.aurora.executor.common.sandbox import DirectorySandbox
+from apache.aurora.executor.thermos_statuses import (
+ INTERNAL_ERROR,
+ INVALID_TASK,
+ TERMINAL_TASK,
+ UNKNOWN_ERROR,
+ UNKNOWN_USER
+)
from apache.aurora.executor.thermos_task_runner import ThermosTaskRunner
+from gen.apache.thermos.ttypes import TaskState
+
TASK = MesosTaskInstance(
instance=0,
role=getpass.getuser(),
@@ -91,6 +101,29 @@ class TestThermosTaskRunnerIntegration(object):
__sleep=sleep,
__exit_code=exit_code)
+ def yield_command(self, fake_runner_command, exit_state=TaskState.SUCCESS):
+ class SkipWaitStartThermosTaskRunner(ThermosTaskRunner):
+ def wait_start(_, timeout=None):
+ pass
+
+ def _cmdline(_):
+ return ['bash', '-c', fake_runner_command]
+
+ def task_state(_):
+ return exit_state
+
+ return self.yield_runner(SkipWaitStartThermosTaskRunner, command='ignore')
+
+ @contextlib.contextmanager
+ def exit_with_status(self, exit_code, exit_state=TaskState.SUCCESS):
+ with self.yield_command('exit %d' % exit_code, exit_state=exit_state) as task_runner:
+ task_runner.start()
+
+ while task_runner.is_alive:
+ task_runner._dead.wait(timeout=0.1)
+
+ yield task_runner
+
def run_to_completion(self, runner, max_wait=Amount(10, Time.SECONDS)):
poll_interval = Amount(100, Time.MILLISECONDS)
total_time = Amount(0, Time.SECONDS)
@@ -106,13 +139,13 @@ class TestThermosTaskRunnerIntegration(object):
self.run_to_completion(task_runner)
assert task_runner.status is not None
- assert TaskState.Name(task_runner.status.status) == 'TASK_FINISHED'
+ assert task_runner.status.status == mesos_pb2.TASK_FINISHED
# no-op
task_runner.stop()
assert task_runner.status is not None
- assert TaskState.Name(task_runner.status.status) == 'TASK_FINISHED'
+ assert task_runner.status.status == mesos_pb2.TASK_FINISHED
def test_integration_failed(self):
with self.yield_sleepy(ThermosTaskRunner, sleep=0, exit_code=1) as task_runner:
@@ -122,13 +155,13 @@ class TestThermosTaskRunnerIntegration(object):
self.run_to_completion(task_runner)
assert task_runner.status is not None
- assert TaskState.Name(task_runner.status.status) == 'TASK_FAILED'
+ assert task_runner.status.status == mesos_pb2.TASK_FAILED
# no-op
task_runner.stop()
assert task_runner.status is not None
- assert TaskState.Name(task_runner.status.status) == 'TASK_FAILED'
+ assert task_runner.status.status == mesos_pb2.TASK_FAILED
def test_integration_stop(self):
with self.yield_sleepy(ThermosTaskRunner, sleep=1000, exit_code=0) as task_runner:
@@ -140,7 +173,7 @@ class TestThermosTaskRunnerIntegration(object):
task_runner.stop()
assert task_runner.status is not None
- assert TaskState.Name(task_runner.status.status) == 'TASK_KILLED'
+ assert task_runner.status.status == mesos_pb2.TASK_KILLED
def test_integration_lose(self):
with self.yield_sleepy(ThermosTaskRunner, sleep=1000, exit_code=0) as task_runner:
@@ -153,7 +186,7 @@ class TestThermosTaskRunnerIntegration(object):
task_runner.stop()
assert task_runner.status is not None
- assert TaskState.Name(task_runner.status.status) == 'TASK_LOST'
+ assert task_runner.status.status == mesos_pb2.TASK_LOST
def test_integration_quitquitquit(self):
ignorant_script = ';'.join([
@@ -173,4 +206,77 @@ class TestThermosTaskRunnerIntegration(object):
task_runner.forked.wait()
task_runner.stop(timeout=Amount(5, Time.SECONDS))
assert task_runner.status is not None
- assert TaskState.Name(task_runner.status.status) == 'TASK_KILLED'
+ assert task_runner.status.status == mesos_pb2.TASK_KILLED
+
+ def test_thermos_normal_exit_status(self):
+ with self.exit_with_status(0, TaskState.SUCCESS) as task_runner:
+ assert task_runner._popen_signal == 0
+ assert task_runner._popen_rc == 0
+ status = task_runner.compute_status()
+ assert status.reason == 'Task finished.'
+ assert status.status is mesos_pb2.TASK_FINISHED
+
+ with self.exit_with_status(0, None) as task_runner:
+ assert task_runner._popen_signal == 0
+ assert task_runner._popen_rc == 0
+ status = task_runner.compute_status()
+ assert status.status is mesos_pb2.TASK_LOST
+
+ def test_thermos_abnormal_exit_statuses(self):
+ with self.exit_with_status(UNKNOWN_USER) as task_runner:
+ assert task_runner._popen_signal == 0
+ assert task_runner._popen_rc == UNKNOWN_USER
+ status = task_runner.compute_status()
+ assert 'unknown user' in status.reason
+ assert status.status is mesos_pb2.TASK_FAILED
+
+ with self.exit_with_status(INTERNAL_ERROR) as task_runner:
+ assert task_runner._popen_signal == 0
+ assert task_runner._popen_rc == INTERNAL_ERROR
+ status = task_runner.compute_status()
+ assert 'internal error' in status.reason
+ assert status.status is mesos_pb2.TASK_LOST
+
+ with self.exit_with_status(INVALID_TASK) as task_runner:
+ assert task_runner._popen_signal == 0
+ assert task_runner._popen_rc == INVALID_TASK
+ status = task_runner.compute_status()
+ assert 'invalid task' in status.reason
+ assert status.status is mesos_pb2.TASK_FAILED
+
+ with self.exit_with_status(TERMINAL_TASK, exit_state=TaskState.FAILED) as task_runner:
+ assert task_runner._popen_signal == 0
+ assert task_runner._popen_rc == TERMINAL_TASK
+ status = task_runner.compute_status()
+ assert 'Task failed' in status.reason
+ assert status.status is mesos_pb2.TASK_FAILED
+
+ with self.exit_with_status(UNKNOWN_ERROR) as task_runner:
+ assert task_runner._popen_signal == 0
+ assert task_runner._popen_rc == UNKNOWN_ERROR
+ status = task_runner.compute_status()
+ assert 'unknown error' in status.reason
+ assert status.status is mesos_pb2.TASK_LOST
+
+ with self.exit_with_status(13) as task_runner:
+ assert task_runner._popen_signal == 0
+ assert task_runner._popen_rc == 13
+ status = task_runner.compute_status()
+ assert 'unknown reason' in status.reason
+ assert 'exit status: 13' in status.reason
+ assert status.status is mesos_pb2.TASK_LOST
+
+ def test_thermos_runner_killed(self):
+ with self.yield_command('sleep 1000') as task_runner:
+ task_runner.start()
+
+ os.kill(task_runner._popen.pid, signal.SIGKILL)
+
+ while task_runner.is_alive:
+ task_runner._dead.wait(timeout=0.1)
+
+ assert task_runner._popen_signal == 9
+ assert task_runner._popen_rc == 0 # ??
+ status = task_runner.compute_status()
+ assert 'killed by signal 9' in status.reason
+ assert status.status is mesos_pb2.TASK_KILLED