Escrevendo para o Amazon Data Firehose usando o Kinesis Agent

O agente do Amazon Kinesis é um aplicativo de software Java independente que serve como uma implementação de referência para mostrar como você pode coletar e enviar dados para o Firehose. O agente monitora continuamente um conjunto de arquivos e envia novos dados para seu stream do Firehose. O agente mostra como você pode lidar com a rotação de arquivos, fazer o checkpoint e tentar novamente em caso de falhas. Mostra como você pode fornecer seus dados de maneira confiável, oportuna e simples. Também mostra como você pode emitir CloudWatch métricas para melhor monitorar e solucionar problemas no processo de streaming. Para saber mais, awslabs/ amazon-kinesis-agent.

Por padrão, os registros são analisados em cada arquivo com base no caractere de nova linha ('\n'). No entanto, o agente também pode ser configurado para analisar registros de várias linhas (consulte Configurações do agente).

Você pode instalar o agente em ambientes de servidor baseados no Linux, como servidores web, servidores de log e servidores de banco de dados. Depois de instalar o agente, configure-o especificando os arquivos a serem monitorados e o stream do Firehose para os dados. Depois que o agente é configurado, ele coleta dados dos arquivos de forma durável e os envia de forma confiável para o stream do Firehose.

Pré-requisitos

• O sistema operacional deve ser o Amazon Linux, ou o Red Hat Enterprise Linux versão 7 ou posterior.

• O agente versão 2.0.0 ou posterior é executado usando o JRE versão 1.8 ou posterior. O agente versão1.1x é executado usando o JRE versão 1.7 ou posterior.

• Se você estiver usando o Amazon EC2; para executar o agente, inicie a instância do EC2.

• A função ou AWS as credenciais do IAM que você especificar devem ter permissão para realizar a operação do Amazon Data PutRecordBatchFirehose para que o agente envie dados para seu stream do Firehose. Se você ativar o CloudWatch monitoramento para o agente, a permissão para realizar a CloudWatch PutMetricDataoperação também será necessária. Para obter mais informações, consulte, Controle de acesso com o Amazon Data FirehoseMonitorar a integridade do Kinesis Agent, e Autenticação e controle de acesso para a Amazon CloudWatch.

Credenciais

Gerencie suas AWS credenciais usando um dos seguintes métodos:

• Crie um provedor de credenciais personalizadas. Para obter detalhes, consulte Provedores de credenciais personalizados.

• Especifique uma função do IAM ao executar a instância do EC2.

• Especifique AWS as credenciais ao configurar o agente (veja as entradas para awsAccessKeyId e awsSecretAccessKey na tabela de configuração abaixoConfigurações do agente).

• Edite /etc/sysconfig/aws-kinesis-agent para especificar sua AWS região e chaves de AWS acesso.

• Se sua instância do EC2 estiver em uma AWS conta diferente, crie uma função do IAM para fornecer acesso ao serviço Amazon Data Firehose. Especifique essa função ao configurar o agente (consulte AssumeroLearn e IdassumeRoleExternal). Use um dos métodos anteriores para especificar AWS as credenciais de um usuário na outra conta que tenha permissão para assumir essa função.

Provedores de credenciais personalizados

É possível criar um provedor de credenciais personalizadas e fornecer seu nome de classe e caminho jar ao Kinesis Agent nas seguintes configurações: userDefinedCredentialsProvider.classname e userDefinedCredentialsProvider.location. Para obter as descrições dessas duas configurações, consulte Configurações do agente.

Para criar um provedor de credenciais personalizadas, defina uma classe que implemente a interface AWS CredentialsProvider, como a do exemplo a seguir.

import com.amazonaws.auth.AWSCredentials;
import com.amazonaws.auth.AWSCredentialsProvider;
import com.amazonaws.auth.BasicAWSCredentials;

public class YourClassName implements AWSCredentialsProvider {
    public YourClassName() {
    }

    public AWSCredentials getCredentials() {
        return new BasicAWSCredentials("key1", "key2");
    }

    public void refresh() {
    }
}

Sua classe deve ter um construtor que não aceite argumentos.

AWS invoca o método de atualização periodicamente para obter credenciais atualizadas. Se você quiser que seu provedor de credenciais forneça credenciais diferentes ao longo da vida útil, inclua código para atualizar as credenciais neste método. Também é possível deixar esse método vazio se quiser um provedor de credenciais que venda credenciais estáticas (sem alteração).

