# 帳票設計書 13-StatsDメトリクスレポート

## 概要

本ドキュメントは、Apache FlinkのStatsDメトリクスレポーター機能に関する設計書である。FlinkのメトリクスをStatsDサーバーへUDP経由で送信するための仕組みを定義する。

### 本帳票の処理概要

StatsDメトリクスレポートは、Apache Flinkで収集された各種メトリクス（Counter、Gauge、Histogram、Meter）をStatsDプロトコル形式でUDPパケットとして送信する機能である。

**業務上の目的・背景**：StatsDは軽量で高速なメトリクス収集プロトコルとして広く普及している。UDPベースのため、ネットワークオーバーヘッドが小さく、メトリクス送信がアプリケーションのパフォーマンスに与える影響を最小限に抑えられる。Graphite、Datadog、InfluxDB、Prometheusなど多くの監視バックエンドがStatsDプロトコルをサポートしており、既存の監視インフラとの統合が容易である。

**帳票の利用シーン**：本レポーターは以下のシーンで利用される。(1)軽量なメトリクス収集が必要な高トラフィック環境、(2)Graphiteバックエンドへのメトリクス転送、(3)Datadogエージェント経由でのメトリクス送信、(4)複数の監視バックエンドへの同時メトリクス配信（StatsD aggregator経由）、(5)コネクションレスなUDP送信による低遅延メトリクス転送。

**主要な出力内容**：
1. Counter値 - name:value|g 形式でカウント数を送信
2. Gauge値 - name:value|g 形式で現在値を送信
3. Histogram統計値 - count, min, max, mean, stddev, p50, p75, p95, p98, p99, p999をそれぞれ送信
4. Meter値 - rate, countをそれぞれ送信

**帳票の出力タイミング**：スケジュールされた間隔（デフォルトで設定可能）で定期的にメトリクスがStatsDサーバーへプッシュされる。`Scheduled`インターフェースの実装により、Flinkのメトリクスフレームワークが`report()`メソッドを周期的に呼び出す。

**帳票の利用者**：Flinkクラスタ運用者、SRE、インフラエンジニア、監視システム管理者。

## 帳票種別

StatsDプロトコルメッセージ（UDP）

## 利用画面

| 画面No | 画面名 | URL/ルーティング | 出力操作 |
|--------|--------|-----------------|---------|
| N/A | 設定ファイル | flink-conf.yaml | metrics.reporter.stsd.* 設定 |

## 出力形式

### 基本仕様

| 項目 | 内容 |
|-----|------|
| ファイル形式 | StatsDプロトコル（UDP） |
| 用紙サイズ | N/A |
| 向き | N/A |
| ファイル名 | N/A（ネットワーク送信） |
| 出力方法 | UDP DatagramPacket送信 |
| 文字コード | UTF-8 |

### StatsD固有設定

| 項目 | 内容 |
|-----|------|
| プロトコル | UDP |
| メッセージ形式 | name:value|g |
| 負値の処理 | 0リセット後に値適用 |

## 帳票レイアウト

### レイアウト概要

StatsDプロトコルフォーマットでデータが構造化される。

```
metric_name:value|g
```

### メトリクス種別ごとの出力構成

#### Counter

| No | 項目名 | 説明 | データ取得元 | 形式 |
|----|-------|------|-------------|------|
| 1 | {name} | カウンター値 | counter.getCount() | {name}:{count}|g |

#### Gauge

| No | 項目名 | 説明 | データ取得元 | 形式 |
|----|-------|------|-------------|------|
| 1 | {name} | ゲージ値 | gauge.getValue() | {name}:{value}|g |

#### Histogram

