# 機能設計書 36-インジェストパイプライン

## 概要

本ドキュメントは、OpenSearchのインジェストパイプライン機能の設計を記述する。インジェストパイプラインはドキュメント登録前にデータ変換・加工を行う機能であり、プロセッサのチェーンを定義して順次適用する。

### 本機能の処理概要

インジェストパイプラインは、ドキュメントがインデックスに格納される前に、一連のプロセッサを順次適用してドキュメントの変換・加工・エンリッチメントを行う機能である。パイプラインの作成・更新・削除・取得・シミュレーションのAPI、およびドキュメント登録時のパイプライン実行を提供する。

**業務上の目的・背景**：ドキュメントの前処理をサーバーサイドで一元的に管理するニーズに対応する。フィールド名の変更、データ型の変換、日付のパース、外部データの付与、不要フィールドの除去等をクライアント側の変更なしに実現する。ETL（Extract, Transform, Load）パイプラインのTransformに相当する機能。

**機能の利用シーン**：
- パイプラインAPIでパイプラインをCRUD管理
- ドキュメント登録時にpipelineパラメータでパイプラインを指定
- インデックス設定のdefault_pipelineで自動適用を設定
- バルク操作時にパイプラインを適用

**主要な処理内容**：
1. Put Pipeline APIでパイプライン定義をクラスタ状態に保存
2. Get Pipeline APIでパイプライン定義を取得
3. Delete Pipeline APIでパイプラインを削除
4. Simulate Pipeline APIでサンプルドキュメントに対するパイプライン処理をテスト
5. ドキュメントインデックス時にIngestServiceがパイプラインを実行
6. CompoundProcessorが各プロセッサを順次実行
7. on_failureハンドラによるエラー処理

**関連システム・外部連携**：クラスタ状態（IngestMetadata）でパイプライン定義を永続化。IngestPluginインターフェースでプロセッサプラグインを統合。

**権限による制御**：パイプラインの管理にはcluster:admin/ingest/pipeline権限が必要。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| 3 | ドキュメント登録・更新 | 補助機能 | ドキュメント登録前のインジェストパイプライン適用処理 |
| 4 | ドキュメント作成 | 補助機能 | ドキュメント作成前のインジェストパイプライン適用処理 |
| 12 | バルク操作 | 補助機能 | バルク操作時のインジェストパイプライン適用処理 |
| 131 | パイプライン取得 | 主機能 | インジェストパイプラインの定義を返す処理 |
| 132 | パイプライン作成・更新 | 主機能 | インジェストパイプラインを作成または更新する処理 |
| 133 | パイプライン削除 | 主機能 | インジェストパイプラインを削除する処理 |
| 134 | パイプラインシミュレーション | 主機能 | サンプルドキュメントでパイプラインの処理結果をシミュレーション |

## 機能種別

データ連携（データ変換・加工パイプライン）/ CRUD操作

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| id | String | Yes（PUT/DELETE） | パイプラインID | 512バイト以内 |
| description | String | No | パイプラインの説明 | - |
| version | Integer | No | パイプラインバージョン | - |
| processors | Array | Yes（PUT） | プロセッサ定義の配列 | 少なくとも1つのプロセッサ |
| on_failure | Array | No | エラー時のフォールバックプロセッサ配列 | - |
| pipeline（Index API） | String | No | 適用するパイプラインID | 存在するパイプライン |
| docs（Simulate） | Array | Yes（Simulate） | シミュレーション対象ドキュメント | - |

### 入力データソース

- REST APIリクエスト（パイプライン管理API）
- ドキュメントインデックスリクエスト（pipelineパラメータ）
- インデックス設定（default_pipeline, final_pipeline）

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| acknowledged | Boolean | パイプラインCRUD操作の成功/失敗 |
| pipelines | Array | パイプライン定義一覧（Get API） |
| docs | Array | シミュレーション結果（Simulate API） |
| ingest_info | Object | 利用可能なプロセッサ情報 |

### 出力先

REST APIレスポンス。パイプライン定義はクラスタ状態（IngestMetadata）に永続化。

## 処理フロー

### 処理シーケンス

```
1. Put Pipeline APIリクエスト受信
   └─ PutPipelineRequestをパース
2. パイプライン定義の検証
   └─ プロセッサファクトリによる各プロセッサの検証
3. クラスタ状態の更新
   └─ IngestMetadataにパイプライン定義を保存
4. ドキュメントインデックスリクエスト受信
   └─ pipelineパラメータまたはdefault_pipelineを解決
5. IngestService.executePipeline()
   └─ パイプライン取得、CompoundProcessor実行
6. 各プロセッサの順次実行
   └─ IngestDocumentに対して変換処理を適用
7. エラー発生時
   └─ on_failureプロセッサを実行
8. 変換済みドキュメントのインデックス
   └─ 通常のインデックス処理に引き継ぎ
```

