| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | package metrix |
| 4 | |
| 5 | import ( |
| 6 | "maps" |
| 7 | "math" |
| 8 | "slices" |
| 9 | "sort" |
| 10 | "sync" |
| 11 | ) |
| 12 | |
| 13 | type storeReader struct { |
| 14 | snap *readSnapshot |
| 15 | raw bool // true => ReadRaw semantics (no freshness filtering) |
| 16 | flattened bool |
| 17 | hostScopeKey string |
| 18 | seriesOnce sync.Once |
| 19 | series map[string]*committedSeries |
| 20 | indexOnce sync.Once |
| 21 | index map[string][]*committedSeries |
| 22 | } |
| 23 | |
| 24 | type familyView struct { |
| 25 | name string |
| 26 | reader *storeReader |
| 27 | } |
| 28 | |
| 29 | // FlattenedRead reports whether this reader was created with ReadFlatten(). |
| 30 | func (r *storeReader) FlattenedRead() bool { |
| 31 | return r.flattened |
| 32 | } |
| 33 | |
| 34 | func (r *storeReader) Value(name string, labels Labels) (SampleValue, bool) { |
| 35 | s, ok := r.lookup(name, labels) |
| 36 | if !ok || !r.visible(s) { |
| 37 | return 0, false |
| 38 | } |
| 39 | if s.desc == nil || !isScalarKind(s.desc.kind) { |
| 40 | return 0, false |
| 41 | } |
| 42 | return s.value, true |
| 43 | } |
| 44 | |
| 45 | func (r *storeReader) Delta(name string, labels Labels) (SampleValue, bool) { |
| 46 | s, ok := r.lookup(name, labels) |
| 47 | if !ok || !r.visible(s) { |
| 48 | return 0, false |
| 49 | } |
| 50 | if s.desc == nil || s.desc.kind != kindCounter { |
| 51 | return 0, false |
| 52 | } |
| 53 | if !s.counterHasPrev || s.counterCurrentSeq != s.counterPreviousSeq+1 { |
| 54 | return 0, false |
| 55 | } |
| 56 | if s.counterCurrent < s.counterPrevious { |
| 57 | return s.counterCurrent, true |
| 58 | } |
| 59 | return s.counterCurrent - s.counterPrevious, true |
| 60 | } |
| 61 | |
| 62 | func (r *storeReader) Histogram(name string, labels Labels) (HistogramPoint, bool) { |
| 63 | if r.flattened { |
| 64 | return HistogramPoint{}, false |
| 65 | } |
| 66 | |
| 67 | s, ok := r.lookup(name, labels) |
| 68 | if !ok || !r.visible(s) { |
| 69 | return HistogramPoint{}, false |
| 70 | } |
| 71 | if s.desc == nil || s.desc.kind != kindHistogram || s.desc.histogram == nil { |
| 72 | return HistogramPoint{}, false |
| 73 | } |
| 74 | if len(s.histogramCumulative) != len(s.desc.histogram.bounds) { |
| 75 | return HistogramPoint{}, false |
| 76 | } |
| 77 | |
| 78 | point := HistogramPoint{ |
| 79 | Count: s.histogramCount, |
| 80 | Sum: s.histogramSum, |
| 81 | Buckets: make([]BucketPoint, len(s.desc.histogram.bounds)), |
| 82 | } |
| 83 | for i, ub := range s.desc.histogram.bounds { |
| 84 | point.Buckets[i] = BucketPoint{ |
| 85 | UpperBound: ub, |
| 86 | CumulativeCount: s.histogramCumulative[i], |
| 87 | } |
| 88 | } |
| 89 | return point, true |
| 90 | } |
| 91 | |
| 92 | func (r *storeReader) Summary(name string, labels Labels) (SummaryPoint, bool) { |
| 93 | if r.flattened { |
| 94 | return SummaryPoint{}, false |
| 95 | } |
| 96 | |
| 97 | s, ok := r.lookup(name, labels) |
| 98 | if !ok || !r.visible(s) { |
| 99 | return SummaryPoint{}, false |
| 100 | } |
| 101 | if s.desc == nil || s.desc.kind != kindSummary { |
| 102 | return SummaryPoint{}, false |
| 103 | } |
| 104 | |
| 105 | point := SummaryPoint{ |
| 106 | Count: s.summaryCount, |
| 107 | Sum: s.summarySum, |
| 108 | } |
| 109 | |
| 110 | if schema := s.desc.summary; schema != nil && len(schema.quantiles) > 0 { |
| 111 | if len(s.summaryQuantiles) != len(schema.quantiles) { |
| 112 | return SummaryPoint{}, false |
| 113 | } |
| 114 | point.Quantiles = make([]QuantilePoint, len(schema.quantiles)) |
| 115 | for i, q := range schema.quantiles { |
| 116 | point.Quantiles[i] = QuantilePoint{ |
| 117 | Quantile: q, |
| 118 | Value: s.summaryQuantiles[i], |
| 119 | } |
| 120 | } |
| 121 | } |
| 122 | |
| 123 | return point, true |
| 124 | } |
| 125 | |
| 126 | func (r *storeReader) StateSet(name string, labels Labels) (StateSetPoint, bool) { |
| 127 | if r.flattened { |
| 128 | return StateSetPoint{}, false |
| 129 | } |
| 130 | |
| 131 | s, ok := r.lookup(name, labels) |
| 132 | if !ok || !r.visible(s) { |
| 133 | return StateSetPoint{}, false |
| 134 | } |
| 135 | if s.desc == nil || s.desc.kind != kindStateSet { |
| 136 | return StateSetPoint{}, false |
| 137 | } |
| 138 | return StateSetPoint{States: cloneStateMap(s.stateSetValues)}, true |
| 139 | } |
| 140 | |
| 141 | func (r *storeReader) MeasureSet(name string, labels Labels) (MeasureSetPoint, bool) { |
| 142 | if r.flattened { |
| 143 | return MeasureSetPoint{}, false |
| 144 | } |
| 145 | |
| 146 | s, ok := r.lookup(name, labels) |
| 147 | if !ok || !r.visible(s) { |
| 148 | return MeasureSetPoint{}, false |
| 149 | } |
| 150 | if s.desc == nil || s.desc.kind != kindMeasureSet || s.desc.measureSet == nil { |
| 151 | return MeasureSetPoint{}, false |
| 152 | } |
| 153 | if len(s.measureSetValues) != len(s.desc.measureSet.fields) { |
| 154 | return MeasureSetPoint{}, false |
| 155 | } |
| 156 | return MeasureSetPoint{Values: append([]SampleValue(nil), s.measureSetValues...)}, true |
| 157 | } |
| 158 | |
| 159 | func (r *storeReader) SeriesMeta(name string, labels Labels) (SeriesMeta, bool) { |
| 160 | s, ok := r.lookup(name, labels) |
| 161 | if !ok || !r.visible(s) { |
| 162 | return SeriesMeta{}, false |
| 163 | } |
| 164 | return s.meta, true |
| 165 | } |
| 166 | |
| 167 | func (r *storeReader) MetricMeta(name string) (MetricMeta, bool) { |
| 168 | index := r.byNameIndex() |
| 169 | series := index[name] |
| 170 | if len(series) == 0 { |
| 171 | return MetricMeta{}, false |
| 172 | } |
| 173 | for _, s := range series { |
| 174 | if s.desc == nil { |
| 175 | continue |
| 176 | } |
| 177 | if r.raw || r.visible(s) { |
| 178 | return s.desc.meta, true |
| 179 | } |
| 180 | } |
| 181 | for _, s := range series { |
| 182 | if s.desc != nil { |
| 183 | return s.desc.meta, true |
| 184 | } |
| 185 | } |
| 186 | return MetricMeta{}, false |
| 187 | } |
| 188 | |
| 189 | func (r *storeReader) CollectMeta() CollectMeta { |
| 190 | return r.snap.collectMeta |
| 191 | } |
| 192 | |
| 193 | func (r *storeReader) HostScopes() []HostScope { |
| 194 | // Scope discovery intentionally ignores this reader's active scope filter so |
| 195 | // jobruntime can enumerate all host partitions from one snapshot. |
| 196 | scopes := make(map[string]HostScope) |
| 197 | for _, s := range r.seriesView() { |
| 198 | scopes[s.hostScopeKey] = cloneHostScope(s.hostScope) |
| 199 | } |
| 200 | return sortedHostScopes(scopes) |
| 201 | } |
| 202 | |
| 203 | func flattenSnapshot(src *readSnapshot) *readSnapshot { |
| 204 | series := snapshotSeriesView(src) |
| 205 | dst := &readSnapshot{ |
| 206 | collectMeta: src.collectMeta, |
| 207 | series: make(map[string]*committedSeries, len(series)), |
| 208 | byName: make(map[string][]*committedSeries), |
| 209 | } |
| 210 | |
| 211 | for _, s := range series { |
| 212 | if s.desc == nil { |
| 213 | continue |
| 214 | } |
| 215 | switch s.desc.kind { |
| 216 | case kindGauge, kindCounter: |
| 217 | dst.series[s.key] = cloneCommittedSeries(s) |
| 218 | case kindHistogram: |
| 219 | appendFlattenedHistogramSeries(dst, s) |
| 220 | case kindSummary: |
| 221 | appendFlattenedSummarySeries(dst, s) |
| 222 | case kindStateSet: |
| 223 | appendFlattenedStateSetSeries(dst, s) |
| 224 | case kindMeasureSet: |
| 225 | appendFlattenedMeasureSetSeries(dst, s) |
| 226 | } |
| 227 | } |
| 228 | |
| 229 | dst.byName = buildByName(dst.series) |
| 230 | return dst |
| 231 | } |
| 232 | |
| 233 | func appendFlattenedHistogramSeries(dst *readSnapshot, src *committedSeries) { |
| 234 | schema := src.desc.histogram |
| 235 | if schema == nil { |
| 236 | return |
| 237 | } |
| 238 | if len(src.histogramCumulative) != len(schema.bounds) { |
| 239 | // Defensive guard against malformed snapshots. |
| 240 | // Histogram() follows the same rule and returns unavailable. |
| 241 | return |
| 242 | } |
| 243 | |
| 244 | for i, ub := range schema.bounds { |
| 245 | labelsMap := make(map[string]string, len(src.labels)+1) |
| 246 | for _, lbl := range src.labels { |
| 247 | labelsMap[lbl.Key] = lbl.Value |
| 248 | } |
| 249 | labelsMap[HistogramBucketLabel] = formatHistogramBucketLabel(ub) |
| 250 | |
| 251 | labels, labelsKey, err := canonicalizeLabels(labelsMap) |
| 252 | if err != nil { |
| 253 | continue |
| 254 | } |
| 255 | |
| 256 | name := src.name + "_bucket" |
| 257 | key := makeSeriesKey(src.hostScopeKey, name, labelsKey) |
| 258 | dst.series[key] = &committedSeries{ |
| 259 | id: SeriesID(key), |
| 260 | hash64: seriesIDHash(SeriesID(key)), |
| 261 | key: key, |
| 262 | name: name, |
| 263 | hostScopeKey: src.hostScopeKey, |
| 264 | hostScope: cloneHostScope(src.hostScope), |
| 265 | labels: labels, |
| 266 | labelsKey: labelsKey, |
| 267 | desc: &instrumentDescriptor{ |
| 268 | name: name, |
| 269 | kind: kindCounter, |
| 270 | mode: src.desc.mode, |
| 271 | freshness: src.desc.freshness, |
| 272 | window: src.desc.window, |
| 273 | meta: src.desc.meta, |
| 274 | }, |
| 275 | value: src.histogramCumulative[i], |
| 276 | meta: flattenedSeriesMeta( |
| 277 | src.meta, |
| 278 | MetricKindCounter, |
| 279 | MetricKindHistogram, |
| 280 | FlattenRoleHistogramBucket, |
| 281 | ), |
| 282 | } |
| 283 | } |
| 284 | |
| 285 | infMap := make(map[string]string, len(src.labels)+1) |
| 286 | for _, lbl := range src.labels { |
| 287 | infMap[lbl.Key] = lbl.Value |
| 288 | } |
| 289 | infMap[HistogramBucketLabel] = formatHistogramBucketLabel(math.Inf(1)) |
| 290 | infLabels, infLabelsKey, err := canonicalizeLabels(infMap) |
| 291 | if err == nil { |
| 292 | infName := src.name + "_bucket" |
| 293 | infKey := makeSeriesKey(src.hostScopeKey, infName, infLabelsKey) |
| 294 | dst.series[infKey] = &committedSeries{ |
| 295 | id: SeriesID(infKey), |
| 296 | hash64: seriesIDHash(SeriesID(infKey)), |
| 297 | key: infKey, |
| 298 | name: infName, |
| 299 | hostScopeKey: src.hostScopeKey, |
| 300 | hostScope: cloneHostScope(src.hostScope), |
| 301 | labels: infLabels, |
| 302 | labelsKey: infLabelsKey, |
| 303 | desc: &instrumentDescriptor{ |
| 304 | name: infName, |
| 305 | kind: kindCounter, |
| 306 | mode: src.desc.mode, |
| 307 | freshness: src.desc.freshness, |
| 308 | window: src.desc.window, |
| 309 | meta: src.desc.meta, |
| 310 | }, |
| 311 | value: src.histogramCount, |
| 312 | meta: flattenedSeriesMeta( |
| 313 | src.meta, |
| 314 | MetricKindCounter, |
| 315 | MetricKindHistogram, |
| 316 | FlattenRoleHistogramBucket, |
| 317 | ), |
| 318 | } |
| 319 | } |
| 320 | |
| 321 | appendFlattenedHistogramScalar( |
| 322 | dst, |
| 323 | src, |
| 324 | src.name+"_count", |
| 325 | src.labels, |
| 326 | src.histogramCount, |
| 327 | flattenedSeriesMeta(src.meta, MetricKindCounter, MetricKindHistogram, FlattenRoleHistogramCount), |
| 328 | src.desc, |
| 329 | ) |
| 330 | appendFlattenedHistogramScalar( |
| 331 | dst, |
| 332 | src, |
| 333 | src.name+"_sum", |
| 334 | src.labels, |
| 335 | src.histogramSum, |
| 336 | flattenedSeriesMeta(src.meta, MetricKindCounter, MetricKindHistogram, FlattenRoleHistogramSum), |
| 337 | src.desc, |
| 338 | ) |
| 339 | } |
| 340 | |
| 341 | func appendFlattenedHistogramScalar(dst *readSnapshot, src *committedSeries, name string, labels []Label, value SampleValue, meta SeriesMeta, desc *instrumentDescriptor) { |
| 342 | labelsMap := make(map[string]string, len(labels)) |
| 343 | for _, lbl := range labels { |
| 344 | labelsMap[lbl.Key] = lbl.Value |
| 345 | } |
| 346 | items, labelsKey, err := canonicalizeLabels(labelsMap) |
| 347 | if err != nil { |
| 348 | return |
| 349 | } |
| 350 | key := makeSeriesKey(src.hostScopeKey, name, labelsKey) |
| 351 | dst.series[key] = &committedSeries{ |
| 352 | id: SeriesID(key), |
| 353 | hash64: seriesIDHash(SeriesID(key)), |
| 354 | key: key, |
| 355 | name: name, |
| 356 | hostScopeKey: src.hostScopeKey, |
| 357 | hostScope: cloneHostScope(src.hostScope), |
| 358 | labels: items, |
| 359 | labelsKey: labelsKey, |
| 360 | desc: &instrumentDescriptor{ |
| 361 | name: name, |
| 362 | kind: kindCounter, |
| 363 | mode: desc.mode, |
| 364 | freshness: desc.freshness, |
| 365 | window: desc.window, |
| 366 | meta: desc.meta, |
| 367 | }, |
| 368 | value: value, |
| 369 | meta: meta, |
| 370 | } |
| 371 | } |
| 372 | |
| 373 | func appendFlattenedSummarySeries(dst *readSnapshot, src *committedSeries) { |
| 374 | appendFlattenedHistogramScalar( |
| 375 | dst, |
| 376 | src, |
| 377 | src.name+"_count", |
| 378 | src.labels, |
| 379 | src.summaryCount, |
| 380 | flattenedSeriesMeta(src.meta, MetricKindCounter, MetricKindSummary, FlattenRoleSummaryCount), |
| 381 | src.desc, |
| 382 | ) |
| 383 | appendFlattenedHistogramScalar( |
| 384 | dst, |
| 385 | src, |
| 386 | src.name+"_sum", |
| 387 | src.labels, |
| 388 | src.summarySum, |
| 389 | flattenedSeriesMeta(src.meta, MetricKindCounter, MetricKindSummary, FlattenRoleSummarySum), |
| 390 | src.desc, |
| 391 | ) |
| 392 | |
| 393 | schema := src.desc.summary |
| 394 | if schema == nil { |
| 395 | return |
| 396 | } |
| 397 | if len(src.summaryQuantiles) != len(schema.quantiles) { |
| 398 | return |
| 399 | } |
| 400 | |
| 401 | for i, q := range schema.quantiles { |
| 402 | labelsMap := make(map[string]string, len(src.labels)+1) |
| 403 | for _, lbl := range src.labels { |
| 404 | labelsMap[lbl.Key] = lbl.Value |
| 405 | } |
| 406 | labelsMap[SummaryQuantileLabel] = formatSummaryQuantileLabel(q) |
| 407 | |
| 408 | labels, labelsKey, err := canonicalizeLabels(labelsMap) |
| 409 | if err != nil { |
| 410 | continue |
| 411 | } |
| 412 | key := makeSeriesKey(src.hostScopeKey, src.name, labelsKey) |
| 413 | dst.series[key] = &committedSeries{ |
| 414 | id: SeriesID(key), |
| 415 | hash64: seriesIDHash(SeriesID(key)), |
| 416 | key: key, |
| 417 | name: src.name, |
| 418 | hostScopeKey: src.hostScopeKey, |
| 419 | hostScope: cloneHostScope(src.hostScope), |
| 420 | labels: labels, |
| 421 | labelsKey: labelsKey, |
| 422 | desc: &instrumentDescriptor{ |
| 423 | name: src.name, |
| 424 | kind: kindGauge, |
| 425 | mode: src.desc.mode, |
| 426 | freshness: src.desc.freshness, |
| 427 | window: src.desc.window, |
| 428 | meta: src.desc.meta, |
| 429 | }, |
| 430 | value: src.summaryQuantiles[i], |
| 431 | meta: flattenedSeriesMeta( |
| 432 | src.meta, |
| 433 | MetricKindGauge, |
| 434 | MetricKindSummary, |
| 435 | FlattenRoleSummaryQuantile, |
| 436 | ), |
| 437 | } |
| 438 | } |
| 439 | } |
| 440 | |
| 441 | func appendFlattenedStateSetSeries(dst *readSnapshot, src *committedSeries) { |
| 442 | schema := src.desc.stateSet |
| 443 | if schema == nil { |
| 444 | return |
| 445 | } |
| 446 | |
| 447 | for _, state := range schema.states { |
| 448 | value := SampleValue(0) |
| 449 | if src.stateSetValues[state] { |
| 450 | value = 1 |
| 451 | } |
| 452 | |
| 453 | labelsMap := make(map[string]string, len(src.labels)+1) |
| 454 | for _, lbl := range src.labels { |
| 455 | labelsMap[lbl.Key] = lbl.Value |
| 456 | } |
| 457 | labelsMap[src.name] = state |
| 458 | |
| 459 | labels, labelsKey, err := canonicalizeLabels(labelsMap) |
| 460 | if err != nil { |
| 461 | continue |
| 462 | } |
| 463 | |
| 464 | key := makeSeriesKey(src.hostScopeKey, src.name, labelsKey) |
| 465 | dst.series[key] = &committedSeries{ |
| 466 | id: SeriesID(key), |
| 467 | hash64: seriesIDHash(SeriesID(key)), |
| 468 | key: key, |
| 469 | name: src.name, |
| 470 | hostScopeKey: src.hostScopeKey, |
| 471 | hostScope: cloneHostScope(src.hostScope), |
| 472 | labels: labels, |
| 473 | labelsKey: labelsKey, |
| 474 | desc: &instrumentDescriptor{ |
| 475 | name: src.name, |
| 476 | kind: kindGauge, |
| 477 | mode: src.desc.mode, |
| 478 | freshness: src.desc.freshness, |
| 479 | window: src.desc.window, |
| 480 | meta: src.desc.meta, |
| 481 | }, |
| 482 | value: value, |
| 483 | meta: flattenedSeriesMeta( |
| 484 | src.meta, |
| 485 | MetricKindGauge, |
| 486 | MetricKindStateSet, |
| 487 | FlattenRoleStateSetState, |
| 488 | ), |
| 489 | } |
| 490 | } |
| 491 | } |
| 492 | |
| 493 | func appendFlattenedMeasureSetSeries(dst *readSnapshot, src *committedSeries) { |
| 494 | schema := src.desc.measureSet |
| 495 | if schema == nil || len(src.measureSetValues) != len(schema.fields) { |
| 496 | return |
| 497 | } |
| 498 | |
| 499 | kind := MetricKindGauge |
| 500 | descKind := kindGauge |
| 501 | if schema.semantics == MeasureSetSemanticsCounter { |
| 502 | kind = MetricKindCounter |
| 503 | descKind = kindCounter |
| 504 | } |
| 505 | |
| 506 | for i, field := range schema.fields { |
| 507 | labelsMap := make(map[string]string, len(src.labels)+1) |
| 508 | for _, lbl := range src.labels { |
| 509 | labelsMap[lbl.Key] = lbl.Value |
| 510 | } |
| 511 | labelsMap[MeasureSetFieldLabel] = field.Name |
| 512 | |
| 513 | labels, labelsKey, err := canonicalizeLabels(labelsMap) |
| 514 | if err != nil { |
| 515 | continue |
| 516 | } |
| 517 | |
| 518 | name := src.name + "_" + field.Name |
| 519 | key := makeSeriesKey(src.hostScopeKey, name, labelsKey) |
| 520 | meta := src.desc.meta |
| 521 | meta.Float = field.Float |
| 522 | |
| 523 | series := &committedSeries{ |
| 524 | id: SeriesID(key), |
| 525 | hash64: seriesIDHash(SeriesID(key)), |
| 526 | key: key, |
| 527 | name: name, |
| 528 | hostScopeKey: src.hostScopeKey, |
| 529 | hostScope: cloneHostScope(src.hostScope), |
| 530 | labels: labels, |
| 531 | labelsKey: labelsKey, |
| 532 | desc: &instrumentDescriptor{ |
| 533 | name: name, |
| 534 | kind: descKind, |
| 535 | mode: src.desc.mode, |
| 536 | freshness: src.desc.freshness, |
| 537 | window: src.desc.window, |
| 538 | meta: meta, |
| 539 | }, |
| 540 | value: src.measureSetValues[i], |
| 541 | meta: flattenedSeriesMeta( |
| 542 | src.meta, |
| 543 | kind, |
| 544 | MetricKindMeasureSet, |
| 545 | FlattenRoleMeasureSetField, |
| 546 | ), |
| 547 | } |
| 548 | if schema.semantics == MeasureSetSemanticsCounter { |
| 549 | series.counterCurrent = src.measureSetValues[i] |
| 550 | series.counterCurrentSeq = src.measureSetCurrentSeq |
| 551 | if src.measureSetHasPrev && len(src.measureSetPreviousValues) == len(schema.fields) { |
| 552 | series.counterHasPrev = true |
| 553 | series.counterPrevious = src.measureSetPreviousValues[i] |
| 554 | series.counterPreviousSeq = src.measureSetPreviousSeq |
| 555 | } |
| 556 | } |
| 557 | dst.series[key] = series |
| 558 | } |
| 559 | } |
| 560 | |
| 561 | func (r *storeReader) Family(name string) (FamilyView, bool) { |
| 562 | index := r.byNameIndex() |
| 563 | if slices.ContainsFunc(index[name], r.visible) { |
| 564 | return familyView{name: name, reader: r}, true |
| 565 | } |
| 566 | return nil, false |
| 567 | } |
| 568 | |
| 569 | func (r *storeReader) ForEachByName(name string, fn func(labels LabelView, v SampleValue)) { |
| 570 | index := r.byNameIndex() |
| 571 | for _, s := range index[name] { |
| 572 | if r.visible(s) { |
| 573 | fn(labelView{items: s.labels}, s.value) |
| 574 | } |
| 575 | } |
| 576 | } |
| 577 | |
| 578 | func (r *storeReader) ForEachSeries(fn func(name string, labels LabelView, v SampleValue)) { |
| 579 | r.ForEachSeriesIdentity(func(_ SeriesIdentity, _ SeriesMeta, name string, labels LabelView, v SampleValue) { |
| 580 | fn(name, labels, v) |
| 581 | }) |
| 582 | } |
| 583 | |
| 584 | func (r *storeReader) ForEachSeriesIdentity(fn func(identity SeriesIdentity, meta SeriesMeta, name string, labels LabelView, v SampleValue)) { |
| 585 | r.ForEachSeriesIdentityRaw(func(identity SeriesIdentity, meta SeriesMeta, name string, labels []Label, v SampleValue) { |
| 586 | fn(identity, meta, name, labelView{items: labels}, v) |
| 587 | }) |
| 588 | } |
| 589 | |
| 590 | func (r *storeReader) ForEachSeriesIdentityRaw(fn func(identity SeriesIdentity, meta SeriesMeta, name string, labels []Label, v SampleValue)) { |
| 591 | index := r.byNameIndex() |
| 592 | names := make([]string, 0, len(index)) |
| 593 | for name := range index { |
| 594 | names = append(names, name) |
| 595 | } |
| 596 | sort.Strings(names) |
| 597 | for _, name := range names { |
| 598 | for _, s := range index[name] { |
| 599 | if r.visible(s) { |
| 600 | hash := s.hash64 |
| 601 | if hash == 0 { |
| 602 | hash = seriesIDHash(s.id) |
| 603 | } |
| 604 | fn(SeriesIdentity{ |
| 605 | ID: s.id, |
| 606 | Hash64: hash, |
| 607 | }, s.meta, name, s.labels, s.value) |
| 608 | } |
| 609 | } |
| 610 | } |
| 611 | } |
| 612 | |
| 613 | func (r *storeReader) ForEachMatch(name string, match func(labels LabelView) bool, fn func(labels LabelView, v SampleValue)) { |
| 614 | index := r.byNameIndex() |
| 615 | for _, s := range index[name] { |
| 616 | if !r.visible(s) { |
| 617 | continue |
| 618 | } |
| 619 | lv := labelView{items: s.labels} |
| 620 | if match(lv) { |
| 621 | fn(lv, s.value) |
| 622 | } |
| 623 | } |
| 624 | } |
| 625 | |
| 626 | func (f familyView) Name() string { |
| 627 | return f.name |
| 628 | } |
| 629 | |
| 630 | func (f familyView) ForEach(fn func(labels LabelView, v SampleValue)) { |
| 631 | f.reader.ForEachByName(f.name, fn) |
| 632 | } |
| 633 | |
| 634 | func (r *storeReader) lookup(name string, labels Labels) (*committedSeries, bool) { |
| 635 | items, labelsKey, err := canonicalizeLabels(labels) |
| 636 | if err != nil { |
| 637 | _ = items |
| 638 | return nil, false |
| 639 | } |
| 640 | key := makeSeriesKey(r.hostScopeKey, name, labelsKey) |
| 641 | return lookupSnapshotSeries(r.snap, key) |
| 642 | } |
| 643 | |
| 644 | func (r *storeReader) byNameIndex() map[string][]*committedSeries { |
| 645 | if r.snap.runtimeBase == nil && r.snap.byName != nil { |
| 646 | return r.snap.byName |
| 647 | } |
| 648 | r.indexOnce.Do(func() { |
| 649 | r.index = buildByName(r.seriesView()) |
| 650 | }) |
| 651 | return r.index |
| 652 | } |
| 653 | |
| 654 | func (r *storeReader) seriesView() map[string]*committedSeries { |
| 655 | if r.snap.runtimeBase == nil { |
| 656 | return r.snap.series |
| 657 | } |
| 658 | r.seriesOnce.Do(func() { |
| 659 | r.series = materializeRuntimeSeries(r.snap) |
| 660 | }) |
| 661 | return r.series |
| 662 | } |
| 663 | |
| 664 | func lookupSnapshotSeries(snap *readSnapshot, key string) (*committedSeries, bool) { |
| 665 | for curr := snap; curr != nil; curr = curr.runtimeBase { |
| 666 | if s, ok := curr.series[key]; ok { |
| 667 | return s, true |
| 668 | } |
| 669 | } |
| 670 | return nil, false |
| 671 | } |
| 672 | |
| 673 | func snapshotSeriesView(snap *readSnapshot) map[string]*committedSeries { |
| 674 | if snap.runtimeBase == nil { |
| 675 | return snap.series |
| 676 | } |
| 677 | return materializeRuntimeSeries(snap) |
| 678 | } |
| 679 | |
| 680 | func materializeRuntimeSeries(snap *readSnapshot) map[string]*committedSeries { |
| 681 | chain := make([]*readSnapshot, 0, snap.runtimeDepth+1) |
| 682 | for curr := snap; curr != nil; curr = curr.runtimeBase { |
| 683 | chain = append(chain, curr) |
| 684 | } |
| 685 | if len(chain) == 0 { |
| 686 | return nil |
| 687 | } |
| 688 | // Chain is leaf->root; root map gives the best starting capacity hint. |
| 689 | series := make(map[string]*committedSeries, len(chain[len(chain)-1].series)) |
| 690 | for i := len(chain) - 1; i >= 0; i-- { |
| 691 | maps.Copy(series, chain[i].series) |
| 692 | } |
| 693 | return series |
| 694 | } |
| 695 | |
| 696 | // visible applies freshness policy for Read(); Read(ReadRaw()) bypasses it. |
| 697 | func (r *storeReader) visible(s *committedSeries) bool { |
| 698 | if s.hostScopeKey != r.hostScopeKey { |
| 699 | return false |
| 700 | } |
| 701 | if r.raw { |
| 702 | return true |
| 703 | } |
| 704 | if s.desc == nil { |
| 705 | return false |
| 706 | } |
| 707 | if s.desc.freshness == FreshnessCommitted { |
| 708 | return true |
| 709 | } |
| 710 | meta := r.snap.collectMeta |
| 711 | return meta.LastAttemptStatus == CollectStatusSuccess && s.meta.LastSeenSuccessSeq == meta.LastSuccessSeq |
| 712 | } |