251 lines
7.4 KiB
Go
251 lines
7.4 KiB
Go
package metrics
|
|
|
|
import (
|
|
"fmt"
|
|
"net/http"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
)
|
|
|
|
// Metrics 监控指标接口
|
|
type Metrics interface {
|
|
// 增加请求计数
|
|
IncRequestCount()
|
|
// 增加错误计数
|
|
IncErrorCount(err error)
|
|
// 观察请求持续时间
|
|
ObserveRequestDuration(seconds float64)
|
|
// 增加活跃连接数
|
|
IncActiveConnections()
|
|
// 减少活跃连接数
|
|
DecActiveConnections()
|
|
// 设置后端健康状态
|
|
SetBackendHealth(backend string, healthy bool)
|
|
// 设置后端响应时间
|
|
SetBackendResponseTime(backend string, duration time.Duration)
|
|
// 观察请求字节数
|
|
ObserveRequestBytes(bytes int64)
|
|
// 观察响应字节数
|
|
ObserveResponseBytes(bytes int64)
|
|
// 添加传输字节数
|
|
AddBytesTransferred(direction string, bytes int64)
|
|
// 增加缓存命中计数
|
|
IncCacheHit()
|
|
// 获取指标处理器
|
|
GetHandler() http.Handler
|
|
}
|
|
|
|
// SimpleMetrics 简单指标实现
|
|
type SimpleMetrics struct {
|
|
// 请求计数
|
|
requestCount int64
|
|
// 错误计数
|
|
errorCount int64
|
|
// 活跃连接数
|
|
activeConnections int64
|
|
// 累计响应时间
|
|
totalResponseTime int64
|
|
// 传输字节数
|
|
bytesTransferred map[string]int64
|
|
// 后端健康状态
|
|
backendHealth map[string]bool
|
|
// 后端响应时间
|
|
backendResponseTime map[string]time.Duration
|
|
// 缓存命中计数
|
|
cacheHits int64
|
|
// 互斥锁
|
|
mu sync.Mutex
|
|
}
|
|
|
|
// NewSimpleMetrics 创建简单指标
|
|
func NewSimpleMetrics() *SimpleMetrics {
|
|
return &SimpleMetrics{
|
|
bytesTransferred: make(map[string]int64),
|
|
backendHealth: make(map[string]bool),
|
|
backendResponseTime: make(map[string]time.Duration),
|
|
}
|
|
}
|
|
|
|
// IncRequestCount 增加请求计数
|
|
func (m *SimpleMetrics) IncRequestCount() {
|
|
atomic.AddInt64(&m.requestCount, 1)
|
|
}
|
|
|
|
// IncErrorCount 增加错误计数
|
|
func (m *SimpleMetrics) IncErrorCount(err error) {
|
|
atomic.AddInt64(&m.errorCount, 1)
|
|
}
|
|
|
|
// ObserveRequestDuration 观察请求持续时间
|
|
func (m *SimpleMetrics) ObserveRequestDuration(seconds float64) {
|
|
nsec := int64(seconds * float64(time.Second))
|
|
atomic.AddInt64(&m.totalResponseTime, nsec)
|
|
}
|
|
|
|
// IncActiveConnections 增加活跃连接数
|
|
func (m *SimpleMetrics) IncActiveConnections() {
|
|
atomic.AddInt64(&m.activeConnections, 1)
|
|
}
|
|
|
|
// DecActiveConnections 减少活跃连接数
|
|
func (m *SimpleMetrics) DecActiveConnections() {
|
|
atomic.AddInt64(&m.activeConnections, -1)
|
|
}
|
|
|
|
// SetBackendHealth 设置后端健康状态
|
|
func (m *SimpleMetrics) SetBackendHealth(backend string, healthy bool) {
|
|
m.backendHealth[backend] = healthy
|
|
}
|
|
|
|
// SetBackendResponseTime 设置后端响应时间
|
|
func (m *SimpleMetrics) SetBackendResponseTime(backend string, duration time.Duration) {
|
|
m.backendResponseTime[backend] = duration
|
|
}
|
|
|
|
// ObserveRequestBytes 观察请求字节数
|
|
func (m *SimpleMetrics) ObserveRequestBytes(bytes int64) {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
m.bytesTransferred["request"] += bytes
|
|
}
|
|
|
|
// ObserveResponseBytes 观察响应字节数
|
|
func (m *SimpleMetrics) ObserveResponseBytes(bytes int64) {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
m.bytesTransferred["response"] += bytes
|
|
}
|
|
|
|
// AddBytesTransferred 添加传输字节数
|
|
func (m *SimpleMetrics) AddBytesTransferred(direction string, bytes int64) {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
m.bytesTransferred[direction] += bytes
|
|
}
|
|
|
|
// IncCacheHit 增加缓存命中计数
|
|
func (m *SimpleMetrics) IncCacheHit() {
|
|
atomic.AddInt64(&m.cacheHits, 1)
|
|
}
|
|
|
|
// GetHandler 获取指标处理器
|
|
func (m *SimpleMetrics) GetHandler() http.Handler {
|
|
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
w.Header().Set("Content-Type", "text/plain")
|
|
|
|
// 输出基本指标
|
|
w.Write([]byte("# HELP proxy_requests_total 代理请求总数\n"))
|
|
w.Write([]byte("# TYPE proxy_requests_total counter\n"))
|
|
w.Write([]byte(fmt.Sprintf("proxy_requests_total %d\n", m.requestCount)))
|
|
|
|
w.Write([]byte("# HELP proxy_errors_total 代理错误总数\n"))
|
|
w.Write([]byte("# TYPE proxy_errors_total counter\n"))
|
|
w.Write([]byte(fmt.Sprintf("proxy_errors_total %d\n", m.errorCount)))
|
|
|
|
w.Write([]byte("# HELP proxy_active_connections 当前活跃连接数\n"))
|
|
w.Write([]byte("# TYPE proxy_active_connections gauge\n"))
|
|
w.Write([]byte(fmt.Sprintf("proxy_active_connections %d\n", m.activeConnections)))
|
|
|
|
// 输出缓存命中数据
|
|
w.Write([]byte("# HELP proxy_cache_hits_total 缓存命中总数\n"))
|
|
w.Write([]byte("# TYPE proxy_cache_hits_total counter\n"))
|
|
w.Write([]byte(fmt.Sprintf("proxy_cache_hits_total %d\n", m.cacheHits)))
|
|
|
|
// 输出传输字节数
|
|
for direction, bytes := range m.bytesTransferred {
|
|
w.Write([]byte(fmt.Sprintf("# HELP proxy_bytes_transferred_%s 代理传输字节数(%s)\n", direction, direction)))
|
|
w.Write([]byte(fmt.Sprintf("# TYPE proxy_bytes_transferred_%s counter\n", direction)))
|
|
w.Write([]byte(fmt.Sprintf("proxy_bytes_transferred_%s %d\n", direction, bytes)))
|
|
}
|
|
|
|
// 输出后端健康状态
|
|
for backend, healthy := range m.backendHealth {
|
|
healthValue := 0
|
|
if healthy {
|
|
healthValue = 1
|
|
}
|
|
w.Write([]byte(fmt.Sprintf("# HELP proxy_backend_health 后端健康状态\n")))
|
|
w.Write([]byte(fmt.Sprintf("# TYPE proxy_backend_health gauge\n")))
|
|
w.Write([]byte(fmt.Sprintf("proxy_backend_health{backend=\"%s\"} %d\n", backend, healthValue)))
|
|
}
|
|
|
|
// 输出后端响应时间
|
|
for backend, duration := range m.backendResponseTime {
|
|
w.Write([]byte(fmt.Sprintf("# HELP proxy_backend_response_time 后端响应时间\n")))
|
|
w.Write([]byte(fmt.Sprintf("# TYPE proxy_backend_response_time gauge\n")))
|
|
w.Write([]byte(fmt.Sprintf("proxy_backend_response_time{backend=\"%s\"} %f\n", backend, float64(duration)/float64(time.Second))))
|
|
}
|
|
|
|
// 平均响应时间
|
|
if m.requestCount > 0 {
|
|
avgTime := float64(m.totalResponseTime) / float64(m.requestCount) / float64(time.Second)
|
|
w.Write([]byte("# HELP proxy_average_response_time 平均响应时间\n"))
|
|
w.Write([]byte("# TYPE proxy_average_response_time gauge\n"))
|
|
w.Write([]byte(fmt.Sprintf("proxy_average_response_time %f\n", avgTime)))
|
|
}
|
|
})
|
|
}
|
|
|
|
// PrometheusMetrics Prometheus指标实现
|
|
type PrometheusMetrics struct {
|
|
// 可以通过引入prometheus客户端库实现更完整的指标收集
|
|
// 此处省略具体实现
|
|
}
|
|
|
|
// MetricsMiddleware 指标中间件
|
|
type MetricsMiddleware struct {
|
|
metrics Metrics
|
|
}
|
|
|
|
// NewMetricsMiddleware 创建指标中间件
|
|
func NewMetricsMiddleware(metrics Metrics) *MetricsMiddleware {
|
|
return &MetricsMiddleware{
|
|
metrics: metrics,
|
|
}
|
|
}
|
|
|
|
// Middleware 中间件处理函数
|
|
func (m *MetricsMiddleware) Middleware(next http.Handler) http.Handler {
|
|
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
start := time.Now()
|
|
|
|
// 包装响应写入器，用于捕获状态码
|
|
rw := &responseWriter{ResponseWriter: w, statusCode: http.StatusOK}
|
|
|
|
// 继续处理请求
|
|
next.ServeHTTP(rw, r)
|
|
|
|
// 记录请求指标
|
|
duration := time.Since(start)
|
|
m.metrics.ObserveRequestDuration(duration.Seconds())
|
|
})
|
|
}
|
|
|
|
// responseWriter 包装的响应写入器
|
|
type responseWriter struct {
|
|
http.ResponseWriter
|
|
statusCode int
|
|
written int64
|
|
}
|
|
|
|
// WriteHeader 写入状态码
|
|
func (rw *responseWriter) WriteHeader(statusCode int) {
|
|
rw.statusCode = statusCode
|
|
rw.ResponseWriter.WriteHeader(statusCode)
|
|
}
|
|
|
|
// Write 写入数据
|
|
func (rw *responseWriter) Write(b []byte) (int, error) {
|
|
n, err := rw.ResponseWriter.Write(b)
|
|
rw.written += int64(n)
|
|
return n, err
|
|
}
|
|
|
|
// Flush 刷新数据
|
|
func (rw *responseWriter) Flush() {
|
|
if flusher, ok := rw.ResponseWriter.(http.Flusher); ok {
|
|
flusher.Flush()
|
|
}
|
|
}
|