转自：eBayTechRecruiting

基于Kafka/Storm的实时流量控制系统

量大货足5000字，超级技术干货

认真阅读10分钟，给你十分收获

这篇干货，你能看到什么？

流量控制对于保证Web服务的安全性和可靠性至关重要。在安全性方面，需要阻止黑客频繁访问某些API而获取大量信息。在可靠性方面，任何服务在有限资源的情况下能处理的TPS都有上限。如果超过上限，Service的SLA会急剧下降，甚至服务不可用。根据队列理论，越多的流量，就会导致更多的延迟。所以为了保证Service的SLA，必须进行流量控制。本文介绍了一个基于Kafka和Storm的 异步通用的流量控制方案；同时描述了如何根据数据倾斜程度来自动切换处理流程，以确保系统灵活性和延展性。最后，性能测试结果验证了该方案在高吞吐量时也能将计算延迟控制在6ms左右。

“许云博士，eBay中国研发中心平台架构部门安全领域负责人，同时也是eBay网站流量控制系统（Rate Limiter）和证书管理系统(CMS)的开发先锋，致力于产品架构，算法，大数据实时处理等研究。”

流量控制有话说

流量控制主要目的是用来限定Server所能承载的最大（高并发）流量峰值，以免在峰值时Server过载而宕机。对于WEB系统而言，通常是分布式部署，如果请求并发量很大，会导致整个集群崩溃，也就是通常所说的“雪崩效应”。所以不仅在网络代理层面（比如nginx）设置流量控制以抵抗、拒止溢出流量，还应该在App Server层面有一定的自我保护策略，确保当前JVM的负载应该在可控范围之内，对于JVM承载能力之外的请求，应该被合理管理。在安全和业务方面App也会需要限制访问频率，例如:

一个IP每天限制创建20个帐号

 

一部手机每天只允许5次信用卡失败交易

  一个大客户每天只能访问某个API 10M次

eBay利用Rate Limiter设置各种策略来防止API被过度访问以及防御DoS攻击。eBay要求Rate Limiter不仅可以基于IP进行流量控制，同时也要求支持对App、用户、设备号等进行流量控制。eBay Rate Limiter的基本需求如下：

   1.滑动窗口

Fix Window简单有效，但有两个致命的弱点：

（a）用户如果将计算窗口 resize，例如从10分钟变成1分钟，Fix Window只能清0重新计数，因为完法区分最近1分钟访问了多少次。

（b）如果限制1分钟只能访问100次，用户可以 在第一个窗口的最后10秒访问99次，然后在下一个窗口的前10秒访问99次，也就是说用户在20秒内访问了198次而不触发policy。

滑动窗口可以有效解决  Fix Window的不足。

2.多计数窗口

App经常需要进行多窗口计数。比如登录要求限制1分钟内3次，1小时内20次，1天内50次。

3.Flexible Policy

支持复杂灵活的Policy。比如可以设置某个API的访问比例不能超过这个服务所有API访问次数的50%。

4.有效期

Policy一旦被触发并返回Block或Captcha后，在生效期间内所有请求都将被Block或要求输入验证码。

5.验证码服务

对于Web App, Rate Limiter可以返回验证码而不是Block请求。这样 WEB App不需要单独与验证码服务进行集成。比如eBay的Sign in界面，由Rate Limiter来决定是否需要让用户输入验证码。

目前处理Rate Limiter的常用方式

    目前常用的Rate Limiter算法有两种。一种是Token Bucket，按指定的方式向Bucket里添加token，每个请求消费一个token，如果没有token则请求被拒绝。另一种是Leaky Bucket，用户请求都会先存放在Bucket中，然后Bucket控制流出量。如果Bucket满了，则请求被拒绝，这个算法具有流量整形的功能。然而这些算法不能直接应用于ebay Rate Limiter中。因为ebay的策略充许一个Resource关联多个Policy，同时一个Policy也可能引用多个Resource。目前常用的RateLimiter方法有Google Guava中的Rate Limiter和Redis。

2.1 Guava Rate Limiter

