Migrar dados do Elasticsearch autogerenciado para o Alibaba Cloud Elasticsearch com Logstash - Elasticsearch: Central de Documentação do Alibaba Cloud

Este tópico descreve como migrar dados completos ou incrementais de um cluster Elasticsearch autogerenciado para o Alibaba Cloud Elasticsearch. O processo envolve a implantação do Logstash em uma instância ECS e a configuração de um pipeline de migração.

Considerações

A instância ECS que hospeda o Logstash deve estar na mesma VPC do cluster Alibaba Cloud Elasticsearch e ter acesso de rede tanto ao cluster de origem quanto ao de destino.
Se sua aplicação gravar ou atualizar dados continuamente, execute primeiro uma migração completa e, em seguida, uma migração incremental baseada em timestamp ou outro campo identificador. Caso contrário, dados antigos podem sobrescrever os novos no cluster de destino. Se o destino já possuir todos os dados existentes, apenas a migração incremental será necessária.

Procedimento

Etapa 1: Preparar o ambiente e as instâncias

Crie um cluster Alibaba Cloud Elasticsearch, implante o Elasticsearch e o Logstash autogerenciados em uma instância ECS e prepare os dados para migração.
Etapa 2 (Opcional): Migrar metadados dos índices (configurações e mapeamentos)

Execute um script Python na instância ECS para migrar os metadados dos índices.
Etapa 3: Executar uma migração de dados completa

Use o Logstash para migrar todos os dados do cluster autogerenciado para o Alibaba Cloud Elasticsearch.
Etapa 4: Executar uma migração de dados incremental
Etapa 5: Verificar os resultados da migração

Etapa 1: Preparar o ambiente e as instâncias

Crie uma instância Alibaba Cloud Elasticsearch.

Criar uma instância do Alibaba Cloud Elasticsearch. O ambiente de teste usa a seguinte configuração:

Parâmetro	Descrição
Região	China (Hangzhou).
Edição	Standard Edition 7.10.0.
Especificações da instância	Três zonas, três nós de dados. Cada nó possui 4 vCPUs, 16 GB de memória e um SSD aprimorado (ESSD) de 100 GB.

Crie uma instância ECS para as instâncias autogerenciadas de Elasticsearch, Kibana e Logstash.

Criar uma instância usando o assistente. O ambiente de teste usa a seguinte configuração:

Parâmetro	Descrição
Região	China (Hangzhou).
Tipo de instância	4 vCPUs, 16 GiB de memória.
Imagem	Imagem pública, CentOS 7.9 64 bits.
Armazenamento	Disco do sistema, SSD aprimorado (ESSD) de 100 GiB.
Rede	Selecione a mesma Virtual Private Cloud (VPC) do seu cluster Alibaba Cloud Elasticsearch. Selecione Assign Public IPv4 Addresses, defina o método de faturamento como pay-by-traffic e configure a largura de banda máxima para 100 Mbit/s.
Grupo de segurança	Adicione uma regra de entrada para permitir acesso na porta 5601 (porta padrão do Kibana). Defina o objeto de autorização como o endereço IP do seu cliente. Importante Se o seu cliente estiver em uma rede doméstica ou corporativa, use o IP público de saída da rede, e não o IP privado do seu computador. Consulte seu IP público em https://www.whatismyip.com. Definir 0.0.0.0/0 como objeto de autorização permite todos os endereços IPv4, mas expõe sua instância ECS à internet pública. Evite essa configuração em ambientes de produção.

Implante o cluster Elasticsearch autogerenciado.

Este tópico usa um cluster Elasticsearch 7.6.2 autogerenciado com um nó de dados.

Conecte-se à instância ECS.

Conectar-se a uma instância Linux usando o Workbench.
Como usuário root, crie um novo usuário chamado elastic.
```
useradd elastic
```
Defina a senha para o usuário elastic.
```
passwd elastic
```
Siga as instruções para inserir e confirmar a nova senha.
Mude para o usuário elastic.
```
su -l elastic
```

Baixe e extraia o pacote de instalação do Elasticsearch.

wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-7.6.2-linux-x86_64.tar.gz
tar -zvxf elasticsearch-7.6.2-linux-x86_64.tar.gz

Inicie o Elasticsearch.
Acesse o diretório de instalação do Elasticsearch e inicie o service em segundo plano.
```
cd elasticsearch-7.6.2
./bin/elasticsearch -d
```

Verifique se o service Elasticsearch está em execução.

cd ~ 
curl localhost:9200

Uma resposta bem-sucedida inclui o número da versão do Elasticsearch e a tagline "You Know, for Search".

[elastic@vm01 ~]$ curl localhost:9200
{
  "name" : "vm01",
  "cluster_name" : "elasticsearch",
  "cluster_uuid" : "SRB4pnk4SmS-YHzsrxxx",
  "version" : {
    "number" : "7.6.2",
    "build_flavor" : "default",
    "build_type" : "tar",
    "build_hash" : "ef48eb35cf30adf4db14086e8aabd07ef6xxx",
    "build_date" : "2020-03-26T06:34:37.794943Z",
    "build_snapshot" : false,
    "lucene_version" : "8.4.0",
    "minimum_wire_compatibility_version" : "6.8.0",
    "minimum_index_compatibility_version" : "6.0.0-beta1"
  },
  "tagline" : "You Know, for Search"
}

Implante uma instância Kibana autogerenciada e prepare dados de exemplo.

Este tópico usa uma instância Kibana 7.6.2 autogerenciada.
1. Conecte-se à instância ECS.
  
  Conectar-se a uma instância Linux usando o Workbench.
  
  Nota
  As etapas deste tópico pressupõem que você esteja executando comandos como um usuário não root, salvo indicação em contrário.
2. Baixe e extraia o pacote de instalação do Kibana.
```
wget https://artifacts.elastic.co/downloads/kibana/kibana-7.6.2-linux-x86_64.tar.gz
tar -zvxf kibana-7.6.2-linux-x86_64.tar.gz
```
3. Edite o arquivo de configuração do Kibana config/kibana.yml e adicione server.host: "0.0.0.0" para habilitar o acesso remoto.
  Acesse o diretório de instalação do Kibana e edite o arquivo kibana.yml.
```
cd kibana-7.6.2-linux-x86_64
vi config/kibana.yml
```
  Defina o valor de server.host como "0.0.0.0" para permitir conexões remotas. A configuração principal no arquivo atualizado fica da seguinte forma:
```
# Kibana is served by a back end server. This setting specifies the port to use.
#server.port: 5601

# Specifies the address to which the Kibana server will bind. IP addresses and host names are both valid values.
# The default is 'localhost', which usually means remote machines will not be able to connect.
# To allow connections from remote users, set this parameter to a non-loopback address.
#server.host: "localhost"
server.host: "0.0.0.0"
# Enables you to specify a path to mount Kibana at if you are running behind a proxy.
# Use the `server.rewriteBasePath` setting to tell Kibana if it should remove the basePath
# from requests it receives, and to prevent a deprecation warning at startup.
# This setting cannot end in a slash.
#server.basePath: ""
```
4. Inicie o Kibana como um usuário não root.
```
sudo nohup ./bin/kibana &
```
5. Faça login no console do Kibana e adicione dados de exemplo.
  1. Acesse o console do Kibana usando o endereço IP público da instância ECS.
    
    A URL segue o formato: http://<your_ecs_instance_public_ip>:5601/app/kibana#/home.
  2. Na página home do Kibana, clique em Try our sample data.
  3. Na aba Sample data, localize o cartão Sample web logs e clique em Add data na parte inferior do cartão para adicionar os dados de exemplo.

Implante uma instância Logstash autogerenciada.

Este tópico usa uma instância Logstash 7.10.0 autogerenciada com um nó.

Conecte-se à instância ECS.

Conectar-se a uma instância Linux usando o Workbench.

Nota
As etapas deste tópico pressupõem que você esteja executando comandos como um usuário não root.

Retorne ao diretório home, baixe e extraia o pacote de instalação do Logstash.

cd ~
wget https://artifacts.elastic.co/downloads/logstash/logstash-7.10.0-linux-x86_64.tar.gz
tar -zvxf logstash-7.10.0-linux-x86_64.tar.gz

