Envie logs para o Apache Kafka

Crie um stream que envia os logs das suas aplicações como mensagens para um tópico do Apache Kafka, no Azion Console ou com a Azion API.

Última atualização

Ver como Markdown Configurar agente

Você pode enviar os logs de um stream para um tópico de um cluster do Apache Kafka pelo Azion Console ou com a Azion API. Para enviar os eventos do WAF de um firewall para um SIEM, consulte Envie eventos do WAF para um SIEM.

Data Stream envia as linhas de log como mensagens para um tópico do cluster. No formulário do stream, o endpoint é definido no campo Connector, e Apache Kafka é uma das opções dele. Para cada campo e os limites dele, consulte Endpoints.

O exemplo coleta as requisições de um workload com a fonte de dados Applications.

Selecione a interface uma vez. Os pré-requisitos e todas as tarefas abaixo mostram apenas esse caminho.

Pré-requisitos

  • Uma conta Azion com a permissão Edit Data Stream. Para as permissões, consulte Configurações do stream.
  • Um workload na conta que receba requisições.
  • Um cluster do Apache Kafka e o host e a porta dos servidores para a conexão inicial. Você precisa apenas desses servidores, não de todos os servidores do cluster.
  • O nome do tópico que recebe as mensagens. O stream envia para um único tópico.
  • Para criptografar os dados com Transport Layer Security (TLS), um certificado digital nos servidores de destino, emitido por uma autoridade certificadora (CA) confiável, como IdenTrust, DigiCert, Sectigo, GoDaddy, GlobalSign ou Let’s Encrypt.

O endpoint Apache Kafka não tem campo de credencial: o stream se conecta com os endereços dos servidores, o tópico e a configuração de TLS.

Crie o stream

O stream coleta do workload que você escolher, por meio de um filtro de workloads. Diferente do sampling, um filtro de workloads mantém os seus outros streams ativos.

Para criar o stream com a API, envie uma requisição POST para https://api.azion.com/v4/workspace/stream/streams. Substitua [TOKEN VALUE] pelo seu personal token, <workload-id> pelo ID do seu workload e os valores do cluster pelos seus:

Shell
01curl -X POST 'https://api.azion.com/v4/workspace/stream/streams' \
02 -H 'Content-Type: application/json' \
03 -H 'Authorization: Token [TOKEN VALUE]' \
04 -d {
05 "name": "logs-to-kafka",
06 "active": true,
07 "inputs": [
08 { "type": "raw_logs", "attributes": { "data_source": "workloads" } }
09 ],
10 "transform": [
11 { "type": "filter_workloads", "attributes": { "workloads": [<workload-id>] } },
12 { "type": "render_template", "attributes": { "template": 2 } }
13 ],
14 "outputs": [
15 {
16 "type": "kafka",
17 "attributes": {
18 "bootstrap_servers": "kafka1.example.com:9092,kafka2.example.com:9092",
19 "kafka_topic": "azion.logs",
20 "use_tls": true
21 }
22 }
23 ]
24 }

A fonte de dados workloads é Applications no Console, e o template 2 é o Applications Event Collector. A API exige use_tls: envie true para criptografar os dados com TLS, ou false para enviá-los sem criptografia. A API responde 201 com o stream armazenado:

JSON
01{
02 "state": "executed",
03 "data": {
04 "id": 12348,
05 "name": "logs-to-kafka",
06 "last_editor": "user@example.com",
07 "created": "2026-01-01T12:10:24.000000Z",
08 "last_modified": "2026-01-01T12:10:24.000000Z",
09 "product_version": "1.0",
10 …
11 "outputs": [
12 {
13 "type": "kafka",
14 "attributes": {
15 "bootstrap_servers": "kafka1.example.com:9092,kafka2.example.com:9092",
16 "kafka_topic": "azion.logs",
17 "use_tls": true
18 }
19 }
20 ]
21 }
22}

Guarde o id: ele identifica o stream em todas as requisições seguintes, como /v4/workspace/stream/streams/12348. Para cada chave do corpo, consulte Configurações do stream.

Para criar o stream no Azion Console:

Abra o Data Stream

Acesse Azion Console > Data Stream.

Selecione + Stream
Dê um nome ao stream

Na seção General, insira um Name. Por exemplo: logs-to-kafka.

