From e1847dd58bd0d529eb3af874fc43b49f3af1073d Mon Sep 17 00:00:00 2001 From: Pipeline Agent Date: Sat, 15 Aug 2026 01:43:26 +0800 Subject: [PATCH] =?UTF-8?q?develop:=20=E8=A1=A5=E9=BD=90=20Gateway=20HTTP?= =?UTF-8?q?=20=E4=B8=8A=E6=8A=A5/=E6=9F=A5=E8=AF=A2=E6=9C=8D=E5=8A=A1?= =?UTF-8?q?=E4=B8=8E=20cmd=20=E5=85=A5=E5=8F=A3=E5=8F=8A=E6=B5=8B=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 107 +++++++++ cmd/agent/main.go | 38 ++++ cmd/gateway/main.go | 32 +++ internal/checkpoint/file_test.go | 69 ++++++ internal/gateway/server.go | 369 +++++++++++++++++++++++++++++++ internal/gateway/server_test.go | 313 ++++++++++++++++++++++++++ internal/gateway/store.go | 6 +- 7 files changed, 931 insertions(+), 3 deletions(-) create mode 100644 cmd/agent/main.go create mode 100644 cmd/gateway/main.go create mode 100644 internal/checkpoint/file_test.go create mode 100644 internal/gateway/server.go create mode 100644 internal/gateway/server_test.go diff --git a/README.md b/README.md index 0e5b2f2..366290e 100644 --- a/README.md +++ b/README.md @@ -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 ./... +``` diff --git a/cmd/agent/main.go b/cmd/agent/main.go new file mode 100644 index 0000000..39e6b22 --- /dev/null +++ b/cmd/agent/main.go @@ -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") +} diff --git a/cmd/gateway/main.go b/cmd/gateway/main.go new file mode 100644 index 0000000..bc45df2 --- /dev/null +++ b/cmd/gateway/main.go @@ -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) + } +} diff --git a/internal/checkpoint/file_test.go b/internal/checkpoint/file_test.go new file mode 100644 index 0000000..ce541ad --- /dev/null +++ b/internal/checkpoint/file_test.go @@ -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") + } +} diff --git a/internal/gateway/server.go b/internal/gateway/server.go new file mode 100644 index 0000000..a7146fe --- /dev/null +++ b/internal/gateway/server.go @@ -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 +} diff --git a/internal/gateway/server_test.go b/internal/gateway/server_test.go new file mode 100644 index 0000000..0cf8a93 --- /dev/null +++ b/internal/gateway/server_test.go @@ -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") + } +} diff --git a/internal/gateway/store.go b/internal/gateway/store.go index 45fc2e3..17479f5 100644 --- a/internal/gateway/store.go +++ b/internal/gateway/store.go @@ -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