時系列の整列と集約

実装: observability/timeseries/ / 実行: go test ./observability/timeseries/

同じ数字でも、監視の現場にはサーバの数だけ別々の系列として届く。1本の線にするには、まず時間方向に揃え、次に系列方向にまとめる。どちらも数を1つに潰す操作なので混同されやすいが、順序が結果を変える。「下から99%目の遅さ」を台ごとに出してから平均すると、遅い1台が速い多数に薄められて消える。分布のまま足してから取れば消えない。同じデータで違う結論が出る。

この章で作るもの

メトリクスとヒストグラムの章では、1台のプロセスが値をどう持つかを見た。カウンタは増える一方の数を持ち、ヒストグラムはバケットに数を持つ。だが、画面に線が出るまでにはもう2段ある。

まず、系列がたくさんある。同じ request_latencies でも、同じサービスの複製(Kubernetes なら Pod)が 30 個動いていれば、30 本の別々の時系列として届く。ゾーンやリビジョンやレスポンスコードでも分かれるので、実際には数百本になる。

そして、点の時刻が揃っていない。各複製は自分の都合で値を送るので、同じ「10 秒 05 分」の点はどこにも無い。

この2つを片付けないと線が引けない。片付け方が2段あって、時間方向に揃えるのが整列(alignment)、系列方向にまとめるのが集約(reduction)になる。

  届く生データ(点の時刻はばらばら)

  pod-a  ・ ・  ・   ・ ・ ・    ・
  pod-b    ・  ・ ・   ・    ・ ・  ・
  pod-c  ・   ・    ・ ・  ・   ・

         ①整列(aligner)      時間方向に潰す。窓ごとに1点へ
         ↓

  pod-a  ┃    ┃    ┃    ┃      窓の中を mean / max / delta / p99 …
  pod-b  ┃    ┃    ┃    ┃
  pod-c  ┃    ┃    ┃    ┃

         ②集約(reducer)      系列方向に潰す。残すラベルを決める
         ↓

  all    ●    ●    ●    ●      同じ時刻の3点を mean / max / sum / p99 …

時間方向に揃えてから、系列方向にまとめる。どちらも数を潰す操作だが、潰す向きが違う

順に見ていく。

1. 種類が読み方を決める: 累計は差を取らないと読めない。読み方はデータ自身が持っている
2. 分位点は平均できない: 潰す順序を間違えると、遅い1台が消える
3. 平均も窓も、何かを隠す: 平均は刺さった1本を、広い窓は短いスパイクを

① 種類が読み方を決める

同じ「数」でも、3つの種類がある:

// Kind は値が何を表すかの区別。これが整列の意味を決める。
type Kind int

const (
	// Gauge はその瞬間の値。CPU 使用率、キューの長さ。そのまま読める。
	Gauge Kind = iota
	// Delta は前回の点からの増分。区間の合計に意味がある。
	Delta
	// Cumulative は計測開始からの累計。差を取らなければ読めない。
	Cumulative
)

func (k Kind) String() string {
	switch k {
	case Gauge:
		return "GAUGE"
	case Delta:
		return "DELTA"
	case Cumulative:
		return "CUMULATIVE"
	}
	return "UNKNOWN"
}

// Point は1つの観測。分布値のときは D にヒストグラムが入り、V は使わない。
type Point struct {
	T int
	V float64
	D *metrics.Histogram
}

// Series は1本の時系列。Labels がこの系列を他と区別する。
//
// 同じメトリクスでも、Pod やゾーンの数だけ系列が存在する。ラベルの組が
// 違えば別の系列で、集約とは「どのラベルを残すか」を決めることになる。
type Series struct {
	Labels map[string]string
	Kind   Kind
	Points []Point
}

// Distribution はこの系列が分布値を持つかを返す。
func (s Series) Distribution() bool {
	return len(s.Points) > 0 && s.Points[0].D != nil
}

