# 機能設計書 60-Graphite

## 概要

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

### 本機能の処理概要

**業務上の目的・背景**：Graphite は時系列データの保存・可視化に特化した監視システムであり、Carbon（データ収集）、Whisper（データストレージ）、Graphite-Web（可視化）から構成される。Flink メトリクスを Graphite にエクスポートすることで、時系列データベースへのメトリクス蓄積、長期トレンド分析、Grafana との統合による可視化が可能になる。

**機能の利用シーン**：
- Graphite ベースの監視基盤を使用している場合
- 時系列メトリクスの長期保存が必要な場合
- Carbon を使用したメトリクス収集環境がある場合
- Grafana で Graphite をデータソースとして使用している場合

**主要な処理内容**：
1. GraphiteReporterFactory による GraphiteReporter インスタンスの生成
2. Dropwizard Metrics ライブラリを使用したメトリクス管理
3. TCP または UDP プロトコルでの Graphite サーバーへの送信
4. 定期的なメトリクスレポーティング（Scheduled インターフェース）
5. Flink メトリクスから Dropwizard メトリクスへの変換

**関連システム・外部連携**：
- Graphite/Carbon サーバー（メトリクス受信）
- Grafana（可視化）
- Whisper（データストレージ）

**権限による制御**：ネットワークアクセス制御に依存

## 関連画面

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

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

## 機能種別

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

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| host | String | Yes | Graphite サーバーホスト | null または空文字列は不可 |
| port | Integer | Yes | Graphite サーバーポート | 1以上 |
| protocol | String | No | 送信プロトコル（TCP/UDP） | デフォルト: TCP |
| prefix | String | No | メトリクス名プレフィックス | 任意 |
| rateConversion | String | No | レート変換単位 | TimeUnit 列挙値 |
| durationConversion | String | No | 期間変換単位 | TimeUnit 列挙値 |

### 入力データソース

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

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| Counter | Dropwizard Counter | FlinkCounterWrapper でラップ |
| Gauge | Dropwizard Gauge | FlinkGaugeWrapper でラップ |
| Histogram | Dropwizard Histogram | FlinkHistogramWrapper でラップ |
| Meter | Dropwizard Meter | FlinkMeterWrapper でラップ |

### 出力先

Graphite/Carbon サーバー（TCP または UDP）

## 処理フロー

### 処理シーケンス

```
1. GraphiteReporterFactory.createMetricReporter(Properties) 呼び出し
   └─ GraphiteReporter インスタンスを生成
2. open(MetricConfig) で getReporter() を呼び出し
   └─ host, port を検証
   └─ prefix, conversionRate, conversionDuration を読み込み
   └─ protocol を判定（TCP/UDP）
   └─ Dropwizard GraphiteReporter を構築
3. notifyOfAddedMetric() でメトリクス追加
   └─ メトリクス名を生成（MetricGroup.getMetricIdentifier）
   └─ Flink メトリクスを Dropwizard メトリクスにラップ
   └─ MetricRegistry に登録
4. report() で定期レポーティング
   └─ registry からすべてのメトリクスを取得
   └─ Dropwizard GraphiteReporter.report() を呼び出し
5. notifyOfRemovedMetric() でメトリクス削除
   └─ registry から削除
6. close() で ScheduledReporter を停止
```

### フローチャート

