# 帳票設計書 11-InfluxDBメトリクスレポート

## 概要

本ドキュメントは、Apache FlinkのInfluxDBメトリクスレポーター機能に関する設計書である。FlinkのメトリクスをInfluxDB時系列データベースへエクスポートするための仕組みを定義する。

### 本帳票の処理概要

InfluxDBメトリクスレポートは、Apache Flinkで収集された各種メトリクス（Counter、Gauge、Histogram、Meter）をInfluxDBデータベースへ定期的に送信する機能である。

**業務上の目的・背景**：分散ストリーム処理基盤であるFlinkにおいて、システムの健全性監視とパフォーマンス分析は運用上不可欠である。InfluxDBは時系列データの保存に最適化されたデータベースであり、メトリクスデータの長期保存と高速クエリを可能にする。本レポーターにより、Flinkクラスタの運用者はGrafanaなどの可視化ツールと連携して、リアルタイムのダッシュボードを構築し、システムの状態を継続的に監視できる。

**帳票の利用シーン**：本レポーターは以下のシーンで利用される。(1)Flinkジョブの実行状況のリアルタイム監視、(2)クラスタリソース使用率の時系列分析、(3)パフォーマンスボトルネックの特定、(4)キャパシティプランニングのためのヒストリカルデータ収集、(5)アラート条件の評価用データ提供。

**主要な出力内容**：
1. Counter値（カウント数）- タスクの処理レコード数など
2. Gauge値（現在値）- メモリ使用量、CPU使用率など
3. Histogram統計値（count, min, max, mean, stddev, p50, p75, p95, p98, p99, p999）
4. Meter値（count, rate）- スループット測定用

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

**帳票の利用者**：Flinkクラスタ運用者、SRE（Site Reliability Engineer）、DevOpsエンジニア、パフォーマンスエンジニア。

## 帳票種別

時系列メトリクスデータ（InfluxDB Line Protocol形式）

## 利用画面

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

## 出力形式

### 基本仕様

| 項目 | 内容 |
|-----|------|
| ファイル形式 | InfluxDB Line Protocol（HTTP経由） |
| 用紙サイズ | N/A（データ送信） |
| 向き | N/A |
| ファイル名 | N/A（データベース書き込み） |
| 出力方法 | HTTP POST（InfluxDB API） |
| 文字コード | UTF-8 |

### InfluxDB固有設定

| 項目 | 内容 |
|-----|------|
| スキーム | HTTP / HTTPS（デフォルト: HTTP） |
| デフォルトポート | 8086 |
| 認証 | ユーザー名/パスワード（オプション） |
| 接続タイムアウト | 10000ms（デフォルト） |
| 書き込みタイムアウト | 10000ms（デフォルト） |
| 一貫性レベル | ONE（デフォルト） |

## 帳票レイアウト

### レイアウト概要

InfluxDB Line Protocolフォーマットでデータが構造化される。

```
measurement_name,tag1=value1,tag2=value2 field1=value1,field2=value2 timestamp
```

### メトリクス種別ごとのフィールド構成

#### Counter

| No | 項目名 | 説明 | データ取得元 | 表示形式 |
|----|-------|------|-------------|---------|
| 1 | count | カウンター値 | Counter.getCount() | Long |

#### Gauge

| No | 項目名 | 説明 | データ取得元 | 表示形式 |
|----|-------|------|-------------|---------|
| 1 | value | ゲージ値 | Gauge.getValue() | Number/String |

#### Histogram

| No | 項目名 | 説明 | データ取得元 | 表示形式 |
|----|-------|------|-------------|---------|
| 1 | count | サンプル数 | statistics.size() | Long |
| 2 | min | 最小値 | statistics.getMin() | Long |
| 3 | max | 最大値 | statistics.getMax() | Long |
| 4 | mean | 平均値 | statistics.getMean() | Double |
| 5 | stddev | 標準偏差 | statistics.getStdDev() | Double |
| 6 | p50 | 50パーセンタイル | statistics.getQuantile(.50) | Double |
| 7 | p75 | 75パーセンタイル | statistics.getQuantile(.75) | Double |
| 8 | p95 | 95パーセンタイル | statistics.getQuantile(.95) | Double |
| 9 | p98 | 98パーセンタイル | statistics.getQuantile(.98) | Double |
| 10 | p99 | 99パーセンタイル | statistics.getQuantile(.99) | Double |
| 11 | p999 | 99.9パーセンタイル | statistics.getQuantile(.999) | Double |

#### Meter

| No | 項目名 | 説明 | データ取得元 | 表示形式 |
|----|-------|------|-------------|---------|
| 1 | count | 累積カウント | Meter.getCount() | Long |
| 2 | rate | レート（件/秒） | Meter.getRate() | Double |

## 出力条件

### 抽出条件

