-
Notifications
You must be signed in to change notification settings - Fork 487
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Flow: Improve Kubernetes log collection (#5623)
* component/prometheus: fix panic in interceptor when child isn't set This commit fixes a panic in prometheus.Interceptor where an interceptor which doesn't forward samples to another appendable panics when appending data. Co-authored-by: Edward Welch <edward.welch@grafana.com> * loki.source.kubernetes: improve detection of rolled log files Versions of Kubernetes that do not contain kubernetes/kubernetes#115702 will fail to detect rolled log files, causing the API to stop sending logs to the agent for processing. To work around this, this commit intorduces a rolling average calculator to determine the average delta between log entries per target. If 3x the normal delta time has elapsed since the last entry, the tailer is restarted. False positives here are acceptable, but false negatives mean that log lines may not appear for an extended period of time until the rolling detection succeeds. Closes #5040 Co-authored-by: Edward Welch <edward.welch@grafana.com> * loki.source.kubernetes: support clustering Add support for loki.source.kubernetes to distribute targets using clustering. Closes #4502 Co-authored-by: Edward Welch <edward.welch@grafana.com> * loki.source.podlogs: support clustering Add support for loki.source.podlogs to distribute targets using clustering. * service/cluster: add common block for clustering arguments * remove irrelevant TODO comment #5623 (comment) --------- Co-authored-by: Edward Welch <edward.welch@grafana.com>
- Loading branch information
Showing
16 changed files
with
406 additions
and
39 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
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,89 @@ | ||
| package kubetail | ||
|
|
||
| import ( | ||
| "sync" | ||
| "time" | ||
| ) | ||
|
|
||
| // rollingAverageCalculator calculates a rolling average between points in | ||
| // time. | ||
| // | ||
| // rollingAverageCalculator stores a circular buffer where the difference | ||
| // between timestamps are kept. | ||
| type rollingAverageCalculator struct { | ||
| mtx sync.Mutex | ||
| window []time.Duration | ||
| windowSize int | ||
|
|
||
| minEntries int | ||
| minDuration time.Duration | ||
| defaultDuration time.Duration | ||
|
|
||
| currentIndex int | ||
| prevTimestamp time.Time | ||
| } | ||
|
|
||
| func newRollingAverageCalculator(windowSize, minEntries int, minDuration, defaultDuration time.Duration) *rollingAverageCalculator { | ||
| return &rollingAverageCalculator{ | ||
| windowSize: windowSize, | ||
| window: make([]time.Duration, windowSize), | ||
| minEntries: minEntries, | ||
| minDuration: minDuration, | ||
| defaultDuration: defaultDuration, | ||
| currentIndex: -1, | ||
| } | ||
| } | ||
|
|
||
| // AddTimestamp adds a new timestamp to the rollingAverageCalculator. If there | ||
| // is a previous timestamp, the difference between timestamps is calculated and | ||
| // stored in the window. | ||
| func (r *rollingAverageCalculator) AddTimestamp(timestamp time.Time) { | ||
| r.mtx.Lock() | ||
| defer func() { | ||
| r.prevTimestamp = timestamp | ||
| r.mtx.Unlock() | ||
| }() | ||
|
|
||
| // First timestamp | ||
| if r.currentIndex == -1 && r.prevTimestamp.Equal(time.Time{}) { | ||
| return | ||
| } | ||
|
|
||
| r.currentIndex++ | ||
| if r.currentIndex >= r.windowSize { | ||
| r.currentIndex = 0 | ||
| } | ||
|
|
||
| r.window[r.currentIndex] = timestamp.Sub(r.prevTimestamp) | ||
| } | ||
|
|
||
| // GetAverage calculates the average of all the durations in the window. | ||
| func (r *rollingAverageCalculator) GetAverage() time.Duration { | ||
| r.mtx.Lock() | ||
| defer r.mtx.Unlock() | ||
|
|
||
| var total time.Duration | ||
| count := 0 | ||
| for _, v := range r.window { | ||
| if v != 0 { | ||
| total += v | ||
| count++ | ||
| } | ||
| } | ||
| if count == 0 || count < r.minEntries { | ||
| return r.defaultDuration | ||
| } | ||
| d := total / time.Duration(count) | ||
| if d < r.minDuration { | ||
| return r.minDuration | ||
| } | ||
| return d | ||
| } | ||
|
|
||
| // GetLast gets the last timestamp added to the rollingAverageCalculator. | ||
| func (r *rollingAverageCalculator) GetLast() time.Time { | ||
| r.mtx.Lock() | ||
| defer r.mtx.Unlock() | ||
|
|
||
| return r.prevTimestamp | ||
| } |
86 changes: 86 additions & 0 deletions
86
component/loki/source/kubernetes/kubetail/tail_utils_test.go
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,86 @@ | ||
| package kubetail | ||
|
|
||
| import ( | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/stretchr/testify/assert" | ||
| ) | ||
|
|
||
| func Test(t *testing.T) { | ||
| tests := []struct { | ||
| name string | ||
| windowSize int | ||
| minEntries int | ||
| input []int64 | ||
| expected time.Duration | ||
| }{ | ||
| { | ||
| name: "empty, expect default", | ||
| windowSize: 10, | ||
| input: []int64{}, | ||
| expected: time.Duration(10), | ||
| }, | ||
| { | ||
| name: "one sample, not enough, expect default", | ||
| windowSize: 5, | ||
| input: []int64{10}, | ||
| expected: time.Duration(10), | ||
| }, | ||
| { | ||
| name: "partially full", | ||
| windowSize: 10, | ||
| input: []int64{10, 20, 30, 40, 50}, | ||
| expected: time.Duration(10), | ||
| }, | ||
| { | ||
| name: "completely full", | ||
| windowSize: 5, | ||
| input: []int64{10, 20, 30, 40, 50, 60}, | ||
| expected: time.Duration(10), | ||
| }, | ||
| { | ||
| name: "rollover simple", | ||
| windowSize: 5, | ||
| input: []int64{10, 20, 30, 40, 50, 60}, | ||
| expected: time.Duration(10), | ||
| }, | ||
| { | ||
| name: "rollover complex: make sure first value is ignored", | ||
| windowSize: 5, | ||
| input: []int64{0, 40, 50, 60, 70, 80, 90}, | ||
| expected: time.Duration(10), | ||
| }, | ||
| { | ||
| name: "complex", | ||
| windowSize: 5, | ||
| // 40 +1 +4 +45 +5 = 95, 95/5 = 19 | ||
| input: []int64{10, 50, 51, 55, 100, 105}, | ||
| expected: time.Duration(19), | ||
| }, | ||
| { | ||
| name: "complex 2", | ||
| windowSize: 10, | ||
| // outside of window | | ||
| // 40 +1 +4 +45 +|5 +5 +90 +100 +150 +300 +5 +45 +50 +149 = 899 | ||
| input: []int64{10, 50, 51, 55, 100, 105, 110, 200, 300, 450, 750, 755, 800, 850, 999}, | ||
| expected: time.Duration(89), // Integer math result is truncated not rounded. | ||
| }, | ||
| { | ||
| name: "below min duration", | ||
| windowSize: 5, | ||
| input: []int64{0, 1, 2, 3, 4, 5, 6}, | ||
| expected: time.Duration(2), | ||
| }, | ||
| } | ||
| for _, test := range tests { | ||
| t.Run(test.name, func(t *testing.T) { | ||
| c := newRollingAverageCalculator(test.windowSize, test.minEntries, time.Duration(2), time.Duration(10)) | ||
| for _, v := range test.input { | ||
| c.AddTimestamp(time.Unix(0, v)) | ||
| } | ||
| avg := c.GetAverage() | ||
| assert.Equal(t, test.expected, avg) | ||
| }) | ||
| } | ||
| } |
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.