develop: 重做 host-metrics-collector 采集模块（遵守模块开发规范）

internal/agent/scheduler_test.go

@@ -0,0 +1,104 @@
+package agent
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"git.opencomputing.cn/yumoqing/host-metrics-collector/internal/model"
+)
+
+func TestCollectWithTimeoutReturnsResult(t *testing.T) {
+	ctx := context.Background()
+	samples, err := collectWithTimeout(ctx, 2*time.Second, func() ([]model.Sample, error) {
+		return []model.Sample{{Name: "cpu.usage", Value: 1.0}}, nil
+	})
+	if err != nil {
+		t.Fatalf("collectWithTimeout returned error: %v", err)
+	}
+	if len(samples) != 1 || samples[0].Name != "cpu.usage" {
+		t.Fatalf("samples = %+v", samples)
+	}
+}
+
+func TestCollectWithTimeoutReturnsError(t *testing.T) {
+	ctx := context.Background()
+	wantErr := errors.New("collect failed")
+	_, err := collectWithTimeout(ctx, 2*time.Second, func() ([]model.Sample, error) {
+		return nil, wantErr
+	})
+	if !errors.Is(err, wantErr) {
+		t.Fatalf("err = %v, want %v", err, wantErr)
+	}
+}
+
+func TestCollectWithTimeoutTimesOut(t *testing.T) {
+	ctx := context.Background()
+	_, err := collectWithTimeout(ctx, 20*time.Millisecond, func() ([]model.Sample, error) {
+		time.Sleep(200 * time.Millisecond)
+		return nil, nil
+	})
+	if !errors.Is(err, context.DeadlineExceeded) {
+		t.Fatalf("err = %v, want context.DeadlineExceeded", err)
+	}
+}
+
+func TestCollectWithTimeoutRespectsContextCancel(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	_, err := collectWithTimeout(ctx, 2*time.Second, func() ([]model.Sample, error) {
+		return nil, nil
+	})
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("err = %v, want context.Canceled", err)
+	}
+}
+
+func TestRunGroupEmitsSamplesAndStops(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	out := make(chan model.Sample, 16)
+	errc := make(chan error, 16)
+	fn := func() ([]model.Sample, error) {
+		return []model.Sample{{Name: "mem.used_percent", Value: 50}}, nil
+	}
+
+	go runGroup(ctx, 20*time.Millisecond, time.Second, "core", fn, out, errc)
+
+	select {
+	case s := <-out:
+		if s.Name != "mem.used_percent" {
+			t.Fatalf("sample = %+v", s)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatalf("timed out waiting for sample")
+	}
+
+	cancel()
+}
+
+func TestRunGroupReportsCollectError(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	out := make(chan model.Sample, 1)
+	errc := make(chan error, 1)
+	fn := func() ([]model.Sample, error) {
+		return nil, errors.New("boom")
+	}
+
+	go runGroup(ctx, 20*time.Millisecond, time.Second, "core", fn, out, errc)
+
+	select {
+	case err := <-errc:
+		if err == nil {
+			t.Fatalf("expected non-nil error")
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatalf("timed out waiting for error")
+	}
+
+	cancel()
+}

internal/collect/logtail_test.go

@@ -0,0 +1,103 @@
+package collect
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"sync"
+	"testing"
+	"time"
+
+	"git.opencomputing.cn/yumoqing/host-metrics-collector/internal/checkpoint"
+	"git.opencomputing.cn/yumoqing/host-metrics-collector/internal/model"
+)
+
+func writeTempLog(t *testing.T, content string) string {
+	t.Helper()
+	path := filepath.Join(t.TempDir(), "app.log")
+	if err := os.WriteFile(path, []byte(content), 0o644); err != nil {
+		t.Fatalf("write temp log: %v", err)
+	}
+	return path
+}
+
+func TestLogTailReadOnceEmitsCompleteLines(t *testing.T) {
+	path := writeTempLog(t, "line one\nline two\n")
+
+	var mu sync.Mutex
+	var got []model.LogRecord
+	tail := NewLogTail(path, checkpoint.NewMemoryStore(), time.Second, func(r model.LogRecord) {
+		mu.Lock()
+		defer mu.Unlock()
+		got = append(got, r)
+	})
+
+	offset, partial := tail.readOnce(0, nil)
+	if offset != int64(len("line one\nline two\n")) {
+		t.Fatalf("offset = %d", offset)
+	}
+	if len(partial) != 0 {
+		t.Fatalf("partial = %q, want empty", partial)
+	}
+
+	mu.Lock()
+	defer mu.Unlock()
+	if len(got) != 2 || got[0].Message != "line one" || got[1].Message != "line two" {
+		t.Fatalf("records = %+v", got)
+	}
+	if got[0].Level != "INFO" {
+		t.Fatalf("level = %q, want INFO", got[0].Level)
+	}
+}
+
+func TestLogTailReadOnceKeepsPartialLine(t *testing.T) {
+	path := writeTempLog(t, "partial line")
+
+	tail := NewLogTail(path, checkpoint.NewMemoryStore(), time.Second, nil)
+	offset, partial := tail.readOnce(0, nil)
+	if offset != 0 {
+		t.Fatalf("offset = %d, want 0 (no newline yet)", offset)
+	}
+	if string(partial) != "partial line" {
+		t.Fatalf("partial = %q", partial)
+	}
+}
+
+func TestLogTailReadOnceHandlesTruncation(t *testing.T) {
+	path := writeTempLog(t, "old content longer\n")
+	tail := NewLogTail(path, checkpoint.NewMemoryStore(), time.Second, nil)
+
+	// 模拟文件被轮转/截断：offset 超出文件大小，应从头读取。
+	if err := os.WriteFile(path, []byte("new\n"), 0o644); err != nil {
+		t.Fatalf("truncate: %v", err)
+	}
+	offset, partial := tail.readOnce(100, []byte("tail"))
+	if offset != int64(len("new\n")) {
+		t.Fatalf("offset = %d, want %d", offset, len("new\n"))
+	}
+	if len(partial) != 0 {
+		t.Fatalf("partial = %q", partial)
+	}
+}
+
+func TestLogTailRunStopsOnContextCancel(t *testing.T) {
+	path := writeTempLog(t, "hello\n")
+	tail := NewLogTail(path, checkpoint.NewMemoryStore(), 10*time.Millisecond, nil)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() {
+		tail.Run(ctx)
+		close(done)
+	}()
+
+	// 等它至少消费一轮，再取消。
+	time.Sleep(50 * time.Millisecond)
+	cancel()
+
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatalf("LogTail.Run did not stop after cancel")
+	}
+}

