blob: 8019c4072842c4048caa8af8faa701c59707b8ae [file] [log] [blame]
import json
from mysos.common.decorators import logged
from .task_runner import TaskError
from mesos.interface import Executor
import mesos.interface.mesos_pb2 as mesos_pb2
from twitter.common import log
from twitter.common.concurrent import defer
class MysosExecutor(Executor):
MysosExecutor is a fine-grained executor, i.e., one executor executes a single task.
def __init__(self, runner_provider, sandbox):
:param runner_provider: An implementation of TaskRunnerProvider.
:param sandbox: The path to the sandbox where all files the executor reads/writes are located.
self._runner_provider = runner_provider
self._runner = None # A singleton task runner created by launchTask().
self._driver = None # Assigned in registered().
self._killed = False # True if the executor's singleton task is killed by the scheduler.
self._sandbox = sandbox
# --- Mesos methods. ---
def registered(self, driver, executorInfo, frameworkInfo, slaveInfo):'Registered with slave: %s' % slaveInfo)
self._driver = driver # Cache the driver to kill later.
def reregistered(self, driver, slaveInfo):'Reregistered with slave: %s' % slaveInfo)
def disconnected(self, driver):"ExecutorDriver disconnected from Mesos slave")
def launchTask(self, driver, task):
if self._runner:
log.error("Executor allows only one task")
update = mesos_pb2.TaskStatus()
update.state = mesos_pb2.TASK_FAILED
# Create the runner here in the driver thread so subsequent task launches are rejected.
self._runner = self._runner_provider.from_task(task, self._sandbox)
except (TaskError, ValueError) as e:
# TODO(jyx): These should really all be 'ValueError's from all providers because they are
# simply factory methods.
log.error("Failed to create TaskRunner: %s" % e.message)
self._send_update(task.task_id.value, mesos_pb2.TASK_FAILED, e.message)
# Run the task in a separate daemon thread.
defer(lambda: self._run_task(task))
def _run_task(self, task):
assert self._runner, "_runner should be created before this method is called"
self._runner.start()"Task runner for task %s started" % task.task_id)
self._send_update(task.task_id.value, mesos_pb2.TASK_RUNNING)
except TaskError as e:
log.error("Task runner for task %s failed to start: %s" % (task.task_id, str(e)))
# Send TASK_FAILED if the task failed to start.
self._send_update(task.task_id.value, mesos_pb2.TASK_FAILED)
except Exception as e:
log.error("Error occurred while executing the task: %s" % e)
# Send TASK_LOST for unknown errors.
self._send_update(task.task_id.value, mesos_pb2.TASK_LOST)
# Wait for the task's return code (when it terminates).
returncode = self._runner.join()
# Regardless of the return code, if '_runner' terminates, it failed!
log.error("Task process terminated with return code %s" % returncode)
except TaskError as e:
log.error("Task terminated: %s" % e)
if self._killed:
self._send_update(task.task_id.value, mesos_pb2.TASK_KILLED)
self._send_update(task.task_id.value, mesos_pb2.TASK_FAILED)
def frameworkMessage(self, driver, message):
if not self._runner:'Ignoring framework message because no task is running yet')
defer(lambda: self._framework_message(message))
def _framework_message(self, message):
master_epoch = message # The log position request is for electing the master of this 'epoch'.
position = self._runner.get_log_position()'Obtained log position %s for epoch %s' % (position, master_epoch))
assert self._driver
# TODO(jyx): Define the message in ProtoBuf or Thrift.
'epoch': master_epoch, # Send the epoch back without parsing it.
'position': position
except TaskError as e:
# Log the error and do not reply to the framework.
log.error("Committing suicide due to failure to process framework message: %s" % e)
def killTask(self, driver, taskId):
# Killing the task also kills the executor because there is one task per executor."Asked to kill task %s" % taskId)
self._killed = True
def _kill(self):
if self._runner:
self._runner.stop() # It could be already stopped. If so, self._runner.stop() is a no-op.
assert self._driver
# TODO(jyx): Fix
def shutdown(self, driver):"Asked to shut down")
self._killed = True
def error(self, driver, message):
log.error("Shutting down due to error: %s" % message)
self._killed = True
def _send_update(self, task_id, state, message=None):
update = mesos_pb2.TaskStatus()
if not isinstance(state, int):
raise TypeError('Invalid state type %s, should be int.' % type(state))
if state not in [
raise ValueError('Invalid state: %s' % state)
update.state = state
update.task_id.value = task_id
if message:
update.message = str(message)'Updating %s => %s. Reason: %s' % (task_id, mesos_pb2.TaskState.Name(state), message))