Shulou(Shulou.com)06/01 Report--

It is believed that many inexperienced people are at a loss about how Uber in open source Chaperone carries out end-to-end audit of Kafka. Therefore, this article summarizes the causes and solutions of the problem. Through this article, I hope you can solve this problem.

As the scale of the Uber business continues to grow, our system continues to generate more events, messages and logs between services. This data needs to go through Kafka before it can be processed. So how does our platform audit this data in real time?

To monitor the health of the Kafka data pipeline and audit every message that flows through Kafka, we rely entirely on our audit system, Chaperone. Since Chaperone became Uber's cross-data center infrastructure in January 2016, it has processed trillions of messages every day. Next we will describe how it works and explain why we built Chaperone.

Overview of Kafka data pipeline of Uber

Uber's services run in multiple data centers in a dual-active mode. Apache Kafka and uReplicator are message buses that connect various parts of the Uber ecosystem.

An overview of Uber's Kafka data pipeline as of November 2016. Data is aggregated from two data centers to a Kafka cluster.

It is difficult to get Uber's Kafka to respond instantly to downstream consumers. To ensure throughput, we use batches as much as possible and rely heavily on asynchronous processing. The service uses its own client to publish messages to the Kafka broker, which forwards the messages to the local Kafka cluster in batches. Some Kafka topics will be consumed directly by the local cluster, while most of the rest will be combined into an aggregated Kafka cluster with data from other data centers. We use uReplicator to do this work for large-scale streams or batches.

Uber's Kafka data pipeline can be divided into four layers that span multiple data centers. The Kafka agent and its client are the second layer and the * layer, respectively. They are used as gateways for messages to enter the third layer, that is, the local Kafka cluster for each data center. Part of the data from the local cluster is copied to the aggregation cluster, that is, the * layer of the data pipeline.

The data of the Kafka data pipeline will be batched and confirmed (send confirmation):

An overview of the path through which data flows through the Kafka data pipeline.

The flow of Uber data from the proxy client to Kafka takes several stages:

The application sends a message to the proxy client by calling the produce method of the proxy client.

The proxy client puts the received message in the client's buffer and lets the method call return.

The proxy client batches the messages in the buffer and sends them to the proxy server.

The proxy server puts the message in the producer buffer and acknowledges the proxy client. At this time, the message batch has been divided into areas and placed in the corresponding buffer according to the name of the topic.

The proxy server batches the messages in the buffer and sends them to the local Kafka server.

The local Kafka server appends the message to the local log and acks=1 the proxy server.

UReplicator gets the message from the local Kafka server and sends it to the aggregation server.

The aggregation server appends the message to the local log and acks=1 the uReplicator.

We made some tradeoffs to get Kafka to support high throughput. Thousands of microservices using Kafka to handle hundreds of thousands of concurrent traffic (and growing) pose potential problems. The goal of Chaperone is to catch every message in every stage of data flow through the data pipeline, count the amount of data in a certain period of time, and accurately detect the loss, delay and repetition of data as soon as possible.

Overview of Chaperone

Chaperone consists of four components: AuditLibrary, ChaperoneService, ChaperoneCollector, and WebService.

Chaperone architectures: AuditLibrary, ChaperoneService, ChaperoneCollector, and WebService, which collect data, perform calculations, automatically detect missing and delayed data, and display audit results.

AuditLibrary implements the audit algorithm, which periodically collects and prints statistical time windows. This library is dependent on the other three components. Its output module is pluggable (can use Kafka, HTTP, etc.). In the agent client, audit metrics are sent to the Kafka agent. In other layers, metrics are sent directly to specific Kafka topics.

The audit algorithm is the core of AuditLibrary, and Chaperone uses a 10-minute scrolling window to continuously collect messages from each topic. The event timestamp in the message is used to determine which time window the message should be placed in. For messages in the same time window, Chaperone calculates their number and p99 latency. Chaperone periodically wraps the statistics for each time window as audit messages and sends them to the pluggable backend, which may be Kafka agents or the Kafka server mentioned earlier.

Chaperone aggregates messages into scrolling time windows based on their event timestamps.

The tier field in the audit message is important to know where the audit took place and whether the message has reached a certain place. By comparing the number of messages between different layers in a certain period of time, we can know whether the messages generated during this period have been successfully delivered.

ChaperoneService is a component of the workload and is always in a state of hunger. It consumes every message from Kafka and records a timestamp. ChaperoneService is built on uReplicator's HelixKafkaConsumer, a consumer component that has proven to be more reliable and easier to use than Kafka's native consumer component (Kafka 0.8.2). ChaperoneService records the status by periodically generating audit messages to specific Kafka topics.

