refactor, fix metrics and upgrade go to 1.24.0

internal/metrics/period/entries.go

@@ -1,26 +1,28 @@
 package period
 
-import "time"
+import (
+	"time"
+)
 
 type Entries[T any] struct {
 	entries  [maxEntries]*T
 	index    int
 	count    int
-	interval int64
-	lastAdd  int64
+	interval time.Duration
+	lastAdd  time.Time
 }
 
-const maxEntries = 500
+const maxEntries = 200
 
-func newEntries[T any](interval int64) *Entries[T] {
+func newEntries[T any](duration time.Duration) *Entries[T] {
 	return &Entries[T]{
-		interval: interval,
-		lastAdd:  time.Now().Unix(),
+		interval: duration / maxEntries,
+		lastAdd:  time.Now(),
 	}
 }
 
-func (e *Entries[T]) Add(now int64, info *T) {
-	if now-e.lastAdd < e.interval {
+func (e *Entries[T]) Add(now time.Time, info *T) {
+	if now.Sub(e.lastAdd) < e.interval {
 		return
 	}
 	e.entries[e.index] = info

internal/metrics/period/handler.go

@@ -2,11 +2,11 @@ package period
 
 import (
 	"net/http"
+	"time"
 
 	"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) {
@@ -42,8 +42,35 @@ func (p *Poller[T, AggregateT]) ServeHTTP(w http.ResponseWriter, r *http.Request
 	}
 }
 
-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())
-	})
+func (p *Poller[T, AggregateT]) ServeWS(allowedDomains []string, w http.ResponseWriter, r *http.Request) {
+	query := r.URL.Query()
+	period := query.Get("period")
+	intervalStr := query.Get("interval")
+	interval, err := time.ParseDuration(intervalStr)
+
+	minInterval := p.interval()
+	if err != nil || interval < minInterval {
+		interval = minInterval
+	}
+
+	if period == "" {
+		utils.PeriodicWS(allowedDomains, w, r, interval, func(conn *websocket.Conn) error {
+			return wsjson.Write(r.Context(), conn, p.GetLastResult())
+		})
+	} else {
+		periodFilter := Filter(period)
+		if !periodFilter.IsValid() {
+			http.Error(w, "invalid period", http.StatusBadRequest)
+			return
+		}
+		if p.aggregator != nil {
+			utils.PeriodicWS(allowedDomains, w, r, interval, func(conn *websocket.Conn) error {
+				return wsjson.Write(r.Context(), conn, p.aggregator(p.Get(periodFilter)...))
+			})
+		} else {
+			utils.PeriodicWS(allowedDomains, w, r, interval, func(conn *websocket.Conn) error {
+				return wsjson.Write(r.Context(), conn, p.Get(periodFilter))
+			})
+		}
+	}
 }

internal/metrics/period/period.go

@@ -24,17 +24,17 @@ const (
 
 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),
+		FifteenMinutes: newEntries[T](15 * time.Minute),
+		OneHour:        newEntries[T](1 * time.Hour),
+		OneDay:         newEntries[T](24 * time.Hour),
+		OneMonth:       newEntries[T](30 * 24 * time.Hour),
 	}
 }
 
 func (p *Period[T]) Add(info *T) {
 	p.mu.Lock()
 	defer p.mu.Unlock()
-	now := time.Now().Unix()
+	now := time.Now()
 	p.FifteenMinutes.Add(now, info)
 	p.OneHour.Add(now, info)
 	p.OneDay.Add(now, info)

internal/metrics/period/poller.go

@@ -18,7 +18,6 @@ type (
 		poll       PollFunc[T]
 		aggregator AggregateFunc[T, AggregateT]
 		period     *Period[T]
-		interval   time.Duration
 		lastResult *T
 		errs       []pollErr
 	}
@@ -36,10 +35,9 @@ func NewPoller[T any](
 	poll PollFunc[T],
 ) *Poller[T, T] {
 	return &Poller[T, T]{
-		name:     name,
-		poll:     poll,
-		period:   NewPeriod[T](),
-		interval: interval,
+		name:   name,
+		poll:   poll,
+		period: NewPeriod[T](),
 	}
 }
 
@@ -54,10 +52,13 @@ func NewPollerWithAggregator[T, AggregateT any](
 		poll:       poll,
 		aggregator: aggregator,
 		period:     NewPeriod[T](),
-		interval:   interval,
 	}
 }
 
