Migrasi data Elasticsearch yang dikelola sendiri ke Alibaba Cloud Elasticsearch dengan Logstash - Elasticsearch

Topik ini menjelaskan cara melakukan migrasi data penuh atau inkremental dari kluster Elasticsearch yang dikelola sendiri ke Alibaba Cloud Elasticsearch dengan men-deploy Logstash pada instance ECS dan mengonfigurasi pipeline migrasi.

Pertimbangan

Instance ECS yang menjalankan Logstash harus berada dalam VPC yang sama dengan kluster Alibaba Cloud Elasticsearch dan memiliki akses jaringan ke kedua kluster sumber dan tujuan.
Jika aplikasi Anda terus-menerus menulis atau memperbarui data, lakukan migrasi penuh terlebih dahulu, lalu lanjutkan dengan migrasi inkremental berdasarkan timestamp atau field pengenal lainnya. Jika tidak, data lama dapat menimpa data baru di kluster tujuan. Jika kluster tujuan sudah berisi semua data yang ada, cukup lakukan migrasi inkremental.

Prosedur

Langkah 1: Siapkan lingkungan dan instans

Buat kluster Alibaba Cloud Elasticsearch, deploy Elasticsearch dan Logstash yang dikelola sendiri pada instance ECS, serta siapkan data migrasi.
Langkah 2 (Opsional): Migrasi metadata indeks (pengaturan dan pemetaan)

Jalankan skrip Python pada instance ECS untuk migrasi metadata indeks.
Langkah 3: Lakukan migrasi data penuh

Gunakan Logstash untuk migrasi seluruh data dari kluster yang dikelola sendiri ke Alibaba Cloud Elasticsearch.
Langkah 4: Lakukan migrasi data inkremental
Langkah 5: Verifikasi hasil migrasi

Langkah 1: Siapkan lingkungan dan instans

Buat instance Alibaba Cloud Elasticsearch.

Buat instance Alibaba Cloud Elasticsearch. Lingkungan pengujian menggunakan konfigurasi berikut.

Parameter	Deskripsi
Region	China (Hangzhou).
Edition	Standard Edition 7.10.0.
Spesifikasi instans	Tiga zona, tiga node data. Setiap node memiliki 4 vCPU, memori 16 GB, dan SSD ditingkatkan (ESSD) 100 GB.

Buat instance ECS untuk instans Elasticsearch, Kibana, dan Logstash yang dikelola sendiri.

Buat instans menggunakan wizard. Lingkungan pengujian menggunakan konfigurasi berikut.

Parameter	Deskripsi
Region	China (Hangzhou).
Tipe instans	4 vCPU, memori 16 GiB.
Image	Image publik, CentOS 7.9 64-bit.
Storage	Disk sistem, SSD ditingkatkan (ESSD) 100 GiB.
Jaringan	Pilih virtual private cloud (VPC) yang sama dengan kluster Alibaba Cloud Elasticsearch Anda. Pilih Assign Public IPv4 Addresses, atur metode penagihan menjadi pay-by-traffic, dan atur bandwidth puncak menjadi 100 Mbit/s.
Security group	Tambahkan aturan masuk untuk mengizinkan akses pada port 5601 (port default Kibana). Atur objek otorisasi ke alamat IP klien Anda. Penting Jika klien Anda berada di jaringan rumah atau perusahaan, gunakan IP egress publik jaringan tersebut, bukan IP privat komputer Anda. Temukan IP publik Anda di https://www.whatismyip.com. Mengatur 0.0.0.0/0 sebagai objek otorisasi mengizinkan semua alamat IPv4 tetapi membuka instance ECS Anda ke internet publik. Hindari hal ini di lingkungan produksi.

Deploy kluster Elasticsearch yang dikelola sendiri.

Topik ini menggunakan kluster Elasticsearch 7.6.2 yang dikelola sendiri dengan satu node data.

Hubungkan ke instance ECS.

Hubungkan ke instans Linux menggunakan Workbench.
Sebagai pengguna root, buat pengguna baru bernama elastic.
```
useradd elastic
```
Atur password untuk pengguna elastic.
```
passwd elastic
```
Ikuti prompt untuk memasukkan dan mengonfirmasi password baru.
Beralih ke pengguna elastic.
```
su -l elastic
```

Unduh dan ekstrak paket instalasi 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

Jalankan Elasticsearch.
Navigasi ke direktori instalasi Elasticsearch dan jalankan layanan di latar belakang.
```
cd elasticsearch-7.6.2
./bin/elasticsearch -d
```

