Help us learn about your current experience with the documentation. Take the survey.

高级搜索迁移风格指南

创建新的高级搜索迁移

此功能仅支持在 GitLab 13.0 及更高版本中创建的索引。

使用脚本

执行 scripts/elastic-migration 并按照提示创建：

定义迁移的迁移文件：ee/elastic/migrate/YYYYMMDDHHMMSS_migration_name.rb
测试迁移的规范文件：ee/spec/elastic/migrate/YYYYMMDDHHMMSS_migration_name_spec.rb
标识迁移的字典文件：ee/elastic/docs/YYYYMMDDHHMMSS_migration_name.yml

手动创建

在 ee/elastic/migrate/ 文件夹中，创建一个新文件，文件名格式为 YYYYMMDDHHMMSS_migration_name.rb。此格式与 Rails 数据库迁移相同。

# frozen_string_literal: true

class MigrationName < Elastic::Migration
  # 重要：任何对 Elasticsearch 索引映射的更新都必须在相应的配置文件中复制：
  #   - `Elastic::Latest::Config`，用于主索引。
  #   - `Elastic::Latest::<Type>Config`，用于独立索引。

  def migrate
  end

  # 检查迁移是否完成
  # 如果完成返回 true，否则返回 false
  def completed?
  end
end

已应用的迁移存储在 gitlab-#{RAILS_ENV}-migrations 索引中。所有未执行的迁移都会由 Elastic::MigrationWorker 定时任务按顺序应用。

要更新 Elasticsearch 索引映射，请将配置应用到相应文件：

对于主索引：Elastic::Latest::Config。
对于独立索引：Elastic::Latest::<Type>Config。

迁移可以设置重试限制，并具有失败并标记为暂停的能力。支持迁移重试所需的所有数据或索引清理都应在迁移中处理。

跳过的迁移

您可以通过添加一个返回 true 或 false 的 skip_if proc 来跳过迁移：

class MigrationName < Elastic::Migration
  skip_if ->() { true|false }

只有当条件为 false 时才会执行迁移。跳过的迁移不会显示为待处理迁移的一部分。

跳过的迁移可以标记为已废弃，但必须保留 skip_if 条件，以便这些迁移始终被跳过。一旦跳过的迁移被标记为已废弃，唯一应用更改的方法是从头重新创建索引。

更新跳过的迁移的文档文件，添加以下属性：

skippable: true
skip_condition: '<描述>'

索引设置和映射变更的迁移

索引设置和映射的变更不会立即应用于现有索引，而是应用于新创建的索引。

要应用设置更改，例如添加分析器，可以：

使用零停机重新索引迁移
在功能问题中添加发布说明，提醒用户通过零停机重新索引或重新创建索引来应用更改

要应用映射更改，可以：

使用零停机重新索引迁移
使用更新映射迁移来更改现有索引的映射，并可选择后续的回填迁移来确保索引中的所有文档都填充了该字段。

零停机重新索引迁移

为目标索引创建一个新索引，并复制现有文档。

class MigrationName < Elastic::Migration
  def migrate
    Elastic::ReindexingTask.create!(targets: %w[Issue], options: { skip_pending_migrations_check: true })
  end

  def completed?
    true
  end
end

规范支持助手

以下辅助方法在 ee/spec/support/helpers/elasticsearch_helpers.rb 的 ElasticsearchHelpers 中可用。当使用任何 Elasticsearch 规范元数据时，ElasticsearchHelpers 会自动包含。

`assert_names_in_query`

验证命名查询是否存在（with）或不存在（without）于 Elasticsearch 查询中。

`assert_fields_in_query`

验证 Elasticsearch 查询是否包含指定的字段。

`assert_named_queries`

此方法需要向 Elasticsearch 发送搜索请求。使用 assert_names_in_query 来测试直接生成的查询。

验证是否向 Elasticsearch 发送了包含命名查询的请求。使用 without 来验证命名查询不在请求中。

`assert_routing_field`

