# 機能設計書 70-コンテナ

## 概要

本ドキュメントは、Apache Flinkのコンテナデプロイメント機能（flink-container モジュール）の設計仕様を記載する。

### 本機能の処理概要

コンテナデプロイメント機能は、Dockerコンテナ等のコンテナ環境でFlinkアプリケーションを実行するためのエントリーポイントを提供する。StandaloneApplicationClusterEntryPointにより、ジョブを埋め込んだコンテナイメージからFlinkクラスターを起動できる。

**業務上の目的・背景**：コンテナ技術の普及に伴い、アプリケーションとランタイムを一体化したイミュータブルなデプロイメントが標準的になっている。本機能によりFlinkジョブをDockerイメージとしてパッケージ化し、Kubernetes、Docker Compose、Mesos等の環境でデプロイできる。

**機能の利用シーン**：
- Dockerイメージへのジョブ埋め込み
- Kubernetes上でのApplicationモード実行
- CI/CDパイプラインでのFlinkジョブデプロイ
- マイクロサービスとしてのFlinkジョブ運用

**主要な処理内容**：
1. コマンドライン引数のパース
2. アーティファクト（jarファイル）のフェッチ
3. PackagedProgramの構築
4. ApplicationClusterEntryPointとしての起動
5. StandaloneResourceManagerによるリソース管理

**関連システム・外部連携**：
- Docker/コンテナランタイム
- Kubernetes（kubectl apply等）
- アーティファクトストレージ（HDFS、S3等）

**権限による制御**：コンテナ実行環境の権限

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| - | - | - | 本機能は画面を持たないインフラ機能 |

## 機能種別

クラスターエントリーポイント（コンテナアプリケーションモード）

## 入力仕様

### 入力パラメータ（コマンドライン引数）

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| --configDir, -c | String | Yes | Flink設定ディレクトリ | - |
| --job-classname, -j | String | No | ジョブのメインクラス名 | - |
| --job-id, -jid | String | No | ジョブID（Hex形式） | JobID.fromHexString() |
| --jars | String[] | No | jarファイルパス（カンマ区切り） | - |
| -s, --fromSavepoint | String | No | セーブポイントパス | - |
| --allowNonRestoredState | Boolean | No | 復元不可状態を許可 | - |
| -D | Properties | No | 動的プロパティ | - |

### 入力データソース

- 設定ファイル（conf/config.yaml）
- jarファイル（ローカルまたはリモート）
- セーブポイント（オプション）

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| Flink Cluster | Process | 起動されたFlinkクラスター |
| JobGraph | Object | 実行されるジョブグラフ |
| Web UI | HTTP | Flink Web インターフェース |

### 出力先

- コンソール（ログ出力）
- REST API（Web UI）

## 処理フロー

### 処理シーケンス

```
1. 起動処理（main）
   └─ EnvironmentInformation.logEnvironmentInfo()
   └─ SignalHandler.register()
   └─ JvmShutdownSafeguard.installAsShutdownHook()

2. 設定パース
   └─ ClusterEntrypointUtils.parseParametersOrExit()
   └─ StandaloneApplicationClusterConfigurationParserFactory

3. 設定ロード
   └─ loadConfigurationFromClusterConfig()
   └─ SavepointRestoreSettings適用
   └─ JobID設定（オプション）

4. プラグイン初期化
   └─ PluginUtils.createPluginManagerFromRootFolder()
   └─ FileSystem.initialize()

5. セキュリティコンテキスト
   └─ installSecurityContext()
   └─ getPackagedProgram()

6. アーティファクトフェッチ
   └─ ArtifactFetchManager.fetchArtifacts()

7. PackagedProgram構築
   └─ DefaultPackagedProgramRetriever.create()
   └─ programRetriever.getPackagedProgram()

8. エントリーポイント起動
   └─ new StandaloneApplicationClusterEntryPoint()
   └─ ClusterEntrypoint.runClusterEntrypoint()
```

