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

Author: xboard

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 {

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)
 	}

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":

internal/monitor/monitor.go

@@ -2,13 +2,16 @@ package monitor
 
 import (
 	"runtime"
+	"strings"
+	"sync"
 	"time"
 
 	"github.com/cedar2025/xboard-node/internal/nlog"
 	"github.com/shirou/gopsutil/v4/cpu"
 	"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)

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)
 }