# 帳票設計書 14-OpenTelemetryメトリクスレポート

## 概要

本ドキュメントは、Apache FlinkのOpenTelemetryメトリクスレポーター機能に関する設計書である。FlinkのメトリクスをOpenTelemetry CollectorへOTLP（OpenTelemetry Protocol）を使用してエクスポートするための仕組みを定義する。

### 本帳票の処理概要

OpenTelemetryメトリクスレポートは、Apache Flinkで収集された各種メトリクス（Counter、Gauge、Histogram、Meter）をOpenTelemetry SDKを使用してOTLP形式に変換し、gRPCまたはHTTP経由でOpenTelemetry Collectorに送信する機能である。

**業務上の目的・背景**：OpenTelemetryは、メトリクス、トレース、ログを統一的に扱うための業界標準オブザーバビリティフレームワークである。CNCFのincubatingプロジェクトとして広く採用されており、Prometheus、Jaeger、Datadog、New Relic、Splunkなど多数のバックエンドとの相互運用性を持つ。本レポーターにより、Flinkを含む分散システム全体でのエンドツーエンドの可観測性を実現し、マイクロサービスアーキテクチャでの統一的な監視基盤を構築できる。

**帳票の利用シーン**：本レポーターは以下のシーンで利用される。(1)OpenTelemetry Collectorを中核とした統合監視基盤への接続、(2)クラウドネイティブ環境でのベンダー中立なメトリクス収集、(3)トレースとメトリクスの相関分析、(4)Kubernetesクラスタでの標準的なオブザーバビリティスタック統合、(5)複数の監視バックエンドへの同時データ配信（Collector経由）。

**主要な出力内容**：
1. Counter - Delta AggregationTemporalityでLongSumとして出力
2. Gauge - LongGaugeまたはDoubleGaugeとして出力
3. Histogram - DoubleSummaryとして出力（min, max, p50, p75, p95, p99パーセンタイル含む）
4. Meter - count（Delta Sum）とrate（DoubleGauge）の2メトリクスとして出力

**帳票の出力タイミング**：スケジュールされた間隔で定期的にメトリクスがCollectorへプッシュされる。`Scheduled`インターフェースの実装により、`report()`メソッドが周期的に呼び出され、`collectAllMetrics()`でメトリクスを収集し、`exporter.export()`で送信する。

**帳票の利用者**：Flinkクラスタ運用者、SRE、プラットフォームエンジニア、クラウドネイティブアーキテクト。

## 帳票種別

OTLP（OpenTelemetry Protocol）メトリクスデータ

## 利用画面

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

## 出力形式

### 基本仕様

| 項目 | 内容 |
|-----|------|
| ファイル形式 | OTLP（Protocol Buffers） |
| 用紙サイズ | N/A |
| 向き | N/A |
| ファイル名 | N/A（ネットワーク送信） |
| 出力方法 | gRPC / HTTP POST |
| 文字コード | UTF-8 |

### OpenTelemetry固有設定

| 項目 | 内容 |
|-----|------|
| プロトコル | gRPC（デフォルト）/ HTTP |
| エンドポイント | 必須設定 |
| タイムアウト | オプション（Duration形式） |
| サービス名 | オプション（ResourceAttributesに設定） |
| サービスバージョン | オプション（ResourceAttributesに設定） |

## 帳票レイアウト

### レイアウト概要

OpenTelemetry MetricData形式でメトリクスが構造化される。

```
Resource (service.name, service.version)
└── InstrumentationScopeInfo (io.confluent.flink.common.metrics)
    └── MetricData
        ├── name: flink.<scope>.<metricName>
        ├── type: LongSum / LongGauge / DoubleGauge / DoubleSummary
        └── DataPoint (startEpochNanos, epochNanos, attributes, value)
```

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

#### Counter (LongSum)

| No | 項目名 | 説明 | データ取得元 | 型 |
|----|-------|------|-------------|-----|
| 1 | name | メトリクス名 | flink.<scope>.<metricName> | String |
| 2 | value | デルタ値（現在値 - 前回値） | counter.getCount() - previousCount | Long |
| 3 | aggregationTemporality | 集約方式 | DELTA | AggregationTemporality |
| 4 | isMonotonic | 単調増加フラグ | true | Boolean |

#### Gauge (LongGauge / DoubleGauge)

| No | 項目名 | 説明 | データ取得元 | 型 |
|----|-------|------|-------------|-----|
| 1 | name | メトリクス名 | flink.<scope>.<metricName> | String |
| 2 | value | 現在値 | gauge.getValue() | Long/Double |

#### Histogram (DoubleSummary)

