Composable ACKer #19632
Merged
Composable ACKer #19632
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
urso
added
needs_backport
botelastic
bot
added
needs_team
💔 Tests Failed
Expand to view the summary
Build stats
Test stats 🧪
Test errorsExpand to view the tests failures
Steps errorsExpand to view the steps failures
Log outputExpand to view the last 100 lines of log output
|
|
jenkins run the tests please |
|
Pinging @elastic/integrations-services (Team:Services) |
kvch
reviewed
Jul 3, 2020
added 7 commits
July 4, 2020 01:35
This change replaces the ACK handler functions with a single interface that makes it easier to combine ACK handlers. The global ACK handler is removed from the pipeline, requiring Beats to wrap and compose per input ACK handlers with their own ones.
kvch
reviewed
Jul 6, 2020
libbeat/common/acker/acker.go
Outdated
|
|
||
| // ConnectionOnly ensures that the given ACKer is only used for as long as the | ||
| // pipeline Client is active. Once the Client is closed, the ACKer will drop | ||
| // it's internal state and no more ACK events will be processed. |
There was a problem hiding this comment.
Suggested change
| // it's internal state and no more ACK events will be processed. | |
| // its internal state and no more ACK events will be processed. |
kvch
approved these changes
Jul 6, 2020
kvch
approved these changes
Jul 6, 2020
4 tasks
urso
pushed a commit
to urso/beats
that referenced
this pull request
Jul 8, 2020
This change replaces the ACK handler functions with a single interface that makes it easier to combine ACK handlers. The global ACK handler is removed from the pipeline, requiring Beats to wrap and compose per input ACK handlers with their own ones. Although the PR is quite big, the main difference is that the `ACKCount`, `ACKEvents`, and `ACKLastEvents` handlers have been replaced by a single interface (`beat.ACKer`). The original ACKer implementations from `libbeat/publisher/pipeline/acker.go` and `libbeat/publisher/pipeline/client_acker.go` have been moved `libbeat/common/acker`. The former private implementation is now exposed as Helpers for writing and combining ACK handlers. Support for global ACK handlers has been removed. The `acker.Combine` and `acker.ConnectionOnly` are the only new additions to the code base. The global ACK handler support was introduced for filebeat, that did require some support for combine events from multiple inputs before applying state updates. With the introduction of the v2 input API this requirement will go away, as per input type managers are responsible for handling state update and ACKs. In order to run old and new architecture in parallel, we need to combine ACK handling from input managers, existing input, custom registrar ACKer in filebeat, and event counting support (also via ACK handling) for shutdown. Exposing the interface and providing combinators (acker.Combine) for merging ACK handlers into one helps with the integration. The v2 Input API gives implementors more flexibility in how to handle event publishing, coordination, and state handling shall be implemented. With the original ACK support the callbacks have been deregistered the moment inputs are stopped automatically. But for cursor based inputs we need to continue handling ACKs, even after the input is gone. The interface and helpers provide greater control over ACK handling after shutdown, which is required for the journald, winlog, and file/log inputs. (cherry picked from commit bb89344)
urso
pushed a commit
that referenced
this pull request
Jul 8, 2020
This change replaces the ACK handler functions with a single interface that makes it easier to combine ACK handlers. The global ACK handler is removed from the pipeline, requiring Beats to wrap and compose per input ACK handlers with their own ones. Although the PR is quite big, the main difference is that the `ACKCount`, `ACKEvents`, and `ACKLastEvents` handlers have been replaced by a single interface (`beat.ACKer`). The original ACKer implementations from `libbeat/publisher/pipeline/acker.go` and `libbeat/publisher/pipeline/client_acker.go` have been moved `libbeat/common/acker`. The former private implementation is now exposed as Helpers for writing and combining ACK handlers. Support for global ACK handlers has been removed. The `acker.Combine` and `acker.ConnectionOnly` are the only new additions to the code base. The global ACK handler support was introduced for filebeat, that did require some support for combine events from multiple inputs before applying state updates. With the introduction of the v2 input API this requirement will go away, as per input type managers are responsible for handling state update and ACKs. In order to run old and new architecture in parallel, we need to combine ACK handling from input managers, existing input, custom registrar ACKer in filebeat, and event counting support (also via ACK handling) for shutdown. Exposing the interface and providing combinators (acker.Combine) for merging ACK handlers into one helps with the integration. The v2 Input API gives implementors more flexibility in how to handle event publishing, coordination, and state handling shall be implemented. With the original ACK support the callbacks have been deregistered the moment inputs are stopped automatically. But for cursor based inputs we need to continue handling ACKs, even after the input is gone. The interface and helpers provide greater control over ACK handling after shutdown, which is required for the journald, winlog, and file/log inputs. (cherry picked from commit bb89344)
melchiormoulin
pushed a commit
to melchiormoulin/beats
that referenced
this pull request
Oct 14, 2020
This change replaces the ACK handler functions with a single interface that makes it easier to combine ACK handlers. The global ACK handler is removed from the pipeline, requiring Beats to wrap and compose per input ACK handlers with their own ones. Although the PR is quite big, the main difference is that the `ACKCount`, `ACKEvents`, and `ACKLastEvents` handlers have been replaced by a single interface (`beat.ACKer`). The original ACKer implementations from `libbeat/publisher/pipeline/acker.go` and `libbeat/publisher/pipeline/client_acker.go` have been moved `libbeat/common/acker`. The former private implementation is now exposed as Helpers for writing and combining ACK handlers. Support for global ACK handlers has been removed. The `acker.Combine` and `acker.ConnectionOnly` are the only new additions to the code base. The global ACK handler support was introduced for filebeat, that did require some support for combine events from multiple inputs before applying state updates. With the introduction of the v2 input API this requirement will go away, as per input type managers are responsible for handling state update and ACKs. In order to run old and new architecture in parallel, we need to combine ACK handling from input managers, existing input, custom registrar ACKer in filebeat, and event counting support (also via ACK handling) for shutdown. Exposing the interface and providing combinators (acker.Combine) for merging ACK handlers into one helps with the integration. The v2 Input API gives implementors more flexibility in how to handle event publishing, coordination, and state handling shall be implemented. With the original ACK support the callbacks have been deregistered the moment inputs are stopped automatically. But for cursor based inputs we need to continue handling ACKs, even after the input is gone. The interface and helpers provide greater control over ACK handling after shutdown, which is required for the journald, winlog, and file/log inputs.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
What does this PR do?
This change replaces the ACK handler functions with a single interface
that makes it easier to combine ACK handlers.
The global ACK handler is removed from the pipeline, requiring Beats to
wrap and compose per input ACK handlers with their own ones.
Review Notes
Although the PR is quite big, the main difference is that the
ACKCount,ACKEvents, andACKLastEventshandlers have been replaced by a single interface (beat.ACKer). The original ACKer implementations fromlibbeat/publisher/pipeline/acker.goandlibbeat/publisher/pipeline/client_acker.gohave been movedlibbeat/common/acker. The former private implementation is now exposed as Helpers for writing and combining ACK handlers. Support for global ACK handlers has been removed. Theacker.Combineandacker.ConnectionOnlyare the only new additions to the code base.Why is it important?
tl;dr This change is required to integrate the v2 input API.
The global ACK handler support was introduced for filebeat, that did require some support for combine events from multiple inputs before applying state updates. With the introduction of the v2 input API this requirement will go away, as per input type managers are responsible for handling state update and ACKs.
In order to run old and new architecture in parallel, we need to combine ACK handling from input managers, existing input, custom registrar ACKer in filebeat, and event counting support (also via ACK handling) for shutdown. Exposing the interface and providing combinators (acker.Combine) for merging ACK handlers into one helps with the integration.
The v2 Input API gives implementors more flexibility in how to handle event publishing, coordination, and state handling shall be implemented. With the original ACK support the callbacks have been deregistered the moment inputs are stopped automatically. But for cursor based inputs we need to continue handling ACKs, even after the input is gone. The interface and helpers provide greater control over ACK handling after shutdown, which is required for the journald, winlog, and file/log inputs.
Checklist
- [ ] I have made corresponding changes to the documentation- [ ] I have made corresponding change to the default configuration filesCHANGELOG.next.asciidocorCHANGELOG-developer.next.asciidoc.Author's Checklist
How to test this PR locally
ACKer implementations should be handled mostly by unit tests. Still, this change can have great impact and some Beats should be tested to double check that we have not introduced any regressions here.
drop_eventwith conditional. Check non-filtered events are publishedRelated issues
Dev Docs:
Calls to
pipeline.ConnectWiththat used to setup an ACK callback need to use the acker helpers. Previously the callbacks have been ignored if the input is shutdown, but not all events have been ACKed yet. For end-to-end ACKers that loose the connection to the source-system on shutdown this behavior can be preserved by usingack.ConnectionOnly(<acker>).Instead of directly passing callbacks, the callbacks should be wrapped using some of the utility functions in the libbeat/common/acker package:
ACKCount: func(n int) { ... }withACKHandler: acker.Counting(func(n int) { ... }).ACKEvents: func(private []interface{}) { ... }withACKHandler: acker.EventPrivateReporter(func(_ int, private []interface{}) { ... }).ACKLastEvent: func(private interface{}) { ... }withACKHandler: acker.LastEventPrivateReporter(func(_ int, interface{}) { ... })The
(beat.Pipeline).SetACKHandlermethod has been removed.libbeat/common/ackerandlibbeat/publisher/pipetoolprovide some helpers to modify and combine ACKers for all newbeat.Clientconnections. For example this will use the global and local ACK handler for each event published.The
WithACKerhelper can be used arbitrarily often. ACKers are combined level by level viaacker.Combine.