Migrate self-managed Elasticsearch data to Alibaba Cloud Elasticsearch with Logstash - Elasticsearch

Considerations

The ECS instance that hosts Logstash must be in the same VPC as the Alibaba Cloud Elasticsearch cluster and must have network access to both the source and destination clusters.
If your application continuously writes or updates data, perform a full migration first, then an incremental migration based on a timestamp or another identifying field. Otherwise, old data may overwrite new data in the destination cluster. If the destination already has all existing data, only an incremental migration is needed.

Procedure

Step 1: Prepare the environment and instances

Create an Alibaba Cloud Elasticsearch cluster, deploy self-managed Elasticsearch and Logstash on an ECS instance, and prepare migration data.
Step 2 (Optional): Migrate index metadata (settings and mappings)

Run a Python script on the ECS instance to migrate index metadata.
Step 3: Perform a full data migration

Use Logstash to migrate all data from the self-managed cluster to Alibaba Cloud Elasticsearch.
Step 4: Perform an incremental data migration
Step 5: Verify the migration results

Step 1: Prepare the environment and instances

Create an Alibaba Cloud Elasticsearch instance.

Create an Alibaba Cloud Elasticsearch instance. The test environment uses the following configuration.

Parameter	Description
Region	China (Hangzhou).
Edition	Standard Edition 7.10.0.
Instance specifications	Three zones, three data nodes. Each node has 4 vCPUs, 16 GB of memory, and a 100 GB enhanced SSD (ESSD).

Create an ECS instance for the self-managed Elasticsearch, Kibana, and Logstash instances.

Create an instance by using the wizard. The test environment uses the following configuration.

Parameter	Description
Region	China (Hangzhou).
Instance type	4 vCPUs, 16 GiB of memory.
Image	Public image, CentOS 7.9 64-bit.
Storage	System disk, 100 GiB enhanced SSD (ESSD).
Network	Select the same virtual private cloud (VPC) as your Alibaba Cloud Elasticsearch cluster. Select Assign Public IPv4 Addresses, set the billing method to pay-by-traffic, and set the peak bandwidth to 100 Mbit/s.
Security group	Add an inbound rule to allow access on port 5601 (default Kibana port). Set the authorization object to your client's IP address. Important If your client is on a home or corporate network, use the network's public egress IP, not your computer's private IP. Find your public IP at https://www.whatismyip.com. Setting 0.0.0.0/0 as the authorization object allows all IPv4 addresses but exposes your ECS instance to the public internet. Avoid this in production.

Deploy the self-managed Elasticsearch cluster.

This topic uses a self-managed Elasticsearch 7.6.2 cluster with one data node.

Connect to the ECS instance.

Connect to a Linux instance by using Workbench.
As the root user, create a new user named elastic.
```
useradd elastic
```
Set the password for the elastic user.
```
passwd elastic
```
Follow the prompts to enter and confirm the new password.
Switch to the elastic user.
```
su -l elastic
```

Download and extract the Elasticsearch installation package.

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

Start Elasticsearch.
Navigate to the Elasticsearch installation directory and start the service in the background.
```
cd elasticsearch-7.6.2
./bin/elasticsearch -d
```

Verify that the Elasticsearch service is running.

cd ~ 
curl localhost:9200

A successful response includes the Elasticsearch version number and the 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"
}

Deploy a self-managed Kibana instance and prepare sample data.

This topic uses a self-managed Kibana 7.6.2 instance.
1. Connect to the ECS instance.
  
  Connect to a Linux instance by using Workbench.
  
  Note
  The steps in this topic assume you are running commands as a non-root user unless otherwise specified.
2. Download and extract the Kibana installation package.
```
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. Edit the Kibana configuration file config/kibana.yml and add server.host: "0.0.0.0" to enable remote access.
  Navigate to the Kibana installation directory and edit kibana.yml.
```
cd kibana-7.6.2-linux-x86_64
vi config/kibana.yml
```
  Set the value of server.host to "0.0.0.0" to allow remote connections. The key configuration in the updated file is as follows:
```
# 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. Start Kibana as a non-root user.
```
sudo nohup ./bin/kibana &
```
5. Log on to the Kibana console and add sample data.
  1. Access the Kibana console by using the public IP address of the ECS instance.
    
    The URL is formatted as: http://<your_ecs_instance_public_ip>:5601/app/kibana#/home.
  2. On the Kibana home page, click Try our sample data.
  3. On the Sample data tab, find the Sample web logs card, and click Add data at the bottom of the card to add the sample data.

Deploy a self-managed Logstash instance.

This topic uses a self-managed Logstash 7.10.0 instance with one node.

Connect to the ECS instance.

Connect to a Linux instance by using Workbench.

