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

develop: 补齐 Gateway HTTP 上报/查询服务与 cmd 入口及测试

Browse Source
This commit is contained in:
Pipeline Agent 2026-08-15 01:43:26 +08:00
parent fb801e2c66
commit e1847dd58b
7 changed files with 931 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

107
README.md
View File

@ -1,2 +1,109 @@
# host-metrics-collector
主机监控系统（HMS）的主机指标与日志采集模块。包含两部分：
- **Agent（`cmd/agent`）**：定时采集 CPU / 内存 / 磁盘 / 网络 / 进程指标，tail 采集文件日志，本地 WAL 断点续传，批量上报。
- **Collector Gateway（`cmd/gateway`）**：在 `4318` 端口接收 Agent 的 protobuf 上报，提供指标 / 日志查询 REST API 与健康检查。
## 目录结构
```
api/collector.proto # protobuf 数据契约（MetricBatch / LogBatch / Ack 等）
pkg/collectorpb/ # protoc 生成代码（已提交）
pkg/convert/ # 领域模型 <-> protobuf 转换
internal/model/ # 领域模型与指标命名规范
internal/config/ # 环境变量优先的配置加载
internal/collect/ # gopsutil 系统指标采集 + 文件日志 tail
internal/checkpoint/ # 断点续传检查点（memory / file / redis）
internal/wal/ # Agent 本地 WAL（追加式，按 seq 恢复）
internal/agent/ # Agent 调度、缓冲、上报、心跳
internal/gateway/ # Gateway 存储与 HTTP 处理器（上报 + 查询 API）
cmd/agent/ # Agent 入口
cmd/gateway/ # Gateway 入口
```
## 快速开始
### 构建
```bash
go build ./cmd/agent
go build ./cmd/gateway
```
### 运行 Gateway（4318）
```bash
HMS_AGENT_TOKEN=dev-token ./gateway
```
### 运行 Agent（采集 + 上报）
```bash
HMS_HOST_ID=host-001 \
HMS_AGENT_TOKEN=dev-token \
HMS_GATEWAY_URL=http://127.0.0.1:4318 \
HMS_LOG_FILES=/var/log/app.log \
HMS_CHECKPOINT_DRIVER=file \
./agent
```
## 配置（环境变量）
| 变量 | 默认值 | 说明 |
|---|---|---|
| `HMS_HOST_ID` | `host-001` | Agent 主机标识 |
| `HMS_AGENT_VERSION` | `0.1.0` | Agent 版本 |
| `HMS_SERVICE` | `host` | 服务标签 |
| `HMS_GATEWAY_URL` | `http://127.0.0.1:4318` | 上报地址 |
| `HMS_AGENT_TOKEN` | `dev-token` | 上报 Token |
| `HMS_LOG_FILES` | 空 | 逗号分隔的日志文件路径 |
| `HMS_CORE_INTERVAL` | `15s` | 核心指标采集频率 |
| `HMS_DISK_INTERVAL` | `30s` | 磁盘指标采集频率 |
| `HMS_NETWORK_INTERVAL` | `30s` | 网络指标采集频率 |
| `HMS_PROCESS_INTERVAL` | `60s` | 进程指标采集频率 |
| `HMS_BATCH_SIZE` | `128` | 指标批量大小 |
| `HMS_FLUSH_INTERVAL` | `10s` | 上报刷新间隔 |
| `HMS_CHECKPOINT_DRIVER` | `file` | `memory` / `file` / `redis` |
| `HMS_CHECKPOINT_FILE` | `checkpoint.json` | file 驱动检查点文件 |
| `HMS_REDIS_URL` | `redis://127.0.0.1:6379/0` | redis 驱动地址 |
| `HMS_GATEWAY_ADDR` | `0.0.0.0:4318` | Gateway 监听地址 |
| `HMS_MAX_METRICS_PER_HOST` | `20000` | 内存中每主机指标保留上限 |
| `HMS_MAX_LOGS_PER_HOST` | `20000` | 内存中每主机日志保留上限 |
## 断点续传
- 日志文件 offset 与指标 / 日志上报 seq 通过 `checkpoint.Store` 持久化。
- 生产环境使用 `HMS_CHECKPOINT_DRIVER=redis`，Redis 键：
- 日志 offset：`hms:agent:logoffset:{file}`
- 已确认 seq：`hms:agent:ackedseq:{host_id}`
- 上报成功收到 `Ack{seq}` 后推进检查点；进程重启后从未确认点继续，保证至少一次投递。
## API
### Agent 上报（protobuf，`X-Agent-Token` 认证）
| Method | Path | 说明 |
|---|---|---|
| POST | `/v1/agent/heartbeat` | 心跳 / 注册 / 配置下发 |
| POST | `/v1/agent/metrics` | 指标批量上报（`MetricBatch` → `Ack`） |
| POST | `/v1/agent/logs` | 日志批量上报（`LogBatch` → `Ack`） |
| GET | `/v1/agent/checkpoint/{host_id}` | 查询已确认序号 |
### 查询 REST API（JSON）
| Method | Path | 说明 |
|---|---|---|
| GET | `/healthz` | 存活探针 |
| GET | `/readyz` | 就绪探针 |
| GET | `/api/v1/hosts/{host_id}/metrics` | 单主机指标时序 |
| GET | `/api/v1/hosts/{host_id}/logs` | 单主机日志 |
| GET | `/api/v1/metrics/query` | 时序范围查询 |
| GET | `/api/v1/metrics/query/instant` | 即时查询 |
| POST | `/api/v1/logs/search` | 跨主机日志检索 |
## 测试
```bash
go test ./...
```