### フローチャート

```mermaid
flowchart TD
    A[開始: main] --> B[環境情報ログ出力]
    B --> C[コマンドライン引数パース]
    C --> D[設定ロード]
    D --> E[プラグインマネージャー初期化]
    E --> F[FileSystem初期化]
    F --> G[セキュリティコンテキストインストール]
    G --> H{jarファイルあり?}
    H -->|Yes| I[ArtifactFetchManager.fetchArtifacts]
    H -->|No| J[userLibDir使用]
    I --> K[PackagedProgramRetriever.create]
    J --> K
    K --> L[PackagedProgram取得]
    L --> M[configureExecution]
    M --> N[StandaloneApplicationClusterEntryPoint作成]
    N --> O[runClusterEntrypoint]
    O --> Z[終了]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-01 | configDir必須 | --configDirは必須パラメータ | 起動時 |
| BR-02 | JobID形式 | JobIDはHex文字列形式 | --job-id指定時 |
| BR-03 | jarsカンマ区切り | 複数jarはカンマで区切る | --jars指定時 |
| BR-04 | Reactive Mode対応 | supportsReactiveMode()はtrue | 常時 |
| BR-05 | StandaloneResourceManager | リソースマネージャーはStandalone版 | 常時 |
| BR-06 | Recovery Claim Mode警告 | LEGACYモードは非推奨 | LEGACY使用時 |

### 計算ロジック

特に複雑な計算ロジックはなし。設定とjarからPackagedProgramを構築。

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

本機能はデータベースを使用しない。

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| FlinkParseException | パースエラー | コマンドライン引数不正 | 引数確認 |
| - | プログラム作成エラー | PackagedProgram作成失敗 | jar/クラス名確認 |
| - | 設定適用エラー | configureExecution失敗 | 設定確認 |
| IllegalArgumentException | JobIDエラー | 不正なJobID形式 | Hex形式確認 |

### リトライ仕様

- アーティファクトフェッチはArtifactFetchManagerの設定に依存
- クラスターエントリーポイントは再起動で対応

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

- ジョブ実行はFlinkのチェックポイント機構で保護
- セーブポイントからの復元をサポート

## パフォーマンス要件

- 起動時間はjar取得とクラスロードに依存
- Reactiveモードによる動的スケーリング対応

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

- セキュリティコンテキストによる認証
- アーティファクト取得時のアクセス制御
- コンテナ環境のセキュリティ設定

## 備考

- ApplicationClusterEntryPointを継承
- StandaloneResourceManagerFactory使用（YARN/K8s非依存）
- flink-daemon.sh standalonejobからも起動可能

---

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

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

### 推奨読解順序

#### Step 1: 設定クラスを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | StandaloneApplicationClusterConfiguration.java | `flink-container/src/main/java/org/apache/flink/container/entrypoint/` | 設定データクラス |
| 1-2 | StandaloneApplicationClusterConfigurationParserFactory.java | `flink-container/src/main/java/org/apache/flink/container/entrypoint/` | 引数パーサー |

**読解のコツ（StandaloneApplicationClusterConfiguration.java）**:
- **35-42行目**: フィールド定義（SavepointRestoreSettings、JobID、jobClassName、jars）
- **43-57行目**: コンストラクタ
- **59-81行目**: ゲッターメソッド

**読解のコツ（StandaloneApplicationClusterConfigurationParserFactory.java）**:
- **46-72行目**: コマンドラインオプション定義（JOB_CLASS_NAME、JOB_ID、JARS）
- **74-86行目**: getOptions()でオプション一覧
- **88-108行目**: createResult()でパース結果生成

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | StandaloneApplicationClusterEntryPoint.java | `flink-container/src/main/java/org/apache/flink/container/entrypoint/` | メインエントリーポイント |

**主要処理フロー（StandaloneApplicationClusterEntryPoint.java）**:
- **49-54行目**: クラス定義（ApplicationClusterEntryPoint継承）
- **51-54行目**: コンストラクタ（StandaloneResourceManagerFactory使用）
- **56-107行目**: main()メソッド
  - **58-61行目**: 環境情報ログ・シグナルハンドラ
  - **63-67行目**: 引数パース
  - **69行目**: 設定ロード
  - **82-93行目**: PackagedProgram取得
  - **96-101行目**: 設定適用
  - **103-106行目**: エントリーポイント起動
- **109-112行目**: supportsReactiveMode() = true
- **115-122行目**: loadConfigurationFromClusterConfig()
- **124-150行目**: getPackagedProgram()（アーティファクトフェッチ含む）

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

```
StandaloneApplicationClusterEntryPoint.main()
    │
    ├─ EnvironmentInformation.logEnvironmentInfo()
    │
    ├─ ClusterEntrypointUtils.parseParametersOrExit()
    │      └─ StandaloneApplicationClusterConfigurationParserFactory
    │             └─ createResult() → StandaloneApplicationClusterConfiguration
    │
    ├─ loadConfigurationFromClusterConfig()
    │      ├─ loadConfiguration()
    │      ├─ setStaticJobId()
    │      └─ SavepointRestoreSettings.toConfiguration()
    │
    ├─ PluginUtils.createPluginManagerFromRootFolder()
    │
    ├─ FileSystem.initialize()
    │
    ├─ installSecurityContext()
    │
    ├─ getPackagedProgram()
    │      ├─ ArtifactFetchManager.fetchArtifacts()
    │      └─ DefaultPackagedProgramRetriever.create()
    │             └─ getPackagedProgram()
    │
    ├─ configureExecution()
    │
    └─ ClusterEntrypoint.runClusterEntrypoint()
           │
           └─ ApplicationClusterEntryPoint
                  │
                  └─ StandaloneResourceManagerFactory
