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

Jump to bottom

[filebeat][azure-blob-storage] - Fixed concurrency & flakey tests issue #36124

Merged
merged 14 commits into from
Aug 4, 2023
Merged

[filebeat][azure-blob-storage] - Fixed concurrency & flakey tests issue #36124

Show file tree
Hide file tree
Changes from 12 commits
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
3281397
added fix for concurrency issues and flakey tests
ShourieG Jul 20, 2023
cb1c22e
Merge remote-tracking branch 'upstream/main' into abs/concurrency_fix
ShourieG Jul 20, 2023
0e5acb3
updated changelog
ShourieG Jul 20, 2023
d97a84d
fixed changelog
ShourieG Jul 20, 2023
758cfcf
Merge remote-tracking branch 'upstream/main' into abs/concurrency_fix
ShourieG Jul 31, 2023
97351d9
added concurrency tests
ShourieG Jul 31, 2023
dadced8
Merge remote-tracking branch 'upstream/main' into abs/concurrency_fix
ShourieG Jul 31, 2023
16cb787
assignment order of done & cp in job.go updated
ShourieG Jul 31, 2023
aff1d96
added custom client for concurrency test
ShourieG Aug 2, 2023
19d795a
Merge remote-tracking branch 'upstream/main' into abs/concurrency_fix
ShourieG Aug 2, 2023
9180f5c
added nolint tags
ShourieG Aug 2, 2023
b49a4c7
removed channels from test publish struct and updated with suggestions
ShourieG Aug 2, 2023
685e486
updated concurrency test code according to PR suggestions
ShourieG Aug 4, 2023
61c1d7b
Merge remote-tracking branch 'upstream/main' into abs/concurrency_fix
ShourieG Aug 4, 2023
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 @@ -104,6 +104,7 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
- Fix handling of NUL-terminated log lines in Fortinet Firewall module. {issue}36026[36026] {pull}36027[36027]
- Make redact field configuration recommended in CEL input and log warning if missing. {pull}36008[36008]
- Fix handling of region name configuration in awss3 input {pull}36034[36034]
- Fixed concurrency and flakey tests issue in azure blob storage input. {issue}35983[35983] {pull}36124[36124]
- Fix panic when sqs input metrics getter is invoked {pull}36101[36101] {issue}36077[36077]
- Make CEL input's `now` global variable static for evaluation lifetime. {pull}36107[36107]
- Update mito CEL extension library to v1.5.0. {pull}36146[36146]
Expand Down
16 changes: 10 additions & 6 deletions x-pack/filebeat/input/azureblobstorage/input.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@ import (

v2 "github.com/elastic/beats/v7/filebeat/input/v2"
cursor "github.com/elastic/beats/v7/filebeat/input/v2/input-cursor"

"github.com/elastic/beats/v7/libbeat/feature"
conf "github.com/elastic/elastic-agent-libs/config"
"github.com/elastic/elastic-agent-libs/logp"
Expand Down Expand Up @@ -48,6 +49,7 @@ func configure(cfg *conf.C) ([]cursor.Source, cursor.Input, error) {
return nil, nil, err
}

//nolint:prealloc // No need to preallocate the slice here
var sources []cursor.Source
for _, c := range config.Containers {
container := tryOverrideOrDefault(config, c)
Expand Down Expand Up @@ -111,20 +113,22 @@ func (input *azurebsInput) Test(src cursor.Source, ctx v2.TestContext) error {
}

func (input *azurebsInput) Run(inputCtx v2.Context, src cursor.Source, cursor cursor.Cursor, publisher cursor.Publisher) error {
currentSource := src.(*Source)

log := inputCtx.Logger.With("account_name", currentSource.AccountName).With("container_name", currentSource.ContainerName)
log.Infof("Running azure blob storage for account: %s", input.config.AccountName)

var cp *Checkpoint
st := newState()
if !cursor.IsNew() {
if err := cursor.Unpack(&cp); err != nil {
return err
}

st.setCheckpoint(cp)
}
return input.run(inputCtx, src, st, publisher)
}

func (input *azurebsInput) run(inputCtx v2.Context, src cursor.Source, st *state, publisher cursor.Publisher) error {
currentSource := src.(*Source)

log := inputCtx.Logger.With("account_name", currentSource.AccountName).With("container_name", currentSource.ContainerName)
log.Infof("Running azure blob storage for account: %s", input.config.AccountName)

ctx, cancel := context.WithCancel(context.Background())
go func() {
Expand Down
Loading