move application code into beszel folder

2025-10-20 02:09:28 +08:00 · 2024-08-11 13:41:57 -04:00
parent ea71492d13
commit 9da1e5751a
103 changed files with 84 additions and 7 deletions
--- a/beszel/internal/agent/agent.go
+++ b/beszel/internal/agent/agent.go
@@ -0,0 +1,458 @@
+package agent
+
+import (
+	"beszel/internal/entities/container"
+	"beszel/internal/entities/system"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"log"
+	"math"
+	"net"
+	"net/http"
+	"net/url"
+	"os"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/shirou/gopsutil/v4/cpu"
+	"github.com/shirou/gopsutil/v4/disk"
+	"github.com/shirou/gopsutil/v4/host"
+	"github.com/shirou/gopsutil/v4/mem"
+
+	sshServer "github.com/gliderlabs/ssh"
+	psutilNet "github.com/shirou/gopsutil/v4/net"
+)
+
+var containerStatsMap = make(map[string]*container.PrevContainerStats)
+
+type Agent struct {
+	port                string
+	pubKey              []byte
+	sem                 chan struct{}
+	containerStatsMutex *sync.Mutex
+	diskIoStats         system.DiskIoStats
+	netIoStats          system.NetIoStats
+}
+
+func NewAgent(pubKey []byte, port string) *Agent {
+	return &Agent{
+		pubKey:              pubKey,
+		sem:                 make(chan struct{}, 15),
+		port:                port,
+		containerStatsMutex: &sync.Mutex{},
+		diskIoStats:         system.DiskIoStats{},
+		netIoStats:          system.NetIoStats{},
+	}
+}
+
+func (a *Agent) acquireSemaphore() {
+	a.sem <- struct{}{}
+}
+
+func (a *Agent) releaseSemaphore() {
+	<-a.sem
+}
+
+// client for docker engine api
+var dockerClient = newDockerClient()
+
+func (a *Agent) getSystemStats() (*system.SystemInfo, *system.SystemStats) {
+	systemStats := &system.SystemStats{}
+
+	// cpu percent
+	cpuPct, err := cpu.Percent(0, false)
+	if err != nil {
+		log.Println("Error getting cpu percent:", err)
+	} else if len(cpuPct) > 0 {
+		systemStats.Cpu = twoDecimals(cpuPct[0])
+	}
+
+	// memory
+	if v, err := mem.VirtualMemory(); err == nil {
+		systemStats.Mem = bytesToGigabytes(v.Total)
+		systemStats.MemUsed = bytesToGigabytes(v.Used)
+		systemStats.MemBuffCache = bytesToGigabytes(v.Total - v.Free - v.Used)
+		systemStats.MemPct = twoDecimals(v.UsedPercent)
+		systemStats.Swap = bytesToGigabytes(v.SwapTotal)
+		systemStats.SwapUsed = bytesToGigabytes(v.SwapTotal - v.SwapFree)
+	}
+
+	// disk usage
+	if d, err := disk.Usage("/"); err == nil {
+		systemStats.Disk = bytesToGigabytes(d.Total)
+		systemStats.DiskUsed = bytesToGigabytes(d.Used)
+		systemStats.DiskPct = twoDecimals(d.UsedPercent)
+	}
+
+	// disk i/o
+	if io, err := disk.IOCounters(a.diskIoStats.Filesystem); err == nil {
+		for _, d := range io {
+			// add to systemStats
+			secondsElapsed := time.Since(a.diskIoStats.Time).Seconds()
+			readPerSecond := float64(d.ReadBytes-a.diskIoStats.Read) / secondsElapsed
+			systemStats.DiskRead = bytesToMegabytes(readPerSecond)
+			writePerSecond := float64(d.WriteBytes-a.diskIoStats.Write) / secondsElapsed
+			systemStats.DiskWrite = bytesToMegabytes(writePerSecond)
+			// update diskIoStats
+			a.diskIoStats.Time = time.Now()
+			a.diskIoStats.Read = d.ReadBytes
+			a.diskIoStats.Write = d.WriteBytes
+		}
+	}
+
+	// network stats
+	if netIO, err := psutilNet.IOCounters(true); err == nil {
+		bytesSent := uint64(0)
+		bytesRecv := uint64(0)
+		for _, v := range netIO {
+			if skipNetworkInterface(&v) {
+				continue
+			}
+			// log.Printf("%+v: %+v recv, %+v sent\n", v.Name, v.BytesRecv, v.BytesSent)
+			bytesSent += v.BytesSent
+			bytesRecv += v.BytesRecv
+		}
+		// add to systemStats
+		secondsElapsed := time.Since(a.netIoStats.Time).Seconds()
+		sentPerSecond := float64(bytesSent-a.netIoStats.BytesSent) / secondsElapsed
+		recvPerSecond := float64(bytesRecv-a.netIoStats.BytesRecv) / secondsElapsed
+		systemStats.NetworkSent = bytesToMegabytes(sentPerSecond)
+		systemStats.NetworkRecv = bytesToMegabytes(recvPerSecond)
+		// update netIoStats
+		a.netIoStats.BytesSent = bytesSent
+		a.netIoStats.BytesRecv = bytesRecv
+		a.netIoStats.Time = time.Now()
+	}
+
+	systemInfo := &system.SystemInfo{
+		Cpu:     systemStats.Cpu,
+		MemPct:  systemStats.MemPct,
+		DiskPct: systemStats.DiskPct,
+	}
+
+	// add host info
+	if info, err := host.Info(); err == nil {
+		systemInfo.Uptime = info.Uptime
+		// systemInfo.Os = info.OS
+	}
+	// add cpu stats
+	if info, err := cpu.Info(); err == nil && len(info) > 0 {
+		systemInfo.CpuModel = info[0].ModelName
+	}
+	if cores, err := cpu.Counts(false); err == nil {
+		systemInfo.Cores = cores
+	}
+	if threads, err := cpu.Counts(true); err == nil {
+		systemInfo.Threads = threads
+	}
+
+	return systemInfo, systemStats
+
+}
+
+func (a *Agent) getDockerStats() ([]*container.ContainerStats, error) {
+	resp, err := dockerClient.Get("http://localhost/containers/json")
+	if err != nil {
+		closeIdleConnections(err)
+		return []*container.ContainerStats{}, err
+	}
+	defer resp.Body.Close()
+
+	var containers []*container.Container
+	if err := json.NewDecoder(resp.Body).Decode(&containers); err != nil {
+		log.Printf("Error decoding containers: %+v\n", err)
+		return []*container.ContainerStats{}, err
+	}
+
+	containerStats := make([]*container.ContainerStats, 0, len(containers))
+
+	// store valid ids to clean up old container ids from map
+	validIds := make(map[string]struct{}, len(containers))
+
+	var wg sync.WaitGroup
+
+	for _, ctr := range containers {
+		ctr.IdShort = ctr.Id[:12]
+		validIds[ctr.IdShort] = struct{}{}
+		// check if container is less than 1 minute old (possible restart)
+		// note: can't use Created field because it's not updated on restart
+		if strings.HasSuffix(ctr.Status, "seconds") {
+			// if so, remove old container data
+			a.deleteContainerStatsSync(ctr.IdShort)
+		}
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			cstats, err := a.getContainerStats(ctr)
+			if err != nil {
+				// Check if the error is a network timeout
+				if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
+					// Close idle connections to prevent reuse of stale connections
+					closeIdleConnections(err)
+				} else {
+					// otherwise delete container from map
+					a.deleteContainerStatsSync(ctr.IdShort)
+				}
+				// retry once
+				cstats, err = a.getContainerStats(ctr)
+				if err != nil {
+					log.Printf("Error getting container stats: %+v\n", err)
+					return
+				}
+			}
+			containerStats = append(containerStats, cstats)
+		}()
+	}
+
+	wg.Wait()
+
+	for id := range containerStatsMap {
+		if _, exists := validIds[id]; !exists {
+			// log.Printf("Removing container cpu map entry: %+v\n", id)
+			delete(containerStatsMap, id)
+		}
+	}
+
+	return containerStats, nil
+}
+
+func (a *Agent) getContainerStats(ctr *container.Container) (*container.ContainerStats, error) {
+	// use semaphore to limit concurrency
+	a.acquireSemaphore()
+	defer a.releaseSemaphore()
+	resp, err := dockerClient.Get("http://localhost/containers/" + ctr.IdShort + "/stats?stream=0&one-shot=1")
+	if err != nil {
+		return &container.ContainerStats{}, err
+	}
+	defer resp.Body.Close()
+
+	var statsJson system.CStats
+	if err := json.NewDecoder(resp.Body).Decode(&statsJson); err != nil {
+		panic(err)
+	}
+
+	name := ctr.Names[0][1:]
+
+	// memory (https://docs.docker.com/reference/cli/docker/container/stats/)
+	memCache := statsJson.MemoryStats.Stats["inactive_file"]
+	if memCache == 0 {
+		memCache = statsJson.MemoryStats.Stats["cache"]
+	}
+	usedMemory := statsJson.MemoryStats.Usage - memCache
+
+	a.containerStatsMutex.Lock()
+	defer a.containerStatsMutex.Unlock()
+
+	// add empty values if they doesn't exist in map
+	stats, initialized := containerStatsMap[ctr.IdShort]
+	if !initialized {
+		stats = &container.PrevContainerStats{}
+		containerStatsMap[ctr.IdShort] = stats
+	}
+
+	// cpu
+	cpuDelta := statsJson.CPUStats.CPUUsage.TotalUsage - stats.Cpu[0]
+	systemDelta := statsJson.CPUStats.SystemUsage - stats.Cpu[1]
+	cpuPct := float64(cpuDelta) / float64(systemDelta) * 100
+	if cpuPct > 100 {
+		return &container.ContainerStats{}, fmt.Errorf("%s cpu pct greater than 100: %+v", name, cpuPct)
+	}
+	stats.Cpu = [2]uint64{statsJson.CPUStats.CPUUsage.TotalUsage, statsJson.CPUStats.SystemUsage}
+
+	// network
+	var total_sent, total_recv uint64
+	for _, v := range statsJson.Networks {
+		total_sent += v.TxBytes
+		total_recv += v.RxBytes
+	}
+	var sent_delta, recv_delta float64
+	// prevent first run from sending all prev sent/recv bytes
+	if initialized {
+		secondsElapsed := time.Since(stats.Net.Time).Seconds()
+		sent_delta = float64(total_sent-stats.Net.Sent) / secondsElapsed
+		recv_delta = float64(total_recv-stats.Net.Recv) / secondsElapsed
+		// log.Printf("sent delta: %+v, recv delta: %+v\n", sent_delta, recv_delta)
+	}
+	stats.Net.Sent = total_sent
+	stats.Net.Recv = total_recv
+	stats.Net.Time = time.Now()
+
+	cStats := &container.ContainerStats{
+		Name:        name,
+		Cpu:         twoDecimals(cpuPct),
+		Mem:         bytesToMegabytes(float64(usedMemory)),
+		NetworkSent: bytesToMegabytes(sent_delta),
+		NetworkRecv: bytesToMegabytes(recv_delta),
+	}
+	return cStats, nil
+}
+
+// delete container stats from map using mutex
+func (a *Agent) deleteContainerStatsSync(id string) {
+	a.containerStatsMutex.Lock()
+	defer a.containerStatsMutex.Unlock()
+	delete(containerStatsMap, id)
+}
+
+func (a *Agent) gatherStats() *system.SystemData {
+	systemInfo, systemStats := a.getSystemStats()
+	stats := &system.SystemData{
+		Stats:      systemStats,
+		Info:       systemInfo,
+		Containers: []*container.ContainerStats{},
+	}
+	containerStats, err := a.getDockerStats()
+	if err == nil {
+		stats.Containers = containerStats
+	}
+	// fmt.Printf("%+v\n", stats)
+	return stats
+}
+
+func (a *Agent) startServer(addr string, pubKey []byte) {
+	sshServer.Handle(func(s sshServer.Session) {
+		stats := a.gatherStats()
+		var jsonStats []byte
+		jsonStats, _ = json.Marshal(stats)
+		io.WriteString(s, string(jsonStats))
+		s.Exit(0)
+	})
+
+	log.Printf("Starting SSH server on %s", addr)
+	if err := sshServer.ListenAndServe(addr, nil, sshServer.NoPty(),
+		sshServer.PublicKeyAuth(func(ctx sshServer.Context, key sshServer.PublicKey) bool {
+			data := []byte(pubKey)
+			allowed, _, _, _, _ := sshServer.ParseAuthorizedKey(data)
+			return sshServer.KeysEqual(key, allowed)
+		}),
+	); err != nil {
+		log.Fatal(err)
+	}
+}
+
+func (a *Agent) Run() {
+
+	if filesystem, exists := os.LookupEnv("FILESYSTEM"); exists {
+		a.diskIoStats.Filesystem = filesystem
+	} else {
+		a.diskIoStats.Filesystem = findDefaultFilesystem()
+	}
+
+	a.initializeDiskIoStats()
+	a.initializeNetIoStats()
+
+	a.startServer(a.port, a.pubKey)
+}
+
+func bytesToMegabytes(b float64) float64 {
+	return twoDecimals(b / 1048576)
+}
+
+func bytesToGigabytes(b uint64) float64 {
+	return twoDecimals(float64(b) / 1073741824)
+}
+
+func twoDecimals(value float64) float64 {
+	return math.Round(value*100) / 100
+}
+
+func findDefaultFilesystem() string {
+	if partitions, err := disk.Partitions(false); err == nil {
+		for _, v := range partitions {
+			if v.Mountpoint == "/" {
+				log.Printf("Using filesystem: %+v\n", v.Device)
+				return v.Device
+			}
+		}
+	}
+	return ""
+}
+
+func skipNetworkInterface(v *psutilNet.IOCountersStat) bool {
+	switch {
+	case strings.HasPrefix(v.Name, "lo"),
+		strings.HasPrefix(v.Name, "docker"),
+		strings.HasPrefix(v.Name, "br-"),
+		strings.HasPrefix(v.Name, "veth"),
+		v.BytesRecv == 0,
+		v.BytesSent == 0:
+		return true
+	default:
+		return false
+	}
+}
+
+func (a *Agent) initializeDiskIoStats() {
+	if io, err := disk.IOCounters(a.diskIoStats.Filesystem); err == nil {
+		for _, d := range io {
+			a.diskIoStats.Time = time.Now()
+			a.diskIoStats.Read = d.ReadBytes
+			a.diskIoStats.Write = d.WriteBytes
+		}
+	}
+}
+
+func (a *Agent) initializeNetIoStats() {
+	if netIO, err := psutilNet.IOCounters(true); err == nil {
+		bytesSent := uint64(0)
+		bytesRecv := uint64(0)
+		for _, v := range netIO {
+			if skipNetworkInterface(&v) {
+				continue
+			}
+			log.Printf("Found network interface: %+v (%+v recv, %+v sent)\n", v.Name, v.BytesRecv, v.BytesSent)
+			bytesSent += v.BytesSent
+			bytesRecv += v.BytesRecv
+		}
+		a.netIoStats.BytesSent = bytesSent
+		a.netIoStats.BytesRecv = bytesRecv
+		a.netIoStats.Time = time.Now()
+	}
+}
+
+func newDockerClient() *http.Client {
+	dockerHost := "unix:///var/run/docker.sock"
+	if dockerHostEnv, exists := os.LookupEnv("DOCKER_HOST"); exists {
+		dockerHost = dockerHostEnv
+	}
+
+	parsedURL, err := url.Parse(dockerHost)
+	if err != nil {
+		log.Fatal("Error parsing DOCKER_HOST: " + err.Error())
+	}
+
+	transport := &http.Transport{
+		ForceAttemptHTTP2:   false,
+		IdleConnTimeout:     90 * time.Second,
+		DisableCompression:  true,
+		MaxIdleConnsPerHost: 20,
+		DisableKeepAlives:   false,
+	}
+
+	switch parsedURL.Scheme {
+	case "unix":
+		transport.DialContext = func(ctx context.Context, proto, addr string) (net.Conn, error) {
+			return (&net.Dialer{}).DialContext(ctx, "unix", parsedURL.Path)
+		}
+	case "tcp", "http", "https":
+		log.Println("Using DOCKER_HOST: " + dockerHost)
+		transport.DialContext = func(ctx context.Context, proto, addr string) (net.Conn, error) {
+			return (&net.Dialer{}).DialContext(ctx, "tcp", parsedURL.Host)
+		}
+	default:
+		log.Fatal("Unsupported DOCKER_HOST: " + parsedURL.Scheme)
+	}
+
+	return &http.Client{
+		Timeout:   time.Second,
+		Transport: transport,
+	}
+}
+
+func closeIdleConnections(err error) {
+	log.Printf("Closing idle connections. Error: %+v\n", err)
+	dockerClient.Transport.(*http.Transport).CloseIdleConnections()
+}
--- a/beszel/internal/alerts/alerts.go
+++ b/beszel/internal/alerts/alerts.go
@@ -0,0 +1,155 @@
+package alerts
+
+import (
+	"beszel/internal/entities/email"
+	"beszel/internal/entities/system"
+	"fmt"
+	"net/mail"
+
+	"github.com/pocketbase/dbx"
+	"github.com/pocketbase/pocketbase"
+	"github.com/pocketbase/pocketbase/models"
+	"github.com/pocketbase/pocketbase/tools/mailer"
+)
+
+type AlertManager struct {
+	app        *pocketbase.PocketBase
+	mailClient mailer.Mailer
+}
+
+func NewAlertManager(app *pocketbase.PocketBase) *AlertManager {
+	return &AlertManager{
+		app:        app,
+		mailClient: app.NewMailClient(),
+	}
+}
+
+func (am *AlertManager) HandleSystemAlerts(newStatus string, newRecord *models.Record, oldRecord *models.Record) {
+	alertRecords, err := am.app.Dao().FindRecordsByExpr("alerts",
+		dbx.NewExp("system = {:system}", dbx.Params{"system": oldRecord.GetId()}),
+	)
+	if err != nil || len(alertRecords) == 0 {
+		// log.Println("no alerts found for system")
+		return
+	}
+	// log.Println("found alerts", len(alertRecords))
+	var systemInfo *system.SystemInfo
+	for _, alertRecord := range alertRecords {
+		name := alertRecord.GetString("name")
+		switch name {
+		case "Status":
+			am.handleStatusAlerts(newStatus, oldRecord, alertRecord)
+		case "CPU", "Memory", "Disk":
+			if newStatus != "up" {
+				continue
+			}
+			if systemInfo == nil {
+				systemInfo = getSystemInfo(newRecord)
+			}
+			if name == "CPU" {
+				am.handleSlidingValueAlert(newRecord, alertRecord, name, systemInfo.Cpu)
+			} else if name == "Memory" {
+				am.handleSlidingValueAlert(newRecord, alertRecord, name, systemInfo.MemPct)
+			} else if name == "Disk" {
+				am.handleSlidingValueAlert(newRecord, alertRecord, name, systemInfo.DiskPct)
+			}
+		}
+	}
+}
+
+func getSystemInfo(record *models.Record) *system.SystemInfo {
+	var SystemInfo system.SystemInfo
+	record.UnmarshalJSONField("info", &SystemInfo)
+	return &SystemInfo
+}
+
+func (am *AlertManager) handleSlidingValueAlert(newRecord *models.Record, alertRecord *models.Record, name string, curValue float64) {
+	triggered := alertRecord.GetBool("triggered")
+	threshold := alertRecord.GetFloat("value")
+	// fmt.Println(name, curValue, "threshold", threshold, "triggered", triggered)
+	var subject string
+	var body string
+	if !triggered && curValue > threshold {
+		alertRecord.Set("triggered", true)
+		systemName := newRecord.GetString("name")
+		subject = fmt.Sprintf("%s usage threshold exceeded on %s", name, systemName)
+		body = fmt.Sprintf("%s usage on %s is %.1f%%.\n\n%s\n\n- Beszel", name, systemName, curValue, am.app.Settings().Meta.AppUrl+"/system/"+systemName)
+	} else if triggered && curValue <= threshold {
+		alertRecord.Set("triggered", false)
+		systemName := newRecord.GetString("name")
+		subject = fmt.Sprintf("%s usage returned below threshold on %s", name, systemName)
+		body = fmt.Sprintf("%s usage on %s is below threshold at %.1f%%.\n\n%s\n\n- Beszel", name, systemName, curValue, am.app.Settings().Meta.AppUrl+"/system/"+systemName)
+	} else {
+		// fmt.Println(name, "not triggered")
+		return
+	}
+	if err := am.app.Dao().SaveRecord(alertRecord); err != nil {
+		// app.Logger().Error("failed to save alert record", "err", err.Error())
+		return
+	}
+	// expand the user relation and send the alert
+	if errs := am.app.Dao().ExpandRecord(alertRecord, []string{"user"}, nil); len(errs) > 0 {
+		// app.Logger().Error("failed to expand user relation", "errs", errs)
+		return
+	}
+	if user := alertRecord.ExpandedOne("user"); user != nil {
+		am.sendAlert(email.NewEmailData(
+			user.GetString("email"),
+			subject,
+			body,
+		))
+	}
+}
+
+func (am *AlertManager) handleStatusAlerts(newStatus string, oldRecord *models.Record, alertRecord *models.Record) error {
+	var alertStatus string
+	switch newStatus {
+	case "up":
+		if oldRecord.GetString("status") == "down" {
+			alertStatus = "up"
+		}
+	case "down":
+		if oldRecord.GetString("status") == "up" {
+			alertStatus = "down"
+		}
+	}
+	if alertStatus == "" {
+		return nil
+	}
+	// expand the user relation
+	if errs := am.app.Dao().ExpandRecord(alertRecord, []string{"user"}, nil); len(errs) > 0 {
+		return fmt.Errorf("failed to expand: %v", errs)
+	}
+	user := alertRecord.ExpandedOne("user")
+	if user == nil {
+		return nil
+	}
+	emoji := "\U0001F534"
+	if alertStatus == "up" {
+		emoji = "\u2705"
+	}
+	// send alert
+	systemName := oldRecord.GetString("name")
+	am.sendAlert(email.NewEmailData(
+		user.GetString("email"),
+		fmt.Sprintf("Connection to %s is %s %v", systemName, alertStatus, emoji),
+		fmt.Sprintf("Connection to %s is %s\n\n- Beszel", systemName, alertStatus),
+	))
+	return nil
+}
+
+func (am *AlertManager) sendAlert(data *email.EmailData) {
+	// fmt.Println("sending alert", "to", data.to, "subj", data.subj, "body", data.body)
+	message := &mailer.Message{
+		From: mail.Address{
+			Address: am.app.Settings().Meta.SenderAddress,
+			Name:    am.app.Settings().Meta.SenderName,
+		},
+		To:      []mail.Address{{Address: data.To()}},
+		Subject: data.Subject(),
+		Text:    data.Body(),
+	}
+	if err := am.mailClient.Send(message); err != nil {
+		am.app.Logger().Error("Failed to send alert: ", "err", err.Error())
+	}
+}
--- a/beszel/internal/entities/container/stats.go
+++ b/beszel/internal/entities/container/stats.go
@@ -0,0 +1,42 @@
+package container
+
+import "time"
+
+type Container struct {
+	Id      string
+	IdShort string
+	Names   []string
+	Status  string
+	// Image   string
+	// ImageID string
+	// Command string
+	// Created int64
+	// Ports      []Port
+	// SizeRw     int64 `json:",omitempty"`
+	// SizeRootFs int64 `json:",omitempty"`
+	// Labels     map[string]string
+	// State      string
+	// HostConfig struct {
+	// 	NetworkMode string            `json:",omitempty"`
+	// 	Annotations map[string]string `json:",omitempty"`
+	// }
+	// NetworkSettings *SummaryNetworkSettings
+	// Mounts          []MountPoint
+}
+
+type ContainerStats struct {
+	Name        string  `json:"n"`
+	Cpu         float64 `json:"c"`
+	Mem         float64 `json:"m"`
+	NetworkSent float64 `json:"ns"`
+	NetworkRecv float64 `json:"nr"`
+}
+
+type PrevContainerStats struct {
+	Cpu [2]uint64
+	Net struct {
+		Sent uint64
+		Recv uint64
+		Time time.Time
+	}
+}
--- a/beszel/internal/entities/email/email.go
+++ b/beszel/internal/entities/email/email.go
@@ -0,0 +1,27 @@
+package email
+
+type EmailData struct {
+	to   string
+	subj string
+	body string
+}
+
+func NewEmailData(to, subj, body string) *EmailData {
+	return &EmailData{
+		to:   to,
+		subj: subj,
+		body: body,
+	}
+}
+
+func (e *EmailData) To() string {
+	return e.to
+}
+
+func (e *EmailData) Subject() string {
+	return e.subj
+}
+
+func (e *EmailData) Body() string {
+	return e.body
+}
--- a/beszel/internal/entities/server/server.go
+++ b/beszel/internal/entities/server/server.go
@@ -0,0 +1,17 @@
+package server
+
+import "golang.org/x/crypto/ssh"
+
+type Server struct {
+	Host   string
+	Port   string
+	Status string
+	Client *ssh.Client
+}
+
+func NewServer(host, port string) *Server {
+	return &Server{
+		Host: host,
+		Port: port,
+	}
+}
--- a/beszel/internal/entities/system/stats.go
+++ b/beszel/internal/entities/system/stats.go
@@ -0,0 +1,121 @@
+package system
+
+import "time"
+
+type SystemStats struct {
+	Cpu          float64 `json:"cpu"`
+	Mem          float64 `json:"m"`
+	MemUsed      float64 `json:"mu"`
+	MemPct       float64 `json:"mp"`
+	MemBuffCache float64 `json:"mb"`
+	Swap         float64 `json:"s"`
+	SwapUsed     float64 `json:"su"`
+	Disk         float64 `json:"d"`
+	DiskUsed     float64 `json:"du"`
+	DiskPct      float64 `json:"dp"`
+	DiskRead     float64 `json:"dr"`
+	DiskWrite    float64 `json:"dw"`
+	NetworkSent  float64 `json:"ns"`
+	NetworkRecv  float64 `json:"nr"`
+}
+
+type DiskIoStats struct {
+	Read       uint64
+	Write      uint64
+	Time       time.Time
+	Filesystem string
+}
+
+type NetIoStats struct {
+	BytesRecv uint64
+	BytesSent uint64
+	Time      time.Time
+	Name      string
+}
+
+type CPUStats struct {
+	// CPU Usage. Linux and Windows.
+	CPUUsage CPUUsage `json:"cpu_usage"`
+
+	// System Usage. Linux only.
+	SystemUsage uint64 `json:"system_cpu_usage,omitempty"`
+
+	// Online CPUs. Linux only.
+	// OnlineCPUs uint32 `json:"online_cpus,omitempty"`
+
+	// Throttling Data. Linux only.
+	// ThrottlingData ThrottlingData `json:"throttling_data,omitempty"`
+}
+
+type CPUUsage struct {
+	// Total CPU time consumed.
+	// Units: nanoseconds (Linux)
+	// Units: 100's of nanoseconds (Windows)
+	TotalUsage uint64 `json:"total_usage"`
+
+	// Total CPU time consumed per core (Linux). Not used on Windows.
+	// Units: nanoseconds.
+	// PercpuUsage []uint64 `json:"percpu_usage,omitempty"`
+
+	// Time spent by tasks of the cgroup in kernel mode (Linux).
+	// Time spent by all container processes in kernel mode (Windows).
+	// Units: nanoseconds (Linux).
+	// Units: 100's of nanoseconds (Windows). Not populated for Hyper-V Containers.
+	// UsageInKernelmode uint64 `json:"usage_in_kernelmode"`
+
+	// Time spent by tasks of the cgroup in user mode (Linux).
+	// Time spent by all container processes in user mode (Windows).
+	// Units: nanoseconds (Linux).
+	// Units: 100's of nanoseconds (Windows). Not populated for Hyper-V Containers
+	// UsageInUsermode uint64 `json:"usage_in_usermode"`
+}
+
+type CStats struct {
+	// Common stats
+	// Read    time.Time `json:"read"`
+	// PreRead time.Time `json:"preread"`
+
+	// Linux specific stats, not populated on Windows.
+	// PidsStats  PidsStats  `json:"pids_stats,omitempty"`
+	// BlkioStats BlkioStats `json:"blkio_stats,omitempty"`
+
+	// Windows specific stats, not populated on Linux.
+	// NumProcs uint32 `json:"num_procs"`
+	// StorageStats StorageStats `json:"storage_stats,omitempty"`
+	// Networks request version >=1.21
+	Networks map[string]NetworkStats
+
+	// Shared stats
+	CPUStats CPUStats `json:"cpu_stats,omitempty"`
+	// PreCPUStats CPUStats    `json:"precpu_stats,omitempty"` // "Pre"="Previous"
+	MemoryStats MemoryStats `json:"memory_stats,omitempty"`
+}
+
+type MemoryStats struct {
+
+	// current res_counter usage for memory
+	Usage uint64 `json:"usage,omitempty"`
+	Cache uint64 `json:"cache,omitempty"`
+	// maximum usage ever recorded.
+	// MaxUsage uint64 `json:"max_usage,omitempty"`
+	// TODO(vishh): Export these as stronger types.
+	// all the stats exported via memory.stat.
+	Stats map[string]uint64 `json:"stats,omitempty"`
+	// number of times memory usage hits limits.
+	// Failcnt uint64 `json:"failcnt,omitempty"`
+	// Limit   uint64 `json:"limit,omitempty"`
+
+	// // committed bytes
+	// Commit uint64 `json:"commitbytes,omitempty"`
+	// // peak committed bytes
+	// CommitPeak uint64 `json:"commitpeakbytes,omitempty"`
+	// // private working set
+	// PrivateWorkingSet uint64 `json:"privateworkingset,omitempty"`
+}
+
+type NetworkStats struct {
+	// Bytes received. Windows and Linux.
+	RxBytes uint64 `json:"rx_bytes"`
+	// Bytes sent. Windows and Linux.
+	TxBytes uint64 `json:"tx_bytes"`
+}
--- a/beszel/internal/entities/system/system.go
+++ b/beszel/internal/entities/system/system.go
@@ -0,0 +1,20 @@
+package system
+
+import "beszel/internal/entities/container"
+
+type SystemInfo struct {
+	Cores    int    `json:"c"`
+	Threads  int    `json:"t"`
+	CpuModel string `json:"m"`
+	// Os       string  `json:"o"`
+	Uptime  uint64  `json:"u"`
+	Cpu     float64 `json:"cpu"`
+	MemPct  float64 `json:"mp"`
+	DiskPct float64 `json:"dp"`
+}
+
+type SystemData struct {
+	Stats      *SystemStats                `json:"stats"`
+	Info       *SystemInfo                 `json:"info"`
+	Containers []*container.ContainerStats `json:"container"`
+}
--- a/beszel/internal/hub/hub.go
+++ b/beszel/internal/hub/hub.go
@@ -0,0 +1,466 @@
+package hub
+
+import (
+	"beszel/internal/alerts"
+	"beszel/internal/entities/server"
+	"beszel/internal/entities/system"
+	"beszel/internal/records"
+	"beszel/site"
+	"bytes"
+	"crypto/ed25519"
+	"encoding/json"
+	"encoding/pem"
+	"errors"
+	"fmt"
+	"log"
+	"net/http"
+	"net/http/httputil"
+	"net/url"
+	"os"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/labstack/echo/v5"
+	"github.com/pocketbase/pocketbase"
+	"github.com/pocketbase/pocketbase/apis"
+	"github.com/pocketbase/pocketbase/core"
+	"github.com/pocketbase/pocketbase/models"
+	"github.com/pocketbase/pocketbase/plugins/migratecmd"
+	"github.com/pocketbase/pocketbase/tools/cron"
+	"golang.org/x/crypto/ssh"
+)
+
+type Hub struct {
+	app                   *pocketbase.PocketBase
+	serverConnectionsLock *sync.Mutex
+	serverConnections     map[string]*server.Server
+}
+
+func NewHub(app *pocketbase.PocketBase) *Hub {
+	return &Hub{
+		app:                   app,
+		serverConnectionsLock: &sync.Mutex{},
+		serverConnections:     make(map[string]*server.Server),
+	}
+}
+
+func (h *Hub) Run() {
+
+	rm := records.NewRecordManager(h.app)
+	am := alerts.NewAlertManager(h.app)
+
+	// loosely check if it was executed using "go run"
+	isGoRun := strings.HasPrefix(os.Args[0], os.TempDir())
+
+	// // enable auto creation of migration files when making collection changes in the Admin UI
+	migratecmd.MustRegister(h.app, h.app.RootCmd, migratecmd.Config{
+		// (the isGoRun check is to enable it only during development)
+		Automigrate: isGoRun,
+	})
+
+	// set auth settings
+	h.app.OnBeforeServe().Add(func(e *core.ServeEvent) error {
+		usersCollection, err := h.app.Dao().FindCollectionByNameOrId("users")
+		if err != nil {
+			return err
+		}
+		usersAuthOptions := usersCollection.AuthOptions()
+		usersAuthOptions.AllowUsernameAuth = false
+		if os.Getenv("DISABLE_PASSWORD_AUTH") == "true" {
+			usersAuthOptions.AllowEmailAuth = false
+		} else {
+			usersAuthOptions.AllowEmailAuth = true
+		}
+		usersCollection.SetOptions(usersAuthOptions)
+		if err := h.app.Dao().SaveCollection(usersCollection); err != nil {
+			return err
+		}
+		return nil
+	})
+
+	// serve site
+	h.app.OnBeforeServe().Add(func(e *core.ServeEvent) error {
+		switch isGoRun {
+		case true:
+			proxy := httputil.NewSingleHostReverseProxy(&url.URL{
+				Scheme: "http",
+				Host:   "localhost:5173",
+			})
+			e.Router.GET("/static/*", apis.StaticDirectoryHandler(os.DirFS("./site/public/static"), false))
+			e.Router.Any("/*", echo.WrapHandler(proxy))
+			// e.Router.Any("/", echo.WrapHandler(proxy))
+		default:
+			e.Router.GET("/static/*", apis.StaticDirectoryHandler(site.Static, false))
+			e.Router.Any("/*", apis.StaticDirectoryHandler(site.Dist, true))
+		}
+		return nil
+	})
+
+	// set up cron jobs / ticker for system updates
+	h.app.OnBeforeServe().Add(func(e *core.ServeEvent) error {
+		// 15 second ticker for system updates
+		go h.startSystemUpdateTicker()
+		// cron job to delete old records
+		scheduler := cron.New()
+		scheduler.MustAdd("delete old records", "8 * * * *", func() {
+			collections := []string{"system_stats", "container_stats"}
+			rm.DeleteOldRecords(collections, "1m", time.Hour)
+			rm.DeleteOldRecords(collections, "10m", 12*time.Hour)
+			rm.DeleteOldRecords(collections, "20m", 24*time.Hour)
+			rm.DeleteOldRecords(collections, "120m", 7*24*time.Hour)
+			rm.DeleteOldRecords(collections, "480m", 30*24*time.Hour)
+		})
+		scheduler.Start()
+		return nil
+	})
+
+	// ssh key setup
+	h.app.OnBeforeServe().Add(func(e *core.ServeEvent) error {
+		// create ssh key if it doesn't exist
+		h.getSSHKey()
+		// api route to return public key
+		e.Router.GET("/api/beszel/getkey", func(c echo.Context) error {
+			requestData := apis.RequestInfo(c)
+			if requestData.AuthRecord == nil {
+				return apis.NewForbiddenError("Forbidden", nil)
+			}
+			key, err := os.ReadFile(h.app.DataDir() + "/id_ed25519.pub")
+			if err != nil {
+				return err
+			}
+			return c.JSON(http.StatusOK, map[string]string{"key": strings.TrimSuffix(string(key), "\n")})
+		})
+		return nil
+	})
+
+	// other api routes
+	h.app.OnBeforeServe().Add(func(e *core.ServeEvent) error {
+		// check if first time setup on login page
+		e.Router.GET("/api/beszel/first-run", func(c echo.Context) error {
+			adminNum, err := h.app.Dao().TotalAdmins()
+			if err != nil {
+				return err
+			}
+			return c.JSON(http.StatusOK, map[string]bool{"firstRun": adminNum == 0})
+		})
+		return nil
+	})
+
+	// user creation - set default role to user if unset
+	h.app.OnModelBeforeCreate("users").Add(func(e *core.ModelEvent) error {
+		user := e.Model.(*models.Record)
+		if user.GetString("role") == "" {
+			user.Set("role", "user")
+		}
+		return nil
+	})
+
+	// system creation defaults
+	h.app.OnModelBeforeCreate("systems").Add(func(e *core.ModelEvent) error {
+		record := e.Model.(*models.Record)
+		record.Set("info", system.SystemInfo{})
+		record.Set("status", "pending")
+		return nil
+	})
+
+	// immediately create connection for new servers
+	h.app.OnModelAfterCreate("systems").Add(func(e *core.ModelEvent) error {
+		go h.updateSystem(e.Model.(*models.Record))
+		return nil
+	})
+
+	// do things after a systems record is updated
+	h.app.OnModelAfterUpdate("systems").Add(func(e *core.ModelEvent) error {
+		newRecord := e.Model.(*models.Record)
+		oldRecord := newRecord.OriginalCopy()
+		newStatus := newRecord.GetString("status")
+
+		// if server is disconnected and connection exists, remove it
+		if newStatus == "down" || newStatus == "paused" {
+			h.deleteServerConnection(newRecord)
+		}
+
+		// if server is set to pending (unpause), try to connect immediately
+		if newStatus == "pending" {
+			go h.updateSystem(newRecord)
+		}
+
+		// alerts
+		am.HandleSystemAlerts(newStatus, newRecord, oldRecord)
+		return nil
+	})
+
+	// do things after a systems record is deleted
+	h.app.OnModelAfterDelete("systems").Add(func(e *core.ModelEvent) error {
+		// if server connection exists, close it
+		h.deleteServerConnection(e.Model.(*models.Record))
+		return nil
+	})
+
+	h.app.OnModelAfterCreate("system_stats").Add(func(e *core.ModelEvent) error {
+		rm.CreateLongerRecords("system_stats", e.Model.(*models.Record))
+		return nil
+	})
+
+	h.app.OnModelAfterCreate("container_stats").Add(func(e *core.ModelEvent) error {
+		rm.CreateLongerRecords("container_stats", e.Model.(*models.Record))
+		return nil
+	})
+
+	if err := h.app.Start(); err != nil {
+		log.Fatal(err)
+	}
+}
+
+func (h *Hub) startSystemUpdateTicker() {
+	ticker := time.NewTicker(15 * time.Second)
+	for range ticker.C {
+		h.updateSystems()
+	}
+}
+
+func (h *Hub) updateSystems() {
+	records, err := h.app.Dao().FindRecordsByFilter(
+		"2hz5ncl8tizk5nx",    // collection
+		"status != 'paused'", // filter
+		"updated",            // sort
+		-1,                   // limit
+		0,                    // offset
+	)
+	// log.Println("records", len(records))
+	if err != nil || len(records) == 0 {
+		// h.app.Logger().Error("Failed to query systems")
+		return
+	}
+	fiftySecondsAgo := time.Now().UTC().Add(-50 * time.Second)
+	batchSize := len(records)/4 + 1
+	done := 0
+	for _, record := range records {
+		// break if batch size reached or if the system was updated less than 50 seconds ago
+		if done >= batchSize || record.GetDateTime("updated").Time().After(fiftySecondsAgo) {
+			break
+		}
+		// don't increment for down systems to avoid them jamming the queue
+		// because they're always first when sorted by least recently updated
+		if record.GetString("status") != "down" {
+			done++
+		}
+		go h.updateSystem(record)
+	}
+}
+
+func (h *Hub) updateSystem(record *models.Record) {
+	var s *server.Server
+	// check if server connection data exists
+	if _, ok := h.serverConnections[record.Id]; ok {
+		s = h.serverConnections[record.Id]
+	} else {
+		// create server connection struct
+		s = server.NewServer(
+			record.GetString("host"),
+			record.GetString("port"))
+		client, err := h.getServerConnection(s)
+		if err != nil {
+			h.app.Logger().Error("Failed to connect:", "err", err.Error(), "server", s.Host, "port", s.Port)
+			h.updateServerStatus(record, "down")
+			return
+		}
+		s.Client = client
+		h.serverConnectionsLock.Lock()
+		h.serverConnections[record.Id] = s
+		h.serverConnectionsLock.Unlock()
+	}
+	// get server stats from agent
+	systemData, err := requestJson(s)
+	if err != nil {
+		if err.Error() == "retry" {
+			// if previous connection was closed, try again
+			h.app.Logger().Error("Existing SSH connection closed. Retrying...", "host", s.Host, "port", s.Port)
+			h.deleteServerConnection(record)
+			h.updateSystem(record)
+			return
+		}
+		h.app.Logger().Error("Failed to get server stats: ", "err", err.Error())
+		h.updateServerStatus(record, "down")
+		return
+	}
+	// update system record
+	record.Set("status", "up")
+	record.Set("info", systemData.Info)
+	if err := h.app.Dao().SaveRecord(record); err != nil {
+		h.app.Logger().Error("Failed to update record: ", "err", err.Error())
+	}
+	// add new system_stats record
+	system_stats, _ := h.app.Dao().FindCollectionByNameOrId("system_stats")
+	system_stats_record := models.NewRecord(system_stats)
+	system_stats_record.Set("system", record.Id)
+	system_stats_record.Set("stats", systemData.Stats)
+	system_stats_record.Set("type", "1m")
+	if err := h.app.Dao().SaveRecord(system_stats_record); err != nil {
+		h.app.Logger().Error("Failed to save record: ", "err", err.Error())
+	}
+	// add new container_stats record
+	if len(systemData.Containers) > 0 {
+		container_stats, _ := h.app.Dao().FindCollectionByNameOrId("container_stats")
+		container_stats_record := models.NewRecord(container_stats)
+		container_stats_record.Set("system", record.Id)
+		container_stats_record.Set("stats", systemData.Containers)
+		container_stats_record.Set("type", "1m")
+		if err := h.app.Dao().SaveRecord(container_stats_record); err != nil {
+			h.app.Logger().Error("Failed to save record: ", "err", err.Error())
+		}
+	}
+}
+
+// set server to status down and close connection
+func (h *Hub) updateServerStatus(record *models.Record, status string) {
+	// if in map, close connection and remove from map
+	// this is now down automatically in an after update hook
+	// if status == "down" || status == "paused" {
+	// 	deleteServerConnection(record)
+	// }
+	if record.GetString("status") != status {
+		record.Set("status", status)
+		if err := h.app.Dao().SaveRecord(record); err != nil {
+			h.app.Logger().Error("Failed to update record: ", "err", err.Error())
+		}
+	}
+}
+
+func (h *Hub) deleteServerConnection(record *models.Record) {
+	if _, ok := h.serverConnections[record.Id]; ok {
+		if h.serverConnections[record.Id].Client != nil {
+			h.serverConnections[record.Id].Client.Close()
+		}
+		h.serverConnectionsLock.Lock()
+		defer h.serverConnectionsLock.Unlock()
+		delete(h.serverConnections, record.Id)
+	}
+}
+
+func (h *Hub) getServerConnection(server *server.Server) (*ssh.Client, error) {
+	// h.app.Logger().Debug("new ssh connection", "server", server.Host)
+	key, err := h.getSSHKey()
+	if err != nil {
+		h.app.Logger().Error("Failed to get SSH key: ", "err", err.Error())
+		return nil, err
+	}
+
+	// Create the Signer for this private key.
+	signer, err := ssh.ParsePrivateKey(key)
+	if err != nil {
+		return nil, err
+	}
+
+	config := &ssh.ClientConfig{
+		User: "u",
+		Auth: []ssh.AuthMethod{
+			ssh.PublicKeys(signer),
+		},
+		HostKeyCallback: ssh.InsecureIgnoreHostKey(),
+		Timeout:         5 * time.Second,
+	}
+
+	client, err := ssh.Dial("tcp", fmt.Sprintf("%s:%s", server.Host, server.Port), config)
+	if err != nil {
+		return nil, err
+	}
+
+	return client, nil
+}
+
+func requestJson(server *server.Server) (system.SystemData, error) {
+	session, err := server.Client.NewSession()
+	if err != nil {
+		return system.SystemData{}, errors.New("retry")
+	}
+	defer session.Close()
+
+	// Create a buffer to capture the output
+	var outputBuffer bytes.Buffer
+	session.Stdout = &outputBuffer
+
+	if err := session.Shell(); err != nil {
+		return system.SystemData{}, err
+	}
+
+	err = session.Wait()
+	if err != nil {
+		return system.SystemData{}, err
+	}
+
+	// Unmarshal the output into our struct
+	var systemData system.SystemData
+	err = json.Unmarshal(outputBuffer.Bytes(), &systemData)
+	if err != nil {
+		return system.SystemData{}, err
+	}
+
+	return systemData, nil
+}
+
+func (h *Hub) getSSHKey() ([]byte, error) {
+	dataDir := h.app.DataDir()
+	// check if the key pair already exists
+	existingKey, err := os.ReadFile(dataDir + "/id_ed25519")
+	if err == nil {
+		return existingKey, nil
+	}
+
+	// Generate the Ed25519 key pair
+	pubKey, privKey, err := ed25519.GenerateKey(nil)
+	if err != nil {
+		// h.app.Logger().Error("Error generating key pair:", "err", err.Error())
+		return nil, err
+	}
+
+	// Get the private key in OpenSSH format
+	privKeyBytes, err := ssh.MarshalPrivateKey(privKey, "")
+	if err != nil {
+		// h.app.Logger().Error("Error marshaling private key:", "err", err.Error())
+		return nil, err
+	}
+
+	// Save the private key to a file
+	privateFile, err := os.Create(dataDir + "/id_ed25519")
+	if err != nil {
+		// h.app.Logger().Error("Error creating private key file:", "err", err.Error())
+		return nil, err
+	}
+	defer privateFile.Close()
+
+	if err := pem.Encode(privateFile, privKeyBytes); err != nil {
+		// h.app.Logger().Error("Error writing private key to file:", "err", err.Error())
+		return nil, err
+	}
+
+	// Generate the public key in OpenSSH format
+	publicKey, err := ssh.NewPublicKey(pubKey)
+	if err != nil {
+		return nil, err
+	}
+
+	pubKeyBytes := ssh.MarshalAuthorizedKey(publicKey)
+
+	// Save the public key to a file
+	publicFile, err := os.Create(dataDir + "/id_ed25519.pub")
+	if err != nil {
+		return nil, err
+	}
+	defer publicFile.Close()
+
+	if _, err := publicFile.Write(pubKeyBytes); err != nil {
+		return nil, err
+	}
+
+	h.app.Logger().Info("ed25519 SSH key pair generated successfully.")
+	h.app.Logger().Info("Private key saved to: " + dataDir + "/id_ed25519")
+	h.app.Logger().Info("Public key saved to: " + dataDir + "/id_ed25519.pub")
+
+	existingKey, err = os.ReadFile(dataDir + "/id_ed25519")
+	if err == nil {
+		return existingKey, nil
+	}
+	return nil, err
+}
--- a/beszel/internal/records/records.go
+++ b/beszel/internal/records/records.go
@@ -0,0 +1,178 @@
+package records
+
+import (
+	"beszel/internal/entities/container"
+	"beszel/internal/entities/system"
+	"fmt"
+	"math"
+	"reflect"
+	"time"
+
+	"github.com/pocketbase/dbx"
+	"github.com/pocketbase/pocketbase"
+	"github.com/pocketbase/pocketbase/daos"
+	"github.com/pocketbase/pocketbase/models"
+)
+
+type RecordManager struct {
+	app *pocketbase.PocketBase
+}
+
+func NewRecordManager(app *pocketbase.PocketBase) *RecordManager {
+	return &RecordManager{app: app}
+}
+
+func (rm *RecordManager) CreateLongerRecords(collectionName string, shorterRecord *models.Record) {
+	shorterRecordType := shorterRecord.GetString("type")
+	systemId := shorterRecord.GetString("system")
+	// fmt.Println("create longer records", "recordType", shorterRecordType, "systemId", systemId)
+	var longerRecordType string
+	var timeAgo time.Duration
+	var expectedShorterRecords int
+	switch shorterRecordType {
+	case "1m":
+		longerRecordType = "10m"
+		timeAgo = -10 * time.Minute
+		expectedShorterRecords = 10
+	case "10m":
+		longerRecordType = "20m"
+		timeAgo = -20 * time.Minute
+		expectedShorterRecords = 2
+	case "20m":
+		longerRecordType = "120m"
+		timeAgo = -120 * time.Minute
+		expectedShorterRecords = 6
+	default:
+		longerRecordType = "480m"
+		timeAgo = -480 * time.Minute
+		expectedShorterRecords = 4
+	}
+
+	longerRecordPeriod := time.Now().UTC().Add(timeAgo + 10*time.Second).Format("2006-01-02 15:04:05")
+	// check creation time of last 10m record
+	lastLongerRecord, err := rm.app.Dao().FindFirstRecordByFilter(
+		collectionName,
+		"type = {:type} && system = {:system} && created > {:created}",
+		dbx.Params{"type": longerRecordType, "system": systemId, "created": longerRecordPeriod},
+	)
+	// return if longer record exists
+	if err == nil || lastLongerRecord != nil {
+		// log.Println("longer record found. returning")
+		return
+	}
+	// get shorter records from the past x minutes
+	// shorterRecordPeriod := time.Now().UTC().Add(timeAgo + time.Second).Format("2006-01-02 15:04:05")
+	allShorterRecords, err := rm.app.Dao().FindRecordsByFilter(
+		collectionName,
+		"type = {:type} && system = {:system} && created > {:created}",
+		"-created",
+		-1,
+		0,
+		dbx.Params{"type": shorterRecordType, "system": systemId, "created": longerRecordPeriod},
+	)
+	// return if not enough shorter records
+	if err != nil || len(allShorterRecords) < expectedShorterRecords {
+		// log.Println("not enough shorter records. returning")
+		return
+	}
+	// average the shorter records and create longer record
+	var stats interface{}
+	switch collectionName {
+	case "system_stats":
+		stats = rm.AverageSystemStats(allShorterRecords)
+	case "container_stats":
+		stats = rm.AverageContainerStats(allShorterRecords)
+	}
+	collection, _ := rm.app.Dao().FindCollectionByNameOrId(collectionName)
+	longerRecord := models.NewRecord(collection)
+	longerRecord.Set("system", systemId)
+	longerRecord.Set("stats", stats)
+	longerRecord.Set("type", longerRecordType)
+	if err := rm.app.Dao().SaveRecord(longerRecord); err != nil {
+		fmt.Println("failed to save longer record", "err", err.Error())
+	}
+
+}
+
+// calculate the average stats of a list of system_stats records
+func (rm *RecordManager) AverageSystemStats(records []*models.Record) system.SystemStats {
+	count := float64(len(records))
+	sum := reflect.New(reflect.TypeOf(system.SystemStats{})).Elem()
+
+	for _, record := range records {
+		var stats system.SystemStats
+		record.UnmarshalJSONField("stats", &stats)
+		statValue := reflect.ValueOf(stats)
+		for i := 0; i < statValue.NumField(); i++ {
+			field := sum.Field(i)
+			field.SetFloat(field.Float() + statValue.Field(i).Float())
+		}
+	}
+
+	average := reflect.New(reflect.TypeOf(system.SystemStats{})).Elem()
+	for i := 0; i < sum.NumField(); i++ {
+		average.Field(i).SetFloat(twoDecimals(sum.Field(i).Float() / count))
+	}
+
+	return average.Interface().(system.SystemStats)
+}
+
+// calculate the average stats of a list of container_stats records
+func (rm *RecordManager) AverageContainerStats(records []*models.Record) (stats []container.ContainerStats) {
+	sums := make(map[string]*container.ContainerStats)
+	count := float64(len(records))
+	for _, record := range records {
+		var stats []container.ContainerStats
+		record.UnmarshalJSONField("stats", &stats)
+		for _, stat := range stats {
+			if _, ok := sums[stat.Name]; !ok {
+				sums[stat.Name] = &container.ContainerStats{Name: stat.Name, Cpu: 0, Mem: 0}
+			}
+			sums[stat.Name].Cpu += stat.Cpu
+			sums[stat.Name].Mem += stat.Mem
+			sums[stat.Name].NetworkSent += stat.NetworkSent
+			sums[stat.Name].NetworkRecv += stat.NetworkRecv
+		}
+	}
+	for _, value := range sums {
+		stats = append(stats, container.ContainerStats{
+			Name:        value.Name,
+			Cpu:         twoDecimals(value.Cpu / count),
+			Mem:         twoDecimals(value.Mem / count),
+			NetworkSent: twoDecimals(value.NetworkSent / count),
+			NetworkRecv: twoDecimals(value.NetworkRecv / count),
+		})
+	}
+	return stats
+}
+
+/* Round float to two decimals */
+func twoDecimals(value float64) float64 {
+	return math.Round(value*100) / 100
+}
+
+/* Delete records of specified collections and type that are older than timeLimit */
+func (rm *RecordManager) DeleteOldRecords(collections []string, recordType string, timeLimit time.Duration) {
+	timeLimitStamp := time.Now().UTC().Add(-timeLimit).Format("2006-01-02 15:04:05")
+
+	// db query
+	expType := dbx.NewExp("type = {:type}", dbx.Params{"type": recordType})
+	expCreated := dbx.NewExp("created < {:created}", dbx.Params{"created": timeLimitStamp})
+
+	var records []*models.Record
+	for _, collection := range collections {
+		if collectionRecords, err := rm.app.Dao().FindRecordsByExpr(collection, expType, expCreated); err == nil {
+			records = append(records, collectionRecords...)
+		}
+	}
+
+	rm.app.Dao().RunInTransaction(func(txDao *daos.Dao) error {
+		for _, record := range records {
+			err := txDao.DeleteRecord(record)
+			if err != nil {
+				return err
+			}
+		}
+		return nil
+	})
+}
--- a/beszel/internal/update/update.go
+++ b/beszel/internal/update/update.go
@@ -0,0 +1,100 @@
+package update
+
+import (
+	"beszel"
+	"fmt"
+	"os"
+	"strings"
+
+	"github.com/blang/semver"
+	"github.com/rhysd/go-github-selfupdate/selfupdate"
+	"github.com/spf13/cobra"
+)
+
+func UpdateBeszel(cmd *cobra.Command, args []string) {
+	var latest *selfupdate.Release
+	var found bool
+	var err error
+	currentVersion := semver.MustParse(beszel.Version)
+	fmt.Println("beszel", currentVersion)
+	fmt.Println("Checking for updates...")
+	updater, _ := selfupdate.NewUpdater(selfupdate.Config{
+		Filters: []string{"beszel_"},
+	})
+	latest, found, err = updater.DetectLatest("henrygd/beszel")
+
+	if err != nil {
+		fmt.Println("Error checking for updates:", err)
+		os.Exit(1)
+	}
+
+	if !found {
+		fmt.Println("No updates found")
+		os.Exit(0)
+	}
+
+	fmt.Println("Latest version:", latest.Version)
+
+	if latest.Version.LTE(currentVersion) {
+		fmt.Println("You are up to date")
+		return
+	}
+
+	var binaryPath string
+	fmt.Printf("Updating from %s to %s...\n", currentVersion, latest.Version)
+	binaryPath, err = os.Executable()
+	if err != nil {
+		fmt.Println("Error getting binary path:", err)
+		os.Exit(1)
+	}
+	err = selfupdate.UpdateTo(latest.AssetURL, binaryPath)
+	if err != nil {
+		fmt.Println("Please try rerunning with sudo. Error:", err)
+		os.Exit(1)
+	}
+	fmt.Printf("Successfully updated to %s\n\n%s\n", latest.Version, strings.TrimSpace(latest.ReleaseNotes))
+}
+
+func UpdateBeszelAgent() {
+	var latest *selfupdate.Release
+	var found bool
+	var err error
+	currentVersion := semver.MustParse(beszel.Version)
+	fmt.Println("beszel-agent", currentVersion)
+	fmt.Println("Checking for updates...")
+	updater, _ := selfupdate.NewUpdater(selfupdate.Config{
+		Filters: []string{"beszel-agent"},
+	})
+	latest, found, err = updater.DetectLatest("henrygd/beszel")
+
+	if err != nil {
+		fmt.Println("Error checking for updates:", err)
+		os.Exit(1)
+	}
+
+	if !found {
+		fmt.Println("No updates found")
+		os.Exit(0)
+	}
+
+	fmt.Println("Latest version:", latest.Version)
+
+	if latest.Version.LTE(currentVersion) {
+		fmt.Println("You are up to date")
+		return
+	}
+
+	var binaryPath string
+	fmt.Printf("Updating from %s to %s...\n", currentVersion, latest.Version)
+	binaryPath, err = os.Executable()
+	if err != nil {
+		fmt.Println("Error getting binary path:", err)
+		os.Exit(1)
+	}
+	err = selfupdate.UpdateTo(latest.AssetURL, binaryPath)
+	if err != nil {
+		fmt.Println("Please try rerunning with sudo. Error:", err)
+		os.Exit(1)
+	}
+	fmt.Printf("Successfully updated to %s\n\n%s\n", latest.Version, strings.TrimSpace(latest.ReleaseNotes))
+}