Migrer des données Elasticsearch auto-gérées vers Alibaba Cloud Elasticsearch avec Logstash - Elasticsearch

Cette rubrique explique comment migrer la totalité ou une partie incrémentielle des données d'un cluster Elasticsearch auto-géré vers Alibaba Cloud Elasticsearch en déployant Logstash sur une instance ECS et en configurant un pipeline de migration.

Prérequis

L'instance ECS qui héberge Logstash doit se trouver dans le même VPC que le cluster Alibaba Cloud Elasticsearch et doit avoir un accès réseau aux clusters source et de destination.
Si votre application écrit ou met à jour des données en continu, effectuez d'abord une migration complète, puis une migration incrémentielle basée sur un horodatage ou un autre champ identifiant. Sinon, les anciennes données risquent d'écraser les nouvelles données dans le cluster de destination. Si le cluster de destination contient déjà toutes les données existantes, seule une migration incrémentielle est nécessaire.

Procédure

Étape 1 : Préparer l'environnement et les instances

Créez un cluster Alibaba Cloud Elasticsearch, déployez Elasticsearch auto-géré et Logstash sur une instance ECS, et préparez les données à migrer.
Étape 2 (Facultative) : Migrer les métadonnées d'index (paramètres et mappages)

Exécutez un script Python sur l'instance ECS pour migrer les métadonnées d'index.
Étape 3 : Effectuer une migration complète des données

Utilisez Logstash pour migrer toutes les données du cluster auto-géré vers Alibaba Cloud Elasticsearch.
Étape 4 : Effectuer une migration incrémentielle des données
Étape 5 : Vérifier les résultats de la migration

Étape 1 : Préparer l'environnement et les instances

Créez une instance Alibaba Cloud Elasticsearch.

Créer une instance Alibaba Cloud Elasticsearch. L'environnement de test utilise la configuration suivante.

Paramètre

Description

Région

Chine (Hangzhou).

Edition

Edition Standard 7.10.0.

Spécifications de l'instance

Trois zones, trois nœuds de données. Chaque nœud dispose de 4 vCPU, de 16 Go de mémoire et d'un disque ESSD (Enhanced SSD) de 100 Go.

Créez une instance ECS pour les instances Elasticsearch auto-gérées, Kibana et Logstash.

Créer une instance à l'aide de l'assistant. L'environnement de test utilise la configuration suivante.

Paramètre	Description
Région	Chine (Hangzhou).
Type d'instance	4 vCPU, 16 GiB de mémoire.
Image	Image publique, CentOS 7.9 64 bits.
Stockage	Disque système, disque ESSD (Enhanced SSD) de 100 GiB.
Réseau	Sélectionnez le même cloud privé virtuel (VPC) que votre cluster Alibaba Cloud Elasticsearch. Sélectionnez Assign Public IPv4 Addresses, définissez la méthode de facturation sur paiement au trafic et définissez la bande passante maximale à 100 Mbit/s.
Groupe de sécurité	Ajoutez une règle entrante pour autoriser l'accès sur le port 5601 (port Kibana par défaut). Définissez l'objet d'autorisation sur l'adresse IP de votre client. Important Si votre client se trouve sur un réseau domestique ou d'entreprise, utilisez l'adresse IP publique de sortie du réseau, et non l'adresse IP privée de votre ordinateur. Trouvez votre adresse IP publique à l'adresse https://www.whatismyip.com. La définition de 0.0.0.0/0 comme objet d'autorisation permet à toutes les adresses IPv4 d'accéder à l'instance, mais expose votre instance ECS à Internet public. Évitez cette configuration en production.

Déployez le cluster Elasticsearch auto-géré.

Cette rubrique utilise un cluster Elasticsearch 7.6.2 auto-géré avec un seul nœud de données.

Connectez-vous à l'instance ECS.

Se connecter à une instance Linux à l'aide de Workbench.
En tant qu'utilisateur root, créez un nouvel utilisateur nommé elastic.
```
useradd elastic
```
Définissez le mot de passe de l'utilisateur elastic.
```
passwd elastic
```
Suivez les instructions pour saisir et confirmer le nouveau mot de passe.
Passez à l'utilisateur elastic.
```
su -l elastic
```

Téléchargez et extrayez le package d'installation d'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

Démarrez Elasticsearch.
Accédez au répertoire d'installation d'Elasticsearch et démarrez le service en arrière-plan.
```
cd elasticsearch-7.6.2
./bin/elasticsearch -d
```

Vérifiez que le service Elasticsearch est en cours d'exécution.

cd ~ 
curl localhost:9200

