72 lines
1.6 KiB
Go
72 lines
1.6 KiB
Go
package agent
|
||
|
||
import (
|
||
"context"
|
||
"math/rand"
|
||
"time"
|
||
|
||
"git.opencomputing.cn/yumoqing/host-metrics-collector/internal/model"
|
||
)
|
||
|
||
// collectFn 是某一指标组的采集函数。
|
||
type collectFn func() ([]model.Sample, error)
|
||
|
||
// runGroup 按固定间隔运行采集，并把样本投递到 out 通道。
|
||
//
|
||
// - 启动时加入随机 jitter（0 ~ interval/2），避免大量主机同秒上报（惊群）。
|
||
// - 每次采集带超时控制，超时或出错时记录错误并继续下一轮。
|
||
func runGroup(ctx context.Context, interval, timeout time.Duration, name string, fn collectFn, out chan<- model.Sample, errc chan<- error) {
|
||
// 首次延迟打散。
|
||
first := time.Duration(rand.Int63n(int64(interval / 2)))
|
||
timer := time.NewTimer(first)
|
||
defer timer.Stop()
|
||
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case <-timer.C:
|
||
}
|
||
|
||
samples, err := collectWithTimeout(ctx, timeout, fn)
|
||
if err != nil {
|
||
select {
|
||
case errc <- err:
|
||
default:
|
||
}
|
||
}
|
||
for _, s := range samples {
|
||
select {
|
||
case out <- s:
|
||
case <-ctx.Done():
|
||
return
|
||
}
|
||
}
|
||
|
||
timer.Reset(interval)
|
||
}
|
||
}
|
||
|
||
// collectWithTimeout 在超时约束下执行采集函数。
|
||
func collectWithTimeout(ctx context.Context, timeout time.Duration, fn collectFn) ([]model.Sample, error) {
|
||
type result struct {
|
||
samples []model.Sample
|
||
err error
|
||
}
|
||
ch := make(chan result, 1)
|
||
|
||
go func() {
|
||
samples, err := fn()
|
||
ch <- result{samples: samples, err: err}
|
||
}()
|
||
|
||
select {
|
||
case r := <-ch:
|
||
return r.samples, r.err
|
||
case <-ctx.Done():
|
||
return nil, ctx.Err()
|
||
case <-time.After(timeout):
|
||
return nil, context.DeadlineExceeded
|
||
}
|
||
}
|