# 機能設計書 59-Prometheus

## 概要

本ドキュメントは、Apache Flink の Prometheus メトリクスレポーターの機能設計書である。flink-metrics-prometheus モジュールが提供する Prometheus へのメトリクスエクスポート機能について、処理内容、設定方法、およびコードの構造を詳細に記載する。

### 本機能の処理概要

**業務上の目的・背景**：Prometheus は Cloud Native Computing Foundation（CNCF）のプロジェクトであり、Kubernetes 環境における事実上の標準メトリクス収集システムである。Flink メトリクスを Prometheus にエクスポートすることで、Grafana でのダッシュボード可視化、AlertManager によるアラート通知など、モダンな可観測性スタックとの統合が可能になる。

**機能の利用シーン**：
- Kubernetes 上で Flink を運用する場合
- Grafana でメトリクスダッシュボードを構築する場合
- AlertManager でメトリクスベースのアラートを設定する場合
- Prometheus エコシステムとの統合が必要な場合

**主要な処理内容**：
1. PrometheusReporterFactory による PrometheusReporter インスタンスの生成
2. HTTP サーバーの起動（Prometheus がスクレイプするエンドポイント）
3. Prometheus クライアントライブラリを使用したメトリクス登録
4. Counter, Gauge, Histogram 用の Prometheus Collector 生成
5. ラベル（次元）の管理と不正文字の置換

**関連システム・外部連携**：
- Prometheus サーバー（スクレイピング）
- PushGateway（オプション）
- Grafana（可視化）

**権限による制御**：HTTP エンドポイントへのアクセス制御は外部（ネットワークポリシー等）に依存

## 関連画面

本機能は画面を持たないバックエンド機能である。

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| - | - | - | - |

## 機能種別

監視・可観測性 / メトリクスレポーター

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| port | String | No | HTTPサーバーポート（デフォルト: 9249） | ポート番号または範囲 |
| filterLabelValueCharacters | Boolean | No | ラベル値の不正文字をフィルタするか | true/false（デフォルト: true） |

### 入力データソース

- Flink Configuration（flink-conf.yaml）
- MetricConfig（レポーター設定）

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| HTTP エンドポイント | /metrics | Prometheus 形式のメトリクス |
| Counter/Gauge | Gauge | Prometheus Gauge として登録 |
| Meter | Gauge | rate として Gauge 登録 |
| Histogram | Summary | quantile 付き Summary |

### 出力先

HTTP エンドポイント（Prometheus がスクレイプ）

## 処理フロー

### 処理シーケンス

```
1. PrometheusReporterFactory.createMetricReporter(Properties) 呼び出し
   └─ ポート範囲を解析
2. PrometheusReporter コンストラクタ
   └─ 利用可能なポートで HTTPServer を起動
3. open(MetricConfig) でラベル値フィルタ設定を読み込み
4. notifyOfAddedMetric() でメトリクス追加
   └─ 次元キー・値を抽出
   └─ スコープ付きメトリクス名を生成
   └─ Prometheus Collector を生成・登録
   └─ メトリクスを Collector に追加
5. notifyOfRemovedMetric() でメトリクス削除
   └─ Collector からメトリクスを削除
   └─ 参照カウントが0になったら Collector を登録解除
6. close() で HTTPServer を停止
```

### フローチャート