```mermaid
flowchart TD
    A[開始: createMetricReporter] --> B[GraphiteReporter生成]
    B --> C[open: getReporter呼び出し]
    C --> D{host/port検証}
    D -->|無効| E[IllegalArgumentException]
    D -->|有効| F[prefix設定]
    F --> G[conversionRate設定]
    G --> H[conversionDuration設定]
    H --> I{protocol判定}
    I -->|TCP| J[Graphite TCP生成]
    I -->|UDP| K[GraphiteUDP生成]
    J --> L[Dropwizard GraphiteReporter構築]
    K --> L
    L --> M{メトリクス操作}
    M -->|追加| N[notifyOfAddedMetric]
    N --> O[getMetricIdentifier]
    O --> P{メトリクス型判定}
    P -->|Counter| Q1[FlinkCounterWrapper]
    P -->|Gauge| Q2[FlinkGaugeWrapper]
    P -->|Histogram| Q3[FlinkHistogramWrapper]
    P -->|Meter| Q4[FlinkMeterWrapper]
    Q1 --> R[registry.register]
    Q2 --> R
    Q3 --> R
    Q4 --> R
    R --> M
    M -->|レポート| S[report]
    S --> T[registry.getXxx]
    T --> U[reporter.report]
    U --> M
    M -->|削除| V[notifyOfRemovedMetric]
    V --> W[registry.remove]
    W --> M
    M -->|終了| X[close]
    X --> Y[reporter.stop]
    Y --> Z[終了]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-60-01 | ホスト・ポート必須 | host と port は必須パラメータ | レポーター初期化時 |
| BR-60-02 | プロトコルデフォルト | protocol が無効な場合は TCP にフォールバック | 常時 |
| BR-60-03 | 文字変換 | '.' を '-' に、'"' を削除 | メトリクス名生成時 |
| BR-60-04 | 定期レポート | Scheduled インターフェースにより定期的にレポート | 常時 |

### 計算ロジック

**レート・期間変換**：
- rateConversion: メトリクスのレートを指定した TimeUnit に変換
- durationConversion: メトリクスの期間を指定した TimeUnit に変換

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

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

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| IllegalArgumentException | パラメータ不正 | host が null/空、または port が 1 未満 | host/port を正しく設定 |
| IllegalArgumentException | プロトコル不正 | protocol が TCP/UDP 以外（警告のみ、TCPにフォールバック） | TCP または UDP を指定 |

### リトライ仕様

リトライは行わない。接続エラー時は次回のレポートサイクルで再試行。

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

該当なし

## パフォーマンス要件

- メトリクス追加/削除は synchronized で同期化
- MetricRegistry は ConcurrentMap で実装（レポート時はロック不要）
- TCP/UDP の選択により信頼性とパフォーマンスのトレードオフを調整可能

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

- Graphite サーバーへの接続は認証なし（Carbon の標準動作）
- 本番環境ではネットワークレベルでアクセス制限を推奨
- 機密情報をメトリクス名に含めないこと

## 備考

- Dropwizard Metrics ライブラリ（com.codahale.metrics）を使用
- TCP は信頼性が高く、UDP は高スループット向け
- Grafana との組み合わせで強力な可視化が可能

---

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

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

### 推奨読解順序

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

GraphiteReporterFactory の実装を理解する。

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

**主要処理フロー**:
- **27行目**: MetricReporterFactory を実装
- **29-32行目**: createMetricReporter() で GraphiteReporter を生成

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

GraphiteReporter の実装を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | GraphiteReporter.java | `flink-metrics/flink-metrics-graphite/src/main/java/org/apache/flink/metrics/graphite/GraphiteReporter.java` | レポーター実装 |

**主要処理フロー**:
1. **36行目**: ScheduledDropwizardReporter を継承
2. **38行目**: ARG_PROTOCOL = "protocol" 定義
3. **40-43行目**: Protocol 列挙型（TCP, UDP）定義
4. **46-98行目**: getReporter() で Dropwizard GraphiteReporter を構築
   - 47-48行目: host, port を取得
   - 50-53行目: host/port のバリデーション
   - 55-58行目: prefix, conversionRate, conversionDuration, protocol を取得
   - 60-61行目: Dropwizard GraphiteReporter.Builder を生成
   - 63-73行目: prefix, conversionRate, conversionDuration を設定
   - 75-84行目: protocol を判定（無効な場合は TCP にフォールバック）
   - 91-97行目: TCP/UDP に応じた Graphite インスタンスを生成

#### Step 3: 基底クラスを理解する

ScheduledDropwizardReporter の実装を理解する。

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

**主要処理フロー**:
1. **55-56行目**: MetricReporter, Scheduled, Reporter, CharacterFilter を実装
2. **60-64行目**: ARG_HOST, ARG_PORT, ARG_PREFIX, ARG_CONVERSION_RATE, ARG_CONVERSION_DURATION 定義
3. **68行目**: MetricRegistry をフィールドに保持
4. **70行目**: ScheduledReporter をフィールドに保持
5. **72-75行目**: gauges, counters, histograms, meters マップでメトリクスを管理
6. **79-81行目**: コンストラクタで MetricRegistry を生成
7. **111-114行目**: open() で getReporter() を呼び出し
8. **116-119行目**: close() で reporter.stop() を呼び出し
9. **125-169行目**: notifyOfAddedMetric() でメトリクスを追加
   - 127行目: getMetricIdentifier() でフル名を生成
   - 130-167行目: メトリクス型に応じた Wrapper を生成し registry に登録
10. **171-197行目**: notifyOfRemovedMetric() でメトリクスを削除
11. **199-229行目**: filterCharacters() で文字変換（'.' → '-', '"' → 削除）
12. **235-248行目**: report() で定期レポーティング

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

```
GraphiteReporterFactory.createMetricReporter(Properties)
    │
    └─ new GraphiteReporter()
           │
           ├─ [継承] ScheduledDropwizardReporter()
           │      └─ new MetricRegistry()
           │
           ├─ open(MetricConfig)
           │      └─ getReporter(config)
           │             │
           │             ├─ config.getString(ARG_HOST)
           │             ├─ config.getInteger(ARG_PORT)
           │             ├─ config.getString(ARG_PREFIX)
           │             ├─ config.getString(ARG_CONVERSION_RATE)
           │             ├─ config.getString(ARG_CONVERSION_DURATION)
           │             ├─ config.getString(ARG_PROTOCOL)
           │             │
           │             ├─ GraphiteReporter.Builder.forRegistry(registry)
           │             │      ├─ .prefixedWith(prefix)
           │             │      ├─ .convertRatesTo(TimeUnit)
           │             │      └─ .convertDurationsTo(TimeUnit)
           │             │
           │             └─ switch(protocol)
           │                    ├─ UDP: builder.build(new GraphiteUDP(host, port))
           │                    └─ TCP: builder.build(new Graphite(host, port))
           │
           ├─ notifyOfAddedMetric(Metric, name, MetricGroup)
           │      │
           │      ├─ group.getMetricIdentifier(metricName, this)
           │      │
           │      └─ switch(metric.getMetricType())
           │             ├─ COUNTER: registry.register(fullName, new FlinkCounterWrapper)
           │             ├─ GAUGE: registry.register(fullName, FlinkGaugeWrapper.fromGauge)
           │             ├─ HISTOGRAM: registry.register(fullName, FlinkHistogramWrapper)
           │             └─ METER: registry.register(fullName, FlinkMeterWrapper)
           │
           ├─ report()
           │      │
           │      ├─ registry.getGauges()
           │      ├─ registry.getCounters()
           │      ├─ registry.getHistograms()
           │      ├─ registry.getMeters()
           │      ├─ registry.getTimers()
           │      │
           │      └─ reporter.report(gauges, counters, histograms, meters, timers)
           │
           ├─ notifyOfRemovedMetric(Metric, name, MetricGroup)
           │      └─ registry.remove(fullName)
           │
           └─ close()
                  └─ reporter.stop()
