# 画面設計書 132-パイプライン作成・更新

## 概要

本ドキュメントは、OpenSearchのIngestパイプライン作成・更新API（`PUT /_ingest/pipeline/{id}`）の画面設計書である。インジェストパイプラインを新規作成または既存パイプラインを更新するためのREST APIエンドポイントの仕様を定義する。

### 本画面の処理概要

本APIは、OpenSearchクラスタにインジェストパイプラインを登録・更新するエンドポイントである。パイプラインはプロセッサの連鎖で構成され、ドキュメントがインデックスに登録される前にデータを加工・変換する。

**業務上の目的・背景**：インジェストパイプラインは、ドキュメント登録時にデータの正規化、エンリッチメント、フィルタリングなどの前処理を自動化する仕組みである。ログデータのパース（GrokプロセッサによるApacheログの構造化など）、日付フォーマットの変換、不要フィールドの削除など、データ取り込み時の変換ロジックを一元管理するために使用される。

**画面へのアクセス方法**：HTTPクライアントから `PUT /_ingest/pipeline/{id}` へパイプライン定義をJSON形式のリクエストボディとともに送信する。

**主要な操作・処理内容**：
1. クライアントがパイプラインIDとパイプライン定義（プロセッサ配列等）をPUTリクエストで送信する
2. RestPutPipelineActionがリクエストを受け取り、PutPipelineRequestを構築する
3. IngestServiceがパイプライン定義をバリデーションし、クラスタ状態を更新する
4. クラスタ状態の更新が全ノードに伝播された後、確認レスポンスを返却する

**画面遷移**：パイプライン取得API（131）で既存定義を確認した後に本APIで更新を行う。作成後はパイプラインシミュレーションAPI（134）でテストし、ドキュメント登録・更新API（3）でパイプラインを適用する流れが一般的である。

**権限による表示制御**：クラスタのインジェストパイプライン管理権限（cluster:admin/ingest/pipeline/put）が必要。権限不足の場合は403エラーが返却される。

## 関連機能

| 機能No | 機能名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| 36 | インジェストパイプライン | 主機能 | インジェストパイプラインを作成または更新する処理 |

## 画面種別

登録・編集（REST API PUTエンドポイント）

## URL/ルーティング

| メソッド | パス | 説明 |
|---------|------|------|
| PUT | `/_ingest/pipeline/{id}` | パイプラインを作成または更新する |

## 入出力項目

### パスパラメータ

| パラメータ名 | 型 | 必須 | 説明 |
|-------------|-----|------|------|
| id | string | はい | パイプラインID |

### クエリパラメータ

| パラメータ名 | 型 | 必須 | デフォルト | 説明 |
|-------------|-----|------|-----------|------|
| cluster_manager_timeout | time | いいえ | 30s | クラスタマネージャノードへの接続タイムアウト |
| master_timeout | time | いいえ | 30s | （非推奨）cluster_manager_timeoutを使用すること |
| timeout | time | いいえ | 30s | 操作の明示的タイムアウト |

### リクエストボディ（必須）

| 項目名 | 型 | 必須 | 説明 |
|--------|-----|------|------|
| description | string | いいえ | パイプラインの説明 |
| processors | array | はい | プロセッサ定義の配列 |
| version | integer | いいえ | パイプラインのバージョン番号 |
| on_failure | array | いいえ | 失敗時のフォールバックプロセッサ定義 |

## 表示項目

### レスポンスボディ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| acknowledged | boolean | クラスタ状態の更新が全ノードに承認されたか |

## イベント仕様

### 1-パイプライン作成・更新リクエスト

クライアントからPUTリクエストを受信すると、RestPutPipelineActionの`prepareRequest`メソッドが呼び出される。リクエストボディからコンテンツとメディアタイプを抽出し、PutPipelineRequestオブジェクトを生成する。クラスタマネージャタイムアウトと操作タイムアウトを設定した後、`client.admin().cluster().putPipeline()`を呼び出す。IngestServiceは以下の処理を実行する：(1)パイプライン定義のパース・バリデーション、(2)プロセッサの生成と初期化検証、(3)AckedClusterStateUpdateTaskを通じたクラスタ状態の更新。

## データベース更新仕様

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

| 操作（イベント） | 対象テーブル | 操作種別 | 概要 |
|----------------|-------------|---------|------|
| パイプライン作成・更新 | ClusterState（IngestMetadata） | INSERT/UPDATE | クラスタ状態にパイプライン定義を登録または更新 |

### テーブル別更新項目詳細

#### ClusterState - IngestMetadata

| 操作 | 項目（カラム名） | 更新値・取得条件 | 備考 |
|-----|-----------------|-----------------|------|
| INSERT/UPDATE | pipelines[{id}] | リクエストボディのパイプライン定義 | 同一IDのパイプラインが存在する場合は上書き |

## メッセージ仕様