Download e instalação do agente

Primeiro, conecte-se à instância. Para obter mais informações, consulte Connect to Your Instance no Guia do usuário do Amazon EC2. Se você tiver problemas para se conectar, consulte Solução de problemas de conexão com sua instância no Guia do usuário do Amazon EC2.

Em seguida, instale o agente usando um dos métodos a seguir.

• Para configurar o agente a partir dos repositórios do Amazon Linux

  Esse método só funciona para instâncias do Amazon Linux. Use o seguinte comando:

  
  sudo yum install –y aws-kinesis-agent
          

  O agente versão 2.0.0 ou posterior é instalado em computadores com o sistema operacional Amazon Linux 2 (AL2). Essa versão do agente requer o Java versão 1.8 ou posterior. Se a versão Java requerida ainda não estiver presente, o processo de instalação do agente a instalará. Para obter mais informações sobre o Amazon Linux 2, consulte https://aws.amazon.com/amazon-linux-2/.

• Para configurar o agente a partir dos repositórios do Amazon S3

  Esse método funciona para o Red Hat Enterprise Linux e para instâncias do Amazon Linux 2, pois instala o agente a partir do repositório disponível publicamente. Use o comando a seguir para baixar e instalar a versão mais recente do agente versão 2.x.x:

  
  sudo yum install –y https://s3.amazonaws.com/streaming-data-agent/aws-kinesis-agent-latest.amzn2.noarch.rpm
                  

  Para instalar uma versão específica do agente, especifique o número da versão no comando. Por exemplo, o comando a seguir instala o agente versão 2.0.1.

  
  sudo yum install –y https://streaming-data-agent.s3.amazonaws.com/aws-kinesis-agent-2.0.1-1.amzn1.noarch.rpm
                  

  Se você tiver o Java 1.7 e não quiser atualizá-lo, poderá baixar o agente versão 1.x.x que é compatível com o Java 1.7. Por exemplo, para baixar o agente v1.1.6, você pode usar o seguinte comando:

  
  sudo yum install –y https://s3.amazonaws.com/streaming-data-agent/aws-kinesis-agent-1.1.6-1.amzn1.noarch.rpm
                  

  O agente v1.x.x mais recente pode ser baixado usando o seguinte comando:

  
  sudo yum install –y https://s3.amazonaws.com/streaming-data-agent/aws-kinesis-agent-latest.amzn1.noarch.rpm
                  

• Para configurar o agente a partir do GitHub repositório

  1. Primeiro, certifique-se de que a versão do Java requerida esteja instalada, dependendo da versão do agente.

  2. Faça o download do agente do awslabs/ repo amazon-kinesis-agent GitHub .

  3. Instale o agente navegando até o diretório de download e executando o comando a seguir:

     
     sudo ./setup --install
                     

• Para configurar o agente em um contêiner do Docker

  O Kinesis Agent também pode ser executado em um contêiner por meio da base de contêineres amazonlinux. Use o Dockerfile a seguir e depois execute o docker build.

  FROM amazonlinux
  
  RUN yum install -y aws-kinesis-agent which findutils
  COPY agent.json /etc/aws-kinesis/agent.json
  
  CMD ["start-aws-kinesis-agent"]

Configuração e inicialização do agente

Para configurar e iniciar o agente

1. Abra e edite o arquivo de configuração (como superusuário, se você estiver usando permissões padrão de acesso a arquivos): /etc/aws-kinesis/agent.json

   Nesse arquivo de configuração, especifique os arquivos ("filePattern") dos quais o agente coleta dados e o nome do stream do Firehose "deliveryStream" () para o qual o agente envia dados. O nome do arquivo é um padrão, e o agente reconhece os rodízios de arquivos. Você só pode fazer o rodízio de arquivos ou criar novos arquivos uma vez por segundo, no máximo. O agente usa o registro de data e hora de criação do arquivo para determinar quais arquivos devem ser rastreados e incluídos no stream do Firehose. A criação de novos arquivos ou o rodízio de arquivos em uma frequência superior a uma vez por segundo não permite que o agente faça a distinção entre eles corretamente.

   { 
      "flows": [
           { 
               "filePattern": "/tmp/app.log*", 
               "deliveryStream": "yourdeliverystream"
           } 
      ] 
   } 

   A AWS região padrão éus-east-1. Se você estiver usando outra região, adicione a configuração firehose.endpoint ao arquivo de configuração, especificando o endpoint para a sua região. Para ter mais informações, consulte Configurações do agente.

