| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | package runtimechartemit |
| 4 | |
| 5 | import ( |
| 6 | "bytes" |
| 7 | "fmt" |
| 8 | "io" |
| 9 | "log/slog" |
| 10 | "maps" |
| 11 | "sort" |
| 12 | "sync/atomic" |
| 13 | "time" |
| 14 | |
| 15 | "github.com/netdata/netdata/go/plugins/logger" |
| 16 | "github.com/netdata/netdata/go/plugins/pkg/metrix" |
| 17 | "github.com/netdata/netdata/go/plugins/pkg/netdataapi" |
| 18 | "github.com/netdata/netdata/go/plugins/plugin/framework/chartemit" |
| 19 | "github.com/netdata/netdata/go/plugins/plugin/framework/chartengine" |
| 20 | "github.com/netdata/netdata/go/plugins/plugin/framework/tickstate" |
| 21 | ) |
| 22 | |
| 23 | type runtimeComponentState struct { |
| 24 | spec componentSpec |
| 25 | engine *chartengine.Engine |
| 26 | prev time.Time |
| 27 | knownCharts map[string]chartengine.ChartMeta |
| 28 | } |
| 29 | |
| 30 | type runtimePreparedEmission struct { |
| 31 | attempt chartengine.PlanAttempt |
| 32 | nextPrev time.Time |
| 33 | nextKnownCharts map[string]chartengine.ChartMeta |
| 34 | } |
| 35 | |
| 36 | type runtimeTickCommitStep struct { |
| 37 | attempt chartengine.PlanAttempt |
| 38 | finalize func() |
| 39 | } |
| 40 | |
| 41 | type runtimeTickCommit struct { |
| 42 | steps []runtimeTickCommitStep |
| 43 | } |
| 44 | |
| 45 | func (c *runtimeTickCommit) add(step runtimeTickCommitStep) { |
| 46 | c.steps = append(c.steps, step) |
| 47 | } |
| 48 | |
| 49 | func (c *runtimeTickCommit) abort() { |
| 50 | for _, step := range c.steps { |
| 51 | step.attempt.Abort() |
| 52 | } |
| 53 | } |
| 54 | |
| 55 | func (c *runtimeTickCommit) commit() error { |
| 56 | for i, step := range c.steps { |
| 57 | if err := step.attempt.Commit(); err != nil { |
| 58 | for _, remaining := range c.steps[i+1:] { |
| 59 | remaining.attempt.Abort() |
| 60 | } |
| 61 | return err |
| 62 | } |
| 63 | if step.finalize != nil { |
| 64 | step.finalize() |
| 65 | } |
| 66 | } |
| 67 | return nil |
| 68 | } |
| 69 | |
| 70 | type runtimeMetricsJob struct { |
| 71 | *logger.Logger |
| 72 | |
| 73 | out io.Writer |
| 74 | registry *componentRegistry |
| 75 | |
| 76 | running atomic.Bool |
| 77 | stop chan struct{} |
| 78 | tick chan int |
| 79 | |
| 80 | buf *bytes.Buffer |
| 81 | api *netdataapi.API |
| 82 | |
| 83 | components map[string]*runtimeComponentState |
| 84 | skipTracker tickstate.SkipTracker |
| 85 | } |
| 86 | |
| 87 | func newRuntimeMetricsJob(out io.Writer, reg *componentRegistry, log *logger.Logger) *runtimeMetricsJob { |
| 88 | if out == nil { |
| 89 | out = io.Discard |
| 90 | } |
| 91 | if log == nil { |
| 92 | log = logger.New().With(slog.String("component", "runtime-metrics-job")) |
| 93 | } |
| 94 | var buf bytes.Buffer |
| 95 | return &runtimeMetricsJob{ |
| 96 | Logger: log, |
| 97 | out: out, |
| 98 | registry: reg, |
| 99 | stop: make(chan struct{}), |
| 100 | tick: make(chan int, 1), |
| 101 | buf: &buf, |
| 102 | api: netdataapi.New(&buf), |
| 103 | components: make(map[string]*runtimeComponentState), |
| 104 | } |
| 105 | } |
| 106 | |
| 107 | func (j *runtimeMetricsJob) Tick(clock int) { |
| 108 | select { |
| 109 | case j.tick <- clock: |
| 110 | default: |
| 111 | skip := j.skipTracker.MarkSkipped() |
| 112 | |
| 113 | if skip.RunStarted.IsZero() { |
| 114 | j.Warning("skipping runtime metrics tick: waiting for first run to start") |
| 115 | return |
| 116 | } |
| 117 | if skip.Count == 1 { |
| 118 | j.Warningf("skipping runtime metrics tick: previous run still in progress for %s", time.Since(skip.RunStarted)) |
| 119 | return |
| 120 | } |
| 121 | j.Debugf("skipping runtime metrics tick: previous run still in progress for %s (skipped %d ticks)", time.Since(skip.RunStarted), skip.Count) |
| 122 | } |
| 123 | } |
| 124 | |
| 125 | func (j *runtimeMetricsJob) Start() { |
| 126 | j.running.Store(true) |
| 127 | j.Info("runtime metrics job started") |
| 128 | defer func() { |
| 129 | j.running.Store(false) |
| 130 | j.Info("runtime metrics job stopped") |
| 131 | }() |
| 132 | |
| 133 | LOOP: |
| 134 | for { |
| 135 | select { |
| 136 | case <-j.stop: |
| 137 | break LOOP |
| 138 | case clock := <-j.tick: |
| 139 | resume := j.skipTracker.MarkRunStart(time.Now()) |
| 140 | if resume.Skipped > 0 { |
| 141 | if resume.RunStopped.IsZero() || resume.RunStarted.IsZero() { |
| 142 | j.Infof("runtime metrics tick resumed (skipped %d ticks)", resume.Skipped) |
| 143 | } else { |
| 144 | j.Infof( |
| 145 | "runtime metrics tick resumed after %s (skipped %d ticks)", |
| 146 | resume.RunStopped.Sub(resume.RunStarted), |
| 147 | resume.Skipped, |
| 148 | ) |
| 149 | } |
| 150 | } |
| 151 | j.runOnce(clock) |
| 152 | j.skipTracker.MarkRunStop(time.Now()) |
| 153 | } |
| 154 | } |
| 155 | j.stop <- struct{}{} |
| 156 | } |
| 157 | |
| 158 | func (j *runtimeMetricsJob) Stop() { |
| 159 | j.stop <- struct{}{} |
| 160 | <-j.stop |
| 161 | } |
| 162 | |
| 163 | func (j *runtimeMetricsJob) runOnce(clock int) { |
| 164 | specs := j.registry.snapshot() |
| 165 | seen := make(map[string]struct{}, len(specs)) |
| 166 | now := time.Now() |
| 167 | var commit runtimeTickCommit |
| 168 | |
| 169 | for _, spec := range specs { |
| 170 | seen[spec.Name] = struct{}{} |
| 171 | |
| 172 | if spec.UpdateEvery > 1 && clock%spec.UpdateEvery != 0 { |
| 173 | continue |
| 174 | } |
| 175 | |
| 176 | step, buf, ok := j.prepareComponentStep(spec, now) |
| 177 | if !ok { |
| 178 | continue |
| 179 | } |
| 180 | if buf != nil && buf.Len() > 0 { |
| 181 | _, _ = j.buf.Write(buf.Bytes()) |
| 182 | } |
| 183 | commit.add(step) |
| 184 | } |
| 185 | |
| 186 | for name := range j.components { |
| 187 | if _, ok := seen[name]; ok { |
| 188 | continue |
| 189 | } |
| 190 | state := j.components[name] |
| 191 | step, buf, ok := j.prepareRemovalStep(name, state) |
| 192 | if !ok { |
| 193 | continue |
| 194 | } |
| 195 | if buf != nil && buf.Len() > 0 { |
| 196 | _, _ = j.buf.Write(buf.Bytes()) |
| 197 | } |
| 198 | commit.add(step) |
| 199 | } |
| 200 | |
| 201 | if j.buf.Len() > 0 { |
| 202 | _, _ = io.Copy(j.out, j.buf) |
| 203 | } |
| 204 | if err := commit.commit(); err != nil { |
| 205 | j.Warningf("runtime metrics commit failed: %v", err) |
| 206 | } |
| 207 | j.buf.Reset() |
| 208 | } |
| 209 | |
| 210 | func (j *runtimeMetricsJob) newComponentState(spec componentSpec) (*runtimeComponentState, error) { |
| 211 | engineLog := j.Logger.With(slog.String("runtime_component", spec.Name)) |
| 212 | engine, err := chartengine.New( |
| 213 | chartengine.WithRuntimeStore(nil), // Two-engine policy: observer engine has no self-metrics. |
| 214 | chartengine.WithSeriesSelectionAllVisible(), |
| 215 | chartengine.WithRuntimePlannerMode(), |
| 216 | chartengine.WithEmitTypeIDBudgetPrefix(spec.EmitEnv.TypeID), |
| 217 | chartengine.WithEnginePolicy(chartengine.EnginePolicy{Autogen: &spec.Autogen}), |
| 218 | chartengine.WithLogger(engineLog), |
| 219 | ) |
| 220 | if err != nil { |
| 221 | return nil, fmt.Errorf("create engine: %w", err) |
| 222 | } |
| 223 | if err := engine.LoadYAML(spec.TemplateYAML, spec.Generation); err != nil { |
| 224 | return nil, fmt.Errorf("load template: %w", err) |
| 225 | } |
| 226 | |
| 227 | state := &runtimeComponentState{ |
| 228 | spec: spec, |
| 229 | engine: engine, |
| 230 | knownCharts: make(map[string]chartengine.ChartMeta), |
| 231 | } |
| 232 | return state, nil |
| 233 | } |
| 234 | |
| 235 | func (j *runtimeMetricsJob) prepareComponentStep(spec componentSpec, now time.Time) (runtimeTickCommitStep, *bytes.Buffer, bool) { |
| 236 | current := j.components[spec.Name] |
| 237 | if current != nil && current.spec.Generation == spec.Generation { |
| 238 | emission, buf, ok := j.prepareEmissionToBuffer(current, now) |
| 239 | if !ok { |
| 240 | return runtimeTickCommitStep{}, nil, false |
| 241 | } |
| 242 | return runtimeTickCommitStep{ |
| 243 | attempt: emission.attempt, |
| 244 | finalize: func() { |
| 245 | current.prev = emission.nextPrev |
| 246 | current.knownCharts = emission.nextKnownCharts |
| 247 | }, |
| 248 | }, buf, true |
| 249 | } |
| 250 | |
| 251 | next, err := j.newComponentState(spec) |
| 252 | if err != nil { |
| 253 | j.Warningf("runtime metrics component %q init failed: %v", spec.Name, err) |
| 254 | return runtimeTickCommitStep{}, nil, false |
| 255 | } |
| 256 | |
| 257 | var buf bytes.Buffer |
| 258 | api := netdataapi.New(&buf) |
| 259 | if err := j.emitComponentObsolete(api, current); err != nil { |
| 260 | j.Warningf("runtime metrics component %q obsolete emit failed: %v", spec.Name, err) |
| 261 | return runtimeTickCommitStep{}, nil, false |
| 262 | } |
| 263 | emission, ok := j.prepareEmission(api, next, now) |
| 264 | if !ok { |
| 265 | return runtimeTickCommitStep{}, nil, false |
| 266 | } |
| 267 | return runtimeTickCommitStep{ |
| 268 | attempt: emission.attempt, |
| 269 | finalize: func() { |
| 270 | next.prev = emission.nextPrev |
| 271 | next.knownCharts = emission.nextKnownCharts |
| 272 | j.components[spec.Name] = next |
| 273 | }, |
| 274 | }, &buf, true |
| 275 | } |
| 276 | |
| 277 | func (j *runtimeMetricsJob) prepareRemovalStep(name string, state *runtimeComponentState) (runtimeTickCommitStep, *bytes.Buffer, bool) { |
| 278 | if state == nil { |
| 279 | return runtimeTickCommitStep{}, nil, false |
| 280 | } |
| 281 | |
| 282 | var buf bytes.Buffer |
| 283 | api := netdataapi.New(&buf) |
| 284 | if err := j.emitComponentObsolete(api, state); err != nil { |
| 285 | j.Warningf("runtime metrics component %q obsolete emit failed: %v", state.spec.Name, err) |
| 286 | return runtimeTickCommitStep{}, nil, false |
| 287 | } |
| 288 | return runtimeTickCommitStep{ |
| 289 | finalize: func() { |
| 290 | delete(j.components, name) |
| 291 | }, |
| 292 | }, &buf, true |
| 293 | } |
| 294 | |
| 295 | func (j *runtimeMetricsJob) prepareEmissionToBuffer(state *runtimeComponentState, now time.Time) (runtimePreparedEmission, *bytes.Buffer, bool) { |
| 296 | var buf bytes.Buffer |
| 297 | api := netdataapi.New(&buf) |
| 298 | emission, ok := j.prepareEmission(api, state, now) |
| 299 | if !ok { |
| 300 | return runtimePreparedEmission{}, nil, false |
| 301 | } |
| 302 | return emission, &buf, true |
| 303 | } |
| 304 | |
| 305 | func (j *runtimeMetricsJob) prepareEmission(api *netdataapi.API, state *runtimeComponentState, now time.Time) (runtimePreparedEmission, bool) { |
| 306 | if state == nil { |
| 307 | return runtimePreparedEmission{}, false |
| 308 | } |
| 309 | |
| 310 | reader := state.spec.Store.Read(metrix.ReadRaw(), metrix.ReadFlatten()) |
| 311 | attempt, err := state.engine.PreparePlan(reader) |
| 312 | if err != nil { |
| 313 | j.Warningf("runtime metrics component %q build plan failed: %v", state.spec.Name, err) |
| 314 | return runtimePreparedEmission{}, false |
| 315 | } |
| 316 | |
| 317 | plan := attempt.Plan() |
| 318 | env := cloneEmitEnv(state.spec.EmitEnv) |
| 319 | env.MSSinceLast = calcRuntimeSinceLast(now, state.prev) |
| 320 | if err := chartemit.ApplyPlan(api, plan, env); err != nil { |
| 321 | attempt.Abort() |
| 322 | j.Warningf("runtime metrics component %q apply plan failed: %v", state.spec.Name, err) |
| 323 | return runtimePreparedEmission{}, false |
| 324 | } |
| 325 | |
| 326 | return runtimePreparedEmission{ |
| 327 | attempt: attempt, |
| 328 | nextPrev: now, |
| 329 | nextKnownCharts: applyEffectiveChartSet(state.knownCharts, plan), |
| 330 | }, true |
| 331 | } |
| 332 | |
| 333 | func (j *runtimeMetricsJob) emitComponentObsolete(api *netdataapi.API, state *runtimeComponentState) error { |
| 334 | if state == nil || len(state.knownCharts) == 0 { |
| 335 | return nil |
| 336 | } |
| 337 | |
| 338 | chartIDs := make([]string, 0, len(state.knownCharts)) |
| 339 | for chartID := range state.knownCharts { |
| 340 | chartIDs = append(chartIDs, chartID) |
| 341 | } |
| 342 | sort.Strings(chartIDs) |
| 343 | |
| 344 | actions := make([]chartengine.EngineAction, 0, len(chartIDs)) |
| 345 | for _, chartID := range chartIDs { |
| 346 | meta := state.knownCharts[chartID] |
| 347 | actions = append(actions, chartengine.RemoveChartAction{ |
| 348 | ChartID: chartID, |
| 349 | Meta: meta, |
| 350 | }) |
| 351 | } |
| 352 | |
| 353 | env := cloneEmitEnv(state.spec.EmitEnv) |
| 354 | env.MSSinceLast = 0 |
| 355 | return chartemit.ApplyPlan(api, chartengine.Plan{Actions: actions}, env) |
| 356 | } |
| 357 | |
| 358 | func applyEffectiveChartSet(known map[string]chartengine.ChartMeta, plan chartengine.Plan) map[string]chartengine.ChartMeta { |
| 359 | out := maps.Clone(known) |
| 360 | if out == nil { |
| 361 | out = make(map[string]chartengine.ChartMeta) |
| 362 | } |
| 363 | |
| 364 | createCharts := make(map[string]chartengine.ChartMeta) |
| 365 | dimensionOnlyCharts := make(map[string]chartengine.ChartMeta) |
| 366 | for _, action := range plan.Actions { |
| 367 | switch v := action.(type) { |
| 368 | case chartengine.CreateChartAction: |
| 369 | createCharts[v.ChartID] = v.Meta |
| 370 | case chartengine.CreateDimensionAction: |
| 371 | if _, ok := createCharts[v.ChartID]; ok { |
| 372 | continue |
| 373 | } |
| 374 | if _, ok := dimensionOnlyCharts[v.ChartID]; !ok { |
| 375 | dimensionOnlyCharts[v.ChartID] = v.ChartMeta |
| 376 | } |
| 377 | case chartengine.RemoveChartAction: |
| 378 | delete(out, v.ChartID) |
| 379 | } |
| 380 | } |
| 381 | maps.Copy(out, createCharts) |
| 382 | for chartID, meta := range dimensionOnlyCharts { |
| 383 | if _, ok := out[chartID]; ok { |
| 384 | continue |
| 385 | } |
| 386 | out[chartID] = meta |
| 387 | } |
| 388 | return out |
| 389 | } |
| 390 | |
| 391 | func calcRuntimeSinceLast(cur, prev time.Time) int { |
| 392 | if prev.IsZero() { |
| 393 | return 0 |
| 394 | } |
| 395 | // Keep parity with module.Job calcSinceLastRun() units. |
| 396 | return int((cur.UnixNano() - prev.UnixNano()) / 1000) |
| 397 | } |