yumoqing/host-metrics-collector
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

develop: 开发 host-metrics-collector 采集模块

Browse Source
This commit is contained in:
Pipeline Agent 2026-08-15 01:36:51 +08:00
parent 7d60597db3
commit fb801e2c66
10 changed files with 1255 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
go.mod
View File

@ -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
)

44
go.sum
View File

@ -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=

388
internal/agent/agent.go Normal file
View File

@ -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
}

136
internal/agent/reporter.go Normal file
View File

@ -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
}

71
internal/agent/scheduler.go Normal file
View File

@ -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
}
}

10
internal/collect/logtail.go
View File

@ -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。

292
internal/gateway/store.go Normal file
View File

@ -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()
}

146
internal/wal/wal.go Normal file
View File

@ -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
}

94
internal/wal/wal_test.go Normal file
View File

@ -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)
}
}

60
pkg/convert/convert.go Normal file
View File

@ -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,
}
}