-
Notifications
You must be signed in to change notification settings - Fork 4.9k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
add support for queue settings under outputs #36788
Conversation
This pull request does not have a backport label.
To fixup this pull request, you need to add the backport labels for the needed
|
441c87c
to
52b0111
Compare
4b506ee
to
ca41e03
Compare
Sorry, I had trouble re-opening #36693, so this a new PR number but is really a continuation. |
libbeat/cmd/instance/beat_test.go
Outdated
tests := map[string]struct { | ||
input []byte | ||
memEvents int | ||
expectValidationError bool |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Is this field being used?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
nope, thanks.
libbeat/cmd/instance/beat_test.go
Outdated
mem: | ||
events: 8096 | ||
`), | ||
memEvents: 8096}, |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Nit: Could totally be just me, but I found these test cases a bit hard to read. I think it might've helped to have the opening and closing braces of each test case struct on their own lines so the struct fields all lined up nicely.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
See if the new fmt is better. I'm open to any suggestions, I just want to be sure the yaml looks like yaml.
// Fail helper can be used by output factories, to create a failure response when | ||
// loading an output must return an error. | ||
func Fail(err error) (Group, error) { return Group{}, err } | ||
|
||
// Success create a valid output Group response for a set of client instances. | ||
func Success(batchSize, retry int, clients ...Client) (Group, error) { | ||
func Success(cfg config.Namespace, batchSize, retry int, clients ...Client) (Group, error) { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Let's document here that cfg
is expected to contain queue configuration, since its datatype here config.Namespace
doesn't quite make that obvious.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Same for SuccessNet
below.
@@ -135,14 +135,15 @@ func NewQueue( | |||
logger *logp.Logger, | |||
ackCallback func(eventCount int), | |||
settings Settings, | |||
inputQueueSize int, |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It's a bit odd not to have inputQueueSize
be part of settings
as before. I agree with removing the special override in the publisher pipeline code but I think that can be done while leaving inputQueueSize
as part of settings
so this constructor here has still has the same signature as before, with all queue settings being in settings
. For that you'll need to set settings.InputQueueSize
wherever you call this constructor from (unless, of course, you want to leave that field at it's zero-value, e.g. in test code).
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The only thing I don't like about that is it is hard to tell if you are supposed to set settings.InputQueueSize
before you pass it into FactoryForSettings
or allow FactoryForSettings
to set it. Also what happens if you pass in a Settings with that field set? Do you override or keep?
I don't like changing the signature, but I'm not fond of the ambiguous nature of who sets InputQueueSize if we leave it in settings. I can be convinced to go either way, do you have a strong opinion?
This PR seems to consist of three types of unrelated changes: some refactoring/cleaning, queue setting changes, idle connection timeout setting changes. That's making it a bit hard to review the impact of each change in isolation. Would it be a lot of work to break this up into three smaller PRs? In general, I think that's a better approach, not just to make reviews easier but also to allow rolling back changes in isolation, if necessary. |
Good idea, I'll split out the idle connection timeout. Unfortunately all the other changes are necessary for the queue under outputs change. I'll add some more comments as to why. |
@@ -1480,3 +1485,42 @@ func sanitizeIPs(ips []string) []string { | |||
} | |||
return validIPs | |||
} | |||
|
|||
func promoteOutputQueueSettings(bc *beatConfig) error { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
promoteOutputQueueSettings handles the use case where the beat is started with a configuration file that has outputs defined (normal stand alone beat).
#36843 now contains the |
@@ -41,20 +41,20 @@ func makeES( | |||
) (outputs.Group, error) { | |||
log := logp.NewLogger(logSelector) | |||
if !cfg.HasField("bulk_max_size") { | |||
cfg.SetInt("bulk_max_size", -1, defaultBulkSize) | |||
_ = cfg.SetInt("bulk_max_size", -1, defaultBulkSize) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
You might have a merge conflict here once #36843 is merged.
- add support for `idle_connection_timeout` for ES output - add support for queue settings under output Closes elastic#35615
505fa7c
to
4dc26ba
Compare
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
LGTM.
This is a confusing change for folks building custom output plugins. Where do I put the configuration for queues now?
If I can leave them as a global config... how do I access the config using the output factory function: beats/libbeat/outputs/output_reg.go Lines 31 to 35 in 433a1a1
|
Looks like only one is allowed:
|
* add support for queue settings under outputs
Proposed commit message
Checklist
CHANGELOG.next.asciidoc
orCHANGELOG-developer.next.asciidoc
.Author's Checklist
How to test this PR locally
Stand alone Filebeat
Start Filebeat with following config file:
Should show
queue.max_events
in the metrics to be 1024.Agent
queue.max_events
in the metrics is 1024Related issues