Verifikasi bahwa layanan Elasticsearch sedang berjalan.

cd ~ 
curl localhost:9200

Tanggapan sukses mencakup nomor versi Elasticsearch dan 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 instans Kibana yang dikelola sendiri dan siapkan data sampel.

Topik ini menggunakan instans Kibana 7.6.2 yang dikelola sendiri.
1. Hubungkan ke instance ECS.
  
  Hubungkan ke instans Linux menggunakan Workbench.
  
  Catatan
  Langkah-langkah dalam topik ini mengasumsikan Anda menjalankan perintah sebagai pengguna non-root kecuali dinyatakan lain.
2. Unduh dan ekstrak paket instalasi 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. Edit file konfigurasi Kibana config/kibana.yml dan tambahkan server.host: "0.0.0.0" untuk mengaktifkan akses remote.
  Navigasi ke direktori instalasi Kibana dan edit kibana.yml.
```
cd kibana-7.6.2-linux-x86_64
vi config/kibana.yml
```
  Atur nilai server.host menjadi "0.0.0.0" untuk mengizinkan koneksi jarak jauh. Konfigurasi utama dalam file yang diperbarui adalah sebagai berikut:
```
# 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. Jalankan Kibana sebagai pengguna non-root.
```
sudo nohup ./bin/kibana &
```
5. Masuk ke konsol Kibana dan tambahkan data sampel.
  1. Akses konsol Kibana menggunakan alamat IP publik instance ECS.
    
    URL-nya diformat sebagai: http://<your_ecs_instance_public_ip>:5601/app/kibana#/home.
  2. Pada halaman utama Kibana, klik Try our sample data.
  3. Pada tab Sample data, temukan kartu Sample web logs, dan klik Add data di bagian bawah kartu untuk menambahkan data sampel.

Deploy instans Logstash yang dikelola sendiri.

Topik ini menggunakan instans Logstash 7.10.0 yang dikelola sendiri dengan satu node.

Hubungkan ke instance ECS.

Hubungkan ke instans Linux menggunakan Workbench.

Catatan
Langkah-langkah dalam topik ini mengasumsikan Anda menjalankan perintah sebagai pengguna non-root.

Kembali ke direktori home, lalu unduh dan ekstrak paket instalasi 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

Sesuaikan ukuran heap Logstash.

Ukuran heap default adalah 1 GB. Sesuaikan berdasarkan spesifikasi instance ECS Anda untuk meningkatkan kinerja migrasi.

Navigasi ke direktori instalasi Logstash dan edit config/jvm.options untuk mengatur ukuran heap awal dan maksimum menjadi 8 GB (-Xms8g dan -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

Ubah ukuran batch Logstash.

Menulis data dalam batch berukuran 5 MB hingga 15 MB mempercepat migrasi data.

Edit config/pipelines.yml dan ubah pipeline.batch.size dari 125 menjadi 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

Verifikasi bahwa Logstash berfungsi dengan benar.

Jalankan pipeline sederhana yang mengambil input standar dan mengirimkannya ke output standar.
```
bin/logstash -e 'input { stdin { } } output { stdout {} }'
```

Setelah pipeline dimulai, ketik "Hello world!" dan tekan Enter.

Jika Logstash berfungsi, ia akan mencetak pesan log terstruktur yang berisi "Hello world!" ke konsol.

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

Langkah 2 (Opsional): Migrasi metadata indeks

Logstash secara otomatis membuat indeks jika indeks tersebut tidak ada di kluster tujuan, tetapi pengaturan dan pemetaan yang dibuat otomatis mungkin berbeda dari sumber. Untuk memastikan struktur indeks yang konsisten, buat indeks tujuan secara manual sebelum migrasi.

Gunakan skrip Python berikut untuk membuat indeks tujuan.

Hubungkan ke instance ECS.

Hubungkan ke instans Linux menggunakan Workbench.

Catatan
Langkah-langkah dalam topik ini mengasumsikan Anda menjalankan perintah sebagai pengguna non-root.
Buat dan buka file skrip Python. Topik ini menggunakan indiceCreate.py sebagai nama file.
```
sudo vi indiceCreate.py
```

Salin kode berikut ke dalam file skrip Python dan ganti nilai placeholder untuk titik akhir kluster, username, dan password dengan kredensial aktual Anda.

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

Jalankan skrip Python untuk membuat indeks tujuan.
```
sudo /usr/bin/python indiceCreate.py
```
Masuk ke konsol Kibana kluster tujuan dan verifikasi bahwa indeks telah dibuat.
```
GET /_cat/indices?v
```

Langkah 3: Migrasi data penuh

Hubungkan ke instance ECS.

Hubungkan ke instans Linux menggunakan Workbench.
Di direktori config, buat dan buka file konfigurasi Logstash.
```
cd logstash-7.10.0/config
vi es2es_all.conf
```

Tambahkan konfigurasi berikut ke file tersebut.

Catatan

Parameter konfigurasi Logstash berubah di versi 8.5. Topik ini menyediakan contoh konfigurasi untuk versi 7.10.0 dan versi 8.5.1.
Untuk memastikan akurasi data, buat file konfigurasi pipeline Logstash terpisah dan migrasi data secara batch.

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

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

Plugin input Elasticsearch berhenti setelah membaca semua data. Di beberapa lingkungan, Logstash mungkin restart secara otomatis, menyebabkan penulisan duplikat. Gunakan parameter schedule dengan ekspresi cron untuk menjalankan tugas pada waktu tertentu dan mencegah hal ini (Scheduling).

Contohnya, untuk menjalankan tugas pada pukul 13.20 tanggal 5 Maret:

schedule => "20 13 5 3 *"

Navigasi ke direktori Logstash.
```
cd ~/logstash-7.10.0
```

Mulai tugas migrasi data penuh.

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

Langkah 4: Migrasi data inkremental

Hubungkan ke instance ECS. Di direktori config, buat dan buka file konfigurasi Logstash baru untuk migrasi inkremental.
```
cd config
vi es2es_kibana_sample_data_logs.conf
```
Catatan
Langkah-langkah dalam topik ini mengasumsikan Anda menjalankan perintah sebagai pengguna non-root.

Tambahkan konfigurasi berikut ke file tersebut.

Berikut adalah contoh konfigurasi untuk versi 7.10.0.

Catatan

Untuk Logstash 8.5 ke atas, Anda harus menghapus baris document_type => "%{[@metadata][_type]}" karena tipe dokumen sudah tidak digunakan lagi.
Setelah mengonfigurasi file, menjalankan tugas Logstash terjadwal akan memicu migrasi inkremental.

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

Penting

Logstash menggunakan timestamp UTC. Jika data sumber Anda menggunakan zona waktu berbeda, sesuaikan rentang kueri tersebut. Nilai now-5m pada field @timestamp didasarkan pada jam UTC server.
Indeks sumber Anda harus berisi field waktu untuk sinkronisasi inkremental. Jika tidak, gunakan ingest pipeline dengan field metadata _ingest.timestamp untuk menambahkan @timestamp ke dokumen selama pengindeksan.

Navigasi ke direktori Logstash.
```
cd ~/logstash-7.10.0
```

Jalankan tugas migrasi data inkremental.

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

Di konsol Kibana kluster Elasticsearch tujuan, kueri catatan terbaru untuk memverifikasi bahwa data inkremental sedang disinkronkan.

Kueri berikut menemukan catatan dalam indeks kibana_sample_data_logs dari lima menit terakhir.

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

Langkah 5: Verifikasi hasil migrasi

Verifikasi migrasi data penuh.

Periksa informasi jumlah indeks dan dokumen di kluster sumber yang dikelola sendiri.

GET _cat/indices?v

Hasil berikut adalah contoh.

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

Periksa jumlah indeks dan dokumen di kluster tujuan Alibaba Cloud sebelum migrasi.

Berikut adalah contoh informasi indeks di kluster tujuan Alibaba Cloud Elasticsearch sebelum migrasi.

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

Setelah migrasi data penuh, periksa kembali informasi jumlah indeks dan dokumen di kluster tujuan Alibaba Cloud.

Jumlah dokumen harus sesuai dengan jumlah di kluster sumber. Di Kibana Dev Tools, jalankan perintah GET _cat/indices?v. Hasil menunjukkan bahwa semua indeks kluster memiliki health green dan status open. Indeks kibana_sample_data_logs memiliki docs.count sebesar 14.074 dan store.size sebesar 9,4 MB, yang mengonfirmasi bahwa data telah berhasil dimigrasi ke kluster tujuan.

Verifikasi migrasi data inkremental.

Periksa catatan terbaru di kluster sumber yang dikelola sendiri.

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

Hasil berikut adalah contoh.

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

Jalankan kueri yang sama di konsol Kibana kluster tujuan. Hasil yang sesuai mengonfirmasi sinkronisasi inkremental berhasil.