浅尝Apache Mesos

文章目录

- 1. Mesos是什么
- 2. 共享集群
- 3. Apache Mesos
- - 3.1 Mesos主节点
  - 3.2 Mesos代理
  - 3.3 Mesos框架
- 4. 资源管理
- - 4.1 资源提供
  - 4.2 资源角色
  - 4.3 资源预留
  - 4.4 资源权重与配额
- 5. 实现框架
- - 5.1 框架主类
  - 5.3 实现执行器
- 6. 小结
- 参考

1. Mesos是什么

Mesos是什么，Mesos是一个分布式的系统内核。

Mesos的构建原理与Linux内核相同，只是抽象级别不同。Mesos的内核在每台机器上运行并为应用程序（例如Hadoop、Spark、Kafka、ElasticSearch）提供API跨整个数据中心和云环境的资源管理和调度。

Mesos主要有以下特性：

线性可扩展
1. 经过业界验证，可以轻松扩展到数万个节点
高可用性
1. 使用Zookeeper实现主服务器和代理服务器容错副本，无中断升级
容器化
1. 原生支持使用Docker和AppC镜像启动容器
可插拔隔离
1. 对CPU、内存、磁盘、端口、GPU和自定义资源隔离模块的一流隔离支持
两级调度
1. 支持使用可插入的调度策略在同一集群中运行云原生和遗留应用程序
接口
1. 用于开发新的分布式应用程序、操作集群和监控的HTTP API
Web UI
1. 内置Web UI，用于查看集群状态和容器沙箱导航
跨平台
1. 可在Linux、OSX和Windows上运行，与云服务商无关

我们通常会将各种应用程序部署在同一台机器集群上。而Apache Mesos是一个允许此类程序之间共享资源的平台。

2. 共享集群

许多应用程序需要共享集群，总的来说有两种常见的方法：

对集群静态分区并在每个分区上运行一个应用程序
为应用程序分配一组机器

上述方法虽然允许应用程序彼此独立运行，但并不能实现较高的资源利用率：比如某个应用程序只运行了一小段时间，之后处于非活动状态，由于我们为该应用程序分配了静态机器或分区，因此在非活动状态期间，存在未使用的资源。

我们可以通过将非活动状态期间的共享资源重新分配给其他应用程序来优化资源利用率，而Apache Mesos则可以帮助应用程序之间的资源动态分配。

3. Apache Mesos

基于我们上面讨论的两种共享集群的方法，应用程序只知道他们正在运行的特定分区或机器的资源，然而Apache Mesos为应用程序提供了集群中所有资源的抽象视图。

Mesos会充当机器与应用程序之间的接口，为应用程序提供集群中所有机器上的可使用资源，它会经常更新此信息从而获取完成状态的应用程序释放的资源。它允许应用程序做出关于哪台机器上执行哪个任务的最佳决定。

为了理解Mesos的工作原理，我们来看下它的架构：

Mesos架构

上图展示了Mesos的主要组件，Mesos由一个管理运行在每个集群节点上的代理守护进程和主守护进程以及在这些代理上运行任务的Mesos框架组成。

主节点通过提供资源实现跨框架的细粒度资源共享(CPU、RAM等)。每个提供的资源包含一个列表<agent ID, resource1: amount1, resource2: amount2, ...>，主节点根据给定的组织策略(例如公平共享和严格优先级)决定向每个框架提供多少资源。为了支持多样化的策略集，主节点才用模块化架构，可通过插件机制来轻松添加新的分配模块。

在Mesos上运行的框架由两个组件组成：

向主节点注册以获取资源的调度程序
在代理节点上启动以运行框架任务的执行程序进程

主节点确定向每个框架提供多少资源，而框架的调度程序则选择使用提供资源中的哪些资源。当框架接受提供的资源时，它会将要在这些资源上运行的任务的描述传递给Mesos，反过来，Mesos会在相应的代理上启动任务。

3.1 Mesos主节点

Master是Mesos中的核心组件，用于存储集群中资源的当前状态，此外，它还通过传递有关资源和任务等信息，从而充当代理和应用程序之间的协调器。

由于主服务器发生任何故障都会导致资源和任务状态丢失，因此我们将其部署在高可用性配置中。如上图所示，Mesos部署了主服务器守护进程和一个Master，这些守护进程依靠Zookeeper在发生故障时恢复状态。

