# 機能設計書 44-ファイルシステムインジェスト

## 概要

本ドキュメントは、ローカルファイルシステムからOpenSearchへのデータ取り込みを行うファイルシステムインジェストプラグインの機能設計書である。

### 本機能の処理概要

**業務上の目的・背景**：ストリーミングインジェスト機能の開発・テスト時に、Apache KafkaやAmazon Kinesisなどの外部インフラを用意せずにローカルファイルからデータ取り込みをシミュレーションできる環境が必要である。本プラグインはNDJSON形式のファイルからデータを読み取り、OpenSearchインデックスに取り込む機能を提供する。

**機能の利用シーン**：開発環境やテスト環境において、ストリーミングインジェスト機能の動作確認やE2Eテストを実施する場面で利用される。本番環境での使用は想定されていない。

**主要な処理内容**：
1. ファイルベースのデータ読み取り：NDJSON形式のファイルから行単位でデータを読み取る
2. 行番号ベースのオフセット管理：ファイルの行番号をオフセットとして管理
3. ラグ計算：未読行数をラグとして計算

**関連システム・外部連携**：ローカルファイルシステム。`${baseDir}/${stream}/${shardId}.ndjson`のパス規則でファイルを読み取る。

**権限による制御**：@SuppressForbiddenアノテーションにより、Java標準のファイルI/O APIが許可されている（通常のOpenSearchプラグインでは禁止される操作）。

## 関連画面

本機能は画面機能マッピングにおいて直接的な画面関連はない。

## 機能種別

データ連携（テスト用）

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| stream | String | Yes | ストリーム名（サブディレクトリ名） | - |
| base_directory | String | Yes | ベースディレクトリパス | - |

### 入力データソース

ローカルファイルシステム上のNDJSONファイル（`${base_directory}/${stream}/${shardId}.ndjson`）

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| offset | FileOffset (long) | 行番号（0始まり） |
| payload | byte[] | 行の内容（UTF-8バイト配列） |
| timestamp | long | 読み取り時のシステムタイムスタンプ |

### 出力先

OpenSearchインデックスへのドキュメント登録（IngestionShardConsumerインターフェース経由）

## 処理フロー

### 処理シーケンス

```
1. FilePlugin登録
   └─ IngestionConsumerPluginとしてTYPE="FILE"でFileConsumerFactoryを登録
2. FilePartitionConsumer生成
   └─ FileSourceConfigとshardIdを指定
3. ファイルパス構築
   └─ ${baseDir}/${stream}/${shardId}.ndjson
4. メッセージ読み取り（readNext）
   └─ BufferedReaderで行を読み取り、ReadResultに変換
5. オフセット管理
   └─ 行番号をFileOffsetとして管理
```

### フローチャート

