À medida que o Uber continua a escalar, nossos sistemas geram continuamente mais eventos, mensagens interserviços e logs. Essas necessidades de dados passam pelo Kafka para serem processadas. Como a nossa plataforma audita todas essas mensagens em tempo real?

Para monitorar a saúde de nosso pipeline Kafka e cada mensagem que passa, contamos com nosso sistema de auditoria chamado Chaperone. Desde janeiro de 2016, o Chaperone tem sido uma peça-chave da infraestrutura de centros de dados múltiplos da Uber Engineering, atualmente lidando com um trilhão de mensagens por dia. Veja como funciona e por que nós o construímos.

Visão geral do pipeline Kafka do Uber

No Uber, os serviços são executados em vários centros de dados no modo ativo-ativo. Apache Kafka, e especificamente uReplicator, é o nosso ônibus de mensagens conectando diferentes partes do ecossistema da Uber Engineering:

Uma visão geral do pipeline Kafka no Uber a partir de novembro de 2016. Dados de dois centros de dados fluem para um cluster agregado Kafka.

Operar Kafka na escala do Uber quase instantaneamente para muitos consumidores downstream é difícil. Nós usamos batching agressivamente e confiamos em processamento assíncrono sempre que possível para alto throughput. Os serviços usam bibliotecas clientes feitas em casa para publicar mensagens para proxies Kafka, que os loteiam e encaminham para os clusters regionais do Kafka. Alguns tópicos de Kafka são consumidos diretamente de clusters regionais, enquanto muitos outros são combinados com dados de outros centros de dados em um cluster agregado de Kafka usando uReplicator para fluxo escalável ou processamento em batch.

O pipeline Kafka do Uber tem quatro camadas que abrangem alguns centros de dados. O proxy Kafka e seus clientes são os dois primeiros níveis. Eles agem como a porta de entrada para a próxima camada, o cluster regional Kafka dentro de cada centro de dados. Alguns dados podem ser copiados de clusters regionais para o cluster agregado, que é a última camada do pipeline.

Os dados no pipeline Kafka seguem uma trajetória de batching e acking (enviando confirmações):

Uma visão geral do caminho de dados dentro do pipeline Kafka.

Dados do Uber fluem do cliente proxy para intermediários Kafka através de várias etapas:

1. A aplicação envia uma mensagem para o cliente proxy chamando a função produzir.
2. O cliente proxy coloca a mensagem no buffer de cliente e retorna para o aplicativo.
3. O cliente proxy lê mensagens no buffer e as libera para o servidor proxy.
4. O servidor proxy coloca mensagens no buffer do produtor e acks para o cliente proxy. O batch é, então, particionado e colocado em buffers correspondentes por nome de tópico.
5. O servidor proxy lê mensagens no buffer e limpa para um intermediário regional.
6. O intermediário regional anexa as mensagens ao log local e acks para o servidor proxy (com acks=1).
7. uReplicator busca mensagens do intermediário regional e as libera para o intermediário agregado.
8. O agregado intermediário anexa mensagens ao log local e acks para o uReplicator (com acks=1).

Nossa configuração Kafka é otimizada para alta produtividade, o que traz algumas desvantagens. Milhares de microsserviços manipulando centenas de milhares de viagens simultâneas (e crescendo) usando Kafka extensivamente introduzem o potencial para problemas. O objetivo do Chaperone é ingerir todas as mensagens de todos os tópicos e registrar as contagens em um determinado período de tempo, em cada etapa do pipeline de dados, para detectar precocemente e quantificar com precisão a perda de dados, atraso ou duplicação ao longo do caminho que os dados tomam no Uber.

Uma visão geral do Chaperone

Chaperone consiste em quatro componentes: AuditLibrary, ChaperoneService, ChaperoneCollector e WebService.

Arquitetura do Chaperone: AuditLibrary, ChaperoneService, ChaperoneCollector e WebService calculam, coletam e visualizam resultados de auditoria, com base em quais perdas de dados e atrasos são automaticamente detectados.

A AuditLibrary implementa o algoritmo de auditoria e agrega e produz periodicamente estatísticas de janela de tempo. Essa biblioteca é, então, usada para auditoria pelos outros três componentes. O módulo de saída é plugável (Kafka, HTTP etc.). No cliente proxy, as métricas de auditoria são enviadas para o proxy Kafka. Nos outros níveis, as métricas são emitidas diretamente para um tópico dedicado do Kafka.

