From fb801e2c66b42fda08cb6ea67ef9dbe54457ffa6 Mon Sep 17 00:00:00 2001 From: Pipeline Agent Date: Sat, 15 Aug 2026 01:36:51 +0800 Subject: [PATCH] =?UTF-8?q?develop:=20=E5=BC=80=E5=8F=91=20host-metrics-co?= =?UTF-8?q?llector=20=E9=87=87=E9=9B=86=E6=A8=A1=E5=9D=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- go.mod | 19 +- go.sum | 44 ++++ internal/agent/agent.go | 388 ++++++++++++++++++++++++++++++++++++ internal/agent/reporter.go | 136 +++++++++++++ internal/agent/scheduler.go | 71 +++++++ internal/collect/logtail.go | 10 +- internal/gateway/store.go | 292 +++++++++++++++++++++++++++ internal/wal/wal.go | 146 ++++++++++++++ internal/wal/wal_test.go | 94 +++++++++ pkg/convert/convert.go | 60 ++++++ 10 files changed, 1255 insertions(+), 5 deletions(-) create mode 100644 internal/agent/agent.go create mode 100644 internal/agent/reporter.go create mode 100644 internal/agent/scheduler.go create mode 100644 internal/gateway/store.go create mode 100644 internal/wal/wal.go create mode 100644 internal/wal/wal_test.go create mode 100644 pkg/convert/convert.go diff --git a/go.mod b/go.mod index 56ab4bb..a090268 100644 --- a/go.mod +++ b/go.mod @@ -2,6 +2,21 @@ module git.opencomputing.cn/yumoqing/host-metrics-collector go 1.22 -require google.golang.org/protobuf v1.34.2 +require ( + github.com/redis/go-redis/v9 v9.5.1 + github.com/shirou/gopsutil/v4 v4.24.6 + google.golang.org/protobuf v1.34.2 +) -require github.com/google/go-cmp v0.6.0 // indirect +require ( + github.com/cespare/xxhash/v2 v2.2.0 // indirect + github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect + github.com/go-ole/go-ole v1.2.6 // indirect + github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect + github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect + github.com/shoenig/go-m1cpu v0.1.6 // indirect + github.com/tklauser/go-sysconf v0.3.12 // indirect + github.com/tklauser/numcpus v0.6.1 // indirect + github.com/yusufpapurcu/wmi v1.2.4 // indirect + golang.org/x/sys v0.20.0 // indirect +) diff --git a/go.sum b/go.sum index dd5797d..7137d69 100644 --- a/go.sum +++ b/go.sum @@ -1,4 +1,48 @@ +github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= +github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= +github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= +github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0= +github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44= +github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78= +github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc= +github.com/go-ole/go-ole v1.2.6 h1:/Fpf6oFPoeFik9ty7siob0G6Ke8QvQEuVcuChpwXzpY= +github.com/go-ole/go-ole v1.2.6/go.mod h1:pprOEPIfldk/42T2oK7lQ4v4JSDwmV0As9GaiUsvbm0= +github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI= github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= +github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 h1:6E+4a0GO5zZEnZ81pIr0yLvtUWk2if982qA3F3QD6H4= +github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0/go.mod h1:zJYVVT2jmtg6P3p1VtQj7WsuWi/y4VnjVBn7F8KPB3I= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c h1:ncq/mPwQF4JjgDlrVEn3C11VoGHZN7m8qihwgMEtzYw= +github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c/go.mod h1:OmDBASR4679mdNQnz2pUhc2G8CO2JrUAVFDRBDP/hJE= +github.com/redis/go-redis/v9 v9.5.1 h1:H1X4D3yHPaYrkL5X06Wh6xNVM/pX0Ft4RV0vMGvLBh8= +github.com/redis/go-redis/v9 v9.5.1/go.mod h1:hdY0cQFCN4fnSYT6TkisLufl/4W5UIXyv0b/CLO2V2M= +github.com/shirou/gopsutil/v4 v4.24.6 h1:9qqCSYF2pgOU+t+NgJtp7Co5+5mHF/HyKBUckySQL64= +github.com/shirou/gopsutil/v4 v4.24.6/go.mod h1:aoebb2vxetJ/yIDZISmduFvVNPHqXQ9SEJwRXxkf0RA= +github.com/shoenig/go-m1cpu v0.1.6 h1:nxdKQNcEB6vzgA2E2bvzKIYRuNj7XNJ4S/aRSwKzFtM= +github.com/shoenig/go-m1cpu v0.1.6/go.mod h1:1JJMcUBvfNwpq05QDQVAnx3gUHr9IYF7GNg9SUEw2VQ= +github.com/shoenig/test v0.6.4 h1:kVTaSd7WLz5WZ2IaoM0RSzRsUD+m8wRR+5qvntpn4LU= +github.com/shoenig/test v0.6.4/go.mod h1:byHiCGXqrVaflBLAMq/srcZIHynQPQgeyvkvXnjqq0k= +github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg= +github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +github.com/tklauser/go-sysconf v0.3.12 h1:0QaGUFOdQaIVdPgfITYzaTegZvdCjmYO52cSFAEVmqU= +github.com/tklauser/go-sysconf v0.3.12/go.mod h1:Ho14jnntGE1fpdOqQEEaiKRpvIavV0hSfmBq8nJbHYI= +github.com/tklauser/numcpus v0.6.1 h1:ng9scYS7az0Bk4OZLvrNXNSAO2Pxr1XXRAPyjhIx+Fk= +github.com/tklauser/numcpus v0.6.1/go.mod h1:1XfjsgE2zo8GVw7POkMbHENHzVg3GzmoZ9fESEdAacY= +github.com/yusufpapurcu/wmi v1.2.4 h1:zFUKzehAFReQwLys1b/iSMl+JQGSCSjtVqQn9bBrPo0= +github.com/yusufpapurcu/wmi v1.2.4/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0= +golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20201204225414-ed752295db88/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.11.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.20.0 h1:Od9JTbYCk261bKm4M/mw7AklTlFYIa0bIp9BgSm1S8Y= +golang.org/x/sys v0.20.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= google.golang.org/protobuf v1.34.2 h1:6xV6lTsCfpGD21XK49h7MhtcApnLqkfYgPcdHftf6hg= google.golang.org/protobuf v1.34.2/go.mod h1:qYOHts0dSfpeUzUFpOMr/WGzszTmLH+DiWniOlNbLDw= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/internal/agent/agent.go b/internal/agent/agent.go new file mode 100644 index 0000000..7e814dd --- /dev/null +++ b/internal/agent/agent.go @@ -0,0 +1,388 @@ +// 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 +} diff --git a/internal/agent/reporter.go b/internal/agent/reporter.go new file mode 100644 index 0000000..f1b71a1 --- /dev/null +++ b/internal/agent/reporter.go @@ -0,0 +1,136 @@ +package agent + +import ( + "bytes" + "context" + "fmt" + "io" + "net/http" + "time" + + "google.golang.org/protobuf/proto" + + "git.opencomputing.cn/yumoqing/host-metrics-collector/pkg/collectorpb" +) + +// Reporter 负责通过 HTTP/protobuf 向 Collector Gateway（4318）上报数据。 +type Reporter struct { + client *http.Client + gatewayURL string + agentToken string + hostID string + agentVersion string + + maxAttempts int + baseBackoff time.Duration +} + +// NewReporter 创建上报客户端。 +func NewReporter(gatewayURL, agentToken, hostID, agentVersion string) *Reporter { + return &Reporter{ + client: &http.Client{ + Timeout: 10 * time.Second, + }, + gatewayURL: gatewayURL, + agentToken: agentToken, + hostID: hostID, + agentVersion: agentVersion, + maxAttempts: 5, + baseBackoff: time.Second, + } +} + +// Heartbeat 发送心跳，返回服务端时间与下发配置。 +func (r *Reporter) Heartbeat(ctx context.Context, configVersion int64) (*collectorpb.HeartbeatResponse, error) { + req := &collectorpb.HeartbeatRequest{ + HostId: r.hostID, + AgentVersion: r.agentVersion, + ConfigVersion: configVersion, + } + out := &collectorpb.HeartbeatResponse{} + if err := r.sendWithRetry(ctx, "/v1/agent/heartbeat", req, out); err != nil { + return nil, err + } + return out, nil +} + +// SendMetrics 上报指标批次，返回服务端 Ack。 +func (r *Reporter) SendMetrics(ctx context.Context, batch *collectorpb.MetricBatch) (*collectorpb.Ack, error) { + out := &collectorpb.Ack{} + if err := r.sendWithRetry(ctx, "/v1/agent/metrics", batch, out); err != nil { + return nil, err + } + return out, nil +} + +// SendLogs 上报日志批次，返回服务端 Ack。 +func (r *Reporter) SendLogs(ctx context.Context, batch *collectorpb.LogBatch) (*collectorpb.Ack, error) { + out := &collectorpb.Ack{} + if err := r.sendWithRetry(ctx, "/v1/agent/logs", batch, out); err != nil { + return nil, err + } + return out, nil +} + +func (r *Reporter) sendWithRetry(ctx context.Context, path string, req proto.Message, out proto.Message) error { + var lastErr error + for attempt := 0; attempt < r.maxAttempts; attempt++ { + if attempt > 0 { + backoff := r.baseBackoff << (attempt - 1) + if backoff > 60*time.Second { + backoff = 60 * time.Second + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(backoff): + } + } + + err := r.post(ctx, path, req, out) + if err == nil { + if ack, ok := out.(*collectorpb.Ack); ok && !ack.Ok { + lastErr = fmt.Errorf("gateway ack error: %s", ack.Error) + continue + } + return nil + } + lastErr = err + } + return fmt.Errorf("report %s failed after %d attempts: %w", path, r.maxAttempts, lastErr) +} + +func (r *Reporter) post(ctx context.Context, path string, req proto.Message, out proto.Message) error { + body, err := proto.Marshal(req) + if err != nil { + return fmt.Errorf("marshal request: %w", err) + } + + httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, r.gatewayURL+path, bytes.NewReader(body)) + if err != nil { + return fmt.Errorf("new request: %w", err) + } + httpReq.Header.Set("Content-Type", "application/x-protobuf") + httpReq.Header.Set("X-Agent-Token", r.agentToken) + httpReq.Header.Set("X-Host-Id", r.hostID) + + resp, err := r.client.Do(httpReq) + if err != nil { + return fmt.Errorf("do request: %w", err) + } + defer resp.Body.Close() + + data, err := io.ReadAll(io.LimitReader(resp.Body, 4<<20)) + if err != nil { + return fmt.Errorf("read response: %w", err) + } + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("gateway returned %d: %s", resp.StatusCode, string(data)) + } + if out != nil && len(data) > 0 { + if err := proto.Unmarshal(data, out); err != nil { + return fmt.Errorf("unmarshal response: %w", err) + } + } + return nil +} diff --git a/internal/agent/scheduler.go b/internal/agent/scheduler.go new file mode 100644 index 0000000..70e465b --- /dev/null +++ b/internal/agent/scheduler.go @@ -0,0 +1,71 @@ +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 + } +} diff --git a/internal/collect/logtail.go b/internal/collect/logtail.go index 6771fff..d77fdd7 100644 --- a/internal/collect/logtail.go +++ b/internal/collect/logtail.go @@ -108,13 +108,17 @@ func (t *LogTail) emit(line []byte, offset int64) { File: t.path, Offset: offset, } + + // 先投递给上层（进入内存缓冲），再推进 offset；若进程在两步之间退出， + // 重启后会从旧 offset 重读，得到「至少一次」语义，避免丢日志。 + if t.onEntry != nil { + t.onEntry(rec) + } + // 检查点写入失败不阻断采集；下轮会按旧 offset 重读（至少一次语义）。 if err := t.store.SetLogOffset(t.path, offset); err != nil { // 可选：由上层记录日志，这里保持轻量。 } - if t.onEntry != nil { - t.onEntry(rec) - } } // detectLevel 从日志内容中识别常见级别，未命中默认 INFO。 diff --git a/internal/gateway/store.go b/internal/gateway/store.go new file mode 100644 index 0000000..45fc2e3 --- /dev/null +++ b/internal/gateway/store.go @@ -0,0 +1,292 @@ +// Package gateway 实现 Collector Gateway：接收 Agent 上报、提供指标/日志 +// 查询 REST API，并维护内存存储与已确认序号。 +package gateway + +import ( + "fmt" + "sort" + "strings" + "sync" + "time" + + "git.opencomputing.cn/yumoqing/host-metrics-collector/internal/model" +) + +// MetricPoint 是时序查询返回的单点。 +type MetricPoint struct { + Timestamp int64 `json:"timestamp"` + Value float64 `json:"value"` +} + +// HostMetricSeries 是一组相同 metric + labels 的时序数据。 +type HostMetricSeries struct { + Metric string `json:"metric"` + Labels map[string]string `json:"labels"` + Values []MetricPoint `json:"values"` +} + +// LogEntryDTO 是日志查询返回的单条日志。 +type LogEntryDTO struct { + Timestamp int64 `json:"timestamp"` + Level string `json:"level"` + Source string `json:"source"` + Message string `json:"message"` + Fields map[string]string `json:"fields,omitempty"` + File string `json:"file"` + Offset int64 `json:"offset"` +} + +// Store 是 Gateway 的线程安全内存存储。生产环境可替换为 Kafka sink + +// VictoriaMetrics/Elasticsearch，本模块为独立可运行实现。 +type Store struct { + mu sync.RWMutex + + metrics map[string][]model.Sample + logs map[string][]model.LogRecord + ackedSeqs map[string]int64 + lastSeqByHost map[string]int64 + + maxMetricsPerHost int + maxLogsPerHost int +} + +// NewStore 创建内存存储。 +func NewStore(maxMetricsPerHost, maxLogsPerHost int) *Store { + return &Store{ + metrics: map[string][]model.Sample{}, + logs: map[string][]model.LogRecord{}, + ackedSeqs: map[string]int64{}, + lastSeqByHost: map[string]int64{}, + maxMetricsPerHost: maxMetricsPerHost, + maxLogsPerHost: maxLogsPerHost, + } +} + +// AddMetrics 追加指标样本并记录最新 seq。 +func (s *Store) AddMetrics(hostID string, seq int64, samples []model.Sample) { + s.mu.Lock() + defer s.mu.Unlock() + + existing := s.metrics[hostID] + existing = append(existing, samples...) + if max := s.maxMetricsPerHost; max > 0 && len(existing) > max { + existing = existing[len(existing)-max:] + } + s.metrics[hostID] = existing + if seq > s.lastSeqByHost[hostID] { + s.lastSeqByHost[hostID] = seq + } +} + +// AddLogs 追加日志记录并记录最新 seq。 +func (s *Store) AddLogs(hostID string, seq int64, entries []model.LogRecord) { + s.mu.Lock() + defer s.mu.Unlock() + + existing := s.logs[hostID] + existing = append(existing, entries...) + if max := s.maxLogsPerHost; max > 0 && len(existing) > max { + existing = existing[len(existing)-max:] + } + s.logs[hostID] = existing + if seq > s.lastSeqByHost[hostID] { + s.lastSeqByHost[hostID] = seq + } +} + +// LastAckedSeq 返回服务端已确认（最近收到）的序号。 +func (s *Store) LastAckedSeq(hostID string) int64 { + s.mu.RLock() + defer s.mu.RUnlock() + return s.ackedSeqs[hostID] +} + +// SetLastAckedSeq 推进服务端已确认序号。 +func (s *Store) SetLastAckedSeq(hostID string, seq int64) { + s.mu.Lock() + defer s.mu.Unlock() + if seq > s.ackedSeqs[hostID] { + s.ackedSeqs[hostID] = seq + } +} + +// HostMetrics 查询单主机指标时序；metric 为空则返回全部指标。 +func (s *Store) HostMetrics(hostID, metric string, from, to time.Time) ([]HostMetricSeries, error) { + s.mu.RLock() + samples := append([]model.Sample(nil), s.metrics[hostID]...) + s.mu.RUnlock() + return groupSeries(samples, metric, nil, from, to), nil +} + +// QueryMetrics 跨主机按 metric 与标签过滤查询时序。 +func (s *Store) QueryMetrics(metric string, labels map[string]string, from, to time.Time) ([]HostMetricSeries, error) { + s.mu.RLock() + var samples []model.Sample + for _, ss := range s.metrics { + samples = append(samples, ss...) + } + s.mu.RUnlock() + return groupSeries(samples, metric, labels, from, to), nil +} + +// InstantMetrics 返回每个匹配序列的最新点。 +func (s *Store) InstantMetrics(metric string, labels map[string]string) ([]MetricPoint, error) { + series, err := s.QueryMetrics(metric, labels, time.Time{}, time.Now().Add(time.Hour)) + if err != nil { + return nil, err + } + out := make([]MetricPoint, 0, len(series)) + for _, ser := range series { + if len(ser.Values) == 0 { + continue + } + out = append(out, ser.Values[len(ser.Values)-1]) + } + return out, nil +} + +// HostLogs 查询单主机日志（支持级别、关键字、时间范围、分页）。 +func (s *Store) HostLogs(hostID, level, keyword string, from, to time.Time, page, pageSize int) (int, []LogEntryDTO, error) { + s.mu.RLock() + entries := append([]model.LogRecord(nil), s.logs[hostID]...) + s.mu.RUnlock() + return filterLogs(entries, level, keyword, from, to, page, pageSize) +} + +// SearchLogs 跨主机检索日志。 +func (s *Store) SearchLogs(hostIDs []string, level, keyword string, from, to time.Time, page, pageSize int) (int, []LogEntryDTO, error) { + s.mu.RLock() + var entries []model.LogRecord + if len(hostIDs) == 0 { + for _, es := range s.logs { + entries = append(entries, es...) + } + } else { + for _, id := range hostIDs { + entries = append(entries, s.logs[id]...) + } + } + s.mu.RUnlock() + return filterLogs(entries, level, keyword, from, to, page, pageSize) +} + +func groupSeries(samples []model.Sample, metric string, labels map[string]string, from, to time.Time) []HostMetricSeries { + type key struct { + metric string + labels string + } + groups := map[key]*HostMetricSeries{} + var order []key + + for _, s := range samples { + if metric != "" && s.Name != metric { + continue + } + if !matchLabels(s.Labels, labels) { + continue + } + ts := s.Timestamp + if !from.IsZero() && ts < from.Unix() { + continue + } + if !to.IsZero() && ts > to.Unix() { + continue + } + + k := key{metric: s.Name, labels: labelsString(s.Labels)} + ser, ok := groups[k] + if !ok { + ser = &HostMetricSeries{Metric: s.Name, Labels: s.Labels} + groups[k] = ser + order = append(order, k) + } + ser.Values = append(ser.Values, MetricPoint{Timestamp: s.Timestamp, Value: s.Value}) + } + + out := make([]HostMetricSeries, 0, len(order)) + for _, k := range order { + ser := groups[k] + sort.Slice(ser.Values, func(i, j int) bool { return ser.Values[i].Timestamp < ser.Values[j].Timestamp }) + out = append(out, *ser) + } + return out +} + +func filterLogs(entries []model.LogRecord, level, keyword string, from, to time.Time, page, pageSize int) (int, []LogEntryDTO, error) { + if page < 1 { + page = 1 + } + if pageSize < 1 { + pageSize = 20 + } + + var matched []model.LogRecord + for _, e := range entries { + if level != "" && !strings.EqualFold(e.Level, level) { + continue + } + if keyword != "" && !strings.Contains(e.Message, keyword) { + continue + } + if !from.IsZero() && e.Timestamp < from.Unix() { + continue + } + if !to.IsZero() && e.Timestamp > to.Unix() { + continue + } + matched = append(matched, e) + } + + total := len(matched) + start := (page - 1) * pageSize + if start > total { + start = total + } + end := start + pageSize + if end > total { + end = total + } + + items := make([]LogEntryDTO, 0, end-start) + for _, e := range matched[start:end] { + items = append(items, LogEntryDTO{ + Timestamp: e.Timestamp, + Level: e.Level, + Source: e.Source, + Message: e.Message, + Fields: e.Fields, + File: e.File, + Offset: e.Offset, + }) + } + return total, items, nil +} + +func matchLabels(got, want map[string]string) bool { + if len(want) == 0 { + return true + } + for k, v := range want { + if got[k] != v { + return false + } + } + return true +} + +func labelsString(labels map[string]string) string { + if len(labels) == 0 { + return "" + } + keys := make([]string, 0, len(labels)) + for k := range labels { + keys = append(keys, k) + } + sort.Strings(keys) + + var b strings.Builder + for _, k := range keys { + fmt.Fprintf(&b, "%s=%s;", k, labels[k]) + } + return b.String() +} diff --git a/internal/wal/wal.go b/internal/wal/wal.go new file mode 100644 index 0000000..71e183b --- /dev/null +++ b/internal/wal/wal.go @@ -0,0 +1,146 @@ +// Package wal 实现 Agent 本地 WAL（Write-Ahead Log）。 +// +// WAL 以追加方式持久化待上报的批次，配合 checkpoint.Store 记录的已确认 +// 序号，实现 Agent 进程重启后从未确认点继续上报（断点续传）。每个批次 +// 使用单调递增的 seq 标识，写入顺序与发送顺序一致。 +package wal + +import ( + "encoding/binary" + "fmt" + "io" + "os" + "sync" +) + +// Record 是 WAL 中的一条记录。Payload 为已序列化的批次（protobuf 字节）。 +type Record struct { + Seq int64 + Payload []byte +} + +// WAL 是一个简单的追加式文件日志。记录格式： +// +// [seq:8字节 big-endian][len:4字节 big-endian][payload:len字节] +type WAL struct { + path string + + mu sync.Mutex + f *os.File +} + +// Open 打开（不存在则创建）指定路径的 WAL 文件。 +func Open(path string) (*WAL, error) { + f, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o644) + if err != nil { + return nil, fmt.Errorf("wal: open %s: %w", path, err) + } + return &WAL{path: path, f: f}, nil +} + +// Append 追加一条记录并立即刷盘，保证进程崩溃后可恢复。 +func (w *WAL) Append(seq int64, payload []byte) error { + w.mu.Lock() + defer w.mu.Unlock() + + var hdr [12]byte + binary.BigEndian.PutUint64(hdr[0:8], uint64(seq)) + binary.BigEndian.PutUint32(hdr[8:12], uint32(len(payload))) + + if _, err := w.f.Write(hdr[:]); err != nil { + return fmt.Errorf("wal: write header: %w", err) + } + if _, err := w.f.Write(payload); err != nil { + return fmt.Errorf("wal: write payload: %w", err) + } + if err := w.f.Sync(); err != nil { + return fmt.Errorf("wal: sync: %w", err) + } + return nil +} + +// Load 读取 WAL 中的全部记录，按写入顺序返回。 +func (w *WAL) Load() ([]Record, error) { + w.mu.Lock() + defer w.mu.Unlock() + return w.loadLocked() +} + +func (w *WAL) loadLocked() ([]Record, error) { + if _, err := w.f.Seek(0, io.SeekStart); err != nil { + return nil, fmt.Errorf("wal: seek: %w", err) + } + + var out []Record + for { + var hdr [12]byte + _, err := io.ReadFull(w.f, hdr[:]) + switch { + case err == io.EOF: + return out, nil + case err == io.ErrUnexpectedEOF: + // 尾部存在半条记录（进程被强杀），视为损坏并忽略。 + return out, nil + case err != nil: + return nil, fmt.Errorf("wal: read header: %w", err) + } + + seq := int64(binary.BigEndian.Uint64(hdr[0:8])) + n := binary.BigEndian.Uint32(hdr[8:12]) + payload := make([]byte, n) + if _, err := io.ReadFull(w.f, payload); err != nil { + return nil, fmt.Errorf("wal: read payload: %w", err) + } + out = append(out, Record{Seq: seq, Payload: payload}) + } +} + +// TruncateBefore 删除 seq 小于 keepSeq 的记录（即保留 seq >= keepSeq）。 +// 通常在服务端确认后调用，以避免 WAL 无限增长。 +func (w *WAL) TruncateBefore(keepSeq int64) error { + w.mu.Lock() + defer w.mu.Unlock() + + recs, err := w.loadLocked() + if err != nil { + return err + } + + kept := make([]Record, 0, len(recs)) + for _, r := range recs { + if r.Seq >= keepSeq { + kept = append(kept, r) + } + } + + if err := w.f.Truncate(0); err != nil { + return fmt.Errorf("wal: truncate: %w", err) + } + if _, err := w.f.Seek(0, io.SeekStart); err != nil { + return fmt.Errorf("wal: seek: %w", err) + } + for _, r := range kept { + var hdr [12]byte + binary.BigEndian.PutUint64(hdr[0:8], uint64(r.Seq)) + binary.BigEndian.PutUint32(hdr[8:12], uint32(len(r.Payload))) + if _, err := w.f.Write(hdr[:]); err != nil { + return fmt.Errorf("wal: rewrite header: %w", err) + } + if _, err := w.f.Write(r.Payload); err != nil { + return fmt.Errorf("wal: rewrite payload: %w", err) + } + } + return w.f.Sync() +} + +// Close 关闭底层文件。 +func (w *WAL) Close() error { + w.mu.Lock() + defer w.mu.Unlock() + if w.f == nil { + return nil + } + err := w.f.Close() + w.f = nil + return err +} diff --git a/internal/wal/wal_test.go b/internal/wal/wal_test.go new file mode 100644 index 0000000..6cb2d4d --- /dev/null +++ b/internal/wal/wal_test.go @@ -0,0 +1,94 @@ +package wal + +import ( + "path/filepath" + "testing" +) + +func TestWALAppendLoad(t *testing.T) { + path := filepath.Join(t.TempDir(), "wal.bin") + w, err := Open(path) + if err != nil { + t.Fatalf("Open: %v", err) + } + defer w.Close() + + if err := w.Append(1, []byte("a")); err != nil { + t.Fatalf("Append: %v", err) + } + if err := w.Append(2, []byte("bb")); err != nil { + t.Fatalf("Append: %v", err) + } + + recs, err := w.Load() + if err != nil { + t.Fatalf("Load: %v", err) + } + if len(recs) != 2 { + t.Fatalf("len(recs)=%d, want 2", len(recs)) + } + if recs[0].Seq != 1 || string(recs[0].Payload) != "a" { + t.Fatalf("recs[0]=%+v", recs[0]) + } + if recs[1].Seq != 2 || string(recs[1].Payload) != "bb" { + t.Fatalf("recs[1]=%+v", recs[1]) + } +} + +func TestWALTruncateBefore(t *testing.T) { + path := filepath.Join(t.TempDir(), "wal.bin") + w, err := Open(path) + if err != nil { + t.Fatalf("Open: %v", err) + } + defer w.Close() + + for i := int64(1); i <= 5; i++ { + if err := w.Append(i, []byte{byte(i)}); err != nil { + t.Fatalf("Append %d: %v", i, err) + } + } + + if err := w.TruncateBefore(3); err != nil { + t.Fatalf("TruncateBefore: %v", err) + } + recs, err := w.Load() + if err != nil { + t.Fatalf("Load: %v", err) + } + if len(recs) != 3 { + t.Fatalf("len(recs)=%d, want 3", len(recs)) + } + for i, want := range []int64{3, 4, 5} { + if recs[i].Seq != want { + t.Fatalf("recs[%d].Seq=%d, want %d", i, recs[i].Seq, want) + } + } +} + +func TestWALReopenPersists(t *testing.T) { + path := filepath.Join(t.TempDir(), "wal.bin") + w, err := Open(path) + if err != nil { + t.Fatalf("Open: %v", err) + } + if err := w.Append(7, []byte("persist")); err != nil { + t.Fatalf("Append: %v", err) + } + if err := w.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + + w2, err := Open(path) + if err != nil { + t.Fatalf("Reopen: %v", err) + } + defer w2.Close() + recs, err := w2.Load() + if err != nil { + t.Fatalf("Load: %v", err) + } + if len(recs) != 1 || recs[0].Seq != 7 || string(recs[0].Payload) != "persist" { + t.Fatalf("recs=%+v", recs) + } +} diff --git a/pkg/convert/convert.go b/pkg/convert/convert.go new file mode 100644 index 0000000..3cd7f2c --- /dev/null +++ b/pkg/convert/convert.go @@ -0,0 +1,60 @@ +// Package convert 提供领域模型（internal/model）与 protobuf 数据契约 +// （pkg/collectorpb）之间的转换函数，供 Agent 上报与 Gateway 接收两端共用。 +package convert + +import ( + "git.opencomputing.cn/yumoqing/host-metrics-collector/internal/model" + "git.opencomputing.cn/yumoqing/host-metrics-collector/pkg/collectorpb" +) + +// SampleToPB 将内部指标样本转换为 protobuf MetricSample。 +func SampleToPB(s model.Sample) *collectorpb.MetricSample { + return &collectorpb.MetricSample{ + Name: s.Name, + Value: s.Value, + Timestamp: s.Timestamp, + Labels: s.Labels, + } +} + +// SampleFromPB 将 protobuf MetricSample 转换为内部指标样本。 +func SampleFromPB(s *collectorpb.MetricSample) model.Sample { + if s == nil { + return model.Sample{} + } + return model.Sample{ + Name: s.Name, + Value: s.Value, + Timestamp: s.Timestamp, + Labels: s.Labels, + } +} + +// LogToPB 将内部日志记录转换为 protobuf LogEntry。 +func LogToPB(r model.LogRecord) *collectorpb.LogEntry { + return &collectorpb.LogEntry{ + Timestamp: r.Timestamp, + Level: r.Level, + Source: r.Source, + Message: r.Message, + Fields: r.Fields, + File: r.File, + Offset: r.Offset, + } +} + +// LogFromPB 将 protobuf LogEntry 转换为内部日志记录。 +func LogFromPB(e *collectorpb.LogEntry) model.LogRecord { + if e == nil { + return model.LogRecord{} + } + return model.LogRecord{ + Timestamp: e.Timestamp, + Level: e.Level, + Source: e.Source, + Message: e.Message, + Fields: e.Fields, + File: e.File, + Offset: e.Offset, + } +}