ELK迁移Amazon OpenSearch Service最佳实践

摘要：随着业务的快速发展，线下自己维护的 ELK 集群会面临运维成本高、扩展性受限、资源利用效率低等挑战。为了解决这些问题，同时充分利用云服务的优势，许多客户在迁移上云时，会决定将现有的自建 ELK 架构迁移到 Amazon OpenSearch Service。本文将详细分享一些迁移经验，包括方案设计、实施过程、遇到的挑战以及解决方案，希望能为面临类似场景的客户提供有价值的参考。

01 一、引言

02 二、Part 1：OpenSearch Service Domain 的选择与配置

03 三、Part 2：Amazon OpenSearch Ingestion还是Logstash

04 四、Part 3：数据迁移

05 五、附录

06 六、结语

一、引言

随着业务的快速发展，线下自己维护的 ELK 集群会面临运维成本高、扩展性受限、资源利用效率低等挑战。为了解决这些问题，同时充分利用云服务的优势，许多客户在迁移上云时，会决定将现有的自建 ELK 架构迁移到 Amazon OpenSearch Service。本文将详细分享一些迁移经验，包括方案设计、实施过程、遇到的挑战以及解决方案，希望能为面临类似场景的客户提供有价值的参考。

ELK 是一个缩略词，用来描述由三个常见项目组成的堆栈：Elasticsearch、Logstash 和 Kibana。它使您能够聚合来自所有系统和应用程序的日志，分析这些日志，并创建可视化来进行应用程序和基础设施监控、更快的故障排除、安全分析等；具体来说，Logstash 将摄取、转换数据并将其发送到正确的目的地；Elasticsearch 将为摄取的数据编制索引，并且分析和搜索这些数据；最终，Kibana 会将分析结果可视化。在AWS中，您可以选择在 EC2 上自行部署和管理 ELK集群。但是，通过向上和向下扩展来满足您的业务要求，或者实现安全性和合规性，对于管理来说也是一个挑战。Amazon OpenSearch Service 是一种完全托管的开源替代方案，可让您更轻松地在 AWS 上安全、经济高效地部署、操作和扩展 OpenSearch 集群。

从ELK迁移到 Amazon OpenSearch Service 直观上看起来并不复杂。作为云上的托管服务，Amazon OpenSearch Service包括多个组成部分：OpenSearch Service Domain, Amazon OpenSearch Ingestion, OpenSearch Dashboard，并且可以简单地理解为Elasticsearch、Logstash 和 Kibana的对应集成模块。但是，在迁移的过程中却有非常多需要关注的细节点，才能用好合适的云上方案。

以下是典型的ELK迁移至Amazon OpenSearch Service的云上方案架构图。

二、Part 1：OpenSearch Service Domain 的选择与配置

在Amazon OpenSearch Service 中，Domain等同于一个 OpenSearch 集群，具体来说是一个具有您指定的设置、实例类型、实例数量和存储资源的集群。Amazon OpenSearch Service 支持 OpenSearch 和传统的 ElasticSearch OSS（到 7.10 版本）。在创建域时，您可以选择使用具体哪种搜索引擎。
与此同时，Amazon OpenSearch Serverless 是一种针对Amazon OpenSearch Service的按需自动缩放配置。与需要手动容量管理的预配置 OpenSearch Domain不同， OpenSearch Serverless 集合会根据应用程序的需求自动扩展计算资源，不存在集群或者节点的概念。

可以看到，OpenSearch Serverless 为不频繁、间歇性或不可预测的工作负载提供了经济实惠的解决方案。它通过根据应用程序的使用情况自动扩展计算容量来优化成本。它们之间的功能差别以及收费不同可以详见：https://docs.aws.amazon.com/zh_cn/opensearch-service/latest/developerguide/serverless-comparison.html。但是需要注意的是，对于数据量非常大的客户，在本文所写之时，OpenSearch Serverless 目前存在3.0 TB 的数据上限限制。

从ELK迁移到Amazon Opensearch Service时，首先要了解您的存储需求、分片数量，从而对Domain实例类型设置做出有根据的初步估计。可以参考 https://docs.aws.amazon.com/opensearch-service/latest/developerguide/sizing-domains.html 做出有针对性的选择。无论任何时候，选择最新一代的实例类型，选择最新的Amazon EBS gp3卷，在稳定运行之后选择购买预留实例RI，永远是不错的选择。

