| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | package runtimechartemit |
| 4 | |
| 5 | import ( |
| 6 | "fmt" |
| 7 | "io" |
| 8 | "log/slog" |
| 9 | "sort" |
| 10 | "strings" |
| 11 | "sync" |
| 12 | "time" |
| 13 | |
| 14 | "github.com/netdata/netdata/go/plugins/logger" |
| 15 | "github.com/netdata/netdata/go/plugins/pkg/ticker" |
| 16 | "github.com/netdata/netdata/go/plugins/plugin/framework/runtimecomp" |
| 17 | ) |
| 18 | |
| 19 | // Service owns runtime/internal metrics components and their cadence-driven |
| 20 | // chart emission job. |
| 21 | type Service struct { |
| 22 | *logger.Logger |
| 23 | |
| 24 | mu sync.Mutex |
| 25 | pluginName string |
| 26 | out io.Writer |
| 27 | started bool |
| 28 | tickEvery time.Duration |
| 29 | |
| 30 | registry *componentRegistry |
| 31 | job *runtimeMetricsJob |
| 32 | producers map[string]runtimeProducer |
| 33 | tkStop chan struct{} |
| 34 | tkDone chan struct{} |
| 35 | } |
| 36 | |
| 37 | func New(log *logger.Logger) *Service { |
| 38 | if log == nil { |
| 39 | log = logger.New().With(slog.String("component", "runtime metrics service")) |
| 40 | } |
| 41 | return &Service{ |
| 42 | Logger: log, |
| 43 | registry: newComponentRegistry(), |
| 44 | producers: make(map[string]runtimeProducer), |
| 45 | tickEvery: time.Second, |
| 46 | } |
| 47 | } |
| 48 | |
| 49 | // SetTickEvery updates the service scheduler interval. Non-positive values are |
| 50 | // ignored. Changing this value while running does not affect the current ticker |
| 51 | // and applies on next Start. |
| 52 | func (s *Service) SetTickEvery(interval time.Duration) { |
| 53 | if s == nil || interval <= 0 { |
| 54 | return |
| 55 | } |
| 56 | s.mu.Lock() |
| 57 | s.tickEvery = interval |
| 58 | s.mu.Unlock() |
| 59 | } |
| 60 | |
| 61 | // Start initializes the runtime metrics service and starts the runtime emitter |
| 62 | // job. Calling Start multiple times is safe. |
| 63 | func (s *Service) Start(pluginName string, out io.Writer) { |
| 64 | if s == nil { |
| 65 | return |
| 66 | } |
| 67 | |
| 68 | s.mu.Lock() |
| 69 | if p := strings.TrimSpace(pluginName); p != "" { |
| 70 | s.pluginName = p |
| 71 | } |
| 72 | if out != nil { |
| 73 | s.out = out |
| 74 | } |
| 75 | if s.out == nil { |
| 76 | s.out = io.Discard |
| 77 | } |
| 78 | if s.started { |
| 79 | s.mu.Unlock() |
| 80 | return |
| 81 | } |
| 82 | |
| 83 | s.job = newRuntimeMetricsJob( |
| 84 | s.out, |
| 85 | s.registry, |
| 86 | s.Logger.With(slog.String("component", "runtime metrics job")), |
| 87 | ) |
| 88 | job := s.job |
| 89 | stopCh := make(chan struct{}) |
| 90 | doneCh := make(chan struct{}) |
| 91 | interval := s.tickEvery |
| 92 | if interval <= 0 { |
| 93 | interval = time.Second |
| 94 | } |
| 95 | s.tkStop = stopCh |
| 96 | s.tkDone = doneCh |
| 97 | s.started = true |
| 98 | s.mu.Unlock() |
| 99 | |
| 100 | go job.Start() |
| 101 | go s.runTicker(interval, stopCh, doneCh) |
| 102 | } |
| 103 | |
| 104 | // Stop terminates the runtime emitter job. Calling Stop multiple times is safe. |
| 105 | func (s *Service) Stop() { |
| 106 | if s == nil { |
| 107 | return |
| 108 | } |
| 109 | |
| 110 | s.mu.Lock() |
| 111 | if !s.started { |
| 112 | s.mu.Unlock() |
| 113 | return |
| 114 | } |
| 115 | job := s.job |
| 116 | stopCh := s.tkStop |
| 117 | doneCh := s.tkDone |
| 118 | s.job = nil |
| 119 | s.tkStop = nil |
| 120 | s.tkDone = nil |
| 121 | s.started = false |
| 122 | s.mu.Unlock() |
| 123 | |
| 124 | if stopCh != nil { |
| 125 | close(stopCh) |
| 126 | } |
| 127 | if doneCh != nil { |
| 128 | <-doneCh |
| 129 | } |
| 130 | if job != nil { |
| 131 | job.Stop() |
| 132 | } |
| 133 | } |
| 134 | |
| 135 | // Tick advances runtime producers and runtime emitter job on scheduler cadence. |
| 136 | func (s *Service) Tick(clock int) { |
| 137 | if s == nil { |
| 138 | return |
| 139 | } |
| 140 | |
| 141 | s.mu.Lock() |
| 142 | if !s.started { |
| 143 | s.mu.Unlock() |
| 144 | return |
| 145 | } |
| 146 | producers := make([]runtimeProducer, 0, len(s.producers)) |
| 147 | for _, producer := range s.producers { |
| 148 | producers = append(producers, producer) |
| 149 | } |
| 150 | sort.Slice(producers, func(i, j int) bool { |
| 151 | return producers[i].Name() < producers[j].Name() |
| 152 | }) |
| 153 | job := s.job |
| 154 | s.mu.Unlock() |
| 155 | |
| 156 | for _, producer := range producers { |
| 157 | if producer == nil { |
| 158 | continue |
| 159 | } |
| 160 | if err := producer.Tick(); err != nil { |
| 161 | s.Warningf("runtime producer %q tick failed: %v", producer.Name(), err) |
| 162 | } |
| 163 | } |
| 164 | if job != nil { |
| 165 | job.Tick(clock) |
| 166 | } |
| 167 | } |
| 168 | |
| 169 | func (s *Service) runTicker(interval time.Duration, stop <-chan struct{}, done chan<- struct{}) { |
| 170 | defer close(done) |
| 171 | |
| 172 | tk := ticker.New(interval) |
| 173 | defer tk.Stop() |
| 174 | |
| 175 | for { |
| 176 | select { |
| 177 | case <-stop: |
| 178 | return |
| 179 | case clock := <-tk.C: |
| 180 | s.Tick(clock) |
| 181 | } |
| 182 | } |
| 183 | } |
| 184 | |
| 185 | func (s *Service) RegisterComponent(cfg runtimecomp.ComponentConfig) error { |
| 186 | if s == nil { |
| 187 | return fmt.Errorf("runtimemgr: nil service") |
| 188 | } |
| 189 | |
| 190 | s.mu.Lock() |
| 191 | pluginName := firstNotEmpty(s.pluginName, "go.d") |
| 192 | s.mu.Unlock() |
| 193 | |
| 194 | spec, err := normalizeComponent(cfg, pluginName) |
| 195 | if err != nil { |
| 196 | return err |
| 197 | } |
| 198 | s.registry.upsert(spec) |
| 199 | return nil |
| 200 | } |
| 201 | |
| 202 | func (s *Service) UnregisterComponent(name string) { |
| 203 | if s == nil { |
| 204 | return |
| 205 | } |
| 206 | name = strings.TrimSpace(name) |
| 207 | if name == "" { |
| 208 | return |
| 209 | } |
| 210 | s.registry.remove(name) |
| 211 | } |
| 212 | |
| 213 | func (s *Service) RegisterProducer(name string, tickFn func() error) error { |
| 214 | if s == nil { |
| 215 | return fmt.Errorf("runtimemgr: nil service") |
| 216 | } |
| 217 | |
| 218 | name = strings.TrimSpace(name) |
| 219 | if name == "" { |
| 220 | return fmt.Errorf("runtimemgr: runtime producer name is required") |
| 221 | } |
| 222 | if tickFn == nil { |
| 223 | return fmt.Errorf("runtimemgr: runtime producer %q tick function is required", name) |
| 224 | } |
| 225 | |
| 226 | s.mu.Lock() |
| 227 | if s.producers == nil { |
| 228 | s.producers = make(map[string]runtimeProducer) |
| 229 | } |
| 230 | s.producers[name] = runtimeProducerFunc{ |
| 231 | name: name, |
| 232 | tick: tickFn, |
| 233 | } |
| 234 | s.mu.Unlock() |
| 235 | return nil |
| 236 | } |
| 237 | |
| 238 | func (s *Service) UnregisterProducer(name string) { |
| 239 | if s == nil { |
| 240 | return |
| 241 | } |
| 242 | name = strings.TrimSpace(name) |
| 243 | if name == "" { |
| 244 | return |
| 245 | } |
| 246 | s.mu.Lock() |
| 247 | delete(s.producers, name) |
| 248 | s.mu.Unlock() |
| 249 | } |