# 機能設計書 59-セグメントレプリケーション

## 概要

本ドキュメントは、OpenSearchにおけるセグメントレプリケーション機能の設計を記述する。セグメントレプリケーションは、プライマリシャードからレプリカシャードへLuceneセグメントを直接複製する機能である。

### 本機能の処理概要

セグメントレプリケーション機能は、従来のドキュメントレプリケーション（各レプリカでインデックス操作を再実行）に代わり、プライマリシャードで作成されたLuceneセグメントファイルをレプリカに直接コピーする。これによりインデックス性能が向上し、レプリカの負荷が軽減される。

**業務上の目的・背景**：大量のインデックス操作を行うワークロードにおいて、各レプリカでインデックス操作を再実行するオーバーヘッドを削減する。特に書き込み負荷の高いユースケースで、インデックススループットの向上とレプリカノードのCPU使用率削減を実現する。

**機能の利用シーン**：
- 書き込み負荷の高いログ収集・分析システム
- リアルタイムデータ取り込みパイプライン
- 大量バルクインデックス処理
- レプリカノードのリソース節約が必要な環境

**主要な処理内容**：
1. プライマリシャードでのチェックポイント発行
2. レプリカへのチェックポイント通知
3. セグメントファイルの取得リクエスト
4. セグメントファイルのコピー・適用
5. 可視チェックポイントの更新

**関連システム・外部連携**：RemoteStoreと連携してリモートストレージ経由でのセグメント複製をサポート。RecoverySettingsと連携してレート制限・タイムアウトを制御。

**権限による制御**：内部通信のため、直接的なアクション権限は不要。インデックス設定の変更には適切な権限が必要。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| 1 | インデックス作成 | 設定画面 | セグメントレプリケーションの有効化 |
| - | Cat Segment Replication API | 参照画面 | レプリケーション状態の確認 |

## 機能種別

データレプリケーション / インデックス管理

## 入力仕様

### インデックス設定

| パラメータ名 | 型 | 必須 | 説明 | デフォルト値 |
|-------------|-----|-----|------|-------------|
| index.replication.type | String | No | レプリケーションタイプ | DOCUMENT |

### 設定値

| 値 | 説明 |
|----|------|
| DOCUMENT | 従来のドキュメントレプリケーション |
| SEGMENT | セグメントレプリケーション |

### クラスタ設定

| パラメータ名 | 型 | 必須 | 説明 | デフォルト値 |
|-------------|-----|-----|------|-------------|
| cluster.indices.replication.strategy | String | No | クラスタデフォルトのレプリケーション戦略 | DOCUMENT |

### REST API（統計取得）

- エンドポイント: `GET /_cat/segment_replication`
- エンドポイント: `GET /_cat/segment_replication/{index}`

### 入力データソース

- ReplicationCheckpoint（プライマリからの通知）
- SegmentInfos（Luceneセグメント情報）
- StoreFileMetadata（ファイルメタデータ）

## 出力仕様

### セグメントレプリケーション統計

| 項目名 | 型 | 説明 |
|--------|-----|------|
| shardId | ShardId | シャードID |
| target_node | String | レプリカノード名 |
| target_host | String | レプリカホスト |
| checkpoints_behind | long | 遅延チェックポイント数 |
| bytes_behind | long | 遅延バイト数 |
| current_lag | TimeValue | 現在のラグ時間 |
| last_completed_lag | TimeValue | 最後に完了したレプリケーションのラグ |
| rejected_requests | long | 拒否されたリクエスト数 |

### SegmentReplicationState

| 項目名 | 型 | 説明 |
|--------|-----|------|
| stage | Stage | 現在のステージ |
| replicationId | long | レプリケーションID |
| sourceDescription | String | ソースの説明 |
| timingData | Map | タイミング情報 |

### Stage列挙型

| 値 | 説明 |
|----|------|
| INIT | 初期化 |
| REPLICATING | レプリケーション中 |
| GET_CHECKPOINT_INFO | チェックポイント情報取得 |
| FILE_DIFF | ファイル差分計算 |
| GET_FILES | ファイル取得 |
| FINALIZE_REPLICATION | 最終化 |
| DONE | 完了 |

### 出力先

- REST APIレスポンス（JSON形式）
- レプリカシャードのLuceneインデックス

## 処理フロー

### 処理シーケンス

```
1. プライマリでのリフレッシュ/フラッシュ
   └─ 新しいチェックポイント生成
2. チェックポイント発行
   └─ SegmentReplicationCheckpointPublisher.publish()
3. レプリカでのチェックポイント受信
   └─ SegmentReplicationTargetService.onNewCheckpoint()
4. レプリケーション開始判定
   └─ shouldProcessCheckpoint()
5. チェックポイント情報取得
   └─ GET_CHECKPOINT_INFO → CheckpointInfoResponse
6. ファイル差分計算
   └─ 必要なセグメントファイルを特定
7. セグメントファイル取得
   └─ GET_SEGMENT_FILES → FileChunk転送
8. セグメント適用
   └─ Luceneインデックスに適用
9. 可視チェックポイント更新
   └─ UPDATE_VISIBLE_CHECKPOINT
```