一般来说，对于日志类数据、时间序列分析、点击流分析，我们会选择使用滚动索引策略，也就是数据持续流入一组具有索引周期和保留期限的索引中。这里我们会要用到 Index State Management(ISM)，它是 OpenSearch 中的一个插件，允许您根据索引年龄、索引大小或文档数量的变化触发这些定期管理操作，从而实现自动化。使用 ISM 插件，您可以定义策略，根据用例自动处理索引的滚动更新或删除操作。我们接下来介绍一下如何具体配置：

首先，我们需要了解触发滚动索引的相关参数，这几个条件可以同时设置，只要有其中一项条件达到了，就会触发建立新的索引的动作。

参数	说明
min_index_age	滚动索引所需的最短时间。索引时间是指从创建到现在的时间。支持的单位包括 d（天）、h（小时）、m（分钟）、s（秒）、ms（毫秒）和 micros（微秒）。
min_doc_count	滚动索引所需的最小文档数。
min_size	滚动索引所需的主分片存储空间（不计副本）的最小大小。例如，如果您将 min_size 设置为 100 GiB，并且您的索引包含 5 个主分片和 5 个副本分片，每个分片大小为 20 GiB，则所有主分片的总大小为 100 GiB，因此会发生滚动。
min_primary_shard_size	滚动索引所需的单个主分片的最小存储大小。例如，如果您将 min_primary_shard_size 设置为 30 GiB，而索引中某个主分片的大小大于该条件，则会发生滚动。

了解了这些参数的作用之后，可以按照如下步骤来实现OpenSearch滚动索引的配置：

1. 我们需要创建一个索引模版（Index Template），它允许用户预先定义索引的配置，以便在创建新索引时自动应用这些配置。

PUT _index_template/<index_template>
{
    "index_patterns":[
        "nginx-log-*"
    ],
    "template":{
        "settings":{
            "refresh_interval":"10s",
            "number_of_shards":"3",
            "number_of_replicas":"1",
            "index.codec":"zstd",
            "plugins.index_state_management.rollover_alias":"<index-alias>",
            "routing":{
                "rebalance":{
                    "enable":"none"
                }
            },
            "translog":{
                "flush_threshold_size":"3072mb",
                "sync_interval":"30s",
                "durability":"async"
            }
        }
    },
    "aliases": { 
        "<index-alias>": {} 
    }
}

指定压缩算法 "index.codec": "zstd",，测试可以压缩 30%的空间。
指定分片数量，对于日志场景，我们对单个分片存储的数据量建议不超过 50GB，而索引的分片数量在创建之后，就无法修改，这时对于数据量增加后，就可以通过索引模版来修改分片数量，在新建的索引就可以应用这个配置。

2. 创建滚动索引策略

如下创建的滚动策略，我们设置了索引滚动的规则：主分片大小120gb，单个主分片 40gb，最小滚动时间 1天，最小文档数 1000000。另外，还指定了热数据存储 7天，温存储 14天，冷存储 30天，超过 30天的数据会被删除。

PUT _plugins/_ism/policies/<custom-name-policy>
{
    "policy":{
        "description":"rollover policy.",
        "default_state":"hot",
        "ism_template":{
            "index_patterns":[
                "<index-name>-*"
            ],
            "priority":100
        },
        "states":[
            {
                "name":"hot",
                "actions":[
                    {
                        "rollover":{
                            "min_size": "120gb",
                            "min_doc_count": 1000000,
                            "min_index_age": "1d",
                            "min_primary_shard_size": "40gb"
                        }
                    }
                ],
                "transitions":[
                    {
                        "state_name":"hot-warm",
                        "conditions": {
                            "min_index_age": "7d"
                        }
                    }
                ]
            },
            {
                "name":"hot-warm",
                "actions":[
                    {
                        "warm_migration":{

                        }
                    }
                ],
                "transitions":[
                    {
                        "state_name":"warm-cold",
                        "conditions":{
                            "min_index_age":"14d"
                        }
                    }
                ]
            },
            {
                "name":"warm-cold",
                "actions":[
                    {
                        "cold_migration":{
                            "timestamp_field":"@timestamp"
                        }
                    }
                ],
                "transitions":[
                    {
                        "state_name":"delete",
                        "conditions":{
                            "min_index_age":"30d"
                        }
                    }
                ]
            },
            {
                "name":"delete",
                "actions":[
                    {
                        "cold_delete":{

                        }
                    }
                ]
            }
        ]
    }
}

如上，实现数据从热（7天）→温（14天）→ 冷（30天）的迁移策略，使用了 Transitions 来定义。

Transitions(转换)的作用是

1. 定义一个状态需要改变的条件。当当前状态的所有 actions 执行完成后，策略就会开始检查 transitions 中的条件

