feat: add xhttp transport, network monitoring and kernel auto-switch

This commit is contained in:
xboard
2026-04-18 01:39:49 +08:00
parent 65d97f3da1
commit 71c53e1071
5 changed files with 175 additions and 1 deletions
+41
View File
@@ -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 {
+15
View File
@@ -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)
}
+29
View File
@@ -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":
+78
View File
@@ -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)
+12 -1
View File
@@ -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)
}