![]()
本文为阿里云SLS徐可甲在 GIAC 2022 全球互联网架构大会 分享时的议题内容。数据总线作为大数据架构下的流量中枢，在不同的大数据组件之间承载着数据桥梁的作用。通过数据总线，可以实时接入来自服务器、K8s、APP、Web、IoT/移动端等产生的各类异构数据，进行统一数据管理，进而实现与下游系统的解耦；之后可以异步实现数据清洗、数据分发、实时计算、离线计算等计算过程，进而将结构化后的数据投递到下游的分析、归档系统，进而达到构建清晰的数据流的目的。广义上，数据采集与接入、传输链路、存储队列、消费计算、投递等都属于数据总线的范畴，整体上可以分为采集接入层、管道层、计算层。解耦生产者与消费者：消费方完全可以不用感知写入方的任何细节，降低系统对接复杂性，提升系统可靠性。
应对流量洪峰，使数据的生产异步于数据的消费，消峰填谷。
定义统一格式与操作语义：接入多种异构数据，通过数据处理构建统一的格式。
举一个简单的例子，在计算广告检索系统中，广告的点展数据至关重要。一份点展数据往往会被多方订阅消费，且应用场景各异，有精度到秒级的实时计算业务，也有类似于Hadoop的小时级或天级的批处理任务。如果数据直接对接，就需要考虑各种异常场景，会让系统变得极其复杂。而通过数据总线，可以大大降低系统复杂度，提高系统可靠性，这样可以保证任意一个数据订阅系统即使经历了下线维护或者宕机，也可以在重新上线后从之前的断点处继续进行数据处理。面对每天几百亿次读写、近百PB数据流量、万级用户的场景时，构建高可用的数据总线将会是一件非常有挑战的事情。这里简单列举一些场景的流量场景：经过几十年的飞速发展，整个开发模式、系统架构、部署模式、基础设施等也都经过了几次颠覆性的变革，这些变革带来了更快的开发和部署效率。但随之而来整个的系统与网络环境也更加的复杂、部署模式和运行环境也更加动态和不确定、接入的数据源与数据量大幅增加、流量波动等不确定因素变大、接入难度与原始结构差异化变大，这些都是云原生时代也给数据总线带来的新的要求。总结下来，云原生时代数据总线的技术挑战可以从采集接入层、管道层、计算层三方面展开。采集接入层重点关注数据源的接入丰富度、接入易用性、资源开销与数据采集可靠性；管道层重点关注网络质量、带宽与水平扩展能力、安全与隔离性、生态丰富度等；计算层重点关注及计算语法能力、流量处理带宽与扩展性等。目前行业上主流的大数据架构，大概可以分为5个部分，其中前3部分构成了数据总线。采集端：承载可观测数据采集及一部分前置的数据处理功能。随着云原生的发展，采集端也需要适应时代潮流，提供对K8s采集的友好支持。常见的采集端有Filebeat、FluentD/Fluent-bIt、Telegraf，以及我们开源的iLogtail。
消息队列：采集Agent往往不会直接将采集到的数据发送到存储系统，而是写入消息队列，起到削峰填谷的作用，避免流量洪峰导致存储系统宕机。常见消息队列为Kafka、RabbitMQ等。
计算：用于消费消息队列中的数据，经过处理、聚合后输出到存储系统。比较常见的为Flink、Logstash等。值得注意的是，也开始有公司将iLogtail开源版作为计算层部署在日志架构中。
存储分析引擎：提供采集数据持久化存储能力，并提供查询分析能力。常见的存储分析引擎为Elasticsearch、ClickHouse、Loki以及时序的Prometheus、influxdb等。
- 可视化：借助Kibana和Grafana提供采集数据的可视化能力。
用户可以利用上述的采集端、消息队列、计算三部分提供的开源组件搭建出一套数据总线。虽然基于开源组件搭建数据总线技术上时可行的，但是整体实施复杂度很高，需要维护多套系统进行协同。此外，也无法完全满足云原生上文提到的云原生场景下面临的技术挑战，例如网络质量与地域规划就是一道比较难以逾越的鸿沟。可观测平台不仅需要解决好数据如何获取、查询的问题，也应该提供具体业务场景的应用能力，帮助客户从碎片化、低信息的数据中挖掘更大的数据价值。一个典型的云原生可观测平台从下至上可以分为四个层面：数据总线、存储分析、工具、应用。其中，数据总线是整个可观测平台的数据基础，为数据分析及上层业务应用提供数据保障。正因为数据总线足够基础，所以在企业数字化过程中必须足够可靠、稳定、确保数据的通畅，并且能够弹性满足流量变化需求。接下来，我们将重点分享阿里云SLS数据总线的架构与实践。![]()
在本文第一部分，我们介绍到一个典型的数据总线可以分为采集接入层、管道层、计算层三个层次，对应的SLS的数据总线的架构也可以类似划分。采集接入层：承载了数据总线所有数据（支持Log、Metric、Trace、Event等）的接入，并且与阿里云服务有深度的集成。接入手段以SDK/API为基础，扩展了端上可观测采集器 iLogtail、数据导入服务、开源标准协议等多种接入方式。
管道层：以 LogHub 作为数据总线的核心流量中枢，功能可以完全替代 Kafka。通过 Restful API 对外提供接入服务，通过消费组的方式提供实时消费服务。
- 计算层：作为 LogHub 的下游服务，基于实时消费组，提供了各类数据处理及投递服务；并且生态上支持主流开源流计算对接。
以上，我们对SLS数据总线有了一个整体的认识，接下来我们将从数据接入、流量中枢、数据处理三个维度进行详细介绍。LogHub 作为数据总线的核心流量中枢，默认提供 HTTP/HTTPS 协议的 API 写入能力，同时，也提供了众多语言的 SDK，用以简化接入场景、增强可靠性，SDK 覆盖从Java、Go、C++等服务端应用，到Android、IOS等移动端场景，甚至 JavaScript 等前端场景。自研的开源可观测数据采集器 iLogtail承载了服务器场景、容器场景的可观测数据采集能力，覆盖Windows、Linux操作系统及X86、ARM架构。借助于阿里云的优势，无缝支持阿里云上各类主流云服务的日志、指标、安全审计类数据的采集。对于通用协议等支持也比较丰富，兼容 Syslog，Kafka、Promethous、JDBC、OpenTelemertry 等开源协议；支持众多开源采集工具，例如 Logstash、Fluentd、Telegraf 等。针对Java、Go等大数据、高并发场景下，我们在SDK基础上提供了Java Producer[1]、Go Producer[2]；针对IoT/嵌入式设备，推出了C Producer[3]。使用 Producer 相对于直接通过 API 或 SDK 发送数据，提供了更上层的封装，性能、可靠性都有大幅提升。线程安全：Producer 内所有的方法以及暴露的接口都是线程安全的。
异步发送：客户端计算与 I/O 逻辑分离。调用 Producer 的发送接口通常能够立即返回，Producer 内部会缓存并合并发送数据，然后批量发送以提高吞吐量。
失败重试：对可重试的异常，Producer 会根据用户配置的最大重试次数和重试退避时间进行重试。
优雅关闭：用户调用关闭方法进行关闭时，Producer 会将所有其缓存的数据进行发送，防止日志丢失。
高性能：通过多线程、缓存策略、批量发送等手段，有效提升发送效率。
iLogtail是数据总线数据接入的重要流量来源，拥有的轻量级、高性能、自动化配置等诸多生产级别特性，可以署于物理机、虚拟机、Kubernetes等多种环境中来采集遥测数据。iLogtail的核心定位是可观测数据的采集器，帮助开发者构建统一的数据采集层，助力可观测平台打造各种上层的应用场景。通过iLogtail可以很好的解决数据总线数据接入的问题。得益于阿里巴巴/蚂蚁集团与公有云场景的不断锤炼，iLogtail和开源Agent（例如Fluentd、Logstash、Beats）相比，性能、资源消耗、可靠性、多租户隔离、K8s支持等硬指标上较为领先，可以满足多种业务场景下的苛刻要求。目前iLogtail已于2022年6月29日完整开源，吸引了众多开发者的关注，GitHub Star数也于2022年11月7日突破1K。采集高性能、低开销是iLogtail的核心优势之一。iLogtail在Linux下使用inotify作为文件监控的主要手段，提供了毫秒级延时的数据发现能力，同时为了兼顾不同操作系统以及支持各类特殊采集场景，iLogtail同时使用了轮询作为的数据的发现方式。通过使用轮询与事件并存的混合方式，iLogtail打造了一套兼具性能优势同时不失鲁棒性的文件发现机制。此外，iLogtail 采用了无锁化事件处理模型。与业界其他开源Agent为每个配置分配独立线程/Goroutine读取数据不同，iLogtail数据的读取只配置了一个线程。由于数据读取的瓶颈并不在于计算而是磁盘，单线程足以完成所有配置的事件处理以及数据读取。使用单线程使得iLogtail的事件处理和数据读取都可以在无锁环境下运行，数据结构更加轻量化，从而取得了相对多线程处理更优的性价比。在生产环境中，一台服务存在数百个采集配置属于常态，每个配置的优先级、日志产生速度、处理方式、上传目的地址等都有可能不同，因此必须有效解决如何隔离各种自定义配置，保证采集配置QoS不因部分配置异常而受到影响的问题。iLogtail采用基于时间片的采集调度、多级高低水位反馈队列、事件非阻塞处理、流控/停采策略以及配置动态更新等多项关键技术，融合实现了兼具隔离性、公平性、可靠性、可控性、性价比五大特性的多租户隔离方案。经历了多年双11流量高峰期的考验，这套方案已经被证明相比其他开源具备较大的稳定性和性价比优势。数据源的多样性毫不夸张的说可以是数据总线的生命线，否则就是巧妇难为无米之炊。iLogtail通过插件化的设计，突破了单纯文件采集的范畴，有效扩展了上下游生态，使得iLogtail成为真正的可观测采集器。目前iLogtail已经支持了众多数据源的接入，数据源类型覆盖Log、Metric、Trace，数据源除了文件的采集，还包括了标准协议的支持，例如HTTP、Mysql Binlog、Prometheus、Skywalking、Syslog等；iLogtail也通过eBPF支持，实现了无侵入的网路数据采集能力。数据输出生态也从SLS逐步扩展到Kafka、gPRC等，未来通过开源社区共建也会支持ClickHouse、ElasticSearch等。面对众多异构数据的接入，数据总线一项职责就是通过数据处理构建统一的数据格式。iLogtail也提供了强大的数据处理能力，可以前置完成数据格式规整、数据过滤、上下文关联等。iLogtail整体采用PipeLine的设计，首先通过Input插件采集到数据，会经过采集配置中设定的Processor处理，之后经过Aggregator插件打包，最终通过Flusher发送到存储系统。数据处理环境包含数据切分、字段提取、过滤、数据增强等环节，所有插件可以自由组合。随着云原生落地，Logtail已全面支持Kubernetes，目前支持Docker、Containerd两种主流的容器运行时。iLogtail通过实时监控容器列表并维护容器和日志采集路径映射，结合高效的文件采集能力提供了极致的容器数据采集体验。iLogtail支持使用容器标签、环境变量、K8s标签、Pod名称、命名空间等多种方式进行容器筛选，为用户提供了便利的采集源配置能力。支持DaemonSet、Sidecar、CRD等多种部署方式，为应对不同使用场景提供了灵活的部署能力。此外，对于CICD自动化部署和运维程度要求较高的用户，iLogtail还提供了K8s原生支持，支持通过CRD的方式进行采集配置管理。该方案中，iLogtail K8s新增了一个CustomResourceDefinition扩展，名为AliyunLogConfig。同时开发了alibaba-log-controller用于监听AliyunLogConfig事件并自动创建iLogtail的采集配置，进而完成日志采集工作。LogHub作为SLS数据总线的流量中枢，是一个高吞吐的数据通道，支持海量可观测数据的实时接入与消费能力。在可观测数据场景下完全可以Kafka等消息队列产品，并且在性能、易用性和稳定性上更佳。LogHub可以理解为 append-only 的 log 队列结构，通过多个 shard 组合实现 IO 和存储的水平扩展。并且在队列基础上创建多层索引，可快速定位到每一条数据在队列中的位置，赋予队列模型随机查询能力。Logstore/Metricstore是SLS可观测数据的采集、存储和查询单元，LogHub作为Logstore/Metricstore的队列模型，提供了数据实时写入和流式消费的能力。同时在此基础上，进行了模型扩展，进而构建出了统一的可观测分析引擎。
LogHub 是阿里云上的基础设施，借助阿里云等全球化部署优先，与阿里云Region保持同步，这也是其他开源数据总线无法比拟的优势。可以确保全球化业务就近选择Region，也可以满足一些国家或组织相关的法律约束数据不能出境的要求。LogHub 联合 CDN 推出了全球自动上传加速方案，基于阿里云CDN硬件资源（覆盖2800+节点，70+国家），全球数据就近接入边缘节点，通过内部高速通道路由至LogHub，大大降低网络延迟和抖动。在生产中我们往往会面临流量峰值和低值的情况，也会遇到因业务层映射不均衡，导致某一个分区（shard）有非常大流量的场景，弹性伸缩（Merge/Split）就是解决该问题的机制。Shard分裂的原理
Shard分裂操作，是流量扩容的一个重要手段，是把一个readwrite的Shard分裂成两个readwrite的Shard；同时原Shard变成readonly状态，之上的数据可继续被消费读取。合并操作和分裂操作相反，是把两个相邻的readwrite的Shard合并成一个readwrite Shard，同时原来的两个Shard变成readonly状态。负载均衡：根据峰值、底值弹性扩容，控制成本。
下图最早只有Shard0提供服务；后随着晚间流量高峰单Shard不足以支撑业务流量，Shard0拆分为了Shard1、Shard2提供服务；等流量再次稳定，出于成本考虑，合并成一个新的Shard3提供服务。日志保序处理：将不同实例映射到不同分区，调整分区处理能力。
对于一些业务场景有保序的需求，数据写入时往往通过Hash规则，将不同业务的数据映射到固定的Shard上。下图中的业务场景，Shard1承载了3个DB的流量不堪重负，此时可以通过进行Shard拆分的方式，实现流量的均衡。拆分后，原来的Shard1变成只读，数据仍可被消费，但是不再接收新的数据写入，该由新拆分出的Shard3、Shard4对外提供服务。那对于Shard调整前后的边界数据，如何保序？Loghub提供了Shard顺序消费的能力，即Shard分裂后，先消费原Shard数据（即Shard1的数据），然后同时消费由该Shard分裂的Shard数据（Hash策略保证同一业务落到一个Shard上）。同理，Shard合并场景，也是先消费原Shard数据，然后消费由原Shard合并后的新Shard数据。当然，对于不严格保序的场景，为了提升消费速率，可以关闭Shard顺序消费功能，以达到所有Shard可以同时消费的目的。粗粒度流控：Project级别
Project全局流控最主要的目的是限制用户整体资源用量，在前端就拒绝掉请求，防止流量穿透后端把整个集群打爆。细粒度流控：Shard级别
Shard级别流控则更加精细、语义（错误码）更加明确、可控性更强。每个Shard明确定义处理能力， 如5MB/sec写入，10MB/sec的读取。
当Shard队列出现堵塞，根据Shard流量是否超过quota，返回用户是限流还是系统错误（返回的Http错误码是403还是500），同时将Shard限流信息通知QuotaServer。
QuotaServer接收到限流信息后，可瞬时将限流信息同步至所有的Nginx。
- Nginx端获得Shard的流控信息之后，对Shard进行精确的流控。
从上文数据接入部分我们可以看到，数据总线作为流量中枢，承载着各种异构数据的接入。虽然 iLogtail 具备较强的前置数据处理能力，但是往往会有一些数据因为历史原因或者接入方式的差异导致了数据格式可能千差万别，难以做到统一的格式规范，给后续分析带来了极大的困难。因此数据总线往往也需要一个数据处理环节，为此我们提供了以数据加工服务和定时SQL（Scheduled SQL）为主的数据处理能力。此外，为了更好的支持开源生态，也支持通过自定义消费或者Flink流式消费的能力。![]()
规整：这是使用频率最高的场景，比如从文本日志数据中提取出关键信息，将其转为规范化的数据。
富化：比如用户的点击数据中只包含商品ID，在分析时需要从数据库关联起详细信息。
脱敏：随着我国信息安全相关法律的完善，对于敏感数据（比如个人信息）的处理要求越来越高。
分裂：在数据写出时，出于性能、便捷性考虑会将多条数据合并输出，在分析前则需要拆分出独立数据条目。
- 分发：将不同类型的数据分别写到不同的特定目标中，以供下游定制化使用。
整体架构
数据加工服务基于实时消费组实现从源Logstore中负载均衡的消费数据，源Logstore的Shard数决定了数据加工任务的并发度上限。当数据加工调度器启动一个新的作业时，会自动创建并绑定到源 Logstore 的一个消费组，消费组内部的多个消费者独立负责处理不同Shard中的数据。随着数据量增多，源Logstore需要拆分出更多的Shard，同时数据加工作业便可以扩展更多的消费者独立工作。当作业需要关联外部资源的时候，每一个消费者独立维护一份资源的拷贝，实现高性能关联计算。弹性机制
日志数据除了数据量巨大，另一个特点是数据量呈周期性波动，而且波动波峰极高极窄。比如对于直播平台，其一天的90%流量都来源于 21:00至23:00 这一个休闲时段，这就导致这一时间段的数据量是平时的数十倍。面对这样的极端场景，数据加工就需要能够实现计算力的弹性伸缩，保障用户作业高性能运转的同时，尽可能减少资源浪费。
数据加工在任务调度中的弹性伸缩是基于LogHub的消费组和K8s的HPA机制实现的。在系统内部通过存储计算分离实现数据加工计算力自由伸缩，在用户侧看到的则是服务化计算平台，按量付费，无需关心繁杂的细节。原生的K8S HPA内置仅支持CPU和内存两个指标，这对于绝大多数数据密集型作业来说已经足够了。但是在数据加工中，某些客户会对数据编排，进行跨地域数据流转，面对这类网络/IO敏感场景，内置的HPA指标就无法满足。所以我们引入了更全面的HPA指标，以支撑多种多样业务场景下的作业需求。除此之外，我们还将使用智能算法对HPA指标进行续升级优化， 比如基于作业的历史运行特性，提前做资源分配，更进一步提升作业运行的稳定性。 Cloud Data Processing Language（DPL）
数据加工服务基于Python语法设计用户侧接口，并提供完善的内置数据处理函数，这里我们将其称为 Cloud Data Processing Language (DPL)。借助DPL，可以通过极少数代码完成非常复杂的数据处理逻辑。以下是一个信息富化的例子：在http请求中，通过RDS维表的关联，将请求状态码（http_status字段）的详细描述（http_code_desc字段）添加到原始数据中。基于时间的数据（日志、指标）在日积月累后的数量是惊人的。以 Nginx 访问日志为例，每一个HTTP/HTTPS 访问请求会记录一条 access log，假设每天产生1000万条数据，则一年为36亿条数据。长时间大量的数据存储，不仅会占用大量的存储空间，对数据数据分析处理也会造成极大的性能压力。Schedule SQL可以对数据进行定时的聚合分析处理（支持标准SQL、SLS 查询和分析语句），按照调度规则周期性执行，并将运行结果写入到目标库中。相比较与自建调用API方式，具备以下优势：云上服务（例如OSS、SLB等）往往是由云厂商托管，用户是无法直接登陆服务器进行日志查看的。数据分派技术就是基于SLS的数据总线技术实现的关键服务模块，在用户授权的前提下，实现将可观测数据从云服务侧投递到各个用户侧的目的。实现机制为云产品服务侧通过预先的采集、清理（处理）流程，将日志存入集中的数据中枢（LogHub），再通过分发服务，将数据以多租户的形式分发到用户侧。其具备以下优势：数据分派技术只是解决了可观测数据从云产品服务侧到用户侧的流转问题，但是要实现云上服务日志的采集，仍然需要云资产归属、多账号体系认证等需要一些上层的服务来维护。为此，我们构建了多个基础服务，包括资产同步、自动化采集服务、多账号体系等，统称为统一接入服务。之后，基于统一接入服务，进一步针对各个云产品提供统一的交互与功能组件，构建了更上层的APP能力 -- Alibaba Cloud Lens。Alibaba Cloud Lens 基于 SLS 构建统一云产品可观测能力，提供阿里云云产品的用量分析、性能监控、安全分析、数据保护、异常检测、访问分析等服务。从成本、性能、安全、数据保护、稳定性、访问分析六个纬度， 提供对云产品的运维辅助分析能力，有效降低云产品可观测的门槛。采集链路自动编排是Cloud Lens的一大特色。通过图形化的方式，就可以实现一键式、自动化的资产发现与自动化数据采集（到用户侧），其背后实际是是依赖于DSL代码实现采集逻辑的编排。未来已来，SLS数据总线技术也会在采集接入、管道、计算三个层次不断耕耘，为可观测平台持续不断的提供稳定可靠的数据基础。附录：
SLS ：打造可观察性统一引擎
五年双十一：SLS数据管道发展之路
数字IT基础-数据采集总线：https://developer.aliyun.com/article/672576
使用日志服务LogHub替换Kafka：https://developer.aliyun.com/article/35979
日志服务（原SLS）新功能发布(1)--支持保序写入和消费：https://developer.aliyun.com/article/5581
日志服务（原SLS）新功能发布(2)--弹性伸缩(Merge/Split)：https://developer.aliyun.com/article/277425
参考链接：
[1]https://github.com/aliyun/aliyun-log-java-producer
[2]https://github.com/aliyun/aliyun-log-go-sdk/tree/master/producer
[3]https://github.com/aliyun/aliyun-log-c-sdk