| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | package metrix |
| 4 | |
| 5 | import ( |
| 6 | "fmt" |
| 7 | "hash/fnv" |
| 8 | "math" |
| 9 | "sort" |
| 10 | "strconv" |
| 11 | ) |
| 12 | |
| 13 | const SummaryQuantileLabel = "quantile" |
| 14 | const defaultSummaryReservoirSize = 1024 |
| 15 | const initialSummaryReservoirCapacity = 64 |
| 16 | |
| 17 | // snapshotSummaryInstrument writes sampled full summary points. |
| 18 | type snapshotSummaryInstrument struct { |
| 19 | backend meterBackend |
| 20 | desc *instrumentDescriptor |
| 21 | scope HostScope |
| 22 | base []LabelSet |
| 23 | } |
| 24 | |
| 25 | // statefulSummaryInstrument writes observed samples into maintained summary state. |
| 26 | type statefulSummaryInstrument struct { |
| 27 | backend meterBackend |
| 28 | desc *instrumentDescriptor |
| 29 | scope HostScope |
| 30 | base []LabelSet |
| 31 | } |
| 32 | |
| 33 | // stagedSummary holds one in-cycle summary sample for a single series identity. |
| 34 | type stagedSummary struct { |
| 35 | key string |
| 36 | name string |
| 37 | hostScopeKey string |
| 38 | hostScope HostScope |
| 39 | labels []Label |
| 40 | labelsKey string |
| 41 | desc *instrumentDescriptor |
| 42 | |
| 43 | count SampleValue |
| 44 | sum SampleValue |
| 45 | quantileValues []SampleValue |
| 46 | sketch *summaryQuantileSketch |
| 47 | } |
| 48 | |
| 49 | // Summary declares or reuses a snapshot summary under this meter. |
| 50 | func (m *snapshotMeter) Summary(name string, opts ...InstrumentOption) SnapshotSummary { |
| 51 | desc, err := m.backend.registerInstrument(metricName(m.prefix, name), kindSummary, modeSnapshot, opts...) |
| 52 | if err != nil { |
| 53 | panic(err) |
| 54 | } |
| 55 | return &snapshotSummaryInstrument{ |
| 56 | backend: m.backend, |
| 57 | desc: desc, |
| 58 | scope: m.scope, |
| 59 | base: appendLabelSets(m.sets, nil), |
| 60 | } |
| 61 | } |
| 62 | |
| 63 | // Summary declares or reuses a stateful summary under this meter. |
| 64 | func (m *statefulMeter) Summary(name string, opts ...InstrumentOption) StatefulSummary { |
| 65 | desc, err := m.backend.registerInstrument(metricName(m.prefix, name), kindSummary, modeStateful, opts...) |
| 66 | if err != nil { |
| 67 | panic(err) |
| 68 | } |
| 69 | return &statefulSummaryInstrument{ |
| 70 | backend: m.backend, |
| 71 | desc: desc, |
| 72 | scope: m.scope, |
| 73 | base: appendLabelSets(m.sets, nil), |
| 74 | } |
| 75 | } |
| 76 | |
| 77 | // ObservePoint writes one full summary point for this collect cycle. |
| 78 | func (s *snapshotSummaryInstrument) ObservePoint(p SummaryPoint, labels ...LabelSet) { |
| 79 | s.backend.recordSummaryObservePoint(s.desc, s.scope, p, appendLabelSets(s.base, labels)) |
| 80 | } |
| 81 | |
| 82 | // Observe adds one sample to a stateful summary for this collect cycle. |
| 83 | func (s *statefulSummaryInstrument) Observe(v SampleValue, labels ...LabelSet) { |
| 84 | s.backend.recordSummaryObserve(s.desc, s.scope, v, appendLabelSets(s.base, labels)) |
| 85 | } |
| 86 | |
| 87 | // recordSummaryObservePoint writes one full summary point into the active frame. |
| 88 | func (c *storeCore) recordSummaryObservePoint(desc *instrumentDescriptor, scope HostScope, point SummaryPoint, sets []LabelSet) { |
| 89 | c.mu.Lock() |
| 90 | defer c.mu.Unlock() |
| 91 | |
| 92 | if c.active == nil { |
| 93 | panic(errCycleInactive) |
| 94 | } |
| 95 | |
| 96 | labels, labelsKey, err := labelsFromSet(sets, c) |
| 97 | if err != nil { |
| 98 | panic(err) |
| 99 | } |
| 100 | if labelsContainKey(labels, SummaryQuantileLabel) { |
| 101 | panic(errSummaryLabelKey) |
| 102 | } |
| 103 | scope, ok := c.prepareHostScopeForWriteLocked(scope) |
| 104 | if !ok { |
| 105 | return |
| 106 | } |
| 107 | |
| 108 | count, sum, quantiles := normalizeSummaryPoint(point, desc.summary) |
| 109 | |
| 110 | key := makeSeriesKey(scope.ScopeKey, desc.name, labelsKey) |
| 111 | entry, ok := c.active.summaries[key] |
| 112 | if !ok { |
| 113 | entry = &stagedSummary{ |
| 114 | key: key, |
| 115 | name: desc.name, |
| 116 | hostScopeKey: scope.ScopeKey, |
| 117 | hostScope: scope, |
| 118 | labels: labels, |
| 119 | labelsKey: labelsKey, |
| 120 | desc: desc, |
| 121 | } |
| 122 | c.active.summaries[key] = entry |
| 123 | } |
| 124 | |
| 125 | entry.count = count |
| 126 | entry.sum = sum |
| 127 | entry.quantileValues = append(entry.quantileValues[:0], quantiles...) |
| 128 | entry.sketch = nil |
| 129 | } |
| 130 | |
| 131 | // recordSummaryObserve adds one sample to a stateful summary in the active frame. |
| 132 | func (c *storeCore) recordSummaryObserve(desc *instrumentDescriptor, scope HostScope, value SampleValue, sets []LabelSet) { |
| 133 | mustFiniteSample(value) |
| 134 | |
| 135 | c.mu.Lock() |
| 136 | defer c.mu.Unlock() |
| 137 | |
| 138 | if c.active == nil { |
| 139 | panic(errCycleInactive) |
| 140 | } |
| 141 | |
| 142 | labels, labelsKey, err := labelsFromSet(sets, c) |
| 143 | if err != nil { |
| 144 | panic(err) |
| 145 | } |
| 146 | if labelsContainKey(labels, SummaryQuantileLabel) { |
| 147 | panic(errSummaryLabelKey) |
| 148 | } |
| 149 | scope, ok := c.prepareHostScopeForWriteLocked(scope) |
| 150 | if !ok { |
| 151 | return |
| 152 | } |
| 153 | |
| 154 | key := makeSeriesKey(scope.ScopeKey, desc.name, labelsKey) |
| 155 | entry, ok := c.active.summaries[key] |
| 156 | if !ok { |
| 157 | entry = &stagedSummary{ |
| 158 | key: key, |
| 159 | name: desc.name, |
| 160 | hostScopeKey: scope.ScopeKey, |
| 161 | hostScope: scope, |
| 162 | labels: labels, |
| 163 | labelsKey: labelsKey, |
| 164 | desc: desc, |
| 165 | } |
| 166 | if desc.window == WindowCumulative { |
| 167 | if existing := c.snapshot.Load().series[key]; existing != nil && existing.desc != nil && existing.desc.kind == kindSummary { |
| 168 | entry.count = existing.summaryCount |
| 169 | entry.sum = existing.summarySum |
| 170 | if len(desc.summaryQuantiles()) > 0 && existing.summarySketch != nil { |
| 171 | entry.sketch = existing.summarySketch.clone() |
| 172 | } |
| 173 | } |
| 174 | } |
| 175 | if len(desc.summaryQuantiles()) > 0 && entry.sketch == nil { |
| 176 | entry.sketch = newSummaryQuantileSketch(desc.summaryReservoirSize(), summarySketchSeed(key)) |
| 177 | } |
| 178 | c.active.summaries[key] = entry |
| 179 | } |
| 180 | |
| 181 | entry.count++ |
| 182 | entry.sum += value |
| 183 | if len(desc.summaryQuantiles()) > 0 { |
| 184 | entry.sketch.observe(value) |
| 185 | } |
| 186 | } |
| 187 | |
| 188 | func normalizeSummaryPoint(point SummaryPoint, schema *summarySchema) (SampleValue, SampleValue, []SampleValue) { |
| 189 | mustFiniteSample(point.Count) |
| 190 | mustFiniteSample(point.Sum) |
| 191 | |
| 192 | if point.Count < 0 { |
| 193 | panic(fmt.Errorf("%w: negative count", errSummaryPoint)) |
| 194 | } |
| 195 | |
| 196 | if schema == nil || len(schema.quantiles) == 0 { |
| 197 | if len(point.Quantiles) > 0 { |
| 198 | panic(fmt.Errorf("%w: quantiles are not configured for this instrument", errSummaryPoint)) |
| 199 | } |
| 200 | return point.Count, point.Sum, nil |
| 201 | } |
| 202 | |
| 203 | if len(point.Quantiles) != len(schema.quantiles) { |
| 204 | panic(fmt.Errorf("%w: quantile count mismatch", errSummaryPoint)) |
| 205 | } |
| 206 | |
| 207 | values := make([]SampleValue, len(schema.quantiles)) |
| 208 | seen := make(map[float64]struct{}, len(schema.quantiles)) |
| 209 | for _, q := range point.Quantiles { |
| 210 | if math.IsNaN(q.Quantile) || q.Quantile < 0 || q.Quantile > 1 { |
| 211 | panic(fmt.Errorf("%w: invalid quantile %v", errSummaryPoint, q.Quantile)) |
| 212 | } |
| 213 | if _, ok := seen[q.Quantile]; ok { |
| 214 | panic(fmt.Errorf("%w: duplicate quantile %v", errSummaryPoint, q.Quantile)) |
| 215 | } |
| 216 | seen[q.Quantile] = struct{}{} |
| 217 | |
| 218 | idx := summaryQuantileIndex(schema.quantiles, q.Quantile) |
| 219 | if idx == -1 { |
| 220 | panic(fmt.Errorf("%w: quantile %v is not declared", errSummaryPoint, q.Quantile)) |
| 221 | } |
| 222 | // A summary may report a NaN quantile value (e.g. an empty observation window). Store it; |
| 223 | // chartengine renders a non-finite dimension value as a gap (SETEMPTY). Reject only Inf. |
| 224 | if math.IsInf(float64(q.Value), 0) { |
| 225 | panic(fmt.Errorf("%w: infinite quantile value %v", errSummaryPoint, q.Value)) |
| 226 | } |
| 227 | values[idx] = q.Value |
| 228 | } |
| 229 | |
| 230 | if len(seen) != len(schema.quantiles) { |
| 231 | panic(fmt.Errorf("%w: missing quantiles in point", errSummaryPoint)) |
| 232 | } |
| 233 | |
| 234 | return point.Count, point.Sum, values |
| 235 | } |
| 236 | |
| 237 | func summaryQuantileIndex(schema []float64, q float64) int { |
| 238 | i := sort.SearchFloat64s(schema, q) |
| 239 | if i < len(schema) && schema[i] == q { |
| 240 | return i |
| 241 | } |
| 242 | return -1 |
| 243 | } |
| 244 | |
| 245 | func nanSummaryQuantiles(quantiles []float64) []SampleValue { |
| 246 | if len(quantiles) == 0 { |
| 247 | return nil |
| 248 | } |
| 249 | out := make([]SampleValue, len(quantiles)) |
| 250 | for i := range out { |
| 251 | out[i] = math.NaN() |
| 252 | } |
| 253 | return out |
| 254 | } |
| 255 | |
| 256 | func formatSummaryQuantileLabel(v float64) string { |
| 257 | return strconv.FormatFloat(v, 'g', -1, 64) |
| 258 | } |
| 259 | |
| 260 | func (d *instrumentDescriptor) summaryQuantiles() []float64 { |
| 261 | if d == nil || d.summary == nil { |
| 262 | return nil |
| 263 | } |
| 264 | return d.summary.quantiles |
| 265 | } |
| 266 | |
| 267 | func (d *instrumentDescriptor) summaryReservoirSize() int { |
| 268 | if d == nil || d.summary == nil || d.summary.reservoirSize <= 0 { |
| 269 | return defaultSummaryReservoirSize |
| 270 | } |
| 271 | return d.summary.reservoirSize |
| 272 | } |
| 273 | |
| 274 | // summaryQuantileSketch keeps bounded-memory approximate quantiles. |
| 275 | // It uses reservoir sampling, which is deterministic per series key seed. |
| 276 | type summaryQuantileSketch struct { |
| 277 | capacity int |
| 278 | count uint64 |
| 279 | rng uint64 |
| 280 | values []SampleValue |
| 281 | scratch []SampleValue |
| 282 | } |
| 283 | |
| 284 | func newSummaryQuantileSketch(capacity int, seed uint64) *summaryQuantileSketch { |
| 285 | if capacity <= 0 { |
| 286 | capacity = defaultSummaryReservoirSize |
| 287 | } |
| 288 | if seed == 0 { |
| 289 | seed = 1 |
| 290 | } |
| 291 | initCap := min(capacity, initialSummaryReservoirCapacity) |
| 292 | return &summaryQuantileSketch{ |
| 293 | capacity: capacity, |
| 294 | rng: seed, |
| 295 | values: make([]SampleValue, 0, initCap), |
| 296 | } |
| 297 | } |
| 298 | |
| 299 | func (s *summaryQuantileSketch) clone() *summaryQuantileSketch { |
| 300 | if s == nil { |
| 301 | return nil |
| 302 | } |
| 303 | cp := *s |
| 304 | cp.values = append([]SampleValue(nil), s.values...) |
| 305 | cp.scratch = nil |
| 306 | return &cp |
| 307 | } |
| 308 | |
| 309 | func (s *summaryQuantileSketch) observe(v SampleValue) { |
| 310 | // Not safe for concurrent use. Callers must hold the owning store mutex. |
| 311 | s.count++ |
| 312 | if len(s.values) < s.capacity { |
| 313 | s.values = append(s.values, v) |
| 314 | return |
| 315 | } |
| 316 | |
| 317 | j := s.next() % s.count |
| 318 | if j < uint64(s.capacity) { |
| 319 | s.values[j] = v |
| 320 | } |
| 321 | } |
| 322 | |
| 323 | func (s *summaryQuantileSketch) quantiles(targets []float64) []SampleValue { |
| 324 | if len(targets) == 0 { |
| 325 | return nil |
| 326 | } |
| 327 | |
| 328 | out := make([]SampleValue, len(targets)) |
| 329 | if len(s.values) == 0 { |
| 330 | for i := range out { |
| 331 | out[i] = math.NaN() |
| 332 | } |
| 333 | return out |
| 334 | } |
| 335 | |
| 336 | s.scratch = growCopy(s.scratch, s.values) |
| 337 | sort.Float64s(s.scratch) |
| 338 | for i, q := range targets { |
| 339 | out[i] = sampleQuantileLinear(s.scratch, q) |
| 340 | } |
| 341 | return out |
| 342 | } |
| 343 | |
| 344 | func (s *summaryQuantileSketch) next() uint64 { |
| 345 | x := s.rng |
| 346 | x ^= x << 13 |
| 347 | x ^= x >> 7 |
| 348 | x ^= x << 17 |
| 349 | s.rng = x |
| 350 | return x |
| 351 | } |
| 352 | |
| 353 | func growCopy(dst, src []SampleValue) []SampleValue { |
| 354 | if cap(dst) < len(src) { |
| 355 | dst = make([]SampleValue, len(src)) |
| 356 | } else { |
| 357 | dst = dst[:len(src)] |
| 358 | } |
| 359 | copy(dst, src) |
| 360 | return dst |
| 361 | } |
| 362 | |
| 363 | func sampleQuantileLinear(sorted []SampleValue, q float64) SampleValue { |
| 364 | last := len(sorted) - 1 |
| 365 | if q <= 0 { |
| 366 | return sorted[0] |
| 367 | } |
| 368 | if q >= 1 { |
| 369 | return sorted[last] |
| 370 | } |
| 371 | pos := q * float64(last) |
| 372 | low := int(math.Floor(pos)) |
| 373 | high := int(math.Ceil(pos)) |
| 374 | if low == high { |
| 375 | return sorted[low] |
| 376 | } |
| 377 | w := pos - float64(low) |
| 378 | return sorted[low]*(1-w) + sorted[high]*w |
| 379 | } |
| 380 | |
| 381 | func summarySketchSeed(key string) uint64 { |
| 382 | h := fnv.New64a() |
| 383 | _, _ = h.Write([]byte(key)) |
| 384 | seed := h.Sum64() |
| 385 | if seed == 0 { |
| 386 | return 1 |
| 387 | } |
| 388 | return seed |
| 389 | } |