Process.start() method does not call Process._notify()

[revz79]

The start() method of the Process class does not call the _notify() method as other methods (i.e. stop()) do. This prevents the executor to connect back to the agent and start processing events, because the method Process._run() never assigns the field _master and therefore never creates a Connection instance. As a result the launched task remains in STAGING state until it fails a couple of minutes after.

A workaround to this issue is to extend the MesosExecutorDriver override the start() method adding a call to _notify().

Is this the intended behavior?

[ariesdevil]

@revz79 Could you please provide any reproduce codes and which version you used? By run examples/scheduler.py and I see everything is OK even if I kill master manually when scheduler running.

[revz79]

Hello,

Here is the scheduler, i've got it from the examples, removing the 'addict' module dependency:

import logging
import os
import signal
import socket
import sys
import threading
import typing
import time
import uuid

import pymesos

logging.basicConfig(level=logging.DEBUG)

TASK_CPU = 4
TASK_MEM = 32
EXECUTOR_CPUS = 0.1
EXECUTOR_MEM = 32

class TestScheduler(pymesos.Scheduler):

    def __init__(self, executor: typing.Dict):
        self._executor: typing.Dict = executor

    def registered(self, driver, frameworkId, masterInfo):
        logging.debug(
            f'Registered (framework: {frameworkId}, master info: {masterInfo}).')

    def resourceOffers(self, driver, offers):

        filters = {'refuse_seconds': 5}

        for offer in offers:
            cpus = self.getResource(offer['resources'], 'cpus')
            mem = self.getResource(offer['resources'], 'mem')
            if cpus < TASK_CPU or mem < TASK_MEM:
                continue

            task_id = str(uuid.uuid4())
            task = {
                'task_id': {
                    'value': task_id
                },
                'agent_id': {
                    'value': offer['agent_id']['value']
                },
                'name': f'task {task_id}',
                'executor': self._executor,
                'data': pymesos.encode_data(f'Hello from task {task_id}!'.encode())
            }

            task['resources'] = [
                {
                    'name': 'cpus',
                    'type': 'SCALAR',
                    'scalar': {'value': TASK_CPU}
                },
                {
                    'name': 'mem',
                    'type': 'SCALAR',
                    'scalar': {'value': TASK_MEM}
                }
            ]

            driver.launchTasks(offer['id'], [task], filters)

    def getResource(self, res, name):
        resource = list(filter(lambda x: x['name'] == name, res))[0]
        return resource.get('scalar', {'value': 0.0}).get('value', 0.0)

    def statusUpdate(self, driver, update):
        logging.debug(
            f'Status update TID {update["task_id"]["value"]} {update["state"]}')

def main(master):

    executor = {
        'executor_id': {
            'value': str(uuid.uuid4())
        },
        'name': 'TestExecutor',
        'command': {
            'value': f'python3 /opt/python/executor.py'
        },
        'resources': [
            {
                'name': 'mem',
                'type': 'SCALAR',
                'scalar': {
                    'value': EXECUTOR_MEM
                }
            },
            {
                'name': 'cpus',
                'type': 'SCALAR',
                'scalar': {
                    'value': EXECUTOR_CPUS
                }
            }
        ]
    }

    driver = pymesos.MesosSchedulerDriver(
        sched=TestScheduler(executor=executor),
        framework={
            'user': 'test_user',
            'name': 'test_name'
        },
        master_uri=master
    )

    def signal_handler(signal, frame):
        driver.stop()

    def run_driver_thread():
        driver.run()

    driver_thread = threading.Thread(target=run_driver_thread, args=())
    driver_thread.start()

    logging.info('Scheduler running, Ctrl+C to quit.')
    signal.signal(signal.SIGINT, signal_handler)

    while driver_thread.is_alive():
        time.sleep(1)

if __name__ == '__main__':
    main(master='127.0.0.1:5050')

Here below is the executor code which doesn't launch the task:

import logging
import sys
import time
import threading
import typing

import pymesos

LOG = logging.getLogger(__name__)
LOG.setLevel(logging.DEBUG)
LOG.addHandler(logging.FileHandler('task.log'))

class TestExecutor(pymesos.Executor):
    def registered(self, driver, executorInfo, frameworkInfo, slaveInfo):
        LOG.debug(
            f'Executor regitered (driver: {driver}, executorInfo: {executorInfo}, frameworkInfo: {frameworkInfo}, slaveInfo:  {slaveInfo}).')

    def launchTask(self, driver, task):
        def run_task(task: typing.Dict):
            update = {
                'task_id': {
                    'value': task['task_id']['value'],
                },
                'state': 'TASK_RUNNING',
                'timestmp': time.time(),
            }

            driver.sendStatusUpdate(update)

            LOG.info(pymesos.decode_data(task['data']))
            time.sleep(30)

            update = {
                'task_id': {
                    'value': task['task_id']['value'],
                },
                'state': 'TASK_FINISHED',
                'timestmp': time.time(),
            }

            driver.sendStatusUpdate(update)

        thread = threading.Thread(target=run_task, args=(task,))
        thread.start()

if __name__ == '__main__':

    driver = pymesos.MesosExecutorDriver(TestExecutor())
    driver.run()

I added some logging and i saw that the the the script is launched by the agent (the if __name__ == '__main__': branch is executed) but the launchTask metchod is never called.

Extending the pymesos.MesosExecutorDriver class and overriding the start() method, solved the problem:

class TestMesosExecutorDriver(pymesos.MesosExecutorDriver):

    def start(self):
        super().start()
        self._notify()

The executor code became:

if __name__ == '__main__':

    driver = TestMesosExecutorDriver(TestExecutor())
    driver.run()

I'm using the 0.35 release.

Thank you,

[ariesdevil]

@revz79 Still not reproduced use your codes in our test cluster, what's your python version? And if you want to fix it as soon, could you please provide an environment like this ?

[mattleaverton]

I've experienced the same issue using the example program. The "Process" class takes a 'master' argument, but assigns it to _new_master. _new_master is only transferred to _master via the _notify method as mentioned, which is called by 'change_master' when a new master is set. Either the MesosExecutorDriver code should a) call change_master by default like the MesosSchedulerDriver does in its 'start' function (as seems to be related to the discussion above) or b) save the 'master' argument to the Process argument in self._master so that the user can specify an initial master at startup if they want, or leave as None.

Change this

class Process(object):
    def __init__(self, master=None, timeout=DAY):
        self._master = None
        ...
        self._new_master = master
        ...

to this

class Process(object):
    def __init__(self, master=None, timeout=DAY):
        self._master = master
        ...
        self._new_master = master
        ...

[ariesdevil]

@revz79 @mattleaverton I guess PR: https://github.com/douban/pymesos/pull/111 solves this problem, plz look at it.

[ariesdevil]

Released: https://github.com/douban/pymesos/releases/tag/0.3.6

[mattleaverton]

Yes v0.3.6 fixes this issue for me - thank you.