| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | package metrix |
| 4 | |
| 5 | import ( |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "maps" |
| 9 | "math" |
| 10 | "sort" |
| 11 | "strings" |
| 12 | "sync" |
| 13 | "sync/atomic" |
| 14 | ) |
| 15 | |
| 16 | type metricKind uint8 |
| 17 | |
| 18 | type metricMode uint8 |
| 19 | |
| 20 | const ( |
| 21 | kindGauge metricKind = iota |
| 22 | kindCounter |
| 23 | kindHistogram |
| 24 | kindSummary |
| 25 | kindStateSet |
| 26 | kindMeasureSet |
| 27 | ) |
| 28 | |
| 29 | const ( |
| 30 | modeSnapshot metricMode = iota |
| 31 | modeStateful |
| 32 | ) |
| 33 | |
| 34 | type instrumentDescriptor struct { |
| 35 | name string |
| 36 | kind metricKind |
| 37 | mode metricMode |
| 38 | freshness FreshnessPolicy // visibility policy used by Read() |
| 39 | window MetricWindow |
| 40 | histogram *histogramSchema // set for kindHistogram only |
| 41 | summary *summarySchema // set for kindSummary only |
| 42 | stateSet *stateSetSchema // set for kindStateSet only |
| 43 | measureSet *measureSetSchema // set for kindMeasureSet only |
| 44 | meta MetricMeta |
| 45 | } |
| 46 | |
| 47 | type histogramSchema struct { |
| 48 | bounds []float64 |
| 49 | } |
| 50 | |
| 51 | type summarySchema struct { |
| 52 | quantiles []float64 |
| 53 | reservoirSize int |
| 54 | } |
| 55 | |
| 56 | type stateSetSchema struct { |
| 57 | mode StateSetMode |
| 58 | states []string |
| 59 | index map[string]struct{} |
| 60 | } |
| 61 | |
| 62 | type measureSetSchema struct { |
| 63 | semantics MeasureSetSemantics |
| 64 | fields []MeasureFieldSpec |
| 65 | index map[string]int |
| 66 | } |
| 67 | |
| 68 | type committedSeries struct { |
| 69 | id SeriesID |
| 70 | hash64 uint64 |
| 71 | key string |
| 72 | name string |
| 73 | // hostScope is immutable after publish and partitions otherwise-identical series. |
| 74 | hostScopeKey string |
| 75 | hostScope HostScope |
| 76 | // labels are immutable after series publish and can be safely shared across snapshots. |
| 77 | labels []Label |
| 78 | labelsKey string |
| 79 | desc *instrumentDescriptor |
| 80 | value SampleValue // last committed sample value |
| 81 | // Internal successful-cycle clock used only for retention aging. |
| 82 | lastSeenSuccessCycle uint64 |
| 83 | // Internal runtime clock (unix nanos) used only by runtime retention. |
| 84 | runtimeLastSeenUnixNano int64 |
| 85 | |
| 86 | // Counter two-sample state (used by Delta()). |
| 87 | counterCurrent SampleValue |
| 88 | counterPrevious SampleValue |
| 89 | counterHasPrev bool |
| 90 | counterCurrentSeq uint64 |
| 91 | counterPreviousSeq uint64 |
| 92 | |
| 93 | // Histogram current sample (used by Histogram()). |
| 94 | histogramCount SampleValue |
| 95 | histogramSum SampleValue |
| 96 | histogramCumulative []SampleValue |
| 97 | |
| 98 | // Summary current sample (used by Summary()). |
| 99 | summaryCount SampleValue |
| 100 | summarySum SampleValue |
| 101 | summaryQuantiles []SampleValue |
| 102 | summarySketch *summaryQuantileSketch // cumulative stateful quantile estimator |
| 103 | |
| 104 | // StateSet current sample (used by StateSet()). |
| 105 | stateSetValues map[string]bool |
| 106 | |
| 107 | // MeasureSet current sample (used by MeasureSet()). |
| 108 | measureSetValues []SampleValue |
| 109 | measureSetPreviousValues []SampleValue |
| 110 | measureSetHasPrev bool |
| 111 | measureSetCurrentSeq uint64 |
| 112 | measureSetPreviousSeq uint64 |
| 113 | |
| 114 | meta SeriesMeta |
| 115 | } |
| 116 | |
| 117 | type readSnapshot struct { |
| 118 | collectMeta CollectMeta |
| 119 | series map[string]*committedSeries // key => series |
| 120 | byName map[string][]*committedSeries // metric name => stable ordered series list |
| 121 | // runtimeBase links runtime snapshots in overlay mode (nil for materialized snapshots). |
| 122 | runtimeBase *readSnapshot |
| 123 | // runtimeDepth tracks overlay chain depth for runtime compaction heuristics. |
| 124 | runtimeDepth int |
| 125 | } |
| 126 | |
| 127 | type cycleFrame struct { |
| 128 | seq uint64 |
| 129 | err error |
| 130 | hostScopes map[string]HostScope |
| 131 | gauges map[string]*stagedGauge |
| 132 | counters map[string]*stagedCounter |
| 133 | histograms map[string]*stagedHistogram |
| 134 | summaries map[string]*stagedSummary |
| 135 | stateSet map[string]*stagedStateSet |
| 136 | measureSetGauges map[string]*stagedMeasureSet |
| 137 | measureSetCounters map[string]*stagedMeasureSet |
| 138 | } |
| 139 | |
| 140 | type storeCore struct { |
| 141 | mu sync.RWMutex |
| 142 | |
| 143 | sequence uint64 |
| 144 | successSeq uint64 |
| 145 | active *cycleFrame |
| 146 | instruments map[string]*instrumentDescriptor // metric name => descriptor (mode/kind locked) |
| 147 | // Captured schema for snapshot histograms declared without explicit bounds. |
| 148 | // Accessed only under c.mu during cycle commit. |
| 149 | snapshotHistogramSchema map[string]*histogramSchema // metric name => captured bounds |
| 150 | retention collectorRetentionPolicy |
| 151 | |
| 152 | snapshot atomic.Pointer[readSnapshot] // atomically swapped immutable read view |
| 153 | } |
| 154 | |
| 155 | type collectorRetentionPolicy struct { |
| 156 | expireAfterSuccessCycles uint64 |
| 157 | maxSeries int |
| 158 | } |
| 159 | |
| 160 | const ( |
| 161 | defaultCollectorExpireAfterSuccessCycles uint64 = 10 |
| 162 | defaultCollectorMaxSeries = 0 // disabled |
| 163 | ) |
| 164 | |
| 165 | type storeView struct { |
| 166 | core *storeCore |
| 167 | } |
| 168 | |
| 169 | type managedStore struct { |
| 170 | core *storeCore |
| 171 | } |
| 172 | |
| 173 | type storeCycleController struct { |
| 174 | core *storeCore |
| 175 | } |
| 176 | |
| 177 | // NewCollectorStore creates a collection store with staged writes and immutable read snapshots. |
| 178 | func NewCollectorStore() CollectorStore { |
| 179 | core := &storeCore{ |
| 180 | instruments: make(map[string]*instrumentDescriptor), |
| 181 | snapshotHistogramSchema: make(map[string]*histogramSchema), |
| 182 | retention: collectorRetentionPolicy{ |
| 183 | expireAfterSuccessCycles: defaultCollectorExpireAfterSuccessCycles, |
| 184 | maxSeries: defaultCollectorMaxSeries, |
| 185 | }, |
| 186 | } |
| 187 | core.snapshot.Store(&readSnapshot{ |
| 188 | collectMeta: CollectMeta{LastAttemptStatus: CollectStatusUnknown}, |
| 189 | series: make(map[string]*committedSeries), |
| 190 | byName: make(map[string][]*committedSeries), |
| 191 | }) |
| 192 | return &storeView{core: core} |
| 193 | } |
| 194 | |
| 195 | // AsCycleManagedStore exposes runtime cycle control for stores created by NewCollectorStore. |
| 196 | // This is intended for runtime internals, not collector code. |
| 197 | func AsCycleManagedStore(s CollectorStore) (CycleManagedStore, bool) { |
| 198 | switch v := s.(type) { |
| 199 | case *managedStore: |
| 200 | return v, true |
| 201 | case *storeView: |
| 202 | return &managedStore{core: v.core}, true |
| 203 | default: |
| 204 | return nil, false |
| 205 | } |
| 206 | } |
| 207 | |
| 208 | func (s *storeView) Read(opts ...ReadOption) Reader { |
| 209 | cfg := resolveReadConfig(opts...) |
| 210 | snap := s.core.snapshot.Load() |
| 211 | if cfg.flatten { |
| 212 | snap = flattenSnapshot(snap) |
| 213 | } |
| 214 | return &storeReader{snap: snap, raw: cfg.raw, flattened: cfg.flatten, hostScopeKey: cfg.hostScopeKey} |
| 215 | } |
| 216 | |
| 217 | func (s *storeView) Write() Writer { |
| 218 | return &writeView{backend: s.core} |
| 219 | } |
| 220 | |
| 221 | func (s *managedStore) Read(opts ...ReadOption) Reader { |
| 222 | return (&storeView{core: s.core}).Read(opts...) |
| 223 | } |
| 224 | |
| 225 | func (s *managedStore) Write() Writer { |
| 226 | return (&storeView{core: s.core}).Write() |
| 227 | } |
| 228 | |
| 229 | func (s *managedStore) CycleController() CycleController { |
| 230 | return &storeCycleController{core: s.core} |
| 231 | } |
| 232 | |
| 233 | // BeginCycle opens a new staged frame for collection writes. |
| 234 | func (c *storeCycleController) BeginCycle() { |
| 235 | c.core.mu.Lock() |
| 236 | defer c.core.mu.Unlock() |
| 237 | |
| 238 | if c.core.active != nil { |
| 239 | panic(errCycleActive) |
| 240 | } |
| 241 | |
| 242 | c.core.sequence++ |
| 243 | c.core.active = &cycleFrame{ |
| 244 | seq: c.core.sequence, |
| 245 | hostScopes: make(map[string]HostScope), |
| 246 | gauges: make(map[string]*stagedGauge), |
| 247 | counters: make(map[string]*stagedCounter), |
| 248 | histograms: make(map[string]*stagedHistogram), |
| 249 | summaries: make(map[string]*stagedSummary), |
| 250 | stateSet: make(map[string]*stagedStateSet), |
| 251 | measureSetGauges: make(map[string]*stagedMeasureSet), |
| 252 | measureSetCounters: make(map[string]*stagedMeasureSet), |
| 253 | } |
| 254 | } |
| 255 | |
| 256 | // CommitCycleSuccess publishes staged writes into a new committed snapshot. |
| 257 | func (c *storeCycleController) CommitCycleSuccess() error { |
| 258 | c.core.mu.Lock() |
| 259 | defer c.core.mu.Unlock() |
| 260 | |
| 261 | if c.core.active == nil { |
| 262 | panic(errCycleMissing) |
| 263 | } |
| 264 | |
| 265 | oldSnap := c.core.snapshot.Load() |
| 266 | if c.core.active.err != nil { |
| 267 | abortSnap := &readSnapshot{ |
| 268 | collectMeta: oldSnap.collectMeta, |
| 269 | series: oldSnap.series, |
| 270 | byName: oldSnap.byName, |
| 271 | } |
| 272 | abortSnap.collectMeta.LastAttemptSeq = c.core.active.seq |
| 273 | abortSnap.collectMeta.LastAttemptStatus = CollectStatusFailed |
| 274 | c.core.snapshot.Store(abortSnap) |
| 275 | err := c.core.active.err |
| 276 | c.core.active = nil |
| 277 | return err |
| 278 | } |
| 279 | successSeq := c.core.successSeq + 1 |
| 280 | next := &readSnapshot{ |
| 281 | collectMeta: oldSnap.collectMeta, |
| 282 | series: make(map[string]*committedSeries, len(oldSnap.series)), |
| 283 | byName: nil, |
| 284 | } |
| 285 | commitHostScopes := make(map[string]HostScope) |
| 286 | |
| 287 | maps.Copy(next.series, oldSnap.series) |
| 288 | |
| 289 | for key, staged := range c.core.active.gauges { |
| 290 | commitHostScopes[staged.hostScopeKey] = staged.hostScope |
| 291 | series := getOrCreateCommitSeries(oldSnap, next, key, staged.name, staged.hostScopeKey, staged.hostScope, staged.labels, staged.labelsKey, staged.desc) |
| 292 | series.value = staged.value |
| 293 | markSeriesSeen(series, c.core.active.seq, successSeq) |
| 294 | } |
| 295 | |
| 296 | for key, staged := range c.core.active.counters { |
| 297 | commitHostScopes[staged.hostScopeKey] = staged.hostScope |
| 298 | series := getOrCreateCommitSeries(oldSnap, next, key, staged.name, staged.hostScopeKey, staged.hostScope, staged.labels, staged.labelsKey, staged.desc) |
| 299 | |
| 300 | hadCurrent := series.desc != nil && series.desc.kind == kindCounter && series.counterCurrentSeq > 0 |
| 301 | if hadCurrent { |
| 302 | series.counterPrevious = series.counterCurrent |
| 303 | series.counterPreviousSeq = series.counterCurrentSeq |
| 304 | series.counterHasPrev = true |
| 305 | } else { |
| 306 | series.counterPrevious = 0 |
| 307 | series.counterPreviousSeq = 0 |
| 308 | series.counterHasPrev = false |
| 309 | } |
| 310 | |
| 311 | series.counterCurrent = staged.current |
| 312 | series.counterCurrentSeq = c.core.active.seq |
| 313 | series.value = staged.current // Value() for counters returns current total. |
| 314 | markSeriesSeen(series, c.core.active.seq, successSeq) |
| 315 | } |
| 316 | |
| 317 | for key, staged := range c.core.active.histograms { |
| 318 | commitHostScopes[staged.hostScopeKey] = staged.hostScope |
| 319 | series := getOrCreateCommitSeries(oldSnap, next, key, staged.name, staged.hostScopeKey, staged.hostScope, staged.labels, staged.labelsKey, staged.desc) |
| 320 | |
| 321 | if series.desc == nil { |
| 322 | panic("metrix: missing histogram descriptor") |
| 323 | } |
| 324 | if series.desc.histogram == nil { |
| 325 | schema := c.core.snapshotHistogramSchema[series.name] |
| 326 | if schema == nil { |
| 327 | schema = &histogramSchema{bounds: append([]float64(nil), staged.bounds...)} |
| 328 | c.core.snapshotHistogramSchema[series.name] = schema |
| 329 | } else if !equalHistogramBounds(schema.bounds, staged.bounds) { |
| 330 | panic("metrix: histogram schema drift detected") |
| 331 | } |
| 332 | // Descriptor pointers can be shared across published snapshots. |
| 333 | // Never mutate shared descriptor state in-place; attach schema via a cloned descriptor. |
| 334 | series.desc = cloneInstrumentDescriptorWithHistogram(series.desc, schema.bounds) |
| 335 | } else if !equalHistogramBounds(series.desc.histogram.bounds, staged.bounds) { |
| 336 | panic("metrix: histogram schema drift detected") |
| 337 | } |
| 338 | |
| 339 | series.histogramCount = staged.count |
| 340 | series.histogramSum = staged.sum |
| 341 | series.histogramCumulative = append(series.histogramCumulative[:0], staged.cumulative...) |
| 342 | markSeriesSeen(series, c.core.active.seq, successSeq) |
| 343 | } |
| 344 | |
| 345 | for key, staged := range c.core.active.summaries { |
| 346 | commitHostScopes[staged.hostScopeKey] = staged.hostScope |
| 347 | series := getOrCreateCommitSeries(oldSnap, next, key, staged.name, staged.hostScopeKey, staged.hostScope, staged.labels, staged.labelsKey, staged.desc) |
| 348 | |
| 349 | if staged.desc.mode == modeStateful && len(staged.desc.summaryQuantiles()) > 0 { |
| 350 | if staged.sketch != nil { |
| 351 | staged.quantileValues = staged.sketch.quantiles(staged.desc.summaryQuantiles()) |
| 352 | } else { |
| 353 | // Defensive fallback for malformed staged state. |
| 354 | staged.quantileValues = nanSummaryQuantiles(staged.desc.summaryQuantiles()) |
| 355 | } |
| 356 | } |
| 357 | |
| 358 | series.summaryCount = staged.count |
| 359 | series.summarySum = staged.sum |
| 360 | if len(staged.quantileValues) > 0 { |
| 361 | series.summaryQuantiles = append(series.summaryQuantiles[:0], staged.quantileValues...) |
| 362 | } else { |
| 363 | series.summaryQuantiles = nil |
| 364 | } |
| 365 | if staged.sketch != nil && series.desc != nil && series.desc.window == WindowCumulative { |
| 366 | series.summarySketch = staged.sketch.clone() |
| 367 | } else { |
| 368 | series.summarySketch = nil |
| 369 | } |
| 370 | markSeriesSeen(series, c.core.active.seq, successSeq) |
| 371 | } |
| 372 | |
| 373 | for key, staged := range c.core.active.stateSet { |
| 374 | commitHostScopes[staged.hostScopeKey] = staged.hostScope |
| 375 | series := getOrCreateCommitSeries(oldSnap, next, key, staged.name, staged.hostScopeKey, staged.hostScope, staged.labels, staged.labelsKey, staged.desc) |
| 376 | |
| 377 | series.stateSetValues = cloneStateMap(staged.states) |
| 378 | markSeriesSeen(series, c.core.active.seq, successSeq) |
| 379 | } |
| 380 | |
| 381 | for key, staged := range c.core.active.measureSetGauges { |
| 382 | commitHostScopes[staged.hostScopeKey] = staged.hostScope |
| 383 | series := getOrCreateCommitSeries(oldSnap, next, key, staged.name, staged.hostScopeKey, staged.hostScope, staged.labels, staged.labelsKey, staged.desc) |
| 384 | series.measureSetValues = append(series.measureSetValues[:0], staged.values...) |
| 385 | markSeriesSeen(series, c.core.active.seq, successSeq) |
| 386 | } |
| 387 | |
| 388 | for key, staged := range c.core.active.measureSetCounters { |
| 389 | commitHostScopes[staged.hostScopeKey] = staged.hostScope |
| 390 | series := getOrCreateCommitSeries(oldSnap, next, key, staged.name, staged.hostScopeKey, staged.hostScope, staged.labels, staged.labelsKey, staged.desc) |
| 391 | |
| 392 | if series.desc != nil && series.desc.kind == kindMeasureSet && series.desc.measureSet != nil && series.desc.measureSet.semantics == MeasureSetSemanticsCounter && series.measureSetCurrentSeq > 0 { |
| 393 | series.measureSetPreviousValues = append(series.measureSetPreviousValues[:0], series.measureSetValues...) |
| 394 | series.measureSetPreviousSeq = series.measureSetCurrentSeq |
| 395 | series.measureSetHasPrev = true |
| 396 | } else { |
| 397 | series.measureSetPreviousValues = nil |
| 398 | series.measureSetPreviousSeq = 0 |
| 399 | series.measureSetHasPrev = false |
| 400 | } |
| 401 | |
| 402 | series.measureSetValues = append(series.measureSetValues[:0], staged.values...) |
| 403 | series.measureSetCurrentSeq = c.core.active.seq |
| 404 | markSeriesSeen(series, c.core.active.seq, successSeq) |
| 405 | } |
| 406 | |
| 407 | refreshCommittedHostScopes(oldSnap, next, commitHostScopes) |
| 408 | applyCollectorRetention(next.series, c.core.retention, successSeq) |
| 409 | next.collectMeta.LastAttemptSeq = c.core.active.seq |
| 410 | next.collectMeta.LastAttemptStatus = CollectStatusSuccess |
| 411 | next.collectMeta.LastSuccessSeq = c.core.active.seq |
| 412 | |
| 413 | c.core.snapshot.Store(next) |
| 414 | c.core.successSeq = successSeq |
| 415 | c.core.active = nil |
| 416 | return nil |
| 417 | } |
| 418 | |
| 419 | func newCommittedSeries(key, name, hostScopeKey string, hostScope HostScope, labels []Label, labelsKey string, desc *instrumentDescriptor) *committedSeries { |
| 420 | return &committedSeries{ |
| 421 | id: SeriesID(key), |
| 422 | hash64: seriesIDHash(SeriesID(key)), |
| 423 | key: key, |
| 424 | name: name, |
| 425 | hostScopeKey: hostScopeKey, |
| 426 | hostScope: cloneHostScope(hostScope), |
| 427 | labels: append([]Label(nil), labels...), |
| 428 | labelsKey: labelsKey, |
| 429 | desc: desc, |
| 430 | meta: baseSeriesMeta(desc), |
| 431 | } |
| 432 | } |
| 433 | |
| 434 | func ensureCommitSeriesMutable(old, next *readSnapshot, key string) *committedSeries { |
| 435 | series := next.series[key] |
| 436 | if series == nil { |
| 437 | return nil |
| 438 | } |
| 439 | if oldSeries, ok := old.series[key]; ok && oldSeries == series { |
| 440 | series = cloneCommittedSeries(series) |
| 441 | next.series[key] = series |
| 442 | } |
| 443 | ensureSeriesMeta(series.desc, &series.meta) |
| 444 | return series |
| 445 | } |
| 446 | |
| 447 | func getOrCreateCommitSeries(old, next *readSnapshot, key, name, hostScopeKey string, hostScope HostScope, labels []Label, labelsKey string, desc *instrumentDescriptor) *committedSeries { |
| 448 | series := ensureCommitSeriesMutable(old, next, key) |
| 449 | if series != nil { |
| 450 | return series |
| 451 | } |
| 452 | series = newCommittedSeries(key, name, hostScopeKey, hostScope, labels, labelsKey, desc) |
| 453 | next.series[key] = series |
| 454 | return series |
| 455 | } |
| 456 | |
| 457 | func refreshCommittedHostScopes(old, next *readSnapshot, scopes map[string]HostScope) { |
| 458 | if len(scopes) == 0 { |
| 459 | return |
| 460 | } |
| 461 | for key, series := range next.series { |
| 462 | scope, ok := scopes[series.hostScopeKey] |
| 463 | if !ok || hostScopeEqual(series.hostScope, scope) { |
| 464 | continue |
| 465 | } |
| 466 | series = ensureCommitSeriesMutable(old, next, key) |
| 467 | if series == nil { |
| 468 | continue |
| 469 | } |
| 470 | series.hostScope = cloneHostScope(scope) |
| 471 | } |
| 472 | } |
| 473 | |
| 474 | func markSeriesSeen(series *committedSeries, attemptSeq, successSeq uint64) { |
| 475 | series.meta.LastSeenSuccessSeq = attemptSeq |
| 476 | series.lastSeenSuccessCycle = successSeq |
| 477 | } |
| 478 | |
| 479 | // AbortCycle discards staged writes and publishes metadata-only failed-attempt status. |
| 480 | func (c *storeCycleController) AbortCycle() { |
| 481 | c.core.mu.Lock() |
| 482 | defer c.core.mu.Unlock() |
| 483 | |
| 484 | if c.core.active == nil { |
| 485 | panic(errCycleMissing) |
| 486 | } |
| 487 | |
| 488 | oldSnap := c.core.snapshot.Load() |
| 489 | // Alias previous committed maps directly. Safe by invariant: |
| 490 | // committed series/snapshots are immutable after publish. |
| 491 | abortSnap := &readSnapshot{ |
| 492 | collectMeta: oldSnap.collectMeta, |
| 493 | series: oldSnap.series, |
| 494 | byName: oldSnap.byName, |
| 495 | } |
| 496 | |
| 497 | abortSnap.collectMeta.LastAttemptSeq = c.core.active.seq |
| 498 | abortSnap.collectMeta.LastAttemptStatus = CollectStatusFailed |
| 499 | c.core.snapshot.Store(abortSnap) |
| 500 | |
| 501 | c.core.active = nil |
| 502 | } |
| 503 | |
| 504 | // buildByName builds deterministic per-name iteration lists for snapshot readers. |
| 505 | func buildByName(series map[string]*committedSeries) map[string][]*committedSeries { |
| 506 | byName := make(map[string][]*committedSeries) |
| 507 | for _, s := range series { |
| 508 | if s.desc == nil || !isScalarKind(s.desc.kind) { |
| 509 | continue |
| 510 | } |
| 511 | byName[s.name] = append(byName[s.name], s) |
| 512 | } |
| 513 | for _, lst := range byName { |
| 514 | sort.Slice(lst, func(i, j int) bool { |
| 515 | if lst[i].hostScopeKey != lst[j].hostScopeKey { |
| 516 | return lst[i].hostScopeKey < lst[j].hostScopeKey |
| 517 | } |
| 518 | return lst[i].labelsKey < lst[j].labelsKey |
| 519 | }) |
| 520 | } |
| 521 | return byName |
| 522 | } |
| 523 | |
| 524 | func applyCollectorRetention(series map[string]*committedSeries, policy collectorRetentionPolicy, successSeq uint64) { |
| 525 | if policy.expireAfterSuccessCycles > 0 { |
| 526 | for key, s := range series { |
| 527 | seen := s.lastSeenSuccessCycle |
| 528 | if seen == 0 || successSeq < seen { |
| 529 | continue |
| 530 | } |
| 531 | if successSeq-seen >= policy.expireAfterSuccessCycles { |
| 532 | delete(series, key) |
| 533 | } |
| 534 | } |
| 535 | } |
| 536 | |
| 537 | evictOldestSeries(series, policy.maxSeries, func(s *committedSeries) uint64 { |
| 538 | return s.lastSeenSuccessCycle |
| 539 | }, nil) |
| 540 | } |
| 541 | |
| 542 | func defaultFreshness(mode metricMode) FreshnessPolicy { |
| 543 | if mode == modeSnapshot { |
| 544 | return FreshnessCycle |
| 545 | } |
| 546 | return FreshnessCommitted |
| 547 | } |
| 548 | |
| 549 | func (c *storeCore) registerInstrument(name string, kind metricKind, mode metricMode, opts ...InstrumentOption) (*instrumentDescriptor, error) { |
| 550 | cfg := instrumentConfig{} |
| 551 | for _, opt := range opts { |
| 552 | if opt != nil { |
| 553 | opt.apply(&cfg) |
| 554 | } |
| 555 | } |
| 556 | |
| 557 | if cfg.windowSet && !isWindowAllowed(kind, mode) { |
| 558 | return nil, fmt.Errorf("metrix: WithWindow is valid only for stateful histogram/summary") |
| 559 | } |
| 560 | if len(cfg.histogramBounds) > 0 && kind != kindHistogram { |
| 561 | return nil, fmt.Errorf("metrix: histogram bounds are invalid for this instrument kind") |
| 562 | } |
| 563 | if len(cfg.summaryQuantile) > 0 && kind != kindSummary { |
| 564 | return nil, fmt.Errorf("metrix: summary quantiles are invalid for this instrument kind") |
| 565 | } |
| 566 | if cfg.summaryReservoirSet && !(kind == kindSummary && mode == modeStateful) { |
| 567 | return nil, fmt.Errorf("metrix: summary reservoir size is valid only for stateful summaries") |
| 568 | } |
| 569 | if (len(cfg.states) > 0 || cfg.stateSetMode != nil) && kind != kindStateSet { |
| 570 | return nil, fmt.Errorf("metrix: stateset options are invalid for this instrument kind") |
| 571 | } |
| 572 | if (len(cfg.measureSetFields) > 0 || cfg.measureSetSemantics != nil) && kind != kindMeasureSet { |
| 573 | return nil, fmt.Errorf("metrix: measureset options are invalid for this instrument kind") |
| 574 | } |
| 575 | |
| 576 | window := WindowCumulative |
| 577 | if cfg.windowSet { |
| 578 | window = cfg.window |
| 579 | } |
| 580 | |
| 581 | fresh := defaultFreshness(mode) |
| 582 | if cfg.freshnessSet { |
| 583 | fresh = cfg.freshness |
| 584 | } |
| 585 | if mode == modeStateful && window == WindowCycle && (kind == kindHistogram || kind == kindSummary) { |
| 586 | if cfg.freshnessSet && fresh != FreshnessCycle { |
| 587 | return nil, fmt.Errorf("metrix: window=cycle requires FreshnessCycle") |
| 588 | } |
| 589 | fresh = FreshnessCycle |
| 590 | } |
| 591 | if mode == modeSnapshot && fresh == FreshnessCommitted { |
| 592 | return nil, fmt.Errorf("metrix: snapshot instruments cannot use FreshnessCommitted") |
| 593 | } |
| 594 | |
| 595 | metricMeta := MetricMeta{ |
| 596 | Description: strings.TrimSpace(cfg.description), |
| 597 | ChartFamily: strings.TrimSpace(cfg.chartFamily), |
| 598 | ChartPriority: cfg.chartPriority, |
| 599 | Unit: strings.TrimSpace(cfg.unit), |
| 600 | Float: cfg.float, |
| 601 | } |
| 602 | |
| 603 | var histogram *histogramSchema |
| 604 | if kind == kindHistogram { |
| 605 | s, err := buildHistogramSchema(cfg, mode) |
| 606 | if err != nil { |
| 607 | return nil, err |
| 608 | } |
| 609 | histogram = s |
| 610 | } |
| 611 | |
| 612 | var summary *summarySchema |
| 613 | if kind == kindSummary { |
| 614 | s, err := buildSummarySchema(cfg) |
| 615 | if err != nil { |
| 616 | return nil, err |
| 617 | } |
| 618 | summary = s |
| 619 | } |
| 620 | |
| 621 | var schema *stateSetSchema |
| 622 | if kind == kindStateSet { |
| 623 | s, err := buildStateSetSchema(cfg) |
| 624 | if err != nil { |
| 625 | return nil, err |
| 626 | } |
| 627 | schema = s |
| 628 | } |
| 629 | |
| 630 | var measureSet *measureSetSchema |
| 631 | if kind == kindMeasureSet { |
| 632 | s, err := buildMeasureSetSchema(cfg) |
| 633 | if err != nil { |
| 634 | return nil, err |
| 635 | } |
| 636 | measureSet = s |
| 637 | } |
| 638 | |
| 639 | c.mu.Lock() |
| 640 | defer c.mu.Unlock() |
| 641 | |
| 642 | if d, ok := c.instruments[name]; ok { |
| 643 | if d.kind != kind { |
| 644 | return nil, fmt.Errorf("metrix: instrument kind mismatch for %s", name) |
| 645 | } |
| 646 | if d.mode != mode { |
| 647 | return nil, fmt.Errorf("metrix: instrument mode mismatch for %s", name) |
| 648 | } |
| 649 | if d.freshness != fresh { |
| 650 | return nil, fmt.Errorf("metrix: instrument freshness mismatch for %s", name) |
| 651 | } |
| 652 | if d.window != window { |
| 653 | return nil, fmt.Errorf("metrix: instrument window mismatch for %s", name) |
| 654 | } |
| 655 | if kind == kindHistogram { |
| 656 | if !(mode == modeSnapshot && histogram == nil) && !equalHistogramSchema(d.histogram, histogram) { |
| 657 | return nil, fmt.Errorf("metrix: histogram schema mismatch for %s", name) |
| 658 | } |
| 659 | } |
| 660 | if kind == kindSummary && !equalSummarySchema(d.summary, summary) { |
| 661 | return nil, fmt.Errorf("metrix: summary schema mismatch for %s", name) |
| 662 | } |
| 663 | if kind == kindStateSet && !equalStateSetSchema(d.stateSet, schema) { |
| 664 | return nil, fmt.Errorf("metrix: stateset schema mismatch for %s", name) |
| 665 | } |
| 666 | if kind == kindMeasureSet && !equalMeasureSetSchema(d.measureSet, measureSet) { |
| 667 | return nil, fmt.Errorf("metrix: measureset schema mismatch for %s", name) |
| 668 | } |
| 669 | if cfg.descriptionSet && d.meta.Description != metricMeta.Description { |
| 670 | return nil, fmt.Errorf("metrix: metric description mismatch for %s", name) |
| 671 | } |
| 672 | if cfg.chartFamilySet && d.meta.ChartFamily != metricMeta.ChartFamily { |
| 673 | return nil, fmt.Errorf("metrix: metric chart family mismatch for %s", name) |
| 674 | } |
| 675 | if cfg.chartPrioritySet && d.meta.ChartPriority != metricMeta.ChartPriority { |
| 676 | return nil, fmt.Errorf("metrix: metric chart priority mismatch for %s", name) |
| 677 | } |
| 678 | if cfg.unitSet && d.meta.Unit != metricMeta.Unit { |
| 679 | return nil, fmt.Errorf("metrix: metric unit mismatch for %s", name) |
| 680 | } |
| 681 | if cfg.floatSet && d.meta.Float != metricMeta.Float { |
| 682 | return nil, fmt.Errorf("metrix: metric float mismatch for %s", name) |
| 683 | } |
| 684 | return d, nil |
| 685 | } |
| 686 | |
| 687 | d := &instrumentDescriptor{ |
| 688 | name: name, |
| 689 | kind: kind, |
| 690 | mode: mode, |
| 691 | freshness: fresh, |
| 692 | window: window, |
| 693 | histogram: histogram, |
| 694 | summary: summary, |
| 695 | stateSet: schema, |
| 696 | measureSet: measureSet, |
| 697 | meta: metricMeta, |
| 698 | } |
| 699 | c.instruments[name] = d |
| 700 | return d, nil |
| 701 | } |
| 702 | |
| 703 | func (c *storeCore) prepareHostScopeForWriteLocked(scope HostScope) (HostScope, bool) { |
| 704 | scope = mustNormalizeHostScope(scope) |
| 705 | if c.active == nil { |
| 706 | return scope, true |
| 707 | } |
| 708 | if existing, ok := c.active.hostScopes[scope.ScopeKey]; ok { |
| 709 | if hostScopeEqual(existing, scope) { |
| 710 | return existing, true |
| 711 | } |
| 712 | c.recordCycleErrorLocked(fmt.Errorf("%w: scope_key=%q", ErrHostScopeConflict, scope.ScopeKey)) |
| 713 | return scope, false |
| 714 | } |
| 715 | c.active.hostScopes[scope.ScopeKey] = cloneHostScope(scope) |
| 716 | return scope, true |
| 717 | } |
| 718 | |
| 719 | func (c *storeCore) recordCycleErrorLocked(err error) { |
| 720 | if err == nil || c.active == nil { |
| 721 | return |
| 722 | } |
| 723 | c.active.err = errors.Join(c.active.err, err) |
| 724 | } |
| 725 | |
| 726 | // makeSeriesKey joins host scope, metric name, and canonical label key into one stable identity key. |
| 727 | func makeSeriesKey(hostScopeKey, name, labelsKey string) string { |
| 728 | base := name |
| 729 | if labelsKey == "" { |
| 730 | base = name |
| 731 | } else { |
| 732 | base = name + "\xfe" + labelsKey |
| 733 | } |
| 734 | if hostScopeKey == "" { |
| 735 | return base |
| 736 | } |
| 737 | return hostScopeKey + "\xff" + base |
| 738 | } |
| 739 | |
| 740 | func cloneCommittedSeries(s *committedSeries) *committedSeries { |
| 741 | cp := *s |
| 742 | ensureSeriesMeta(cp.desc, &cp.meta) |
| 743 | cp.hostScope = cloneHostScope(s.hostScope) |
| 744 | // cp.labels intentionally reuses the original immutable label slice. |
| 745 | // Label identity is part of the series key and is never mutated after publish. |
| 746 | if s.stateSetValues != nil { |
| 747 | cp.stateSetValues = cloneStateMap(s.stateSetValues) |
| 748 | } |
| 749 | if len(s.measureSetValues) > 0 { |
| 750 | cp.measureSetValues = append([]SampleValue(nil), s.measureSetValues...) |
| 751 | } |
| 752 | if len(s.measureSetPreviousValues) > 0 { |
| 753 | cp.measureSetPreviousValues = append([]SampleValue(nil), s.measureSetPreviousValues...) |
| 754 | } |
| 755 | if len(s.histogramCumulative) > 0 { |
| 756 | cp.histogramCumulative = append([]SampleValue(nil), s.histogramCumulative...) |
| 757 | } |
| 758 | if len(s.summaryQuantiles) > 0 { |
| 759 | cp.summaryQuantiles = append([]SampleValue(nil), s.summaryQuantiles...) |
| 760 | } |
| 761 | if s.summarySketch != nil { |
| 762 | cp.summarySketch = s.summarySketch.clone() |
| 763 | } |
| 764 | return &cp |
| 765 | } |
| 766 | |
| 767 | func cloneInstrumentDescriptorWithHistogram(desc *instrumentDescriptor, bounds []float64) *instrumentDescriptor { |
| 768 | cp := *desc |
| 769 | cp.histogram = &histogramSchema{bounds: append([]float64(nil), bounds...)} |
| 770 | return &cp |
| 771 | } |
| 772 | |
| 773 | func cloneStateMap(in map[string]bool) map[string]bool { |
| 774 | if in == nil { |
| 775 | return nil |
| 776 | } |
| 777 | out := make(map[string]bool, len(in)) |
| 778 | maps.Copy(out, in) |
| 779 | return out |
| 780 | } |
| 781 | |
| 782 | func isScalarKind(kind metricKind) bool { |
| 783 | return kind == kindGauge || kind == kindCounter |
| 784 | } |
| 785 | |
| 786 | func buildHistogramSchema(cfg instrumentConfig, mode metricMode) (*histogramSchema, error) { |
| 787 | bounds, err := normalizeHistogramBounds(cfg.histogramBounds) |
| 788 | if err != nil { |
| 789 | return nil, err |
| 790 | } |
| 791 | if mode == modeStateful && len(bounds) == 0 { |
| 792 | return nil, fmt.Errorf("%w for stateful histogram", errHistogramBounds) |
| 793 | } |
| 794 | if len(bounds) == 0 { |
| 795 | return nil, nil |
| 796 | } |
| 797 | return &histogramSchema{bounds: bounds}, nil |
| 798 | } |
| 799 | |
| 800 | func buildSummarySchema(cfg instrumentConfig) (*summarySchema, error) { |
| 801 | if cfg.summaryReservoirSet && cfg.summaryReservoir <= 0 { |
| 802 | return nil, fmt.Errorf("metrix: summary reservoir size must be > 0") |
| 803 | } |
| 804 | |
| 805 | qs, err := normalizeSummaryQuantiles(cfg.summaryQuantile) |
| 806 | if err != nil { |
| 807 | return nil, err |
| 808 | } |
| 809 | |
| 810 | if len(qs) == 0 { |
| 811 | return nil, nil |
| 812 | } |
| 813 | |
| 814 | size := defaultSummaryReservoirSize |
| 815 | if cfg.summaryReservoirSet { |
| 816 | size = cfg.summaryReservoir |
| 817 | } |
| 818 | |
| 819 | return &summarySchema{ |
| 820 | quantiles: qs, |
| 821 | reservoirSize: size, |
| 822 | }, nil |
| 823 | } |
| 824 | |
| 825 | func buildStateSetSchema(cfg instrumentConfig) (*stateSetSchema, error) { |
| 826 | if len(cfg.states) == 0 { |
| 827 | return nil, fmt.Errorf("metrix: stateset requires WithStateSetStates") |
| 828 | } |
| 829 | |
| 830 | mode := ModeBitSet |
| 831 | if cfg.stateSetMode != nil { |
| 832 | mode = *cfg.stateSetMode |
| 833 | } |
| 834 | |
| 835 | seen := make(map[string]struct{}, len(cfg.states)) |
| 836 | states := make([]string, 0, len(cfg.states)) |
| 837 | for _, st := range cfg.states { |
| 838 | if st == "" { |
| 839 | return nil, fmt.Errorf("metrix: stateset state cannot be empty") |
| 840 | } |
| 841 | if _, ok := seen[st]; ok { |
| 842 | return nil, fmt.Errorf("metrix: duplicate stateset state %q", st) |
| 843 | } |
| 844 | seen[st] = struct{}{} |
| 845 | states = append(states, st) |
| 846 | } |
| 847 | |
| 848 | return &stateSetSchema{ |
| 849 | mode: mode, |
| 850 | states: states, |
| 851 | index: seen, |
| 852 | }, nil |
| 853 | } |
| 854 | |
| 855 | func equalStateSetSchema(a, b *stateSetSchema) bool { |
| 856 | if a == nil || b == nil { |
| 857 | return a == b |
| 858 | } |
| 859 | if a.mode != b.mode || len(a.states) != len(b.states) { |
| 860 | return false |
| 861 | } |
| 862 | for i := range a.states { |
| 863 | if a.states[i] != b.states[i] { |
| 864 | return false |
| 865 | } |
| 866 | } |
| 867 | return true |
| 868 | } |
| 869 | |
| 870 | func buildMeasureSetSchema(cfg instrumentConfig) (*measureSetSchema, error) { |
| 871 | if len(cfg.measureSetFields) == 0 { |
| 872 | return nil, fmt.Errorf("metrix: measureset requires WithMeasureSetFields") |
| 873 | } |
| 874 | if cfg.measureSetSemantics == nil { |
| 875 | return nil, fmt.Errorf("metrix: measureset semantics are missing") |
| 876 | } |
| 877 | |
| 878 | fields := make([]MeasureFieldSpec, 0, len(cfg.measureSetFields)) |
| 879 | index := make(map[string]int, len(cfg.measureSetFields)) |
| 880 | for i, field := range cfg.measureSetFields { |
| 881 | name := strings.TrimSpace(field.Name) |
| 882 | if name == "" { |
| 883 | return nil, fmt.Errorf("metrix: measureset field name cannot be empty") |
| 884 | } |
| 885 | if _, ok := index[name]; ok { |
| 886 | return nil, fmt.Errorf("metrix: duplicate measureset field %q", name) |
| 887 | } |
| 888 | field.Name = name |
| 889 | fields = append(fields, field) |
| 890 | index[name] = i |
| 891 | } |
| 892 | |
| 893 | return &measureSetSchema{ |
| 894 | semantics: *cfg.measureSetSemantics, |
| 895 | fields: fields, |
| 896 | index: index, |
| 897 | }, nil |
| 898 | } |
| 899 | |
| 900 | func equalMeasureSetSchema(a, b *measureSetSchema) bool { |
| 901 | if a == nil || b == nil { |
| 902 | return a == b |
| 903 | } |
| 904 | if a.semantics != b.semantics || len(a.fields) != len(b.fields) { |
| 905 | return false |
| 906 | } |
| 907 | for i := range a.fields { |
| 908 | if a.fields[i].Name != b.fields[i].Name || a.fields[i].Float != b.fields[i].Float { |
| 909 | return false |
| 910 | } |
| 911 | } |
| 912 | return true |
| 913 | } |
| 914 | |
| 915 | func equalHistogramSchema(a, b *histogramSchema) bool { |
| 916 | if a == nil || b == nil { |
| 917 | return a == b |
| 918 | } |
| 919 | return equalHistogramBounds(a.bounds, b.bounds) |
| 920 | } |
| 921 | |
| 922 | func equalSummarySchema(a, b *summarySchema) bool { |
| 923 | if a == nil || b == nil { |
| 924 | return a == b |
| 925 | } |
| 926 | if a.reservoirSize != b.reservoirSize || len(a.quantiles) != len(b.quantiles) { |
| 927 | return false |
| 928 | } |
| 929 | for i := range a.quantiles { |
| 930 | if a.quantiles[i] != b.quantiles[i] { |
| 931 | return false |
| 932 | } |
| 933 | } |
| 934 | return true |
| 935 | } |
| 936 | |
| 937 | func equalHistogramBounds(a, b []float64) bool { |
| 938 | if len(a) != len(b) { |
| 939 | return false |
| 940 | } |
| 941 | for i := range a { |
| 942 | if a[i] != b[i] { |
| 943 | return false |
| 944 | } |
| 945 | } |
| 946 | return true |
| 947 | } |
| 948 | |
| 949 | func normalizeHistogramBounds(in []float64) ([]float64, error) { |
| 950 | if len(in) == 0 { |
| 951 | return nil, nil |
| 952 | } |
| 953 | |
| 954 | bounds := append([]float64(nil), in...) |
| 955 | out := make([]float64, 0, len(bounds)) |
| 956 | prev := math.Inf(-1) |
| 957 | for i, b := range bounds { |
| 958 | if math.IsNaN(b) || math.IsInf(b, -1) { |
| 959 | return nil, fmt.Errorf("%w: invalid upper bound", errHistogramPoint) |
| 960 | } |
| 961 | if math.IsInf(b, +1) { |
| 962 | if i != len(bounds)-1 { |
| 963 | return nil, fmt.Errorf("%w: +Inf bucket must be last", errHistogramPoint) |
| 964 | } |
| 965 | break // +Inf is implicit. |
| 966 | } |
| 967 | if b <= prev { |
| 968 | return nil, fmt.Errorf("%w: bounds must be strictly increasing", errHistogramPoint) |
| 969 | } |
| 970 | out = append(out, b) |
| 971 | prev = b |
| 972 | } |
| 973 | return out, nil |
| 974 | } |
| 975 | |
| 976 | func normalizeSummaryQuantiles(in []float64) ([]float64, error) { |
| 977 | if len(in) == 0 { |
| 978 | return nil, nil |
| 979 | } |
| 980 | |
| 981 | qs := append([]float64(nil), in...) |
| 982 | prev := -1.0 |
| 983 | for _, q := range qs { |
| 984 | if math.IsNaN(q) || q < 0 || q > 1 { |
| 985 | return nil, fmt.Errorf("metrix: invalid summary quantile %v", q) |
| 986 | } |
| 987 | if q <= prev { |
| 988 | return nil, fmt.Errorf("metrix: summary quantiles must be strictly increasing") |
| 989 | } |
| 990 | prev = q |
| 991 | } |
| 992 | return qs, nil |
| 993 | } |
| 994 | |
| 995 | func isWindowAllowed(kind metricKind, mode metricMode) bool { |
| 996 | return mode == modeStateful && (kind == kindHistogram || kind == kindSummary) |
| 997 | } |