07 ‐ Ingestão de Dados no Elasticsearch - BktechBrazil/elastic-eck GitHub Wiki

Ingestão de dados é o processo de coletar informações de diversas fontes, transformá-las se necessário, e indexá-las no Elasticsearch. O Elastic Stack oferece quatro abordagens principais para isso, cada uma com características e casos de uso distintos.

O Ingest Pipeline é um recurso nativo do Elasticsearch que permite processar documentos dentro do próprio cluster, antes de indexá-los. É executado no momento da indexação, diretamente no node ingest, sem necessidade de ferramentas externas.

• Quando usar: transformações leves e pontuais — extrair campos, converter formatos de data, remover campos desnecessários, enriquecer com GeoIP.
• Quando não usar: processamento complexo, múltiplas fontes de dados, roteamento condicional para destinos diferentes.

Beats são agentes leves instalados diretamente nos servidores para coletar dados e enviá-los ao Elasticsearch ou ao Logstash. Cada Beat é especializado em um tipo de dado.

• Quando usar: coleta simples e direta de logs, métricas ou eventos — sem necessidade de transformações complexas.
• Quando não usar: quando o dado precisa de enriquecimento pesado ou roteamento para múltiplos destinos.

O Elastic Agent é a evolução dos Beats — um único agente que substitui múltiplos Beats e é gerenciado centralmente pelo Fleet no Kibana. É a abordagem recomendada em novas implementações com Elasticsearch 9.x.

• Quando usar: novas implementações, ambientes onde gestão centralizada é importante, integrações com observabilidade e segurança.
• Quando não usar: ambientes muito restritos onde um agente leve e minimal é mandatório.

Logstash é uma ferramenta de processamento de dados com três estágios — input, filter e output. É a opção mais poderosa para transformações complexas e integração com fontes de dados heterogêneas.

• Quando usar: parsing complexo de logs, ingestão de bancos de dados via JDBC, múltiplos destinos de output, lógica condicional avançada.
• Quando não usar: quando a simplicidade importa e o dado não precisa de transformação — use Beats ou Elastic Agent.

Possíveis arquiteturas:

No Kubernetes, Beats, Elastic Agent e Logstash são gerenciados pelo mesmo operator ECK que gerencia o Elasticsearch e o Kibana, cada um com seu próprio recurso (CRD):

Nos manifestos, o campo elasticsearchRef (ou elasticsearchRefs) aponta para o cluster de destino. A partir dele, o operator configura sozinho o endereço do Elasticsearch, as credenciais e a CA do TLS: não há senha nem certificado para copiar à mão. Os Ingest Pipelines não mudam: são configurados pela API do Elasticsearch, como em qualquer outra instalação.

Os Beats são agentes leves instalados diretamente nos servidores para coletar dados e enviá-los ao Elasticsearch ou ao Logstash. Cada Beat é especializado em um tipo de dado:

Cada Beat tem sua própria configuração YAML, seus próprios módulos e precisa ser configurado e gerenciado individualmente. No ECK, essa configuração fica no campo spec.config do recurso Beat.

O Elastic Agent foi criado para resolver os problemas operacionais dos Beats individuais:

• Uma instalação, múltiplos propósitos: em vez de instalar Filebeat, Metricbeat e Auditbeat separadamente em cada servidor, o Elastic Agent substitui todos com uma única instalação
• Gestão centralizada via Fleet: todas as configurações, políticas e atualizações são feitas no Kibana e propagadas automaticamente para todos os agents — sem precisar acessar cada servidor individualmente
• Integrations: o Elastic Agent usa o sistema de Integrations do Kibana, que oferece centenas de integrações prontas (Nginx, MySQL, Windows, AWS, Azure, etc.) com configuração via interface gráfica
• Atualizações remotas: é possível atualizar todos os agents de um ambiente inteiro diretamente pelo Kibana

Em novas implementações com Elasticsearch 9.x, use sempre o Elastic Agent. Os Beats individuais continuam disponíveis e suportados, mas o foco de desenvolvimento e de novas funcionalidades da Elastic está no Elastic Agent. Na prática, o Filebeat e o Metricbeat que você veria sendo configurados manualmente no passado são hoje substituídos pelas integrações do Elastic Agent gerenciadas pelo Fleet.

O Elastic Agent funciona de dois modos:

• Standalone: a configuração de coleta fica no próprio manifesto (spec.config do recurso Agent). É simples, versionável em Git e não depende do Kibana.
• Gerenciado pelo Fleet: o agente se registra em um Fleet Server e recebe as políticas de coleta definidas na interface do Fleet, no Kibana. É o modo mais usado em produção, porque centraliza a gestão de muitos agentes. No ECK, o Fleet Server é apenas outro recurso Agent, com mode: fleet e fleetServerEnabled: true.

Para ler os logs dos containers, os pods de coleta montam diretórios do node (/var/log/containers e /var/log/pods) com volumes hostPath e rodam como root (runAsUser: 0). Para enriquecer os eventos com dados do Kubernetes (nome do pod, namespace, labels), eles usam uma ServiceAccount com permissões de leitura na API do cluster (RBAC). Por isso, os manifestos dos laboratórios trazem, além do recurso Beat ou Agent, uma ServiceAccount, um ClusterRole e um ClusterRoleBinding.

Os laboratórios usam o cluster lab-es com um node e o Kibana lab-kb. Os comandos de terminal são executados na VM, no diretório de trabalho onde fica a pasta manifests; as consultas, no Console do Kibana.

Um coletor por vez. Filebeat e Elastic Agent pedem de 200 a 500 MiB cada; o Logstash, cerca de 1 GiB; o Fleet Server com o Agent, mais ainda. Cada laboratório termina removendo o que criou antes de passar ao próximo.

kubectl -n elastic get elasticsearch,kibana
kubectl get crd | grep -E 'beats|agents|logstashes'

Resultado esperado: lab-es e lab-kb com HEALTH green, e os três CRDs instalados junto com o operator: agents.agent.k8s.elastic.co, beats.beat.k8s.elastic.co e logstashes.logstash.k8s.elastic.co.

PASSWORD=$(kubectl -n elastic get secret lab-es-es-elastic-user \
  -o go-template='{{.data.elastic | base64decode}}')
echo "Senha do elastic: $PASSWORD"

pkill -f "port-forward.*5601" || true
nohup kubectl -n elastic port-forward --address 0.0.0.0 service/lab-kb-kb-http 5601 > /tmp/pf-kibana.log 2>&1 &
sleep 2
cat /tmp/pf-kibana.log

Resultado esperado: a senha do usuário elastic e a mensagem Forwarding from 0.0.0.0:5601 -> 5601.

O túnel fica ligado ao pod do Kibana que estava rodando quando ele foi aberto. Se esse pod for reiniciado ou recriado, o túnel cai e o Kibana deixa de responder no navegador: execute este bloco novamente.

No navegador, abra https://<IP_DA_VM>:5601, faça login com o usuário elastic e abra o Console (Dev Tools).

Réplicas nos laboratórios: os índices e data streams criados pelos coletores nascem com 1 réplica. Como o lab-es tem um único node, a réplica não tem onde ser alocada e o índice fica yellow. Cada laboratório mostra como ajustar para 0 réplicas depois que os dados chegam.

Neste laboratório, um Filebeat roda como DaemonSet, lê os logs de todos os containers do node e envia direto para o lab-es.

