# 帳票設計書 4-StatsdSink メトリクスレポート

## 概要

本ドキュメントは、Apache Spark の MetricsSystem における StatsdSink が出力する StatsD 形式メトリクスレポートの設計書である。カスタム実装の StatsdReporter を利用し、Spark の各コンポーネントのメトリクス情報を StatsD サーバへ UDP プロトコルで定期的に送信する。

### 本帳票の処理概要

StatsdSink メトリクスレポートは、Spark の内部メトリクスデータを StatsD サーバへ UDP で定期送信する帳票機能である。

**業務上の目的・背景**：StatsD は Etsy が開発した軽量なメトリクス集約デーモンであり、多くのモニタリングバックエンド（Graphite、InfluxDB、Datadog 等）と連携可能である。UDP ベースの通信により、送信元のアプリケーションに対するオーバーヘッドが極めて小さく、高頻度のメトリクス送信に適している。Spark クラスタのメトリクスを既存の StatsD ベースのモニタリングインフラに統合する際に利用する。

**帳票の利用シーン**：(1) StatsD + Graphite/InfluxDB/Datadog ベースの既存モニタリング基盤への Spark メトリクス統合、(2) UDP ベースの軽量メトリクス送信が求められる環境、(3) 複数アプリケーションのメトリクスを StatsD で集約管理する場合。

**主要な出力内容**：
1. Gauge メトリクス（ゲージ型: 現在値の送信）
2. Counter メトリクス（カウンタ型: 累積値の送信）
3. Histogram メトリクス（count, min, max, mean, stddev, パーセンタイルをタイマー型で送信）
4. Meter メトリクス（count とレート情報をタイマー型で送信）
5. Timer メトリクス（Histogram + Meter の複合データを送信）

**帳票の出力タイミング**：MetricsSystem の start() 呼び出し時に定期送信が開始され、設定された間隔（デフォルト10秒）で自動送信される。

**帳票の利用者**：Spark クラスタ管理者、SRE チーム、DevOps エンジニア。

## 帳票種別

メトリクスレポート（UDP 送信型の時系列データ）

## 利用画面

| 画面No | 画面名 | URL/ルーティング | 出力操作 |
|--------|--------|-----------------|---------|
| N/A | 設定ファイル駆動 | N/A | metrics.properties の設定により StatsD サーバへ自動送信 |

## 出力形式

### 基本仕様

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

### StatsD 固有設定

| 項目 | 内容 |
|-----|------|
| プロトコル | UDP |
| データフォーマット | `{prefix}.{metric_name}:{value}\|{type}` |
| メトリクスタイプ | c(counter), g(gauge), ms(timer), s(set) |
| 時間単位（Duration） | ミリ秒（MILLISECONDS） |
| レート単位 | 秒あたり（SECONDS） |

## 帳票レイアウト

### レイアウト概要

StatsD プロトコルに従い、メトリクスごとに個別の UDP パケットを送信する。

```
{prefix}.{metric_name}:{value}|{type}
```

### ヘッダー部

| No | 項目名 | 説明 | データ取得元 | 表示形式 |
|----|-------|------|-------------|---------|
| N/A | ヘッダーなし | StatsD プロトコルにヘッダーは存在しない | N/A | N/A |

### 明細部

**Gauge メトリクス:**

| No | 項目名 | 説明 | データ取得元 | 表示形式 | 列幅 |
|----|-------|------|-------------|---------|-----|
| 1 | name | prefix 付きメトリクス名 | MetricRegistry.name(prefix, key) | 文字列 | 可変 |
| 2 | value | ゲージ値 | Gauge.getValue | 数値（%2.2f） | 可変 |
| 3 | type | メトリクスタイプ | 固定値 "g" | 文字列 | 1 |

**Counter メトリクス:**

