python 编写mesos framework -- helloworld

系统管理员 2022-03-28 14:28 259阅读 0赞

#### 主备 ####

--------------------

*  Python2.7 pymesos（第三方的一个Python模块）
 *  mesos 1.7.0 集群
 *  http服务器或者ftp服务器

#### framework编写 ####

--------------------

framework精确的说是包括scheduler、SchedulerDriver、Executor、ExecutorDriver。下图是大致的工作流程：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ2ODYzOTk_size_16_color_FFFFFF_t_70]

#### 调度器代码 ####

--------------------

> 在远端应该有个存放执行器代码的服务器，http或者ftp等。用于mesos fetcher拉取代码。

from pymesos import MesosSchedulerDriver, Scheduler, encode_data
    from addict import Dict
    import datetime
    import uuid
    import logging
    import getpass
    import socket
    from threading import Thread
    import signal
    import time
    import sys
    
    TASK_CPU = 0.1
    TASK_MEM = 32
    EXECUTOR_CPUS = 0.1
    EXECUTOR_MEM = 32
    
    class HellowWorldScheduler(Scheduler):
    
        def __init__(self, executor):
            self.executor = executor
            self.switch = True
    
        def get_time(self):
            return datetime.datetime.now().strftime("%Y-%m-%d %H:%M")
    
        def get_resource(self, res, name):
            for r in res:
                if r.name == name:
                    return r.scalar.value
            return 0.0
    
        def handle_offer(self, driver, offers):
            filters = {'refuse_seconds': 5}
            for part in offers:
                # print(self.get_time(), 'get offers', part)
                cpus = self.get_resource(part.resources, 'cpus')
                mem = self.get_resource(part.resources, 'mem')
                if cpus < TASK_CPU or mem < TASK_MEM:
                    continue
                logging.info('get offer %s - %s - %s',
                             self.get_time(),
                             '\033[1;32m' + 'success' + '\033[0m',
                             "cpu: {0} mem: {1} agent: {2}".format(cpus, mem, part.agent_id.value))
                task = Dict()
                task_id = str(uuid.uuid4())
                task.task_id.value = task_id
                task.agent_id.value = part.agent_id.value
                task.name = 'task {}'.format(task_id)
                task.executor = self.executor
                task.data = encode_data('{0} Hello from task {1}!'.format(self.get_time(), task_id))
                task.health_check.type = 'TCP'
                task.health_check.interval_seconds = 1
                task.health_check.tcp.port = 8000
                task.resources = [
                    dict(name='cpus', type='SCALAR', scalar={'value': TASK_CPU}),
                    dict(name='mem', type='SCALAR', scalar={'value': TASK_MEM}),
                ]
                driver.launchTasks(part.id, [task], filters)
                break
    
        def resourceOffers(self, driver, offers):
            if self.switch:
                self.handle_offer(driver, offers)
    
        def statusUpdate(self, driver, update):
            # print(update)
            logging.info('Status update TID %s - %s - %s - %s - %s',
                          update.task_id.value,
                          update.state, update.message, update.reason, update.healthy)
            if update.state == 'TASK_RUNNING':
                logging.info('http server running on %s' % update.slave_id.value)
                self.switch = False
            else:
                self.switch = True
    
    if __name__ == '__main__':
        def main(master):
            executor = Dict()
            executor.executor_id.value = str(uuid.uuid4())
            executor.name = 'HellowWorldExecutor'
            executor.command.uris.value = 'http://192.168.0.12:8000/executor.py'
            executor.command.value = "python executor.py"
            executor.resources = [
                dict(name='mem', type='SCALAR', scalar={'value': EXECUTOR_MEM}),
                dict(name='cpus', type='SCALAR', scalar={'value': EXECUTOR_CPUS}),
            ]
    
            framework = Dict()
            framework.user = getpass.getuser()
            framework.name = "HelloWorldFramework"
            framework.hostname = socket.gethostname()
    
            driver = MesosSchedulerDriver(
                HellowWorldScheduler(executor),
                framework,
                master,
                use_addict=True,
            )
    
            def signal_handler(signal, frame):
                driver.stop()
    
            def run_driver_thread():
                driver.run()
    
            driver_thread = Thread(target=run_driver_thread, args=())
            driver_thread.start()
    
            print('Scheduler running, Ctrl+C to quit.')
            signal.signal(signal.SIGINT, signal_handler)
            while driver_thread.is_alive():
                time.sleep(1)
    
        logging.basicConfig(level=logging.DEBUG)
        if len(sys.argv) != 2:
            print("Usage: {} <mesos_master>".format(sys.argv[0]))
            sys.exit(1)
        else:
            main(sys.argv[1])

