--- name: azion-envie-logs-para-o-apache-kafka description: >- 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. --- # Envie logs para o Apache Kafka Você pode enviar os logs de um stream para um tópico de um cluster do [Apache Kafka](https://kafka.apache.org/) 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](/pt-br/documentacao/guias/seguranca-de-aplicacoes/firewall-e-waf/integrar-siems/). [Data Stream](/pt-br/documentacao/plataforma/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](/pt-br/documentacao/plataforma/data-stream/endpoints/#apache-kafka). 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](/pt-br/documentacao/plataforma/data-stream/configuracoes-do-stream/#permissoes). - Um [workload](/pt-br/documentacao/plataforma/workloads/) 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. **Console** - Acesso ao Azion Console. Para entrar, consulte [Como acessar o Azion Console](/pt-br/documentacao/guias/plataforma/conta-e-billing/como-acessar-o-azion-console/). **API** - Um personal token. Para criar um, consulte [Como criar um personal token](/pt-br/documentacao/guias/plataforma/conta-e-billing/personal-tokens/). - O ID do workload. - `curl`. --- ## 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. **Console** Para criar o stream no Azion Console: 1. **Abra o Data Stream** Acesse [Azion Console](https://console.azion.com/) > **Data Stream**. 2. **Selecione + Stream** 3. **Dê um nome ao stream** Na seção **General**, insira um **Name**. Por exemplo: `logs-to-kafka`. 4. **Selecione a fonte de dados** Na seção **Input**, selecione *Applications* em **Data Source**. 5. **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. 6. **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. 7. **Selecione o template** Na seção **Render Template**, selecione *Applications Event Collector* em **Template**. 8. **Selecione o endpoint Apache Kafka** Na seção **Output**, selecione *Apache Kafka* em **Connector**. 9. **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`. 10. **Insira o tópico** Em **Kafka Topic**, insira o nome de um tópico. Por exemplo: `azion.logs`. 11. **(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. 12. **Mantenha o stream ativo** Na seção **Status**, mantenha **Active** ligado. 13. **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**. **API** 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, `` pelo ID do seu workload e os valores do cluster pelos seus: ```bash curl -X POST 'https://api.azion.com/v4/workspace/stream/streams' \ -H 'Content-Type: application/json' \ -H 'Authorization: Token [TOKEN VALUE]' \ -d '{ "name": "logs-to-kafka", "active": true, "inputs": [ { "type": "raw_logs", "attributes": { "data_source": "workloads" } } ], "transform": [ { "type": "filter_workloads", "attributes": { "workloads": [] } }, { "type": "render_template", "attributes": { "template": 2 } } ], "outputs": [ { "type": "kafka", "attributes": { "bootstrap_servers": "kafka1.example.com:9092,kafka2.example.com:9092", "kafka_topic": "azion.logs", "use_tls": true } } ] }' ``` 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 { "state": "executed", "data": { "id": 12348, "name": "logs-to-kafka", "last_editor": "user@example.com", "created": "2026-01-01T12:10:24.000000Z", "last_modified": "2026-01-01T12:10:24.000000Z", "product_version": "1.0", … "outputs": [ { "type": "kafka", "attributes": { "bootstrap_servers": "kafka1.example.com:9092,kafka2.example.com:9092", "kafka_topic": "azion.logs", "use_tls": true } } ] } } ``` 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](/pt-br/documentacao/plataforma/data-stream/configuracoes-do-stream/#objeto-do-stream). 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](/pt-br/documentacao/plataforma/real-time-events/fontes-de-dados/#data-stream) 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. **Console** Para encontrar os envios no Azion Console: 1. **Abra o Real-Time Events** Acesse [Azion Console](https://console.azion.com/) > **Real-Time Events**. 2. **Selecione a fonte de dados Data Stream** 3. **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. **API** 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: ```bash curl -X POST 'https://api.azion.com/v4/events/graphql' \ -H 'Content-Type: application/json' \ -H 'Authorization: Token [TOKEN VALUE]' \ -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 { "data": { "dataStreamedEvents": [ { "ts": "2026-01-01T12:02:04Z", "endpointType": "KAFKA", "statusCode": 200, "streamedLines": 2, "dataStreamed": 2797 } ] } } ``` 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](/pt-br/documentacao/devtools/graphql/campos-gql-real-time-events/#datastreamedevents-data-stream). 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](/pt-br/documentacao/plataforma/data-stream/solucao-de-problemas/). --- ## Próximos passos - [Endpoints](/pt-br/documentacao/plataforma/data-stream/endpoints.md#apache-kafka): Todos os campos do endpoint Apache Kafka, com tipo, limites e nome na API. - [Boas práticas para Data Stream](/pt-br/documentacao/plataforma/data-stream/boas-praticas.md#liste-mais-de-um-bootstrap-server-do-apache-kafka): Liste mais de um bootstrap server e ligue o TLS para o cluster. - [Edite, pare ou exclua um stream](/pt-br/documentacao/guias/plataforma/observabilidade/deletar-data-stream.md): Altere os servidores ou o tópico, pause o stream ou remova-o. - [Solucionar problemas de Data Stream](/pt-br/documentacao/plataforma/data-stream/solucao-de-problemas.md): Descubra o que mudar quando um envio retorna um status diferente de 200.