2. ISM会按照transitions定义的顺序依次评估。一旦找到一个评估为true的transition，就会停止并将下一个状态设置为该transition中定义的状态

接着，需要附加策略给别名样例：

POST _opendistro/_ism/add/<index-alias>
{
    "policy_id": "<custom-name-policy>"
}

最后，需要手动创建第一个索引

PUT %3C<index-name>-%7Bnow%7Byyyy.MM.dd%7D%7D-000001%3E
{
    "aliases": {
        "<index-alias>": {
            "is_write_index": true
            }
       } 
}

上面的创建索引，使用date math来自动根据时间生成索引，rollover 发生的时候会根据时间生成索引名字，这样通过和 ISM 策略配合，实现索引按天或者小时+数据大小进行滚动。这里生成的索引的日期格式是到天，也可以到小时，只需要调整日期格式即可yyyy.MM.dd.HH。

三、Part 2：Amazon OpenSearch Ingestion还是Logstash

将数据摄入 OpenSearch 的方法可以像之前ELK部分一样通过Filebeat + Logstash的方式，配置简单灵活。

首先，安装Filebeat和Logstash opensearch plugin：

# Step 1. 安装 Filebeat
wget https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-oss-7.16.2-x86_64.rpm
rpm -ivh filebeat-oss-7.16.2-x86_64.rpm

# Step 2. 安装 Logstash
# 2.1. 下载 logstash
wget https://artifacts.opensearch.org/logstash/logstash-oss-with-opensearch-output-plugin-7.16.3-macos-x64.tar.gz

# 2.2. 解压
tar -zxvf logstash-oss-with-opensearch-output-plugin-7.16.3-macos-x64.tar.gz
cd logstash-7.16.3/

# 2.3. 安装logstash opensearch plugin
./bin/logstash-plugin install --preserve logstash-input-opensearch

其次进行Filebeat配置：Filebeat 部署于需要收集日志的服务器节点，通过 paths 参数指定需要采集的日志文件。

通过 Filebeat 直接向 OpenSearch 摄入数据