mkdir -p manifests
cat <<'EOF' > manifests/filebeat.yaml
# Filebeat via CRD Beat do ECK — coleta logs de containers em todos os nós (DaemonSet)
# Namespace: elastic. RBAC incluído para o add_kubernetes_metadata enriquecer os eventos.
---
apiVersion: v1
kind: ServiceAccount
metadata:
  name: filebeat
  namespace: elastic
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: filebeat
rules:
  - apiGroups: [""]
    resources: ["namespaces", "pods", "nodes"]
    verbs: ["get", "watch", "list"]
  - apiGroups: ["apps"]
    resources: ["replicasets"]
    verbs: ["get", "list", "watch"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: filebeat
subjects:
  - kind: ServiceAccount
    name: filebeat
    namespace: elastic
roleRef:
  kind: ClusterRole
  name: filebeat
  apiGroup: rbac.authorization.k8s.io
---
apiVersion: beat.k8s.elastic.co/v1beta1
kind: Beat
metadata:
  name: filebeat
  namespace: elastic
spec:
  type: filebeat
  version: 9.4.2
  elasticsearchRef:
    name: lab-es            # operator injeta host, usuário, senha e CA automaticamente
  kibanaRef:
    name: lab-kb            # habilita o setup de dashboards do Filebeat
  config:
    filebeat.inputs:
      - type: filestream
        id: container-logs
        paths:
          - /var/log/containers/*.log
        parsers:
          - container: {}
        prospector:
          scanner:
            fingerprint.enabled: true
            symlinks: true
        file_identity.fingerprint: {}
    processors:
      - add_kubernetes_metadata:
          host: ${NODE_NAME}
          matchers:
            - logs_path:
                logs_path: "/var/log/containers/"
  daemonSet:
    podTemplate:
      spec:
        serviceAccountName: filebeat
        automountServiceAccountToken: true
        terminationGracePeriodSeconds: 30
        hostNetwork: true
        dnsPolicy: ClusterFirstWithHostNet
        securityContext:
          runAsUser: 0
        containers:
          - name: filebeat
            resources:
              requests:
                memory: 200Mi
                cpu: 100m
              limits:
                memory: 300Mi
            env:
              - name: NODE_NAME
                valueFrom:
                  fieldRef:
                    fieldPath: spec.nodeName
            volumeMounts:
              - name: varlogcontainers
                mountPath: /var/log/containers
              - name: varlogpods
                mountPath: /var/log/pods
              - name: varlibdockercontainers
                mountPath: /var/lib/docker/containers
        volumes:
          - name: varlogcontainers
            hostPath:
              path: /var/log/containers
          - name: varlogpods
            hostPath:
              path: /var/log/pods
          - name: varlibdockercontainers
            hostPath:
              path: /var/lib/docker/containers
EOF

Os pontos principais do manifesto:

• elasticsearchRef: lab-es: o operator configura a saída para o lab-es, com usuário, senha e CA;
• kibanaRef: lab-kb: permite ao Filebeat carregar os seus dashboards no Kibana;
• input filestream com o parser container: lê os arquivos de log no formato do runtime de containers. Os arquivos em /var/log/containers são links simbólicos para /var/log/pods, por isso symlinks: true;
• add_kubernetes_metadata: acrescenta a cada evento o pod, o namespace e os labels de origem, usando a ServiceAccount filebeat e o RBAC do início do arquivo;
• daemonSet: um pod do Filebeat em cada node do cluster.

O input container, que aparece em exemplos antigos, é um atalho do input log, que vem desativado desde a versão 9.0. Na 9.4.2, use sempre o filestream com o parser container.

kubectl apply -f manifests/filebeat.yaml
kubectl -n elastic get beat filebeat -w

Resultado esperado: depois de alguns segundos, HEALTH green, AVAILABLE 1 e EXPECTED 1. Pressione Ctrl+C e confira o pod:

kubectl -n elastic get pods | grep filebeat

Resultado esperado: um pod filebeat-beat-filebeat-<sufixo> em 1/1 Running.

No Console:

GET /_data_stream/filebeat-*

O Filebeat grava no data stream filebeat-9.4.2. Se a saúde (status) aparecer YELLOW, ajuste as réplicas dos índices já criados:

PUT /filebeat-*/_settings
{
  "index.number_of_replicas": 0
}

Confira a quantidade de eventos:

GET /filebeat-*/_count

kubectl delete -f manifests/filebeat.yaml

Resultado esperado: a ServiceAccount, o ClusterRole, o ClusterRoleBinding e o Beat filebeat são removidos. Os dados já indexados continuam no lab-es.

Neste laboratório, o Logstash entra no meio do caminho: um Filebeat envia os logs para o Logstash, que acrescenta um campo e grava no Elasticsearch.

cat <<'EOF' > manifests/logstash.yaml
# Logstash via CRD Logstash do ECK — pipeline central beats -> filtro -> elasticsearch
# O operator injeta as variáveis LAB_ES_ES_* a partir do elasticsearchRefs (clusterName: lab-es).
apiVersion: logstash.k8s.elastic.co/v1alpha1
kind: Logstash
metadata:
  name: lab-logstash
  namespace: elastic
spec:
  count: 1
  version: 9.4.2
  elasticsearchRefs:
    - name: lab-es
      clusterName: lab-es          # gera env: LAB_ES_ES_HOSTS, LAB_ES_ES_USER, LAB_ES_ES_PASSWORD, LAB_ES_ES_SSL_CERTIFICATE_AUTHORITY
  pipelines:
    - pipeline.id: main
      config.string: |
        input {
          beats { port => 5044 }
        }
        filter {
          mutate { add_field => { "[origem]" => "logstash-lab" } }
        }
        output {
          elasticsearch {
            hosts    => [ "${LAB_ES_ES_HOSTS}" ]
            user     => "${LAB_ES_ES_USER}"
            password => "${LAB_ES_ES_PASSWORD}"
            ssl_enabled                 => true
            ssl_certificate_authorities => "${LAB_ES_ES_SSL_CERTIFICATE_AUTHORITY}"
            index    => "logstash-lab-%{+YYYY.MM.dd}"
          }
        }
  # Service para os Beats enviarem eventos (porta 5044)
  services:
    - name: beats
      service:
        spec:
          type: ClusterIP
          ports:
            - port: 5044
              name: filebeat
              protocol: TCP
              targetPort: 5044
  podTemplate:
    spec:
      containers:
        - name: logstash
          resources:
            requests:
              memory: 1Gi
              cpu: 500m
            limits:
              memory: 1536Mi
EOF

Os pontos principais do manifesto:

• elasticsearchRefs com clusterName: lab-es: o operator cria no pod as variáveis de ambiente LAB_ES_ES_HOSTS, LAB_ES_ES_USER, LAB_ES_ES_PASSWORD e LAB_ES_ES_SSL_CERTIFICATE_AUTHORITY. O prefixo vem do clusterName, em maiúsculas e com - trocado por _;
• pipelines: o pipeline do Logstash (o equivalente ao arquivo .conf) fica no próprio manifesto, em config.string, e usa essas variáveis na saída elasticsearch;
• services: cria o Service lab-logstash-ls-beats, na porta 5044, por onde o Filebeat vai enviar os eventos. O nome segue o padrão <nome do Logstash>-ls-<nome do service>.

O usuário que o operator cria para o Logstash (papel eck_logstash_user_role) só pode escrever em índices com nomes como logstash-*, logs-* e metrics-*. Para gravar em outros nomes, é preciso ampliar esse papel no Elasticsearch. Por isso, todos os índices do Logstash neste capítulo começam com logstash-.

kubectl apply -f manifests/logstash.yaml
kubectl -n elastic get pods -w

Resultado esperado: o pod lab-logstash-ls-0 chega a 1/1 Running em um ou dois minutos. Pressione Ctrl+C e confira o recurso e os Services:

kubectl -n elastic get logstash lab-logstash
kubectl -n elastic get svc | grep lab-logstash

Resultado esperado: o Logstash com AVAILABLE 1 e EXPECTED 1, e os Services lab-logstash-ls-api (porta 9600, API de monitoramento) e lab-logstash-ls-beats (porta 5044).

cat <<'EOF' > manifests/filebeat-para-logstash.yaml
# Variante do Filebeat que envia para o LOGSTASH (não direto ao ES).
# Demonstra o fluxo Beats -> Logstash -> Elasticsearch.
# O Service criado pelo CRD Logstash chama-se <nome>-ls-<service> => lab-logstash-ls-beats:5044
---
apiVersion: v1
kind: ServiceAccount
metadata:
  name: filebeat-ls
  namespace: elastic
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: filebeat-ls
rules:
  - apiGroups: [""]
    resources: ["namespaces", "pods", "nodes"]
    verbs: ["get", "watch", "list"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: filebeat-ls
subjects:
  - kind: ServiceAccount
    name: filebeat-ls
    namespace: elastic
roleRef:
  kind: ClusterRole
  name: filebeat-ls
  apiGroup: rbac.authorization.k8s.io
---
apiVersion: beat.k8s.elastic.co/v1beta1
kind: Beat
metadata:
  name: filebeat-ls
  namespace: elastic
spec:
  type: filebeat
  version: 9.4.2
  # SEM elasticsearchRef: a saída vai para o Logstash, definida no config abaixo.
  config:
    filebeat.inputs:
      - type: filestream
        id: container-logs
        paths:
          - /var/log/containers/*.log
        parsers:
          - container: {}
        prospector:
          scanner:
            fingerprint.enabled: true
            symlinks: true
        file_identity.fingerprint: {}
    output.elasticsearch.enabled: false
    output.logstash:
      hosts: ["lab-logstash-ls-beats:5044"]
  daemonSet:
    podTemplate:
      spec:
        serviceAccountName: filebeat-ls
        automountServiceAccountToken: true
        hostNetwork: true
        dnsPolicy: ClusterFirstWithHostNet
        securityContext:
          runAsUser: 0
        containers:
          - name: filebeat
            resources:
              requests:
                memory: 200Mi
                cpu: 100m
              limits:
                memory: 300Mi
            volumeMounts:
              - name: varlogcontainers
                mountPath: /var/log/containers
              - name: varlogpods
                mountPath: /var/log/pods
        volumes:
          - name: varlogcontainers
            hostPath:
              path: /var/log/containers
          - name: varlogpods
            hostPath:
              path: /var/log/pods
EOF

Este Beat não tem elasticsearchRef: a saída é o output.logstash, apontando para o Service lab-logstash-ls-beats. A entrada é a mesma do laboratório anterior.

kubectl apply -f manifests/filebeat-para-logstash.yaml
kubectl -n elastic get beat filebeat-ls -w

Resultado esperado: HEALTH green, AVAILABLE 1 e EXPECTED 1. Pressione Ctrl+C.

No Console:

GET /_cat/indices/logstash-lab-*?v

Ajuste as réplicas, se o índice aparecer yellow:

PUT /logstash-lab-*/_settings
{
  "index.number_of_replicas": 0
}

Confira que os eventos passaram pelo filtro do Logstash:

GET /logstash-lab-*/_search
{
  "size": 1,
  "query": {
    "term": { "origem.keyword": "logstash-lab" }
  }
}

O campo origem: logstash-lab, acrescentado pelo mutate, prova que os eventos passaram pelo Logstash. Se quiser explorar no Discover, crie um data view com o padrão logstash-lab-*, como no laboratório anterior.

kubectl delete -f manifests/filebeat-para-logstash.yaml
kubectl delete -f manifests/logstash.yaml

Neste laboratório, o Logstash lê um banco SQLite via JDBC e indexa os dados de forma incremental. O banco de exemplo (Chinook) e o driver JDBC precisam estar dentro do pod do Logstash. No ECK, isso é feito com um initContainer: um container que roda antes do Logstash, baixa os arquivos para um volume persistente e termina. O Logstash monta o mesmo volume e encontra os arquivos prontos.

Os pods precisam de acesso à internet para baixar o banco e o driver do GitHub.

No Console, crie o índice com 0 réplicas antes de subir o Logstash:

PUT /logstash-chinook
{
  "settings": {
    "number_of_shards": 1,
    "number_of_replicas": 0
  }
}

cat <<'EOF' > manifests/logstash-jdbc.yaml
apiVersion: logstash.k8s.elastic.co/v1alpha1
kind: Logstash
metadata:
  name: lab-logstash-jdbc
  namespace: elastic
spec:
  count: 1
  version: 9.4.2
  elasticsearchRefs:
    - name: lab-es
      clusterName: lab-es
  pipelines:
    - pipeline.id: chinook_pipeline
      config.string: |
        input {
          jdbc {
            jdbc_driver_library    => "/usr/share/logstash/datasets/sqlite-jdbc-3.7.2.jar"
            jdbc_driver_class      => "org.sqlite.JDBC"
            jdbc_connection_string => "jdbc:sqlite:/usr/share/logstash/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                       => [ "${LAB_ES_ES_HOSTS}" ]
            user                        => "${LAB_ES_ES_USER}"
            password                    => "${LAB_ES_ES_PASSWORD}"
            ssl_enabled                 => true
            ssl_certificate_authorities => "${LAB_ES_ES_SSL_CERTIFICATE_AUTHORITY}"
            index                       => "logstash-chinook"
            data_stream                 => false
            ilm_enabled                 => false
            manage_template             => false
          }
        }
  volumeClaimTemplates:
    - metadata:
        name: datasets
      spec:
        accessModes:
          - ReadWriteOnce
        resources:
          requests:
            storage: 50Mi
  podTemplate:
    spec:
      initContainers:
        - name: download-datasets
          command: ["/bin/sh"]
          args:
            - "-c"
            - >-
              curl -sSL -o /data/chinook.db https://raw.githubusercontent.com/tornis/esstackenterprise/master/datasets/chinook.db &&
              curl -sSL -o /data/sqlite-jdbc-3.7.2.jar https://github.com/tornis/esstackenterprise/raw/master/datasets/sqlite-jdbc-3.7.2.jar
          volumeMounts:
            - name: datasets
              mountPath: /data
      containers:
        - name: logstash
          volumeMounts:
            - name: datasets
              mountPath: /usr/share/logstash/datasets
          resources:
            requests:
              memory: 1Gi
              cpu: 500m
            limits:
              memory: 1536Mi
EOF

Explicação dos principais parâmetros do pipeline:

• jdbc_driver_library: caminho para o driver JDBC do banco de dados, no volume datasets
• jdbc_driver_class: classe Java do driver
• jdbc_connection_string: string de conexão com o banco — para SQLite, aponta para o arquivo .db
• jdbc_user: usuário do banco (vazio para SQLite, que não usa autenticação)
• schedule: define a frequência de execução no formato cron — * * * * * executa a cada minuto
• tracking_column + use_column_value: controla a ingestão incremental — o Logstash guarda o último valor lido da coluna de rastreamento e na próxima execução busca apenas registros novos. Esse último valor fica no volume persistente logstash-data, que o ECK cria para todo Logstash, e sobrevive a um reinício do pod
• date: usa a data da fatura (invoicedate) como @timestamp do documento
• data_stream, ilm_enabled e manage_template desligados: o Logstash grava no índice logstash-chinook, criado no Passo 1, sem criar data stream, alias de ILM ou template próprio

E no manifesto:

• volumeClaimTemplates: cria o volume persistente datasets, de 50 MiB;
• initContainers: o container download-datasets baixa o banco chinook.db e o driver sqlite-jdbc-3.7.2.jar para o volume. Sem o campo image, ele usa a mesma imagem do Logstash;
• o container logstash monta o volume em /usr/share/logstash/datasets, o caminho usado no pipeline;
• a senha do Elasticsearch não aparece em lugar nenhum: vem das variáveis LAB_ES_ES_*, criadas pelo operator.

kubectl apply -f manifests/logstash-jdbc.yaml
kubectl -n elastic get pods -w

Resultado esperado: o pod lab-logstash-jdbc-ls-0 passa por Init:0/1 (o download) e chega a 1/1 Running. Pressione Ctrl+C e acompanhe o log:

kubectl -n elastic logs -f lab-logstash-jdbc-ls-0

Resultado esperado: a cada minuto, uma linha com a consulta SELECT * FROM Customer c JOIN Invoice i ... executada pelo input jdbc. Pressione Ctrl+C.

No Console:

GET /_cat/indices/logstash-chinook?v

GET /logstash-chinook/_count

Repita o _count depois de um minuto: a contagem aumenta de 5 em 5, o LIMIT da consulta. Veja um documento:

GET /logstash-chinook/_search
{
  "size": 1,
  "sort": [{ "@timestamp": "desc" }]
}

Os nomes das colunas chegam em minúsculas (invoicedate, firstname, country...) e o @timestamp é a data da fatura.

kubectl delete -f manifests/logstash-jdbc.yaml

Neste laboratório vamos usar o Centralized Pipeline Management do Logstash — um recurso que permite criar e gerenciar pipelines diretamente pelo Kibana, sem precisar reiniciar o Logstash para aplicar mudanças. O pipeline não fica no manifesto: o Logstash busca no Elasticsearch os pipelines cadastrados pelo Kibana.

O Centralized Pipeline Management é um recurso pago, que não está disponível na licença Basic. Para o laboratório, vamos ativar a licença trial de 30 dias.

No ECK, a trial pode ser ativada de duas formas:

Para este laboratório, basta a trial do cluster, ativada pelo Kibana:

1. No Kibana, abra a página License Management pelo menu Stack Management ou digitando License Management na busca global.
2. A página mostra a licença atual, Basic. No quadro de trial, clique em Start trial e confirme na janela que se abre.
3. A página passa a mostrar a licença Trial, com a data de expiração.

Confira também no Console:

GET /_license

O campo type deve ser trial e o status, active.

Para ativar:

POST _license/start_trial?acknowledge=true

Alternativa — trial pelo operator ECK: para ativar a trial também para o operator, crie o Secret abaixo no terminal da VM, em vez de usar o Kibana:

cat <<'EOF' | kubectl apply -f -
apiVersion: v1
kind: Secret
metadata:
  name: eck-trial-license
  namespace: elastic-system
  labels:
    license.k8s.elastic.co/type: enterprise_trial
  annotations:
    elastic.co/eula: accepted
EOF

A anotação elastic.co/eula: accepted indica que você aceita o EULA da Elastic.

Atenção: a trial do Elasticsearch pode ser ativada uma única vez e dura 30 dias. Durante esse período, os recursos pagos (Platinum e Enterprise) ficam disponíveis também para os demais laboratórios do curso. Ao fim do prazo, esses recursos passam a funcionar em modo degradado.

O dataset de qualidade do ar será gerado na VM com um script Python. Isso garante que o arquivo tem exatamente a estrutura esperada pelo pipeline e pelo mapping do índice.

Criando o script gerador:

mkdir -p datasets
cat <<'EOF' > datasets/gerar_air_quality.py
import csv
import random
from datetime import datetime, timedelta
 
stations = [
    ("ST001", "PM10",  "Particulate Matter 10", "µg/m³", "CETESB"),
    ("ST001", "PM2.5", "Particulate Matter 2.5", "µg/m³", "CETESB"),
    ("ST002", "NO2",   "Nitrogen Dioxide",       "µg/m³", "CETESB"),
    ("ST002", "O3",    "Ozone",                  "µg/m³", "CETESB"),
    ("ST003", "CO",    "Carbon Monoxide",         "mg/m³", "INMET"),
    ("ST003", "SO2",   "Sulfur Dioxide",          "µg/m³", "INMET"),
    ("ST004", "PM10",  "Particulate Matter 10",   "µg/m³", "INMET"),
    ("ST004", "NO2",   "Nitrogen Dioxide",        "µg/m³", "INMET"),
]
 
start_date = datetime(2024, 1, 1, 0, 0, 0)
rows = []
 
for day in range(90):
    for hour in range(0, 24, 3):
        dt = start_date + timedelta(days=day, hours=hour)
        for station_id, code, name, dimension, source in stations:
            value = round(random.uniform(5.0, 150.0), 2)
            rows.append([
                station_id,
                code,
                name,
                dimension,
                value,
                dt.strftime("%Y-%m-%d %H:%M:%S"),
                source
            ])
 
output_path = "datasets/air_quality.csv"
with open(output_path, "w", newline="") as f:
    writer = csv.writer(f)
    writer.writerow(["id_station", "indicator_code", "indicator_name",
                     "indicator_dimension", "value", "datetime", "source"])
    writer.writerows(rows)
 
print(f"Arquivo gerado: {output_path}")
print(f"Total de registros: {len(rows)}")
EOF

Executando o script:

python3 datasets/gerar_air_quality.py
head -5 datasets/air_quality.csv

Resultado esperado:

Arquivo gerado: datasets/air_quality.csv
Total de registros: 5760

Seguido do cabeçalho e das quatro primeiras linhas do arquivo.

O arquivo precisa estar dentro do pod do Logstash. Como tem menos de 1 MiB, cabe em um ConfigMap, que será montado como arquivo no pod:

kubectl -n elastic create configmap air-quality-csv --from-file=air_quality.csv=datasets/air_quality.csv
kubectl -n elastic get configmap air-quality-csv

Resultado esperado: configmap/air-quality-csv created e o ConfigMap listado com DATA 1.

O ConfigMap é criado com kubectl create, e não com apply: o apply guarda uma cópia do conteúdo em uma anotação, que tem limite de tamanho menor que o do ConfigMap.

No Console:

PUT /logstash-air-quality
{
  "settings": {
    "number_of_shards": 1,
    "number_of_replicas": 0
  },
  "mappings": {
    "properties": {
      "id_station":          { "type": "keyword" },
      "indicator_code":      { "type": "keyword" },
      "indicator_name": {
        "type": "text",
        "fields": { "keyword": { "type": "keyword" } }
      },
      "indicator_dimension": { "type": "keyword" },
      "value":               { "type": "float" },
      "datetime":            { "type": "date" },
      "source":              { "type": "keyword" }
    }
  }
}

As configurações que ficariam no logstash.yml vão no campo spec.config do recurso Logstash:

cat <<'EOF' > manifests/logstash-cpm.yaml
apiVersion: logstash.k8s.elastic.co/v1alpha1
kind: Logstash
metadata:
  name: lab-logstash-cpm
  namespace: elastic
spec:
  count: 1
  version: 9.4.2
  elasticsearchRefs:
    - name: lab-es
      clusterName: lab-es
  config:
    xpack.management.enabled: true
    xpack.management.pipeline.id: ["*"]
    xpack.management.elasticsearch.hosts: ["${LAB_ES_ES_HOSTS}"]
    xpack.management.elasticsearch.username: "${LAB_ES_ES_USER}"
    xpack.management.elasticsearch.password: "${LAB_ES_ES_PASSWORD}"
    xpack.management.elasticsearch.ssl.certificate_authority: "${LAB_ES_ES_SSL_CERTIFICATE_AUTHORITY}"
  podTemplate:
    spec:
      containers:
        - name: logstash
          volumeMounts:
            - name: air-quality-csv
              mountPath: /usr/share/logstash/datasets
              readOnly: true
          resources:
            requests:
              memory: 1Gi
              cpu: 500m
            limits:
              memory: 1536Mi
      volumes:
        - name: air-quality-csv
          configMap:
            name: air-quality-csv
EOF

• xpack.management.enabled: true: liga o Centralized Pipeline Management;
• xpack.management.pipeline.id: ["*"]: o Logstash aceita qualquer pipeline criado pelo Kibana, sem precisar listar os IDs;
• xpack.management.elasticsearch.*: onde e com qual usuário o Logstash busca os pipelines. Os valores vêm das variáveis LAB_ES_ES_*, criadas pelo operator. A documentação do Logstash pede um usuário com os papéis logstash_admin e logstash_system; o papel que o operator cria para o Logstash (eck_logstash_user_role) inclui os privilégios manage_logstash_pipelines e monitor;
• o volume air-quality-csv monta o ConfigMap: o CSV aparece em /usr/share/logstash/datasets/air_quality.csv.

kubectl apply -f manifests/logstash-cpm.yaml
kubectl -n elastic get pods -w

Resultado esperado: o pod lab-logstash-cpm-ls-0 chega a 1/1 Running. Pressione Ctrl+C e veja o log:

kubectl -n elastic logs lab-logstash-cpm-ls-0 | tail -20

Resultado esperado: com o Centralized Pipeline Management ativo e nenhum pipeline criado no Kibana ainda, o log indica que nenhum pipeline foi encontrado — isso é esperado.

Acesse Menu → Stack Management → Logstash Pipelines e clique em Create pipeline:

• Pipeline ID: air_quality_pipeline
• Pipeline: cole o conteúdo abaixo

input {
  file {
    path           => "/usr/share/logstash/datasets/air_quality.csv"
    start_position => "beginning"
    sincedb_path   => "/dev/null"
  }
}
 
filter {
  csv {
    separator   => ","
    skip_header => "true"
    columns     => [
      "id_station", "indicator_code", "indicator_name", "indicator_dimension",
      "value", "datetime", "source"
    ]
  }
 
  mutate {
    convert      => { "value" => "float" }
    remove_field => ["log", "host", "event", "message", "@version"]
  }
 
  date {
    match  => ["datetime", "yyyy-MM-dd HH:mm:ss"]
    target => "datetime"
  }
}
 
output {
  elasticsearch {
    hosts                       => [ "${LAB_ES_ES_HOSTS}" ]
    user                        => "${LAB_ES_ES_USER}"
    password                    => "${LAB_ES_ES_PASSWORD}"
    ssl_enabled                 => true
    ssl_certificate_authorities => "${LAB_ES_ES_SSL_CERTIFICATE_AUTHORITY}"
    index                       => "logstash-air-quality"
    data_stream                 => false
    ilm_enabled                 => false
    manage_template             => false
  }
}

Clique em Create and deploy:

O Logstash detectará o novo pipeline automaticamente e iniciará a ingestão do CSV.

Explicação dos filtros:

• csv: interpreta cada linha do arquivo como um registro CSV, mapeando as colunas para os campos definidos
• mutate/convert: converte o campo value para float, garantindo que seja indexado com o tipo correto
• mutate/remove_field: remove campos internos do Logstash que não têm utilidade no índice final
• date: converte o campo datetime do formato textual para uma data. Como target => "datetime", o resultado é gravado no próprio campo datetime; o @timestamp continua sendo o momento da leitura

No Console:

GET /_cat/indices/logstash-air-quality?v

GET /logstash-air-quality/_count

A contagem deve chegar a 5760, o total de linhas do CSV sem o cabeçalho. Para explorar os dados no Discover, crie um data view com o padrão logstash-air-quality e o campo de tempo datetime.

Para ver o recurso central em ação, edite o pipeline no Kibana (por exemplo, acrescente mutate { add_field => { "estado" => "SP" } } no filtro) e salve: o Logstash recarrega o pipeline sozinho, sem reiniciar o pod.

kubectl delete -f manifests/logstash-cpm.yaml
kubectl -n elastic delete configmap air-quality-csv

A licença trial continua ativa. Se quiser apagar também o pipeline cadastrado, use Stack Management → Logstash Pipelines.

Neste laboratório, o Elastic Agent roda como DaemonSet no modo standalone e coleta métricas de sistema (CPU, memória, rede e filesystem). A configuração de coleta fica inteira no manifesto.

Atenção: COMO ESPECIALISTAS ELASTIC, A BKTECH, NÃO RECOMENDA TRABALHAR COM ELASTIC AGENT STANDALONE. Faz mais sentido subir um filebeat+metricbeat do que um Elastic Agent StandAlone, ou seguir para o próximo laboratório e instalar o Fleet Server com um Elastic Agent que é a forma utilizada pelo o mercado e a forma que mais extrai os recursos de um Elastic Agent.

cat <<'EOF' > manifests/elastic-agent-standalone.yaml
# Elastic Agent STANDALONE via CRD Agent do ECK — coleta métricas do sistema (DaemonSet).
# A configuração de coleta vive em spec.config (sem Fleet/UI). O operator cria a saída
# "default" automaticamente a partir do elasticsearchRefs.
---
apiVersion: v1
kind: ServiceAccount
metadata:
  name: elastic-agent
  namespace: elastic
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: elastic-agent
rules:
  - apiGroups: [""]
    resources: ["nodes", "namespaces", "pods", "events", "services"]
    verbs: ["get", "list", "watch"]
  - apiGroups: ["apps"]
    resources: ["replicasets", "deployments", "daemonsets", "statefulsets"]
    verbs: ["get", "list", "watch"]
  - nonResourceURLs: ["/metrics"]
    verbs: ["get"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: elastic-agent
subjects:
  - kind: ServiceAccount
    name: elastic-agent
    namespace: elastic
roleRef:
  kind: ClusterRole
  name: elastic-agent
  apiGroup: rbac.authorization.k8s.io
---
apiVersion: agent.k8s.elastic.co/v1alpha1
kind: Agent
metadata:
  name: elastic-agent
  namespace: elastic
spec:
  version: 9.4.2
  elasticsearchRefs:
    - name: lab-es          # cria a saída "default" com host/credenciais/CA automaticamente
  daemonSet:
    podTemplate:
      spec:
        serviceAccountName: elastic-agent
        automountServiceAccountToken: true
        securityContext:
          runAsUser: 0
        containers:
          - name: agent
            resources:
              requests:
                memory: 350Mi
                cpu: 100m
              limits:
                memory: 500Mi
  config:
    id: lab-agent
    agent:
      monitoring:
        enabled: true
        use_output: default
        logs: true
        metrics: true
    inputs:
      - id: system-metrics
        type: system/metrics
        use_output: default
        data_stream:
          namespace: default
        streams:
          - id: cpu
            data_stream: { dataset: system.cpu, type: metrics }
            metricsets: [cpu]
            period: 10s
          - id: memory
            data_stream: { dataset: system.memory, type: metrics }
            metricsets: [memory]
            period: 10s
          - id: network
            data_stream: { dataset: system.network, type: metrics }
            metricsets: [network]
            period: 10s
          - id: filesystem
            data_stream: { dataset: system.filesystem, type: metrics }
            metricsets: [filesystem]
            period: 1m
EOF

Os pontos principais do manifesto:

• elasticsearchRefs: lab-es: o operator cria uma saída chamada default, com endereço, credenciais e CA do lab-es;
• config.inputs: um input system/metrics com quatro streams (CPU, memória, rede e filesystem). Cada stream grava em um data stream próprio, com o nome no formato <type>-<dataset>-<namespace>: metrics-system.cpu-default, metrics-system.memory-default e assim por diante;
• agent.monitoring: o próprio Agent envia seus logs e métricas de funcionamento;
• runAsUser: 0: o Agent guarda o seu estado local em um volume hostPath no node, e para isso precisa rodar como root.

kubectl apply -f manifests/elastic-agent-standalone.yaml
kubectl -n elastic get agent elastic-agent -w

Resultado esperado: HEALTH green, AVAILABLE 1 e EXPECTED 1. Pressione Ctrl+C.

No Console, liste os data streams criados pelo Agent:

GET /_data_stream/metrics-system.*

Se algum aparecer com status YELLOW, ajuste as réplicas dos data streams do Agent:

PUT /metrics-*,logs-*/_settings
{
  "index.number_of_replicas": 0
}

Veja a última leitura de CPU:

GET /metrics-system.cpu-default/_search
{
  "size": 1,
  "sort": [{ "@timestamp": "desc" }]
}

No Discover, crie um data view com o padrão metrics-system.* para explorar as métricas.

kubectl delete -f manifests/elastic-agent-standalone.yaml

Neste laboratório, o Elastic Agent deixa de ter a configuração no manifesto: um Fleet Server passa a distribuir as políticas de coleta, definidas no Kibana. O fluxo tem três partes:

1. o Kibana recebe as configurações do Fleet (xpack.fleet.*), com as políticas pré-configuradas;
2. um recurso Agent sobe o Fleet Server, com fleetServerEnabled: true;
3. outro recurso Agent sobe o Elastic Agent de coleta, que se registra no Fleet Server com fleetServerRef.

Recursos: o Fleet Server, o Agent e as integrações pesam. Faça este laboratório com os coletores dos laboratórios anteriores removidos. O Kibana também precisa de acesso à internet para baixar os pacotes das integrações.

Chaves de criptografia do Kibana: o Fleet exige que o Kibana tenha chaves de criptografia configuradas (xpack.encryptedSavedObjects.encryptionKey e outras). No ECK, o operator gera essas chaves automaticamente ao criar o Kibana; não há nada a fazer. Para conferir:

kubectl -n elastic get secret lab-kb-kb-config -o jsonpath='{ .data.kibana\.yml }' | base64 --decode | grep -A1 encryptedSavedObjects

Resultado esperado: a linha xpack.encryptedSavedObjects: seguida da chave encryptionKey gerada pelo operator.

O manifesto abaixo é o lab-kb com a seção config do Fleet acrescentada:

cat <<'EOF' > manifests/kibana-fleet.yaml
apiVersion: kibana.k8s.elastic.co/v1
kind: Kibana
metadata:
  name: lab-kb
spec:
  version: 9.4.2
  count: 1
  elasticsearchRef:
    name: lab-es
  config:
    xpack.fleet.agents.elasticsearch.hosts: ["https://lab-es-es-http.elastic.svc:9200"]
    xpack.fleet.agents.fleet_server.hosts:  ["https://fleet-server-agent-http.elastic.svc:8220"]
    xpack.fleet.packages:
      - name: system
        version: latest
      - name: elastic_agent
        version: latest
      - name: fleet_server
        version: latest
    xpack.fleet.agentPolicies:
      - name: Fleet Server on ECK policy
        id: eck-fleet-server
        is_default_fleet_server: true
        package_policies:
          - name: fleet_server-1
            package: { name: fleet_server }
      - name: Elastic Agent on ECK policy
        id: eck-agent
        is_default: true
        package_policies:
          - name: system-1
            package: { name: system }
  podTemplate:
    spec:
      containers:
        - name: kibana
          resources:
            requests:
              memory: 1Gi
              cpu: 500m
            limits:
              memory: 2Gi
EOF

• xpack.fleet.agents.elasticsearch.hosts: para onde os agentes enviam os dados — o Service do lab-es;
• xpack.fleet.agents.fleet_server.hosts: onde os agentes encontram o Fleet Server. O ECK cria esse Service com o nome <nome do Agent>-agent-http, na porta 8220;
• xpack.fleet.packages: as integrações instaladas ao subir o Kibana;
• xpack.fleet.agentPolicies: duas políticas pré-configuradas, eck-fleet-server (para o Fleet Server) e eck-agent (com a integração System, para os agentes de coleta).

kubectl apply -n elastic -f manifests/kibana-fleet.yaml
kubectl -n elastic get pods -w

Resultado esperado: um novo pod lab-kb-kb-<sufixo> é criado e o antigo é removido. Aguarde o novo ficar 1/1 Running e pressione Ctrl+C.

O pod do Kibana foi recriado, então o túnel caiu. Abra-o novamente:

pkill -f "port-forward.*5601" || true
nohup kubectl -n elastic port-forward --address 0.0.0.0 service/lab-kb-kb-http 5601 > /tmp/pf-kibana.log 2>&1 &
sleep 2
cat /tmp/pf-kibana.log

Resultado esperado: Forwarding from 0.0.0.0:5601 -> 5601.

cat <<'EOF' > manifests/fleet.yaml
# ==============================================================================
# Elastic Agent gerenciado por FLEET no ECK.
# O Fleet Server é apenas outro recurso Agent (mode: fleet, fleetServerEnabled).
# ==============================================================================
---
apiVersion: v1
kind: ServiceAccount
metadata:
  name: fleet-server
  namespace: elastic
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: fleet-server
rules:
  - apiGroups: [""]
    resources: ["pods", "namespaces", "nodes"]
    verbs: ["get", "watch", "list"]
  - apiGroups: ["coordination.k8s.io"]
    resources: ["leases"]
    verbs: ["get", "create", "update"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: fleet-server
subjects:
  - kind: ServiceAccount
    name: fleet-server
    namespace: elastic
roleRef:
  kind: ClusterRole
  name: fleet-server
  apiGroup: rbac.authorization.k8s.io
---
apiVersion: v1
kind: ServiceAccount
metadata:
  name: elastic-agent
  namespace: elastic
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: elastic-agent
rules:
  - apiGroups: [""]
    resources: ["pods", "nodes", "namespaces"]
    verbs: ["get", "watch", "list"]
  - apiGroups: ["coordination.k8s.io"]
    resources: ["leases"]
    verbs: ["get", "create", "update"]
  - apiGroups: ["apps"]
    resources: ["replicasets"]
    verbs: ["list", "watch"]
  - apiGroups: ["batch"]
    resources: ["jobs"]
    verbs: ["list", "watch"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: elastic-agent
subjects:
  - kind: ServiceAccount
    name: elastic-agent
    namespace: elastic
roleRef:
  kind: ClusterRole
  name: elastic-agent
  apiGroup: rbac.authorization.k8s.io
---
apiVersion: agent.k8s.elastic.co/v1alpha1
kind: Agent
metadata:
  name: fleet-server
  namespace: elastic
spec:
  version: 9.4.2
  kibanaRef:
    name: lab-kb
  elasticsearchRefs:
    - name: lab-es
  mode: fleet
  fleetServerEnabled: true
  policyID: eck-fleet-server
  deployment:
    replicas: 1
    podTemplate:
      spec:
        serviceAccountName: fleet-server
        automountServiceAccountToken: true
        securityContext:
          runAsUser: 0
---
apiVersion: agent.k8s.elastic.co/v1alpha1
kind: Agent
metadata:
  name: elastic-agent-fleet
  namespace: elastic
spec:
  version: 9.4.2
  kibanaRef:
    name: lab-kb
  fleetServerRef:
    name: fleet-server
  mode: fleet
  policyID: eck-agent
  daemonSet:
    podTemplate:
      spec:
        serviceAccountName: elastic-agent
        automountServiceAccountToken: true
        hostNetwork: true
        dnsPolicy: ClusterFirstWithHostNet
        securityContext:
          runAsUser: 0
EOF

• fleet-server: recurso Agent com mode: fleet e fleetServerEnabled: true, rodando como Deployment com a política eck-fleet-server. Usa kibanaRef para se registrar no Fleet e elasticsearchRefs para gravar no lab-es;
• elastic-agent-fleet: recurso Agent com mode: fleet e fleetServerRef: fleet-server, rodando como DaemonSet com a política eck-agent. O operator cuida do registro (enrollment) do agente no Fleet Server: não há token para copiar;
• a ServiceAccount, o ClusterRole e o ClusterRoleBinding elastic-agent fazem parte do arquivo, para que o laboratório não dependa de recursos criados em outro manifesto.

kubectl apply -f manifests/fleet.yaml
kubectl -n elastic get agent -w

Resultado esperado: primeiro o fleet-server e depois o elastic-agent-fleet chegam a HEALTH green, com AVAILABLE 1 e EXPECTED 1. O Fleet Server pode levar alguns minutos. Pressione Ctrl+C.

1. No Kibana, abra Management → Fleet.
2. Na aba Agents, aparecem dois agentes com status Healthy: o do Fleet Server, com a política Fleet Server on ECK policy, e o de coleta, com a política Elastic Agent on ECK policy.
3. Na aba Agent policies, abra a Elastic Agent on ECK policy e veja a integração System (system-1). É nessa tela que se acrescentam novas integrações: a mudança chega ao agente sem alterar nenhum manifesto.

No Console, confira os data streams da integração System:

GET /_data_stream/metrics-system.*,logs-system.*

Se algum aparecer YELLOW, ajuste as réplicas como no laboratório anterior.

kubectl delete -f manifests/fleet.yaml
kubectl apply -n elastic -f manifests/kibana.yaml
kubectl -n elastic get pods -w

Resultado esperado: os pods do Fleet Server e do Agent são removidos e o Kibana é recriado sem a configuração do Fleet. Aguarde o novo pod lab-kb-kb-<sufixo> ficar 1/1 Running, pressione Ctrl+C e abra o túnel novamente, com o mesmo bloco do Passo 2.

Os agentes removidos continuam listados no Fleet como Offline, e as políticas e integrações instaladas permanecem no Kibana. Isso não interfere nos laboratórios seguintes.

Os Ingest Pipelines são processadores de dados que rodam dentro do próprio Elasticsearch, executados no momento da indexação de cada documento. Diferentemente do Logstash e dos Beats, os pipelines não buscam dados ativamente — eles recebem os documentos no momento em que são enviados ao cluster e os transformam antes de indexá-los.

Via Kibana: Menu → Stack Management → Ingest Pipelines → Create Pipeline

Via API:

PUT /_ingest/pipeline/nome_do_pipeline
{
  "description": "Descrição do pipeline",
  "processors": [
    { "set":    { "field": "campo", "value": "valor" } },
    { "remove": { "field": "campo_desnecessario", "ignore_missing": true } }
  ]
}

Os processors são executados sequencialmente, na ordem em que são definidos.

POST /_ingest/pipeline/nome_do_pipeline/_simulate
{
  "docs": [
    {
      "_source": {
        "campo": "valor de teste",
        "outro_campo": "outro valor",
        "campo_desnecessario": "este campo será removido"
      }
    }
  ]
}

Neste laboratório vamos criar um pipeline que combina três processors diferentes: set para definir um valor fixo, dissect para extrair campos de uma string e html_strip para limpar tags HTML.

PUT /_ingest/pipeline/meu_pipeline
{
  "description": "Pipeline de demonstração com set, dissect e html_strip",
  "processors": [
    {
      "set": {
        "field": "idade",
        "value": "40"
      }
    },
    {
      "dissect": {
        "field": "classificacao",
        "pattern": "%{fornecedor_status} %{fornecedor_negativado} %{fornecedor_ativo_receita}"
      }
    },
    {
      "html_strip": {
        "field": "descricao_fornecedor"
      }
    }
  ]
}

Explicação:

• set: define o campo idade com o valor fixo "40" em todos os documentos processados
• dissect: divide o campo classificacao em três novos campos separados por espaço — fornecedor_status, fornecedor_negativado e fornecedor_ativo_receita
• html_strip: remove as tags HTML do campo descricao_fornecedor, deixando apenas o texto puro

POST /_ingest/pipeline/meu_pipeline/_simulate
{
  "docs": [
    {
      "_source": {
        "codigo": 124,
        "nome_fornecedor": "GOL Linhas Aereas",
        "classificacao": "sim nao ativo",
        "descricao_fornecedor": "<br>teste de remocao html</br>"
      }
    }
  ]
}

A resposta mostra o documento como ele seria indexado: com o campo idade, os três campos extraídos pelo dissect e a descricao_fornecedor sem as tags HTML. Nada é gravado no cluster.

Crie o índice de teste:

PUT /fornecedor_pipeline
{
  "settings": {
    "number_of_shards": 1,
    "number_of_replicas": 0
  }
}

Indique o pipeline na própria requisição de indexação:

POST /fornecedor_pipeline/_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 nao ativo",
  "descricao_fornecedor": "<br>teste de remocao html</br>"
}

GET /fornecedor_pipeline/_doc/3

Em vez de especificar o pipeline em cada requisição, é possível configurá-lo como padrão do índice — todos os documentos indexados nele passarão pelo pipeline automaticamente:

PUT /fornecedor_teste
{
  "settings": {
    "number_of_shards": 1,
    "number_of_replicas": 0,
    "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 nao ativo",
  "descricao_fornecedor": "<br>teste de remocao html</br>"
}

GET /fornecedor_teste/_doc/1

O documento foi gravado com os campos criados pelo pipeline, sem o parâmetro pipeline na requisição.

Neste laboratório vamos usar um pipeline combinado com _update_by_query para enriquecer documentos que já estão indexados — adicionando um novo campo calculado a partir de campos existentes.

PUT /informacoes_pessoas
{
  "settings": {
    "number_of_shards": 1,
    "number_of_replicas": 0
  },
  "mappings": {
    "properties": {
      "nome":             { "type": "text" },
      "idade":            { "type": "integer" },
      "sexo":             { "type": "keyword" },
      "ddd":              { "type": "keyword" },
      "numero_telefone":  { "type": "keyword" },
      "uf":               { "type": "keyword" }
    }
  }
}

POST /_bulk?refresh=true
{"index": {"_index": "informacoes_pessoas", "_id": 1}}
{"nome": "Ana Silva",       "idade": 28, "sexo": "F", "ddd": "11", "numero_telefone": "912345678", "uf": "SP"}
{"index": {"_index": "informacoes_pessoas", "_id": 2}}
{"nome": "Bruno Pereira",   "idade": 35, "sexo": "M", "ddd": "21", "numero_telefone": "923456789", "uf": "RJ"}
{"index": {"_index": "informacoes_pessoas", "_id": 3}}
{"nome": "Carlos Almeida",  "idade": 42, "sexo": "M", "ddd": "31", "numero_telefone": "934567890", "uf": "MG"}
{"index": {"_index": "informacoes_pessoas", "_id": 4}}
{"nome": "Daniela Costa",   "idade": 29, "sexo": "F", "ddd": "41", "numero_telefone": "945678901", "uf": "PR"}
{"index": {"_index": "informacoes_pessoas", "_id": 5}}
{"nome": "Eduardo Lima",    "idade": 33, "sexo": "M", "ddd": "51", "numero_telefone": "956789012", "uf": "RS"}
{"index": {"_index": "informacoes_pessoas", "_id": 6}}
{"nome": "Fernanda Souza",  "idade": 26, "sexo": "F", "ddd": "61", "numero_telefone": "967890123", "uf": "DF"}
{"index": {"_index": "informacoes_pessoas", "_id": 7}}
{"nome": "Gabriel Oliveira","idade": 39, "sexo": "M", "ddd": "71", "numero_telefone": "978901234", "uf": "BA"}
{"index": {"_index": "informacoes_pessoas", "_id": 8}}
{"nome": "Helena Martins",  "idade": 31, "sexo": "F", "ddd": "81", "numero_telefone": "989012345", "uf": "PE"}
{"index": {"_index": "informacoes_pessoas", "_id": 9}}
{"nome": "Igor Santos",     "idade": 37, "sexo": "M", "ddd": "91", "numero_telefone": "990123456", "uf": "PA"}
{"index": {"_index": "informacoes_pessoas", "_id": 10}}
{"nome": "Juliana Ramos",   "idade": 25, "sexo": "F", "ddd": "65", "numero_telefone": "901234567", "uf": "MT"}

GET /informacoes_pessoas/_search

PUT /_ingest/pipeline/numero_completo_pipeline
{
  "description": "Combina DDD e número de telefone em um único campo",
  "processors": [
    {
      "script": {
        "source": "ctx.numero_completo = ctx.ddd + ctx.numero_telefone"
      }
    }
  ]
}

O processor script usa a linguagem Painless — a linguagem de scripting nativa do Elasticsearch. ctx representa o documento atual sendo processado, e ctx.campo acessa ou define campos nele.

O _update_by_query com um pipeline permite reaplicar transformações em documentos já indexados — sem precisar reindexá-los do zero:

POST /informacoes_pessoas/_update_by_query?pipeline=numero_completo_pipeline&refresh=true

GET /informacoes_pessoas/_search

Confirme que todos os documentos agora possuem o campo numero_completo com o DDD e o número concatenados.

Confira que nenhum coletor ficou rodando:

kubectl -n elastic get beat,agent,logstash

Resultado esperado: No resources found in elastic namespace.

Os volumes persistentes dos Logstash podem continuar existindo depois que os recursos foram removidos. Liste e remova os que sobraram:

kubectl -n elastic get pvc | grep logstash
kubectl -n elastic get pvc -o name | grep logstash | xargs -r kubectl -n elastic delete

Resultado esperado: a remoção de cada PVC com nome iniciado por logstash-data- ou datasets-; se não sobrou nenhum, o segundo comando não faz nada.

No Console, remova os índices e pipelines dos laboratórios de Logstash e de Ingest Pipelines:

DELETE /logstash-chinook,logstash-air-quality,fornecedor_pipeline,fornecedor_teste,informacoes_pessoas

DELETE /_ingest/pipeline/meu_pipeline

DELETE /_ingest/pipeline/numero_completo_pipeline

Os data streams do Filebeat e do Elastic Agent (filebeat-*, metrics-*, logs-*) e o índice logstash-lab-* podem ficar: eles servem de dados de exemplo para explorar no Kibana.

• O Elastic Stack oferece quatro caminhos de ingestão: Ingest Pipelines (dentro do Elasticsearch), Beats, Elastic Agent e Logstash. Em novas implementações, o Elastic Agent é o coletor recomendado.
• No ECK, Beats, Elastic Agent e Logstash são recursos do mesmo operator (Beat, Agent e Logstash). O elasticsearchRef configura sozinho endereço, credenciais e CA do cluster de destino.
• Coletores de logs e métricas rodam como DaemonSet, com volumes hostPath, runAsUser: 0 e RBAC; o Fleet Server roda como Deployment e o Logstash como StatefulSet, com volume persistente.
• Na 9.x, o Filebeat lê logs de containers com o input filestream e o parser container; o input container vem desativado.
• No Logstash do ECK, o pipeline fica no manifesto (pipelines) e usa as variáveis <CLUSTERNAME>_ES_*. Arquivos auxiliares chegam ao pod por ConfigMap ou por um volume preenchido por initContainer. O usuário criado pelo operator só escreve em índices como logstash-*, logs-* e metrics-*.
• O Centralized Pipeline Management move o pipeline do Logstash para o Kibana e exige licença paga. No ECK, a trial do cluster pode ser ativada pelo Kibana (License Management); o Secret eck-trial-license ativa também a trial do operator.
• O Elastic Agent funciona em modo standalone (configuração no manifesto) ou gerenciado pelo Fleet (políticas no Kibana). No ECK, o Fleet Server é outro recurso Agent, e as chaves de criptografia exigidas pelo Fleet são geradas pelo operator.
• Ingest Pipelines transformam documentos na indexação com processors como set, dissect, html_strip e script; podem ser testados com _simulate, definidos como default_pipeline de um índice e reaplicados a documentos existentes com _update_by_query.

7.1 e 7.2 - Fundamentos, Beats e Elastic Agent

• Ingestion tools overview
• Beats
• Elastic Agent
• Logstash
• Orchestrate other Elastic applications (ECK)
• Standalone Elastic Agent (ECK)
• Fleet-managed Elastic Agent (ECK)

7.4 e 7.5 - Filebeat e Logstash

• Quickstart: Beats on ECK
• Configuration: Beats on ECK
• Receita oficial do ECK 3.4: Filebeat sem autodiscover
• Container input (desativado na 9.0)
• Filestream input
• Add Kubernetes metadata
• Quickstart: Logstash on ECK
• Configuration: Logstash on ECK (elasticsearchRefs, variáveis de ambiente, volumes, services)

7.6 - Logstash com JDBC

• Logstash plugins on ECK (initContainer e volumes)
• Logstash configuration examples on ECK (índice customizado)
• Jdbc input plugin
• Elasticsearch output plugin
• Date filter plugin

7.7 - Logstash com CSV e Centralized Pipeline Management

• Manage your license in ECK (trial)
• Manage your license in a self-managed cluster (License Management no Kibana)
• Start a trial API
• Código do operator ECK 3.4: preservação da trial iniciada no Kibana
• Centralized Pipeline Management
• Configure Centralized Pipeline Management
• File input plugin
• Csv filter plugin
• Mutate filter plugin
• ConfigMaps

7.8 e 7.9 - Elastic Agent standalone e Fleet

• Quickstart: standalone Elastic Agent on ECK
• Configuration: standalone Elastic Agent on ECK
• Quickstart: Fleet on ECK
• Configuration: Fleet-managed Elastic Agent on ECK
• Known limitations: Elastic Agent e Fleet on ECK
• Kibana encryption keys (ECK)
• Data streams do Elastic Agent (nomenclatura)

7.10 a 7.12 - Ingest Pipelines

• Ingest pipelines
• Ingest processor reference
• Simulate a pipeline API
• Index setting default_pipeline
• Update by query API
• Painless scripting language