将 10 亿条日志行从 OpenSearch 迁移到 Elasticsearch

作者：Ugo Sangiorgi

当前从 OpenSearch 迁移到 Elasticsearch® 的选项有哪些？

OpenSearch 是 Elasticsearch 7.10 的一个分支，最近与自身有很大分歧，导致了一组不同的功能和不同的性能，正如该基准测试所示（提示：它目前比 Elasticsearch 慢得多）。

鉴于这两种解决方案之间的差异，从 OpenSearch 恢复快照是不可能的，从远程重新索引也是不可能的，因此我们唯一的选择是使用介于两者之间的东西，从 OpenSearch 读取并写入 Elasticsearch。

本博客将向你展示从 OpenSearch 迁移到 Elasticsearch 以获得更好的性能和更少的磁盘使用量是多么容易！

10 亿条日志行

我们将使用用于基准测试的部分数据集，该数据集在磁盘上占用约 0.5 TB（包括副本），时间跨度超过一周（2023 年 1 月 1 日至 7 日）。

我们总共有 1,009,165,775 个文档，在 OpenSearch 中占用了 453.5GB 的空间，包括副本。 即每个文档 241.2KB。 稍后当我们在 Elasticsearch 中启用一些优化时，这将变得很重要，这将在不牺牲性能的情况下降低总大小！

这个十亿日志行数据集分布在九个索引上，这些索引是我们称为 “logs-myapplication-prod” 的数据流的一部分。 根据最佳分片大小调整的最佳实践，我们的主分片大小约为 25GB。 GET _cat/indices 向我们显示我们正在处理的索引：

index                              docs.count pri rep pri.store.size store.size
.ds-logs-myapplication-prod-000049  102519334   1   1         22.1gb     44.2gb
.ds-logs-myapplication-prod-000048  114273539   1   1         26.1gb     52.3gb
.ds-logs-myapplication-prod-000044  111093596   1   1         25.4gb     50.8gb
.ds-logs-myapplication-prod-000043  113821016   1   1         25.7gb     51.5gb
.ds-logs-myapplication-prod-000042  113859174   1   1         24.8gb     49.7gb
.ds-logs-myapplication-prod-000041  112400019   1   1         25.7gb     51.4gb
.ds-logs-myapplication-prod-000040  113362823   1   1         25.9gb     51.9gb
.ds-logs-myapplication-prod-000038  110994116   1   1         25.3gb     50.7gb
.ds-logs-myapplication-prod-000037  116842158   1   1         25.4gb     50.8gb

OpenSearch 和 Elasticsearch 集群具有相同的配置：3 个节点，64GB RAM 和 12 个 CPU 核心。 就像基准测试一样，集群在 Kubernetes 中运行。

将数据从 A 移动到 B

通常，将数据从一个 Elasticsearch 集群移动到另一个集群非常简单，如果集群彼此版本兼容，则可以轻松进行快照和恢复；如果你需要实时同步并最大限度地减少停机时间，则可以从远程重新建立索引。 当将数据从 OpenSearch 迁移到 Elasticsearch 时，这些方法不适用，因为这些项目与 7.10 分支有显着差异。 然而，有一种方法可行：滚动（scrolling）。

滚动 (scrolling)

滚动涉及使用外部工具（例如 Logstash®）从源集群读取数据并将其写入目标集群。 这种方法提供了高度的定制性，允许我们在需要时在迁移过程中转换数据。 以下是使用 Logstash 的几个优点：

• 轻松并行化：编写可以从索引的不同 “切片” 读取的并发作业非常容易，从本质上最大化了我们的吞吐量。
• 排队：Logstash 在发送前自动对文档进行排队。
• 自动重试：如果数据传输过程中出现失败或错误，Logstash 会自动尝试重新发送数据； 此外，它将停止频繁查询源集群，直到重新建立连接，所有这些都无需手动干预。

滚动允许我们进行初始搜索，并不断从 Elasticsearch 中提取批量结果，直到没有更多结果为止，类似于关系数据库中 “光标” 的工作方式。

滚动搜索 (scrolled search) 通过冻结构成索引的段直到发出请求时来及时拍摄快照，从而防止这些段合并。 因此，在发出初始搜索请求后，滚动看不到对索引所做的任何更改。