| No | 項目名 | 説明 | データ取得元 | 形式 |
|----|-------|------|-------------|------|
| 1 | {name}.count | サンプル数 | histogram.getCount() | {name}.count:{value}|g |
| 2 | {name}.max | 最大値 | statistics.getMax() | {name}.max:{value}|g |
| 3 | {name}.min | 最小値 | statistics.getMin() | {name}.min:{value}|g |
| 4 | {name}.mean | 平均値 | statistics.getMean() | {name}.mean:{value}|g |
| 5 | {name}.stddev | 標準偏差 | statistics.getStdDev() | {name}.stddev:{value}|g |
| 6 | {name}.p50 | 50パーセンタイル | statistics.getQuantile(0.5) | {name}.p50:{value}|g |
| 7 | {name}.p75 | 75パーセンタイル | statistics.getQuantile(0.75) | {name}.p75:{value}|g |
| 8 | {name}.p95 | 95パーセンタイル | statistics.getQuantile(0.95) | {name}.p95:{value}|g |
| 9 | {name}.p98 | 98パーセンタイル | statistics.getQuantile(0.98) | {name}.p98:{value}|g |
| 10 | {name}.p99 | 99パーセンタイル | statistics.getQuantile(0.99) | {name}.p99:{value}|g |
| 11 | {name}.p999 | 99.9パーセンタイル | statistics.getQuantile(0.999) | {name}.p999:{value}|g |

#### Meter

| No | 項目名 | 説明 | データ取得元 | 形式 |
|----|-------|------|-------------|------|
| 1 | {name}.rate | レート（件/秒） | meter.getRate() | {name}.rate:{value}|g |
| 2 | {name}.count | 累積カウント | meter.getCount() | {name}.count:{value}|g |

## 出力条件

### 抽出条件

| 条件名 | 説明 | 必須 |
|-------|------|-----|
| host | StatsDサーバーホスト | Yes |
| port | StatsDサーバーポート | Yes |

### ソート順

| 優先度 | 項目 | 昇順/降順 |
|-------|------|---------|
| N/A | 送信順 | 登録順 |

### 改ページ条件

N/A（ネットワーク送信）

## データベース参照仕様

### 参照テーブル一覧

| テーブル名 | 用途 | 結合条件 |
|-----------|------|---------|
| N/A（メモリ内メトリクスレジストリ） | メトリクス収集 | N/A |

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

#### メトリクスレジストリ（AbstractReporter内部Map）

| 参照項目 | 帳票項目との対応 | 取得条件 | 備考 |
|---------|----------------|---------|------|
| gauges | Gauge出力 | 全登録Gauge | Map<Gauge<?>, String> |
| counters | Counter出力 | 全登録Counter | Map<Counter, String> |
| histograms | Histogram出力 | 全登録Histogram | Map<Histogram, String> |
| meters | Meter出力 | 全登録Meter | Map<Meter, String> |

## 計算仕様

### 計算項目一覧

| 項目名 | 計算式 | 端数処理 | 備考 |
|-------|-------|---------|------|
| メトリクス名サフィックス | prefix(baseName, suffix) | N/A | "."で連結 |
| パーセンタイル | statistics.getQuantile(percentile) | N/A | 統計ライブラリ依存 |

## 処理フロー

### 出力フロー

```mermaid
flowchart TD
    A[Scheduledによるreport呼び出し] --> B{closed?}
    B -->|Yes| C[終了]
    B -->|No| D[Gauges処理]
    D --> E{closed?}
    E -->|Yes| C
    E -->|No| F[reportGauge実行]
    F --> G[Counters処理]
    G --> H{closed?}
    H -->|Yes| C
    H -->|No| I[reportCounter実行]
    I --> J[Histograms処理]
    J --> K[reportHistogram実行]
    K --> L[Meters処理]
    L --> M[reportMeter実行]
    M --> C
```

### send処理フロー