Guava Rate Limiter是使用最多的第三方Class。但这个是单机版的，只能在一个JVM里有效，不能同时针对一个pool里多台机器。也不支持复杂的policy和及多窗口计数。

2.2 Redis

Redis作为distributed cache，提供了原子性的计数功能。所以很多解决方案都结合使用Redis的计数和cache过期这两个功能。但是Redis如果要实现复杂的policy及多滑动窗口计数需要很多额外工作，最重要的是增加了很多远程访问，导致延迟大幅增加。基于eBay对于Rate Limiter的需求，这两种方案并不适用。

eBay如何解决Rate Limiter痛点

为了解决eBay Rate Limiter痛点，本文提出了可扩展的Cloud Native解决方案。该方案使用Apache Storm进行大数据实时处理 。该方案有3个重要前提：

1、通用解决方案，对于所有的HTTP APP都可以使用。

2、对于Policy触发的阈值不要求严格匹配。比如Policy定义1分钟只允许访问  1000次，APP接受在第1003次的时候才开始Block。对于秒杀这种业务场景，就   不适合这条假设。

3、对用户请求造成的延迟小于3ms。

3.1 Architecture

基于上述前提，eBay Rate Limiter使用了异步处理方案。如图1所示，Rate Limiter包含如下主要模块：

Fig 1: Architecture of Next Gen RateLimiter

SMC用户使用SMC创建和编缉流量控制策略。

Instrument  运行在用户的Pool中，将每次用户的URL请求转换成RateLimiter Event发送给Rate Limiter Service。

RateLimiter Service 将收到的Event匹配Blacklist和Whitelist，如果匹配成功，则直接返回用户。如果匹配不成功，则计算Event需要评估哪些policy，然后将相应的信息插入Kafka中；同时从cache中查找对应的状态并返回给用户。

RateLimiter  BackEnd  Kafka Spout消费Event, 通过normalize来确定需要哪些计数窗口，然后经过 metering和decision进行计数和判断状态。最后将结果双写到Kafka和Remote cache。

3.2 RateLimiter Service

RateLimiter Service 是Raptor Ginger 应用服务，用于分析 Event匹配哪些Policy。如果Event匹配Whitelist或Blacklist，则直接返回结果，否则计算这个Event需要计算哪些policy，然后将相应的信息保存在Kafka的ratelimiter-event topic中。同时Service后台有个线程会从Kafka的ratelimiter-result topic消费Backend计算结果并更新Local cache。Service从Local cache获取Event对应的decision信息并返回，如果没有从本地cache中获取到decision，则从remote cache中获取decision。

3.3 RateLimiter Backend

RateLimiter Backend 是apache storm程序。如图1所示，它主要包含 Policy spout, Kafka spout, normalizer bolt, metering bolt, decision bolt and Kafka bolt。

Policy spout 初始化时加载所有的policy并且增量加载变化的policy。将policy信息发送给所有的normalizer和decision bolt。

Kafka spout 消费Kafka中的ratelimiter-event topic, 然后将Event消息发送给normalizer  bolt。

Normalizer bolt 将Kafka中的消息解析成rate limiter event对象，然后根据对应的policy计算出所有的计数窗口。

Metering bolt 支持多窗口的sliding window与fix window计数。

Decision bolt 根据metering与policy信息进行decision计算。

Kafka bolt 将计算结果保存在Kafka的ratelimiter-result topic中。

3.3.1Metering

Metering同时支持fix window和sliding window。通过cache将sliding window的算法复杂度从O(n)优化到O(1)。假设某个policy需要对每个IP分别进行15秒与30秒的计数，并且每个time slot是5秒，则 Tc表示当前的time slot。如公式所示：

      比如，表示15秒的计数结果； 表示30秒的计数结果。

图2到图5显示了如何使用sliding window计算多窗口的4个连续状态。

根据上述公式，如果需要对图2中15秒与30秒进行计数，则分别使用如下公式：

上述公式计算某个窗口需要循环累加所有time slot的值，它的计算复杂度是 O(n)。假设一个time slot是5秒，则24小时的计数窗口则需要每次累加17280个 time slot。为了降低计算复杂度，缓存除了当前slot的数量之和。则上述公式可变换成计算复杂度为O(1)的公式：