38
cmd/agent/main.go Normal file
View File

@ -0,0 +1,38 @@
// Command agent 是 host-metrics-collector 的 Agent 端入口：定时采集主机
// CPU/内存/磁盘/网络/进程指标与文件日志，经本地 WAL 断点续传，批量上报到
// Collector Gateway（默认 http://127.0.0.1:4318）。
package main
import (
"context"
"log"
"os"
"os/signal"
"syscall"
"git.opencomputing.cn/yumoqing/host-metrics-collector/internal/agent"
"git.opencomputing.cn/yumoqing/host-metrics-collector/internal/config"
)
func main() {
cfg := config.DefaultAgentConfig()
log.Printf("[agent] starting host=%s version=%s gateway=%s", cfg.HostID, cfg.AgentVersion, cfg.GatewayURL)
a, err := agent.New(cfg)
if err != nil {
log.Fatalf("[agent] init failed: %v", err)
}
defer func() {
if err := a.Close(); err != nil {
log.Printf("[agent] close: %v", err)
}
}()
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop()
if err := a.Run(ctx); err != nil {
log.Printf("[agent] run: %v", err)
}
log.Printf("[agent] stopped")
}

32
cmd/gateway/main.go Normal file
View File

@ -0,0 +1,32 @@
// Command gateway 是 host-metrics-collector 的 Collector Gateway 入口：
// 在 4318 端口接收 Agent 的 protobuf 上报，并暴露指标/日志查询 REST API。
package main
import (
"log"
"net/http"
"time"
"git.opencomputing.cn/yumoqing/host-metrics-collector/internal/config"
"git.opencomputing.cn/yumoqing/host-metrics-collector/internal/gateway"
)
func main() {
cfg := config.DefaultGatewayConfig()
store := gateway.NewStore(cfg.MaxMetricsPerHost, cfg.MaxLogsPerHost)
srv := gateway.NewServer(cfg, store)
httpSrv := &http.Server{
Addr: cfg.Addr,
Handler: srv.Handler(),
ReadHeaderTimeout: 5 * time.Second,
ReadTimeout: 30 * time.Second,
WriteTimeout: 30 * time.Second,
IdleTimeout: 60 * time.Second,
}
log.Printf("[gateway] listening on %s (agent token set=%v)", cfg.Addr, cfg.AgentToken != "")
if err := httpSrv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
log.Fatalf("[gateway] listen: %v", err)
}
}

69
internal/checkpoint/file_test.go Normal file
View File