| No | 項目名 | 説明 | データ取得元 | 型 |
|----|-------|------|-------------|-----|
| 1 | name | メトリクス名 | flink.<scope>.<metricName> | String |
| 2 | count | サンプル数 | histogram.getCount() | Long |
| 3 | sum | 合計値 | mean * count | Double |
| 4 | quantiles | パーセンタイル | [0(min), 0.5, 0.75, 0.95, 0.99, 1(max)] | List<ValueAtQuantile> |

#### Meter (LongSum + DoubleGauge)

| No | 項目名 | 説明 | データ取得元 | 型 |
|----|-------|------|-------------|-----|
| 1 | {name}.count | デルタカウント | meter.getCount() - previousCount | Long (LongSum) |
| 2 | {name}.rate | レート | meter.getRate() | Double (DoubleGauge) |

## 出力条件

### 抽出条件

| 条件名 | 説明 | 必須 |
|-------|------|-----|
| exporter.endpoint | CollectorエンドポイントURL | Yes |
| exporter.protocol | 通信プロトコル（gRPC/HTTP） | No（デフォルト: gRPC） |
| exporter.timeout | タイムアウト（Duration形式） | No |
| service.name | サービス名（Resourceに設定） | No |
| service.version | サービスバージョン（Resourceに設定） | No |

### ソート順

| 優先度 | 項目 | 昇順/降順 |
|-------|------|---------|
| N/A | 収集順 | 登録順 |

### 改ページ条件

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

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

### 参照テーブル一覧

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

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

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

| 参照項目 | 帳票項目との対応 | 取得条件 | 備考 |
|---------|----------------|---------|------|
| gauges | Gauge出力 | 全登録Gauge | Map<Gauge<?>, MetricMetadata> |
| counters | Counter出力 | 全登録Counter | Map<Counter, MetricMetadata> |
| histograms | Histogram出力 | 全登録Histogram | Map<Histogram, MetricMetadata> |
| meters | Meter出力 | 全登録Meter | Map<Meter, MetricMetadata> |
| lastValueSnapshots | 前回値スナップショット | Counter/Meterの前回値 | Map<Metric, Long> |

## 計算仕様

### 計算項目一覧

| 項目名 | 計算式 | 端数処理 | 備考 |
|-------|-------|---------|------|
| デルタ値 | currentCount - previousCount | N/A | 非単調は警告ログ |
| 合計値（Histogram） | mean * count | N/A | サマリー用 |
| 現在時刻（ナノ秒） | TimeUnit.SECONDS.toNanos(epochSecond) + nano | N/A | Clock.systemUTC() |

## 処理フロー

### 出力フロー

```mermaid
flowchart TD
    A[Scheduledによるreport呼び出し] --> B[collectAllMetrics実行]
    B --> C[getCurrentTimeNanos取得]
    C --> D[CollectionMetadata作成]
    D --> E[takeLastValueSnapshots実行]
    E --> F{Countersループ}
    F --> G[convertCounter実行]
    G --> H{Gaugesループ}
    H --> I[convertGauge実行]
    I --> J{Metersループ}
    J --> K[convertMeter実行]
    K --> L{Histogramsループ}
    L --> M[convertHistogram実行]
    M --> N[lastValueSnapshots更新]
    N --> O[exporter.export実行]
    O --> P{成功?}
    P -->|Yes| Q[デバッグログ出力]
    P -->|No| R[警告ログ出力]
    Q --> S[終了]
    R --> S
```

### open処理フロー

```mermaid
flowchart TD
    A[open呼び出し] --> B[super.open実行]
    B --> C[Resource設定]
    C --> D{protocol判定}
    D -->|http| E[OtlpHttpMetricExporter.builder]
    D -->|gRPC/default| F[OtlpGrpcMetricExporter.builder]
    E --> G[endpoint/timeout設定]
    F --> G
    G --> H[exporter = builder.build]
    H --> I[終了]
```

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 表示メッセージ | 対処方法 |
|----------|---------|--------------|---------|
| 設定エラー | endpoint未設定 | Must set exporter.endpoint | endpoint設定の追加 |
| 非単調Counter | 現在値 < 前回値 | Non-monotonic counter {name}: current count {x} is less than previous count {y} | データ確認/無視 |
| エクスポート失敗 | export()失敗 | Failed to export {count} metrics using {exporter} | Collector確認 |
| export例外 | export()例外 | Failed to call export for {count} metrics using {exporter} | ネットワーク確認 |

## パフォーマンス要件

| 項目 | 内容 |
|-----|------|
| 想定データ件数 | メトリクス数に依存（数百〜数千） |
| 目標出力時間 | タイムアウト設定に依存 |
| 同時出力数上限 | 1（シングルスレッド送信、非同期完了） |

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

- gRPC/HTTPSによる通信暗号化を推奨
- OpenTelemetry Collectorでの認証・認可設定
- ネットワークセグメンテーションによるアクセス制御
- service.name/service.versionでのメトリクス分離

## 備考

