# バッチ設計書 23-IndexingMemoryController

## 概要

本ドキュメントは、OpenSearchのIndexingMemoryControllerが提供するインデクシングバッファのメモリ使用状況監視バッチ処理の設計を記述する。このバッチはノード全体のインデクシングバッファメモリ使用量を定期的にチェックし、閾値超過時にシャードのバッファをディスクへ書き出す。

### 本バッチの処理概要

IndexingMemoryControllerは、ShardsIndicesStatusCheckerという内部Runnableを定期的に実行し、ノード上の全アクティブシャードのインデクシングバッファRAM使用量を監視する。使用量が設定された上限（デフォルトはヒープの10%）を超過した場合、最もメモリを使用しているシャードから順にインデクシングバッファのディスク書き出し（writeIndexingBuffer）を非同期で実行する。さらに、書き出し中のバイト数と使用中のバイト数の合計がバッファサイズの1.5倍を超えた場合、インデクシングのスロットリングを有効にして書き込み速度を制限する。

**業務上の目的・背景**：インデクシング操作はメモリバッファにデータを蓄積し、定期的にLuceneセグメントとしてディスクに書き出す。多数のシャードが同時にインデクシングを行うと、バッファが肥大化しヒープメモリを圧迫する。IndexingMemoryControllerは、ヒープ枯渇によるOutOfMemoryErrorやGCの頻発を防止するために、バッファメモリの使用量を適切に制御する重要な役割を担う。

**バッチの実行タイミング**：indices.memory.interval設定（デフォルト5秒）の間隔でスケジューラにより定期実行される。加えて、インデクシング操作のバイト数がバッファサイズの1/30を超えた場合にも即座にチェックが走る。

**主要な処理内容**：
1. 全アクティブシャードのインデクシングバッファRAM使用量を集計する
2. 各シャードのアイドル状態をチェックし、アイドルシャードをフラッシュする
3. 使用量がバッファサイズ上限を超過した場合、最大使用量のシャードから順にディスクへの書き出しを実行する
4. 書き出し中バイト数+使用中バイト数がバッファサイズの1.5倍超の場合、スロットリングを有効化する
5. スロットリング条件が解消された場合、スロットリングを解除する

**前後の処理との関連**：IndexShardのindex/delete操作がIndexingOperationListenerとして本コントローラに通知される。スロットリング有効化によりIndexShardのインデクシング速度が制限される。writeIndexingBufferAsync()はREFRESHスレッドプールで非同期実行される。

**影響範囲**：ノード上の全アクティブインデックスシャードのインデクシング性能に影響する。スロットリングが発動するとインデクシングスループットが低下する。

## バッチ種別

リソース監視・メモリ管理

## 実行スケジュール

| 項目 | 内容 |
|-----|------|
| 実行頻度 | indices.memory.interval（デフォルト5秒）+ インデクシングバイト数による即時トリガー |
| 実行時刻 | ノード起動後から継続的に実行 |
| 実行曜日 | 該当なし（常時） |
| 実行日 | 該当なし（常時） |
| トリガー | ThreadPool.scheduleWithFixedDelay + バイト書き込み量トリガー |

## 実行条件

### 前提条件

| 条件 | 説明 |
|-----|------|
| アクティブシャードの存在 | RECOVERING、POST_RECOVERY、STARTEDのいずれかの状態のシャードが対象 |

### 実行可否判定

スケジューラにより常に実行される。バイト書き込み量によるトリガーはReentrantLock.tryLock()で排他制御され、同時に複数回実行されることを防止する。

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | デフォルト値 | 説明 |
|-------------|-----|-----|-------------|------|
| indices.memory.index_buffer_size | ByteSizeValue | No | 10%（ヒープの割合） | インデクシングバッファの合計サイズ上限 |
| indices.memory.min_index_buffer_size | ByteSizeValue | No | 48MB | バッファサイズの下限（%指定時のみ適用） |
| indices.memory.max_index_buffer_size | ByteSizeValue | No | 無制限 | バッファサイズの上限（%指定時のみ適用） |
| indices.memory.shard_inactive_time | TimeValue | No | 5m | シャードをアイドルとみなすまでの時間 |
| indices.memory.interval | TimeValue | No | 5s | メモリチェックの実行間隔 |

### 入力データソース

| データソース | 形式 | 説明 |
|-------------|------|------|
| IndexShard.getIndexBufferRAMBytesUsed() | long | 各シャードのバッファRAM使用量 |
| IndexShard.getWritingBytes() | long | 各シャードの書き出し中バイト数 |

## 出力仕様

### 出力データ

| 出力先 | 形式 | 説明 |
|-------|------|------|
| IndexShard.writeIndexingBuffer() | 非同期実行 | バッファからディスクへの書き出し |
| IndexShard.activateThrottling() | メソッド呼び出し | インデクシングスロットリングの有効化 |
| IndexShard.deactivateThrottling() | メソッド呼び出し | インデクシングスロットリングの解除 |