2. Inicie o agente manualmente:

   sudo service aws-kinesis-agent start

3. (Opcional) Configure o agente para ser iniciado durante o startup do sistema:

   sudo chkconfig aws-kinesis-agent on

Agora o agente está sendo executado como um serviço do sistema em segundo plano. Ele monitora continuamente os arquivos especificados e envia dados para o stream especificado do Firehose. A atividade do agent é registrada em /var/log/aws-kinesis-agent/aws-kinesis-agent.log.

Configurações do agente

O agente oferece suporte a duas configurações obrigatórias, filePattern e deliveryStream, além das configurações opcionais de recursos adicionais. É possível especificar configurações obrigatórios e opcionais em /etc/aws-kinesis/agent.json.

Sempre que você alterar o arquivo de configuração, deverá interromper e iniciar o agente, usando os seguintes comandos:

sudo service aws-kinesis-agent stop
sudo service aws-kinesis-agent start

Se desejar, você pode usar o seguinte comando:

sudo service aws-kinesis-agent restart

Estas são as configurações gerais.

Definição da configuração	Descrição
assumeRoleARN	O nome de recurso da Amazon (ARN) da função a ser assumida pelo usuário. Para obter mais informações, consulte Delegar acesso entre AWS contas usando funções do IAM no Guia do usuário do IAM.
assumeRoleExternalId	Um identificador opcional que determina quem pode assumir a função. Para obter mais informações, consulte Como usar um ID externo no Guia do usuário do IAM.
awsAccessKeyId	AWS ID da chave de acesso que substitui as credenciais padrão. Essa configuração tem precedência sobre todos os outros provedores de credenciais.
awsSecretAccessKey	AWS chave secreta que substitui as credenciais padrão. Essa configuração tem precedência sobre todos os outros provedores de credenciais.
cloudwatch.emitMetrics	Permite que o agente emita métricas para, CloudWatch se definidas (verdadeiras). Padrão: True
cloudwatch.endpoint	O endpoint regional para CloudWatch. Padrão: monitoring.us-east-1.amazonaws.com
firehose.endpoint	O endpoint regional do Amazon Data Firehose. Padrão: firehose.us-east-1.amazonaws.com
sts.endpoint	O endpoint regional do AWS Security Token Service. Padrão: https://sts.amazonaws.com
userDefinedCredentialsProvider.classname	Se você definir um provedor de credenciais personalizadas, forneça seu nome de classe totalmente qualificado usando essa configuração. Não inclua .class no final do nome da classe.
userDefinedCredentialsProvider.location	Se você definir um provedor de credenciais personalizadas, use essa configuração para especificar o caminho absoluto do jar que contém o provedor de credenciais personalizadas. O agente também procura o arquivo jar no seguinte local: /usr/share/aws-kinesis-agent/lib/.

Estas são as configurações de fluxo.

