# 通知設計書 17-PersistentTasksNodeService通知

## 概要

本ドキュメントは、永続タスク（Persistent Tasks）のクラスタ状態変更を監視し、ノード上のタスクライフサイクル（開始・停止・再割り当て）を管理・通知するPersistentTasksNodeService通知の設計を記述する。

### 本通知の処理概要

PersistentTasksNodeServiceは、ClusterStateListenerを実装してクラスタ状態の変化を監視し、永続タスクの割り当て状態に応じてローカルノード上のタスクの開始・キャンセル・削除を管理する。タスクの失敗時にはクラスタマネージャへの通知も行う。

**業務上の目的・背景**：OpenSearchの永続タスク（例：クロスクラスタレプリケーション、MLタスク等）は、クラスタ状態として管理され、クラスタマネージャによって特定のノードに割り当てられる。各ノードは自身に割り当てられたタスクを実行し、タスクの完了・失敗・キャンセルを適切に管理する必要がある。PersistentTasksNodeServiceはこの分散タスク管理のノード側コンポーネントである。

**通知の送信タイミング**：clusterChanged(ClusterChangedEvent)がトリガーとなり、クラスタ状態が変化するたびに呼び出される。状態がSTATE_NOT_RECOVERED_BLOCKでブロックされている場合はスキップされる。

**通知の受信者**：(1) PersistentTasksExecutor（タスク実行者）がタスク開始・キャンセルの通知を受け取る。(2) クラスタマネージャがタスク失敗の通知を受け取る（PersistentTasksService経由）。

**通知内容の概要**：タスクの開始時はタスクパラメータと状態が渡される。タスク失敗時は例外情報がクラスタマネージャに送信される。タスクキャンセル時はキャンセル理由が通知される。

**期待されるアクション**：タスク開始 - PersistentTasksExecutorがタスクを実行。タスク失敗 - クラスタマネージャがタスクを再割り当てまたは削除。タスクキャンセル - TaskManagerがローカルタスクをキャンセル。

## 通知種別

プロセス内コールバック通知 + クラスタ間通知（ActionListener経由）

## 送信仕様

### 基本情報

| 項目 | 内容 |
|-----|------|
| 送信方式 | 同期（clusterChanged内）+ 非同期（クラスタマネージャ通知） |
| 優先度 | 高（タスクライフサイクル管理） |
| リトライ | なし（失敗時はログ出力） |

### 送信先決定ロジック

クラスタ状態のPersistentTasksCustomMetadataから、ローカルノードに割り当てられたタスクを抽出し、ローカルのrunningTasksマップと比較して差分処理を行う。

## 通知テンプレート

### メール通知の場合

該当なし。

### 本文テンプレート

```
該当なし（Javaオブジェクト渡し）
```

### 添付ファイル

該当なし。

## テンプレート変数

| 変数名 | 説明 | データ取得元 | 必須 |
|--------|------|-------------|-----|
| taskInProgress.getTaskName() | タスク名 | PersistentTasksCustomMetadata | Yes |
| taskInProgress.getId() | タスクID | PersistentTasksCustomMetadata | Yes |
| taskInProgress.getAllocationId() | 割り当てID | PersistentTasksCustomMetadata | Yes |
| taskInProgress.getParams() | タスクパラメータ | PersistentTasksCustomMetadata | Yes |
| taskInProgress.getState() | タスク状態 | PersistentTasksCustomMetadata | No |

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| クラスタ状態変化 | clusterChanged() | タスクメタデータの変更またはノード変更 | PersistentTasksCustomMetadataまたはノード構成が変化した場合 |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| STATE_NOT_RECOVERED_BLOCK | ゲートウェイがディスクから回復していない場合はスキップ |
| タスクメタデータ変更なし | 前回と同じメタデータでノード変更もない場合は処理をスキップ |

## 処理フロー

### 送信フロー