| No | 項目名 | 説明 | データ取得元 | 表示形式 | 列幅 |
|----|-------|------|-------------|---------|-----|
| 1 | name | prefix 付きメトリクス名 | MetricRegistry.name(prefix, key) | 文字列 | 可変 |
| 2 | count | カウント値 | Counter.getCount | 数値 | 可変 |
| 3 | type | メトリクスタイプ | 固定値 "c" | 文字列 | 1 |

**Histogram メトリクス（複数パケット）:**

| No | 項目名 | 説明 | データ取得元 | 表示形式 | 列幅 |
|----|-------|------|-------------|---------|-----|
| 1 | count | サンプル数 | Histogram.getCount | gauge(g) | 可変 |
| 2 | max | 最大値 | Snapshot.getMax | timer(ms) | 可変 |
| 3 | mean | 平均値 | Snapshot.getMean | timer(ms) | 可変 |
| 4 | min | 最小値 | Snapshot.getMin | timer(ms) | 可変 |
| 5 | stddev | 標準偏差 | Snapshot.getStdDev | timer(ms) | 可変 |
| 6 | p50 | 中央値 | Snapshot.getMedian | timer(ms) | 可変 |
| 7 | p75 | 75パーセンタイル | Snapshot.get75thPercentile | timer(ms) | 可変 |
| 8 | p95 | 95パーセンタイル | Snapshot.get95thPercentile | timer(ms) | 可変 |
| 9 | p98 | 98パーセンタイル | Snapshot.get98thPercentile | timer(ms) | 可変 |
| 10 | p99 | 99パーセンタイル | Snapshot.get99thPercentile | timer(ms) | 可変 |
| 11 | p999 | 99.9パーセンタイル | Snapshot.get999thPercentile | timer(ms) | 可変 |

### フッター部

| No | 項目名 | 説明 | データ取得元 | 表示形式 |
|----|-------|------|-------------|---------|
| N/A | フッターなし | N/A | N/A | N/A |

## 出力条件

### 抽出条件

| 条件名 | 説明 | 必須 |
|-------|------|-----|
| MetricRegistry 登録 | MetricRegistry に登録されているメトリクスが対象 | Yes |
| 正規表現フィルタ | regex 設定がある場合、マッチするメトリクスのみ送信 | No |

### ソート順

| 優先度 | 項目 | 昇順/降順 |
|-------|------|---------|
| 1 | メトリクス名 | 昇順（SortedMap） |

### 改ページ条件

ネットワーク送信のため改ページなし。

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

### 参照テーブル一覧

本帳票はデータベースを参照しない。

| テーブル名 | 用途 | 結合条件 |
|-----------|------|---------|
| N/A（MetricRegistry） | メトリクスデータの取得元 | N/A |

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

#### MetricRegistry（インメモリ）

| 参照項目 | 帳票項目との対応 | 取得条件 | 備考 |
|---------|----------------|---------|------|
| Gauges | Gauge パケット送信 | registry.getGauges() | filter 適用後 |
| Counters | Counter パケット送信 | registry.getCounters() | filter 適用後 |
| Histograms | Histogram パケット群送信 | registry.getHistograms() | filter 適用後 |
| Meters | Meter パケット群送信 | registry.getMeters() | filter 適用後 |
| Timers | Timer パケット群送信 | registry.getTimers() | filter 適用後 |

## 計算仕様

### 計算項目一覧

| 項目名 | 計算式 | 端数処理 | 備考 |
|-------|-------|---------|------|
| レート変換 | 元値 / 秒 | %2.2f | convertRate() |
| Duration変換 | 元値 -> ms | %2.2f | convertDuration() |
| 数値フォーマット | Float/Double を %2.2f、Number を toString | %2.2f | formatAny() メソッド |

## 処理フロー

### 出力フロー

