摘要：如何分配集群内的计算资源是云计算数据中心的重要问题。为了提高集群内服务器的利用率，实现动态资源共享，需要一个中间件进行集群内的任务调度。Torca作为Typhoon云平台的关键系统也就应运而生，已经在网页搜索、广告等广泛应用。本文主要介绍Torca的设计目标、系统架构、主要特性等。 1. 背景 Typhoon（台风）是由基础架构部开发的集分布式存储和分布式计算于一体的云计算平台。它由提供高可靠性的分布式文件系统XFS、集群调度系统Torca、分布式半结构化存储系统XCube、分布式计算框架MapReduce以及认证组件Taas等组件构成。 图1. Typhoon overview. 从台风架构上看，Torca处于比较底层的位置，上面可以运行xcube、Mapreduce、其他应用程序，同时，Torca也依赖TAAS、xfs、poppy等更底层的系统。简单说来，可以把Torca看作一个OS，它可以运行程序，管理cpu、内存等资源。 2. Torca的设计目标 1) 提高资源利用率 下图是是公司网管系统（TMP）上看到的几种主流服务器的月平均利用率与月峰值平均利用率。 图2. 公司主流服务器的平均使用率 图3. 公司机型分布 由于C类的机型最多（图3），我们特别看一下C类机器，月峰值利用率不到30%，月平均利用率不到10%！ 图4. C1类服务器月平均利用率与月峰值平均利用率 资源利用率本质上就是成本问题，公司日益增长的服务器数量虽然一定程度上反映了我们业务规模的不断扩大，但是资源使用率也要引起我们的特别关注。简单分析一下利用率低的原因： a. 为了满足峰值资源需求，业务只能以最大资源部署应用； b. 业务间都是隔离封闭的，资源无共享，无法错峰交谷； c. 整个业务存在这样的情况，单个进程也常以最大资源需求部署； d. 不同业务都有自己的buffer，公司总的空闲机器就会有很多，这些buffer对业务迅速扩容很有意义，但是各个buffer加起来的资源空闲不能忽视。 Torca可以实现多业务之间的资源共享，也可以实现基于单个机器上的多任务资源共享，并有弹性管理机制，这样就可以根据不同业务的需要，像有一个聪明的大脑一样实现着调度，提高资源利用率。 2) 自动发布部署 台风云平台由很多复杂组件构成，让用户从传统的独占机器资源方式过渡到云平台上，对业务的代价有多少呢？是否为了获得更高的资源利用率，节省更多的成本，对业务的要求更高，发布部署方式更加复杂呢？ 恰恰相反，Torca不止有操作系统那样可以启动进程、管理资源这样基本的功能，也提供了便于用户操作的方式。如下图，只有第一步“资源申请”需要用户做的，申请方式也只是一个在开机上的一条命令而已。而接下来的资源准备、启动程序、监控程序、展示程序等等都由Torca为你完成，用户甚至不需要登陆到机器上。将所有的资源以及部署、监控等运维工作都交给Torca，业务方可以更加关注自己的业务。 图5. Applications on Torca 3) 容灾 下图是googlejeff dean的一页ppt，在我们的观念里，也开始把故障作为一种理所当然的常态了，尤其是在大型分布式环境中，没有无故障的侥幸存在。Torca更是在设计之初就充分考虑了Torca自身的容灾，以及如何为上层业务提供自动化的容灾。 图6. The joys of real hardware 4) 资源共享及任务隔离 如前所述，只有在资源共享才有机会带来更高的资源利用率。不同业务在同一个集群上运行，他们各自可以获取多少资源？有无优先级只分？如果有不同优先级的话，低优先级可否被高优先级抢占？同一个机器上来自不同的业务的进程，各自使用哪个磁盘分区？server允许监听哪些端口，一个进程需要的资源如何保证等等，如何在不需要用户参与的情况下自动调度就是Torca的一个核心能力。 资源共享创造了一个大同的资源池，然而作为业务A，可能会担心：我的进程会不会和业务B的进程共享被调度到同一服务器呢？万一B的进程有内存泄漏之类会不会引发我的进程都被oom kill呢？B进程该不会有引发系统崩溃的危险操作吧？ 另一个场景：我想提交一个测试进程到云平台上，但这个测试程序还没有通过完备的测试，万一程序有问题，该不会伤害到一个未知的程序吧？这个测试敢不敢提交呢？ 这些担心在Torca上都是不需要的。Torca既然做了资源共享，就也为个各个进程提供了隔离的方式。 3. 系统架构 下图是Torca的一个总体系统架构。 图7. Architecture of Torca. Submitter Submitter是Torca的客户端，在集群外的用户机器（如开发机器或管理机器）上运行，用于提交任务，查询集群机器资源/任务状态，对任务进行操作（如暂停/终止）。Submitter可向权限范围内的所有集群提交任务。当用户从Submitter向某个集群提交任务时，Submitter会向Name Service查询那个集群Central Manager的IP地址，然后把任务提交给那个Central Manager。提交任务时，用户可选择下列3种方式部署软件：1. 可执行文件从Submitter拷贝到共享文件系统，供Execute Server拉取；2. 预先把可执行文件及.so库部署在共享文件系统，供Execute Server拉取。 任务描述文件 每个任务都有任务描述文件以描述任务的属性及对资源的需求。用户编写任务描述文件，通过提交任务描述文件提交任务。一个任务对应一个可执行文件，可能以多个进程在多台机器上运行。任务对资源的需求分Requirement和Rank。Requirement为必须满足的条件；Rank为Requirement满足之后，优先考虑的条件。任务描述文件可包括以下要点： 可执行文件的路径，对于有Setup/TearDown要求的任务，可用Linux脚本进行封装，如pre-script/post-script； 任务的运行环境； 任务需要独占机器，还是可以与其他应用复用机器； 任务的扩容/收缩策略； 任务需要多少台机器，或多少个核；为简化实现，Torca按机器分配任务； 任务对硬件的要求，如CPU型号，Memory Size，等； 任务对软件的要求，如OS版本，glibc版本，软件包，等 任务描述文件举例： 图8. Torca jdf(job description file). 为了程序调用的方便，我们也提供了protobuf格式的任务描述方式： JobDescriptionjob_des; job_des.set_cmd(“simple_job”); job_des.set_transfer_files(“/data/home/mavisluo/test_submitter/job/simple_job”); job_des.set_job_name=“test”; job_des.set_task_count=10 Central Manager Central Manager是集群任务调度的中心，包括以下模块： Master Daemon：负责启动/重启其他进程。 Scheduler：管理多个优先级的任务队列，根据任务描述文件生成任务ClassAd（分类广告），通过collector匹配任务与资源，下发任务至Execute Server。 Collector：集群的数据中心，负责从Execute Server收集机器及任务状态。机器的动态信息由Execute Server上报，一些静态信息及机器无法上报的信息如机位从CMDB拉取；同时，也是集群的匹配中心，对任务与资源的ClassAd进行匹配（MatchMaking）。 HTTP Server：CM也运行HTTP Server，提供集群及任务信息给浏览器查询，如集群有多少台机器，有多少任务，任务在哪些机器上运行，等等。 Execute Server Execute Server是集群内运行任务的机器，包括以下模块： Master Daemon：负责启动/重启其他进程。 Start Daemon: 接受Central Manager下发的任务，产生一个进程去执行任务；接受Central Manager对任务暂停/终止/继续的命令，发相应Signal给任务进程；把任务Exit Status上报给Central Manager；定期上报机器状态给Central Manager的Collector，机器状态为类似Linux Top命令的信息, 默认值为每分钟上报一次（参考Linux CPU load average 采样频率为1/5/15分钟）。 HTTP Server: Start Daemon也运行HTTP Server，提供机器状态及任务情况（占用CPU，Heap/Stack Size）给浏览器查询。 Shared File System 为避免数据在不同机器间迁移，Torca的应用应使用共享文件系统，如分布式文件系统XFS或网络文件系统NFS，读写数据。任务运行的Log文件也可写到共享文件系统中。软件包也可预发布在共享文件系统，供Execute Server拉取。 作业执行流程如下图：用户通过submitter和jdf提交job，如果作业依赖的文件在提交机本地，则submitter自动将其copy到xfs，并且用digest做标记，同时对job的各个属性进行解析，以及进行有效性检查，如果没有问题后，将生成最终的作业描述，发给CM上的scheduler。scheduler执行一定的调度策略，当决定这个job可以调度时，就将其发给collector，则collector会返回给scheduler满足条件的机器列表，scheduler就可以向这些机器发出启动task的流程了。Execute server根据job描述，准备相应的作业执行环境，分配资源，创建container等，就会启动task，并且在zk上记入该task相应的name信息。（本逻辑在2.0中有改动，记录name信息在CM上完成） 图9. Torca 作业执行流程图. 4. Torca特性
图10. Resource management on Torca. 1) 调度和资源管理 如前所述，调度是Torca的一个核心特性，调度涉及到下面几个方面： a) Resource Allocation 使用中心化的方式来将job放到各个资源，这样的好处是，central manager中有整个集群的所有机器的信息，可以在整个集群范围内做最优化的资源调度。 常用的调度算法一般有两类：Naive fair: stride, (weighted)round robin，capacity, HTB等；Resource-aware fair: dominant resource fairness (DRF)等。对于Torca，显然naïve fair算法是不够的，如图，在只考虑作业数的情况下，很容易使得资源利用不合理，导致某一个job使用了更多的资源，也是一种unfair。而资源感知的DRF可以做到最终两个业务可以达到一种资源使用上的平衡。 图11. Naïve fair vs DRF. b) Resource Container Torca采用container的技术来做同一机器上的资源隔离，LXC是一种进入了Linux内核计划的Resource container方案。此方案由IBM Linux技术中心、Intel Linux团队等的工程师进行开发和维护。从linux2.6.24内核开始，源于google的cgroup资源管理框架开始出现在内核中，并逐渐成熟和稳定。LXC所实现的OS-level虚拟化，正是建立在cgroup提供的按照进程组实现资源分配和控制的基础之上。同时，LXC还在namespace框架下提供了网络配置，利用pivot_root实现了对用户文件系统的隔离，并使用file POSIX capabilities对用户权限进行了细粒度划分和控制，从而为用户呈现了一个完整的虚拟服务器视图。 LXC相关的构造和运行工作基本集中在用户态。最终运行的容器也只是在Host OS中的一个进程，和普通进程的区别在于赋予了namespace和文件系统的隔离，以及利用了cgroup来进行资源的复用和控制。因此可以推断，相比起构架在hypervisor上层的虚拟机，LXC具有更低的开销和更高的效率。 图12. Resource container. c) Resource Partition 做好资源调度，首先要对资源进行划分，Torca的资源划分粒度很小，也就比较灵活。与虚拟机的方式不同，Torca采用了container的技术来为task提供隔离的环境。如一个ES上有4 core，8G memory，job 1的需求是2 core， 4Gmemory，则我们就为其创建相应容量的container。对于需要1 core、2G memory的job 2也是一样。分配完job 1和job 2之后，这个机器上还有1 core， 2G memory，还可以创建几个container呢？也许是两个（0.5，1）的container，也需是若干个（0.05， 20M）的container。cpu允许设置小数个，memory最小单位为1M。 图13. Resource partition. 这里又引出了另一个问题，在上图中，job 1声明需要4G内存，然后我们给它创建了一个有4G内存的container，正如一个进程独占一个机器一样，如果它在峰值需要60G内存，我们可能要给它B6的机器，而平均情况它是用不到峰值的声明资源的，也就是job 1在大多数用不到4G内存，那么对于这种资源“浪费”问题，Torca也是有策略处理的，我们采用了树状container的机制，后续会在资源管理的专题进行更详尽的讨论。 d) Resource Mapping 根据作业的资源需求，central manager从集群中选择合适的节点来运行这个作业的过程。通常的mapping策略有first-fit，random-fit，group-fit等。不论采用何种算法，resource fragment 和irregularly-shaped usage都是一个让人头疼的问题，直接影响着作业是否可以匹配，最终影响集群的资源利用率。 解释一下，如下图，假设一个server上有两类资源，Resource 1和Resource 2，不妨可以认为是cpu和memory。对于server i和server j，没有被使用的资源量见白色部分，此时，有一个job k提交到集群中，server i满足job k对Resource 1的资源需求，但是不满足对Resource 2的资源需求，而server j与之相反，但同样不符合job k的要求。而实际上集群的总的资源剩余量是足够的。 图14. Resource mapping. e) Other aspects… 云平台的调度本身就是一个复杂的问题，在我们公司众多业务的背景下，又有一些其他要考虑的方面，一起来看下： Quota 不同的业务共享cluster，各个业务都要在一定的quota限制下，防止某一个业务占用集群的过多资源，quota的总量，使用率等如何影响调度，当超过quota时，对该业务的后续作业如何处理等。 Priority 即使是同一个业务，不同的作业之间也会有不同的优先级，如在线服务和离线服务的优先级区分，可能某一个离线的job几天运行一次就ok，优先级不是很高。当集群上运行了很多低优先级的作业时，高优先级作业进入集群时，是否要抢占低优先级作业？什么样的抢占策略才是最合适的，我们也将会在专题中进行讨论。 Task Resource utilization 很多task的资源使用率也是有峰谷之分的，固定的hardlimit机制下的container在资源使用低谷时如果给其他进程共享资源，是否所有类型的job都可以用container中还没有用尽的资源？container暂借出的资源在资源消耗高峰时如何回收？借用别人的资源的task在资源归还后何去何从？这里也留下一个悬念，期待我们的基于ES资源调度的专题。 图15. Resource utilization of one task. Cluster Resource utilization 单个task进程的资源使用有峰谷之别，同样，对于集群来说，cluster整体的资源消耗也是有峰谷的，拿一个tjd1-log集群来说，这个集群主要处理一些报表任务，一般是凌晨1点开始，到9点前都是资源高峰期，相对的，白天任务量就少一些。目前Torca也正在调研如何能在对用户透明的前提下，做cluster间的负载均衡。 图16. Resource utilization of one cluster. 2) 名字服务TNS（Torca naming service） 为什么需要名字服务？ 用户在使用Torca运行自己的作业时，是将作业提交到了一朵“云”上，没有了ip的概念，当然并不是说机器没有ip，我们的每一个ES还是用ip标识的，只是用户并不知道他的作业会运行在哪些ip上，那这个时候，名字服务就应运而生了。 TNS的原理 应用在Torca下的用户视图如下： 图17. Torca 上的任务. 用户在提交作业时给自己的作业起一个好记的名字，那么这个job就获得了一个唯一的名字标识，组成为/$cluster/$role/$job_name（role代表业务），task的标识为/$cluster/$role/$job_name/$task_id，Torca在启动用户进程时（进程与task对应），会将该task的ip、port信息写入zk。 TNS的使用方式 Server：对于要提供服务的server进程来说，它要向Torca申请端口，并且可能有多个端口，我们提供宏的方式让用户提出端口申请，并且对端口也可以取名字，方便多个端口时，client端方便区分。如下，server的程序就可以直接用my_ip和my_port等这些flag了。 图18. 任务向Torca申请端口的几种方式. Client：Torca提供了一组TNS api，应用软件Client端在联系Server端前，用户可以通过名字获取server的ip和端口。并且api有监听方式，可以注册回调函数，如果发现task有迁移，可以采取用户自定义的动作。当然，poppy现在也兼容了tns路径格式，可以对用户透明的实现对迁移的适应。 2) 容灾 Torca系统自身的容灾 和台风中的其他系统一样，Torca本身需要很强的容灾能力，提供高可用。 所依赖的外部系统 Zookeeper：zookeeper本身是server组机制，有2N+1个server，在最多N个server不可用都可以提供服务。详见：http://hadoop.apache.org/zookeeper 图19. Zookeeper servers. Xfs： Torca对xfs的依赖主要作为作业依赖文件的中转，所以即使xfs不可用，对我们的影响也只是新的job可能无法启动，但是我们有重试，所以短暂的不可用也是可以接受的。另外，Torca对发布包的依赖正在开始转向ES自己组成的p2p网络，所以很快对xfs的依赖也会进一步降低。 Torca内部组件： Central manager：与xfs的master的HA机制类似，Torca的master也采用了主备机制。ZooKeeper提供了强一致性的服务。ZooKeeper提供了分布式系统常需要的协调同步服务，可用来实现分布式锁，分布式选举等机制。我们就是用了这种选举和锁服务来实现了主备的选举和单个leader服务。而一些状态信息我们也有定期的checkpoint以及mutation log机制，限于篇幅，此处也不做展开。 Execute server：Torca对每个ES都有监控，当发现有ES宕机或者网络不可达时，central manager会将出问题的ES上的所有task进行再调度，迁移到正常的机器上运行。 应用程序的容灾 在Torca上运行的应用程序的容灾有下面几个方面： i. 以类似Torca master或者xfs master的方式，采用leader选举模式保证高可用，如在Torca上运行的xcube的master job，包含了3个task，这3个task中只有一个作为leader； ii. 依赖Torca的retry/migrate 机制。通过设置作业属性，Torca可以在task意外退出时，自动将task重新在本地拉起，如abort、非0退出、oom等情况，这对于在线服务很有意义。除了本地retry的重新拉起，Torca还提供了跨机的migrate 重试，包括ES宕机等的migrate以及task本身的错误可能导致的migrate； iii.自动监控：只要在提交job时将alarm_id赋值，Torca就会将task进程的所有异常情况以rtx、mail、短信等方式通知用户，也可以接入NOC。异常情况包括：task被oom/ood，coredump，preempted，exit abnormally，migrate.etc. 3) ISSU（In-Service Software Upgrade） 在Torca的实现中，用户的task进程是Torca系统进程的子进程，这样Torca通过waitpid对task进程进行即时的监控。但是同时也会带来Torca系统升级的不便。 目前业界广泛使用的系统升级的方式大致分为两种：第一种为冷升级，即关闭老版本的服务，重新部署新版本的服务。第二种为灰度升级方式，将老版本上的在线服务请求暂时转移到其它服务器，启动部署新服务后，再将服务请求迁移过来进行服务。第一种冷升级的方式，会导致服务中断，当所有的新服务全部启动成功后，才可以继续对外提供服务。第二种灰度升级方式，虽然在一定程度上可以避免服务的中断，但是服务请求的迁移势必带来一些服务质量的损失，服务吞吐量受到比较大的影响。 我们目前有两种升级方式： 冷升级：重启Torca进程，上面的task也会被自动迁移。这种升级方式我们一般会采用灰度方式，one by one的执行。可以保证作业不会丢，缺点是集群的所有task都会被重启一遍。在遇到集群的业务也同时变更时，可以做并发冷升级，1000台规模的集群可以在5分钟之内完成升级。 热升级：为了消除Torca冷升级对上层应用可能造成的迁移影响，ISSU提供一种微小粒度中断服务的系统升级方法，相比灰度升级会更小粒度的影响在线服务质量，在系统升级时，对已有的业务不构成任何影响，升级时间控制在毫秒级别。 在系统热升级时，首先给待升级的服务进程发送升级信号，当进程收到升级信号后，暂停服务，并采用一种扩展性和兼容性强的序列化工具来记录checkpoint。并将checkpoint写入共享内存。 服务进程调用exec系统调用，使用新的可执行文件来改变自身的进程空间。 当新的进程启动后，读取共享内存中的checkpoint数据，恢复现场，继续服务。 新的进程沿用了旧进程的进程号，以及相应的父子进程关系。且由于纯内存操作，因此中断服务质量的时间控制在毫秒级别。且对已有的业务不构成影响。 在现网运营中，我们大多数情况下采用ISSU热升级方式，用户几乎完全感知不到Torca系统的变更，对平台的bug fix，紧急特性上线等都带来了极大的便利，不需要等待大版本的发布。 4) HOT（Hadoop on Torca） 5) Web monitor 为了方便Torca用户对自己进程的观察和调试，我们开发了web monitor。这样用户就不需要的登陆到各个机器上，task的运行情况一目了然。这里列举几个视图： 1) 集群各个机器的资源情况等信息 图20. Torca机器概览. 2) 各个业务的quota以及quot a使用情况 图21. 各个业务的quota情况. 2) 集群中各类作业队列情况
图22. 作业队列. 4) 集群的配置管理 图23. 配置管理. 5) task的历史运行状况 图24.task历史资源占用. 6) coredump的task的stacktrace以及在线调试 图25.Task stacktrace. 7) 运行中的task的在线debug 图26. Task 在线debug. 8) Task的本地目录展示以及下载 图27.Task working dir. 9) Task日志的在线展示 图28.task 日志. 10) 作业的各个task的运行情况 图29. Job & task概览. 注：web monitor的很多页面都有权限管理，并非所有人都可以去remove、debug别人的进程。 6) 实时任务扩容/收缩 Torca提供了对job/task的很丰富的api，业务端可以利用这些api做各种策略，最常见的就是对任务进行扩容和缩容以及监控等等。这里可以拿HOT的例子来做介绍。因为在Torca之上的hadoop本质上也是Torca的作业，即是一种应用。 将Hadoop集群中的TaskTracker划分为两类，guaranteedTT和bestEffortTT，其中guaranteedTT是固定的部分，永远不退出，具有较高的作业优先级，占用quota。而bestEffortTT是动态伸缩的部分，优先级最低，不占用quota。当Torca集群空闲时，并且Hadoop集群资源紧张时，可以灵活扩展，提高Hadoop处理能力，当Torca集群有更高优先级job要运行，且Hadoop集群相对比较空闲时，又可以收缩，用guaranteedTT来保证Hadoop的资源需求。 这种利用了Torca client的api来做容量探测和操作的方式可以使得资源的利用得到最优化，同时，利用了Torca的自动容灾、监控、以及tns等服务，也可以使hadoop的运营更加自动和稳定。 图30. Hadoop on Torca 6. 应用情况 目前Torca已经在天津、西安、上海三地idc部署了集群，。接入的role已经达到130+。并可同时支持离线计算和在线应用。如广告的数据处理、网页搜索、日志处理、Hadoop类应用，以及在线的新闻，广告服务，下载服务以及微信的消息群发服务等等。 最大集群规模已经达到1000+，16000+核的调度能力，Torca的单个集群容量的设计目标是10000台。 图31.tjd2-websearch Torca cluster 7. Future Work 限于篇幅，本文对大部分内容都没有做展开，以后会发出各个专题文章与大家讨论，未来，Torca还有很多工作，我们将进一步关注稳定性，为业务提供稳健的平台；强化调度能力，更好的满足不同业务的调度需求；进一步优化资源管理及隔离，提高资源利用率，并且在多个资源纬度上提供管理；关注用户体验，提高系统的易用性，做更贴近用户的服务。 |
