389 lines
9.5 KiB
Go
389 lines
9.5 KiB
Go
// Package agent 实现 host-metrics-collector 的 Agent 端：定时采集、文件
|
||
// 日志 tail、本地 WAL、HTTP/protobuf 上报与断点续传。
|
||
package agent
|
||
|
||
import (
|
||
"context"
|
||
"log"
|
||
"sync"
|
||
"time"
|
||
|
||
"google.golang.org/protobuf/proto"
|
||
|
||
"git.opencomputing.cn/yumoqing/host-metrics-collector/internal/checkpoint"
|
||
"git.opencomputing.cn/yumoqing/host-metrics-collector/internal/collect"
|
||
"git.opencomputing.cn/yumoqing/host-metrics-collector/internal/config"
|
||
"git.opencomputing.cn/yumoqing/host-metrics-collector/internal/model"
|
||
"git.opencomputing.cn/yumoqing/host-metrics-collector/internal/wal"
|
||
"git.opencomputing.cn/yumoqing/host-metrics-collector/pkg/collectorpb"
|
||
"git.opencomputing.cn/yumoqing/host-metrics-collector/pkg/convert"
|
||
)
|
||
|
||
// 检查点键：指标与日志使用独立的 seq 空间，均以 host_id 为前缀。
|
||
const (
|
||
metricsSeqKey = "metrics:seq:"
|
||
logsSeqKey = "logs:seq:"
|
||
)
|
||
|
||
// Agent 聚合采集、缓冲、WAL 与上报。
|
||
type Agent struct {
|
||
cfg config.AgentConfig
|
||
store checkpoint.Store
|
||
collector *collect.Collector
|
||
reporter *Reporter
|
||
|
||
metricsWAL *wal.WAL
|
||
logsWAL *wal.WAL
|
||
|
||
metricsCh chan model.Sample
|
||
logBuf *logBuffer
|
||
errCh chan error
|
||
|
||
mu sync.Mutex
|
||
metricsSeq int64
|
||
logsSeq int64
|
||
}
|
||
|
||
// New 根据配置构建 Agent。
|
||
func New(cfg config.AgentConfig) (*Agent, error) {
|
||
store, err := checkpoint.New(cfg.CheckpointDriver, cfg.CheckpointFile, cfg.RedisURL)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
metricsWAL, err := wal.Open(cfg.CheckpointFile + ".metrics.wal")
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
logsWAL, err := wal.Open(cfg.CheckpointFile + ".logs.wal")
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
a := &Agent{
|
||
cfg: cfg,
|
||
store: store,
|
||
collector: collect.NewCollector(nil, cfg.HostID, cfg.Service),
|
||
reporter: NewReporter(cfg.GatewayURL, cfg.AgentToken, cfg.HostID, cfg.AgentVersion),
|
||
metricsWAL: metricsWAL,
|
||
logsWAL: logsWAL,
|
||
metricsCh: make(chan model.Sample, 4096),
|
||
logBuf: newLogBuffer(),
|
||
errCh: make(chan error, 16),
|
||
}
|
||
if err := a.initSeq(); err != nil {
|
||
return nil, err
|
||
}
|
||
return a, nil
|
||
}
|
||
|
||
// Run 启动 Agent 并阻塞直到 ctx 取消；取消时会尽力 flush 剩余数据。
|
||
func (a *Agent) Run(ctx context.Context) error {
|
||
// 恢复未确认批次（断点续传）。
|
||
if err := a.drainMetrics(ctx); err != nil {
|
||
log.Printf("[agent] initial metrics drain: %v", err)
|
||
}
|
||
if err := a.drainLogs(ctx); err != nil {
|
||
log.Printf("[agent] initial logs drain: %v", err)
|
||
}
|
||
|
||
go a.heartbeatLoop(ctx)
|
||
go a.logErrors(ctx)
|
||
|
||
// 指标采集：核心 15s / 磁盘 30s / 网络 30s / 进程 60s。
|
||
go runGroup(ctx, a.cfg.CoreInterval, a.cfg.CollectTimeout, "core", a.collector.CollectCore, a.metricsCh, a.errCh)
|
||
go runGroup(ctx, a.cfg.DiskInterval, a.cfg.CollectTimeout, "disk", a.collector.CollectDisk, a.metricsCh, a.errCh)
|
||
go runGroup(ctx, a.cfg.NetworkInterval, a.cfg.CollectTimeout, "network", a.collector.CollectNetwork, a.metricsCh, a.errCh)
|
||
go runGroup(ctx, a.cfg.ProcessInterval, a.cfg.CollectTimeout, "process", a.collector.CollectProcess, a.metricsCh, a.errCh)
|
||
|
||
// 日志采集：每个配置的日志文件一个 tail goroutine。
|
||
for _, file := range a.cfg.LogFiles {
|
||
file := file
|
||
tail := collect.NewLogTail(file, a.store, time.Second, a.logBuf.Add)
|
||
go tail.Run(ctx)
|
||
}
|
||
|
||
// 上报循环。
|
||
go a.flushMetricsLoop(ctx)
|
||
go a.flushLogsLoop(ctx)
|
||
|
||
<-ctx.Done()
|
||
|
||
// 最后 flush：使用独立的超时上下文，避免因父 ctx 取消立即失败。
|
||
finalCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||
defer cancel()
|
||
|
||
select {
|
||
case s := <-a.metricsCh:
|
||
_ = s // 通道在关闭时不会再投递，这里仅吞掉可能的残余。
|
||
default:
|
||
}
|
||
if err := a.flushMetrics(finalCtx, nil); err != nil {
|
||
log.Printf("[agent] final metrics flush: %v", err)
|
||
}
|
||
if err := a.flushLogs(finalCtx, nil); err != nil {
|
||
log.Printf("[agent] final logs flush: %v", err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// Close 释放 WAL 与检查点存储。
|
||
func (a *Agent) Close() error {
|
||
if err := a.metricsWAL.Close(); err != nil {
|
||
return err
|
||
}
|
||
if err := a.logsWAL.Close(); err != nil {
|
||
return err
|
||
}
|
||
return a.store.Close()
|
||
}
|
||
|
||
func (a *Agent) heartbeatLoop(ctx context.Context) {
|
||
ticker := time.NewTicker(30 * time.Second)
|
||
defer ticker.Stop()
|
||
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case <-ticker.C:
|
||
if _, err := a.reporter.Heartbeat(ctx, 0); err != nil {
|
||
log.Printf("[agent] heartbeat: %v", err)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
func (a *Agent) logErrors(ctx context.Context) {
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case err := <-a.errCh:
|
||
log.Printf("[agent] collect: %v", err)
|
||
}
|
||
}
|
||
}
|
||
|
||
// initSeq 从检查点与 WAL 恢复下一个可分配的 seq。
|
||
func (a *Agent) initSeq() error {
|
||
a.mu.Lock()
|
||
defer a.mu.Unlock()
|
||
|
||
ackedM, err := a.store.GetAckedSeq(metricsSeqKey + a.cfg.HostID)
|
||
if err != nil {
|
||
ackedM = 0
|
||
}
|
||
a.metricsSeq = ackedM
|
||
if recs, err := a.metricsWAL.Load(); err == nil {
|
||
for _, r := range recs {
|
||
if r.Seq > a.metricsSeq {
|
||
a.metricsSeq = r.Seq
|
||
}
|
||
}
|
||
}
|
||
|
||
ackedL, err := a.store.GetAckedSeq(logsSeqKey + a.cfg.HostID)
|
||
if err != nil {
|
||
ackedL = 0
|
||
}
|
||
a.logsSeq = ackedL
|
||
if recs, err := a.logsWAL.Load(); err == nil {
|
||
for _, r := range recs {
|
||
if r.Seq > a.logsSeq {
|
||
a.logsSeq = r.Seq
|
||
}
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (a *Agent) nextMetricsSeq() int64 {
|
||
a.mu.Lock()
|
||
defer a.mu.Unlock()
|
||
a.metricsSeq++
|
||
return a.metricsSeq
|
||
}
|
||
|
||
func (a *Agent) nextLogsSeq() int64 {
|
||
a.mu.Lock()
|
||
defer a.mu.Unlock()
|
||
a.logsSeq++
|
||
return a.logsSeq
|
||
}
|
||
|
||
// flushMetrics 将内存缓冲样本写入 WAL 并尝试上报全部未确认批次。
|
||
func (a *Agent) flushMetrics(ctx context.Context, samples []model.Sample) error {
|
||
if len(samples) > 0 {
|
||
seq := a.nextMetricsSeq()
|
||
batch := &collectorpb.MetricBatch{
|
||
HostId: a.cfg.HostID,
|
||
AgentVersion: a.cfg.AgentVersion,
|
||
Seq: seq,
|
||
SentAt: time.Now().Unix(),
|
||
Samples: make([]*collectorpb.MetricSample, 0, len(samples)),
|
||
}
|
||
for _, s := range samples {
|
||
batch.Samples = append(batch.Samples, convert.SampleToPB(s))
|
||
}
|
||
payload, err := proto.Marshal(batch)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if err := a.metricsWAL.Append(seq, payload); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return a.drainMetrics(ctx)
|
||
}
|
||
|
||
func (a *Agent) drainMetrics(ctx context.Context) error {
|
||
key := metricsSeqKey + a.cfg.HostID
|
||
acked, err := a.store.GetAckedSeq(key)
|
||
if err != nil {
|
||
acked = 0
|
||
}
|
||
recs, err := a.metricsWAL.Load()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
for _, r := range recs {
|
||
if r.Seq <= acked {
|
||
continue
|
||
}
|
||
var batch collectorpb.MetricBatch
|
||
if err := proto.Unmarshal(r.Payload, &batch); err != nil {
|
||
return err
|
||
}
|
||
if _, err := a.reporter.SendMetrics(ctx, &batch); err != nil {
|
||
return err
|
||
}
|
||
acked = r.Seq
|
||
if err := a.store.SetAckedSeq(key, acked); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return a.metricsWAL.TruncateBefore(acked + 1)
|
||
}
|
||
|
||
func (a *Agent) flushMetricsLoop(ctx context.Context) {
|
||
ticker := time.NewTicker(a.cfg.FlushInterval)
|
||
defer ticker.Stop()
|
||
|
||
buf := make([]model.Sample, 0, a.cfg.BatchSize)
|
||
for {
|
||
select {
|
||
case s := <-a.metricsCh:
|
||
buf = append(buf, s)
|
||
if len(buf) >= a.cfg.BatchSize {
|
||
if err := a.flushMetrics(ctx, buf); err != nil {
|
||
log.Printf("[agent] flush metrics: %v", err)
|
||
}
|
||
buf = buf[:0]
|
||
}
|
||
case <-ticker.C:
|
||
if err := a.flushMetrics(ctx, buf); err != nil {
|
||
log.Printf("[agent] flush metrics: %v", err)
|
||
}
|
||
buf = buf[:0]
|
||
case <-ctx.Done():
|
||
return
|
||
}
|
||
}
|
||
}
|
||
|
||
// flushLogs 将内存缓冲日志写入 WAL 并尝试上报全部未确认批次。
|
||
func (a *Agent) flushLogs(ctx context.Context, records []model.LogRecord) error {
|
||
if len(records) > 0 {
|
||
seq := a.nextLogsSeq()
|
||
batch := &collectorpb.LogBatch{
|
||
HostId: a.cfg.HostID,
|
||
Seq: seq,
|
||
Entries: make([]*collectorpb.LogEntry, 0, len(records)),
|
||
}
|
||
for _, r := range records {
|
||
batch.Entries = append(batch.Entries, convert.LogToPB(r))
|
||
}
|
||
payload, err := proto.Marshal(batch)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if err := a.logsWAL.Append(seq, payload); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return a.drainLogs(ctx)
|
||
}
|
||
|
||
func (a *Agent) drainLogs(ctx context.Context) error {
|
||
key := logsSeqKey + a.cfg.HostID
|
||
acked, err := a.store.GetAckedSeq(key)
|
||
if err != nil {
|
||
acked = 0
|
||
}
|
||
recs, err := a.logsWAL.Load()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
for _, r := range recs {
|
||
if r.Seq <= acked {
|
||
continue
|
||
}
|
||
var batch collectorpb.LogBatch
|
||
if err := proto.Unmarshal(r.Payload, &batch); err != nil {
|
||
return err
|
||
}
|
||
if _, err := a.reporter.SendLogs(ctx, &batch); err != nil {
|
||
return err
|
||
}
|
||
acked = r.Seq
|
||
if err := a.store.SetAckedSeq(key, acked); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return a.logsWAL.TruncateBefore(acked + 1)
|
||
}
|
||
|
||
func (a *Agent) flushLogsLoop(ctx context.Context) {
|
||
ticker := time.NewTicker(a.cfg.FlushInterval)
|
||
defer ticker.Stop()
|
||
|
||
for {
|
||
select {
|
||
case <-ticker.C:
|
||
records := a.logBuf.TakeAll()
|
||
if err := a.flushLogs(ctx, records); err != nil {
|
||
log.Printf("[agent] flush logs: %v", err)
|
||
}
|
||
case <-ctx.Done():
|
||
return
|
||
}
|
||
}
|
||
}
|
||
|
||
// logBuffer 是日志记录的内存缓冲，避免 tail goroutine 在通道满时阻塞或丢日志。
|
||
type logBuffer struct {
|
||
mu sync.Mutex
|
||
items []model.LogRecord
|
||
}
|
||
|
||
func newLogBuffer() *logBuffer {
|
||
return &logBuffer{items: make([]model.LogRecord, 0, 4096)}
|
||
}
|
||
|
||
func (b *logBuffer) Add(r model.LogRecord) {
|
||
b.mu.Lock()
|
||
defer b.mu.Unlock()
|
||
b.items = append(b.items, r)
|
||
}
|
||
|
||
func (b *logBuffer) TakeAll() []model.LogRecord {
|
||
b.mu.Lock()
|
||
defer b.mu.Unlock()
|
||
out := b.items
|
||
b.items = make([]model.LogRecord, 0, 4096)
|
||
return out
|
||
}
|