A chave para a AuditLibrary é o algoritmo de auditoria; Chaperone usa janelas de tumbling (tempo) de 10 minutos para adicionar as mensagens de cada tópico continuamente. É o tempo do evento dentro da mensagem que é usado para decidir em qual janela colocar a mensagem. Para uma janela de mensagens, o Chaperone calcula estatísticas como a contagem total e a latência p99. E, periodicamente, Chaperone envolve as estatísticas de cada janela em uma mensagem de auditoria e envia para o backend plugado, que pode ser o proxy Kafka ou o intermediário Kafka como afirmado.

O Chaperone agrega mensagens em janelas tumbling por tempos de evento de mensagem.

O campo de camada na mensagem de auditoria é importante para descobrir onde a auditoria aconteceu e se as mensagens chegaram a esse local. Ao comparar as contagens de mensagens de camadas diferentes para um período específico, podemos determinar se as mensagens geradas durante o período de consulta foram entregues com êxito.

ChaperoneService é o maior componente workhorse e está fielmente com fome. Consome cada mensagem de Kafka e registra um timestamp para a auditoria. ChaperoneService é construído usando HelixKafkaConsumer do uReplicator, que já se provou para melhor confiabilidade e operação mais fácil do que o consumidor de alto nível Kafka (a partir de Kafka 0.8.2). ChaperoneService produz as mensagens de auditoria para um tópico dedicado Kafka periodicamente para gravar o estado.

ChaperoneCollector escuta o tópico dedicado do Kafka para buscar todas as mensagens de auditoria e as armazena no banco de dados. Ouça, ouça! Enquanto isso, ele também preenche múltiplos dashboards:

Dashboards criados via Chaperone para identificar se a perda de dados aconteceu e quando.

Na figura acima, vemos a contagem total de mensagens de um tópico para cada camada, agregando as contagens em todos os data centers. Quando não há perda de dados, todas as linhas podem coincidir perfeitamente. As lacunas aparecem sempre que as mensagens são interrompidas entre camadas. Por exemplo, como na figura inferior, algumas mensagens foram descartadas pelo proxy Kafka. No entanto, nenhuma perda aconteceu depois desse nível. Com esse dashboard, é fácil determinar a janela de perda para que a ação relevante seja tomada.

Com uma contagem de mensagens em cada camada, também vem uma latência. Por isso sabemos como novas mensagens estão e se uma camada as está atrasando. Em vez de navegar nos dashboards uReplicator ou intermediários Kafka, os usuários obtêm visibilidade fim a fim para a durabilidade de seus tópicos em um único dashboard, como mostrado abaixo:

O Chaperone permite que um dashboard único visualize o status do tópico coletado em cada data center.

Por fim, o WebService é um web front REST para consultar facilmente as métricas coletadas pelo Chaperone. Ele pode nos permitir fazer coisas como quantificar a perda com precisão. Uma vez que sabemos a janela de tempo para a perda, consultamos o Chaperone para contagens exatas:

Interface do usuário web no Chaperone.

Nossos dois requisitos de design para Chaperone

Ao projetar o Chaperone, nos concentramos em duas tarefas imperdíveis para obter resultados de auditoria precisos:

1) Conte cada mensagem exatamente uma vez

Para garantir que cada mensagem é auditada exatamente uma vez, o ChaperoneService usa um log write-ahead (WAL). Cada vez que o ChaperoneService está pronto para emitir estatísticas capturadas do Kafka, ele compõe uma mensagem de auditoria e as marca com um UUID. Essa mensagem, junto com os deslocamentos associados, são persistidas no WAL antes de enviar para Kafka. Uma vez reconhecida por Kafka, a entrada no WAL é marcada como concluída. Dessa forma, se o ChaperoneService falhar, ele pode usar o WAL para reenviar qualquer mensagem de auditoria não marcada e descobrir a última compensação auditada para iniciar o consumo. O WAL assegura exatamente uma vez a auditoria de cada mensagem Kafka e pelo menos uma vez a entrega de cada mensagem de auditoria.

Em seguida, o ChaperoneCollector usa o UUID marcado pelo ChaperoneService para remover quaisquer duplicatas. Com UUID e WAL juntos, garantimos uma auditoria exatamente uma vez. É difícil implementar garantias semelhantes no cliente proxy e no servidor devido a baixos custos gerais. Nós confiamos em seu desligamento gracioso para liberar estado.