验证是否向 Elasticsearch 发送了请求使用特定路由。

`ensure_elasticsearch_index!`

为所有 Elastic::ProcessBookkeepingService 类运行 execute 并调用 refresh_index!。此方法索引任何已使用 track! 方法排队等待索引的记录。

`refresh_index!`

对所有索引（包括迁移索引）执行 Elasticsearch 索引刷新。这使得在索引上执行的最近操作可用于搜索。

`set_elasticsearch_migration_to`

将迁移索引中的当前迁移设置为特定迁移（按名称或版本）。迁移默认标记为已完成，可以通过发送 including: false 设置为待处理。

`es_helper`

提供 Gitlab::Elastic::Helper.default 的实例

`warm_elasticsearch_migrations_cache!`

通过为每个迁移调用 migration_has_finished? 来预加载 ::Elastic::DataMigrationService 迁移缓存。

`elastic_wiki_indexer_worker_random_delay_range`

返回 0 到 ElasticWikiIndexerWorker::MAX_JOBS_PER_HOUR 之间的随机延迟

`elastic_delete_group_wiki_worker_random_delay_range`

返回 0 到 Search::Wiki::ElasticDeleteGroupWikiWorker::MAX_JOBS_PER_HOUR 之间的随机延迟

`elastic_group_association_deletion_worker_random_delay_range`

返回 0 到 Search::ElasticGroupAssociationDeletionWorker::MAX_JOBS_PER_HOUR 之间的随机延迟

`items_in_index`

返回存在于提供的索引名称中的 id 数组。

迁移助手

以下迁移助手在 ee/app/workers/concerns/elastic/ 中可用：

`Search::Elastic::MigrationBackfillHelper`

回填索引中的特定字段。

要求：

字段的映射应该已经添加
该字段必须始终有值。如果字段可以为 null，请使用 Search::Elastic::MigrationReindexBasedOnSchemaVersion
对于单个字段，定义 field_name 方法和 DOCUMENT_TYPE 常量
对于多个字段，定义 field_names 方法和 DOCUMENT_TYPE 常量

单个字段示例：

class MigrationName < Elastic::Migration
  include ::Search::Elastic::MigrationBackfillHelper

  DOCUMENT_TYPE = Issue

  private

  def field_name
    :schema_version
  end
end

多个字段示例：

class MigrationName < Elastic::Migration
  include ::Search::Elastic::MigrationBackfillHelper

  DOCUMENT_TYPE = Issue

  private

  def field_names
    %w[schema_version visibility_level]
  end
end

您可以使用 'migration backfills fields' 共享示例来测试此迁移。

describe 'migration', :elastic_delete_by_query, :sidekiq_inline do
  include_examples 'migration backfills fields' do
    let(:expected_throttle_delay) { 1.minute }
    let(:expected_batch_size) { 9000 }
    let(:objects) { create_list(:issues, 3) }
    let(:expected_fields) { schema_version: '25_09' }
  end
end

`Search::Elastic::MigrationUpdateMappingsHelper`

通过使用指定的映射调用 put_mapping 来更新索引中的映射。

需要 new_mappings 方法和 DOCUMENT_TYPE 常量。

class MigrationName < Elastic::Migration
  include ::Search::Elastic::MigrationUpdateMappingsHelper

  DOCUMENT_TYPE = Issue

  private

  def new_mappings
    {
      schema_version: {
        type: 'short'
      }
    }
  end
end

您可以使用 'migration adds mapping' 共享示例来测试此迁移。

describe 'migration', :elastic, :sidekiq_inline do
  include_examples 'migration adds mapping'
end

`Search::Elastic::MigrationRemoveFieldsHelper`

从索引中删除指定的字段。

分批检查是否有与 DOCUMENT_TYPE 匹配的文档在 Elasticsearch 中具有指定的字段。如果存在文档，则使用 Painless 脚本执行 update_by_query。

