Elasticsearch:Logstash を使用してセルフマネージド型 Elasticsearch から Alibaba Cloud Elasticsearch にデータを移行

適切な移行方法の選択

前提条件

開始する前に、以下を確認してください。

• 移行するデータを含むセルフマネージド型 Elasticsearch クラスター (ソース)

• 作成および実行中の Alibaba Cloud Elasticsearch クラスター (宛先)。詳細については、「Alibaba Cloud Elasticsearch クラスターの作成」をご参照ください。

• 宛先クラスターと同じ Virtual Private Cloud (VPC) 内にあり、パブリック IPv4 アドレスが割り当てられた ECS インスタンス

• ECS インスタンスにインストールされたセルフマネージド型 Logstash

• Logstash と両方の Elasticsearch クラスター間のネットワーク接続性の確認

• ポート 5601 (Kibana アクセス用) でのトラフィックを許可するセキュリティグループのインバウンドルール

ステップ 1: 環境設定

このセクションでは、単一の ECS インスタンスにセルフマネージド型 Elasticsearch クラスター、Kibana、および Logstash をデプロイする手順を説明します。これらのコンポーネントがすでに実行されている場合は、「ステップ 2」に進んでください。

ステップ 2: (オプション) インデックスメタデータの移行

Logstash がデータを移行すると、Alibaba Cloud Elasticsearch クラスターの自動インデックス作成機能によってインデックスが自動的に作成されます。ただし、自動作成されたインデックスは、ソースインデックスとは異なる設定とマッピングを持つ場合があります。元のインデックス構造を保持するには、データを移行する前に宛先クラスターにインデックスを手動で作成します。

次の Python スクリプトは、ソースクラスターからインデックス設定とマッピングを読み取り、宛先クラスターに一致するインデックスを作成します。

1. 一般ユーザーとして ECS インスタンスに接続します。

2. Python スクリプトファイルを作成します。

      sudo vi indiceCreate.py

3. 次のスクリプトを貼り付けます。ソースクラスターと宛先クラスターの両方のホスト、ユーザー名、パスワードの値を置き換えてください。このスクリプトの主な動作:

   • ソースインデックスのプライマリシャードの数を保持します。

   • 初期インデックス作成を高速化するために、レプリカシャードの数を 0 (DEFAULT_REPLICAS = 0) に設定します。移行後にレプリカを増やしてください。

   • システムインデックス (名前が . で始まるもの) をスキップします。必要に応じて、これらを手動で再作成してください。

      #!/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. スクリプトを実行します。

      sudo /usr/bin/python indiceCreate.py

5. 作成されたインデックスを確認します。Alibaba Cloud Elasticsearch クラスターの Kibana コンソールにログインし (「Kibana コンソールへのログイン」をご参照ください)、以下を実行します。

      GET /_cat/indices?v

ステップ 3: 完全データの移行

ソースクラスターでデータの書き込みまたは更新が継続的に発生する場合は、まず完全なデータ移行を実行します。これにより、増分同期を開始する前に、宛先クラスターに完全なベースラインが確立されます。

データ精度を向上させるには、複数の Logstash パイプラインを作成し、異なるインデックスを個別に移行してください。

Logstash の構成フォーマットはバージョン 8.5 で変更されました。このセクションでは、7.x と 8.x の両方の構成を提供します。

1. ECS インスタンスに接続します。

2. Logstash 構成ファイルを作成します。

      cd logstash-7.10.0/config
      vi es2es_all.conf

