# 機能設計書 68-Kubernetes

## 概要

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

### 本機能の処理概要

Kubernetesデプロイメント機能は、FlinkクラスターをKubernetesクラスター上にネイティブにデプロイするための機能である。Kubernetes APIを使用してJobManager/TaskManagerのPodを動的に作成・管理する。

**業務上の目的・背景**：Kubernetesはコンテナオーケストレーションの業界標準であり、多くのクラウドベンダーがマネージドKubernetesサービスを提供している。本機能によりFlinkをKubernetes上でネイティブに実行でき、コンテナベースのインフラと統合可能になる。

**機能の利用シーン**：
- クラウドネイティブ環境でのFlink実行
- Kubernetesの自動スケーリング機能との統合
- コンテナベースのCI/CDパイプラインでのFlink運用
- マイクロサービスアーキテクチャとの統合

**主要な処理内容**：
1. Kubernetes APIクライアントを使用したPod/Service/ConfigMap作成
2. JobManager DeploymentとServiceの作成
3. TaskManager Podの動的作成
4. Pod Templateによるカスタマイズ
5. クラスターリソースのライフサイクル管理

**関連システム・外部連携**：
- Kubernetes API Server
- コンテナレジストリ（Docker Hub等）
- Kubernetes Secrets/ConfigMap

**権限による制御**：Kubernetes RBAC（ServiceAccount）

## 関連画面

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

## 機能種別

クラスターデプロイメント（Kubernetesネイティブ）

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| kubernetes.cluster-id | String | Yes | クラスターID（45文字以下） | 小文字英数字と"-"のみ |
| kubernetes.namespace | String | No | 名前空間（デフォルト: default） | - |
| kubernetes.context | String | No | kubeconfigコンテキスト | - |
| kubernetes.config.file | String | No | kubeconfigファイルパス | - |
| kubernetes.container.image.ref | String | No | Flinkコンテナイメージ | - |
| kubernetes.container.image.pull-policy | Enum | No | イメージプルポリシー（デフォルト: IfNotPresent） | Always/IfNotPresent/Never |
| kubernetes.rest-service.exposed.type | Enum | No | REST Serviceタイプ（デフォルト: ClusterIP） | ClusterIP/NodePort/LoadBalancer |
| kubernetes.jobmanager.cpu.amount | Double | No | JM CPU数（デフォルト: 1.0） | 正数 |
| kubernetes.taskmanager.cpu.amount | Double | No | TM CPU数（デフォルト: スロット数） | 正数 |
| kubernetes.jobmanager.service-account | String | No | JM ServiceAccount（デフォルト: default） | - |
| kubernetes.taskmanager.service-account | String | No | TM ServiceAccount（デフォルト: default） | - |
| kubernetes.entry.path | String | No | エントリーポイントスクリプト | デフォルト: /docker-entrypoint.sh |
| kubernetes.flink.conf.dir | String | No | 設定ディレクトリ | デフォルト: /opt/flink/conf |

### 入力データソース

- Flink設定ファイル（flink-conf.yaml）
- Kubeconfig（~/.kube/config）
- Pod Template YAML

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| clusterId | String | Kubernetesクラスター識別子 |
| WebInterfaceURL | String | Flink Web UIのURL |
| RestEndpoint | Endpoint | REST APIエンドポイント |

### 出力先

- Kubernetes API Server（リソース作成）
- クライアントコンソール（ログ出力）

## 処理フロー

### 処理シーケンス

```
1. クラスターディスクリプタ初期化
   └─ FlinkKubeClient作成
   └─ clusterId検証

2. クラスターデプロイ（deployClusterInternal）
   └─ 実行モード設定（DETACHED/NORMAL）
   └─ ポート固定値設定
   └─ HA設定（有効な場合）

3. JobManager作成
   └─ Pod Template読み込み（オプション）
   └─ KubernetesJobManagerParameters構築
   └─ KubernetesJobManagerFactory.buildKubernetesJobManagerSpecification()
   └─ client.createJobManagerComponent()

4. クライアント接続
   └─ RestEndpoint取得
   └─ RestClusterClient作成

5. TaskManager動的起動（ResourceManagerDriver）
   └─ Pod作成リクエスト
   └─ コンテナ起動
```

### フローチャート

