mirror of
https://github.com/henrygd/beszel.git
synced 2026-09-24 02:17:47 +02:00
Compare commits
7 Commits
b89314889d
...
dev-probes
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3534552d37 | ||
|
|
723401819f | ||
|
|
2ea576c989 | ||
|
|
526a2c6aab | ||
|
|
aaa8eb773f | ||
|
|
099935e78e | ||
|
|
d2eb3b259a |
@@ -221,6 +221,5 @@ func (h *SyncNetworkProbesHandler) Handle(hctx *HandlerContext) error {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
slog.Info("network probes synced", "action", req.Action)
|
|
||||||
return hctx.SendResponse(resp, hctx.RequestID)
|
return hctx.SendResponse(resp, hctx.RequestID)
|
||||||
}
|
}
|
||||||
|
|||||||
129
agent/probe.go
129
agent/probe.go
@@ -17,21 +17,20 @@ import (
|
|||||||
"github.com/henrygd/beszel/internal/entities/probe"
|
"github.com/henrygd/beszel/internal/entities/probe"
|
||||||
)
|
)
|
||||||
|
|
||||||
// Probe functionality overview:
|
|
||||||
// Probes run at user-defined intervals (e.g., every 10s).
|
// Probes run at user-defined intervals (e.g., every 10s).
|
||||||
// To keep memory usage low and constant, data is stored in two layers:
|
// To keep memory usage low and constant, data is stored in two layers:
|
||||||
// 1. Raw samples: The most recent individual results (kept for probeRawRetention).
|
// 1. Raw samples: The most recent individual results (kept for probeRawRetention).
|
||||||
// 2. Minute buckets: A fixed-size ring buffer of 61 buckets, each representing one
|
// 2. Minute buckets: A ring buffer of 61 buckets, each representing one
|
||||||
// wall-clock minute. Samples collected within the same minute are aggregated
|
// wall-clock minute. Samples collected within the same minute are aggregated
|
||||||
// (sum, min, max, count) into a single bucket.
|
// (sum, min, max, count) into a single bucket.
|
||||||
//
|
//
|
||||||
// Short-term requests (<= 2m) use raw samples for perfect accuracy.
|
// Short-term requests (<= 70s) use raw samples.
|
||||||
// Long-term requests (up to 1h) use the minute buckets to avoid storing thousands
|
// Long-term requests (up to 1h) use the minute buckets to avoid storing thousands
|
||||||
// of individual data points.
|
// of individual data points.
|
||||||
|
|
||||||
const (
|
const (
|
||||||
// probeRawRetention is the duration to keep individual samples for high-precision short-term requests
|
// probeRawRetention is the duration to keep individual samples
|
||||||
probeRawRetention = 70 * time.Second
|
probeRawRetention = 61 * time.Second
|
||||||
// probeMinuteBucketLen is the number of 1-minute buckets to keep (1 hour + 1 for partials)
|
// probeMinuteBucketLen is the number of 1-minute buckets to keep (1 hour + 1 for partials)
|
||||||
probeMinuteBucketLen int32 = 61
|
probeMinuteBucketLen int32 = 61
|
||||||
)
|
)
|
||||||
@@ -147,27 +146,27 @@ func (agg probeAggregate) hasData() bool {
|
|||||||
return agg.totalCount > 0
|
return agg.totalCount > 0
|
||||||
}
|
}
|
||||||
|
|
||||||
// result converts the aggregate into the probe result slice format.
|
// result converts the aggregate into the probe result format.
|
||||||
func (agg probeAggregate) result() probe.Result {
|
func (agg probeAggregate) result() probe.Result {
|
||||||
avg := agg.avgResponse()
|
avg := agg.avgResponse()
|
||||||
minUs := 0.0
|
result := probe.Result{
|
||||||
if agg.successCount > 0 {
|
AvgResponse: avg,
|
||||||
minUs = float64(agg.minUs)
|
MinResponse: agg.minUs,
|
||||||
|
MaxResponse: agg.maxUs,
|
||||||
|
PacketLoss: agg.lossPercentage(),
|
||||||
}
|
}
|
||||||
return probe.Result{
|
if agg.successCount == 0 {
|
||||||
avg,
|
result.MinResponse, result.MaxResponse = 0, 0
|
||||||
minUs,
|
|
||||||
float64(agg.maxUs),
|
|
||||||
agg.lossPercentage(),
|
|
||||||
}
|
}
|
||||||
|
return result
|
||||||
}
|
}
|
||||||
|
|
||||||
// avgResponse returns the rounded average of successful samples.
|
// avgResponse returns the rounded average of successful samples.
|
||||||
func (agg probeAggregate) avgResponse() float64 {
|
func (agg probeAggregate) avgResponse() int64 {
|
||||||
if agg.successCount == 0 {
|
if agg.successCount == 0 {
|
||||||
return 0
|
return 0
|
||||||
}
|
}
|
||||||
return float64(agg.sumUs / agg.successCount)
|
return agg.sumUs / agg.successCount
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -333,7 +332,8 @@ func (pm *ProbeManager) runProbe(task *probeTask, runNow bool) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
stagger := getStagger(interval.Milliseconds())
|
stagger := getStagger(interval.Milliseconds())
|
||||||
slog.Info("starting probe task", "id", task.config.ID, "initial_delay", stagger.String(), "interval", interval.String())
|
|
||||||
|
slog.Debug("starting probe task", "target", task.config.Target, "delay", stagger.String(), "interval", interval.String())
|
||||||
|
|
||||||
if runNow {
|
if runNow {
|
||||||
pm.executeProbe(task)
|
pm.executeProbe(task)
|
||||||
@@ -341,10 +341,9 @@ func (pm *ProbeManager) runProbe(task *probeTask, runNow bool) {
|
|||||||
|
|
||||||
select {
|
select {
|
||||||
case <-task.cancel:
|
case <-task.cancel:
|
||||||
slog.Info("removed probe", "id", task.config.ID)
|
// slog.Info("removed probe", "target", task.config.Target)
|
||||||
return
|
return
|
||||||
case <-time.After(stagger):
|
case <-time.After(stagger):
|
||||||
slog.Info("initial probe execution", "id", task.config.ID)
|
|
||||||
pm.executeProbe(task)
|
pm.executeProbe(task)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -353,16 +352,15 @@ func (pm *ProbeManager) runProbe(task *probeTask, runNow bool) {
|
|||||||
for {
|
for {
|
||||||
select {
|
select {
|
||||||
case <-task.cancel:
|
case <-task.cancel:
|
||||||
slog.Info("removed probe", "id", task.config.ID)
|
// slog.Info("removed probe", "target", task.config.Target)
|
||||||
return
|
return
|
||||||
case <-ticker:
|
case <-ticker:
|
||||||
slog.Info("running probe in main loop", "id", task.config.ID, "interval", interval.String())
|
|
||||||
pm.executeProbe(task)
|
pm.executeProbe(task)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// getStagger returns a random duration between intervalSeconds/2 and intervalSeconds to stagger probe executions
|
// getStagger returns a random duration between intervalSeconds/2 and intervalSeconds to stagger initial probe executions
|
||||||
func getStagger(intervalMilli int64) time.Duration {
|
func getStagger(intervalMilli int64) time.Duration {
|
||||||
intervalMilliInt := int(intervalMilli)
|
intervalMilliInt := int(intervalMilli)
|
||||||
randomDelayInt := rand.Intn(intervalMilliInt)
|
randomDelayInt := rand.Intn(intervalMilliInt)
|
||||||
@@ -383,6 +381,27 @@ func (pm *ProbeManager) runProbeNow(task *probeTask) *probe.Result {
|
|||||||
return &result
|
return &result
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// resultLocked returns the aggregated probe result for the requested duration along with a bool indicating whether any data was available.
|
||||||
|
func (task *probeTask) resultLocked(duration time.Duration, now time.Time) (probe.Result, bool) {
|
||||||
|
agg := task.aggregateLocked(duration, now)
|
||||||
|
hourAgg := task.aggregateLocked(time.Hour, now)
|
||||||
|
if !agg.hasData() {
|
||||||
|
return probe.Result{}, false
|
||||||
|
}
|
||||||
|
|
||||||
|
result := agg.result()
|
||||||
|
|
||||||
|
result.AvgResponse1h = hourAgg.avgResponse()
|
||||||
|
result.MinResponse1h = hourAgg.minUs
|
||||||
|
result.MaxResponse1h = hourAgg.maxUs
|
||||||
|
result.PacketLoss1h = hourAgg.lossPercentage()
|
||||||
|
|
||||||
|
if hourAgg.successCount == 0 {
|
||||||
|
result.MinResponse1h, result.MaxResponse1h = 0, 0
|
||||||
|
}
|
||||||
|
return result, true
|
||||||
|
}
|
||||||
|
|
||||||
// aggregateLocked collects probe data for the requested time window.
|
// aggregateLocked collects probe data for the requested time window.
|
||||||
func (task *probeTask) aggregateLocked(duration time.Duration, now time.Time) probeAggregate {
|
func (task *probeTask) aggregateLocked(duration time.Duration, now time.Time) probeAggregate {
|
||||||
cutoff := now.Add(-duration)
|
cutoff := now.Add(-duration)
|
||||||
@@ -393,40 +412,6 @@ func (task *probeTask) aggregateLocked(duration time.Duration, now time.Time) pr
|
|||||||
return aggregateBucketsSince(task.buckets[:], cutoff, now)
|
return aggregateBucketsSince(task.buckets[:], cutoff, now)
|
||||||
}
|
}
|
||||||
|
|
||||||
// resultLocked returns the aggregated probe result for the requested duration along with a bool indicating whether any data was available.
|
|
||||||
func (task *probeTask) resultLocked(duration time.Duration, now time.Time) (probe.Result, bool) {
|
|
||||||
agg := task.aggregateLocked(duration, now)
|
|
||||||
hourAgg := task.aggregateLocked(time.Hour, now)
|
|
||||||
if !agg.hasData() {
|
|
||||||
return nil, false
|
|
||||||
}
|
|
||||||
|
|
||||||
result := agg.result()
|
|
||||||
|
|
||||||
res := result[0]
|
|
||||||
res1h := hourAgg.avgResponse()
|
|
||||||
resMin := result[1]
|
|
||||||
resMin1h := float64(hourAgg.minUs)
|
|
||||||
resMax := result[2]
|
|
||||||
resMax1h := float64(hourAgg.maxUs)
|
|
||||||
loss := result[3]
|
|
||||||
loss1h := hourAgg.lossPercentage()
|
|
||||||
|
|
||||||
if hourAgg.successCount == 0 {
|
|
||||||
resMin1h, resMax1h = 0, 0
|
|
||||||
}
|
|
||||||
return probe.Result{
|
|
||||||
res,
|
|
||||||
res1h,
|
|
||||||
resMin,
|
|
||||||
resMin1h,
|
|
||||||
resMax,
|
|
||||||
resMax1h,
|
|
||||||
loss,
|
|
||||||
loss1h,
|
|
||||||
}, true
|
|
||||||
}
|
|
||||||
|
|
||||||
// aggregateSamplesSince aggregates raw samples newer than the cutoff.
|
// aggregateSamplesSince aggregates raw samples newer than the cutoff.
|
||||||
func aggregateSamplesSince(samples []probeSample, cutoff time.Time) probeAggregate {
|
func aggregateSamplesSince(samples []probeSample, cutoff time.Time) probeAggregate {
|
||||||
agg := newProbeAggregate()
|
agg := newProbeAggregate()
|
||||||
@@ -485,20 +470,26 @@ func (task *probeTask) addSampleLocked(sample probeSample) {
|
|||||||
|
|
||||||
// executeProbe runs the configured probe and records the sample.
|
// executeProbe runs the configured probe and records the sample.
|
||||||
func (pm *ProbeManager) executeProbe(task *probeTask) {
|
func (pm *ProbeManager) executeProbe(task *probeTask) {
|
||||||
|
// slog.Info("running probe", "id", task.config.ID, "interval", task.config.Interval)
|
||||||
var responseUs int64
|
var responseUs int64
|
||||||
|
var err error
|
||||||
|
|
||||||
switch task.config.Protocol {
|
switch task.config.Protocol {
|
||||||
case "icmp":
|
case "icmp":
|
||||||
responseUs = probeICMP(task.config.Target)
|
responseUs, err = probeICMP(task.config.Target)
|
||||||
case "tcp":
|
case "tcp":
|
||||||
responseUs = probeTCP(task.config.Target, task.config.Port)
|
responseUs, err = probeTCP(task.config.Target, task.config.Port)
|
||||||
case "http":
|
case "http":
|
||||||
responseUs = probeHTTP(pm.httpClient, task.config.Target)
|
responseUs, err = probeHTTP(pm.httpClient, task.config.Target)
|
||||||
default:
|
default:
|
||||||
slog.Warn("unknown probe protocol", "protocol", task.config.Protocol)
|
slog.Warn("unknown probe protocol", "protocol", task.config.Protocol)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if err != nil {
|
||||||
|
slog.Warn("probe failed", "err", err, "target", task.config.Target, "protocol", task.config.Protocol)
|
||||||
|
}
|
||||||
|
|
||||||
sample := probeSample{
|
sample := probeSample{
|
||||||
responseUs: responseUs,
|
responseUs: responseUs,
|
||||||
timestamp: time.Now(),
|
timestamp: time.Now(),
|
||||||
@@ -510,12 +501,12 @@ func (pm *ProbeManager) executeProbe(task *probeTask) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// probeTCP measures pure TCP handshake response (excluding DNS resolution).
|
// probeTCP measures pure TCP handshake response (excluding DNS resolution).
|
||||||
// Returns -1 on failure.
|
// Returns -1 and an error on failure.
|
||||||
func probeTCP(target string, port uint16) int64 {
|
func probeTCP(target string, port uint16) (int64, error) {
|
||||||
// Resolve DNS first, outside the timing window
|
// Resolve DNS first, outside the timing window
|
||||||
ips, err := net.LookupHost(target)
|
ips, err := net.LookupHost(target)
|
||||||
if err != nil || len(ips) == 0 {
|
if err != nil || len(ips) == 0 {
|
||||||
return -1
|
return -1, err
|
||||||
}
|
}
|
||||||
addr := net.JoinHostPort(ips[0], fmt.Sprintf("%d", port))
|
addr := net.JoinHostPort(ips[0], fmt.Sprintf("%d", port))
|
||||||
|
|
||||||
@@ -523,25 +514,25 @@ func probeTCP(target string, port uint16) int64 {
|
|||||||
start := time.Now()
|
start := time.Now()
|
||||||
conn, err := net.DialTimeout("tcp", addr, 3*time.Second)
|
conn, err := net.DialTimeout("tcp", addr, 3*time.Second)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return -1
|
return -1, err
|
||||||
}
|
}
|
||||||
conn.Close()
|
conn.Close()
|
||||||
return time.Since(start).Microseconds()
|
return time.Since(start).Microseconds(), nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// probeHTTP measures HTTP GET request response in microseconds. Returns -1 on failure.
|
// probeHTTP measures HTTP GET request response in microseconds. Returns -1 and an error on failure.
|
||||||
func probeHTTP(client *http.Client, url string) int64 {
|
func probeHTTP(client *http.Client, url string) (int64, error) {
|
||||||
if client == nil {
|
if client == nil {
|
||||||
client = http.DefaultClient
|
client = http.DefaultClient
|
||||||
}
|
}
|
||||||
start := time.Now()
|
start := time.Now()
|
||||||
resp, err := client.Get(url)
|
resp, err := client.Get(url)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return -1
|
return -1, err
|
||||||
}
|
}
|
||||||
resp.Body.Close()
|
resp.Body.Close()
|
||||||
if resp.StatusCode >= 400 {
|
if resp.StatusCode >= 400 {
|
||||||
return -1
|
return -1, fmt.Errorf("HTTP error: %s", resp.Status)
|
||||||
}
|
}
|
||||||
return time.Since(start).Microseconds()
|
return time.Since(start).Microseconds(), nil
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,6 +1,7 @@
|
|||||||
package agent
|
package agent
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"errors"
|
||||||
"math"
|
"math"
|
||||||
"net"
|
"net"
|
||||||
"os"
|
"os"
|
||||||
@@ -27,7 +28,7 @@ type icmpPacketConn interface {
|
|||||||
// icmpMethod tracks which ICMP approach to use. Once a method succeeds or
|
// icmpMethod tracks which ICMP approach to use. Once a method succeeds or
|
||||||
// all native methods fail, the choice is cached so subsequent probes skip
|
// all native methods fail, the choice is cached so subsequent probes skip
|
||||||
// the trial-and-error overhead.
|
// the trial-and-error overhead.
|
||||||
type icmpMethod int
|
type icmpMethod uint8
|
||||||
|
|
||||||
const (
|
const (
|
||||||
icmpUntried icmpMethod = iota // haven't tried yet
|
icmpUntried icmpMethod = iota // haven't tried yet
|
||||||
@@ -76,11 +77,11 @@ var (
|
|||||||
// Supports both IPv4 and IPv6 targets. The ICMP method (raw socket,
|
// Supports both IPv4 and IPv6 targets. The ICMP method (raw socket,
|
||||||
// unprivileged datagram, or exec fallback) is detected once per address
|
// unprivileged datagram, or exec fallback) is detected once per address
|
||||||
// family and cached for subsequent probes.
|
// family and cached for subsequent probes.
|
||||||
// Returns response in microseconds, or -1 on failure.
|
// Returns response in microseconds, or -1 and an error on failure.
|
||||||
func probeICMP(target string) int64 {
|
func probeICMP(target string) (int64, error) {
|
||||||
family, ip := resolveICMPTarget(target)
|
family, ip, err := resolveICMPTarget(target)
|
||||||
if family == nil {
|
if err != nil {
|
||||||
return -1
|
return -1, err
|
||||||
}
|
}
|
||||||
|
|
||||||
icmpModeMu.Lock()
|
icmpModeMu.Lock()
|
||||||
@@ -98,30 +99,30 @@ func probeICMP(target string) int64 {
|
|||||||
case icmpExecFallback:
|
case icmpExecFallback:
|
||||||
return probeICMPExec(target, family.isIPv6)
|
return probeICMPExec(target, family.isIPv6)
|
||||||
default:
|
default:
|
||||||
return -1
|
return -1, errors.New("unsupported ICMP mode")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// resolveICMPTarget resolves a target hostname or IP to determine the address
|
// resolveICMPTarget resolves a target hostname or IP to determine the address
|
||||||
// family and concrete IP address. Prefers IPv4 for dual-stack hostnames.
|
// family and concrete IP address. Prefers IPv4 for dual-stack hostnames.
|
||||||
func resolveICMPTarget(target string) (*icmpFamily, net.IP) {
|
func resolveICMPTarget(target string) (*icmpFamily, net.IP, error) {
|
||||||
if ip := net.ParseIP(target); ip != nil {
|
if ip := net.ParseIP(target); ip != nil {
|
||||||
if ip.To4() != nil {
|
if ip.To4() != nil {
|
||||||
return &icmpV4, ip.To4()
|
return &icmpV4, ip.To4(), nil
|
||||||
}
|
}
|
||||||
return &icmpV6, ip
|
return &icmpV6, ip, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
ips, err := net.LookupIP(target)
|
ips, err := net.LookupIP(target)
|
||||||
if err != nil || len(ips) == 0 {
|
if err != nil || len(ips) == 0 {
|
||||||
return nil, nil
|
return nil, nil, err
|
||||||
}
|
}
|
||||||
for _, ip := range ips {
|
for _, ip := range ips {
|
||||||
if v4 := ip.To4(); v4 != nil {
|
if v4 := ip.To4(); v4 != nil {
|
||||||
return &icmpV4, v4
|
return &icmpV4, v4, nil
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return &icmpV6, ips[0]
|
return &icmpV6, ips[0], nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func detectICMPMode(family *icmpFamily, listen func(network, listenAddr string) (icmpPacketConn, error)) icmpMethod {
|
func detectICMPMode(family *icmpFamily, listen func(network, listenAddr string) (icmpPacketConn, error)) icmpMethod {
|
||||||
@@ -130,31 +131,28 @@ func detectICMPMode(family *icmpFamily, listen func(network, listenAddr string)
|
|||||||
label = "IPv6"
|
label = "IPv6"
|
||||||
}
|
}
|
||||||
|
|
||||||
if conn, err := listen(family.rawNetwork, family.listenAddr); err == nil {
|
conn, err := listen(family.rawNetwork, family.listenAddr)
|
||||||
|
slog.Debug("ICMP raw socket test", "family", label, "err", err)
|
||||||
|
if err == nil {
|
||||||
conn.Close()
|
conn.Close()
|
||||||
slog.Info("ICMP probe using raw socket", "family", label)
|
|
||||||
return icmpRaw
|
return icmpRaw
|
||||||
} else {
|
|
||||||
slog.Debug("ICMP raw socket unavailable", "family", label, "err", err)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
if conn, err := listen(family.dgramNetwork, family.listenAddr); err == nil {
|
conn, err = listen(family.dgramNetwork, family.listenAddr)
|
||||||
|
slog.Debug("ICMP datagram socket test", "family", label, "err", err)
|
||||||
|
if err == nil {
|
||||||
conn.Close()
|
conn.Close()
|
||||||
slog.Info("ICMP probe using unprivileged datagram socket", "family", label)
|
|
||||||
return icmpDatagram
|
return icmpDatagram
|
||||||
} else {
|
|
||||||
slog.Debug("ICMP datagram socket unavailable", "family", label, "err", err)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
slog.Info("ICMP probe falling back to system ping command", "family", label)
|
|
||||||
return icmpExecFallback
|
return icmpExecFallback
|
||||||
}
|
}
|
||||||
|
|
||||||
// probeICMPNative sends an ICMP echo request using Go's x/net/icmp package.
|
// probeICMPNative sends an ICMP echo request using Go's x/net/icmp package.
|
||||||
func probeICMPNative(network string, family *icmpFamily, dst net.Addr) int64 {
|
func probeICMPNative(network string, family *icmpFamily, dst net.Addr) (int64, error) {
|
||||||
conn, err := icmp.ListenPacket(network, family.listenAddr)
|
conn, err := icmp.ListenPacket(network, family.listenAddr)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return -1
|
return -1, err
|
||||||
}
|
}
|
||||||
defer conn.Close()
|
defer conn.Close()
|
||||||
|
|
||||||
@@ -170,7 +168,7 @@ func probeICMPNative(network string, family *icmpFamily, dst net.Addr) int64 {
|
|||||||
}
|
}
|
||||||
msgBytes, err := msg.Marshal(nil)
|
msgBytes, err := msg.Marshal(nil)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return -1
|
return -1, err
|
||||||
}
|
}
|
||||||
|
|
||||||
// Set deadline before sending
|
// Set deadline before sending
|
||||||
@@ -178,7 +176,7 @@ func probeICMPNative(network string, family *icmpFamily, dst net.Addr) int64 {
|
|||||||
|
|
||||||
start := time.Now()
|
start := time.Now()
|
||||||
if _, err := conn.WriteTo(msgBytes, dst); err != nil {
|
if _, err := conn.WriteTo(msgBytes, dst); err != nil {
|
||||||
return -1
|
return -1, err
|
||||||
}
|
}
|
||||||
|
|
||||||
// Read reply
|
// Read reply
|
||||||
@@ -186,23 +184,23 @@ func probeICMPNative(network string, family *icmpFamily, dst net.Addr) int64 {
|
|||||||
for {
|
for {
|
||||||
n, _, err := conn.ReadFrom(buf)
|
n, _, err := conn.ReadFrom(buf)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return -1
|
return -1, err
|
||||||
}
|
}
|
||||||
|
|
||||||
reply, err := icmp.ParseMessage(family.proto, buf[:n])
|
reply, err := icmp.ParseMessage(family.proto, buf[:n])
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return -1
|
return -1, err
|
||||||
}
|
}
|
||||||
|
|
||||||
if reply.Type == family.replyType {
|
if reply.Type == family.replyType {
|
||||||
return time.Since(start).Microseconds()
|
return time.Since(start).Microseconds(), nil
|
||||||
}
|
}
|
||||||
// Ignore non-echo-reply messages (e.g. destination unreachable) and keep reading
|
// Ignore non-echo-reply messages (e.g. destination unreachable) and keep reading
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// probeICMPExec falls back to the system ping command. Returns -1 on failure.
|
// probeICMPExec falls back to the system ping command. Returns -1 and an error on failure.
|
||||||
func probeICMPExec(target string, isIPv6 bool) int64 {
|
func probeICMPExec(target string, isIPv6 bool) (int64, error) {
|
||||||
var cmd *exec.Cmd
|
var cmd *exec.Cmd
|
||||||
switch runtime.GOOS {
|
switch runtime.GOOS {
|
||||||
case "windows":
|
case "windows":
|
||||||
@@ -211,7 +209,7 @@ func probeICMPExec(target string, isIPv6 bool) int64 {
|
|||||||
} else {
|
} else {
|
||||||
cmd = exec.Command("ping", "-n", "1", "-w", "3000", target)
|
cmd = exec.Command("ping", "-n", "1", "-w", "3000", target)
|
||||||
}
|
}
|
||||||
default: // linux, darwin, freebsd
|
default:
|
||||||
if isIPv6 {
|
if isIPv6 {
|
||||||
cmd = exec.Command("ping", "-6", "-c", "1", "-W", "3", target)
|
cmd = exec.Command("ping", "-6", "-c", "1", "-W", "3", target)
|
||||||
} else {
|
} else {
|
||||||
@@ -224,20 +222,20 @@ func probeICMPExec(target string, isIPv6 bool) int64 {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
// If ping fails but we got output, still try to parse
|
// If ping fails but we got output, still try to parse
|
||||||
if len(output) == 0 {
|
if len(output) == 0 {
|
||||||
return -1
|
return -1, err
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
matches := pingTimeRegex.FindSubmatch(output)
|
matches := pingTimeRegex.FindSubmatch(output)
|
||||||
if len(matches) >= 2 {
|
if len(matches) >= 2 {
|
||||||
if ms, err := strconv.ParseFloat(string(matches[1]), 64); err == nil {
|
if ms, err := strconv.ParseFloat(string(matches[1]), 64); err == nil {
|
||||||
return int64(math.Round(ms * 1000))
|
return int64(math.Round(ms * 1000)), nil
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Fallback: use wall clock time if ping succeeded but parsing failed
|
// Fallback: use wall clock time if ping succeeded but parsing failed
|
||||||
if err == nil {
|
if err == nil {
|
||||||
return time.Since(start).Microseconds()
|
return time.Since(start).Microseconds(), nil
|
||||||
}
|
}
|
||||||
return -1
|
return -1, err
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -96,21 +96,24 @@ func TestDetectICMPMode(t *testing.T) {
|
|||||||
|
|
||||||
func TestResolveICMPTarget(t *testing.T) {
|
func TestResolveICMPTarget(t *testing.T) {
|
||||||
t.Run("IPv4 literal", func(t *testing.T) {
|
t.Run("IPv4 literal", func(t *testing.T) {
|
||||||
family, ip := resolveICMPTarget("127.0.0.1")
|
family, ip, err := resolveICMPTarget("127.0.0.1")
|
||||||
|
require.NoError(t, err)
|
||||||
require.NotNil(t, family)
|
require.NotNil(t, family)
|
||||||
assert.False(t, family.isIPv6)
|
assert.False(t, family.isIPv6)
|
||||||
assert.Equal(t, "127.0.0.1", ip.String())
|
assert.Equal(t, "127.0.0.1", ip.String())
|
||||||
})
|
})
|
||||||
|
|
||||||
t.Run("IPv6 literal", func(t *testing.T) {
|
t.Run("IPv6 literal", func(t *testing.T) {
|
||||||
family, ip := resolveICMPTarget("::1")
|
family, ip, err := resolveICMPTarget("::1")
|
||||||
|
require.NoError(t, err)
|
||||||
require.NotNil(t, family)
|
require.NotNil(t, family)
|
||||||
assert.True(t, family.isIPv6)
|
assert.True(t, family.isIPv6)
|
||||||
assert.Equal(t, "::1", ip.String())
|
assert.Equal(t, "::1", ip.String())
|
||||||
})
|
})
|
||||||
|
|
||||||
t.Run("IPv4-mapped IPv6 resolves as IPv4", func(t *testing.T) {
|
t.Run("IPv4-mapped IPv6 resolves as IPv4", func(t *testing.T) {
|
||||||
family, ip := resolveICMPTarget("::ffff:127.0.0.1")
|
family, ip, err := resolveICMPTarget("::ffff:127.0.0.1")
|
||||||
|
require.NoError(t, err)
|
||||||
require.NotNil(t, family)
|
require.NotNil(t, family)
|
||||||
assert.False(t, family.isIPv6)
|
assert.False(t, family.isIPv6)
|
||||||
assert.Equal(t, "127.0.0.1", ip.String())
|
assert.Equal(t, "127.0.0.1", ip.String())
|
||||||
|
|||||||
@@ -24,10 +24,11 @@ func TestProbeTaskAggregateLockedUsesRawSamplesForShortWindows(t *testing.T) {
|
|||||||
require.True(t, agg.hasData())
|
require.True(t, agg.hasData())
|
||||||
assert.Equal(t, int64(2), agg.totalCount)
|
assert.Equal(t, int64(2), agg.totalCount)
|
||||||
assert.Equal(t, int64(1), agg.successCount)
|
assert.Equal(t, int64(1), agg.successCount)
|
||||||
assert.Equal(t, 20.0, agg.result()[0])
|
result := agg.result()
|
||||||
assert.Equal(t, 20.0, agg.result()[1])
|
assert.Equal(t, int64(20), result.AvgResponse)
|
||||||
assert.Equal(t, 20.0, agg.result()[2])
|
assert.Equal(t, int64(20), result.MinResponse)
|
||||||
assert.Equal(t, 50.0, agg.result()[3])
|
assert.Equal(t, int64(20), result.MaxResponse)
|
||||||
|
assert.Equal(t, 50.0, result.PacketLoss)
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestProbeTaskAggregateLockedUsesMinuteBucketsForLongWindows(t *testing.T) {
|
func TestProbeTaskAggregateLockedUsesMinuteBucketsForLongWindows(t *testing.T) {
|
||||||
@@ -44,10 +45,11 @@ func TestProbeTaskAggregateLockedUsesMinuteBucketsForLongWindows(t *testing.T) {
|
|||||||
require.True(t, agg.hasData())
|
require.True(t, agg.hasData())
|
||||||
assert.Equal(t, int64(4), agg.totalCount)
|
assert.Equal(t, int64(4), agg.totalCount)
|
||||||
assert.Equal(t, int64(3), agg.successCount)
|
assert.Equal(t, int64(3), agg.successCount)
|
||||||
assert.Equal(t, 30.0, agg.result()[0])
|
result := agg.result()
|
||||||
assert.Equal(t, 20.0, agg.result()[1])
|
assert.Equal(t, int64(30), result.AvgResponse)
|
||||||
assert.Equal(t, 40.0, agg.result()[2])
|
assert.Equal(t, int64(20), result.MinResponse)
|
||||||
assert.Equal(t, 25.0, agg.result()[3])
|
assert.Equal(t, int64(40), result.MaxResponse)
|
||||||
|
assert.Equal(t, 25.0, result.PacketLoss)
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestProbeTaskAddSampleLockedTrimsRawSamplesButKeepsBucketHistory(t *testing.T) {
|
func TestProbeTaskAddSampleLockedTrimsRawSamplesButKeepsBucketHistory(t *testing.T) {
|
||||||
@@ -64,10 +66,11 @@ func TestProbeTaskAddSampleLockedTrimsRawSamplesButKeepsBucketHistory(t *testing
|
|||||||
require.True(t, agg.hasData())
|
require.True(t, agg.hasData())
|
||||||
assert.Equal(t, int64(2), agg.totalCount)
|
assert.Equal(t, int64(2), agg.totalCount)
|
||||||
assert.Equal(t, int64(2), agg.successCount)
|
assert.Equal(t, int64(2), agg.successCount)
|
||||||
assert.Equal(t, 15.0, agg.result()[0])
|
result := agg.result()
|
||||||
assert.Equal(t, 10.0, agg.result()[1])
|
assert.Equal(t, int64(15), result.AvgResponse)
|
||||||
assert.Equal(t, 20.0, agg.result()[2])
|
assert.Equal(t, int64(10), result.MinResponse)
|
||||||
assert.Equal(t, 0.0, agg.result()[3])
|
assert.Equal(t, int64(20), result.MaxResponse)
|
||||||
|
assert.Equal(t, 0.0, result.PacketLoss)
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestProbeManagerGetResultsIncludesHourResponseRange(t *testing.T) {
|
func TestProbeManagerGetResultsIncludesHourResponseRange(t *testing.T) {
|
||||||
@@ -84,13 +87,14 @@ func TestProbeManagerGetResultsIncludesHourResponseRange(t *testing.T) {
|
|||||||
results := pm.GetResults(uint16(time.Minute / time.Millisecond))
|
results := pm.GetResults(uint16(time.Minute / time.Millisecond))
|
||||||
result, ok := results["probe-1"]
|
result, ok := results["probe-1"]
|
||||||
require.True(t, ok)
|
require.True(t, ok)
|
||||||
require.Len(t, result, 6)
|
assert.Equal(t, int64(30), result.AvgResponse)
|
||||||
assert.Equal(t, 30.0, result[0])
|
assert.Equal(t, int64(25), result.AvgResponse1h)
|
||||||
assert.Equal(t, 25.0, result[1])
|
assert.Equal(t, int64(30), result.MinResponse)
|
||||||
assert.Equal(t, 10.0, result[2])
|
assert.Equal(t, int64(10), result.MinResponse1h)
|
||||||
assert.Equal(t, 40.0, result[3])
|
assert.Equal(t, int64(30), result.MaxResponse)
|
||||||
assert.Equal(t, 50.0, result[4])
|
assert.Equal(t, int64(40), result.MaxResponse1h)
|
||||||
assert.Equal(t, 20.0, result[5])
|
assert.Equal(t, 50.0, result.PacketLoss)
|
||||||
|
assert.Equal(t, 20.0, result.PacketLoss1h)
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestProbeManagerGetResultsIncludesLossOnlyHourData(t *testing.T) {
|
func TestProbeManagerGetResultsIncludesLossOnlyHourData(t *testing.T) {
|
||||||
@@ -104,13 +108,14 @@ func TestProbeManagerGetResultsIncludesLossOnlyHourData(t *testing.T) {
|
|||||||
results := pm.GetResults(uint16(time.Minute / time.Millisecond))
|
results := pm.GetResults(uint16(time.Minute / time.Millisecond))
|
||||||
result, ok := results["probe-1"]
|
result, ok := results["probe-1"]
|
||||||
require.True(t, ok)
|
require.True(t, ok)
|
||||||
require.Len(t, result, 6)
|
assert.Equal(t, int64(0), result.AvgResponse)
|
||||||
assert.Equal(t, 0.0, result[0])
|
assert.Equal(t, int64(0), result.AvgResponse1h)
|
||||||
assert.Equal(t, 0.0, result[1])
|
assert.Equal(t, int64(0), result.MinResponse)
|
||||||
assert.Equal(t, 0.0, result[2])
|
assert.Equal(t, int64(0), result.MinResponse1h)
|
||||||
assert.Equal(t, 0.0, result[3])
|
assert.Equal(t, int64(0), result.MaxResponse)
|
||||||
assert.Equal(t, 100.0, result[4])
|
assert.Equal(t, int64(0), result.MaxResponse1h)
|
||||||
assert.Equal(t, 100.0, result[5])
|
assert.Equal(t, 100.0, result.PacketLoss)
|
||||||
|
assert.Equal(t, 100.0, result.PacketLoss1h)
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestProbeConfigResultKeyUsesSyncedID(t *testing.T) {
|
func TestProbeConfigResultKeyUsesSyncedID(t *testing.T) {
|
||||||
@@ -207,10 +212,9 @@ func TestProbeManagerApplySyncUpsertRunsImmediatelyAndReturnsResult(t *testing.T
|
|||||||
defer pm.Stop()
|
defer pm.Stop()
|
||||||
|
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
require.Len(t, resp.Result, 6)
|
assert.GreaterOrEqual(t, resp.Result.AvgResponse, int64(0))
|
||||||
assert.GreaterOrEqual(t, resp.Result[0], 0.0)
|
assert.Equal(t, 0.0, resp.Result.PacketLoss)
|
||||||
assert.Equal(t, 0.0, resp.Result[4])
|
assert.Equal(t, 0.0, resp.Result.PacketLoss1h)
|
||||||
assert.Equal(t, 0.0, resp.Result[5])
|
|
||||||
|
|
||||||
task := pm.probes["probe-1"]
|
task := pm.probes["probe-1"]
|
||||||
require.NotNil(t, task)
|
require.NotNil(t, task)
|
||||||
@@ -252,7 +256,7 @@ func TestProbeManagerUpsertProbeKeepsHistoryWhenOnlyIntervalChanges(t *testing.T
|
|||||||
require.True(t, agg.hasData())
|
require.True(t, agg.hasData())
|
||||||
assert.Equal(t, int64(2), agg.totalCount)
|
assert.Equal(t, int64(2), agg.totalCount)
|
||||||
assert.Equal(t, int64(2), agg.successCount)
|
assert.Equal(t, int64(2), agg.successCount)
|
||||||
assert.Equal(t, 18.0, agg.avgResponse())
|
assert.Equal(t, int64(18), agg.avgResponse())
|
||||||
|
|
||||||
select {
|
select {
|
||||||
case <-existingTask.cancel:
|
case <-existingTask.cancel:
|
||||||
@@ -299,7 +303,8 @@ func TestProbeHTTP(t *testing.T) {
|
|||||||
}))
|
}))
|
||||||
defer server.Close()
|
defer server.Close()
|
||||||
|
|
||||||
responseUs := probeHTTP(server.Client(), server.URL)
|
responseUs, err := probeHTTP(server.Client(), server.URL)
|
||||||
|
require.NoError(t, err)
|
||||||
assert.GreaterOrEqual(t, responseUs, int64(0))
|
assert.GreaterOrEqual(t, responseUs, int64(0))
|
||||||
})
|
})
|
||||||
|
|
||||||
@@ -309,7 +314,9 @@ func TestProbeHTTP(t *testing.T) {
|
|||||||
}))
|
}))
|
||||||
defer server.Close()
|
defer server.Close()
|
||||||
|
|
||||||
assert.Equal(t, int64(-1), probeHTTP(server.Client(), server.URL))
|
responseUs, err := probeHTTP(server.Client(), server.URL)
|
||||||
|
assert.Equal(t, int64(-1), responseUs)
|
||||||
|
require.Error(t, err)
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -329,7 +336,8 @@ func TestProbeTCP(t *testing.T) {
|
|||||||
}()
|
}()
|
||||||
|
|
||||||
port := uint16(listener.Addr().(*net.TCPAddr).Port)
|
port := uint16(listener.Addr().(*net.TCPAddr).Port)
|
||||||
responseUs := probeTCP("127.0.0.1", port)
|
responseUs, err := probeTCP("127.0.0.1", port)
|
||||||
|
require.NoError(t, err)
|
||||||
assert.GreaterOrEqual(t, responseUs, int64(0))
|
assert.GreaterOrEqual(t, responseUs, int64(0))
|
||||||
<-accepted
|
<-accepted
|
||||||
})
|
})
|
||||||
@@ -341,6 +349,8 @@ func TestProbeTCP(t *testing.T) {
|
|||||||
port := uint16(listener.Addr().(*net.TCPAddr).Port)
|
port := uint16(listener.Addr().(*net.TCPAddr).Port)
|
||||||
require.NoError(t, listener.Close())
|
require.NoError(t, listener.Close())
|
||||||
|
|
||||||
assert.Equal(t, int64(-1), probeTCP("127.0.0.1", port))
|
responseUs, err := probeTCP("127.0.0.1", port)
|
||||||
|
assert.Equal(t, int64(-1), responseUs)
|
||||||
|
require.Error(t, err)
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -51,14 +51,15 @@ type SyncResponse struct {
|
|||||||
// 6: packet loss percentage (0-100)
|
// 6: packet loss percentage (0-100)
|
||||||
//
|
//
|
||||||
// 7: 1h packet loss percentage (0-100)
|
// 7: 1h packet loss percentage (0-100)
|
||||||
type Result []float64
|
type Result struct {
|
||||||
|
AvgResponse int64 `cbor:"0,keyasint,omitempty"`
|
||||||
// Get returns the value at the specified index or 0 if the index is out of range.
|
AvgResponse1h int64 `cbor:"1,keyasint,omitempty"`
|
||||||
func (r Result) Get(index int) float64 {
|
MinResponse int64 `cbor:"2,keyasint,omitempty"`
|
||||||
if index < len(r) {
|
MinResponse1h int64 `cbor:"3,keyasint,omitempty"`
|
||||||
return r[index]
|
MaxResponse int64 `cbor:"4,keyasint,omitempty"`
|
||||||
}
|
MaxResponse1h int64 `cbor:"5,keyasint,omitempty"`
|
||||||
return 0
|
PacketLoss float64 `cbor:"6,keyasint,omitempty"`
|
||||||
|
PacketLoss1h float64 `cbor:"7,keyasint,omitempty"`
|
||||||
}
|
}
|
||||||
|
|
||||||
// Stats holds only 1m values for a single target, which are used for charts.
|
// Stats holds only 1m values for a single target, which are used for charts.
|
||||||
@@ -74,9 +75,9 @@ type Stats []float64
|
|||||||
|
|
||||||
func (s Stats) FromResult(result Result) Stats {
|
func (s Stats) FromResult(result Result) Stats {
|
||||||
return Stats{
|
return Stats{
|
||||||
result.Get(0), // avg response
|
float64(result.AvgResponse),
|
||||||
result.Get(2), // min response
|
float64(result.MinResponse),
|
||||||
result.Get(4), // max response
|
float64(result.MaxResponse),
|
||||||
result.Get(6), // packet loss
|
result.PacketLoss,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -12,7 +12,12 @@ import (
|
|||||||
|
|
||||||
// generateProbeID creates a stable hash ID for a probe based on its configuration and the system it belongs to.
|
// generateProbeID creates a stable hash ID for a probe based on its configuration and the system it belongs to.
|
||||||
func generateProbeID(systemId string, config probe.Config) string {
|
func generateProbeID(systemId string, config probe.Config) string {
|
||||||
return systems.MakeStableHashId(systemId, config.Target, config.Protocol, strconv.FormatUint(uint64(config.Port), 10))
|
args := []string{systemId, config.Target, config.Protocol}
|
||||||
|
// only use port for TCP probes, since for other protocols it's not relevant as standalone value
|
||||||
|
if config.Protocol == "tcp" {
|
||||||
|
args = append(args, strconv.FormatUint(uint64(config.Port), 10))
|
||||||
|
}
|
||||||
|
return systems.MakeStableHashId(args...)
|
||||||
}
|
}
|
||||||
|
|
||||||
// bindNetworkProbesEvents keeps probe records and agent probe state in sync.
|
// bindNetworkProbesEvents keeps probe records and agent probe state in sync.
|
||||||
@@ -48,6 +53,10 @@ func bindNetworkProbesEvents(hub *Hub) {
|
|||||||
// record with the new ID and delete the old one. Otherwise, just update the existing probe on the agent.
|
// record with the new ID and delete the old one. Otherwise, just update the existing probe on the agent.
|
||||||
hub.OnRecordUpdateRequest("network_probes").BindFunc(func(e *core.RecordRequestEvent) error {
|
hub.OnRecordUpdateRequest("network_probes").BindFunc(func(e *core.RecordRequestEvent) error {
|
||||||
systemID := e.Record.GetString("system")
|
systemID := e.Record.GetString("system")
|
||||||
|
// only tcp uses port - set other protocols port to zero
|
||||||
|
if e.Record.GetString("protocol") != "tcp" {
|
||||||
|
e.Record.Set("port", 0)
|
||||||
|
}
|
||||||
ID := generateProbeID(systemID, *probeConfigFromRecord(e.Record))
|
ID := generateProbeID(systemID, *probeConfigFromRecord(e.Record))
|
||||||
if ID != e.Record.Id {
|
if ID != e.Record.Id {
|
||||||
newRecord := copyProbeToNewRecord(e.Record, ID)
|
newRecord := copyProbeToNewRecord(e.Record, ID)
|
||||||
@@ -96,14 +105,12 @@ func probeConfigFromRecord(record *core.Record) *probe.Config {
|
|||||||
|
|
||||||
// setProbeResultFields stores the latest probe result values on the record.
|
// setProbeResultFields stores the latest probe result values on the record.
|
||||||
func setProbeResultFields(record *core.Record, result probe.Result) {
|
func setProbeResultFields(record *core.Record, result probe.Result) {
|
||||||
now := time.Now().UTC()
|
nowString := time.Now().UTC().Format(types.DefaultDateLayout)
|
||||||
nowString := now.Format(types.DefaultDateLayout)
|
record.Set("res", result.AvgResponse)
|
||||||
record.Set("res", result.Get(0))
|
record.Set("resAvg1h", result.AvgResponse1h)
|
||||||
record.Set("resAvg1h", result.Get(1))
|
record.Set("resMin1h", result.MinResponse1h)
|
||||||
record.Set("resMin1h", result.Get(3))
|
record.Set("resMax1h", result.MaxResponse1h)
|
||||||
record.Set("resMax1h", result.Get(5))
|
record.Set("loss1h", result.PacketLoss1h)
|
||||||
// record.Set("loss", result.Get(4))
|
|
||||||
record.Set("loss1h", result.Get(7))
|
|
||||||
record.Set("updated", nowString)
|
record.Set("updated", nowString)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -113,8 +120,11 @@ func setProbeResultFields(record *core.Record, result probe.Result) {
|
|||||||
func copyProbeToNewRecord(oldRecord *core.Record, newID string) *core.Record {
|
func copyProbeToNewRecord(oldRecord *core.Record, newID string) *core.Record {
|
||||||
collection := oldRecord.Collection()
|
collection := oldRecord.Collection()
|
||||||
newRecord := core.NewRecord(collection)
|
newRecord := core.NewRecord(collection)
|
||||||
newRecord.Load(oldRecord.FieldsData())
|
newRecord.Id = newID
|
||||||
newRecord.Set("id", newID)
|
fields := []string{"system", "name", "target", "protocol", "port", "interval", "enabled"}
|
||||||
|
for _, field := range fields {
|
||||||
|
newRecord.Set(field, oldRecord.Get(field))
|
||||||
|
}
|
||||||
return newRecord
|
return newRecord
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -4,7 +4,9 @@ import (
|
|||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
"github.com/henrygd/beszel/internal/entities/probe"
|
"github.com/henrygd/beszel/internal/entities/probe"
|
||||||
|
"github.com/pocketbase/pocketbase/core"
|
||||||
"github.com/stretchr/testify/assert"
|
"github.com/stretchr/testify/assert"
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
)
|
)
|
||||||
|
|
||||||
func TestGenerateProbeID(t *testing.T) {
|
func TestGenerateProbeID(t *testing.T) {
|
||||||
@@ -20,10 +22,21 @@ func TestGenerateProbeID(t *testing.T) {
|
|||||||
config: probe.Config{
|
config: probe.Config{
|
||||||
Protocol: "http",
|
Protocol: "http",
|
||||||
Target: "example.com",
|
Target: "example.com",
|
||||||
Port: 80,
|
Port: 0,
|
||||||
Interval: 60,
|
Interval: 60,
|
||||||
},
|
},
|
||||||
expected: "de7b3647",
|
expected: "a20a5827",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "HTTP probe on example.com with different port",
|
||||||
|
systemID: "sys123",
|
||||||
|
config: probe.Config{
|
||||||
|
Protocol: "http",
|
||||||
|
Target: "example.com",
|
||||||
|
Port: 8080,
|
||||||
|
Interval: 60,
|
||||||
|
},
|
||||||
|
expected: "a20a5827",
|
||||||
},
|
},
|
||||||
{
|
{
|
||||||
name: "HTTP probe on example.com with different system ID",
|
name: "HTTP probe on example.com with different system ID",
|
||||||
@@ -34,7 +47,7 @@ func TestGenerateProbeID(t *testing.T) {
|
|||||||
Port: 80,
|
Port: 80,
|
||||||
Interval: 60,
|
Interval: 60,
|
||||||
},
|
},
|
||||||
expected: "be9e2707",
|
expected: "ab602ae7",
|
||||||
},
|
},
|
||||||
{
|
{
|
||||||
name: "Same probe, different interval",
|
name: "Same probe, different interval",
|
||||||
@@ -45,7 +58,7 @@ func TestGenerateProbeID(t *testing.T) {
|
|||||||
Port: 80,
|
Port: 80,
|
||||||
Interval: 120,
|
Interval: 120,
|
||||||
},
|
},
|
||||||
expected: "be9e2707",
|
expected: "ab602ae7",
|
||||||
},
|
},
|
||||||
{
|
{
|
||||||
name: "ICMP probe on 1.1.1.1",
|
name: "ICMP probe on 1.1.1.1",
|
||||||
@@ -56,7 +69,7 @@ func TestGenerateProbeID(t *testing.T) {
|
|||||||
Port: 0,
|
Port: 0,
|
||||||
Interval: 10,
|
Interval: 10,
|
||||||
},
|
},
|
||||||
expected: "49ec14fc",
|
expected: "6d13a4a4",
|
||||||
}, {
|
}, {
|
||||||
name: "ICMP probe on 1.1.1.1 with different system ID",
|
name: "ICMP probe on 1.1.1.1 with different system ID",
|
||||||
systemID: "sys4567",
|
systemID: "sys4567",
|
||||||
@@ -66,7 +79,7 @@ func TestGenerateProbeID(t *testing.T) {
|
|||||||
Port: 0,
|
Port: 0,
|
||||||
Interval: 10,
|
Interval: 10,
|
||||||
},
|
},
|
||||||
expected: "84921aa3",
|
expected: "ddd6c81",
|
||||||
},
|
},
|
||||||
{
|
{
|
||||||
name: "TCP probe on example.com with port 443",
|
name: "TCP probe on example.com with port 443",
|
||||||
@@ -99,3 +112,44 @@ func TestGenerateProbeID(t *testing.T) {
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestCopyProbeToNewRecordDropsResultFields(t *testing.T) {
|
||||||
|
hub, testApp, err := createTestHub(t)
|
||||||
|
require.NoError(t, err)
|
||||||
|
defer cleanupTestHub(hub, testApp)
|
||||||
|
|
||||||
|
collection, err := hub.FindCachedCollectionByNameOrId("network_probes")
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
oldRecord := core.NewRecord(collection)
|
||||||
|
oldRecord.Load(map[string]any{
|
||||||
|
"system": "sys123",
|
||||||
|
"name": "Example",
|
||||||
|
"target": "https://example.com",
|
||||||
|
"protocol": "http",
|
||||||
|
"port": 443,
|
||||||
|
"interval": 60,
|
||||||
|
"enabled": true,
|
||||||
|
"res": 1200,
|
||||||
|
"resAvg1h": 1300,
|
||||||
|
"resMin1h": 900,
|
||||||
|
"resMax1h": 1600,
|
||||||
|
"loss1h": 5,
|
||||||
|
"updated": "2026-04-29 12:00:00.000Z",
|
||||||
|
})
|
||||||
|
|
||||||
|
newRecord := copyProbeToNewRecord(oldRecord, "next12345")
|
||||||
|
|
||||||
|
assert.Equal(t, "next12345", newRecord.Id)
|
||||||
|
assert.Equal(t, "Example", newRecord.GetString("name"))
|
||||||
|
assert.Equal(t, "https://example.com", newRecord.GetString("target"))
|
||||||
|
assert.Equal(t, "http", newRecord.GetString("protocol"))
|
||||||
|
assert.Equal(t, 443, newRecord.GetInt("port"))
|
||||||
|
assert.True(t, newRecord.GetBool("enabled"))
|
||||||
|
assert.Zero(t, newRecord.GetFloat("res"))
|
||||||
|
assert.Zero(t, newRecord.GetFloat("resAvg1h"))
|
||||||
|
assert.Zero(t, newRecord.GetFloat("resMin1h"))
|
||||||
|
assert.Zero(t, newRecord.GetFloat("resMax1h"))
|
||||||
|
assert.Zero(t, newRecord.GetFloat("loss1h"))
|
||||||
|
assert.Equal(t, "", newRecord.GetString("updated"))
|
||||||
|
}
|
||||||
|
|||||||
@@ -320,7 +320,7 @@ func updateNetworkProbesRecords(app core.App, probeResults map[string]probe.Resu
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
var err error
|
var err error
|
||||||
probeCollectionName := "network_probes"
|
const probeCollectionName = "network_probes"
|
||||||
|
|
||||||
// If realtime updates are active, we save via PocketBase records to trigger realtime events.
|
// If realtime updates are active, we save via PocketBase records to trigger realtime events.
|
||||||
// Otherwise we can do a more efficient direct update via SQL
|
// Otherwise we can do a more efficient direct update via SQL
|
||||||
@@ -345,14 +345,14 @@ func updateNetworkProbesRecords(app core.App, probeResults map[string]probe.Resu
|
|||||||
}
|
}
|
||||||
|
|
||||||
// update network_probes records
|
// update network_probes records
|
||||||
for id, values := range probeResults {
|
for id, result := range probeResults {
|
||||||
probeData := map[string]any{
|
probeData := map[string]any{
|
||||||
"id": id,
|
"id": id,
|
||||||
"res": values.Get(0),
|
"res": result.AvgResponse,
|
||||||
"resAvg1h": values.Get(1),
|
"resAvg1h": result.AvgResponse1h,
|
||||||
"resMin1h": values.Get(3),
|
"resMin1h": result.MinResponse1h,
|
||||||
"resMax1h": values.Get(5),
|
"resMax1h": result.MaxResponse1h,
|
||||||
"loss1h": values.Get(7),
|
"loss1h": result.PacketLoss1h,
|
||||||
"updated": nowString,
|
"updated": nowString,
|
||||||
}
|
}
|
||||||
switch realtimeActive {
|
switch realtimeActive {
|
||||||
@@ -372,12 +372,12 @@ func updateNetworkProbesRecords(app core.App, probeResults map[string]probe.Resu
|
|||||||
}
|
}
|
||||||
|
|
||||||
// handle stats collection as well
|
// handle stats collection as well
|
||||||
statsCollectionName := "network_probe_stats"
|
const statsCollectionName = "network_probe_stats"
|
||||||
|
|
||||||
// we don't need the hour values for the stats collection
|
// we don't need the hour values for the stats collection
|
||||||
stats := make(map[string]probe.Stats, len(probeResults))
|
stats := make(map[string]probe.Stats, len(probeResults))
|
||||||
for key, values := range probeResults {
|
for key, result := range probeResults {
|
||||||
stats[key] = probe.Stats{}.FromResult(values)
|
stats[key] = probe.Stats{}.FromResult(result)
|
||||||
}
|
}
|
||||||
|
|
||||||
statsRecordData := map[string]any{
|
statsRecordData := map[string]any{
|
||||||
|
|||||||
@@ -24,7 +24,7 @@ func (sys *System) UpsertNetworkProbe(config probe.Config, runNow bool) (*probe.
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
if len(resp.Result) == 0 {
|
if resp.Result == (probe.Result{}) {
|
||||||
return nil, nil
|
return nil, nil
|
||||||
}
|
}
|
||||||
result := resp.Result
|
result := resp.Result
|
||||||
|
|||||||
@@ -32,13 +32,13 @@ func TestAverageProbeStats(t *testing.T) {
|
|||||||
recordA, err := tests.CreateRecord(hub, "network_probe_stats", map[string]any{
|
recordA, err := tests.CreateRecord(hub, "network_probe_stats", map[string]any{
|
||||||
"system": system.Id,
|
"system": system.Id,
|
||||||
"type": "1m",
|
"type": "1m",
|
||||||
"stats": `{"icmp:1.1.1.1":[10,80,8,14,1]}`,
|
"stats": `{"icmp:1.1.1.1":[10,5,20,1.5]}`,
|
||||||
})
|
})
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
recordB, err := tests.CreateRecord(hub, "network_probe_stats", map[string]any{
|
recordB, err := tests.CreateRecord(hub, "network_probe_stats", map[string]any{
|
||||||
"system": system.Id,
|
"system": system.Id,
|
||||||
"type": "1m",
|
"type": "1m",
|
||||||
"stats": `{"icmp:1.1.1.1":[40,100,9,50,5]}`,
|
"stats": `{"icmp:1.1.1.1":[22.5,10,60,0]}`,
|
||||||
})
|
})
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
|
|
||||||
@@ -49,10 +49,9 @@ func TestAverageProbeStats(t *testing.T) {
|
|||||||
|
|
||||||
stats, ok := result["icmp:1.1.1.1"]
|
stats, ok := result["icmp:1.1.1.1"]
|
||||||
require.True(t, ok)
|
require.True(t, ok)
|
||||||
require.Len(t, stats, 5)
|
require.Len(t, stats, 4)
|
||||||
assert.Equal(t, 25.0, stats[0])
|
assert.InDelta(t, 16.25, stats[0], 0.001) // avg of avg
|
||||||
assert.Equal(t, 90.0, stats[1])
|
assert.InDelta(t, 5, stats[1], 0.001) // min of mins
|
||||||
assert.Equal(t, 8.0, stats[2])
|
assert.InDelta(t, 60, stats[2], 0.001) // max of maxes
|
||||||
assert.Equal(t, 50.0, stats[3])
|
assert.InDelta(t, 0.75, stats[3], 0.001) // avg of packet loss
|
||||||
assert.Equal(t, 3.0, stats[4])
|
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -532,9 +532,9 @@ func AverageContainerStatsSlice(records [][]container.Stats) []container.Stats {
|
|||||||
|
|
||||||
// AverageProbeStats averages probe stats across multiple records.
|
// AverageProbeStats averages probe stats across multiple records.
|
||||||
// For each probe key: avg of average fields, min of mins, and max of maxes.
|
// For each probe key: avg of average fields, min of mins, and max of maxes.
|
||||||
func (rm *RecordManager) AverageProbeStats(db dbx.Builder, records RecordIds) map[string]probe.Result {
|
func (rm *RecordManager) AverageProbeStats(db dbx.Builder, records RecordIds) map[string]probe.Stats {
|
||||||
type probeValues struct {
|
type probeValues struct {
|
||||||
sums probe.Result
|
sums probe.Stats
|
||||||
counts []int
|
counts []int
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -546,18 +546,18 @@ func (rm *RecordManager) AverageProbeStats(db dbx.Builder, records RecordIds) ma
|
|||||||
for _, rec := range records {
|
for _, rec := range records {
|
||||||
row.Stats = row.Stats[:0]
|
row.Stats = row.Stats[:0]
|
||||||
query.Bind(dbx.Params{"id": rec.Id}).One(&row)
|
query.Bind(dbx.Params{"id": rec.Id}).One(&row)
|
||||||
var rawStats map[string]probe.Result
|
var rawStats map[string]probe.Stats
|
||||||
if err := json.Unmarshal(row.Stats, &rawStats); err != nil {
|
if err := json.Unmarshal(row.Stats, &rawStats); err != nil {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
for key, vals := range rawStats {
|
for key, vals := range rawStats {
|
||||||
s, ok := sums[key]
|
s, ok := sums[key]
|
||||||
if !ok {
|
if !ok {
|
||||||
s = &probeValues{sums: make(probe.Result, len(vals)), counts: make([]int, len(vals))}
|
s = &probeValues{sums: make(probe.Stats, len(vals)), counts: make([]int, len(vals))}
|
||||||
sums[key] = s
|
sums[key] = s
|
||||||
}
|
}
|
||||||
if len(vals) > len(s.sums) {
|
if len(vals) > len(s.sums) {
|
||||||
expandedSums := make(probe.Result, len(vals))
|
expandedSums := make(probe.Stats, len(vals))
|
||||||
copy(expandedSums, s.sums)
|
copy(expandedSums, s.sums)
|
||||||
s.sums = expandedSums
|
s.sums = expandedSums
|
||||||
|
|
||||||
@@ -584,7 +584,7 @@ func (rm *RecordManager) AverageProbeStats(db dbx.Builder, records RecordIds) ma
|
|||||||
}
|
}
|
||||||
|
|
||||||
// compute final averages
|
// compute final averages
|
||||||
result := make(map[string]probe.Result, len(sums))
|
result := make(map[string]probe.Stats, len(sums))
|
||||||
for key, s := range sums {
|
for key, s := range sums {
|
||||||
if len(s.counts) == 0 {
|
if len(s.counts) == 0 {
|
||||||
continue
|
continue
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
import type { CellContext, Column, ColumnDef } from "@tanstack/react-table"
|
import type { CellContext, Column, ColumnDef } from "@tanstack/react-table"
|
||||||
import { Button } from "@/components/ui/button"
|
import { Button } from "@/components/ui/button"
|
||||||
import { cn, copyToClipboard, formatMicroseconds, hourWithSeconds } from "@/lib/utils"
|
import { cn, copyToClipboard, decimalString, formatMicroseconds, hourWithSeconds } from "@/lib/utils"
|
||||||
import {
|
import {
|
||||||
GlobeIcon,
|
GlobeIcon,
|
||||||
TimerIcon,
|
TimerIcon,
|
||||||
@@ -33,11 +33,12 @@ import { SystemStatus } from "@/lib/enums"
|
|||||||
import { Checkbox } from "@/components/ui/checkbox"
|
import { Checkbox } from "@/components/ui/checkbox"
|
||||||
import { useMemo } from "react"
|
import { useMemo } from "react"
|
||||||
import { formatBulkProbeLine } from "@/components/network-probes-table/probe-dialog"
|
import { formatBulkProbeLine } from "@/components/network-probes-table/probe-dialog"
|
||||||
|
import { Badge } from "../ui/badge"
|
||||||
|
|
||||||
const protocolColors: Record<string, string> = {
|
const protocolColors: Record<string, string> = {
|
||||||
icmp: "bg-blue-500/15 text-blue-400",
|
icmp: "bg-blue-500/15! text-blue-600 dark:text-blue-400",
|
||||||
tcp: "bg-purple-500/15 text-purple-400",
|
tcp: "bg-purple-500/15! text-purple-600 dark:text-purple-400",
|
||||||
http: "bg-green-500/15 text-green-400",
|
http: "bg-green-500/15! text-green-700 dark:text-green-400",
|
||||||
}
|
}
|
||||||
|
|
||||||
const SYSTEM_STATUS_COLORS = {
|
const SYSTEM_STATUS_COLORS = {
|
||||||
@@ -97,9 +98,17 @@ export function getProbeColumns(
|
|||||||
header: ({ column }) => <HeaderButton column={column} name={t`Name`} Icon={NetworkIcon} />,
|
header: ({ column }) => <HeaderButton column={column} name={t`Name`} Icon={NetworkIcon} />,
|
||||||
cell: ({ row, getValue }) => {
|
cell: ({ row, getValue }) => {
|
||||||
const probe = row.original
|
const probe = row.original
|
||||||
|
const { status } = useStore($allSystemsById)[probe.system] || {}
|
||||||
|
|
||||||
|
let color = "bg-green-500"
|
||||||
|
if (!probe.enabled || status === SystemStatus.Paused) {
|
||||||
|
color = "bg-primary/40"
|
||||||
|
} else if (status === SystemStatus.Down || status === SystemStatus.Pending) {
|
||||||
|
color = "bg-yellow-500"
|
||||||
|
}
|
||||||
return (
|
return (
|
||||||
<div className="ms-1.5 max-w-40 flex gap-2 items-center tabular-nums">
|
<div className="ms-1.5 max-w-40 flex gap-2 items-center tabular-nums">
|
||||||
<span className={cn("shrink-0 size-2 rounded-full", probe.enabled ? "bg-green-500" : "bg-primary/40")} />
|
<span className={cn("shrink-0 size-2 rounded-full", color)} />
|
||||||
<div className="relative w-fit min-w-0 max-w-full">
|
<div className="relative w-fit min-w-0 max-w-full">
|
||||||
<span className="invisible block overflow-hidden whitespace-nowrap" aria-hidden="true">
|
<span className="invisible block overflow-hidden whitespace-nowrap" aria-hidden="true">
|
||||||
{longestName}
|
{longestName}
|
||||||
@@ -117,7 +126,11 @@ export function getProbeColumns(
|
|||||||
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 || a.original.target).localeCompare(b.original.name || b.original.target)
|
||||||
},
|
},
|
||||||
header: ({ column }) => <HeaderButton column={column} name={t`System`} Icon={ServerIcon} />,
|
header: ({ column }) => <HeaderButton column={column} name={t`System`} Icon={ServerIcon} />,
|
||||||
cell: ({ getValue }) => {
|
cell: ({ getValue }) => {
|
||||||
@@ -162,16 +175,13 @@ export function getProbeColumns(
|
|||||||
header: ({ column }) => <HeaderButton column={column} name={t`Protocol`} Icon={ArrowLeftRightIcon} />,
|
header: ({ column }) => <HeaderButton column={column} name={t`Protocol`} Icon={ArrowLeftRightIcon} />,
|
||||||
cell: ({ getValue }) => {
|
cell: ({ getValue }) => {
|
||||||
const protocol = getValue() as string
|
const protocol = getValue() as string
|
||||||
return (
|
return <Badge className={cn("uppercase", protocolColors[protocol])}>{protocol}</Badge>
|
||||||
<span className={cn("ms-1.5 px-2 py-0.5 rounded text-xs font-medium uppercase", protocolColors[protocol])}>
|
|
||||||
{protocol}
|
|
||||||
</span>
|
|
||||||
)
|
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
{
|
{
|
||||||
id: "interval",
|
id: "interval",
|
||||||
accessorFn: (record) => record.interval,
|
accessorFn: (record) => record.interval,
|
||||||
|
invertSorting: true,
|
||||||
header: ({ column }) => <HeaderButton column={column} name={t`Interval`} Icon={RefreshCwIcon} />,
|
header: ({ column }) => <HeaderButton column={column} name={t`Interval`} Icon={RefreshCwIcon} />,
|
||||||
cell: ({ getValue }) => <span className="ms-1.5 tabular-nums">{getValue() as number}s</span>,
|
cell: ({ getValue }) => <span className="ms-1.5 tabular-nums">{getValue() as number}s</span>,
|
||||||
},
|
},
|
||||||
@@ -226,7 +236,7 @@ export function getProbeColumns(
|
|||||||
return (
|
return (
|
||||||
<span className="ms-1.5 tabular-nums flex gap-2 items-center">
|
<span className="ms-1.5 tabular-nums flex gap-2 items-center">
|
||||||
<span className={cn("shrink-0 size-2 rounded-full", color)} />
|
<span className={cn("shrink-0 size-2 rounded-full", color)} />
|
||||||
{loss1h}%
|
{loss1h === 100 ? loss1h : decimalString(loss1h, loss1h >= 10 ? 1 : 2)}%
|
||||||
</span>
|
</span>
|
||||||
)
|
)
|
||||||
},
|
},
|
||||||
|
|||||||
@@ -37,7 +37,7 @@ import { $allSystemsById, $chartTime, $direction } from "@/lib/stores"
|
|||||||
import { cn, isVisuallyLonger, useBrowserStorage } from "@/lib/utils"
|
import { cn, isVisuallyLonger, useBrowserStorage } from "@/lib/utils"
|
||||||
import type { NetworkProbeRecord } from "@/types"
|
import type { NetworkProbeRecord } from "@/types"
|
||||||
import { AddProbeDialog, EditProbeDialog } from "./probe-dialog"
|
import { AddProbeDialog, EditProbeDialog } from "./probe-dialog"
|
||||||
import { XIcon } from "lucide-react"
|
import { ArrowLeftRightIcon, EthernetPortIcon, GlobeIcon, ServerIcon, XIcon } from "lucide-react"
|
||||||
import { Sheet, SheetContent, SheetDescription, SheetHeader, SheetTitle } from "@/components/ui/sheet"
|
import { Sheet, SheetContent, SheetDescription, SheetHeader, SheetTitle } from "@/components/ui/sheet"
|
||||||
import ChartTimeSelect from "@/components/charts/chart-time-select"
|
import ChartTimeSelect from "@/components/charts/chart-time-select"
|
||||||
import { LossChart, AvgMinMaxResponseChart } from "@/components/routes/system/charts/probes-charts"
|
import { LossChart, AvgMinMaxResponseChart } from "@/components/routes/system/charts/probes-charts"
|
||||||
@@ -501,16 +501,20 @@ function NetworkProbeSheetContent({
|
|||||||
<SheetHeader className="mb-0 border-b p-0 pb-4">
|
<SheetHeader className="mb-0 border-b p-0 pb-4">
|
||||||
<SheetTitle>{probeLabel}</SheetTitle>
|
<SheetTitle>{probeLabel}</SheetTitle>
|
||||||
<SheetDescription className="flex flex-wrap items-center gap-x-2 gap-y-1">
|
<SheetDescription className="flex flex-wrap items-center gap-x-2 gap-y-1">
|
||||||
|
<ServerIcon className="size-3.5 text-muted-foreground" />
|
||||||
<Link className="hover:underline" href={getPagePath($router, "system", { id: system?.id ?? "" })}>
|
<Link className="hover:underline" href={getPagePath($router, "system", { id: system?.id ?? "" })}>
|
||||||
{system?.name ?? ""}
|
{system?.name ?? ""}
|
||||||
</Link>
|
</Link>
|
||||||
<Separator orientation="vertical" className="h-2.5 bg-muted-foreground opacity-70" />
|
<Separator orientation="vertical" className="h-2.5 bg-muted-foreground opacity-70" />
|
||||||
|
<ArrowLeftRightIcon className="size-3.5 text-muted-foreground" />
|
||||||
{probe.protocol.toUpperCase()}
|
{probe.protocol.toUpperCase()}
|
||||||
<Separator orientation="vertical" className="h-2.5 bg-muted-foreground opacity-70" />
|
<Separator orientation="vertical" className="h-2.5 bg-muted-foreground opacity-70" />
|
||||||
|
<GlobeIcon className="size-3.5 text-muted-foreground" />
|
||||||
{probe.target}
|
{probe.target}
|
||||||
{probe.port > 0 && (
|
{probe.protocol === "tcp" && probe.port > 0 && (
|
||||||
<>
|
<>
|
||||||
<Separator orientation="vertical" className="h-2.5 bg-muted-foreground opacity-70" />
|
<Separator orientation="vertical" className="h-2.5 bg-muted-foreground opacity-70" />
|
||||||
|
<EthernetPortIcon className="size-3.5 text-muted-foreground" />
|
||||||
<span>{probe.port}</span>
|
<span>{probe.port}</span>
|
||||||
</>
|
</>
|
||||||
)}
|
)}
|
||||||
|
|||||||
@@ -40,7 +40,7 @@ type NormalizedProbeValues = Omit<ProbeValues, "system" | "interval"> & {
|
|||||||
|
|
||||||
type BulkProbeLineSource = Pick<NetworkProbeRecord, "target" | "protocol" | "port" | "interval" | "name">
|
type BulkProbeLineSource = Pick<NetworkProbeRecord, "target" | "protocol" | "port" | "interval" | "name">
|
||||||
|
|
||||||
const defaultInterval = 20
|
const defaultInterval = 30
|
||||||
|
|
||||||
const ProbeProtocolSchema = v.picklist(["icmp", "tcp", "http"])
|
const ProbeProtocolSchema = v.picklist(["icmp", "tcp", "http"])
|
||||||
|
|
||||||
@@ -58,15 +58,19 @@ const NormalizedProbeValuesSchema = v.pipe(
|
|||||||
}),
|
}),
|
||||||
v.transform((input): NormalizedProbeValues => {
|
v.transform((input): NormalizedProbeValues => {
|
||||||
let { protocol, port } = input
|
let { protocol, port } = input
|
||||||
if (protocol === "icmp") {
|
let httpTarget = input.target
|
||||||
|
if (protocol === "icmp" || protocol === "http") {
|
||||||
|
if (protocol === "http") {
|
||||||
|
httpTarget = normalizeHttpTarget(input.target, port)
|
||||||
|
}
|
||||||
port = 0
|
port = 0
|
||||||
} else if ((protocol === "tcp" || protocol === "http") && !port) {
|
} else if (protocol === "tcp" && !port) {
|
||||||
port = 443
|
port = 443
|
||||||
}
|
}
|
||||||
return {
|
return {
|
||||||
// HTTP probes may be entered as bare hostnames, so normalize them to a
|
// HTTP probes may be entered as bare hostnames, so normalize them to a
|
||||||
// scheme-bearing URL before the payload is sent to PocketBase.
|
// scheme-bearing URL before the payload is sent to PocketBase.
|
||||||
target: protocol === "http" ? normalizeHttpTarget(input.target, port) : input.target,
|
target: protocol === "http" ? httpTarget : input.target,
|
||||||
protocol,
|
protocol,
|
||||||
port,
|
port,
|
||||||
interval: input.interval,
|
interval: input.interval,
|
||||||
@@ -75,7 +79,7 @@ const NormalizedProbeValuesSchema = v.pipe(
|
|||||||
}),
|
}),
|
||||||
v.forward(
|
v.forward(
|
||||||
v.check((input) => {
|
v.check((input) => {
|
||||||
if (input.protocol === "icmp") {
|
if (input.protocol === "icmp" || input.protocol === "http") {
|
||||||
return input.port === 0
|
return input.port === 0
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -95,12 +99,31 @@ const BulkProbeSchema = v.object({
|
|||||||
name: v.optional(v.pipe(v.string(), v.trim())),
|
name: v.optional(v.pipe(v.string(), v.trim())),
|
||||||
})
|
})
|
||||||
|
|
||||||
function normalizeHttpTarget(target: string, port: number) {
|
function normalizeHttpTarget(target: string, port = 0) {
|
||||||
if (/^https?:\/\//i.test(target)) {
|
const useExplicitPort = port > 0 && port !== 80 && port !== 443
|
||||||
|
const hasOriginOnlyTarget = /^https?:\/\/[^/?#]+$/i.test(target)
|
||||||
|
if (!/^https?:\/\//i.test(target)) {
|
||||||
|
const scheme = port === 80 ? "http" : "https"
|
||||||
|
return `${scheme}://${target}${useExplicitPort ? `:${port}` : ""}`
|
||||||
|
}
|
||||||
|
|
||||||
|
let parsedUrl: URL
|
||||||
|
try {
|
||||||
|
parsedUrl = new URL(target)
|
||||||
|
} catch {
|
||||||
return target
|
return target
|
||||||
}
|
}
|
||||||
|
|
||||||
return `${port === 443 ? "https" : "http"}://${target}`
|
if (!parsedUrl.port && useExplicitPort) {
|
||||||
|
parsedUrl.port = `${port}`
|
||||||
|
}
|
||||||
|
|
||||||
|
// avoid converting "http://localhost:8090" to "http://localhost:8090/" - keep the original formatting if the URL is just an origin
|
||||||
|
if (hasOriginOnlyTarget && parsedUrl.pathname === "/" && !parsedUrl.search && !parsedUrl.hash) {
|
||||||
|
return parsedUrl.origin
|
||||||
|
}
|
||||||
|
|
||||||
|
return parsedUrl.toString()
|
||||||
}
|
}
|
||||||
|
|
||||||
function trimTrailingEmptyFields(fields: string[]) {
|
function trimTrailingEmptyFields(fields: string[]) {
|
||||||
@@ -152,12 +175,13 @@ function parseBulkProbeLine(line: string, lineNumber: number, system: string) {
|
|||||||
if (!parsed.success) {
|
if (!parsed.success) {
|
||||||
throw new Error(`Line ${lineNumber}: ${parsed.issues[0]?.message || "invalid probe entry"}`)
|
throw new Error(`Line ${lineNumber}: ${parsed.issues[0]?.message || "invalid probe entry"}`)
|
||||||
}
|
}
|
||||||
|
const protocol = (parsed.output.protocol?.toLowerCase() ||
|
||||||
|
(/^https?:\/\//i.test(parsed.output.target) ? "http" : "icmp")) as ProbeProtocol
|
||||||
|
|
||||||
return buildProbePayload({
|
return buildProbePayload({
|
||||||
system,
|
system,
|
||||||
target: parsed.output.target,
|
target: parsed.output.target,
|
||||||
protocol: (parsed.output.protocol?.toLowerCase() ||
|
protocol,
|
||||||
(/^https?:\/\//i.test(parsed.output.target) ? "http" : "icmp")) as ProbeProtocol,
|
|
||||||
port: parsed.output.port ? Number(parsed.output.port) : 0,
|
port: parsed.output.port ? Number(parsed.output.port) : 0,
|
||||||
interval: parsed.output.interval || `${defaultInterval}`,
|
interval: parsed.output.interval || `${defaultInterval}`,
|
||||||
name: parsed.output.name || undefined,
|
name: parsed.output.name || undefined,
|
||||||
@@ -165,7 +189,7 @@ function parseBulkProbeLine(line: string, lineNumber: number, system: string) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
export function formatBulkProbeLine(probe: BulkProbeLineSource) {
|
export function formatBulkProbeLine(probe: BulkProbeLineSource) {
|
||||||
const port = probe.protocol === "icmp" || probe.port === 443 ? "" : `${probe.port}`
|
const port = probe.protocol !== "tcp" || probe.port === 443 ? "" : `${probe.port}`
|
||||||
const interval = probe.interval === defaultInterval ? "" : `${probe.interval}`
|
const interval = probe.interval === defaultInterval ? "" : `${probe.interval}`
|
||||||
return trimTrailingEmptyFields([probe.target, probe.protocol, port, interval, probe.name?.trim() || ""]).join(",")
|
return trimTrailingEmptyFields([probe.target, probe.protocol, port, interval, probe.name?.trim() || ""]).join(",")
|
||||||
}
|
}
|
||||||
@@ -402,9 +426,7 @@ function ProbeDialogContent({
|
|||||||
}) {
|
}) {
|
||||||
const [protocol, setProtocol] = useState<ProbeProtocol>(probe?.protocol ?? "icmp")
|
const [protocol, setProtocol] = useState<ProbeProtocol>(probe?.protocol ?? "icmp")
|
||||||
const [target, setTarget] = useState(probe?.target ?? "")
|
const [target, setTarget] = useState(probe?.target ?? "")
|
||||||
const [port, setPort] = useState(
|
const [port, setPort] = useState(probe?.protocol === "tcp" && probe.port ? String(probe.port) : "")
|
||||||
(probe?.protocol === "tcp" || probe?.protocol === "http") && probe.port ? String(probe.port) : ""
|
|
||||||
)
|
|
||||||
const [probeInterval, setProbeInterval] = useState(String(probe?.interval ?? defaultInterval))
|
const [probeInterval, setProbeInterval] = useState(String(probe?.interval ?? defaultInterval))
|
||||||
const [name, setName] = useState(probe?.name ?? "")
|
const [name, setName] = useState(probe?.name ?? "")
|
||||||
const [loading, setLoading] = useState(false)
|
const [loading, setLoading] = useState(false)
|
||||||
@@ -423,7 +445,7 @@ function ProbeDialogContent({
|
|||||||
|
|
||||||
setProtocol(probe?.protocol ?? "icmp")
|
setProtocol(probe?.protocol ?? "icmp")
|
||||||
setTarget(probe?.target ?? "")
|
setTarget(probe?.target ?? "")
|
||||||
setPort((probe?.protocol === "tcp" || probe?.protocol === "http") && probe.port ? String(probe.port) : "")
|
setPort(probe?.protocol === "tcp" && probe.port ? String(probe.port) : "")
|
||||||
setProbeInterval(String(probe?.interval ?? defaultInterval))
|
setProbeInterval(String(probe?.interval ?? defaultInterval))
|
||||||
setName(probe?.name ?? "")
|
setName(probe?.name ?? "")
|
||||||
setSelectedSystemId(probe?.system ?? "")
|
setSelectedSystemId(probe?.system ?? "")
|
||||||
@@ -444,7 +466,7 @@ function ProbeDialogContent({
|
|||||||
system: selectedSystem,
|
system: selectedSystem,
|
||||||
target,
|
target,
|
||||||
protocol,
|
protocol,
|
||||||
port: protocol === "tcp" || protocol === "http" ? Number(port) : 0,
|
port: protocol === "tcp" ? Number(port) : 0,
|
||||||
interval: probeInterval,
|
interval: probeInterval,
|
||||||
name,
|
name,
|
||||||
},
|
},
|
||||||
@@ -500,7 +522,7 @@ function ProbeDialogContent({
|
|||||||
<Input
|
<Input
|
||||||
value={target}
|
value={target}
|
||||||
onChange={(e) => setTarget(e.target.value)}
|
onChange={(e) => setTarget(e.target.value)}
|
||||||
placeholder={protocol === "http" ? "https://example.com" : "1.1.1.1"}
|
placeholder={protocol === "http" ? "http://localhost:8090" : "1.1.1.1"}
|
||||||
required
|
required
|
||||||
/>
|
/>
|
||||||
</div>
|
</div>
|
||||||
@@ -520,7 +542,7 @@ function ProbeDialogContent({
|
|||||||
</SelectContent>
|
</SelectContent>
|
||||||
</Select>
|
</Select>
|
||||||
</div>
|
</div>
|
||||||
{(protocol === "tcp" || protocol === "http") && (
|
{protocol === "tcp" && (
|
||||||
<div className="grid gap-2">
|
<div className="grid gap-2">
|
||||||
<Label>
|
<Label>
|
||||||
<Trans>Port</Trans>
|
<Trans>Port</Trans>
|
||||||
|
|||||||
@@ -81,7 +81,7 @@ function ProbeChart({
|
|||||||
return probeStats.filter((record) => visibleKeys.some((id) => record.stats?.[id] != null))
|
return probeStats.filter((record) => visibleKeys.some((id) => record.stats?.[id] != null))
|
||||||
}, [probeStats, visibleKeys])
|
}, [probeStats, visibleKeys])
|
||||||
|
|
||||||
const legend = dataPoints.length < 10 && dataPoints.length > 1
|
const legend = dataPoints.length < 10 && showFilter
|
||||||
|
|
||||||
return (
|
return (
|
||||||
<ChartCard
|
<ChartCard
|
||||||
|
|||||||
@@ -116,6 +116,18 @@ export function useNetworkProbeStats(props: UseNetworkProbeStatsProps) {
|
|||||||
const [probeStats, setProbeStats] = useState<NetworkProbeStatsRecord[]>([])
|
const [probeStats, setProbeStats] = useState<NetworkProbeStatsRecord[]>([])
|
||||||
const requestID = useRef(0)
|
const requestID = useRef(0)
|
||||||
|
|
||||||
|
useEffect(() => {
|
||||||
|
if (!systemId) {
|
||||||
|
setProbeStats([])
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if (chartTime === "1m") {
|
||||||
|
setProbeStats(getCacheValue(systemId, "rt"))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
setProbeStats(getCacheValue(systemId, chartTime))
|
||||||
|
}, [systemId, chartTime])
|
||||||
|
|
||||||
// fetch missing probe stats on load and when chart time changes
|
// fetch missing probe stats on load and when chart time changes
|
||||||
useEffect(() => {
|
useEffect(() => {
|
||||||
if (!systemId || !chartTime || chartTime === "1m") {
|
if (!systemId || !chartTime || chartTime === "1m") {
|
||||||
@@ -148,7 +160,7 @@ export function useNetworkProbeStats(props: UseNetworkProbeStatsProps) {
|
|||||||
setProbeStats(newStats)
|
setProbeStats(newStats)
|
||||||
}
|
}
|
||||||
)
|
)
|
||||||
}, [chartTime])
|
}, [systemId, chartTime])
|
||||||
|
|
||||||
// Subscribe to new probe stats on non-1m chart times (1h, 12h, etc)
|
// Subscribe to new probe stats on non-1m chart times (1h, 12h, etc)
|
||||||
useEffect(() => {
|
useEffect(() => {
|
||||||
@@ -216,20 +228,11 @@ export function useNetworkProbeStats(props: UseNetworkProbeStatsProps) {
|
|||||||
|
|
||||||
return probeStats
|
return probeStats
|
||||||
}
|
}
|
||||||
// function probesToStats(probes: NetworkProbeRecord[]): NetworkProbeStatsRecord["stats"] {
|
async function fetchProbes(system?: string) {
|
||||||
// const stats: NetworkProbeStatsRecord["stats"] = {}
|
|
||||||
// for (const probe of probes) {
|
|
||||||
// // TODO: include only if probe.updated < charttime
|
|
||||||
// stats[probe.id] = [probe.res, probe.resAvg1h, probe.resMin1h, probe.resMax1h, probe.loss, probe.loss1h]
|
|
||||||
// }
|
|
||||||
// return stats
|
|
||||||
// }
|
|
||||||
|
|
||||||
async function fetchProbes(systemId?: string) {
|
|
||||||
try {
|
try {
|
||||||
const res = await pb.collection<NetworkProbeRecord>("network_probes").getList(0, 2000, {
|
const res = await pb.collection<NetworkProbeRecord>("network_probes").getList(0, 2000, {
|
||||||
fields: NETWORK_PROBE_FIELDS,
|
fields: NETWORK_PROBE_FIELDS,
|
||||||
filter: systemId ? pb.filter("system={:system}", { system: systemId }) : undefined,
|
filter: system ? pb.filter("system={:system}", { system }) : undefined,
|
||||||
})
|
})
|
||||||
return res.items
|
return res.items
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
|
|||||||
@@ -443,6 +443,48 @@ export function runOnce<T extends (...args: any[]) => any>(fn: T): T {
|
|||||||
}) as T
|
}) as T
|
||||||
}
|
}
|
||||||
|
|
||||||
|
const visualWidthCache = new Map<string, number>()
|
||||||
|
|
||||||
|
/** Get the visual width of a string, accounting for full-width and narrow punctuation characters.
|
||||||
|
* Don't use for monospaced fonts, use .length instead
|
||||||
|
*/
|
||||||
|
function getVisualStringWidth(str: string): number {
|
||||||
|
const cached = visualWidthCache.get(str)
|
||||||
|
if (cached !== undefined) {
|
||||||
|
return cached
|
||||||
|
}
|
||||||
|
let width = 0
|
||||||
|
for (const char of str) {
|
||||||
|
if (char === ".") {
|
||||||
|
width += 0.7
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
const code = char.codePointAt(0) || 0
|
||||||
|
// Hangul Jamo and Syllables are often slightly thinner than Hanzi/Kanji
|
||||||
|
if ((code >= 0x1100 && code <= 0x115f) || (code >= 0xac00 && code <= 0xd7af)) {
|
||||||
|
width += 1.8
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
// Count CJK and other full-width characters as 2 units, others as 1
|
||||||
|
// Arabic and Cyrillic are counted as 1
|
||||||
|
const isFullWidth =
|
||||||
|
(code >= 0x2e80 && code <= 0x9fff) || // CJK Radicals, Symbols, and Ideographs
|
||||||
|
(code >= 0xf900 && code <= 0xfaff) || // CJK Compatibility Ideographs
|
||||||
|
(code >= 0xfe30 && code <= 0xfe6f) || // CJK Compatibility Forms
|
||||||
|
(code >= 0xff00 && code <= 0xff60) || // Fullwidth Forms
|
||||||
|
(code >= 0xffe0 && code <= 0xffe6) || // Fullwidth Symbols
|
||||||
|
code > 0xffff // Emojis and other supplementary plane characters
|
||||||
|
width += isFullWidth ? 2 : 1
|
||||||
|
}
|
||||||
|
visualWidthCache.set(str, width)
|
||||||
|
return width
|
||||||
|
}
|
||||||
|
|
||||||
|
/** Compare the visual width of two strings imprecisely */
|
||||||
|
export function isVisuallyLonger(str1: string, str2: string): boolean {
|
||||||
|
return getVisualStringWidth(str1) > getVisualStringWidth(str2)
|
||||||
|
}
|
||||||
|
|
||||||
/** Format seconds to hours, minutes, or seconds */
|
/** Format seconds to hours, minutes, or seconds */
|
||||||
export function secondsToString(seconds: number, unit: "hour" | "minute" | "day"): string {
|
export function secondsToString(seconds: number, unit: "hour" | "minute" | "day"): string {
|
||||||
const count = Math.floor(seconds / (unit === "hour" ? 3600 : unit === "minute" ? 60 : 86400))
|
const count = Math.floor(seconds / (unit === "hour" ? 3600 : unit === "minute" ? 60 : 86400))
|
||||||
@@ -472,45 +514,3 @@ export function secondsToUptimeString(seconds: number): string {
|
|||||||
return secondsToString(seconds, "day")
|
return secondsToString(seconds, "day")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
const visualWidthCache = new Map<string, number>()
|
|
||||||
|
|
||||||
/** Get the visual width of a string, accounting for full-width and narrow punctuation characters.
|
|
||||||
* Don't use for monospaced fonts, use .length instead
|
|
||||||
*/
|
|
||||||
export function getVisualStringWidth(str: string): number {
|
|
||||||
const cached = visualWidthCache.get(str)
|
|
||||||
if (cached !== undefined) {
|
|
||||||
return cached
|
|
||||||
}
|
|
||||||
let width = 0
|
|
||||||
for (const char of str) {
|
|
||||||
if (char === ".") {
|
|
||||||
width += 0.7
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
const code = char.codePointAt(0) || 0
|
|
||||||
// Hangul Jamo and Syllables are often slightly thinner than Hanzi/Kanji
|
|
||||||
if ((code >= 0x1100 && code <= 0x115f) || (code >= 0xac00 && code <= 0xd7af)) {
|
|
||||||
width += 1.8
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
// Count CJK and other full-width characters as 2 units, others as 1
|
|
||||||
// Arabic and Cyrillic are counted as 1
|
|
||||||
const isFullWidth =
|
|
||||||
(code >= 0x2e80 && code <= 0x9fff) || // CJK Radicals, Symbols, and Ideographs
|
|
||||||
(code >= 0xf900 && code <= 0xfaff) || // CJK Compatibility Ideographs
|
|
||||||
(code >= 0xfe30 && code <= 0xfe6f) || // CJK Compatibility Forms
|
|
||||||
(code >= 0xff00 && code <= 0xff60) || // Fullwidth Forms
|
|
||||||
(code >= 0xffe0 && code <= 0xffe6) || // Fullwidth Symbols
|
|
||||||
code > 0xffff // Emojis and other supplementary plane characters
|
|
||||||
width += isFullWidth ? 2 : 1
|
|
||||||
}
|
|
||||||
visualWidthCache.set(str, width)
|
|
||||||
|
|
||||||
return width
|
|
||||||
}
|
|
||||||
|
|
||||||
export function isVisuallyLonger(str1: string, str2: string): boolean {
|
|
||||||
return getVisualStringWidth(str1) > getVisualStringWidth(str2)
|
|
||||||
}
|
|
||||||
|
|||||||
Reference in New Issue
Block a user