ChaperoneCollector listens for specific Kafka topics, fetches all audit messages, and stores them in the database. At the same time, it also produces multiple dashboards:

The dashboard created by Chaperone, from which we can see the data loss.

From the figure above, you can see the total number of topic messages for each layer, which is obtained by aggregating messages from all data centers. If there is no data loss, all the lines will overlap. If there is data loss between layers, there will be cracks between lines. For example, as you can see from the following figure, the Kafka agent has lost some messages, but no messages have been lost in subsequent layers. The time window of data loss can be easily seen from the dashboard, so that action can be taken accordingly.

The delay of messages can also be seen from the dashboard, so that we can know the timeliness of messages and whether they have transmission delays at some layers. Users can see the health of the theme directly from this dashboard without having to check the Kafka server or uReplicator dashboard:

Chaperone provides an one-stop dashboard to view the topic status of each data center.

WebService provides a REST interface to query the metrics collected by Chaperone. Through these interfaces, we can accurately calculate the amount of data lost. After knowing the time window of data loss, we can look up the exact number from Chaperone:

Chaperone's Web interface.

Two Design goals of Chaperone

When designing a Chaperone, in order to achieve an accurate audit, we focused on two tasks that must be done:

1) each message is audited only once

To ensure that each message is audited only once, ChaperoneService uses a prewritten log (WAL). Each time ChaperoneService triggers an Kafka audit message, it adds a UUID to the audit message. This message with the associated offset is saved in the WAL before being sent to Kafka. After being confirmed by Kafka, the message in WAL is marked as completed. If ChaperoneService crashes, it can resend unmarked audit messages in WAL after restart, locate the most recent audit offset, and then continue to consume. WAL ensures that each Kafka message is audited only once and that each audit message is sent at least once.

Next, ChaperoneCollector uses the UUID that ChaperoneService added earlier to remove duplicate messages. With UUID and WAL, we can ensure that the audit is one-off. It is difficult to achieve an one-time guarantee on the proxy client and server side because it gives them additional overhead. We rely on their elegant closing operation so that their state can be washed out.

2) use consistent timestamps between tiers

Because Chaperone can see the same Kafka message in multiple layers, it is necessary to embed a timestamp for the message. Without these timestamps, time misalignment will occur when counting. In Uber, most of the data sent to Kafka is either encoded in avro-style schema or in JSON format. For messages encoded with schema, you can get the timestamp directly. For messages in JSON format, the JSON data needs to be decoded to get the timestamp. To speed up this process, we implemented a stream-based JSON message parser that scans the timestamp without decoding the entire message in advance. This parser is very efficient in ChaperoneService, but it still costs a lot for proxy clients and servers. So in these two layers, we use the processing timestamp of the message. Differences in counting between tiers due to inconsistent timestamps may trigger an incorrect warning of data loss. We are working to resolve the timestamp inconsistency and will publish the solution later.

Two uses of Chaperone in Uber

1. Detect data loss

Before Chaperone, the signs of data loss came from data consumers who would complain about data loss. But by the time they come out to complain, it's too late, and we don't know which part of the data pipeline is wrong. With Chaperone, we create a job to detect missing data, which periodically pulls metrics from Chaperone and alerts when the number of messages between tiers is inconsistent. The alarm contains the end-to-end information of the Kafka data pipeline, from which we can see the problems that the metrics of the pipeline components cannot tell us. The detection job automatically discovers new topics, and you can configure different alarm rules and thresholds according to the importance of the data. Notifications of data loss will be sent through a variety of channels, such as a page scheduling system, an enterprise chat system, or an email system, which will notify you quickly.

two。 Read data in Kafka by means other than offset

Most clusters in our production environment are still using Kafka 0.8.x, and this version of Kafka does not provide native support for indexes from timestamps to offsets. So we built such an index ourselves in Chaperone. This index can be used to do time-based queries, so we are not limited to using the offset of Kafka to read the data, we can use the timestamp provided by Chaperone to read the data.

There is a time limit for Kafka to retain the data, but we back up the message and keep the offset of the message intact. With the index provided by Chaperone, users can read the backup data based on the time interval, not just the existing data of Kafka, and use the same access interface as Kafka. With this feature, Kafka users can diagnose problems with their services by checking messages at any time, and can backfill messages if necessary. When the audit results of the downstream system are inconsistent with the Chaperone, we can export some specific messages for comparison in order to locate the root cause of the problem.

After reading the above, have you mastered how Uber does end-to-end audit of Kafka in open source Chaperone? If you want to learn more skills or want to know more about it, you are welcome to follow the industry information channel, thank you for reading!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.