Une réponse réussie inclut le numéro de version d'Elasticsearch et la mention "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"
}

Déployez une instance Kibana auto-gérée et préparez les exemples de données.

Cette rubrique utilise une instance Kibana 7.6.2 auto-gérée.
1. Connectez-vous à l'instance ECS.
  
  Se connecter à une instance Linux à l'aide de Workbench.
  
  Remarque
  Les étapes de cette rubrique supposent que vous exécutez les commandes en tant qu'utilisateur non root, sauf indication contraire.
2. Téléchargez et extrayez le package d'installation de 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. Modifiez le fichier de configuration de Kibana config/kibana.yml et ajoutez server.host: "0.0.0.0" pour activer l'accès distant.
  Accédez au répertoire d'installation de Kibana et modifiez kibana.yml.
```
cd kibana-7.6.2-linux-x86_64
vi config/kibana.yml
```
  Définissez la valeur de server.host sur "0.0.0.0" pour autoriser les connexions distantes. La configuration clé dans le fichier mis à jour est la suivante :
```
# 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. Démarrez Kibana en tant qu'utilisateur non root.
```
sudo nohup ./bin/kibana &
```
5. Connectez-vous à la console Kibana et ajoutez des exemples de données.
  1. Accédez à la console Kibana en utilisant l'adresse IP publique de l'instance ECS.
    
    L'URL est formatée comme suit : http://<your_ecs_instance_public_ip>:5601/app/kibana#/home.
  2. Sur la page d'accueil de Kibana, cliquez sur Try our sample data.
  3. Dans l'onglet Sample data, localisez la carte Sample web logs, puis cliquez sur Add data en bas de la carte pour ajouter les exemples de données.

Déployez une instance Logstash auto-gérée.

Cette rubrique utilise une instance Logstash 7.10.0 auto-gérée avec un seul nœud.

Connectez-vous à l'instance ECS.

Se connecter à une instance Linux à l'aide de Workbench.

Remarque
Les étapes de cette rubrique supposent que vous exécutez les commandes en tant qu'utilisateur non root.

Revenez au répertoire personnel, puis téléchargez et extrayez le package d'installation de 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

Ajustez la taille du tas (heap) de Logstash.

La taille par défaut du tas est de 1 Go. Ajustez-la en fonction des spécifications de votre instance ECS pour de meilleures performances de migration.

Accédez au répertoire d'installation de Logstash et modifiez le fichier config/jvm.options pour définir la taille initiale et maximale du tas (heap) sur 8 Go (-Xms8g et -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

Modifiez la taille des lots Logstash.

L'écriture des données par lots de 5 à 15 Mo accélère la migration des données.

Modifiez le fichier config/pipelines.yml et remplacez la valeur de pipeline.batch.size de 125 à 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

Vérifiez que Logstash fonctionne correctement.

Exécutez un pipeline simple qui prend l'entrée standard et l'envoie vers la sortie standard.
```
bin/logstash -e 'input { stdin { } } output { stdout {} }'
```

Une fois le pipeline démarré, saisissez « Hello world! » et appuyez sur Entrée.

Si Logstash fonctionne, il affiche un message de journal structuré contenant "Hello world!" dans la 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
}

Étape 2 (facultative) : Migrer les métadonnées d'index

Logstash crée automatiquement un index s'il n'existe pas dans le cluster de destination, mais les paramètres et mappages créés automatiquement peuvent différer de ceux de la source. Pour garantir une structure d'index cohérente, créez manuellement l'index de destination avant la migration.

Utilisez le script Python suivant pour créer l'index de destination.

Connectez-vous à l'instance ECS.

Se connecter à une instance Linux à l'aide de Workbench.

Remarque
Les étapes de cette rubrique supposent que vous exécutez les commandes en tant qu'utilisateur non root.
Créez et ouvrez un fichier de script Python. Cette rubrique utilise indiceCreate.py comme nom de fichier.
```
sudo vi indiceCreate.py
```

Copiez le code suivant dans le fichier de script Python et remplacez les valeurs d'espace réservé pour les endpoints du cluster, les noms d'utilisateur et les mots de passe par vos informations d'identification réelles.

#!/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.")

Exécutez le script Python pour créer l'index de destination.
```
sudo /usr/bin/python indiceCreate.py
```
Se connecter à la console Kibana du cluster de destination et vérifiez que l'index a été créé.
```
GET /_cat/indices?v
```

Étape 3 : Migrer toutes les données

Connectez-vous à l'instance ECS.

Se connecter à une instance Linux à l'aide de Workbench.
Dans le répertoire config, créez et ouvrez un fichier de configuration Logstash.
```
cd logstash-7.10.0/config
vi es2es_all.conf
```

Ajoutez la configuration suivante au fichier.

Remarque

Les paramètres de configuration Logstash ont changé dans la version 8.5. Cette rubrique fournit des exemples de configuration pour les versions 7.10.0 et 8.5.1.
Pour garantir l'exactitude des données, créez des fichiers de configuration de pipeline Logstash distincts et migrez les données par lots.

Version 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
    }
}

Version 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
    }
}

Le plugin d'entrée Elasticsearch s'arrête une fois toutes les données lues. Dans certains environnements, Logstash peut redémarrer automatiquement, ce qui entraîne des écritures en double. Utilisez le paramètre schedule avec une expression cron pour exécuter la tâche à un moment précis et éviter ce problème (Planification).

Par exemple, pour exécuter la tâche le 5 mars à 13 h 20 :

schedule => "20 13 5 3 *"

Accédez au répertoire Logstash.
```
cd ~/logstash-7.10.0
```

Démarrez la tâche de migration complète des données.

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

Étape 4 : Migrer les données incrémentielles

Connectez-vous à l'instance ECS. Dans le répertoire config, créez et ouvrez un nouveau fichier de configuration Logstash pour la migration incrémentielle.
```
cd config
vi es2es_kibana_sample_data_logs.conf
```
Remarque
Les étapes décrites dans cette rubrique supposent que vous exécutez les commandes en tant qu'utilisateur non root.

Ajoutez la configuration suivante au fichier.

Voici un exemple de configuration pour la version 7.10.0.

Remarque

Pour Logstash 8.5 et versions ultérieures, vous devez supprimer la ligne document_type => "%{[@metadata][_type]}" car les types de documents sont obsolètes.
Une fois le fichier configuré, le démarrage de la tâche Logstash planifiée déclenche la migration incrémentielle.

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
    }
}

Important

Logstash utilise des horodatages UTC. Si vos données sources utilisent un fuseau horaire différent, ajustez la plage de requête en conséquence. La valeur now-5m dans le champ @timestamp est basée sur l'horloge UTC du serveur.
Votre index source doit contenir un champ temporel pour la synchronisation incrémentielle. Si ce n'est pas le cas, utilisez un pipeline d'ingestion avec le champ de métadonnées _ingest.timestamp pour ajouter @timestamp aux documents lors de l'indexation.

Accédez au répertoire Logstash.
```
cd ~/logstash-7.10.0
```

Démarrez la tâche de migration incrémentielle des données.

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

Dans la console Kibana du cluster Elasticsearch de destination, interrogez les derniers enregistrements pour vérifier que les données incrémentielles sont synchronisées.
La requête suivante recherche les enregistrements de l'index kibana_sample_data_logs datant des cinq dernières minutes.
```
GET kibana_sample_data_logs/_search
{
  "query": {
    "range": {
      "@timestamp": {
        "gte": "now-5m",
        "lte": "now/m"
      }
    }
  },
  "sort": [
    {
      "@timestamp": {
        "order": "desc"
      }
    }
  ]
}
                        