3.2 Mesos代理

Mesos集群必须在每台机器上运行代理，这些代理定期向主节点报告其资源，并依次接收应用程序已经安排运行的任务。在计划任务完成或者丢失后，次循环将重复。

3.3 Mesos框架

Mesos允许应用程序实现一个抽象组件，该组件与Master交互以接收集群中的可用资源，并根据这些资源做出调度决策，这些组件成为框架。

Mesos的框架由两个子组件组成：

调度程序：使应用程序能够根据所有代理上的可用资源来调度任务
执行器：在所有代理上运行，并包含该代理上执行任务计划任务所需的所有信息

框架安排任务运行的示例图：

Mesos框架安排任务运行示意

代理1向主服务器报告其有4个CPU和4GB的可用内存，主服务器随后调度分配策略模块，告知其应该为框架1提供所有可用资源。
主服务器向框架1发送资源邀约，描述代理1有哪些可用资源
框架的调度程序向主服务器回复在代理上运行的两个任务的信息，第一个任务使用<2CPU,1GB RAM>，第二个任务使用<1CPU,2GB RAM>
最后主服务器将任务发送给代理，代理将适当的额资源分配给框架的执行器，执行器则启动这两个任务(图中虚线所示)，由于仍然存在1CPU和1GB RAM未分配，因此分配模块现在可以将它提供给框架2.

此外，当任务完成且新资源变为空闲时，此资源提供过程会重复。

Mesos允许应用程序使用各种编程语言实现自定义调度程序和执行程序。按照Mesos-go中的Scheduler接口的定义如下，具体可参考：Mesos Scheduler interface

// Scheduler a type with callback attributes to be provided by frameworks
// schedulers.
//
// Each callback includes a reference to the scheduler driver that was
// used to run this scheduler. The pointer will not change for the
// duration of a scheduler (i.e., from the point you do
// SchedulerDriver.Start() to the point that SchedulerDriver.Stop()
// returns). This is intended for convenience so that a scheduler
// doesn't need to store a reference to the driver itself.
type Scheduler interface {

	Registered(SchedulerDriver, *mesos.FrameworkID, *mesos.MasterInfo)

	Reregistered(SchedulerDriver, *mesos.MasterInfo)

	Disconnected(SchedulerDriver)

	ResourceOffers(SchedulerDriver, []*mesos.Offer)

	OfferRescinded(SchedulerDriver, *mesos.OfferID)

	StatusUpdate(SchedulerDriver, *mesos.TaskStatus)

	FrameworkMessage(SchedulerDriver, *mesos.ExecutorID, *mesos.SlaveID, string)

	SlaveLost(SchedulerDriver, *mesos.SlaveID)

	ExecutorLost(SchedulerDriver, *mesos.ExecutorID, *mesos.SlaveID, int)

	Error(SchedulerDriver, string)
}

从上述接口的定义可以看出，它主要由各种回调方法组成，主要用于与主机通信。

同样，执行器的实现也需要实现Executor接口，具体可参考：Mesos Executor interface

/**
 * Executor callback interface to be implemented by frameworks' executors. Note
 * that only one callback will be invoked at a time, so it is not
 * recommended that you block within a callback because it may cause a
 * deadlock.
 *
 * Each callback includes an instance to the executor driver that was
 * used to run this executor. The driver will not change for the
 * duration of an executor (i.e., from the point you do
 * ExecutorDriver.Start() to the point that ExecutorDriver.Join()
 * returns). This is intended for convenience so that an executor
 * doesn't need to store a pointer to the driver itself.
 */
type Executor interface {

	Registered(ExecutorDriver, *mesosproto.ExecutorInfo, *mesosproto.FrameworkInfo, *mesosproto.SlaveInfo)

	Reregistered(ExecutorDriver, *mesosproto.SlaveInfo)

	Disconnected(ExecutorDriver)

	LaunchTask(ExecutorDriver, *mesosproto.TaskInfo)

	KillTask(ExecutorDriver, *mesosproto.TaskID)

	FrameworkMessage(ExecutorDriver, string)

	Shutdown(ExecutorDriver)

	Error(ExecutorDriver, string)
}

4. 资源管理

4.1 资源提供