Note
The steps in this topic assume you are running commands as a non-root user.

Return to the home directory, then download and extract the Logstash installation package.

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

Adjust the Logstash heap size.

The default heap size is 1 GB. Adjust it based on your ECS instance specifications for better migration performance.

Navigate to the Logstash installation directory and edit config/jvm.options to set both the initial and maximum heap size to 8 GB (-Xms8g and -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

Modify the Logstash batch size.

Writing data in batches of 5 MB to 15 MB accelerates data migration.

Edit config/pipelines.yml and change the pipeline.batch.size from 125 to 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

Verify that Logstash is functioning correctly.

Run a simple pipeline that takes standard input and sends it to standard output.
```
bin/logstash -e 'input { stdin { } } output { stdout {} }'
```

After the pipeline starts, type "Hello world!" and press Enter.

If Logstash is working, it prints a structured log message containing "Hello world!" to the 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
}

Step 2 (Optional): Migrate index metadata

Logstash automatically creates an index if one does not exist in the destination cluster, but the auto-created settings and mappings may differ from the source. To ensure consistent index structures, manually create the destination index before migration.

Use the following Python script to create the destination index.

Connect to the ECS instance.

Connect to a Linux instance by using Workbench.

Note
The steps in this topic assume you are running commands as a non-root user.
Create and open a Python script file. This topic uses indiceCreate.py as the filename.
```
sudo vi indiceCreate.py
```

Copy the following code into the Python script file and replace the placeholder values for cluster endpoints, usernames, and passwords with your actual credentials.

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

Run the Python script to create the destination index.
```
sudo /usr/bin/python indiceCreate.py
```
Log on to the Kibana console of the destination cluster and verify that the index was created.
```
GET /_cat/indices?v
```

Step 3: Migrate full data

Connect to the ECS instance.

Connect to a Linux instance by using Workbench.
In the config directory, create and open a Logstash configuration file.
```
cd logstash-7.10.0/config
vi es2es_all.conf
```

Add the following configuration to the file.

Note

Logstash configuration parameters changed in version 8.5. This topic provides configuration examples for both version 7.10.0 and version 8.5.1.
To ensure data accuracy, create separate Logstash pipeline configuration files and migrate data in batches.

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

The Elasticsearch input plugin stops after reading all data. In some environments, Logstash may restart automatically, causing duplicate writes. Use the schedule parameter with a cron expression to run the task at a specific time and prevent this (Scheduling).

For example, to run the task at 1:20 PM on March 5:

schedule => "20 13 5 3 *"

Navigate to the Logstash directory.
```
cd ~/logstash-7.10.0
```

Start the full data migration task.

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

Step 4: Migrate incremental data

Connect to the ECS instance. In the config directory, create and open a new Logstash configuration file for incremental migration.
```
cd config
vi es2es_kibana_sample_data_logs.conf
```
Note
The steps in this topic assume you are running commands as a non-root user.

Add the following configuration to the file.

The following is a sample configuration for version 7.10.0.

Note

For Logstash 8.5 and later, you must remove the document_type => "%{[@metadata][_type]}" line because document types are deprecated.
After configuring the file, starting the scheduled Logstash task triggers the incremental migration.

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 uses UTC timestamps. If your source data uses a different time zone, adjust the query range accordingly. The now-5m in the @timestamp field is based on the server's UTC clock.
Your source index must contain a time field for incremental sync. If it does not, use an ingest pipeline with the _ingest.timestamp metadata field to add @timestamp to documents during indexing.

Navigate to the Logstash directory.
```
cd ~/logstash-7.10.0
```

Start the incremental data migration task.

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

In the Kibana console of the destination Elasticsearch cluster, query for the latest records to verify that incremental data is synchronizing.

The following query finds records in the kibana_sample_data_logs index from the last five minutes.

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

Step 5: Verify the migration results

Verify the full data migration.

Check the index and document count information in the self-managed source cluster.

GET _cat/indices?v

The following result is an example.

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

Check the index and document count in the Alibaba Cloud destination cluster before the migration.

The following is an example of the index information in the Alibaba Cloud Elasticsearch destination cluster before 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

After the full data migration, check the index and document count information in the Alibaba Cloud destination cluster again.

The number of documents should match the source cluster count. In Kibana Dev Tools, run the GET _cat/indices?v command. The result shows that all cluster indices have a health of green and a status of open. The kibana_sample_data_logs index has a docs.count of 14074 and a store.size of 9.4 MB, confirming that data has been successfully migrated to the destination cluster.

Verify the incremental data migration.

Check the most recent records in the self-managed source cluster.

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

The following result is an example.

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

Run the same query in the destination cluster's Kibana console. Matching results confirm successful incremental synchronization.