@ -0,0 +1,69 @@
package checkpoint
import (
"path/filepath"
"testing"
)
func TestFileStorePersistsAcrossInstances(t *testing.T) {
path := filepath.Join(t.TempDir(), "cp.json")
s, err := NewFileStore(path)
if err != nil {
t.Fatalf("NewFileStore: %v", err)
}
if err := s.SetLogOffset("/var/log/app.log", 2048); err != nil {
t.Fatalf("SetLogOffset: %v", err)
}
if err := s.SetAckedSeq("h-001", 42); err != nil {
t.Fatalf("SetAckedSeq: %v", err)
}
_ = s.Close()
s2, err := NewFileStore(path)
if err != nil {
t.Fatalf("reopen: %v", err)
}
defer s2.Close()
if got, _ := s2.GetLogOffset("/var/log/app.log"); got != 2048 {
t.Fatalf("log offset = %d, want 2048", got)
}
if got, _ := s2.GetAckedSeq("h-001"); got != 42 {
t.Fatalf("acked seq = %d, want 42", got)
}
}
func TestFileStoreMissingReturnsZero(t *testing.T) {
path := filepath.Join(t.TempDir(), "cp.json")
s, err := NewFileStore(path)
if err != nil {
t.Fatalf("NewFileStore: %v", err)
}
defer s.Close()
if got, err := s.GetLogOffset("nope"); err != nil || got != 0 {
t.Fatalf("got=%d err=%v, want 0 nil", got, err)
}
if got, err := s.GetAckedSeq("nope"); err != nil || got != 0 {
t.Fatalf("got=%d err=%v, want 0 nil", got, err)
}
}
func TestMemoryStoreThreadSafe(t *testing.T) {
s := NewMemoryStore()
for i := int64(0); i < 100; i++ {
if err := s.SetAckedSeq("h", i); err != nil {
t.Fatalf("SetAckedSeq: %v", err)
}
}
if got, _ := s.GetAckedSeq("h"); got != 99 {
t.Fatalf("acked seq = %d, want 99", got)
}
}
func TestNewUnknownDriver(t *testing.T) {
if _, err := New("bogus", "", ""); err == nil {
t.Fatalf("expected error for unknown driver")
}
}

369
internal/gateway/server.go Normal file
View File