上面说到代理节点会将其资源信息发布给主服务器，反过来主服务器将这些资源提供给集群中运行的框架，此过程成为资源提供。

资源信息一般由两部分组成：

资源
1. 资源用于发布代理机器的硬件信息，例如CPU、内存、磁盘等
属性

每个代理有五种预定义资源：

中央处理器
图形处理器
内存
磁盘
端口

这些资源的值可以用以下三种类型之一来定义：

标量：用浮点数来表示数值信息，以允许小数值，例如1.5G内存
范围：用于表示标量值的范围，例如端口号范围
集合：用于表示多个文本值

默认情况下，Mesos代理会尝试从机器检测这些资源。但在某些情况下我们可以在代理上配置自定义资源，此类自定义资源的值，也应为上述任何一种类型。

例如，我们可以使用以下这些资源启动我们的代理：

--resources='cpus:24;gpus:2;mem:2048;disk:40960;ports:[21000-24000];bugs(debug_role):{a,b,c}'

上面我们为代理配置了一些预定义资源和一个名为bugs集合类型的自定义资源。

除了资源之外，代理还可以向主服务器发布键值属性，这些属性充当代理的附加元数据，并帮助框架进行调度决策。

一个有用的例子是将带来添加到不同的机架或者区域，然后在同一个机架或者区域上安排各种任务以实现数据本地化

--attributes='rack:abc;zone:sg;os:centos5;level:10;keys:[1000-1500]'

与资源类似，属性的值可以是标量、范围或者文本类型。

4.2 资源角色

现代操作系统许多都支持多用户，同样，Mesos也支持同一个集群中的多个用户，这些用户被称为角色。我们可以将每个决策视为集群中的资源消费者。

由此，Mesos代理可以基于不同的分配策略对不同角色下的资源进行划分，而框架可以在集群内订阅这些角色，从而实现对不同角色下的资源进行细粒度的控制。

例如，假设有一个集群托管应用程序，为组织中不同用户提供服务，因此，通过将资源划分成角色，每个应用程序都可以彼此独立地工作。

此外，框架可以使用这些角色来实现数据局部性。

例如，假设集群中有两个应用程序，分别为生产者和消费者，其中生产者将数据写入持久卷，消费者随后可以读取该卷，我们可以通过生产者共享该卷来优化消费者应用程序。

由于Mesos允许多个应用程序订阅同一角色，我们可以将持久卷与资源角色关联起来，此外，生产者和消费者的框架都将订阅相同的资源角色，因此消费者应用程序现在可以在与生产者应用程序相同的卷上启动数据读取任务。

4.3 资源预留

Mesos如何将集群资源分配给不同的角色，Mesos通过预留来实现资源分配。

预留类型有两种：

静态预留
动态预留

静态预留则是在代理启动时的资源分配：

–resources=“cpus:4;mem:2048;cpus(test_abc):8;mem(test_abc):4096”

与静态预留不同，动态预留允许我们重新调整角色内的资源，Mesos允许框架和集群操作员通过框架消息(作为对资源提供的响应)或者通过HTTP API调用动态更改资源分配。

Mesos将所有不具有任何角色的资源分配给名为(*)的默认角色，Master主服务器向所有框架提供此类资源，无论它们是否订阅了该资源。

4.4 资源权重与配额

Mesos主节点一般使用公平策略来提供资源，它使用加权主导资源公平性来识别缺少资源的角色，然后主服务器向已订阅这些角色的框架提供更多资源。

尽管在应用程序之间公平共享资源是Mesos的一个重要特性，但这并非是必要的。假设一个集群托管着资源占用低的应用程序和资源需求高的应用程序。在样的部署中，我们希望根据应用程序的性质分配资源。

Mesos允许框架通过订阅角色并为该角色添加更高的权重来请求更多资源。因此，如果有两个角色，一个权重为1，另一个为2，则Mesos会为第二个角色分配两倍的公平份额资源。

另外，我们还可以通过HTTP接口调用来配置权重，可参考：Mesos Weights Set by HTTP API

除了确保为具有权重的角色公平分配资源之外，Mesos还确保为角色分配最少的资源。

Mesos允许我们为资源角色添加配额，配额可以指定角色保证接收到的最小资源量。

5. 实现框架

Mesos允许应用程序选择任意语言提供框架实现，即实现Scheduler和Executor的接口从而实现框架。