| 条件名 | 説明 | 必須 |
|-------|------|-----|
| host | InfluxDBサーバーホスト | Yes |
| port | InfluxDBサーバーポート | Yes |
| db | データベース名 | Yes |
| scheme | HTTP/HTTPS | No（デフォルト: HTTP） |
| username | 認証ユーザー名 | No |
| password | 認証パスワード | No |
| retentionPolicy | 保持ポリシー | No |
| consistency | 一貫性レベル | No（デフォルト: ONE） |

### ソート順

| 優先度 | 項目 | 昇順/降順 |
|-------|------|---------|
| N/A | タイムスタンプ | 送信順 |

### 改ページ条件

N/A（データベース書き込み）

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

### 参照テーブル一覧

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

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

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

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

## 計算仕様

### 計算項目一覧

| 項目名 | 計算式 | 端数処理 | 備考 |
|-------|-------|---------|------|
| タイムスタンプ | Instant.now().toEpochMilli() | N/A | ミリ秒精度 |
| パーセンタイル | statistics.getQuantile(percentile) | N/A | 統計ライブラリ依存 |

## 処理フロー

### 出力フロー

```mermaid
flowchart TD
    A[Scheduledによるreport呼び出し] --> B[buildReport実行]
    B --> C[BatchPoints.Builder作成]
    C --> D[database/retentionPolicy/consistency設定]
    D --> E{各メトリクス種別をループ}
    E --> F[Gauges処理]
    E --> G[Counters処理]
    E --> H[Histograms処理]
    E --> I[Meters処理]
    F --> J[MetricMapper.mapでPoint変換]
    G --> J
    H --> J
    I --> J
    J --> K[BatchPointsに追加]
    K --> L{全メトリクス処理完了?}
    L -->|No| E
    L -->|Yes| M[BatchPoints.build]
    M --> N[influxDB.write実行]
    N --> O[終了]
```

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 表示メッセージ | 対処方法 |
|----------|---------|--------------|---------|
| 設定エラー | host/portが無効 | Invalid host/port configuration | 設定値の確認 |
| DB未設定 | db設定なし | 'db' configuration option is not set | db設定の追加 |
| 並行変更 | メトリクス追加/削除中 | N/A（次回レポートで再試行） | 自動リカバリ |
| 接続エラー | InfluxDB接続失敗 | InfluxDB例外 | ネットワーク/サーバー確認 |

## パフォーマンス要件

| 項目 | 内容 |
|-----|------|
| 想定データ件数 | メトリクス数に依存（数百〜数千） |
| 目標出力時間 | 接続/書き込みタイムアウト内（デフォルト各10秒） |
| 同時出力数上限 | 1（シングルスレッド送信） |

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

- 認証情報（username/password）は設定ファイルで管理されるため、適切なファイルパーミッションが必要
- HTTPSスキームの使用により通信の暗号化が可能
- InfluxDBへのネットワークアクセス制御を推奨

## 備考

- OkHttpClientを使用したHTTP通信
- InfluxDB Java Clientライブラリ依存
- ConcurrentModificationExceptionは許容され、次回レポートで再試行

---

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

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

### 推奨読解順序

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

メトリクス情報を保持するデータ構造を理解することが最初のステップとなる。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | MeasurementInfo.java | `flink-metrics/flink-metrics-influxdb/src/main/java/org/apache/flink/metrics/influxdb/MeasurementInfo.java` | メトリクス名とタグを保持するデータクラス（行23-39） |
| 1-2 | MetricInfoProvider.java | `flink-metrics/flink-metrics-influxdb/src/main/java/org/apache/flink/metrics/influxdb/MetricInfoProvider.java` | メトリクス情報を提供するインターフェース |

**読解のコツ**: MeasurementInfoはInfluxDBのmeasurement名とタグ情報をカプセル化している。

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

処理の起点となるファクトリとレポータークラスを確認する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | InfluxdbReporterFactory.java | `flink-metrics/flink-metrics-influxdb/src/main/java/org/apache/flink/metrics/influxdb/InfluxdbReporterFactory.java` | MetricReporterFactoryの実装、レポーターインスタンス生成（行29-32） |
| 2-2 | InfluxdbReporter.java | `flink-metrics/flink-metrics-influxdb/src/main/java/org/apache/flink/metrics/influxdb/InfluxdbReporter.java` | メインレポータークラス、open/close/reportメソッド |

**主要処理フロー**:
1. **行67-69**: コンストラクタでMeasurementInfoProviderを設定
2. **行72-113**: open()で接続設定とInfluxDBクライアント初期化
3. **行124-129**: report()でbuildReport呼び出しとwrite実行
4. **行132-159**: buildReport()で全メトリクスをBatchPointsに変換

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | InfluxdbReporterOptions.java | `flink-metrics/flink-metrics-influxdb/src/main/java/org/apache/flink/metrics/influxdb/InfluxdbReporterOptions.java` | 設定オプション定義（HOST, PORT, DB, USERNAME, PASSWORD等） |

