mirror of
https://github.com/yusing/godoxy.git
synced 2026-04-24 01:08:31 +02:00
api: remove service health from prometheus, implement godoxy metrics
This commit is contained in:
yusing
@@ -1,13 +0,0 @@
|
||||
package metrics
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
|
||||
"github.com/prometheus/client_golang/prometheus/promhttp"
|
||||
)
|
||||
|
||||
func NewHandler() http.Handler {
|
||||
mux := http.NewServeMux()
|
||||
mux.Handle("/metrics", promhttp.Handler())
|
||||
return mux
|
||||
}
|
||||
@@ -9,7 +9,6 @@ type (
|
||||
StreamRouteMetricLabels struct {
|
||||
Service, Visitor string
|
||||
}
|
||||
HealthMetricLabels string
|
||||
)
|
||||
|
||||
func (lbl *HTTPRouteMetricLabels) toPromLabels() prometheus.Labels {
|
||||
@@ -28,9 +27,3 @@ func (lbl *StreamRouteMetricLabels) toPromLabels() prometheus.Labels {
|
||||
"visitor": lbl.Visitor,
|
||||
}
|
||||
}
|
||||
|
||||
func (lbl HealthMetricLabels) toPromLabels() prometheus.Labels {
|
||||
return prometheus.Labels{
|
||||
"service": string(lbl),
|
||||
}
|
||||
}
|
||||
|
||||
45
internal/metrics/period/entries.go
Normal file
45
internal/metrics/period/entries.go
Normal file
@@ -0,0 +1,45 @@
|
||||
package period
|
||||
|
||||
import "time"
|
||||
|
||||
type Entries[T any] struct {
|
||||
entries [maxEntries]*T
|
||||
index int
|
||||
count int
|
||||
interval int64
|
||||
lastAdd int64
|
||||
}
|
||||
|
||||
const maxEntries = 500
|
||||
|
||||
func newEntries[T any](interval int64) *Entries[T] {
|
||||
return &Entries[T]{
|
||||
interval: interval,
|
||||
lastAdd: time.Now().Unix(),
|
||||
}
|
||||
}
|
||||
|
||||
func (e *Entries[T]) Add(now int64, info *T) {
|
||||
if now-e.lastAdd < e.interval {
|
||||
return
|
||||
}
|
||||
e.entries[e.index] = info
|
||||
e.index++
|
||||
if e.index >= maxEntries {
|
||||
e.index = 0
|
||||
}
|
||||
if e.count < maxEntries {
|
||||
e.count++
|
||||
}
|
||||
e.lastAdd = now
|
||||
}
|
||||
|
||||
func (e *Entries[T]) Get() []*T {
|
||||
if e.count < maxEntries {
|
||||
return e.entries[:e.count]
|
||||
}
|
||||
res := make([]*T, maxEntries)
|
||||
copy(res, e.entries[e.index:])
|
||||
copy(res[maxEntries-e.index:], e.entries[:e.index])
|
||||
return res
|
||||
}
|
||||
49
internal/metrics/period/handler.go
Normal file
49
internal/metrics/period/handler.go
Normal file
@@ -0,0 +1,49 @@
|
||||
package period
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
|
||||
"github.com/coder/websocket"
|
||||
"github.com/coder/websocket/wsjson"
|
||||
"github.com/yusing/go-proxy/internal/api/v1/utils"
|
||||
config "github.com/yusing/go-proxy/internal/config/types"
|
||||
)
|
||||
|
||||
func (p *Poller[T, AggregateT]) lastResultHandler(w http.ResponseWriter, r *http.Request) {
|
||||
info := p.GetLastResult()
|
||||
if info == nil {
|
||||
http.Error(w, "no system info", http.StatusNoContent)
|
||||
return
|
||||
}
|
||||
utils.RespondJSON(w, r, info)
|
||||
}
|
||||
|
||||
func (p *Poller[T, AggregateT]) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
period := r.URL.Query().Get("period")
|
||||
if period == "" {
|
||||
p.lastResultHandler(w, r)
|
||||
return
|
||||
}
|
||||
periodFilter := Filter(period)
|
||||
if !periodFilter.IsValid() {
|
||||
http.Error(w, "invalid period", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
rangeData := p.Get(periodFilter)
|
||||
if len(rangeData) == 0 {
|
||||
http.Error(w, "no data", http.StatusNoContent)
|
||||
return
|
||||
}
|
||||
if p.aggregator != nil {
|
||||
aggregated := p.aggregator(rangeData...)
|
||||
utils.RespondJSON(w, r, aggregated)
|
||||
} else {
|
||||
utils.RespondJSON(w, r, rangeData)
|
||||
}
|
||||
}
|
||||
|
||||
func (p *Poller[T, AggregateT]) ServeWS(cfg config.ConfigInstance, w http.ResponseWriter, r *http.Request) {
|
||||
utils.PeriodicWS(cfg, w, r, p.interval, func(conn *websocket.Conn) error {
|
||||
return wsjson.Write(r.Context(), conn, p.GetLastResult())
|
||||
})
|
||||
}
|
||||
67
internal/metrics/period/period.go
Normal file
67
internal/metrics/period/period.go
Normal file
@@ -0,0 +1,67 @@
|
||||
package period
|
||||
|
||||
import (
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
type Period[T any] struct {
|
||||
FifteenMinutes *Entries[T]
|
||||
OneHour *Entries[T]
|
||||
OneDay *Entries[T]
|
||||
OneMonth *Entries[T]
|
||||
mu sync.RWMutex
|
||||
}
|
||||
|
||||
type Filter string
|
||||
|
||||
const (
|
||||
PeriodFifteenMinutes Filter = "15m"
|
||||
PeriodOneHour Filter = "1h"
|
||||
PeriodOneDay Filter = "1d"
|
||||
PeriodOneMonth Filter = "1m"
|
||||
)
|
||||
|
||||
func NewPeriod[T any]() *Period[T] {
|
||||
return &Period[T]{
|
||||
FifteenMinutes: newEntries[T](15 * 60 / maxEntries),
|
||||
OneHour: newEntries[T](60 * 60 / maxEntries),
|
||||
OneDay: newEntries[T](24 * 60 * 60 / maxEntries),
|
||||
OneMonth: newEntries[T](30 * 24 * 60 * 60 / maxEntries),
|
||||
}
|
||||
}
|
||||
|
||||
func (p *Period[T]) Add(info *T) {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
now := time.Now().Unix()
|
||||
p.FifteenMinutes.Add(now, info)
|
||||
p.OneHour.Add(now, info)
|
||||
p.OneDay.Add(now, info)
|
||||
p.OneMonth.Add(now, info)
|
||||
}
|
||||
|
||||
func (p *Period[T]) Get(filter Filter) []*T {
|
||||
p.mu.RLock()
|
||||
defer p.mu.RUnlock()
|
||||
switch filter {
|
||||
case PeriodFifteenMinutes:
|
||||
return p.FifteenMinutes.Get()
|
||||
case PeriodOneHour:
|
||||
return p.OneHour.Get()
|
||||
case PeriodOneDay:
|
||||
return p.OneDay.Get()
|
||||
case PeriodOneMonth:
|
||||
return p.OneMonth.Get()
|
||||
default:
|
||||
panic("invalid period filter")
|
||||
}
|
||||
}
|
||||
|
||||
func (filter Filter) IsValid() bool {
|
||||
switch filter {
|
||||
case PeriodFifteenMinutes, PeriodOneHour, PeriodOneDay, PeriodOneMonth:
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
132
internal/metrics/period/poller.go
Normal file
132
internal/metrics/period/poller.go
Normal file
@@ -0,0 +1,132 @@
|
||||
package period
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/yusing/go-proxy/internal/logging"
|
||||
"github.com/yusing/go-proxy/internal/task"
|
||||
)
|
||||
|
||||
type (
|
||||
PollFunc[T any] func(ctx context.Context) (*T, error)
|
||||
AggregateFunc[T, AggregateT any] func(entries ...*T) AggregateT
|
||||
Poller[T, AggregateT any] struct {
|
||||
name string
|
||||
poll PollFunc[T]
|
||||
aggregator AggregateFunc[T, AggregateT]
|
||||
period *Period[T]
|
||||
interval time.Duration
|
||||
lastResult *T
|
||||
errs []pollErr
|
||||
}
|
||||
pollErr struct {
|
||||
err error
|
||||
count int
|
||||
}
|
||||
)
|
||||
|
||||
const gatherErrsInterval = 30 * time.Second
|
||||
|
||||
func NewPoller[T any](
|
||||
name string,
|
||||
interval time.Duration,
|
||||
poll PollFunc[T],
|
||||
) *Poller[T, T] {
|
||||
return &Poller[T, T]{
|
||||
name: name,
|
||||
poll: poll,
|
||||
period: NewPeriod[T](),
|
||||
interval: interval,
|
||||
}
|
||||
}
|
||||
|
||||
func NewPollerWithAggregator[T, AggregateT any](
|
||||
name string,
|
||||
interval time.Duration,
|
||||
poll PollFunc[T],
|
||||
aggregator AggregateFunc[T, AggregateT],
|
||||
) *Poller[T, AggregateT] {
|
||||
return &Poller[T, AggregateT]{
|
||||
name: name,
|
||||
poll: poll,
|
||||
aggregator: aggregator,
|
||||
period: NewPeriod[T](),
|
||||
interval: interval,
|
||||
}
|
||||
}
|
||||
|
||||
func (p *Poller[T, AggregateT]) appendErr(err error) {
|
||||
if len(p.errs) == 0 {
|
||||
p.errs = []pollErr{
|
||||
{err: err, count: 1},
|
||||
}
|
||||
return
|
||||
}
|
||||
for i, e := range p.errs {
|
||||
if e.err.Error() == err.Error() {
|
||||
p.errs[i].count++
|
||||
return
|
||||
}
|
||||
}
|
||||
p.errs = append(p.errs, pollErr{err: err, count: 1})
|
||||
}
|
||||
|
||||
func (p *Poller[T, AggregateT]) gatherErrs() (string, bool) {
|
||||
if len(p.errs) == 0 {
|
||||
return "", false
|
||||
}
|
||||
title := fmt.Sprintf("Poller %s has encountered %d errors in the last %s seconds:", p.name, len(p.errs), gatherErrsInterval)
|
||||
errs := make([]string, 0, len(p.errs)+1)
|
||||
errs = append(errs, title)
|
||||
for _, e := range p.errs {
|
||||
errs = append(errs, fmt.Sprintf("%s: %d times", e.err.Error(), e.count))
|
||||
}
|
||||
return strings.Join(errs, "\n"), true
|
||||
}
|
||||
|
||||
func (p *Poller[T, AggregateT]) pollWithTimeout(ctx context.Context) (*T, error) {
|
||||
ctx, cancel := context.WithTimeout(ctx, p.interval)
|
||||
defer cancel()
|
||||
return p.poll(ctx)
|
||||
}
|
||||
|
||||
func (p *Poller[T, AggregateT]) Start() {
|
||||
go func() {
|
||||
ctx := task.RootContext()
|
||||
ticker := time.NewTicker(p.interval)
|
||||
gatherErrsTicker := time.NewTicker(gatherErrsInterval)
|
||||
defer ticker.Stop()
|
||||
defer gatherErrsTicker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
data, err := p.pollWithTimeout(ctx)
|
||||
if err != nil {
|
||||
p.appendErr(err)
|
||||
continue
|
||||
}
|
||||
p.period.Add(data)
|
||||
p.lastResult = data
|
||||
case <-gatherErrsTicker.C:
|
||||
errs, ok := p.gatherErrs()
|
||||
if ok {
|
||||
logging.Error().Msg(errs)
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
func (p *Poller[T, AggregateT]) Get(filter Filter) []*T {
|
||||
return p.period.Get(filter)
|
||||
}
|
||||
|
||||
func (p *Poller[T, AggregateT]) GetLastResult() *T {
|
||||
return p.lastResult
|
||||
}
|
||||
@@ -1,80 +0,0 @@
|
||||
package metrics
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"time"
|
||||
|
||||
"github.com/shirou/gopsutil/v4/cpu"
|
||||
"github.com/shirou/gopsutil/v4/disk"
|
||||
"github.com/shirou/gopsutil/v4/mem"
|
||||
"github.com/shirou/gopsutil/v4/net"
|
||||
"github.com/shirou/gopsutil/v4/sensors"
|
||||
"github.com/yusing/go-proxy/internal/utils/strutils"
|
||||
)
|
||||
|
||||
type (
|
||||
SystemInfo struct {
|
||||
CPUAverage float64
|
||||
Memory *mem.VirtualMemoryStat
|
||||
Disk *disk.UsageStat
|
||||
Network *net.IOCountersStat
|
||||
Sensors []sensors.TemperatureStat
|
||||
}
|
||||
)
|
||||
|
||||
func GetSystemInfo(ctx context.Context) (*SystemInfo, error) {
|
||||
memoryInfo, err := mem.VirtualMemory()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
cpuAverage, err := cpu.PercentWithContext(ctx, time.Second, false)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
diskInfo, err := disk.Usage("/")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
networkInfo, err := net.IOCounters(false)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
sensors, err := sensors.SensorsTemperatures()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &SystemInfo{
|
||||
CPUAverage: cpuAverage[0],
|
||||
Memory: memoryInfo,
|
||||
Disk: diskInfo,
|
||||
Network: &networkInfo[0],
|
||||
Sensors: sensors,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (info *SystemInfo) MarshalJSON() ([]byte, error) {
|
||||
return json.Marshal(map[string]interface{}{
|
||||
"cpu_average": info.CPUAverage,
|
||||
"memory": map[string]interface{}{
|
||||
"total": strutils.FormatByteSize(info.Memory.Total),
|
||||
"available": strutils.FormatByteSize(info.Memory.Available),
|
||||
"used": strutils.FormatByteSize(info.Memory.Used),
|
||||
"used_percent": info.Memory.UsedPercent,
|
||||
"free": strutils.FormatByteSize(info.Memory.Free),
|
||||
},
|
||||
"disk": map[string]interface{}{
|
||||
"total": strutils.FormatByteSize(info.Disk.Total),
|
||||
"used": strutils.FormatByteSize(info.Disk.Used),
|
||||
"used_percent": info.Disk.UsedPercent,
|
||||
"free": strutils.FormatByteSize(info.Disk.Free),
|
||||
"fs_type": info.Disk.Fstype,
|
||||
},
|
||||
"network": map[string]interface{}{
|
||||
"bytes_sent": strutils.FormatByteSize(info.Network.BytesSent),
|
||||
"bytes_recv": strutils.FormatByteSize(info.Network.BytesRecv),
|
||||
},
|
||||
"sensors": info.Sensors,
|
||||
})
|
||||
}
|
||||
60
internal/metrics/systeminfo/system_info.go
Normal file
60
internal/metrics/systeminfo/system_info.go
Normal file
@@ -0,0 +1,60 @@
|
||||
package systeminfo
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"github.com/shirou/gopsutil/v4/cpu"
|
||||
"github.com/shirou/gopsutil/v4/disk"
|
||||
"github.com/shirou/gopsutil/v4/mem"
|
||||
"github.com/shirou/gopsutil/v4/net"
|
||||
"github.com/shirou/gopsutil/v4/sensors"
|
||||
"github.com/yusing/go-proxy/internal/metrics/period"
|
||||
)
|
||||
|
||||
type SystemInfo struct {
|
||||
Timestamp time.Time
|
||||
CPUAverage float64
|
||||
Memory *mem.VirtualMemoryStat
|
||||
Disk *disk.UsageStat
|
||||
Network *net.IOCountersStat
|
||||
Sensors []sensors.TemperatureStat
|
||||
}
|
||||
|
||||
var Poller = period.NewPoller("system_info", 1*time.Second, getSystemInfo)
|
||||
|
||||
func init() {
|
||||
Poller.Start()
|
||||
}
|
||||
|
||||
func getSystemInfo(ctx context.Context) (*SystemInfo, error) {
|
||||
memoryInfo, err := mem.VirtualMemory()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
cpuAverage, err := cpu.PercentWithContext(ctx, 150*time.Millisecond, false)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
diskInfo, err := disk.Usage("/")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
networkInfo, err := net.IOCounters(false)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
sensors, err := sensors.SensorsTemperatures()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &SystemInfo{
|
||||
Timestamp: time.Now(),
|
||||
CPUAverage: cpuAverage[0],
|
||||
Memory: memoryInfo,
|
||||
Disk: diskInfo,
|
||||
Network: &networkInfo[0],
|
||||
Sensors: sensors,
|
||||
}, nil
|
||||
}
|
||||
73
internal/metrics/uptime/uptime.go
Normal file
73
internal/metrics/uptime/uptime.go
Normal file
@@ -0,0 +1,73 @@
|
||||
package uptime
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"github.com/yusing/go-proxy/internal/metrics/period"
|
||||
"github.com/yusing/go-proxy/internal/route/routes/routequery"
|
||||
"github.com/yusing/go-proxy/internal/watcher/health"
|
||||
)
|
||||
|
||||
type (
|
||||
Statuses struct {
|
||||
Statuses map[string]health.Status
|
||||
Timestamp int64
|
||||
}
|
||||
Status struct {
|
||||
Status health.Status
|
||||
Timestamp int64
|
||||
}
|
||||
Aggregated map[string][]Status
|
||||
)
|
||||
|
||||
var Poller = period.NewPollerWithAggregator("uptime", 1*time.Second, getStatuses, aggregateStatuses)
|
||||
|
||||
func init() {
|
||||
Poller.Start()
|
||||
}
|
||||
|
||||
func getStatuses(ctx context.Context) (*Statuses, error) {
|
||||
return &Statuses{
|
||||
Statuses: routequery.HealthStatuses(),
|
||||
Timestamp: time.Now().Unix(),
|
||||
}, nil
|
||||
}
|
||||
|
||||
func aggregateStatuses(entries ...*Statuses) any {
|
||||
aggregated := make(Aggregated)
|
||||
for _, entry := range entries {
|
||||
for alias, status := range entry.Statuses {
|
||||
aggregated[alias] = append(aggregated[alias], Status{
|
||||
Status: status,
|
||||
Timestamp: entry.Timestamp,
|
||||
})
|
||||
}
|
||||
}
|
||||
return aggregated.finalize()
|
||||
}
|
||||
|
||||
func (a Aggregated) calculateUptime(alias string) float64 {
|
||||
aggregated := a[alias]
|
||||
if len(aggregated) == 0 {
|
||||
return 0
|
||||
}
|
||||
uptime := 0
|
||||
for _, status := range aggregated {
|
||||
if status.Status == health.StatusHealthy {
|
||||
uptime++
|
||||
}
|
||||
}
|
||||
return float64(uptime) / float64(len(aggregated))
|
||||
}
|
||||
|
||||
func (a Aggregated) finalize() map[string]map[string]interface{} {
|
||||
result := make(map[string]map[string]interface{}, len(a))
|
||||
for alias, statuses := range a {
|
||||
result[alias] = map[string]interface{}{
|
||||
"uptime": a.calculateUptime(alias),
|
||||
"statuses": statuses,
|
||||
}
|
||||
}
|
||||
return result
|
||||
}
|
||||
Reference in New Issue
Block a user