迁移策略

如果不进行优化，从 A 读取数据并写入 B 可能会很慢，因为它涉及对结果进行分页、通过网络将每个批次传输到 Logstash，Logstash 将在另一批次中组装文档，然后再次通过网络将这些批次传输到 Elasticsearch（文档将被索引的位置）。 因此，当涉及到如此大的数据集时，我们必须非常高效，并尽可能地提取每一点性能。

让我们从事实开始 —— 我们对需要传输的数据了解多少？ 我们在数据流中有 9 个索引，每个索引大约有 1 亿个文档。 让我们仅使用其中一个索引进行测试并测量索引率，看看迁移需要多长时间。 通过激活 Elastic® 中的监控功能，然后导航到你想要检查的索引，可以看到索引率。

在深处滚动

传输日志行的最简单方法是让 Elasticsearch 滚动整个数据集，并在完成后检查它。 这里我们将介绍前两个变量：PAGE_SIZE 和 BATCH_SIZE。 前者是我们每次需要时将从源获取多少记录，后者是 Logstash 将有多少文档组装在一起并写入目标索引。

对于如此大的数据集，随着深度分页的进行，滚动速度会减慢。 索引速率从 6,000 个文档/秒开始，然后稳步下降到 700 个文档/秒，因为分页变得非常深。 如果不进行任何优化，我们需要 19 天（！）才能迁移 10 亿个文档。 我们可以做得更好！

切我好

我们可以通过使用一种称为 “切片滚动 (Sliced scroll)” 的方法来优化滚动，其中我们将索引分成不同的切片以独立地使用它们。

这里我们将介绍最后两个变量：SLICES 和 WORKERS。 切片的数量不能太小，因为随着时间的推移，性能会急剧下降，并且切片的数量不能太大，因为维护滚动的开销会抵消较小搜索的好处。

让我们首先迁移具有不同参数的单个索引（我们拥有的九个索引中的一个），看看哪种组合可以为我们提供最高的吞吐量。

看起来我们有一组很好的候选方案可以最大化单个索引的吞吐量，每秒处理 12K 到 14K 文档。 这并不意味着我们已经达到了上限。 尽管搜索操作是单线程的，并且每个切片都会触发顺序搜索操作来读取数据，但这并不妨碍我们并行读取多个索引。

默认情况下，打开 scrolls 的最大数量为 500 — 可以使用 search.max_open_scroll_context 集群设置更新此限制，但默认值对于此特定迁移来说已经足够了。

让我们迁移吧

准备我们的目的地索引

我们将创建一个名为 logs-myapplication-reindex 的数据流来写入数据，但在对任何数据进行索引之前，让我们确保正确设置索引模板和索引生命周期管理配置。 索引模板充当创建新索引的蓝图，允许你定义应在索引中一致应用的各种设置。

索引生命周期管理策略

索引生命周期管理 (ILM) 同样重要，因为它可以在索引的整个生命周期中自动管理索引。 借助 ILM，你可以定义策略来确定数据应保留多长时间、何时应将其转入新索引以及何时应删除或归档旧索引。 我们的政策非常简单：

PUT _ilm/policy/logs-myapplication-lifecycle-policy
{
  "policy": {
    "phases": {
      "hot": {
        "actions": {
          "rollover": {
            "max_primary_shard_size": "25gb"
          }
        }
      },
      "warm": {
        "min_age": "0d",
        "actions": {
          "forcemerge": {
            "max_num_segments": 1
          }
        }
      }
    }
  }
}

索引模板（节省 23% 的磁盘空间）

既然我们在这里，我们将继续启用 Synthetic Source，这是一个聪明的功能，它允许我们存储和丢弃原始 JSON 文档，同时仍然在需要时从存储的字段重建它。

对于我们的示例，启用 Synthetic Source 使存储效率显着提高了 23.4%，将存储单个文档所需的大小从 OpenSearch 中的 241.2KB 减少到 Elasticsearch 中的 185KB。

因此，我们的完整索引模板是：