对于单个字段，定义 field_to_remove 方法和 DOCUMENT_TYPE 常量
对于多个字段，定义 fields_to_remove 方法和 DOCUMENT_TYPE 常量

class MigrationName < Elastic::Migration
  include ::Search::Elastic::MigrationRemoveFieldsHelper

  batched!
  throttle_delay 1.minute

  DOCUMENT_TYPE = User

  private

  def fields_to_remove
    %w[two_factor_enabled has_projects]
  end
end

默认批处理大小为 10_000。您可以通过指定 BATCH_SIZE 来覆盖此值：

class MigrationName < Elastic::Migration
  include ::Search::Elastic::MigrationRemoveFieldsHelper

  batched!
  BATCH_SIZE = 100

  ...
end

您可以使用 'migration removes field' 共享示例来测试此迁移。

include_examples 'migration removes field' do
  let(:expected_throttle_delay) { 1.minute }
  let(:objects) { create_list(:work_item, 6) }
  let(:index_name) { ::Search::Elastic::Types::WorkItem.index_name }
  let(:field) { :correct_work_item_type_id }
  let(:type) { 'long' }
end

如果映射包含多个 type，则省略 type 变量并定义 mapping：

include_examples 'migration removes field' do
  let(:expected_throttle_delay) { 1.minute }
  let(:objects) { create_list(:work_item, 6) }
  let(:index_name) { ::Search::Elastic::Types::WorkItem.index_name }
  let(:field) { :embedding_0 }
  let(:mapping) { { type: 'dense_vector', dims: 768, index: true, similarity: 'cosine' } }
end

如果您收到 expecting token of type [VALUE_NUMBER] but found [FIELD_NAME] 错误，请定义 value 变量：

include_examples 'migration removes field' do
  let(:expected_throttle_delay) { 1.minute }
  let(:objects) { create_list(:work_item, 6) }
  let(:index_name) { ::Search::Elastic::Types::WorkItem.index_name }
  let(:field) { :embedding_0 }
  let(:mapping) { { type: 'dense_vector', dims: 768, index: true, similarity: 'cosine' } }
  let(:value) { Array.new(768, 1) }
end

`Search::Elastic::MigrationObsolete`

当迁移不再需要时，将其标记为已废弃。

class MigrationName < Elastic::Migration
  include ::Search::Elastic::MigrationObsolete
end

将可跳过的迁移标记为已废弃时，必须保留 skip_if 条件。

您可以使用 'a deprecated Advanced Search migration' 共享示例来测试此迁移。遵循将迁移标记为已废弃的过程。

`Search::Elastic::MigrationCreateIndexHelper`

创建一个新索引。

需要：

target_class 和 document_type 方法
该类的映射和索引设置

您必须在同一里程碑中执行后续迁移来填充索引。

class MigrationName < Elastic::Migration
  include ::Search::Elastic::MigrationCreateIndexHelper

  retry_on_failure

  def document_type
    :epic
  end

  def target_class
    Epic
  end
end

您可以使用 'migration creates a new index' 共享示例来测试此迁移。

it_behaves_like 'migration creates a new index', 20240501134252, WorkItem

`Search::Elastic::MigrationReindexTaskHelper`

创建一个重新索引任务，该任务创建一个新索引并将数据复制到新索引中。

需要：

targets 方法。

class MigrationName < Elastic::Migration
  include ::Search::Elastic::MigrationReindexTaskHelper

  def targets
    %w[MergeRequest]
  end
end

您可以使用以下规范来测试此迁移。

let(:migration) { described_class.new(version) }
let(:task) { Search::Elastic::ReindexingTask.last }
let(:targets) { %w[MergeRequest] }

it 'does not have migration options set', :aggregate_failures do
  expect(migration).not_to be_batched
  expect(migration).not_to be_retry_on_failure
end

describe '#migrate', :aggregate_failures do
  it 'creates reindexing task with correct target and options' do
    expect { migration.migrate }.to change { Search::Elastic::ReindexingTask.count }.by(1)
    expect(task.targets).to eq(targets)
    expect(task.options).to eq('skip_pending_migrations_check' => true)
  end