```mermaid
flowchart TD
    A[MetricsSystem.start] --> B[StatsdSink.start]
    B --> C[StatsdReporter.start - 定期スケジュール登録]
    C --> D{定期タイマー発火}
    D --> E[StatsdReporter.report]
    E --> F[DatagramSocket 生成]
    F --> G[MetricRegistry から取得]
    G --> H[各メトリクスを UDP パケットとして送信]
    H --> I[DatagramSocket close]
    I --> D
    J[MetricsSystem.stop] --> K[StatsdSink.stop]
    K --> L[StatsdReporter.stop]
```

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 表示メッセージ | 対処方法 |
|----------|---------|--------------|---------|
| ポーリング間隔エラー | ポーリング間隔が1秒未満 | IllegalArgumentException | 設定値を1秒以上に変更 |
| ソケット生成失敗 | DatagramSocket の生成に失敗 | WARNING: StatsD datagram socket construction failed | ネットワーク設定を確認 |
| パケット送信失敗 | UDP パケットの送信に失敗 | DEBUG: Unable to send packets to StatsD | StatsD サーバの稼働状態を確認 |
| ソケットクローズ失敗 | DatagramSocket のクローズに失敗 | DEBUG: Error when close socket to StatsD | 通常は無視可能 |

## パフォーマンス要件

| 項目 | 内容 |
|-----|------|
| 想定データ件数 | メトリクス数に依存（regex フィルタで絞込可能） |
| 目標出力時間 | 定期間隔内に送信完了（デフォルト10秒以内） |
| 同時出力数上限 | 1（単一スレッドでの定期レポート） |

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

- UDP 通信は暗号化されていないため、機密性の高いメトリクスを送信する場合はネットワーク分離を検討すること
- デフォルトの送信先は 127.0.0.1:8125（ローカルホスト）であり、リモートサーバへの送信時はファイアウォール設定に注意

## 備考

- 設定は `metrics.properties` ファイルで行う。設定例：
  ```
  *.sink.statsd.class=org.apache.spark.metrics.sink.StatsdSink
  *.sink.statsd.host=127.0.0.1
  *.sink.statsd.port=8125
  *.sink.statsd.period=10
  *.sink.statsd.unit=SECONDS
  *.sink.statsd.prefix=spark
  *.sink.statsd.regex=.*
  ```
- StatsdReporter は Spark 独自のカスタム実装であり、Dropwizard の ScheduledReporter を継承している
- 各 report() 呼び出しごとに新しい DatagramSocket を生成し、送信後にクローズする方式である
- ホワイトスペースはハイフンに置換される（sanitize メソッド）

---

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

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

### 推奨読解順序

#### Step 1: データ構造を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | Sink.scala | `core/src/main/scala/org/apache/spark/metrics/sink/Sink.scala` | Sink トレイトの3メソッド定義（行20-24） |
| 1-2 | StatsdReporter.scala | `core/src/main/scala/org/apache/spark/metrics/sink/StatsdReporter.scala` | StatsdMetricType オブジェクト（行38-43）で StatsD のメトリクスタイプ定義を確認。COUNTER="c", GAUGE="g", TIMER="ms", Set="s" |

**読解のコツ**: StatsdReporter は Dropwizard の ScheduledReporter を継承しているため、report() メソッドのシグネチャは親クラスで定義されている。

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | MetricsSystem.scala | `core/src/main/scala/org/apache/spark/metrics/MetricsSystem.scala` | registerSinks()（行199-240）と start()（行97-106） |

#### Step 3: StatsdSink の実装を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | StatsdSink.scala | `core/src/main/scala/org/apache/spark/metrics/sink/StatsdSink.scala` | コンパニオンオブジェクト（行29-42）でデフォルト値定義。クラス本体（行44-82）で設定読み込みとレポーター生成 |

**主要処理フロー**:
- **行29-42**: StatsdSink コンパニオンオブジェクトでデフォルト値定義（host=127.0.0.1, port=8125, period=10, unit=SECONDS, prefix=""）
- **行48-56**: Properties から host, port, period, unit, prefix を読み込み
- **行58-65**: regex 設定に基づいて MetricFilter を生成
- **行67**: MetricsSystem.checkMinimalPollingPeriod() でポーリング間隔検証
- **行69**: StatsdReporter インスタンスを生成
- **行71-73**: start() で定期レポート開始、ログ出力
- **行76-78**: stop() で停止、ログ出力

