06 ‐ Ingestão de Dados no Elasticsearch
A ingestão de dados é um aspecto crucial do uso do Elasticsearch, permitindo que você colete, enriqueça, transforme e indexe seus dados de várias fontes. Várias ferramentas podem ser usadas para esse fim, cada uma com suas particularidades: Ingest Pipelines, Beats, Elastic Agent e Logstash.
O Ingest Pipeline é um recurso do Elasticsearch que permite definir uma série de processadores que transformam, enriquecem ou manipulam dados antes de serem indexados. Esses pipelines são executados no momento da indexação, diretamente no nó do Elasticsearch, proporcionando uma maneira conveniente e eficiente de pré-processar os dados.
- Funcionalidades: Pode incluir a remoção de campos desnecessários, conversão de formatos de data, extração de partes de strings, entre outras transformações.
- Aplicabilidade: Ideal para transformações leves e quando não se necessita de um processamento externo complexo.
Beats são agentes leves de envio de dados que são instalados nas máquinas para coletar dados e enviá-los diretamente para o Elasticsearch ou para o Logstash para processamento adicional. Existem vários Beats para diferentes propósitos, como Filebeat para logs, Metricbeat para métricas, Packetbeat para tráfego de rede, etc.
- Funcionalidades: Cada Beat é projetado para uma fonte de dados específica, coletando dados e metadados relevantes.
- Aplicabilidade: Útil quando você deseja coletar dados diretamente das fontes sem transformações complexas.
O Elastic Agent é uma ferramenta mais recente e unificada para observabilidade e segurança, podendo gerenciar vários Beats e integrar-se com o Fleet para gerenciamento centralizado. O Elastic Agent simplifica a coleta, transformação e envio de dados para o cluster do Elasticsearch.
- Funcionalidades: Oferece uma maneira unificada de implantar e gerenciar a coleta de dados em sua infraestrutura, podendo substituir o uso de múltiplos Beats.
- Aplicabilidade: Ideal para cenários onde é necessário uma solução de ingestão e gestão mais holística e integrada.
Logstash é uma ferramenta robusta de ingestão de dados que pode coletar, transformar e enviar dados para o Elasticsearch. Diferentemente dos Beats, o Logstash oferece uma ampla variedade de plugins de entrada, filtros e saídas, permitindo processamentos mais complexos e transformações de dados.
- Funcionalidades: Pode filtrar, analisar e transformar dados de diversas formas, com uma vasta gama de plugins disponíveis.
- Aplicabilidade: Recomendado para situações em que é necessário um processamento de dados mais complexo ou quando se trabalha com uma variedade mais ampla de fontes de dados.
-
Beats e Elastic Agent: Direcionados para a coleta rápida e eficiente de dados de várias fontes, com o Elastic Agent oferecendo uma abordagem mais unificada e gerenciável.
-
Logstash e Ingest Pipelines: Enquanto o Logstash é adequado para processamentos mais pesados e complexos, os Ingest Pipelines oferecem uma solução mais leve e direta dentro do próprio Elasticsearch.
-
Integração: Todos esses componentes podem ser combinados de várias maneiras, dependendo dos requisitos específicos de ingestão, processamento e análise de dados.
-
A título de complemento da abrangência das possibilidades de instalação do nosso pacote de serviços. Estamos disponibilizando uma tabela com as referências de instalação para as principais versões do Linux e seus comandos:
Resumo descritivo:
A tabela abaixo foi incluída com o proposito de demonstrar as possibilidade de instalação, considerando as várias versões do Linux onde pode ser instalado o pacote de serviços da Elasticsearch:
| Referência | Comando para Download e Versão do Linux | Comando para Instalação e Versão do Linux |
|---|---|---|
| 1 | wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-8.15.1-linux-x86_64.tar.gz |
tar -xzf elasticsearch-8.15.1-linux-x86_64.tar.gz Linux genérico (Tarball) |
| 2 | wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-8.15.1-x86_64.rpm |
sudo rpm -ivh elasticsearch-8.15.1-x86_64.rpm Red Hat/CentOS/Fedora/Oracle Linux (RPM) |
| 3 | wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-8.15.1-amd64.deb |
sudo dpkg -i elasticsearch-8.15.1-amd64.deb Debian/Ubuntu (DEB) |
| 4 | wget https://artifacts.elastic.co/downloads/kibana/kibana-8.15.1-linux-x86_64.tar.gz |
tar -xzf kibana-8.15.1-linux-x86_64.tar.gz |
| 5 | wget https://artifacts.elastic.co/downloads/kibana/kibana-8.15.1-x86_64.rpm |
sudo rpm -ivh kibana-8.15.1-x86_64.rpm Red Hat/CentOS/Fedora/Oracle Linux (RPM) |
| 6 | wget https://artifacts.elastic.co/downloads/kibana/kibana-8.15.1-amd64.deb |
sudo dpkg -i kibana-8.15.1-amd64.deb Debian/Ubuntu (DEB) |
| 7 | wget https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-8.15.1-linux-x86_64.tar.gz |
tar -xzf filebeat-8.15.1-linux-x86_64.tar.gz Linux genérico (Tarball) |
| 8 | wget https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-8.15.1-x86_64.rpm |
sudo rpm -ivh filebeat-8.15.1-x86_64.rpm Red Hat/CentOS/Fedora/Oracle Linux (RPM) |
| 9 | wget https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-8.15.1-amd64.deb |
sudo dpkg -i filebeat-8.15.1-amd64.deb Debian/Ubuntu (DEB) |
| 10 | wget https://artifacts.elastic.co/downloads/beats/metricbeat/metricbeat-8.15.1-linux-x86_64.tar.gz |
tar -xzf metricbeat-8.15.1-linux-x86_64.tar.gz Linux genérico (Tarball) |
| 11 | wget https://artifacts.elastic.co/downloads/beats/metricbeat/metricbeat-8.15.1-x86_64.rpm |
sudo rpm -ivh metricbeat-8.15.1-x86_64.rpm Red Hat/CentOS/Fedora/Oracle Linux (RPM) |
| 12 | wget https://artifacts.elastic.co/downloads/beats/metricbeat/metricbeat-8.15.1-amd64.deb |
sudo dpkg -i metricbeat-8.15.1-amd64.deb Debian/Ubuntu (DEB) |
Possíveis arquiteturas:
Neste laboratório, você aprenderá a instalar, configurar e utilizar o Filebeat e o Metricbeat no Linux Ubuntu, incluindo a configuração para a coleta de logs do sistema e logs do Tomcat, além de métricas do sistema. Também cobriremos o uso do Keystore para proteger informações sensíveis como senhas.
Nosso objetivo no laboratório a partir desse ponto será entender as funcionalidades das outras ferramentas. Estaremos trabalhando apenas com uma instância do elasticsearch e do kibana.
wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-8.15.1-amd64.deb
sudo dpkg -i elasticsearch-8.15.1-amd64.deb- Onde está instalado o elasticsearch
Quando da instalação do Elasticsearch via pacote deb. O mesmo é instalado no diretório:
/etc/elasticsearch
A estrutura de diretórios já é conhecida
Serão alterados os arquivos:
-
elasticsearch.yml
-
jvm.options
Após as devidas configurações o Elasticsearch deverá ser inicializado através dos comandos:
systemctl start elasticsearch
Formas de verificação de a instancia está no ar:
systemctl status elasticsearch
- Verificando se o serviço está no ar:
ss -lntp | grep 9200
O resultado deverá ser as informações de que a porta 9200 está escutando
- Verificação do log
Caso tenhamos alguma intercorrência uma das formas de verificação das informações referente a saúde do serviço será através do log
tail -f 100 /var/log/elasticsearch/cluster-nome_do_cluster
- Verificação do log: syslog
tail -f 100 /var/log/sys
- Verificando log: Syslog
tail -f 1000 /var/log/syslog | grep elastic
wget https://artifacts.elastic.co/downloads/kibana/kibana-8.15.1-amd64.deb
sudo dpkg -i kibana-8.15.1-amd64.deb
- Instale o Filebeat:
-
Faça o Download no diretório /tmp
-
Utilizando curl
curl -L -O https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-8.15.1-amd64.deb sudo dpkg -i filebeat-8.15.1-amd64.deb
-
Utilizando wget
wget https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-8.15.1-amd64.deb
sudo dpkg -i filebeat-8.15.1-amd64.deb-
Instale o Metricbeat:
curl -L -O https://artifacts.elastic.co/downloads/beats/metricbeat/metricbeat-8.15.1-amd64.deb sudo dpkg -i metricbeat-8.15.1-amd64.deb
wget https://artifacts.elastic.co/downloads/beats/metricbeat/metricbeat-8.15.1-amd64.deb
sudo dpkg -i metricbeat-8.15.1-amd64.deb-
Crie e configure o Keystore para o Filebeat:
sudo filebeat keystore create sudo filebeat keystore add ES_PASSWORD
-
Crie e configure o Keystore para o Metricbeat:
sudo metricbeat keystore create sudo metricbeat keystore add ES_PASSWORD
-
Configure o Filebeat para coletar logs do sistema e logs do Tomcat:
-
Habilite o módulo system:
sudo filebeat modules enable system -
Habilite a coleta do módulo conforme exemplo aaixo:
vi /etc/filebeat/modules.d/system.yml - module: system # Syslog syslog: enabled: true auth: enabled: true
-
Baixe um arquivo de log do Tomcat para teste:
wget https://raw.githubusercontent.com/tornis/esstackenterprise/master/cap04-ingestao/tomcat.log
-
Configure o Filebeat para ler o log do Tomcat com suporte a multiline:
Edite
/etc/filebeat/filebeat.yml:filebeat.inputs: - type: filestream paths: - /tmp/tomcat.log multiline.pattern: '^[[:space:]]+(at|\.{3})[[:space:]]+\b|^Caused by:' multiline.negate: false multiline.match: after setup.kibana: host: "https://localhost:5601" ssl.verification_mode: "none" output.elasticsearch: hosts: ["localhost:9200","localhost:9201","localhost:9202"] preset: balanced protocol: "https" username: "elastic" password: "${ES_PASSWORD}" ssl.verification_mode: "none"
-
-
Configure o Metricbeat para coletar métricas do sistema:
-
Habilite o módulo system:
sudo metricbeat modules enable system -
Atualize a configuração em
/etc/metricbeat/metricbeat.yml:setup.kibana: host: "https://localhost:5601" ssl.verification_mode: "none" output.elasticsearch: hosts: ["localhost:9200","localhost:9201","localhost:9202"] preset: balanced protocol: "https" username: "elastic" password: "${ES_PASSWORD}" ssl.verification_mode: "none"
-
-
Teste as configurações e a conectividade:
-
Filebeat:
sudo filebeat test config sudo filebeat test output
-
Metricbeat:
sudo metricbeat test config sudo metricbeat test modules sudo metricbeat test output
-
-
Execute o setup inicial para carregar dashboards Kibana, index templates e ingest pipelines:
-
Filebeat:
sudo filebeat setup
-
Metricbeat:
sudo metricbeat setup
-
-
Inicie os serviços:
sudo systemctl start filebeat sudo systemctl start metricbeat
Após completar esses passos, você terá o Filebeat configurado para coletar logs do sistema e do Tomcat, utilizando o recurso de multiline para logs do Tomcat, e o Metricbeat configurado para coletar métricas do sistema. Ambos os serviços estão seguramente configurados para comunicar com o Elasticsearch, usando senhas armazenadas no Keystore.
Neste laboratório, você aprenderá a instalar e configurar o Elastic Agent e o Fleet Server no Linux Ubuntu. O Elastic Agent é uma ferramenta unificada para coleta de dados que pode ser gerenciada centralmente pelo Kibana usando o Fleet. Vamos configurar o Elastic Agent para coletar logs e métricas do sistema. Para isso siga os passos abaixo:
- Configurando o Fleet Server
Inicialmente acesse o menu e vá para o Menu>Stack Management>Fleet.
Agora clique em no botão Add Fleet Server
Preencha os campos conforme modelo abaixo. Não esqueça que o endereço ip deve estar acessível pelos agentes e com o https no endereço ok?
Após definido o endereço ip, o Fleet irá instalar a Policy conforme figura abaixo:
Agora ele fornece um passo a passo de como instar o Elastic Agent para que o o Fleet Server fique disponível. Siga os passo abaixo:
Durante a instalação o Elastic Agent irá perguntar se ele será um serviço do Linux e nesse caso digite "Y" para conforme que Sim:
Após instalação o Elastic Agent irá estabelecer a conexão com o Elasticsearch e validar as configurações. A tela abaixo deverá ser apresentada confirmado que o Elatic Agent e o Fleet Server está configurado e funcional:
Após feito a confirmação podemos voltar a tela do Fleet no qual podemos constatar que o Elastic Agent está ativo e configurado com a policy Fleet Server.
Neste tópico, vamos explorar como indexar dados de um banco de dados SQLite no Elasticsearch usando o Logstash. Vamos seguir um passo a passo para preparar o ambiente, configurar o Logstash e finalmente indexar os dados.
-
Instalação do Logstash e SQLite:
Como root, instale o SQLite no Ubuntu:
sudo apt-get update sudo apt-get install sqlite3
Como usuário elastic no Ubuntu:
su - elastic bash cd /opt/elastic wget https://artifacts.elastic.co/downloads/logstash/logstash-8.15.1-linux-x86_64.tar.gz tar zxfv logstash-8.15.1-linux-x86_64.tar.gz cd logstash-8.15.1
-
Criação da Pasta Datasets:
Mude para o usuário
elastice crie a pastadatasetsem/opt/elastic:sudo -u elastic mkdir -p /opt/elastic/datasets
-
Download do Banco de Dados de Exemplo:
Baixe o banco de dados SQLite de exemplo para a pasta
/opt/elastic/datasets:sudo -u elastic wget -P /opt/elastic/datasets https://raw.githubusercontent.com/tornis/esstackenterprise/master/datasets/chinook.db
-
Download do Driver JDBC do SQLite:
Baixe o driver JDBC do SQLite para a pasta
/opt/elastic/datasets:sudo -u elastic wget -P /opt/elastic/datasets https://github.com/tornis/esstackenterprise/raw/master/datasets/sqlite-jdbc-3.7.2.jar
-
Download e Descompactação do Logstash:
Baixe e descompacte o Logstash:
sudo -u elastic wget -P /opt/elastic https://artifacts.elastic.co/downloads/logstash/logstash-8.12.1-linux-x86_64.tar.gz cd /opt/elastic sudo -u elastic tar -xzf logstash-8.12.1-linux-x86_64.tar.gz cd logstash-8.12.1
Após preparar o ambiente, vamos configurar o pipeline do Logstash para indexar os dados no Elasticsearch.
-
Criação do Arquivo de Configuração:
Crie o arquivo
config/database.confno diretório do Logstash com o seguinte conteúdo:sudo -u elastic vi config/database.conf
input { jdbc { jdbc_driver_library => "/opt/elastic/datasets/sqlite-jdbc-3.7.2.jar" jdbc_driver_class => "org.sqlite.JDBC" jdbc_connection_string => "jdbc:sqlite:/opt/elastic/datasets/chinook.db" jdbc_user => "" schedule => "* * * * *" tracking_column => "invoicedate" tracking_column_type => "timestamp" use_column_value => true statement => "SELECT * from Customer c JOIN Invoice i on i.CustomerId = c.CustomerId WHERE i.InvoiceDate > :sql_last_value ORDER BY i.InvoiceDate ASC limit 5" } } filter { date { match => [ "invoicedate", "yyyy-MM-dd HH:mm:ss" ] } } output { elasticsearch { hosts => ["https://localhost:9200","localhost:9201","https://localhost:9202"] ssl_enabled => true ssl_verification_mode => "none" index => "chinook-%{+YYYY.MM.dd}" user => "elastic" password => "123456" } }Explicação:
-
input/jdbc: Define a configuração para coletar dados do banco SQLite usando JDBC.
-
jdbc_driver_library: Caminho para o driver JDBC do SQLite. -
jdbc_driver_class: Classe do driver JDBC. -
jdbc_connection_string: String de conexão com o banco de dados. -
jdbc_user: Nome do usuário para conexão (vazio para SQLite). -
schedule: Cronograma para executar a consulta. -
tracking_column: Coluna usada para rastrear as últimas alterações. -
statement: Consulta SQL para coletar dados.
-
-
filter/date: Configura o filtro para analisar o campo de data.
-
output/elasticsearch: Define como os dados serão enviados para o Elasticsearch.
-
hosts: Endereços dos nós do Elasticsearch. -
ssl_enabledessl_verification_mode: Configurações para o uso de SSL. -
index: Padrão de nomeação do índice no Elasticsearch. -
userepassword: Credenciais para o Elasticsearch.
-
-
Para ajustar o Logstash para carregar a configuração específica do arquivo database.conf, precisamos editar o arquivo pipelines.yml, que gerencia múltiplos pipelines dentro do Logstash. Veja como proceder:
-
Editar o Arquivo
pipelines.yml:No diretório de instalação do Logstash, edite ou crie o arquivo
pipelines.yml. Especifique o caminho do arquivo de configuraçãodatabase.conf:cd /opt/elastic/logstash-8.12.1 sudo -u elastic vi config/pipelines.yml- pipeline.id: chinook_pipeline path.config: "/opt/elastic/logstash-8.12.1/config/database.conf"
Aqui,
pipeline.idé um identificador único para o pipeline, epath.configaponta para o arquivo de configuração que acabamos de criar.
Após configurar o arquivo pipelines.yml, podemos iniciar o Logstash para processar os dados conforme definido.
-
Iniciar o Logstash:
Navegue até o diretório do Logstash e execute o seguinte comando para iniciar o processo de ingestão:
sudo -u elastic /opt/elastic/logstash-8.12.1/bin/logstash
Esse comando inicia o Logstash e carrega a configuração especificada, iniciando a coleta de dados do banco de dados SQLite e indexando-os no Elasticsearch.
Ao executar esses passos, o Logstash irá periodicamente consultar o banco de dados SQLite conforme definido na consulta SQL em database.conf, e indexar os resultados no Elasticsearch. Você pode verificar o sucesso da operação observando os logs do Logstash e também verificando se os dados aparecem no índice especificado no Elasticsearch.
Nesta seção, exploraremos uma outra abordagem de ingestão e tratamento de dados: os Ingest Pipelines. Uma distinção importante dos Ingest Pipelines em relação ao Logstash e aos Beats é que eles não buscam ativamente os dados para tratamento; em vez disso, os Ingest Pipelines recebem e processam as informações diretamente.
Esta técnica oferece uma maneira eficiente de realizar transformações e preparações nos dados antes de serem indexados no Elasticsearch. Ao compreender e utilizar os Ingest Pipelines, podemos otimizar o fluxo de dados e garantir que estejam prontos para serem consultados e analisados de forma eficaz.
No Kibana, abra o menu principal e clique em Stack Management > Ingest Pipelines. Para criar um pipeline, clique em Create Pipeline > New Pipeline. Também é possível usar as APIs de ingestão no Dev Tools para criar e gerenciar pipelines. Sempre importante lembrar que os processors definidos são executados sequencialmente na ordem especificada.
Antes de usar um pipeline em produção, é recomendado que você teste ele utilizando documentos de amostra. Ao criar ou editar um pipeline no Kibana, clique em "Add documents". Na aba Documents, forneça documentos de amostra e clique em "Run the pipeline". Da mesma forma, o Dev Tools também permite testar pipelines usando a API de simulação de pipeline. Você pode especificar um pipeline configurado no caminho da solicitação.
PUT _ingest/pipeline/meu_pipeline
{
"processors": [
{
"set": {
"field": "idade",
"value": "40"
}
},
{
"dissect": {
"field": "classificacao",
"pattern": "%{fornecedor_status} %{fornecedor_negativado} %{fornecedor_ativo_receita}"
}
},
{
"html_strip": {
"field": "descricao_fornecedor"
}
}
]
}
- Para cada documento:
POST fornecedor/_doc/3?pipeline=meu_pipeline
{
"codigo": 124,
"nome_fornecedor": "GOL Linhas Aereas",
"cnpj": "12222334000102",
"endereco": "Rua tal e etc",
"bairro": "Fulano",
"cidade": "Caxias",
"uf": "RJ",
"classificacao": "sim não ativo",
"descricao_fornecedor": "<br>teste de remoçao html</br>"
}- Para o índice:
PUT fornecedor_teste
{
"settings": {
"default_pipeline": "meu_pipeline"
}
}POST fornecedor_teste/_doc/1
{
"codigo": 124,
"nome_fornecedor": "GOL Linhas Aereas",
"cnpj": "12222334000102",
"endereco": "Rua tal e etc",
"bairro": "Fulano",
"cidade": "Caxias",
"uf": "RJ",
"classificacao": "sim não ativo",
"descricao_fornecedor": "<br>teste de remoçao html</br>"
}