PUT _index_template/logs-myapplication-reindex
{
  "index_patterns": [
    "logs-myapplication-reindex"
  ],
  "priority": 500,
  "data_stream": {},
  "template": {
    "settings": {
      "index": {
        "lifecycle.name": "logs-myapplication-lifecycle-policy",
        "codec": "best_compression",
        "number_of_shards": "1",
        "number_of_replicas": "1",
        "query": {
          "default_field": [
            "message"
          ]
        }
      }
    },
    "mappings": {
      "_source": {
        "mode": "synthetic"
      },
      "_data_stream_timestamp": {
        "enabled": true
      },
      "date_detection": false,
      "properties": {
        "@timestamp": {
          "type": "date"
        },
        "agent": {
          "properties": {
            "ephemeral_id": {
              "type": "keyword",
              "ignore_above": 1024
            },
            "id": {
              "type": "keyword",
              "ignore_above": 1024
            },
            "name": {
              "type": "keyword",
              "ignore_above": 1024
            },
            "type": {
              "type": "keyword",
              "ignore_above": 1024
            },
            "version": {
              "type": "keyword",
              "ignore_above": 1024
            }
          }
        },
        "aws": {
          "properties": {
            "cloudwatch": {
              "properties": {
                "ingestion_time": {
                  "type": "keyword",
                  "ignore_above": 1024
                },
                "log_group": {
                  "type": "keyword",
                  "ignore_above": 1024
                },
                "log_stream": {
                  "type": "keyword",
                  "ignore_above": 1024
                }
              }
            }
          }
        },
        "cloud": {
          "properties": {
            "region": {
              "type": "keyword",
              "ignore_above": 1024
            }
          }
        },
        "data_stream": {
          "properties": {
            "dataset": {
              "type": "keyword",
              "ignore_above": 1024
            },
            "namespace": {
              "type": "keyword",
              "ignore_above": 1024
            },
            "type": {
              "type": "keyword",
              "ignore_above": 1024
            }
          }
        },
        "ecs": {
          "properties": {
            "version": {
              "type": "keyword",
              "ignore_above": 1024
            }
          }
        },
        "event": {
          "properties": {
            "dataset": {
              "type": "keyword",
              "ignore_above": 1024
            },
            "id": {
              "type": "keyword",
              "ignore_above": 1024
            },
            "ingested": {
              "type": "date"
            }
          }
        },
        "host": {
          "type": "object"
        },
        "input": {
          "properties": {
            "type": {
              "type": "keyword",
              "ignore_above": 1024
            }
          }
        },
        "log": {
          "properties": {
            "file": {
              "properties": {
                "path": {
                  "type": "keyword",
                  "ignore_above": 1024
                }
              }
            }
          }
        },
        "message": {
          "type": "match_only_text"
        },
        "meta": {
          "properties": {
            "file": {
              "type": "keyword",
              "ignore_above": 1024
            }
          }
        },
        "metrics": {
          "properties": {
            "size": {
              "type": "long"
            },
            "tmin": {
              "type": "long"
            }
          }
        },
        "process": {
          "properties": {
            "name": {
              "type": "keyword",
              "ignore_above": 1024
            }
          }
        },
        "tags": {
          "type": "keyword",
          "ignore_above": 1024
        }
      }
    }
  }
}

构建自定义 Logstash 镜像

我们将使用容器化 Logstash 进行此迁移，因为两个集群都位于 Kubernetes 基础设施上，因此更容易启动一个与两个集群通信的 Pod。

由于 OpenSearch 不是官方的 Logstash 输入，因此我们必须构建一个包含 logstash-input-opensearch 插件的自定义 Logstash 映像。 让我们使用 docker.elastic.co/logstash/logstash:8.10.0 中的基本映像并安装插件：

FROM docker.elastic.co/logstash/logstash:8.10.0

USER logstash
WORKDIR /usr/share/logstash
RUN bin/logstash-plugin install logstash-input-opensearch

编写 Logstash 管道

现在我们有了 Logstash Docker 镜像，我们需要编写一个从 OpenSearch 读取数据并写入 Elasticsearch 的管道。

input

