# バッチ設計書 20-PersistentTasksClusterService (PeriodicRechecker)

## 概要

本ドキュメントは、未割り当ての永続タスクを定期的に再チェック・再割り当てするバッチ処理「PersistentTasksClusterService (PeriodicRechecker)」の設計仕様を記載する。

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

PersistentTasksClusterServiceのPeriodicRecheckerは、クラスタマネージャノードでのみ動作し、未割り当て状態の永続タスク（PersistentTask）を定期的に検出して適切なノードへの再割り当てを試みるバッチ処理である。

**業務上の目的・背景**：OpenSearchの永続タスク（Persistent Tasks）は、ノードの追加・削除、クラスタ状態の変更に伴い、割り当て先ノードが利用不可能になることがある。例えば、タスクを実行していたノードがクラスタから離脱した場合、そのタスクは未割り当て状態になる。本バッチは、このような未割り当てタスクを検出し、利用可能なノードへの再割り当てを試みることで、永続タスクの継続的な実行を保証する。クラスタ状態変更イベント（clusterChanged）でも即座に再割り当てが試みられるが、割り当て可能なノードが一時的に存在しない場合に備え、定期的な再チェックが必要となる。

**バッチの実行タイミング**：`cluster.persistent_tasks.allocation.recheck_interval`設定に基づく定期実行（デフォルト30秒間隔、最小10秒）。未割り当てタスクが存在する場合にのみスケジュールされる。

**主要な処理内容**：
1. ローカルノードがクラスタマネージャであることを確認
2. PersistentTasksCustomMetadataから未割り当てタスクの有無を確認
3. 未割り当てタスクが存在する場合、reassignPersistentTasks()を呼び出し
4. ClusterStateUpdateTaskとして再割り当てロジックを実行
5. 各タスクについてneedsReassignment()で再割り当て必要性を判定
6. createAssignment()で新しい割り当てを決定
7. 割り当てが変更される場合のみクラスタ状態を更新

**前後の処理との関連**：PersistentTasksClusterServiceはClusterStateListenerも実装しており、クラスタ状態変更時にもshouldReassignPersistentTasks()で再割り当て判定を行う。即座の再割り当てが失敗した場合やタスクがまだ未割り当ての場合に、PeriodicRecheckerが定期的にフォローアップする。EnableAssignmentDeciderが割り当て可否の判定を行う。

**影響範囲**：クラスタ状態の永続タスクメタデータに影響。再割り当てによりクラスタ状態が更新され、対応するノードでタスクが開始される。

## バッチ種別

クラスタ管理 / タスクスケジューリング

## 実行スケジュール

| 項目 | 内容 |
|-----|------|
| 実行頻度 | 定期実行（デフォルト30秒間隔） |
| 実行時刻 | 未割り当てタスク検出時にスケジュール開始 |
| 実行曜日 | 該当なし |
| 実行日 | 該当なし |
| トリガー | AbstractAsyncTask（rescheduleIfNecessary） |

## 実行条件

### 前提条件

| 条件 | 説明 |
|-----|------|
| ローカルノードがクラスタマネージャであること | isClusterManagerNode()で確認 |
| 未割り当ての永続タスクが存在すること | isAnyTaskUnassigned()で確認 |

### 実行可否判定

PeriodicRecheckerはAbstractAsyncTaskを継承しており、rescheduleIfNecessary()が呼ばれた場合にのみスケジュールされる。以下のケースでスケジュールされる：
1. createPersistentTask()後に未割り当てタスクが存在する場合
2. reassignPersistentTasks()完了後に未割り当てタスクが残っている場合
3. reassignPersistentTasks()が失敗した場合（NotClusterManagerExceptionを除く）

clusterChanged()でクラスタマネージャでなくなった場合はperiodicRechecker.cancel()で停止される。

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | デフォルト値 | 説明 |
|-------------|-----|-----|-------------|------|
| cluster.persistent_tasks.allocation.recheck_interval | TimeValue | No | 30s | 再チェック間隔（最小10秒） |

### 入力データソース

| データソース | 形式 | 説明 |
|-------------|------|------|
| PersistentTasksCustomMetadata | クラスタメタデータ | 永続タスクの割り当て状態 |
| DiscoveryNodes | クラスタ状態 | 利用可能なノード一覧 |
| PersistentTasksExecutorRegistry | レジストリ | タスク種別ごとのExecutor |

## 出力仕様

### 出力データ

| 出力先 | 形式 | 説明 |
|-------|------|------|
| ClusterState | クラスタ状態更新 | 永続タスクの割り当て情報の更新 |

### 出力ファイル仕様

ファイル出力はなし。

## 処理フロー

### 処理シーケンス

```
1. PeriodicRechecker.runInternal()実行
   └─ ローカルノードがクラスタマネージャか確認
2. isAnyTaskUnassigned()チェック
   └─ 未割り当てタスクがなければ終了
3. reassignPersistentTasks()呼び出し
   └─ ClusterStateUpdateTaskを投入
4. reassignTasks()実行（ClusterState更新ロジック）
   ├─ 全PersistentTaskをイテレーション
   ├─ needsReassignment()で再割り当て必要性判定
   │   └─ タスクが未割り当て、または割り当て先ノードが存在しない場合true
   ├─ createAssignment()で新しい割り当てを決定
   │   ├─ EnableAssignmentDeciderで割り当て可否チェック
   │   └─ PersistentTasksExecutor.getAssignment()で割り当て先決定
   └─ 割り当てが変更される場合、ClusterStateを更新
5. clusterStateProcessed()コールバック
   └─ 未割り当てタスクが残っている場合、periodicRechecker.rescheduleIfNecessary()
6. onFailure()コールバック
   └─ NotClusterManagerException以外の場合、periodicRechecker.rescheduleIfNecessary()
```

