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

Use server-side filtering when retrieving Cloud Foundry logs #33456

Merged
merged 11 commits into from
Nov 8, 2022
Merged

Use server-side filtering when retrieving Cloud Foundry logs #33456

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
971dd5d
Expose Cloud Foundry filtering into user configuration
jsoriano Oct 25, 2022
20c7ddc
Fix changelog
jsoriano Oct 25, 2022
7fc340d
Linting and fix on docs
jsoriano Oct 25, 2022
fae0b0a
Add both server and client side filtering
MichaelKatsoulis Nov 7, 2022
1a1ff0f
Merge remote-tracking branch 'upstream/main' into cloudfoundry-config…
MichaelKatsoulis Nov 7, 2022
989dfde
Edit changelog
MichaelKatsoulis Nov 7, 2022
7b6afaf
Merge remote-tracking branch 'origin/main' into cloudfoundry-config-f…
jsoriano Nov 7, 2022
17d262b
Remove all-logs option
jsoriano Nov 7, 2022
3deb848
Remove option, simplify
jsoriano Nov 7, 2022
07e002b
Merge remote-tracking branch 'upstream/main' into cloudfoundry-config…
MichaelKatsoulis Nov 8, 2022
f0ca944
Merge upstream
MichaelKatsoulis Nov 8, 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 @@ -150,6 +150,7 @@ https://github.com/elastic/beats/compare/v8.2.0\...main[Check the HEAD diff]
- Allow http_endpoint instances to share ports. {issue}32578[32578] {pull}33377[33377]
- Improve httpjson documentation for split processor. {pull}33473[33473]
- Added separation of transform context object inside httpjson. Introduced new clause `.parent_last_response.*` {pull}33499[33499]
- 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 @@ -8,7 +8,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 @@ -51,7 +51,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