@ -0,0 +1,369 @@
// Package gateway 实现 Collector Gateway：接收 Agent 上报、提供指标/日志
// 查询 REST API，并维护内存存储与已确认序号。
package gateway
import (
"encoding/json"
"fmt"
"io"
"net/http"
"strconv"
"strings"
"time"
"google.golang.org/protobuf/proto"
"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/pkg/collectorpb"
"git.opencomputing.cn/yumoqing/host-metrics-collector/pkg/convert"
)
const (
headerAgentToken = "X-Agent-Token"
headerHostID = "X-Host-Id"
contentTypeProtobuf = "application/x-protobuf"
contentTypeJSON = "application/json"
)
// Server 是 Gateway 的 HTTP 处理器，同时暴露：
// - Agent 上报接口（protobuf）：/v1/agent/*
// - REST 查询接口（JSON）：/api/v1/*
// - 健康检查：/healthz、/readyz
type Server struct {
cfg config.GatewayConfig
store *Store
}
// NewServer 构建 Gateway HTTP 处理器。
func NewServer(cfg config.GatewayConfig, store *Store) *Server {
return &Server{cfg: cfg, store: store}
}
// Handler 返回路由处理器。
func (s *Server) Handler() http.Handler {
mux := http.NewServeMux()
// Agent 上报接口（protobuf，Token 认证）。
mux.HandleFunc("POST /v1/agent/heartbeat", s.requireAgentToken(s.handleHeartbeat))
mux.HandleFunc("POST /v1/agent/metrics", s.requireAgentToken(s.handleAgentMetrics))
mux.HandleFunc("POST /v1/agent/logs", s.requireAgentToken(s.handleAgentLogs))
mux.HandleFunc("GET /v1/agent/checkpoint/{host_id}", s.requireAgentToken(s.handleCheckpoint))
// 健康检查。
mux.HandleFunc("GET /healthz", s.handleHealthz)
mux.HandleFunc("GET /readyz", s.handleReadyz)
// REST 查询接口（JSON）。
mux.HandleFunc("GET /api/v1/hosts/{host_id}/metrics", s.handleHostMetrics)
mux.HandleFunc("GET /api/v1/hosts/{host_id}/logs", s.handleHostLogs)
mux.HandleFunc("GET /api/v1/metrics/query/instant", s.handleMetricsInstant)
mux.HandleFunc("GET /api/v1/metrics/query", s.handleMetricsQuery)
mux.HandleFunc("POST /api/v1/logs/search", s.handleLogsSearch)
return mux
}
func (s *Server) requireAgentToken(next http.HandlerFunc) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
if s.cfg.AgentToken != "" && r.Header.Get(headerAgentToken) != s.cfg.AgentToken {
writeProto(w, http.StatusUnauthorized, &collectorpb.Ack{Ok: false, Error: "unauthorized"})
return
}
next(w, r)
}
}
func (s *Server) handleHeartbeat(w http.ResponseWriter, r *http.Request) {
body, err := io.ReadAll(io.LimitReader(r.Body, 1<<20))
if err != nil {
writeProto(w, http.StatusBadRequest, &collectorpb.HeartbeatResponse{Ok: false})
return
}
req := &collectorpb.HeartbeatRequest{}
if err := proto.Unmarshal(body, req); err != nil {
writeProto(w, http.StatusBadRequest, &collectorpb.HeartbeatResponse{Ok: false})
return
}
writeProto(w, http.StatusOK, &collectorpb.HeartbeatResponse{
Ok: true,
ServerTime: time.Now().Unix(),
Config: map[string]string{},
})
}
func (s *Server) handleAgentMetrics(w http.ResponseWriter, r *http.Request) {
body, err := io.ReadAll(io.LimitReader(r.Body, 64<<20))
if err != nil {
writeProto(w, http.StatusBadRequest, &collectorpb.Ack{Ok: false, Error: "read body"})
return
}
batch := &collectorpb.MetricBatch{}
if err := proto.Unmarshal(body, batch); err != nil {
writeProto(w, http.StatusBadRequest, &collectorpb.Ack{Ok: false, Error: "invalid protobuf"})
return
}
if batch.HostId == "" {
writeProto(w, http.StatusBadRequest, &collectorpb.Ack{Ok: false, Seq: batch.Seq, Error: "host_id is required"})
return
}
samples := make([]model.Sample, 0, len(batch.Samples))
for _, pb := range batch.Samples {
samples = append(samples, convert.SampleFromPB(pb))
}
s.store.AddMetrics(batch.HostId, batch.Seq, samples)
s.store.SetLastAckedSeq(batch.HostId, batch.Seq)
writeProto(w, http.StatusOK, &collectorpb.Ack{Ok: true, Seq: batch.Seq})
}
func (s *Server) handleAgentLogs(w http.ResponseWriter, r *http.Request) {
body, err := io.ReadAll(io.LimitReader(r.Body, 64<<20))
if err != nil {
writeProto(w, http.StatusBadRequest, &collectorpb.Ack{Ok: false, Error: "read body"})
return
}
batch := &collectorpb.LogBatch{}
if err := proto.Unmarshal(body, batch); err != nil {
writeProto(w, http.StatusBadRequest, &collectorpb.Ack{Ok: false, Error: "invalid protobuf"})
return
}
if batch.HostId == "" {
writeProto(w, http.StatusBadRequest, &collectorpb.Ack{Ok: false, Seq: batch.Seq, Error: "host_id is required"})
return
}
entries := make([]model.LogRecord, 0, len(batch.Entries))
for _, pb := range batch.Entries {
entries = append(entries, convert.LogFromPB(pb))
}
s.store.AddLogs(batch.HostId, batch.Seq, entries)
s.store.SetLastAckedSeq(batch.HostId, batch.Seq)
writeProto(w, http.StatusOK, &collectorpb.Ack{Ok: true, Seq: batch.Seq})
}
func (s *Server) handleCheckpoint(w http.ResponseWriter, r *http.Request) {
hostID := r.PathValue("host_id")
writeProto(w, http.StatusOK, &collectorpb.CheckpointResponse{LastAckedSeq: s.store.LastAckedSeq(hostID)})
}
func (s *Server) handleHealthz(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, 0, "ok", map[string]string{"status": "ok"})
}
func (s *Server) handleReadyz(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, 0, "ok", map[string]string{"status": "ready"})
}
func (s *Server) handleHostMetrics(w http.ResponseWriter, r *http.Request) {
hostID := r.PathValue("host_id")
q := r.URL.Query()
from, to, err := parseRange(q.Get("from"), q.Get("to"))
if err != nil {
writeJSON(w, http.StatusBadRequest, 40001, err.Error(), nil)
return
}
series, err := s.store.HostMetrics(hostID, q.Get("metric"), from, to)
if err != nil {
writeJSON(w, http.StatusInternalServerError, 50001, err.Error(), nil)
return
}
writeJSON(w, http.StatusOK, 0, "ok", series)
}
func (s *Server) handleHostLogs(w http.ResponseWriter, r *http.Request) {
hostID := r.PathValue("host_id")
q := r.URL.Query()
from, to, err := parseRange(q.Get("from"), q.Get("to"))
if err != nil {
writeJSON(w, http.StatusBadRequest, 40001, err.Error(), nil)
return
}
page, pageSize, err := parsePagination(q.Get("page"), q.Get("page_size"))
if err != nil {
writeJSON(w, http.StatusBadRequest, 40001, err.Error(), nil)
return
}
total, items, err := s.store.HostLogs(hostID, q.Get("level"), q.Get("keyword"), from, to, page, pageSize)
if err != nil {
writeJSON(w, http.StatusInternalServerError, 50001, err.Error(), nil)
return
}
writeJSON(w, http.StatusOK, 0, "ok", map[string]any{"total": total, "items": items})
}
func (s *Server) handleMetricsQuery(w http.ResponseWriter, r *http.Request) {
q := r.URL.Query()
from, to, err := parseRange(q.Get("from"), q.Get("to"))
if err != nil {
writeJSON(w, http.StatusBadRequest, 40001, err.Error(), nil)
return
}
labels, err := parseLabels(q.Get("labels"))
if err != nil {
writeJSON(w, http.StatusBadRequest, 40001, err.Error(), nil)
return
}
series, err := s.store.QueryMetrics(q.Get("metric"), labels, from, to)
if err != nil {
writeJSON(w, http.StatusInternalServerError, 50001, err.Error(), nil)
return
}
writeJSON(w, http.StatusOK, 0, "ok", map[string]any{"resultType": "matrix", "result": series})
}
func (s *Server) handleMetricsInstant(w http.ResponseWriter, r *http.Request) {
q := r.URL.Query()
labels, err := parseLabels(q.Get("labels"))
if err != nil {
writeJSON(w, http.StatusBadRequest, 40001, err.Error(), nil)
return
}
points, err := s.store.InstantMetrics(q.Get("metric"), labels)
if err != nil {
writeJSON(w, http.StatusInternalServerError, 50001, err.Error(), nil)
return
}
writeJSON(w, http.StatusOK, 0, "ok", map[string]any{"resultType": "vector", "result": points})
}
func (s *Server) handleLogsSearch(w http.ResponseWriter, r *http.Request) {
var req struct {
HostIDs []string `json:"host_ids"`
Level string `json:"level"`
Keyword string `json:"keyword"`
From string `json:"from"`
To string `json:"to"`
Page int `json:"page"`
PageSize int `json:"page_size"`
}
if err := json.NewDecoder(io.LimitReader(r.Body, 1<<20)).Decode(&req); err != nil {
writeJSON(w, http.StatusBadRequest, 40001, "invalid json body", nil)
return
}
from, to, err := parseRange(req.From, req.To)
if err != nil {
writeJSON(w, http.StatusBadRequest, 40001, err.Error(), nil)
return
}
page, pageSize := req.Page, req.PageSize
if page < 1 {
page = 1
}
if pageSize < 1 {
pageSize = 20
}
total, items, err := s.store.SearchLogs(req.HostIDs, req.Level, req.Keyword, from, to, page, pageSize)
if err != nil {
writeJSON(w, http.StatusInternalServerError, 50001, err.Error(), nil)
return
}
writeJSON(w, http.StatusOK, 0, "ok", map[string]any{"total": total, "items": items})
}
// --- 序列化辅助 ---
func writeProto(w http.ResponseWriter, status int, msg proto.Message) {
data, err := proto.Marshal(msg)
if err != nil {
w.Header().Set("Content-Type", "text/plain; charset=utf-8")
w.WriteHeader(http.StatusInternalServerError)
_, _ = io.WriteString(w, "marshal response failed")
return
}
w.Header().Set("Content-Type", contentTypeProtobuf)
w.WriteHeader(status)
_, _ = w.Write(data)
}
func writeJSON(w http.ResponseWriter, status, code int, msg string, data any) {
w.Header().Set("Content-Type", contentTypeJSON)
w.WriteHeader(status)
_ = json.NewEncoder(w).Encode(map[string]any{
"code": code,
"message": msg,
"data": data,
})
}
func parseRange(fromStr, toStr string) (time.Time, time.Time, error) {
from, err := parseTime(fromStr)
if err != nil {
return time.Time{}, time.Time{}, fmt.Errorf("invalid from: %w", err)
}
to, err := parseTime(toStr)
if err != nil {
return time.Time{}, time.Time{}, fmt.Errorf("invalid to: %w", err)
}
if !from.IsZero() && !to.IsZero() && from.After(to) {
return time.Time{}, time.Time{}, fmt.Errorf("from must be before to")
}
return from, to, nil
}
func parseTime(s string) (time.Time, error) {
if s == "" {
return time.Time{}, nil
}
// 优先按 Unix 秒解析，便于与内部时间戳一致。
if n, err := strconv.ParseInt(s, 10, 64); err == nil {
return time.Unix(n, 0), nil
}
if t, err := time.Parse(time.RFC3339, s); err == nil {
return t, nil
}
if t, err := time.Parse(time.RFC3339Nano, s); err == nil {
return t, nil
}
return time.Time{}, fmt.Errorf("invalid time %q", s)
}
func parsePagination(pageStr, sizeStr string) (int, int, error) {
page := 1
if pageStr != "" {
n, err := strconv.Atoi(pageStr)
if err != nil || n < 1 {
return 0, 0, fmt.Errorf("invalid page %q", pageStr)
}
page = n
}
pageSize := 20
if sizeStr != "" {
n, err := strconv.Atoi(sizeStr)
if err != nil || n < 1 || n > 200 {
return 0, 0, fmt.Errorf("invalid page_size %q", sizeStr)
}
pageSize = n
}
return page, pageSize, nil
}
func parseLabels(s string) (map[string]string, error) {
if strings.TrimSpace(s) == "" {
return nil, nil
}
out := map[string]string{}
for _, part := range strings.Split(s, ",") {
part = strings.TrimSpace(part)
if part == "" {
continue
}
kv := strings.SplitN(part, "=", 2)
if len(kv) != 2 {
return nil, fmt.Errorf("invalid labels %q, expect k=v", s)
}
out[strings.TrimSpace(kv[0])] = strings.TrimSpace(kv[1])
}
return out, nil
}