#### 执行器代码 ####

--------------------

> 这个执行器，是运行一个简单的httpserver，访问这个server会看到操作系统上的文件

import time
    from threading import Thread
    import socket
    from pymesos import MesosExecutorDriver, Executor, decode_data
    from addict import Dict
    import SimpleHTTPServer
    import SocketServer
    import cgi
    
    PORT = 8000
    
    class ServerHandler(SimpleHTTPServer.SimpleHTTPRequestHandler):
    
        def do_GET(self):
            SimpleHTTPServer.SimpleHTTPRequestHandler.do_GET(self)
            #host_name = socket.gethostname()
            #self.wfile.write('<h1>Hello world Mesos-%s</h1>' % host_name)
    
        def do_POST(self):
            form = cgi.FieldStorage()
            SimpleHTTPServer.SimpleHTTPRequestHandler.do_GET(self)
    
    class HelloWorldExecutor(Executor):
    
        def run_hello_world(self):
            Handler = ServerHandler
            httpd = SocketServer.TCPServer(("", PORT), Handler)
            print("serving at port", PORT)
            httpd.serve_forever()
    
        def check_port(self, host, port):
            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            s.settimeout(2)
            ADDR = (str(host), int(port))
            status = s.connect_ex(ADDR)
            if status != 0:
                return False
            else:
                return True
    
        def launchTask(self, driver, task):
            def run_task(task):
                time.sleep(2)
                update = Dict()
                update.task_id.value = task.task_id.value
                update.timestamp = time.time()
                if self.check_port('127.0.0.1', 8000):
                    update.state = 'TASK_RUNNING'
                    update.healthy = True
                else:
                    update.state = 'TASK_FAILED'
                    update.healthy = False
                driver.sendStatusUpdate(update)
            thread2 = Thread(target=run_task, args=(task,))
            thread1 = Thread(target=self.run_hello_world, args=())
            thread1.start()
            thread2.start()
    		# 此处不用join，驱动器会帮我们join
    		
    if __name__ == '__main__':
        import logging
        logging.basicConfig(level=logging.DEBUG)
        driver = MesosExecutorDriver(HelloWorldExecutor(), use_addict=True)
        driver.run()

#### 运行 ####

--------------------

*  python test\_framework.py 192.168.0.13 即可运行了  
    ![在这里插入图片描述][20190110193650958.png]
 *  mesos UI  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ2ODYzOTk_size_16_color_FFFFFF_t_70 1]
 *  访问httpserver  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ2ODYzOTk_size_16_color_FFFFFF_t_70 2]

--------------------

> 先写到这里了，有问题进QQ群630300475

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ2ODYzOTk_size_16_color_FFFFFF_t_70]: /images/20220328/2c60639cb3a54cefaa92e88e84717c8c.png
[20190110193650958.png]: /images/20220328/97c7306944d24ba2b20fff509267b088.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ2ODYzOTk_size_16_color_FFFFFF_t_70 1]: /images/20220328/ca41d66465294f28923b852981e9ec1f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ2ODYzOTk_size_16_color_FFFFFF_t_70 2]: /images/20220328/5cc157860da941f79689bff39a728bf5.png