internal/config/config_test.go

@@ -0,0 +1,83 @@
+package config
+
+import (
+	"testing"
+	"time"
+)
+
+func TestDefaultAgentConfig(t *testing.T) {
+	cfg := DefaultAgentConfig()
+
+	if cfg.CoreInterval != 15*time.Second {
+		t.Fatalf("CoreInterval = %v, want 15s", cfg.CoreInterval)
+	}
+	if cfg.DiskInterval != 30*time.Second {
+		t.Fatalf("DiskInterval = %v, want 30s", cfg.DiskInterval)
+	}
+	if cfg.NetworkInterval != 30*time.Second {
+		t.Fatalf("NetworkInterval = %v, want 30s", cfg.NetworkInterval)
+	}
+	if cfg.ProcessInterval != 60*time.Second {
+		t.Fatalf("ProcessInterval = %v, want 60s", cfg.ProcessInterval)
+	}
+	if cfg.BatchSize != 128 {
+		t.Fatalf("BatchSize = %d, want 128", cfg.BatchSize)
+	}
+	if cfg.CheckpointDriver != "file" {
+		t.Fatalf("CheckpointDriver = %q, want file", cfg.CheckpointDriver)
+	}
+}
+
+func TestDefaultGatewayConfig(t *testing.T) {
+	cfg := DefaultGatewayConfig()
+	if cfg.Addr != "0.0.0.0:4318" {
+		t.Fatalf("Addr = %q, want 0.0.0.0:4318", cfg.Addr)
+	}
+	if cfg.MaxMetricsPerHost != 20000 {
+		t.Fatalf("MaxMetricsPerHost = %d, want 20000", cfg.MaxMetricsPerHost)
+	}
+}
+
+func TestDurEnv(t *testing.T) {
+	if got := durEnv("HMS_TEST_DUR_EMPTY", 3*time.Second); got != 3*time.Second {
+		t.Fatalf("durEnv empty = %v, want 3s", got)
+	}
+
+	t.Setenv("HMS_TEST_DUR", "7s")
+	if got := durEnv("HMS_TEST_DUR", 3*time.Second); got != 7*time.Second {
+		t.Fatalf("durEnv set = %v, want 7s", got)
+	}
+
+	t.Setenv("HMS_TEST_DUR_BAD", "not-a-duration")
+	if got := durEnv("HMS_TEST_DUR_BAD", 3*time.Second); got != 3*time.Second {
+		t.Fatalf("durEnv bad = %v, want fallback 3s", got)
+	}
+}
+
+func TestIntEnv(t *testing.T) {
+	if got := intEnv("HMS_TEST_INT_EMPTY", 42); got != 42 {
+		t.Fatalf("intEnv empty = %d, want 42", got)
+	}
+
+	t.Setenv("HMS_TEST_INT", "99")
+	if got := intEnv("HMS_TEST_INT", 42); got != 99 {
+		t.Fatalf("intEnv set = %d, want 99", got)
+	}
+
+	t.Setenv("HMS_TEST_INT_BAD", "bad")
+	if got := intEnv("HMS_TEST_INT_BAD", 42); got != 42 {
+		t.Fatalf("intEnv bad = %d, want fallback 42", got)
+	}
+}
+
+func TestSplitCSV(t *testing.T) {
+	got := splitCSV("")
+	if got != nil {
+		t.Fatalf("splitCSV(empty) = %v, want nil", got)
+	}
+
+	got = splitCSV("/var/log/a.log, /var/log/b.log ,")
+	if len(got) != 2 || got[0] != "/var/log/a.log" || got[1] != "/var/log/b.log" {
+		t.Fatalf("splitCSV = %v", got)
+	}
+}

