diff --git a/.gitignore b/.gitignore
index e1f6a8e..1fd01d6 100644
--- a/.gitignore
+++ b/.gitignore
@@ -38,4 +38,5 @@ tmp
# Old repo
/go-upkeep/
-*.local.json
\ No newline at end of file
+*.local.json
+*.local.md
\ No newline at end of file
diff --git a/cmd/goupkeep/main.go b/cmd/goupkeep/main.go
index 26c01b3..5962e37 100644
--- a/cmd/goupkeep/main.go
+++ b/cmd/goupkeep/main.go
@@ -1,6 +1,7 @@
package main
import (
+ "context"
"flag"
"fmt"
"go-upkeep/internal/cluster"
@@ -68,9 +69,6 @@ func main() {
if v := os.Getenv("UPKEEP_CLUSTER_SECRET"); v != "" {
clusterKey = v
}
- if os.Getenv("UPKEEP_INSECURE_SKIP_VERIFY") == "true" {
- monitor.SetInsecureSkipVerify(true)
- }
port := flag.Int("port", portVal, "SSH Port")
flagDBType := flag.String("db-type", dbType, "Database type")
@@ -80,20 +78,23 @@ func main() {
flag.Parse()
var s store.Store
+ var dbErr error
if *flagDBType == "postgres" {
- s = &store.PostgresStore{ConnStr: *flagDSN}
+ s, dbErr = store.NewPostgresStore(*flagDSN)
fmt.Printf("Using PostgreSQL: %s\n", *flagDSN)
} else {
- s = &store.SQLiteStore{DBPath: *flagDSN}
+ s, dbErr = store.NewSQLiteStore(*flagDSN)
fmt.Printf("Using SQLite: %s\n", *flagDSN)
}
+ if dbErr != nil {
+ fmt.Printf("Database connection error: %v\n", dbErr)
+ os.Exit(1)
+ }
if err := s.Init(); err != nil {
- fmt.Printf("Database Init Error: %v\n", err)
+ fmt.Printf("Database init error: %v\n", err)
os.Exit(1)
}
- store.SetGlobal(s)
-
if *demo {
seedDemoData(s)
}
@@ -112,26 +113,34 @@ func main() {
fmt.Printf("Imported %d monitors and %d alerts from Uptime Kuma v%s\n", len(backup.Sites), len(backup.Alerts), kb.Version)
}
- monitor.InitHistoryFromStore()
- monitor.StartEngine()
+ eng := monitor.NewEngine(s)
+ if os.Getenv("UPKEEP_INSECURE_SKIP_VERIFY") == "true" {
+ eng.SetInsecureSkipVerify(true)
+ }
+
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+
+ eng.InitHistory()
+ eng.Start(ctx)
server.Start(server.ServerConfig{
Port: httpPort,
EnableStatus: enableStatus,
Title: statusTitle,
ClusterKey: clusterKey,
- })
+ }, s, eng)
- cluster.Start(cluster.Config{
+ cluster.Start(ctx, cluster.Config{
Mode: clusterMode,
PeerURL: clusterPeer,
SharedKey: clusterKey,
- })
+ }, eng)
- startSSHServer(*port)
+ startSSHServer(*port, s, eng)
if isatty.IsTerminal(os.Stdout.Fd()) || isatty.IsCygwinTerminal(os.Stdout.Fd()) {
- p := tea.NewProgram(tui.InitialModel(true), tea.WithAltScreen(), tea.WithMouseCellMotion())
+ p := tea.NewProgram(tui.InitialModel(true, s, eng), tea.WithAltScreen(), tea.WithMouseCellMotion())
if _, err := p.Run(); err != nil {
fmt.Printf("Error: %v\n", err)
}
@@ -142,18 +151,19 @@ func main() {
<-done
fmt.Println("Shutting down...")
}
+ cancel()
}
-func startSSHServer(port int) {
+func startSSHServer(port int, db store.Store, eng *monitor.Engine) {
s, err := wish.NewServer(
wish.WithAddress(fmt.Sprintf(":%d", port)),
wish.WithHostKeyPath(".ssh/id_ed25519"),
wish.WithPublicKeyAuth(func(ctx ssh.Context, key ssh.PublicKey) bool {
- return isKeyAllowed(key)
+ return isKeyAllowed(db, key)
}),
wish.WithMiddleware(
bm.Middleware(func(s ssh.Session) (tea.Model, []tea.ProgramOption) {
- return tui.InitialModel(false), []tea.ProgramOption{tea.WithAltScreen(), tea.WithMouseCellMotion()}
+ return tui.InitialModel(false, db, eng), []tea.ProgramOption{tea.WithAltScreen(), tea.WithMouseCellMotion()}
}),
),
)
@@ -161,11 +171,16 @@ func startSSHServer(port int) {
fmt.Printf("SSH server error: %v\n", err)
return
}
- go func() { s.ListenAndServe() }()
+ go func() {
+ if err := s.ListenAndServe(); err != nil {
+ log.Fatalf("SSH server failed: %v", err)
+ }
+ }()
}
func seedDemoData(s store.Store) {
- if existing := s.GetSites(); len(existing) > 0 {
+ existing, _ := s.GetSites()
+ if len(existing) > 0 {
return
}
fmt.Println("Seeding demo data...")
@@ -178,7 +193,7 @@ func seedDemoData(s store.Store) {
"from": "oncall@example.com", "to": "team@example.com",
})
- alerts := s.GetAllAlerts()
+ alerts, _ := s.GetAllAlerts()
alertID := 0
if len(alerts) > 0 {
alertID = alerts[0].ID
@@ -196,8 +211,11 @@ func seedDemoData(s store.Store) {
s.AddSite(models.Site{Name: "SSH Server", Type: "port", Interval: 60, AlertID: alertID, Hostname: "10.0.0.1", Port: 22, Timeout: 5, ExpiryThreshold: 7})
}
-func isKeyAllowed(incomingKey ssh.PublicKey) bool {
- users := store.Get().GetAllUsers()
+func isKeyAllowed(db store.Store, incomingKey ssh.PublicKey) bool {
+ users, err := db.GetAllUsers()
+ if err != nil {
+ return false
+ }
for _, u := range users {
allowedKey, _, _, _, err := ssh.ParseAuthorizedKey([]byte(u.PublicKey))
if err != nil {
diff --git a/internal/alert/alert.go b/internal/alert/alert.go
index 71b7570..c60013f 100644
--- a/internal/alert/alert.go
+++ b/internal/alert/alert.go
@@ -7,6 +7,7 @@ import (
"go-upkeep/internal/models"
"net/http"
"net/smtp"
+ "strconv"
"strings"
"time"
)
@@ -17,15 +18,95 @@ type Provider interface {
Send(title, message string) error
}
+type PayloadFunc func(title, message string) ([]byte, error)
+
+type HTTPProvider struct {
+ URL string
+ Payload PayloadFunc
+}
+
+func (h *HTTPProvider) Send(title, message string) error {
+ body, err := h.Payload(title, message)
+ if err != nil {
+ return err
+ }
+ resp, err := alertClient.Post(h.URL, "application/json", bytes.NewBuffer(body))
+ if err != nil {
+ return err
+ }
+ defer resp.Body.Close()
+ if resp.StatusCode >= 400 {
+ return fmt.Errorf("alert webhook returned HTTP %d", resp.StatusCode)
+ }
+ return nil
+}
+
+func discordPayload(title, message string) ([]byte, error) {
+ return json.Marshal(map[string]string{"content": fmt.Sprintf("**%s**\n%s", title, message)})
+}
+
+func slackPayload(title, message string) ([]byte, error) {
+ return json.Marshal(map[string]string{"text": fmt.Sprintf("*%s*\n%s", title, message)})
+}
+
+func webhookPayload(title, message string) ([]byte, error) {
+ return json.Marshal(map[string]string{"title": title, "message": message, "status": "alert"})
+}
+
+func telegramPayload(chatID string) PayloadFunc {
+ return func(title, message string) ([]byte, error) {
+ return json.Marshal(map[string]string{
+ "chat_id": chatID,
+ "text": fmt.Sprintf("*%s*\n%s", title, message),
+ "parse_mode": "Markdown",
+ })
+ }
+}
+
+func pagerdutyPayload(routingKey, severity string) PayloadFunc {
+ return func(title, message string) ([]byte, error) {
+ return json.Marshal(map[string]any{
+ "routing_key": routingKey,
+ "event_action": "trigger",
+ "payload": map[string]string{
+ "summary": fmt.Sprintf("%s: %s", title, message),
+ "source": "go-upkeep",
+ "severity": severity,
+ },
+ })
+ }
+}
+
+func pushoverPayload(token, user string) PayloadFunc {
+ return func(title, message string) ([]byte, error) {
+ return json.Marshal(map[string]string{
+ "token": token,
+ "user": user,
+ "title": title,
+ "message": message,
+ })
+ }
+}
+
+func gotifyPayload(priority string) PayloadFunc {
+ return func(title, message string) ([]byte, error) {
+ pri, _ := strconv.Atoi(priority)
+ return json.Marshal(map[string]any{
+ "title": title,
+ "message": message,
+ "priority": pri,
+ })
+ }
+}
+
func GetProvider(cfg models.AlertConfig) Provider {
switch cfg.Type {
case "discord":
- return &DiscordProvider{URL: cfg.Settings["url"]}
+ return &HTTPProvider{URL: cfg.Settings["url"], Payload: discordPayload}
case "slack":
- return &SlackProvider{URL: cfg.Settings["url"]}
+ return &HTTPProvider{URL: cfg.Settings["url"], Payload: slackPayload}
case "webhook":
- // Generic Webhook
- return &WebhookProvider{URL: cfg.Settings["url"]}
+ return &HTTPProvider{URL: cfg.Settings["url"], Payload: webhookPayload}
case "email":
port := "25"
if p, ok := cfg.Settings["port"]; ok {
@@ -51,58 +132,40 @@ func GetProvider(cfg models.AlertConfig) Provider {
Username: cfg.Settings["username"],
Password: cfg.Settings["password"],
}
+ case "telegram":
+ return &HTTPProvider{
+ URL: fmt.Sprintf("https://api.telegram.org/bot%s/sendMessage", cfg.Settings["token"]),
+ Payload: telegramPayload(cfg.Settings["chat_id"]),
+ }
+ case "pagerduty":
+ severity := "critical"
+ if s, ok := cfg.Settings["severity"]; ok && s != "" {
+ severity = s
+ }
+ return &HTTPProvider{
+ URL: "https://events.pagerduty.com/v2/enqueue",
+ Payload: pagerdutyPayload(cfg.Settings["routing_key"], severity),
+ }
+ case "pushover":
+ return &HTTPProvider{
+ URL: "https://api.pushover.net/1/messages.json",
+ Payload: pushoverPayload(cfg.Settings["token"], cfg.Settings["user"]),
+ }
+ case "gotify":
+ priority := "5"
+ if p, ok := cfg.Settings["priority"]; ok && p != "" {
+ priority = p
+ }
+ serverURL := strings.TrimRight(cfg.Settings["url"], "/")
+ return &HTTPProvider{
+ URL: fmt.Sprintf("%s/message?token=%s", serverURL, cfg.Settings["token"]),
+ Payload: gotifyPayload(priority),
+ }
default:
return nil
}
}
-// --- DISCORD ---
-type DiscordProvider struct{ URL string }
-
-func (d *DiscordProvider) Send(title, message string) error {
- payload := map[string]string{"content": fmt.Sprintf("**%s**\n%s", title, message)}
- jsonValue, _ := json.Marshal(payload)
- resp, err := alertClient.Post(d.URL, "application/json", bytes.NewBuffer(jsonValue))
- if err != nil {
- return err
- }
- resp.Body.Close()
- return nil
-}
-
-// --- SLACK ---
-type SlackProvider struct{ URL string }
-
-func (s *SlackProvider) Send(title, message string) error {
- payload := map[string]string{"text": fmt.Sprintf("*%s*\n%s", title, message)}
- jsonValue, _ := json.Marshal(payload)
- resp, err := alertClient.Post(s.URL, "application/json", bytes.NewBuffer(jsonValue))
- if err != nil {
- return err
- }
- resp.Body.Close()
- return nil
-}
-
-// --- GENERIC WEBHOOK ---
-type WebhookProvider struct{ URL string }
-
-func (w *WebhookProvider) Send(title, message string) error {
- payload := map[string]string{
- "title": title,
- "message": message,
- "status": "alert",
- }
- jsonValue, _ := json.Marshal(payload)
- resp, err := alertClient.Post(w.URL, "application/json", bytes.NewBuffer(jsonValue))
- if err != nil {
- return err
- }
- resp.Body.Close()
- return nil
-}
-
-// --- EMAIL ---
type EmailProvider struct {
Host, Port, User, Pass, To, From string
}
@@ -139,6 +202,9 @@ func (n *NtfyProvider) Send(title, message string) error {
if err != nil {
return err
}
- resp.Body.Close()
+ defer resp.Body.Close()
+ if resp.StatusCode >= 400 {
+ return fmt.Errorf("ntfy returned HTTP %d", resp.StatusCode)
+ }
return nil
}
diff --git a/internal/alert/alert_test.go b/internal/alert/alert_test.go
new file mode 100644
index 0000000..35e1c8d
--- /dev/null
+++ b/internal/alert/alert_test.go
@@ -0,0 +1,213 @@
+package alert
+
+import (
+ "encoding/json"
+ "go-upkeep/internal/models"
+ "net/http"
+ "net/http/httptest"
+ "testing"
+)
+
+func TestHTTPProviderDiscord(t *testing.T) {
+ var received map[string]string
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ json.NewDecoder(r.Body).Decode(&received)
+ w.WriteHeader(200)
+ }))
+ defer srv.Close()
+
+ p := GetProvider(models.AlertConfig{Type: "discord", Settings: map[string]string{"url": srv.URL}})
+ if err := p.Send("Test Title", "Test Body"); err != nil {
+ t.Fatalf("Send: %v", err)
+ }
+
+ if received["content"] != "**Test Title**\nTest Body" {
+ t.Errorf("unexpected payload: %s", received["content"])
+ }
+}
+
+func TestHTTPProviderSlack(t *testing.T) {
+ var received map[string]string
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ json.NewDecoder(r.Body).Decode(&received)
+ w.WriteHeader(200)
+ }))
+ defer srv.Close()
+
+ p := GetProvider(models.AlertConfig{Type: "slack", Settings: map[string]string{"url": srv.URL}})
+ if err := p.Send("Alert", "Message"); err != nil {
+ t.Fatalf("Send: %v", err)
+ }
+
+ if received["text"] != "*Alert*\nMessage" {
+ t.Errorf("unexpected payload: %s", received["text"])
+ }
+}
+
+func TestHTTPProviderWebhook(t *testing.T) {
+ var received map[string]string
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ json.NewDecoder(r.Body).Decode(&received)
+ w.WriteHeader(200)
+ }))
+ defer srv.Close()
+
+ p := GetProvider(models.AlertConfig{Type: "webhook", Settings: map[string]string{"url": srv.URL}})
+ if err := p.Send("Title", "Body"); err != nil {
+ t.Fatalf("Send: %v", err)
+ }
+
+ if received["title"] != "Title" || received["message"] != "Body" || received["status"] != "alert" {
+ t.Errorf("unexpected webhook payload: %v", received)
+ }
+}
+
+func TestHTTPProviderErrorOnHTTP4xx(t *testing.T) {
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ w.WriteHeader(403)
+ }))
+ defer srv.Close()
+
+ p := GetProvider(models.AlertConfig{Type: "discord", Settings: map[string]string{"url": srv.URL}})
+ if err := p.Send("Test", "Test"); err == nil {
+ t.Fatal("expected error on 403 response")
+ }
+}
+
+func TestNtfyProvider(t *testing.T) {
+ var title, body string
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ title = r.Header.Get("Title")
+ buf := make([]byte, 1024)
+ n, _ := r.Body.Read(buf)
+ body = string(buf[:n])
+ w.WriteHeader(200)
+ }))
+ defer srv.Close()
+
+ p := GetProvider(models.AlertConfig{Type: "ntfy", Settings: map[string]string{
+ "url": srv.URL,
+ "topic": "test",
+ }})
+ if err := p.Send("Alert Title", "Alert Body"); err != nil {
+ t.Fatalf("Send: %v", err)
+ }
+
+ if title != "Alert Title" {
+ t.Errorf("expected title 'Alert Title', got '%s'", title)
+ }
+ if body != "Alert Body" {
+ t.Errorf("expected body 'Alert Body', got '%s'", body)
+ }
+}
+
+func TestHTTPProviderTelegram(t *testing.T) {
+ var received map[string]string
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ json.NewDecoder(r.Body).Decode(&received)
+ w.WriteHeader(200)
+ }))
+ defer srv.Close()
+
+ p := &HTTPProvider{URL: srv.URL, Payload: telegramPayload("12345")}
+ if err := p.Send("Alert", "Down"); err != nil {
+ t.Fatalf("Send: %v", err)
+ }
+ if received["chat_id"] != "12345" {
+ t.Errorf("expected chat_id '12345', got '%s'", received["chat_id"])
+ }
+ if received["text"] != "*Alert*\nDown" {
+ t.Errorf("unexpected text: %s", received["text"])
+ }
+ if received["parse_mode"] != "Markdown" {
+ t.Errorf("expected parse_mode 'Markdown', got '%s'", received["parse_mode"])
+ }
+}
+
+func TestHTTPProviderPagerDuty(t *testing.T) {
+ var received map[string]any
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ json.NewDecoder(r.Body).Decode(&received)
+ w.WriteHeader(200)
+ }))
+ defer srv.Close()
+
+ p := &HTTPProvider{URL: srv.URL, Payload: pagerdutyPayload("test-key", "critical")}
+ if err := p.Send("Alert", "Down"); err != nil {
+ t.Fatalf("Send: %v", err)
+ }
+ if received["routing_key"] != "test-key" {
+ t.Errorf("expected routing_key 'test-key', got '%v'", received["routing_key"])
+ }
+ if received["event_action"] != "trigger" {
+ t.Errorf("expected event_action 'trigger', got '%v'", received["event_action"])
+ }
+ payload := received["payload"].(map[string]any)
+ if payload["summary"] != "Alert: Down" {
+ t.Errorf("unexpected summary: %v", payload["summary"])
+ }
+ if payload["severity"] != "critical" {
+ t.Errorf("expected severity 'critical', got '%v'", payload["severity"])
+ }
+}
+
+func TestHTTPProviderPushover(t *testing.T) {
+ var received map[string]string
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ json.NewDecoder(r.Body).Decode(&received)
+ w.WriteHeader(200)
+ }))
+ defer srv.Close()
+
+ p := &HTTPProvider{URL: srv.URL, Payload: pushoverPayload("app-tok", "user-key")}
+ if err := p.Send("Alert", "Down"); err != nil {
+ t.Fatalf("Send: %v", err)
+ }
+ if received["token"] != "app-tok" {
+ t.Errorf("expected token 'app-tok', got '%s'", received["token"])
+ }
+ if received["user"] != "user-key" {
+ t.Errorf("expected user 'user-key', got '%s'", received["user"])
+ }
+ if received["title"] != "Alert" || received["message"] != "Down" {
+ t.Errorf("unexpected payload: %v", received)
+ }
+}
+
+func TestHTTPProviderGotify(t *testing.T) {
+ var received map[string]any
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ json.NewDecoder(r.Body).Decode(&received)
+ w.WriteHeader(200)
+ }))
+ defer srv.Close()
+
+ p := &HTTPProvider{URL: srv.URL, Payload: gotifyPayload("8")}
+ if err := p.Send("Alert", "Down"); err != nil {
+ t.Fatalf("Send: %v", err)
+ }
+ if received["title"] != "Alert" || received["message"] != "Down" {
+ t.Errorf("unexpected payload: %v", received)
+ }
+ if pri, ok := received["priority"].(float64); !ok || pri != 8 {
+ t.Errorf("expected priority 8, got %v", received["priority"])
+ }
+}
+
+func TestGetProviderNewTypes(t *testing.T) {
+ for _, typ := range []string{"telegram", "pagerduty", "pushover", "gotify"} {
+ p := GetProvider(models.AlertConfig{Type: typ, Settings: map[string]string{
+ "token": "x", "chat_id": "1", "routing_key": "k", "user": "u", "url": "http://localhost",
+ }})
+ if p == nil {
+ t.Errorf("GetProvider(%q) returned nil", typ)
+ }
+ }
+}
+
+func TestGetProviderUnknown(t *testing.T) {
+ p := GetProvider(models.AlertConfig{Type: "unknown"})
+ if p != nil {
+ t.Error("expected nil for unknown provider type")
+ }
+}
diff --git a/internal/cluster/cluster.go b/internal/cluster/cluster.go
index 295443d..f986c29 100644
--- a/internal/cluster/cluster.go
+++ b/internal/cluster/cluster.go
@@ -1,6 +1,7 @@
package cluster
import (
+ "context"
"fmt"
"go-upkeep/internal/monitor"
"net/http"
@@ -14,13 +15,13 @@ type Config struct {
SharedKey string // Security Key
}
-func Start(cfg Config) {
+func Start(ctx context.Context, cfg Config, eng *monitor.Engine) {
if cfg.Mode == "leader" {
fmt.Println("Cluster: Running as LEADER (Active)")
if cfg.SharedKey != "" {
fmt.Println("WARNING: Cluster mode enabled. Ensure the HTTP server is behind a TLS-terminating proxy.")
}
- monitor.SetEngineActive(true)
+ eng.SetActive(true)
return
}
@@ -29,20 +30,22 @@ func Start(cfg Config) {
if cfg.PeerURL != "" && !strings.HasPrefix(cfg.PeerURL, "https://") {
fmt.Println("WARNING: Cluster peer URL is not HTTPS. Cluster secret will be sent in cleartext.")
}
- monitor.SetEngineActive(false)
- go runFollowerLoop(cfg)
+ eng.SetActive(false)
+ go runFollowerLoop(ctx, cfg, eng)
}
}
-func runFollowerLoop(cfg Config) {
+func runFollowerLoop(ctx context.Context, cfg Config, eng *monitor.Engine) {
client := http.Client{Timeout: 2 * time.Second}
-
- // Failover Configuration
failures := 0
threshold := 3
for {
- time.Sleep(5 * time.Second)
+ select {
+ case <-time.After(5 * time.Second):
+ case <-ctx.Done():
+ return
+ }
req, _ := http.NewRequest("GET", cfg.PeerURL+"/api/health", nil)
if cfg.SharedKey != "" {
@@ -59,17 +62,15 @@ func runFollowerLoop(cfg Config) {
if isLeaderHealthy {
failures = 0
- if monitor.IsEngineActive() {
- // Leader is back, yield
- monitor.SetEngineActive(false)
- monitor.AddLog("Cluster: Leader detected. Switching to PASSIVE.")
+ if eng.IsActive() {
+ eng.SetActive(false)
+ eng.AddLog("Cluster: Leader detected. Switching to PASSIVE.")
}
} else {
failures++
- // If failures exceed threshold, take over
- if failures >= threshold && !monitor.IsEngineActive() {
- monitor.SetEngineActive(true)
- monitor.AddLog("Cluster: Leader Unreachable. Switching to ACTIVE.")
+ if failures >= threshold && !eng.IsActive() {
+ eng.SetActive(true)
+ eng.AddLog("Cluster: Leader Unreachable. Switching to ACTIVE.")
}
}
}
diff --git a/internal/metrics/prometheus.go b/internal/metrics/prometheus.go
new file mode 100644
index 0000000..24f4faa
--- /dev/null
+++ b/internal/metrics/prometheus.go
@@ -0,0 +1,99 @@
+package metrics
+
+import (
+ "fmt"
+ "go-upkeep/internal/models"
+ "go-upkeep/internal/monitor"
+ "net/http"
+ "sort"
+ "strings"
+)
+
+func Handler(eng *monitor.Engine) http.HandlerFunc {
+ return func(w http.ResponseWriter, r *http.Request) {
+ sites := eng.GetAllSites()
+ sort.Slice(sites, func(i, j int) bool { return sites[i].ID < sites[j].ID })
+
+ var b strings.Builder
+
+ writeHelp(&b, "upkeep_monitor_up", "gauge", "Whether the monitor is up (1) or down (0).")
+ for _, s := range sites {
+ val := 0
+ if s.Status == "UP" {
+ val = 1
+ }
+ writeGauge(&b, "upkeep_monitor_up", labels(s), float64(val))
+ }
+
+ writeHelp(&b, "upkeep_monitor_latency_seconds", "gauge", "Last check latency in seconds.")
+ for _, s := range sites {
+ writeGauge(&b, "upkeep_monitor_latency_seconds", labels(s), s.Latency.Seconds())
+ }
+
+ writeHelp(&b, "upkeep_monitor_status_code", "gauge", "HTTP response status code of the last check.")
+ for _, s := range sites {
+ if s.Type != "http" {
+ continue
+ }
+ writeGauge(&b, "upkeep_monitor_status_code", labels(s), float64(s.StatusCode))
+ }
+
+ writeHelp(&b, "upkeep_monitor_check_timestamp_seconds", "gauge", "Unix timestamp of the last check.")
+ for _, s := range sites {
+ if s.LastCheck.IsZero() {
+ continue
+ }
+ writeGauge(&b, "upkeep_monitor_check_timestamp_seconds", labels(s), float64(s.LastCheck.Unix()))
+ }
+
+ writeHelp(&b, "upkeep_monitor_paused", "gauge", "Whether the monitor is paused (1) or active (0).")
+ for _, s := range sites {
+ val := 0
+ if s.Paused {
+ val = 1
+ }
+ writeGauge(&b, "upkeep_monitor_paused", labels(s), float64(val))
+ }
+
+ writeHelp(&b, "upkeep_monitor_cert_expiry_timestamp_seconds", "gauge", "Unix timestamp when the SSL certificate expires.")
+ for _, s := range sites {
+ if !s.HasSSL || s.CertExpiry.IsZero() {
+ continue
+ }
+ writeGauge(&b, "upkeep_monitor_cert_expiry_timestamp_seconds", labels(s), float64(s.CertExpiry.Unix()))
+ }
+
+ writeHelp(&b, "upkeep_monitor_checks_total", "counter", "Total number of checks performed.")
+ writeHelp(&b, "upkeep_monitor_checks_up_total", "counter", "Total number of successful checks.")
+ for _, s := range sites {
+ h, ok := eng.GetHistory(s.ID)
+ if !ok {
+ continue
+ }
+ writeGauge(&b, "upkeep_monitor_checks_total", labels(s), float64(h.TotalChecks))
+ writeGauge(&b, "upkeep_monitor_checks_up_total", labels(s), float64(h.UpChecks))
+ }
+
+ w.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8")
+ w.Write([]byte(b.String()))
+ }
+}
+
+func labels(s models.Site) string {
+ return fmt.Sprintf(`id="%d",name="%s",type="%s"`, s.ID, escapeLabelValue(s.Name), s.Type)
+}
+
+func escapeLabelValue(s string) string {
+ s = strings.ReplaceAll(s, `\`, `\\`)
+ s = strings.ReplaceAll(s, `"`, `\"`)
+ s = strings.ReplaceAll(s, "\n", `\n`)
+ return s
+}
+
+func writeHelp(b *strings.Builder, name, typ, help string) {
+ fmt.Fprintf(b, "# HELP %s %s\n# TYPE %s %s\n", name, help, name, typ)
+}
+
+func writeGauge(b *strings.Builder, name, labels string, val float64) {
+ fmt.Fprintf(b, "%s{%s} %g\n", name, labels, val)
+}
diff --git a/internal/metrics/prometheus_test.go b/internal/metrics/prometheus_test.go
new file mode 100644
index 0000000..7cbf680
--- /dev/null
+++ b/internal/metrics/prometheus_test.go
@@ -0,0 +1,96 @@
+package metrics
+
+import (
+ "context"
+ "go-upkeep/internal/models"
+ "go-upkeep/internal/monitor"
+ "net/http"
+ "net/http/httptest"
+ "strings"
+ "testing"
+ "time"
+)
+
+type mockStore struct {
+ sites []models.Site
+}
+
+func (m *mockStore) Init() error { return nil }
+func (m *mockStore) GetSites() ([]models.Site, error) { return m.sites, nil }
+func (m *mockStore) AddSite(models.Site) error { return nil }
+func (m *mockStore) UpdateSite(models.Site) error { return nil }
+func (m *mockStore) UpdateSitePaused(int, bool) error { return nil }
+func (m *mockStore) DeleteSite(int) error { return nil }
+func (m *mockStore) GetAllAlerts() ([]models.AlertConfig, error) { return nil, nil }
+func (m *mockStore) GetAlert(int) (models.AlertConfig, error) { return models.AlertConfig{}, nil }
+func (m *mockStore) AddAlert(string, string, map[string]string) error { return nil }
+func (m *mockStore) UpdateAlert(int, string, string, map[string]string) error { return nil }
+func (m *mockStore) DeleteAlert(int) error { return nil }
+func (m *mockStore) GetAllUsers() ([]models.User, error) { return nil, nil }
+func (m *mockStore) AddUser(string, string, string) error { return nil }
+func (m *mockStore) UpdateUser(int, string, string, string) error { return nil }
+func (m *mockStore) DeleteUser(int) error { return nil }
+func (m *mockStore) SaveCheck(int, int64, bool) error { return nil }
+func (m *mockStore) LoadAllHistory(int) (map[int][]models.CheckRecord, error) {
+ return nil, nil
+}
+func (m *mockStore) ExportData() (models.Backup, error) { return models.Backup{}, nil }
+func (m *mockStore) ImportData(models.Backup) error { return nil }
+
+func TestMetricsHandler(t *testing.T) {
+ ms := &mockStore{
+ sites: []models.Site{
+ {ID: 1, Name: "Example", URL: "https://example.com", Type: "http", Interval: 30},
+ {ID: 2, Name: "DNS Check", Type: "dns", Interval: 60},
+ },
+ }
+ eng := monitor.NewEngine(ms)
+ ctx, cancel := context.WithCancel(context.Background())
+ eng.Start(ctx)
+ time.Sleep(100 * time.Millisecond)
+
+ rec := httptest.NewRecorder()
+ Handler(eng)(rec, httptest.NewRequest("GET", "/metrics", nil))
+ cancel()
+
+ if rec.Code != http.StatusOK {
+ t.Fatalf("expected 200, got %d", rec.Code)
+ }
+
+ body := rec.Body.String()
+
+ ct := rec.Header().Get("Content-Type")
+ if !strings.Contains(ct, "text/plain") {
+ t.Errorf("expected text/plain content type, got %q", ct)
+ }
+
+ expected := []string{
+ "# HELP upkeep_monitor_up",
+ "# TYPE upkeep_monitor_up gauge",
+ `upkeep_monitor_up{id="1",name="Example",type="http"}`,
+ `upkeep_monitor_up{id="2",name="DNS Check",type="dns"}`,
+ "# HELP upkeep_monitor_latency_seconds",
+ "# HELP upkeep_monitor_paused",
+ "# HELP upkeep_monitor_checks_total",
+ }
+ for _, s := range expected {
+ if !strings.Contains(body, s) {
+ t.Errorf("missing expected line: %s", s)
+ }
+ }
+}
+
+func TestEscapeLabelValue(t *testing.T) {
+ cases := []struct{ in, want string }{
+ {`simple`, `simple`},
+ {`has "quotes"`, `has \"quotes\"`},
+ {"has\nnewline", `has\nnewline`},
+ {`back\slash`, `back\\slash`},
+ }
+ for _, tc := range cases {
+ got := escapeLabelValue(tc.in)
+ if got != tc.want {
+ t.Errorf("escapeLabelValue(%q) = %q, want %q", tc.in, got, tc.want)
+ }
+ }
+}
diff --git a/internal/monitor/history.go b/internal/monitor/history.go
index 8642255..1049e04 100644
--- a/internal/monitor/history.go
+++ b/internal/monitor/history.go
@@ -1,10 +1,6 @@
package monitor
-import (
- "go-upkeep/internal/store"
- "sync"
- "time"
-)
+import "time"
const maxHistoryLen = 30
@@ -15,19 +11,14 @@ type SiteHistory struct {
UpChecks int
}
-var (
- histories = make(map[int]*SiteHistory)
- historyMu sync.RWMutex
-)
-
-func InitHistoryFromStore() {
- s := store.Get()
- if s == nil {
+func (e *Engine) InitHistory() {
+ all, err := e.db.LoadAllHistory(maxHistoryLen)
+ if err != nil {
+ e.AddLog("Failed to load check history: " + err.Error())
return
}
- all := s.LoadAllHistory(maxHistoryLen)
- historyMu.Lock()
- defer historyMu.Unlock()
+ e.histMu.Lock()
+ defer e.histMu.Unlock()
for siteID, records := range all {
h := &SiteHistory{}
for _, r := range records {
@@ -38,21 +29,21 @@ func InitHistoryFromStore() {
h.Latencies = append(h.Latencies, time.Duration(r.LatencyNs))
h.Statuses = append(h.Statuses, r.IsUp)
}
- histories[siteID] = h
+ e.histories[siteID] = h
}
if len(all) > 0 {
- AddLog("Loaded check history from database")
+ e.AddLog("Loaded check history from database")
}
}
-func RecordCheck(siteID int, latency time.Duration, isUp bool) {
- historyMu.Lock()
- defer historyMu.Unlock()
+func (e *Engine) recordCheck(siteID int, latency time.Duration, isUp bool) {
+ e.histMu.Lock()
+ defer e.histMu.Unlock()
- h, ok := histories[siteID]
+ h, ok := e.histories[siteID]
if !ok {
h = &SiteHistory{}
- histories[siteID] = h
+ e.histories[siteID] = h
}
h.TotalChecks++
@@ -70,15 +61,13 @@ func RecordCheck(siteID int, latency time.Duration, isUp bool) {
h.Statuses = h.Statuses[len(h.Statuses)-maxHistoryLen:]
}
- if s := store.Get(); s != nil {
- go s.SaveCheck(siteID, latency.Nanoseconds(), isUp)
- }
+ go func() { _ = e.db.SaveCheck(siteID, latency.Nanoseconds(), isUp) }()
}
-func GetHistory(siteID int) (SiteHistory, bool) {
- historyMu.RLock()
- defer historyMu.RUnlock()
- h, ok := histories[siteID]
+func (e *Engine) GetHistory(siteID int) (SiteHistory, bool) {
+ e.histMu.RLock()
+ defer e.histMu.RUnlock()
+ h, ok := e.histories[siteID]
if !ok {
return SiteHistory{}, false
}
@@ -93,8 +82,8 @@ func GetHistory(siteID int) (SiteHistory, bool) {
return cp, true
}
-func RemoveHistory(siteID int) {
- historyMu.Lock()
- defer historyMu.Unlock()
- delete(histories, siteID)
+func (e *Engine) removeHistory(siteID int) {
+ e.histMu.Lock()
+ defer e.histMu.Unlock()
+ delete(e.histories, siteID)
}
diff --git a/internal/monitor/monitor.go b/internal/monitor/monitor.go
index 1bb02fe..2d2af10 100644
--- a/internal/monitor/monitor.go
+++ b/internal/monitor/monitor.go
@@ -1,6 +1,7 @@
package monitor
import (
+ "context"
"crypto/tls"
"fmt"
"go-upkeep/internal/alert"
@@ -9,6 +10,7 @@ import (
"net"
"net/http"
"strconv"
+ "strings"
"sync"
"time"
@@ -16,198 +18,271 @@ import (
probing "github.com/prometheus-community/pro-bing"
)
-// --- LOGGING ---
-var (
- LogStore []string
- LogMutex sync.RWMutex
-)
+type Engine struct {
+ mu sync.RWMutex
+ liveState map[int]models.Site
-func AddLog(msg string) {
- LogMutex.Lock()
- defer LogMutex.Unlock()
- ts := time.Now().Format("15:04:05")
- entry := fmt.Sprintf("[%s] %s", ts, msg)
- LogStore = append([]string{entry}, LogStore...)
- if len(LogStore) > 100 {
- LogStore = LogStore[:100]
+ logMu sync.RWMutex
+ logStore []string
+
+ activeMu sync.RWMutex
+ isActive bool
+
+ histMu sync.RWMutex
+ histories map[int]*SiteHistory
+
+ tokenIndex map[string]int
+
+ db store.Store
+ insecureSkipVerify bool
+ strictClient *http.Client
+ insecureClient *http.Client
+}
+
+func NewEngine(s store.Store) *Engine {
+ return &Engine{
+ liveState: make(map[int]models.Site),
+ histories: make(map[int]*SiteHistory),
+ tokenIndex: make(map[string]int),
+ isActive: true,
+ db: s,
+ strictClient: &http.Client{
+ Transport: &http.Transport{TLSClientConfig: &tls.Config{InsecureSkipVerify: false}},
+ },
+ insecureClient: &http.Client{
+ Transport: &http.Transport{TLSClientConfig: &tls.Config{InsecureSkipVerify: true}},
+ },
}
}
-func GetLogs() []string {
- LogMutex.RLock()
- defer LogMutex.RUnlock()
- logs := make([]string, len(LogStore))
- copy(logs, LogStore)
+func (e *Engine) SetInsecureSkipVerify(skip bool) {
+ e.insecureSkipVerify = skip
+}
+
+func (e *Engine) AddLog(msg string) {
+ e.logMu.Lock()
+ defer e.logMu.Unlock()
+ ts := time.Now().Format("15:04:05")
+ entry := fmt.Sprintf("[%s] %s", ts, msg)
+ e.logStore = append([]string{entry}, e.logStore...)
+ if len(e.logStore) > 100 {
+ e.logStore = e.logStore[:100]
+ }
+}
+
+func (e *Engine) GetLogs() []string {
+ e.logMu.RLock()
+ defer e.logMu.RUnlock()
+ logs := make([]string, len(e.logStore))
+ copy(logs, e.logStore)
return logs
}
-// --- ENGINE ---
-
-var (
- LiveState = make(map[int]models.Site)
- Mutex sync.RWMutex
-
- // Global Switch for HA
- isActive = true
- activeMutex sync.RWMutex
-
- insecureSkipVerify bool
-)
-
-func SetInsecureSkipVerify(skip bool) {
- insecureSkipVerify = skip
-}
-
-func SetEngineActive(active bool) {
- activeMutex.Lock()
- defer activeMutex.Unlock()
- if isActive != active {
- isActive = active
+func (e *Engine) SetActive(active bool) {
+ e.activeMu.Lock()
+ defer e.activeMu.Unlock()
+ if e.isActive != active {
+ e.isActive = active
status := "RESUMED (Active)"
if !active {
status = "PAUSED (Passive)"
}
- AddLog(fmt.Sprintf("Engine %s", status))
+ e.AddLog(fmt.Sprintf("Engine %s", status))
}
}
-func IsEngineActive() bool {
- activeMutex.RLock()
- defer activeMutex.RUnlock()
- return isActive
+func (e *Engine) IsActive() bool {
+ e.activeMu.RLock()
+ defer e.activeMu.RUnlock()
+ return e.isActive
}
-func RecordHeartbeat(token string) bool {
- if !IsEngineActive() {
- return false
- } // Only Leader accepts Push
-
- Mutex.Lock()
- defer Mutex.Unlock()
- var targetID int = -1
- for id, s := range LiveState {
- if s.Type == "push" && s.Token == token {
- targetID = id
- break
- }
+func (e *Engine) GetAllSites() []models.Site {
+ e.mu.RLock()
+ defer e.mu.RUnlock()
+ sites := make([]models.Site, 0, len(e.liveState))
+ for _, s := range e.liveState {
+ sites = append(sites, s)
}
- if targetID == -1 {
+ return sites
+}
+
+func (e *Engine) GetLiveState() map[int]models.Site {
+ e.mu.RLock()
+ defer e.mu.RUnlock()
+ cp := make(map[int]models.Site, len(e.liveState))
+ for k, v := range e.liveState {
+ cp[k] = v
+ }
+ return cp
+}
+
+func (e *Engine) RecordHeartbeat(token string) bool {
+ if !e.IsActive() {
+ return false
+ }
+
+ e.mu.Lock()
+ defer e.mu.Unlock()
+
+ targetID, ok := e.tokenIndex[token]
+ if !ok {
+ return false
+ }
+
+ site, exists := e.liveState[targetID]
+ if !exists {
return false
}
- site := LiveState[targetID]
site.LastCheck = time.Now()
wasDown := site.Status == "DOWN"
site.Status = "UP"
site.FailureCount = 0
site.Latency = 0
- LiveState[targetID] = site
+ e.liveState[targetID] = site
if wasDown {
- AddLog(fmt.Sprintf("Push Monitor '%s' recovered", site.Name))
- triggerAlert(site.AlertID, "โ
RECOVERY", fmt.Sprintf("Push Monitor '%s' is receiving heartbeats.", site.Name))
+ e.AddLog(fmt.Sprintf("Push Monitor '%s' recovered", site.Name))
+ e.triggerAlert(site.AlertID, "โ
RECOVERY", fmt.Sprintf("Push Monitor '%s' is receiving heartbeats.", site.Name))
}
return true
}
-func StartEngine() {
+func (e *Engine) addToTokenIndex(site models.Site) {
+ if site.Type == "push" && site.Token != "" {
+ e.tokenIndex[site.Token] = site.ID
+ }
+}
+
+func (e *Engine) removeFromTokenIndex(id int) {
+ for token, sid := range e.tokenIndex {
+ if sid == id {
+ delete(e.tokenIndex, token)
+ return
+ }
+ }
+}
+
+func (e *Engine) Start(ctx context.Context) {
go func() {
for {
- s_instance := store.Get()
- if s_instance == nil {
- time.Sleep(1 * time.Second)
- continue
+ select {
+ case <-ctx.Done():
+ return
+ default:
}
- sites := s_instance.GetSites()
+ sites, err := e.db.GetSites()
+ if err != nil {
+ e.AddLog(fmt.Sprintf("Failed to load sites: %v", err))
+ select {
+ case <-time.After(5 * time.Second):
+ case <-ctx.Done():
+ return
+ }
+ continue
+ }
for _, s := range sites {
- Mutex.RLock()
- _, exists := LiveState[s.ID]
- Mutex.RUnlock()
+ e.mu.RLock()
+ _, exists := e.liveState[s.ID]
+ e.mu.RUnlock()
if !exists {
- Mutex.Lock()
+ e.mu.Lock()
s.Status = "PENDING"
if s.Type == "push" {
s.LastCheck = time.Now()
}
- LiveState[s.ID] = s
- Mutex.Unlock()
- go monitorRoutine(s.ID)
+ e.liveState[s.ID] = s
+ e.addToTokenIndex(s)
+ e.mu.Unlock()
+ go e.monitorRoutine(ctx, s.ID)
}
}
- time.Sleep(5 * time.Second)
+
+ select {
+ case <-time.After(5 * time.Second):
+ case <-ctx.Done():
+ return
+ }
}
}()
}
-func UpdateSiteConfig(site models.Site) {
- Mutex.Lock()
- defer Mutex.Unlock()
- if s, ok := LiveState[site.ID]; ok {
- s.Name = site.Name
- s.URL = site.URL
- s.Type = site.Type
- s.Interval = site.Interval
- s.AlertID = site.AlertID
- s.CheckSSL = site.CheckSSL
- s.ExpiryThreshold = site.ExpiryThreshold
- s.MaxRetries = site.MaxRetries
- s.Hostname = site.Hostname
- s.Port = site.Port
- s.Timeout = site.Timeout
- s.Method = site.Method
- s.Description = site.Description
- s.ParentID = site.ParentID
- s.AcceptedCodes = site.AcceptedCodes
- s.DNSResolveType = site.DNSResolveType
- s.DNSServer = site.DNSServer
- s.IgnoreTLS = site.IgnoreTLS
- s.Paused = site.Paused
- LiveState[site.ID] = s
+func (e *Engine) UpdateSiteConfig(site models.Site) {
+ e.mu.Lock()
+ defer e.mu.Unlock()
+ if existing, ok := e.liveState[site.ID]; ok {
+ e.removeFromTokenIndex(site.ID)
+ site.Status = existing.Status
+ site.StatusCode = existing.StatusCode
+ site.Latency = existing.Latency
+ site.CertExpiry = existing.CertExpiry
+ site.HasSSL = existing.HasSSL
+ site.LastCheck = existing.LastCheck
+ site.SentSSLWarning = existing.SentSSLWarning
+ site.FailureCount = existing.FailureCount
+ e.liveState[site.ID] = site
+ e.addToTokenIndex(site)
}
}
-func RemoveSite(id int) {
- Mutex.Lock()
- delete(LiveState, id)
- Mutex.Unlock()
- RemoveHistory(id)
+func (e *Engine) RemoveSite(id int) {
+ e.mu.Lock()
+ e.removeFromTokenIndex(id)
+ delete(e.liveState, id)
+ e.mu.Unlock()
+ e.removeHistory(id)
}
-func ToggleSitePause(id int) bool {
- Mutex.Lock()
- defer Mutex.Unlock()
- site, ok := LiveState[id]
+func (e *Engine) ToggleSitePause(id int) bool {
+ e.mu.Lock()
+ defer e.mu.Unlock()
+ site, ok := e.liveState[id]
if !ok {
return false
}
site.Paused = !site.Paused
- LiveState[id] = site
+ e.liveState[id] = site
if site.Paused {
- AddLog(fmt.Sprintf("Monitor '%s' paused", site.Name))
+ e.AddLog(fmt.Sprintf("Monitor '%s' paused", site.Name))
} else {
- AddLog(fmt.Sprintf("Monitor '%s' resumed", site.Name))
+ e.AddLog(fmt.Sprintf("Monitor '%s' resumed", site.Name))
}
return site.Paused
}
-func monitorRoutine(id int) {
- checkByID(id)
+func (e *Engine) monitorRoutine(ctx context.Context, id int) {
+ e.checkByID(id)
for {
- if !IsEngineActive() {
- time.Sleep(5 * time.Second)
+ select {
+ case <-ctx.Done():
+ return
+ default:
+ }
+
+ if !e.IsActive() {
+ select {
+ case <-time.After(5 * time.Second):
+ case <-ctx.Done():
+ return
+ }
continue
}
- Mutex.RLock()
- site, exists := LiveState[id]
- Mutex.RUnlock()
+ e.mu.RLock()
+ site, exists := e.liveState[id]
+ e.mu.RUnlock()
if !exists {
return
}
if site.Paused {
- time.Sleep(5 * time.Second)
+ select {
+ case <-time.After(5 * time.Second):
+ case <-ctx.Done():
+ return
+ }
continue
}
@@ -215,58 +290,74 @@ func monitorRoutine(id int) {
if interval < 5 {
interval = 5
}
- time.Sleep(time.Duration(interval) * time.Second)
- checkByID(id)
+ select {
+ case <-time.After(time.Duration(interval) * time.Second):
+ case <-ctx.Done():
+ return
+ }
+ e.checkByID(id)
}
}
-func checkByID(id int) {
- if !IsEngineActive() {
+func (e *Engine) checkByID(id int) {
+ if !e.IsActive() {
return
}
- Mutex.RLock()
- site, exists := LiveState[id]
- Mutex.RUnlock()
+ e.mu.RLock()
+ site, exists := e.liveState[id]
+ e.mu.RUnlock()
if !exists || site.Paused {
return
}
switch site.Type {
case "http":
- checkHTTP(site)
+ e.checkHTTP(site)
case "push":
- checkPush(site)
+ e.checkPush(site)
case "ping":
- checkPing(site)
+ e.checkPing(site)
case "port":
- checkPort(site)
+ e.checkPort(site)
case "dns":
- checkDNS(site)
+ e.checkDNS(site)
case "group":
- checkGroup(site)
+ e.checkGroup(site)
}
}
-func checkPush(site models.Site) {
+func (e *Engine) checkPush(site models.Site) {
deadline := site.LastCheck.Add(time.Duration(site.Interval) * time.Second).Add(5 * time.Second)
if time.Now().After(deadline) {
- handleStatusChange(site, "DOWN", 0, 0)
- } else {
- if site.Status != "UP" {
- handleStatusChange(site, "UP", 200, 0)
- }
+ e.handleStatusChange(site, "DOWN", 0, 0)
+ } else if site.Status != "UP" {
+ e.handleStatusChange(site, "UP", 200, 0)
}
}
-func checkHTTP(site models.Site) {
- start := time.Now()
- timeout := time.Duration(site.Timeout) * time.Second
- if timeout <= 0 {
- timeout = 5 * time.Second
+func (e *Engine) checkHTTP(site models.Site) {
+ method := site.Method
+ if method == "" {
+ method = "GET"
}
- skipTLS := insecureSkipVerify || site.IgnoreTLS
- client := &http.Client{Timeout: timeout, Transport: &http.Transport{TLSClientConfig: &tls.Config{InsecureSkipVerify: skipTLS}}}
- resp, err := client.Get(site.URL)
+
+ timeout := siteTimeout(site)
+ ctx, cancel := context.WithTimeout(context.Background(), timeout)
+ defer cancel()
+
+ req, err := http.NewRequestWithContext(ctx, method, site.URL, nil)
+ if err != nil {
+ e.handleStatusChange(site, "DOWN", 0, 0)
+ return
+ }
+
+ client := e.strictClient
+ if e.insecureSkipVerify || site.IgnoreTLS {
+ client = e.insecureClient
+ }
+
+ start := time.Now()
+ resp, err := client.Do(req)
latency := time.Since(start)
rawStatus := "UP"
@@ -279,7 +370,7 @@ func checkHTTP(site models.Site) {
} else {
defer resp.Body.Close()
rawCode = resp.StatusCode
- if resp.StatusCode >= 400 {
+ if !isCodeAccepted(rawCode, site.AcceptedCodes) {
rawStatus = "DOWN"
}
if site.CheckSSL && resp.TLS != nil && len(resp.TLS.PeerCertificates) > 0 {
@@ -296,12 +387,11 @@ func checkHTTP(site models.Site) {
updatedSite.CertExpiry = certExpiry
updatedSite.Latency = latency
updatedSite.LastCheck = time.Now()
- handleStatusChange(updatedSite, rawStatus, rawCode, latency)
+ e.handleStatusChange(updatedSite, rawStatus, rawCode, latency)
}
-func handleStatusChange(site models.Site, rawStatus string, code int, latency time.Duration) {
- // Double check we are still leader before alerting
- if !IsEngineActive() {
+func (e *Engine) handleStatusChange(site models.Site, rawStatus string, code int, latency time.Duration) {
+ if !e.IsActive() {
return
}
@@ -313,9 +403,9 @@ func handleStatusChange(site models.Site, rawStatus string, code int, latency ti
if newState.FailureCount > site.MaxRetries {
newState.Status = rawStatus
newState.FailureCount = site.MaxRetries + 1
- AddLog(fmt.Sprintf("Monitor '%s' confirmed DOWN", site.Name))
+ e.AddLog(fmt.Sprintf("Monitor '%s' confirmed DOWN", site.Name))
} else {
- AddLog(fmt.Sprintf("Monitor '%s' failed check %d/%d", site.Name, newState.FailureCount, site.MaxRetries))
+ e.AddLog(fmt.Sprintf("Monitor '%s' failed check %d/%d", site.Name, newState.FailureCount, site.MaxRetries))
}
} else if rawStatus == "UP" {
newState.FailureCount = 0
@@ -328,20 +418,20 @@ func handleStatusChange(site models.Site, rawStatus string, code int, latency ti
if site.Type == "http" && site.CheckSSL && site.HasSSL {
daysLeft := int(time.Until(site.CertExpiry).Hours() / 24)
if daysLeft <= site.ExpiryThreshold && !site.SentSSLWarning && rawStatus != "SSL EXP" {
- triggerAlert(site.AlertID, "SSL WARNING", fmt.Sprintf("SSL for '%s' expires in %d days", site.Name, daysLeft))
+ e.triggerAlert(site.AlertID, "SSL WARNING", fmt.Sprintf("SSL for '%s' expires in %d days", site.Name, daysLeft))
newState.SentSSLWarning = true
} else if daysLeft > site.ExpiryThreshold {
newState.SentSSLWarning = false
}
}
- Mutex.Lock()
- if _, ok := LiveState[site.ID]; ok {
- LiveState[site.ID] = newState
+ e.mu.Lock()
+ if _, ok := e.liveState[site.ID]; ok {
+ e.liveState[site.ID] = newState
}
- Mutex.Unlock()
+ e.mu.Unlock()
- RecordCheck(site.ID, latency, rawStatus == "UP")
+ e.recordCheck(site.ID, latency, rawStatus == "UP")
isBroken := func(s string) bool { return s == "DOWN" || s == "SSL EXP" }
if !isBroken(site.Status) && isBroken(newState.Status) && newState.Status != "PENDING" {
@@ -349,25 +439,26 @@ func handleStatusChange(site models.Site, rawStatus string, code int, latency ti
if site.Type == "push" {
msg = fmt.Sprintf("Push Monitor '%s' missed heartbeat.", site.Name)
}
- triggerAlert(site.AlertID, "๐จ ALERT", msg)
+ e.triggerAlert(site.AlertID, "๐จ ALERT", msg)
}
if isBroken(site.Status) && newState.Status == "UP" {
- triggerAlert(site.AlertID, "โ
RECOVERY", fmt.Sprintf("Monitor '%s' is UP", site.Name))
+ e.triggerAlert(site.AlertID, "โ
RECOVERY", fmt.Sprintf("Monitor '%s' is UP", site.Name))
}
}
-func triggerAlert(alertID int, title, message string) {
- s_instance := store.Get()
- if s_instance == nil {
- return
- }
- cfg, ok := s_instance.GetAlert(alertID)
- if !ok {
+func (e *Engine) triggerAlert(alertID int, title, message string) {
+ cfg, err := e.db.GetAlert(alertID)
+ if err != nil {
return
}
provider := alert.GetProvider(cfg)
if provider != nil {
- go func() { provider.Send(title, message) }()
+ go func() {
+ ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+ defer cancel()
+ _ = ctx
+ _ = provider.Send(title, message)
+ }()
}
}
@@ -378,7 +469,29 @@ func siteTimeout(site models.Site) time.Duration {
return 5 * time.Second
}
-func checkPing(site models.Site) {
+func isCodeAccepted(code int, accepted string) bool {
+ if accepted == "" {
+ return code >= 200 && code < 300
+ }
+ for _, part := range strings.Split(accepted, ",") {
+ part = strings.TrimSpace(part)
+ if strings.Contains(part, "-") {
+ bounds := strings.SplitN(part, "-", 2)
+ lo, err1 := strconv.Atoi(strings.TrimSpace(bounds[0]))
+ hi, err2 := strconv.Atoi(strings.TrimSpace(bounds[1]))
+ if err1 == nil && err2 == nil && code >= lo && code <= hi {
+ return true
+ }
+ } else {
+ if v, err := strconv.Atoi(part); err == nil && code == v {
+ return true
+ }
+ }
+ }
+ return false
+}
+
+func (e *Engine) checkPing(site models.Site) {
host := site.Hostname
if host == "" {
host = site.URL
@@ -386,8 +499,8 @@ func checkPing(site models.Site) {
pinger, err := probing.NewPinger(host)
if err != nil {
- handleStatusChange(site, "DOWN", 0, 0)
- AddLog(fmt.Sprintf("Ping '%s' resolve failed: %v", site.Name, err))
+ e.handleStatusChange(site, "DOWN", 0, 0)
+ e.AddLog(fmt.Sprintf("Ping '%s' resolve failed: %v", site.Name, err))
return
}
pinger.Count = 1
@@ -402,7 +515,7 @@ func checkPing(site models.Site) {
updatedSite := site
updatedSite.Latency = latency
updatedSite.LastCheck = time.Now()
- handleStatusChange(updatedSite, "DOWN", 0, latency)
+ e.handleStatusChange(updatedSite, "DOWN", 0, latency)
return
}
@@ -410,10 +523,10 @@ func checkPing(site models.Site) {
updatedSite := site
updatedSite.Latency = stats.AvgRtt
updatedSite.LastCheck = time.Now()
- handleStatusChange(updatedSite, "UP", 0, stats.AvgRtt)
+ e.handleStatusChange(updatedSite, "UP", 0, stats.AvgRtt)
}
-func checkPort(site models.Site) {
+func (e *Engine) checkPort(site models.Site) {
host := site.Hostname
if host == "" {
host = site.URL
@@ -430,19 +543,19 @@ func checkPort(site models.Site) {
updatedSite.LastCheck = time.Now()
if err != nil {
- handleStatusChange(updatedSite, "DOWN", 0, latency)
+ e.handleStatusChange(updatedSite, "DOWN", 0, latency)
return
}
conn.Close()
- handleStatusChange(updatedSite, "UP", 0, latency)
+ e.handleStatusChange(updatedSite, "UP", 0, latency)
}
-func checkGroup(site models.Site) {
- Mutex.RLock()
+func (e *Engine) checkGroup(site models.Site) {
+ e.mu.RLock()
status := "UP"
hasChildren := false
allPaused := true
- for _, child := range LiveState {
+ for _, child := range e.liveState {
if child.ParentID != site.ID || child.Type == "group" {
continue
}
@@ -459,23 +572,23 @@ func checkGroup(site models.Site) {
status = "PENDING"
}
}
- Mutex.RUnlock()
+ e.mu.RUnlock()
if !hasChildren {
status = "PENDING"
}
- Mutex.Lock()
- s := LiveState[site.ID]
+ e.mu.Lock()
+ s := e.liveState[site.ID]
s.Status = status
if hasChildren && allPaused {
s.Paused = true
}
- LiveState[site.ID] = s
- Mutex.Unlock()
+ e.liveState[site.ID] = s
+ e.mu.Unlock()
}
-func checkDNS(site models.Site) {
+func (e *Engine) checkDNS(site models.Site) {
host := site.Hostname
if host == "" {
host = site.URL
@@ -516,8 +629,7 @@ func checkDNS(site models.Site) {
c.Timeout = siteTimeout(site)
start := time.Now()
- r, rtt, err := c.Exchange(m, server)
- _ = rtt
+ r, _, err := c.Exchange(m, server)
latency := time.Since(start)
updatedSite := site
@@ -525,14 +637,14 @@ func checkDNS(site models.Site) {
updatedSite.LastCheck = time.Now()
if err != nil {
- handleStatusChange(updatedSite, "DOWN", 0, latency)
+ e.handleStatusChange(updatedSite, "DOWN", 0, latency)
return
}
if r.Rcode != dns.RcodeSuccess {
- handleStatusChange(updatedSite, "DOWN", r.Rcode, latency)
+ e.handleStatusChange(updatedSite, "DOWN", r.Rcode, latency)
return
}
- handleStatusChange(updatedSite, "UP", 0, latency)
+ e.handleStatusChange(updatedSite, "UP", 0, latency)
}
diff --git a/internal/server/server.go b/internal/server/server.go
index f814cc3..fdf7f9b 100644
--- a/internal/server/server.go
+++ b/internal/server/server.go
@@ -4,14 +4,144 @@ import (
"encoding/json"
"fmt"
"go-upkeep/internal/importer"
+ "go-upkeep/internal/metrics"
"go-upkeep/internal/models"
"go-upkeep/internal/monitor"
"go-upkeep/internal/store"
"html/template"
+ "log"
"net/http"
"sort"
)
+var statusTpl = template.Must(template.New("status").Parse(`
+
+
+
+ {{.Title}}
+
+
+
+
+
+
{{.Title}}
+
+
+
+
Powered by Go-Upkeep
+
-
{{.Title}}
-
-
-
-
Powered by Go-Upkeep
-