```mermaid
flowchart TD
    A[開始: deploySessionCluster/deployApplicationCluster] --> B{クラスター存在?}
    B -->|Yes| C[ClusterDeploymentException]
    B -->|No| D[deployClusterInternal]
    D --> E[ポート設定固定化]
    E --> F{HA有効?}
    F -->|Yes| G[HA設定追加]
    F -->|No| H[Pod Template読み込み]
    G --> H
    H --> I[KubernetesJobManagerFactory]
    I --> J[client.createJobManagerComponent]
    J --> K{作成成功?}
    K -->|No| L[クリーンアップ]
    L --> C
    K -->|Yes| M[RestEndpoint取得]
    M --> N[RestClusterClient返却]
    N --> Z[終了]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-01 | ClusterID必須 | kubernetes.cluster-idは必須設定 | 常時 |
| BR-02 | ClusterID形式 | 45文字以下、小文字英数字と"-"のみ | 常時 |
| BR-03 | ポート固定化 | RPC/Blob/RESTポートは固定値に設定 | デプロイ時 |
| BR-04 | 既存クラスター検出 | 同名Serviceが存在する場合はエラー | Application Cluster時 |
| BR-05 | ServiceAccount使用 | デフォルトは"default" ServiceAccount | 常時 |
| BR-06 | イメージプルポリシー | デフォルトはIfNotPresent | 常時 |

### 計算ロジック

CPU/メモリリミット計算:
```
limit_cpu = cpu * cpu_limit_factor
limit_memory = memory * memory_limit_factor
```

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

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

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| ClusterDeploymentException | デプロイエラー | 同名クラスター存在 | 別のclusterIdを使用 |
| ClusterDeploymentException | 設定エラー | 不正なdeployment.target | 設定修正 |
| ClusterRetrieveException | 取得エラー | RestEndpoint取得失敗 | Service状態確認 |
| FlinkException | 削除エラー | クラスター削除失敗 | 手動クリーンアップ |

### リトライ仕様

- トランザクショナル操作は最大15回リトライ
- 初期リトライ間隔: 50ms
- 最大リトライ間隔: 1分

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

- ConfigMap操作はKubernetesのoptimistic lockingを使用
- リソース作成は逐次実行

## パフォーマンス要件

- Fabric8 Kubernetes Clientを使用
- 非同期API呼び出しによる効率化
- ConfigMapベースのHA（ZooKeeperなし可能）

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

- Kubernetes RBAC（Role-Based Access Control）対応
- ServiceAccountによる権限制御
- Secretsのマウント対応
- imagePullSecretsによるプライベートレジストリアクセス

## 備考

- Session ClusterとApplication Clusterの2つのデプロイモードをサポート
- Pod Templateによる高度なカスタマイズが可能
- Owner Referenceによるカスケード削除対応

---

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

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

### 推奨読解順序

#### Step 1: 設定オプションを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | KubernetesConfigOptions.java | `flink-kubernetes/src/main/java/org/apache/flink/kubernetes/configuration/KubernetesConfigOptions.java` | K8s固有の設定項目定義 |
| 1-2 | KubernetesDeploymentTarget.java | `flink-kubernetes/src/main/java/org/apache/flink/kubernetes/configuration/KubernetesDeploymentTarget.java` | デプロイモード（Session/Application） |

**読解のコツ**: ConfigOptionで定義される各設定項目の意味とデフォルト値を確認。特にCLUSTER_ID、NAMESPACE、CONTAINER_IMAGE、REST_SERVICE_EXPOSED_TYPE等が重要。

#### Step 2: クラスターディスクリプタを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | KubernetesClusterDescriptor.java | `flink-kubernetes/src/main/java/org/apache/flink/kubernetes/KubernetesClusterDescriptor.java` | メインデプロイメントクラス |

**主要処理フロー（KubernetesClusterDescriptor.java）**:
- **70-98行目**: クラス定義とコンストラクタ（FlinkKubeClient作成）
- **105-139行目**: createClusterClientProvider()でRestEndpoint取得
- **157-169行目**: retrieve()で既存クラスター取得
- **172-187行目**: deploySessionCluster()でセッションクラスター起動
- **190-247行目**: deployApplicationCluster()でアプリケーションクラスター起動
- **249-311行目**: deployClusterInternal()でメインデプロイメント処理
- **313-320行目**: killCluster()でクラスター削除
- **322-329行目**: close()でクライアントクローズ

#### Step 3: Kubernetesクライアントを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | FlinkKubeClient.java | `flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/FlinkKubeClient.java` | K8sクライアントインターフェース |
| 3-2 | Fabric8FlinkKubeClient.java | `flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClient.java` | Fabric8実装 |

#### Step 4: JobManagerファクトリを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | KubernetesJobManagerFactory.java | `flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/factory/KubernetesJobManagerFactory.java` | JM Pod/Service生成 |
| 4-2 | KubernetesJobManagerParameters.java | `flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesJobManagerParameters.java` | JMパラメータ |

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

```
KubernetesClusterDescriptor
    │
    ├─ deploySessionCluster() / deployApplicationCluster()
    │      │
    │      └─ deployClusterInternal()
    │             │
    │             ├─ ポート設定固定化
    │             │      └─ KubernetesUtils.checkAndUpdatePortConfigOption()
    │             │
    │             ├─ KubernetesJobManagerParameters
    │             │      └─ ClusterSpecification
    │             │
    │             ├─ Pod Template読み込み
    │             │      └─ KubernetesUtils.loadPodFromTemplateFile()
    │             │
    │             └─ KubernetesJobManagerFactory
    │                    │
    │                    └─ buildKubernetesJobManagerSpecification()
    │                           │
    │                           ├─ Deployment
    │                           ├─ Service (internal/external)
    │                           └─ ConfigMap
    │
    └─ FlinkKubeClient.createJobManagerComponent()
           │
           └─ Fabric8FlinkKubeClient
                  │
                  ├─ createDeployment()
                  ├─ createService()
                  └─ createConfigMap()
