Compare commits

..

7 Commits

19 changed files with 356 additions and 101 deletions

View File

@@ -11,6 +11,7 @@ import (
"strings" "strings"
"sync/atomic" "sync/atomic"
"testing" "testing"
"testing/synctest"
beszelTests "github.com/henrygd/beszel/internal/tests" beszelTests "github.com/henrygd/beszel/internal/tests"
pbTests "github.com/pocketbase/pocketbase/tests" pbTests "github.com/pocketbase/pocketbase/tests"
@@ -533,6 +534,20 @@ func TestSendTestNotification(t *testing.T) {
for _, url := range []string{localURL, "smtp://user:pass@127.0.0.1/?fromAddress=sender@example.com&toAddresses=recipient@example.com", "mqtt://127.0.0.1/topic"} { for _, url := range []string{localURL, "smtp://user:pass@127.0.0.1/?fromAddress=sender@example.com&toAddresses=recipient@example.com", "mqtt://127.0.0.1/topic"} {
scenarios = append(scenarios, beszelTests.ApiScenario{ scenarios = append(scenarios, beszelTests.ApiScenario{
BeforeTestFunc: func(tb testing.TB, _ *pbTests.TestApp, e *core.ServeEvent) {
if !strings.HasPrefix(url, "mqtt://") {
return
}
// Keep the real MQTT rejection path, but advance its library's
// fixed timeout using virtual time instead of waiting 10 seconds.
e.Router.BindFunc(func(re *core.RequestEvent) error {
var err error
synctest.Test(tb.(*testing.T), func(t *testing.T) {
err = re.Next()
})
return err
})
},
Name: "readonly cannot send to " + url, Name: "readonly cannot send to " + url,
Method: http.MethodPost, Method: http.MethodPost,
URL: "/api/beszel/test-notification", URL: "/api/beszel/test-notification",

View File

@@ -11,6 +11,7 @@ import (
"strings" "strings"
"sync/atomic" "sync/atomic"
"testing" "testing"
"testing/synctest"
"github.com/nicholas-fedor/shoutrrr/pkg/types" "github.com/nicholas-fedor/shoutrrr/pkg/types"
"golang.org/x/net/dns/dnsmessage" "golang.org/x/net/dns/dnsmessage"
@@ -178,39 +179,45 @@ func TestPublicNotificationTCP(t *testing.T) {
} { } {
t.Run(rawURL, func(t *testing.T) { t.Run(rawURL, func(t *testing.T) {
t.Parallel() t.Parallel()
// MQTT waits for a fixed library timeout even after a dial failure.
// Virtual time preserves the full send/cleanup path without that delay.
t.Run("internal destination", func(t *testing.T) { t.Run("internal destination", func(t *testing.T) {
err := sendPublicNotification(strings.ReplaceAll(rawURL, "HOST", "127.0.0.1"), "test") synctest.Test(t, func(t *testing.T) {
if !errors.Is(err, errInternalDestination) { err := sendPublicNotification(strings.ReplaceAll(rawURL, "HOST", "127.0.0.1"), "test")
t.Fatalf("expected blocked destination, got %v", err) if !errors.Is(err, errInternalDestination) {
} t.Fatalf("expected blocked destination, got %v", err)
}
})
}) })
t.Run("public destination uses injected dialer", func(t *testing.T) { t.Run("public destination uses injected dialer", func(t *testing.T) {
var calls atomic.Int32 synctest.Test(t, func(t *testing.T) {
stopped := errors.New("test dial stopped") var calls atomic.Int32
service, err := newPublicNotificationService(strings.ReplaceAll(rawURL, "HOST", "8.8.8.8"), types.SenderOptions{ stopped := errors.New("test dial stopped")
DialContext: func(ctx context.Context, network, address string) (net.Conn, error) { service, err := newPublicNotificationService(strings.ReplaceAll(rawURL, "HOST", "8.8.8.8"), types.SenderOptions{
calls.Add(1) DialContext: func(ctx context.Context, network, address string) (net.Conn, error) {
if network != "tcp" || !strings.HasPrefix(address, "8.8.8.8:") { calls.Add(1)
t.Errorf("unexpected dial: %s %s", network, address) if network != "tcp" || !strings.HasPrefix(address, "8.8.8.8:") {
} t.Errorf("unexpected dial: %s %s", network, address)
if err := checkNotificationAddress(address); err != nil { }
t.Error(err) if err := checkNotificationAddress(address); err != nil {
} t.Error(err)
return nil, stopped }
}, return nil, stopped
},
})
if err != nil {
t.Fatal(err)
}
if closer, ok := service.(io.Closer); ok {
defer closer.Close()
}
if err := service.Send("test", &types.Params{}); err == nil {
t.Fatal("expected dial failure")
}
if calls.Load() == 0 {
t.Fatal("custom dialer was not used")
}
}) })
if err != nil {
t.Fatal(err)
}
if closer, ok := service.(io.Closer); ok {
defer closer.Close()
}
if err := service.Send("test", &types.Params{}); err == nil {
t.Fatal("expected dial failure")
}
if calls.Load() == 0 {
t.Fatal("custom dialer was not used")
}
}) })
}) })
} }

View File

@@ -12,7 +12,7 @@ RUN apk add --no-cache ca-certificates && update-ca-certificates
# Build # Build
ARG TARGETOS TARGETARCH ARG TARGETOS TARGETARCH
RUN CGO_ENABLED=0 GOGC=75 GOOS=$TARGETOS GOARCH=$TARGETARCH go build -ldflags "-w -s" -o /agent ./internal/cmd/agent RUN CGO_ENABLED=0 GOOS=$TARGETOS GOARCH=$TARGETARCH go build -ldflags "-w -s" -o /agent ./internal/cmd/agent
RUN rm -rf /tmp/* RUN rm -rf /tmp/*

View File

@@ -10,14 +10,14 @@ COPY . ./
# Build # Build
ARG TARGETOS TARGETARCH ARG TARGETOS TARGETARCH
RUN CGO_ENABLED=0 GOGC=75 GOOS=$TARGETOS GOARCH=$TARGETARCH go build -ldflags "-w -s" -o /agent ./internal/cmd/agent RUN CGO_ENABLED=0 GOOS=$TARGETOS GOARCH=$TARGETARCH go build -ldflags "-w -s" -o /agent ./internal/cmd/agent
RUN rm -rf /tmp/* RUN rm -rf /tmp/*
# -------------------------- # --------------------------
# Final image: default scratch-based agent # Final image: default scratch-based agent
# -------------------------- # --------------------------
FROM alpine:3.23 FROM alpine:3.24
COPY --from=builder /agent /agent COPY --from=builder /agent /agent
# AMD GPU name lookup (used by agent on Linux when /usr/share/libdrm/amdgpu.ids is read) # AMD GPU name lookup (used by agent on Linux when /usr/share/libdrm/amdgpu.ids is read)
@@ -28,4 +28,4 @@ RUN apk add --no-cache smartmontools zfs
# Ensure data persistence across container recreations # Ensure data persistence across container recreations
VOLUME ["/var/lib/beszel-agent"] VOLUME ["/var/lib/beszel-agent"]
ENTRYPOINT ["/agent"] ENTRYPOINT ["/agent"]

View File

@@ -10,13 +10,13 @@ COPY . ./
# Build # Build
ARG TARGETOS TARGETARCH ARG TARGETOS TARGETARCH
RUN CGO_ENABLED=0 GOGC=75 GOOS=$TARGETOS GOARCH=$TARGETARCH go build -ldflags "-w -s" -o /agent ./internal/cmd/agent RUN CGO_ENABLED=0 GOOS=$TARGETOS GOARCH=$TARGETARCH go build -ldflags "-w -s" -o /agent ./internal/cmd/agent
# -------------------------- # --------------------------
# Final image # Final image
# Note: must cap_add: [CAP_PERFMON] and mount /dev/dri/ as volume # Note: must cap_add: [CAP_PERFMON] and mount /dev/dri/ as volume
# -------------------------- # --------------------------
FROM alpine:3.23 FROM alpine:3.24
COPY --from=builder /agent /agent COPY --from=builder /agent /agent

View File

@@ -10,7 +10,7 @@ COPY . ./
# Build # Build
ARG TARGETOS TARGETARCH ARG TARGETOS TARGETARCH
RUN CGO_ENABLED=0 GOGC=75 GOOS=$TARGETOS GOARCH=$TARGETARCH go build -tags glibc -ldflags "-w -s" -o /agent ./internal/cmd/agent RUN CGO_ENABLED=0 GOOS=$TARGETOS GOARCH=$TARGETARCH go build -tags glibc -ldflags "-w -s" -o /agent ./internal/cmd/agent
# -------------------------- # --------------------------
# Smartmontools builder stage # Smartmontools builder stage

View File

@@ -17,7 +17,7 @@ RUN set -eux; \
if [ "$TARGETARCH" = "arm" ] && [ -n "$TARGETVARIANT" ]; then \ if [ "$TARGETARCH" = "arm" ] && [ -n "$TARGETVARIANT" ]; then \
export GOARM="${TARGETVARIANT#v}"; \ export GOARM="${TARGETVARIANT#v}"; \
fi; \ fi; \
CGO_ENABLED=0 GOGC=75 GOOS=$TARGETOS GOARCH=$TARGETARCH \ CGO_ENABLED=0 GOOS=$TARGETOS GOARCH=$TARGETARCH \
go build -tags glibc -ldflags "-w -s" -o /agent ./internal/cmd/agent go build -tags glibc -ldflags "-w -s" -o /agent ./internal/cmd/agent
# -------------------------- # --------------------------

View File

@@ -17,7 +17,7 @@ RUN update-ca-certificates
# Build # Build
ARG TARGETOS TARGETARCH ARG TARGETOS TARGETARCH
RUN CGO_ENABLED=0 GOGC=75 GOOS=$TARGETOS GOARCH=$TARGETARCH go build -ldflags "-w -s" -o /beszel ./internal/cmd/hub RUN CGO_ENABLED=0 GOOS=$TARGETOS GOARCH=$TARGETARCH go build -ldflags "-w -s" -o /beszel ./internal/cmd/hub
# ? ------------------------- # ? -------------------------
FROM scratch FROM scratch
@@ -31,4 +31,4 @@ VOLUME ["/beszel_data"]
EXPOSE 8090 EXPOSE 8090
ENTRYPOINT [ "/beszel" ] ENTRYPOINT [ "/beszel" ]
CMD ["serve", "--http=0.0.0.0:8090"] CMD ["serve", "--http=0.0.0.0:8090"]

View File

@@ -978,11 +978,18 @@ func TestAgentWebSocketIntegration(t *testing.T) {
} }
} }
// Verify system status // A connected WebSocket does not mean the hub has finished verifying
updatedSystemRecord, err := testApp.FindRecordById("systems", systemRecord.Id) // the agent and updating the system. Wait for the database state rather
require.NoError(t, err) // than assuming that work completes within a fixed sleep under load.
status := updatedSystemRecord.GetString("status") var status string
assert.Equal(t, tc.expectSystemStatus, status, "System status should match expected value") require.EventuallyWithT(t, func(c *assert.CollectT) {
updatedSystemRecord, err := testApp.FindRecordById("systems", systemRecord.Id)
if !assert.NoError(c, err) {
return
}
status = updatedSystemRecord.GetString("status")
assert.Equal(c, tc.expectSystemStatus, status, "System status should match expected value")
}, 5*time.Second, 20*time.Millisecond)
t.Logf("%s - System status: %s, Fingerprint: %s", tc.description, status, finalFingerprint) t.Logf("%s - System status: %s, Fingerprint: %s", tc.description, status, finalFingerprint)
}) })
@@ -1142,42 +1149,43 @@ func TestMultipleSystemsWithSameUniversalToken(t *testing.T) {
// Verify system creation/reuse behavior // Verify system creation/reuse behavior
if tc.expectConnection { if tc.expectConnection {
// Count systems after connection expectedSystemsAfter := systemsBeforeCount
systemsAfter, err := testApp.FindRecordsByFilter("systems", "users ~ {:userId}", "", -1, 0, map[string]any{"userId": userRecord.Id})
require.NoError(t, err)
systemsAfterCount := len(systemsAfter)
if tc.expectNewSystem { if tc.expectNewSystem {
// Should have created a new system expectedSystemsAfter++
systemCount++ systemCount++
assert.Equal(t, systemsBeforeCount+1, systemsAfterCount, "Should have created a new system")
assert.Equal(t, systemCount, systemsAfterCount, "Total system count should match expected")
} else {
// Should have reused existing system
assert.Equal(t, systemsBeforeCount, systemsAfterCount, "Should not have created a new system")
assert.Equal(t, systemCount, systemsAfterCount, "Total system count should remain the same")
} }
time.Sleep(20 * time.Millisecond) // WebSocket connection precedes the hub's asynchronous system
// setup. Re-read all database state until setup is complete.
var systemId, status string
require.EventuallyWithT(t, func(c *assert.CollectT) {
systemsAfter, err := testApp.FindRecordsByFilter("systems", "users ~ {:userId}", "", -1, 0, map[string]any{"userId": userRecord.Id})
if !assert.NoError(c, err) {
return
}
assert.Len(c, systemsAfter, expectedSystemsAfter, "System creation/reuse should match expected behavior")
assert.Len(c, systemsAfter, systemCount, "Total system count should match expected")
// Verify that a fingerprint record exists for this fingerprint fingerprints, err := testApp.FindRecordsByFilter("fingerprints", "token = {:token} && fingerprint = {:fingerprint}", "", -1, 0, map[string]any{
fingerprints, err := testApp.FindRecordsByFilter("fingerprints", "token = {:token} && fingerprint = {:fingerprint}", "", -1, 0, map[string]any{ "token": universalToken,
"token": universalToken, "fingerprint": tc.agentFingerprint,
"fingerprint": tc.agentFingerprint, })
}) if !assert.NoError(c, err) || !assert.Len(c, fingerprints, 1, "Should have exactly one fingerprint record for this token+fingerprint combination") {
require.NoError(t, err) return
require.Len(t, fingerprints, 1, "Should have exactly one fingerprint record for this token+fingerprint combination") }
fingerprint := fingerprints[0] fingerprint := fingerprints[0]
assert.Equal(t, universalToken, fingerprint.GetString("token"), "Fingerprint should have the universal token") assert.Equal(c, universalToken, fingerprint.GetString("token"), "Fingerprint should have the universal token")
assert.Equal(t, tc.agentFingerprint, fingerprint.GetString("fingerprint"), "Fingerprint should match agent's fingerprint") assert.Equal(c, tc.agentFingerprint, fingerprint.GetString("fingerprint"), "Fingerprint should match agent's fingerprint")
// Verify system status systemId = fingerprint.GetString("system")
systemId := fingerprint.GetString("system") system, err := testApp.FindRecordById("systems", systemId)
system, err := testApp.FindRecordById("systems", systemId) if !assert.NoError(c, err) {
require.NoError(t, err) return
status := system.GetString("status") }
assert.Equal(t, tc.expectSystemStatus, status, "System status should match expected value") status = system.GetString("status")
assert.Equal(c, tc.expectSystemStatus, status, "System status should match expected value")
}, 5*time.Second, 20*time.Millisecond)
t.Logf("%s - System ID: %s, Status: %s, New System: %v", tc.description, systemId, status, tc.expectNewSystem) t.Logf("%s - System ID: %s, Status: %s, New System: %v", tc.description, systemId, status, tc.expectNewSystem)
} }

View File

@@ -0,0 +1,159 @@
//go:build testing
package systems
import (
"context"
"crypto/ed25519"
"crypto/rand"
"net"
"sync/atomic"
"testing"
"time"
"github.com/fxamacker/cbor/v2"
"github.com/henrygd/beszel/internal/common"
"github.com/henrygd/beszel/internal/entities/monitor"
esystem "github.com/henrygd/beszel/internal/entities/system"
"github.com/henrygd/beszel/internal/hub/expirymap"
"github.com/pocketbase/pocketbase/core"
"github.com/stretchr/testify/require"
"golang.org/x/crypto/ssh"
)
func TestSSHNetworkMonitorReconnectSync(t *testing.T) {
sys, app := newTestSystemWithHub(t)
sys.manager.zfsFetchMap = expirymap.New[zfsFetchState](time.Hour)
t.Cleanup(sys.manager.zfsFetchMap.StopCleaner)
sys.ctx = context.Background()
sys.Status = up
_, key, err := ed25519.GenerateKey(rand.Reader)
require.NoError(t, err)
signer, err := ssh.NewSignerFromKey(key)
require.NoError(t, err)
config := &ssh.ServerConfig{NoClientAuth: true, ServerVersion: "SSH-2.0-beszel_0.20.0"}
config.AddHostKey(signer)
listener, err := net.Listen("tcp", "127.0.0.1:0")
require.NoError(t, err)
t.Cleanup(func() { _ = listener.Close() })
sys.Host, sys.Port, err = net.SplitHostPort(listener.Addr().String())
require.NoError(t, err)
sys.manager.sshConfig = &ssh.ClientConfig{User: "test", HostKeyCallback: ssh.InsecureIgnoreHostKey(), Timeout: time.Second}
t.Cleanup(sys.closeSSHConnection)
requests := make(chan monitor.SyncRequest, 10)
var failSync atomic.Bool
go func() {
for {
conn, err := listener.Accept()
if err != nil {
return
}
go func() {
server, channels, reqs, err := ssh.NewServerConn(conn, config)
if err != nil {
_ = conn.Close()
return
}
defer server.Close()
go ssh.DiscardRequests(reqs)
for channel := range channels {
ch, reqs, err := channel.Accept()
if err != nil {
return
}
go func() {
defer ch.Close()
for req := range reqs {
if req.Type != "shell" {
_ = req.Reply(false, nil)
continue
}
_ = req.Reply(true, nil)
var request common.HubRequest[cbor.RawMessage]
if cbor.NewDecoder(ch).Decode(&request) != nil {
return
}
response := common.AgentResponse{}
switch request.Action {
case common.GetData:
response.SystemData = &esystem.CombinedData{}
case common.SyncNetworkMonitors:
var syncReq monitor.SyncRequest
if cbor.Unmarshal(request.Data, &syncReq) != nil {
return
}
requests <- syncReq
if failSync.Load() {
response.Error = "test sync failure"
} else {
response.Data, _ = cbor.Marshal(monitor.SyncResponse{})
}
}
_ = cbor.NewEncoder(ch).Encode(response)
_, _ = ch.SendRequest("exit-status", false, ssh.Marshal(struct{ Status uint32 }{0}))
return
}
}()
}
}()
}
}()
collection, err := app.FindCachedCollectionByNameOrId("network_monitors")
require.NoError(t, err)
probe := core.NewRecord(collection)
probe.Load(map[string]any{"system": sys.Id, "target": "localhost", "protocol": "tcp", "port": 80, "interval": 60, "enabled": true})
require.NoError(t, app.SaveNoValidate(probe))
fetch := func() {
t.Helper()
_, err := sys.fetchDataFromAgent(common.DataRequestOptions{})
require.NoError(t, err, "monitor sync failure must not fail stats fetching")
}
receive := func() monitor.SyncRequest {
t.Helper()
select {
case req := <-requests:
require.Equal(t, monitor.SyncActionReplace, req.Action)
return req
case <-time.After(time.Second):
t.Fatal("missing full monitor sync")
return monitor.SyncRequest{}
}
}
fetch()
require.Equal(t, probe.Id, receive().Configs[0].ID)
require.False(t, sys.monitorsNeedSync.Load())
fetch()
require.Empty(t, requests, "steady-state fetch must not resync")
// Simulate loss of the agent process/connection and its in-memory monitors.
require.NoError(t, sys.client.Load().Close())
fetch()
require.Equal(t, probe.Id, receive().Configs[0].ID)
require.False(t, sys.monitorsNeedSync.Load())
// Failed replacements are retried on the next successful stats fetch.
require.NoError(t, sys.client.Load().Close())
failSync.Store(true)
fetch()
receive()
require.True(t, sys.monitorsNeedSync.Load())
failSync.Store(false)
fetch()
receive()
require.False(t, sys.monitorsNeedSync.Load())
probe.Set("enabled", false)
require.NoError(t, app.SaveNoValidate(probe))
require.NoError(t, sys.client.Load().Close())
fetch()
require.Empty(t, receive().Configs, "empty replacement must clear stale monitors")
}
func TestPendingNetworkMonitorSyncQueryFailure(t *testing.T) {
sys, app := newTestSystemWithHub(t)
_, err := app.DB().NewQuery("DROP TABLE network_monitors").Execute()
require.NoError(t, err)
sys.monitorsNeedSync.Store(true)
sys.syncPendingNetworkMonitors()
require.True(t, sys.monitorsNeedSync.Load())
}

View File

@@ -6,6 +6,7 @@ import (
"net/http" "net/http"
"net/http/httptest" "net/http/httptest"
"strings" "strings"
"sync/atomic"
"testing" "testing"
"time" "time"
@@ -13,6 +14,7 @@ import (
"github.com/fxamacker/cbor/v2" "github.com/fxamacker/cbor/v2"
"github.com/henrygd/beszel/internal/common" "github.com/henrygd/beszel/internal/common"
"github.com/henrygd/beszel/internal/entities/monitor" "github.com/henrygd/beszel/internal/entities/monitor"
esystem "github.com/henrygd/beszel/internal/entities/system"
"github.com/henrygd/beszel/internal/hub/ws" "github.com/henrygd/beszel/internal/hub/ws"
"github.com/lxzan/gws" "github.com/lxzan/gws"
"github.com/pocketbase/pocketbase/core" "github.com/pocketbase/pocketbase/core"
@@ -22,17 +24,31 @@ import (
type monitorSyncClient struct { type monitorSyncClient struct {
gws.BuiltinEventHandler gws.BuiltinEventHandler
requests chan common.HubRequest[monitor.SyncRequest] requests chan common.HubRequest[monitor.SyncRequest]
failSync atomic.Bool
} }
func (c *monitorSyncClient) OnMessage(conn *gws.Conn, message *gws.Message) { func (c *monitorSyncClient) OnMessage(conn *gws.Conn, message *gws.Message) {
defer message.Close() defer message.Close()
var req common.HubRequest[monitor.SyncRequest] var req common.HubRequest[cbor.RawMessage]
if err := cbor.Unmarshal(message.Bytes(), &req); err != nil { if err := cbor.Unmarshal(message.Bytes(), &req); err != nil {
return return
} }
c.requests <- req resp := common.AgentResponse{Id: req.Id}
data, _ := cbor.Marshal(monitor.SyncResponse{}) if req.Action == common.GetData {
response, _ := cbor.Marshal(common.AgentResponse{Id: req.Id, Data: data}) resp.SystemData = &esystem.CombinedData{}
} else {
var data monitor.SyncRequest
if err := cbor.Unmarshal(req.Data, &data); err != nil {
return
}
c.requests <- common.HubRequest[monitor.SyncRequest]{Id: req.Id, Action: req.Action, Data: data}
if c.failSync.Load() {
resp.Error = "test sync failure"
} else {
resp.Data, _ = cbor.Marshal(monitor.SyncResponse{})
}
}
response, _ := cbor.Marshal(resp)
_ = conn.WriteMessage(gws.OpcodeBinary, response) _ = conn.WriteMessage(gws.OpcodeBinary, response)
} }
@@ -57,7 +73,7 @@ func TestNetworkMonitorSyncSkipsOlderAgents(t *testing.T) {
} }
func TestNetworkMonitorReconnectSync(t *testing.T) { func TestNetworkMonitorReconnectSync(t *testing.T) {
for _, change := range []string{"delete", "disable"} { for _, change := range []string{"delete", "disable", "retry"} {
t.Run(change, func(t *testing.T) { t.Run(change, func(t *testing.T) {
sys, app := newTestSystemWithHub(t) sys, app := newTestSystemWithHub(t)
record, err := app.FindRecordById("systems", sys.Id) record, err := app.FindRecordById("systems", sys.Id)
@@ -120,9 +136,33 @@ func TestNetworkMonitorReconnectSync(t *testing.T) {
} }
} }
client.failSync.Store(change == "retry")
initial := connect() initial := connect()
require.Len(t, initial.Configs, 1) require.Len(t, initial.Configs, 1)
require.Equal(t, probe.Id, initial.Configs[0].ID) require.Equal(t, probe.Id, initial.Configs[0].ID)
if change == "retry" {
system, err := sm.GetSystem(sys.Id)
require.NoError(t, err)
require.Eventually(t, system.monitorsNeedSync.Load, time.Second, time.Millisecond)
// A second failed sync must not fail the stats fetch or clear pending state.
_, err = system.fetchDataFromAgent(common.DataRequestOptions{})
require.NoError(t, err)
require.True(t, system.monitorsNeedSync.Load())
require.Len(t, client.requests, 1)
<-client.requests
client.failSync.Store(false)
_, err = system.fetchDataFromAgent(common.DataRequestOptions{})
require.NoError(t, err)
require.False(t, system.monitorsNeedSync.Load())
require.Len(t, client.requests, 1)
retry := <-client.requests
require.Equal(t, monitor.SyncActionReplace, retry.Data.Action)
require.Equal(t, initial.Configs, retry.Data.Configs)
_, err = system.fetchDataFromAgent(common.DataRequestOptions{})
require.NoError(t, err)
require.Empty(t, client.requests, "successful sync must not repeat on every fetch")
return
}
require.NoError(t, sm.RemoveSystem(sys.Id)) require.NoError(t, sm.RemoveSystem(sys.Id))
if change == "delete" { if change == "delete" {
require.NoError(t, app.Delete(probe)) require.NoError(t, app.Delete(probe))

View File

@@ -2,6 +2,7 @@ package systems
import ( import (
"context" "context"
"fmt"
"time" "time"
"github.com/henrygd/beszel" "github.com/henrygd/beszel"
@@ -9,6 +10,27 @@ import (
"github.com/henrygd/beszel/internal/entities/monitor" "github.com/henrygd/beszel/internal/entities/monitor"
) )
// syncPendingNetworkMonitors runs on WebSocket connect and after successful stats
// fetches. Failed syncs retry on the next update without taking the system down.
func (sys *System) syncPendingNetworkMonitors() {
if !sys.monitorsNeedSync.Swap(false) {
return
}
if err := sys.syncAllNetworkMonitors(); err != nil {
sys.monitorsNeedSync.Store(true)
sys.manager.hub.Logger().Warn("failed to sync monitors to agent", "system", sys.Id, "err", err)
}
}
func (sys *System) syncAllNetworkMonitors() error {
configs, err := sys.manager.GetMonitorConfigsForSystem(sys.Id)
if err != nil {
return fmt.Errorf("failed to load monitors: %w", err)
}
// An empty set must also replace probes retained across a disconnect.
return sys.SyncNetworkMonitors(configs)
}
// SyncNetworkMonitors sends monitor configurations to the agent. // SyncNetworkMonitors sends monitor configurations to the agent.
func (sys *System) SyncNetworkMonitors(configs []monitor.Config) error { func (sys *System) SyncNetworkMonitors(configs []monitor.Config) error {
_, err := sys.syncNetworkMonitors(monitor.SyncRequest{Action: monitor.SyncActionReplace, Configs: configs}) _, err := sys.syncNetworkMonitors(monitor.SyncRequest{Action: monitor.SyncActionReplace, Configs: configs})

View File

@@ -56,6 +56,9 @@ type System struct {
smartInterval time.Duration // Interval for periodic SMART data updates smartInterval time.Duration // Interval for periodic SMART data updates
zfsFetching atomic.Bool // True if ZFS pools are currently being fetched zfsFetching atomic.Bool // True if ZFS pools are currently being fetched
zfsInterval time.Duration // Interval for periodic ZFS detail data updates zfsInterval time.Duration // Interval for periodic ZFS detail data updates
// A fresh connection needs a full monitor configuration sync.
monitorsNeedSync atomic.Bool
// Serialize persistence from scheduled updates and resumes through commit. // Serialize persistence from scheduled updates and resumes through commit.
recordsMu sync.Mutex recordsMu sync.Mutex
// Protected by recordsMu; realtime reads don't consume probes. // Protected by recordsMu; realtime reads don't consume probes.
@@ -630,7 +633,10 @@ func (sys *System) request(ctx context.Context, action common.WebSocketAction, r
err := sys.sshTransport.RequestWithRetry(ctx, action, req, dest, 1) err := sys.sshTransport.RequestWithRetry(ctx, action, req, dest, 1)
// Keep legacy SSH client/version fields in sync for other code paths. // Keep legacy SSH client/version fields in sync for other code paths.
if sys.sshTransport != nil { if sys.sshTransport != nil {
sys.client.Store(sys.sshTransport.GetClient()) client := sys.sshTransport.GetClient()
if previous := sys.client.Swap(client); client != nil && client != previous {
sys.monitorsNeedSync.Store(true)
}
sys.agentVersion = sys.sshTransport.GetAgentVersion() sys.agentVersion = sys.sshTransport.GetAgentVersion()
} }
return err return err
@@ -688,6 +694,7 @@ func (sys *System) fetchDataFromAgent(options common.DataRequestOptions) (*syste
if sys.WsConn != nil && sys.WsConn.IsConnected() { if sys.WsConn != nil && sys.WsConn.IsConnected() {
wsData, err := sys.fetchDataViaWebSocket(options) wsData, err := sys.fetchDataViaWebSocket(options)
if err == nil { if err == nil {
sys.syncPendingNetworkMonitors()
return wsData, nil return wsData, nil
} }
// close the WebSocket connection if error and try SSH // close the WebSocket connection if error and try SSH
@@ -698,6 +705,7 @@ func (sys *System) fetchDataFromAgent(options common.DataRequestOptions) (*syste
if err != nil { if err != nil {
return nil, err return nil, err
} }
sys.syncPendingNetworkMonitors()
return sshData, nil return sshData, nil
} }
@@ -932,6 +940,7 @@ func (s *System) createSSHClient() error {
return err return err
} }
s.agentVersion, _ = extractAgentVersion(string(client.Conn.ServerVersion())) s.agentVersion, _ = extractAgentVersion(string(client.Conn.ServerVersion()))
s.monitorsNeedSync.Store(true)
s.manager.resetFailedSmartFetchState(s.Id) s.manager.resetFailedSmartFetchState(s.Id)
s.manager.resetFailedZfsFetchState(s.Id) s.manager.resetFailedZfsFetchState(s.Id)
return nil return nil

View File

@@ -349,23 +349,14 @@ func (sm *SystemManager) AddWebSocketSystem(systemId string, agentVersion semver
system := sm.NewSystem(systemId) system := sm.NewSystem(systemId)
system.WsConn = wsConn system.WsConn = wsConn
system.agentVersion = agentVersion system.agentVersion = agentVersion
system.monitorsNeedSync.Store(true)
if err := sm.AddRecord(systemRecord, system); err != nil { if err := sm.AddRecord(systemRecord, system); err != nil {
return err return err
} }
// Sync network monitors to the newly connected agent // Sync network monitors to the newly connected agent
go func() { go system.syncPendingNetworkMonitors()
configs, err := sm.GetMonitorConfigsForSystem(systemId)
if err != nil {
sm.hub.Logger().Warn("failed to load monitors for agent", "system", systemId, "err", err)
return
}
// An empty set must also replace any probes retained across a disconnect.
if err := system.SyncNetworkMonitors(configs); err != nil {
sm.hub.Logger().Warn("failed to sync monitors to agent", "system", systemId, "err", err)
}
}()
return nil return nil
} }

View File

@@ -59,7 +59,11 @@ export const containerChartCols: ColumnDef<ContainerRecord>[] = [
const allSystems = $allSystemsById.get() const allSystems = $allSystemsById.get()
const systemNameA = allSystems[a.original.system]?.name ?? "" const systemNameA = allSystems[a.original.system]?.name ?? ""
const systemNameB = allSystems[b.original.system]?.name ?? "" const systemNameB = allSystems[b.original.system]?.name ?? ""
return systemNameA.localeCompare(systemNameB) const primary = systemNameA.localeCompare(systemNameB)
if (primary !== 0) {
return primary
}
return a.original.name.localeCompare(b.original.name)
}, },
header: ({ column }) => <HeaderButton column={column} name={t`System`} Icon={ServerIcon} />, header: ({ column }) => <HeaderButton column={column} name={t`System`} Icon={ServerIcon} />,
cell: ({ getValue }) => { cell: ({ getValue }) => {

View File

@@ -2,9 +2,9 @@ apiVersion: v1
description: Installs beszel-agent in kubernetes description: Installs beszel-agent in kubernetes
home: https://github.com/henrygd/beszel/tree/main/supplemental/helm/beszel-agent home: https://github.com/henrygd/beszel/tree/main/supplemental/helm/beszel-agent
name: beszel-agent name: beszel-agent
appVersion: "0.19.0" appVersion: "0.20.0"
# Bump this version when publishing chart changes. # Bump this version when publishing chart changes.
version: 0.1.6 version: 0.1.7
sources: sources:
- https://github.com/henrygd/beszel/tree/main/supplemental/helm/beszel-agent - https://github.com/henrygd/beszel/tree/main/supplemental/helm/beszel-agent
- https://www.beszel.dev/ - https://www.beszel.dev/

View File

@@ -80,7 +80,7 @@ Essential parameters to configure:
| `secret.sshKey` | `ssh-key` | Key name in the secret for the SSH public key | | `secret.sshKey` | `ssh-key` | Key name in the secret for the SSH public key |
| `secret.tokenKey` | `token` | Key name in the secret for the authentication token | | `secret.tokenKey` | `token` | Key name in the secret for the authentication token |
| `image.repository` | `henrygd/beszel-agent` | Container image | | `image.repository` | `henrygd/beszel-agent` | Container image |
| `image.tag` | Chart AppVersion (0.19.0) | Image version | | `image.tag` | Chart AppVersion (0.20.0) | Image version |
| `hostNetwork` | `false` | Use host network for network monitoring | | `hostNetwork` | `false` | Use host network for network monitoring |
| `tolerations` | Allows all taints | Tolerations for running on tainted nodes | | `tolerations` | Allows all taints | Tolerations for running on tainted nodes |
@@ -385,7 +385,7 @@ helm upgrade beszel-agent ./beszel-agent \
# Change image version # Change image version
helm upgrade beszel-agent ./beszel-agent \ helm upgrade beszel-agent ./beszel-agent \
--set image.tag="0.19.0" --set image.tag="0.20.0"
``` ```
### Restart All Agents ### Restart All Agents
@@ -522,7 +522,7 @@ kubectl get secret beszel-agent -o jsonpath='{.data.ssh-key}' | base64 -d
## Chart Information ## Chart Information
- **Chart Version**: 0.1.0 - **Chart Version**: 0.1.0
- **App Version**: 0.19.0 - **App Version**: 0.20.0
- **Kubernetes Version**: 1.19+ - **Kubernetes Version**: 1.19+
- **Maintainer**: cloudwithdan (nikoloskid@pm.me) - **Maintainer**: cloudwithdan (nikoloskid@pm.me)

View File

@@ -2,9 +2,9 @@ apiVersion: v1
description: Installs beszel-hub in kubernetes description: Installs beszel-hub in kubernetes
home: https://github.com/henrygd/beszel/tree/main/supplemental/helm/beszel-hub home: https://github.com/henrygd/beszel/tree/main/supplemental/helm/beszel-hub
name: beszel-hub name: beszel-hub
appVersion: "0.19.0" appVersion: "0.20.0"
# Bump this version when publishing chart changes. # Bump this version when publishing chart changes.
version: 0.1.6 version: 0.1.7
sources: sources:
- https://github.com/henrygd/beszel/tree/main/supplemental/helm/beszel-hub - https://github.com/henrygd/beszel/tree/main/supplemental/helm/beszel-hub
- https://www.beszel.dev/ - https://www.beszel.dev/

View File

@@ -47,7 +47,7 @@ Key configuration options in `values.yaml`:
|-----------|---------|-------------| |-----------|---------|-------------|
| `replicaCount` | `1` | Number of Beszel Hub replicas | | `replicaCount` | `1` | Number of Beszel Hub replicas |
| `image.repository` | `henrygd/beszel` | Container image repository | | `image.repository` | `henrygd/beszel` | Container image repository |
| `image.tag` | Chart AppVersion (0.19.0) | Container image tag | | `image.tag` | Chart AppVersion (0.20.0) | Container image tag |
| `image.pullPolicy` | `IfNotPresent` | Image pull policy | | `image.pullPolicy` | `IfNotPresent` | Image pull policy |
| `service.port` | `8090` | Service port | | `service.port` | `8090` | Service port |
| `persistentVolumeClaim.enabled` | `true` | Enable persistent volume | | `persistentVolumeClaim.enabled` | `true` | Enable persistent volume |
@@ -169,7 +169,7 @@ tolerations:
```yaml ```yaml
replicaCount: 3 replicaCount: 3
image: image:
tag: "0.19.0" tag: "0.20.0"
service: service:
type: LoadBalancer type: LoadBalancer
ingress: ingress:
@@ -330,7 +330,7 @@ By default, Beszel Hub uses a PersistentVolumeClaim for data storage. Ensure you
## Chart Information ## Chart Information
- **Chart Version**: 0.1.0 - **Chart Version**: 0.1.0
- **App Version**: 0.19.0 - **App Version**: 0.20.0
- **Kubernetes Version**: 1.19+ - **Kubernetes Version**: 1.19+
- **Maintainer**: cloudwithdan (nikoloskid@pm.me) - **Maintainer**: cloudwithdan (nikoloskid@pm.me)