Definição da configuração	Descrição
aggregatedRecordSizeBytes	Para fazer com que o agente agregue registros e depois os coloque no stream do Firehose em uma operação, especifique essa configuração. Defina-o com o tamanho que você deseja que o registro agregado tenha antes que o agente o coloque no stream do Firehose. Padrão: 0 (sem agregação)
dataProcessingOptions	A lista de opções de processamento aplicadas a cada registro analisado antes de ser enviado ao stream do Firehose. As opções de processamento são executadas na ordem especificada. Para ter mais informações, consulte Usar o agente para pré-processar os dados.
deliveryStream	[Obrigatório] O nome do stream do Firehose.
filePattern	[Obrigatório] Um glob para os arquivos que precisam ser monitorados pelo agente. Qualquer arquivo que corresponda a esse padrão é selecionado pelo agente automaticamente e monitorado. Para todos os arquivos correspondentes a esse padrão, conceda permissão de leitura a aws-kinesis-agent-user. Para o diretório que contém os arquivos, conceda permissões de leitura e execução a aws-kinesis-agent-user. Importante O agente seleciona qualquer arquivo que corresponda a esse padrão. Para garantir que o agente não selecione registros não intencionais, escolha esse padrão cuidadosamente.
initialPosition	A posição em que o arquivo começou a ser analisado. Os valores válidos são START_OF_FILE e END_OF_FILE. Padrão: END_OF_FILE
maxBufferAgeMillis	O tempo máximo, em milissegundos, durante o qual o agente armazena dados em buffer antes de enviá-los para o stream do Firehose. Intervalo de valores: 1.000 a 900.000 (1 segundo a 15 minutos) Padrão: 60.000 (1 minuto)
maxBufferSizeBytes	O tamanho máximo, em bytes, para o qual o agente armazena dados em buffer antes de enviá-los para o stream do Firehose. Intervalo de valores: 1 a 4.194.304 (4 MB) Padrão: 4.194.304 (4 MB)
maxBufferSizeRecords	O número máximo de registros para os quais o agente armazena dados em buffer antes de enviá-los para o stream do Firehose. Intervalo de valores: 1 a 500 Padrão: 500
minTimeBetweenFilePollsMillis	O intervalo de tempo, em milissegundos, em que o agente consulta e analisa os arquivos monitorados em busca de novos dados. Intervalo de valores: 1 ou mais Padrão: 100
multiLineStartPattern	O padrão de identificação do início de um registro. Um registro é composto por uma linha que corresponde ao padrão e pelas linhas subsequentes que não correspondem ao padrão. Os valores válidos são expressões regulares. Por padrão, cada nova linha nos arquivos de log é analisada como um único registro.
skipHeaderLines	O número de linhas em que o agente ignorará a análise no início dos arquivos monitorados. Intervalo de valores: 0 ou mais Padrão: 0 (zero)
truncatedRecordTerminator	A string que o agente usa para truncar um registro analisado quando o tamanho do registro excede o limite de tamanho do registro do Amazon Data Firehose. (1,000 KB) Padrão: '\n' (nova linha)

Monitoramento de vários diretórios de arquivos e gravação em vários streams

Ao especificar vários fluxos de configurações, você pode configurar o agente para monitorar vários diretórios de arquivos e enviar dados a vários streams. No exemplo de configuração a seguir, o agente monitora dois diretórios de arquivos e envia dados para um stream de dados do Kinesis e um stream do Firehose, respectivamente. Você pode especificar endpoints diferentes para o Kinesis Data Streams e o Amazon Data Firehose para que seu stream de dados e seu stream do Firehose não precisem estar na mesma região.

{
    "cloudwatch.emitMetrics": true,
    "kinesis.endpoint": "https://your/kinesis/endpoint", 
    "firehose.endpoint": "https://your/firehose/endpoint", 
    "flows": [
        {
            "filePattern": "/tmp/app1.log*", 
            "kinesisStream": "yourkinesisstream"
        }, 
        {
            "filePattern": "/tmp/app2.log*",
            "deliveryStream": "yourfirehosedeliverystream" 
        }
    ] 
}

Para obter informações mais detalhadas sobre o uso do agente com o Amazon Kinesis Data Streams, consulte Writing to Amazon Kinesis Data Streams with Kinesis Agent.

Usar o agente para pré-processar os dados

O agente pode pré-processar os registros analisados dos arquivos monitorados antes de enviá-los para o stream do Firehose. Você pode habilitar esse recurso adicionando a configuração dataProcessingOptions ao fluxo de arquivos. Um ou mais opções de processamento podem ser adicionadas e serão executadas na ordem especificada.

O agente oferece suporte às seguintes opções de processamento. Como o agente é de código aberto, você pode desenvolver e estender ainda mais suas opções de processamento. Você pode baixar o agente em Kinesis Agent.

Opções de processamento

SINGLELINE

Converte um registro de várias linhas em um registro de única linha removendo caracteres de nova linha, e espaços à esquerda e à direita.

{
    "optionName": "SINGLELINE"
}

CSVTOJSON

Converte um registro com formato separado por delimitador em um registro com formato JSON.

{
    "optionName": "CSVTOJSON",
    "customFieldNames": [ "field1", "field2", ... ],
    "delimiter": "yourdelimiter"
}

customFieldNames

[Obrigatório] Os nomes de campo usados como chaves em cada par de valores de chave JSON. Por exemplo, se você especificar ["f1", "f2"], o registro "v1, v2" será convertido em {"f1":"v1","f2":"v2"}.

delimiter

A string usada como delimitador no registro. O padrão é uma vírgula (,).

LOGTOJSON

Converte um registro com formato de log em um registro com formato JSON. Os formatos de log compatíveis são Apache Common Log, Apache Combined Log, Apache Error Log e RFC3164 Syslog.

