370 lines
11 KiB
Go
370 lines
11 KiB
Go
// 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
|
||
}
|