| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | package metrix |
| 4 | |
| 5 | import ( |
| 6 | "math" |
| 7 | "sync" |
| 8 | "testing" |
| 9 | "time" |
| 10 | |
| 11 | "github.com/stretchr/testify/require" |
| 12 | ) |
| 13 | |
| 14 | func TestRuntimeStoreScenarios(t *testing.T) { |
| 15 | tests := map[string]struct { |
| 16 | run func(t *testing.T) |
| 17 | }{ |
| 18 | "stateful writes are immediate commit with runtime metadata sequence": { |
| 19 | run: func(t *testing.T) { |
| 20 | s := NewRuntimeStore() |
| 21 | m := s.Write().StatefulMeter("runtime") |
| 22 | g := m.Gauge("heap_bytes") |
| 23 | c := m.Counter("jobs_total") |
| 24 | |
| 25 | g.Set(5) |
| 26 | mustValue(t, s.Read(), "runtime.heap_bytes", nil, 5) |
| 27 | meta := s.Read().CollectMeta() |
| 28 | require.Equal(t, CollectStatusSuccess, meta.LastAttemptStatus, "unexpected metadata after gauge set: %#v", meta) |
| 29 | require.Equal(t, uint64(1), meta.LastAttemptSeq, "unexpected metadata after gauge set: %#v", meta) |
| 30 | require.Equal(t, uint64(1), meta.LastSuccessSeq, "unexpected metadata after gauge set: %#v", meta) |
| 31 | |
| 32 | c.Add(10) |
| 33 | mustValue(t, s.Read(), "runtime.jobs_total", nil, 10) |
| 34 | mustNoDelta(t, s.Read(), "runtime.jobs_total", nil) |
| 35 | meta = s.Read().CollectMeta() |
| 36 | require.Equal(t, uint64(2), meta.LastAttemptSeq, "unexpected metadata after first counter add: %#v", meta) |
| 37 | require.Equal(t, uint64(2), meta.LastSuccessSeq, "unexpected metadata after first counter add: %#v", meta) |
| 38 | |
| 39 | c.Add(4) |
| 40 | mustValue(t, s.Read(), "runtime.jobs_total", nil, 14) |
| 41 | mustDelta(t, s.Read(), "runtime.jobs_total", nil, 4) |
| 42 | meta = s.Read().CollectMeta() |
| 43 | require.Equal(t, uint64(3), meta.LastAttemptSeq, "unexpected metadata after second counter add: %#v", meta) |
| 44 | require.Equal(t, uint64(3), meta.LastSuccessSeq, "unexpected metadata after second counter add: %#v", meta) |
| 45 | }, |
| 46 | }, |
| 47 | "runtime rejects snapshot-style constraints": { |
| 48 | run: func(t *testing.T) { |
| 49 | s := NewRuntimeStore() |
| 50 | |
| 51 | expectPanic(t, func() { |
| 52 | _ = s.Write().StatefulMeter("runtime").Gauge("x", WithFreshness(FreshnessCycle)) |
| 53 | }) |
| 54 | expectPanic(t, func() { |
| 55 | _ = s.Write().StatefulMeter("runtime").Summary("s", WithSummaryQuantiles(0.5), WithWindow(WindowCycle)) |
| 56 | }) |
| 57 | }, |
| 58 | }, |
| 59 | "runtime summary read and flatten are chart-compatible": { |
| 60 | run: func(t *testing.T) { |
| 61 | s := NewRuntimeStore() |
| 62 | sum := s.Write().StatefulMeter("runtime").Summary("latency", WithSummaryQuantiles(0.5, 1.0)) |
| 63 | |
| 64 | sum.Observe(1) |
| 65 | sum.Observe(3) |
| 66 | |
| 67 | p, ok := s.Read().Summary("runtime.latency", nil) |
| 68 | require.True(t, ok, "expected runtime summary point") |
| 69 | require.Equal(t, SampleValue(2), p.Count) |
| 70 | require.Equal(t, SampleValue(4), p.Sum) |
| 71 | require.Len(t, p.Quantiles, 2) |
| 72 | require.False(t, math.IsNaN(p.Quantiles[0].Value) || math.IsNaN(p.Quantiles[1].Value), "expected finite quantile values: %#v", p.Quantiles) |
| 73 | |
| 74 | fr := s.Read(ReadFlatten()) |
| 75 | mustValue(t, fr, "runtime.latency_count", nil, 2) |
| 76 | mustValue(t, fr, "runtime.latency_sum", nil, 4) |
| 77 | _, ok = fr.Summary("runtime.latency", nil) |
| 78 | require.False(t, ok, "expected flattened view to hide typed summary getter") |
| 79 | }, |
| 80 | }, |
| 81 | "runtime histogram and stateset are readable and flattenable": { |
| 82 | run: func(t *testing.T) { |
| 83 | s := NewRuntimeStore() |
| 84 | m := s.Write().StatefulMeter("runtime") |
| 85 | h := m.Histogram("req_duration", WithHistogramBounds(1, 2)) |
| 86 | ss := m.StateSet("mode", WithStateSetStates("maintenance", "operational"), WithStateSetMode(ModeEnum)) |
| 87 | |
| 88 | h.Observe(0.5) |
| 89 | h.Observe(3) |
| 90 | ss.Enable("operational") |
| 91 | |
| 92 | mustHistogram(t, s.Read(), "runtime.req_duration", nil, HistogramPoint{ |
| 93 | Count: 2, Sum: 3.5, |
| 94 | Buckets: []BucketPoint{ |
| 95 | {UpperBound: 1, CumulativeCount: 1}, |
| 96 | {UpperBound: 2, CumulativeCount: 1}, |
| 97 | }, |
| 98 | }) |
| 99 | mustStateSet(t, s.Read(), "runtime.mode", nil, map[string]bool{ |
| 100 | "maintenance": false, |
| 101 | "operational": true, |
| 102 | }) |
| 103 | |
| 104 | fr := s.Read(ReadFlatten()) |
| 105 | mustValue(t, fr, "runtime.req_duration_bucket", Labels{"le": "1"}, 1) |
| 106 | mustValue(t, fr, "runtime.req_duration_bucket", Labels{"le": "2"}, 1) |
| 107 | mustValue(t, fr, "runtime.req_duration_bucket", Labels{"le": "+Inf"}, 2) |
| 108 | mustValue(t, fr, "runtime.mode", Labels{"runtime.mode": "maintenance"}, 0) |
| 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() |
| 173 | c := s.Write().StatefulMeter("runtime").Counter("events_total") |
| 174 | |
| 175 | const workers = 8 |
| 176 | const perWorker = 200 |
| 177 | |
| 178 | var wg sync.WaitGroup |
| 179 | wg.Add(workers) |
| 180 | for range workers { |
| 181 | go func() { |
| 182 | defer wg.Done() |
| 183 | for range perWorker { |
| 184 | c.Add(1) |
| 185 | } |
| 186 | }() |
| 187 | } |
| 188 | wg.Wait() |
| 189 | |
| 190 | mustValue(t, s.Read(), "runtime.events_total", nil, workers*perWorker) |
| 191 | }, |
| 192 | }, |
| 193 | "runtime counter delta works with intervening writes to other metrics": { |
| 194 | run: func(t *testing.T) { |
| 195 | s := NewRuntimeStore() |
| 196 | m := s.Write().StatefulMeter("runtime") |
| 197 | c := m.Counter("events_total") |
| 198 | g := m.Gauge("heap_bytes") |
| 199 | |
| 200 | c.Add(10) |
| 201 | g.Set(5) // intervening non-counter write |
| 202 | c.Add(4) |
| 203 | |
| 204 | mustValue(t, s.Read(), "runtime.events_total", nil, 14) |
| 205 | mustDelta(t, s.Read(), "runtime.events_total", nil, 4) |
| 206 | }, |
| 207 | }, |
| 208 | "runtime wall-clock retention evicts stale series": { |
| 209 | run: func(t *testing.T) { |
| 210 | s := NewRuntimeStore() |
| 211 | view := runtimeStoreViewForTest(t, s) |
| 212 | now := time.Unix(1_700_000_000, 0) |
| 213 | view.backend.now = func() time.Time { return now } |
| 214 | view.backend.retention = runtimeRetentionPolicy{ |
| 215 | ttl: 5 * time.Second, |
| 216 | maxSeries: 0, |
| 217 | } |
| 218 | view.backend.compaction = runtimeCompactionPolicy{ |
| 219 | maxOverlayDepth: 1, |
| 220 | maxOverlayWrites: 1, |
| 221 | } |
| 222 | |
| 223 | m := s.Write().StatefulMeter("runtime") |
| 224 | g := m.Gauge("queue_depth") |
| 225 | lsa := m.LabelSet(Label{Key: "id", Value: "a"}) |
| 226 | lsb := m.LabelSet(Label{Key: "id", Value: "b"}) |
| 227 | |
| 228 | g.Set(10, lsa) |
| 229 | now = now.Add(2 * time.Second) |
| 230 | g.Set(20, lsb) |
| 231 | mustValue(t, s.Read(ReadRaw()), "runtime.queue_depth", Labels{"id": "a"}, 10) |
| 232 | mustValue(t, s.Read(ReadRaw()), "runtime.queue_depth", Labels{"id": "b"}, 20) |
| 233 | |
| 234 | now = now.Add(4 * time.Second) |
| 235 | g.Set(21, lsb) // triggers retention sweep |
| 236 | mustNoValue(t, s.Read(ReadRaw()), "runtime.queue_depth", Labels{"id": "a"}) |
| 237 | mustValue(t, s.Read(ReadRaw()), "runtime.queue_depth", Labels{"id": "b"}, 21) |
| 238 | }, |
| 239 | }, |
| 240 | "runtime max-series cap evicts oldest series deterministically": { |
| 241 | run: func(t *testing.T) { |
| 242 | s := NewRuntimeStore() |
| 243 | view := runtimeStoreViewForTest(t, s) |
| 244 | now := time.Unix(1_700_000_000, 0) |
| 245 | view.backend.now = func() time.Time { return now } |
| 246 | view.backend.retention = runtimeRetentionPolicy{ |
| 247 | ttl: 0, |
| 248 | maxSeries: 2, |
| 249 | } |
| 250 | view.backend.compaction = runtimeCompactionPolicy{ |
| 251 | maxOverlayDepth: 1, |
| 252 | maxOverlayWrites: 1, |
| 253 | } |
| 254 | |
| 255 | m := s.Write().StatefulMeter("runtime") |
| 256 | g := m.Gauge("queue_depth") |
| 257 | lsa := m.LabelSet(Label{Key: "id", Value: "a"}) |
| 258 | lsb := m.LabelSet(Label{Key: "id", Value: "b"}) |
| 259 | lsc := m.LabelSet(Label{Key: "id", Value: "c"}) |
| 260 | |
| 261 | g.Set(10, lsa) |
| 262 | g.Set(20, lsb) |
| 263 | g.Set(30, lsc) |
| 264 | |
| 265 | mustNoValue(t, s.Read(ReadRaw()), "runtime.queue_depth", Labels{"id": "a"}) |
| 266 | mustValue(t, s.Read(ReadRaw()), "runtime.queue_depth", Labels{"id": "b"}, 20) |
| 267 | mustValue(t, s.Read(ReadRaw()), "runtime.queue_depth", Labels{"id": "c"}, 30) |
| 268 | }, |
| 269 | }, |
| 270 | } |
| 271 | |
| 272 | for name, tc := range tests { |
| 273 | t.Run(name, tc.run) |
| 274 | } |
| 275 | } |
| 276 | |
| 277 | func runtimeStoreViewForTest(t *testing.T, s RuntimeStore) *runtimeStoreView { |
| 278 | t.Helper() |
| 279 | v, ok := s.(*runtimeStoreView) |
| 280 | require.True(t, ok, "unexpected runtime store implementation: %T", s) |
| 281 | return v |
| 282 | } |