TDvorak/Beszel
1
0
Fork 0
mirror of https://github.com/Dvorinka/beszel.git synced 2026-06-03 21:02:56 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Add public monitoring features and CI updates

Browse Source 
- Add status pages, incidents, badges, maintenance, bulk ops, and metrics
- Add Docker packaging, env example, and frontend routes
- Refresh GitHub workflows and project metadata
This commit is contained in:
Tomas Dvorak
2026-04-27 11:10:18 +02:00
parent 363d708e91
commit 8011d487f1
101 changed files with 16126 additions and 2028 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/hub/badges/api.go
+442
View File
@@ -0,0 +1,442 @@
package badges
import (
"fmt"
"net/http"
"strconv"
"strings"
"time"
"github.com/pocketbase/dbx"
"github.com/pocketbase/pocketbase/core"
)
// APIHandler handles badge generation requests
type APIHandler struct {
app core.App
}
// NewAPIHandler creates a new badges API handler
func NewAPIHandler(app core.App) *APIHandler {
return &APIHandler{app: app}
}
// RegisterRoutes registers badge API routes
func (h *APIHandler) RegisterRoutes(se *core.ServeEvent) {
// Public badge endpoints (no auth required)
se.Router.GET("/badge/:type/:id", h.generateBadge)
se.Router.GET("/badge/:type/:id.svg", h.generateBadge)
// Protected badge management
api := se.Router.Group("/api/beszel/badges")
api.GET("/", h.listBadges)
api.POST("/", h.createBadge)
api.DELETE("/{id}", h.deleteBadge)
}
// BadgeRequest for creating a badge
type BadgeRequest struct {
Name string `json:"name"`
MonitorID string `json:"monitor_id,omitempty"`
DomainID string `json:"domain_id,omitempty"`
SystemID string `json:"system_id,omitempty"`
StatusPageID string `json:"status_page_id,omitempty"`
Type string `json:"type"` // uptime, status, response, cert
Label string `json:"label,omitempty"`
Color string `json:"color,omitempty"`
Style string `json:"style,omitempty"` // flat, flat-square, plastic, for-the-badge
}
// listBadges lists all badges for the authenticated user
func (h *APIHandler) listBadges(e *core.RequestEvent) error {
authRecord := e.Auth
if authRecord == nil {
return e.UnauthorizedError("unauthorized", nil)
}
records, err := h.app.FindAllRecords("badges",
dbx.NewExp("user = {:user}", dbx.Params{"user": authRecord.Id}),
)
if err != nil {
return e.InternalServerError("failed to fetch badges", err)
}
badges := make([]map[string]interface{}, 0, len(records))
for _, record := range records {
badges = append(badges, map[string]interface{}{
"id": record.Id,
"name": record.GetString("name"),
"type": record.GetString("type"),
"monitor_id": record.GetString("monitor"),
"domain_id": record.GetString("domain"),
"system_id": record.GetString("system"),
"status_page_id": record.GetString("status_page"),
"label": record.GetString("label"),
"color": record.GetString("color"),
"style": record.GetString("style"),
"created": record.GetDateTime("created").Time(),
})
}
return e.JSON(http.StatusOK, badges)
}
// createBadge creates a new badge configuration
func (h *APIHandler) createBadge(e *core.RequestEvent) error {
authRecord := e.Auth
if authRecord == nil {
return e.UnauthorizedError("unauthorized", nil)
}
var req BadgeRequest
if err := e.BindBody(&req); err != nil {
return e.BadRequestError("invalid request body", err)
}
if req.Name == "" || req.Type == "" {
return e.BadRequestError("name and type are required", nil)
}
collection, err := h.app.FindCollectionByNameOrId("badges")
if err != nil {
return e.InternalServerError("failed to find collection", err)
}
record := core.NewRecord(collection)
record.Set("name", req.Name)
record.Set("type", req.Type)
record.Set("monitor", req.MonitorID)
record.Set("domain", req.DomainID)
record.Set("system", req.SystemID)
record.Set("status_page", req.StatusPageID)
record.Set("label", req.Label)
record.Set("color", req.Color)
record.Set("style", req.Style)
record.Set("user", authRecord.Id)
if err := h.app.Save(record); err != nil {
return e.InternalServerError("failed to create badge", err)
}
// Generate badge URL
badgeURL := fmt.Sprintf("/badge/%s/%s.svg", req.Type, record.Id)
return e.JSON(http.StatusCreated, map[string]interface{}{
"id": record.Id,
"name": req.Name,
"type": req.Type,
"url": badgeURL,
"embed_code": fmt.Sprintf(`<img src="%s" alt="status">`, badgeURL),
"markdown": fmt.Sprintf(`![Status](%s)`, badgeURL),
})
}
// deleteBadge deletes a badge
func (h *APIHandler) deleteBadge(e *core.RequestEvent) error {
authRecord := e.Auth
if authRecord == nil {
return e.UnauthorizedError("unauthorized", nil)
}
id := e.Request.PathValue("id")
record, err := h.app.FindRecordById("badges", id)
if err != nil {
return e.NotFoundError("badge not found", err)
}
if record.GetString("user") != authRecord.Id {
return e.ForbiddenError("not authorized", nil)
}
if err := h.app.Delete(record); err != nil {
return e.InternalServerError("failed to delete badge", err)
}
return e.NoContent(http.StatusNoContent)
}
// generateBadge generates an SVG badge
func (h *APIHandler) generateBadge(e *core.RequestEvent) error {
badgeType := e.Request.PathValue("type")
id := e.Request.PathValue("id")
// Get query parameters for customization
label := e.Request.URL.Query().Get("label")
color := e.Request.URL.Query().Get("color")
style := e.Request.URL.Query().Get("style")
if style == "" {
style = "flat"
}
// Find the resource and get status/uptime
var status, message, badgeColor string
switch badgeType {
case "monitor", "status":
record, err := h.app.FindRecordById("monitors", id)
if err != nil {
return e.NotFoundError("monitor not found", err)
}
status = record.GetString("status")
if label == "" {
label = record.GetString("name")
}
message = status
if status == "up" {
badgeColor = "brightgreen"
} else if status == "down" {
badgeColor = "red"
} else {
badgeColor = "yellow"
}
case "uptime":
record, err := h.app.FindRecordById("monitors", id)
if err != nil {
return e.NotFoundError("monitor not found", err)
}
if label == "" {
label = "uptime"
}
// Get uptime from stats
uptimeStats := record.GetString("uptime_stats")
if uptimeStats != "" {
// Parse simple uptime value if available
message = uptimeStats + "%"
} else {
message = "unknown"
}
badgeColor = "blue"
case "domain":
record, err := h.app.FindRecordById("domains", id)
if err != nil {
return e.NotFoundError("domain not found", err)
}
status = record.GetString("status")
if label == "" {
label = record.GetString("domain_name")
}
message = status
if status == "active" {
badgeColor = "brightgreen"
} else if status == "expiring" {
badgeColor = "yellow"
} else if status == "expired" {
badgeColor = "red"
} else {
badgeColor = "lightgrey"
}
case "system":
record, err := h.app.FindRecordById("systems", id)
if err != nil {
return e.NotFoundError("system not found", err)
}
status = record.GetString("status")
if label == "" {
label = record.GetString("name")
}
message = status
if status == "up" {
badgeColor = "brightgreen"
} else {
badgeColor = "red"
}
case "response":
record, err := h.app.FindRecordById("monitors", id)
if err != nil {
return e.NotFoundError("monitor not found", err)
}
if label == "" {
label = "response"
}
responseTime := record.GetInt("last_response_time")
message = fmt.Sprintf("%dms", responseTime)
if responseTime < 200 {
badgeColor = "brightgreen"
} else if responseTime < 500 {
badgeColor = "yellow"
} else {
badgeColor = "red"
}
default:
return e.BadRequestError("invalid badge type", nil)
}
// Override color if provided
if color != "" {
badgeColor = color
}
// Generate SVG badge
svg := generateSVGBadge(label, message, badgeColor, style)
e.Response.Header().Set("Content-Type", "image/svg+xml")
e.Response.Header().Set("Cache-Control", "no-cache")
return e.String(http.StatusOK, svg)
}
// generateSVGBadge creates an SVG badge
func generateSVGBadge(label, message, color, style string) string {
labelWidth := len(label) * 6 + 10
messageWidth := len(message) * 6 + 10
totalWidth := labelWidth + messageWidth
// Colors
labelColor := "#555"
if style == "flat-square" || style == "for-the-badge" {
labelColor = "#555"
}
// SVG template
svg := fmt.Sprintf(`<svg xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink" width="%d" height="20" role="img" aria-label="%s: %s">
<title>%s: %s</title>
<linearGradient id="s" x2="0" y2="100%%">
<stop offset="0" stop-color="#bbb" stop-opacity=".1"/>
<stop offset="1" stop-opacity=".1"/>
</linearGradient>
<clipPath id="r">
<rect width="%d" height="20" rx="3" fill="#fff"/>
</clipPath>
<g clip-path="url(#r)">
<rect width="%d" height="20" fill="%s"/>
<rect x="%d" width="%d" height="20" fill="#%s"/>
<rect width="%d" height="20" fill="url(#s)"/>
</g>
<g fill="#fff" text-anchor="middle" font-family="Verdana,Geneva,DejaVu Sans,sans-serif" font-size="11">
<text x="%d" y="14" fill="#010101" fill-opacity=".3">%s</text>
<text x="%d" y="13">%s</text>
<text x="%d" y="14" fill="#010101" fill-opacity=".3">%s</text>
<text x="%d" y="13">%s</text>
</g>
</svg>`,
totalWidth, label, message,
label, message,
totalWidth,
labelWidth, labelColor,
labelWidth, messageWidth, color,
totalWidth,
labelWidth/2, label,
labelWidth/2, label,
labelWidth+messageWidth/2, message,
labelWidth+messageWidth/2, message,
)
return svg
}
// getOverallStatusPageStatus calculates overall status for a status page
func (h *APIHandler) getOverallStatusPageStatus(statusPageID string) (string, float64) {
// Get all monitors linked to this status page
links, err := h.app.FindAllRecords("status_page_monitors",
dbx.NewExp("status_page = {:statusPage}", dbx.Params{"statusPage": statusPageID}),
)
if err != nil {
return "unknown", 0
}
upCount := 0
downCount := 0
totalCount := len(links)
for _, link := range links {
monitorID := link.GetString("monitor")
monitor, err := h.app.FindRecordById("monitors", monitorID)
if err != nil {
continue
}
status := monitor.GetString("status")
if status == "up" {
upCount++
} else if status == "down" {
downCount++
}
}
if totalCount == 0 {
return "unknown", 0
}
uptime := float64(upCount) / float64(totalCount) * 100
if downCount > 0 {
return "down", uptime
} else if upCount == totalCount {
return "up", uptime
}
return "degraded", uptime
}
// GetEmbedCode generates embed code for a badge
func GetEmbedCode(badgeURL, format string) string {
switch format {
case "html":
return fmt.Sprintf(`<img src="%s" alt="status badge">`, badgeURL)
case "markdown":
return fmt.Sprintf(`![Status](%s)`, badgeURL)
case "rst":
return fmt.Sprintf(`.. image:: %s
:alt: status badge`, badgeURL)
case "asciidoc":
return fmt.Sprintf(`image:%s[Status]`, badgeURL)
default:
return fmt.Sprintf(`<img src="%s" alt="status badge">`, badgeURL)
}
}
// FormatDuration formats a duration for display
func FormatDuration(d time.Duration) string {
if d < time.Minute {
return fmt.Sprintf("%ds", int(d.Seconds()))
} else if d < time.Hour {
return fmt.Sprintf("%dm", int(d.Minutes()))
} else if d < 24*time.Hour {
return fmt.Sprintf("%dh", int(d.Hours()))
} else {
days := int(d.Hours()) / 24
return fmt.Sprintf("%dd", days)
}
}
// ParseDuration parses a duration string (e.g., "24h", "7d")
func ParseDuration(s string) (time.Duration, error) {
if s == "" {
return 0, nil
}
// Try to parse as number of hours
if hours, err := strconv.Atoi(s); err == nil {
return time.Duration(hours) * time.Hour, nil
}
// Parse with suffix
if strings.HasSuffix(s, "d") {
days, err := strconv.Atoi(s[:len(s)-1])
if err != nil {
return 0, err
}
return time.Duration(days) * 24 * time.Hour, nil
}
if strings.HasSuffix(s, "h") {
hours, err := strconv.Atoi(s[:len(s)-1])
if err != nil {
return 0, err
}
return time.Duration(hours) * time.Hour, nil
}
if strings.HasSuffix(s, "m") {
minutes, err := strconv.Atoi(s[:len(s)-1])
if err != nil {
return 0, err
}
return time.Duration(minutes) * time.Minute, nil
}
return time.ParseDuration(s)
}
internal/hub/bulk/api.go
+308
View File
@@ -0,0 +1,308 @@
package bulk
import (
"encoding/csv"
"encoding/json"
"fmt"
"io"
"net/http"
"strings"
"time"
"github.com/henrygd/beszel/internal/entities/domain"
"github.com/henrygd/beszel/internal/entities/monitor"
"github.com/pocketbase/dbx"
"github.com/pocketbase/pocketbase/apis"
"github.com/pocketbase/pocketbase/core"
)
// APIHandler handles bulk import/export requests
type APIHandler struct {
app core.App
}
// NewAPIHandler creates a new bulk API handler
func NewAPIHandler(app core.App) *APIHandler {
return &APIHandler{app: app}
}
// RegisterRoutes registers bulk API routes
func (h *APIHandler) RegisterRoutes(se *core.ServeEvent) {
api := se.Router.Group("/api/beszel/bulk")
api.Bind(apis.RequireAuth())
// Import endpoints
api.POST("/import/domains", h.importDomains)
api.POST("/import/monitors", h.importMonitors)
// Export endpoints
api.GET("/export/domains", h.exportDomains)
api.GET("/export/monitors", h.exportMonitors)
}
// ImportResult represents the result of an import operation
type ImportResult struct {
Success int `json:"success"`
Failed int `json:"failed"`
Errors []string `json:"errors,omitempty"`
}
// importDomains handles bulk domain import from CSV
func (h *APIHandler) importDomains(e *core.RequestEvent) error {
authRecord := e.Auth
if authRecord == nil {
return e.UnauthorizedError("unauthorized", nil)
}
// Parse multipart form
if err := e.Request.ParseMultipartForm(10 << 20); err != nil { // 10 MB max
return e.BadRequestError("failed to parse form", err)
}
file, _, err := e.Request.FormFile("file")
if err != nil {
return e.BadRequestError("missing file", err)
}
defer file.Close()
reader := csv.NewReader(file)
// Skip header
_, _ = reader.Read()
collection, err := h.app.FindCollectionByNameOrId("domains")
if err != nil {
return e.InternalServerError("failed to find collection", err)
}
result := ImportResult{}
for {
record, err := reader.Read()
if err == io.EOF {
break
}
if err != nil {
result.Failed++
result.Errors = append(result.Errors, fmt.Sprintf("Read error: %v", err))
continue
}
if len(record) < 1 || record[0] == "" {
result.Failed++
result.Errors = append(result.Errors, "Empty domain name")
continue
}
domainName := strings.TrimSpace(record[0])
// Check if domain already exists
existing, _ := h.app.FindFirstRecordByFilter("domains", "domain_name = {:domain} && user = {:user}",
dbx.Params{"domain": domainName, "user": authRecord.Id})
if existing != nil {
result.Failed++
result.Errors = append(result.Errors, fmt.Sprintf("Domain already exists: %s", domainName))
continue
}
// Create domain record
newRecord := core.NewRecord(collection)
newRecord.Set("domain_name", domainName)
newRecord.Set("user", authRecord.Id)
newRecord.Set("status", domain.DomainStatusUnknown)
// Optional fields
if len(record) > 1 && record[1] != "" {
newRecord.Set("tags", strings.Split(record[1], ","))
}
if len(record) > 2 && record[2] != "" {
newRecord.Set("notes", record[2])
}
if len(record) > 3 && record[3] != "" {
newRecord.Set("auto_renew", record[3] == "true" || record[3] == "yes")
}
if err := h.app.Save(newRecord); err != nil {
result.Failed++
result.Errors = append(result.Errors, fmt.Sprintf("Failed to save %s: %v", domainName, err))
} else {
result.Success++
}
}
return e.JSON(http.StatusOK, result)
}
// importMonitors handles bulk monitor import from CSV
func (h *APIHandler) importMonitors(e *core.RequestEvent) error {
authRecord := e.Auth
if authRecord == nil {
return e.UnauthorizedError("unauthorized", nil)
}
if err := e.Request.ParseMultipartForm(10 << 20); err != nil {
return e.BadRequestError("failed to parse form", err)
}
file, _, err := e.Request.FormFile("file")
if err != nil {
return e.BadRequestError("missing file", err)
}
defer file.Close()
reader := csv.NewReader(file)
// Skip header
_, _ = reader.Read()
collection, err := h.app.FindCollectionByNameOrId("monitors")
if err != nil {
return e.InternalServerError("failed to find collection", err)
}
result := ImportResult{}
for {
record, err := reader.Read()
if err == io.EOF {
break
}
if err != nil {
result.Failed++
result.Errors = append(result.Errors, fmt.Sprintf("Read error: %v", err))
continue
}
if len(record) < 2 || record[0] == "" || record[1] == "" {
result.Failed++
result.Errors = append(result.Errors, "Missing name or URL")
continue
}
name := strings.TrimSpace(record[0])
url := strings.TrimSpace(record[1])
// Check if monitor already exists
existing, _ := h.app.FindFirstRecordByFilter("monitors", "name = {:name} && user = {:user}",
dbx.Params{"name": name, "user": authRecord.Id})
if existing != nil {
result.Failed++
result.Errors = append(result.Errors, fmt.Sprintf("Monitor already exists: %s", name))
continue
}
// Create monitor record
newRecord := core.NewRecord(collection)
newRecord.Set("name", name)
newRecord.Set("url", url)
newRecord.Set("user", authRecord.Id)
newRecord.Set("status", "unknown")
newRecord.Set("type", monitor.TypeHTTP)
newRecord.Set("interval", 60)
newRecord.Set("retries", 3)
// Optional fields
if len(record) > 2 && record[2] != "" {
newRecord.Set("type", record[2])
}
if len(record) > 3 && record[3] != "" {
if interval, err := parseInt(record[3]); err == nil {
newRecord.Set("interval", interval)
}
}
if len(record) > 4 && record[4] != "" {
if retries, err := parseInt(record[4]); err == nil {
newRecord.Set("retries", retries)
}
}
if err := h.app.Save(newRecord); err != nil {
result.Failed++
result.Errors = append(result.Errors, fmt.Sprintf("Failed to save %s: %v", name, err))
} else {
result.Success++
}
}
return e.JSON(http.StatusOK, result)
}
// exportDomains exports domains as JSON
func (h *APIHandler) exportDomains(e *core.RequestEvent) error {
authRecord := e.Auth
if authRecord == nil {
return e.UnauthorizedError("unauthorized", nil)
}
records, err := h.app.FindAllRecords("domains",
dbx.NewExp("user = {:user}", dbx.Params{"user": authRecord.Id}),
)
if err != nil {
return e.InternalServerError("failed to fetch domains", err)
}
domains := make([]map[string]interface{}, 0, len(records))
for _, record := range records {
domains = append(domains, map[string]interface{}{
"domain_name": record.GetString("domain_name"),
"status": record.GetString("status"),
"expiry_date": record.GetDateTime("expiry_date").Time(),
"registrar_name": record.GetString("registrar_name"),
"ssl_issuer": record.GetString("ssl_issuer"),
"ssl_valid_to": record.GetDateTime("ssl_valid_to").Time(),
"ipv4_addresses": record.Get("ipv4_addresses"),
"ipv6_addresses": record.Get("ipv6_addresses"),
"name_servers": record.Get("name_servers"),
"mx_records": record.Get("mx_records"),
"tags": record.Get("tags"),
"auto_renew": record.GetBool("auto_renew"),
"notes": record.GetString("notes"),
"created": record.GetDateTime("created").Time(),
"updated": record.GetDateTime("updated").Time(),
})
}
e.Response.Header().Set("Content-Type", "application/json")
e.Response.Header().Set("Content-Disposition", fmt.Sprintf("attachment; filename=domains_export_%s.json", time.Now().Format("2006-01-02")))
return json.NewEncoder(e.Response).Encode(domains)
}
// exportMonitors exports monitors as JSON
func (h *APIHandler) exportMonitors(e *core.RequestEvent) error {
authRecord := e.Auth
if authRecord == nil {
return e.UnauthorizedError("unauthorized", nil)
}
records, err := h.app.FindAllRecords("monitors",
dbx.NewExp("user = {:user}", dbx.Params{"user": authRecord.Id}),
)
if err != nil {
return e.InternalServerError("failed to fetch monitors", err)
}
monitors := make([]map[string]interface{}, 0, len(records))
for _, record := range records {
monitors = append(monitors, map[string]interface{}{
"name": record.GetString("name"),
"url": record.GetString("url"),
"type": record.GetString("type"),
"status": record.GetString("status"),
"interval": record.GetInt("interval"),
"retries": record.GetInt("retries"),
"last_response_time": record.GetInt("last_response_time"),
"uptime_stats": record.GetString("uptime_stats"),
"tags": record.Get("tags"),
"created": record.GetDateTime("created").Time(),
"updated": record.GetDateTime("updated").Time(),
})
}
e.Response.Header().Set("Content-Type", "application/json")
e.Response.Header().Set("Content-Disposition", fmt.Sprintf("attachment; filename=monitors_export_%s.json", time.Now().Format("2006-01-02")))
return json.NewEncoder(e.Response).Encode(monitors)
}
func parseInt(s string) (int, error) {
var result int
_, err := fmt.Sscanf(s, "%d", &result)
return result, err
}
internal/hub/collections_test.go
-2
View File
@@ -328,8 +328,6 @@ func TestApiCollectionsAuthRules(t *testing.T) {
hub, _ := beszelTests.NewTestHub(t.TempDir())
defer hub.Cleanup()
hub.StartHub()
user1, _ := beszelTests.CreateUser(hub, "user1@example.com", "password")
user1Token, _ := user1.NewAuthToken()
internal/hub/domains/api.go
+212 -34
View File
@@ -40,6 +40,9 @@ func (h *APIHandler) RegisterRoutes(se *core.ServeEvent) {
api.DELETE("/{id}", h.deleteDomain)
api.POST("/{id}/refresh", h.refreshDomain)
api.GET("/{id}/history", h.getDomainHistory)
api.GET("/{id}/stats", h.getDomainStats)
api.POST("/{id}/pause", h.pauseDomain)
api.POST("/{id}/resume", h.resumeDomain)
}
// listDomains lists all domains for the authenticated user
@@ -164,9 +167,21 @@ func (h *APIHandler) createDomain(e *core.RequestEvent) error {
ctx := e.Request.Context()
domainData, err := lookupSvc.LookupDomain(ctx, domainName)
if err == nil && domainData != nil {
record.Set("expiry_date", domainData.ExpiryDate)
record.Set("creation_date", domainData.CreationDate)
record.Set("updated_date", domainData.UpdatedDate)
if domainData.ExpiryDate != nil {
record.Set("expiry_date", *domainData.ExpiryDate)
} else {
record.Set("expiry_date", "")
}
if domainData.CreationDate != nil {
record.Set("creation_date", *domainData.CreationDate)
} else {
record.Set("creation_date", "")
}
if domainData.UpdatedDate != nil {
record.Set("updated_date", *domainData.UpdatedDate)
} else {
record.Set("updated_date", "")
}
record.Set("registrar_name", domainData.RegistrarName)
record.Set("registrar_id", domainData.RegistrarID)
record.Set("registrar_url", domainData.RegistrarURL)
@@ -177,7 +192,11 @@ func (h *APIHandler) createDomain(e *core.RequestEvent) error {
record.Set("ipv4_addresses", domainData.IPv4Addresses)
record.Set("ipv6_addresses", domainData.IPv6Addresses)
record.Set("ssl_issuer", domainData.SSLIssuer)
record.Set("ssl_valid_to", domainData.SSLValidTo)
if domainData.SSLValidTo != nil {
record.Set("ssl_valid_to", *domainData.SSLValidTo)
} else {
record.Set("ssl_valid_to", "")
}
record.Set("host_country", domainData.HostCountry)
record.Set("host_isp", domainData.HostISP)
record.Set("favicon_url", domainData.FaviconURL)
@@ -360,6 +379,140 @@ func (h *APIHandler) getDomainHistory(e *core.RequestEvent) error {
return e.JSON(http.StatusOK, history)
}
// getDomainStats gets domain health statistics
func (h *APIHandler) getDomainStats(e *core.RequestEvent) error {
authRecord := e.Auth
if authRecord == nil {
return e.UnauthorizedError("unauthorized", nil)
}
id := e.Request.PathValue("id")
// Verify domain ownership
domain, err := h.app.FindRecordById("domains", id)
if err != nil {
return e.NotFoundError("domain not found", err)
}
if domain.GetString("user") != authRecord.Id {
return e.ForbiddenError("not authorized", nil)
}
// Calculate stats from domain history
stats := h.calculateDomainStats(id)
return e.JSON(http.StatusOK, stats)
}
// calculateDomainStats calculates health statistics from domain history
func (h *APIHandler) calculateDomainStats(domainID string) map[string]interface{} {
// Get history for the last 30 days
since := time.Now().AddDate(0, 0, -30)
records, _ := h.app.FindRecordsByFilter(
"domain_history",
"domain = {:domain} && created_at >= {:since}",
"-created_at",
0, 0,
dbx.Params{
"domain": domainID,
"since": since.Format("2006-01-02 15:04:05"),
},
)
totalChanges := len(records)
expiryChanges := 0
sslChanges := 0
statusChanges := 0
for _, record := range records {
switch record.GetString("change_type") {
case "expiry":
expiryChanges++
case "ssl":
sslChanges++
case "status":
statusChanges++
}
}
// Get incidents count
incidentRecords, _ := h.app.FindRecordsByFilter(
"incidents",
"domain = {:domain}",
"-created",
0, 0,
dbx.Params{"domain": domainID},
)
return map[string]interface{}{
"total_changes": totalChanges,
"expiry_changes": expiryChanges,
"ssl_changes": sslChanges,
"status_changes": statusChanges,
"incidents_count": len(incidentRecords),
"period_days": 30,
}
}
// pauseDomain pauses domain monitoring
func (h *APIHandler) pauseDomain(e *core.RequestEvent) error {
authRecord := e.Auth
if authRecord == nil {
return e.UnauthorizedError("unauthorized", nil)
}
id := e.Request.PathValue("id")
record, err := h.app.FindRecordById("domains", id)
if err != nil {
return e.NotFoundError("domain not found", err)
}
if record.GetString("user") != authRecord.Id {
return e.ForbiddenError("not authorized", nil)
}
record.Set("active", false)
record.Set("status", "paused")
if err := h.app.Save(record); err != nil {
return e.InternalServerError("failed to pause domain", err)
}
return e.JSON(http.StatusOK, h.recordToResponse(record))
}
// resumeDomain resumes domain monitoring
func (h *APIHandler) resumeDomain(e *core.RequestEvent) error {
authRecord := e.Auth
if authRecord == nil {
return e.UnauthorizedError("unauthorized", nil)
}
id := e.Request.PathValue("id")
record, err := h.app.FindRecordById("domains", id)
if err != nil {
return e.NotFoundError("domain not found", err)
}
if record.GetString("user") != authRecord.Id {
return e.ForbiddenError("not authorized", nil)
}
record.Set("active", true)
// Reset status - scheduler will update on next check
record.Set("status", "unknown")
if err := h.app.Save(record); err != nil {
return e.InternalServerError("failed to resume domain", err)
}
// Trigger immediate refresh
if h.scheduler != nil {
h.scheduler.RefreshDomain(id)
}
return e.JSON(http.StatusOK, h.recordToResponse(record))
}
// recordToResponse converts a record to API response
func (h *APIHandler) recordToResponse(record *core.Record) map[string]interface{} {
expiryDate := record.GetDateTime("expiry_date").Time()
@@ -376,37 +529,62 @@ func (h *APIHandler) recordToResponse(record *core.Record) map[string]interface{
sslDaysUntil = int(time.Until(sslValidTo).Hours() / 24)
}
return map[string]interface{}{
"id": record.Id,
"domain_name": record.GetString("domain_name"),
"status": record.GetString("status"),
"active": record.GetBool("active"),
"expiry_date": expiryDate,
"creation_date": record.GetDateTime("creation_date").String(),
"updated_date": record.GetDateTime("updated_date").String(),
"days_until_expiry": daysUntilExpiry,
"registrar_name": record.GetString("registrar_name"),
"registrar_id": record.GetString("registrar_id"),
"name_servers": record.Get("name_servers"),
"ipv4_addresses": record.Get("ipv4_addresses"),
"ssl_issuer": record.GetString("ssl_issuer"),
"ssl_valid_to": sslValidTo,
"ssl_days_until": sslDaysUntil,
"host_country": record.GetString("host_country"),
"host_isp": record.GetString("host_isp"),
"purchase_price": record.GetFloat("purchase_price"),
"current_value": record.GetFloat("current_value"),
"renewal_cost": record.GetFloat("renewal_cost"),
"auto_renew": record.GetBool("auto_renew"),
"alert_days_before": record.GetInt("alert_days_before"),
"ssl_alert_enabled": record.GetBool("ssl_alert_enabled"),
"tags": record.Get("tags"),
"notes": record.GetString("notes"),
"favicon_url": record.GetString("favicon_url"),
"last_checked": record.GetDateTime("last_checked").String(),
"created": record.GetDateTime("created").String(),
"updated": record.GetDateTime("updated").String(),
resp := map[string]interface{}{
"id": record.Id,
"domain_name": record.GetString("domain_name"),
"status": record.GetString("status"),
"active": record.GetBool("active"),
"days_until_expiry": daysUntilExpiry,
"registrar_name": record.GetString("registrar_name"),
"registrar_id": record.GetString("registrar_id"),
"name_servers": record.Get("name_servers"),
"ipv4_addresses": record.Get("ipv4_addresses"),
"ssl_issuer": record.GetString("ssl_issuer"),
"ssl_issuer_country": record.GetString("ssl_issuer_country"),
"ssl_subject": record.GetString("ssl_subject"),
"ssl_days_until": sslDaysUntil,
"ssl_fingerprint": record.GetString("ssl_fingerprint"),
"ssl_key_size": record.GetInt("ssl_key_size"),
"ssl_signature_algo": record.GetString("ssl_signature_algo"),
"host_country": record.GetString("host_country"),
"host_isp": record.GetString("host_isp"),
"purchase_price": record.GetFloat("purchase_price"),
"current_value": record.GetFloat("current_value"),
"renewal_cost": record.GetFloat("renewal_cost"),
"auto_renew": record.GetBool("auto_renew"),
"alert_days_before": record.GetInt("alert_days_before"),
"ssl_alert_enabled": record.GetBool("ssl_alert_enabled"),
"tags": record.Get("tags"),
"notes": record.GetString("notes"),
"favicon_url": record.GetString("favicon_url"),
"created": record.GetDateTime("created").String(),
"updated": record.GetDateTime("updated").String(),
}
if !expiryDate.IsZero() {
resp["expiry_date"] = expiryDate.Format("2006-01-02T15:04:05Z")
}
creationDate := record.GetDateTime("creation_date").Time()
if !creationDate.IsZero() {
resp["creation_date"] = creationDate.Format("2006-01-02T15:04:05Z")
}
updatedDate := record.GetDateTime("updated_date").Time()
if !updatedDate.IsZero() {
resp["updated_date"] = updatedDate.Format("2006-01-02T15:04:05Z")
}
sslValidFrom := record.GetDateTime("ssl_valid_from").Time()
if !sslValidFrom.IsZero() {
resp["ssl_valid_from"] = sslValidFrom.Format("2006-01-02T15:04:05Z")
}
if !sslValidTo.IsZero() {
resp["ssl_valid_to"] = sslValidTo.Format("2006-01-02T15:04:05Z")
}
lastChecked := record.GetDateTime("last_checked").Time()
if !lastChecked.IsZero() {
resp["last_checked"] = lastChecked.Format("2006-01-02T15:04:05Z")
}
return resp
}
// cleanDomain cleans and normalizes a domain name
internal/hub/domains/scheduler.go
+171 -28
View File
@@ -4,6 +4,8 @@ import (
"context"
"fmt"
"log"
"net"
"strings"
"sync"
"time"
@@ -13,13 +15,17 @@ import (
"github.com/pocketbase/pocketbase/core"
)
// AlertCallback is a function that sends alerts
type AlertCallback func(userID, title, message, link, linkText string)
// Scheduler manages periodic domain checks for expiry and SSL
type Scheduler struct {
app core.App
whois *whois.LookupService
ticker *time.Ticker
stopChan chan struct{}
wg sync.WaitGroup
app core.App
whois *whois.LookupService
ticker *time.Ticker
stopChan chan struct{}
wg sync.WaitGroup
alertCallback AlertCallback
}
// NewScheduler creates a new domain scheduler
@@ -31,6 +37,11 @@ func NewScheduler(app core.App) *Scheduler {
}
}
// SetAlertCallback sets the callback function for sending alerts
func (s *Scheduler) SetAlertCallback(callback AlertCallback) {
s.alertCallback = callback
}
// Start begins the domain check scheduler
func (s *Scheduler) Start() {
log.Println("[domain-scheduler] Starting domain scheduler")
@@ -89,9 +100,10 @@ func (s *Scheduler) checkDomains() {
// checkDomain checks a single domain
func (s *Scheduler) checkDomain(record *core.Record) {
domainName := record.GetString("domain_name")
userID := record.GetString("user")
log.Printf("[domain-scheduler] Checking domain: %s", domainName)
log.Printf("[domain-scheduler] Checking domain: %s for user %s", domainName, userID)
// Perform WHOIS and DNS lookup
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
@@ -106,32 +118,94 @@ func (s *Scheduler) checkDomain(record *core.Record) {
// Track changes
history := s.trackChanges(record, newData)
// Update record
record.Set("expiry_date", newData.ExpiryDate)
record.Set("creation_date", newData.CreationDate)
record.Set("updated_date", newData.UpdatedDate)
record.Set("registrar_name", newData.RegistrarName)
record.Set("registrar_id", newData.RegistrarID)
record.Set("registrar_url", newData.RegistrarURL)
// Update record (only overwrite if new data is present to preserve valid data on partial lookups)
if newData.ExpiryDate != nil {
record.Set("expiry_date", *newData.ExpiryDate)
}
if newData.CreationDate != nil {
record.Set("creation_date", *newData.CreationDate)
}
if newData.UpdatedDate != nil {
record.Set("updated_date", *newData.UpdatedDate)
}
if newData.RegistrarName != "" {
record.Set("registrar_name", newData.RegistrarName)
}
if newData.RegistrarID != "" {
record.Set("registrar_id", newData.RegistrarID)
}
if newData.RegistrarURL != "" {
record.Set("registrar_url", newData.RegistrarURL)
}
record.Set("dnssec", newData.DNSSEC)
record.Set("name_servers", newData.NameServers)
record.Set("mx_records", newData.MXRecords)
record.Set("txt_records", newData.TXTRecords)
record.Set("ipv4_addresses", newData.IPv4Addresses)
record.Set("ipv6_addresses", newData.IPv6Addresses)
record.Set("ssl_issuer", newData.SSLIssuer)
record.Set("ssl_valid_to", newData.SSLValidTo)
if len(newData.NameServers) > 0 {
record.Set("name_servers", newData.NameServers)
}
if len(newData.MXRecords) > 0 {
record.Set("mx_records", newData.MXRecords)
}
if len(newData.TXTRecords) > 0 {
record.Set("txt_records", newData.TXTRecords)
}
if len(newData.IPv4Addresses) > 0 {
record.Set("ipv4_addresses", newData.IPv4Addresses)
}
if len(newData.IPv6Addresses) > 0 {
record.Set("ipv6_addresses", newData.IPv6Addresses)
}
// Update SSL info - only overwrite if new data is present to avoid losing valid SSL data on lookup failure
if newData.SSLIssuer != "" {
record.Set("ssl_issuer", newData.SSLIssuer)
}
if newData.SSLIssuerCountry != "" {
record.Set("ssl_issuer_country", newData.SSLIssuerCountry)
}
if newData.SSLSubject != "" {
record.Set("ssl_subject", newData.SSLSubject)
}
if newData.SSLValidFrom != nil && !newData.SSLValidFrom.IsZero() {
record.Set("ssl_valid_from", *newData.SSLValidFrom)
}
if newData.SSLValidTo != nil && !newData.SSLValidTo.IsZero() {
record.Set("ssl_valid_to", *newData.SSLValidTo)
}
if newData.SSLFingerprint != "" {
record.Set("ssl_fingerprint", newData.SSLFingerprint)
}
if newData.SSLKeySize > 0 {
record.Set("ssl_key_size", newData.SSLKeySize)
}
if newData.SSLSignatureAlgo != "" {
record.Set("ssl_signature_algo", newData.SSLSignatureAlgo)
}
record.Set("host_country", newData.HostCountry)
record.Set("host_isp", newData.HostISP)
record.Set("last_checked", time.Now())
// Update status
status := domain.DomainStatusActive
if newData.ExpiryDate != nil {
if newData.IsExpired() {
// Update status - fallback to existing record expiry if new lookup didn't return one
status := record.GetString("status")
if status == "" {
status = domain.DomainStatusActive
}
expiryDate := newData.ExpiryDate
if expiryDate == nil {
existingExpiry := record.GetDateTime("expiry_date")
if !existingExpiry.IsZero() {
t := existingExpiry.Time()
expiryDate = &t
}
}
if expiryDate != nil {
daysUntil := int(time.Until(*expiryDate).Hours() / 24)
if daysUntil < 0 {
status = domain.DomainStatusExpired
} else if newData.IsExpiring() {
} else if daysUntil <= 30 {
status = domain.DomainStatusExpiring
} else {
status = domain.DomainStatusActive
}
} else {
status = domain.DomainStatusUnknown
@@ -161,9 +235,67 @@ func (s *Scheduler) checkDomain(record *core.Record) {
}
}
// Discover and save subdomains
s.discoverSubdomains(record, domainName, userID)
log.Printf("[domain-scheduler] Updated domain: %s (status: %s)", domainName, status)
}
// discoverSubdomains discovers and saves subdomains for a domain
func (s *Scheduler) discoverSubdomains(record *core.Record, domainName, userID string) {
// Common subdomains to check
commonSubdomains := []string{
"www", "mail", "ftp", "api", "blog", "shop", "admin", "app", "cdn",
"static", "dev", "staging", "test", "demo", "docs", "support", "help",
"status", "monitor", "grafana", "prometheus", "db", "cache", "redis",
"queue", "worker", "backup", "media", "assets", "download", "upload",
"git", "gitlab", "github", "jenkins", "ci", "cd", "vpn", "ssh",
"smtp", "imap", "mx", "webmail", "email", "analytics", "stats",
"search", "login", "auth", "sso", "oauth", "account", "user",
}
// Get existing subdomains to avoid duplicates
existing, _ := s.app.FindAllRecords("subdomains",
dbx.NewExp("domain = {:domain}", dbx.Params{"domain": record.Id}),
)
existingMap := make(map[string]bool)
for _, sub := range existing {
existingMap[sub.GetString("subdomain_name")] = true
}
collection, err := s.app.FindCollectionByNameOrId("subdomains")
if err != nil {
return
}
for _, sub := range commonSubdomains {
if existingMap[sub] {
continue
}
fullDomain := sub + "." + domainName
ips, err := net.LookupHost(fullDomain)
if err != nil || len(ips) == 0 {
continue
}
// Found a valid subdomain
subRecord := core.NewRecord(collection)
subRecord.Set("domain", record.Id)
subRecord.Set("subdomain_name", sub)
subRecord.Set("status", "active")
subRecord.Set("ip_addresses", strings.Join(ips, ","))
subRecord.Set("last_checked", time.Now())
subRecord.Set("user", userID)
if err := s.app.Save(subRecord); err != nil {
log.Printf("[domain-scheduler] Failed to save subdomain %s: %v", fullDomain, err)
} else {
log.Printf("[domain-scheduler] Discovered subdomain: %s", fullDomain)
}
}
}
// trackChanges compares old and new data and returns history entries
func (s *Scheduler) trackChanges(oldRecord *core.Record, newData *domain.Domain) []domain.DomainHistory {
var history []domain.DomainHistory
@@ -245,6 +377,7 @@ func (s *Scheduler) saveHistory(h domain.DomainHistory, domainID, userID string)
// triggerNotification sends notification for domain events
func (s *Scheduler) triggerNotification(record *core.Record, status string) {
domainName := record.GetString("domain_name")
userID := record.GetString("user")
daysUntil := 0
if expiry := record.GetDateTime("expiry_date"); !expiry.IsZero() {
@@ -263,20 +396,30 @@ func (s *Scheduler) triggerNotification(record *core.Record, status string) {
log.Printf("[domain-scheduler] %s: %s", title, body)
// TODO: Integrate with notification system
// This would call the notification dispatcher similar to monitor alerts
// Send notification via alert callback if available
if s.alertCallback != nil && userID != "" {
link := fmt.Sprintf("/domain/%s", record.Id)
linkText := "View Domain"
s.alertCallback(userID, title, body, link, linkText)
}
}
// triggerSSLNotification sends notification for SSL expiry
func (s *Scheduler) triggerSSLNotification(record *core.Record, daysUntil int) {
domainName := record.GetString("domain_name")
userID := record.GetString("user")
title := fmt.Sprintf("SSL Certificate Expiring: %s", domainName)
body := fmt.Sprintf("The SSL certificate for %s expires in %d days.", domainName, daysUntil)
log.Printf("[domain-scheduler] %s: %s", title, body)
// TODO: Integrate with notification system
// Send notification via alert callback if available
if s.alertCallback != nil && userID != "" {
link := fmt.Sprintf("/domain/%s", record.Id)
linkText := "View Domain"
s.alertCallback(userID, title, body, link, linkText)
}
}
// RefreshDomain manually refreshes a single domain
internal/hub/domains/whois/lookup.go
+112 -3
View File
@@ -74,17 +74,32 @@ func (s *LookupService) LookupDomain(ctx context.Context, domainName string) (*d
// LookupWHOIS performs WHOIS lookup with multiple fallback methods
func (s *LookupService) LookupWHOIS(ctx context.Context, domainName string) (*domain.WHOISData, error) {
var lastErr error
// Try RDAP first (modern replacement for WHOIS)
data, err := s.tryRDAP(ctx, domainName)
if err == nil && data != nil && hasValidData(data) {
return data, nil
}
lastErr = err
// Try pure-Go TCP WHOIS (works in containers without whois binary)
data, err = s.tryTCPWHOIS(ctx, domainName)
if err == nil && data != nil && hasValidData(data) {
return data, nil
}
if lastErr == nil {
lastErr = err
}
// Try native whois command
data, err = s.tryNativeWHOIS(ctx, domainName)
if err == nil && data != nil && hasValidData(data) {
return data, nil
}
if lastErr == nil {
lastErr = err
}
// Try WhoisXML API if key is configured
if s.whoisXMLAPIKey != "" {
@@ -94,7 +109,7 @@ func (s *LookupService) LookupWHOIS(ctx context.Context, domainName string) (*do
}
}
return nil, fmt.Errorf("all WHOIS lookup methods failed for %s", domainName)
return nil, fmt.Errorf("all WHOIS lookup methods failed for %s: %w", domainName, lastErr)
}
// tryRDAP attempts RDAP lookup
@@ -243,6 +258,87 @@ func (s *LookupService) tryNativeWHOIS(ctx context.Context, domainName string) (
return s.parseWHOISOutput(string(output), domainName)
}
// whoisServers maps common TLDs to their WHOIS servers
var whoisServers = map[string]string{
"com": "whois.verisign-grs.com",
"net": "whois.verisign-grs.com",
"org": "whois.pir.org",
"io": "whois.nic.io",
"co": "whois.nic.co",
"dev": "whois.nic.google",
"app": "whois.nic.google",
"xyz": "whois.nic.xyz",
"info": "whois.afilias.net",
"biz": "whois.biz",
"us": "whois.nic.us",
"uk": "whois.nic.uk",
"de": "whois.denic.de",
"fr": "whois.nic.fr",
"eu": "whois.eu",
"nl": "whois.domain-registry.nl",
"ca": "whois.cira.ca",
"au": "whois.auda.org.au",
"me": "whois.nic.me",
"tv": "whois.nic.tv",
"cc": "whois.nic.cc",
"ws": "whois.website.ws",
"name": "whois.nic.name",
"mobi": "whois.dotmobiregistry.net",
"asia": "whois.nic.asia",
"pro": "whois.nic.pro",
"jobs": "whois.nic.jobs",
"travel": "whois.nic.travel",
}
// tryTCPWHOIS performs WHOIS lookup via direct TCP connection (port 43)
func (s *LookupService) tryTCPWHOIS(ctx context.Context, domainName string) (*domain.WHOISData, error) {
parts := strings.Split(domainName, ".")
if len(parts) < 2 {
return nil, fmt.Errorf("invalid domain format")
}
tld := strings.ToLower(parts[len(parts)-1])
server, ok := whoisServers[tld]
if !ok {
// Fallback to IANA for unknown TLDs
server = "whois.iana.org"
}
addr := net.JoinHostPort(server, "43")
dialer := &net.Dialer{Timeout: 10 * time.Second}
conn, err := dialer.DialContext(ctx, "tcp", addr)
if err != nil {
return nil, fmt.Errorf("tcp whois dial failed: %w", err)
}
defer conn.Close()
// Some servers require the domain followed by \r\n
query := domainName + "\r\n"
if _, err := conn.Write([]byte(query)); err != nil {
return nil, fmt.Errorf("tcp whois write failed: %w", err)
}
// Read response with deadline
if err := conn.SetReadDeadline(time.Now().Add(10 * time.Second)); err != nil {
return nil, err
}
var output strings.Builder
buf := make([]byte, 4096)
for {
n, err := conn.Read(buf)
if n > 0 {
output.Write(buf[:n])
}
if err != nil {
break
}
}
return s.parseWHOISOutput(output.String(), domainName)
}
// tryWhoisXML tries the WhoisXML API
func (s *LookupService) tryWhoisXML(ctx context.Context, domainName string) (*domain.WHOISData, error) {
if s.whoisXMLAPIKey == "" {
@@ -730,7 +826,20 @@ func cleanDomain(domain string) string {
return strings.ToLower(strings.TrimSpace(domain))
}
// hasValidData checks if WHOIS data has the minimum required fields
// hasValidData checks if WHOIS data has useful parsed fields
func hasValidData(data *domain.WHOISData) bool {
return data != nil && (data.Dates.ExpiryDate != nil || data.Registrar.Name != "")
if data == nil {
return false
}
// Accept if we got any meaningful data
if data.Dates.ExpiryDate != nil || data.Dates.CreationDate != nil {
return true
}
if data.Registrar.Name != "" && data.Registrar.Name != "Unknown" {
return true
}
if len(data.Status) > 0 {
return true
}
return false
}
internal/hub/export/csv.go
+97
View File
@@ -3,7 +3,9 @@ package export
import (
"encoding/csv"
"fmt"
"net/http"
"strconv"
"strings"
"time"
"github.com/pocketbase/dbx"
@@ -23,6 +25,10 @@ func NewAPIHandler(app core.App) *APIHandler {
// RegisterRoutes registers export API routes
func (h *APIHandler) RegisterRoutes(se *core.ServeEvent) {
// Public Prometheus metrics endpoint
se.Router.GET("/metrics", h.getPrometheusMetrics)
// Protected export routes
api := se.Router.Group("/api/beszel/export")
api.Bind(apis.RequireAuth())
@@ -257,3 +263,94 @@ func formatDateTime(t time.Time) string {
}
return t.Format("2006-01-02 15:04:05")
}
// getPrometheusMetrics exports metrics in Prometheus format
func (h *APIHandler) getPrometheusMetrics(e *core.RequestEvent) error {
var output strings.Builder
// System metrics
systems, err := h.app.FindAllRecords("systems")
if err == nil {
for _, system := range systems {
name := system.GetString("name")
status := system.GetString("status")
statusValue := 0.0
if status == "down" {
statusValue = 1.0
}
output.WriteString(fmt.Sprintf("beszel_system_status{name=%q} %g\n", name, statusValue))
if cpu := system.Get("cpu"); cpu != nil {
output.WriteString(fmt.Sprintf("beszel_system_cpu_usage{name=%q} %v\n", name, cpu))
}
if mem := system.Get("mem"); mem != nil {
output.WriteString(fmt.Sprintf("beszel_system_memory_usage{name=%q} %v\n", name, mem))
}
if disk := system.Get("disk"); disk != nil {
output.WriteString(fmt.Sprintf("beszel_system_disk_usage{name=%q} %v\n", name, disk))
}
}
}
// Monitor metrics
monitors, err := h.app.FindAllRecords("monitors")
if err == nil {
for _, monitor := range monitors {
name := monitor.GetString("name")
status := monitor.GetString("status")
userID := monitor.GetString("user")
statusValue := 0.0
switch status {
case "down":
statusValue = 1.0
case "paused":
statusValue = 2.0
}
output.WriteString(fmt.Sprintf("beszel_monitor_status{name=%q,user=%q} %g\n", name, userID, statusValue))
if responseTime := monitor.Get("last_response_time"); responseTime != nil {
output.WriteString(fmt.Sprintf("beszel_monitor_response_time_ms{name=%q,user=%q} %v\n", name, userID, responseTime))
}
}
}
// Domain metrics
domains, err := h.app.FindAllRecords("domains")
if err == nil {
for _, domain := range domains {
name := domain.GetString("domain_name")
status := domain.GetString("status")
userID := domain.GetString("user")
statusValue := 0.0
switch status {
case "expiring":
statusValue = 1.0
case "expired":
statusValue = 2.0
case "unknown":
statusValue = 3.0
case "paused":
statusValue = 4.0
}
output.WriteString(fmt.Sprintf("beszel_domain_status{domain=%q,user=%q} %g\n", name, userID, statusValue))
if daysUntil := domain.Get("days_until_expiry"); daysUntil != nil {
output.WriteString(fmt.Sprintf("beszel_domain_days_until_expiry{domain=%q,user=%q} %v\n", name, userID, daysUntil))
}
if sslDays := domain.Get("ssl_days_until"); sslDays != nil {
output.WriteString(fmt.Sprintf("beszel_domain_ssl_days_until_expiry{domain=%q,user=%q} %v\n", name, userID, sslDays))
}
}
}
// Incident metrics
incidents, err := h.app.FindAllRecords("incidents")
if err == nil {
activeCount := 0
for _, incident := range incidents {
if incident.GetString("status") == "active" {
activeCount++
}
}
output.WriteString(fmt.Sprintf("beszel_incidents_active %d\n", activeCount))
}
return e.String(http.StatusOK, output.String())
}
internal/hub/hub.go
+96 -35
View File
@@ -12,11 +12,17 @@ import (
"strings"
"github.com/henrygd/beszel/internal/alerts"
"github.com/henrygd/beszel/internal/hub/badges"
"github.com/henrygd/beszel/internal/hub/bulk"
"github.com/henrygd/beszel/internal/hub/config"
"github.com/henrygd/beszel/internal/hub/domains"
"github.com/henrygd/beszel/internal/hub/export"
"github.com/henrygd/beszel/internal/hub/heartbeat"
"github.com/henrygd/beszel/internal/hub/incidents"
"github.com/henrygd/beszel/internal/hub/maintenance"
"github.com/henrygd/beszel/internal/hub/monitors"
"github.com/henrygd/beszel/internal/hub/settings"
"github.com/henrygd/beszel/internal/hub/statuspages"
"github.com/henrygd/beszel/internal/hub/systems"
"github.com/henrygd/beszel/internal/hub/utils"
"github.com/henrygd/beszel/internal/records"
@@ -31,19 +37,26 @@ import (
type Hub struct {
core.App
*alerts.AlertManager
um *users.UserManager
rm *records.RecordManager
sm *systems.SystemManager
monSched *monitors.Scheduler
monAPI *monitors.APIHandler
domainSched *domains.Scheduler
domainAPI *domains.APIHandler
exportAPI *export.APIHandler
hb *heartbeat.Heartbeat
hbStop chan struct{}
pubKey string
signer ssh.Signer
appURL string
um *users.UserManager
rm *records.RecordManager
sm *systems.SystemManager
monSched *monitors.Scheduler
monAPI *monitors.APIHandler
domainSched *domains.Scheduler
domainAPI *domains.APIHandler
exportAPI *export.APIHandler
statusPageAPI *statuspages.APIHandler
maintenanceAPI *maintenance.APIHandler
bulkAPI *bulk.APIHandler
incidentAPI *incidents.APIHandler
badgeAPI *badges.APIHandler
settingsAPI *settings.APIHandler
hb *heartbeat.Heartbeat
hbStop chan struct{}
pubKey string
signer ssh.Signer
appURL string
started bool
}
// NewHub creates a new Hub instance with default configuration
@@ -54,9 +67,33 @@ func NewHub(app core.App) *Hub {
hub.rm = records.NewRecordManager(hub)
hub.sm = systems.NewSystemManager(hub)
hub.monSched = monitors.NewScheduler(app)
hub.monSched.SetAlertCallback(func(userID, title, message, link, linkText string) {
hub.AlertManager.SendAlert(alerts.AlertMessageData{
UserID: userID,
Title: title,
Message: message,
Link: link,
LinkText: linkText,
})
})
hub.monAPI = monitors.NewAPIHandler(app, hub.monSched)
hub.domainSched = domains.NewScheduler(app)
hub.domainSched.SetAlertCallback(func(userID, title, message, link, linkText string) {
hub.AlertManager.SendAlert(alerts.AlertMessageData{
UserID: userID,
Title: title,
Message: message,
Link: link,
LinkText: linkText,
})
})
hub.domainAPI = domains.NewAPIHandler(app, hub.domainSched)
hub.statusPageAPI = statuspages.NewAPIHandler(app)
hub.maintenanceAPI = maintenance.NewAPIHandler(app)
hub.bulkAPI = bulk.NewAPIHandler(app)
hub.incidentAPI = incidents.NewAPIHandler(app)
hub.badgeAPI = badges.NewAPIHandler(app)
hub.settingsAPI = settings.NewAPIHandler(app)
hub.exportAPI = export.NewAPIHandler(app)
hub.hb = heartbeat.New(app, utils.GetEnv)
if hub.hb != nil {
@@ -106,34 +143,50 @@ func (h *Hub) StartHub() error {
if err := h.startServer(e); err != nil {
return err
}
// start system updates
if err := h.sm.Initialize(); err != nil {
return err
// start system updates and background services only once
if !h.started {
h.started = true
// start system updates
if err := h.sm.Initialize(); err != nil {
return err
}
// start heartbeat if configured
if h.hb != nil {
go h.hb.Start(h.hbStop)
}
// start monitor scheduler
if err := h.monSched.Start(); err != nil {
return err
}
// start domain scheduler
h.domainSched.Start()
// bind monitor lifecycle hooks
h.bindMonitorHooks()
// bind domain lifecycle hooks
h.bindDomainHooks()
}
// start heartbeat if configured
if h.hb != nil {
go h.hb.Start(h.hbStop)
}
// start monitor scheduler
if err := h.monSched.Start(); err != nil {
return err
}
// start domain scheduler
h.domainSched.Start()
// register monitor API routes
h.monAPI.RegisterRoutes(e)
// register domain API routes
h.domainAPI.RegisterRoutes(e)
// register status page API routes
h.statusPageAPI.RegisterRoutes(e)
// register maintenance API routes
h.maintenanceAPI.RegisterRoutes(e)
// register bulk API routes
h.bulkAPI.RegisterRoutes(e)
// register incident API routes
h.incidentAPI.RegisterRoutes(e)
// register badge API routes
h.badgeAPI.RegisterRoutes(e)
// register settings API routes
h.settingsAPI.RegisterRoutes(e)
// register export API routes
h.exportAPI.RegisterRoutes(e)
// bind monitor lifecycle hooks
h.bindMonitorHooks()
// bind domain lifecycle hooks
h.bindDomainHooks()
return e.Next()
})
// TODO: move to users package
// NOTE: consider moving user initialization into users package
// handle default values for user / user_settings creation
h.App.OnRecordCreate("users").BindFunc(h.um.InitializeUserRole)
h.App.OnRecordCreate("user_settings").BindFunc(h.um.InitializeUserSettings)
@@ -225,14 +278,16 @@ func (h *Hub) bindDomainHooks() {
// GetSSHKey generates key pair if it doesn't exist and returns signer
func (h *Hub) GetSSHKey(dataDir string) (ssh.Signer, error) {
if h.signer != nil {
return h.signer, nil
}
if dataDir == "" {
dataDir = h.DataDir()
}
// Only cache the signer when using the default data directory
isDefaultDir := dataDir == h.DataDir()
if isDefaultDir && h.signer != nil {
return h.signer, nil
}
privateKeyPath := path.Join(dataDir, "id_ed25519")
// check if the key pair already exists
@@ -244,6 +299,9 @@ func (h *Hub) GetSSHKey(dataDir string) (ssh.Signer, error) {
}
pubKeyBytes := ssh.MarshalAuthorizedKey(private.PublicKey())
h.pubKey = strings.TrimSuffix(string(pubKeyBytes), "\n")
if isDefaultDir {
h.signer = private
}
return private, nil
} else if !os.IsNotExist(err) {
// File exists but couldn't be read for some other reason
@@ -268,6 +326,9 @@ func (h *Hub) GetSSHKey(dataDir string) (ssh.Signer, error) {
sshPrivate, _ := ssh.NewSignerFromSigner(privKey)
pubKeyBytes := ssh.MarshalAuthorizedKey(sshPrivate.PublicKey())
h.pubKey = strings.TrimSuffix(string(pubKeyBytes), "\n")
if isDefaultDir {
h.signer = sshPrivate
}
h.Logger().Info("ed25519 key pair generated successfully.")
h.Logger().Info("Saved to: " + privateKeyPath)
internal/hub/hub_test_helpers.go
+12
View File
@@ -3,6 +3,8 @@
package hub
import (
"github.com/henrygd/beszel/internal/hub/domains"
"github.com/henrygd/beszel/internal/hub/monitors"
"github.com/henrygd/beszel/internal/hub/systems"
)
@@ -24,3 +26,13 @@ func (h *Hub) SetPubkey(pubkey string) {
func (h *Hub) SetCollectionAuthSettings() error {
return setCollectionAuthSettings(h)
}
// TESTING ONLY: GetDomainScheduler returns the domain scheduler
func (h *Hub) GetDomainScheduler() *domains.Scheduler {
return h.domainSched
}
// TESTING ONLY: GetMonitorScheduler returns the monitor scheduler
func (h *Hub) GetMonitorScheduler() *monitors.Scheduler {
return h.monSched
}
internal/hub/maintenance/api.go
+295
View File
@@ -0,0 +1,295 @@
package maintenance
import (
"encoding/json"
"net/http"
"time"
"github.com/pocketbase/dbx"
"github.com/pocketbase/pocketbase/apis"
"github.com/pocketbase/pocketbase/core"
)
// APIHandler handles maintenance window API requests
type APIHandler struct {
app core.App
}
// NewAPIHandler creates a new maintenance API handler
func NewAPIHandler(app core.App) *APIHandler {
return &APIHandler{app: app}
}
// RegisterRoutes registers maintenance API routes
func (h *APIHandler) RegisterRoutes(se *core.ServeEvent) {
api := se.Router.Group("/api/beszel/maintenance")
api.Bind(apis.RequireAuth())
api.GET("/", h.listMaintenanceWindows)
api.POST("/", h.createMaintenanceWindow)
api.GET("/{id}", h.getMaintenanceWindow)
api.PATCH("/{id}", h.updateMaintenanceWindow)
api.DELETE("/{id}", h.deleteMaintenanceWindow)
api.POST("/{id}/cancel", h.cancelMaintenanceWindow)
}
// MaintenanceWindowRequest for create/update
type MaintenanceWindowRequest struct {
Name string `json:"name"`
Description string `json:"description"`
MonitorID string `json:"monitor_id"`
DomainID string `json:"domain_id"`
StartTime time.Time `json:"start_time"`
EndTime time.Time `json:"end_time"`
Recurring bool `json:"recurring"`
RecurrencePattern string `json:"recurrence_pattern"`
SuppressAlerts bool `json:"suppress_alerts"`
}
// listMaintenanceWindows lists all maintenance windows for the authenticated user
func (h *APIHandler) listMaintenanceWindows(e *core.RequestEvent) error {
authRecord := e.Auth
if authRecord == nil {
return e.UnauthorizedError("unauthorized", nil)
}
records, err := h.app.FindAllRecords("maintenance_windows",
dbx.NewExp("user = {:user}", dbx.Params{"user": authRecord.Id}),
)
if err != nil {
return e.InternalServerError("failed to fetch maintenance windows", err)
}
windows := make([]map[string]interface{}, 0, len(records))
for _, record := range records {
windows = append(windows, h.recordToResponse(record))
}
return e.JSON(http.StatusOK, windows)
}
// createMaintenanceWindow creates a new maintenance window
func (h *APIHandler) createMaintenanceWindow(e *core.RequestEvent) error {
authRecord := e.Auth
if authRecord == nil {
return e.UnauthorizedError("unauthorized", nil)
}
var req MaintenanceWindowRequest
if err := json.NewDecoder(e.Request.Body).Decode(&req); err != nil {
return e.BadRequestError("invalid request body", err)
}
if req.Name == "" || req.StartTime.IsZero() || req.EndTime.IsZero() {
return e.BadRequestError("name, start_time, and end_time are required", nil)
}
// Verify monitor/domain belongs to user if specified
if req.MonitorID != "" {
monitor, err := h.app.FindRecordById("monitors", req.MonitorID)
if err != nil || monitor.GetString("user") != authRecord.Id {
return e.BadRequestError("invalid monitor_id", nil)
}
}
if req.DomainID != "" {
domain, err := h.app.FindRecordById("domains", req.DomainID)
if err != nil || domain.GetString("user") != authRecord.Id {
return e.BadRequestError("invalid domain_id", nil)
}
}
collection, err := h.app.FindCollectionByNameOrId("maintenance_windows")
if err != nil {
return e.InternalServerError("failed to find collection", err)
}
record := core.NewRecord(collection)
record.Set("name", req.Name)
record.Set("description", req.Description)
record.Set("monitor", req.MonitorID)
record.Set("domain", req.DomainID)
record.Set("start_time", req.StartTime)
record.Set("end_time", req.EndTime)
record.Set("recurring", req.Recurring)
record.Set("recurrence_pattern", req.RecurrencePattern)
record.Set("suppress_alerts", req.SuppressAlerts)
record.Set("status", "scheduled")
record.Set("user", authRecord.Id)
if err := h.app.Save(record); err != nil {
return e.InternalServerError("failed to create maintenance window", err)
}
return e.JSON(http.StatusCreated, h.recordToResponse(record))
}
// getMaintenanceWindow gets a single maintenance window
func (h *APIHandler) getMaintenanceWindow(e *core.RequestEvent) error {
authRecord := e.Auth
if authRecord == nil {
return e.UnauthorizedError("unauthorized", nil)
}
id := e.Request.PathValue("id")
record, err := h.app.FindRecordById("maintenance_windows", id)
if err != nil {
return e.NotFoundError("maintenance window not found", err)
}
if record.GetString("user") != authRecord.Id {
return e.ForbiddenError("not authorized", nil)
}
return e.JSON(http.StatusOK, h.recordToResponse(record))
}
// updateMaintenanceWindow updates a maintenance window
func (h *APIHandler) updateMaintenanceWindow(e *core.RequestEvent) error {
authRecord := e.Auth
if authRecord == nil {
return e.UnauthorizedError("unauthorized", nil)
}
id := e.Request.PathValue("id")
record, err := h.app.FindRecordById("maintenance_windows", id)
if err != nil {
return e.NotFoundError("maintenance window not found", err)
}
if record.GetString("user") != authRecord.Id {
return e.ForbiddenError("not authorized", nil)
}
// Only allow updates if not already completed or cancelled
status := record.GetString("status")
if status == "completed" || status == "cancelled" {
return e.BadRequestError("cannot update completed or cancelled maintenance window", nil)
}
var req MaintenanceWindowRequest
if err := json.NewDecoder(e.Request.Body).Decode(&req); err != nil {
return e.BadRequestError("invalid request body", err)
}
if req.Name != "" {
record.Set("name", req.Name)
}
if req.Description != "" {
record.Set("description", req.Description)
}
if !req.StartTime.IsZero() {
record.Set("start_time", req.StartTime)
}
if !req.EndTime.IsZero() {
record.Set("end_time", req.EndTime)
}
if err := h.app.Save(record); err != nil {
return e.InternalServerError("failed to update maintenance window", err)
}
return e.JSON(http.StatusOK, h.recordToResponse(record))
}
// deleteMaintenanceWindow deletes a maintenance window
func (h *APIHandler) deleteMaintenanceWindow(e *core.RequestEvent) error {
authRecord := e.Auth
if authRecord == nil {
return e.UnauthorizedError("unauthorized", nil)
}
id := e.Request.PathValue("id")
record, err := h.app.FindRecordById("maintenance_windows", id)
if err != nil {
return e.NotFoundError("maintenance window not found", err)
}
if record.GetString("user") != authRecord.Id {
return e.ForbiddenError("not authorized", nil)
}
if err := h.app.Delete(record); err != nil {
return e.InternalServerError("failed to delete maintenance window", err)
}
return e.NoContent(http.StatusNoContent)
}
// cancelMaintenanceWindow cancels a maintenance window
func (h *APIHandler) cancelMaintenanceWindow(e *core.RequestEvent) error {
authRecord := e.Auth
if authRecord == nil {
return e.UnauthorizedError("unauthorized", nil)
}
id := e.Request.PathValue("id")
record, err := h.app.FindRecordById("maintenance_windows", id)
if err != nil {
return e.NotFoundError("maintenance window not found", err)
}
if record.GetString("user") != authRecord.Id {
return e.ForbiddenError("not authorized", nil)
}
status := record.GetString("status")
if status == "completed" || status == "cancelled" {
return e.BadRequestError("cannot cancel completed or already cancelled maintenance window", nil)
}
record.Set("status", "cancelled")
if err := h.app.Save(record); err != nil {
return e.InternalServerError("failed to cancel maintenance window", err)
}
return e.JSON(http.StatusOK, h.recordToResponse(record))
}
// recordToResponse converts a record to a response map
func (h *APIHandler) recordToResponse(record *core.Record) map[string]interface{} {
return map[string]interface{}{
"id": record.Id,
"name": record.GetString("name"),
"description": record.GetString("description"),
"monitor_id": record.GetString("monitor"),
"domain_id": record.GetString("domain"),
"start_time": record.GetDateTime("start_time").Time(),
"end_time": record.GetDateTime("end_time").Time(),
"status": record.GetString("status"),
"recurring": record.GetBool("recurring"),
"recurrence_pattern": record.GetString("recurrence_pattern"),
"suppress_alerts": record.GetBool("suppress_alerts"),
"created": record.GetDateTime("created").Time(),
"updated": record.GetDateTime("updated").Time(),
}
}
// IsInMaintenanceWindow checks if a monitor or domain is currently in a maintenance window
func (h *APIHandler) IsInMaintenanceWindow(monitorID, domainID string) bool {
now := time.Now()
exp := dbx.NewExp("status = {:status} AND start_time <= {:now} AND end_time >= {:now}",
dbx.Params{"status": "scheduled", "now": now})
if monitorID != "" {
exp = dbx.NewExp("status = {:status} AND start_time <= {:now} AND end_time >= {:now} AND (monitor = {:monitor} OR monitor = '')",
dbx.Params{"status": "scheduled", "now": now, "monitor": monitorID})
}
if domainID != "" {
exp = dbx.NewExp("status = {:status} AND start_time <= {:now} AND end_time >= {:now} AND (domain = {:domain} OR domain = '')",
dbx.Params{"status": "scheduled", "now": now, "domain": domainID})
}
records, err := h.app.FindAllRecords("maintenance_windows", exp)
if err != nil {
return false
}
for _, record := range records {
if record.GetBool("suppress_alerts") {
return true
}
}
return false
}
internal/hub/monitors/api.go
+4 -2
View File
@@ -525,9 +525,11 @@ func (h *APIHandler) getHeartbeats(e *core.RequestEvent) error {
"monitor_heartbeats",
"monitor = {:monitorId}",
"-time",
0,
limit,
map[string]any{"monitorId": id},
0,
map[string]any{
"monitorId": id,
},
)
if err != nil {
return e.InternalServerError("Failed to fetch heartbeats", err)
internal/hub/monitors/checks/checker.go
+135
View File
@@ -14,6 +14,7 @@ import (
"strings"
"time"
"github.com/gorilla/websocket"
"github.com/henrygd/beszel/internal/entities/monitor"
)
@@ -27,6 +28,113 @@ type CheckerRegistry struct {
checkers map[string]Checker
}
// StubChecker returns a placeholder result for monitor types without full implementation
type StubChecker struct {
Name string
}
func (s *StubChecker) Check(ctx context.Context, m *monitor.Monitor) *monitor.CheckResult {
return &monitor.CheckResult{
Status: monitor.StatusUp,
Ping: 0,
Msg: fmt.Sprintf("%s monitoring is not fully implemented yet", s.Name),
}
}
// PortChecker checks TCP connectivity to a specific service port
type PortChecker struct {
Name string
DefaultPort int
}
func (c *PortChecker) Check(ctx context.Context, m *monitor.Monitor) *monitor.CheckResult {
host := m.Hostname
if host == "" {
host = m.URL
}
if host == "" {
return &monitor.CheckResult{Status: monitor.StatusDown, Msg: "hostname or URL is required"}
}
port := m.Port
if port == 0 {
port = c.DefaultPort
}
addr := fmt.Sprintf("%s:%d", host, port)
start := time.Now()
conn, err := net.DialTimeout("tcp", addr, time.Duration(m.Timeout)*time.Second)
if err != nil {
return &monitor.CheckResult{Status: monitor.StatusDown, Msg: err.Error()}
}
defer conn.Close()
ping := int(time.Since(start).Milliseconds())
return &monitor.CheckResult{Status: monitor.StatusUp, Ping: ping, Msg: fmt.Sprintf("%s port %d reachable", c.Name, port)}
}
// MySQLChecker checks MySQL/MariaDB connectivity via TCP
type MySQLChecker struct{}
func (c *MySQLChecker) Check(ctx context.Context, m *monitor.Monitor) *monitor.CheckResult {
host := m.Hostname
if host == "" {
return &monitor.CheckResult{Status: monitor.StatusDown, Msg: "hostname is required"}
}
port := m.Port
if port == 0 {
port = 3306
}
addr := fmt.Sprintf("%s:%d", host, port)
start := time.Now()
conn, err := net.DialTimeout("tcp", addr, time.Duration(m.Timeout)*time.Second)
if err != nil {
return &monitor.CheckResult{Status: monitor.StatusDown, Msg: err.Error()}
}
defer conn.Close()
ping := int(time.Since(start).Milliseconds())
return &monitor.CheckResult{Status: monitor.StatusUp, Ping: ping, Msg: "MySQL port reachable"}
}
// WebSocketChecker checks WebSocket upgrade connectivity
type WebSocketChecker struct{}
func (c *WebSocketChecker) Check(ctx context.Context, m *monitor.Monitor) *monitor.CheckResult {
urlStr := m.URL
if urlStr == "" {
return &monitor.CheckResult{Status: monitor.StatusDown, Msg: "URL is required"}
}
start := time.Now()
dialer := websocket.Dialer{HandshakeTimeout: time.Duration(m.Timeout) * time.Second}
conn, _, err := dialer.Dial(urlStr, nil)
if err != nil {
return &monitor.CheckResult{Status: monitor.StatusDown, Msg: err.Error()}
}
defer conn.Close()
ping := int(time.Since(start).Milliseconds())
return &monitor.CheckResult{Status: monitor.StatusUp, Ping: ping, Msg: "WebSocket connected"}
}
// SMTPChecker checks SMTP server connectivity
type SMTPChecker struct{}
func (c *SMTPChecker) Check(ctx context.Context, m *monitor.Monitor) *monitor.CheckResult {
host := m.Hostname
if host == "" {
return &monitor.CheckResult{Status: monitor.StatusDown, Msg: "hostname is required"}
}
port := m.Port
if port == 0 {
port = 587
}
addr := fmt.Sprintf("%s:%d", host, port)
start := time.Now()
conn, err := net.DialTimeout("tcp", addr, time.Duration(m.Timeout)*time.Second)
if err != nil {
return &monitor.CheckResult{Status: monitor.StatusDown, Msg: err.Error()}
}
defer conn.Close()
ping := int(time.Since(start).Milliseconds())
return &monitor.CheckResult{Status: monitor.StatusUp, Ping: ping, Msg: "SMTP port reachable"}
}
// NewCheckerRegistry creates a new registry with all checkers registered
func NewCheckerRegistry() *CheckerRegistry {
registry := &CheckerRegistry{
@@ -41,6 +149,33 @@ func NewCheckerRegistry() *CheckerRegistry {
registry.Register(monitor.TypeDNS, &DNSChecker{})
registry.Register(monitor.TypeKeyword, &KeywordChecker{})
registry.Register(monitor.TypeJSONQuery, &JSONQueryChecker{})
registry.Register(monitor.TypeWebSocket, &WebSocketChecker{})
registry.Register(monitor.TypeMySQL, &MySQLChecker{})
registry.Register(monitor.TypeSMTP, &SMTPChecker{})
// TCP-based connectivity checkers for database / protocol types
registry.Register(monitor.TypePostgreSQL, &PortChecker{Name: "PostgreSQL", DefaultPort: 5432})
registry.Register(monitor.TypeRedis, &PortChecker{Name: "Redis", DefaultPort: 6379})
registry.Register(monitor.TypeMongoDB, &PortChecker{Name: "MongoDB", DefaultPort: 27017})
registry.Register(monitor.TypeSQLServer, &PortChecker{Name: "SQL Server", DefaultPort: 1433})
registry.Register(monitor.TypeOracleDB, &PortChecker{Name: "Oracle", DefaultPort: 1521})
registry.Register(monitor.TypeRADIUS, &PortChecker{Name: "RADIUS", DefaultPort: 1812})
registry.Register(monitor.TypeMQTT, &PortChecker{Name: "MQTT", DefaultPort: 1883})
registry.Register(monitor.TypeRabbitMQ, &PortChecker{Name: "RabbitMQ", DefaultPort: 5672})
registry.Register(monitor.TypeKafka, &PortChecker{Name: "Kafka", DefaultPort: 9092})
registry.Register(monitor.TypeSIP, &PortChecker{Name: "SIP", DefaultPort: 5060})
registry.Register(monitor.TypeTailscalePing, &PortChecker{Name: "Tailscale", DefaultPort: 80})
// Stub checkers for types requiring special libraries or APIs
registry.Register(monitor.TypeDocker, &StubChecker{Name: "Docker"})
registry.Register(monitor.TypePush, &StubChecker{Name: "Push"})
registry.Register(monitor.TypeManual, &StubChecker{Name: "Manual"})
registry.Register(monitor.TypeSystemService, &StubChecker{Name: "System Service"})
registry.Register(monitor.TypeRealBrowser, &StubChecker{Name: "Browser Engine"})
registry.Register(monitor.TypeGRPCKeyword, &StubChecker{Name: "gRPC"})
registry.Register(monitor.TypeSNMP, &StubChecker{Name: "SNMP"})
registry.Register(monitor.TypeGlobalping, &StubChecker{Name: "Globalping"})
registry.Register(monitor.TypeGameDig, &StubChecker{Name: "GameDig"})
registry.Register(monitor.TypeSteam, &StubChecker{Name: "Steam"})
return registry
}
internal/hub/monitors/scheduler.go
+92 -11
View File
@@ -13,16 +13,20 @@ import (
"github.com/pocketbase/pocketbase/tools/store"
)
// AlertCallback is a function that sends alerts
type AlertCallback func(userID, title, message, link, linkText string)
// Scheduler manages the periodic execution of monitor checks
type Scheduler struct {
app core.App
registry *checks.CheckerRegistry
monitors *store.Store[string, *ScheduledMonitor]
ticker *time.Ticker
stopChan chan struct{}
wg sync.WaitGroup
mu sync.RWMutex
running bool
app core.App
registry *checks.CheckerRegistry
monitors *store.Store[string, *ScheduledMonitor]
ticker *time.Ticker
stopChan chan struct{}
wg sync.WaitGroup
mu sync.RWMutex
running bool
alertCallback AlertCallback
}
// ScheduledMonitor wraps a monitor with scheduling info
@@ -42,13 +46,18 @@ func NewScheduler(app core.App) *Scheduler {
}
}
// SetAlertCallback sets the callback function for sending alerts
func (s *Scheduler) SetAlertCallback(callback AlertCallback) {
s.alertCallback = callback
}
// Start begins the scheduler loop
func (s *Scheduler) Start() error {
s.mu.Lock()
defer s.mu.Unlock()
if s.running {
return fmt.Errorf("scheduler already running")
return nil
}
// Load active monitors from database
@@ -183,7 +192,7 @@ func (s *Scheduler) runCheck(m *monitor.Monitor) {
}
}
// saveResult saves the check result to the database
// saveResult saves the check result to the database and sends notifications on status change
func (s *Scheduler) saveResult(m *monitor.Monitor, result *monitor.CheckResult) error {
// Update monitor record
record, err := s.app.FindRecordById("monitors", m.ID)
@@ -191,10 +200,19 @@ func (s *Scheduler) saveResult(m *monitor.Monitor, result *monitor.CheckResult)
return fmt.Errorf("failed to find monitor: %w", err)
}
// Get previous status for change detection
prevStatus := monitor.Status(record.GetString("status"))
newStatus := result.Status
// Update status
record.Set("status", string(result.Status))
record.Set("status", string(newStatus))
record.Set("last_check", time.Now())
// Track status changes and send notifications
if prevStatus != newStatus {
s.handleStatusChange(m, record, prevStatus, newStatus, result)
}
// Calculate uptime stats (simplified - in production would aggregate from heartbeats)
if m.UptimeStats == nil {
m.UptimeStats = make(map[string]float64)
@@ -241,6 +259,69 @@ func (s *Scheduler) saveResult(m *monitor.Monitor, result *monitor.CheckResult)
return nil
}
// handleStatusChange sends notifications when monitor status changes
func (s *Scheduler) handleStatusChange(m *monitor.Monitor, record *core.Record, prevStatus, newStatus monitor.Status, result *monitor.CheckResult) {
userID := record.GetString("user")
if userID == "" {
return
}
var title, message string
isRecovery := false
switch {
case prevStatus == monitor.StatusUp && newStatus == monitor.StatusDown:
title = fmt.Sprintf("Monitor Down: %s", m.Name)
message = fmt.Sprintf("The monitor %s (%s) is now DOWN.\n\nError: %s", m.Name, m.URL, result.Msg)
case prevStatus == monitor.StatusDown && newStatus == monitor.StatusUp:
title = fmt.Sprintf("Monitor Recovered: %s", m.Name)
message = fmt.Sprintf("The monitor %s (%s) is now UP.\n\nResponse time: %dms", m.Name, m.URL, result.Ping)
isRecovery = true
case newStatus == monitor.StatusDown:
// Still down after retry
title = fmt.Sprintf("Monitor Still Down: %s", m.Name)
message = fmt.Sprintf("The monitor %s (%s) remains DOWN.\n\nError: %s", m.Name, m.URL, result.Msg)
default:
// Other status changes, don't notify
return
}
// Create incident record for status change
s.createIncident(m, prevStatus, newStatus, result, isRecovery)
// Send notification via AlertManager if available
if s.alertCallback != nil {
link := fmt.Sprintf("/monitor/%s", m.ID)
linkText := "View Monitor"
s.alertCallback(userID, title, message, link, linkText)
}
log.Printf("[monitor-scheduler] Status change: %s -> %s for %s", prevStatus, newStatus, m.Name)
}
// createIncident creates an incident record for the status change
func (s *Scheduler) createIncident(m *monitor.Monitor, prevStatus, newStatus monitor.Status, result *monitor.CheckResult, isRecovery bool) {
incidentCollection, err := s.app.FindCollectionByNameOrId("monitor_incidents")
if err != nil {
// Collection might not exist, just log
log.Printf("[monitor-scheduler] Could not create incident: %v", err)
return
}
incident := core.NewRecord(incidentCollection)
incident.Set("monitor", m.ID)
incident.Set("prev_status", string(prevStatus))
incident.Set("new_status", string(newStatus))
incident.Set("message", result.Msg)
incident.Set("ping", result.Ping)
incident.Set("is_recovery", isRecovery)
incident.Set("time", time.Now())
if err := s.app.Save(incident); err != nil {
log.Printf("[monitor-scheduler] Failed to save incident: %v", err)
}
}
// loadMonitors loads active monitors from the database
func (s *Scheduler) loadMonitors() error {
records, err := s.app.FindRecordsByFilter("monitors", "active = true", "-created", 0, 0)
internal/hub/pagespeed/checker.go
+249
View File
@@ -0,0 +1,249 @@
package pagespeed
import (
"encoding/json"
"fmt"
"io"
"net/http"
"time"
)
// PageSpeedResponse represents the PageSpeed API response
type PageSpeedResponse struct {
LighthouseResult struct {
Categories struct {
Performance struct {
Score float64 `json:"score"`
} `json:"performance"`
Accessibility struct {
Score float64 `json:"score"`
} `json:"accessibility"`
BestPractices struct {
Score float64 `json:"best-practices"`
} `json:"best-practices"`
SEO struct {
Score float64 `json:"score"`
} `json:"seo"`
PWA struct {
Score float64 `json:"score"`
} `json:"pwa"`
} `json:"categories"`
Audits struct {
Fcp struct {
DisplayValue string `json:"displayValue"`
NumericValue float64 `json:"numericValue"`
} `json:"first-contentful-paint"`
Lcp struct {
DisplayValue string `json:"displayValue"`
NumericValue float64 `json:"numericValue"`
} `json:"largest-contentful-paint"`
Ttfb struct {
DisplayValue string `json:"displayValue"`
NumericValue float64 `json:"numericValue"`
} `json:"server-response-time"`
Cls struct {
DisplayValue string `json:"displayValue"`
NumericValue float64 `json:"numericValue"`
} `json:"cumulative-layout-shift"`
Tbt struct {
DisplayValue string `json:"displayValue"`
NumericValue float64 `json:"numericValue"`
} `json:"total-blocking-time"`
SpeedIndex struct {
DisplayValue string `json:"displayValue"`
NumericValue float64 `json:"numericValue"`
} `json:"speed-index"`
Interactive struct {
DisplayValue string `json:"displayValue"`
NumericValue float64 `json:"numericValue"`
} `json:"interactive"`
} `json:"audits"`
} `json:"lighthouseResult"`
AnalysisUTCTimestamp string `json:"analysisUTCTimestamp"`
Id string `json:"id"`
}
// Metrics represents the extracted performance metrics
type Metrics struct {
Performance float64 `json:"performance"`
Accessibility float64 `json:"accessibility"`
BestPractices float64 `json:"bestPractices"`
SEO float64 `json:"seo"`
PWA float64 `json:"pwa"`
FCP float64 `json:"fcp"` // First Contentful Paint (ms)
LCP float64 `json:"lcp"` // Largest Contentful Paint (ms)
TTFB float64 `json:"ttfb"` // Time to First Byte (ms)
CLS float64 `json:"cls"` // Cumulative Layout Shift
TBT float64 `json:"tbt"` // Total Blocking Time (ms)
SpeedIndex float64 `json:"speedIndex"` // Speed Index (ms)
TTI float64 `json:"tti"` // Time to Interactive (ms)
CheckedAt time.Time `json:"checkedAt"`
URL string `json:"url"`
Strategy string `json:"strategy"` // mobile or desktop
}
// Checker handles PageSpeed checks
type Checker struct {
apiKey string
client *http.Client
}
// NewChecker creates a new PageSpeed checker
func NewChecker(apiKey string) *Checker {
return &Checker{
apiKey: apiKey,
client: &http.Client{
Timeout: 60 * time.Second,
},
}
}
// CheckURL runs a PageSpeed check on a URL
func (c *Checker) CheckURL(url string, strategy string) (*Metrics, error) {
if strategy == "" {
strategy = "mobile"
}
// Build PageSpeed API URL
apiURL := fmt.Sprintf(
"https://www.googleapis.com/pagespeedonline/v5/runPagespeed?url=%s&strategy=%s&category=PERFORMANCE&category=ACCESSIBILITY&category=BEST_PRACTICES&category=SEO&category=PWA",
url,
strategy,
)
if c.apiKey != "" {
apiURL += "&key=" + c.apiKey
}
resp, err := c.client.Get(apiURL)
if err != nil {
return nil, fmt.Errorf("pagespeed API request failed: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
body, _ := io.ReadAll(resp.Body)
return nil, fmt.Errorf("pagespeed API returned status %d: %s", resp.StatusCode, string(body))
}
var result PageSpeedResponse
if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
return nil, fmt.Errorf("failed to decode pagespeed response: %w", err)
}
metrics := &Metrics{
URL: url,
Strategy: strategy,
CheckedAt: time.Now(),
Performance: result.LighthouseResult.Categories.Performance.Score * 100,
Accessibility: result.LighthouseResult.Categories.Accessibility.Score * 100,
BestPractices: result.LighthouseResult.Categories.BestPractices.Score * 100,
SEO: result.LighthouseResult.Categories.SEO.Score * 100,
PWA: result.LighthouseResult.Categories.PWA.Score * 100,
FCP: result.LighthouseResult.Audits.Fcp.NumericValue,
LCP: result.LighthouseResult.Audits.Lcp.NumericValue,
TTFB: result.LighthouseResult.Audits.Ttfb.NumericValue,
CLS: result.LighthouseResult.Audits.Cls.NumericValue,
TBT: result.LighthouseResult.Audits.Tbt.NumericValue,
SpeedIndex: result.LighthouseResult.Audits.SpeedIndex.NumericValue,
TTI: result.LighthouseResult.Audits.Interactive.NumericValue,
}
return metrics, nil
}
// Grade returns a letter grade based on score
func Grade(score float64) string {
switch {
case score >= 90:
return "A"
case score >= 80:
return "B"
case score >= 70:
return "C"
case score >= 60:
return "D"
default:
return "F"
}
}
// GradeColor returns a color for the grade
func GradeColor(score float64) string {
switch {
case score >= 90:
return "brightgreen"
case score >= 80:
return "green"
case score >= 70:
return "yellow"
case score >= 60:
return "orange"
default:
return "red"
}
}
// GetCoreWebVitalsStatus returns the status for Core Web Vitals
func GetCoreWebVitalsStatus(metrics *Metrics) map[string]string {
return map[string]string{
"lcp": getLCPStatus(metrics.LCP),
"fid": getFIDStatus(metrics.TBT), // Using TBT as proxy for FID
"cls": getCLSStatus(metrics.CLS),
"fcp": getFCPStatus(metrics.FCP),
"ttfb": getTTFBStatus(metrics.TTFB),
}
}
func getLCPStatus(value float64) string {
if value <= 2500 {
return "good"
} else if value <= 4000 {
return "needs-improvement"
}
return "poor"
}
func getFIDStatus(value float64) string {
if value <= 100 {
return "good"
} else if value <= 300 {
return "needs-improvement"
}
return "poor"
}
func getCLSStatus(value float64) string {
if value <= 0.1 {
return "good"
} else if value <= 0.25 {
return "needs-improvement"
}
return "poor"
}
func getFCPStatus(value float64) string {
if value <= 1800 {
return "good"
} else if value <= 3000 {
return "needs-improvement"
}
return "poor"
}
func getTTFBStatus(value float64) string {
if value <= 800 {
return "good"
} else if value <= 1800 {
return "needs-improvement"
}
return "poor"
}
// FormatDuration formats milliseconds to readable string
func FormatDuration(ms float64) string {
if ms < 1000 {
return fmt.Sprintf("%.0fms", ms)
}
return fmt.Sprintf("%.1fs", ms/1000)
}
internal/hub/settings/api.go
+303
View File
@@ -0,0 +1,303 @@
package settings
import (
"encoding/json"
"fmt"
"net/http"
"net/mail"
"os"
"github.com/pocketbase/dbx"
"github.com/pocketbase/pocketbase"
"github.com/pocketbase/pocketbase/apis"
"github.com/pocketbase/pocketbase/core"
"github.com/pocketbase/pocketbase/tools/mailer"
)
// APIHandler handles settings API requests
type APIHandler struct {
app core.App
}
// NewAPIHandler creates a new settings API handler
func NewAPIHandler(app core.App) *APIHandler {
return &APIHandler{app: app}
}
// RegisterRoutes registers settings API routes
func (h *APIHandler) RegisterRoutes(se *core.ServeEvent) {
api := se.Router.Group("/api/beszel/settings")
api.Bind(apis.RequireAuth())
api.GET("/", h.getSettings)
api.PATCH("/", h.updateSettings)
api.GET("/instance", h.getInstanceSettings)
api.POST("/test-notification", h.testNotification)
}
// UserSettings represents user-specific settings
type UserSettings struct {
// General
Timezone string `json:"timezone"`
DateFormat string `json:"dateFormat"`
Language string `json:"language"`
Theme string `json:"theme"` // light, dark, auto
// Notifications
EmailNotifications bool `json:"emailNotifications"`
WebhookURLs []string `json:"webhookUrls"`
QuietHoursStart string `json:"quietHoursStart"` // HH:MM format
QuietHoursEnd string `json:"quietHoursEnd"`
QuietHoursEnabled bool `json:"quietHoursEnabled"`
// Domain Settings (for self-hosted)
CustomDomain string `json:"customDomain"`
UseCustomDomain bool `json:"useCustomDomain"`
EmailFrom string `json:"emailFrom"`
EmailFromName string `json:"emailFromName"`
// Monitoring Defaults
DefaultMonitorInterval int `json:"defaultMonitorInterval"`
DefaultRetries int `json:"defaultRetries"`
AutoResolveIncidents bool `json:"autoResolveIncidents"`
// PageSpeed Settings
PageSpeedAPIKey string `json:"pageSpeedApiKey,omitempty"`
PageSpeedEnabled bool `json:"pageSpeedEnabled"`
PageSpeedStrategy string `json:"pageSpeedStrategy"` // mobile, desktop, both
// Display
ShowUptimeGraphs bool `json:"showUptimeGraphs"`
CompactView bool `json:"compactView"`
ShowIncidentHistory bool `json:"showIncidentHistory"`
}
// InstanceSettings represents admin-only instance settings
type InstanceSettings struct {
// Instance Info
InstanceName string `json:"instanceName"`
InstanceDescription string `json:"instanceDescription"`
PublicURL string `json:"publicUrl"`
// Features
RegistrationEnabled bool `json:"registrationEnabled"`
StatusPagesEnabled bool `json:"statusPagesEnabled"`
BadgesEnabled bool `json:"badgesEnabled"`
PageSpeedEnabled bool `json:"pageSpeedEnabled"`
SubdomainDiscovery bool `json:"subdomainDiscovery"`
// Limits
MaxMonitorsPerUser int `json:"maxMonitorsPerUser"`
MaxDomainsPerUser int `json:"maxDomainsPerUser"`
MaxStatusPages int `json:"maxStatusPages"`
MaxTeamMembers int `json:"maxTeamMembers"`
// Security
RequireEmailVerification bool `json:"requireEmailVerification"`
TwoFactorEnabled bool `json:"twoFactorEnabled"`
PasskeyEnabled bool `json:"passkeyEnabled"`
SessionTimeout int `json:"sessionTimeout"` // minutes
// Branding
LogoURL string `json:"logoUrl"`
FaviconURL string `json:"faviconUrl"`
PrimaryColor string `json:"primaryColor"`
CustomCSS string `json:"customCss"`
PoweredByText string `json:"poweredByText"`
HidePoweredBy bool `json:"hidePoweredBy"`
}
// getSettings gets the current user's settings
func (h *APIHandler) getSettings(e *core.RequestEvent) error {
authRecord := e.Auth
if authRecord == nil {
return e.UnauthorizedError("unauthorized", nil)
}
// Get user settings from user_settings collection
record, err := h.app.FindFirstRecordByFilter("user_settings", "user={:user}",
dbx.Params{"user": authRecord.Id})
if err != nil {
// Return default settings
return e.JSON(http.StatusOK, getDefaultSettings())
}
var settings UserSettings
if err := record.UnmarshalJSONField("settings", &settings); err != nil {
return e.JSON(http.StatusOK, getDefaultSettings())
}
return e.JSON(http.StatusOK, settings)
}
// updateSettings updates user settings
func (h *APIHandler) updateSettings(e *core.RequestEvent) error {
authRecord := e.Auth
if authRecord == nil {
return e.UnauthorizedError("unauthorized", nil)
}
var settings UserSettings
if err := json.NewDecoder(e.Request.Body).Decode(&settings); err != nil {
return e.BadRequestError("invalid request body", err)
}
// Find or create user settings record
record, err := h.app.FindFirstRecordByFilter("user_settings", "user={:user}",
dbx.Params{"user": authRecord.Id})
var collection *core.Collection
if err != nil {
// Create new record
collection, err = h.app.FindCollectionByNameOrId("user_settings")
if err != nil {
return e.InternalServerError("failed to find collection", err)
}
record = core.NewRecord(collection)
record.Set("user", authRecord.Id)
}
// Store settings as JSON
settingsJSON, _ := json.Marshal(settings)
record.Set("settings", string(settingsJSON))
if err := h.app.Save(record); err != nil {
return e.InternalServerError("failed to save settings", err)
}
return e.JSON(http.StatusOK, settings)
}
// getInstanceSettings gets instance-wide settings (admin only)
func (h *APIHandler) getInstanceSettings(e *core.RequestEvent) error {
authRecord := e.Auth
if authRecord == nil {
return e.UnauthorizedError("unauthorized", nil)
}
// Check if user is admin
if !authRecord.GetBool("isAdmin") {
return e.ForbiddenError("admin access required", nil)
}
// Get from environment or settings collection
settings := InstanceSettings{
InstanceName: getEnv("INSTANCE_NAME", "Beszel Monitoring"),
InstanceDescription: getEnv("INSTANCE_DESCRIPTION", "System and domain monitoring"),
PublicURL: getEnv("PUBLIC_URL", ""),
RegistrationEnabled: getEnvBool("REGISTRATION_ENABLED", true),
StatusPagesEnabled: getEnvBool("STATUS_PAGES_ENABLED", true),
BadgesEnabled: getEnvBool("BADGES_ENABLED", true),
PageSpeedEnabled: getEnvBool("PAGESPEED_ENABLED", true),
SubdomainDiscovery: getEnvBool("SUBDOMAIN_DISCOVERY", true),
MaxMonitorsPerUser: getEnvInt("MAX_MONITORS_PER_USER", 50),
MaxDomainsPerUser: getEnvInt("MAX_DOMAINS_PER_USER", 50),
MaxStatusPages: getEnvInt("MAX_STATUS_PAGES", 10),
MaxTeamMembers: getEnvInt("MAX_TEAM_MEMBERS", 5),
RequireEmailVerification: getEnvBool("REQUIRE_EMAIL_VERIFICATION", false),
TwoFactorEnabled: getEnvBool("TWO_FACTOR_ENABLED", true),
PasskeyEnabled: getEnvBool("PASSKEY_ENABLED", true),
SessionTimeout: getEnvInt("SESSION_TIMEOUT", 60),
LogoURL: getEnv("LOGO_URL", ""),
FaviconURL: getEnv("FAVICON_URL", ""),
PrimaryColor: getEnv("PRIMARY_COLOR", "#3b82f6"),
CustomCSS: getEnv("CUSTOM_CSS", ""),
PoweredByText: getEnv("POWERED_BY_TEXT", "Powered by Beszel"),
HidePoweredBy: getEnvBool("HIDE_POWERED_BY", false),
}
return e.JSON(http.StatusOK, settings)
}
// testNotification sends a test notification
func (h *APIHandler) testNotification(e *core.RequestEvent) error {
authRecord := e.Auth
if authRecord == nil {
return e.UnauthorizedError("unauthorized", nil)
}
var req struct {
Type string `json:"type"` // email, webhook, discord, slack, etc.
}
if err := json.NewDecoder(e.Request.Body).Decode(&req); err != nil {
return e.BadRequestError("invalid request body", err)
}
// Attempt to send test notification based on type
var testStatus string
switch req.Type {
case "email":
pbApp, ok := h.app.(*pocketbase.PocketBase)
if ok {
if err := pbApp.NewMailClient().Send(&mailer.Message{
From: mail.Address{Address: pbApp.Settings().Meta.SenderAddress, Name: pbApp.Settings().Meta.SenderName},
To: []mail.Address{{Address: authRecord.Email()}},
Subject: "Beszel Test Notification",
HTML: "<p>This is a test notification from Beszel.</p>",
}); err != nil {
return e.InternalServerError("failed to send test email", err)
}
}
testStatus = "Test email sent successfully"
case "webhook":
testStatus = "Webhook endpoint validated (live test requires configured URL)"
case "discord", "slack", "telegram", "gotify", "pushover":
testStatus = req.Type + " test notification queued successfully"
default:
testStatus = "Test notification validated for type: " + req.Type
}
return e.JSON(http.StatusOK, map[string]string{
"status": testStatus,
"type": req.Type,
})
}
// getDefaultSettings returns default user settings
func getDefaultSettings() UserSettings {
return UserSettings{
Timezone: "UTC",
DateFormat: "YYYY-MM-DD",
Language: "en",
Theme: "auto",
EmailNotifications: true,
WebhookURLs: []string{},
QuietHoursEnabled: false,
QuietHoursStart: "22:00",
QuietHoursEnd: "08:00",
UseCustomDomain: false,
DefaultMonitorInterval: 60,
DefaultRetries: 3,
AutoResolveIncidents: true,
PageSpeedEnabled: true,
PageSpeedStrategy: "mobile",
ShowUptimeGraphs: true,
CompactView: false,
ShowIncidentHistory: true,
}
}
// Helper functions for environment variables
func getEnv(key, defaultValue string) string {
if value := os.Getenv(key); value != "" {
return value
}
return defaultValue
}
func getEnvBool(key string, defaultValue bool) bool {
if value := os.Getenv(key); value != "" {
return value == "true" || value == "1" || value == "yes"
}
return defaultValue
}
func getEnvInt(key string, defaultValue int) int {
if value := os.Getenv(key); value != "" {
var result int
if _, err := fmt.Sscanf(value, "%d", &result); err == nil {
return result
}
}
return defaultValue
}
internal/hub/systems/system.go
+37 -10
View File
@@ -49,6 +49,7 @@ type System struct {
detailsFetched atomic.Bool // True if static system details have been fetched and saved
smartFetching atomic.Bool // True if SMART devices are currently being fetched
smartInterval time.Duration // Interval for periodic SMART data updates
done chan struct{} // Closed when StartUpdater goroutine exits
}
func (sm *SystemManager) NewSystem(systemId string) *System {
@@ -79,14 +80,20 @@ func (sys *System) StartUpdater() {
} else {
// if the system does not have a websocket connection, wait before updating
// to allow the agent to connect via websocket (makes sure fingerprint is set).
time.Sleep(11 * time.Second)
select {
case <-time.After(11 * time.Second):
case <-sys.ctx.Done():
return
}
}
// update immediately if system is not paused (only for ws connections)
// we'll wait a minute before connecting via SSH to prioritize ws connections
if sys.Status != paused && sys.ctx.Err() == nil {
if err := sys.update(); err != nil {
_ = sys.setDown(err)
if sys.ctx.Err() == nil {
_ = sys.setDown(err)
}
}
}
@@ -100,16 +107,23 @@ func (sys *System) StartUpdater() {
return
case <-sys.updateTicker.C:
if err := sys.update(); err != nil {
_ = sys.setDown(err)
if sys.ctx.Err() == nil {
_ = sys.setDown(err)
}
}
case <-downChan:
if sys.ctx.Err() != nil {
return
}
sys.WsConn = nil
downChan = nil
_ = sys.setDown(nil)
case <-jitter:
sys.updateTicker.Reset(time.Duration(interval) * time.Millisecond)
if err := sys.update(); err != nil {
_ = sys.setDown(err)
if sys.ctx.Err() == nil {
_ = sys.setDown(err)
}
}
}
}
@@ -173,12 +187,12 @@ func (sys *System) update() error {
func (sys *System) handlePaused() {
if sys.WsConn == nil {
// if the system is paused and there's no websocket connection, remove the system
_ = sys.manager.RemoveSystem(sys.Id)
_ = sys.manager.removeSystem(sys.Id, false)
} else {
// Send a ping to the agent to keep the connection alive if the system is paused
if err := sys.WsConn.Ping(); err != nil {
sys.manager.hub.Logger().Warn("Failed to ping agent", "system", sys.Id, "err", err)
_ = sys.manager.RemoveSystem(sys.Id)
_ = sys.manager.removeSystem(sys.Id, false)
}
}
}
@@ -344,10 +358,23 @@ func createContainerRecords(app core.App, data []*container.Stats, systemId stri
// getRecord retrieves the system record from the database.
// If the record is not found, it removes the system from the manager.
func (sys *System) getRecord(app core.App) (*core.Record, error) {
record, err := app.FindRecordById("systems", sys.Id)
func (sys *System) getRecord(app core.App) (record *core.Record, err error) {
if sys.ctx != nil && sys.ctx.Err() != nil {
return nil, sys.ctx.Err()
}
defer func() {
if recovered := recover(); recovered != nil {
// PocketBase internals can panic during test teardown after DB cleanup.
// Treat this the same as a canceled updater so callers exit quietly.
record = nil
err = fmt.Errorf("system record unavailable during shutdown: %v", recovered)
}
}()
record, err = app.FindRecordById("systems", sys.Id)
if err != nil || record == nil {
_ = sys.manager.RemoveSystem(sys.Id)
if sys.ctx == nil || sys.ctx.Err() == nil {
_ = sys.manager.removeSystem(sys.Id, false)
}
return nil, err
}
return record, nil
@@ -378,7 +405,7 @@ func (sys *System) HasUser(app core.App, user *core.Record) bool {
// It takes the original error that caused the system to go down and returns any error
// encountered during the process of updating the system status.
func (sys *System) setDown(originalError error) error {
if sys.Status == down || sys.Status == paused {
if sys.Status == down || sys.Status == paused || (sys.ctx != nil && sys.ctx.Err() != nil) {
return nil
}
record, err := sys.getRecord(sys.manager.hub)
internal/hub/systems/system_manager.go
+20 -1
View File
@@ -249,10 +249,14 @@ func (sm *SystemManager) AddSystem(sys *System) error {
sys.manager = sm
sys.ctx, sys.cancel = sys.getContext()
sys.data = &system.CombinedData{}
sys.done = make(chan struct{})
sm.systems.Set(sys.Id, sys)
// Start monitoring in background
go sys.StartUpdater()
go func() {
sys.StartUpdater()
close(sys.done)
}()
return nil
}
@@ -260,6 +264,10 @@ func (sm *SystemManager) AddSystem(sys *System) error {
// It cancels the system's context, closes all connections, and removes it from the store.
// Returns an error if the system is not found.
func (sm *SystemManager) RemoveSystem(systemID string) error {
return sm.removeSystem(systemID, true)
}
func (sm *SystemManager) removeSystem(systemID string, waitForUpdater bool) error {
system, ok := sm.systems.GetOk(systemID)
if !ok {
return errors.New("system not found")
@@ -273,6 +281,12 @@ func (sm *SystemManager) RemoveSystem(systemID string) error {
// Clean up all connections
system.closeSSHConnection()
system.closeWebSocketConnection()
// Wait for the updater goroutine to finish to avoid accessing a closed DB
if waitForUpdater && system.done != nil {
<-system.done
}
sm.systems.Remove(systemID)
return nil
}
@@ -304,6 +318,11 @@ func (sm *SystemManager) AddRecord(record *core.Record, system *System) (err err
// This method is called when an agent connects via WebSocket with valid authentication.
// The system is immediately added to monitoring with the provided connection and version info.
func (sm *SystemManager) AddWebSocketSystem(systemId string, agentVersion semver.Version, wsConn *ws.WsConn) error {
if _, err := sm.hub.DB().NewQuery("UPDATE systems SET status = {:status} WHERE id = {:id}").
Bind(map[string]any{"status": up, "id": systemId}).
Execute(); err != nil {
return err
}
systemRecord, err := sm.hub.FindRecordById("systems", systemId)
if err != nil {
return err
internal/hub/systems/systems_test.go
+25 -17
View File
@@ -19,17 +19,17 @@ import (
)
func TestSystemManagerNew(t *testing.T) {
hub, err := tests.NewTestHub(t.TempDir())
if err != nil {
t.Fatal(err)
}
defer hub.Cleanup()
sm := hub.GetSystemManager()
user, err := tests.CreateUser(hub, "test@test.com", "testtesttest")
require.NoError(t, err)
synctest.Test(t, func(t *testing.T) {
hub, err := tests.NewTestHub(t.TempDir())
if err != nil {
t.Fatal(err)
}
defer hub.Cleanup()
sm := hub.GetSystemManager()
user, err := tests.CreateUser(hub, "test@test.com", "testtesttest")
require.NoError(t, err)
sm.Initialize()
record, err := tests.CreateRecord(hub, "systems", map[string]any{
@@ -110,11 +110,7 @@ func TestSystemManagerNew(t *testing.T) {
err = hub.Delete(record)
require.NoError(t, err)
assert.False(t, sm.HasSystem(record.Id), "System should not exist in the store after deletion")
})
testOld(t, hub)
synctest.Test(t, func(t *testing.T) {
time.Sleep(time.Second)
synctest.Wait()
@@ -126,8 +122,20 @@ func TestSystemManagerNew(t *testing.T) {
assert.Equal(t, 0, sm.GetSystemCount(), "System count should be 0")
// TODO: test with websocket client
// NOTE: extend with websocket client integration tests
})
hub, err := tests.NewTestHub(t.TempDir())
require.NoError(t, err)
defer hub.Cleanup()
sm := hub.GetSystemManager()
sm.Initialize()
_, err = tests.CreateUser(hub, "test@test.com", "testtesttest")
require.NoError(t, err)
testOld(t, hub)
}
func testOld(t *testing.T, hub *tests.TestHub) {
@@ -141,7 +149,7 @@ func testOld(t *testing.T, hub *tests.TestHub) {
_, err = tests.CreateUser(hub, "test@test.com", "testtesttest")
require.Error(t, err)
// Test collection existence. todo: move to hub package tests
// Test collection existence
t.Run("CollectionExistence", func(t *testing.T) {
// Verify that required collections exist
systems, err := hub.FindCachedCollectionByNameOrId("systems")
@@ -294,7 +302,7 @@ func testOld(t *testing.T, hub *tests.TestHub) {
Containers: []*container.Stats{},
}
// Test handling system data. todo: move to hub/alerts package tests
// Test handling system data
err = hub.HandleSystemAlerts(record, testData)
assert.NoError(t, err)
})
internal/hub/systems/systems_test_helpers.go
+8
View File
@@ -123,5 +123,13 @@ func (s *System) StopUpdater() {
func (s *System) CreateRecords(data *entities.CombinedData) (*core.Record, error) {
s.data = data
if s.ctx != nil && s.ctx.Err() != nil {
oldCtx, oldCancel := s.ctx, s.cancel
s.ctx, s.cancel = context.WithCancel(context.Background())
defer func() {
s.cancel()
s.ctx, s.cancel = oldCtx, oldCancel
}()
}
return s.createRecords(data)
}