# 機能設計書 41-ストリーミングインジェスト

## 概要

本ドキュメントは、OpenSearchのストリーミングインジェスト機能について、外部データソースからのプルベースのデータ取り込みに関する一時停止（Pause）、再開（Resume）、状態取得（Get State）、状態更新（Update State）の管理機能を設計書としてまとめたものである。

### 本機能の処理概要

**業務上の目的・背景**：OpenSearchにおいて、Apache KafkaやAmazon Kinesis、ファイルシステムなどの外部データソースからストリーミング形式でデータを取り込む際、取り込みの一時停止・再開・状態監視などの運用管理が必要となる。本機能は、ストリーミングインジェストのライフサイクル管理を提供し、運用時のデータパイプライン制御を可能にする。

**機能の利用シーン**：メンテナンス時やデータソース側の障害時にインジェストを一時停止する場面、障害復旧後にインジェストを再開する場面、各シャードのインジェスト状態をモニタリングする場面で利用される。

**主要な処理内容**：
1. インジェスト一時停止（Pause）：指定インデックスのストリーミングインジェストポーラーを一時停止する
2. インジェスト再開（Resume）：一時停止中のインジェストポーラーを再開する
3. インジェスト状態取得（Get State）：各シャードのポーラー状態、エラーポリシー、バッチ開始ポインタ等を取得する
4. インジェスト状態更新（Update State）：シャードのインジェスト状態を更新する

**関連システム・外部連携**：Apache Kafka、Amazon Kinesis、ファイルシステム等の外部データソースと連携する。各データソースプラグイン（ingestion-kafka、ingestion-kinesis、ingestion-fs）がIngestionConsumerPluginインターフェースを実装する。

**権限による制御**：クラスタマネージャノードへの操作が必要であり、METADATA_WRITEレベルのクラスタブロックチェックが行われる。DestructiveOperationsによる破壊的操作の制御も適用される。

## 関連画面

本機能は画面機能マッピングにおいて直接的な画面関連はない（API経由での操作が主体）。

## 機能種別

データ連携 / CRUD操作

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| indices | String[] | Yes | 対象インデックス名の配列 | 空でないこと |
| indicesOptions | IndicesOptions | No | インデックス解決オプション | strictExpandOpen（デフォルト） |
| shards | int[] | No（Get State） | 対象シャードID配列 | - |
| pageParams | PageParams | No（Get State） | ページネーションパラメータ | デフォルトページサイズ1000 |
| ingestionPaused | boolean | Yes（Update State） | 一時停止フラグ | - |

### 入力データソース

REST API経由でのリクエスト。クラスタマネージャノードに転送される。

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| acknowledged | boolean | 操作が承認されたか |
| shards_acknowledged | boolean | 全シャードで操作が成功したか |
| failures | IngestionStateShardFailure[] | シャード単位の失敗情報 |
| error_message | String | エラーメッセージ |
| shard_ingestion_states | ShardIngestionState[] | 各シャードのインジェスト状態（Get State時） |

### 出力先

REST APIレスポンス（JSON形式）

## 処理フロー

### 処理シーケンス

```
1. REST APIリクエスト受信
   └─ PauseIngestionRequest / ResumeIngestionRequest / GetIngestionStateRequest を生成
2. クラスタマネージャノードへ転送
   └─ TransportClusterManagerNodeAction を経由
3. インデックス名解決
   └─ IndexNameExpressionResolver で具体的なインデックスを解決
4. クラスタブロックチェック
   └─ METADATA_WRITE レベルのブロックを確認
5. MetadataStreamingIngestionStateService による状態更新
   └─ updateIngestionPollerState でポーラー状態を更新
6. レスポンス生成
   └─ 各シャードの結果を集約してレスポンスを返却
```

### フローチャート

