| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | package sd |
| 4 | |
| 5 | import ( |
| 6 | "context" |
| 7 | "fmt" |
| 8 | "maps" |
| 9 | "sync" |
| 10 | "time" |
| 11 | |
| 12 | "github.com/netdata/netdata/go/plugins/logger" |
| 13 | "github.com/netdata/netdata/go/plugins/plugin/agent/discovery/sd/pipeline" |
| 14 | "github.com/netdata/netdata/go/plugins/plugin/framework/confgroup" |
| 15 | "github.com/netdata/netdata/go/plugins/plugin/framework/dyncfg" |
| 16 | ) |
| 17 | |
| 18 | const ( |
| 19 | restartGracePeriod = 1 * time.Minute |
| 20 | ) |
| 21 | |
| 22 | // PipelineManager manages the lifecycle of discovery pipelines. |
| 23 | // It handles starting, stopping, and restarting pipelines, tracks sources |
| 24 | // for cleanup, and implements a grace period mechanism for restarts. |
| 25 | type PipelineManager struct { |
| 26 | *logger.Logger |
| 27 | |
| 28 | newPipeline func(cfg pipeline.Config) (sdPipeline, error) |
| 29 | send func(ctx context.Context, groups []*confgroup.Group) |
| 30 | |
| 31 | mux sync.Mutex |
| 32 | pipelines map[string]*runningPipeline // [pipelineKey] |
| 33 | pipelineSources map[string]map[string]struct{} // [pipelineKey][source] |
| 34 | pendingRemovals map[string]*pendingRemoval // [pipelineKey] |
| 35 | } |
| 36 | |
| 37 | type runningPipeline struct { |
| 38 | cfg pipeline.Config |
| 39 | cancel context.CancelFunc |
| 40 | done chan struct{} |
| 41 | } |
| 42 | |
| 43 | type pendingRemoval struct { |
| 44 | sources map[string]struct{} |
| 45 | timestamp time.Time |
| 46 | } |
| 47 | |
| 48 | // NewPipelineManager creates a new PipelineManager. |
| 49 | func NewPipelineManager( |
| 50 | log *logger.Logger, |
| 51 | newPipeline func(cfg pipeline.Config) (sdPipeline, error), |
| 52 | send func(ctx context.Context, groups []*confgroup.Group), |
| 53 | ) *PipelineManager { |
| 54 | return &PipelineManager{ |
| 55 | Logger: log, |
| 56 | newPipeline: newPipeline, |
| 57 | send: send, |
| 58 | pipelines: make(map[string]*runningPipeline), |
| 59 | pipelineSources: make(map[string]map[string]struct{}), |
| 60 | pendingRemovals: make(map[string]*pendingRemoval), |
| 61 | } |
| 62 | } |
| 63 | |
| 64 | // Start starts a new pipeline with the given key and config. |
| 65 | // If a pipeline with the same key is already running, it will be stopped first. |
| 66 | func (m *PipelineManager) Start(ctx context.Context, key string, cfg pipeline.Config) error { |
| 67 | m.mux.Lock() |
| 68 | |
| 69 | // Stop existing pipeline if any (no grace period - this is initial start or replace) |
| 70 | sp := m.removePipelineLocked(key, true) |
| 71 | |
| 72 | m.mux.Unlock() |
| 73 | |
| 74 | // Wait for old pipeline and cleanup outside the lock |
| 75 | if sp != nil { |
| 76 | m.waitAndCleanup(key, sp) |
| 77 | } |
| 78 | |
| 79 | m.mux.Lock() |
| 80 | defer m.mux.Unlock() |
| 81 | |
| 82 | return m.startPipelineLocked(ctx, key, cfg) |
| 83 | } |
| 84 | |
| 85 | // Stop stops a pipeline and sends removal groups for all its tracked sources. |
| 86 | func (m *PipelineManager) Stop(key string) { |
| 87 | m.mux.Lock() |
| 88 | sp := m.removePipelineLocked(key, true) |
| 89 | m.mux.Unlock() |
| 90 | |
| 91 | // Wait for pipeline and cleanup outside the lock |
| 92 | if sp != nil { |
| 93 | m.waitAndCleanup(key, sp) |
| 94 | } |
| 95 | } |
| 96 | |
| 97 | // Restart stops a pipeline and starts it with new config, using grace period |
| 98 | // to avoid removing discovered jobs that will be re-discovered. |
| 99 | func (m *PipelineManager) Restart(ctx context.Context, key string, cfg pipeline.Config) error { |
| 100 | // Validate new config first by creating the pipeline (outside lock) |
| 101 | pl, err := m.newPipeline(cfg) |
| 102 | if err != nil { |
| 103 | return dyncfg.MarkNonDisruptiveUpdate(fmt.Errorf("failed to create new pipeline config: %w", err)) |
| 104 | } |
| 105 | |
| 106 | m.mux.Lock() |
| 107 | |
| 108 | // Mark current sources as pending removal (grace period) |
| 109 | // Merge with existing pending removals to avoid losing sources from previous restarts |
| 110 | if sources, ok := m.pipelineSources[key]; ok && len(sources) > 0 { |
| 111 | if existing, ok := m.pendingRemovals[key]; ok { |
| 112 | // Merge: add current sources to existing pending removals |
| 113 | for src := range sources { |
| 114 | existing.sources[src] = struct{}{} |
| 115 | } |
| 116 | existing.timestamp = time.Now() // Reset grace period |
| 117 | m.Debugf("pipeline '%s': merged %d sources into pending removal (now %d total)", key, len(sources), len(existing.sources)) |
| 118 | } else { |
| 119 | m.pendingRemovals[key] = &pendingRemoval{ |
| 120 | sources: copySourcesMap(sources), |
| 121 | timestamp: time.Now(), |
| 122 | } |
| 123 | m.Debugf("pipeline '%s': marked %d sources for pending removal (grace period)", key, len(sources)) |
| 124 | } |
| 125 | } |
| 126 | |
| 127 | // Stop old pipeline without cleanup (sources are pending, not removed) |
| 128 | sp := m.removePipelineLocked(key, false) |
| 129 | |
| 130 | m.mux.Unlock() |
| 131 | |
| 132 | // Wait for old pipeline outside the lock |
| 133 | if sp != nil { |
| 134 | m.waitForPipeline(key, sp) |
| 135 | } |
| 136 | |
| 137 | m.mux.Lock() |
| 138 | defer m.mux.Unlock() |
| 139 | |
| 140 | // Start the already-created new pipeline |
| 141 | return m.startPipelineWithInstanceLocked(ctx, key, cfg, pl) |
| 142 | } |
| 143 | |
| 144 | // StopAll stops all running pipelines with cleanup. |
| 145 | func (m *PipelineManager) StopAll() { |
| 146 | // Collect and remove all pipelines while holding the lock |
| 147 | m.mux.Lock() |
| 148 | toStop := make(map[string]*stoppedPipeline, len(m.pipelines)) |
| 149 | for key := range m.pipelines { |
| 150 | if sp := m.removePipelineLocked(key, true); sp != nil { |
| 151 | toStop[key] = sp |
| 152 | } |
| 153 | } |
| 154 | m.mux.Unlock() |
| 155 | |
| 156 | // Wait for all pipelines and cleanup outside the lock |
| 157 | for key, sp := range toStop { |
| 158 | m.waitAndCleanup(key, sp) |
| 159 | } |
| 160 | } |
| 161 | |
| 162 | // RunGracePeriodCleanup runs the grace period cleanup loop. |
| 163 | // It should be called as a goroutine and will run until ctx is cancelled. |
| 164 | func (m *PipelineManager) RunGracePeriodCleanup(ctx context.Context) { |
| 165 | tk := time.NewTicker(5 * time.Second) |
| 166 | defer tk.Stop() |
| 167 | |
| 168 | for { |
| 169 | select { |
| 170 | case <-ctx.Done(): |
| 171 | return |
| 172 | case <-tk.C: |
| 173 | m.processGracePeriodRemovals(ctx) |
| 174 | } |
| 175 | } |
| 176 | } |
| 177 | |
| 178 | // IsRunning returns true if a pipeline with the given key is running. |
| 179 | func (m *PipelineManager) IsRunning(key string) bool { |
| 180 | m.mux.Lock() |
| 181 | defer m.mux.Unlock() |
| 182 | |
| 183 | _, ok := m.pipelines[key] |
| 184 | return ok |
| 185 | } |
| 186 | |
| 187 | // Keys returns the keys of all running pipelines. |
| 188 | func (m *PipelineManager) Keys() []string { |
| 189 | m.mux.Lock() |
| 190 | defer m.mux.Unlock() |
| 191 | |
| 192 | keys := make([]string, 0, len(m.pipelines)) |
| 193 | for k := range m.pipelines { |
| 194 | keys = append(keys, k) |
| 195 | } |
| 196 | return keys |
| 197 | } |
| 198 | |
| 199 | func (m *PipelineManager) startPipelineLocked(ctx context.Context, key string, cfg pipeline.Config) error { |
| 200 | pl, err := m.newPipeline(cfg) |
| 201 | if err != nil { |
| 202 | return err |
| 203 | } |
| 204 | |
| 205 | return m.startPipelineWithInstanceLocked(ctx, key, cfg, pl) |
| 206 | } |
| 207 | |
| 208 | func (m *PipelineManager) startPipelineWithInstanceLocked(ctx context.Context, key string, cfg pipeline.Config, pl sdPipeline) error { |
| 209 | // No check for existing pipeline needed here: |
| 210 | // All operations for the same pipeline key are processed sequentially |
| 211 | // in ServiceDiscovery.run()'s select loop (both file config events and |
| 212 | // dyncfg commands), so concurrent Start/Restart calls for the same key |
| 213 | // cannot occur. |
| 214 | |
| 215 | plCtx, cancel := context.WithCancel(ctx) |
| 216 | done := make(chan struct{}) |
| 217 | |
| 218 | rp := &runningPipeline{ |
| 219 | cfg: cfg, |
| 220 | cancel: cancel, |
| 221 | done: done, |
| 222 | } |
| 223 | |
| 224 | m.pipelines[key] = rp |
| 225 | m.pipelineSources[key] = make(map[string]struct{}) |
| 226 | |
| 227 | go func() { |
| 228 | defer close(done) |
| 229 | m.runPipeline(plCtx, key, pl) |
| 230 | }() |
| 231 | |
| 232 | m.Infof("pipeline '%s' started", key) |
| 233 | return nil |
| 234 | } |
| 235 | |
| 236 | func (m *PipelineManager) runPipeline(ctx context.Context, key string, pl sdPipeline) { |
| 237 | groups := make(chan []*confgroup.Group) |
| 238 | done := make(chan struct{}) |
| 239 | |
| 240 | go func() { |
| 241 | defer close(done) |
| 242 | pl.Run(ctx, groups) |
| 243 | }() |
| 244 | |
| 245 | for { |
| 246 | select { |
| 247 | case <-ctx.Done(): |
| 248 | select { |
| 249 | case <-done: |
| 250 | case <-time.After(10 * time.Second): |
| 251 | m.Warningf("pipeline '%s': timeout waiting for shutdown", key) |
| 252 | } |
| 253 | return |
| 254 | case <-done: |
| 255 | return |
| 256 | case grps := <-groups: |
| 257 | m.onGroupsReceived(ctx, key, grps) |
| 258 | } |
| 259 | } |
| 260 | } |
| 261 | |
| 262 | func (m *PipelineManager) onGroupsReceived(ctx context.Context, key string, groups []*confgroup.Group) { |
| 263 | m.mux.Lock() |
| 264 | |
| 265 | // Ignore groups if pipeline is no longer tracked (was removed) |
| 266 | if _, exists := m.pipelines[key]; !exists { |
| 267 | m.mux.Unlock() |
| 268 | return |
| 269 | } |
| 270 | |
| 271 | // Track sources |
| 272 | for _, grp := range groups { |
| 273 | if m.pipelineSources[key] == nil { |
| 274 | m.pipelineSources[key] = make(map[string]struct{}) |
| 275 | } |
| 276 | m.pipelineSources[key][grp.Source] = struct{}{} |
| 277 | |
| 278 | // Cancel pending removal for re-discovered sources |
| 279 | if pending, ok := m.pendingRemovals[key]; ok { |
| 280 | if _, wasPending := pending.sources[grp.Source]; wasPending { |
| 281 | delete(pending.sources, grp.Source) |
| 282 | m.Debugf("pipeline '%s': source '%s' re-discovered, cancelled pending removal", key, grp.Source) |
| 283 | } |
| 284 | } |
| 285 | } |
| 286 | |
| 287 | m.mux.Unlock() |
| 288 | |
| 289 | // Forward groups |
| 290 | m.send(ctx, groups) |
| 291 | } |
| 292 | |
| 293 | // stoppedPipeline holds info needed to complete pipeline shutdown outside the lock. |
| 294 | type stoppedPipeline struct { |
| 295 | rp *runningPipeline |
| 296 | sourcesToRemove []string |
| 297 | } |
| 298 | |
| 299 | // removePipelineLocked removes a pipeline from the map, cancels it, and optionally |
| 300 | // collects sources for cleanup. Returns info needed to complete shutdown outside the lock. |
| 301 | // Must be called with m.mux held. |
| 302 | func (m *PipelineManager) removePipelineLocked(key string, cleanup bool) *stoppedPipeline { |
| 303 | rp, ok := m.pipelines[key] |
| 304 | if !ok { |
| 305 | return nil |
| 306 | } |
| 307 | |
| 308 | // Cancel the pipeline (it will stop asynchronously) |
| 309 | rp.cancel() |
| 310 | |
| 311 | // Remove from map so it's not visible to other operations |
| 312 | delete(m.pipelines, key) |
| 313 | |
| 314 | result := &stoppedPipeline{rp: rp} |
| 315 | |
| 316 | if cleanup { |
| 317 | result.sourcesToRemove = m.collectSourcesForCleanupLocked(key) |
| 318 | } |
| 319 | |
| 320 | return result |
| 321 | } |
| 322 | |
| 323 | // waitForPipeline waits for a pipeline to finish without sending removal notifications. |
| 324 | // Used when sources are in pending removal state (grace period) and shouldn't be cleaned up. |
| 325 | // Must be called without holding m.mux. |
| 326 | func (m *PipelineManager) waitForPipeline(key string, sp *stoppedPipeline) { |
| 327 | <-sp.rp.done |
| 328 | m.Infof("pipeline '%s' stopped", key) |
| 329 | } |
| 330 | |
| 331 | // waitAndCleanup waits for pipeline to finish and sends removal notifications. |
| 332 | // Must be called without holding m.mux. |
| 333 | func (m *PipelineManager) waitAndCleanup(key string, sp *stoppedPipeline) { |
| 334 | <-sp.rp.done |
| 335 | m.Infof("pipeline '%s' stopped", key) |
| 336 | |
| 337 | // Send removals outside the lock |
| 338 | if len(sp.sourcesToRemove) > 0 { |
| 339 | ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) |
| 340 | defer cancel() |
| 341 | |
| 342 | for _, source := range sp.sourcesToRemove { |
| 343 | m.Debugf("pipeline '%s': sending removal for source '%s'", key, source) |
| 344 | m.send(ctx, []*confgroup.Group{{Source: source}}) |
| 345 | } |
| 346 | } |
| 347 | } |
| 348 | |
| 349 | // collectSourcesForCleanupLocked collects all sources that need removal notifications, |
| 350 | // including tracked sources and pending removals. Must be called with m.mux held. |
| 351 | func (m *PipelineManager) collectSourcesForCleanupLocked(key string) []string { |
| 352 | sourceSet := make(map[string]struct{}) |
| 353 | |
| 354 | // Collect tracked sources |
| 355 | if sources, ok := m.pipelineSources[key]; ok { |
| 356 | for src := range sources { |
| 357 | sourceSet[src] = struct{}{} |
| 358 | } |
| 359 | } |
| 360 | |
| 361 | // Collect pending removal sources (these would otherwise be orphaned) |
| 362 | if pending, ok := m.pendingRemovals[key]; ok { |
| 363 | for src := range pending.sources { |
| 364 | sourceSet[src] = struct{}{} |
| 365 | } |
| 366 | } |
| 367 | |
| 368 | // Clean up maps |
| 369 | delete(m.pipelineSources, key) |
| 370 | delete(m.pendingRemovals, key) |
| 371 | |
| 372 | // Convert to slice |
| 373 | sources := make([]string, 0, len(sourceSet)) |
| 374 | for src := range sourceSet { |
| 375 | sources = append(sources, src) |
| 376 | } |
| 377 | return sources |
| 378 | } |
| 379 | |
| 380 | func (m *PipelineManager) processGracePeriodRemovals(ctx context.Context) { |
| 381 | // Collect expired removals while holding the lock |
| 382 | type removal struct { |
| 383 | key string |
| 384 | source string |
| 385 | } |
| 386 | var toRemove []removal |
| 387 | |
| 388 | m.mux.Lock() |
| 389 | now := time.Now() |
| 390 | |
| 391 | for key, pending := range m.pendingRemovals { |
| 392 | if now.Sub(pending.timestamp) < restartGracePeriod { |
| 393 | continue |
| 394 | } |
| 395 | |
| 396 | // Grace period expired - collect sources that weren't re-discovered |
| 397 | for source := range pending.sources { |
| 398 | toRemove = append(toRemove, removal{key: key, source: source}) |
| 399 | |
| 400 | // Remove from tracked sources |
| 401 | if sources, ok := m.pipelineSources[key]; ok { |
| 402 | delete(sources, source) |
| 403 | } |
| 404 | } |
| 405 | |
| 406 | delete(m.pendingRemovals, key) |
| 407 | } |
| 408 | m.mux.Unlock() |
| 409 | |
| 410 | // Send removals outside the lock to avoid blocking other operations |
| 411 | for _, r := range toRemove { |
| 412 | m.Infof("pipeline '%s': grace period expired, removing source '%s'", r.key, r.source) |
| 413 | m.send(ctx, []*confgroup.Group{{Source: r.source}}) |
| 414 | } |
| 415 | } |
| 416 | |
| 417 | func copySourcesMap(src map[string]struct{}) map[string]struct{} { |
| 418 | dst := make(map[string]struct{}, len(src)) |
| 419 | maps.Copy(dst, src) |
| 420 | return dst |
| 421 | } |