internal/gateway/server_test.go

@@ -83,7 +83,7 @@ func TestAgentMetricsRoundTrip(t *testing.T) {
 		t.Fatalf("unmarshal ack: %v", err)
 	}
 	if !ack.Ok || ack.Seq != 42 {
-		t.Fatalf("ack = %+v", ack)
+		t.Fatalf("ack ok=%v seq=%d, want ok=true seq=42", ack.Ok, ack.Seq)
 	}
 }
 

internal/model/model_test.go

@@ -0,0 +1,49 @@
+package model
+
+import (
+	"testing"
+	"time"
+)
+
+func TestNewSampleDefaultsLabels(t *testing.T) {
+	at := time.Unix(1700000000, 0)
+	s := NewSample(MetricCPUUsage, 42.5, at, nil)
+	if s.Name != MetricCPUUsage || s.Value != 42.5 || s.Timestamp != 1700000000 {
+		t.Fatalf("sample = %+v", s)
+	}
+	if s.Labels == nil {
+		t.Fatalf("labels should not be nil after NewSample")
+	}
+}
+
+func TestNewSampleKeepsProvidedLabels(t *testing.T) {
+	at := time.Now()
+	labels := map[string]string{LabelHostID: "h-001", "service": "web"}
+	s := NewSample(MetricMemUsedPercent, 70, at, labels)
+	if s.Labels[LabelHostID] != "h-001" || s.Labels["service"] != "web" {
+		t.Fatalf("labels = %v", s.Labels)
+	}
+}
+
+func TestMetricNameConstants(t *testing.T) {
+	// 与架构设计 5.1.3 的 <domain>.<metric> 命名规范对齐。
+	want := map[string]string{
+		"cpu.usage":           MetricCPUUsage,
+		"mem.used_percent":    MetricMemUsedPercent,
+		"load.1m":             MetricLoad1m,
+		"disk.used_percent":   MetricDiskUsedPercent,
+		"disk.io.read_bytes":  MetricDiskIOReadBytes,
+		"disk.io.write_bytes": MetricDiskIOWriteBytes,
+		"net.bytes_sent":      MetricNetBytesSent,
+		"net.bytes_recv":      MetricNetBytesRecv,
+		"net.pkt_drop":        MetricNetPktDrop,
+		"process.cpu":         MetricProcessCPU,
+		"process.mem":         MetricProcessMem,
+		"process.count":       MetricProcessCount,
+	}
+	for name, constant := range want {
+		if name != constant {
+			t.Fatalf("metric name mismatch: %q != %q", name, constant)
+		}
+	}
+}

pkg/convert/convert_test.go

@@ -0,0 +1,59 @@
+package convert
+
+import (
+	"testing"
+
+	"git.opencomputing.cn/yumoqing/host-metrics-collector/internal/model"
+)
+
+func TestSampleRoundTrip(t *testing.T) {
+	in := model.Sample{
+		Name:      "cpu.usage",
+		Value:     42.5,
+		Timestamp: 1700000000,
+		Labels:    map[string]string{"host_id": "h-001"},
+	}
+	out := SampleFromPB(SampleToPB(in))
+	if out.Name != in.Name || out.Value != in.Value || out.Timestamp != in.Timestamp {
+		t.Fatalf("round trip = %+v, want %+v", out, in)
+	}
+	if out.Labels["host_id"] != "h-001" {
+		t.Fatalf("labels = %v", out.Labels)
+	}
+}
+
+func TestSampleFromPBNil(t *testing.T) {
+	out := SampleFromPB(nil)
+	if out.Name != "" || out.Labels != nil {
+		t.Fatalf("nil conversion = %+v", out)
+	}
+}
+
+func TestLogRoundTrip(t *testing.T) {
+	in := model.LogRecord{
+		Timestamp: 1700000000,
+		Level:     "ERROR",
+		Source:    "/var/log/app.log",
+		Message:   "boom",
+		Fields:    map[string]string{"trace_id": "tr-1"},
+		File:      "/var/log/app.log",
+		Offset:    1024,
+	}
+	out := LogFromPB(LogToPB(in))
+	if out.Timestamp != in.Timestamp || out.Level != in.Level || out.Source != in.Source {
+		t.Fatalf("round trip = %+v, want %+v", out, in)
+	}
+	if out.Message != "boom" || out.Offset != 1024 {
+		t.Fatalf("round trip = %+v", out)
+	}
+	if out.Fields["trace_id"] != "tr-1" {
+		t.Fatalf("fields = %v", out.Fields)
+	}
+}
+
+func TestLogFromPBNil(t *testing.T) {
+	out := LogFromPB(nil)
+	if out.Message != "" || out.Fields != nil {
+		t.Fatalf("nil conversion = %+v", out)
+	}
+}