+func (p *Poller[T, AggregateT]) interval() time.Duration {
+	return p.period.FifteenMinutes.interval
+}
+
 func (p *Poller[T, AggregateT]) appendErr(err error) {
 	if len(p.errs) == 0 {
 		p.errs = []pollErr{
@@ -87,32 +88,36 @@ func (p *Poller[T, AggregateT]) gatherErrs() (string, bool) {
 	return strings.Join(errs, "\n"), true
 }
 
-func (p *Poller[T, AggregateT]) pollWithTimeout(ctx context.Context) (*T, error) {
-	ctx, cancel := context.WithTimeout(ctx, p.interval)
+func (p *Poller[T, AggregateT]) pollWithTimeout(ctx context.Context) {
+	ctx, cancel := context.WithTimeout(ctx, p.interval())
 	defer cancel()
-	return p.poll(ctx)
+	data, err := p.poll(ctx)
+	if err != nil {
+		p.appendErr(err)
+		return
+	}
+	p.period.Add(data)
+	p.lastResult = data
 }
 
 func (p *Poller[T, AggregateT]) Start() {
 	go func() {
 		ctx := task.RootContext()
-		ticker := time.NewTicker(p.interval)
+		ticker := time.NewTicker(p.interval())
 		gatherErrsTicker := time.NewTicker(gatherErrsInterval)
 		defer ticker.Stop()
 		defer gatherErrsTicker.Stop()
 
+		logging.Debug().Msgf("Starting poller %s with interval %s", p.name, p.interval())
+
+		p.pollWithTimeout(ctx)
+
 		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
+				p.pollWithTimeout(ctx)
 			case <-gatherErrsTicker.C:
 				errs, ok := p.gatherErrs()
 				if ok {

internal/metrics/uptime/uptime.go

@@ -2,21 +2,23 @@ package uptime
 
 import (
 	"context"
+	"encoding/json"
 	"time"
 
 	"github.com/yusing/go-proxy/internal/metrics/period"
 	"github.com/yusing/go-proxy/internal/route/routes/routequery"
+	"github.com/yusing/go-proxy/internal/utils/strutils"
 	"github.com/yusing/go-proxy/internal/watcher/health"
 )
 
 type (
 	Statuses struct {
 		Statuses  map[string]health.Status
-		Timestamp int64
+		Timestamp time.Time
 	}
 	Status struct {
 		Status    health.Status
-		Timestamp int64
+		Timestamp time.Time
 	}
 	Aggregated map[string][]Status
 )
@@ -30,7 +32,7 @@ func init() {
 func getStatuses(ctx context.Context) (*Statuses, error) {
 	return &Statuses{
 		Statuses:  routequery.HealthStatuses(),
-		Timestamp: time.Now().Unix(),
+		Timestamp: time.Now(),
 	}, nil
 }
 
@@ -71,3 +73,19 @@ func (a Aggregated) finalize() map[string]map[string]interface{} {
 	}
 	return result
 }
+
+func (s *Status) MarshalJSON() ([]byte, error) {
+	return json.Marshal(map[string]interface{}{
+		"status":    s.Status.String(),
+		"timestamp": s.Timestamp.Unix(),
+		"tooltip":   strutils.FormatTime(s.Timestamp),
+	})
+}
+
+func (s *Statuses) MarshalJSON() ([]byte, error) {
+	return json.Marshal(map[string]interface{}{
+		"statuses":  s.Statuses,
+		"timestamp": s.Timestamp.Unix(),
+		"tooltip":   strutils.FormatTime(s.Timestamp),
+	})
+}