### フローチャート

```mermaid
flowchart TD
    A[ドキュメントインデックスリクエスト] --> B{pipelineパラメータ?}
    B -->|Yes| C[指定パイプライン取得]
    B -->|No| D{default_pipeline設定?}
    D -->|Yes| C
    D -->|No| E[パイプラインなしでインデックス]
    C --> F[CompoundProcessor.execute]
    F --> G[プロセッサ1実行]
    G --> H{成功?}
    H -->|Yes| I[プロセッサ2実行]
    H -->|No| J{on_failure定義?}
    J -->|Yes| K[on_failureプロセッサ実行]
    J -->|No| L[エラー返却]
    I --> M[... 残りのプロセッサ]
    M --> N[変換済みドキュメントのインデックス]
    K --> N
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-36-01 | パイプラインID長 | パイプラインIDは512バイト以内 | Put Pipeline |
| BR-36-02 | NOOP パイプライン | pipeline="_none"でパイプラインを無効化 | ドキュメントインデックス |
| BR-36-03 | default_pipeline | インデックス設定のdefault_pipelineが自動適用 | ドキュメントインデックス |
| BR-36-04 | final_pipeline | final_pipelineはdefault_pipelineの後に必ず実行 | ドキュメントインデックス |
| BR-36-05 | プロセッサ数上限 | cluster.ingest.max_number_processorsで制限 | Put Pipeline |
| BR-36-06 | システムパイプライン | cluster.ingest.system_pipeline_enabledで有効/無効 | システムパイプライン |
| BR-36-07 | on_failure | プロセッサエラー時にon_failureチェーンを実行 | プロセッサエラー発生 |

### 計算ロジック

特になし（データ変換のためのルーティングロジック）。

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

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

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| Put Pipeline | クラスタ状態（IngestMetadata） | UPDATE | パイプライン定義をクラスタ状態に保存 |
| Delete Pipeline | クラスタ状態（IngestMetadata） | DELETE | パイプライン定義をクラスタ状態から削除 |
| Get Pipeline | クラスタ状態（IngestMetadata） | SELECT | パイプライン定義を取得 |

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

#### クラスタ状態（IngestMetadata）

| 操作 | 項目 | 更新値・取得条件 | 備考 |
|-----|------|-----------------|------|
| UPDATE | パイプライン定義 | PipelineConfigurationオブジェクト | AckedClusterStateUpdateTask経由 |
| DELETE | パイプライン定義 | パイプラインIDで指定 | AckedClusterStateUpdateTask経由 |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| 404 | resource_not_found_exception | 存在しないパイプラインを参照 | 正しいパイプラインIDを指定 |
| 400 | illegal_argument_exception | パイプラインID長超過（512バイト超） | 短いIDに変更 |
| 400 | parse_exception | プロセッサ定義のパースエラー | 正しいプロセッサ設定を指定 |
| 500 | ingest_process_exception | プロセッサ実行時のエラー | on_failureで処理するか、プロセッサ設定を修正 |

### リトライ仕様

パイプラインの管理操作はクラスタ状態更新タスクとして実行され、ClusterManagerTaskThrottlerでスロットリングされる。ドキュメント処理のリトライはクライアント側で実施。

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

パイプライン定義の更新はクラスタ状態のアトミック更新として実行される。ドキュメントのインジェスト処理自体はドキュメント単位で独立して実行。

## パフォーマンス要件

- プロセッサ数に比例したレイテンシ増加
- バルク操作時は各ドキュメントに対してパイプラインが個別に適用
- SystemIngestPipelineCacheでシステムパイプラインの結果をキャッシュ

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

- パイプライン管理APIにはクラスタ管理権限が必要
- スクリプトプロセッサを含むパイプラインはスクリプト実行権限に依存
- IngestDocumentのフィールドアクセスにはセキュリティ制御なし

## 備考

- IngestServiceはClusterStateApplierを実装し、クラスタ状態変更時にパイプラインを再構築
- PipelineProcessorにより別パイプラインをネストして呼び出すことが可能
- ConditionalProcessorで条件付きプロセッサ実行が可能
- TrackingResultProcessorでシミュレーション時のプロセッサ追跡が可能

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | IngestDocument.java | `server/src/main/java/org/opensearch/ingest/IngestDocument.java` | ドキュメントのラッパークラス。フィールドアクセス・変更メソッド |
| 1-2 | PipelineConfiguration.java | `server/src/main/java/org/opensearch/ingest/PipelineConfiguration.java` | パイプライン設定の永続化形式 |
| 1-3 | IngestMetadata.java | `server/src/main/java/org/opensearch/ingest/IngestMetadata.java` | クラスタ状態内のインジェストメタデータ |
| 1-4 | Processor.java | `server/src/main/java/org/opensearch/ingest/Processor.java` | プロセッサインターフェース |

**読解のコツ**: Processorインターフェースのexecute(IngestDocument)メソッドがプロセッサの核心。非同期バージョンのexecute(IngestDocument, BiConsumer)もある。Processor.Factoryがプロセッサのインスタンス化を担当。

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | IngestService.java | `server/src/main/java/org/opensearch/ingest/IngestService.java` | インジェストサービスの中核クラス |

**主要処理フロー**:
- **118行目**: IngestServiceクラス定義。ClusterStateApplier, ReportingServiceを実装
- **120行目**: NOOP_PIPELINE_NAME = "_none"
- **128-135行目**: MAX_NUMBER_OF_INGEST_PROCESSORS設定（デフォルトInteger.MAX_VALUE）
- **140-145行目**: SYSTEM_INGEST_PIPELINE_ENABLED設定（デフォルトtrue）
- **151行目**: processorFactoriesマップでプロセッサファクトリを管理
- **157行目**: pipelinesマップでパイプラインインスタンスを管理
- **169-199行目**: コンストラクタでClusterService、ScriptService、プロセッサファクトリを初期化

#### Step 3: パイプライン実行ロジックを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | Pipeline.java | `server/src/main/java/org/opensearch/ingest/Pipeline.java` | パイプラインクラス。id, description, version, CompoundProcessor |
| 3-2 | CompoundProcessor.java | `server/src/main/java/org/opensearch/ingest/CompoundProcessor.java` | プロセッサチェーンの順次実行。on_failureハンドリング |

**主要処理フロー**:
- **56行目**: Pipelineクラス定義
- **58-61行目**: DESCRIPTION_KEY, PROCESSORS_KEY, VERSION_KEY, ON_FAILURE_KEY定数
- **63-70行目**: id, description, version, compoundProcessor, metrics, relativeTimeProviderフィールド

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

```
TransportBulkAction / TransportIndexAction
    |
    +-- IngestService.executeBulkRequest()
          |
          +-- Pipeline.execute(IngestDocument)
                |
                +-- CompoundProcessor.execute(IngestDocument)
                      |
                      +-- Processor[0].execute(IngestDocument)
                      +-- Processor[1].execute(IngestDocument)
                      +-- ...
                      +-- [on_failure] CompoundProcessor.execute()
                            +-- FailureProcessor.execute()