5.1 框架主类

在实现调度程序和执行程序之前，需要先实现框架的入口点：

向master注册
向代理执行器提供运行时信息
启动调度程序

下面程序来自Mesos官方提供的示例，具体可参考：Mesos Framework Python test examples

import os
import sys
import time

import mesos.interface
from mesos.interface import mesos_pb2
from mesos.scheduler import MesosSchedulerDriver

TOTAL_TASKS = 5

TASK_CPUS = 1
TASK_MEM = 128

class TestScheduler(mesos.interface.Scheduler):
    def __init__(self, implicitAcknowledgements, executor, framework):
        self.implicitAcknowledgements = implicitAcknowledgements
        self.executor = executor
        self.framework = framework
        self.taskData = {}
        self.tasksLaunched = 0
        self.tasksFinished = 0
        self.messagesSent = 0
        self.messagesReceived = 0
		# 注册
    def registered(self, driver, frameworkId, masterInfo):
        print "Registered with framework ID %s" % frameworkId.value
        self.framework.id.CopyFrom(frameworkId)
        driver.updateFramework(framework, [])

    def resourceOffers(self, driver, offers):# 资源申请
        for offer in offers:
            tasks = []
            offerCpus = 0
            offerMem = 0
            for resource in offer.resources:
                if resource.name == "cpus":
                    offerCpus += resource.scalar.value
                elif resource.name == "mem":
                    offerMem += resource.scalar.value

            print "Received offer %s with cpus: %s and mem: %s" \
                  % (offer.id.value, offerCpus, offerMem)

            remainingCpus = offerCpus
            remainingMem = offerMem

            while self.tasksLaunched < TOTAL_TASKS and \
                  remainingCpus >= TASK_CPUS and \
                  remainingMem >= TASK_MEM:
                tid = self.tasksLaunched
                self.tasksLaunched += 1

                print "Launching task %d using offer %s" \
                      % (tid, offer.id.value)

                task = mesos_pb2.TaskInfo()
                task.task_id.value = str(tid)
                task.slave_id.value = offer.slave_id.value
                task.name = "task %d" % tid
                task.executor.MergeFrom(self.executor)

                cpus = task.resources.add()
                cpus.name = "cpus"
                cpus.type = mesos_pb2.Value.SCALAR
                cpus.scalar.value = TASK_CPUS # CPU资源

                mem = task.resources.add()
                mem.name = "mem"
                mem.type = mesos_pb2.Value.SCALAR
                mem.scalar.value = TASK_MEM # 内存资源

                tasks.append(task)
                self.taskData[task.task_id.value] = (
                    offer.slave_id, task.executor.executor_id)

                remainingCpus -= TASK_CPUS
                remainingMem -= TASK_MEM

            operation = mesos_pb2.Offer.Operation()
            operation.type = mesos_pb2.Offer.Operation.LAUNCH
            operation.launch.task_infos.extend(tasks)

            driver.acceptOffers([offer.id], [operation])

    def statusUpdate(self, driver, update):# 状态更新
        print "Task %s is in state %s" % \
            (update.task_id.value, mesos_pb2.TaskState.Name(update.state))

        # Ensure the binary data came through.
        if update.data != "data with a \0 byte":
            print "The update data did not match!"
            print "  Expected: 'data with a \\x00 byte'"
            print "  Actual:  ", repr(str(update.data))
            sys.exit(1)

        if update.state == mesos_pb2.TASK_FINISHED:
            self.tasksFinished += 1
            if self.tasksFinished == TOTAL_TASKS:
                print "All tasks done, waiting for final framework message"

            slave_id, executor_id = self.taskData[update.task_id.value]

            self.messagesSent += 1
            driver.sendFrameworkMessage(
                executor_id,
                slave_id,
                'data with a \0 byte')

        if update.state == mesos_pb2.TASK_LOST or \
           update.state == mesos_pb2.TASK_KILLED or \
           update.state == mesos_pb2.TASK_FAILED:
            print "Aborting because task %s is in unexpected state %s with message '%s'" \
                % (update.task_id.value, mesos_pb2.TaskState.Name(update.state), update.message)
            driver.abort()

        # Explicitly acknowledge the update if implicit acknowledgements
        # are not being used.
        if not self.implicitAcknowledgements:
            driver.acknowledgeStatusUpdate(update)
		# 框架消息处理
    def frameworkMessage(self, driver, executorId, slaveId, message):
        self.messagesReceived += 1

        # The message bounced back as expected.
        if message != "data with a \0 byte":
            print "The returned message data did not match!"
            print "  Expected: 'data with a \\x00 byte'"
            print "  Actual:  ", repr(str(message))
            sys.exit(1)
        print "Received message:", repr(str(message))

        if self.messagesReceived == TOTAL_TASKS:
            if self.messagesReceived != self.messagesSent:
                print "Sent", self.messagesSent,
                print "but received", self.messagesReceived
                sys.exit(1)
            print "All tasks done, and all messages received, exiting"
            driver.stop()