```mermaid
flowchart TD
    A[send呼び出し] --> B{負値?}
    B -->|Yes| C[send name:0 g]
    B -->|No| D[直接送信]
    C --> D
    D --> E[String.format name:value g]
    E --> F[getBytes UTF-8]
    F --> G[DatagramPacket作成]
    G --> H[socket.send]
    H --> I[終了]
```

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 表示メッセージ | 対処方法 |
|----------|---------|--------------|---------|
| 設定エラー | host/portが無効 | Invalid host/port configuration. Host: {host} Port: {port} | 設定値の確認 |
| SocketException | DatagramSocket作成失敗 | Could not create datagram socket. | ネットワーク設定確認 |
| IOException | パケット送信失敗 | unable to send packet to statsd at '{host}:{port}' | ネットワーク/サーバー確認 |
| 並行変更 | メトリクス追加/削除中 | N/A（次回レポートで再試行） | 自動リカバリ |

## パフォーマンス要件

| 項目 | 内容 |
|-----|------|
| 想定データ件数 | メトリクス数に依存（数百〜数千） |
| 目標出力時間 | UDPのため即時（ファイア・アンド・フォーゲット） |
| 同時出力数上限 | 1（シングルスレッド送信） |

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

- UDP通信は暗号化されないため、機密性の高いメトリクスには注意が必要
- StatsDサーバーへのネットワークアクセス制御を推奨
- ファイアウォールルールでUDPポートを適切に制限

## 備考

- ReadyTalk社のmetrics-statsdを参考に実装
- DatagramSocketはポート0（自動割り当て）で作成
- 負値は直接送信できないため、0にリセット後に適用（負値は減算として解釈されるため）
- filterCharactersメソッドで":"を"-"に置換

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | StatsDReporterFactory.java | `flink-metrics/flink-metrics-statsd/src/main/java/org/apache/flink/metrics/statsd/StatsDReporterFactory.java` | ファクトリ実装（行29-35） |
| 1-2 | StatsDReporter.java | `flink-metrics/flink-metrics-statsd/src/main/java/org/apache/flink/metrics/statsd/StatsDReporter.java` | クラス定義、AbstractReporter継承、Scheduled実装（行52） |

**読解のコツ**: StatsDReporterはAbstractReporterを継承しており、メトリクス登録の基本処理は親クラスに委譲される。

#### Step 2: 初期化処理を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | StatsDReporter.java | `flink-metrics/flink-metrics-statsd/src/main/java/org/apache/flink/metrics/statsd/StatsDReporter.java` | open()メソッド（行65-82） |

**主要処理フロー**:
1. **行66-67**: 設定からhost/port取得
2. **行69-72**: 設定値バリデーション
3. **行74**: InetSocketAddress作成
4. **行76-80**: DatagramSocket作成（ポート0）
5. **行81**: 設定ログ出力

#### Step 3: レポート処理を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | StatsDReporter.java | `flink-metrics/flink-metrics-statsd/src/main/java/org/apache/flink/metrics/statsd/StatsDReporter.java` | report()メソッド（行95-125） |

**主要処理フロー**:
- **行100-105**: Gaugesループ処理（closedチェック付き）
- **行107-112**: Countersループ処理（closedチェック付き）
- **行114-116**: Histogramsループ処理
- **行118-120**: Metersループ処理
- **行121-124**: ConcurrentModificationException許容

#### Step 4: 個別メトリクス送信処理を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | StatsDReporter.java | `flink-metrics/flink-metrics-statsd/src/main/java/org/apache/flink/metrics/statsd/StatsDReporter.java` | reportCounter/reportGauge/reportHistogram/reportMeter（行129-172） |

**主要処理フロー**:
- **行129-131**: reportCounter - カウント値送信
- **行133-144**: reportGauge - null/Number判定、値送信
- **行146-165**: reportHistogram - 11メトリクス送信（count, max, min, mean, stddev, p50-p999）
- **行167-172**: reportMeter - rate/count送信

#### Step 5: send処理を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 5-1 | StatsDReporter.java | `flink-metrics/flink-metrics-statsd/src/main/java/org/apache/flink/metrics/statsd/StatsDReporter.java` | send系メソッド（行188-216） |