### フローチャート

```mermaid
flowchart TD
    A[PeriodicRechecker.runInternal] --> B{クラスタマネージャ?}
    B -->|No| C[終了]
    B -->|Yes| D{未割り当てタスクあり?}
    D -->|No| C
    D -->|Yes| E[reassignPersistentTasks]
    E --> F[ClusterStateUpdateTask投入]
    F --> G[各タスクを評価]
    G --> H{再割り当て必要?}
    H -->|No| I[次のタスク]
    H -->|Yes| J[createAssignment]
    J --> K{割り当て変更あり?}
    K -->|Yes| L[ClusterState更新]
    K -->|No| I
    L --> I
    I --> M{未割り当て残存?}
    M -->|Yes| N[再スケジュール]
    M -->|No| O[完了]
```

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

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

| 処理 | 対象 | 操作種別 | 概要 |
|-----|------|---------|------|
| タスク再割り当て | PersistentTasksCustomMetadata | UPDATE | タスクの割り当て情報を更新 |

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

#### PersistentTasksCustomMetadata

| 操作 | 項目 | 更新値・取得条件 | 備考 |
|-----|------|-----------------|------|
| UPDATE | assignment | 新しいAssignment（executorNode, reason） | needsReassignment()がtrueかつ新しい割り当てが異なる場合 |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| なし | NotClusterManagerException | クラスタマネージャ変更中の再割り当て | 再スケジュールせず終了 |
| なし | Exception | reassignPersistentTasks()の失敗 | 警告ログ出力、periodicRechecker.rescheduleIfNecessary() |

### リトライ仕様

| 項目 | 内容 |
|-----|------|
| リトライ回数 | 明示的なリトライなし（PeriodicRecheckerにより自動再試行） |
| リトライ間隔 | recheck_interval（デフォルト30秒） |
| リトライ対象エラー | NotClusterManagerException以外の全例外 |

### 障害時対応

再割り当てに繰り返し失敗する場合は、タスクの割り当て条件や対象ノードの状態を確認する。EnableAssignmentDeciderの設定（cluster.persistent_tasks.allocation.enable）でタスク割り当てが無効化されている可能性もある。

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

| 項目 | 内容 |
|-----|------|
| トランザクション範囲 | ClusterStateUpdateTask全体 |
| コミットタイミング | ClusterState更新時 |
| ロールバック条件 | execute()内で例外発生時はonFailure()が呼ばれ、状態更新は行われない |

## パフォーマンス要件

| 項目 | 内容 |
|-----|------|
| 想定処理件数 | 永続タスク数（通常は少数） |
| 目標処理時間 | ミリ秒単位 |
| メモリ使用量上限 | 微小（タスク割り当て情報のみ） |

## 排他制御

PeriodicRecheckerはAbstractAsyncTaskを継承しており、内部でスケジュール管理を行う。clusterChanged()でのperiodicRechecker.cancel()とrescheduleIfNecessary()は、AbstractAsyncTaskの内部同期メカニズムにより安全に動作する。reassignPersistentTasks()はClusterStateUpdateTaskとして投入されるため、クラスタ状態の更新は直列化される。

## ログ出力

| ログ種別 | 出力タイミング | 出力内容 |
|---------|--------------|---------|
| トレースログ | 再チェック実行時 | "periodic persistent task assignment check running for cluster state {version}" |
| トレースログ | clusterChanged時 | "checking task reassignment for cluster state {version}" |
| トレースログ | タスク再割り当て時 | "reassigning task {id} from node {old} to node {new}" |
| トレースログ | 割り当て変更なし | "ignoring task {id} because assignment is the same {assignment}" |
| 警告ログ | 再割り当て失敗時 | "failed to reassign persistent tasks" |

## 監視・アラート

| 監視項目 | 閾値 | アラート先 |
|---------|-----|----------|
| 未割り当て永続タスク数 | 0（理想） | PersistentTasksCustomMetadata |
| 再割り当て失敗回数 | 連続失敗 | OpenSearchログ |

## 備考

- 実装ファイル: `server/src/main/java/org/opensearch/persistent/PersistentTasksClusterService.java`
- PeriodicRecheckerはPersistentTasksClusterService内部のインナークラスとしてAbstractAsyncTaskを継承して実装
- クラスタマネージャノードでのみClusterStateListenerが登録される（DiscoveryNode.isClusterManagerNode()で判定）
- recheck_intervalは動的に変更可能（Property.Dynamic）
- 最小値は10秒
- mustReschedule()はtrueを返すが、実際のスケジュールはrescheduleIfNecessary()の呼び出しに依存
- EnableAssignmentDeciderのcanAssign()がNOを返すと、タスクは未割り当てのまま保持される
- ClusterManagerTaskThrottlerによるスロットリングが適用される（createPersistentTaskKey等）
