| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | package sd |
| 4 | |
| 5 | import ( |
| 6 | "context" |
| 7 | "errors" |
| 8 | "fmt" |
| 9 | "sync" |
| 10 | "sync/atomic" |
| 11 | "testing" |
| 12 | "time" |
| 13 | |
| 14 | "github.com/netdata/netdata/go/plugins/logger" |
| 15 | "github.com/netdata/netdata/go/plugins/plugin/agent/discovery/sd/pipeline" |
| 16 | "github.com/netdata/netdata/go/plugins/plugin/framework/confgroup" |
| 17 | |
| 18 | "github.com/stretchr/testify/assert" |
| 19 | "github.com/stretchr/testify/require" |
| 20 | ) |
| 21 | |
| 22 | func TestPipelineManager_Start(t *testing.T) { |
| 23 | tests := map[string]struct { |
| 24 | setup func(m *PipelineManager, ctx context.Context) |
| 25 | key string |
| 26 | cfg pipeline.Config |
| 27 | wantErr bool |
| 28 | wantRunning bool |
| 29 | }{ |
| 30 | "start new pipeline": { |
| 31 | key: "test-pipeline", |
| 32 | cfg: pipeline.Config{Name: "test"}, |
| 33 | wantRunning: true, |
| 34 | }, |
| 35 | "start replaces existing pipeline": { |
| 36 | setup: func(m *PipelineManager, ctx context.Context) { |
| 37 | _ = m.Start(ctx, "test-pipeline", pipeline.Config{Name: "old"}) |
| 38 | }, |
| 39 | key: "test-pipeline", |
| 40 | cfg: pipeline.Config{Name: "new"}, |
| 41 | wantRunning: true, |
| 42 | }, |
| 43 | "start with invalid config fails": { |
| 44 | key: "test-pipeline", |
| 45 | cfg: pipeline.Config{Name: "invalid"}, |
| 46 | wantErr: true, |
| 47 | }, |
| 48 | } |
| 49 | |
| 50 | for name, tc := range tests { |
| 51 | t.Run(name, func(t *testing.T) { |
| 52 | ctx := t.Context() |
| 53 | |
| 54 | var sentGroups []*confgroup.Group |
| 55 | var mu sync.Mutex |
| 56 | |
| 57 | m := NewPipelineManager( |
| 58 | logger.New(), |
| 59 | mockNewPipeline, |
| 60 | func(_ context.Context, groups []*confgroup.Group) { |
| 61 | mu.Lock() |
| 62 | sentGroups = append(sentGroups, groups...) |
| 63 | mu.Unlock() |
| 64 | }, |
| 65 | ) |
| 66 | |
| 67 | if tc.setup != nil { |
| 68 | tc.setup(m, ctx) |
| 69 | } |
| 70 | |
| 71 | err := m.Start(ctx, tc.key, tc.cfg) |
| 72 | |
| 73 | if tc.wantErr { |
| 74 | assert.Error(t, err) |
| 75 | } else { |
| 76 | assert.NoError(t, err) |
| 77 | } |
| 78 | assert.Equal(t, tc.wantRunning, m.IsRunning(tc.key)) |
| 79 | }) |
| 80 | } |
| 81 | } |
| 82 | |
| 83 | func TestPipelineManager_Stop(t *testing.T) { |
| 84 | t.Run("stop sends removal for tracked sources", func(t *testing.T) { |
| 85 | ctx := t.Context() |
| 86 | |
| 87 | var sentGroups []*confgroup.Group |
| 88 | var mu sync.Mutex |
| 89 | |
| 90 | m := NewPipelineManager( |
| 91 | logger.New(), |
| 92 | mockNewPipelineWithGroups( |
| 93 | []*confgroup.Group{ |
| 94 | {Source: "source1", Configs: []confgroup.Config{}}, |
| 95 | {Source: "source2", Configs: []confgroup.Config{}}, |
| 96 | }, |
| 97 | ), |
| 98 | func(_ context.Context, groups []*confgroup.Group) { |
| 99 | mu.Lock() |
| 100 | sentGroups = append(sentGroups, groups...) |
| 101 | mu.Unlock() |
| 102 | }, |
| 103 | ) |
| 104 | |
| 105 | err := m.Start(ctx, "test-pipeline", pipeline.Config{Name: "test"}) |
| 106 | require.NoError(t, err) |
| 107 | |
| 108 | // Wait for groups to be received |
| 109 | time.Sleep(100 * time.Millisecond) |
| 110 | |
| 111 | m.Stop("test-pipeline") |
| 112 | |
| 113 | // Wait for stop to complete |
| 114 | time.Sleep(100 * time.Millisecond) |
| 115 | |
| 116 | assert.False(t, m.IsRunning("test-pipeline")) |
| 117 | |
| 118 | mu.Lock() |
| 119 | // Should have initial groups + removal groups |
| 120 | // Removal groups have nil Configs (not empty slice) |
| 121 | var removalSources []string |
| 122 | for _, g := range sentGroups { |
| 123 | if g.Configs == nil { |
| 124 | removalSources = append(removalSources, g.Source) |
| 125 | } |
| 126 | } |
| 127 | mu.Unlock() |
| 128 | |
| 129 | assert.ElementsMatch(t, []string{"source1", "source2"}, removalSources) |
| 130 | }) |
| 131 | |
| 132 | t.Run("stop non-existent pipeline is no-op", func(t *testing.T) { |
| 133 | m := NewPipelineManager( |
| 134 | logger.New(), |
| 135 | mockNewPipeline, |
| 136 | func(_ context.Context, _ []*confgroup.Group) {}, |
| 137 | ) |
| 138 | |
| 139 | // Should not panic |
| 140 | m.Stop("non-existent") |
| 141 | assert.False(t, m.IsRunning("non-existent")) |
| 142 | }) |
| 143 | } |
| 144 | |
| 145 | func TestPipelineManager_Restart(t *testing.T) { |
| 146 | t.Run("restart uses grace period for overlapping sources", func(t *testing.T) { |
| 147 | ctx := t.Context() |
| 148 | |
| 149 | var sentGroups []*confgroup.Group |
| 150 | var mu sync.Mutex |
| 151 | |
| 152 | // First pipeline discovers source1 and source2 |
| 153 | firstPipelineGroups := []*confgroup.Group{ |
| 154 | {Source: "source1", Configs: []confgroup.Config{}}, |
| 155 | {Source: "source2", Configs: []confgroup.Config{}}, |
| 156 | } |
| 157 | |
| 158 | // Second pipeline re-discovers source1 but not source2 |
| 159 | secondPipelineGroups := []*confgroup.Group{ |
| 160 | {Source: "source1", Configs: []confgroup.Config{}}, |
| 161 | } |
| 162 | |
| 163 | callCount := 0 |
| 164 | m := NewPipelineManager( |
| 165 | logger.New(), |
| 166 | func(cfg pipeline.Config) (sdPipeline, error) { |
| 167 | callCount++ |
| 168 | if callCount == 1 { |
| 169 | return newMockPipelineWithGroups(cfg.Name, firstPipelineGroups), nil |
| 170 | } |
| 171 | return newMockPipelineWithGroups(cfg.Name, secondPipelineGroups), nil |
| 172 | }, |
| 173 | func(_ context.Context, groups []*confgroup.Group) { |
| 174 | mu.Lock() |
| 175 | sentGroups = append(sentGroups, groups...) |
| 176 | mu.Unlock() |
| 177 | }, |
| 178 | ) |
| 179 | |
| 180 | // Start first pipeline |
| 181 | err := m.Start(ctx, "test-pipeline", pipeline.Config{Name: "v1"}) |
| 182 | require.NoError(t, err) |
| 183 | |
| 184 | // Wait for first pipeline to send groups |
| 185 | time.Sleep(100 * time.Millisecond) |
| 186 | |
| 187 | // Restart with new config |
| 188 | err = m.Restart(ctx, "test-pipeline", pipeline.Config{Name: "v2"}) |
| 189 | require.NoError(t, err) |
| 190 | |
| 191 | // Wait for second pipeline to send groups |
| 192 | time.Sleep(100 * time.Millisecond) |
| 193 | |
| 194 | assert.True(t, m.IsRunning("test-pipeline")) |
| 195 | |
| 196 | // source1 should NOT be in pending removals (re-discovered) |
| 197 | // source2 should be in pending removals (not re-discovered) |
| 198 | mu.Lock() |
| 199 | // At this point, no removals should have been sent yet (within grace period) |
| 200 | // Removal groups have nil Configs (not empty slice) |
| 201 | var removalSources []string |
| 202 | for _, g := range sentGroups { |
| 203 | if g.Configs == nil { |
| 204 | removalSources = append(removalSources, g.Source) |
| 205 | } |
| 206 | } |
| 207 | mu.Unlock() |
| 208 | |
| 209 | assert.Empty(t, removalSources, "no removals should be sent within grace period") |
| 210 | |
| 211 | // Verify pending removals state |
| 212 | m.mux.Lock() |
| 213 | pending, ok := m.pendingRemovals["test-pipeline"] |
| 214 | m.mux.Unlock() |
| 215 | |
| 216 | assert.True(t, ok, "should have pending removals") |
| 217 | if ok { |
| 218 | _, hasSource2 := pending.sources["source2"] |
| 219 | _, hasSource1 := pending.sources["source1"] |
| 220 | assert.True(t, hasSource2, "source2 should be pending removal") |
| 221 | assert.False(t, hasSource1, "source1 should NOT be pending (re-discovered)") |
| 222 | } |
| 223 | }) |
| 224 | |
| 225 | t.Run("restart with invalid config keeps old pipeline", func(t *testing.T) { |
| 226 | ctx := t.Context() |
| 227 | |
| 228 | callCount := 0 |
| 229 | m := NewPipelineManager( |
| 230 | logger.New(), |
| 231 | func(cfg pipeline.Config) (sdPipeline, error) { |
| 232 | callCount++ |
| 233 | if cfg.Name == "invalid" { |
| 234 | return nil, errors.New("invalid config") |
| 235 | } |
| 236 | return newMockPipeline(cfg.Name), nil |
| 237 | }, |
| 238 | func(_ context.Context, _ []*confgroup.Group) {}, |
| 239 | ) |
| 240 | |
| 241 | // Start first pipeline |
| 242 | err := m.Start(ctx, "test-pipeline", pipeline.Config{Name: "v1"}) |
| 243 | require.NoError(t, err) |
| 244 | assert.True(t, m.IsRunning("test-pipeline")) |
| 245 | |
| 246 | // Try to restart with invalid config |
| 247 | err = m.Restart(ctx, "test-pipeline", pipeline.Config{Name: "invalid"}) |
| 248 | assert.Error(t, err) |
| 249 | |
| 250 | // Old pipeline should still be running |
| 251 | assert.True(t, m.IsRunning("test-pipeline")) |
| 252 | }) |
| 253 | } |
| 254 | |
| 255 | func TestPipelineManager_StopAll(t *testing.T) { |
| 256 | t.Run("stops all pipelines and sends removals", func(t *testing.T) { |
| 257 | ctx := t.Context() |
| 258 | |
| 259 | var sentGroups []*confgroup.Group |
| 260 | var mu sync.Mutex |
| 261 | |
| 262 | m := NewPipelineManager( |
| 263 | logger.New(), |
| 264 | mockNewPipelineWithGroups( |
| 265 | []*confgroup.Group{{Source: "source1", Configs: []confgroup.Config{}}}, |
| 266 | ), |
| 267 | func(_ context.Context, groups []*confgroup.Group) { |
| 268 | mu.Lock() |
| 269 | sentGroups = append(sentGroups, groups...) |
| 270 | mu.Unlock() |
| 271 | }, |
| 272 | ) |
| 273 | |
| 274 | // Start multiple pipelines |
| 275 | _ = m.Start(ctx, "pipeline1", pipeline.Config{Name: "p1"}) |
| 276 | _ = m.Start(ctx, "pipeline2", pipeline.Config{Name: "p2"}) |
| 277 | _ = m.Start(ctx, "pipeline3", pipeline.Config{Name: "p3"}) |
| 278 | |
| 279 | // Wait for pipelines to send groups |
| 280 | time.Sleep(100 * time.Millisecond) |
| 281 | |
| 282 | assert.Len(t, m.Keys(), 3) |
| 283 | |
| 284 | m.StopAll() |
| 285 | |
| 286 | // Wait for stop to complete |
| 287 | time.Sleep(100 * time.Millisecond) |
| 288 | |
| 289 | assert.Empty(t, m.Keys()) |
| 290 | assert.False(t, m.IsRunning("pipeline1")) |
| 291 | assert.False(t, m.IsRunning("pipeline2")) |
| 292 | assert.False(t, m.IsRunning("pipeline3")) |
| 293 | }) |
| 294 | } |
| 295 | |
| 296 | func TestPipelineManager_RunGracePeriodCleanup(t *testing.T) { |
| 297 | t.Run("expired pending removals are cleaned up", func(t *testing.T) { |
| 298 | ctx := t.Context() |
| 299 | |
| 300 | var sentGroups []*confgroup.Group |
| 301 | var mu sync.Mutex |
| 302 | |
| 303 | m := NewPipelineManager( |
| 304 | logger.New(), |
| 305 | mockNewPipeline, |
| 306 | func(_ context.Context, groups []*confgroup.Group) { |
| 307 | mu.Lock() |
| 308 | sentGroups = append(sentGroups, groups...) |
| 309 | mu.Unlock() |
| 310 | }, |
| 311 | ) |
| 312 | |
| 313 | // Manually add a pending removal with expired timestamp |
| 314 | m.mux.Lock() |
| 315 | m.pendingRemovals["test-pipeline"] = &pendingRemoval{ |
| 316 | sources: map[string]struct{}{"expired-source": {}}, |
| 317 | timestamp: time.Now().Add(-65 * time.Second), // older than 1 minute grace period |
| 318 | } |
| 319 | m.pipelineSources["test-pipeline"] = map[string]struct{}{"expired-source": {}} |
| 320 | m.mux.Unlock() |
| 321 | |
| 322 | // Run one iteration of cleanup |
| 323 | m.processGracePeriodRemovals(ctx) |
| 324 | |
| 325 | // Check that removal was sent |
| 326 | // Removal groups have nil Configs |
| 327 | mu.Lock() |
| 328 | var removalSources []string |
| 329 | for _, g := range sentGroups { |
| 330 | if g.Configs == nil { |
| 331 | removalSources = append(removalSources, g.Source) |
| 332 | } |
| 333 | } |
| 334 | mu.Unlock() |
| 335 | |
| 336 | assert.Contains(t, removalSources, "expired-source") |
| 337 | |
| 338 | // Pending removal should be cleared |
| 339 | m.mux.Lock() |
| 340 | _, hasPending := m.pendingRemovals["test-pipeline"] |
| 341 | m.mux.Unlock() |
| 342 | assert.False(t, hasPending) |
| 343 | }) |
| 344 | |
| 345 | t.Run("non-expired pending removals are preserved", func(t *testing.T) { |
| 346 | ctx := t.Context() |
| 347 | |
| 348 | var sentGroups []*confgroup.Group |
| 349 | var mu sync.Mutex |
| 350 | |
| 351 | m := NewPipelineManager( |
| 352 | logger.New(), |
| 353 | mockNewPipeline, |
| 354 | func(_ context.Context, groups []*confgroup.Group) { |
| 355 | mu.Lock() |
| 356 | sentGroups = append(sentGroups, groups...) |
| 357 | mu.Unlock() |
| 358 | }, |
| 359 | ) |
| 360 | |
| 361 | // Manually add a pending removal with recent timestamp |
| 362 | m.mux.Lock() |
| 363 | m.pendingRemovals["test-pipeline"] = &pendingRemoval{ |
| 364 | sources: map[string]struct{}{"recent-source": {}}, |
| 365 | timestamp: time.Now(), // just now - not expired |
| 366 | } |
| 367 | m.mux.Unlock() |
| 368 | |
| 369 | // Run one iteration of cleanup |
| 370 | m.processGracePeriodRemovals(ctx) |
| 371 | |
| 372 | // No removal should be sent |
| 373 | // Removal groups have nil Configs |
| 374 | mu.Lock() |
| 375 | var removalSources []string |
| 376 | for _, g := range sentGroups { |
| 377 | if g.Configs == nil { |
| 378 | removalSources = append(removalSources, g.Source) |
| 379 | } |
| 380 | } |
| 381 | mu.Unlock() |
| 382 | |
| 383 | assert.Empty(t, removalSources) |
| 384 | |
| 385 | // Pending removal should still exist |
| 386 | m.mux.Lock() |
| 387 | _, hasPending := m.pendingRemovals["test-pipeline"] |
| 388 | m.mux.Unlock() |
| 389 | assert.True(t, hasPending) |
| 390 | }) |
| 391 | } |
| 392 | |
| 393 | func TestPipelineManager_IsRunning(t *testing.T) { |
| 394 | ctx := t.Context() |
| 395 | |
| 396 | m := NewPipelineManager( |
| 397 | logger.New(), |
| 398 | mockNewPipeline, |
| 399 | func(_ context.Context, _ []*confgroup.Group) {}, |
| 400 | ) |
| 401 | |
| 402 | assert.False(t, m.IsRunning("test")) |
| 403 | |
| 404 | _ = m.Start(ctx, "test", pipeline.Config{Name: "test"}) |
| 405 | assert.True(t, m.IsRunning("test")) |
| 406 | |
| 407 | m.Stop("test") |
| 408 | time.Sleep(50 * time.Millisecond) |
| 409 | assert.False(t, m.IsRunning("test")) |
| 410 | } |
| 411 | |
| 412 | func TestPipelineManager_Keys(t *testing.T) { |
| 413 | ctx := t.Context() |
| 414 | |
| 415 | m := NewPipelineManager( |
| 416 | logger.New(), |
| 417 | mockNewPipeline, |
| 418 | func(_ context.Context, _ []*confgroup.Group) {}, |
| 419 | ) |
| 420 | |
| 421 | assert.Empty(t, m.Keys()) |
| 422 | |
| 423 | _ = m.Start(ctx, "p1", pipeline.Config{Name: "p1"}) |
| 424 | _ = m.Start(ctx, "p2", pipeline.Config{Name: "p2"}) |
| 425 | |
| 426 | keys := m.Keys() |
| 427 | assert.Len(t, keys, 2) |
| 428 | assert.ElementsMatch(t, []string{"p1", "p2"}, keys) |
| 429 | } |
| 430 | |
| 431 | func TestPipelineManager_ConcurrentOperations(t *testing.T) { |
| 432 | // Note: Concurrent operations on the SAME key are not supported and cannot |
| 433 | // happen in production (ServiceDiscovery.run() processes events sequentially). |
| 434 | // This test verifies concurrent operations on DIFFERENT keys work correctly. |
| 435 | |
| 436 | ctx := t.Context() |
| 437 | |
| 438 | // Track created and stopped pipelines to detect leaks |
| 439 | var created, stopped atomic.Int64 |
| 440 | |
| 441 | mockFactory := func(cfg pipeline.Config) (sdPipeline, error) { |
| 442 | created.Add(1) |
| 443 | return &trackingMockPipeline{stopped: &stopped}, nil |
| 444 | } |
| 445 | |
| 446 | m := NewPipelineManager( |
| 447 | logger.New(), |
| 448 | mockFactory, |
| 449 | func(_ context.Context, _ []*confgroup.Group) {}, |
| 450 | ) |
| 451 | |
| 452 | var wg sync.WaitGroup |
| 453 | |
| 454 | // Concurrent starts for different keys |
| 455 | for i := range 10 { |
| 456 | wg.Add(1) |
| 457 | go func(i int) { |
| 458 | defer wg.Done() |
| 459 | key := fmt.Sprintf("pipeline-%d", i) |
| 460 | _ = m.Start(ctx, key, pipeline.Config{Name: key}) |
| 461 | }(i) |
| 462 | } |
| 463 | |
| 464 | // Concurrent IsRunning checks |
| 465 | for i := range 10 { |
| 466 | wg.Add(1) |
| 467 | go func(i int) { |
| 468 | defer wg.Done() |
| 469 | _ = m.IsRunning(fmt.Sprintf("pipeline-%d", i)) |
| 470 | }(i) |
| 471 | } |
| 472 | |
| 473 | // Concurrent Keys checks |
| 474 | for range 10 { |
| 475 | wg.Go(func() { |
| 476 | _ = m.Keys() |
| 477 | }) |
| 478 | } |
| 479 | |
| 480 | wg.Wait() |
| 481 | |
| 482 | // Should have 10 pipelines running (one per unique key) |
| 483 | assert.Len(t, m.Keys(), 10) |
| 484 | for i := range 10 { |
| 485 | assert.True(t, m.IsRunning(fmt.Sprintf("pipeline-%d", i))) |
| 486 | } |
| 487 | |
| 488 | // Stop all pipelines |
| 489 | m.StopAll() |
| 490 | |
| 491 | // Wait for all pipelines to stop |
| 492 | assert.Eventually(t, func() bool { |
| 493 | return created.Load() == stopped.Load() |
| 494 | }, time.Second*5, time.Millisecond*100, |
| 495 | "leaked pipelines: created=%d, stopped=%d", created.Load(), stopped.Load()) |
| 496 | } |
| 497 | |
| 498 | // mockNewPipeline creates a mock pipeline that does nothing. |
| 499 | func mockNewPipeline(cfg pipeline.Config) (sdPipeline, error) { |
| 500 | if cfg.Name == "invalid" { |
| 501 | return nil, errors.New("invalid config") |
| 502 | } |
| 503 | return newMockPipeline(cfg.Name), nil |
| 504 | } |
| 505 | |
| 506 | // mockNewPipelineWithGroups creates a factory that produces pipelines that send specific groups. |
| 507 | func mockNewPipelineWithGroups(groups []*confgroup.Group) func(cfg pipeline.Config) (sdPipeline, error) { |
| 508 | return func(cfg pipeline.Config) (sdPipeline, error) { |
| 509 | if cfg.Name == "invalid" { |
| 510 | return nil, errors.New("invalid config") |
| 511 | } |
| 512 | return newMockPipelineWithGroups(cfg.Name, groups), nil |
| 513 | } |
| 514 | } |
| 515 | |
| 516 | type testMockPipeline struct { |
| 517 | name string |
| 518 | groups []*confgroup.Group |
| 519 | } |
| 520 | |
| 521 | func newMockPipeline(name string) *testMockPipeline { |
| 522 | return &testMockPipeline{name: name} |
| 523 | } |
| 524 | |
| 525 | func newMockPipelineWithGroups(name string, groups []*confgroup.Group) *testMockPipeline { |
| 526 | return &testMockPipeline{name: name, groups: groups} |
| 527 | } |
| 528 | |
| 529 | func (p *testMockPipeline) Run(ctx context.Context, out chan<- []*confgroup.Group) { |
| 530 | // Send initial groups if any |
| 531 | if len(p.groups) > 0 { |
| 532 | select { |
| 533 | case out <- p.groups: |
| 534 | case <-ctx.Done(): |
| 535 | return |
| 536 | } |
| 537 | } |
| 538 | |
| 539 | // Wait for cancellation |
| 540 | <-ctx.Done() |
| 541 | } |
| 542 | |
| 543 | // trackingMockPipeline tracks when it stops for leak detection |
| 544 | type trackingMockPipeline struct { |
| 545 | stopped *atomic.Int64 |
| 546 | } |
| 547 | |
| 548 | func (p *trackingMockPipeline) Run(ctx context.Context, _ chan<- []*confgroup.Group) { |
| 549 | <-ctx.Done() |
| 550 | p.stopped.Add(1) |
| 551 | } |