```mermaid
flowchart TD
    A[clusterChanged イベント] --> B{STATE_NOT_RECOVERED_BLOCK?}
    B -->|Yes| C[スキップ]
    B -->|No| D[PersistentTasksCustomMetadata取得]
    D --> E{メタデータ変更あり?}
    E -->|No| C
    E -->|Yes| F[ローカルノードの割り当てタスクを走査]
    F --> G{ローカルで未実行?}
    G -->|Yes| H[startTask: タスク開始]
    G -->|No| I[notVisitedTasksから除外]
    H --> J[TaskManager登録]
    J --> K[PersistentTasksExecutor実行]
    F --> L[notVisitedTasksを処理]
    L --> M{completed?}
    M -->|Yes| N[runningTasksから削除]
    M -->|No| O[cancelTask: キャンセル]
    O --> P[PersistentTasksService.sendCancelRequest]
```

## データベース参照・更新仕様

### 参照テーブル一覧

| テーブル名 | 用途 | 備考 |
|-----------|------|------|
| PersistentTasksCustomMetadata | タスク割り当て情報の参照 | クラスタ状態メタデータ |

### テーブル別参照項目詳細

#### PersistentTasksCustomMetadata

| 参照項目（カラム名） | 用途 | 取得条件 |
|-------------------|------|---------|
| tasks() | 全永続タスク一覧 | なし |
| getExecutorNode() | タスク実行ノード | ローカルノードIDと一致 |
| getAllocationId() | タスク割り当てID | なし |

### 更新テーブル一覧

| テーブル名 | 操作 | 概要 |
|-----------|------|------|
| runningTasks (Map) | PUT/REMOVE | ローカル実行中タスクの管理 |

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 対処方法 |
|----------|---------|---------|
| タスク開始失敗 | startTask()で例外発生 | ログ出力、クラスタマネージャに失敗通知（notifyMasterOfFailedTask） |
| タスク登録失敗 | TaskManager.register()で例外発生 | ログ出力、クラスタマネージャに失敗通知、return |
| タスク実行失敗 | executeTask()で例外発生 | task.markAsFailed(e)を呼び出し |
| キャンセル失敗 | sendCancelRequest()の応答で失敗 | 警告ログのみ出力 |
| 失敗通知の送信失敗 | sendCompletionRequest()の応答で失敗 | 警告ログのみ出力 |

### リトライ仕様

| 項目 | 内容 |
|-----|------|
| リトライ回数 | なし |
| リトライ間隔 | なし |
| リトライ対象エラー | なし（次回のclusterChangedで再処理される可能性あり） |

## 配信設定

### レート制限

| 項目 | 内容 |
|-----|------|
| 1分あたり上限 | なし（クラスタ状態変化頻度に依存） |
| 1日あたり上限 | なし |

### 配信時間帯

制限なし。

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

- クラスタマネージャへの通知はトランスポート層を経由するため、TLS/SSL設定に依存する
- タスクパラメータにはアプリケーション固有の情報が含まれる場合がある

## 備考

- タスクの状態遷移：NULL -> STARTED -> COMPLETED、または NULL -> STARTED -> PENDING_CANCEL -> COMPLETED
- ゲートウェイ回復前はクラスタ状態変更を無視する（92-96行目）
- notifyMasterOfFailedTaskはActionListenerパターンで非同期にクラスタマネージャに通知
- Statusクラス（Task.Status実装）はタスクのシリアライズ可能な状態を表す

---

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

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

### 推奨読解順序

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

永続タスクの状態モデルとメタデータ構造を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | PersistentTasksNodeService.java | `server/src/main/java/org/opensearch/persistent/PersistentTasksNodeService.java` | 99-121行目のコメントで状態遷移モデル（Cluster State / Local State / Local Action）を理解 |

**読解のコツ**: コメント内の状態遷移表を参照し、クラスタ状態とローカル状態の組み合わせによるアクション決定を理解する。

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

clusterChanged()メソッドの処理フローを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | PersistentTasksNodeService.java | `server/src/main/java/org/opensearch/persistent/PersistentTasksNodeService.java` | clusterChanged()（91行目〜）のゲートウェイ回復チェック、タスク走査、差分処理 |