// Label はラベルの値を返す(無ければ空文字)。
func (s Series) Label(k string) string { return s.Labels[k] }

Gauge はその瞬間の値で、CPU 使用率やキューの長さがこれになる。そのまま読める。

Delta は前回からの増分で、区間の合計に意味がある。

Cumulative が厄介で、計測開始からの累計になっている。そのまま描くと、ひたすら右上がりの線しか見えない。リクエスト数が増えているのか減っているのかは、傾きを見なければ分からない。

だから差を取る:

// Aligner は1本の系列の中で、不揃いな点を等間隔の窓に揃える方法。
type Aligner int

const (
	AlignNone  Aligner = iota
	AlignMean          // 窓の中の平均
	AlignMax           // 窓の中の最大
	AlignMin           // 窓の中の最小
	AlignSum           // 窓の中の合計
	AlignDelta         // 窓ぶんの増分。累計を読める形にする
	AlignRate          // 増分を時間で割る。秒あたりに直す
	AlignP50           // 窓の中の中央値
	AlignP99           // 窓の中の 99 パーセンタイル
)

func (a Aligner) String() string {
	return [...]string{"ALIGN_NONE", "ALIGN_MEAN", "ALIGN_MAX", "ALIGN_MIN",
		"ALIGN_SUM", "ALIGN_DELTA", "ALIGN_RATE", "ALIGN_PERCENTILE_50",
		"ALIGN_PERCENTILE_99"}[a]
}

// Align は period ごとの窓に点をまとめる。出力の時刻は窓の終わりになる。
//
// 分布値を持つ系列に AlignDelta を使うと、窓の中のヒストグラムを足し合わせた
// 分布が出る。値が分布のまま残るのが大事なところで、ここで分位点にしてしまうと
// 後の集約で正しく足せなくなる。
func Align(s Series, a Aligner, period int) Series {
	out := Series{Labels: copyLabels(s.Labels), Kind: s.Kind}
	if a == AlignNone || period <= 0 {
		out.Points = append(out.Points, s.Points...)
		return out
	}

	if s.Distribution() {
		// 分布値の系列。窓の中のヒストグラムを足し合わせる。
		// 分位点や平均を指定したときだけ数値に潰れ、それ以外は分布のまま残る。
		for _, w := range windows(s, period) {
			h := mergeHists(w.points)
			switch a {
			case AlignP50:
				out.Points = append(out.Points, Point{T: w.end, V: h.Quantile(0.5)})
			case AlignP99:
				out.Points = append(out.Points, Point{T: w.end, V: h.Quantile(0.99)})
			case AlignMean:
				out.Points = append(out.Points, Point{T: w.end, V: h.Mean()})
			default:
				out.Points = append(out.Points, Point{T: w.end, D: h})
			}
		}
		if collapses(a) {
			out.Kind = Gauge
		}
		return out
	}

	if a == AlignDelta || a == AlignRate {
		return alignChange(s, a, period)
	}

	for _, w := range windows(s, period) {
		out.Points = append(out.Points, Point{T: w.end, V: reduceValues(values(w.points), a)})
	}
	if collapses(a) {
		out.Kind = Gauge // 分位点や平均は、もう増分でも累計でもない
	}
	return out
}

// collapses は、その整列が「1つの数」に潰す種類かを返す。
func collapses(a Aligner) bool { return a == AlignP50 || a == AlignP99 || a == AlignMean }

