| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | package runtimechartemit |
| 4 | |
| 5 | import ( |
| 6 | "bytes" |
| 7 | "errors" |
| 8 | "strings" |
| 9 | "sync" |
| 10 | "testing" |
| 11 | "time" |
| 12 | |
| 13 | "github.com/netdata/netdata/go/plugins/plugin/framework/chartemit" |
| 14 | "github.com/netdata/netdata/go/plugins/plugin/framework/chartengine" |
| 15 | "github.com/stretchr/testify/assert" |
| 16 | "github.com/stretchr/testify/require" |
| 17 | |
| 18 | "github.com/netdata/netdata/go/plugins/pkg/metrix" |
| 19 | ) |
| 20 | |
| 21 | type safeBuffer struct { |
| 22 | mu sync.Mutex |
| 23 | buf bytes.Buffer |
| 24 | } |
| 25 | |
| 26 | func (b *safeBuffer) Write(p []byte) (int, error) { |
| 27 | b.mu.Lock() |
| 28 | defer b.mu.Unlock() |
| 29 | return b.buf.Write(p) |
| 30 | } |
| 31 | |
| 32 | func (b *safeBuffer) Len() int { |
| 33 | b.mu.Lock() |
| 34 | defer b.mu.Unlock() |
| 35 | return b.buf.Len() |
| 36 | } |
| 37 | |
| 38 | func (b *safeBuffer) String() string { |
| 39 | b.mu.Lock() |
| 40 | defer b.mu.Unlock() |
| 41 | return b.buf.String() |
| 42 | } |
| 43 | |
| 44 | type failingWriter struct{} |
| 45 | |
| 46 | func (failingWriter) Write(_ []byte) (int, error) { |
| 47 | return 0, errors.New("boom") |
| 48 | } |
| 49 | |
| 50 | func requireInOrder(t *testing.T, text string, parts ...string) { |
| 51 | t.Helper() |
| 52 | |
| 53 | offset := 0 |
| 54 | for _, part := range parts { |
| 55 | idx := strings.Index(text[offset:], part) |
| 56 | require.NotEqualf(t, -1, idx, "missing ordered fragment %q in %q", part, text) |
| 57 | offset += idx + len(part) |
| 58 | } |
| 59 | } |
| 60 | |
| 61 | func TestRuntimeMetricsJobStartStopLifecycle(t *testing.T) { |
| 62 | tests := map[string]struct { |
| 63 | clock int |
| 64 | }{ |
| 65 | "start tick stop emits output and toggles running state": { |
| 66 | clock: 1, |
| 67 | }, |
| 68 | } |
| 69 | |
| 70 | for name, tc := range tests { |
| 71 | t.Run(name, func(t *testing.T) { |
| 72 | reg := newComponentRegistry() |
| 73 | store := metrix.NewRuntimeStore() |
| 74 | store.Write().StatefulMeter("component").Gauge("load").Set(5) |
| 75 | |
| 76 | reg.upsert(componentSpec{ |
| 77 | Name: "component", |
| 78 | Store: store, |
| 79 | TemplateYAML: []byte(runtimeGaugeTemplateYAML()), |
| 80 | UpdateEvery: 1, |
| 81 | EmitEnv: chartemit.EmitEnv{ |
| 82 | TypeID: "netdata.go.d.internal.component", |
| 83 | UpdateEvery: 1, |
| 84 | Plugin: "go.d", |
| 85 | Module: "internal", |
| 86 | JobName: "component", |
| 87 | }, |
| 88 | }) |
| 89 | |
| 90 | out := &safeBuffer{} |
| 91 | job := newRuntimeMetricsJob(out, reg, nil) |
| 92 | |
| 93 | done := make(chan struct{}) |
| 94 | go func() { |
| 95 | job.Start() |
| 96 | close(done) |
| 97 | }() |
| 98 | |
| 99 | require.Eventually(t, func() bool { return job.running.Load() }, time.Second, 10*time.Millisecond) |
| 100 | |
| 101 | job.Tick(tc.clock) |
| 102 | require.Eventually(t, func() bool { return out.Len() > 0 }, time.Second, 10*time.Millisecond) |
| 103 | |
| 104 | job.Stop() |
| 105 | select { |
| 106 | case <-done: |
| 107 | case <-time.After(time.Second): |
| 108 | t.Fatal("runtime metrics job did not stop") |
| 109 | } |
| 110 | |
| 111 | assert.False(t, job.running.Load()) |
| 112 | requireInOrder(t, out.String(), "HOST ''", "BEGIN") |
| 113 | }) |
| 114 | } |
| 115 | } |
| 116 | |
| 117 | func TestRuntimeMetricsJobTickSkipWhenBusy(t *testing.T) { |
| 118 | tests := map[string]struct { |
| 119 | skippedTicks int |
| 120 | }{ |
| 121 | "one tick skipped when queue is full once": { |
| 122 | skippedTicks: 1, |
| 123 | }, |
| 124 | "multiple ticks skipped while queue remains full": { |
| 125 | skippedTicks: 2, |
| 126 | }, |
| 127 | } |
| 128 | |
| 129 | for name, tc := range tests { |
| 130 | t.Run(name, func(t *testing.T) { |
| 131 | job := newRuntimeMetricsJob(&safeBuffer{}, newComponentRegistry(), nil) |
| 132 | |
| 133 | // Fill single-slot queue to force skip path. |
| 134 | job.tick <- 1 |
| 135 | for i := 0; i < tc.skippedTicks; i++ { |
| 136 | job.Tick(2 + i) |
| 137 | } |
| 138 | |
| 139 | resume := job.skipTracker.MarkRunStart(time.Now()) |
| 140 | assert.Equal(t, tc.skippedTicks, resume.Skipped) |
| 141 | }) |
| 142 | } |
| 143 | } |
| 144 | |
| 145 | func TestRuntimeMetricsJobTransactionalScenarios(t *testing.T) { |
| 146 | tests := map[string]struct { |
| 147 | run func(t *testing.T) |
| 148 | }{ |
| 149 | "flush failure does not block component state advancement": { |
| 150 | run: func(t *testing.T) { |
| 151 | reg := newComponentRegistry() |
| 152 | store := metrix.NewRuntimeStore() |
| 153 | store.Write().StatefulMeter("component").Gauge("load").Set(5) |
| 154 | |
| 155 | reg.upsert(componentSpec{ |
| 156 | Name: "component", |
| 157 | Store: store, |
| 158 | TemplateYAML: []byte(runtimeGaugeTemplateYAML()), |
| 159 | UpdateEvery: 1, |
| 160 | EmitEnv: chartemit.EmitEnv{ |
| 161 | TypeID: "netdata.go.d.internal.component", |
| 162 | UpdateEvery: 1, |
| 163 | Plugin: "go.d", |
| 164 | Module: "internal", |
| 165 | JobName: "component", |
| 166 | }, |
| 167 | }) |
| 168 | |
| 169 | job := newRuntimeMetricsJob(failingWriter{}, reg, nil) |
| 170 | job.runOnce(1) |
| 171 | |
| 172 | state := job.components["component"] |
| 173 | require.NotNil(t, state) |
| 174 | require.False(t, state.prev.IsZero()) |
| 175 | require.NotEmpty(t, state.knownCharts) |
| 176 | |
| 177 | var out safeBuffer |
| 178 | job.out = &out |
| 179 | job.runOnce(2) |
| 180 | |
| 181 | state = job.components["component"] |
| 182 | require.NotNil(t, state) |
| 183 | require.False(t, state.prev.IsZero()) |
| 184 | require.NotEmpty(t, state.knownCharts) |
| 185 | requireInOrder(t, out.String(), "HOST ''", "BEGIN") |
| 186 | assert.NotContains(t, out.String(), "CHART 'component_load'") |
| 187 | }, |
| 188 | }, |
| 189 | "effective chart tracking includes dimension only creation": { |
| 190 | run: func(t *testing.T) { |
| 191 | plan := chartengine.Plan{ |
| 192 | Actions: []chartengine.EngineAction{ |
| 193 | chartengine.CreateDimensionAction{ |
| 194 | ChartID: "component_load", |
| 195 | ChartMeta: chartengine.ChartMeta{ |
| 196 | Title: "Component Load", |
| 197 | Context: "netdata.go.plugin.component.component_load", |
| 198 | Units: "load", |
| 199 | }, |
| 200 | Name: "value", |
| 201 | }, |
| 202 | }, |
| 203 | } |
| 204 | |
| 205 | known := applyEffectiveChartSet(nil, plan) |
| 206 | require.Contains(t, known, "component_load") |
| 207 | |
| 208 | known = applyEffectiveChartSet(known, chartengine.Plan{ |
| 209 | Actions: []chartengine.EngineAction{ |
| 210 | chartengine.RemoveChartAction{ |
| 211 | ChartID: "component_load", |
| 212 | Meta: chartengine.ChartMeta{ |
| 213 | Title: "Component Load", |
| 214 | Context: "netdata.go.plugin.component.component_load", |
| 215 | Units: "load", |
| 216 | }, |
| 217 | }, |
| 218 | }, |
| 219 | }) |
| 220 | require.NotContains(t, known, "component_load") |
| 221 | }, |
| 222 | }, |
| 223 | "generation replacement keeps old state when obsolete emit fails": { |
| 224 | run: func(t *testing.T) { |
| 225 | reg := newComponentRegistry() |
| 226 | store := metrix.NewRuntimeStore() |
| 227 | store.Write().StatefulMeter("component").Gauge("load").Set(7) |
| 228 | reg.upsert(componentSpec{ |
| 229 | Name: "component", |
| 230 | Store: store, |
| 231 | TemplateYAML: []byte(runtimeGaugeTemplateYAML()), |
| 232 | UpdateEvery: 1, |
| 233 | EmitEnv: chartemit.EmitEnv{ |
| 234 | TypeID: "netdata.go.d.internal.component", |
| 235 | UpdateEvery: 1, |
| 236 | Plugin: "go.d", |
| 237 | Module: "internal", |
| 238 | JobName: "component", |
| 239 | }, |
| 240 | }) |
| 241 | |
| 242 | job := newRuntimeMetricsJob(&safeBuffer{}, reg, nil) |
| 243 | current := &runtimeComponentState{ |
| 244 | spec: componentSpec{ |
| 245 | Name: "component", |
| 246 | Generation: 0, |
| 247 | EmitEnv: chartemit.EmitEnv{ |
| 248 | TypeID: " ", |
| 249 | UpdateEvery: 1, |
| 250 | Plugin: "go.d", |
| 251 | Module: "internal", |
| 252 | JobName: "component", |
| 253 | }, |
| 254 | }, |
| 255 | prev: time.Unix(1, 0), |
| 256 | knownCharts: map[string]chartengine.ChartMeta{ |
| 257 | "component_load": { |
| 258 | Title: "Component Load", |
| 259 | Context: "netdata.go.plugin.component.component_load", |
| 260 | Units: "load", |
| 261 | }, |
| 262 | }, |
| 263 | } |
| 264 | job.components["component"] = current |
| 265 | |
| 266 | job.runOnce(1) |
| 267 | |
| 268 | require.Same(t, current, job.components["component"]) |
| 269 | assert.Equal(t, uint64(0), job.components["component"].spec.Generation) |
| 270 | }, |
| 271 | }, |
| 272 | "removal keeps old state when obsolete emit fails": { |
| 273 | run: func(t *testing.T) { |
| 274 | job := newRuntimeMetricsJob(&safeBuffer{}, newComponentRegistry(), nil) |
| 275 | current := &runtimeComponentState{ |
| 276 | spec: componentSpec{ |
| 277 | Name: "component", |
| 278 | Generation: 1, |
| 279 | EmitEnv: chartemit.EmitEnv{ |
| 280 | TypeID: " ", |
| 281 | UpdateEvery: 1, |
| 282 | Plugin: "go.d", |
| 283 | Module: "internal", |
| 284 | JobName: "component", |
| 285 | }, |
| 286 | }, |
| 287 | prev: time.Unix(1, 0), |
| 288 | knownCharts: map[string]chartengine.ChartMeta{ |
| 289 | "component_load": { |
| 290 | Title: "Component Load", |
| 291 | Context: "netdata.go.plugin.component.component_load", |
| 292 | Units: "load", |
| 293 | }, |
| 294 | }, |
| 295 | } |
| 296 | job.components["component"] = current |
| 297 | |
| 298 | job.runOnce(1) |
| 299 | |
| 300 | require.Same(t, current, job.components["component"]) |
| 301 | require.Contains(t, job.components["component"].knownCharts, "component_load") |
| 302 | }, |
| 303 | }, |
| 304 | } |
| 305 | |
| 306 | for name, tc := range tests { |
| 307 | t.Run(name, tc.run) |
| 308 | } |
| 309 | } |