input {
    opensearch {
      hosts => ["os-cluster:9200"]
      ssl => true
      ca_file => "/etc/logstash/certificates/opensearch-ca.crt"
      user => "${OPENSEARCH_USERNAME}"
      password => "${OPENSEARCH_PASSWORD}"
      index => "${SOURCE_INDEX_NAME}"
      slices => "${SOURCE_SLICES}"
      size => "${SOURCE_PAGE_SIZE}"
      scroll => "5m"
      docinfo => true
      docinfo_target => "[@metadata][doc]"
    }
}

让我们分解最重要的输入参数。 这里的值全部表示为环境变量：

• Hosts：指定 OpenSearch 集群的主机和端口。 在本例中，它连接到端口 9200 上的 “os-cluster”。
• index：指定 OpenSearch 集群中检索日志的索引。 在本例中，它是 “logs-myapplication-prod”，它是包含当前索引的数据流（例如，.ds-logs-myapplication-prod-000049）。
• size：指定每个请求中要检索的最大日志数。
• scroll：定义搜索上下文在 OpenSearch 服务器上保持打开状态的时间。 在本例中，它设置为 “5m”，这意味着每个请求都必须在五分钟内得到答复并询问新的 “page”。
• docinfo 和 docinfo_target：这些设置控制文档元数据是否应包含在 Logstash 输出中以及应存储在何处。 在这种情况下，文档元数据存储在 [@metadata][doc] 字段中 - 这很重要，因为文档的 _id 也将用作目标 id。

如果你要从不同基础设施（单独的云提供商）中的集群迁移，则强烈建议使用 ssl 和 ca_file。 如果你的 TLS 证书由公共机构签名，则无需指定 ca_file（如果你使用 SaaS 并且可通过 Internet 访问你的端点，则可能会出现这种情况）。 在这种情况下，只需 ssl => true 就足够了。 在我们的例子中，我们所有的 TLS 证书都是自签名的，因此我们还必须提供证书颁发机构 (CA) 证书。

（可选）filter

如果我们愿意，我们可以使用它来删除或更改要写入 Elasticsearch 的文档，但我们不会这样做，因为我们希望按原样迁移文档。 我们仅删除 Logstash 在所有文档中包含的额外元数据字段，例如 “@version” 和 “host”。 我们还删除了原始的 “data_stream”，因为它包含源数据流名称，该名称在目标中可能不同。

filter {
    mutate {
        remove_field => ["@version", "host", "data_stream"]
    }
}

output

输出非常简单 —— 我们将数据流命名为 logs-myapplication-reindex，并在 document_id 中使用原始文档的文档 ID，以确保没有重复的文档。 在 Elasticsearch 中，数据流名称遵循约定 <type>-<dataset>-<namespace>，因此我们的 logs-myapplication-reindex 数据流将 “myapplication” 作为数据集，将 “prod” 作为命名空间。

elasticsearch {
    hosts => "${ELASTICSEARCH_HOST}"

    user => "${ELASTICSEARCH_USERNAME}"
    password => "${ELASTICSEARCH_PASSWORD}"

    document_id => "%{[@metadata][doc][_id]}"

    data_stream => "true"
    data_stream_type => "logs"
    data_stream_dataset => "myapplication"
    data_stream_namespace => "prod"
}

部署 Logstash

我们有几种部署 Logstash 的选项：它可以从命令行本地部署、作为 systemd 服务、通过 docker 或在 Kubernetes 上部署。

由于我们的两个集群都部署在 Kubernetes 环境中，因此我们将引用之前创建的 Docker 映像将 Logstash 部署为 Pod。 让我们将管道以及一些配置文件（pipelines.yml 和 config.yml）放入 ConfigMap 中。

在下面的配置中，我们将 SOURCE_INDEX_NAME、SOURCE_SLICES、SOURCE_PAGE_SIZE、LOGSTASH_WORKERS 和 LOGSTASH_BATCH_SIZE 方便地公开为环境变量，因此你只需填写它们即可。

apiVersion: v1
kind: Pod
metadata:
  name: logstash-1