Ajuste o tamanho do heap do Logstash.

O tamanho padrão do heap é 1 GB. Ajuste-o conforme as especificações da sua instância ECS para obter melhor desempenho na migração.

Acesse o diretório de instalação do Logstash e edite o arquivo config/jvm.options para definir os tamanhos inicial e máximo do heap como 8 GB (-Xms8g e -Xmx8g).

cd logstash-7.10.0
sudo vi config/jvm.options

## JVM configuration

# Xms represents the initial size of total heap space
# Xmx represents the maximum size of total heap space

-Xms8g
-Xmx8g

################################################################
## Expert settings
################################################################
##
## All settings below this section are considered
## expert settings. Don't tamper with them unless
## you understand what you are doing
##
################################################################

## GC configuration
-XX:+UseConcMarkSweepGC
-XX:CMSInitiatingOccupancyFraction=75
-XX:+UseCMSInitiatingOccupancyOnly

## Locale

Modifique o tamanho do lote do Logstash.

Gravar dados em lotes de 5 MB a 15 MB acelera a migração.

Edite o arquivo config/pipelines.yml e altere o parâmetro pipeline.batch.size de 125 para 5000.

vi config/pipelines.yml

#    # The path from where to read the configuration text
#    path.config: "/etc/conf.d/logstash/myconfig.cfg"
#
#    # How many worker threads execute the Filters+Outputs stage of the pipeline
#    pipeline.workers: 1 (actually defaults to number of CPUs)
#
#    # How many events to retrieve from inputs before sending to filters+workers
     pipeline.batch.size: 5000
#
#    # How long to wait in milliseconds while polling for the next event
#    # before dispatching an undersized batch to filters+outputs
#    pipeline.batch.delay: 50
#
#    # Internal queuing model, "memory" for legacy in-memory based queuing and
#    # "persisted" for disk-based acked queuing. Defaults is memory
#    queue.type: memory

Verifique se o Logstash está funcionando corretamente.

Execute um pipeline simples que receba a entrada padrão e a envie para a saída padrão.
```
bin/logstash -e 'input { stdin { } } output { stdout {} }'
```

Após o início do pipeline, digite "Hello world!" e pressione Enter.

Se o Logstash estiver operando corretamente, ele imprimirá uma mensagem de log estruturada contendo "Hello world!" no console.

