forked from elastic/beats
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Filebeat in memory states registry improvements (elastic#6346)
* Filebeat in memory states registry improvements - make States type easier to find by moving into separate states.go file - Ensure provides States constructor is actually used - Add ID->array index, index for faster lookups on update and find operations. When updating states in a big registry, the Updates converged to quadratic complexity. The index helps in keeping the complexity about linear in number of state updates. - Debug will print number of states subject to future cleanups (if state TTL > 0) - Add title to states unit tests * Fix godocs * update state index in Cleanup * Refine index handling + reintroduce debug message * review
- Loading branch information
Showing
6 changed files
with
180 additions
and
133 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,157 @@ | ||
| package file | ||
|
|
||
| import ( | ||
| "sync" | ||
| "time" | ||
|
|
||
| "github.com/elastic/beats/libbeat/logp" | ||
| ) | ||
|
|
||
| // States handles list of FileState. One must use NewStates to instantiate a | ||
| // file states regisry. Using the zero-value is not safe. | ||
| type States struct { | ||
| sync.RWMutex | ||
|
|
||
| // states store | ||
| states []State | ||
|
|
||
| // idx maps state IDs to state indexes for fast lookup and modifications. | ||
| idx map[string]int | ||
| } | ||
|
|
||
| // NewStates generates a new states registry. | ||
| func NewStates() *States { | ||
| return &States{ | ||
| states: nil, | ||
| idx: map[string]int{}, | ||
| } | ||
| } | ||
|
|
||
| // Update updates a state. If previous state didn't exist, new one is created | ||
| func (s *States) Update(newState State) { | ||
| s.UpdateWithTs(newState, time.Now()) | ||
| } | ||
|
|
||
| // UpdateWithTs updates a state, assigning the given timestamp. | ||
| // If previous state didn't exist, new one is created | ||
| func (s *States) UpdateWithTs(newState State, ts time.Time) { | ||
| s.Lock() | ||
| defer s.Unlock() | ||
|
|
||
| id := newState.ID() | ||
| index := s.findPrevious(id) | ||
| newState.Timestamp = ts | ||
|
|
||
| if index >= 0 { | ||
| s.states[index] = newState | ||
| } else { | ||
| // No existing state found, add new one | ||
| s.idx[id] = len(s.states) | ||
| s.states = append(s.states, newState) | ||
| logp.Debug("input", "New state added for %s", newState.Source) | ||
| } | ||
| } | ||
|
|
||
| // FindPrevious lookups a registered state, that matching the new state. | ||
| // Returns a zero-state if no match is found. | ||
| func (s *States) FindPrevious(newState State) State { | ||
| s.RLock() | ||
| defer s.RUnlock() | ||
| i := s.findPrevious(newState.ID()) | ||
| if i < 0 { | ||
| return State{} | ||
| } | ||
| return s.states[i] | ||
| } | ||
|
|
||
| // findPrevious returns the previous state for the file. | ||
| // In case no previous state exists, index -1 is returned | ||
| func (s *States) findPrevious(id string) int { | ||
| if i, exists := s.idx[id]; exists { | ||
| return i | ||
| } | ||
| return -1 | ||
| } | ||
|
|
||
| // Cleanup cleans up the state array. All states which are older then `older` are removed | ||
| // The number of states that were cleaned up and number of states that can be | ||
| // cleaned up in the future is returned. | ||
| func (s *States) Cleanup() (int, int) { | ||
| s.Lock() | ||
| defer s.Unlock() | ||
|
|
||
| currentTime := time.Now() | ||
| statesBefore := len(s.states) | ||
| numCanExpire := 0 | ||
|
|
||
| L := len(s.states) | ||
| for i := 0; i < L; { | ||
| state := &s.states[i] | ||
| canExpire := state.TTL > 0 | ||
| expired := (canExpire && currentTime.Sub(state.Timestamp) > state.TTL) | ||
|
|
||
| if state.TTL == 0 || expired { | ||
| if !state.Finished { | ||
| logp.Err("State for %s should have been dropped, but couldn't as state is not finished.", state.Source) | ||
| i++ | ||
| continue | ||
| } | ||
|
|
||
| delete(s.idx, state.ID()) | ||
| logp.Debug("state", "State removed for %v because of older: %v", state.Source, state.TTL) | ||
|
|
||
| L-- | ||
| if L != i { | ||
| s.states[i] = s.states[L] | ||
| s.idx[s.states[i].ID()] = i | ||
| } | ||
| } else { | ||
| i++ | ||
| if canExpire { | ||
| numCanExpire++ | ||
| } | ||
| } | ||
| } | ||
|
|
||
| s.states = s.states[:L] | ||
| return statesBefore - L, numCanExpire | ||
| } | ||
|
|
||
| // Count returns number of states | ||
| func (s *States) Count() int { | ||
| s.RLock() | ||
| defer s.RUnlock() | ||
|
|
||
| return len(s.states) | ||
| } | ||
|
|
||
| // GetStates creates copy of the file states. | ||
| func (s *States) GetStates() []State { | ||
| s.RLock() | ||
| defer s.RUnlock() | ||
|
|
||
| newStates := make([]State, len(s.states)) | ||
| copy(newStates, s.states) | ||
|
|
||
| return newStates | ||
| } | ||
|
|
||
| // SetStates overwrites all internal states with the given states array | ||
| func (s *States) SetStates(states []State) { | ||
| s.Lock() | ||
| defer s.Unlock() | ||
| s.states = states | ||
|
|
||
| // create new index | ||
| s.idx = map[string]int{} | ||
| for i := range states { | ||
| s.idx[states[i].ID()] = i | ||
| } | ||
| } | ||
|
|
||
| // Copy create a new copy of the states object | ||
| func (s *States) Copy() *States { | ||
| new := NewStates() | ||
| new.SetStates(s.GetStates()) | ||
| return new | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.