// alignChange は累計や増分を、窓ぶんの変化量に直す。
//
// 累計は差を取る。ここで前の窓より小さくなっていたら、プロセスが再起動して
// 0 から数え直したということなので、現在値そのものを増分とみなす。この検出を
// 忘れると、再起動のたびに大きな負の値が出る。
func alignChange(s Series, a Aligner, period int) Series {
	out := Series{Labels: copyLabels(s.Labels), Kind: Gauge}
	prev := 0.0
	first := true
	for _, w := range windows(s, period) {
		var v float64
		if s.Kind == Cumulative {
			last := w.points[len(w.points)-1].V
			switch {
			case first:
				v = 0 // 起点が無いので、最初の窓では変化量を出せない
			case last < prev:
				v = last // 数え直しが起きた
			default:
				v = last - prev
			}
			prev, first = last, false
		} else {
			for _, p := range w.points {
				v += p.V
			}
		}
		if a == AlignRate {
			v /= float64(period)
		}
		out.Points = append(out.Points, Point{T: w.end, V: v})
	}
	return out
}

type window struct {
	end    int
	points []Point
}

// windows は点を period ごとの窓に振り分ける。空の窓は作らない。
func windows(s Series, period int) []window {
	byIdx := map[int][]Point{}
	var idxs []int
	for _, p := range s.Points {
		i := p.T / period
		if _, ok := byIdx[i]; !ok {
			idxs = append(idxs, i)
		}
		byIdx[i] = append(byIdx[i], p)
	}
	sort.Ints(idxs)
	out := make([]window, 0, len(idxs))
	for _, i := range idxs {
		out = append(out, window{end: (i + 1) * period, points: byIdx[i]})
	}
	return out
}

func values(ps []Point) []float64 {
	out := make([]float64, 0, len(ps))
	for _, p := range ps {
		out = append(out, p.V)
	}
	return out
}

// reduceValues は数値の並びを1つにまとめる。整列にも集約にも同じ計算を使う。
func reduceValues(vs []float64, a Aligner) float64 {
	if len(vs) == 0 {
		return 0
	}
	switch a {
	case AlignMax:
		m := vs[0]
		for _, v := range vs {
			if v > m {
				m = v
			}
		}
		return m
	case AlignMin:
		m := vs[0]
		for _, v := range vs {
			if v < m {
				m = v
			}
		}
		return m
	case AlignSum:
		s := 0.0
		for _, v := range vs {
			s += v
		}
		return s
	case AlignP50:
		return nearestRank(vs, 0.5)
	case AlignP99:
		return nearestRank(vs, 0.99)
	default: // AlignMean
		s := 0.0
		for _, v := range vs {
			s += v
		}
		return s / float64(len(vs))
	}
}

// nearestRank は並べ替えて下から q の位置の値を返す。
func nearestRank(vs []float64, q float64) float64 {
	sorted := append([]float64(nil), vs...)
	sort.Float64s(sorted)
	i := int(q*float64(len(sorted))+0.999999) - 1
	if i < 0 {
		i = 0
	}
	if i >= len(sorted) {
		i = len(sorted) - 1
	}
	return sorted[i]
}

AlignDelta が窓ぶんの増分、AlignRate がそれを時間で割った秒あたりの値になる。ここまでは素直だが、1つ罠がある。

プロセスが再起動すると、累計は 0 から数え直す。素朴に引き算すると、そこで大きな負の値が出る。「リクエストが毎秒マイナス 1200 件」という線が引かれることになる。だから、前の窓より小さくなっていたら数え直しとみなして、現在値そのものを増分として扱う。テストで、数え直しを挟んでも負の値が出ないことを固定した。

分布値を持つ系列に AlignDelta を使うと、窓の中のヒストグラムを足し合わせた分布が出る。ここで分布のまま残るのが大事で、その理由が次になる。

② 分位点は平均できない

集約はこうなる:

// Reducer は同じ時刻の複数の系列を1つにまとめる方法。
type Reducer int

const (
	ReduceNone Reducer = iota
	ReduceMean
	ReduceSum
	ReduceMax
	ReduceMin
	ReduceP50
	ReduceP99
)

func (r Reducer) String() string {
	return [...]string{"REDUCE_NONE", "REDUCE_MEAN", "REDUCE_SUM", "REDUCE_MAX",
		"REDUCE_MIN", "REDUCE_PERCENTILE_50", "REDUCE_PERCENTILE_99"}[r]
}