end

describe '#completed?' do
  it 'always returns true' do
    expect(migration.completed?).to be(true)
  end
end

`Search::Elastic::MigrationReindexBasedOnSchemaVersion`

重新索引索引中存储的指定文档类型的所有文档，并更新 schema_version。

需要 DOCUMENT_TYPE 和 NEW_SCHEMA_VERSION 常量。索引映射必须有一个 YYWW（年/周）格式的 schema_version 整数字段。

之前索引映射 schema_version 使用 YYMM 格式。新版本应使用 YYWW 格式。

class MigrationName < Elastic::Migration
  include Search::Elastic::MigrationReindexBasedOnSchemaVersion

  batched!
  batch_size 9_000
  throttle_delay 1.minute

  DOCUMENT_TYPE = WorkItem
  NEW_SCHEMA_VERSION = 24_46
  UPDATE_BATCH_SIZE = 100
end

您可以使用 'migration reindex based on schema_version' 共享示例来测试此迁移。

include_examples 'migration reindex based on schema_version' do
  let(:expected_throttle_delay) { 1.minute }
  let(:expected_batch_size) { 9_000 }
  let(:objects) { create_list(:project, 3) }
end

`Search::Elastic::MigrationDeleteBasedOnSchemaVersion`

删除索引中存储的指定文档类型且 schema_version 小于给定值的所有文档。

需要 DOCUMENT_TYPE 常量和 schema_version 方法。索引映射必须有一个 YYWW（年/周）格式的 schema_version 整数字段。

之前索引映射 schema_version 使用 YYMM 格式。新版本应使用 YYWW 格式。

class MigrationName < Elastic::Migration
  include ::Search::Elastic::MigrationDeleteBasedOnSchemaVersion

  DOCUMENT_TYPE = Issue

  batch_size 10_000
  batched!
  throttle_delay 1.minute
  retry_on_failure

  def schema_version
    23_12
  end
end

您可以使用 'migration deletes documents based on schema version' 共享示例来测试此迁移。

include_examples 'migration deletes documents based on schema version' do
  let(:objects) { create_list(:issue, 3) }
  let(:expected_throttle_delay) { 1.minute }
  let(:expected_batch_size) { 20000 }
end

`Search::Elastic::MigrationDatabaseBackfillHelper`

将数据库中的所有文档重新索引到 elasticsearch 索引中，同时尊重 limited_indexing 设置。

需要 DOCUMENT_TYPE 常量和 respect_limited_indexing? 方法。

class MigrationName < Elastic::Migration
  include ::Search::Elastic::MigrationDatabaseBackfillHelper

  batch_size 10_000
  batched!
  throttle_delay 1.minute
  retry_on_failure

  DOCUMENT_TYPE = Issue

  def respect_limited_indexing?
    true
  end
end

`Search::Elastic::MigrationHelper`

包含当迁移不适合前面示例时可以使用的方法。

class MigrationName < Elastic::Migration
  include ::Search::Elastic::MigrationHelper

  def migrate
  ...
  end

  def completed?
  ...
  end
end

`Elastic::MigrationWorker` 支持的迁移选项

Elastic::MigrationWorker 支持以下迁移选项：