[elastic@vm01 logstash-7.10.0]$ bin/logstash -e 'input { stdin { } } output { stdout {} }'
Using bundled JDK: /home/elastic/logstash-7.10.0/jdk
OpenJDK 64-Bit Server VM warning: Option UseConcMarkSweepGC was deprecated in version 9.0 a
WARNING: An illegal reflective access operation has occurred
WARNING: Illegal reflective access by org.jruby.ext.openssl.SecurityHelper (file:/tmp/jruby
WARNING: Please consider reporting this to the maintainers of org.jruby.ext.openssl.Securit
WARNING: Use --illegal-access-warn to enable warnings of further illegal reflective access
WARNING: All illegal access operations will be denied in a future release
Sending Logstash logs to /home/elastic/logstash-7.10.0/logs which is now configured via log
[2022-03-21T15:39:24,470][INFO ][logstash.runner          ] Starting Logstash {"logstash.ve
inux-x86_64]"}
[2022-03-21T15:39:24,606][INFO ][logstash.setting.writabledirectory] Creating directory {:s
[2022-03-21T15:39:24,618][INFO ][logstash.setting.writabledirectory] Creating directory {:s
[2022-03-21T15:39:24,845][WARN ][logstash.config.source.multilocal] Ignoring the 'pipelines
[2022-03-21T15:39:24,865][INFO ][logstash.agent           ] No persistent UUID file found.
[2022-03-21T15:39:25,961][INFO ][org.reflections.Reflections] Reflections took 36 ms to sca
[2022-03-21T15:39:26,356][INFO ][logstash.javapipeline    ][main] Starting pipeline {:pipel
s"=>["config string"], :thread=>"#<Thread:0x75693a9 run>"}
[2022-03-21T15:39:26,997][INFO ][logstash.javapipeline    ][main] Pipeline Java execution i
[2022-03-21T15:39:27,032][INFO ][logstash.javapipeline    ][main] Pipeline started {"pipeli
The stdin plugin is now waiting for input:
[2022-03-21T15:39:27,073][INFO ][logstash.agent           ] Pipelines running {:count=>1, :
[2022-03-21T15:39:27,211][INFO ][logstash.agent           ] Successfully started Logstash A
Hello world!
{
       "host" => "vm01",
    "@version" => "1",
    "message" => "\"Hello world!\"",
  "@timestamp" => 2022-03-21T07:39:46.598Z
}

Etapa 2 (Opcional): Migrar metadados dos índices

O Logstash cria automaticamente um índice caso ele não exista no cluster de destino, mas as configurações e mapeamentos gerados automaticamente podem diferir da source. Para garantir estruturas de índice consistentes, crie manualmente o índice de destino antes da migração.

Use o seguinte script Python para criar o índice de destino.

Conecte-se à instância ECS.

Conectar-se a uma instância Linux usando o Workbench.

Nota
As etapas deste tópico pressupõem que você esteja executando comandos como um usuário não root.
Crie e abra um arquivo de script Python. Este tópico usa indiceCreate.py como nome do arquivo.
```
sudo vi indiceCreate.py
```

Copie o código abaixo para o arquivo de script Python e substitua os valores de espaço reservado para endpoints de cluster, nomes de usuário e senhas pelas suas credenciais reais.

#!/usr/bin/python
# -*- coding: UTF-8 -*-
# Filename: indiceCreate.py
import sys
import base64
import time
import httplib
import json
## Host of the source cluster.
oldClusterHost = "localhost:9200"
## Username for the source cluster. Can be left empty.
oldClusterUserName = "elastic"
## Password for the source cluster. Can be left empty.
oldClusterPassword = "xxxxxx"
## Host of the destination cluster. You can find this on the Basic Information page of your Alibaba Cloud Elasticsearch instance.
newClusterHost = "es-cn-zvp2m4bko0009****.elasticsearch.aliyuncs.com:9200"
## Username for the destination cluster.
newClusterUser = "elastic"
## Password for the destination cluster.
newClusterPassword = "xxxxxx"
DEFAULT_REPLICAS = 0
def httpRequest(method, host, endpoint, params="", username="", password=""):
    conn = httplib.HTTPConnection(host)
    headers = {}
    if (username != "") :
        'Hello {name}, your age is {age} !'.format(name = 'Tom', age = '20')
        base64string = base64.encodestring('{username}:{password}'.format(username = username, password = password)).replace('\n', '')
        headers["Authorization"] = "Basic %s" % base64string;
    if "GET" == method:
        headers["Content-Type"] = "application/x-www-form-urlencoded"
        conn.request(method=method, url=endpoint, headers=headers)
    else :
        headers["Content-Type"] = "application/json"
        conn.request(method=method, url=endpoint, body=params, headers=headers)
    response = conn.getresponse()
    res = response.read()
    return res
def httpGet(host, endpoint, username="", password=""):
    return httpRequest("GET", host, endpoint, "", username, password)
def httpPost(host, endpoint, params, username="", password=""):
    return httpRequest("POST", host, endpoint, params, username, password)
def httpPut(host, endpoint, params, username="", password=""):
    return httpRequest("PUT", host, endpoint, params, username, password)
def getIndices(host, username="", password=""):
    endpoint = "/_cat/indices"
    indicesResult = httpGet(oldClusterHost, endpoint, oldClusterUserName, oldClusterPassword)
    indicesList = indicesResult.split("\n")
    indexList = []
    for indices in indicesList:
        if (indices.find("open") > 0):
            indexList.append(indices.split()[2])
    return indexList
def getSettings(index, host, username="", password=""):
    endpoint = "/" + index + "/_settings"
    indexSettings = httpGet(host, endpoint, username, password)
    print (index + "  Original settings:\n" + indexSettings)
    settingsDict = json.loads(indexSettings)
    ## The number of shards defaults to matching the source index.
    number_of_shards = settingsDict[index]["settings"]["index"]["number_of_shards"]
    ## The default number of replicas is 0.
    number_of_replicas = DEFAULT_REPLICAS
    newSetting = "\"settings\": {\"number_of_shards\": %s, \"number_of_replicas\": %s}" % (number_of_shards, number_of_replicas)
    return newSetting
def getMapping(index, host, username="", password=""):
    endpoint = "/" + index + "/_mapping"
    indexMapping = httpGet(host, endpoint, username, password)
    print (index + " Original mapping:\n" + indexMapping)
    mappingDict = json.loads(indexMapping)
    mappings = json.dumps(mappingDict[index]["mappings"])
    newMapping = "\"mappings\" : " + mappings
    return newMapping
def createIndexStatement(oldIndexName):
    settingStr = getSettings(oldIndexName, oldClusterHost, oldClusterUserName, oldClusterPassword)
    mappingStr = getMapping(oldIndexName, oldClusterHost, oldClusterUserName, oldClusterPassword)
    createstatement = "{\n" + str(settingStr) + ",\n" + str(mappingStr) + "\n}"
    return createstatement
def createIndex(oldIndexName, newIndexName=""):
    if (newIndexName == "") :
        newIndexName = oldIndexName
    createstatement = createIndexStatement(oldIndexName)
    print ("Settings and mapping for the new index " + newIndexName + ":\n" + createstatement)
    endpoint = "/" + newIndexName
    createResult = httpPut(newClusterHost, endpoint, createstatement, newClusterUser, newClusterPassword)
    print ("Result of creating new index " + newIndexName + ": " + createResult)
## main
indexList = getIndices(oldClusterHost, oldClusterUserName, oldClusterPassword)
systemIndex = []
for index in indexList:
    if (index.startswith(".")):
        systemIndex.append(index)
    else :
        createIndex(index, index)
if (len(systemIndex) > 0) :
    for index in systemIndex:
        print (index + " might be a system index and will not be recreated. If required, handle it separately.")

Execute o script Python para criar o índice de destino.
```
sudo /usr/bin/python indiceCreate.py
```
Fazer login no console do Kibana do cluster de destino e verifique se o índice foi criado.
```
GET /_cat/indices?v
```

Etapa 3: Migrar dados completos

Conecte-se à instância ECS.

Conectar-se a uma instância Linux usando o Workbench.
No diretório config, crie e abra um arquivo de configuração do Logstash.
```
cd logstash-7.10.0/config
vi es2es_all.conf
```

Adicione a seguinte configuração ao arquivo.

Nota

Os parâmetros de configuração do Logstash foram alterados na versão 8.5. Este tópico fornece exemplos de configuração tanto para a versão 7.10.0 quanto para a 8.5.1.
Para garantir a precisão dos dados, crie arquivos de configuração de pipeline do Logstash separados e migre os dados em lotes.

Versão 7.10.0

input{
    elasticsearch{
        # Endpoints of the source Elasticsearch cluster.
        hosts =>  ["http://localhost:9200"]
        # Username and password for the source cluster.
        user => "xxxxxx"
        password => "xxxxxx"
        # List of indices to migrate. Separate multiple indices with commas (,).
        index => "kibana_sample_data_*"
        # The following three items can be left as default. They relate to the thread count, migration data size, and Logstash JVM configuration.
        docinfo=>true
        slices => 5
        size => 5000
    }
}

filter {
  # Remove metadata fields added by Logstash.
  mutate {
    remove_field => ["@timestamp", "@version"]
  }
}

output{
    elasticsearch{
        # Endpoints of the destination cluster. You can find this on the Basic Information page of your Alibaba Cloud Elasticsearch instance.
        hosts => ["http://es-cn-zvp2m4bko0009****.elasticsearch.aliyuncs.com:9200"]
        # Username and password for the destination cluster.
        user => "elastic"
        password => "xxxxxx"
        # Name of the destination index. This configuration keeps the index name the same as the source.
        index => "%{[@metadata][_index]}"
        # Type of the destination index. This configuration keeps the index type the same as the source.
        document_type => "%{[@metadata][_type]}"
        # The ID for the data in the destination cluster. To improve performance, you can remove this line if you do not need to preserve the original document IDs.
        document_id => "%{[@metadata][_id]}"
        ilm_enabled => false
        manage_template => false
    }
}

Versão 8.5.1

input{
    elasticsearch{
        # Endpoints of the source Elasticsearch cluster.
        hosts =>  ["http://es-cn-uqm3811160002***.elasticsearch.aliyuncs.com:9200"]
        # Username and password for the source cluster.
        user => "elastic"
        password => ""
        # List of indices to migrate. Separate multiple indices with commas (,).
        index => "test_ecommerce"
        # The following items can be left as default. They relate to the thread count, migration data size, and Logstash JVM configuration.
        docinfo => true
        size => 10000
        docinfo_target => "[@metadata]"
    }
}

filter {
  # Remove metadata fields added by Logstash.
  mutate {
    remove_field => ["@timestamp","@version"]
  }
}

output{
    elasticsearch{
        # Endpoints of the destination cluster. You can find this on the Basic Information page of your Alibaba Cloud Elasticsearch instance.
        hosts => ["http://es-cn-nwy38aixp0001****.elasticsearch.aliyuncs.com:9200"]
        # Username and password for the destination cluster.
        user => "elastic"
        password => ""
        # Name of the destination index. This configuration keeps the index name the same as the source.
        index => "%{[@metadata][_index]}"
        # The ID for the data in the destination cluster. To improve performance, you can remove this line if you do not need to preserve the original document IDs.
        document_id => "%{[@metadata][_id]}"
        ilm_enabled => false
        manage_template => false
    }
}

O plugin de entrada do Elasticsearch interrompe a execução após ler todos os dados. Em alguns ambientes, o Logstash pode reiniciar automaticamente, causando gravações duplicadas. Use o parâmetro schedule com uma expressão cron para executar a tarefa em um horário específico e evitar esse comportamento (Agendamento).

Por exemplo, para executar a tarefa às 13h20 do dia 5 de março:

schedule => "20 13 5 3 *"

Acesse o diretório do Logstash.
```
cd ~/logstash-7.10.0
```

Inicie a tarefa de migração de dados completos.

nohup bin/logstash -f config/es2es_all.conf >/dev/null 2>&1 &

Etapa 4: Migrar dados incrementais

Conecte-se à instância ECS. No diretório config, crie e abra um novo arquivo de configuração do Logstash para a migração incremental.
```
cd config
vi es2es_kibana_sample_data_logs.conf
```
Nota
As etapas deste tópico pressupõem que você esteja executando comandos como um usuário não root.

Adicione a seguinte configuração ao arquivo.

Veja a seguir um exemplo de configuração para a versão 7.10.0.

Nota

Para o Logstash 8.5 e versões posteriores, remova a linha document_type => "%{[@metadata][_type]}", pois os tipos de documento foram descontinuados.
Após configurar o arquivo, iniciar a tarefa agendada do Logstash acionará a migração incremental.

input{
    elasticsearch{
        # Endpoints of the source Elasticsearch cluster.
        hosts =>  ["http://localhost:9200"]
        # Username and password for the source cluster.
        user => "xxxxxx"
        password => "xxxxxx"
        # List of indices to migrate. Separate multiple indices with commas (,).
        index => "kibana_sample_data_logs"
        # Query incremental data within a time range. The following configuration queries data from the last five minutes.
        query => '{"query":{"range":{"@timestamp":{"gte":"now-5m","lte":"now/m"}}}}'
        # Scheduled task. The following configuration runs the task every minute.
        schedule => "* * * * *"
        scroll => "5m"
        docinfo=>true
        size => 5000
    }
}

filter {
  # Remove metadata fields added by Logstash.
  mutate {
    remove_field => ["@timestamp", "@version"]
  }
}

output{
    elasticsearch{
        # Endpoints of the destination cluster. You can find this on the Basic Information page of your Alibaba Cloud Elasticsearch instance.
        hosts => ["http://es-cn-zvp2m4bko0009****.elasticsearch.aliyuncs.com:9200"]
        # Username and password for the destination cluster.
        user => "elastic"
        password => "xxxxxx"
        # Name of the destination index. This configuration keeps the index name the same as the source.
        index => "%{[@metadata][_index]}"
        # Type of the destination index. This configuration keeps the index type the same as the source.
        document_type => "%{[@metadata][_type]}"
        # The ID for the data in the destination cluster. To improve performance, you can remove this line if you do not need to preserve the original document IDs.
        document_id => "%{[@metadata][_id]}"
        ilm_enabled => false
        manage_template => false
    }
}

Importante

O Logstash usa timestamps UTC. Se seus dados de source usarem um fuso horário diferente, ajuste o intervalo de consulta adequadamente. O valor now-5m no campo @timestamp baseia-se no relógio UTC do servidor.
Seu índice de source deve conter um campo de tempo para a sincronização incremental. Caso contrário, use um pipeline de ingestão com o campo de metadados _ingest.timestamp para adicionar @timestamp aos documentos durante a indexação.

Acesse o diretório do Logstash.
```
cd ~/logstash-7.10.0
```

Inicie a tarefa de migração de dados incrementais.

sudo nohup bin/logstash -f config/es2es_kibana_sample_data_logs.conf >/dev/null 2>&1 &

No console do Kibana do cluster Elasticsearch de destino, consulte os registros mais recentes para verificar se os dados incrementais estão sendo sincronizados.

A consulta a seguir localiza registros no índice kibana_sample_data_logs dos últimos cinco minutos.

GET kibana_sample_data_logs/_search
{
  "query": {
    "range": {
      "@timestamp": {
        "gte": "now-5m",
        "lte": "now/m"
      }
    }
  },
  "sort": [
    {
      "@timestamp": {
        "order": "desc"
      }
    }
  ]
}

Etapa 5: Verificar os resultados da migração

Verifique a migração de dados completos.

Verifique as informações de índice e contagem de documentos no cluster de source autogerenciado.

GET _cat/indices?v

O resultado a seguir é um exemplo.

GET _cat/indices?v

health status index                    uuid                   pri rep docs.count docs.deleted store.size pri.store.size
green  open   .kibana_task_manager_1   CxAx5J2sT0qHPsWV       1   0   2          0            6.6kb      6.6kb
green  open   .apm-agent-configuration dYz5bh4dTomjtDP3       1   0   0          0            283b       283b
green  open   kibana_sample_data_logs  PUBQrSkJRMGyI-cV       1   0   14074      0            11.6mb     11.6mb
green  open   .kibana_1                MXhG2XbYTYSORB8G       1   0   49         4            139.5kb    139.5kb

Verifique o índice e a contagem de documentos no cluster de destino Alibaba Cloud antes da migração.

Veja a seguir um exemplo das informações de índice no cluster de destino Alibaba Cloud Elasticsearch antes da migração.

GET _cat/indices?v

health status index                          uuid                 pri rep docs.count docs.deleted store.size pri.store.size
green  open   .aliyun-limiter-group          5K4N8YNUSxeJZCXPxxx   1   1          0            0       522b           261b
green  open   .apm-agent-configuration       vaVC28KVQMCsABwuxxx   1   1          0            0       522b           261b
green  open   .monitoring-es-7-2022.03.19    9NUdZCaAQw-426Zrxxx   1   1     207485        15328    229.8kb          9.9kb
green  open   highlight_unified              PubNS7HIRR2B5FIfxxx   1   1          2            0     19.8kb          9.9kb
green  open   .monitoring-es-7-2022.03.18    kEP-0LeeSh01-kg2xxx   1   1     117792            0    132.3mb         60.3mb
green  open   .aliyun-limiter-config         6SJImN0bRoap3fYMxxx   1   1          0            0       522b           261b
green  open   .kibana_1                      0RRrLWLCT4aaT-1fxxx   1   1         27            4     20.8mb         10.4mb
green  open   .security-7                    D7Ux5eq7S5WtYH_Yxxx   1   1         55            0    397.7kb        198.4kb
green  open   .monitoring-es-7-2022.03.21    n6DZS66KRmW1zaN7xxx   1   1      85969         5244    102.5mb         51.5mb
green  open   .apm-custom-link               SBnBUOojSd-Vt3xxxxx   1   1          0            0       522b           261b
green  open   .monitoring-kibana-7-2022.03.20 eHPFB1h4Q8yxYbxAxxx  1   1      17278            0      5.8mb          2.8mb
green  open   .kibana_task_manager_1         iDK1EK-iR22Gkhfxxxx   1   1          6           68      157kb         66.1kb
green  open   .monitoring-kibana-7-2022.03.21 YIivw66dSBi0_Rwuxxx  1   1       6062            0      4.5mb          2.2mb
green  open   kibana_sample_data_logs        1zaN5Ji7RWqbFwKZxxx   1   0          0            0       208b           208b
green  open   .kibana-event-log-7.16.0-000001 ImZU-V4KRq2K3EUxxxx 1   1          1            0     11.4kb          5.7kb
green  open   highlight_fvh                  sErtUXXpToiiPSaSxxx   1   1          2            0     23.5kb         11.7kb
green  open   .monitoring-es-7-2022.03.20    SyOns3d-QU6ysbFDxxx   1   1     224781        43812    246.6mb        124.1mb
green  open   .monitoring-kibana-7-2022.03.18 gOvcKvRlQ9O-PipQxxx  1   1      10700            0      3.4mb          1.7mb
green  open   .monitoring-kibana-7-2022.03.19 IwSi_UIYQ5eFSyUJxxx  1   1      17280            0      5.8mb          2.9mb

Após a migração de dados completos, verifique novamente as informações de índice e contagem de documentos no cluster de destino Alibaba Cloud.

A quantidade de documentos deve corresponder à contagem do cluster de source. Nas Dev Tools do Kibana, execute o comando GET _cat/indices?v. O resultado mostra que todos os índices do cluster possuem health igual a green e status igual a open. O índice kibana_sample_data_logs apresenta docs.count de 14074 e store.size de 9,4 MB, confirmando que os dados foram migrados com sucesso para o cluster de destino.

Verifique a migração de dados incrementais.

Consulte os registros mais recentes no cluster de source autogerenciado.

GET kibana_sample_data_logs/_search
{
  "query": {
    "range": {
      "@timestamp": {
        "gte": "now-5m",
        "lte": "now/m"
      }
    }
  },
  "sort": [
    {
      "@timestamp": {
        "order": "desc"
      }
    }
  ]
}

O resultado a seguir é um exemplo.

{
  "_source" : {
    "agent" : "Mozilla/5.0 (X11; Linux x86_64; rv:6.0a1) Gecko/20110421 Firefox/6.0a1",
    "bytes" : 658,
    "clientip" : "171.66.xxx",
    "extension" : "",
    "geo" : {
      "srcdest" : "CN:US",
      "src" : "CN",
      "dest" : "US",
      "coordinates" : {
        "lat" : 45.54039389,
        "lon" : -122.9498258
      }
    },
    "host" : "www.elastic.co",
    "index" : "kibana_sample_data_logs",
    "ip" : "171.66.xxx",
    "machine" : {
      "ram" : 3221225xxx,
      "os" : "win 7"
    },
    "memory" : null,
    "message" : "171.66.xxx - - [2018-07-30T09:23:11.012Z] \"GET /security-analytics Gecko/20110421 Firefox/6.0a1\"",
    "phpmemory" : null,
    "referer" : "http://www.elastic-elastic-elastic.com/success/albert-sacco",
    "request" : "/security-analytics",
    "response" : 200,
    "tags" : [
      "success",
      "security"
    ],
    "timestamp" : "2022-03-21T09:23:11.012Z",
    "url" : "https://www.elastic.co/solutions/security-analytics",
    "utc_time" : "2022-03-21T09:23:11.012Z",
    "event" : {
      "dataset" : "sample_web_logs"
    }
  },
  "sort" : [
    1647854591012
  ]
}

Execute a mesma consulta no console do Kibana do cluster de destino. Resultados correspondentes confirmam o êxito da sincronização incremental.