xql的流量治理

[cjuexuan]

引子

已经很久没有更新过博客了，一方面的原因是事情变多了，但更主要的原因是人变懒了。不过从这个8月开始，我还是会继续不定期的更新下我的博客 这个月就从xql开始说起

背景

我大概是2016年入职喜马拉雅没多久就开始做xql了，到现在已经过了3年多了，xql也成功的走到了第3个大版本。 之前其实已经分享过xql的一些设计， 但xql也一直没停止前进，所以来分享下xql到3.0为止，我们还在做的一些有趣的事情。今天我会分享下流量治理相关的内容，后续还会分享一些其他功能点

如何治理流量

多实例架构

之前我们在xql1的时候其实是一个单体的应用，有着一个proxy和一个engine，这两者的通信是基于akka remote做的，这个架构大概跑了有1年左右。 随着用户规模的增加，我们也给自己提了更高的要求，比如我的发布部署应该最小程度的影响用户，再比如大家都知道spark应用其实不一定非常稳定，有可能还是会因为一些sql把 整个yarn上的application跑死，这种情况下我们都希望最好用户都是无感知的。所以到了2017年我们就开始做xql的集群，得益于之前akka的抽象，我们也很自然的将我们的协议从 akka remote升级到akka cluster，这样就拥有了多个计算实例，而且可以做到弹性的，在我们检测到一个实例（对应一个spark的application）有问题后，我们会自动的摘掉这个实例， 将运行在它上面的流量动态的分配到其他节点，最后也可以自动的起一个新的健康的节点来顶替掉之前的有问题的节点。

流量的分配

做完这些之后，xql有了初步的高可用，不过新的挑战也来了，因为有了多个实例，自然会涉及到流量的分配问题。最开始17年刚上线的时候，我们其实也比较简单，用的就是round robin。 没过多久，我们就发现这样分配还是非常不好的，因为每一个sql的运行时间都不同，有快有慢，如果用轮询进行分配，会导致有些流量分配常常出现一些不均衡，所以我们很快的又用上了最小在途请求。

这样大概跑了半年左右，用户反馈xql跑的变慢了，web页面提交的ad-hoc查询要等一会才能响应，我们又开始想新的优化思路了。这里提一下，xql的流量其实有几个来源，一个是用户在web页面上提交的查询请求，这类请求用户希望最快时间的给他返回结果；一个是来自api的调用，比如我们的调度系统将etl的任务也会打过来，还有些application里面有访问数据需求，他们也会调用我们的client，将请求打过来，这一类请求因为是程序调用，特别是etl任务的情况下，用户其实不太关心是不是能马上给他跑，他只要最终在规定的时间内能跑完他的任务就可以了。所以我们从18年年初开始就做了一个读写分离的功能，我们把web用户的query请求标记成读流量，把etl的任务标记成写请求，然后在后台对我们的engine也打上了一些读写的标签。一个sql只能跑在标签能匹配的实例上。另外由于夜间其实没有什么读的请求（因为没有来自web页面的查询了2333），但有着非常多的etl请求，所以我们为了提高我们的资源利用率，夜间会关闭这个强隔离的策略

这样又跑了大半年之后，我们发现有些大的读请求也会导致整个实例的load比较高，影响到这个实例上的其他请求。这一类请求的有非常多的都是读了一个很大跨度的分区表，比如读一个月或者更久的数仓底表。另外我司开发在公司环境查询线上的mysql也是走xql，这一类请求都是响应很快的。针对这些场景，我们决定进一步细化我们的流量分配策略，首先我们做了xql的analyzer，来分析分析用户的sql是否带有了分区条件，并且分区条件的区间内的数据量是否在我们允许的范围内，如果不带分区条件或者超过了我们允许的范围，我们就会将此类sql标记成bigsql；另外我们还标记出了那些只访问jdbc数据源的sql，标记成jdbc；接着我们考虑到用户的调用来源，将流量区分成web和script；最后我们充分利用这一类标签信息来作为我们的流量划分依据，这个时候我们也做到了不同场景的实例有着不同的规格。做完这些之后，整个xql又顺滑了很多

路由选择的过程

我们将路由的策略分成filter和selector，路由过程中我们有n个filter和1个selector

filter会从候选集中过滤掉不满足条件的实例，比如JobIsolationFilter会过滤掉那些标签和sql标签不匹配的实例，MinimumRunningJobFilter会过滤那些请求比较多的实例

selector会从最后满足条件条件的候选集中选择1个实例

具体的流程是我们会先走完所有的filter，如果此时还有实例在，就让selector从里面最多选出1个实例来最终提交

trait RouteSelector extends RouteStrategy {

  def select(engines: List[EngineRefExt], job: ExecutableJob): Option[EngineRefExt]

}

trait RouteFilter extends RouteStrategy {

  def enable: Boolean = true

  def filter(chainData: ChainData): ChainData
}

class ChainData(context: ChainContext, job: ExecutableJob, candidates: List[EngineRefExt])

class RouteStrategyChain(filters: List[RouteFilter], selector: RouteSelector) {

  def select(engines: List[EngineRefExt], job: ExecutableJob): Option[EngineRefExt] = {
    @tailrec
    def loop(chainData: ChainData, filters: List[RouteFilter], selector: RouteSelector): Option[EngineRefExt] = {
      filters match {
        case head :: tail ⇒
          val newChainData = if (head.enable) {
            head.filter(chainData)
          } else {
            chainData
          }
          if (newChainData.candidates.nonEmpty) {
            loop(newChainData, tail, selector)
          } else {
            None
          }
        case Nil ⇒
          selector.select(chainData.candidates, chainData.job)
      }
    }
    val rpcContext = ChainContext.init()
    val chainData = ChainData(rpcContext, job, engines)
    loop(chainData, filters, selector)
  }
}