```mermaid
flowchart TD
    A[REST API Request] --> B{Action Type}
    B -->|Pause| C[TransportPauseIngestionAction]
    B -->|Resume| D[TransportResumeIngestionAction]
    B -->|Get State| E[GetIngestionStateAction]
    C --> F[Resolve Indices]
    D --> F
    F --> G[Check Cluster Blocks]
    G --> H[Update Ingestion Poller State]
    H --> I{All Shards Succeeded?}
    I -->|Yes| J[Return acknowledged=true]
    I -->|No| K[Return with failures]
    E --> L[Broadcast to Shards]
    L --> M[Collect ShardIngestionState]
    M --> N[Return GetIngestionStateResponse]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-41-01 | インデックス必須 | 対象インデックスが空の場合はバリデーションエラー | Pause/Resume操作時 |
| BR-41-02 | 重複インデックス禁止 | 同一インデックス名の重複指定は禁止 | Get State操作時 |
| BR-41-03 | 破壊的操作チェック | DestructiveOperationsによる保護対象チェック | Pause操作時 |
| BR-41-04 | 実験的API | @ExperimentalApiアノテーション付きの実験的機能 | 全操作 |

### 計算ロジック

特になし。

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

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

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| Pause | ClusterState (Metadata) | UPDATE | インジェストポーラーの一時停止状態をクラスタ状態に反映 |
| Resume | ClusterState (Metadata) | UPDATE | インジェストポーラーの再開状態をクラスタ状態に反映 |
| Get State | ShardIngestionState | SELECT | 各シャードのインジェスト状態を取得 |

### テーブル別操作詳細

#### ClusterState (Metadata)

| 操作 | 項目 | 更新値・取得条件 | 備考 |
|-----|------|-----------------|------|
| UPDATE | ingestionPaused | true/false | ポーラーの一時停止状態 |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| - | ValidationException | インデックス名が空 | リクエストパラメータを確認 |
| - | ClusterBlockException | METADATA_WRITEブロックが設定されている | クラスタブロックを解除 |
| - | IngestionStateShardFailure | シャード単位での状態更新失敗 | 個別シャードの状態を確認 |

### リトライ仕様

クラスタマネージャノードタスクとして実行されるため、タスクスロットリングによるリトライが適用される。

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

クラスタ状態更新タスクとして原子的に実行される。全シャードの結果が集約されてレスポンスが返却される。

## パフォーマンス要件

クラスタマネージャノードへの転送・処理であり、タイムアウトはリクエストのclusterManagerNodeTimeoutに依存する。

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

- クラスタ管理権限が必要
- DestructiveOperationsによる保護が適用される
- @ExperimentalApiとして提供されており、APIの安定性は保証されない

## 備考

本機能は@ExperimentalApiアノテーションが付与された実験的機能であり、将来のバージョンで変更される可能性がある。バージョン3.3.0以降でShardIngestionStateにisPrimaryおよびnodeNameフィールドが追加されている。

---

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

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

### 推奨読解順序

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

まず、シャードインジェスト状態を表現するデータ構造を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | ShardIngestionState.java | `server/src/main/java/org/opensearch/action/admin/indices/streamingingestion/state/ShardIngestionState.java` | シャードのインジェスト状態を表現するデータ構造。pollerState, errorPolicy, isPollerPaused, isWriteBlockEnabled, batchStartPointer等のフィールドを理解する |
| 1-2 | IngestionStateShardFailure.java | `server/src/main/java/org/opensearch/action/admin/indices/streamingingestion/IngestionStateShardFailure.java` | シャード単位の失敗情報を表現する |
| 1-3 | IngestionUpdateStateResponse.java | `server/src/main/java/org/opensearch/action/admin/indices/streamingingestion/IngestionUpdateStateResponse.java` | 状態更新レスポンスの共通データ構造 |

**読解のコツ**: ShardIngestionStateのバージョン互換性処理（V_3_3_0以降でisPrimary, nodeNameが追加）に注意。**65-72行目**でバージョンチェックによる条件分岐がある。

#### Step 2: エントリーポイントを理解する

Pause操作のトランスポートアクションをエントリーポイントとして理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | TransportPauseIngestionAction.java | `server/src/main/java/org/opensearch/action/admin/indices/streamingingestion/pause/TransportPauseIngestionAction.java` | Pause操作のメイン処理。TransportClusterManagerNodeActionを継承 |
| 2-2 | PauseIngestionRequest.java | `server/src/main/java/org/opensearch/action/admin/indices/streamingingestion/pause/PauseIngestionRequest.java` | Pauseリクエストのバリデーション。indices必須チェック |

**主要処理フロー**:
1. **84-87行目**: doExecuteでDestructiveOperationsチェック後、親クラスのdoExecuteを呼び出し
2. **105-150行目**: clusterManagerOperationでインデックス解決後、MetadataStreamingIngestionStateServiceのupdateIngestionPollerStateを呼び出し
3. **117-122行目**: UpdateIngestionStateRequestを生成し、ingestionPaused=trueを設定

#### Step 3: Resume操作を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | TransportResumeIngestionAction.java | `server/src/main/java/org/opensearch/action/admin/indices/streamingingestion/resume/TransportResumeIngestionAction.java` | Resume操作のメイン処理。Pauseとほぼ同じ構造でingestionPaused=falseを設定 |
| 3-2 | ResumeIngestionRequest.java | `server/src/main/java/org/opensearch/action/admin/indices/streamingingestion/resume/ResumeIngestionRequest.java` | Resumeリクエスト |

#### Step 4: 状態取得・更新を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | GetIngestionStateRequest.java | `server/src/main/java/org/opensearch/action/admin/indices/streamingingestion/state/GetIngestionStateRequest.java` | 状態取得リクエスト。ページネーション対応（DEFAULT_PAGE_SIZE=1000） |
| 4-2 | UpdateIngestionStateRequest.java | `server/src/main/java/org/opensearch/action/admin/indices/streamingingestion/state/UpdateIngestionStateRequest.java` | 状態更新リクエスト |

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

```
REST API (Pause/Resume/GetState)
    |
    +-- TransportPauseIngestionAction / TransportResumeIngestionAction
    |       |
    |       +-- TransportClusterManagerNodeAction.doExecute()
    |       |       +-- DestructiveOperations.failDestructive()
    |       |
    |       +-- clusterManagerOperation()
    |               +-- IndexNameExpressionResolver.concreteResolvedIndices()
    |               +-- MetadataStreamingIngestionStateService.updateIngestionPollerState()
    |                       +-- UpdateIngestionStateRequest
    |                       +-- ActionListener<UpdateIngestionStateResponse>
    |
    +-- GetIngestionStateAction (Broadcast)
            +-- ShardIngestionState.groupShardStateByIndex()