filebeat.inputs:
  - type: log
    enabled: true
    paths:
      - /var/log/*.log
    tags: "var-log-all"

filebeat.config.modules:
  path: ${path.config}/modules.d/*.yml
  reload.enabled: false

setup.template.enabled: true
setup.template.name: "nginx_log_template"
setup.template.pattern: "nginx-log-*"
setup.template.fields: "fields.yml"

processors:
  - drop_fields:
      fields: ["agent","host","input","ecs"]
      ignore_missing: false

        #output.elasticsearch:
        hosts: ["vpc-xxxxx.us-east-1.es.amazonaws.com:443"]
        protocol: "https"
        username: "admin"
        password: "xxx"
        indices:
           - index: "var-log-all-%{+yyyy.MM.dd}"
           when.contains:
               tags: "var-log-all"
           - index: "nginx-log-%{+yyyy.MM.dd}"
           when.contains:
            tags: "nginx-log"
           - index: "nginx-log-%{+yyyy.MM.dd}"
           when.contains:
           tags: "nginx-log"

setup.ilm.enabled: false
ilm.enabled: false

Filebeat 向 Logstash 摄入数据。这种情况下，在向 logstash 摄入数据之前，需要先启动 logstash 程序，这样会有一个监听端口【默认5044】启动。

filebeat.inputs:
  - type: log
    enabled: true
    paths:
      - /var/log/*.log
    tags: "var-log-all"

filebeat.config.modules:
  path: ${path.config}/modules.d/*.yml
  reload.enabled: false

setup.template.enabled: true
setup.template.name: "nginx_log_template"
setup.template.pattern: "nginx-log-*"
setup.template.fields: "fields.yml"

processors:
  - drop_fields:
      fields: ["agent","host","input","ecs"]
      ignore_missing: false

output.logstash:
  # The Logstash hosts
  hosts: ["localhost:5044"]

setup.ilm.enabled: false
ilm.enabled: false

其中，我们可以通过drop_fields字段来控制哪些filebeat字段不显示

processors:
  - drop_fields:
      fields: ["agent","host","input","ecs"]
      ignore_missing: false

最后，启动Logstash

export LS_JAVA_HOME=/usr/lib/jvm/jre-17
~/logstash-7.16.3/bin/logstash -f /etc/logstash/conf.d/logstash-beat.conf

logstash的配置如下：

# Sample Logstash configuration for creating a simple
# Beats -> Logstash -> Elasticsearch pipeline.

input {
  beats {
    port => 5044
  }
}


output {
  opensearch {
    hosts => ["https://vpc-storage-layer-wuc4242nw3vy3u6cdtjinremiq.us-east-1.es.amazonaws.com:443"]
    index => "opensearch-logstash-beat-%{+YYYY.MM.dd}"
    user => "admin"
    password => "xxx"
  }
}
input {
   # 从filebeat取数据，端口与filebeat配置文件一致
   beats {
     port => 5044
   }
}

filter {
    # 只对nginx的json日志做json解析，系统message为其他格式，无需处理
    if [fields][log_type] == "nginx"{
      json {
         source => "message"
         remove_field => ["beat","offset","tags","prospector"] #移除字段，不需要采集
      }
      date {
        match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"] #匹配timestamp字段
        target => "@timestamp"  #将匹配到的数据写到@timestamp字段中
      }
  }
}

output {
       # 输出es，这的filetype就是在filebeat那边新增的自定义字段名
       if [fields][log_type] == "es" {
         opensearch {
            hosts => ["https://vpc-storage-layer-wuc4242nw3vy3u6cdtjinremiq.us-east-1.es.amazonaws.com:443"]
            index => "es-%{+YYYY.MM}"
            user => "admin"
            password => "xxx"
        }
       }
       if [fields][log_type] == "nginx" {
         opensearch {
            hosts => ["https://vpc-storage-layer-wuc4242nw3vy3u6cdtjinremiq.us-east-1.es.amazonaws.com:443"]
            index => "nginx-%{+YYYY.MM}"
            user => "admin"
            password => "xxx"
        }       
       }

}

在 logstash 中根据不同的类型分 index 存储

input {
   # 从filebeat取数据，端口与filebeat配置文件一致
   beats {
     port => 5044
   }
}

filter {
    # 只对nginx的json日志做json解析，系统message为其他格式，无需处理
    if [fields][log_type] == "nginx"{
      json {
         source => "message"
         remove_field => ["beat","offset","tags","prospector"] #移除字段，不需要采集
      }
      date {
        match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"] #匹配timestamp字段
        target => "@timestamp"  #将匹配到的数据写到@timestamp字段中
      }
  }
}

output {
       # 输出es，这的filetype就是在filebeat那边新增的自定义字段名
       if [fields][log_type] == "es" {
         opensearch {
            hosts => ["https://vpc-storage-layer-wuc4242nw3vy3u6cdtjinremiq.us-east-1.es.amazonaws.com:443"]
            index => "es-%{+YYYY.MM}"
            user => "admin"
            password => "xxx"
        }
       }
       if [fields][log_type] == "nginx" {
         opensearch {
            hosts => ["https://vpc-storage-layer-wuc4242nw3vy3u6cdtjinremiq.us-east-1.es.amazonaws.com:443"]
            index => "nginx-%{+YYYY.MM}"
            user => "admin"
            password => "xxx"
        }       
       }

}

同时，在 filebeat 中定义关联到在 OpenSearch 上创建好的模版。这样从 filebeat 开始运行时，会根据 OpenSearch 的模版定义来创建 index。但是注意修改策略/索引模版，只针对新建索引生效。

setup.template.enabled: true
setup.template.name: "nginx_log_template"
setup.template.pattern: "nginx-log-*"
setup.template.fields: "fields.yml"

也可以使用 Amazon OpenSearch Service 托管的数据集成服务 OpenSearch Ingestion Pipeline 来实现数据的写入，Amazon OpenSearch Ingestion 是一个完全托管的、无服务器数据收集器，它将实时日志、指标和跟踪数据传送到 Amazon OpenSearch Service 域和 OpenSearch Serverless 集合。使用 OpenSearch Ingestion，就不再需要使用第三方解决方案（如 Logstash 或 Jaeger）将数据导入到 OpenSearch Service 域和 OpenSearch Serverless 集合中。只需配置数据生产者将数据发送到 OpenSearch Ingestion，然后它会自动将数据传送到您指定的域或集合。您还可以配置 OpenSearch Ingestion 在传送数据之前对其进行转换。此外，使用 OpenSearch Ingestion，您无需担心配置服务器、管理和修补软件，或扩展服务器集群。您可以直接在 AWS 管理控制台中配置摄取管道，OpenSearch Ingestion 会负责管理和扩展它们。

version: "2"
msk-pipeline:

  source:
    kafka:
      # Prevent data loss by only considering logs to be processed successfully after they are received by the opensearch sink
      acknowledgments: true
      topics:
         - name: "nginx"
           group_id: "aos_ingestion_nginx_group"
           serde_format: "json"                 # Remove, if Schema Registry is used. (Other option is "plaintext")
         - name: "nginx-error"
           group_id: "aos_ingestion_nginx_group"
           serde_format: "json"                 # Remove, if Schema Registry is used. (Other option is "plaintext")

           # The fields below are optional and can be tuned as needed
           # fetch_max_bytes: 52428800
           # fetch_max_wait: 500                (in msecs)
           # fetch_min_bytes: 1                 (in MB)
           # max_partition_fetch_bytes: 1048576
           # consumer_max_poll_records: 500
           # auto_offset_reset: "earliest"
           # key_mode: include_as_field
     # Enable this configuration if Glue schema registry is used
      # schema:
      #   type: aws_glue

      aws:
        # Provide the Role ARN with access to MSK. This role should have a trust relationship with osis-pipelines.amazonaws.com
        sts_role_arn: "arn:aws:iam::<account id>:role/OpenSearch-Role-20231224"
        # Provide the region of the domain.
        region: "us-east-1"
        msk:
          # Provide the MSK ARN.
          arn: "arn:aws:kafka:us-east-1:<account id>:cluster/kafka-cluster-02/1070092a-6988-4095-91fb-18bce30e5dc7-8"

  sink:
      - opensearch:
          # Provide an AWS OpenSearch Service domain endpoint
          hosts: [ "https://vpc-mycluster-kqsemxfudexm7fqkqsmsfaxvm4.us-east-1.es.amazonaws.com" ]
          aws:
            # Provide a Role ARN with access to the domain. This role should have a trust relationship with osis-pipelines.amazonaws.com
            sts_role_arn: "arn:aws:iam::<account id>:role/OpenSearch-Role-20231224"
            # Provide the region of the domain.
            region: "us-east-1"
            # Enable the 'serverless' flag if the sink is an Amazon OpenSearch Serverless collection
            serverless: false
            # serverless_options:
              # Specify a name here to create or update network policy for the serverless collection
              # network_policy_name: "network-policy-name"
            # Index names can be auto-generated from topic name. In such case topic name should adhere to OpenSearch Index naming conventions.
          index: "index-${fields/log_type}-%{yyyy.MM.dd}"
          # Enable the 'distribution_version' setting if the AWS OpenSearch Service domain is of version Elasticsearch 6.x
          # distribution_version: "es6"
          # Enable and switch the 'enable_request_compression' flag if the default compression setting is changed in the domain. See https://docs.aws.amazon.com/opensearch-service/latest/developerguide/gzip.html
          # enable_request_compression: true/false
          # Optional: Enable the S3 DLQ to capture any failed requests in an S3 bucket. Delete this entire block if you don't want a DLQ.

注意Logstash到Opensearch的部分写法可能会存在差异，需要比对结果是否存在异常。

和Logstash不同，OpenSearch Ingestion Pipeline通过对OpenSearch Ingestion 管道中提取、转换和路由数据所需的计算收费。计算容量以 OpenSearch 计算单位 (OCU) 来衡量。OCU 的数量直接对应于摄取数据或对数据进行转换所需的 CPU 和内存。一个 OCU 由 8GB 的 RAM 和 2 个 vCPU 组成。OCU 按小时计费，粒度为每分钟。因此，您可以配置每个管道的最小和最大 OCU 数量以控制成本。同时，OpenSearch Ingestion 允许您完全暂停未使用的管道，管道暂停时不消耗 OCU，不使用的Ingestion如果无需向管道中提取数据，您可以在相应时段停止管道。最后，经过验证，合并多个Ingestion Pipeline从而降低Ingestion OCU成本存在可行性。具体来说，多个 topic，可以通过 filebeat 加fields.标签，写到不同的 index 中。这样可以减少多个pipeline本身的最小capacity（https://docs.aws.amazon.com/opensearch-service/latest/developerguide/ingestion-scaling.html）部分造成的最低成本。

另外，一些使用OpenSearch Ingestion Pipeline的其他最佳实践可以查看：https://docs.aws.amazon.com/zh_cn/opensearch-service/latest/developerguide/osis-best-practices.html

四、Part 3：数据迁移

1. 从Kibana到OpenSearch Dashboard的迁移

从ELK迁移到Amazon OpenSearch上，会使用集成的Opensearch Dashboard来替换原来使用的Kibana。一般来说，在Kibana上面通过Management→Stack Management → Saved Objects可以export出对应的报表到一个NDJSON文件，并且通过Opensearch Dashboard上面Management→Dashboard Management → Saved Objects可以import刚刚导出的NDJSON文件。