if __name__ == "__main__":
    if len(sys.argv) != 2:
        print "Usage: %s master" % sys.argv[0]
        sys.exit(1)

    executor = mesos_pb2.ExecutorInfo()
    executor.executor_id.value = "default"
    executor.command.value = os.path.abspath("./test-executor")
    executor.name = "Test Executor (Python)"
    executor.source = "python_test"

    framework = mesos_pb2.FrameworkInfo()
    framework.user = "" # Have Mesos fill in the current user.
    framework.name = "Test Framework (Python)"
    framework.checkpoint = True
    framework.role = "*"

    implicitAcknowledgements = 1
    if os.getenv("MESOS_EXPLICIT_ACKNOWLEDGEMENTS"):
        print "Enabling explicit status update acknowledgements"
        implicitAcknowledgements = 0

    if os.getenv("MESOS_EXAMPLE_AUTHENTICATE"):
        print "Enabling authentication for the framework"

        if not os.getenv("MESOS_EXAMPLE_PRINCIPAL"):
            print "Expecting authentication principal in the environment"
            sys.exit(1);

        credential = mesos_pb2.Credential()
        credential.principal = os.getenv("MESOS_EXAMPLE_PRINCIPAL")

        if os.getenv("MESOS_EXAMPLE_SECRET"):
            credential.secret = os.getenv("MESOS_EXAMPLE_SECRET")

        framework.principal = os.getenv("MESOS_EXAMPLE_PRINCIPAL")

    else:
        framework.principal = "test-framework-python"
        credential = None

    # Subscribe with all roles suppressed to test updateFramework() method
    driver = MesosSchedulerDriver(
        TestScheduler(implicitAcknowledgements, executor, framework),
        framework,
        sys.argv[1],
        implicitAcknowledgements,
        credential,
        [framework.role])

    status = 0 if driver.run() == mesos_pb2.DRIVER_STOPPED else 1

    # Ensure that the driver process terminates.
    driver.stop();

    sys.exit(status)

调度程序可能会由于缺乏资源而无法再代理上启动任务，则会被拒绝，就是上面的DRIVER_STOPPED。

5.3 实现执行器

框架的执行器组件负责在Mesos代理上执行应用程序服务。

import sys
import threading
import time

import mesos.interface
from mesos.interface import mesos_pb2
from mesos.executor import MesosExecutorDriver

class MyExecutor(mesos.interface.Executor):
    def launchTask(self, driver, task):
        # Create a thread to run the task. Tasks should always be run in new
        # threads or processes, rather than inside launchTask itself.
        def run_task():
            print "Running task %s" % task.task_id.value
            update = mesos_pb2.TaskStatus()
            update.task_id.value = task.task_id.value
            update.state = mesos_pb2.TASK_RUNNING
            update.data = 'data with a \0 byte'
            driver.sendStatusUpdate(update)

            # This is where one would perform the requested task.

            print "Sending status update..."
            update = mesos_pb2.TaskStatus()
            update.task_id.value = task.task_id.value
            update.state = mesos_pb2.TASK_FINISHED
            update.data = 'data with a \0 byte'
            driver.sendStatusUpdate(update)
            print "Sent status update"

        thread = threading.Thread(target=run_task)
        thread.start()

    def frameworkMessage(self, driver, message):
        # Send it back to the scheduler.
        driver.sendFrameworkMessage(message)

if __name__ == "__main__":
    print "Starting executor"
    driver = MesosExecutorDriver(MyExecutor())
    sys.exit(0 if driver.run() == mesos_pb2.DRIVER_STOPPED else 1)