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

add mutex to listinfInfo #30131

Merged
merged 6 commits into from
Feb 8, 2022
Merged

add mutex to listinfInfo #30131

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
ead6097
add mutex to listinfInfo
Feb 1, 2022
9196892
back to mutex
Feb 2, 2022
a39d44f
add warn logging for totObject not matchig sum of storedObjects and e…
Feb 3, 2022
40474e8
changelog
Feb 3, 2022
a794b4d
Merge branch 'master' into race-condition-on-s3-poller-filebeat-issue…
Feb 3, 2022
eea234a
fix merge
Feb 4, 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 @@ -98,6 +98,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
*Filebeat*

- Fix using log_group_name_prefix in aws-cloudwatch input. {pull}29695[29695]
- aws-s3: fix race condition in states used by s3-poller. {issue}30123[30123] {pull}30131[30131]

*Heartbeat*

Expand Down
17 changes: 16 additions & 1 deletion x-pack/filebeat/input/awss3/states.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,9 @@ const (
)

type listingInfo struct {
totObjects int
totObjects int

mu sync.Mutex
storedObjects int
errorObjects int
finalCheck bool
Expand Down Expand Up @@ -105,11 +107,19 @@ func (s *states) Delete(id string) {
func (s *states) IsListingFullyStored(listingID string) bool {
info, _ := s.listingInfo.Load(listingID)
listingInfo := info.(*listingInfo)
listingInfo.mu.Lock()
defer listingInfo.mu.Unlock()
if listingInfo.finalCheck {
return false
}

listingInfo.finalCheck = (listingInfo.storedObjects + listingInfo.errorObjects) == listingInfo.totObjects

if (listingInfo.storedObjects + listingInfo.errorObjects) > listingInfo.totObjects {
s.log.Warnf("unexepected mixmatch between storedObjects (%d), errorObjects (%d) and totObjects (%d)",
listingInfo.storedObjects, listingInfo.errorObjects, listingInfo.totObjects)
}

return listingInfo.finalCheck
}

Expand Down Expand Up @@ -154,6 +164,9 @@ func (s *states) Update(newState state, listingID string) {
// here we increase the number of stored object
info, _ := s.listingInfo.Load(listingID)
listingInfo := info.(*listingInfo)

listingInfo.mu.Lock()

if newState.Stored {
listingInfo.storedObjects++
}
Expand All @@ -162,6 +175,8 @@ func (s *states) Update(newState state, listingID string) {
listingInfo.errorObjects++
}

listingInfo.mu.Unlock()

if _, ok := s.statesByListingID[listingID]; !ok {
s.statesByListingID[listingID] = make([]state, 0)
}
Expand Down