CL-Softwares
/
telegraf
8
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(regression): Fixes problem with metrics not exposed by plugins. (#12016)

This commit is contained in:
Paweł Żak 2022-10-14 16:48:47 +02:00 committed by GitHub
parent 90d8f426b3
commit ab293e853c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 48 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
agent/agent.go
View File

@ -280,20 +280,8 @@ func (a *Agent) runInputs(
unit *inputUnit, unit *inputUnit,
) { ) {
var wg sync.WaitGroup var wg sync.WaitGroup
tickers := make([]Ticker, len(unit.inputs))
for _, input := range unit.inputs { for _, input := range unit.inputs {
a.runInput(ctx, startTime, unit, input, &wg)
}
wg.Wait()
log.Printf("D! [agent] Stopping service inputs")
stopServiceInputs(unit.inputs)
close(unit.dst)
log.Printf("D! [agent] Input channel closed")
}
func (a *Agent) runInput(ctx context.Context, startTime time.Time, unit *inputUnit, input *models.RunningInput, wg *sync.WaitGroup) {
// Overwrite agent interval if this plugin has its own. // Overwrite agent interval if this plugin has its own.
interval := time.Duration(a.Config.Agent.Interval) interval := time.Duration(a.Config.Agent.Interval)
if input.Config.Interval != 0 { if input.Config.Interval != 0 {
@ -324,7 +312,7 @@ func (a *Agent) runInput(ctx context.Context, startTime time.Time, unit *inputUn
} else { } else {
ticker = NewUnalignedTicker(interval, jitter, offset) ticker = NewUnalignedTicker(interval, jitter, offset)
} }
defer ticker.Stop() tickers = append(tickers, ticker)
acc := NewAccumulator(input, unit.dst) acc := NewAccumulator(input, unit.dst)
acc.SetPrecision(getPrecision(precision, interval)) acc.SetPrecision(getPrecision(precision, interval))
@ -334,6 +322,15 @@ func (a *Agent) runInput(ctx context.Context, startTime time.Time, unit *inputUn
defer wg.Done() defer wg.Done()
a.gatherLoop(ctx, acc, input, ticker, interval) a.gatherLoop(ctx, acc, input, ticker, interval)
}(input) }(input)
}
defer stopTickers(tickers)
wg.Wait()
log.Printf("D! [agent] Stopping service inputs")
stopServiceInputs(unit.inputs)
close(unit.dst)
log.Printf("D! [agent] Input channel closed")
} }
// testStartInputs is a variation of startInputs for use in --test and --once // testStartInputs is a variation of startInputs for use in --test and --once
@ -1110,3 +1107,9 @@ func panicRecover(input *models.RunningInput) {
"https://github.com/influxdata/telegraf/issues/new/choose") "https://github.com/influxdata/telegraf/issues/new/choose")
} }
} }
func stopTickers(tickers []Ticker) {
for _, ticker := range tickers {
ticker.Stop()
}
}