@cryptotaxi247 / netdata-1 / commits / da7606586

feat(go/plugin/metrix): add MeasureSet structured family support (#21909)

Ilya Mashchenko committed Mar 8, 2026 at 00:46 UTC da76065869a5dc6b14d113e4cde9f5294ef87320
26 files changed +2176 -84
src/go/pkg/metrix/README.md
+117 -17
@@ -20,7 +20,7 @@
20 - **Cycle-scoped collector writes** — `CollectorStore` writes are staged between `BeginCycle` and `CommitCycleSuccess`. Nothing is visible to readers until commit.
21 - **Stateful runtime writes** — `RuntimeStore` writes are committed immediately (no cycle API). Each write produces a new overlay snapshot.
22 - **Label canonicalization** — Label maps (`map[string]string`) are sorted and encoded into a canonical key that uniquely identifies a series (metric name + labels).
23 -- **Typed + flattened views** — Reader supports canonical typed families (Histogram, Summary, StateSet) and a flattened scalar view where complex types are projected into individual scalar series.
23 +- **Typed + flattened views** — Reader supports canonical typed families (Histogram, Summary, StateSet, MeasureSet) and a flattened scalar view where complex types are projected into individual scalar series.
24
25 ## Key Definitions
26
@@ -33,12 +33,12 @@
33
34 ## Stores and Interfaces
35
36 -| Interface | Key methods | Notes |
37 -|---------------------|----------------------------------------------|-----------------------------------------|
38 -| `CollectorStore` | `Read(...)`, `Write()` | Default collector-facing store |
39 -| `RuntimeStore` | `Read(...)`, `Write()` | Stateful-only writes |
40 -| `CycleManagedStore` | `CycleController()` | Runtime/orchestrator-only cycle control |
41 -| `Reader` | `Value/Delta/Histogram/Summary/StateSet/...` | Immutable snapshot read API |
36 +| Interface | Key methods | Notes |
37 +|---------------------|---------------------------------------------------------|-----------------------------------------|
38 +| `CollectorStore` | `Read(...)`, `Write()` | Default collector-facing store |
39 +| `RuntimeStore` | `Read(...)`, `Write()` | Stateful-only writes |
40 +| `CycleManagedStore` | `CycleController()` | Runtime/orchestrator-only cycle control |
41 +| `Reader` | `Value/Delta/Histogram/Summary/StateSet/MeasureSet/...` | Immutable snapshot read API |
42
43 ## Write Model
44
@@ -72,23 +72,107 @@
72
73 ## Instrument Options
74
75 -| Option | Scope |
76 -|-----------------------------------------------------------------|---------------------------------------------------------------------------------------------|
77 -| `WithFreshness(...)` | Freshness policy override (subject to mode constraints) |
78 -| `WithWindow(...)` | Stateful histogram/summary window mode |
79 -| `WithHistogramBounds(...)` | Histogram bucket boundaries |
80 -| `WithSummaryQuantiles(...)` | Summary quantile output (required for quantile series in flattened view) |
81 -| `WithSummaryReservoirSize(...)` | Stateful summary estimator size |
82 -| `WithStateSetStates(...)` | StateSet allowed states |
83 -| `WithStateSetMode(...)` | `ModeBitSet` (multiple simultaneous active states) or `ModeEnum` (exactly one active state) |
75 +| Option | Scope |
76 +|-----------------------------------------------------------------------------------|------------------------------------------------------------------------------------------------|
77 +| `WithFreshness(...)` | Freshness policy override (subject to mode constraints) |
78 +| `WithWindow(...)` | Stateful histogram/summary window mode |
79 +| `WithHistogramBounds(...)` | Histogram bucket boundaries |
80 +| `WithSummaryQuantiles(...)` | Summary quantile output (required for quantile series in flattened view) |
81 +| `WithSummaryReservoirSize(...)` | Stateful summary estimator size |
82 +| `WithStateSetStates(...)` | StateSet allowed states |
83 +| `WithStateSetMode(...)` | `ModeBitSet` (multiple simultaneous active states) or `ModeEnum` (exactly one active state) |
84 +| `WithMeasureSetFields(...)` | MeasureSet fixed ordered field schema (required for MeasureSet instruments) |
85 | `WithDescription(...)`, `WithChartFamily(...)`, `WithUnit(...)`, `WithFloat(...)` | Metric metadata hints for downstream consumers (e.g., autogen chart identity + float SET mode) |
86
87 +## MeasureSet
88 +
89 +- **Structured numeric family** — `MeasureSet` stores one logical metric family with a fixed ordered list of named numeric fields.
90 +- **Family-level semantics** — one `MeasureSet` family is either gauge-like or counter-like; semantics are never mixed per field.
91 +- **Family-level metadata** — `Description`, `ChartFamily`, and `Unit` apply to the whole family.
92 +- **Field-level schema** — `MeasureFieldSpec` declares per-field `Name` and `Float`.
93 +- **Chartengine integration** — chart autogen treats `MeasureSet` as a structured family, similar to `StateSet`; flatten remains the generic reader/tooling path.
94 +
95 +### Writers
96 +
97 +| Mode | Gauge-like family | Counter-like family |
98 +|----------|---------------------------------------------------------------------------------------------------------------------------------|---------------------------------------------------------------------------------------------------------------|
99 +| Snapshot | `MeasureSetGauge(...).ObservePoint(...)` or preferred `ObserveFields(...)` | `MeasureSetCounter(...).ObserveTotalPoint(...)` or preferred `ObserveTotalFields(...)` |
100 +| Stateful | `MeasureSetGauge(...).SetPoint(...)`, `AddPoint(...)`, `SetFields(...)`, `AddFields(...)`, `SetField(...)`, `AddField(...)` | `MeasureSetCounter(...).AddPoint(...)`, `AddFields(...)`, `AddField(...)` |
101 +
102 +### Phase-1 write contract
103 +
104 +- **Preferred collector-facing API** — use named write helpers instead of raw positional `MeasureSetPoint` values whenever practical.
105 +- **Snapshot handles** support:
106 + - full-family positional writes (`ObservePoint(...)`, `ObserveTotalPoint(...)`)
107 + - full-family named writes (`ObserveFields(...)`, `ObserveTotalFields(...)`)
108 +- **Stateful handles** support:
109 + - full-family positional writes
110 + - full-family named writes
111 + - singular field writes (`SetField(...)`, `AddField(...)`)
112 +- **Snapshot singular field writes do not exist in phase 1.**
113 + - `MeasureSet` still models one sampled family point per collect cycle in snapshot mode.
114 + - Partial snapshot field visibility/completeness semantics are intentionally deferred.
115 +- **Named full-family writes require the exact declared field set.**
116 + - Missing fields panic.
117 + - Unknown extra fields panic.
118 +- **Stateful singular field writes update only the addressed field.**
119 + - Gauge-like `SetField(...)` overwrites one field on top of the committed/staged family.
120 + - Gauge-like `AddField(...)` and counter-like `AddField(...)` apply a delta to one field.
121 +
122 +### Schema example
123 +
124 +```go
125 +store := metrix.NewCollectorStore()
126 +meter := store.Write().SnapshotMeter("svc")
127 +latency := meter.MeasureSetGauge(
128 + "latency",
129 + metrix.WithMeasureSetFields(
130 + metrix.MeasureFieldSpec{Name: "value"},
131 + metrix.MeasureFieldSpec{Name: "ratio", Float: true},
132 + ),
133 + metrix.WithUnit("seconds"),
134 +)
135 +latency.ObserveFields(map[string]metrix.SampleValue{
136 + "value": 1.5,
137 + "ratio": 0.5,
138 +})
139 +```
140 +
141 +Re-registering the same metric name with a different `MeasureSet` schema is rejected like other structured families.
142 +
143 +### Stateful singular-write example
144 +
145 +```go
146 +store := metrix.NewRuntimeStore()
147 +meter := store.Write().StatefulMeter("svc")
148 +usage := meter.MeasureSetGauge(
149 + "usage",
150 + metrix.WithMeasureSetFields(
151 + metrix.MeasureFieldSpec{Name: "value"},
152 + metrix.MeasureFieldSpec{Name: "limit"},
153 + ),
154 +)
155 +
156 +usage.SetFields(map[string]metrix.SampleValue{
157 + "value": 10,
158 + "limit": 20,
159 +})
160 +usage.SetField("value", 15)
161 +usage.AddField("limit", 3)
162 +```
163 +
164 +This yields a committed `MeasureSet` point equivalent to:
165 +
166 +```go
167 +metrix.MeasureSetPoint{Values: []metrix.SampleValue{15, 23}}
168 +```
169 +
170 ## Read Modes
171
172 `Read(...)` accepts option functions that control two independent axes:
173
174 - **Raw** (`ReadRaw()`) — bypasses freshness filtering, returning all committed series regardless of when they were last observed.
91 -- **Flatten** (`ReadFlatten()`) — projects complex types (Histogram, Summary, StateSet) into individual scalar series.
175 +- **Flatten** (`ReadFlatten()`) — projects complex types (Histogram, Summary, StateSet, MeasureSet) into individual scalar series.
176
177 | Read options | Visibility | Shape |
178 |----------------------------------|----------------------|--------------------------|
@@ -106,9 +190,12 @@
190 | Histogram | `<name>_bucket{le=...}`, `<name>_count`, `<name>_sum` |
191 | Summary | `<name>_count`, `<name>_sum` (always); `<name>{quantile=...}` (only when `WithSummaryQuantiles()` is configured) |
192 | StateSet | `<name>{<name>=state}` with scalar 0/1 values |
193 +| MeasureSet | `<name>_<field>{measure_field=field}`; flattened kind follows family semantics (`Gauge` or `Counter`) |
194
195 Flatten metadata is exposed via `SeriesMeta.Kind`, `SeriesMeta.SourceKind`, and `SeriesMeta.FlattenRole`.
196
197 +`MeasureSet` flattening keeps per-field metric names for `MetricMeta(name)` compatibility and also adds a synthetic `measure_field=<field>` label. This gives chartengine explicit field identity without widening the reader metadata API.
198 +
199 ## Minimal Usage Snippets
200
201 ### Collector write path
@@ -129,6 +216,15 @@ _ = value
216 _ = ok
217 ```
218
219 +### Direct MeasureSet read
220 +
221 +```go
222 +reader := store.Read()
223 +point, ok := reader.MeasureSet("svc.latency", nil)
224 +_ = point
225 +_ = ok
226 +```
227 +
228 For a complete collector integration pattern (cycle management, error handling),
229 see [how-to-write-a-collector.md](/src/go/plugin/go.d/docs/how-to-write-a-collector.md).
230
@@ -141,8 +237,12 @@ see [how-to-write-a-collector.md](/src/go/plugin/go.d/docs/how-to-write-a-collec
237 - **Snapshot freshness** — Snapshot-mode instruments cannot use `FreshnessCommitted`.
238 - **Runtime writes** — `RuntimeStore` rejects snapshot-mode instrument registration with an error.
239 Calling snapshot-mode record methods (`ObserveTotal`, `ObservePoint`) **panics**.
240 +- **MeasureSet runtime writes** — `RuntimeStore` supports both gauge-like and counter-like `MeasureSet` families, but only through `StatefulMeter(...)`.
241 +- **MeasureSet named writes** — `ObserveFields(...)`, `ObserveTotalFields(...)`, `SetFields(...)`, and `AddFields(...)` require the exact declared field set. Snapshot singular field writes are intentionally absent in phase 1.
242 - **Window/freshness coupling** — Stateful histogram/summary with `WindowCycle` requires (and silently forces) `FreshnessCycle`. Setting an explicit non-Cycle freshness with `WindowCycle` returns an error.
243 - **Schema stability** — Re-registering an existing metric name with different kind/mode/schema returns an error (or panics in strict runtime paths).
244 +- **MeasureSet flatten naming** — Flattened `MeasureSet` series use per-field metric names like `<name>_<field>` and also carry a synthetic `measure_field=<field>` label.
245 +- **MeasureSet counter semantics** — Stateful counter-like `MeasureSet` families reject negative `AddPoint(...)` deltas, just like scalar counters.
246 - **Collector retention** — `CollectorStore` evicts series not seen for 10 successful cycles by default.
247
248 ## Internal Architecture Notes
src/go/pkg/metrix/backend.go
+6
@@ -17,6 +17,12 @@ type meterBackend interface {
17 recordSummaryObservePoint(desc *instrumentDescriptor, point SummaryPoint, sets []LabelSet)
18 recordSummaryObserve(desc *instrumentDescriptor, value SampleValue, sets []LabelSet)
19 recordStateSetObserve(desc *instrumentDescriptor, point StateSetPoint, sets []LabelSet)
20 + recordMeasureSetGaugeObservePoint(desc *instrumentDescriptor, point MeasureSetPoint, sets []LabelSet)
21 + recordMeasureSetGaugeSetPoint(desc *instrumentDescriptor, point MeasureSetPoint, sets []LabelSet)
22 + recordMeasureSetGaugeAddPoint(desc *instrumentDescriptor, delta MeasureSetPoint, sets []LabelSet)
23 + recordMeasureSetGaugeSetField(desc *instrumentDescriptor, field string, value SampleValue, sets []LabelSet)
24 + recordMeasureSetCounterObserveTotalPoint(desc *instrumentDescriptor, point MeasureSetPoint, sets []LabelSet)
25 + recordMeasureSetCounterAddPoint(desc *instrumentDescriptor, delta MeasureSetPoint, sets []LabelSet)
26 }
27
28 var _ meterBackend = (*storeCore)(nil)
src/go/pkg/metrix/collector_store.go
+140 -30
@@ -21,6 +21,7 @@ const (
21 kindHistogram
22 kindSummary
23 kindStateSet
24 + kindMeasureSet
25 )
26
27 const (
@@ -29,15 +30,16 @@ const (
30 )
31
32 type instrumentDescriptor struct {
32 - name string
33 - kind metricKind
34 - mode metricMode
35 - freshness FreshnessPolicy // visibility policy used by Read()
36 - window MetricWindow
37 - histogram *histogramSchema // set for kindHistogram only
38 - summary *summarySchema // set for kindSummary only
39 - stateSet *stateSetSchema // set for kindStateSet only
40 - meta MetricMeta
33 + name string
34 + kind metricKind
35 + mode metricMode
36 + freshness FreshnessPolicy // visibility policy used by Read()
37 + window MetricWindow
38 + histogram *histogramSchema // set for kindHistogram only
39 + summary *summarySchema // set for kindSummary only
40 + stateSet *stateSetSchema // set for kindStateSet only
41 + measureSet *measureSetSchema // set for kindMeasureSet only
42 + meta MetricMeta
43 }
44
45 type histogramSchema struct {
@@ -55,6 +57,12 @@ type stateSetSchema struct {
57 index map[string]struct{}
58 }
59
60 +type measureSetSchema struct {
61 + semantics MeasureSetSemantics
62 + fields []MeasureFieldSpec
63 + index map[string]int
64 +}
65 +
66 type committedSeries struct {
67 id SeriesID
68 hash64 uint64
@@ -91,6 +99,13 @@ type committedSeries struct {
99 // StateSet current sample (used by StateSet()).
100 stateSetValues map[string]bool
101
102 + // MeasureSet current sample (used by MeasureSet()).
103 + measureSetValues []SampleValue
104 + measureSetPreviousValues []SampleValue
105 + measureSetHasPrev bool
106 + measureSetCurrentSeq uint64
107 + measureSetPreviousSeq uint64
108 +
109 meta SeriesMeta
110 }
111
@@ -105,12 +120,14 @@ type readSnapshot struct {
120 }
121
122 type cycleFrame struct {
108 - seq uint64
109 - gauges map[string]*stagedGauge
110 - counters map[string]*stagedCounter
111 - histograms map[string]*stagedHistogram
112 - summaries map[string]*stagedSummary
113 - stateSet map[string]*stagedStateSet
123 + seq uint64
124 + gauges map[string]*stagedGauge
125 + counters map[string]*stagedCounter
126 + histograms map[string]*stagedHistogram
127 + summaries map[string]*stagedSummary
128 + stateSet map[string]*stagedStateSet
129 + measureSetGauges map[string]*stagedMeasureSet
130 + measureSetCounters map[string]*stagedMeasureSet
131 }
132
133 type storeCore struct {
@@ -217,12 +234,14 @@ func (c *storeCycleController) BeginCycle() {
234
235 c.core.sequence++
236 c.core.active = &cycleFrame{
220 - seq: c.core.sequence,
221 - gauges: make(map[string]*stagedGauge),
222 - counters: make(map[string]*stagedCounter),
223 - histograms: make(map[string]*stagedHistogram),
224 - summaries: make(map[string]*stagedSummary),
225 - stateSet: make(map[string]*stagedStateSet),
237 + seq: c.core.sequence,
238 + gauges: make(map[string]*stagedGauge),
239 + counters: make(map[string]*stagedCounter),
240 + histograms: make(map[string]*stagedHistogram),
241 + summaries: make(map[string]*stagedSummary),
242 + stateSet: make(map[string]*stagedStateSet),
243 + measureSetGauges: make(map[string]*stagedMeasureSet),
244 + measureSetCounters: make(map[string]*stagedMeasureSet),
245 }
246 }
247
@@ -334,6 +353,30 @@ func (c *storeCycleController) CommitCycleSuccess() {
353 markSeriesSeen(series, c.core.active.seq, successSeq)
354 }
355
356 + for key, staged := range c.core.active.measureSetGauges {
357 + series := getOrCreateCommitSeries(oldSnap, next, key, staged.name, staged.labels, staged.labelsKey, staged.desc)
358 + series.measureSetValues = append(series.measureSetValues[:0], staged.values...)
359 + markSeriesSeen(series, c.core.active.seq, successSeq)
360 + }
361 +
362 + for key, staged := range c.core.active.measureSetCounters {
363 + series := getOrCreateCommitSeries(oldSnap, next, key, staged.name, staged.labels, staged.labelsKey, staged.desc)
364 +
365 + if series.desc != nil && series.desc.kind == kindMeasureSet && series.desc.measureSet != nil && series.desc.measureSet.semantics == MeasureSetSemanticsCounter && series.measureSetCurrentSeq > 0 {
366 + series.measureSetPreviousValues = append(series.measureSetPreviousValues[:0], series.measureSetValues...)
367 + series.measureSetPreviousSeq = series.measureSetCurrentSeq
368 + series.measureSetHasPrev = true
369 + } else {
370 + series.measureSetPreviousValues = nil
371 + series.measureSetPreviousSeq = 0
372 + series.measureSetHasPrev = false
373 + }
374 +
375 + series.measureSetValues = append(series.measureSetValues[:0], staged.values...)
376 + series.measureSetCurrentSeq = c.core.active.seq
377 + markSeriesSeen(series, c.core.active.seq, successSeq)
378 + }
379 +
380 applyCollectorRetention(next.series, c.core.retention, successSeq)
381 next.collectMeta.LastAttemptSeq = c.core.active.seq
382 next.collectMeta.LastAttemptStatus = CollectStatusSuccess
@@ -475,6 +518,9 @@ func (c *storeCore) registerInstrument(name string, kind metricKind, mode metric
518 if (len(cfg.states) > 0 || cfg.stateSetMode != nil) && kind != kindStateSet {
519 return nil, fmt.Errorf("metrix: stateset options are invalid for this instrument kind")
520 }
521 + if (len(cfg.measureSetFields) > 0 || cfg.measureSetSemantics != nil) && kind != kindMeasureSet {
522 + return nil, fmt.Errorf("metrix: measureset options are invalid for this instrument kind")
523 + }
524
525 window := WindowCumulative
526 if cfg.windowSet {
@@ -529,6 +575,15 @@ func (c *storeCore) registerInstrument(name string, kind metricKind, mode metric
575 schema = s
576 }
577
578 + var measureSet *measureSetSchema
579 + if kind == kindMeasureSet {
580 + s, err := buildMeasureSetSchema(cfg)
581 + if err != nil {
582 + return nil, err
583 + }
584 + measureSet = s
585 + }
586 +
587 c.mu.Lock()
588 defer c.mu.Unlock()
589
@@ -556,6 +611,9 @@ func (c *storeCore) registerInstrument(name string, kind metricKind, mode metric
611 if kind == kindStateSet && !equalStateSetSchema(d.stateSet, schema) {
612 return nil, fmt.Errorf("metrix: stateset schema mismatch for %s", name)
613 }
614 + if kind == kindMeasureSet && !equalMeasureSetSchema(d.measureSet, measureSet) {
615 + return nil, fmt.Errorf("metrix: measureset schema mismatch for %s", name)
616 + }
617 if cfg.descriptionSet && d.meta.Description != metricMeta.Description {
618 return nil, fmt.Errorf("metrix: metric description mismatch for %s", name)
619 }
@@ -572,15 +630,16 @@ func (c *storeCore) registerInstrument(name string, kind metricKind, mode metric
630 }
631
632 d := &instrumentDescriptor{
575 - name: name,
576 - kind: kind,
577 - mode: mode,
578 - freshness: fresh,
579 - window: window,
580 - histogram: histogram,
581 - summary: summary,
582 - stateSet: schema,
583 - meta: metricMeta,
633 + name: name,
634 + kind: kind,
635 + mode: mode,
636 + freshness: fresh,
637 + window: window,
638 + histogram: histogram,
639 + summary: summary,
640 + stateSet: schema,
641 + measureSet: measureSet,
642 + meta: metricMeta,
643 }
644 c.instruments[name] = d
645 return d, nil
@@ -602,6 +661,12 @@ func cloneCommittedSeries(s *committedSeries) *committedSeries {
661 if s.stateSetValues != nil {
662 cp.stateSetValues = cloneStateMap(s.stateSetValues)
663 }
664 + if len(s.measureSetValues) > 0 {
665 + cp.measureSetValues = append([]SampleValue(nil), s.measureSetValues...)
666 + }
667 + if len(s.measureSetPreviousValues) > 0 {
668 + cp.measureSetPreviousValues = append([]SampleValue(nil), s.measureSetPreviousValues...)
669 + }
670 if len(s.histogramCumulative) > 0 {
671 cp.histogramCumulative = append([]SampleValue(nil), s.histogramCumulative...)
672 }
@@ -719,6 +784,51 @@ func equalStateSetSchema(a, b *stateSetSchema) bool {
784 return true
785 }
786
787 +func buildMeasureSetSchema(cfg instrumentConfig) (*measureSetSchema, error) {
788 + if len(cfg.measureSetFields) == 0 {
789 + return nil, fmt.Errorf("metrix: measureset requires WithMeasureSetFields")
790 + }
791 + if cfg.measureSetSemantics == nil {
792 + return nil, fmt.Errorf("metrix: measureset semantics are missing")
793 + }
794 +
795 + fields := make([]MeasureFieldSpec, 0, len(cfg.measureSetFields))
796 + index := make(map[string]int, len(cfg.measureSetFields))
797 + for i, field := range cfg.measureSetFields {
798 + name := strings.TrimSpace(field.Name)
799 + if name == "" {
800 + return nil, fmt.Errorf("metrix: measureset field name cannot be empty")
801 + }
802 + if _, ok := index[name]; ok {
803 + return nil, fmt.Errorf("metrix: duplicate measureset field %q", name)
804 + }
805 + field.Name = name
806 + fields = append(fields, field)
807 + index[name] = i
808 + }
809 +
810 + return &measureSetSchema{
811 + semantics: *cfg.measureSetSemantics,
812 + fields: fields,
813 + index: index,
814 + }, nil
815 +}
816 +
817 +func equalMeasureSetSchema(a, b *measureSetSchema) bool {
818 + if a == nil || b == nil {
819 + return a == b
820 + }
821 + if a.semantics != b.semantics || len(a.fields) != len(b.fields) {
822 + return false
823 + }
824 + for i := range a.fields {
825 + if a.fields[i].Name != b.fields[i].Name || a.fields[i].Float != b.fields[i].Float {
826 + return false
827 + }
828 + }
829 + return true
830 +}
831 +
832 func equalHistogramSchema(a, b *histogramSchema) bool {
833 if a == nil || b == nil {
834 return a == b
src/go/pkg/metrix/errors.go
+5
@@ -25,6 +25,11 @@ var (
25 errStateSetEnumCount = errors.New("metrix: stateset enum mode requires exactly one active state")
26 errStateSetUnknownState = errors.New("metrix: stateset point contains undeclared state")
27 errStateSetLabelKey = errors.New("metrix: stateset flatten label key collides with existing label")
28 + errMeasureSetLabelKey = errors.New("metrix: measureset flatten label key collides with existing label")
29 + errMeasureSetSchema = errors.New("metrix: measureset schema is missing")
30 + errMeasureSetPoint = errors.New("metrix: invalid measureset point")
31 + errMeasureSetFields = errors.New("metrix: invalid measureset fields")
32 + errMeasureSetField = errors.New("metrix: unknown measureset field")
33 errRuntimeSnapshotWrite = errors.New("metrix: runtime store supports stateful writes only")
34 errRuntimeFreshness = errors.New("metrix: runtime store freshness is fixed to FreshnessCommitted")
35 errRuntimeWindowCycle = errors.New("metrix: runtime store does not support window=cycle")
src/go/pkg/metrix/flatten_meta_test.go
+29
@@ -71,6 +71,35 @@ func TestFlattenSeriesMetaCarriesOriginType(t *testing.T) {
71 assert.Equal(t, FlattenRoleStateSetState, flatMeta.FlattenRole)
72 },
73 },
74 + "measureset flatten series carry source kind and role": {
75 + run: func(t *testing.T) {
76 + s := NewCollectorStore()
77 + cc := cycleController(t, s)
78 +
79 + ms := s.Write().SnapshotMeter("svc").MeasureSetGauge(
80 + "latency",
81 + WithMeasureSetFields(
82 + MeasureFieldSpec{Name: "value"},
83 + MeasureFieldSpec{Name: "limit"},
84 + ),
85 + )
86 + cc.BeginCycle()
87 + ms.ObservePoint(MeasureSetPoint{Values: []SampleValue{1, 2}})
88 + cc.CommitCycleSuccess()
89 +
90 + rawMeta, ok := s.Read().SeriesMeta("svc.latency", nil)
91 + require.True(t, ok)
92 + assert.Equal(t, MetricKindMeasureSet, rawMeta.Kind)
93 + assert.Equal(t, MetricKindMeasureSet, rawMeta.SourceKind)
94 + assert.Equal(t, FlattenRoleNone, rawMeta.FlattenRole)
95 +
96 + flatMeta, ok := s.Read(ReadFlatten()).SeriesMeta("svc.latency_value", measureSetFieldLabels("value"))
97 + require.True(t, ok)
98 + assert.Equal(t, MetricKindGauge, flatMeta.Kind)
99 + assert.Equal(t, MetricKindMeasureSet, flatMeta.SourceKind)
100 + assert.Equal(t, FlattenRoleMeasureSetField, flatMeta.FlattenRole)
101 + },
102 + },
103 }
104
105 for name, tc := range tests {
src/go/pkg/metrix/histogram.go
+3 -3
@@ -9,7 +9,7 @@ import (
9 "strconv"
10 )
11
12 -const histogramBucketLabel = "le"
12 +const HistogramBucketLabel = "le"
13
14 // snapshotHistogramInstrument writes sampled full histogram points.
15 type snapshotHistogramInstrument struct {
@@ -87,7 +87,7 @@ func (c *storeCore) recordHistogramObservePoint(desc *instrumentDescriptor, poin
87 if err != nil {
88 panic(err)
89 }
90 - if labelsContainKey(labels, histogramBucketLabel) {
90 + if labelsContainKey(labels, HistogramBucketLabel) {
91 panic(errHistogramLabelKey)
92 }
93
@@ -140,7 +140,7 @@ func (c *storeCore) recordHistogramObserve(desc *instrumentDescriptor, value Sam
140 if err != nil {
141 panic(err)
142 }
143 - if labelsContainKey(labels, histogramBucketLabel) {
143 + if labelsContainKey(labels, HistogramBucketLabel) {
144 panic(errHistogramLabelKey)
145 }
146
src/go/pkg/metrix/interfaces.go
+54
@@ -38,6 +38,7 @@ type Reader interface {
38 Histogram(name string, labels Labels) (HistogramPoint, bool)
39 Summary(name string, labels Labels) (SummaryPoint, bool)
40 StateSet(name string, labels Labels) (StateSetPoint, bool)
41 + MeasureSet(name string, labels Labels) (MeasureSetPoint, bool)
42 SeriesMeta(name string, labels Labels) (SeriesMeta, bool)
43 // MetricMeta resolves metadata by metric name in the active reader view.
44 // With Read(ReadFlatten()), lookups use flattened scalar series names.
@@ -94,6 +95,8 @@ type SnapshotMeter interface {
95 Histogram(name string, opts ...InstrumentOption) SnapshotHistogram
96 Summary(name string, opts ...InstrumentOption) SnapshotSummary
97 StateSet(name string, opts ...InstrumentOption) StateSetInstrument
98 + MeasureSetGauge(name string, opts ...InstrumentOption) SnapshotMeasureSetGauge
99 + MeasureSetCounter(name string, opts ...InstrumentOption) SnapshotMeasureSetCounter
100 LabelSet(labels ...Label) LabelSet
101 }
102
@@ -104,6 +107,8 @@ type SnapshotVecMeter interface {
107 Histogram(name string, opts ...InstrumentOption) SnapshotHistogramVec
108 Summary(name string, opts ...InstrumentOption) SnapshotSummaryVec
109 StateSet(name string, opts ...InstrumentOption) SnapshotStateSetVec
110 + MeasureSetGauge(name string, opts ...InstrumentOption) SnapshotMeasureSetGaugeVec
111 + MeasureSetCounter(name string, opts ...InstrumentOption) SnapshotMeasureSetCounterVec
112 }
113
114 // StatefulMeter declares stateful-mode instruments under a metric-name prefix.
@@ -117,6 +122,8 @@ type StatefulMeter interface {
122 Histogram(name string, opts ...InstrumentOption) StatefulHistogram
123 Summary(name string, opts ...InstrumentOption) StatefulSummary
124 StateSet(name string, opts ...InstrumentOption) StateSetInstrument
125 + MeasureSetGauge(name string, opts ...InstrumentOption) StatefulMeasureSetGauge
126 + MeasureSetCounter(name string, opts ...InstrumentOption) StatefulMeasureSetCounter
127 LabelSet(labels ...Label) LabelSet
128 }
129
@@ -127,6 +134,8 @@ type StatefulVecMeter interface {
134 Histogram(name string, opts ...InstrumentOption) StatefulHistogramVec
135 Summary(name string, opts ...InstrumentOption) StatefulSummaryVec
136 StateSet(name string, opts ...InstrumentOption) StatefulStateSetVec
137 + MeasureSetGauge(name string, opts ...InstrumentOption) StatefulMeasureSetGaugeVec
138 + MeasureSetCounter(name string, opts ...InstrumentOption) StatefulMeasureSetCounterVec
139 }
140
141 // SnapshotGauge writes sampled absolute values; last write wins in a cycle.
@@ -221,6 +230,51 @@ type StateSetInstrument interface {
230 Enable(actives ...string)
231 }
232
233 +type SnapshotMeasureSetGauge interface {
234 + ObservePoint(p MeasureSetPoint, labels ...LabelSet)
235 + ObserveFields(fields map[string]SampleValue, labels ...LabelSet)
236 +}
237 +
238 +type SnapshotMeasureSetGaugeVec interface {
239 + GetWithLabelValues(labelValues ...string) (SnapshotMeasureSetGauge, error)
240 + WithLabelValues(labelValues ...string) SnapshotMeasureSetGauge
241 +}
242 +
243 +type SnapshotMeasureSetCounter interface {
244 + ObserveTotalPoint(p MeasureSetPoint, labels ...LabelSet)
245 + ObserveTotalFields(fields map[string]SampleValue, labels ...LabelSet)
246 +}
247 +
248 +type SnapshotMeasureSetCounterVec interface {
249 + GetWithLabelValues(labelValues ...string) (SnapshotMeasureSetCounter, error)
250 + WithLabelValues(labelValues ...string) SnapshotMeasureSetCounter
251 +}
252 +
253 +type StatefulMeasureSetGauge interface {
254 + SetPoint(p MeasureSetPoint, labels ...LabelSet)
255 + SetFields(fields map[string]SampleValue, labels ...LabelSet)
256 + SetField(field string, value SampleValue, labels ...LabelSet)
257 + AddPoint(delta MeasureSetPoint, labels ...LabelSet)
258 + AddFields(delta map[string]SampleValue, labels ...LabelSet)
259 + AddField(field string, delta SampleValue, labels ...LabelSet)
260 +}
261 +
262 +type StatefulMeasureSetGaugeVec interface {
263 + GetWithLabelValues(labelValues ...string) (StatefulMeasureSetGauge, error)
264 + WithLabelValues(labelValues ...string) StatefulMeasureSetGauge
265 +}
266 +
267 +type StatefulMeasureSetCounter interface {
268 + AddPoint(delta MeasureSetPoint, labels ...LabelSet)
269 + AddFields(delta map[string]SampleValue, labels ...LabelSet)
270 + AddField(field string, delta SampleValue, labels ...LabelSet)
271 +}
272 +
273 +type StatefulMeasureSetCounterVec interface {
274 + GetWithLabelValues(labelValues ...string) (StatefulMeasureSetCounter, error)
275 + WithLabelValues(labelValues ...string) StatefulMeasureSetCounter
276 +}
277 +
278 // SnapshotStateSetVec provides labeled series handles for snapshot statesets.
279 type SnapshotStateSetVec interface {
280 GetWithLabelValues(labelValues ...string) (StateSetInstrument, error)
src/go/pkg/metrix/measureset.go new
+448
@@ -0,0 +1,448 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package metrix
4 +
5 +const MeasureSetFieldLabel = "measure_field"
6 +
7 +// stagedMeasureSet holds one in-cycle MeasureSet sample for a single series identity.
8 +type stagedMeasureSet struct {
9 + key string
10 + name string
11 + labels []Label
12 + labelsKey string
13 + desc *instrumentDescriptor
14 + values []SampleValue
15 +}
16 +
17 +// snapshotMeasureSetGaugeInstrument writes sampled MeasureSet gauge points.
18 +type snapshotMeasureSetGaugeInstrument struct {
19 + backend meterBackend
20 + desc *instrumentDescriptor
21 + base []LabelSet
22 +}
23 +
24 +// snapshotMeasureSetCounterInstrument writes sampled MeasureSet counter totals.
25 +type snapshotMeasureSetCounterInstrument struct {
26 + backend meterBackend
27 + desc *instrumentDescriptor
28 + base []LabelSet
29 +}
30 +
31 +// statefulMeasureSetGaugeInstrument writes maintained MeasureSet gauge points.
32 +type statefulMeasureSetGaugeInstrument struct {
33 + backend meterBackend
34 + desc *instrumentDescriptor
35 + base []LabelSet
36 +}
37 +
38 +// statefulMeasureSetCounterInstrument writes maintained MeasureSet counter deltas.
39 +type statefulMeasureSetCounterInstrument struct {
40 + backend meterBackend
41 + desc *instrumentDescriptor
42 + base []LabelSet
43 +}
44 +
45 +func appendMeasureSetSemantics(opts []InstrumentOption, semantics MeasureSetSemantics) []InstrumentOption {
46 + out := make([]InstrumentOption, 0, len(opts)+1)
47 + out = append(out, withMeasureSetSemantics(semantics))
48 + out = append(out, opts...)
49 + return out
50 +}
51 +
52 +// MeasureSetGauge declares or reuses a snapshot MeasureSet with gauge semantics.
53 +func (m *snapshotMeter) MeasureSetGauge(name string, opts ...InstrumentOption) SnapshotMeasureSetGauge {
54 + desc, err := m.backend.registerInstrument(metricName(m.prefix, name), kindMeasureSet, modeSnapshot, appendMeasureSetSemantics(opts, MeasureSetSemanticsGauge)...)
55 + if err != nil {
56 + panic(err)
57 + }
58 + return &snapshotMeasureSetGaugeInstrument{
59 + backend: m.backend,
60 + desc: desc,
61 + base: appendLabelSets(m.sets, nil),
62 + }
63 +}
64 +
65 +// MeasureSetCounter declares or reuses a snapshot MeasureSet with counter semantics.
66 +func (m *snapshotMeter) MeasureSetCounter(name string, opts ...InstrumentOption) SnapshotMeasureSetCounter {
67 + desc, err := m.backend.registerInstrument(metricName(m.prefix, name), kindMeasureSet, modeSnapshot, appendMeasureSetSemantics(opts, MeasureSetSemanticsCounter)...)
68 + if err != nil {
69 + panic(err)
70 + }
71 + return &snapshotMeasureSetCounterInstrument{
72 + backend: m.backend,
73 + desc: desc,
74 + base: appendLabelSets(m.sets, nil),
75 + }
76 +}
77 +
78 +// MeasureSetGauge declares or reuses a stateful MeasureSet with gauge semantics.
79 +func (m *statefulMeter) MeasureSetGauge(name string, opts ...InstrumentOption) StatefulMeasureSetGauge {
80 + desc, err := m.backend.registerInstrument(metricName(m.prefix, name), kindMeasureSet, modeStateful, appendMeasureSetSemantics(opts, MeasureSetSemanticsGauge)...)
81 + if err != nil {
82 + panic(err)
83 + }
84 + return &statefulMeasureSetGaugeInstrument{
85 + backend: m.backend,
86 + desc: desc,
87 + base: appendLabelSets(m.sets, nil),
88 + }
89 +}
90 +
91 +// MeasureSetCounter declares or reuses a stateful MeasureSet with counter semantics.
92 +func (m *statefulMeter) MeasureSetCounter(name string, opts ...InstrumentOption) StatefulMeasureSetCounter {
93 + desc, err := m.backend.registerInstrument(metricName(m.prefix, name), kindMeasureSet, modeStateful, appendMeasureSetSemantics(opts, MeasureSetSemanticsCounter)...)
94 + if err != nil {
95 + panic(err)
96 + }
97 + return &statefulMeasureSetCounterInstrument{
98 + backend: m.backend,
99 + desc: desc,
100 + base: appendLabelSets(m.sets, nil),
101 + }
102 +}
103 +
104 +func (m *snapshotMeasureSetGaugeInstrument) ObservePoint(p MeasureSetPoint, labels ...LabelSet) {
105 + m.backend.recordMeasureSetGaugeObservePoint(m.desc, p, appendLabelSets(m.base, labels))
106 +}
107 +
108 +func (m *snapshotMeasureSetGaugeInstrument) ObserveFields(fields map[string]SampleValue, labels ...LabelSet) {
109 + m.ObservePoint(measureSetPointFromFields(fields, m.desc.measureSet), labels...)
110 +}
111 +
112 +func (m *snapshotMeasureSetCounterInstrument) ObserveTotalPoint(p MeasureSetPoint, labels ...LabelSet) {
113 + m.backend.recordMeasureSetCounterObserveTotalPoint(m.desc, p, appendLabelSets(m.base, labels))
114 +}
115 +
116 +func (m *snapshotMeasureSetCounterInstrument) ObserveTotalFields(fields map[string]SampleValue, labels ...LabelSet) {
117 + m.ObserveTotalPoint(measureSetPointFromFields(fields, m.desc.measureSet), labels...)
118 +}
119 +
120 +func (m *statefulMeasureSetGaugeInstrument) SetPoint(p MeasureSetPoint, labels ...LabelSet) {
121 + m.backend.recordMeasureSetGaugeSetPoint(m.desc, p, appendLabelSets(m.base, labels))
122 +}
123 +
124 +func (m *statefulMeasureSetGaugeInstrument) SetFields(fields map[string]SampleValue, labels ...LabelSet) {
125 + m.SetPoint(measureSetPointFromFields(fields, m.desc.measureSet), labels...)
126 +}
127 +
128 +func (m *statefulMeasureSetGaugeInstrument) SetField(field string, value SampleValue, labels ...LabelSet) {
129 + m.backend.recordMeasureSetGaugeSetField(m.desc, field, value, appendLabelSets(m.base, labels))
130 +}
131 +
132 +func (m *statefulMeasureSetGaugeInstrument) AddPoint(delta MeasureSetPoint, labels ...LabelSet) {
133 + m.backend.recordMeasureSetGaugeAddPoint(m.desc, delta, appendLabelSets(m.base, labels))
134 +}
135 +
136 +func (m *statefulMeasureSetGaugeInstrument) AddFields(delta map[string]SampleValue, labels ...LabelSet) {
137 + m.AddPoint(measureSetPointFromFields(delta, m.desc.measureSet), labels...)
138 +}
139 +
140 +func (m *statefulMeasureSetGaugeInstrument) AddField(field string, delta SampleValue, labels ...LabelSet) {
141 + m.AddPoint(singleMeasureSetPoint(field, delta, m.desc.measureSet), labels...)
142 +}
143 +
144 +func (m *statefulMeasureSetCounterInstrument) AddPoint(delta MeasureSetPoint, labels ...LabelSet) {
145 + m.backend.recordMeasureSetCounterAddPoint(m.desc, delta, appendLabelSets(m.base, labels))
146 +}
147 +
148 +func (m *statefulMeasureSetCounterInstrument) AddFields(delta map[string]SampleValue, labels ...LabelSet) {
149 + m.AddPoint(measureSetPointFromFields(delta, m.desc.measureSet), labels...)
150 +}
151 +
152 +func (m *statefulMeasureSetCounterInstrument) AddField(field string, delta SampleValue, labels ...LabelSet) {
153 + m.AddPoint(singleMeasureSetPoint(field, delta, m.desc.measureSet), labels...)
154 +}
155 +
156 +func normalizeMeasureSetPoint(point MeasureSetPoint, schema *measureSetSchema) []SampleValue {
157 + if schema == nil {
158 + panic(errMeasureSetSchema)
159 + }
160 + if len(point.Values) != len(schema.fields) {
161 + panic(errMeasureSetPoint)
162 + }
163 +
164 + values := make([]SampleValue, len(point.Values))
165 + for i, v := range point.Values {
166 + mustFiniteSample(v)
167 + values[i] = v
168 + }
169 + return values
170 +}
171 +
172 +func measureSetPointFromFields(fields map[string]SampleValue, schema *measureSetSchema) MeasureSetPoint {
173 + return MeasureSetPoint{Values: normalizeMeasureSetFields(fields, schema)}
174 +}
175 +
176 +func normalizeMeasureSetFields(fields map[string]SampleValue, schema *measureSetSchema) []SampleValue {
177 + if schema == nil {
178 + panic(errMeasureSetSchema)
179 + }
180 + if len(fields) != len(schema.fields) {
181 + panic(errMeasureSetFields)
182 + }
183 +
184 + values := make([]SampleValue, len(schema.fields))
185 + for field, value := range fields {
186 + idx, ok := schema.index[field]
187 + if !ok {
188 + panic(errMeasureSetField)
189 + }
190 + mustFiniteSample(value)
191 + values[idx] = value
192 + }
193 + return values
194 +}
195 +
196 +func singleMeasureSetPoint(field string, value SampleValue, schema *measureSetSchema) MeasureSetPoint {
197 + values := make([]SampleValue, len(schema.fields))
198 + idx := mustMeasureSetFieldIndex(field, schema)
199 + mustFiniteSample(value)
200 + values[idx] = value
201 + return MeasureSetPoint{Values: values}
202 +}
203 +
204 +func mustMeasureSetFieldIndex(field string, schema *measureSetSchema) int {
205 + if schema == nil {
206 + panic(errMeasureSetSchema)
207 + }
208 + idx, ok := schema.index[field]
209 + if !ok {
210 + panic(errMeasureSetField)
211 + }
212 + return idx
213 +}
214 +
215 +func normalizeMeasureSetCounterDelta(delta MeasureSetPoint, schema *measureSetSchema) []SampleValue {
216 + values := normalizeMeasureSetPoint(delta, schema)
217 + for _, v := range values {
218 + if v < 0 {
219 + panic(errCounterNegativeDelta)
220 + }
221 + }
222 + return values
223 +}
224 +
225 +func (c *storeCore) recordMeasureSetGaugeObservePoint(desc *instrumentDescriptor, point MeasureSetPoint, sets []LabelSet) {
226 + c.recordMeasureSetGaugeSetPoint(desc, point, sets)
227 +}
228 +
229 +func (c *storeCore) recordMeasureSetGaugeSetPoint(desc *instrumentDescriptor, point MeasureSetPoint, sets []LabelSet) {
230 + schema := desc.measureSet
231 + if schema == nil || schema.semantics != MeasureSetSemanticsGauge {
232 + panic(errMeasureSetSchema)
233 + }
234 +
235 + values := normalizeMeasureSetPoint(point, schema)
236 +
237 + c.mu.Lock()
238 + defer c.mu.Unlock()
239 +
240 + if c.active == nil {
241 + panic(errCycleInactive)
242 + }
243 +
244 + labels, labelsKey, err := labelsFromSet(sets, c)
245 + if err != nil {
246 + panic(err)
247 + }
248 + if labelsContainKey(labels, MeasureSetFieldLabel) {
249 + panic(errMeasureSetLabelKey)
250 + }
251 +
252 + key := makeSeriesKey(desc.name, labelsKey)
253 + entry, ok := c.active.measureSetGauges[key]
254 + if !ok {
255 + entry = &stagedMeasureSet{
256 + key: key,
257 + name: desc.name,
258 + labels: labels,
259 + labelsKey: labelsKey,
260 + desc: desc,
261 + }
262 + c.active.measureSetGauges[key] = entry
263 + }
264 + entry.values = append(entry.values[:0], values...)
265 +}
266 +
267 +func (c *storeCore) recordMeasureSetGaugeAddPoint(desc *instrumentDescriptor, delta MeasureSetPoint, sets []LabelSet) {
268 + schema := desc.measureSet
269 + if schema == nil || schema.semantics != MeasureSetSemanticsGauge {
270 + panic(errMeasureSetSchema)
271 + }
272 +
273 + values := normalizeMeasureSetPoint(delta, schema)
274 +
275 + c.mu.Lock()
276 + defer c.mu.Unlock()
277 +
278 + if c.active == nil {
279 + panic(errCycleInactive)
280 + }
281 +
282 + labels, labelsKey, err := labelsFromSet(sets, c)
283 + if err != nil {
284 + panic(err)
285 + }
286 + if labelsContainKey(labels, MeasureSetFieldLabel) {
287 + panic(errMeasureSetLabelKey)
288 + }
289 +
290 + key := makeSeriesKey(desc.name, labelsKey)
291 + entry, ok := c.active.measureSetGauges[key]
292 + if !ok {
293 + baseline := make([]SampleValue, len(schema.fields))
294 + if existing := c.snapshot.Load().series[key]; existing != nil {
295 + baseline = append(baseline[:0], existing.measureSetValues...)
296 + if len(baseline) != len(schema.fields) {
297 + baseline = make([]SampleValue, len(schema.fields))
298 + }
299 + }
300 + entry = &stagedMeasureSet{
301 + key: key,
302 + name: desc.name,
303 + labels: labels,
304 + labelsKey: labelsKey,
305 + desc: desc,
306 + values: baseline,
307 + }
308 + c.active.measureSetGauges[key] = entry
309 + }
310 + for i, deltaValue := range values {
311 + entry.values[i] += deltaValue
312 + }
313 +}
314 +
315 +func (c *storeCore) recordMeasureSetGaugeSetField(desc *instrumentDescriptor, field string, value SampleValue, sets []LabelSet) {
316 + schema := desc.measureSet
317 + if schema == nil || schema.semantics != MeasureSetSemanticsGauge {
318 + panic(errMeasureSetSchema)
319 + }
320 +
321 + fieldIndex := mustMeasureSetFieldIndex(field, schema)
322 + mustFiniteSample(value)
323 +
324 + c.mu.Lock()
325 + defer c.mu.Unlock()
326 +
327 + if c.active == nil {
328 + panic(errCycleInactive)
329 + }
330 +
331 + labels, labelsKey, err := labelsFromSet(sets, c)
332 + if err != nil {
333 + panic(err)
334 + }
335 + if labelsContainKey(labels, MeasureSetFieldLabel) {
336 + panic(errMeasureSetLabelKey)
337 + }
338 +
339 + key := makeSeriesKey(desc.name, labelsKey)
340 + entry, ok := c.active.measureSetGauges[key]
341 + if !ok {
342 + baseline := make([]SampleValue, len(schema.fields))
343 + if existing := c.snapshot.Load().series[key]; existing != nil {
344 + baseline = append(baseline[:0], existing.measureSetValues...)
345 + if len(baseline) != len(schema.fields) {
346 + baseline = make([]SampleValue, len(schema.fields))
347 + }
348 + }
349 + entry = &stagedMeasureSet{
350 + key: key,
351 + name: desc.name,
352 + labels: labels,
353 + labelsKey: labelsKey,
354 + desc: desc,
355 + values: baseline,
356 + }
357 + c.active.measureSetGauges[key] = entry
358 + } else if len(entry.values) == 0 {
359 + entry.values = make([]SampleValue, len(schema.fields))
360 + }
361 + entry.values[fieldIndex] = value
362 +}
363 +
364 +func (c *storeCore) recordMeasureSetCounterObserveTotalPoint(desc *instrumentDescriptor, point MeasureSetPoint, sets []LabelSet) {
365 + schema := desc.measureSet
366 + if schema == nil || schema.semantics != MeasureSetSemanticsCounter {
367 + panic(errMeasureSetSchema)
368 + }
369 +
370 + values := normalizeMeasureSetPoint(point, schema)
371 +
372 + c.mu.Lock()
373 + defer c.mu.Unlock()
374 +
375 + if c.active == nil {
376 + panic(errCycleInactive)
377 + }
378 +
379 + labels, labelsKey, err := labelsFromSet(sets, c)
380 + if err != nil {
381 + panic(err)
382 + }
383 + if labelsContainKey(labels, MeasureSetFieldLabel) {
384 + panic(errMeasureSetLabelKey)
385 + }
386 +
387 + key := makeSeriesKey(desc.name, labelsKey)
388 + entry, ok := c.active.measureSetCounters[key]
389 + if !ok {
390 + entry = &stagedMeasureSet{
391 + key: key,
392 + name: desc.name,
393 + labels: labels,
394 + labelsKey: labelsKey,
395 + desc: desc,
396 + }
397 + c.active.measureSetCounters[key] = entry
398 + }
399 + entry.values = append(entry.values[:0], values...)
400 +}
401 +
402 +func (c *storeCore) recordMeasureSetCounterAddPoint(desc *instrumentDescriptor, delta MeasureSetPoint, sets []LabelSet) {
403 + schema := desc.measureSet
404 + if schema == nil || schema.semantics != MeasureSetSemanticsCounter {
405 + panic(errMeasureSetSchema)
406 + }
407 +
408 + values := normalizeMeasureSetCounterDelta(delta, schema)
409 +
410 + c.mu.Lock()
411 + defer c.mu.Unlock()
412 +
413 + if c.active == nil {
414 + panic(errCycleInactive)
415 + }
416 +
417 + labels, labelsKey, err := labelsFromSet(sets, c)
418 + if err != nil {
419 + panic(err)
420 + }
421 + if labelsContainKey(labels, MeasureSetFieldLabel) {
422 + panic(errMeasureSetLabelKey)
423 + }
424 +
425 + key := makeSeriesKey(desc.name, labelsKey)
426 + entry, ok := c.active.measureSetCounters[key]
427 + if !ok {
428 + baseline := make([]SampleValue, len(schema.fields))
429 + if existing := c.snapshot.Load().series[key]; existing != nil {
430 + baseline = append(baseline[:0], existing.measureSetValues...)
431 + if len(baseline) != len(schema.fields) {
432 + baseline = make([]SampleValue, len(schema.fields))
433 + }
434 + }
435 + entry = &stagedMeasureSet{
436 + key: key,
437 + name: desc.name,
438 + labels: labels,
439 + labelsKey: labelsKey,
440 + desc: desc,
441 + values: baseline,
442 + }
443 + c.active.measureSetCounters[key] = entry
444 + }
445 + for i, deltaValue := range values {
446 + entry.values[i] += deltaValue
447 + }
448 +}
src/go/pkg/metrix/measureset_store_test.go new
+474
@@ -0,0 +1,474 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package metrix
4 +
5 +import (
6 + "testing"
7 +
8 + "github.com/stretchr/testify/require"
9 +)
10 +
11 +func TestMeasureSetDeclarationValidation(t *testing.T) {
12 + tests := map[string]struct {
13 + run func(t *testing.T)
14 + }{
15 + "snapshot MeasureSetGauge declaration requires WithMeasureSetFields": {
16 + run: func(t *testing.T) {
17 + s := NewCollectorStore()
18 + expectPanic(t, func() {
19 + _ = s.Write().SnapshotMeter("svc").MeasureSetGauge("latency")
20 + })
21 + },
22 + },
23 + "stateful MeasureSetCounter declaration requires WithMeasureSetFields": {
24 + run: func(t *testing.T) {
25 + s := NewCollectorStore()
26 + expectPanic(t, func() {
27 + _ = s.Write().StatefulMeter("svc").MeasureSetCounter("requests")
28 + })
29 + },
30 + },
31 + "MeasureSet declaration rejects duplicate field names": {
32 + run: func(t *testing.T) {
33 + s := NewCollectorStore()
34 + expectPanic(t, func() {
35 + _ = s.Write().SnapshotMeter("svc").MeasureSetGauge("latency",
36 + WithMeasureSetFields(
37 + MeasureFieldSpec{Name: "value"},
38 + MeasureFieldSpec{Name: "value"},
39 + ),
40 + )
41 + })
42 + },
43 + },
44 + "MeasureSet declaration rejects empty field names": {
45 + run: func(t *testing.T) {
46 + s := NewCollectorStore()
47 + expectPanic(t, func() {
48 + _ = s.Write().SnapshotMeter("svc").MeasureSetGauge("latency",
49 + WithMeasureSetFields(MeasureFieldSpec{Name: " "}),
50 + )
51 + })
52 + },
53 + },
54 + "MeasureSet schema mismatch panics on field drift": {
55 + run: func(t *testing.T) {
56 + s := NewCollectorStore()
57 + _ = s.Write().SnapshotMeter("svc").MeasureSetGauge("latency",
58 + WithMeasureSetFields(
59 + MeasureFieldSpec{Name: "value"},
60 + MeasureFieldSpec{Name: "max"},
61 + ),
62 + )
63 + expectPanic(t, func() {
64 + _ = s.Write().SnapshotMeter("svc").MeasureSetGauge("latency",
65 + WithMeasureSetFields(
66 + MeasureFieldSpec{Name: "value"},
67 + MeasureFieldSpec{Name: "min"},
68 + ),
69 + )
70 + })
71 + },
72 + },
73 + "MeasureSet schema mismatch panics on field float drift": {
74 + run: func(t *testing.T) {
75 + s := NewCollectorStore()
76 + _ = s.Write().SnapshotMeter("svc").MeasureSetGauge("latency",
77 + WithMeasureSetFields(MeasureFieldSpec{Name: "value", Float: false}),
78 + )
79 + expectPanic(t, func() {
80 + _ = s.Write().SnapshotMeter("svc").MeasureSetGauge("latency",
81 + WithMeasureSetFields(MeasureFieldSpec{Name: "value", Float: true}),
82 + )
83 + })
84 + },
85 + },
86 + "MeasureSet schema mismatch panics on semantics drift": {
87 + run: func(t *testing.T) {
88 + s := NewCollectorStore()
89 + _ = s.Write().SnapshotMeter("svc").MeasureSetGauge("latency",
90 + WithMeasureSetFields(MeasureFieldSpec{Name: "value"}),
91 + )
92 + expectPanic(t, func() {
93 + _ = s.Write().SnapshotMeter("svc").MeasureSetCounter("latency",
94 + WithMeasureSetFields(MeasureFieldSpec{Name: "value"}),
95 + )
96 + })
97 + },
98 + },
99 + "MeasureSet options are invalid for other instrument kinds": {
100 + run: func(t *testing.T) {
101 + s := NewCollectorStore()
102 + expectPanic(t, func() {
103 + _ = s.Write().SnapshotMeter("svc").Gauge("latency",
104 + WithMeasureSetFields(MeasureFieldSpec{Name: "value"}),
105 + )
106 + })
107 + },
108 + },
109 + }
110 +
111 + for name, tc := range tests {
112 + t.Run(name, tc.run)
113 + }
114 +}
115 +
116 +func TestMeasureSetStoreScenarios(t *testing.T) {
117 + tests := map[string]struct {
118 + run func(t *testing.T)
119 + }{
120 + "snapshot MeasureSet gauge read and flatten metadata": {
121 + run: func(t *testing.T) {
122 + s := NewCollectorStore()
123 + cc := cycleController(t, s)
124 + ms := s.Write().SnapshotMeter("svc").MeasureSetGauge(
125 + "latency",
126 + WithMeasureSetFields(
127 + MeasureFieldSpec{Name: "value"},
128 + MeasureFieldSpec{Name: "ratio", Float: true},
129 + ),
130 + WithDescription("Latency"),
131 + WithChartFamily("Service"),
132 + WithUnit("seconds"),
133 + )
134 +
135 + cc.BeginCycle()
136 + ms.ObserveFields(map[string]SampleValue{
137 + "value": 1.5,
138 + "ratio": 0.5,
139 + })
140 + cc.CommitCycleSuccess()
141 +
142 + mustMeasureSet(t, s.Read(), "svc.latency", nil, []SampleValue{1.5, 0.5})
143 +
144 + rawMeta, ok := s.Read().SeriesMeta("svc.latency", nil)
145 + require.True(t, ok)
146 + require.Equal(t, MetricKindMeasureSet, rawMeta.Kind)
147 + require.Equal(t, MetricKindMeasureSet, rawMeta.SourceKind)
148 + require.Equal(t, FlattenRoleNone, rawMeta.FlattenRole)
149 +
150 + flat := s.Read(ReadFlatten())
151 + mustValue(t, flat, "svc.latency_value", measureSetFieldLabels("value"), 1.5)
152 + mustValue(t, flat, "svc.latency_ratio", measureSetFieldLabels("ratio"), 0.5)
153 + _, ok = flat.Value("svc.latency_value", nil)
154 + require.False(t, ok, "expected flattened MeasureSet scalar lookup without synthetic field label to miss")
155 + _, ok = flat.MeasureSet("svc.latency", nil)
156 + require.False(t, ok, "expected flattened view to hide typed MeasureSet getter")
157 +
158 + flatMeta, ok := flat.SeriesMeta("svc.latency_ratio", measureSetFieldLabels("ratio"))
159 + require.True(t, ok)
160 + require.Equal(t, MetricKindGauge, flatMeta.Kind)
161 + require.Equal(t, MetricKindMeasureSet, flatMeta.SourceKind)
162 + require.Equal(t, FlattenRoleMeasureSetField, flatMeta.FlattenRole)
163 +
164 + meta, ok := flat.MetricMeta("svc.latency_ratio")
165 + require.True(t, ok)
166 + require.Equal(t, "Latency", meta.Description)
167 + require.Equal(t, "Service", meta.ChartFamily)
168 + require.Equal(t, "seconds", meta.Unit)
169 + require.True(t, meta.Float)
170 + },
171 + },
172 + "stateful MeasureSet gauge add baselines from committed and remains visible": {
173 + run: func(t *testing.T) {
174 + s := NewCollectorStore()
175 + cc := cycleController(t, s)
176 + ms := s.Write().StatefulMeter("svc").MeasureSetGauge(
177 + "usage",
178 + WithMeasureSetFields(
179 + MeasureFieldSpec{Name: "value"},
180 + MeasureFieldSpec{Name: "limit"},
181 + ),
182 + )
183 +
184 + cc.BeginCycle()
185 + ms.SetPoint(MeasureSetPoint{Values: []SampleValue{10, 20}})
186 + cc.CommitCycleSuccess()
187 +
188 + cc.BeginCycle()
189 + ms.AddPoint(MeasureSetPoint{Values: []SampleValue{2, 3}})
190 + ms.AddPoint(MeasureSetPoint{Values: []SampleValue{1, 0}})
191 + cc.CommitCycleSuccess()
192 + mustMeasureSet(t, s.Read(), "svc.usage", nil, []SampleValue{13, 23})
193 +
194 + cc.BeginCycle()
195 + cc.CommitCycleSuccess()
196 + mustMeasureSet(t, s.Read(), "svc.usage", nil, []SampleValue{13, 23})
197 + mustValue(t, s.Read(ReadFlatten()), "svc.usage_value", measureSetFieldLabels("value"), 13)
198 + mustValue(t, s.Read(ReadFlatten()), "svc.usage_limit", measureSetFieldLabels("limit"), 23)
199 + },
200 + },
201 + "snapshot MeasureSet counter flatten delta and reset-aware semantics": {
202 + run: func(t *testing.T) {
203 + s := NewCollectorStore()
204 + cc := cycleController(t, s)
205 + ms := s.Write().SnapshotMeter("svc").MeasureSetCounter(
206 + "requests",
207 + WithMeasureSetFields(
208 + MeasureFieldSpec{Name: "ok"},
209 + MeasureFieldSpec{Name: "failed"},
210 + ),
211 + )
212 +
213 + cc.BeginCycle()
214 + ms.ObserveTotalFields(map[string]SampleValue{
215 + "ok": 100,
216 + "failed": 40,
217 + })
218 + cc.CommitCycleSuccess()
219 + mustMeasureSet(t, s.Read(), "svc.requests", nil, []SampleValue{100, 40})
220 + mustNoDelta(t, s.Read(ReadFlatten()), "svc.requests_ok", measureSetFieldLabels("ok"))
221 +
222 + cc.BeginCycle()
223 + ms.ObserveTotalFields(map[string]SampleValue{
224 + "ok": 150,
225 + "failed": 50,
226 + })
227 + cc.CommitCycleSuccess()
228 + mustDelta(t, s.Read(ReadFlatten()), "svc.requests_ok", measureSetFieldLabels("ok"), 50)
229 + mustDelta(t, s.Read(ReadFlatten()), "svc.requests_failed", measureSetFieldLabels("failed"), 10)
230 +
231 + cc.BeginCycle()
232 + ms.ObserveTotalFields(map[string]SampleValue{
233 + "ok": 20,
234 + "failed": 5,
235 + })
236 + cc.CommitCycleSuccess()
237 + mustDelta(t, s.Read(ReadFlatten()), "svc.requests_ok", measureSetFieldLabels("ok"), 20)
238 + mustDelta(t, s.Read(ReadFlatten()), "svc.requests_failed", measureSetFieldLabels("failed"), 5)
239 + },
240 + },
241 + "snapshot MeasureSet counter delta unavailable on attempt gap": {
242 + run: func(t *testing.T) {
243 + s := NewCollectorStore()
244 + cc := cycleController(t, s)
245 + ms := s.Write().SnapshotMeter("svc").MeasureSetCounter(
246 + "jobs",
247 + WithMeasureSetFields(MeasureFieldSpec{Name: "done"}),
248 + )
249 +
250 + cc.BeginCycle()
251 + ms.ObserveTotalFields(map[string]SampleValue{"done": 10})
252 + cc.CommitCycleSuccess()
253 +
254 + cc.BeginCycle()
255 + ms.ObserveTotalFields(map[string]SampleValue{"done": 20})
256 + cc.CommitCycleSuccess()
257 + mustDelta(t, s.Read(ReadFlatten()), "svc.jobs_done", measureSetFieldLabels("done"), 10)
258 +
259 + cc.BeginCycle()
260 + ms.ObserveTotalFields(map[string]SampleValue{"done": 30})
261 + cc.AbortCycle()
262 +
263 + cc.BeginCycle()
264 + ms.ObserveTotalFields(map[string]SampleValue{"done": 40})
265 + cc.CommitCycleSuccess()
266 + mustNoDelta(t, s.Read(ReadFlatten()), "svc.jobs_done", measureSetFieldLabels("done"))
267 + },
268 + },
269 + "MeasureSet flatten label key collision panics": {
270 + run: func(t *testing.T) {
271 + s := NewCollectorStore()
272 + cc := cycleController(t, s)
273 + ms := s.Write().SnapshotMeter("svc").
274 + WithLabels(Label{Key: MeasureSetFieldLabel, Value: "already-present"}).
275 + MeasureSetGauge(
276 + "latency",
277 + WithMeasureSetFields(MeasureFieldSpec{Name: "value"}),
278 + )
279 +
280 + cc.BeginCycle()
281 + expectPanic(t, func() {
282 + ms.ObservePoint(MeasureSetPoint{Values: []SampleValue{1}})
283 + })
284 + cc.AbortCycle()
285 + },
286 + },
287 + "stateful MeasureSet counter add accumulates and flattened delta works": {
288 + run: func(t *testing.T) {
289 + s := NewCollectorStore()
290 + cc := cycleController(t, s)
291 + ms := s.Write().StatefulMeter("svc").MeasureSetCounter(
292 + "events",
293 + WithMeasureSetFields(
294 + MeasureFieldSpec{Name: "ok"},
295 + MeasureFieldSpec{Name: "failed"},
296 + ),
297 + )
298 +
299 + cc.BeginCycle()
300 + ms.AddPoint(MeasureSetPoint{Values: []SampleValue{5, 1}})
301 + cc.CommitCycleSuccess()
302 + mustNoDelta(t, s.Read(ReadFlatten()), "svc.events_ok", measureSetFieldLabels("ok"))
303 +
304 + cc.BeginCycle()
305 + ms.AddPoint(MeasureSetPoint{Values: []SampleValue{2, 3}})
306 + ms.AddPoint(MeasureSetPoint{Values: []SampleValue{1, 0}})
307 + cc.CommitCycleSuccess()
308 + mustMeasureSet(t, s.Read(), "svc.events", nil, []SampleValue{8, 4})
309 + mustDelta(t, s.Read(ReadFlatten()), "svc.events_ok", measureSetFieldLabels("ok"), 3)
310 + mustDelta(t, s.Read(ReadFlatten()), "svc.events_failed", measureSetFieldLabels("failed"), 3)
311 + },
312 + },
313 + "stateful MeasureSet counter negative add panics": {
314 + run: func(t *testing.T) {
315 + s := NewCollectorStore()
316 + cc := cycleController(t, s)
317 + ms := s.Write().StatefulMeter("svc").MeasureSetCounter(
318 + "events",
319 + WithMeasureSetFields(
320 + MeasureFieldSpec{Name: "ok"},
321 + MeasureFieldSpec{Name: "failed"},
322 + ),
323 + )
324 +
325 + cc.BeginCycle()
326 + expectPanic(t, func() {
327 + ms.AddPoint(MeasureSetPoint{Values: []SampleValue{1, -1}})
328 + })
329 + cc.AbortCycle()
330 + },
331 + },
332 + "snapshot MeasureSet named full writes require exact field set": {
333 + run: func(t *testing.T) {
334 + s := NewCollectorStore()
335 + cc := cycleController(t, s)
336 + ms := s.Write().SnapshotMeter("svc").MeasureSetGauge(
337 + "latency",
338 + WithMeasureSetFields(
339 + MeasureFieldSpec{Name: "value"},
340 + MeasureFieldSpec{Name: "max"},
341 + ),
342 + )
343 +
344 + cc.BeginCycle()
345 + expectPanic(t, func() {
346 + ms.ObserveFields(map[string]SampleValue{"value": 1})
347 + })
348 + cc.AbortCycle()
349 +
350 + cc.BeginCycle()
351 + expectPanic(t, func() {
352 + ms.ObserveFields(map[string]SampleValue{
353 + "value": 1,
354 + "max": 2,
355 + "min": 0,
356 + })
357 + })
358 + cc.AbortCycle()
359 + },
360 + },
361 + "stateful MeasureSet gauge named writes support full and singular updates": {
362 + run: func(t *testing.T) {
363 + s := NewCollectorStore()
364 + cc := cycleController(t, s)
365 + ms := s.Write().StatefulMeter("svc").MeasureSetGauge(
366 + "usage",
367 + WithMeasureSetFields(
368 + MeasureFieldSpec{Name: "value"},
369 + MeasureFieldSpec{Name: "limit"},
370 + ),
371 + )
372 +
373 + cc.BeginCycle()
374 + ms.SetFields(map[string]SampleValue{
375 + "value": 10,
376 + "limit": 20,
377 + })
378 + cc.CommitCycleSuccess()
379 +
380 + cc.BeginCycle()
381 + ms.SetField("value", 15)
382 + ms.AddField("limit", 3)
383 + cc.CommitCycleSuccess()
384 +
385 + mustMeasureSet(t, s.Read(), "svc.usage", nil, []SampleValue{15, 23})
386 + mustValue(t, s.Read(ReadFlatten()), "svc.usage_value", measureSetFieldLabels("value"), 15)
387 + mustValue(t, s.Read(ReadFlatten()), "svc.usage_limit", measureSetFieldLabels("limit"), 23)
388 + },
389 + },
390 + "stateful MeasureSet counter named writes support full and singular deltas": {
391 + run: func(t *testing.T) {
392 + s := NewCollectorStore()
393 + cc := cycleController(t, s)
394 + ms := s.Write().StatefulMeter("svc").MeasureSetCounter(
395 + "events",
396 + WithMeasureSetFields(
397 + MeasureFieldSpec{Name: "ok"},
398 + MeasureFieldSpec{Name: "failed"},
399 + ),
400 + )
401 +
402 + cc.BeginCycle()
403 + ms.AddFields(map[string]SampleValue{
404 + "ok": 5,
405 + "failed": 1,
406 + })
407 + cc.CommitCycleSuccess()
408 + mustNoDelta(t, s.Read(ReadFlatten()), "svc.events_ok", measureSetFieldLabels("ok"))
409 +
410 + cc.BeginCycle()
411 + ms.AddField("ok", 2)
412 + ms.AddField("failed", 3)
413 + cc.CommitCycleSuccess()
414 +
415 + mustMeasureSet(t, s.Read(), "svc.events", nil, []SampleValue{7, 4})
416 + mustDelta(t, s.Read(ReadFlatten()), "svc.events_ok", measureSetFieldLabels("ok"), 2)
417 + mustDelta(t, s.Read(ReadFlatten()), "svc.events_failed", measureSetFieldLabels("failed"), 3)
418 + },
419 + },
420 + "MeasureSet direct read returns a copy": {
421 + run: func(t *testing.T) {
422 + s := NewCollectorStore()
423 + cc := cycleController(t, s)
424 + ms := s.Write().SnapshotMeter("svc").MeasureSetGauge(
425 + "latency",
426 + WithMeasureSetFields(MeasureFieldSpec{Name: "value"}),
427 + )
428 +
429 + cc.BeginCycle()
430 + ms.ObservePoint(MeasureSetPoint{Values: []SampleValue{7}})
431 + cc.CommitCycleSuccess()
432 +
433 + p, ok := s.Read().MeasureSet("svc.latency", nil)
434 + require.True(t, ok)
435 + p.Values[0] = 99
436 +
437 + mustMeasureSet(t, s.Read(), "svc.latency", nil, []SampleValue{7})
438 + },
439 + },
440 + "MeasureSet point length mismatch panics": {
441 + run: func(t *testing.T) {
442 + s := NewCollectorStore()
443 + cc := cycleController(t, s)
444 + ms := s.Write().SnapshotMeter("svc").MeasureSetGauge(
445 + "latency",
446 + WithMeasureSetFields(
447 + MeasureFieldSpec{Name: "value"},
448 + MeasureFieldSpec{Name: "max"},
449 + ),
450 + )
451 +
452 + cc.BeginCycle()
453 + expectPanic(t, func() {
454 + ms.ObservePoint(MeasureSetPoint{Values: []SampleValue{1}})
455 + })
456 + cc.AbortCycle()
457 + },
458 + },
459 + }
460 +
461 + for name, tc := range tests {
462 + t.Run(name, tc.run)
463 + }
464 +}
465 +
466 +func mustMeasureSet(t *testing.T, r Reader, name string, labels Labels, want []SampleValue) {
467 + t.Helper()
468 + got, ok := r.MeasureSet(name, labels)
469 + require.True(t, ok, "expected measureset for %s", name)
470 + require.Len(t, got.Values, len(want), "unexpected measureset size for %s", name)
471 + for i, w := range want {
472 + require.Equal(t, w, got.Values[i], "unexpected measureset value %d for %s", i, name)
473 + }
474 +}
src/go/pkg/metrix/meta.go
+2
@@ -14,6 +14,8 @@ func metricKindPublic(kind metricKind) MetricKind {
14 return MetricKindSummary
15 case kindStateSet:
16 return MetricKindStateSet
17 + case kindMeasureSet:
18 + return MetricKindMeasureSet
19 default:
20 return MetricKindUnknown
21 }
src/go/pkg/metrix/meter.go
+16
@@ -108,6 +108,14 @@ func (m *snapshotVecMeter) StateSet(name string, opts ...InstrumentOption) Snaps
108 return m.meter.StateSetVec(name, m.labelKeys, opts...)
109 }
110
111 +func (m *snapshotVecMeter) MeasureSetGauge(name string, opts ...InstrumentOption) SnapshotMeasureSetGaugeVec {
112 + return m.meter.MeasureSetGaugeVec(name, m.labelKeys, opts...)
113 +}
114 +
115 +func (m *snapshotVecMeter) MeasureSetCounter(name string, opts ...InstrumentOption) SnapshotMeasureSetCounterVec {
116 + return m.meter.MeasureSetCounterVec(name, m.labelKeys, opts...)
117 +}
118 +
119 func (m *statefulVecMeter) Gauge(name string, opts ...InstrumentOption) StatefulGaugeVec {
120 return m.meter.GaugeVec(name, m.labelKeys, opts...)
121 }
@@ -128,6 +136,14 @@ func (m *statefulVecMeter) StateSet(name string, opts ...InstrumentOption) State
136 return m.meter.StateSetVec(name, m.labelKeys, opts...)
137 }
138
139 +func (m *statefulVecMeter) MeasureSetGauge(name string, opts ...InstrumentOption) StatefulMeasureSetGaugeVec {
140 + return m.meter.MeasureSetGaugeVec(name, m.labelKeys, opts...)
141 +}
142 +
143 +func (m *statefulVecMeter) MeasureSetCounter(name string, opts ...InstrumentOption) StatefulMeasureSetCounterVec {
144 + return m.meter.MeasureSetCounterVec(name, m.labelKeys, opts...)
145 +}
146 +
147 // metricName composes meter prefix with instrument local name.
148 func metricName(prefix, name string) string {
149 if prefix == "" {
src/go/pkg/metrix/options.go
+15
@@ -23,6 +23,8 @@ type instrumentConfig struct {
23 summaryReservoir int
24 states []string
25 stateSetMode *StateSetMode
26 + measureSetFields []MeasureFieldSpec
27 + measureSetSemantics *MeasureSetSemantics
28
29 descriptionSet bool
30 description string
@@ -82,6 +84,19 @@ func WithStateSetMode(mode StateSetMode) InstrumentOption {
84 })
85 }
86
87 +func WithMeasureSetFields(fields ...MeasureFieldSpec) InstrumentOption {
88 + return optionFunc(func(cfg *instrumentConfig) {
89 + cfg.measureSetFields = append([]MeasureFieldSpec(nil), fields...)
90 + })
91 +}
92 +
93 +func withMeasureSetSemantics(semantics MeasureSetSemantics) InstrumentOption {
94 + return optionFunc(func(cfg *instrumentConfig) {
95 + s := semantics
96 + cfg.measureSetSemantics = &s
97 + })
98 +}
99 +
100 // WithDescription sets optional metric-family description metadata.
101 func WithDescription(description string) InstrumentOption {
102 return optionFunc(func(cfg *instrumentConfig) {
src/go/pkg/metrix/reader.go
+89 -3
@@ -135,6 +135,24 @@ func (r *storeReader) StateSet(name string, labels Labels) (StateSetPoint, bool)
135 return StateSetPoint{States: cloneStateMap(s.stateSetValues)}, true
136 }
137
138 +func (r *storeReader) MeasureSet(name string, labels Labels) (MeasureSetPoint, bool) {
139 + if r.flattened {
140 + return MeasureSetPoint{}, false
141 + }
142 +
143 + s, ok := r.lookup(name, labels)
144 + if !ok || !r.visible(s) {
145 + return MeasureSetPoint{}, false
146 + }
147 + if s.desc == nil || s.desc.kind != kindMeasureSet || s.desc.measureSet == nil {
148 + return MeasureSetPoint{}, false
149 + }
150 + if len(s.measureSetValues) != len(s.desc.measureSet.fields) {
151 + return MeasureSetPoint{}, false
152 + }
153 + return MeasureSetPoint{Values: append([]SampleValue(nil), s.measureSetValues...)}, true
154 +}
155 +
156 func (r *storeReader) SeriesMeta(name string, labels Labels) (SeriesMeta, bool) {
157 s, ok := r.lookup(name, labels)
158 if !ok || !r.visible(s) {
@@ -190,6 +208,8 @@ func flattenSnapshot(src *readSnapshot) *readSnapshot {
208 appendFlattenedSummarySeries(dst, s)
209 case kindStateSet:
210 appendFlattenedStateSetSeries(dst, s)
211 + case kindMeasureSet:
212 + appendFlattenedMeasureSetSeries(dst, s)
213 }
214 }
215
@@ -213,7 +233,7 @@ func appendFlattenedHistogramSeries(dst *readSnapshot, src *committedSeries) {
233 for _, lbl := range src.labels {
234 labelsMap[lbl.Key] = lbl.Value
235 }
216 - labelsMap[histogramBucketLabel] = formatHistogramBucketLabel(ub)
236 + labelsMap[HistogramBucketLabel] = formatHistogramBucketLabel(ub)
237
238 labels, labelsKey, err := canonicalizeLabels(labelsMap)
239 if err != nil {
@@ -251,7 +271,7 @@ func appendFlattenedHistogramSeries(dst *readSnapshot, src *committedSeries) {
271 for _, lbl := range src.labels {
272 infMap[lbl.Key] = lbl.Value
273 }
254 - infMap[histogramBucketLabel] = formatHistogramBucketLabel(math.Inf(1))
274 + infMap[HistogramBucketLabel] = formatHistogramBucketLabel(math.Inf(1))
275 infLabels, infLabelsKey, err := canonicalizeLabels(infMap)
276 if err == nil {
277 infName := src.name + "_bucket"
@@ -360,7 +380,7 @@ func appendFlattenedSummarySeries(dst *readSnapshot, src *committedSeries) {
380 for _, lbl := range src.labels {
381 labelsMap[lbl.Key] = lbl.Value
382 }
363 - labelsMap[summaryQuantileLabel] = formatSummaryQuantileLabel(q)
383 + labelsMap[SummaryQuantileLabel] = formatSummaryQuantileLabel(q)
384
385 labels, labelsKey, err := canonicalizeLabels(labelsMap)
386 if err != nil {
@@ -443,6 +463,72 @@ func appendFlattenedStateSetSeries(dst *readSnapshot, src *committedSeries) {
463 }
464 }
465
466 +func appendFlattenedMeasureSetSeries(dst *readSnapshot, src *committedSeries) {
467 + schema := src.desc.measureSet
468 + if schema == nil || len(src.measureSetValues) != len(schema.fields) {
469 + return
470 + }
471 +
472 + kind := MetricKindGauge
473 + descKind := kindGauge
474 + if schema.semantics == MeasureSetSemanticsCounter {
475 + kind = MetricKindCounter
476 + descKind = kindCounter
477 + }
478 +
479 + for i, field := range schema.fields {
480 + labelsMap := make(map[string]string, len(src.labels)+1)
481 + for _, lbl := range src.labels {
482 + labelsMap[lbl.Key] = lbl.Value
483 + }
484 + labelsMap[MeasureSetFieldLabel] = field.Name
485 +
486 + labels, labelsKey, err := canonicalizeLabels(labelsMap)
487 + if err != nil {
488 + continue
489 + }
490 +
491 + name := src.name + "_" + field.Name
492 + key := makeSeriesKey(name, labelsKey)
493 + meta := src.desc.meta
494 + meta.Float = field.Float
495 +
496 + series := &committedSeries{
497 + id: SeriesID(key),
498 + hash64: seriesIDHash(SeriesID(key)),
499 + key: key,
500 + name: name,
501 + labels: labels,
502 + labelsKey: labelsKey,
503 + desc: &instrumentDescriptor{
504 + name: name,
505 + kind: descKind,
506 + mode: src.desc.mode,
507 + freshness: src.desc.freshness,
508 + window: src.desc.window,
509 + meta: meta,
510 + },
511 + value: src.measureSetValues[i],
512 + meta: flattenedSeriesMeta(
513 + src.meta,
514 + kind,
515 + MetricKindMeasureSet,
516 + FlattenRoleMeasureSetField,
517 + ),
518 + }
519 + if schema.semantics == MeasureSetSemanticsCounter {
520 + series.counterCurrent = src.measureSetValues[i]
521 + series.counterCurrentSeq = src.measureSetCurrentSeq
522 + if src.measureSetHasPrev && len(src.measureSetPreviousValues) == len(schema.fields) {
523 + series.counterHasPrev = true
524 + series.counterPrevious = src.measureSetPreviousValues[i]
525 + series.counterPreviousSeq = src.measureSetPreviousSeq
526 + }
527 + }
528 + dst.series[key] = series
529 + }
530 +}
531 +
532 func (r *storeReader) Family(name string) (FamilyView, bool) {
533 index := r.byNameIndex()
534 if len(index[name]) == 0 {
src/go/pkg/metrix/runtime_store.go
+130 -2
@@ -205,7 +205,7 @@ func (r *runtimeStoreBackend) recordHistogramObserve(desc *instrumentDescriptor,
205 if err != nil {
206 panic(err)
207 }
208 - if labelsContainKey(labels, histogramBucketLabel) {
208 + if labelsContainKey(labels, HistogramBucketLabel) {
209 panic(errHistogramLabelKey)
210 }
211
@@ -244,7 +244,7 @@ func (r *runtimeStoreBackend) recordSummaryObserve(desc *instrumentDescriptor, v
244 if err != nil {
245 panic(err)
246 }
247 - if labelsContainKey(labels, summaryQuantileLabel) {
247 + if labelsContainKey(labels, SummaryQuantileLabel) {
248 panic(errSummaryLabelKey)
249 }
250
@@ -297,6 +297,134 @@ func (r *runtimeStoreBackend) recordStateSetObserve(desc *instrumentDescriptor,
297 })
298 }
299
300 +func (r *runtimeStoreBackend) recordMeasureSetGaugeObservePoint(_ *instrumentDescriptor, _ MeasureSetPoint, _ []LabelSet) {
301 + panic(errRuntimeSnapshotWrite)
302 +}
303 +
304 +func (r *runtimeStoreBackend) recordMeasureSetGaugeSetPoint(desc *instrumentDescriptor, point MeasureSetPoint, sets []LabelSet) {
305 + schema := desc.measureSet
306 + if schema == nil || schema.semantics != MeasureSetSemanticsGauge {
307 + panic(errMeasureSetSchema)
308 + }
309 +
310 + values := normalizeMeasureSetPoint(point, schema)
311 +
312 + labels, labelsKey, err := labelsFromSet(sets, r)
313 + if err != nil {
314 + panic(err)
315 + }
316 + if labelsContainKey(labels, MeasureSetFieldLabel) {
317 + panic(errMeasureSetLabelKey)
318 + }
319 + key := makeSeriesKey(desc.name, labelsKey)
320 + r.commitRuntimeWrite(func(old, next *readSnapshot, seq uint64, nowUnixNano int64) {
321 + series := runtimeEnsureSeriesMutable(old, next, key, desc.name, labels, labelsKey, desc)
322 + series.measureSetValues = append(series.measureSetValues[:0], values...)
323 + series.meta.LastSeenSuccessSeq = seq
324 + series.runtimeLastSeenUnixNano = nowUnixNano
325 + })
326 +}
327 +
328 +func (r *runtimeStoreBackend) recordMeasureSetGaugeAddPoint(desc *instrumentDescriptor, delta MeasureSetPoint, sets []LabelSet) {
329 + schema := desc.measureSet
330 + if schema == nil || schema.semantics != MeasureSetSemanticsGauge {
331 + panic(errMeasureSetSchema)
332 + }
333 +
334 + values := normalizeMeasureSetPoint(delta, schema)
335 +
336 + labels, labelsKey, err := labelsFromSet(sets, r)
337 + if err != nil {
338 + panic(err)
339 + }
340 + if labelsContainKey(labels, MeasureSetFieldLabel) {
341 + panic(errMeasureSetLabelKey)
342 + }
343 + key := makeSeriesKey(desc.name, labelsKey)
344 + r.commitRuntimeWrite(func(old, next *readSnapshot, seq uint64, nowUnixNano int64) {
345 + series := runtimeEnsureSeriesMutable(old, next, key, desc.name, labels, labelsKey, desc)
346 + if len(series.measureSetValues) == 0 {
347 + series.measureSetValues = make([]SampleValue, len(schema.fields))
348 + }
349 + for i, deltaValue := range values {
350 + series.measureSetValues[i] += deltaValue
351 + }
352 + series.meta.LastSeenSuccessSeq = seq
353 + series.runtimeLastSeenUnixNano = nowUnixNano
354 + })
355 +}
356 +
357 +func (r *runtimeStoreBackend) recordMeasureSetGaugeSetField(desc *instrumentDescriptor, field string, value SampleValue, sets []LabelSet) {
358 + schema := desc.measureSet
359 + if schema == nil || schema.semantics != MeasureSetSemanticsGauge {
360 + panic(errMeasureSetSchema)
361 + }
362 +
363 + fieldIndex := mustMeasureSetFieldIndex(field, schema)
364 + mustFiniteSample(value)
365 +
366 + labels, labelsKey, err := labelsFromSet(sets, r)
367 + if err != nil {
368 + panic(err)
369 + }
370 + if labelsContainKey(labels, MeasureSetFieldLabel) {
371 + panic(errMeasureSetLabelKey)
372 + }
373 + key := makeSeriesKey(desc.name, labelsKey)
374 + r.commitRuntimeWrite(func(old, next *readSnapshot, seq uint64, nowUnixNano int64) {
375 + series := runtimeEnsureSeriesMutable(old, next, key, desc.name, labels, labelsKey, desc)
376 + if len(series.measureSetValues) == 0 {
377 + series.measureSetValues = make([]SampleValue, len(schema.fields))
378 + }
379 + series.measureSetValues[fieldIndex] = value
380 + series.meta.LastSeenSuccessSeq = seq
381 + series.runtimeLastSeenUnixNano = nowUnixNano
382 + })
383 +}
384 +
385 +func (r *runtimeStoreBackend) recordMeasureSetCounterObserveTotalPoint(_ *instrumentDescriptor, _ MeasureSetPoint, _ []LabelSet) {
386 + panic(errRuntimeSnapshotWrite)
387 +}
388 +
389 +func (r *runtimeStoreBackend) recordMeasureSetCounterAddPoint(desc *instrumentDescriptor, delta MeasureSetPoint, sets []LabelSet) {
390 + schema := desc.measureSet
391 + if schema == nil || schema.semantics != MeasureSetSemanticsCounter {
392 + panic(errMeasureSetSchema)
393 + }
394 +
395 + values := normalizeMeasureSetCounterDelta(delta, schema)
396 +
397 + labels, labelsKey, err := labelsFromSet(sets, r)
398 + if err != nil {
399 + panic(err)
400 + }
401 + if labelsContainKey(labels, MeasureSetFieldLabel) {
402 + panic(errMeasureSetLabelKey)
403 + }
404 + key := makeSeriesKey(desc.name, labelsKey)
405 + r.commitRuntimeWrite(func(old, next *readSnapshot, seq uint64, nowUnixNano int64) {
406 + series := runtimeEnsureSeriesMutable(old, next, key, desc.name, labels, labelsKey, desc)
407 + if len(series.measureSetValues) == 0 {
408 + series.measureSetValues = make([]SampleValue, len(schema.fields))
409 + }
410 + if series.measureSetCurrentSeq > 0 {
411 + series.measureSetPreviousValues = append(series.measureSetPreviousValues[:0], series.measureSetValues...)
412 + series.measureSetPreviousSeq = series.measureSetCurrentSeq
413 + series.measureSetHasPrev = true
414 + } else {
415 + series.measureSetPreviousValues = nil
416 + series.measureSetPreviousSeq = 0
417 + series.measureSetHasPrev = false
418 + }
419 + for i, deltaValue := range values {
420 + series.measureSetValues[i] += deltaValue
421 + }
422 + series.measureSetCurrentSeq++
423 + series.meta.LastSeenSuccessSeq = seq
424 + series.runtimeLastSeenUnixNano = nowUnixNano
425 + })
426 +}
427 +
428 func (r *runtimeStoreBackend) commitRuntimeWrite(apply func(old, next *readSnapshot, seq uint64, nowUnixNano int64)) {
429 r.core.mu.Lock()
430 defer r.core.mu.Unlock()
src/go/pkg/metrix/runtime_store_test.go
+58
@@ -109,6 +109,64 @@ func TestRuntimeStoreScenarios(t *testing.T) {
109 mustValue(t, fr, "runtime.mode", Labels{"runtime.mode": "operational"}, 1)
110 },
111 },
112 + "runtime MeasureSet gauge and counter are readable and flattenable": {
113 + run: func(t *testing.T) {
114 + s := NewRuntimeStore()
115 + m := s.Write().StatefulMeter("runtime")
116 + g := m.MeasureSetGauge(
117 + "usage",
118 + WithMeasureSetFields(
119 + MeasureFieldSpec{Name: "value"},
120 + MeasureFieldSpec{Name: "limit"},
121 + ),
122 + )
123 + c := m.MeasureSetCounter(
124 + "events",
125 + WithMeasureSetFields(
126 + MeasureFieldSpec{Name: "ok"},
127 + MeasureFieldSpec{Name: "failed"},
128 + ),
129 + )
130 +
131 + g.SetFields(map[string]SampleValue{
132 + "value": 10,
133 + "limit": 20,
134 + })
135 + g.SetField("value", 11)
136 + g.AddField("limit", 2)
137 + mustMeasureSet(t, s.Read(), "runtime.usage", nil, []SampleValue{11, 22})
138 + mustValue(t, s.Read(ReadFlatten()), "runtime.usage_value", measureSetFieldLabels("value"), 11)
139 + mustValue(t, s.Read(ReadFlatten()), "runtime.usage_limit", measureSetFieldLabels("limit"), 22)
140 +
141 + c.AddFields(map[string]SampleValue{
142 + "ok": 5,
143 + "failed": 1,
144 + })
145 + mustNoDelta(t, s.Read(ReadFlatten()), "runtime.events_ok", measureSetFieldLabels("ok"))
146 + c.AddField("ok", 2)
147 + mustDelta(t, s.Read(ReadFlatten()), "runtime.events_ok", measureSetFieldLabels("ok"), 2)
148 + mustDelta(t, s.Read(ReadFlatten()), "runtime.events_failed", measureSetFieldLabels("failed"), 0)
149 + c.AddField("failed", 3)
150 + mustMeasureSet(t, s.Read(), "runtime.events", nil, []SampleValue{7, 4})
151 + mustDelta(t, s.Read(ReadFlatten()), "runtime.events_ok", measureSetFieldLabels("ok"), 0)
152 + mustDelta(t, s.Read(ReadFlatten()), "runtime.events_failed", measureSetFieldLabels("failed"), 3)
153 + },
154 + },
155 + "runtime MeasureSet flatten label key collision panics": {
156 + run: func(t *testing.T) {
157 + s := NewRuntimeStore()
158 + ms := s.Write().StatefulMeter("runtime").
159 + WithLabels(Label{Key: MeasureSetFieldLabel, Value: "already-present"}).
160 + MeasureSetGauge(
161 + "usage",
162 + WithMeasureSetFields(MeasureFieldSpec{Name: "value"}),
163 + )
164 +
165 + expectPanic(t, func() {
166 + ms.SetPoint(MeasureSetPoint{Values: []SampleValue{1}})
167 + })
168 + },
169 + },
170 "runtime counter is thread-safe for concurrent writers": {
171 run: func(t *testing.T) {
172 s := NewRuntimeStore()
src/go/pkg/metrix/summary.go
+3 -3
@@ -10,7 +10,7 @@ import (
10 "strconv"
11 )
12
13 -const summaryQuantileLabel = "quantile"
13 +const SummaryQuantileLabel = "quantile"
14 const defaultSummaryReservoirSize = 1024
15 const initialSummaryReservoirCapacity = 64
16
@@ -90,7 +90,7 @@ func (c *storeCore) recordSummaryObservePoint(desc *instrumentDescriptor, point
90 if err != nil {
91 panic(err)
92 }
93 - if labelsContainKey(labels, summaryQuantileLabel) {
93 + if labelsContainKey(labels, SummaryQuantileLabel) {
94 panic(errSummaryLabelKey)
95 }
96
@@ -130,7 +130,7 @@ func (c *storeCore) recordSummaryObserve(desc *instrumentDescriptor, value Sampl
130 if err != nil {
131 panic(err)
132 }
133 - if labelsContainKey(labels, summaryQuantileLabel) {
133 + if labelsContainKey(labels, SummaryQuantileLabel) {
134 panic(errSummaryLabelKey)
135 }
136
src/go/pkg/metrix/test_helpers_test.go
+4
@@ -42,3 +42,7 @@ func mustNoDelta(t *testing.T, r Reader, name string, labels Labels) {
42 _, ok := r.Delta(name, labels)
43 require.False(t, ok, "expected no delta for %s", name)
44 }
45 +
46 +func measureSetFieldLabels(field string) Labels {
47 + return Labels{MeasureSetFieldLabel: field}
48 +}
src/go/pkg/metrix/types.go
+23
@@ -68,6 +68,7 @@ const (
68 MetricKindHistogram
69 MetricKindSummary
70 MetricKindStateSet
71 + MetricKindMeasureSet
72 )
73
74 // FlattenRole describes synthetic scalar roles produced by Read(ReadFlatten()).
@@ -82,6 +83,7 @@ const (
83 FlattenRoleSummarySum
84 FlattenRoleSummaryQuantile
85 FlattenRoleStateSetState
86 + FlattenRoleMeasureSetField
87 )
88
89 type CollectMeta struct {
@@ -116,6 +118,27 @@ type StateSetPoint struct {
118 States map[string]bool
119 }
120
121 +type MeasureFieldSpec struct {
122 + Name string
123 + Float bool
124 +}
125 +
126 +type MeasureSetPoint struct {
127 + Values []SampleValue
128 +}
129 +
130 +type MeasureSetSemantics int
131 +
132 +const (
133 + MeasureSetSemanticsGauge MeasureSetSemantics = iota
134 + MeasureSetSemanticsCounter
135 +)
136 +
137 +type MeasureSetSchema struct {
138 + Semantics MeasureSetSemantics
139 + Fields []MeasureFieldSpec
140 +}
141 +
142 type StateSetMode int
143
144 const (
src/go/pkg/metrix/vec.go
+156
@@ -142,6 +142,26 @@ type statefulStateSetVec struct {
142 cache *vecCache[*statefulStateSetInstrument]
143 }
144
145 +// snapshotMeasureSetGaugeVec caches snapshot MeasureSet gauge series handles by vec label values.
146 +type snapshotMeasureSetGaugeVec struct {
147 + cache *vecCache[*snapshotMeasureSetGaugeInstrument]
148 +}
149 +
150 +// snapshotMeasureSetCounterVec caches snapshot MeasureSet counter series handles by vec label values.
151 +type snapshotMeasureSetCounterVec struct {
152 + cache *vecCache[*snapshotMeasureSetCounterInstrument]
153 +}
154 +
155 +// statefulMeasureSetGaugeVec caches stateful MeasureSet gauge series handles by vec label values.
156 +type statefulMeasureSetGaugeVec struct {
157 + cache *vecCache[*statefulMeasureSetGaugeInstrument]
158 +}
159 +
160 +// statefulMeasureSetCounterVec caches stateful MeasureSet counter series handles by vec label values.
161 +type statefulMeasureSetCounterVec struct {
162 + cache *vecCache[*statefulMeasureSetCounterInstrument]
163 +}
164 +
165 // GaugeVec declares or reuses a snapshot gauge and exposes a label-values lookup API.
166 func (m *snapshotMeter) GaugeVec(name string, labelKeys []string, opts ...InstrumentOption) SnapshotGaugeVec {
167 desc := mustRegisterInstrument(m.backend, metricName(m.prefix, name), kindGauge, modeSnapshot, opts...)
@@ -302,6 +322,70 @@ func (m *statefulMeter) StateSetVec(name string, labelKeys []string, opts ...Ins
322 }
323 }
324
325 +// MeasureSetGaugeVec declares or reuses a snapshot MeasureSet gauge and exposes a label-values lookup API.
326 +func (m *snapshotMeter) MeasureSetGaugeVec(name string, labelKeys []string, opts ...InstrumentOption) SnapshotMeasureSetGaugeVec {
327 + desc := mustRegisterInstrument(m.backend, metricName(m.prefix, name), kindMeasureSet, modeSnapshot, appendMeasureSetSemantics(opts, MeasureSetSemanticsGauge)...)
328 + keys := mustNormalizeVecLabelKeys(labelKeys)
329 + base := appendLabelSets(m.sets, nil)
330 + return &snapshotMeasureSetGaugeVec{
331 + cache: newVecCache(m.backend, base, keys, func(base []LabelSet, vecSet LabelSet) *snapshotMeasureSetGaugeInstrument {
332 + return &snapshotMeasureSetGaugeInstrument{
333 + backend: m.backend,
334 + desc: desc,
335 + base: appendVecSet(base, vecSet),
336 + }
337 + }),
338 + }
339 +}
340 +
341 +// MeasureSetCounterVec declares or reuses a snapshot MeasureSet counter and exposes a label-values lookup API.
342 +func (m *snapshotMeter) MeasureSetCounterVec(name string, labelKeys []string, opts ...InstrumentOption) SnapshotMeasureSetCounterVec {
343 + desc := mustRegisterInstrument(m.backend, metricName(m.prefix, name), kindMeasureSet, modeSnapshot, appendMeasureSetSemantics(opts, MeasureSetSemanticsCounter)...)
344 + keys := mustNormalizeVecLabelKeys(labelKeys)
345 + base := appendLabelSets(m.sets, nil)
346 + return &snapshotMeasureSetCounterVec{
347 + cache: newVecCache(m.backend, base, keys, func(base []LabelSet, vecSet LabelSet) *snapshotMeasureSetCounterInstrument {
348 + return &snapshotMeasureSetCounterInstrument{
349 + backend: m.backend,
350 + desc: desc,
351 + base: appendVecSet(base, vecSet),
352 + }
353 + }),
354 + }
355 +}
356 +
357 +// MeasureSetGaugeVec declares or reuses a stateful MeasureSet gauge and exposes a label-values lookup API.
358 +func (m *statefulMeter) MeasureSetGaugeVec(name string, labelKeys []string, opts ...InstrumentOption) StatefulMeasureSetGaugeVec {
359 + desc := mustRegisterInstrument(m.backend, metricName(m.prefix, name), kindMeasureSet, modeStateful, appendMeasureSetSemantics(opts, MeasureSetSemanticsGauge)...)
360 + keys := mustNormalizeVecLabelKeys(labelKeys)
361 + base := appendLabelSets(m.sets, nil)
362 + return &statefulMeasureSetGaugeVec{
363 + cache: newVecCache(m.backend, base, keys, func(base []LabelSet, vecSet LabelSet) *statefulMeasureSetGaugeInstrument {
364 + return &statefulMeasureSetGaugeInstrument{
365 + backend: m.backend,
366 + desc: desc,
367 + base: appendVecSet(base, vecSet),
368 + }
369 + }),
370 + }
371 +}
372 +
373 +// MeasureSetCounterVec declares or reuses a stateful MeasureSet counter and exposes a label-values lookup API.
374 +func (m *statefulMeter) MeasureSetCounterVec(name string, labelKeys []string, opts ...InstrumentOption) StatefulMeasureSetCounterVec {
375 + desc := mustRegisterInstrument(m.backend, metricName(m.prefix, name), kindMeasureSet, modeStateful, appendMeasureSetSemantics(opts, MeasureSetSemanticsCounter)...)
376 + keys := mustNormalizeVecLabelKeys(labelKeys)
377 + base := appendLabelSets(m.sets, nil)
378 + return &statefulMeasureSetCounterVec{
379 + cache: newVecCache(m.backend, base, keys, func(base []LabelSet, vecSet LabelSet) *statefulMeasureSetCounterInstrument {
380 + return &statefulMeasureSetCounterInstrument{
381 + backend: m.backend,
382 + desc: desc,
383 + base: appendVecSet(base, vecSet),
384 + }
385 + }),
386 + }
387 +}
388 +
389 // GetWithLabelValues returns a snapshot gauge handle for the provided vec label values.
390 func (v *snapshotGaugeVec) GetWithLabelValues(labelValues ...string) (SnapshotGauge, error) {
391 inst, err := v.cache.get(labelValues...)
@@ -482,6 +566,78 @@ func (v *statefulStateSetVec) WithLabelValues(labelValues ...string) StateSetIns
566 return inst
567 }
568
569 +// GetWithLabelValues returns a snapshot MeasureSet gauge handle for the provided vec label values.
570 +func (v *snapshotMeasureSetGaugeVec) GetWithLabelValues(labelValues ...string) (SnapshotMeasureSetGauge, error) {
571 + inst, err := v.cache.get(labelValues...)
572 + if err != nil {
573 + return nil, err
574 + }
575 + return inst, nil
576 +}
577 +
578 +// WithLabelValues returns a snapshot MeasureSet gauge handle and panics on invalid label values.
579 +func (v *snapshotMeasureSetGaugeVec) WithLabelValues(labelValues ...string) SnapshotMeasureSetGauge {
580 + inst, err := v.GetWithLabelValues(labelValues...)
581 + if err != nil {
582 + panic(err)
583 + }
584 + return inst
585 +}
586 +
587 +// GetWithLabelValues returns a snapshot MeasureSet counter handle for the provided vec label values.
588 +func (v *snapshotMeasureSetCounterVec) GetWithLabelValues(labelValues ...string) (SnapshotMeasureSetCounter, error) {
589 + inst, err := v.cache.get(labelValues...)
590 + if err != nil {
591 + return nil, err
592 + }
593 + return inst, nil
594 +}
595 +
596 +// WithLabelValues returns a snapshot MeasureSet counter handle and panics on invalid label values.
597 +func (v *snapshotMeasureSetCounterVec) WithLabelValues(labelValues ...string) SnapshotMeasureSetCounter {
598 + inst, err := v.GetWithLabelValues(labelValues...)
599 + if err != nil {
600 + panic(err)
601 + }
602 + return inst
603 +}
604 +
605 +// GetWithLabelValues returns a stateful MeasureSet gauge handle for the provided vec label values.
606 +func (v *statefulMeasureSetGaugeVec) GetWithLabelValues(labelValues ...string) (StatefulMeasureSetGauge, error) {
607 + inst, err := v.cache.get(labelValues...)
608 + if err != nil {
609 + return nil, err
610 + }
611 + return inst, nil
612 +}
613 +
614 +// WithLabelValues returns a stateful MeasureSet gauge handle and panics on invalid label values.
615 +func (v *statefulMeasureSetGaugeVec) WithLabelValues(labelValues ...string) StatefulMeasureSetGauge {
616 + inst, err := v.GetWithLabelValues(labelValues...)
617 + if err != nil {
618 + panic(err)
619 + }
620 + return inst
621 +}
622 +
623 +// GetWithLabelValues returns a stateful MeasureSet counter handle for the provided vec label values.
624 +func (v *statefulMeasureSetCounterVec) GetWithLabelValues(labelValues ...string) (StatefulMeasureSetCounter, error) {
625 + inst, err := v.cache.get(labelValues...)
626 + if err != nil {
627 + return nil, err
628 + }
629 + return inst, nil
630 +}
631 +
632 +// WithLabelValues returns a stateful MeasureSet counter handle and panics on invalid label values.
633 +func (v *statefulMeasureSetCounterVec) WithLabelValues(labelValues ...string) StatefulMeasureSetCounter {
634 + inst, err := v.GetWithLabelValues(labelValues...)
635 + if err != nil {
636 + panic(err)
637 + }
638 + return inst
639 +}
640 +
641 // normalizeVecLabelKeys validates and copies vec label keys in declared order.
642 func normalizeVecLabelKeys(labelKeys []string) ([]string, error) {
643 keys := append([]string(nil), labelKeys...)
src/go/plugin/framework/chartengine/README.md
+38 -7
@@ -39,7 +39,7 @@ For `ModuleV2` collectors, the runtime integration expects:
39 | `WithRuntimeStore(...)` | Override/disable self-metrics store |
40 | `WithSeriesSelectionAllVisible()` | Process all visible series instead of filtering to latest successful collect cycle. Intended for runtime/internal stores that commit immediately (no cycle boundaries). |
41 | `WithEmitTypeIDBudgetPrefix(...)` | Set the effective type-id prefix used by autogen budget checks |
42 -| `WithRuntimePlannerMode(...)` | Enable runtime planner mode with no-write-tick semantics, for jobs/tests that drive planning directly from runtime metrics instead of collect-cycle boundaries. |
42 +| `WithRuntimePlannerMode(...)` | Enable runtime planner mode with no-write-tick semantics, for jobs/tests that drive planning directly from runtime metrics instead of collect-cycle boundaries. |
43
44 ## End-to-End Example (Single Flow)
45
@@ -75,7 +75,8 @@ groups:
75
76 // 3) Build plan from flattened+raw reader and emit.
77 // ReadFlatten() is included even for templates with static dimensions
78 -// because it is required for inferred dimensions and is the standard pattern.
78 +// because it is required for inferred dimensions, structured-family autogen,
79 +// and is the standard pattern.
80 plan, err := engine.BuildPlan(store.Read(metrix.ReadRaw(), metrix.ReadFlatten()))
81 // handle err
82
@@ -107,11 +108,12 @@ Terms like "materialized state" and "route cache" are defined in the Engine Stat
108
109 ## Reader Requirements
110
110 -| Scenario | Required reader mode |
111 -|------------------------------------------------------------|----------------------------------------------------|
112 -| Static named dimensions only | `Read(...)` is sufficient (no flatten needed) |
113 -| Inferred dimensions (`name` and `name_from_label` omitted) | Must use flattened reader metadata (`ReadFlatten`) |
114 -| Runtime/default `ModuleV2` path | `Read(ReadRaw(), ReadFlatten())` |
111 +| Scenario | Required reader mode |
112 +|--------------------------------------------------------------------------------|----------------------------------------------------------------------------|
113 +| Static named dimensions only | `Read(...)` is sufficient (no flatten needed) |
114 +| Inferred dimensions (`name` and `name_from_label` omitted) | Must use flattened reader metadata (`ReadFlatten`) |
115 +| Structured autogen families (`Histogram`, `Summary`, `StateSet`, `MeasureSet`) | Must use flattened reader metadata (`ReadFlatten`) or they are not visible |
116 +| Runtime/default `ModuleV2` path | `Read(ReadRaw(), ReadFlatten())` |
117
118 If inferred dimensions are present without flattened reader metadata, `BuildPlan` returns an explicit error.
119
@@ -161,10 +163,39 @@ Default lifecycle policy when template omits lifecycle:
163 | Topic | Behavior |
164 |-----------------------|------------------------------------------------------------------------------------------------------------------------------------------------|
165 | Trigger | Unmatched series only when autogen is enabled |
166 +| Structured families | Autogen has dedicated source builders for flattened `Histogram`, `Summary`, `StateSet`, and `MeasureSet` families |
167 | Metric metadata usage | Uses `metrix.MetricMeta` hints for title/family/unit where allowed |
168 | Type ID budget | Enforced via `AutogenPolicy.MaxTypeIDLen` + effective emit type-id prefix (`WithEmitTypeIDBudgetPrefix(...)`) |
169 | Lifecycle | Autogen applies `ExpireAfterSuccessCycles` to **both** chart and dimension expiry (unlike template lifecycle where they default independently) |
170
171 +`MeasureSet` autogen specifics:
172 +
173 +- chartengine treats `MeasureSet` as a structured family, similar to `StateSet`, not as grouped scalar coincidence
174 +- flattened `MeasureSet` inputs are expected to carry:
175 + - `SourceKind = MetricKindMeasureSet`
176 + - `FlattenRole = FlattenRoleMeasureSetField`
177 + - per-field metric names like `<name>_<field>`
178 + - a synthetic reserved field label (`measure_field=<field>`)
179 +- the synthetic `measure_field` label is the authoritative field-identity channel; the per-field metric-name suffix remains for `MetricMeta(name)` compatibility
180 +- gauge-like `MeasureSet` fields autogen with absolute algorithm behavior; counter-like `MeasureSet` fields autogen with incremental algorithm behavior
181 +
182 +### Reserved Flattened Label Keys
183 +
184 +These label keys are treated specially by chartengine when consuming flattened structured-family or distribution inputs:
185 +
186 +| Key / Pattern | Meaning |
187 +|-------------------|---------|
188 +| `le` | Histogram bucket bound label |
189 +| `quantile` | Summary quantile label |
190 +| `measure_field` | `MeasureSet` field identity label |
191 +| `<metric-name>` | `StateSet` special case: the flattened state name is carried under a synthetic label whose key is the base metric name |
192 +
193 +Notes:
194 +
195 +- `le`, `quantile`, and `measure_field` are static reserved flattened-label keys in chartengine.
196 +- `StateSet` is different: it does not use a global static key; it uses the base metric name itself as the synthetic flattened label key.
197 +- These keys are part of the flatten contract between `metrix` and chartengine. Reusing them as ordinary user labels on those flattened inputs is not supported.
198 +
199 ## Runtime Metrics
200
201 `chartengine` self-instruments to a runtime store by default (disable with `WithRuntimeStore(nil)`).
src/go/plugin/framework/chartengine/autogen.go
+75 -9
@@ -46,9 +46,10 @@ type autogenRoleBuilder func(
46 ) (autogenRoute, bool, error)
47
48 var autogenSourceBuilders = map[metrix.MetricKind]autogenSourceBuilder{
49 - metrix.MetricKindHistogram: buildHistogramAutogenRoute,
50 - metrix.MetricKindSummary: buildSummaryAutogenRoute,
51 - metrix.MetricKindStateSet: buildStateSetAutogenRoute,
49 + metrix.MetricKindHistogram: buildHistogramAutogenRoute,
50 + metrix.MetricKindSummary: buildSummaryAutogenRoute,
51 + metrix.MetricKindStateSet: buildStateSetAutogenRoute,
52 + metrix.MetricKindMeasureSet: buildMeasureSetAutogenRoute,
53 }
54
55 var histogramRoleBuilders = map[metrix.FlattenRole]autogenRoleBuilder{
@@ -267,12 +268,12 @@ func buildHistogramBucketAutogenRoute(
268 if baseName == "" {
269 baseName = metricName
270 }
270 - upperBound, ok := labels.Get(histogramBucketLabel)
271 + upperBound, ok := labels.Get(metrix.HistogramBucketLabel)
272 if !ok || strings.TrimSpace(upperBound) == "" {
273 return autogenRoute{}, false, nil
274 }
275 chartID := buildJoinedLabelAutogenID(baseName, labels, map[string]struct{}{
275 - histogramBucketLabel: {},
276 + metrix.HistogramBucketLabel: {},
277 })
278 if !fitsTypeIDBudget(policy.MaxTypeIDLen, typeIDPrefix, chartID) {
279 return autogenRoute{}, false, nil
@@ -281,7 +282,7 @@ func buildHistogramBucketAutogenRoute(
282 chartID: chartID,
283 chartName: baseName,
284 dimensionName: "bucket_" + upperBound,
284 - dimensionKeyLabel: histogramBucketLabel,
285 + dimensionKeyLabel: metrix.HistogramBucketLabel,
286 algorithm: program.AlgorithmIncremental,
287 units: "observations/s",
288 chartType: program.ChartTypeLine,
@@ -323,12 +324,12 @@ func buildSummaryQuantileAutogenRoute(
324 policy AutogenPolicy,
325 typeIDPrefix string,
326 ) (autogenRoute, bool, error) {
326 - quantile, ok := labels.Get(summaryQuantileLabel)
327 + quantile, ok := labels.Get(metrix.SummaryQuantileLabel)
328 if !ok || strings.TrimSpace(quantile) == "" {
329 return autogenRoute{}, false, nil
330 }
331 chartID := buildJoinedLabelAutogenID(metricName, labels, map[string]struct{}{
331 - summaryQuantileLabel: {},
332 + metrix.SummaryQuantileLabel: {},
333 })
334 if !fitsTypeIDBudget(policy.MaxTypeIDLen, typeIDPrefix, chartID) {
335 return autogenRoute{}, false, nil
@@ -338,7 +339,7 @@ func buildSummaryQuantileAutogenRoute(
339 chartID: chartID,
340 chartName: metricName,
341 dimensionName: "quantile_" + quantile,
341 - dimensionKeyLabel: summaryQuantileLabel,
342 + dimensionKeyLabel: metrix.SummaryQuantileLabel,
343 algorithm: program.AlgorithmAbsolute,
344 units: units,
345 chartType: chartTypeFromUnits(units),
@@ -434,6 +435,71 @@ func buildStateSetAutogenRoute(
435 }, true, nil
436 }
437
438 +func buildMeasureSetAutogenRoute(
439 + metricName string,
440 + labels metrix.LabelView,
441 + meta metrix.SeriesMeta,
442 + policy AutogenPolicy,
443 + typeIDPrefix string,
444 +) (autogenRoute, bool, error) {
445 + if meta.FlattenRole != metrix.FlattenRoleMeasureSetField {
446 + return autogenRoute{}, false, nil
447 + }
448 + sourceName, fieldName, ok := resolveMeasureSetAutogenSource(metricName, labels)
449 + if !ok {
450 + return autogenRoute{}, false, nil
451 + }
452 + chartID := buildJoinedLabelAutogenID(sourceName, labels, map[string]struct{}{
453 + metrix.MeasureSetFieldLabel: {},
454 + })
455 + if !fitsTypeIDBudget(policy.MaxTypeIDLen, typeIDPrefix, chartID) {
456 + return autogenRoute{}, false, nil
457 + }
458 + algorithm := program.AlgorithmAbsolute
459 + units := getAutogenGaugeUnits(sourceName)
460 + if meta.Kind == metrix.MetricKindCounter {
461 + algorithm = program.AlgorithmIncremental
462 + units = getAutogenCounterUnits(sourceName)
463 + }
464 + return autogenRoute{
465 + chartID: chartID,
466 + chartName: sourceName,
467 + dimensionName: fieldName,
468 + dimensionKeyLabel: metrix.MeasureSetFieldLabel,
469 + algorithm: algorithm,
470 + units: units,
471 + chartType: chartTypeFromUnits(units),
472 + family: getAutogenChartFamily(sourceName),
473 + contextName: sourceName,
474 + staticDimension: false,
475 + }, true, nil
476 +}
477 +
478 +func resolveMeasureSetAutogenSource(metricName string, labels metrix.LabelView) (string, string, bool) {
479 + if strings.TrimSpace(metricName) == "" {
480 + return "", "", false
481 + }
482 + if labels == nil {
483 + return "", "", false
484 + }
485 +
486 + fieldName, ok := labels.Get(metrix.MeasureSetFieldLabel)
487 + if !ok || strings.TrimSpace(fieldName) == "" {
488 + return "", "", false
489 + }
490 +
491 + suffix := "_" + fieldName
492 + if !strings.HasSuffix(metricName, suffix) {
493 + return "", "", false
494 + }
495 +
496 + sourceName := strings.TrimSuffix(metricName, suffix)
497 + if strings.TrimSpace(sourceName) == "" {
498 + return "", "", false
499 + }
500 + return sourceName, fieldName, true
501 +}
502 +
503 func buildScalarAutogenRoute(
504 metricName string,
505 labels metrix.LabelView,
src/go/plugin/framework/chartengine/autogen_test.go
+121 -2
@@ -29,6 +29,9 @@ func TestAutogenRouteBuilderScenarios(t *testing.T) {
29 "build state-set autogen route": {
30 run: runTestBuildStateSetAutogenRoute,
31 },
32 + "build measure-set autogen route": {
33 + run: runTestBuildMeasureSetAutogenRoute,
34 + },
35 }
36
37 for name, tc := range tests {
@@ -124,7 +127,7 @@ func runTestBuildHistogramBucketAutogenRoute(t *testing.T) {
127
128 assert.Equal(t, tc.wantID, route.chartID)
129 assert.Equal(t, tc.wantDim, route.dimensionName)
127 - assert.Equal(t, histogramBucketLabel, route.dimensionKeyLabel)
130 + assert.Equal(t, metrix.HistogramBucketLabel, route.dimensionKeyLabel)
131 assert.Equal(t, program.AlgorithmIncremental, route.algorithm)
132 assert.False(t, route.staticDimension)
133 })
@@ -163,7 +166,7 @@ func runTestBuildSummaryQuantileAutogenRoute(t *testing.T) {
166
167 assert.Equal(t, tc.wantID, route.chartID)
168 assert.Equal(t, tc.wantDim, route.dimensionName)
166 - assert.Equal(t, summaryQuantileLabel, route.dimensionKeyLabel)
169 + assert.Equal(t, metrix.SummaryQuantileLabel, route.dimensionKeyLabel)
170 assert.Equal(t, program.AlgorithmAbsolute, route.algorithm)
171 assert.False(t, route.staticDimension)
172 })
@@ -212,6 +215,122 @@ func runTestBuildStateSetAutogenRoute(t *testing.T) {
215 }
216 }
217
218 +func runTestBuildMeasureSetAutogenRoute(t *testing.T) {
219 + tests := map[string]struct {
220 + metricName string
221 + labels map[string]string
222 + meta metrix.SeriesMeta
223 + wantOK bool
224 + wantID string
225 + wantDim string
226 + wantAlg program.Algorithm
227 + wantUnits string
228 + }{
229 + "MeasureSet gauge uses synthetic field label and absolute algorithm": {
230 + metricName: "service_latency_seconds_value",
231 + labels: map[string]string{
232 + "instance": "db1",
233 + metrix.MeasureSetFieldLabel: "value",
234 + },
235 + meta: metrix.SeriesMeta{
236 + Kind: metrix.MetricKindGauge,
237 + SourceKind: metrix.MetricKindMeasureSet,
238 + FlattenRole: metrix.FlattenRoleMeasureSetField,
239 + },
240 + wantOK: true,
241 + wantID: "service_latency_seconds-instance=db1",
242 + wantDim: "value",
243 + wantAlg: program.AlgorithmAbsolute,
244 + wantUnits: "seconds",
245 + },
246 + "MeasureSet counter uses synthetic field label and incremental algorithm": {
247 + metricName: "svc_requests_total_ok",
248 + labels: map[string]string{
249 + "instance": "db1",
250 + metrix.MeasureSetFieldLabel: "ok",
251 + },
252 + meta: metrix.SeriesMeta{
253 + Kind: metrix.MetricKindCounter,
254 + SourceKind: metrix.MetricKindMeasureSet,
255 + FlattenRole: metrix.FlattenRoleMeasureSetField,
256 + },
257 + wantOK: true,
258 + wantID: "svc_requests_total-instance=db1",
259 + wantDim: "ok",
260 + wantAlg: program.AlgorithmIncremental,
261 + wantUnits: "requests/s",
262 + },
263 + "MeasureSet ignores unrelated matching labels and uses reserved field label": {
264 + metricName: "svc_requests_total_ok",
265 + labels: map[string]string{
266 + "instance": "db1",
267 + "svc_requests": "total_ok",
268 + metrix.MeasureSetFieldLabel: "ok",
269 + },
270 + meta: metrix.SeriesMeta{
271 + Kind: metrix.MetricKindCounter,
272 + SourceKind: metrix.MetricKindMeasureSet,
273 + FlattenRole: metrix.FlattenRoleMeasureSetField,
274 + },
275 + wantOK: true,
276 + wantID: "svc_requests_total-instance=db1-svc_requests=total_ok",
277 + wantDim: "ok",
278 + wantAlg: program.AlgorithmIncremental,
279 + wantUnits: "requests/s",
280 + },
281 + "MeasureSet without reserved field label does not route": {
282 + metricName: "svc_requests_total_ok",
283 + labels: map[string]string{
284 + "instance": "db1",
285 + "svc_requests": "total_ok",
286 + },
287 + meta: metrix.SeriesMeta{
288 + Kind: metrix.MetricKindCounter,
289 + SourceKind: metrix.MetricKindMeasureSet,
290 + FlattenRole: metrix.FlattenRoleMeasureSetField,
291 + },
292 + wantOK: false,
293 + },
294 + "MeasureSet with mismatched reserved field label does not route": {
295 + metricName: "svc_requests_total_ok",
296 + labels: map[string]string{
297 + "instance": "db1",
298 + metrix.MeasureSetFieldLabel: "failed",
299 + },
300 + meta: metrix.SeriesMeta{
301 + Kind: metrix.MetricKindCounter,
302 + SourceKind: metrix.MetricKindMeasureSet,
303 + FlattenRole: metrix.FlattenRoleMeasureSetField,
304 + },
305 + wantOK: false,
306 + },
307 + }
308 +
309 + for name, tc := range tests {
310 + t.Run(name, func(t *testing.T) {
311 + route, ok, err := buildMeasureSetAutogenRoute(
312 + tc.metricName,
313 + sortedLabelView(tc.labels),
314 + tc.meta,
315 + AutogenPolicy{Enabled: true, MaxTypeIDLen: defaultMaxTypeIDLen},
316 + "",
317 + )
318 + require.NoError(t, err)
319 + require.Equal(t, tc.wantOK, ok)
320 + if !tc.wantOK {
321 + return
322 + }
323 +
324 + assert.Equal(t, tc.wantID, route.chartID)
325 + assert.Equal(t, tc.wantDim, route.dimensionName)
326 + assert.Equal(t, tc.wantAlg, route.algorithm)
327 + assert.Equal(t, tc.wantUnits, route.units)
328 + assert.Equal(t, metrix.MeasureSetFieldLabel, route.dimensionKeyLabel)
329 + assert.False(t, route.staticDimension)
330 + })
331 + }
332 +}
333 +
334 func TestFitsTypeIDBudget(t *testing.T) {
335 tests := map[string]struct {
336 maxLen int
src/go/plugin/framework/chartengine/compiler.go
+2 -1
@@ -7,6 +7,7 @@ import (
7 "sort"
8 "strings"
9
10 + "github.com/netdata/netdata/go/plugins/pkg/metrix"
11 metrixselector "github.com/netdata/netdata/go/plugins/pkg/metrix/selector"
12 "github.com/netdata/netdata/go/plugins/plugin/framework/chartengine/internal/program"
13 "github.com/netdata/netdata/go/plugins/plugin/framework/charttpl"
@@ -380,7 +381,7 @@ func metricKindsFromNames(names []string) []string {
381 func supportsRuntimeInferredDimension(meta metrixselector.Meta) bool {
382 for _, key := range meta.ConstrainedLabelKeys {
383 switch key {
383 - case histogramBucketLabel, summaryQuantileLabel:
384 + case metrix.HistogramBucketLabel, metrix.SummaryQuantileLabel:
385 return true
386 }
387 }
src/go/plugin/framework/chartengine/planner.go
-5
@@ -12,11 +12,6 @@ import (
12 "github.com/netdata/netdata/go/plugins/plugin/framework/chartengine/internal/program"
13 )
14
15 -const (
16 - histogramBucketLabel = "le"
17 - summaryQuantileLabel = "quantile"
18 -)
19 -
15 type labelSliceView struct {
16 items []metrix.Label
17 }
src/go/plugin/framework/chartengine/planner_dimension.go
+2 -2
@@ -44,9 +44,9 @@ func resolveDimensionName(dim program.Dimension, metricName string, labels metri
44 func inferDimensionLabelKey(metricName string, meta metrix.SeriesMeta) (string, bool, error) {
45 switch meta.FlattenRole {
46 case metrix.FlattenRoleHistogramBucket:
47 - return histogramBucketLabel, true, nil
47 + return metrix.HistogramBucketLabel, true, nil
48 case metrix.FlattenRoleSummaryQuantile:
49 - return summaryQuantileLabel, true, nil
49 + return metrix.SummaryQuantileLabel, true, nil
50 case metrix.FlattenRoleStateSetState:
51 if strings.TrimSpace(metricName) == "" {
52 return "", false, fmt.Errorf("chartengine: stateset inference requires metric family name")
src/go/plugin/framework/chartengine/planner_test.go
+166
@@ -226,6 +226,8 @@ func TestBuildPlanLegacySingleScenarioCases(t *testing.T) {
226 "BuildPlanAutogenCreatesChartForUnmatchedGauge": {run: runTestBuildPlanAutogenCreatesChartForUnmatchedGauge},
227 "BuildPlanAutogenCreatesChartForUnmatchedStateSet": {run: runTestBuildPlanAutogenCreatesChartForUnmatchedStateSet},
228 "BuildPlanAutogenKeepsStateSetUnitsWhenMetricMetaUnitIsSet": {run: runTestBuildPlanAutogenKeepsStateSetUnitsWhenMetricMetaUnitIsSet},
229 + "BuildPlanAutogenCreatesChartForUnmatchedMeasureSetGauge": {run: runTestBuildPlanAutogenCreatesChartForUnmatchedMeasureSetGauge},
230 + "BuildPlanAutogenCreatesChartForUnmatchedMeasureSetCounter": {run: runTestBuildPlanAutogenCreatesChartForUnmatchedMeasureSetCounter},
231 "BuildPlanTemplateWinsOnAutogenChartIDCollisionAcrossSeries": {run: runTestBuildPlanTemplateWinsOnAutogenChartIDCollisionAcrossSeries},
232 "BuildPlanAutogenRemovalLifecycleExpiry": {run: runTestBuildPlanAutogenRemovalLifecycleExpiry},
233 "BuildPlanFirstWriterWinsAndAccumulatesRepeatedRoutes": {run: runTestBuildPlanFirstWriterWinsAndAccumulatesRepeatedRoutes},
@@ -1514,6 +1516,170 @@ groups:
1516 assert.Equal(t, "state", create.Meta.Units)
1517 }
1518
1519 +func runTestBuildPlanAutogenCreatesChartForUnmatchedMeasureSetGauge(t *testing.T) {
1520 + e, err := New(WithEnginePolicy(EnginePolicy{Autogen: &AutogenPolicy{Enabled: true}}))
1521 + require.NoError(t, err)
1522 +
1523 + yaml := `
1524 +version: v1
1525 +groups:
1526 + - family: Service
1527 + metrics:
1528 + - svc.requests_total
1529 + charts:
1530 + - title: Requests
1531 + context: requests
1532 + units: requests/s
1533 + dimensions:
1534 + - selector: svc.requests_total
1535 + name: total
1536 +`
1537 + require.NoError(t, e.LoadYAML([]byte(yaml), 1))
1538 +
1539 + store := metrix.NewCollectorStore()
1540 + cc := mustCycleController(t, store)
1541 + ms := store.Write().SnapshotMeter("svc").MeasureSetGauge(
1542 + "latency_seconds",
1543 + metrix.WithMeasureSetFields(
1544 + metrix.MeasureFieldSpec{Name: "value"},
1545 + metrix.MeasureFieldSpec{Name: "ratio", Float: true},
1546 + ),
1547 + metrix.WithDescription("Latency"),
1548 + metrix.WithChartFamily("Service"),
1549 + metrix.WithUnit("seconds"),
1550 + )
1551 +
1552 + cc.BeginCycle()
1553 + ms.ObservePoint(metrix.MeasureSetPoint{Values: []metrix.SampleValue{1.5, 0.5}})
1554 + cc.CommitCycleSuccess()
1555 +
1556 + plan, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
1557 + require.NoError(t, err)
1558 +
1559 + assert.Equal(t, []ActionKind{
1560 + ActionCreateChart,
1561 + ActionCreateDimension,
1562 + ActionCreateDimension,
1563 + ActionUpdateChart,
1564 + }, actionKinds(plan.Actions))
1565 +
1566 + create := findCreateChartAction(plan)
1567 + require.NotNil(t, create)
1568 + assert.Equal(t, "svc.latency_seconds", create.ChartID)
1569 + assert.Equal(t, "Latency", create.Meta.Title)
1570 + assert.Equal(t, "Service", create.Meta.Family)
1571 + assert.Equal(t, "svc.latency_seconds", create.Meta.Context)
1572 + assert.Equal(t, "seconds", create.Meta.Units)
1573 + _, hasFieldLabel := create.Labels[metrix.MeasureSetFieldLabel]
1574 + assert.False(t, hasFieldLabel)
1575 +
1576 + dims := map[string]CreateDimensionAction{}
1577 + for _, action := range plan.Actions {
1578 + dim, ok := action.(CreateDimensionAction)
1579 + if !ok || dim.ChartID != "svc.latency_seconds" {
1580 + continue
1581 + }
1582 + dims[dim.Name] = dim
1583 + }
1584 + require.Len(t, dims, 2)
1585 + assert.Equal(t, program.AlgorithmAbsolute, dims["value"].Algorithm)
1586 + assert.False(t, dims["value"].Float)
1587 + assert.Equal(t, program.AlgorithmAbsolute, dims["ratio"].Algorithm)
1588 + assert.True(t, dims["ratio"].Float)
1589 +
1590 + update := findUpdateAction(plan)
1591 + require.NotNil(t, update)
1592 + require.Len(t, update.Values, 2)
1593 + names := map[string]struct{}{}
1594 + for _, value := range update.Values {
1595 + names[value.Name] = struct{}{}
1596 + }
1597 + assert.Contains(t, names, "ratio")
1598 + assert.Contains(t, names, "value")
1599 +}
1600 +
1601 +func runTestBuildPlanAutogenCreatesChartForUnmatchedMeasureSetCounter(t *testing.T) {
1602 + e, err := New(WithEnginePolicy(EnginePolicy{Autogen: &AutogenPolicy{Enabled: true}}))
1603 + require.NoError(t, err)
1604 +
1605 + yaml := `
1606 +version: v1
1607 +groups:
1608 + - family: Service
1609 + metrics:
1610 + - svc.requests_total
1611 + charts:
1612 + - title: Requests
1613 + context: requests
1614 + units: requests/s
1615 + dimensions:
1616 + - selector: svc.requests_total
1617 + name: total
1618 +`
1619 + require.NoError(t, e.LoadYAML([]byte(yaml), 1))
1620 +
1621 + store := metrix.NewCollectorStore()
1622 + cc := mustCycleController(t, store)
1623 + ms := store.Write().SnapshotMeter("svc").MeasureSetCounter(
1624 + "requests_total",
1625 + metrix.WithMeasureSetFields(
1626 + metrix.MeasureFieldSpec{Name: "ok"},
1627 + metrix.MeasureFieldSpec{Name: "failed"},
1628 + ),
1629 + metrix.WithDescription("Requests"),
1630 + metrix.WithChartFamily("Service"),
1631 + metrix.WithUnit("requests"),
1632 + )
1633 +
1634 + cc.BeginCycle()
1635 + ms.ObserveTotalPoint(metrix.MeasureSetPoint{Values: []metrix.SampleValue{10, 2}})
1636 + cc.CommitCycleSuccess()
1637 +
1638 + plan, err := e.BuildPlan(store.Read(metrix.ReadFlatten()))
1639 + require.NoError(t, err)
1640 +
1641 + assert.Equal(t, []ActionKind{
1642 + ActionCreateChart,
1643 + ActionCreateDimension,
1644 + ActionCreateDimension,
1645 + ActionUpdateChart,
1646 + }, actionKinds(plan.Actions))
1647 +
1648 + create := findCreateChartAction(plan)
1649 + require.NotNil(t, create)
1650 + assert.Equal(t, "svc.requests_total", create.ChartID)
1651 + assert.Equal(t, "Requests", create.Meta.Title)
1652 + assert.Equal(t, "Service", create.Meta.Family)
1653 + assert.Equal(t, "svc.requests_total", create.Meta.Context)
1654 + assert.Equal(t, "requests/s", create.Meta.Units)
1655 + _, hasFieldLabel := create.Labels[metrix.MeasureSetFieldLabel]
1656 + assert.False(t, hasFieldLabel)
1657 +
1658 + dims := map[string]CreateDimensionAction{}
1659 + for _, action := range plan.Actions {
1660 + dim, ok := action.(CreateDimensionAction)
1661 + if !ok || dim.ChartID != "svc.requests_total" {
1662 + continue
1663 + }
1664 + dims[dim.Name] = dim
1665 + }
1666 + require.Len(t, dims, 2)
1667 + assert.Equal(t, program.AlgorithmIncremental, dims["ok"].Algorithm)
1668 + assert.Equal(t, program.AlgorithmIncremental, dims["failed"].Algorithm)
1669 + assert.False(t, dims["ok"].Float)
1670 + assert.False(t, dims["failed"].Float)
1671 +
1672 + update := findUpdateAction(plan)
1673 + require.NotNil(t, update)
1674 + require.Len(t, update.Values, 2)
1675 + names := map[string]struct{}{}
1676 + for _, value := range update.Values {
1677 + names[value.Name] = struct{}{}
1678 + }
1679 + assert.Contains(t, names, "failed")
1680 + assert.Contains(t, names, "ok")
1681 +}
1682 +
1683 func runTestBuildPlanTemplateWinsOnAutogenChartIDCollisionAcrossSeries(t *testing.T) {
1684 e, err := New(WithEnginePolicy(EnginePolicy{Autogen: &AutogenPolicy{Enabled: true}}))
1685 require.NoError(t, err)