// Reduce は groupBy に挙げたラベルだけを残して、残りが同じ系列を1本にまとめる。
//
// 分布値のままの系列に ReduceP99 を使うと、まずヒストグラムを足し合わせ、
// それから分位点を取る。これが正しい順序になる。すでに分位点になってしまった
// 数値を平均しても、全体の分位点にはならない。
func Reduce(all []Series, r Reducer, groupBy ...string) []Series {
	if r == ReduceNone {
		return append([]Series(nil), all...)
	}

	groups := map[string][]Series{}
	var order []string
	for _, s := range all {
		k := groupKey(s, groupBy)
		if _, ok := groups[k]; !ok {
			order = append(order, k)
		}
		groups[k] = append(groups[k], s)
	}
	sort.Strings(order)

	out := make([]Series, 0, len(order))
	for _, k := range order {
		out = append(out, reduceGroup(groups[k], r, groupBy))
	}
	return out
}

func reduceGroup(group []Series, r Reducer, groupBy []string) Series {
	res := Series{Labels: map[string]string{}, Kind: group[0].Kind}
	for _, k := range groupBy {
		res.Labels[k] = group[0].Labels[k]
	}

	byTime := map[int][]Point{}
	var times []int
	for _, s := range group {
		for _, p := range s.Points {
			if _, ok := byTime[p.T]; !ok {
				times = append(times, p.T)
			}
			byTime[p.T] = append(byTime[p.T], p)
		}
	}
	sort.Ints(times)

	for _, t := range times {
		ps := byTime[t]
		if ps[0].D != nil {
			h := mergeHists(ps)
			switch r {
			case ReduceP50:
				res.Points = append(res.Points, Point{T: t, V: h.Quantile(0.5)})
			case ReduceP99:
				res.Points = append(res.Points, Point{T: t, V: h.Quantile(0.99)})
			case ReduceMean:
				res.Points = append(res.Points, Point{T: t, V: h.Mean()})
			default:
				res.Points = append(res.Points, Point{T: t, D: h})
			}
			continue
		}
		res.Points = append(res.Points, Point{T: t, V: reduceValues(values(ps), asAligner(r))})
	}
	if r == ReduceP50 || r == ReduceP99 || r == ReduceMean {
		res.Kind = Gauge
	}
	return res
}

// asAligner は同じ計算を指す整列側の名前に読み替える。
// 時間方向か系列方向かが違うだけで、数の潰し方は同じになる。
func asAligner(r Reducer) Aligner {
	switch r {
	case ReduceSum:
		return AlignSum
	case ReduceMax:
		return AlignMax
	case ReduceMin:
		return AlignMin
	case ReduceP50:
		return AlignP50
	case ReduceP99:
		return AlignP99
	default:
		return AlignMean
	}
}

func groupKey(s Series, groupBy []string) string {
	k := ""
	for _, g := range groupBy {
		k += g + "=" + s.Labels[g] + ";"
	}
	return k
}

Reduce は groupBy に挙げたラベルだけを残し、残りが同じ系列を1本にまとめる。ラベルを1つも残さなければ全体で1本になり、zone だけ残せばゾーンの数だけ線が出る。集約とは「どのラベルを捨てるか」を決めることになっている。

ここで、この章でいちばん間違えられるところが出てくる。

3台の複製があって、2台は速く、1台だけが遅いとする。遅い台では 1 割のリクエストが 900 ミリ秒かかっている。全体の p99(下から 99% 目の応答時間。前章で作った分位点)を知りたい。

素朴には、各台の p99 を出してから平均すればよさそうに見える。だが平均は「3台ぶんの真ん中」を出すので、遅い1台の値が速い2台に薄められる。テストで、この誤ったやり方では 318 ミリ秒、正しいやり方では 835 ミリ秒になることを固定した。2.6 倍違う。