```

### データフロー図

```
[クライアント側]                                          [Kubernetes側]

flink-conf.yaml ─────┐
                     │
~/.kube/config ──────┼──▶ KubernetesClusterDescriptor
                     │           │
Pod Template ────────┘           │
                                 ▼
                    ┌────────────────────────┐
                    │  FlinkKubeClient       │
                    │  - Fabric8 Client      │
                    └───────────┬────────────┘
                                │
                    ┌───────────▼────────────┐           ┌─────────────────┐
                    │  Kubernetes API        │───────────│  Namespace      │
                    │  Server                │           │                 │
                    └────────────────────────┘           │ ┌─────────────┐ │
                                                         │ │ Deployment  │ │
                                                         │ │ (JobManager)│ │
                                                         │ └──────┬──────┘ │
                                                         │        │        │
                                                         │ ┌──────▼──────┐ │
                                                         │ │ Service     │ │
                                                         │ │ (REST)      │ │
                                                         │ └─────────────┘ │
                                                         │                 │
                                                         │ ┌─────────────┐ │
                                                         │ │ Pod         │ │
                                                         │ │(TaskManager)│ │
                                                         │ └─────────────┘ │
                                                         └─────────────────┘
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| KubernetesClusterDescriptor.java | `flink-kubernetes/src/main/java/org/apache/flink/kubernetes/` | ソース | メインデプロイメントクラス |
| KubernetesConfigOptions.java | `flink-kubernetes/src/main/java/org/apache/flink/kubernetes/configuration/` | ソース | K8s設定オプション |
| KubernetesDeploymentTarget.java | `flink-kubernetes/src/main/java/org/apache/flink/kubernetes/configuration/` | ソース | デプロイモード定義 |
| FlinkKubeClient.java | `flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/` | ソース | K8sクライアントインターフェース |
| Fabric8FlinkKubeClient.java | `flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/` | ソース | Fabric8実装 |
| KubernetesJobManagerFactory.java | `flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/factory/` | ソース | JM Pod/Service生成 |
| KubernetesTaskManagerFactory.java | `flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/factory/` | ソース | TM Pod生成 |
| KubernetesSessionClusterEntrypoint.java | `flink-kubernetes/src/main/java/org/apache/flink/kubernetes/entrypoint/` | ソース | セッションモードエントリーポイント |
| KubernetesApplicationClusterEntrypoint.java | `flink-kubernetes/src/main/java/org/apache/flink/kubernetes/entrypoint/` | ソース | アプリケーションモードエントリーポイント |
| KubernetesResourceManagerDriver.java | `flink-kubernetes/src/main/java/org/apache/flink/kubernetes/` | ソース | TaskManager管理 |