```

### データフロー図

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

コマンドライン引数 ──┐
                    │
                    ▼
           ┌─────────────────────────────┐
           │ ConfigurationParserFactory │
           │   └─ parseParametersOrExit()│
           └────────────┬────────────────┘
                        │
config.yaml ────────────┼───▶ Configuration
                        │
                        ▼
           ┌─────────────────────────────┐
           │ getPackagedProgram()       │
           │   ├─ ArtifactFetchManager  │◀── jar files (local/remote)
           │   └─ PackagedProgramRetriever│
           └────────────┬────────────────┘
                        │
                        ▼
           ┌─────────────────────────────┐
           │ StandaloneApplicationCluster│
           │ EntryPoint                 │
           │   └─ runClusterEntrypoint() │───▶ Flink Cluster
           └─────────────────────────────┘         │
                                                   ├── JobManager
                                                   ├── TaskManager(s)
                                                   └── Web UI
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| StandaloneApplicationClusterEntryPoint.java | `flink-container/src/main/java/org/apache/flink/container/entrypoint/` | ソース | メインエントリーポイント |
| StandaloneApplicationClusterConfiguration.java | `flink-container/src/main/java/org/apache/flink/container/entrypoint/` | ソース | 設定データクラス |
| StandaloneApplicationClusterConfigurationParserFactory.java | `flink-container/src/main/java/org/apache/flink/container/entrypoint/` | ソース | 引数パーサー |
| ApplicationClusterEntryPoint.java | `flink-clients/src/main/java/org/apache/flink/client/deployment/application/` | ソース | 基底クラス |
| StandaloneResourceManagerFactory.java | `flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/` | ソース | リソースマネージャーファクトリ |
| DefaultPackagedProgramRetriever.java | `flink-clients/src/main/java/org/apache/flink/client/program/` | ソース | プログラムリトリーバー |
| ArtifactFetchManager.java | `flink-clients/src/main/java/org/apache/flink/client/program/artifact/` | ソース | アーティファクト取得 |
| ClusterEntrypointUtils.java | `flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/` | ソース | エントリーポイントユーティリティ |