**主要処理フロー**:
- **行39-43**: HOST設定オプション定義
- **行51-55**: PORT設定オプション定義（デフォルト8086）
- **行70-74**: DB設定オプション定義
- **行88-98**: タイムアウト設定（CONNECT_TIMEOUT, WRITE_TIMEOUT）

#### Step 4: メトリクスマッピングを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | MetricMapper.java | `flink-metrics/flink-metrics-influxdb/src/main/java/org/apache/flink/metrics/influxdb/MetricMapper.java` | メトリクスからInfluxDB Pointへの変換 |

**主要処理フロー**:
- **行34-43**: Gauge変換処理
- **行45-47**: Counter変換処理
- **行49-64**: Histogram変換処理（統計値全フィールド追加）
- **行66-71**: Meter変換処理
- **行73-77**: 共通Builder生成（measurement名、タグ、タイムスタンプ設定）

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 5-1 | AbstractReporter.java | `flink-metrics/flink-metrics-influxdb/src/main/java/org/apache/flink/metrics/influxdb/AbstractReporter.java` | メトリクス登録/削除のコールバック実装 |

**主要処理フロー**:
- **行44-47**: 各メトリクス種別のMapフィールド定義
- **行55-78**: notifyOfAddedMetric()でメトリクス登録
- **行81-103**: notifyOfRemovedMetric()でメトリクス削除

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

```
InfluxdbReporterFactory.createMetricReporter()
    │
    └─ new InfluxdbReporter()
           │
           ├─ open(MetricConfig)
           │      ├─ getString/getInteger/getDuration (設定読み込み)
           │      ├─ InfluxDBFactory.connect() (クライアント初期化)
           │      └─ log.info() (設定ログ出力)
           │
           ├─ notifyOfAddedMetric(Metric, String, MetricGroup)
           │      └─ AbstractReporter.notifyOfAddedMetric()
           │             └─ metricInfoProvider.getMetricInfo()
           │                    └─ gauges/counters/histograms/meters.put()
           │
           ├─ report() [Scheduled呼び出し]
           │      └─ buildReport()
           │             ├─ BatchPoints.database().retentionPolicy().consistency()
           │             ├─ [for each gauge] MetricMapper.map()
           │             ├─ [for each counter] MetricMapper.map()
           │             ├─ [for each histogram] MetricMapper.map()
           │             ├─ [for each meter] MetricMapper.map()
           │             └─ report.build()
           │      └─ influxDB.write(report)
           │
           └─ close()
                  └─ influxDB.close()
```

### データフロー図

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

Flink Metrics          InfluxdbReporter              InfluxDB
    │                        │                           │
    │ Counter              │                           │
    │ Gauge     ─────────▶ │ notifyOfAddedMetric()    │
    │ Histogram            │ (メトリクス登録)           │
    │ Meter                │                           │
    │                        │                           │
    │                        │ report() [定期実行]       │
    │                        │     │                     │
    │                        │     ▼                     │
    │                        │ buildReport()            │
    │                        │     │                     │
    │                        │     ▼                     │
    │                        │ MetricMapper.map()       │
    │                        │     │                     │
    │                        │     ▼                     │
    │                        │ BatchPoints              │
    │                        │     │                     │
    │                        │     ▼                     │
    │                        │ influxDB.write() ───────▶│ 時系列データ保存
    │                        │                           │
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| InfluxdbReporter.java | `flink-metrics/flink-metrics-influxdb/src/main/java/org/apache/flink/metrics/influxdb/InfluxdbReporter.java` | ソース | メインレポータークラス |
| InfluxdbReporterFactory.java | `flink-metrics/flink-metrics-influxdb/src/main/java/org/apache/flink/metrics/influxdb/InfluxdbReporterFactory.java` | ソース | ファクトリクラス |
| InfluxdbReporterOptions.java | `flink-metrics/flink-metrics-influxdb/src/main/java/org/apache/flink/metrics/influxdb/InfluxdbReporterOptions.java` | ソース | 設定オプション定義 |
| AbstractReporter.java | `flink-metrics/flink-metrics-influxdb/src/main/java/org/apache/flink/metrics/influxdb/AbstractReporter.java` | ソース | 基底レポータークラス |
| MetricMapper.java | `flink-metrics/flink-metrics-influxdb/src/main/java/org/apache/flink/metrics/influxdb/MetricMapper.java` | ソース | メトリクス変換ユーティリティ |
| MeasurementInfo.java | `flink-metrics/flink-metrics-influxdb/src/main/java/org/apache/flink/metrics/influxdb/MeasurementInfo.java` | ソース | メトリクス情報データクラス |
| MeasurementInfoProvider.java | `flink-metrics/flink-metrics-influxdb/src/main/java/org/apache/flink/metrics/influxdb/MeasurementInfoProvider.java` | ソース | メトリクス情報プロバイダー |
| MetricInfoProvider.java | `flink-metrics/flink-metrics-influxdb/src/main/java/org/apache/flink/metrics/influxdb/MetricInfoProvider.java` | ソース | メトリクス情報プロバイダーインターフェース |