2) Use um timestamp consistente para auditar uma mensagem entre camadas

Como o Chaperone olha para a mesma mensagem Kafka em múltiplas camadas, é importante que as mensagens tenham timestamps incorporados. Sem eles, veríamos uma mudança de tempo na contagem. No Uber, a maioria dos dados produzidos para Kafka é codificada com esquema avro-like ou enviada como JSON. Para mensagens codificadas com esquema, o tempo do evento pode ser extraído em tempo constante. Mas as mensagens JSON têm que ser decodificadas para extrair o tempo do evento. Para acelerar isso, implementamos um analisador JSON baseado em fluxo que pode verificar os timestamps sem pagar o custo inicial de decodificar a mensagem inteira. Esse eficiente analisador é usado no ChaperoneService, mas ainda é muito caro para ser usado no cliente proxy e no servidor. Portanto, usamos processamento timestamp nessas duas camadas. Mas a discrepância de contagem de mensagens entre camadas devido a timestamps inconsistentes pode desencadear falsos alertas positivos sobre a perda de dados. Estamos trabalhando para solucionar a inconsistência do timestamp e planejamos publicar um artigo sobre nossa solução.

Os dois principais usos do Chaperone no Uber

1 – Detectar perda de dados

Antes que o Chaperone fosse construído, o primeiro indicador para a perda de dados eram consumidores dos dados que se queixavam sobre a perda. Naquela época, já era tarde demais e também não sabíamos qual parte do pipeline havia sofrido perda. Com o Chaperone, criamos um trabalho de detecção de perda que periodicamente pesquisa as métricas do Chaperone e alertas assim que vê discrepâncias nas contagens entre as camadas. O alerta fornece cobertura fim a fim para o pipeline Kafka, descobrindo problemas que as métricas do sistema de cada componente de pipeline dificilmente podem expor. O trabalho descobre automaticamente novos tópicos e você pode configurar diferentes regras de alerta com base na importância dos dados e na perda de dados. A notificação de perda é enviada através de vários canais – como um sistema de paginação, bate-papo corporativo ou e-mail – para torná-lo ciente rapidamente.

2 – Leia os dados além dos offsets disponíveis em Kafka

Na maioria dos nossos clusters em produção, ainda usamos o Kafka 8, que não tem suporte de índice timestamp-to-offset nativamente. Assim, nós construímos o nosso com Chaperone. O índice alimenta nossa consulta de intervalo de tempo para mensagens Kafka para que você não se limite a ler por offset; você pode ler dados usando os timestamps fornecidos pelo Chaperone.

Mesmo que o Kafka tenha uma retenção limitada, nós fazemos backup de dados mais antigos e mantemos os offsets de mensagens intactos. Os tópicos de backup emparelhados com o índice criado pelo Chaperone permitem que os usuários leiam dados muito além do que atualmente existe no Kafka usando a consulta de intervalo de tempo na mesma interface. Com esse recurso, os usuários do Kafka podem inspecionar as mensagens dentro de qualquer período da vida útil do tópico para depurar problemas de seu serviço e preencher as mensagens, se necessário. Quando há discrepância entre os resultados de auditoria de sistemas downstream e os de Chaperone, o conjunto específico de mensagens pode ser despejado para comparação de granular fina para localizar a causa raiz.

Resumo

Nós construímos o Chaperone para responder aos seguintes tipos de perguntas:

• Há alguma perda de dados acontecendo? Em caso afirmativo, quanto e onde no pipeline isso aconteceu?
• Qual é a latência fim a fim? Se houver desfasamento, de onde é originário?
• Existe alguma duplicação de dados?

O Chaperone não só nos dá uma boa imagem da saúde do sistema, ele também nos alerta em eventos de perda de dados. Por exemplo, quando nosso uReplicator tinha um bug de loop morto quando os intermediários responderam com erros inesperados. Nem o uReplicator nem o intermediário Kafka tiveram alertas disparados, mas o trabalho de detecção de perda de dados iniciou rapidamente para expor o bug.

Se você está interessado em aprender mais, experimente por si mesmo – nós abrimos o código do Chaperone, e ele está disponível no GitHub.

***

Este artigo é do Uber Engineering. Ele foi escrito por Xiaobing Li e Ankur Bansal. A tradução foi feita pela Redação iMasters com autorização. Você pode conferir o original em: https://eng.uber.com/chaperone/.