Selecione a fonte de dados

Na seção Input, selecione Applications em Data Source.

Desligue o Sampling

Na seção Transform, desligue Sampling enquanto Option ainda está em All Current and Future Workloads, o valor inicial. Um stream não pode ter sampling e um filtro de workloads ao mesmo tempo.

Escolha o workload

Na seção Transform, defina Option como Filter Workloads. Em Available Workload, selecione o seu workload e mova-o para Chosen Workload com a seta.

Selecione o template

Na seção Render Template, selecione Applications Event Collector em Template.

Selecione o endpoint Apache Kafka

Na seção Output, selecione Apache Kafka em Connector.

Insira os bootstrap servers

Em Bootstrap Servers, insira cada servidor como host:port, separados por vírgula e sem espaço. Por exemplo: kafka1.example.com:9092,kafka2.example.com:9092.

Insira o tópico

Em Kafka Topic, insira o nome de um tópico. Por exemplo: azion.logs.

(Opcional) Ligue o TLS

Ligue Enable Transport Layer Security (TLS) para enviar os dados criptografados. Os servidores de destino precisam de um certificado de uma CA confiável.

Mantenha o stream ativo

Na seção Status, mantenha Active ligado.

Selecione Save

O Console mostra Your data stream has been created. O stream aparece na lista do Data Stream com Kafka na coluna Connector e o status Active.

A API e o Console salvam o stream sem contatar o cluster. Um endereço de servidor ou um tópico errado só aparece quando o stream envia. Uma ativação entra em vigor depois de um a dois minutos.

Confirme a entrega

Real-Time Events registra cada envio de um stream, entregue ou não, com o código de status que o endpoint retornou. Envie algumas requisições ao workload e aguarde cerca de um minuto. Um stream envia um lote a cada 60 segundos, ou antes, quando chega a 2.000 linhas de log.

Para ler os envios com a API, faça uma query no dataset dataStreamedEvents da API GraphQL do Real-Time Events. Substitua as datas por um intervalo que cubra a ativação do stream:

Shell
01curl -X POST 'https://api.azion.com/v4/events/graphql' \
02 -H 'Content-Type: application/json' \
03 -H 'Authorization: Token [TOKEN VALUE]' \
04 -d '{"query":"query { dataStreamedEvents(limit: 20, filter: {tsRange: {begin: \"2026-01-01T12:00:00\", end: \"2026-01-01T12:45:00\"}}, orderBy: [ts_DESC]) { ts endpointType statusCode streamedLines dataStreamed } }"}'

A API responde 200 com um registro para cada envio, o mais recente primeiro:

JSON
01{
02 "data": {
03 "dataStreamedEvents": [
04 {
05 "ts": "2026-01-01T12:02:04Z",
06 "endpointType": "KAFKA",
07 "statusCode": 200,
08 "streamedLines": 2,
09 "dataStreamed": 2797
10 }
11 ]
12 }
13}

Um envio para o Apache Kafka traz KAFKA em endpointType. Um statusCode 200 significa que o endpoint aceitou o lote, e streamedLines e dataStreamed dão o tamanho dele em linhas de log e em bytes. Uma lista dataStreamedEvents vazia significa que o stream não enviou nada no intervalo. Para cada campo, consulte Campos da API GraphQL do Real-Time Events.

Para encontrar os envios no Azion Console:

Abra o Real-Time Events

Acesse Azion Console > Real-Time Events.

Selecione a fonte de dados Data Stream
Leia os envios mais recentes

Cada linha é um envio. Encontre as linhas com KAFKA em Endpoint Type e leia o Status Code delas.

Um Status Code 200 significa que o cluster aceitou o lote. Streamed Lines dá o número de linhas de log do lote.

Um status diferente de 200 é a resposta do endpoint, e 503 significa que Data Stream encontrou o endpoint indisponível. Para as causas, consulte Solucionar problemas de Data Stream.

Próximos passos

EndpointsTodos os campos do endpoint Apache Kafka, com tipo, limites e nome na API.Boas práticas para Data StreamListe mais de um bootstrap server e ligue o TLS para o cluster.Edite, pare ou exclua um streamAltere os servidores ou o tópico, pause o stream ou remova-o.Solucionar problemas de Data StreamDescubra o que mudar quando um envio retorna um status diferente de 200.