どちらもバケットの中を線形補間した推定値なので、真の値である 900 ミリ秒とは少しずれる。だがずれの向きが違う。正しいほうは推定の誤差ぶんだけ低く出ているだけで、誤ったほうは仕組みとして低く出ている。台数を増やせば増やすほど、誤ったほうはさらに低くなる。

正しい順序は、分布のまま足してから分位点を取ることになる。ヒストグラムはバケットごとに足せるので、3台ぶんを足せば全体の分布が手に入る。そこから 99 パーセンタイルを取れば、全体の 99 パーセンタイルになる。この足し算ができることが、ヒストグラムを使う理由そのものだった。

順序を式にすると、こうなる。

• 間違い: ALIGN_PERCENTILE_99 してから REDUCE_MEAN(分位点を平均している)
• 正しい: ALIGN_DELTA で分布のまま揃えてから REDUCE_PERCENTILE_99

重みの問題も同じところから来る。1万件を捌いた台と、10 件しか来ていない台を系列として等しく扱うと、10 件のほうが半分の重みを持ってしまう。分布のまま足せば件数がそのまま重みになる。テストで、100 件と 1 件を混ぜたときに件数の多い側へ寄ることを固定した。

なぜ整列が先なのかも、ここで分かる。点の時刻が揃っていないまま集約すると、たまたま同じ時刻に点があった系列だけが足され、無かった系列は抜ける。値が変わっていなくても、線は上下する。テストで、10 秒ごとに送る系列と 30 秒ごとに送る系列を揃えずに足すと値が揺れ、先に窓へ揃えると揺れなくなることを固定した。

③ 平均も窓も、何かを隠す

10 台のうち1台だけが 98% まで張り付いているとする。残り9台は 10%。平均は 18.8% になる。閾値 65% で警告を出す設定にしていても、平均で見ている限り一度も鳴らない。テストで、平均は一度も 65 を超えず、最大なら超えることを固定した。

これは平均が間違っているのではなく、平均という問いに対して正しく答えているだけになる。「全体としてどれくらい使っているか」を知りたいなら平均が正しいし、「困っているところがあるか」を知りたいなら最大が正しい。同じデータに違う問いを投げている。

窓の長さも同じ形の話になる。10 秒だけ 610 まで跳ねる系列を、10 秒窓の平均で見れば 610 のまま見える。600 秒窓の平均で見ると 20 まで落ちて、スパイクは消える。テストで、同じデータが窓の長さだけで見え方を変えること、同じ広い窓でも最大で取れば残ることを固定した。

長い窓には理由がある。窓が短いと点の数が増えて重く、細かい揺れで警告が鳴りやすい。だから運用では窓を広げたくなる。広げると、広げたぶんだけ短い事象が見えなくなる。どちらが正しいということはなく、何を見たいかで決まる。

動かす

下のデモは、同じ生データに違う整列と集約を当てる。左が設定で、右が結果の線になる。分位点を先に取るか後に取るかを切り替えると、同じデータから違う数字が出る。窓の長さを動かすと、スパイクが消えたり戻ったりする。

デモ時系列の整列と集約p99: 正 850ms / 誤 320ms

3つの Pod がそれぞれ 1000 件を捌いた。pod-a と pod-b は全件 4ms、pod-c だけ 1 割が 900ms。 全体の p99 を知りたい

pod-aこの系列の p99 = 5ms

pod-bこの系列の p99 = 5ms

pod-cこの系列の p99 = 950ms

ALIGN_PERCENTILE_99 → REDUCE_MEAN

先に系列ごとの p99 にしてから平均する

320 ms

遅い1台の値が、速い2台に薄められた

ALIGN_DELTA → REDUCE_PERCENTILE_99

分布のまま足してから p99 を取る

850 ms

3000 件ぶんの分布から取った、本当の p99

同じデータから 2.7 倍違う数字が出ている。 分位点は足せないので、潰すのはいちばん最後にする