- OpenTelemetry SDK（io.opentelemetry）を使用
- Delta AggregationTemporalityでCounter/Meterを出力（前回値からの差分）
- lastValueSnapshotsで前回収集時の値を保持
- Clockインジェクションによるテスタビリティ確保
- close()時にflush()とwaitForLastReportToComplete()を実行

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | MetricMetadata.java | `flink-metrics/flink-metrics-otel/src/main/java/org/apache/flink/metrics/otel/MetricMetadata.java` | メトリクス名と変数を保持するデータクラス（行27-50） |
| 1-2 | OpenTelemetryReporterOptions.java | `flink-metrics/flink-metrics-otel/src/main/java/org/apache/flink/metrics/otel/OpenTelemetryReporterOptions.java` | 設定オプション定義（EXPORTER_PROTOCOL, EXPORTER_ENDPOINT等） |

**読解のコツ**: MetricMetadataはFlinkメトリクスからOpenTelemetryメトリクスへの変換に必要な情報を保持する。

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | OpenTelemetryReporterBase.java | `flink-metrics/flink-metrics-otel/src/main/java/org/apache/flink/metrics/otel/OpenTelemetryReporterBase.java` | Resource設定、共通open()処理（行35-70） |

**主要処理フロー**:
- **行41-43**: コンストラクタでResource.getDefault()
- **行45-68**: open()でservice.name/service.version設定

#### Step 3: メインレポータークラスを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | OpenTelemetryMetricReporter.java | `flink-metrics/flink-metrics-otel/src/main/java/org/apache/flink/metrics/otel/OpenTelemetryMetricReporter.java` | 全体構造、フィールド定義（行68-91） |

**主要処理フロー**:
- **行74-77**: gauges/counters/histograms/metersのMap定義
- **行81-82**: lastValueSnapshots/lastCollectTimeNanos（デルタ計算用）
- **行84-91**: コンストラクタ（Clock設定）

#### Step 4: open/close処理を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | OpenTelemetryMetricReporter.java | `flink-metrics/flink-metrics-otel/src/main/java/org/apache/flink/metrics/otel/OpenTelemetryMetricReporter.java` | open()（行94-123）、close()（行126-132） |

**主要処理フロー**:
- **行98-102**: protocol設定取得
- **行104-122**: switch(protocol)でHTTP/gRPCエクスポーター生成
- **行126-132**: close()でflush/wait/close

#### Step 5: メトリクス登録処理を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 5-1 | OpenTelemetryMetricReporter.java | `flink-metrics/flink-metrics-otel/src/main/java/org/apache/flink/metrics/otel/OpenTelemetryMetricReporter.java` | notifyOfAddedMetric（行135-174）、notifyOfRemovedMetric（行176-200） |

**主要処理フロー**:
- **行136-141**: メトリクス名生成（LOGICAL_SCOPE_PREFIX + scope + metricName）
- **行143-148**: 変数マップ変換（VariableNameUtil使用）
- **行151**: MetricMetadata作成
- **行153-173**: switch(metric.getMetricType())でMap登録

#### Step 6: collectAllMetrics処理を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 6-1 | OpenTelemetryMetricReporter.java | `flink-metrics/flink-metrics-otel/src/main/java/org/apache/flink/metrics/otel/OpenTelemetryMetricReporter.java` | collectAllMetrics()（行214-256） |

**主要処理フロー**:
- **行215-220**: CollectionMetadata作成
- **行221-229**: Counterループ（デルタ計算含む）
- **行230-236**: Gaugeループ
- **行237-245**: Meterループ（デルタ計算含む）
- **行246-252**: Histogramループ
- **行253-254**: スナップショット更新

#### Step 7: メトリクス変換処理を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 7-1 | OpenTelemetryMetricAdapter.java | `flink-metrics/flink-metrics-otel/src/main/java/org/apache/flink/metrics/otel/OpenTelemetryMetricAdapter.java` | 各convert*メソッド |

**主要処理フロー**:
- **行60-91**: convertCounter - LongSum（デルタ、単調）
- **行101-143**: convertGauge - LongGauge/DoubleGauge
- **行153-165**: convertMeter - count + rate
- **行175-206**: convertHistogram - DoubleSummary（quantiles含む）

#### Step 8: report処理を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 8-1 | OpenTelemetryMetricReporter.java | `flink-metrics/flink-metrics-otel/src/main/java/org/apache/flink/metrics/otel/OpenTelemetryMetricReporter.java` | report()（行272-296） |

**主要処理フロー**:
- **行273**: collectAllMetrics()呼び出し
- **行275**: exporter.export()実行
- **行276-289**: 完了コールバック（成功/失敗ログ）

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

