acquisition: fix transformation being ignored - #4609
Conversation
|
@blotus: There are no 'kind' label on this PR. You need a 'kind' label to generate the release automatically.
DetailsI am a bot created to help the crowdsecurity developers manage community feedback and contributions. You can check out my manifest file to understand my behavior and what I can do. If you want to use this for your project, you can check out the BirthdayResearch/oss-governance-bot repository. |
|
/kind fix |
|
@blotus: There are no area labels on this PR. You can add as many areas as you see fit.
DetailsI am a bot created to help the crowdsecurity developers manage community feedback and contributions. You can check out my manifest file to understand my behavior and what I can do. If you want to use this for your project, you can check out the BirthdayResearch/oss-governance-bot repository. |
There was a problem hiding this comment.
Pull request overview
Fixes acquisition transform handling introduced by the datasource refactor by ensuring datasources write into the transform channel (when configured) and ensures the transform goroutine terminates cleanly in CAT/replay mode.
Changes:
- Route datasource output to the transform channel when a transform expression is configured, and forward transformed events to the final output channel.
- Ensure the transformer exits when the transform channel is closed (CAT mode), preventing acquisition hangs.
- Add tests validating transform behavior and CAT-mode termination with transforms.
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.
| File | Description |
|---|---|
| pkg/acquisition/acquisition.go | Fixes channel wiring so transforms are applied; closes transform channel in CAT mode so transformer can exit. |
| pkg/acquisition/acquisition_test.go | Adds regression tests for transform application and CAT-mode termination with transforms. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| out := make(chan pipeline.Event) | ||
| acquisTomb := tomb.Tomb{} | ||
|
|
||
| go func() { | ||
| _ = StartAcquisition(ctx, sources, out, &acquisTomb) | ||
| }() | ||
|
|
||
| got := []string{} | ||
|
|
||
| READLOOP: | ||
| for { | ||
| select { | ||
| case evt := <-out: | ||
| got = append(got, evt.Line.Raw) | ||
| case <-time.After(1 * time.Second): | ||
| break READLOOP | ||
| } | ||
| } | ||
|
|
||
| acquisTomb.Kill(nil) | ||
|
|
||
| assert.Equal(t, tc.expected, got) |
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #4609 +/- ##
==========================================
+ Coverage 64.10% 64.77% +0.66%
==========================================
Files 520 503 -17
Lines 39649 38922 -727
==========================================
- Hits 25418 25210 -208
+ Misses 11848 11370 -478
+ Partials 2383 2342 -41
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
Since the datasource refactoring, we were passing the output chan to the datasources even if a transform expression was provided, instead of the transform chan.
This PR also fixes the transform goroutine being kept alive after the acquisition ended in replay mode.