💥 feat(agent)!: 添加主机监控代理首版
* 实现周期采集、快照缓存、本地 HTTP API 和主动上报 * 补齐 Linux 和 Windows 主机、网络、资源、进程和 TCP 采集 * 添加服务生命周期管理、配置加载、构建脚本和测试覆盖 BREAKING CHANGE: network.ips 拆分为 network.ipv4s 和 network.ipv6s。
This commit is contained in:
104
internal/snapshot/snapshot.go
Normal file
104
internal/snapshot/snapshot.go
Normal file
@@ -0,0 +1,104 @@
|
||||
package snapshot
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"mfd-monitor-agent/internal/status"
|
||||
)
|
||||
|
||||
// Collector gathers one subsection of HostStatus into the provided snapshot.
|
||||
type Collector interface {
|
||||
// Name returns the stable collector name used in partial snapshot errors.
|
||||
Name() string
|
||||
// Collect fills its subsection of out and returns an error when that subsection is incomplete.
|
||||
Collect(context.Context, *status.HostStatus) error
|
||||
}
|
||||
|
||||
// CollectorFunc adapts a function into a Collector for tests and small collectors.
|
||||
type CollectorFunc func(context.Context, *status.HostStatus) error
|
||||
|
||||
// Name returns a generic collector name for function-backed collectors.
|
||||
func (fn CollectorFunc) Name() string {
|
||||
return "collector"
|
||||
}
|
||||
|
||||
// Collect calls the wrapped function.
|
||||
func (fn CollectorFunc) Collect(ctx context.Context, out *status.HostStatus) error {
|
||||
return fn(ctx, out)
|
||||
}
|
||||
|
||||
// Cache stores the most recent HostStatus snapshot behind a concurrency-safe interface.
|
||||
type Cache struct {
|
||||
collectors []Collector
|
||||
version string
|
||||
startedAt time.Time
|
||||
mu sync.RWMutex
|
||||
last status.HostStatus
|
||||
hasLast bool
|
||||
}
|
||||
|
||||
// NewCache creates an empty snapshot cache using the provided collectors and agent metadata.
|
||||
func NewCache(collectors []Collector, version string, startedAt time.Time) *Cache {
|
||||
return &Cache{collectors: collectors, version: version, startedAt: startedAt}
|
||||
}
|
||||
|
||||
// Collect runs all collectors once and stores the resulting snapshot even when some collectors fail.
|
||||
func (c *Cache) Collect(ctx context.Context) status.HostStatus {
|
||||
out := status.HostStatus{
|
||||
Agent: status.AgentInfo{
|
||||
Version: c.version,
|
||||
StartedAt: c.startedAt,
|
||||
SnapshotAt: time.Now().UTC(),
|
||||
SnapshotStatus: "ok",
|
||||
},
|
||||
Processes: status.ProcessResult{Status: "ok"},
|
||||
TCPListeners: status.TCPResult{Status: "ok"},
|
||||
}
|
||||
|
||||
for _, collector := range c.collectors {
|
||||
if err := collector.Collect(ctx, &out); err != nil {
|
||||
out.Agent.SnapshotStatus = "partial"
|
||||
out.Errors = append(out.Errors, status.StatusError{
|
||||
Collector: collector.Name(),
|
||||
Message: err.Error(),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
c.mu.Lock()
|
||||
c.last = out
|
||||
c.hasLast = true
|
||||
c.mu.Unlock()
|
||||
return out
|
||||
}
|
||||
|
||||
// Get returns the latest cached snapshot and whether collection has completed at least once.
|
||||
func (c *Cache) Get() (status.HostStatus, bool) {
|
||||
c.mu.RLock()
|
||||
defer c.mu.RUnlock()
|
||||
return c.last, c.hasLast
|
||||
}
|
||||
|
||||
// Run collects immediately and then refreshes the cache until ctx is canceled.
|
||||
func (c *Cache) Run(ctx context.Context, interval time.Duration, timeout time.Duration) {
|
||||
c.collectWithTimeout(ctx, timeout)
|
||||
ticker := time.NewTicker(interval)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
c.collectWithTimeout(ctx, timeout)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// collectWithTimeout runs one collection cycle with an overall deadline.
|
||||
func (c *Cache) collectWithTimeout(ctx context.Context, timeout time.Duration) {
|
||||
collectCtx, cancel := context.WithTimeout(ctx, timeout)
|
||||
defer cancel()
|
||||
c.Collect(collectCtx)
|
||||
}
|
||||
103
internal/snapshot/snapshot_test.go
Normal file
103
internal/snapshot/snapshot_test.go
Normal file
@@ -0,0 +1,103 @@
|
||||
package snapshot
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"mfd-monitor-agent/internal/status"
|
||||
)
|
||||
|
||||
// TestCollectorFailuresReturnPartialSnapshot verifies failed collectors do not fail snapshots.
|
||||
func TestCollectorFailuresReturnPartialSnapshot(t *testing.T) {
|
||||
cache := NewCache([]Collector{
|
||||
CollectorFunc(func(context.Context, *status.HostStatus) error {
|
||||
return nil
|
||||
}),
|
||||
CollectorFunc(func(context.Context, *status.HostStatus) error {
|
||||
return errors.New("permission denied")
|
||||
}),
|
||||
}, "test", time.Now())
|
||||
|
||||
got := cache.Collect(context.Background())
|
||||
|
||||
if got.Agent.SnapshotStatus != "partial" {
|
||||
t.Fatalf("snapshot status = %q", got.Agent.SnapshotStatus)
|
||||
}
|
||||
if len(got.Errors) != 1 || got.Errors[0].Message != "permission denied" {
|
||||
t.Fatalf("errors = %+v", got.Errors)
|
||||
}
|
||||
}
|
||||
|
||||
// TestCacheStoresLastSnapshot verifies successful collection is cached.
|
||||
func TestCacheStoresLastSnapshot(t *testing.T) {
|
||||
cache := NewCache([]Collector{
|
||||
CollectorFunc(func(_ context.Context, out *status.HostStatus) error {
|
||||
out.Host.Hostname = "host-a"
|
||||
return nil
|
||||
}),
|
||||
}, "test", time.Now())
|
||||
|
||||
cache.Collect(context.Background())
|
||||
got, ok := cache.Get()
|
||||
|
||||
if !ok {
|
||||
t.Fatal("expected cached snapshot")
|
||||
}
|
||||
if got.Host.Hostname != "host-a" {
|
||||
t.Fatalf("hostname = %q", got.Host.Hostname)
|
||||
}
|
||||
}
|
||||
|
||||
// TestRunCollectsImmediately verifies Run collects before the first interval tick.
|
||||
func TestRunCollectsImmediately(t *testing.T) {
|
||||
collected := make(chan struct{})
|
||||
cache := NewCache([]Collector{
|
||||
CollectorFunc(func(_ context.Context, out *status.HostStatus) error {
|
||||
out.Host.Hostname = "host-a"
|
||||
close(collected)
|
||||
return nil
|
||||
}),
|
||||
}, "test", time.Now())
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
go cache.Run(ctx, time.Hour, time.Second)
|
||||
|
||||
select {
|
||||
case <-collected:
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("initial collection did not run")
|
||||
}
|
||||
got, ok := cache.Get()
|
||||
if !ok {
|
||||
t.Fatal("expected cached snapshot")
|
||||
}
|
||||
if got.Host.Hostname != "host-a" {
|
||||
t.Fatalf("hostname = %q", got.Host.Hostname)
|
||||
}
|
||||
}
|
||||
|
||||
// TestCollectWithTimeoutMarksCanceledCollectorPartial verifies timeout errors are partial.
|
||||
func TestCollectWithTimeoutMarksCanceledCollectorPartial(t *testing.T) {
|
||||
cache := NewCache([]Collector{
|
||||
CollectorFunc(func(ctx context.Context, _ *status.HostStatus) error {
|
||||
<-ctx.Done()
|
||||
return ctx.Err()
|
||||
}),
|
||||
}, "test", time.Now())
|
||||
|
||||
cache.collectWithTimeout(context.Background(), time.Millisecond)
|
||||
got, ok := cache.Get()
|
||||
|
||||
if !ok {
|
||||
t.Fatal("expected cached snapshot")
|
||||
}
|
||||
if got.Agent.SnapshotStatus != "partial" {
|
||||
t.Fatalf("snapshot status = %q", got.Agent.SnapshotStatus)
|
||||
}
|
||||
if len(got.Errors) != 1 || got.Errors[0].Message != context.DeadlineExceeded.Error() {
|
||||
t.Fatalf("errors = %+v", got.Errors)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user