#### Step 4: StatsdReporter の送信処理を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | StatsdReporter.scala | `core/src/main/scala/org/apache/spark/metrics/sink/StatsdReporter.scala` | report() メソッド（行61-93）で UDP 送信の全体フローを確認。各メトリクスタイプの送信メソッド（行95-139）で StatsD プロトコルへの変換を確認 |

**主要処理フロー**:
- **行45-53**: StatsdReporter コンストラクタ。ScheduledReporter を継承し、registry, host, port, prefix, filter 等を受け取る
- **行58-59**: InetSocketAddress と whitespace 正規表現パターンの初期化
- **行61-93**: report() メソッド。DatagramSocket を生成し、Try-Success/Failure で例外安全に送信処理を実行
- **行95-96**: reportGauge() - 値をフォーマットし GAUGE タイプで送信
- **行98-99**: reportCounter() - カウント値を COUNTER タイプで送信
- **行101-115**: reportHistogram() - count, max, mean, min, stddev, パーセンタイルを個別パケットで送信
- **行117-123**: reportMetered() - count と各レートを送信
- **行125-139**: reportTimer() - Histogram + Meter の複合データを送信
- **行141-146**: send() - StatsD フォーマット文字列を UTF-8 エンコードして DatagramPacket で送信
- **行148**: fullName() - prefix 付きメトリクス名を生成
- **行150**: sanitize() - ホワイトスペースをハイフンに置換
- **行152-161**: format()/formatAny() - 数値を文字列にフォーマット（Float/Double は %2.2f）

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

```
MetricsSystem.start()
    |
    +-- registerSinks()
    |       +-- StatsdSink.<init>(properties, registry)
    |               +-- MetricsSystem.checkMinimalPollingPeriod()
    |               +-- new StatsdReporter(registry, host, port, prefix, filter)
    |
    +-- sinks.foreach(_.start())
            +-- StatsdSink.start()
                    +-- StatsdReporter.start(pollPeriod, pollUnit)
                            +-- [定期スケジュール] StatsdReporter.report()
                                    +-- new DatagramSocket()
                                    +-- reportGauge() -> send() -> DatagramPacket
                                    +-- reportCounter() -> send() -> DatagramPacket
                                    +-- reportHistogram() -> send() x 11
                                    +-- reportMetered() -> send() x 5
                                    +-- reportTimer() -> send() x 10 + reportMetered()
                                    +-- socket.close()
```

### データフロー図

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

MetricRegistry          StatsdSink                     StatsD サーバ
(Gauges,                +-- StatsdReporter              host:port (UDP)
 Counters,                  +-- report()
 Histograms,                    +-- DatagramSocket
 Meters,                        +-- reportGauge()      name:value|g
 Timers)                        +-- reportCounter()    name:count|c
                                +-- reportHistogram()  name.p50:val|ms
Properties                      +-- reportMetered()    name.m1_rate:val|ms
(host, port, prefix,            +-- reportTimer()      (histogram+meter)
 period, unit, regex)           +-- send()
                                    +-- sanitize()
                                    +-- DatagramPacket -> UDP
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| Sink.scala | `core/src/main/scala/org/apache/spark/metrics/sink/Sink.scala` | ソース | Sink トレイト定義 |
| StatsdSink.scala | `core/src/main/scala/org/apache/spark/metrics/sink/StatsdSink.scala` | ソース | StatsD 送信 Sink の設定・管理 |
| StatsdReporter.scala | `core/src/main/scala/org/apache/spark/metrics/sink/StatsdReporter.scala` | ソース | StatsD プロトコルでの UDP 送信レポーター実装 |
| MetricsSystem.scala | `core/src/main/scala/org/apache/spark/metrics/MetricsSystem.scala` | ソース | Sink の登録・管理・起動 |