batched! - 允许迁移以批处理方式运行。如果设置，Elastic::MigrationWorker 会使用下面描述的 throttle_delay 选项设置的延迟重新排队自己。批处理必须在 migrate 方法中处理。此设置仅控制重新排队。
batch_size - 设置在 batched! 迁移运行期间修改的文档数量。此大小应设置为允许更新有足够时间完成的值。可以与下面描述的 throttle_delay 选项一起调整。批处理必须在自定义 migrate 方法或使用 Search::Elastic::MigrationBackfillHelper migrate 方法中处理，该方法使用此设置。默认值为 1000 个文档。
throttle_delay - 设置批处理运行之间的等待时间。此时间应设置得足够高，以允许每个迁移批处理有足够时间完成。此外，时间应小于 5 分钟，因为这是 Elastic::MigrationWorker 定时任务运行的频率。默认值为 3 分钟。
pause_indexing! - 在迁移运行时暂停索引。此设置记录迁移运行前的索引设置，并在迁移完成后将其设置回该值。
space_requirements! - 在迁移运行时验证集群中是否有足够的可用空间。此设置如果迁移运行时没有足够的存储空间，则会停止迁移。迁移必须通过定义 space_required_bytes 方法来提供所需的存储空间（以字节为单位）。
retry_on_failure - 启用失败重试功能。默认情况下，它会重试迁移 30 次。用完重试次数后，迁移被标记为暂停。要自定义重试次数，请传递 max_attempts 参数： retry_on_failure max_attempts: 10

# frozen_string_literal: true

class BatchedMigrationName < Elastic::Migration
  # 声明迁移应以批处理方式运行
  batched!
  throttle_delay 10.minutes
  pause_indexing!
  space_requirements!
  retry_on_failure

  # ...
end

避免迁移中的停机

回滚迁移

如果迁移在 GitLab.com 上失败或暂停，我们倾向于回滚引入迁移的更改。这可以防止自管理客户收到损坏的迁移，并减少回溯的需要。

多版本兼容性

高级搜索迁移，就像其他 GitLab 更改一样，需要支持多个版本的应用程序同时运行的情况。

根据部署顺序，迁移可能已经开始或完成，但仍有一台服务器运行迁移前的应用程序代码。我们需要考虑到这一点，直到我们能够确保所有高级搜索迁移在部署完成后开始。

高风险迁移

由于 Elasticsearch 不支持事务，我们总是需要设计我们的迁移以适应应用程序代码在迁移开始后或完成后被回滚的情况。

因此，我们通常将破坏性操作（例如，在移动某些数据后的删除）推迟到迁移成功完成后的后续合并请求中。为了安全起见，对于自管理客户，如果存在重要数据丢失的风险，我们也应该将其推迟到另一个版本中。

计算迁移运行时间

了解迁移在 GitLab.com 上可能需要多长时间运行很重要。推导出迁移将处理的文档数量。这个数字可能来自查询数据库或现有的 Elasticsearch 索引。使用以下公式计算运行时间：

> batch_size = 9_000
=> 9000
> throttle_delay = 1.minute
=> 1 minute
> number_of_documents = 15_536_906
=> 15536906
> (number_of_documents / batch_size) * throttle_delay
=> 1726 minutes
> (number_of_documents / batch_size) * throttle_delay / 1.hour
=> 28

高级搜索迁移的最佳实践

遵循这些最佳实践以获得最佳结果：

为每种文档类型排序所有迁移，以便任何使用 Search::Elastic::MigrationUpdateMappingsHelper 的迁移在使用 Search::Elastic::MigrationBackfillHelper 的迁移之前执行。这避免了如果所有迁移都被撤销，重新索引相同文档多次的情况，并减少了回填时间。
在批处理工作时，保持批处理大小在 9,000 个文档以下。批量索引器设置为每分钟运行一次，处理 10,000 个文档的批处理。这样，批量索引器就有时间在尝试另一个迁移批处理之前处理记录。
为了确保文档计数是最新的，您应该在检查迁移是否完成之前刷新索引。
在迁移开始时、完成检查发生时以及迁移完成时，为每个迁移添加日志语句。这些日志在调试迁移问题时很有帮助。
如果您使用任何 Elasticsearch Reindex API 操作，请暂停索引。
如果迁移有可能失败，请考虑添加重试限制。这确保了如果出现问题，迁移可以被暂停。

清理高级搜索迁移

由于高级搜索迁移通常需要我们在很长一段时间内支持多个代码路径，因此在我们安全地清理它们时很重要。

我们选择使用 GitLab 必需停止点作为安全时间来删除尚未完全迁移的索引的向后兼容性。我们在升级文档中记录了这一点。