| メッセージID | メッセージ内容 | 種別 | 発生条件 |
|-------------|--------------|------|---------|
| 200 | `{"acknowledged": true}` | 成功 | パイプラインが正常に作成・更新された場合 |
| 400 | パイプライン定義が不正 | エラー | プロセッサ定義にエラーがある場合 |
| 403 | 権限不足 | エラー | 必要な権限がない場合 |
| 408 | タイムアウト | エラー | 操作タイムアウトが発生した場合 |

## 例外処理

- **不正なパイプライン定義**: プロセッサの種別が不明、必須パラメータの欠如などの場合、OpenSearchParseExceptionがスローされHTTPステータス400が返却される
- **リクエストボディの欠如**: リクエストボディが空の場合、必須ボディが存在しないエラーが返却される
- **クラスタマネージャノード接続タイムアウト**: 指定時間内にクラスタマネージャノードから応答がない場合、タイムアウトエラーが返却される
- **クラスタブロック**: クラスタ全体にメタデータ書き込みブロックが設定されている場合、ClusterBlockExceptionが返却される

## 備考

- 同一IDのパイプラインが既に存在する場合は上書き更新される（エラーにはならない）
- パイプライン作成時にプロセッサの初期化バリデーションが実行されるため、プロセッサの構成に問題がある場合はこの時点でエラーとなる
- `master_timeout` パラメータはバージョン2.0.0で非推奨

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | PutPipelineRequest.java | `server/src/main/java/org/opensearch/action/ingest/PutPipelineRequest.java` | パイプライン登録リクエストの構造。id, source, mediaTypeの保持方法 |
| 1-2 | PipelineConfiguration.java | `server/src/main/java/org/opensearch/ingest/PipelineConfiguration.java` | パイプライン構成のシリアライゼーション |

**読解のコツ**: PutPipelineRequestは`AcknowledgedRequest`を継承しており、クラスタ状態の更新を伴うリクエストである。

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | RestPutPipelineAction.java | `server/src/main/java/org/opensearch/rest/action/ingest/RestPutPipelineAction.java` | RESTリクエストからPutPipelineRequestへの変換処理 |

**主要処理フロー**:
1. **行62**: ルート定義 - `PUT /_ingest/pipeline/{id}` を登録
2. **行72**: リクエストボディからコンテンツとメディアタイプを抽出
3. **行73**: PutPipelineRequestを生成（id, source, mediaType）
4. **行74-76**: タイムアウト設定の適用
5. **行77**: `client.admin().cluster().putPipeline()` を呼び出し

#### Step 3: サービス層を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | IngestService.java | `server/src/main/java/org/opensearch/ingest/IngestService.java` | `putPipeline()`メソッドによるクラスタ状態更新処理 |

**主要処理フロー**:
- パイプライン定義のパースとバリデーション
- プロセッサインスタンスの生成・初期化テスト
- AckedClusterStateUpdateTaskによるクラスタ状態の原子的更新

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

```
RestPutPipelineAction.prepareRequest()
    |
    +-- PutPipelineRequest(id, source, mediaType)
    |
    +-- client.admin().cluster().putPipeline()
            |
            +-- IngestService.putPipeline()
                    |
                    +-- validatePipeline() - プロセッサのバリデーション
                    |
                    +-- AckedClusterStateUpdateTask
                            |
                            +-- ClusterState.metadata().putCustom(IngestMetadata)
                    |
                    +-- AcknowledgedResponse -> JSONレスポンス
```

### データフロー図

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

PUT /_ingest/pipeline/{id}
    |
    +-- id (パスパラメータ)    --> RestPutPipelineAction            --> PutPipelineRequest
    +-- リクエストボディ(JSON)        |                                   |
    +-- timeout                      v                                   v
                              IngestService.putPipeline()         バリデーション
                                     |                                   |
                                     v                                   v
                              ClusterState更新                   AcknowledgedResponse
                                                                        |
                                                                        v
                                                              {"acknowledged": true}
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| RestPutPipelineAction.java | `server/src/main/java/org/opensearch/rest/action/ingest/RestPutPipelineAction.java` | ソース | RESTエントリーポイント |
| PutPipelineRequest.java | `server/src/main/java/org/opensearch/action/ingest/PutPipelineRequest.java` | ソース | リクエストデータ構造 |
| PutPipelineAction.java | `server/src/main/java/org/opensearch/action/ingest/PutPipelineAction.java` | ソース | アクション定義 |
| IngestService.java | `server/src/main/java/org/opensearch/ingest/IngestService.java` | ソース | パイプライン管理サービス |
| IngestMetadata.java | `server/src/main/java/org/opensearch/ingest/IngestMetadata.java` | ソース | パイプラインメタデータ管理 |
| PipelineConfiguration.java | `server/src/main/java/org/opensearch/ingest/PipelineConfiguration.java` | ソース | パイプライン構成データ |
| ingest.put_pipeline.json | `rest-api-spec/src/main/resources/rest-api-spec/api/ingest.put_pipeline.json` | API仕様 | REST API仕様定義 |