### 出力ファイル仕様

直接のファイル出力はなし。IndexShardのwriteIndexingBuffer()がLuceneセグメントファイルを生成する。

## 処理フロー

### 処理シーケンス

```
1. 全アクティブシャードを列挙
   └─ RECOVERING/POST_RECOVERY/STARTEDの状態のシャード
2. 各シャードのアイドル判定（flushOnIdle）
   └─ inactiveTime（デフォルト5分）以上操作がなければフラッシュ
3. 各シャードのバッファRAM使用量と書き出し中バイト数を集計
   └─ totalBytesUsed、totalBytesWritingを算出
4. スロットリング判定
   └─ (totalBytesWriting + totalBytesUsed) > 1.5 * indexingBuffer
5. バッファ超過判定
   └─ totalBytesUsed > indexingBuffer の場合、書き出しを開始
6. 優先度キューにシャードを投入（使用量降順）
7. 最大使用量のシャードから順にwriteIndexingBufferAsync()を呼び出し
   └─ totalBytesUsedがバッファサイズ以下になるまで繰り返す
8. スロットリング条件に応じてactivate/deactivateThrottling
```

### フローチャート

```mermaid
flowchart TD
    A[チェッカー起動] --> B[全アクティブシャード列挙]
    B --> C[各シャードのアイドル判定]
    C --> D[バッファ使用量・書き出し中バイト数集計]
    D --> E{totalBytesUsed > indexingBuffer?}
    E -->|No| F{スロットリング解除必要?}
    E -->|Yes| G[優先度キュー構築]
    G --> H[最大使用量シャードを書き出し]
    H --> I{スロットリング条件?}
    I -->|Yes| J[スロットリング有効化]
    I -->|No| K[次のシャードを書き出し]
    J --> K
    K --> L{バッファ以下 or キュー空?}
    L -->|No| H
    L -->|Yes| F
    F -->|Yes| M[全スロットリング解除]
    F -->|No| N[チェック終了]
    M --> N
```

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

本バッチはデータベースを使用しない。Luceneインデックスバッファの管理を行う。

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| - | AlreadyClosedException | シャードがクローズ済みの場合のアイドルチェック | TRACEログ出力してスキップ |
| - | Exception | writeIndexingBufferAsync内でのバッファ書き出し失敗 | WARNログ出力して無視 |

### リトライ仕様

| 項目 | 内容 |
|-----|------|
| リトライ回数 | なし（次回スケジュールで再実行） |
| リトライ間隔 | indices.memory.interval（デフォルト5秒） |
| リトライ対象エラー | 全てのエラー（次回実行で自動リトライ） |

### 障害時対応

ThreadPool.scheduleWithFixedDelayは例外をログに記録し再スケジュールするため、一時的な障害は自動的に回復する。

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

| 項目 | 内容 |
|-----|------|
| トランザクション範囲 | なし |
| コミットタイミング | 該当なし |
| ロールバック条件 | 該当なし |

## パフォーマンス要件

| 項目 | 内容 |
|-----|------|
| 想定処理件数 | ノード上のアクティブシャード数に依存 |
| 目標処理時間 | チェック間隔（5秒）内に完了すること |
| メモリ使用量上限 | indices.memory.index_buffer_size（デフォルトヒープの10%） |

## 排他制御

ShardsIndicesStatusCheckerはReentrantLockで排他制御される。スケジューラによる定期実行とバイト書き込み量によるトリガーが競合した場合、tryLock()で衝突を回避する。throttledセットへの操作はチェッカーのロック内で行われる。

## ログ出力

| ログ種別 | 出力タイミング | 出力内容 |
|---------|--------------|---------|
| DEBUGログ | 起動時 | バッファサイズ、アイドル時間、チェック間隔 |
| DEBUGログ | バッファ書き出し開始時 | 使用量、シャード数 |
| TRACEログ | 各シャードの使用量集計時 | シャードごとの使用量と書き出し中バイト数 |
| INFOログ | スロットリング開始/停止時 | 対象シャード |
| WARNログ | バッファ書き出し失敗時 | シャードIDとエラー詳細 |

## 監視・アラート

| 監視項目 | 閾値 | アラート先 |
|---------|-----|----------|
| スロットリング発動 | 発動時にINFOログ出力 | OpenSearchログ |
| バッファ超過 | indexingBuffer超過時 | OpenSearchログ（DEBUGレベル） |

## 備考

- バイト書き込み量によるトリガーはバッファサイズの1/30を閾値とする近似チェックである
- 書き出しはREFRESHスレッドプールで非同期に実行される
- アイドル判定されたシャードはflushOnIdle()によりフラッシュされる