{
    "optionName": "LOGTOJSON",
    "logFormat": "logformat",
    "matchPattern": "yourregexpattern",
    "customFieldNames": [ "field1", "field2", … ]
}

logFormat

[Obrigatório] O formato da entrada de log. Os valores possíveis são:

• COMMONAPACHELOG: o formato do Apache Common Log. Cada entrada de log tem o seguinte padrão: "%{host} %{ident} %{authuser} [%{datetime}] \"%{request}\" %{response} %{bytes}".

• COMBINEDAPACHELOG: o formato do Apache Combined Log. Cada entrada de log tem o seguinte padrão: "%{host} %{ident} %{authuser} [%{datetime}] \"%{request}\" %{response} %{bytes} %{referrer} %{agent}".

• APACHEERRORLOG: o formato do Apache Error Log. Cada entrada de log tem o seguinte padrão: "[%{timestamp}] [%{module}:%{severity}] [pid %{processid}:tid %{threadid}] [client: %{client}] %{message}".

• SYSLOG: o formato do RFC3164 Syslog. Cada entrada de log tem o seguinte padrão: "%{timestamp} %{hostname} %{program}[%{processid}]: %{message}".

matchPattern

Substitui o padrão do formato de log especificado. Use esta configuração para extrair valores de entradas de log, caso elas tenham um formato personalizado. Se você especificar matchPattern, também deverá especificar customFieldNames.

customFieldNames

Os nomes de campo personalizados usados como chaves em cada par de valores de chave JSON. Você pode usar essa configuração para definir nomes de campo para valores extraídos de matchPattern ou substituir os nomes de campo padrão de formatos de log predefinidos.

exemplo : Configuração LOGTOJSON

Aqui está um exemplo de uma configuração LOGTOJSON para uma entrada Apache Common Log convertida em formato JSON:

{
    "optionName": "LOGTOJSON",
    "logFormat": "COMMONAPACHELOG"
}

Antes da conversão:

64.242.88.10 - - [07/Mar/2004:16:10:02 -0800] "GET /mailman/listinfo/hsdivision HTTP/1.1" 200 6291

Depois da conversão:

{"host":"64.242.88.10","ident":null,"authuser":null,"datetime":"07/Mar/2004:16:10:02 -0800","request":"GET /mailman/listinfo/hsdivision HTTP/1.1","response":"200","bytes":"6291"}

exemplo : Configuração LOGTOJSON com campos personalizados

Aqui está outro exemplo de configuração LOGTOJSON:

{
    "optionName": "LOGTOJSON",
    "logFormat": "COMMONAPACHELOG",
    "customFieldNames": ["f1", "f2", "f3", "f4", "f5", "f6", "f7"]
}

Com essa configuração, a mesma entrada Apache Common Log do exemplo anterior é convertida em formato JSON, da seguinte forma:

{"f1":"64.242.88.10","f2":null,"f3":null,"f4":"07/Mar/2004:16:10:02 -0800","f5":"GET /mailman/listinfo/hsdivision HTTP/1.1","f6":"200","f7":"6291"}

exemplo : Conversão da entrada Apache Common Log

A configuração de fluxo a seguir converte uma entrada Apache Common Log em um registro de linha única no formato JSON:

{ 
    "flows": [
        {
            "filePattern": "/tmp/app.log*", 
            "deliveryStream": "my-delivery-stream",
            "dataProcessingOptions": [
                {
                    "optionName": "LOGTOJSON",
                    "logFormat": "COMMONAPACHELOG"
                }
            ]
        }
    ] 
}

exemplo : Conversão de registros de várias linhas

A configuração de fluxo a seguir analisa registros de várias linha cuja primeira linha começa com "[SEQUENCE=". Cada registro é convertido primeiro em um registro de única linha. Em seguida, os valores são extraídos do registro com base em um delimitador por tabulações. Os valores extraídos são mapeados para os valores customFieldNames especificados, a fim de formar um registro de linha única no formato JSON.

{ 
    "flows": [
        {
            "filePattern": "/tmp/app.log*", 
            "deliveryStream": "my-delivery-stream",
            "multiLineStartPattern": "\\[SEQUENCE=",
            "dataProcessingOptions": [
                {
                    "optionName": "SINGLELINE"
                },
                {
                    "optionName": "CSVTOJSON",
                    "customFieldNames": [ "field1", "field2", "field3" ],
                    "delimiter": "\\t"
                }
            ]
        }
    ] 
}