**主要処理フロー**:
- **行188-190**: send(String, double) - double値送信
- **行192-194**: send(String, long) - long値送信（負値チェック）
- **行196-203**: send(boolean, String, String) - 負値時0リセット処理
- **行205-216**: send(String, String) - UDP送信実行

#### Step 6: 文字フィルタリングを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 6-1 | StatsDReporter.java | `flink-metrics/flink-metrics-statsd/src/main/java/org/apache/flink/metrics/statsd/StatsDReporter.java` | filterCharacters()メソッド（行218-243） |

**主要処理フロー**:
- **行227-231**: ":"を"-"に置換

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

```
StatsDReporterFactory.createMetricReporter(Properties)
    │
    └─ new StatsDReporter()
           │
           ├─ open(MetricConfig)
           │      ├─ config.getString(ARG_HOST) / config.getInteger(ARG_PORT)
           │      ├─ new InetSocketAddress(host, port)
           │      └─ new DatagramSocket(0)
           │
           ├─ notifyOfAddedMetric(Metric, String, MetricGroup)
           │      └─ [AbstractReporter実装を継承]
           │
           ├─ report() [Scheduled呼び出し]
           │      ├─ [for each gauge] reportGauge(name, gauge)
           │      │      └─ send(name, value)
           │      │
           │      ├─ [for each counter] reportCounter(name, counter)
           │      │      └─ send(name, counter.getCount())
           │      │
           │      ├─ [for each histogram] reportHistogram(name, histogram)
           │      │      ├─ send(prefix(name, "count"), histogram.getCount())
           │      │      ├─ send(prefix(name, "max"), statistics.getMax())
           │      │      ├─ ... [min, mean, stddev]
           │      │      └─ send(prefix(name, "p999"), statistics.getQuantile(0.999))
           │      │
           │      └─ [for each meter] reportMeter(name, meter)
           │             ├─ send(prefix(name, "rate"), meter.getRate())
           │             └─ send(prefix(name, "count"), meter.getCount())
           │
           │      send(boolean resetToZero, String name, String value)
           │          │
           │          └─ send(String name, String value)
           │                 ├─ String.format("%s:%s|g", name, value)
           │                 ├─ formatted.getBytes(StandardCharsets.UTF_8)
           │                 └─ socket.send(new DatagramPacket(...))
           │
           └─ close()
                  └─ socket.close()
```

### データフロー図

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

Flink Metrics          StatsDReporter                StatsD Server
    │                        │                           │
    │ Counter              │                           │
    │ Gauge     ─────────▶ │ notifyOfAddedMetric()    │
    │ Histogram            │ (メトリクス登録)           │
    │ Meter                │                           │
    │                        │                           │
    │                        │ report() [定期実行]       │
    │                        │     │                     │
    │                        │     ▼                     │
    │                        │ reportGauge/Counter/...   │
    │                        │     │                     │
    │                        │     ▼                     │
    │                        │ send(name, value)        │
    │                        │     │                     │
    │                        │     ▼                     │
    │                        │ String.format            │
    │                        │ "%s:%s|g"                │
    │                        │     │                     │
    │                        │     ▼                     │
    │                        │ DatagramPacket          │
    │                        │     │                     │
    │                        │     ▼                     │
    │                        │ socket.send() ──────────▶│ UDP受信
    │                        │                           │ name:value|g
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| StatsDReporter.java | `flink-metrics/flink-metrics-statsd/src/main/java/org/apache/flink/metrics/statsd/StatsDReporter.java` | ソース | メインレポータークラス |
| StatsDReporterFactory.java | `flink-metrics/flink-metrics-statsd/src/main/java/org/apache/flink/metrics/statsd/StatsDReporterFactory.java` | ソース | ファクトリクラス |
| AbstractReporter.java | `flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/reporter/AbstractReporter.java` | ソース | 基底レポータークラス |