spec:
  containers:
  - name: logstash
    image: ugosan/logstash-opensearch-input:8.10.0
    imagePullPolicy: Always
    env: 
      - name: SOURCE_INDEX_NAME
        value: ".ds-logs-benchmark-dev-000037"
      - name: SOURCE_SLICES
        value: "10"
      - name: SOURCE_PAGE_SIZE
        value: "500"
      - name: LOGSTASH_WORKERS
        value: "4"
      - name: LOGSTASH_BATCH_SIZE
        value: "1000"
      - name: OPENSEARCH_USERNAME
        valueFrom:
          secretKeyRef:
            name: os-cluster-admin-password
            key: username
      - name: OPENSEARCH_PASSWORD
        valueFrom:
          secretKeyRef:
            name: os-cluster-admin-password
            key: password
      - name: ELASTICSEARCH_USERNAME
        value: "elastic"
      - name: ELASTICSEARCH_PASSWORD
        valueFrom:
          secretKeyRef:
            name: es-cluster-es-elastic-user
            key: elastic
    resources:
      limits:
        memory: "4Gi"
        cpu: "2500m"
      requests: 
        memory: "1Gi"
        cpu: "300m"
    volumeMounts:
      - name: config-volume
        mountPath: /usr/share/logstash/config
      - name: etc
        mountPath: /etc/logstash
        readOnly: true
  volumes:  
  - name: config-volume
    projected:
      sources:
      - configMap:
          name: logstash-configmap
          items:
            - key: pipelines.yml
              path: pipelines.yml
            - key: logstash.yml
              path: logstash.yml
  - name: etc
    projected:
      sources:
      - configMap:
          name: logstash-configmap
          items:
            - key: pipeline.conf
              path: pipelines/pipeline.conf
      - secret:
          name: os-cluster-http-cert
          items:
            - key: ca.crt
              path: certificates/opensearch-ca.crt
      - secret:
          name: es-cluster-es-http-ca-internal 
          items:
            - key: tls.crt
              path: certificates/elasticsearch-ca.crt
---
apiVersion: v1
kind: ConfigMap
metadata:
  name: logstash-configmap
data:
  pipelines.yml: |
    - pipeline.id: reindex-os-es
      path.config: "/etc/logstash/pipelines/pipeline.conf"
      pipeline.batch.size: ${LOGSTASH_BATCH_SIZE}
      pipeline.workers: ${LOGSTASH_WORKERS}
  logstash.yml: |
    log.level: info
    pipeline.unsafe_shutdown: true
    pipeline.ordered: false
  pipeline.conf: |
    input {  
        opensearch {
          hosts => ["os-cluster:9200"]
          ssl => true
          ca_file => "/etc/logstash/certificates/opensearch-ca.crt"
          user => "${OPENSEARCH_USERNAME}"
          password => "${OPENSEARCH_PASSWORD}"
          index => "${SOURCE_INDEX_NAME}"
          slices => "${SOURCE_SLICES}"
          size => "${SOURCE_PAGE_SIZE}"
          scroll => "5m"
          docinfo => true
          docinfo_target => "[@metadata][doc]"
        }
    }

    filter {
        mutate {
            remove_field => ["@version", "host", "data_stream"]
        }
    }

    output {
        elasticsearch {
            hosts => "https://es-cluster-es-http:9200"
            ssl => true
            ssl_certificate_authorities => ["/etc/logstash/certificates/elasticsearch-ca.crt"]
            ssl_verification_mode => "full"

            user => "${ELASTICSEARCH_USERNAME}"
            password => "${ELASTICSEARCH_PASSWORD}"

            document_id => "%{[@metadata][doc][_id]}"

            data_stream => "true"
            data_stream_type => "logs"
            data_stream_dataset => "myapplication"
            data_stream_namespace => "reindex"
        }
    }

就是这样。

几个小时后，我们成功将 10 亿个文档从 OpenSearch 迁移到 Elasticsearch，甚至还节省了 23% 的磁盘存储空间！ 既然我们在 Elasticsearch 中拥有了日志，如何从中提取当前的业务价值呢？ 日志包含如此多有价值的信息 - 我们不仅可以使用 AIOPS 做各种有趣的事情，例如自动分类这些日志，还可以提取业务指标并检测其中的异常情况，尝试一下。

有兴趣尝试 Elasticsearch 吗？ 开始我们的 14 天免费试用。

本文中描述的任何特性或功能的发布和时间安排均由 Elastic 自行决定。 当前不可用的任何特性或功能可能无法按时交付或根本无法交付。