refactor: quality refactor across all subsystems #3
@@ -39,3 +39,4 @@ tmp
|
||||
/go-upkeep/
|
||||
|
||||
*.local.json
|
||||
*.local.md
|
||||
@@ -161,7 +161,11 @@ 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) {
|
||||
|
||||
+16
-7
@@ -61,12 +61,15 @@ 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)
|
||||
jsonValue, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
resp, err := alertClient.Post(d.URL, "application/json", bytes.NewBuffer(jsonValue))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
resp.Body.Close()
|
||||
defer resp.Body.Close()
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -75,12 +78,15 @@ 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)
|
||||
jsonValue, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
resp, err := alertClient.Post(s.URL, "application/json", bytes.NewBuffer(jsonValue))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
resp.Body.Close()
|
||||
defer resp.Body.Close()
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -93,12 +99,15 @@ func (w *WebhookProvider) Send(title, message string) error {
|
||||
"message": message,
|
||||
"status": "alert",
|
||||
}
|
||||
jsonValue, _ := json.Marshal(payload)
|
||||
jsonValue, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
resp, err := alertClient.Post(w.URL, "application/json", bytes.NewBuffer(jsonValue))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
resp.Body.Close()
|
||||
defer resp.Body.Close()
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -139,6 +148,6 @@ func (n *NtfyProvider) Send(title, message string) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
resp.Body.Close()
|
||||
defer resp.Body.Close()
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -52,6 +54,13 @@ var (
|
||||
activeMutex sync.RWMutex
|
||||
|
||||
insecureSkipVerify bool
|
||||
|
||||
strictClient = &http.Client{
|
||||
Transport: &http.Transport{TLSClientConfig: &tls.Config{InsecureSkipVerify: false}},
|
||||
}
|
||||
insecureClient = &http.Client{
|
||||
Transport: &http.Transport{TLSClientConfig: &tls.Config{InsecureSkipVerify: true}},
|
||||
}
|
||||
)
|
||||
|
||||
func SetInsecureSkipVerify(skip bool) {
|
||||
@@ -258,15 +267,51 @@ func checkPush(site models.Site) {
|
||||
}
|
||||
}
|
||||
|
||||
func checkHTTP(site models.Site) {
|
||||
start := time.Now()
|
||||
timeout := time.Duration(site.Timeout) * time.Second
|
||||
if timeout <= 0 {
|
||||
timeout = 5 * time.Second
|
||||
func isCodeAccepted(code int, accepted string) bool {
|
||||
if accepted == "" {
|
||||
return code >= 200 && code < 300
|
||||
}
|
||||
skipTLS := insecureSkipVerify || site.IgnoreTLS
|
||||
client := &http.Client{Timeout: timeout, Transport: &http.Transport{TLSClientConfig: &tls.Config{InsecureSkipVerify: skipTLS}}}
|
||||
resp, err := client.Get(site.URL)
|
||||
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 checkHTTP(site models.Site) {
|
||||
method := site.Method
|
||||
if method == "" {
|
||||
method = "GET"
|
||||
}
|
||||
|
||||
timeout := siteTimeout(site)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), timeout)
|
||||
defer cancel()
|
||||
|
||||
req, err := http.NewRequestWithContext(ctx, method, site.URL, nil)
|
||||
if err != nil {
|
||||
handleStatusChange(site, "DOWN", 0, 0)
|
||||
return
|
||||
}
|
||||
|
||||
client := strictClient
|
||||
if insecureSkipVerify || site.IgnoreTLS {
|
||||
client = insecureClient
|
||||
}
|
||||
|
||||
start := time.Now()
|
||||
resp, err := client.Do(req)
|
||||
latency := time.Since(start)
|
||||
|
||||
rawStatus := "UP"
|
||||
@@ -279,7 +324,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 {
|
||||
|
||||
+140
-135
@@ -8,142 +8,12 @@ import (
|
||||
"go-upkeep/internal/monitor"
|
||||
"go-upkeep/internal/store"
|
||||
"html/template"
|
||||
"log"
|
||||
"net/http"
|
||||
"sort"
|
||||
)
|
||||
|
||||
type ServerConfig struct {
|
||||
Port int
|
||||
EnableStatus bool
|
||||
Title string
|
||||
ClusterKey string // Shared Secret for Security
|
||||
}
|
||||
|
||||
func Start(cfg ServerConfig) {
|
||||
if cfg.ClusterKey == "" {
|
||||
fmt.Println("WARNING: No UPKEEP_CLUSTER_SECRET set. Cluster API endpoints are unauthenticated.")
|
||||
}
|
||||
mux := http.NewServeMux()
|
||||
|
||||
// 1. Push Heartbeat
|
||||
mux.HandleFunc("/api/push", func(w http.ResponseWriter, r *http.Request) {
|
||||
token := r.URL.Query().Get("token")
|
||||
if token == "" {
|
||||
http.Error(w, "Missing token", 400)
|
||||
return
|
||||
}
|
||||
if monitor.RecordHeartbeat(token) {
|
||||
w.WriteHeader(http.StatusOK)
|
||||
w.Write([]byte("OK"))
|
||||
} else {
|
||||
http.Error(w, "Invalid Token", 404)
|
||||
}
|
||||
})
|
||||
|
||||
// 2. Health Check (For Cluster Follower)
|
||||
mux.HandleFunc("/api/health", func(w http.ResponseWriter, r *http.Request) {
|
||||
if cfg.ClusterKey != "" && r.Header.Get("X-Upkeep-Secret") != cfg.ClusterKey {
|
||||
http.Error(w, "Unauthorized", 401)
|
||||
return
|
||||
}
|
||||
w.WriteHeader(http.StatusOK)
|
||||
w.Write([]byte("OK"))
|
||||
})
|
||||
|
||||
// 3. Config Export
|
||||
mux.HandleFunc("/api/backup/export", func(w http.ResponseWriter, r *http.Request) {
|
||||
if cfg.ClusterKey == "" || r.Header.Get("X-Upkeep-Secret") != cfg.ClusterKey {
|
||||
http.Error(w, "Unauthorized: UPKEEP_CLUSTER_SECRET required", 401)
|
||||
return
|
||||
}
|
||||
data := store.Get().ExportData()
|
||||
json.NewEncoder(w).Encode(data)
|
||||
})
|
||||
|
||||
// 4. Config Import
|
||||
mux.HandleFunc("/api/backup/import", func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != "POST" {
|
||||
http.Error(w, "POST required", 405)
|
||||
return
|
||||
}
|
||||
if cfg.ClusterKey == "" || r.Header.Get("X-Upkeep-Secret") != cfg.ClusterKey {
|
||||
http.Error(w, "Unauthorized", 401)
|
||||
return
|
||||
}
|
||||
var data models.Backup
|
||||
if err := json.NewDecoder(r.Body).Decode(&data); err != nil {
|
||||
http.Error(w, "Invalid JSON", 400)
|
||||
return
|
||||
}
|
||||
if err := store.Get().ImportData(data); err != nil {
|
||||
http.Error(w, "Import Failed: "+err.Error(), 500)
|
||||
return
|
||||
}
|
||||
w.Write([]byte("Import Successful"))
|
||||
})
|
||||
|
||||
// 5. Kuma Import
|
||||
mux.HandleFunc("/api/import/kuma", func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != "POST" {
|
||||
http.Error(w, "POST required", 405)
|
||||
return
|
||||
}
|
||||
if cfg.ClusterKey == "" || r.Header.Get("X-Upkeep-Secret") != cfg.ClusterKey {
|
||||
http.Error(w, "Unauthorized", 401)
|
||||
return
|
||||
}
|
||||
var kb importer.KumaBackup
|
||||
if err := json.NewDecoder(r.Body).Decode(&kb); err != nil {
|
||||
http.Error(w, "Invalid Kuma JSON: "+err.Error(), 400)
|
||||
return
|
||||
}
|
||||
backup := importer.ConvertKuma(&kb)
|
||||
if err := store.Get().ImportData(backup); err != nil {
|
||||
http.Error(w, "Import Failed: "+err.Error(), 500)
|
||||
return
|
||||
}
|
||||
w.Write([]byte(fmt.Sprintf("Imported %d monitors, %d alerts from Kuma v%s", len(backup.Sites), len(backup.Alerts), kb.Version)))
|
||||
})
|
||||
|
||||
// 6. Status Page
|
||||
if cfg.EnableStatus {
|
||||
mux.HandleFunc("/status", func(w http.ResponseWriter, r *http.Request) { renderStatusPage(w, cfg.Title) })
|
||||
mux.HandleFunc("/status/json", func(w http.ResponseWriter, r *http.Request) {
|
||||
monitor.Mutex.RLock()
|
||||
defer monitor.Mutex.RUnlock()
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(monitor.LiveState)
|
||||
})
|
||||
}
|
||||
|
||||
go func() {
|
||||
addr := fmt.Sprintf(":%d", cfg.Port)
|
||||
fmt.Printf("HTTP Server listening on %s\n", addr)
|
||||
http.ListenAndServe(addr, mux)
|
||||
}()
|
||||
}
|
||||
|
||||
func renderStatusPage(w http.ResponseWriter, title string) {
|
||||
monitor.Mutex.RLock()
|
||||
var sites []models.Site
|
||||
for _, s := range monitor.LiveState {
|
||||
sites = append(sites, s)
|
||||
}
|
||||
monitor.Mutex.RUnlock()
|
||||
|
||||
sort.Slice(sites, func(i, j int) bool {
|
||||
if sites[i].Status != sites[j].Status {
|
||||
if sites[i].Status == "DOWN" {
|
||||
return true
|
||||
}
|
||||
if sites[j].Status == "DOWN" {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return sites[i].Name < sites[j].Name
|
||||
})
|
||||
|
||||
const tpl = `
|
||||
var statusTpl = template.Must(template.New("status").Parse(`
|
||||
<!DOCTYPE html>
|
||||
<html>
|
||||
<head>
|
||||
@@ -269,12 +139,147 @@ func renderStatusPage(w http.ResponseWriter, title string) {
|
||||
refresh();
|
||||
</script>
|
||||
</body>
|
||||
</html>`
|
||||
</html>`))
|
||||
|
||||
type ServerConfig struct {
|
||||
Port int
|
||||
EnableStatus bool
|
||||
Title string
|
||||
ClusterKey string // Shared Secret for Security
|
||||
}
|
||||
|
||||
func Start(cfg ServerConfig) {
|
||||
if cfg.ClusterKey == "" {
|
||||
fmt.Println("WARNING: No UPKEEP_CLUSTER_SECRET set. Cluster API endpoints are unauthenticated.")
|
||||
}
|
||||
mux := http.NewServeMux()
|
||||
|
||||
// 1. Push Heartbeat
|
||||
mux.HandleFunc("/api/push", func(w http.ResponseWriter, r *http.Request) {
|
||||
token := r.URL.Query().Get("token")
|
||||
if token == "" {
|
||||
http.Error(w, "Missing token", 400)
|
||||
return
|
||||
}
|
||||
if monitor.RecordHeartbeat(token) {
|
||||
w.WriteHeader(http.StatusOK)
|
||||
w.Write([]byte("OK"))
|
||||
} else {
|
||||
http.Error(w, "Invalid Token", 404)
|
||||
}
|
||||
})
|
||||
|
||||
// 2. Health Check (For Cluster Follower)
|
||||
mux.HandleFunc("/api/health", func(w http.ResponseWriter, r *http.Request) {
|
||||
if cfg.ClusterKey != "" && r.Header.Get("X-Upkeep-Secret") != cfg.ClusterKey {
|
||||
http.Error(w, "Unauthorized", 401)
|
||||
return
|
||||
}
|
||||
w.WriteHeader(http.StatusOK)
|
||||
w.Write([]byte("OK"))
|
||||
})
|
||||
|
||||
// 3. Config Export
|
||||
mux.HandleFunc("/api/backup/export", func(w http.ResponseWriter, r *http.Request) {
|
||||
if cfg.ClusterKey == "" || r.Header.Get("X-Upkeep-Secret") != cfg.ClusterKey {
|
||||
http.Error(w, "Unauthorized: UPKEEP_CLUSTER_SECRET required", 401)
|
||||
return
|
||||
}
|
||||
data := store.Get().ExportData()
|
||||
json.NewEncoder(w).Encode(data)
|
||||
})
|
||||
|
||||
// 4. Config Import
|
||||
mux.HandleFunc("/api/backup/import", func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != "POST" {
|
||||
http.Error(w, "POST required", 405)
|
||||
return
|
||||
}
|
||||
if cfg.ClusterKey == "" || r.Header.Get("X-Upkeep-Secret") != cfg.ClusterKey {
|
||||
http.Error(w, "Unauthorized", 401)
|
||||
return
|
||||
}
|
||||
var data models.Backup
|
||||
if err := json.NewDecoder(r.Body).Decode(&data); err != nil {
|
||||
http.Error(w, "Invalid JSON", 400)
|
||||
return
|
||||
}
|
||||
if err := store.Get().ImportData(data); err != nil {
|
||||
log.Printf("Import failed: %v", err)
|
||||
http.Error(w, "Import failed", 500)
|
||||
return
|
||||
}
|
||||
w.Write([]byte("Import Successful"))
|
||||
})
|
||||
|
||||
// 5. Kuma Import
|
||||
mux.HandleFunc("/api/import/kuma", func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != "POST" {
|
||||
http.Error(w, "POST required", 405)
|
||||
return
|
||||
}
|
||||
if cfg.ClusterKey == "" || r.Header.Get("X-Upkeep-Secret") != cfg.ClusterKey {
|
||||
http.Error(w, "Unauthorized", 401)
|
||||
return
|
||||
}
|
||||
var kb importer.KumaBackup
|
||||
if err := json.NewDecoder(r.Body).Decode(&kb); err != nil {
|
||||
log.Printf("Invalid Kuma JSON: %v", err)
|
||||
http.Error(w, "Invalid Kuma JSON", 400)
|
||||
return
|
||||
}
|
||||
backup := importer.ConvertKuma(&kb)
|
||||
if err := store.Get().ImportData(backup); err != nil {
|
||||
log.Printf("Kuma import failed: %v", err)
|
||||
http.Error(w, "Import failed", 500)
|
||||
return
|
||||
}
|
||||
w.Write([]byte(fmt.Sprintf("Imported %d monitors, %d alerts from Kuma v%s", len(backup.Sites), len(backup.Alerts), kb.Version)))
|
||||
})
|
||||
|
||||
// 6. Status Page
|
||||
if cfg.EnableStatus {
|
||||
mux.HandleFunc("/status", func(w http.ResponseWriter, r *http.Request) { renderStatusPage(w, cfg.Title) })
|
||||
mux.HandleFunc("/status/json", func(w http.ResponseWriter, r *http.Request) {
|
||||
monitor.Mutex.RLock()
|
||||
defer monitor.Mutex.RUnlock()
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(monitor.LiveState)
|
||||
})
|
||||
}
|
||||
|
||||
go func() {
|
||||
addr := fmt.Sprintf(":%d", cfg.Port)
|
||||
fmt.Printf("HTTP Server listening on %s\n", addr)
|
||||
if err := http.ListenAndServe(addr, mux); err != nil {
|
||||
log.Fatalf("HTTP server failed: %v", err)
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
func renderStatusPage(w http.ResponseWriter, title string) {
|
||||
monitor.Mutex.RLock()
|
||||
var sites []models.Site
|
||||
for _, s := range monitor.LiveState {
|
||||
sites = append(sites, s)
|
||||
}
|
||||
monitor.Mutex.RUnlock()
|
||||
|
||||
sort.Slice(sites, func(i, j int) bool {
|
||||
if sites[i].Status != sites[j].Status {
|
||||
if sites[i].Status == "DOWN" {
|
||||
return true
|
||||
}
|
||||
if sites[j].Status == "DOWN" {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return sites[i].Name < sites[j].Name
|
||||
})
|
||||
|
||||
t, _ := template.New("status").Parse(tpl)
|
||||
data := struct {
|
||||
Title string
|
||||
Sites []models.Site
|
||||
}{Title: title, Sites: sites}
|
||||
t.Execute(w, data)
|
||||
statusTpl.Execute(w, data)
|
||||
}
|
||||
|
||||
@@ -239,6 +239,7 @@ func (p *PostgresStore) ImportData(data models.Backup) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer tx.Rollback()
|
||||
|
||||
tx.Exec("TRUNCATE TABLE sites RESTART IDENTITY CASCADE")
|
||||
tx.Exec("TRUNCATE TABLE alerts RESTART IDENTITY CASCADE")
|
||||
|
||||
@@ -258,8 +258,8 @@ func (s *SQLiteStore) ImportData(data models.Backup) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer tx.Rollback()
|
||||
|
||||
// Wipe Existing
|
||||
tx.Exec("DELETE FROM sites")
|
||||
tx.Exec("DELETE FROM sqlite_sequence WHERE name='sites'")
|
||||
tx.Exec("DELETE FROM alerts")
|
||||
|
||||
Reference in New Issue
Block a user