```

Étape 5 : Vérifier les résultats de la migration

Vérifiez la migration complète des données.

Vérifiez les informations relatives aux index et au nombre de documents dans le cluster source autogéré.

GET _cat/indices?v

Le résultat suivant est donné à titre d'exemple.

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

Vérifiez les index et le nombre de documents dans le cluster de destination Alibaba Cloud avant la migration.

Voici un exemple des informations d'index dans le cluster Elasticsearch de destination Alibaba Cloud avant la migration.

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

Après la migration complète des données, vérifiez à nouveau les informations relatives aux index et au nombre de documents dans le cluster de destination Alibaba Cloud.

Le nombre de documents doit correspondre au décompte du cluster source. Dans Kibana Dev Tools, exécutez la commande GET _cat/indices?v. Le résultat indique que tous les index du cluster ont un état health défini sur green et un statut status défini sur open. L'index kibana_sample_data_logs affiche un docs.count de 14 074 et une store.size de 9,4 Mo, ce qui confirme que les données ont été migrées avec succès vers le cluster de destination.

Vérifiez la migration incrémentielle des données.

Vérifiez les enregistrements les plus récents dans le cluster source autogéré.

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

Le résultat suivant est donné à titre d'exemple.

{
  "_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
  ]
}

Exécutez la même requête dans la console Kibana du cluster de destination. Des résultats identiques confirment la réussite de la synchronisation incrémentielle.