```
OpenTelemetryMetricReporterFactory.createMetricReporter(Properties)
    │
    └─ new OpenTelemetryMetricReporter()
           │
           ├─ open(MetricConfig)
           │      ├─ super.open() [OpenTelemetryReporterBase]
           │      │      └─ Resource.merge() for service.name/version
           │      │
           │      └─ switch(protocol)
           │             ├─ "http" → OtlpHttpMetricExporter.builder()
           │             └─ "grpc" → OtlpGrpcMetricExporter.builder()
           │                    └─ setEndpoint() / setTimeout()
           │                           └─ build()
           │
           ├─ notifyOfAddedMetric(Metric, String, MetricGroup)
           │      ├─ LOGICAL_SCOPE_PREFIX + getLogicalScope() + metricName
           │      ├─ VariableNameUtil.getVariableName() [変数変換]
           │      ├─ new MetricMetadata(name, variables)
           │      └─ gauges/counters/histograms/meters.put()
           │
           ├─ report() [Scheduled呼び出し]
           │      ├─ collectAllMetrics()
           │      │      ├─ getCurrentTimeNanos()
           │      │      ├─ new CollectionMetadata(resource, lastTime, currentTime)
           │      │      ├─ takeLastValueSnapshots()
           │      │      │
           │      │      ├─ [for each counter]
           │      │      │      └─ OpenTelemetryMetricAdapter.convertCounter()
           │      │      │
           │      │      ├─ [for each gauge]
           │      │      │      └─ OpenTelemetryMetricAdapter.convertGauge()
           │      │      │
           │      │      ├─ [for each meter]
           │      │      │      └─ OpenTelemetryMetricAdapter.convertMeter()
           │      │      │             ├─ convertCounter() for count
           │      │      │             └─ convertGauge() for rate
           │      │      │
           │      │      └─ [for each histogram]
           │      │             └─ OpenTelemetryMetricAdapter.convertHistogram()
           │      │
           │      └─ exporter.export(metricData)
           │             └─ lastResult.whenComplete()
           │
           └─ close()
                  ├─ exporter.flush()
                  ├─ waitForLastReportToComplete()
                  └─ exporter.close()
```

### データフロー図

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

Flink Metrics           OpenTelemetryMetricReporter      OTLP Collector
    │                          │                               │
    │ Counter                 │                               │
    │ Gauge      ───────────▶│ notifyOfAddedMetric()         │
    │ Histogram               │     │                         │
    │ Meter                   │     ▼                         │
    │                          │ MetricMetadata作成           │
    │                          │ (name, variables)            │
    │                          │     │                         │
    │                          │     ▼                         │
    │                          │ Mapに登録                     │
    │                          │                               │
    │                          │ report() [定期実行]           │
    │                          │     │                         │
    │                          │     ▼                         │
    │                          │ collectAllMetrics()          │
    │                          │     │                         │
    │                          │     ▼                         │
    │                          │ OpenTelemetryMetricAdapter   │
    │                          │ convert*()                   │
    │                          │     │                         │
    │                          │     ▼                         │
    │                          │ Collection<MetricData>       │
    │                          │     │                         │
    │                          │     ▼                         │
    │                          │ exporter.export() ──────────▶│ OTLP受信
    │                          │ (gRPC/HTTP)                   │
    │                          │                               │
                                                              ▼
                                                    バックエンド
                                                    (Prometheus,
                                                     Jaeger, etc.)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| OpenTelemetryMetricReporter.java | `flink-metrics/flink-metrics-otel/src/main/java/org/apache/flink/metrics/otel/OpenTelemetryMetricReporter.java` | ソース | メインレポータークラス |
| OpenTelemetryMetricReporterFactory.java | `flink-metrics/flink-metrics-otel/src/main/java/org/apache/flink/metrics/otel/OpenTelemetryMetricReporterFactory.java` | ソース | ファクトリクラス |
| OpenTelemetryReporterBase.java | `flink-metrics/flink-metrics-otel/src/main/java/org/apache/flink/metrics/otel/OpenTelemetryReporterBase.java` | ソース | 基底クラス（Resource設定） |
| OpenTelemetryReporterOptions.java | `flink-metrics/flink-metrics-otel/src/main/java/org/apache/flink/metrics/otel/OpenTelemetryReporterOptions.java` | ソース | 設定オプション定義 |
| OpenTelemetryMetricAdapter.java | `flink-metrics/flink-metrics-otel/src/main/java/org/apache/flink/metrics/otel/OpenTelemetryMetricAdapter.java` | ソース | メトリクス変換アダプター |
| MetricMetadata.java | `flink-metrics/flink-metrics-otel/src/main/java/org/apache/flink/metrics/otel/MetricMetadata.java` | ソース | メトリクスメタデータクラス |
| VariableNameUtil.java | `flink-metrics/flink-metrics-otel/src/main/java/org/apache/flink/metrics/otel/VariableNameUtil.java` | ソース | 変数名変換ユーティリティ |