3つの場面はどれも、データではなく見方の話になっている。分位点は潰す順序で、CPU はどの問いを投げるかで、 スパイクは窓の長さで結論が変わる。Cloud Monitoring では、これが perSeriesAligner と crossSeriesReducer と alignmentPeriod という3つの設定として画面に並んでいる。

実物ではどう呼ばれているか

ここまでの2段は、Cloud Monitoring がそのまま持っている概念になる。API の aggregation には perSeriesAligner と crossSeriesReducer があり、名前も役割も対応している。整列が先で集約が後、という順序まで API の制約として書かれている。

対応の一覧と、Spanner や Pub/Sub といった製品ごとの読み方は、Cloud Monitoring で読み替えるで扱う。

設計の観点

• 潰す前に何を知りたいか決める: 平均も最大も分位点も、それぞれ違う問いへの正しい答えになる。問いを決めずに選ぶと、答えを誤読する
• 潰せる形のまま運ぶ: 分布は足せるが、分位点は足せない。潰すのはいちばん最後にする
• 種類は捨てない: 累計を「ただの数」として扱うと、差を取り忘れるか、再起動で負が出る
• 窓は見たいものの長さで決める: 10 秒の事象を見たいなら窓は 10 秒より短くする。長い窓は軽いが、短い事象を消す
• 平均と最大を並べる: どちらか一方だけを置いたダッシュボードは、必ず片方の見落としを持つ
• ヒストグラムを前提にする: 足せる形で持っておくことが、後の自由度になる

対照と実例

見たいこと	整列	集約	間違えるとどうなるか
秒あたりのリクエスト数	ALIGN_RATE	REDUCE_SUM	累計のまま描くと右上がりの線しか出ない
全体の p99 レイテンシ	ALIGN_DELTA(分布のまま)	REDUCE_PERCENTILE_99	先に p99 にすると遅い1台が薄まる
困っているインスタンスの有無	ALIGN_MAX	REDUCE_MAX	平均だと1台の張り付きが消える
全体としての使用量	ALIGN_MEAN	REDUCE_MEAN	最大だと1台の瞬間値に引っ張られる
エラー率	ALIGN_RATE	REDUCE_SUM を分子と分母それぞれに	系列ごとの比率を平均すると件数の重みが消える
短いスパイクの検出	ALIGN_MAX + 短い窓	REDUCE_MAX	長い窓の平均だと平滑化されて見えない

裏どり:

• Cloud Monitoring の集約: perSeriesAligner が先、crossSeriesReducer が後。系列をまたぐ集約には ALIGN_NONE 以外の整列が必要
• 保持期間: 指標は 6 週間。それより長く見るには API で取り出して BigQuery などに書き出す
• 書き方は 3 通りある: コンソールの GUI、MQL、PromQL。語彙が違うだけで、整列してから集約するという構造は変わらない
• 製品ごとの読み方は別章: Spanner の閾値、Pub/Sub の 2 つ組、BigQuery のスロットといった各論はCloud Monitoring で読み替えるにまとめた

簡略化したこと

• 時刻は整数: 実物はナノ秒精度のタイムスタンプと、区間の始点と終点を持つ
• 欠測を埋めない: 実物は点が無い窓の扱い(補間するか空けるか)を選べる
• 分位点は補間: バケットの中を線形補間するので厳密値ではない。ヒストグラムの章と同じ性質になる
• 数値への分位点は近似順位: 分布値でない系列への AlignP99 は、窓の中の値を並べて位置で取る
• アラートなし: 閾値、継続時間、通知は扱わない。線を出すところまで
• 書き出しなし: 指標をどこかへ持っていく話は文章だけで、実装していない

参考資料

• Filtering and aggregation — 整列と集約の順序、種類ごとの制約
• Aligner / Reducer の一覧 — enum の全一覧と適用条件
• 実装: observability/timeseries