```

### データフロー図

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

Metric追加通知 ────────────▶ notifyOfAddedMetric ────────▶ registry登録
    │                           │
MetricGroup ───────────────▶ getMetricIdentifier ────────▶ "scope.metricName"
    │                           │
                            filterCharacters ────────────▶ "scope-metricName"
    │                           │
Counter ───────────────────▶ FlinkCounterWrapper ────────▶ Dropwizard Counter
Gauge ─────────────────────▶ FlinkGaugeWrapper ──────────▶ Dropwizard Gauge
Histogram ─────────────────▶ FlinkHistogramWrapper ──────▶ Dropwizard Histogram
Meter ─────────────────────▶ FlinkMeterWrapper ──────────▶ Dropwizard Meter
    │                           │
                            report() ────────────────────▶ Graphite Reporter
                                │
                            TCP/UDP ─────────────────────▶ Carbon/Graphite
                                                            │
Grafana ────────────────────────────────────────────────────┘
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| GraphiteReporter.java | `flink-metrics/flink-metrics-graphite/src/main/java/org/apache/flink/metrics/graphite/GraphiteReporter.java` | ソース | レポーター実装 |
| GraphiteReporterFactory.java | `flink-metrics/flink-metrics-graphite/src/main/java/org/apache/flink/metrics/graphite/GraphiteReporterFactory.java` | ソース | ファクトリ実装 |
| ScheduledDropwizardReporter.java | `flink-metrics/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/ScheduledDropwizardReporter.java` | ソース | 基底レポーター |
| FlinkCounterWrapper.java | `flink-metrics/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/metrics/FlinkCounterWrapper.java` | ソース | Counter ラッパー |
| FlinkGaugeWrapper.java | `flink-metrics/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/metrics/FlinkGaugeWrapper.java` | ソース | Gauge ラッパー |
| FlinkHistogramWrapper.java | `flink-metrics/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/metrics/FlinkHistogramWrapper.java` | ソース | Histogram ラッパー |
| FlinkMeterWrapper.java | `flink-metrics/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/metrics/FlinkMeterWrapper.java` | ソース | Meter ラッパー |
| org.apache.flink.metrics.reporter.MetricReporterFactory | `flink-metrics/flink-metrics-graphite/src/main/resources/META-INF/services/` | 設定 | SPIサービス登録 |