**主要処理フロー**:
1. **92-96行目**: STATE_NOT_RECOVERED_BLOCKチェック
2. **97-98行目**: 現在と前回のPersistentTasksCustomMetadata取得
3. **123行目**: メタデータ変更またはノード変更の判定
4. **128-153行目**: ローカルノードのタスク走査、新規タスク開始
5. **156-178行目**: 未訪問タスクの処理（完了なら削除、実行中ならキャンセル）

#### Step 3: タスク開始処理を理解する

startTask()の詳細な処理フローを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | PersistentTasksNodeService.java | `server/src/main/java/org/opensearch/persistent/PersistentTasksNodeService.java` | startTask()（184行目〜）でのTaskManager登録、初期化、実行委譲 |

**主要処理フロー**:
- **185-187行目**: PersistentTasksExecutorの取得
- **210行目**: TaskManager.register()でタスク登録
- **228行目**: task.init()でタスク初期化
- **237行目**: nodePersistentTasksExecutor.executeTask()で実行委譲

#### Step 4: 失敗通知とキャンセル処理を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | PersistentTasksNodeService.java | `server/src/main/java/org/opensearch/persistent/PersistentTasksNodeService.java` | notifyMasterOfFailedTask()（257行目〜）とcancelTask()（295行目〜） |

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

```
ClusterService.clusterChanged(ClusterChangedEvent)
    |
    +-- PersistentTasksNodeService.clusterChanged(event)
            |
            +-- [新規タスク] startTask(taskInProgress)
            |       |
            |       +-- PersistentTasksExecutorRegistry.getPersistentTaskExecutorSafe()
            |       +-- TaskManager.register()
            |       +-- AllocatedPersistentTask.init()
            |       +-- NodePersistentTasksExecutor.executeTask()
            |       |       |
            |       |       +-- PersistentTasksExecutor.nodeOperation()
            |       |
            |       +-- [失敗時] notifyMasterOfFailedTask()
            |               |
            |               +-- PersistentTasksService.sendCompletionRequest()
            |
            +-- [完了タスク] runningTasks.remove()
            |
            +-- [不明タスク] cancelTask(allocationId)
                    |
                    +-- AllocatedPersistentTask.markAsCancelled()
                    +-- PersistentTasksService.sendCancelRequest()
```

### データフロー図

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

ClusterChangedEvent          -->  clusterChanged()                --> startTask()
  PersistentTasksCustomMetadata       |                                  |
  previousTasks                   差分検出                          --> TaskManager登録
  nodesChanged                        |                            --> PersistentTasksExecutor実行
                                  状態遷移判定                      --> cancelTask()
                                                                   --> notifyMasterOfFailedTask()
                                                                        |
                                                                   --> PersistentTasksService
                                                                        (クラスタマネージャ通知)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| PersistentTasksNodeService.java | `server/src/main/java/org/opensearch/persistent/PersistentTasksNodeService.java` | ソース | 通知の主要実装 |
| PersistentTasksCustomMetadata.java | `server/src/main/java/org/opensearch/persistent/PersistentTasksCustomMetadata.java` | ソース | タスクメタデータ定義 |
| AllocatedPersistentTask.java | `server/src/main/java/org/opensearch/persistent/AllocatedPersistentTask.java` | ソース | 割り当て済みタスクの状態管理 |
| PersistentTasksService.java | `server/src/main/java/org/opensearch/persistent/PersistentTasksService.java` | ソース | クラスタマネージャ通知サービス |
| PersistentTasksExecutor.java | `server/src/main/java/org/opensearch/persistent/PersistentTasksExecutor.java` | ソース | タスク実行の抽象基底クラス |
| NodePersistentTasksExecutor.java | `server/src/main/java/org/opensearch/persistent/NodePersistentTasksExecutor.java` | ソース | ノード上でのタスク実行 |
| PersistentTasksNodeServiceTests.java | `server/src/test/java/org/opensearch/persistent/PersistentTasksNodeServiceTests.java` | テスト | ノードサービスのテスト |