exemplo : Configuração LOGTOJSON com padrão de correspondência

Aqui está um exemplo de configuração LOGTOJSON referente a uma entrada Apache Common Log convertida em formato JSON, com o último campo (bytes) omitido:

{
    "optionName": "LOGTOJSON",
    "logFormat": "COMMONAPACHELOG",
    "matchPattern": "^([\\d.]+) (\\S+) (\\S+) \\[([\\w:/]+\\s[+\\-]\\d{4})\\] \"(.+?)\" (\\d{3})",
    "customFieldNames": ["host", "ident", "authuser", "datetime", "request", "response"]
}

Antes da conversão:

123.45.67.89 - - [27/Oct/2000:09:27:09 -0400] "GET /java/javaResources.html HTTP/1.0" 200

Depois da conversão:

{"host":"123.45.67.89","ident":null,"authuser":null,"datetime":"27/Oct/2000:09:27:09 -0400","request":"GET /java/javaResources.html HTTP/1.0","response":"200"}

Comandos da CLI do agente

Inicie automaticamente o agente durante o startup do sistema:

sudo chkconfig aws-kinesis-agent on

Verifique o status do agente:

sudo service aws-kinesis-agent status

Interrompa o agente:

sudo service aws-kinesis-agent stop

Leia o arquivo de log do agente a partir deste local:

/var/log/aws-kinesis-agent/aws-kinesis-agent.log

Desinstale o agente:

sudo yum remove aws-kinesis-agent

Perguntas frequentes

Existe um Kinesis Agent para Windows?

O Kinesis Agent para Windows é um software diferente das plataformas do Kinesis Agent para Linux.

Por que o Kinesis Agent está ficando mais lento e/ou aumentando os RecordSendErrors?

Isso geralmente ocorre devido ao controle de utilização do Kinesis. Verifique a WriteProvisionedThroughputExceeded métrica do Kinesis Data Streams ThrottledRecords ou a métrica dos streams do Firehose. Qualquer aumento de 0 nessas métricas indica que os limites do fluxo precisam ser aumentados. Para obter mais informações, consulte Limites do Kinesis Data Stream e Streams do Firehose.

Depois de descartar o controle de utilização como causa, verifique se o Kinesis Agent está configurado para seguir um número grande de arquivos pequenos. Há um atraso quando o Kinesis Agent exibe os dados do final de um arquivo novo, portanto, o Kinesis Agent deveria estar exibindo os dados do final de um pequeno número de arquivos maiores. Tente consolidar os arquivos de log em arquivos maiores.

Por que estou recebendo exceções java.lang.OutOfMemoryError ?

O Kinesis Agent não tem memória suficiente para lidar com a workload atual. Tente aumentar JAVA_START_HEAP e JAVA_MAX_HEAP no /usr/bin/start-aws-kinesis-agent e reiniciar o agente.

Por que estou recebendo exceções IllegalStateException : connection pool shut down?

O Kinesis Agent não tem conexões suficientes para lidar com a workload atual. Tente aumentar maxConnections e maxSendingThreads nas configurações gerais do agente em /etc/aws-kinesis/agent.json. O valor padrão para esses campos é 12 vezes o número de processadores de runtime disponíveis. Consulte AgentConfiguration.java para saber mais sobre as configurações avançadas do agente.

Como posso depurar outro problema com o Kinesis Agent?

Os logs do nível DEBUG podem ser habilitados em /etc/aws-kinesis/log4j.xml.

Como devo configurar o Kinesis Agent?

Quanto menor o maxBufferSizeBytes, mais frequentemente o Kinesis Agent enviará dados. Isso pode ser bom, pois diminui o tempo de entrega dos registros, mas também aumenta as solicitações por segundo feitas ao Kinesis.

Por que o Kinesis Agent está enviando registros duplicados?

Isso ocorre devido a uma configuração incorreta da exibição dos dados do final dos arquivos. Certifique-se de que cada fileFlow’s filePattern corresponda a apenas um arquivo. Isso também pode ocorrer se o modo logrotate que está sendo usado estiver no modo copytruncate. Tente mudar o modo para o modo padrão ou criar para evitar duplicações. Para obter mais informações sobre como lidar com registros duplicados, consulte Handling Duplicate Records.