通过Mesos、Docker和Go，使用300行代码创建一个分布式系统

构建一个分布式系统是很困难的。它需要可扩展性、容错性、高可用性、一致性、可伸缩以及高效。为了达到这些目的，分布式系统需要很多复杂的组件以一种复杂的方式协同工作。例如，Apache Hadoop在大型集群上并行处理TB级别的数据集时，需要依赖有着高容错的文件系统（HDFS）来达到高吞吐量。

在之前，每一个新的分布式系统，例如Hadoop和Cassandra，都需要构建自己的底层架构，包括消息处理、存储、网络、容错性和可伸缩性。庆幸的是，像Apache Mesos这样的系统，通过给分布式系统的关键构建模块提供类似操作系统的管理服务，简化了构建和管理分布式系统的任务。Mesos抽离了CPU、存储和其它计算资源，因此开发者开发分布式应用程序时能够将整个数据中心集群当做一台巨型机对待。

构建在Mesos上的应用程序被称为框架，它们能解决很多问题：Apache Spark，一种流行的集群式数据分析工具；Chronos，一个类似cron的具有容错性的分布式scheduler，这是两个构建在Mesos上的框架的例子。构建框架可以使用多种语言，包括C++，Go，Python，Java，Haskell和 Scala。

在分布式系统用例上，比特币开采就是一个很好的例子。比特币将为生成 acceptable hash 的挑战转为验证一块事务的可靠性。可能需要几十年，单台笔记本电脑挖一块可能需要花费超过150年。结果是，有许多的“采矿池”允许采矿者将他们的计算资源联合起来以加快挖矿速度。Mesosphere的一个实习生，Derek，写了一个比特币开采框架，利用集群资源的优势来做同样的事情。在接下来的内容中，会以他的代码为例。

1个Mesos框架有1个scheduler 和1个executor组成。scheduler 和Mesos master通信并决定运行什么任务，而executor 运行在slaves上面，执行实际任务。大多数的框架实现了自己的scheduler，并使用1个由Mesos提供的标准executors。当然，框架也可以自己定制executor。在这个例子中即会编写定制的scheduler，并使用标准命令执行器（executor）运行包含我们比特币服务的Docker镜像。

对这里的scheduler来说，需要运行的有两种任务—— 单矿服务器任务和多矿服务器任务。服务器会和一个比特币采矿池通信，并给每个“工人”分配块。“工人”会努力工作，即开采比特币。

任务实际上被封装在executor框架中，因此任务运行意味着告诉Mesos master在其中一个slave上面启动一个executor。由于这里使用的是标准命令执行器（executor），因此可以指定任务是二进制可执行文件、bash脚本或者其他命令。由于Mesos支持Docker，因此在本例中将使用可执行的Docker镜像。Docker是这样一种技术，它允许你将应用程序和它运行时需要的依赖一起打包。

为了在Mesos中使用Docker镜像，这里需要在Docker registry中注册它们的名称： 

const (
    MinerServerDockerImage = "derekchiang/p2pool"
    MinerDaemonDockerImage = "derekchiang/cpuminer"
)

然后定义一个常量，指定每个任务所需资源：

const (
    MemPerDaemonTask = 128  // mining shouldn't be memory-intensive
    MemPerServerTask = 256
    CPUPerServerTask = 1    // a miner server does not use much CPU
)

现在定义一个真正的scheduler，对其跟踪，并确保其正确运行需要的状态：

type MinerScheduler struct {
    // bitcoind RPC credentials
    bitcoindAddr string
    rpcUser      string
    rpcPass      string
    // mutable state
    minerServerRunning  bool
    minerServerHostname string 
    minerServerPort     int    // the port that miner daemons 
                               // connect to
    // unique task ids
    tasksLaunched        int
    currentDaemonTaskIDs []*mesos.TaskID
}

这个scheduler必须实现下面的接口：

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)
} 

现在一起看一个回调函数：

func (s *MinerScheduler) Registered(_ sched.SchedulerDriver, 
      frameworkId *mesos.FrameworkID, masterInfo *mesos.MasterInfo) {
    log.Infoln("Framework registered with Master ", masterInfo)
}
func (s *MinerScheduler) Reregistered(_ sched.SchedulerDriver, 
      masterInfo *mesos.MasterInfo) {
    log.Infoln("Framework Re-Registered with Master ", masterInfo)
}
func (s *MinerScheduler) Disconnected(sched.SchedulerDriver) {
    log.Infoln("Framework disconnected with Master")
}

Registered在scheduler 成功向Mesos master注册之后被调用。