### フローチャート

```mermaid
flowchart TD
    A[プライマリでリフレッシュ] --> B[チェックポイント生成]
    B --> C[レプリカへチェックポイント通知]
    C --> D{レプリカの状態?}

    D -->|CLOSED| E[無視]
    D -->|STARTED| F{処理すべきチェックポイント?}
    D -->|その他| G[保存して後で処理]

    F -->|No| H[チェックポイント保存]
    F -->|Yes| I{進行中のレプリケーション?}

    I -->|Yes, 古いPrimaryTerm| J[進行中をキャンセル]
    I -->|Yes, 同じ/新しいPrimaryTerm| K[無視]
    I -->|No| L[レプリケーション開始]
    J --> L

    L --> M[チェックポイント情報取得]
    M --> N[ファイル差分計算]
    N --> O[セグメントファイル取得]
    O --> P[セグメント適用]
    P --> Q[可視チェックポイント更新]
    Q --> R{保存済みの新しいチェックポイント?}
    R -->|Yes| L
    R -->|No| S[完了]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-01 | プライマリ専用 | セグメントはプライマリシャードでのみ生成 | 常時 |
| BR-02 | チェックポイント比較 | 新しいチェックポイントのみ処理 | 常時 |
| BR-03 | PrimaryTerm優先 | PrimaryTermが古いレプリケーションはキャンセル | PrimaryTerm変更時 |
| BR-04 | シャード状態チェック | STARTEDステートのシャードのみ処理 | 常時 |
| BR-05 | 差分転送 | 既存のセグメントファイルはスキップ | 常時 |

### ReplicationCheckpoint構造

```
ReplicationCheckpoint {
  shardId: ShardId
  primaryTerm: long
  segmentsGen: long
  segmentInfosVersion: long
  length: long
  codec: String
  metadataMap: Map<String, StoreFileMetadata>
}
```

## データベース操作仕様

### 操作別データベース影響一覧

| 操作 | 対象 | 操作種別 | 概要 |
|-----|------|---------|------|
| セグメント適用 | Lucene Index | WRITE | セグメントファイルの書き込み |
| チェックポイント更新 | ReplicationTracker | UPDATE | 可視チェックポイントの更新 |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| - | ReplicationFailedException | セグメントコピー失敗 | 自動リトライ |
| - | CancellableThreads.ExecutionCancelledException | レプリケーションキャンセル | 新しいチェックポイントで再試行 |
| - | IndexShardClosedException | シャードクローズ | 処理中止 |

### リトライ仕様

レプリケーション失敗時は、保存済みの最新チェックポイントで自動再試行する。致命的なエラーの場合はシャード障害としてマークされる。

## トランザクション仕様

セグメントレプリケーションはアトミックではない。チェックポイント単位で一貫性を保証する。中断時は次回チェックポイントで再同期される。

## パフォーマンス要件

- セグメントコピーはRecoverySettings.replicationRateLimiterでレート制限
- 同時レプリケーション数はRecoverySettings.maxConcurrentFileChunksで制御
- マージセグメントの先行コピーによる遅延軽減

## セキュリティ考慮事項

- 内部通信のためノード間認証で保護
- セグメントファイルは暗号化されていない場合、転送中は平文

## 備考

- V2.3.0以降で利用可能
- RemoteStoreと組み合わせることで、リモートストレージ経由のレプリケーションが可能
- ドキュメントレプリケーションとの互換性なし（インデックス作成時に選択）

---

## コードリーディングガイド

本機能を理解するために参照すべきファイルと、推奨する読み解き順序を以下に示す。

### 推奨読解順序

#### Step 1: データ構造を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | ReplicationCheckpoint.java | `server/src/main/java/org/opensearch/indices/replication/checkpoint/ReplicationCheckpoint.java` | チェックポイントデータ構造 |
| 1-2 | SegmentReplicationState.java | `server/src/main/java/org/opensearch/indices/replication/SegmentReplicationState.java` | レプリケーション状態 |
| 1-3 | SegmentReplicationShardStats.java | `server/src/main/java/org/opensearch/index/SegmentReplicationShardStats.java` | シャードごとの統計 |

#### Step 2: レプリカ側（ターゲット）を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | SegmentReplicationTargetService.java | `server/src/main/java/org/opensearch/indices/replication/SegmentReplicationTargetService.java` | レプリカ側メインサービス |
| 2-2 | SegmentReplicationTarget.java | `server/src/main/java/org/opensearch/indices/replication/SegmentReplicationTarget.java` | レプリケーションターゲット |
| 2-3 | SegmentReplicator.java | `server/src/main/java/org/opensearch/indices/replication/SegmentReplicator.java` | レプリケーション実行 |

**SegmentReplicationTargetService主要処理**:
- **298-385行目**: onNewCheckpoint() -- 新しいチェックポイント受信処理
- **523-529行目**: startReplication() -- レプリケーション開始
- **411-476行目**: updateVisibleCheckpoint() -- 可視チェックポイント更新

#### Step 3: プライマリ側（ソース）を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | SegmentReplicationSourceService.java | `server/src/main/java/org/opensearch/indices/replication/SegmentReplicationSourceService.java` | プライマリ側メインサービス |
| 3-2 | SegmentReplicationSourceHandler.java | `server/src/main/java/org/opensearch/indices/replication/SegmentReplicationSourceHandler.java` | ソースハンドラ |
| 3-3 | OngoingSegmentReplications.java | `server/src/main/java/org/opensearch/indices/replication/OngoingSegmentReplications.java` | 進行中のレプリケーション管理 |

#### Step 4: チェックポイント発行を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | SegmentReplicationCheckpointPublisher.java | `server/src/main/java/org/opensearch/indices/replication/checkpoint/SegmentReplicationCheckpointPublisher.java` | チェックポイント発行 |
| 4-2 | PublishCheckpointAction.java | `server/src/main/java/org/opensearch/indices/replication/checkpoint/PublishCheckpointAction.java` | チェックポイント通知アクション |

### プログラム呼び出し階層図

```
Primary Shard (リフレッシュ/フラッシュ)
    |
    +-- SegmentReplicationCheckpointPublisher.publish()
          |
          +-- PublishCheckpointAction (レプリカへ通知)
                |
                +-- SegmentReplicationTargetService.onNewCheckpoint()
                      |
                      +-- shouldProcessCheckpoint()
                      |
                      +-- startReplication()
                            |
                            +-- GET_CHECKPOINT_INFO (プライマリへ)
                            |     +-- SegmentReplicationSourceService.CheckpointInfoRequestHandler
                            |     +-- CheckpointInfoResponse
                            |
                            +-- ファイル差分計算
                            |
                            +-- GET_SEGMENT_FILES (プライマリへ)
                            |     +-- SegmentReplicationSourceService.GetSegmentFilesRequestHandler
                            |     +-- FileChunk転送
                            |
                            +-- セグメント適用
                            |
                            +-- UPDATE_VISIBLE_CHECKPOINT (プライマリへ)
