Import source: internal/api/api
Import the existing local project snapshot. This commit organizes existing files; it does not represent newly developed functionality.
This commit is contained in:
arsam
@@ -0,0 +1,259 @@
|
|||||||
|
// Package api exposes the REST endpoints and the SSE stream.
|
||||||
|
package api
|
||||||
|
|
||||||
|
import (
|
||||||
|
"encoding/json"
|
||||||
|
"fmt"
|
||||||
|
"net/http"
|
||||||
|
"strconv"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"dash/internal/engine"
|
||||||
|
"dash/internal/store"
|
||||||
|
)
|
||||||
|
|
||||||
|
type API struct {
|
||||||
|
engine *engine.Engine
|
||||||
|
store *store.Store
|
||||||
|
hub *engine.Hub
|
||||||
|
}
|
||||||
|
|
||||||
|
func New(e *engine.Engine, s *store.Store, h *engine.Hub) *API {
|
||||||
|
return &API{engine: e, store: s, hub: h}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *API) Register(mux *http.ServeMux) {
|
||||||
|
mux.HandleFunc("GET /api/targets", a.handleTargets)
|
||||||
|
mux.HandleFunc("GET /api/targets/{id}/history", a.handleHistory)
|
||||||
|
mux.HandleFunc("GET /api/incidents", a.handleIncidents)
|
||||||
|
mux.HandleFunc("GET /api/stream", a.handleStream)
|
||||||
|
}
|
||||||
|
|
||||||
|
func writeJSON(w http.ResponseWriter, v any) {
|
||||||
|
w.Header().Set("Content-Type", "application/json")
|
||||||
|
json.NewEncoder(w).Encode(v)
|
||||||
|
}
|
||||||
|
|
||||||
|
// parseRange turns "30m", "24h", "7d" into a duration. Default 24h.
|
||||||
|
func parseRange(s string) (time.Duration, error) {
|
||||||
|
if s == "" {
|
||||||
|
return 24 * time.Hour, nil
|
||||||
|
}
|
||||||
|
if n, err := strconv.Atoi(s[:len(s)-1]); err == nil && n > 0 {
|
||||||
|
switch s[len(s)-1] {
|
||||||
|
case 'm':
|
||||||
|
return time.Duration(n) * time.Minute, nil
|
||||||
|
case 'h':
|
||||||
|
return time.Duration(n) * time.Hour, nil
|
||||||
|
case 'd':
|
||||||
|
return time.Duration(n) * 24 * time.Hour, nil
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return 0, fmt.Errorf("invalid range %q (use e.g. 30m, 24h, 7d)", s)
|
||||||
|
}
|
||||||
|
|
||||||
|
// TargetView is the API shape for one target's current state.
|
||||||
|
type TargetView struct {
|
||||||
|
ID string `json:"id"`
|
||||||
|
Name string `json:"name"`
|
||||||
|
Type string `json:"type"`
|
||||||
|
Endpoint string `json:"endpoint"`
|
||||||
|
IntervalSec int `json:"interval_sec"`
|
||||||
|
State engine.State `json:"state"`
|
||||||
|
Fails int `json:"fails"`
|
||||||
|
Last *store.Check `json:"last,omitempty"`
|
||||||
|
Uptime24h *float64 `json:"uptime_24h,omitempty"`
|
||||||
|
Uptime7d *float64 `json:"uptime_7d,omitempty"`
|
||||||
|
Uptime30d *float64 `json:"uptime_30d,omitempty"`
|
||||||
|
Latency store.LatencyStats `json:"latency"`
|
||||||
|
CertDays *int `json:"cert_days,omitempty"`
|
||||||
|
Incident *store.Incident `json:"incident,omitempty"`
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *API) targetViews() []TargetView {
|
||||||
|
now := time.Now().UnixMilli()
|
||||||
|
day := now - 24*time.Hour.Milliseconds()
|
||||||
|
week := now - 7*24*time.Hour.Milliseconds()
|
||||||
|
month := now - 30*24*time.Hour.Milliseconds()
|
||||||
|
|
||||||
|
snap := a.engine.Snapshot()
|
||||||
|
out := make([]TargetView, 0, len(snap))
|
||||||
|
for _, ts := range snap {
|
||||||
|
t := ts.Target
|
||||||
|
v := TargetView{
|
||||||
|
ID: t.ID,
|
||||||
|
Name: t.Name,
|
||||||
|
Type: string(t.Type),
|
||||||
|
Endpoint: endpointOf(ts),
|
||||||
|
IntervalSec: int(t.Interval.Duration.Seconds()),
|
||||||
|
State: ts.State,
|
||||||
|
Fails: ts.Fails,
|
||||||
|
Incident: ts.Incident,
|
||||||
|
}
|
||||||
|
if ts.Last != nil {
|
||||||
|
v.Last = &store.Check{
|
||||||
|
TS: ts.Last.TS.UnixMilli(), OK: ts.Last.OK,
|
||||||
|
LatencyMS: ts.Last.LatencyMS, Detail: ts.Last.Detail, Err: ts.Last.Err,
|
||||||
|
}
|
||||||
|
if t.Type == "tls" && ts.Last.Detail != "" {
|
||||||
|
if d, err := strconv.Atoi(ts.Last.Detail); err == nil {
|
||||||
|
v.CertDays = &d
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if ok, total, err := a.store.Uptime(t.ID, day); err == nil && total > 0 {
|
||||||
|
pct := 100 * float64(ok) / float64(total)
|
||||||
|
v.Uptime24h = &pct
|
||||||
|
}
|
||||||
|
if ok, total, err := a.store.Uptime(t.ID, week); err == nil && total > 0 {
|
||||||
|
pct := 100 * float64(ok) / float64(total)
|
||||||
|
v.Uptime7d = &pct
|
||||||
|
}
|
||||||
|
if ok, total, err := a.store.Uptime(t.ID, month); err == nil && total > 0 {
|
||||||
|
pct := 100 * float64(ok) / float64(total)
|
||||||
|
v.Uptime30d = &pct
|
||||||
|
}
|
||||||
|
if stats, err := a.store.Latency(t.ID, day); err == nil {
|
||||||
|
v.Latency = stats
|
||||||
|
}
|
||||||
|
out = append(out, v)
|
||||||
|
}
|
||||||
|
return out
|
||||||
|
}
|
||||||
|
|
||||||
|
func endpointOf(ts engine.TargetStatus) string {
|
||||||
|
t := ts.Target
|
||||||
|
switch {
|
||||||
|
case t.URL != "":
|
||||||
|
return t.URL
|
||||||
|
case t.Address != "":
|
||||||
|
return t.Address
|
||||||
|
case t.Container != "":
|
||||||
|
return t.Container
|
||||||
|
default:
|
||||||
|
return t.Host
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *API) handleTargets(w http.ResponseWriter, r *http.Request) {
|
||||||
|
writeJSON(w, a.targetViews())
|
||||||
|
}
|
||||||
|
|
||||||
|
type historyResponse struct {
|
||||||
|
TargetID string `json:"target_id"`
|
||||||
|
Range string `json:"range"`
|
||||||
|
From int64 `json:"from"`
|
||||||
|
To int64 `json:"to"`
|
||||||
|
Points []store.Check `json:"points"`
|
||||||
|
Buckets []store.Bucket `json:"buckets"`
|
||||||
|
BucketMS int64 `json:"bucket_ms"`
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *API) handleHistory(w http.ResponseWriter, r *http.Request) {
|
||||||
|
id := r.PathValue("id")
|
||||||
|
rangeStr := r.URL.Query().Get("range")
|
||||||
|
dur, err := parseRange(rangeStr)
|
||||||
|
if err != nil {
|
||||||
|
http.Error(w, err.Error(), http.StatusBadRequest)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if rangeStr == "" {
|
||||||
|
rangeStr = "24h"
|
||||||
|
}
|
||||||
|
|
||||||
|
to := time.Now().UnixMilli()
|
||||||
|
from := to - dur.Milliseconds()
|
||||||
|
|
||||||
|
points, err := a.store.History(id, from, to, 500)
|
||||||
|
if err != nil {
|
||||||
|
http.Error(w, err.Error(), http.StatusInternalServerError)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
// 24 buckets across the window: 1h each for 24h, scaling with range.
|
||||||
|
bucketMS := dur.Milliseconds() / 24
|
||||||
|
buckets, err := a.store.Buckets(id, from, to, bucketMS)
|
||||||
|
if err != nil {
|
||||||
|
http.Error(w, err.Error(), http.StatusInternalServerError)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if points == nil {
|
||||||
|
points = []store.Check{}
|
||||||
|
}
|
||||||
|
if buckets == nil {
|
||||||
|
buckets = []store.Bucket{}
|
||||||
|
}
|
||||||
|
writeJSON(w, historyResponse{
|
||||||
|
TargetID: id, Range: rangeStr, From: from, To: to,
|
||||||
|
Points: points, Buckets: buckets, BucketMS: bucketMS,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *API) handleIncidents(w http.ResponseWriter, r *http.Request) {
|
||||||
|
dur, err := parseRange(r.URL.Query().Get("range"))
|
||||||
|
if err != nil {
|
||||||
|
http.Error(w, err.Error(), http.StatusBadRequest)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
from := time.Now().Add(-dur).UnixMilli()
|
||||||
|
incidents, err := a.store.Incidents(from)
|
||||||
|
if err != nil {
|
||||||
|
http.Error(w, err.Error(), http.StatusInternalServerError)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
writeJSON(w, incidents)
|
||||||
|
}
|
||||||
|
|
||||||
|
// handleStream is the SSE endpoint. Events:
|
||||||
|
//
|
||||||
|
// snapshot — full target state, sent once on connect
|
||||||
|
// check — every completed probe
|
||||||
|
// state — confirmed up/degraded/down transitions
|
||||||
|
func (a *API) handleStream(w http.ResponseWriter, r *http.Request) {
|
||||||
|
flusher, ok := w.(http.Flusher)
|
||||||
|
if !ok {
|
||||||
|
http.Error(w, "streaming unsupported", http.StatusInternalServerError)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
w.Header().Set("Content-Type", "text/event-stream")
|
||||||
|
w.Header().Set("Cache-Control", "no-cache")
|
||||||
|
w.Header().Set("Connection", "keep-alive")
|
||||||
|
w.Header().Set("X-Accel-Buffering", "no")
|
||||||
|
|
||||||
|
send := func(name string, v any) bool {
|
||||||
|
data, err := json.Marshal(v)
|
||||||
|
if err != nil {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
if _, err := fmt.Fprintf(w, "event: %s\ndata: %s\n\n", name, data); err != nil {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
flusher.Flush()
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
|
||||||
|
if !send("snapshot", a.targetViews()) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
sub := a.hub.Subscribe()
|
||||||
|
defer a.hub.Unsubscribe(sub)
|
||||||
|
|
||||||
|
keepalive := time.NewTicker(25 * time.Second)
|
||||||
|
defer keepalive.Stop()
|
||||||
|
|
||||||
|
for {
|
||||||
|
select {
|
||||||
|
case <-r.Context().Done():
|
||||||
|
return
|
||||||
|
case ev := <-sub:
|
||||||
|
if !send(ev.Name, ev.Data) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
case <-keepalive.C:
|
||||||
|
if _, err := fmt.Fprint(w, ": keepalive\n\n"); err != nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
flusher.Flush()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user