3.3.2Decision

Decision Engine根据metering的信息来判断Event对应的policy是否应该触发。图6显示了Decision Engine的工作流程。如果Event已经触发了一条policy，并且生效时间还没有过期，则直接返回decision信息；否则，按优先级逐条判断相应的policy是否满足条件。如果满足条件，则返回decision信息并且设置policy的有效期。当有效期结束后，需要根据最新的metering信息重新判断policy是否满足条件。

这个流程中最耗时的是计算policy中的Boolean表达式。有三种可选方案，它们分别是JavaScript Engine, J2V8和JEval。JDK1.7的JavaScript Engine基于Mozilla Rhino，而JDK1.8则基于Oracle Nashorn。J2V8则是基于Google的 JavaScript Engine V8，通过不同OS的动态链接库，可以运行在Windows、Linux and Mac OS。JEval则是一个高性能，专门处理数学与Boolean表达式的解析器。它并不支持JavaScript语言，但相比复杂的JavaScript Engine具有明显的性能优势。表1列出了不同Engine之间的性能比较。

Fig 6: Decision flow

3.3.3Process flow

图7显示了Rate Limiter处理普通 traffic的流程。图中每个小方块表示一个Rate Limiter Event，相同颜色表示相同的Event。黑色箭头表示随机发送，彩色箭头表示分组发送。

Backend配置了32个Kafka spouts，其数量等于ratelimiter-eventtopic的partition 数量。Kafka spout将收到的Event消息随机发送给normalizer bolt，因为normalizing是无状态操作。相比其它的分组策略，随机分配具有最好的性能。因为metering与decision是有状态的操作，所以相同的Event必须发送给同一个metering与decision blot。每个Event会立即触发 metering and decision计算以达到最低的延迟。

Fig 7: Process flow for normal case

当数据倾斜很严重时，对上述处理流程是很大的挑战。假设某个IP大量访问同一个resource，则大量的Event会发送到同一个metering和decision bolt，从而导致某个JVM很忙，而其它很空闲。为解决数据倾斜的问题，引入了新的处理流程。如图8所示，与普通流程最大的不同在于多了一个metering aggregation bolt，并且相同的Event可以被发送到一组metering bolt而不是一个bolt。最重要的是，meteringbolt不会立即发送metering计数结果，而是窗口移动时批量下发计数结果，然后由aggregation bolt进行汇总。如果一组metering bolt是10个，而窗口的time slot是5秒的话，那么不论流量有多少，在aggregation bolt上每5秒只会有10个请求。它的本质思想是牺牲了延迟，达到高吞吐量来应对数据倾斜。

系统的后台会实时检测每类Event的流量，并自动进行切换处理流程。对于普通的case追求低延迟，对于attack case追求高吞吐量，达到自我保护的目的。

Fig 8: Process flow for attack case

一次有趣的实验

为了验证该方案的性能，搭建了Backend的LnP测试环境。表2列出了 storm结点的VM信息。

在LnP测试中，前1小时按10K TPS的流量插入Event到Kafka，后30分钟按20K TPS的速率插入Event到Kafka。由于Kafka容量的限制，LnP测试环境不能插入太多的Event。测试中分别收集了三个阶段的性能指标，分别是normalizer，metering和decision。如图9所示, 第一个指标显示normalizing平均耗时在1ms以下；第二个指标显示前两个操作（normalizing和metering）完成后的平均 耗时在3ms左右；第三个指标显示所有操作完成的平均耗时在6ms左右。这其中还包含了bolt之间的网络开销。从LnP测试结果可以推断出，两个supervisor结点可以至少处理20K TPS的Event，因为从10K TPS增加到20K TPS，延迟完全没有增加。

Fig 9: Performance of RateLimiter backend

eBay Rate Limiter是一个基于Kafka和Storm，满足流量控制需求的异步通用方案。它支持多计算窗口和复杂的policy。系统会实时检测每类event的流量，并自动进行切换处理流程。对于普通的case追求低延迟，对于attack case追求高吞吐量，达到自我保护的目的。