293 lines
7.7 KiB
Go
293 lines
7.7 KiB
Go
// Package gateway 实现 Collector Gateway：接收 Agent 上报、提供指标/日志
|
||
// 查询 REST API，并维护内存存储与已确认序号。
|
||
package gateway
|
||
|
||
import (
|
||
"fmt"
|
||
"sort"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
|
||
"git.opencomputing.cn/yumoqing/host-metrics-collector/internal/model"
|
||
)
|
||
|
||
// MetricPoint 是时序查询返回的单点。
|
||
type MetricPoint struct {
|
||
Timestamp int64 `json:"timestamp"`
|
||
Value float64 `json:"value"`
|
||
}
|
||
|
||
// HostMetricSeries 是一组相同 metric + labels 的时序数据。
|
||
type HostMetricSeries struct {
|
||
Metric string `json:"metric"`
|
||
Labels map[string]string `json:"labels"`
|
||
Values []MetricPoint `json:"values"`
|
||
}
|
||
|
||
// LogEntryDTO 是日志查询返回的单条日志。
|
||
type LogEntryDTO struct {
|
||
Timestamp int64 `json:"timestamp"`
|
||
Level string `json:"level"`
|
||
Source string `json:"source"`
|
||
Message string `json:"message"`
|
||
Fields map[string]string `json:"fields,omitempty"`
|
||
File string `json:"file"`
|
||
Offset int64 `json:"offset"`
|
||
}
|
||
|
||
// Store 是 Gateway 的线程安全内存存储。生产环境可替换为 Kafka sink +
|
||
// VictoriaMetrics/Elasticsearch，本模块为独立可运行实现。
|
||
type Store struct {
|
||
mu sync.RWMutex
|
||
|
||
metrics map[string][]model.Sample
|
||
logs map[string][]model.LogRecord
|
||
ackedSeqs map[string]int64
|
||
lastSeqByHost map[string]int64
|
||
|
||
maxMetricsPerHost int
|
||
maxLogsPerHost int
|
||
}
|
||
|
||
// NewStore 创建内存存储。
|
||
func NewStore(maxMetricsPerHost, maxLogsPerHost int) *Store {
|
||
return &Store{
|
||
metrics: map[string][]model.Sample{},
|
||
logs: map[string][]model.LogRecord{},
|
||
ackedSeqs: map[string]int64{},
|
||
lastSeqByHost: map[string]int64{},
|
||
maxMetricsPerHost: maxMetricsPerHost,
|
||
maxLogsPerHost: maxLogsPerHost,
|
||
}
|
||
}
|
||
|
||
// AddMetrics 追加指标样本并记录最新 seq。
|
||
func (s *Store) AddMetrics(hostID string, seq int64, samples []model.Sample) {
|
||
s.mu.Lock()
|
||
defer s.mu.Unlock()
|
||
|
||
existing := s.metrics[hostID]
|
||
existing = append(existing, samples...)
|
||
if max := s.maxMetricsPerHost; max > 0 && len(existing) > max {
|
||
existing = existing[len(existing)-max:]
|
||
}
|
||
s.metrics[hostID] = existing
|
||
if seq > s.lastSeqByHost[hostID] {
|
||
s.lastSeqByHost[hostID] = seq
|
||
}
|
||
}
|
||
|
||
// AddLogs 追加日志记录并记录最新 seq。
|
||
func (s *Store) AddLogs(hostID string, seq int64, entries []model.LogRecord) {
|
||
s.mu.Lock()
|
||
defer s.mu.Unlock()
|
||
|
||
existing := s.logs[hostID]
|
||
existing = append(existing, entries...)
|
||
if max := s.maxLogsPerHost; max > 0 && len(existing) > max {
|
||
existing = existing[len(existing)-max:]
|
||
}
|
||
s.logs[hostID] = existing
|
||
if seq > s.lastSeqByHost[hostID] {
|
||
s.lastSeqByHost[hostID] = seq
|
||
}
|
||
}
|
||
|
||
// LastAckedSeq 返回服务端已确认（最近收到）的序号。
|
||
func (s *Store) LastAckedSeq(hostID string) int64 {
|
||
s.mu.RLock()
|
||
defer s.mu.RUnlock()
|
||
return s.ackedSeqs[hostID]
|
||
}
|
||
|
||
// SetLastAckedSeq 推进服务端已确认序号。
|
||
func (s *Store) SetLastAckedSeq(hostID string, seq int64) {
|
||
s.mu.Lock()
|
||
defer s.mu.Unlock()
|
||
if seq > s.ackedSeqs[hostID] {
|
||
s.ackedSeqs[hostID] = seq
|
||
}
|
||
}
|
||
|
||
// HostMetrics 查询单主机指标时序；metric 为空则返回全部指标。
|
||
func (s *Store) HostMetrics(hostID, metric string, from, to time.Time) ([]HostMetricSeries, error) {
|
||
s.mu.RLock()
|
||
samples := append([]model.Sample(nil), s.metrics[hostID]...)
|
||
s.mu.RUnlock()
|
||
return groupSeries(samples, metric, nil, from, to), nil
|
||
}
|
||
|
||
// QueryMetrics 跨主机按 metric 与标签过滤查询时序。
|
||
func (s *Store) QueryMetrics(metric string, labels map[string]string, from, to time.Time) ([]HostMetricSeries, error) {
|
||
s.mu.RLock()
|
||
var samples []model.Sample
|
||
for _, ss := range s.metrics {
|
||
samples = append(samples, ss...)
|
||
}
|
||
s.mu.RUnlock()
|
||
return groupSeries(samples, metric, labels, from, to), nil
|
||
}
|
||
|
||
// InstantMetrics 返回每个匹配序列的最新点。
|
||
func (s *Store) InstantMetrics(metric string, labels map[string]string) ([]MetricPoint, error) {
|
||
series, err := s.QueryMetrics(metric, labels, time.Time{}, time.Now().Add(time.Hour))
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
out := make([]MetricPoint, 0, len(series))
|
||
for _, ser := range series {
|
||
if len(ser.Values) == 0 {
|
||
continue
|
||
}
|
||
out = append(out, ser.Values[len(ser.Values)-1])
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
// HostLogs 查询单主机日志（支持级别、关键字、时间范围、分页）。
|
||
func (s *Store) HostLogs(hostID, level, keyword string, from, to time.Time, page, pageSize int) (int, []LogEntryDTO, error) {
|
||
s.mu.RLock()
|
||
entries := append([]model.LogRecord(nil), s.logs[hostID]...)
|
||
s.mu.RUnlock()
|
||
return filterLogs(entries, level, keyword, from, to, page, pageSize)
|
||
}
|
||
|
||
// SearchLogs 跨主机检索日志。
|
||
func (s *Store) SearchLogs(hostIDs []string, level, keyword string, from, to time.Time, page, pageSize int) (int, []LogEntryDTO, error) {
|
||
s.mu.RLock()
|
||
var entries []model.LogRecord
|
||
if len(hostIDs) == 0 {
|
||
for _, es := range s.logs {
|
||
entries = append(entries, es...)
|
||
}
|
||
} else {
|
||
for _, id := range hostIDs {
|
||
entries = append(entries, s.logs[id]...)
|
||
}
|
||
}
|
||
s.mu.RUnlock()
|
||
return filterLogs(entries, level, keyword, from, to, page, pageSize)
|
||
}
|
||
|
||
func groupSeries(samples []model.Sample, metric string, labels map[string]string, from, to time.Time) []HostMetricSeries {
|
||
type key struct {
|
||
metric string
|
||
labels string
|
||
}
|
||
groups := map[key]*HostMetricSeries{}
|
||
var order []key
|
||
|
||
for _, s := range samples {
|
||
if metric != "" && s.Name != metric {
|
||
continue
|
||
}
|
||
if !matchLabels(s.Labels, labels) {
|
||
continue
|
||
}
|
||
ts := s.Timestamp
|
||
if !from.IsZero() && ts < from.Unix() {
|
||
continue
|
||
}
|
||
if !to.IsZero() && ts > to.Unix() {
|
||
continue
|
||
}
|
||
|
||
k := key{metric: s.Name, labels: labelsString(s.Labels)}
|
||
ser, ok := groups[k]
|
||
if !ok {
|
||
ser = &HostMetricSeries{Metric: s.Name, Labels: s.Labels}
|
||
groups[k] = ser
|
||
order = append(order, k)
|
||
}
|
||
ser.Values = append(ser.Values, MetricPoint{Timestamp: s.Timestamp, Value: s.Value})
|
||
}
|
||
|
||
out := make([]HostMetricSeries, 0, len(order))
|
||
for _, k := range order {
|
||
ser := groups[k]
|
||
sort.Slice(ser.Values, func(i, j int) bool { return ser.Values[i].Timestamp < ser.Values[j].Timestamp })
|
||
out = append(out, *ser)
|
||
}
|
||
return out
|
||
}
|
||
|
||
func filterLogs(entries []model.LogRecord, level, keyword string, from, to time.Time, page, pageSize int) (int, []LogEntryDTO, error) {
|
||
if page < 1 {
|
||
page = 1
|
||
}
|
||
if pageSize < 1 {
|
||
pageSize = 20
|
||
}
|
||
|
||
var matched []model.LogRecord
|
||
for _, e := range entries {
|
||
if level != "" && !strings.EqualFold(e.Level, level) {
|
||
continue
|
||
}
|
||
if keyword != "" && !strings.Contains(e.Message, keyword) {
|
||
continue
|
||
}
|
||
if !from.IsZero() && e.Timestamp < from.Unix() {
|
||
continue
|
||
}
|
||
if !to.IsZero() && e.Timestamp > to.Unix() {
|
||
continue
|
||
}
|
||
matched = append(matched, e)
|
||
}
|
||
|
||
total := len(matched)
|
||
start := (page - 1) * pageSize
|
||
if start > total {
|
||
start = total
|
||
}
|
||
end := start + pageSize
|
||
if end > total {
|
||
end = total
|
||
}
|
||
|
||
items := make([]LogEntryDTO, 0, end-start)
|
||
for _, e := range matched[start:end] {
|
||
items = append(items, LogEntryDTO{
|
||
Timestamp: e.Timestamp,
|
||
Level: e.Level,
|
||
Source: e.Source,
|
||
Message: e.Message,
|
||
Fields: e.Fields,
|
||
File: e.File,
|
||
Offset: e.Offset,
|
||
})
|
||
}
|
||
return total, items, nil
|
||
}
|
||
|
||
func matchLabels(got, want map[string]string) bool {
|
||
if len(want) == 0 {
|
||
return true
|
||
}
|
||
for k, v := range want {
|
||
if got[k] != v {
|
||
return false
|
||
}
|
||
}
|
||
return true
|
||
}
|
||
|
||
func labelsString(labels map[string]string) string {
|
||
if len(labels) == 0 {
|
||
return ""
|
||
}
|
||
keys := make([]string, 0, len(labels))
|
||
for k := range labels {
|
||
keys = append(keys, k)
|
||
}
|
||
sort.Strings(keys)
|
||
|
||
var b strings.Builder
|
||
for _, k := range keys {
|
||
fmt.Fprintf(&b, "%s=%s;", k, labels[k])
|
||
}
|
||
return b.String()
|
||
}
|