diff --git a/internal/kernel/xray/config.go b/internal/kernel/xray/config.go index 9acb459..1603f7b 100644 --- a/internal/kernel/xray/config.go +++ b/internal/kernel/xray/config.go @@ -529,6 +529,31 @@ func applyStreamSettings(base M, nc *model.NodeSpec, tc kernel.TLSCert) { } ss["httpSettings"] = h2Settings + case "xhttp", "splithttp": + ss["network"] = "xhttp" + xhttpSettings := M{} + if nc.NetworkSettings != nil { + if v, ok := nc.NetworkSettings["path"]; ok { + xhttpSettings["path"] = v + } + if v, ok := nc.NetworkSettings["host"]; ok { + xhttpSettings["host"] = v + } + if v, ok := nc.NetworkSettings["mode"]; ok { + xhttpSettings["mode"] = v + } + if v, ok := nc.NetworkSettings["extra"]; ok { + // PHP sends empty arrays [] instead of {} for empty objects; + // xray rejects [] for fields that expect objects (e.g. sockopt, tlsSettings). + // Recursively strip empty arrays from the extra map before passing to xray. + if m, ok := v.(map[string]interface{}); ok && len(m) > 0 { + sanitizeEmptyArrays(m) + xhttpSettings["extra"] = m + } + } + } + ss["xhttpSettings"] = xhttpSettings + case "tcp": // default, no extra settings } @@ -784,6 +809,22 @@ func copyStrings(src []string) []string { return out } +// sanitizeEmptyArrays recursively removes empty slices from a map. +// PHP encodes empty objects as [] instead of {}; xray-core rejects [] +// for fields that expect struct types (e.g. sockopt, tlsSettings). +func sanitizeEmptyArrays(m map[string]interface{}) { + for k, v := range m { + switch val := v.(type) { + case []interface{}: + if len(val) == 0 { + delete(m, k) + } + case map[string]interface{}: + sanitizeEmptyArrays(val) + } + } +} + // extractECHServerKeys extracts the ECH private key from tls_settings.ech // and converts it from PEM to base64 for xray-core's echServerKeys field. func extractECHServerKeys(tlsSettings map[string]interface{}) string { diff --git a/internal/machine/machine.go b/internal/machine/machine.go index cd94694..1b057fc 100644 --- a/internal/machine/machine.go +++ b/internal/machine/machine.go @@ -11,6 +11,7 @@ import ( "github.com/cedar2025/xboard-node/internal/config" "github.com/cedar2025/xboard-node/internal/controlplane" + "github.com/cedar2025/xboard-node/internal/model" "github.com/cedar2025/xboard-node/internal/monitor" "github.com/cedar2025/xboard-node/internal/nlog" "github.com/cedar2025/xboard-node/internal/panel" @@ -134,6 +135,19 @@ func (o *Orchestrator) startNode(ctx context.Context, mn panel.MachineNode) { perNodeClient := o.client.ForNode(mn.ID) + // Pre-fetch node config to detect transport-based kernel requirements. + // If the transport (e.g. xhttp) is incompatible with the configured kernel + // (e.g. singbox), auto-switch to the required kernel for this node. + if cfgSnapshot, err := perNodeClient.GetConfig(); err == nil && cfgSnapshot != nil { + if resolved := model.ResolveKernelForTransport(cfgSnapshot.Network, nodeCfg.Kernel.Type); resolved != nodeCfg.Kernel.Type { + nlog.Core().Info(fmt.Sprintf("machine: auto-switching kernel for node %d (%sā†’%s, transport=%s)", + mn.ID, nodeCfg.Kernel.Type, resolved, cfgSnapshot.Network)) + nodeCfg.Kernel.Type = resolved + } + } + // Reset cached ETag so the subsequent GetConfig in Initial() gets a full response. + perNodeClient.ResetConfigETag() + var push controlplane.PushClient if o.ws != nil { push = &machineNodePush{ @@ -247,6 +261,7 @@ func (o *Orchestrator) reportMachineStatus() { [2]uint64{s.MemTotal, s.MemUsed}, [2]uint64{s.SwapTotal, s.SwapUsed}, [2]uint64{s.DiskTotal, s.DiskUsed}, + s.NetInSpeed, s.NetOutSpeed, ); err != nil { nlog.Core().Warn("machine status report failed", "error", err) } diff --git a/internal/model/validate.go b/internal/model/validate.go index 506b9de..6910658 100644 --- a/internal/model/validate.go +++ b/internal/model/validate.go @@ -36,9 +36,38 @@ func ValidateNodeSpec(n *NodeSpec, kcfg config.KernelConfig) error { if err := ValidateCustomRouteRules(n.CustomRouteRules, kernelType, availableTags); err != nil { return fmt.Errorf("validate custom route rules: %w", err) } + if err := validateTransportKernel(n.Network, kernelType); err != nil { + return err + } return nil } +// singboxUnsupportedTransports lists transport types that sing-box does not support. +var singboxUnsupportedTransports = map[string]bool{ + "xhttp": true, + "splithttp": true, +} + +func validateTransportKernel(network, kernelType string) error { + net := strings.ToLower(strings.TrimSpace(network)) + if kernelType == "singbox" && singboxUnsupportedTransports[net] { + return fmt.Errorf("transport %q is not supported by sing-box kernel; use xray kernel instead", net) + } + return nil +} + +// ResolveKernelForTransport returns the kernel type required by the given +// transport. If the configured kernel cannot handle the transport, it returns +// the kernel that can. Otherwise it returns configuredKernel unchanged. +// This is used in machine mode to auto-switch kernel per node. +func ResolveKernelForTransport(network, configuredKernel string) string { + net := strings.ToLower(strings.TrimSpace(network)) + if configuredKernel == "singbox" && singboxUnsupportedTransports[net] { + return "xray" + } + return configuredKernel +} + func normalizeKernelType(value string) (string, error) { switch strings.ToLower(strings.TrimSpace(value)) { case "singbox", "sing-box": diff --git a/internal/monitor/monitor.go b/internal/monitor/monitor.go index 2f99f6a..c46d394 100644 --- a/internal/monitor/monitor.go +++ b/internal/monitor/monitor.go @@ -2,6 +2,8 @@ package monitor import ( "runtime" + "strings" + "sync" "time" "github.com/cedar2025/xboard-node/internal/nlog" @@ -9,6 +11,7 @@ import ( "github.com/shirou/gopsutil/v4/disk" "github.com/shirou/gopsutil/v4/load" "github.com/shirou/gopsutil/v4/mem" + "github.com/shirou/gopsutil/v4/net" ) var startTime = time.Now() @@ -18,6 +21,9 @@ func init() { // always returns 0% because it has no prior sample. This throwaway call // seeds the baseline so subsequent Collect() calls return real values. cpu.Percent(500*time.Millisecond, false) + + // Seed the network baseline so the first Collect() can compute rates. + collectNetSpeed() } // Status holds system resource metrics @@ -36,11 +42,81 @@ type Status struct { DiskUsed uint64 Goroutines int + // Network speed (bytes/sec), -1 means unavailable (first sample). + NetInSpeed float64 + NetOutSpeed float64 + // GC metrics (process-wide) NumGC uint32 LastPauseMS float64 } +// netBaseline tracks the previous network counters for rate calculation. +var ( + netMu sync.Mutex + netPrevRecv uint64 + netPrevSent uint64 + netPrevTime time.Time + netHasBase bool +) + +// skipInterface returns true for loopback and common virtual interfaces. +func skipInterface(name string) bool { + lower := strings.ToLower(name) + for _, prefix := range []string{"lo", "docker", "veth", "br-", "virbr", "vnet", "tun", "tap"} { + if strings.HasPrefix(lower, prefix) { + return true + } + } + return false +} + +// collectNetSpeed calculates network in/out bytes per second since last call. +// Returns -1, -1 on first call or if counters decreased (reboot). +func collectNetSpeed() (inSpeed, outSpeed float64) { + counters, err := net.IOCounters(true) // per-interface + if err != nil { + nlog.Core().Debug("failed to get network counters", "error", err) + return -1, -1 + } + + var totalRecv, totalSent uint64 + for _, c := range counters { + if skipInterface(c.Name) { + continue + } + totalRecv += c.BytesRecv + totalSent += c.BytesSent + } + + now := time.Now() + + netMu.Lock() + defer netMu.Unlock() + + if !netHasBase { + netPrevRecv, netPrevSent, netPrevTime, netHasBase = totalRecv, totalSent, now, true + return -1, -1 + } + + elapsed := now.Sub(netPrevTime).Seconds() + if elapsed <= 0 { + return -1, -1 + } + + // Counter decreased ā†’ system reboot or interface reset; reset baseline. + if totalRecv < netPrevRecv || totalSent < netPrevSent { + netPrevRecv, netPrevSent, netPrevTime = totalRecv, totalSent, now + return -1, -1 + } + + inSpeed = float64(totalRecv-netPrevRecv) / elapsed + outSpeed = float64(totalSent-netPrevSent) / elapsed + + netPrevRecv, netPrevSent, netPrevTime = totalRecv, totalSent, now + return inSpeed, outSpeed +} + // Collect gathers current system metrics func Collect() Status { var s Status @@ -79,6 +155,8 @@ func Collect() Status { s.DiskUsed = diskStat.Used } + s.NetInSpeed, s.NetOutSpeed = collectNetSpeed() + // GC metrics var ms runtime.MemStats runtime.ReadMemStats(&ms) diff --git a/internal/panel/client.go b/internal/panel/client.go index 402d6cb..a725afc 100644 --- a/internal/panel/client.go +++ b/internal/panel/client.go @@ -75,6 +75,13 @@ func (c *Client) ForNode(nodeID int) *Client { } } +// ResetConfigETag clears the cached config ETag so the next GetConfig call +// returns a full response instead of 304. Used by machine mode after a +// pre-fetch to probe the transport type. +func (c *Client) ResetConfigETag() { + c.configETag = "" +} + // Handshake calls the new v2 API to get WS config + initial data in one shot. func (c *Client) Handshake() (*HandshakeResponse, error) { resp, err := c.doRequest("POST", "/api/v2/server/handshake", nil, "") @@ -335,13 +342,17 @@ func (c *Client) GetMachineNodes() (*MachineNodesResponse, error) { } // ReportMachineStatus sends machine-level load metrics to the panel. -func (c *Client) ReportMachineStatus(cpu float64, mem, swap, disk [2]uint64) error { +// netIn/netOut are bytes/sec; negative values mean "unavailable" (first sample). +func (c *Client) ReportMachineStatus(cpu float64, mem, swap, disk [2]uint64, netIn, netOut float64) error { payload := map[string]interface{}{ "cpu": cpu, "mem": map[string]interface{}{"total": mem[0], "used": mem[1]}, "swap": map[string]interface{}{"total": swap[0], "used": swap[1]}, "disk": map[string]interface{}{"total": disk[0], "used": disk[1]}, } + if netIn >= 0 && netOut >= 0 { + payload["net"] = map[string]interface{}{"in_speed": netIn, "out_speed": netOut} + } return c.postJSON("/api/v2/server/machine/status", payload) }