```

### データフロー図

```
[入力]                    [処理]                              [出力]

PauseIngestionRequest --> TransportPauseIngestionAction   --> PauseIngestionResponse
                          |                                    (acknowledged, shards_acknowledged, failures)
                          +-> IndexNameExpressionResolver
                          +-> MetadataStreamingIngestionStateService

GetIngestionStateRequest --> BroadcastAction              --> GetIngestionStateResponse
                             |                                 (ShardIngestionState[])
                             +-> Each Shard Node
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| TransportPauseIngestionAction.java | `server/src/main/java/org/opensearch/action/admin/indices/streamingingestion/pause/` | ソース | Pause操作のトランスポートアクション |
| PauseIngestionRequest.java | 同上 | ソース | Pauseリクエスト定義 |
| PauseIngestionResponse.java | 同上 | ソース | Pauseレスポンス定義 |
| PauseIngestionAction.java | 同上 | ソース | Pauseアクション定義 |
| TransportResumeIngestionAction.java | `server/src/main/java/org/opensearch/action/admin/indices/streamingingestion/resume/` | ソース | Resume操作のトランスポートアクション |
| ResumeIngestionRequest.java | 同上 | ソース | Resumeリクエスト定義 |
| ResumeIngestionResponse.java | 同上 | ソース | Resumeレスポンス定義 |
| ShardIngestionState.java | `server/src/main/java/org/opensearch/action/admin/indices/streamingingestion/state/` | ソース | シャードインジェスト状態データ構造 |
| GetIngestionStateRequest.java | 同上 | ソース | 状態取得リクエスト |
| GetIngestionStateResponse.java | 同上 | ソース | 状態取得レスポンス |
| UpdateIngestionStateRequest.java | 同上 | ソース | 状態更新リクエスト |
| UpdateIngestionStateResponse.java | 同上 | ソース | 状態更新レスポンス |
| IngestionStateShardFailure.java | `server/src/main/java/org/opensearch/action/admin/indices/streamingingestion/` | ソース | シャード失敗情報 |
| IngestionUpdateStateResponse.java | 同上 | ソース | 状態更新レスポンス共通 |