313
internal/gateway/server_test.go Normal file
View File

@ -0,0 +1,313 @@
package gateway
import (
"bytes"
"encoding/json"
"io"
"net/http"
"net/http/httptest"
"strings"
"testing"
"time"
"google.golang.org/protobuf/proto"
"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/pkg/collectorpb"
)
func newTestServer() (*httptest.Server, *Store) {
cfg := config.GatewayConfig{Addr: "127.0.0.1:4318", AgentToken: "test-token", MaxMetricsPerHost: 100, MaxLogsPerHost: 100}
store := NewStore(cfg.MaxMetricsPerHost, cfg.MaxLogsPerHost)
srv := httptest.NewServer(NewServer(cfg, store).Handler())
return srv, store
}
func agentRequest(t *testing.T, method, url string, body []byte, token string) *http.Response {
t.Helper()
req, err := http.NewRequest(method, url, bytes.NewReader(body))
if err != nil {
t.Fatalf("new request: %v", err)
}
if token != "" {
req.Header.Set(headerAgentToken, token)
}
if body != nil {
req.Header.Set("Content-Type", contentTypeProtobuf)
}
resp, err := http.DefaultClient.Do(req)
if err != nil {
t.Fatalf("do request: %v", err)
}
return resp
}
func readAll(t *testing.T, resp *http.Response) []byte {
t.Helper()
defer resp.Body.Close()
data, err := io.ReadAll(resp.Body)
if err != nil {
t.Fatalf("read body: %v", err)
}
return data
}
func decodeJSON(t *testing.T, resp *http.Response, out any) {
t.Helper()
defer resp.Body.Close()
if err := json.NewDecoder(resp.Body).Decode(out); err != nil {
t.Fatalf("decode json: %v", err)
}
}
func TestAgentMetricsRoundTrip(t *testing.T) {
ts, _ := newTestServer()
defer ts.Close()
batch := &collectorpb.MetricBatch{
HostId: "h-001",
Seq: 42,
SentAt: time.Now().Unix(),
Samples: []*collectorpb.MetricSample{
{Name: "cpu.usage", Value: 55.5, Timestamp: time.Now().Unix(), Labels: map[string]string{"host_id": "h-001"}},
},
}
body, _ := proto.Marshal(batch)
resp := agentRequest(t, http.MethodPost, ts.URL+"/v1/agent/metrics", body, "test-token")
if resp.StatusCode != http.StatusOK {
t.Fatalf("status = %d, want 200", resp.StatusCode)
}
var ack collectorpb.Ack
if err := proto.Unmarshal(readAll(t, resp), &ack); err != nil {
t.Fatalf("unmarshal ack: %v", err)
}
if !ack.Ok || ack.Seq != 42 {
t.Fatalf("ack = %+v", ack)
}
}
func TestAgentMetricsAuth(t *testing.T) {
ts, _ := newTestServer()
defer ts.Close()
body, _ := proto.Marshal(&collectorpb.MetricBatch{HostId: "h-001", Seq: 1})
resp := agentRequest(t, http.MethodPost, ts.URL+"/v1/agent/metrics", body, "wrong-token")
if resp.StatusCode != http.StatusUnauthorized {
t.Fatalf("status = %d, want 401", resp.StatusCode)
}
}
func TestAgentMetricsStoresSamples(t *testing.T) {
ts, store := newTestServer()
defer ts.Close()
at := time.Now().Unix()
batch := &collectorpb.MetricBatch{
HostId: "h-001",
Seq: 7,
SentAt: at,
Samples: []*collectorpb.MetricSample{
{Name: "cpu.usage", Value: 90, Timestamp: at, Labels: map[string]string{model.LabelHostID: "h-001", model.LabelMetricGroup: model.GroupCore}},
},
}
body, _ := proto.Marshal(batch)
resp := agentRequest(t, http.MethodPost, ts.URL+"/v1/agent/metrics", body, "test-token")
if resp.StatusCode != http.StatusOK {
t.Fatalf("status = %d", resp.StatusCode)
}
readAll(t, resp)
series, err := store.HostMetrics("h-001", "cpu.usage", time.Time{}, time.Time{})
if err != nil {
t.Fatalf("HostMetrics: %v", err)
}
if len(series) != 1 || len(series[0].Values) != 1 || series[0].Values[0].Value != 90 {
t.Fatalf("series = %+v", series)
}
if got := store.LastAckedSeq("h-001"); got != 7 {
t.Fatalf("acked seq = %d, want 7", got)
}
}
func TestAgentLogsStoresEntries(t *testing.T) {
ts, store := newTestServer()
defer ts.Close()
batch := &collectorpb.LogBatch{
HostId: "h-001",
Seq: 3,
Entries: []*collectorpb.LogEntry{
{Timestamp: time.Now().Unix(), Level: "ERROR", Source: "/var/log/app.log", Message: "boom", File: "/var/log/app.log", Offset: 100},
},
}
body, _ := proto.Marshal(batch)
resp := agentRequest(t, http.MethodPost, ts.URL+"/v1/agent/logs", body, "test-token")
if resp.StatusCode != http.StatusOK {
t.Fatalf("status = %d", resp.StatusCode)
}
readAll(t, resp)
total, items, err := store.HostLogs("h-001", "ERROR", "boom", time.Time{}, time.Time{}, 1, 20)
if err != nil {
t.Fatalf("HostLogs: %v", err)
}
if total != 1 || len(items) != 1 || items[0].Message != "boom" {
t.Fatalf("total=%d items=%+v", total, items)
}
}
func TestHeartbeat(t *testing.T) {
ts, _ := newTestServer()
defer ts.Close()
body, _ := proto.Marshal(&collectorpb.HeartbeatRequest{HostId: "h-001", AgentVersion: "0.1.0"})
resp := agentRequest(t, http.MethodPost, ts.URL+"/v1/agent/heartbeat", body, "test-token")
if resp.StatusCode != http.StatusOK {
t.Fatalf("status = %d", resp.StatusCode)
}
var hb collectorpb.HeartbeatResponse
if err := proto.Unmarshal(readAll(t, resp), &hb); err != nil {
t.Fatalf("unmarshal heartbeat: %v", err)
}
if !hb.Ok {
t.Fatalf("heartbeat not ok")
}
}
func TestCheckpoint(t *testing.T) {
ts, store := newTestServer()
defer ts.Close()
store.SetLastAckedSeq("h-001", 99)
resp := agentRequest(t, http.MethodGet, ts.URL+"/v1/agent/checkpoint/h-001", nil, "test-token")
var cp collectorpb.CheckpointResponse
if err := proto.Unmarshal(readAll(t, resp), &cp); err != nil {
t.Fatalf("unmarshal checkpoint: %v", err)
}
if cp.LastAckedSeq != 99 {
t.Fatalf("checkpoint = %d, want 99", cp.LastAckedSeq)
}
}
func TestRESTHostMetrics(t *testing.T) {
ts, store := newTestServer()
defer ts.Close()
at := time.Now().Unix()
store.AddMetrics("h-001", 1, []model.Sample{
{Name: "cpu.usage", Value: 12.5, Timestamp: at, Labels: map[string]string{model.LabelHostID: "h-001"}},
})
resp, err := http.Get(ts.URL + "/api/v1/hosts/h-001/metrics?metric=cpu.usage")
if err != nil {
t.Fatalf("get: %v", err)
}
var envelope struct {
Code int `json:"code"`
Message string `json:"message"`
Data []HostMetricSeries `json:"data"`
}
decodeJSON(t, resp, &envelope)
if envelope.Code != 0 || len(envelope.Data) != 1 || envelope.Data[0].Values[0].Value != 12.5 {
t.Fatalf("envelope = %+v", envelope)
}
}
func TestRESTHostLogsPagination(t *testing.T) {
ts, store := newTestServer()
defer ts.Close()
for i := 0; i < 25; i++ {
store.AddLogs("h-001", int64(i+1), []model.LogRecord{{Timestamp: time.Now().Unix(), Level: "INFO", Message: "line"}})
}
resp, err := http.Get(ts.URL + "/api/v1/hosts/h-001/logs?page=2&page_size=10")
if err != nil {
t.Fatalf("get: %v", err)
}
var envelope struct {
Code int `json:"code"`
Data struct {
Total int `json:"total"`
Items []LogEntryDTO `json:"items"`
} `json:"data"`
}
decodeJSON(t, resp, &envelope)
if envelope.Code != 0 || envelope.Data.Total != 25 || len(envelope.Data.Items) != 10 {
t.Fatalf("envelope = %+v", envelope)
}
}
func TestRESTMetricsQueryInstant(t *testing.T) {
ts, store := newTestServer()
defer ts.Close()
at := time.Now().Unix()
store.AddMetrics("h-001", 1, []model.Sample{
{Name: "mem.used_percent", Value: 88, Timestamp: at, Labels: map[string]string{model.LabelHostID: "h-001", "service": "web"}},
})
resp, err := http.Get(ts.URL + "/api/v1/metrics/query/instant?metric=mem.used_percent&labels=service=web")
if err != nil {
t.Fatalf("get: %v", err)
}
var envelope struct {
Code int `json:"code"`
Data struct {
Result []MetricPoint `json:"result"`
} `json:"data"`
}
decodeJSON(t, resp, &envelope)
if envelope.Code != 0 || len(envelope.Data.Result) != 1 || envelope.Data.Result[0].Value != 88 {
t.Fatalf("envelope = %+v", envelope)
}
}
func TestRESTLogsSearch(t *testing.T) {
ts, store := newTestServer()
defer ts.Close()
store.AddLogs("h-001", 1, []model.LogRecord{{Timestamp: time.Now().Unix(), Level: "ERROR", Message: "disk full"}})
store.AddLogs("h-002", 1, []model.LogRecord{{Timestamp: time.Now().Unix(), Level: "INFO", Message: "ok"}})
body := `{"host_ids":["h-001"],"level":"ERROR","keyword":"disk"}`
resp, err := http.Post(ts.URL+"/api/v1/logs/search", "application/json", strings.NewReader(body))
if err != nil {
t.Fatalf("post: %v", err)
}
var envelope struct {
Code int `json:"code"`
Data struct {
Total int `json:"total"`
Items []LogEntryDTO `json:"items"`
} `json:"data"`
}
decodeJSON(t, resp, &envelope)
if envelope.Code != 0 || envelope.Data.Total != 1 || envelope.Data.Items[0].Message != "disk full" {
t.Fatalf("envelope = %+v", envelope)
}
}
func TestParseTimeAndLabels(t *testing.T) {
if _, err := parseTime("1700000000"); err != nil {
t.Fatalf("unix parse: %v", err)
}
if _, err := parseTime("2025-01-01T00:00:00Z"); err != nil {
t.Fatalf("rfc3339 parse: %v", err)
}
if _, err := parseTime("bad"); err == nil {
t.Fatalf("expected error for bad time")
}
labels, err := parseLabels("host_id=h-001,service=web")
if err != nil {
t.Fatalf("parseLabels: %v", err)
}
if labels["service"] != "web" {
t.Fatalf("labels = %v", labels)
}
if _, err := parseLabels("bad"); err == nil {
t.Fatalf("expected error for bad labels")
}
}

6
internal/gateway/store.go
View File

@ -41,9 +41,9 @@ type LogEntryDTO struct {
type Store struct {
mu sync.RWMutex
metrics map[string][]model.Sample
logs map[string][]model.LogRecord
ackedSeqs map[string]int64
metrics map[string][]model.Sample
logs map[string][]model.LogRecord
ackedSeqs map[string]int64
lastSeqByHost map[string]int64
maxMetricsPerHost int