Skip to content
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.

Sign up for GitHub

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

[7.17](backport #33456) Use server-side filtering when retrieving Cloud Foundry logs #33637

Merged
merged 2 commits into from
Nov 10, 2022
Merged

[7.17](backport #33456) Use server-side filtering when retrieving Cloud Foundry logs #33637

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
421874f
Use server-side filtering when retrieving Cloud Foundry logs (#33456)
jsoriano Nov 8, 2022
6cc6211
Update CHANGELOG.next.asciidoc
MichaelKatsoulis Nov 10, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.next.asciidoc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -96,6 +96,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
- httpjson input: Add request tracing logger. {issue}32402[32402] {pull}32412[32412]
- Add cloudflare R2 to provider list in AWS S3 input. {pull}32620[32620]
- Add support for single string containing multiple relation-types in getRFC5988Link. {pull}32811[32811]
- Cloud Foundry input uses server-side filtering when retrieving logs. {pull}33456[33456]

*Auditbeat*

Expand Down
4 changes: 2 additions & 2 deletions x-pack/filebeat/input/cloudfoundry/v1.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
package cloudfoundry

import (
"github.com/pkg/errors"
"fmt"

v2 "github.com/elastic/beats/v7/filebeat/input/v2"
stateless "github.com/elastic/beats/v7/filebeat/input/v2/input-stateless"
Expand Down Expand Up @@ -48,7 +48,7 @@ func (i *inputV1) Run(ctx v2.Context, publisher stateless.Publisher) error {

consumer, err := hub.DopplerConsumer(callbacks)
if err != nil {
return errors.Wrapf(err, "initializing doppler consumer")
return fmt.Errorf("initializing doppler consumer: %w", err)
}

stopCtx, cancel := ctxtool.WithFunc(ctx.Cancelation, func() {
Expand Down
24 changes: 6 additions & 18 deletions x-pack/libbeat/common/cloudfoundry/dopplerconsumer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,27 +80,15 @@ func (c *DopplerConsumer) Run() {
}

func (c *DopplerConsumer) logsFirehose() {
c.firehose(c.callbacks.Log, consumer.LogMessages)
c.firehose(c.callbacks.Log, filterLogs, consumer.LogMessages)
}

func (c *DopplerConsumer) metricsFirehose() {
c.firehose(c.callbacks.Metric, consumer.Metrics)
}

func (c *DopplerConsumer) firehose(cb func(evt Event), filter consumer.EnvelopeFilter) {
var msgChan <-chan *events.Envelope
var errChan <-chan error
filterFn := filterNoFilter
if filter == consumer.LogMessages {
// We are interested in more envelopes than the ones obtained when filtering
// by log messages, retrieve them all and filter later.
// If this causes performance or other problems, we will have to investigate
// if it is possible to pass different filters to the firehose url.
filterFn = filterLogs
msgChan, errChan = c.consumer.Firehose(c.subscriptionID, "")
} else {
msgChan, errChan = c.consumer.FilteredFirehose(c.subscriptionID, "", filter)
}
c.firehose(c.callbacks.Metric, filterNoFilter, consumer.Metrics)
}

func (c *DopplerConsumer) firehose(cb func(evt Event), filterFn func(*events.Envelope) bool, filter consumer.EnvelopeFilter) {
msgChan, errChan := c.consumer.FilteredFirehose(c.subscriptionID, "", filter)
for {
select {
case env := <-msgChan:
Expand Down