Put Pipeline API:
    RestPutPipelineAction
        +-- TransportPutPipelineAction
              +-- IngestService.putPipeline()
                    +-- ClusterService.submitStateUpdateTask()
                          +-- IngestMetadata更新
```

### データフロー図

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

ドキュメント         +--> IngestService.executePipeline()
(IndexRequest)       |
                     +--> Pipeline取得
パイプライン定義     |     (IngestMetadata → Pipeline)
(クラスタ状態)       |
                     +--> CompoundProcessor.execute()
                     |     +-- Processor[0]: フィールド変換
                     |     +-- Processor[1]: 日付パース
                     |     +-- Processor[N]: ...
                     |
                     +--> on_failure（エラー時）
                     |
                     +--> 変換済みIngestDocument    +--> インデックス処理
                                                         (TransportBulkAction)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| IngestService.java | `server/src/main/java/org/opensearch/ingest/IngestService.java` | ソース | インジェストサービス中核 |
| Pipeline.java | `server/src/main/java/org/opensearch/ingest/Pipeline.java` | ソース | パイプラインクラス |
| CompoundProcessor.java | `server/src/main/java/org/opensearch/ingest/CompoundProcessor.java` | ソース | プロセッサチェーン |
| Processor.java | `server/src/main/java/org/opensearch/ingest/Processor.java` | ソース | プロセッサインターフェース |
| IngestDocument.java | `server/src/main/java/org/opensearch/ingest/IngestDocument.java` | ソース | ドキュメントラッパー |
| IngestMetadata.java | `server/src/main/java/org/opensearch/ingest/IngestMetadata.java` | ソース | クラスタ状態メタデータ |
| PipelineConfiguration.java | `server/src/main/java/org/opensearch/ingest/PipelineConfiguration.java` | ソース | パイプライン設定 |
| ConditionalProcessor.java | `server/src/main/java/org/opensearch/ingest/ConditionalProcessor.java` | ソース | 条件付きプロセッサ |
| PipelineProcessor.java | `server/src/main/java/org/opensearch/ingest/PipelineProcessor.java` | ソース | パイプラインネスト |
| TrackingResultProcessor.java | `server/src/main/java/org/opensearch/ingest/TrackingResultProcessor.java` | ソース | シミュレーション追跡 |
| ConfigurationUtils.java | `server/src/main/java/org/opensearch/ingest/ConfigurationUtils.java` | ソース | 設定ユーティリティ |
| IngestStats.java | `server/src/main/java/org/opensearch/ingest/IngestStats.java` | ソース | インジェスト統計 |