Reregistered在scheduler 与Mesos master断开连接并且再次注册时被调用，例如，在master重启的时候。

Disconnected在scheduler 与Mesos master断开连接时被调用。这个在master挂了的时候会发生。

目前为止，这里仅仅在回调函数中打印了日志信息，因为对于一个像这样的简单框架，大多数回调函数可以空在那里。然而，下一个回调函数就是每一个框架的核心，必须要认真的编写。

ResourceOffers在scheduler 从master那里得到一个offer的时候被调用。每一个offer包含一个集群上可以给框架使用的资源列表。资源通常包括CPU、内存、端口和磁盘。一个框架可以使用它提供的一些资源、所有资源或者一点资源都不给用。

针对每一个offer，现在期望聚集所有的提供的资源并决定是否需要发布一个新的server任务或者一个新的worker任务。这里可以向每个offer发送尽可能多的任务以测试最大容量，但是由于开采比特币是依赖CPU的，所以这里每个offer运行一个开采者任务并使用所有可用的CPU资源。

for i, offer := range offers {
    // … Gather resource being offered and do setup
    if !s.minerServerRunning && mems >= MemPerServerTask &&
            cpus >= CPUPerServerTask && ports >= 2 {
        // … Launch a server task since no server is running and we 
        // have resources to launch it.
    } else if s.minerServerRunning && mems >= MemPerDaemonTask {
        // … Launch a miner since a server is running and we have mem 
        // to launch one.
    }
}

针对每个任务都需要创建一个对应的TaskInfo message ，它包含了运行这个任务需要的信息。

s.tasksLaunched++
taskID = &mesos.TaskID {
    Value: proto.String("miner-server-" + 
                        strconv.Itoa(s.tasksLaunched)),
} 

Task IDs由框架决定，并且每个框架必须是唯一的。

containerType := mesos.ContainerInfo_DOCKER
task = &mesos.TaskInfo {
    Name: proto.String("task-" + taskID.GetValue()),
    TaskId: taskID,
    SlaveId: offer.SlaveId,
    Container: &mesos.ContainerInfo {
        Type: &containerType,
        Docker: &mesos.ContainerInfo_DockerInfo {
            Image: proto.String(MinerServerDockerImage),
        },
    },
    Command: &mesos.CommandInfo {
        Shell: proto.Bool(false),
        Arguments: []string {
            // these arguments will be passed to run_p2pool.py
            "--bitcoind-address", s.bitcoindAddr,
            "--p2pool-port", strconv.Itoa(int(p2poolPort)),
            "-w", strconv.Itoa(int(workerPort)),
            s.rpcUser, s.rpcPass,
        },
    },
    Resources: []*mesos.Resource {
        util.NewScalarResource("cpus", CPUPerServerTask),
        util.NewScalarResource("mem", MemPerServerTask),
    },
}

TaskInfo message指定了一些关于任务的重要元数据信息，它允许Mesos节点运行Docker容器，特别会指定name、task ID、container information以及一些需要给容器传递的参数。这里也会指定任务需要的资源。

现在TaskInfo已经被构建好，因此任务可以这样运行：

driver.LaunchTasks([]*mesos.OfferID{offer.Id}, tasks, &mesos.Filters{RefuseSeconds: proto.Float64(1)})

在框架中，需要处理的最后一件事情是当开采者server关闭时会发生什么。这里可以利用StatusUpdate 函数来处理。

在一个任务的生命周期中，针对不同的阶段有不同类型的状态更新。对这个框架来说，想要确保的是如果开采者server由于某种原因失败，系统会Kill所有开采者worker以避免浪费资源。这里是相关的代码：

if strings.Contains(status.GetTaskId().GetValue(), "server") &&
    (status.GetState() == mesos.TaskState_TASK_LOST ||
        status.GetState() == mesos.TaskState_TASK_KILLED ||
        status.GetState() == mesos.TaskState_TASK_FINISHED ||
        status.GetState() == mesos.TaskState_TASK_ERROR ||
        status.GetState() == mesos.TaskState_TASK_FAILED) {
    s.minerServerRunning = false
    // kill all tasks
    for _, taskID := range s.currentDaemonTaskIDs {
        _, err := driver.KillTask(taskID)
        if err != nil {
            log.Errorf("Failed to kill task %s", taskID)
        }
    }
    s.currentDaemonTaskIDs = make([]*mesos.TaskID, 0)
}

万事大吉！通过努力，这里在Apache Mesos上建立一个正常工作的分布式比特币开采框架，它只用了大约300行GO代码。这证明了使用Mesos 框架的API编写分布式系统是多么快速和简单。