```mermaid
flowchart TD
    A[FilePlugin Loaded] --> B[FileConsumerFactory registered]
    B --> C[FilePartitionConsumer created]
    C --> D[Construct file path]
    D --> E{File exists?}
    E -->|No| F[Log warning]
    E -->|Yes| G{readNext called}
    G --> H{Reader initialized?}
    H -->|No| I[Create BufferedReader]
    H -->|Yes| J{startLine < currentLine?}
    J -->|Yes| K[Reset reader]
    J -->|No| L[Skip to startLine]
    I --> L
    K --> L
    L --> M[Read lines up to maxLines]
    M --> N[Convert to ReadResult]
    N --> O[Return results]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-44-01 | ファイル不存在時の動作 | ファイルが存在しない場合は空のリストを返す | readNext呼び出し時 |
| BR-44-02 | タイムスタンプポインタ非対応 | タイムスタンプベースのポインタはearliest（行0）にフォールバック | pointerFromTimestampMillis呼び出し時 |
| BR-44-03 | ファイルパス規則 | `${baseDir}/${stream}/${shardId}.ndjson`の規則に従う | コンシューマー初期化時 |

### 計算ロジック

ラグ計算: `lag = latestOffset.getLine() - lastReadLine - 1`（159-172行目）

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

本機能はファイルからのデータ読み取りのみを行い、OpenSearchへのドキュメント登録は上位のインジェストフレームワークが担当する。

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| - | RuntimeException | ファイル読み取りI/Oエラー | ファイルパスとアクセス権限を確認 |

### リトライ仕様

上位のインジェストフレームワークによるリトライ制御に依存する。

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

ファイルからのシーケンシャル読み取りであり、トランザクション管理はOpenSearch側で行う。

## パフォーマンス要件

ローカルテスト用途であり、特定のパフォーマンス要件はない。

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

- @SuppressForbiddenアノテーションにより、通常禁止されるJavaファイルAPIが使用されている
- 本番環境での使用は推奨されない
- ファイルパスの入力検証は行われていないため、パストラバーサルのリスクがある

## 備考

本プラグインはローカルテスト用途（"used for local testing"）として位置づけられており、本番環境での使用は想定されていない。

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | FileOffset.java | `plugins/ingestion-fs/src/main/java/org/opensearch/plugin/ingestion/fs/FileOffset.java` | long型のlineフィールドを持つIngestionShardPointer実装 |
| 1-2 | FileMessage.java | `plugins/ingestion-fs/src/main/java/org/opensearch/plugin/ingestion/fs/FileMessage.java` | payload(byte[]), timestamp(long)を持つIngestionMessage実装 |
| 1-3 | FileSourceConfig.java | `plugins/ingestion-fs/src/main/java/org/opensearch/plugin/ingestion/fs/FileSourceConfig.java` | stream, baseDirectoryの設定クラス |

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | FilePlugin.java | `plugins/ingestion-fs/src/main/java/org/opensearch/plugin/ingestion/fs/FilePlugin.java` | TYPE="FILE"でFileConsumerFactoryを登録 |
| 2-2 | FileConsumerFactory.java | `plugins/ingestion-fs/src/main/java/org/opensearch/plugin/ingestion/fs/FileConsumerFactory.java` | FilePartitionConsumerを生成するファクトリ |

#### Step 3: コンシューマー処理を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | FilePartitionConsumer.java | `plugins/ingestion-fs/src/main/java/org/opensearch/plugin/ingestion/fs/FilePartitionConsumer.java` | メインのコンシューマー実装 |

**主要処理フロー**:
- **48-59行目**: コンストラクタでファイルパスを構築し存在チェック
- **73-111行目**: readFromFileで行読み取り。リーダーのリセット・スキップロジックに注意
- **114-136行目**: earliestPointerとlatestPointerの実装。latestPointerではファイル全体を走査

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

```
FilePlugin (IngestionConsumerPlugin)
    |
    +-- FileConsumerFactory (IngestionConsumerFactory)
            |
            +-- FilePartitionConsumer (IngestionShardConsumer)
                    |
                    +-- BufferedReader (Java I/O)
                    |       +-- readLine()
                    |
                    +-- FileSourceConfig
                    +-- FileOffset
                    +-- FileMessage
```

### データフロー図

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

NDJSON File                    --> FilePartitionConsumer       --> ReadResult<FileOffset, FileMessage>
  (${baseDir}/${stream}/            |                              (lineNumber, payload, timestamp)
   ${shardId}.ndjson)               +-> BufferedReader.readLine()
                                    +-> Convert to FileMessage
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| FilePlugin.java | `plugins/ingestion-fs/src/main/java/org/opensearch/plugin/ingestion/fs/` | ソース | プラグインエントリーポイント |
| FileConsumerFactory.java | 同上 | ソース | コンシューマーファクトリ |
| FilePartitionConsumer.java | 同上 | ソース | パーティション単位のコンシューマー |
| FileSourceConfig.java | 同上 | ソース | ファイル接続設定 |
| FileOffset.java | 同上 | ソース | オフセットデータ構造 |
| FileMessage.java | 同上 | ソース | メッセージデータ構造 |
