Elasticsearch:Migrate data from self-managed Elasticsearch to Alibaba Cloud Elasticsearch using Logstash

Choose the right migration method

Prerequisites

Before you begin, make sure that you have:

• A self-managed Elasticsearch cluster (source) with data to migrate

• An Alibaba Cloud Elasticsearch cluster (destination) created and running. For more information, see Create an Alibaba Cloud Elasticsearch cluster

• An ECS instance in the same Virtual Private Cloud (VPC) as the destination cluster, with a public IPv4 address assigned

• Self-managed Logstash installed on the ECS instance

• Network connectivity verified between Logstash and both Elasticsearch clusters

• A security group inbound rule that allows traffic on port 5601 (for Kibana access)

Step 1: Set up the environment

This section walks through deploying a self-managed Elasticsearch cluster, Kibana, and Logstash on a single ECS instance. If you already have these components running, skip to Step 2.

Step 2: (Optional) Migrate index metadata

When Logstash migrates data, the Auto Indexing feature on the Alibaba Cloud Elasticsearch cluster automatically creates indices. However, auto-created indices may have different settings and mappings than the source indices. To preserve the original index structure, create the indices manually on the destination cluster before migrating data.

The following Python script reads index settings and mappings from the source cluster and creates matching indices on the destination cluster.

1. Connect to the ECS instance as a regular user.

2. Create a Python script file:

      sudo vi indiceCreate.py

3. Paste the following script. Replace the host, username, and password values for both the source and destination clusters: Key behaviors of this script:

   • Preserves the number of primary shards from the source index.

   • Sets the number of replica shards to 0 (DEFAULT_REPLICAS = 0) for faster initial indexing. Increase replicas after migration.

   • Skips system indices (names starting with .). Recreate these manually if needed.

      #!/usr/bin/python
      # -*- coding: UTF-8 -*-
      # File name: indiceCreate.py
      import sys
      import base64
      import time
      import httplib
      import json
      ## Source Elasticsearch cluster
      oldClusterHost = "<source-es-host>:9200"
      ## Source cluster username (leave blank if not required)
      oldClusterUserName = "elastic"
      ## Source cluster password (leave blank if not required)
      oldClusterPassword = "<source-es-password>"
      ## Destination Elasticsearch cluster (find on the Basic Information page)
      newClusterHost = "<destination-es-endpoint>:9200"
      ## Destination cluster username
      newClusterUser = "elastic"
      ## Destination cluster password
      newClusterPassword = "<destination-es-password>"
      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)
          ## Number of primary shards matches the source index
          number_of_shards = settingsDict[index]["settings"]["index"]["number_of_shards"]
          ## Default replica count 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 mappings: \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 ("New index " + newIndexName + " Index settings and mappings: \n" + createstatement)
          endpoint = "/" + newIndexName
          createResult = httpPut(newClusterHost, endpoint, createstatement, newClusterUser, newClusterPassword)
          print ("New index " + newIndexName + " Creation result: " + 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 + " It may be a system index and will not be recreated. You can manually recreate the index based on your business requirements.")

4. Run the script:

      sudo /usr/bin/python indiceCreate.py

5. Verify the created indices. Log on to the Kibana console of the Alibaba Cloud Elasticsearch cluster (see Log on to the Kibana console) and run:

      GET /_cat/indices?v

Step 3: Migrate full data

Perform full data migration first if data writes or updates occur continuously on the source cluster. This establishes a complete baseline on the destination cluster before starting incremental synchronization.

To improve data accuracy, create multiple Logstash pipelines and migrate different indices separately.

The Logstash configuration format changed in version 8.5. This section provides configurations for both 7.x and 8.x.

1. Connect to the ECS instance.

2. Create a Logstash configuration file:

      cd logstash-7.10.0/config
      vi es2es_all.conf

3. Add the configuration for your Logstash version.

   Note

   • The configuration parameters for Logstash 8.5 are different. This section provides configurations for both 7.x and 8.x.

   • To improve data accuracy, create multiple Logstash pipelines and migrate different indices separately.

   Logstash 7.x configuration

      input{
          elasticsearch{
              # Source Elasticsearch host
              hosts =>  ["http://<source-es-host>:9200"]
              # Source cluster credentials
              user => "<source-es-username>"
              password => "<source-es-password>"
              # Indices to migrate (supports wildcards)
              index => "kibana_sample_data_*"
              # Performance settings
              docinfo=>true
              slices => 5
              size => 5000
          }
      }
   
      filter {
        # Remove fields added by Logstash
        mutate {
          remove_field => ["@timestamp", "@version"]
        }
      }
   
      output{
          elasticsearch{
              # Destination cluster endpoint (find on the Basic Information page)
              hosts => ["http://<destination-es-endpoint>:9200"]
              # Destination cluster credentials
              user => "elastic"
              password => "<destination-es-password>"
              # Preserve original index names
              index => "%{[@metadata][_index]}"
              # Preserve original document types (7.x only)
              document_type => "%{[@metadata][_type]}"
              # Preserve original document IDs (remove for better performance)
              document_id => "%{[@metadata][_id]}"
              ilm_enabled => false
              manage_template => false
          }
      }

   Logstash 8.x configuration

      input{
          elasticsearch{
              # Source Elasticsearch host
              hosts =>  ["http://<source-es-host>:9200"]
              # Source cluster credentials
              user => "elastic"
              password => "<source-es-password>"
              # Indices to migrate
              index => "<index-name>"
              # Performance settings
              docinfo => true
              size => 10000
              docinfo_target => "[@metadata]"
          }
      }
   
      filter {
        # Remove fields added by Logstash
        mutate {
          remove_field => ["@timestamp","@version"]
        }
      }
   
      output{
          elasticsearch{
              # Destination cluster endpoint (find on the Basic Information page)
              hosts => ["http://<destination-es-endpoint>:9200"]
              # Destination cluster credentials
              user => "elastic"
              password => "<destination-es-password>"
              # Preserve original index names
              index => "%{[@metadata][_index]}"
              # Preserve original document IDs (remove for better performance)
              document_id => "%{[@metadata][_id]}"
              ilm_enabled => false
              manage_template => false
          }
      }

   Prevent duplicate data: The Elasticsearch input plugin reads all data and then stops the Logstash process. Logstash automatically restarts the process, which can cause duplicate writes with a single pipeline. To prevent this, use the schedule parameter with a cron expression to run the pipeline at a specific time. For example, to run the pipeline once at 13:20 on March 5: For more information about cron syntax, see Scheduling in the Logstash documentation.

   In Logstash 8.x, the document_type setting is removed and docinfo_target => "[@metadata]" is required.

      schedule => "20 13 5 3 *"