3. Logstash のバージョンに応じた構成を追加します。

   説明

   • Logstash 8.5 の構成パラメーターは異なります。このセクションでは、7.x と 8.x の両方の構成を提供します。

   • データ精度を向上させるには、複数の Logstash パイプラインを作成し、異なるインデックスを個別に移行してください。

   Logstash 7.x の構成

      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 の構成

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

   重複データの防止: Elasticsearch 入力プラグインはすべてのデータを読み取った後、Logstash プロセスを停止します。Logstash はプロセスを自動的に再起動するため、単一のパイプラインで重複書き込みが発生する可能性があります。これを防ぐには、schedule パラメーターを cron 式とともに使用して、特定の時間にパイプラインを実行します。たとえば、3 月 5 日の 13:20 にパイプラインを 1 回実行するには、次のようにします。cron 構文の詳細については、Logstash ドキュメントの「スケジューリング」をご参照ください。

   Logstash 8.x では、document_type 設定は削除され、docinfo_target => "[@metadata]" が必須です。

      schedule => "20 13 5 3 *"

4. Logstash インストールディレクトリに移動します。

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

5. 移行を開始します。

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

ステップ 4: 増分データの移行

完全なデータ移行後、継続的な変更をキャプチャするために増分同期を設定します。増分パイプラインは、時間範囲クエリと cron スケジュールを使用して、ソースクラスターから新しいデータを定期的にプルします。

このステップは一般ユーザーとして実行してください。

1. 増分移行構成ファイルを作成します。

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

2. 次の構成を追加します。この例では、過去 5 分間に更新されたデータをクエリし、毎分実行します。Logstash 7.x の構成:

   Logstash 8.x の場合は、document_type => "%{[@metadata][_type]}" 設定を削除してください。

   重要

   - Logstash のタイムスタンプは UTC です。ローカルタイムゾーンが UTC + 08:00 の場合は、それに応じてタイムスタンプを変換してください。query パラメーターの @timestamp 範囲フィルターは UTC を使用します。 - ソースインデックスに時間フィールドが含まれていない場合は、ingest パイプラインと _ingest.timestamp パラメーターを使用して 1 つ追加してください。

      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. Logstash インストールディレクトリに移動します。

      cd ~/logstash-7.10.0

4. 増分移行を開始します。

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

5. Alibaba Cloud Elasticsearch クラスターで増分データを確認します。Kibana コンソールで、最近更新されたレコードをクエリします。

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

ステップ 5: 移行結果の確認

パフォーマンスチューニング

移行スループットを最適化するために、次のパラメーターを調整します。

-Xms と -Xmx を同じ値に設定して、JVM ガベージコレクションの一時停止を回避してください。システムの総メモリの 50% を超えないようにしてください。

認証情報のセキュリティ

認証情報を Logstash 構成ファイルに直接埋め込むことは避けてください。次のいずれかの代替方法を使用してください。

• Logstash キーストア: 機密値を Logstash キーストアに保存し、構成ファイルで ${KEY_NAME} 構文で参照します。

• 環境変数: 認証情報を環境変数としてエクスポートし、構成ファイルで参照します。

詳細については、Logstash ドキュメントの「シークレットキーストア」をご参照ください。

トラブルシューティング

移行後のチェックリスト

移行完了後、以下を確認してください。

• [ ] 移行されたすべてのインデックスについて、ソースと宛先間でドキュメント数が一致していること

• [ ] サンプルドキュメントのコンテンツが同一であること (いくつかのドキュメントをスポットチェック)

• [ ] 宛先クラスターのインデックスマッピングが正しいこと

• [ ] Logstash ログ (logs/logstash-plain.log) にエラーがないこと

• [ ] アプリケーションエンドポイントが Alibaba Cloud Elasticsearch クラスターを指すように更新されていること

• [ ] 新しいクラスターへのアプリケーション接続性が確認されていること

• [ ] 切り替え後に増分同期パイプラインが停止されていること

• [ ] ECS インスタンス上の Logstash パイプラインがクリーンアップされていること

• [ ] 宛先クラスターがエラーとパフォーマンスについて監視されていること

参考文献

• Alibaba Cloud Elasticsearch クラスターの作成

• Kibana コンソールへのログイン

• パスワードまたはキーを使用して Linux インスタンスに接続

• Logstash Elasticsearch 入力プラグイン

• Logstash Elasticsearch 出力プラグイン

• Logstash シークレットキーストア