```mermaid
flowchart TD
    A[開始: createMetricReporter] --> B[ポート範囲解析]
    B --> C[HTTPServer起動試行]
    C --> D{起動成功?}
    D -->|Yes| E[PrometheusReporter生成完了]
    D -->|No| F{次のポート有り?}
    F -->|Yes| C
    F -->|No| G[RuntimeException]
    E --> H[open: 設定読み込み]
    H --> I{メトリクス操作}
    I -->|追加| J[notifyOfAddedMetric]
    J --> K[次元キー・値抽出]
    K --> L[getScopedName]
    L --> M{Collector存在?}
    M -->|Yes| N[既存Collectorを使用]
    M -->|No| O[createCollector]
    O --> P[registry.register]
    N --> Q[addMetric]
    P --> Q
    Q --> R[参照カウント増加]
    R --> I
    I -->|削除| S[notifyOfRemovedMetric]
    S --> T[removeMetric]
    T --> U{参照カウント=0?}
    U -->|Yes| V[registry.unregister]
    U -->|No| I
    V --> I
    I -->|終了| W[close: HTTPServer停止]
    W --> X[終了]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-59-01 | メトリクス名プレフィックス | "flink_" プレフィックスを付与 | 常時 |
| BR-59-02 | 文字変換 | [a-zA-Z0-9:_] 以外は "_" に置換 | メトリクス名・ラベルキー |
| BR-59-03 | Histogram処理 | Flink Histogram は Prometheus Summary（quantile付き）として出力 | Histogram型 |
| BR-59-04 | 参照カウント | 同名メトリクスは参照カウントで管理、0になったら削除 | 複数タスク同名メトリクス |

### 計算ロジック

**HistogramSummaryProxy**:
- QUANTILES = [0.5, 0.75, 0.95, 0.98, 0.99, 0.999]
- count と各 quantile の値を出力

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

本機能はデータベース操作を行わない。

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| RuntimeException | ポート使用不可 | 指定ポート範囲がすべて使用中 | 別のポート範囲を指定 |
| IOException | HTTP起動エラー | HTTPServer の起動に失敗 | ログを確認、ポート競合を解消 |

### リトライ仕様

ポート範囲内で順次試行。すべて失敗した場合は例外をスロー。

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

該当なし

## パフォーマンス要件

- 同期化されたメトリクス追加/削除（synchronized）
- Prometheus スクレイピング時にメトリクス値を取得
- collectorsWithCountByMetricName マップで参照カウント管理

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

- HTTP エンドポイントは認証なしで公開される
- 本番環境ではネットワークポリシーでアクセス制限を推奨
- 機密情報をメトリクス名・ラベル値に含めないこと

## 備考

- PrometheusPushGatewayReporter も利用可能（短命ジョブ向け）
- デフォルトポート 9249 は Prometheus エコシステムで慣例的に使用される

---

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

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

### 推奨読解順序

#### Step 1: ファクトリ実装を理解する

Prometheusレポーターファクトリの実装を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | PrometheusReporterFactory.java | `flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/PrometheusReporterFactory.java` | ファクトリ実装 |

**主要処理フロー**:
- **27行目**: MetricReporterFactory を実装
- **29-30行目**: ARG_PORT = "port", DEFAULT_PORT = "9249"
- **33-38行目**: createMetricReporter() でポート設定を解析し PrometheusReporter を生成

#### Step 2: 基底レポーター実装を理解する

AbstractPrometheusReporter の実装を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | AbstractPrometheusReporter.java | `flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/AbstractPrometheusReporter.java` | 基底レポーター |

**主要処理フロー**:
1. **58-59行目**: 不正文字パターン定義 [^a-zA-Z0-9:_]
2. **62-63行目**: SCOPE_SEPARATOR = '_', SCOPE_PREFIX = "flink_"
3. **65-66行目**: collectorsWithCountByMetricName で参照カウント管理
4. **69-74行目**: replaceInvalidChars() で文字変換
5. **81-90行目**: open() でラベル値フィルタ設定を読み込み
6. **98-140行目**: notifyOfAddedMetric() でメトリクス追加
   - 101-108行目: 次元キー・値の抽出
   - 110-111行目: スコープ付きメトリクス名生成
   - 116-135行目: Collector 生成または既存を使用
7. **149-183行目**: createCollector() でメトリクス型に応じた Collector 生成
8. **185-207行目**: addMetric() でメトリクスを Collector に追加
9. **231-261行目**: notifyOfRemovedMetric() でメトリクス削除

#### Step 3: レポーター実装を理解する

PrometheusReporter の実装を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | PrometheusReporter.java | `flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/PrometheusReporter.java` | HTTPサーバー起動 |

**主要処理フロー**:
- **36行目**: AbstractPrometheusReporter を継承
- **47-65行目**: コンストラクタでポート範囲を順次試行して HTTPServer 起動
- **68-74行目**: close() で HTTPServer 停止

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

```
PrometheusReporterFactory.createMetricReporter(Properties)
    │
    └─ new PrometheusReporter(PortRange)
           │
           ├─ [loop] new HTTPServer(port, registry)
           │
           ├─ notifyOfAddedMetric(Metric, name, MetricGroup)
           │      │
           │      ├─ group.getAllVariables() → 次元抽出
           │      │
           │      ├─ getScopedName(name, group)
           │      │      └─ SCOPE_PREFIX + getLogicalScope() + name
           │      │
           │      ├─ createCollector(metric, ...)
           │      │      ├─ GAUGE/COUNTER/METER: io.prometheus.client.Gauge
           │      │      └─ HISTOGRAM: HistogramSummaryProxy
           │      │
           │      ├─ collector.register(registry)
           │      │
           │      └─ addMetric(metric, dimensionValues, collector)
           │             ├─ Gauge.setChild(gaugeFrom(...))
           │             └─ HistogramSummaryProxy.addChild(histogram)
           │
           ├─ notifyOfRemovedMetric(Metric, name, MetricGroup)
           │      │
           │      ├─ removeMetric(metric, dimensionValues, collector)
           │      │
           │      └─ [if count==0] registry.unregister(collector)
           │
           └─ close()
                  └─ httpServer.stop()
```

### データフロー図

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

Metric追加通知 ────────────▶ notifyOfAddedMetric ────────▶ Prometheus Collector登録
    │                           │
MetricGroup ───────────────▶ getAllVariables() ──────────▶ 次元ラベル
    │                           │
                            getScopedName() ─────────────▶ "flink_xxx_yyy_metricName"
    │                           │
Counter/Gauge/Meter ───────▶ gaugeFrom() ────────────────▶ Prometheus Gauge
    │                           │
Histogram ─────────────────▶ HistogramSummaryProxy ──────▶ Prometheus Summary
    │                           │
                            HTTPServer (/metrics) ───────▶ Prometheus Scrape
                                                            │
Prometheus Server ──────────────────────────────────────────┘
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| PrometheusReporter.java | `flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/PrometheusReporter.java` | ソース | HTTPサーバー起動 |
| AbstractPrometheusReporter.java | `flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/AbstractPrometheusReporter.java` | ソース | 基底レポーター |
| PrometheusReporterFactory.java | `flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/PrometheusReporterFactory.java` | ソース | ファクトリ |
| PrometheusPushGatewayReporter.java | `flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/PrometheusPushGatewayReporter.java` | ソース | PushGatewayレポーター |
| PrometheusPushGatewayReporterFactory.java | `flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/PrometheusPushGatewayReporterFactory.java` | ソース | PushGatewayファクトリ |
| org.apache.flink.metrics.reporter.MetricReporterFactory | `flink-metrics/flink-metrics-prometheus/src/main/resources/META-INF/services/` | 設定 | SPIサービス登録 |