4. Navigate to the Logstash installation directory:

      cd ~/logstash-7.10.0

5. Start the migration:

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

Step 4: Migrate incremental data

After full data migration, set up incremental synchronization to capture ongoing changes. The incremental pipeline uses a time-range query and a cron schedule to periodically pull new data from the source cluster.

Run this step as a regular user.

1. Create the incremental migration configuration file:

      cd ~/logstash-7.10.0/config
      vi es2es_kibana_sample_data_logs.conf

2. Add the following configuration. This example queries data updated in the last 5 minutes and runs every minute. Logstash 7.x configuration:

   For Logstash 8.x, remove the document_type => "%{[@metadata][_type]}" setting.

   Important

   - Logstash timestamps are in UTC. If your local timezone is UTC+8, convert timestamps accordingly. The @timestamp range filter in the query parameter uses UTC. - If the source index does not contain a time field, use an ingest pipeline with the _ingest.timestamp parameter to add one.

      input{
          elasticsearch{
              # Source Elasticsearch host
              hosts =>  ["http://<source-es-host>:9200"]
              # Source cluster credentials
              user => "<source-es-username>"
              password => "<source-es-password>"
              # Index to sync
              index => "kibana_sample_data_logs"
              # Query for incremental data (last 5 minutes)
              query => '{"query":{"range":{"@timestamp":{"gte":"now-5m","lte":"now/m"}}}}'
              # Run every minute
              schedule => "* * * * *"
              scroll => "5m"
              docinfo=>true
              size => 5000
          }
      }
   
      filter {
        # Remove fields added by Logstash
        mutate {
          remove_field => ["@timestamp", "@version"]
        }
      }
   
      output{
          elasticsearch{
              # Destination cluster endpoint (find on the Basic Information page)
              hosts => ["http://<destination-es-endpoint>:9200"]
              # Destination cluster credentials
              user => "elastic"
              password => "<destination-es-password>"
              # Preserve original index names
              index => "%{[@metadata][_index]}"
              # Preserve original document types (remove for 8.x)
              document_type => "%{[@metadata][_type]}"
              # Preserve original document IDs
              document_id => "%{[@metadata][_id]}"
              ilm_enabled => false
              manage_template => false
          }
      }

3. Navigate to the Logstash installation directory:

      cd ~/logstash-7.10.0

4. Start the incremental migration:

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

5. Verify incremental data in the Alibaba Cloud Elasticsearch cluster. In the Kibana console, query recently updated records:

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

Step 5: Verify migration results

Performance tuning

Adjust the following parameters to optimize migration throughput:

Set -Xms and -Xmx to the same value to avoid JVM garbage collection pauses. Do not exceed 50% of total system memory.

Credential security

Avoid embedding credentials directly in Logstash configuration files. Use one of these alternatives:

• Logstash keystore: Store sensitive values in the Logstash keystore and reference them in configuration files with ${KEY_NAME} syntax.

• Environment variables: Export credentials as environment variables and reference them in configuration files.

For more information, see Secrets keystore in the Logstash documentation.

Troubleshooting

Post-migration checklist

After migration completes, verify the following:

• [ ] Document counts match between source and destination for all migrated indices

• [ ] Sample document content is identical (spot-check a few documents)

• [ ] Index mappings are correct on the destination cluster

• [ ] No errors in Logstash logs (logs/logstash-plain.log)

• [ ] Application endpoints updated to point to the Alibaba Cloud Elasticsearch cluster

• [ ] Application connectivity to the new cluster verified

• [ ] Incremental sync pipeline stopped after cutover

• [ ] Logstash pipelines cleaned up on the ECS instance

• [ ] Destination cluster monitored for errors and performance

References

• Create an Alibaba Cloud Elasticsearch cluster

• Log on to the Kibana console

• Connect to a Linux instance by using a password or key

• Logstash Elasticsearch input plugin

• Logstash Elasticsearch output plugin

• Logstash secrets keystore