diff --git a/internal/agent/scheduler_test.go b/internal/agent/scheduler_test.go new file mode 100644 index 0000000..288537e --- /dev/null +++ b/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() +} diff --git a/internal/collect/logtail_test.go b/internal/collect/logtail_test.go new file mode 100644 index 0000000..35d1277 --- /dev/null +++ b/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") + } +} diff --git a/internal/config/config_test.go b/internal/config/config_test.go new file mode 100644 index 0000000..cb3a521 --- /dev/null +++ b/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) + } +} diff --git a/internal/gateway/server_test.go b/internal/gateway/server_test.go index 0cf8a93..8052c88 100644 --- a/internal/gateway/server_test.go +++ b/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) } } diff --git a/internal/model/model_test.go b/internal/model/model_test.go new file mode 100644 index 0000000..e7293c5 --- /dev/null +++ b/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 的 . 命名规范对齐。 + 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) + } + } +} diff --git a/pkg/convert/convert_test.go b/pkg/convert/convert_test.go new file mode 100644 index 0000000..288df36 --- /dev/null +++ b/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) + } +}