GitLab Housekeeper 用于自动化清理过程。此过程包括将现有迁移标记为已废弃并删除已废弃的迁移。当迁移被标记为已废弃时，迁移代码被替换为已废弃的迁移代码，测试被替换为已废弃的迁移共享示例，以便：

我们不需要维护任何从我们的高级搜索迁移中调用的代码。
我们不会浪费 CI 时间运行我们不再支持的迁移的测试。
尚未运行此迁移并直接升级到目标版本的操作员会看到一条消息，提示他们从头重新索引。

为了更加安全，我们不清理在最后一个必需停止点之前的最后一个次要版本中创建的迁移。例如，如果最后一个必需停止点是 %14.0，我们不应该清理仅在 %13.12 中添加的迁移。这个额外的安全网允许在 GitLab.com 上可能需要数周才能完成的迁移。由于我们对 GitLab.com 的部署是自动化的，并且我们没有自动检查来防止此清理，因此采取额外的预防措施是合理的。此外，即使我们有自动检查来防止它，我们实际上也不希望因为高级搜索迁移而延迟 GitLab.com 的部署，因为它们可能还需要一周才能完成，这太长了，无法阻止部署。

将迁移标记为已废弃的过程

手动运行 Keeps::MarkOldAdvancedSearchMigrationsAsObsolete Keep 将迁移标记为已废弃。

对于在最后一个必需停止点之前两个版本中创建的每个迁移，该 Keep：

保留迁移的内容并在底部添加一个 prepend：

 ClassName.prepend ::Search::Elastic::MigrationObsolete

将规范文件内容替换为 'a deprecated Advanced Search migration' 共享示例。
随机选择一名 Global Search 后端工程师作为分配人。
更新字典文件以将迁移标记为已废弃。

MR 分配人必须：

确保字典文件具有正确的 marked_obsolete_by_url 和 marked_obsolete_in_milestone。
验证 .rubocop_todo/ 目录中不存在对迁移或规范文件的任何引用。
通过查找 Elastic::DataMigrationService.migration_has_finished?(:migration_name_in_lowercase) 来删除处理此迁移向后兼容性的任何逻辑。
将任何必需的更改推送到合并请求。

删除已废弃迁移的过程

手动运行 Keeps::DeleteObsoleteAdvancedSearchMigrations Keep 删除已废弃的迁移和规范。该 Keep 删除除最近的已废弃迁移之外的所有内容。

选择在最后一个必需停止点之前标记为已废弃的迁移。
如果第一步包括所有已废弃的迁移，保留一个已废弃的迁移作为具有未应用迁移的客户的保障。
删除这些迁移的迁移文件和规范文件。
创建一个合并请求并将其分配给 Global Search 团队成员。

MR 分配人必须：

将默认分支中的迁移备份到迁移墓地
验证 .rubocop_todo/ 目录中不存在对迁移或规范文件的任何引用。
将任何必需的更改推送到合并请求。

监控迁移的 ChatOps 命令

您随时可以从 Slack（或任何支持 ChatOps 的频道）检查迁移状态：

/chatops run search_migrations --help
/chatops run search_migrations list
/chatops run search_migrations get MigrationName
/chatops run search_migrations get VersionNumber

上述命令使用 search_migrations ChatOps 插件来获取当前迁移状态。

高级搜索迁移风格指南

创建新的高级搜索迁移

使用脚本

手动创建

跳过的迁移

索引设置和映射变更的迁移

零停机重新索引迁移

规范支持助手

assert_names_in_query

assert_fields_in_query

assert_named_queries

assert_routing_field

ensure_elasticsearch_index!

refresh_index!

set_elasticsearch_migration_to

es_helper

warm_elasticsearch_migrations_cache!

elastic_wiki_indexer_worker_random_delay_range

elastic_delete_group_wiki_worker_random_delay_range

elastic_group_association_deletion_worker_random_delay_range

items_in_index