```

### データフロー図

```
[プライマリシャード]                    [レプリカシャード]

    リフレッシュ
        |
    チェックポイント生成
        |
        +-----> PublishCheckpointAction -----> onNewCheckpoint()
                                                    |
                                               shouldProcess?
                                                    |
        <----- GET_CHECKPOINT_INFO <---------------+
        |
    CheckpointInfoResponse
        |
        +------------------------------------> ファイル差分計算
                                                    |
        <----- GET_SEGMENT_FILES <-----------------+
        |
    FileChunk転送
        |
        +------------------------------------> セグメント適用
                                                    |
        <----- UPDATE_VISIBLE_CHECKPOINT <---------+
        |
    チェックポイント記録
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| SegmentReplicationTargetService.java | `server/src/main/java/org/opensearch/indices/replication/SegmentReplicationTargetService.java` | ソース | レプリカ側サービス |
| SegmentReplicationSourceService.java | `server/src/main/java/org/opensearch/indices/replication/SegmentReplicationSourceService.java` | ソース | プライマリ側サービス |
| SegmentReplicationTarget.java | `server/src/main/java/org/opensearch/indices/replication/SegmentReplicationTarget.java` | ソース | ターゲット実装 |
| SegmentReplicationSourceHandler.java | `server/src/main/java/org/opensearch/indices/replication/SegmentReplicationSourceHandler.java` | ソース | ソースハンドラ |
| SegmentReplicator.java | `server/src/main/java/org/opensearch/indices/replication/SegmentReplicator.java` | ソース | レプリケータ |
| ReplicationCheckpoint.java | `server/src/main/java/org/opensearch/indices/replication/checkpoint/ReplicationCheckpoint.java` | ソース | チェックポイント |
| SegmentReplicationCheckpointPublisher.java | `server/src/main/java/org/opensearch/indices/replication/checkpoint/SegmentReplicationCheckpointPublisher.java` | ソース | チェックポイント発行 |
| SegmentReplicationState.java | `server/src/main/java/org/opensearch/indices/replication/SegmentReplicationState.java` | ソース | 状態管理 |
| SegmentReplicationPressureService.java | `server/src/main/java/org/opensearch/index/SegmentReplicationPressureService.java` | ソース | バックプレッシャー制御 |
