diff --git a/install.sh b/install.sh index 5b42c97..fdcfcb7 100644 --- a/install.sh +++ b/install.sh @@ -89,6 +89,7 @@ console_log: mqtt: true llm: true sql: true + meshtastic: true EOF chown "${SERVICE_USER}:${SERVICE_USER}" "${CONFIG_DIR}/config.yaml" chmod 0640 "${CONFIG_DIR}/config.yaml" diff --git a/internal/config/config.go b/internal/config/config.go index b92a3a2..de5cd64 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -83,10 +83,11 @@ type AIConfig struct { // ConsoleLogConfig 控制各模块是否在控制台打印日志。后续若新增模块，按需扩展。 type ConsoleLogConfig struct { - Web bool `yaml:"web"` - MQTT bool `yaml:"mqtt"` - LLM bool `yaml:"llm"` - SQL bool `yaml:"sql"` + Web bool `yaml:"web"` + MQTT bool `yaml:"mqtt"` + LLM bool `yaml:"llm"` + SQL bool `yaml:"sql"` + Meshtastic bool `yaml:"meshtastic"` } type rawConfig struct { @@ -99,10 +100,11 @@ type rawConfig struct { } type rawConsoleLogConfig struct { - Web *bool `yaml:"web"` - MQTT *bool `yaml:"mqtt"` - LLM *bool `yaml:"llm"` - SQL *bool `yaml:"sql"` + Web *bool `yaml:"web"` + MQTT *bool `yaml:"mqtt"` + LLM *bool `yaml:"llm"` + SQL *bool `yaml:"sql"` + Meshtastic *bool `yaml:"meshtastic"` } type rawAIConfig struct { @@ -200,10 +202,11 @@ func Default() *Config { DataDir: defaultDataDir(), }, ConsoleLog: ConsoleLogConfig{ - Web: true, - MQTT: true, - LLM: true, - SQL: true, + Web: true, + MQTT: true, + LLM: true, + SQL: true, + Meshtastic: true, }, } } @@ -515,6 +518,11 @@ func normalize(raw rawConfig) (*Config, bool) { } else { cfg.ConsoleLog.SQL = *raw.ConsoleLog.SQL } + if raw.ConsoleLog.Meshtastic == nil { + changed = true + } else { + cfg.ConsoleLog.Meshtastic = *raw.ConsoleLog.Meshtastic + } } return cfg, changed diff --git a/internal/mqttforward/client_stats.go b/internal/mqttforward/client_stats.go new file mode 100644 index 0000000..bf6220e --- /dev/null +++ b/internal/mqttforward/client_stats.go @@ -0,0 +1,74 @@ +package mqttforward + +import "sync" + +// ClientStats 在内存中维护每个 MQTT 客户端的收/发包数量。 +// key 取自 mqtt.Client.ID（broker 内部唯一标识），客户端断开时由调用方调用 Delete 清除， +// 重新连接同一 client_id 会拿到一份新的零值计数，符合"断链就清空"。 +type ClientStats struct { + mu sync.RWMutex + all map[string]*clientCounter +} + +type clientCounter struct { + In int64 // 客户端 → 服务器（broker 收到的报文数） + Out int64 // 服务器 → 客户端（broker 发出的报文数） +} + +// NewClientStats 返回一个空的统计器。 +func NewClientStats() *ClientStats { + return &ClientStats{all: make(map[string]*clientCounter)} +} + +// IncIn 在 broker 收到客户端报文时调用。clientID 为空直接忽略。 +func (s *ClientStats) IncIn(clientID string) { + if s == nil || clientID == "" { + return + } + s.mu.Lock() + c, ok := s.all[clientID] + if !ok { + c = &clientCounter{} + s.all[clientID] = c + } + c.In++ + s.mu.Unlock() +} + +// IncOut 在 broker 向客户端发出报文时调用。 +func (s *ClientStats) IncOut(clientID string) { + if s == nil || clientID == "" { + return + } + s.mu.Lock() + c, ok := s.all[clientID] + if !ok { + c = &clientCounter{} + s.all[clientID] = c + } + c.Out++ + s.mu.Unlock() +} + +// Get 返回指定 clientID 当前的收/发包数量；不存在时返回 0,0。 +func (s *ClientStats) Get(clientID string) (in, out int64) { + if s == nil || clientID == "" { + return 0, 0 + } + s.mu.RLock() + defer s.mu.RUnlock() + if c, ok := s.all[clientID]; ok { + return c.In, c.Out + } + return 0, 0 +} + +// Delete 在客户端断开连接时清除其计数。重新连接同一 clientID 会从 0 重新计起。 +func (s *ClientStats) Delete(clientID string) { + if s == nil || clientID == "" { + return + } + s.mu.Lock() + delete(s.all, clientID) + s.mu.Unlock() +} diff --git a/internal/web/mqtt_status.go b/internal/web/mqtt_status.go index e3e3eeb..e1d0e45 100644 --- a/internal/web/mqtt_status.go +++ b/internal/web/mqtt_status.go @@ -16,11 +16,12 @@ type MQTTStatusProvider interface { // MQTTRuntimeStatus 把 mqtt.Server / 写队列 / 转发统计三个上下文打包成 // 实现 MQTTStatusProvider 的具体类型。供 main 包构造后注入 newRouter。 type MQTTRuntimeStatus struct { - Server *mqtt.Server - Address string - TLS bool - Stats *mqttforwardpkg.Stats - DBQueue *storepkg.WriteQueue + Server *mqtt.Server + Address string + TLS bool + Stats *mqttforwardpkg.Stats + ClientStats *mqttforwardpkg.ClientStats + DBQueue *storepkg.WriteQueue } // AdminMQTTStatus 是 admin 路由 GET /admin/mqtt-status 返回的 JSON 视图。 @@ -51,12 +52,12 @@ type AdminMQTTStatus struct { } type AdminMQTTClient struct { - ClientID string `json:"client_id"` - Username string `json:"username"` - Listener string `json:"listener"` - RemoteAddr string `json:"remote_addr"` - RemoteHost string `json:"remote_host"` - RemotePort string `json:"remote_port"` + ClientID string `json:"client_id"` + Username string `json:"username"` + Listener string `json:"listener"` + RemoteAddr string `json:"remote_addr"` + PacketsIn int64 `json:"packets_in"` // 客户端 → 服务器 + PacketsOut int64 `json:"packets_out"` // 服务器 → 客户端 } // Status 实现 MQTTStatusProvider。 @@ -94,13 +95,14 @@ func (m MQTTRuntimeStatus) Status() AdminMQTTStatus { continue } info := mqttClientInfo(client) + in, out := m.ClientStats.Get(info.ClientID) status.Clients = append(status.Clients, AdminMQTTClient{ ClientID: info.ClientID, Username: info.Username, Listener: info.Listener, RemoteAddr: info.RemoteAddr, - RemoteHost: info.RemoteHost, - RemotePort: info.RemotePort, + PacketsIn: in, + PacketsOut: out, }) } return status @@ -112,39 +114,16 @@ type mqttClientInfoView struct { Username string Listener string RemoteAddr string - RemoteHost string - RemotePort string } func mqttClientInfo(c *mqtt.Client) mqttClientInfoView { if c == nil { return mqttClientInfoView{} } - info := mqttClientInfoView{ + return mqttClientInfoView{ ClientID: c.ID, Username: string(c.Properties.Username), Listener: c.Net.Listener, RemoteAddr: c.Net.Remote, } - host, port := splitHostPort(c.Net.Remote) - info.RemoteHost = host - info.RemotePort = port - return info -} - -func splitHostPort(addr string) (string, string) { - if addr == "" { - return "", "" - } - // 复用 net.SplitHostPort，但要兼容 "host" 这种没端口的情况。 - for i := len(addr) - 1; i >= 0; i-- { - if addr[i] == ':' { - host := addr[:i] - if len(host) >= 2 && host[0] == '[' && host[len(host)-1] == ']' { - host = host[1 : len(host)-1] - } - return host, addr[i+1:] - } - } - return addr, "" } diff --git a/main.go b/main.go index bcf684a..fe0f49b 100644 --- a/main.go +++ b/main.go @@ -20,17 +20,17 @@ import ( "github.com/mochi-mqtt/server/v2/packets" "meshtastic_mqtt_server/internal/ai" - "meshtastic_mqtt_server/internal/autoreply" "meshtastic_mqtt_server/internal/auth" + "meshtastic_mqtt_server/internal/autoreply" blockingpkg "meshtastic_mqtt_server/internal/blocking" botpkg "meshtastic_mqtt_server/internal/bot" configpkg "meshtastic_mqtt_server/internal/config" + "meshtastic_mqtt_server/internal/llm" + "meshtastic_mqtt_server/internal/mqtpp" mqttforwardpkg "meshtastic_mqtt_server/internal/mqttforward" rspkg "meshtastic_mqtt_server/internal/runtimesettings" storepkg "meshtastic_mqtt_server/internal/store" webpkg "meshtastic_mqtt_server/internal/web" - "meshtastic_mqtt_server/internal/llm" - "meshtastic_mqtt_server/internal/mqtpp" ) const ( @@ -46,14 +46,16 @@ const ( type meshtasticFilterHook struct { mqtt.HookBase - key []byte - dbQueue *storepkg.WriteQueue - stats *mqttforwardpkg.Stats - blocking *blockingpkg.Cache - settings *rspkg.Cache - pkiResolver func(toNodeNum, fromNodeNum uint32) ([]byte, []byte, bool) - autoAcker func(record map[string]any) - consoleLog bool // 控制台是否打印 MQTT 连接/订阅事件 + key []byte + dbQueue *storepkg.WriteQueue + stats *mqttforwardpkg.Stats + clientStats *mqttforwardpkg.ClientStats + blocking *blockingpkg.Cache + settings *rspkg.Cache + pkiResolver func(toNodeNum, fromNodeNum uint32) ([]byte, []byte, bool) + autoAcker func(record map[string]any) + consoleLog bool // 控制台是否打印 MQTT 连接/订阅事件 + packetConsoleLog bool // 控制台是否打印 Meshtastic 数据包 } // ID 返回用于识别 Meshtastic payload 过滤器的 hook 名称。 @@ -68,7 +70,9 @@ func (h *meshtasticFilterHook) Provides(b byte) bool { b == mqtt.OnSessionEstablished || b == mqtt.OnDisconnect || b == mqtt.OnSubscribed || - b == mqtt.OnUnsubscribed + b == mqtt.OnUnsubscribed || + b == mqtt.OnPacketRead || + b == mqtt.OnPacketSent } // OnConnect 在 MQTT 会话建立前拒绝命中 IP 屏蔽表的客户端。 @@ -93,6 +97,9 @@ func (h *meshtasticFilterHook) OnSessionEstablished(cl *mqtt.Client, pk packets. // OnDisconnect 在客户端断开时打印日志，含触发原因。 func (h *meshtasticFilterHook) OnDisconnect(cl *mqtt.Client, err error, expire bool) { + if cl != nil { + h.clientStats.Delete(cl.ID) + } if !h.consoleLog { return } @@ -105,6 +112,22 @@ func (h *meshtasticFilterHook) OnDisconnect(cl *mqtt.Client, err error, expire b info.ClientID, info.Username, info.RemoteHost, info.RemotePort, expire, reason) } +// OnPacketRead 在 broker 收到客户端报文时累计入站计数（客户端 → 服务器）。 +// 返回原始 packet 不做修改；该 hook 在 packet 校验前触发。 +func (h *meshtasticFilterHook) OnPacketRead(cl *mqtt.Client, pk packets.Packet) (packets.Packet, error) { + if cl != nil { + h.clientStats.IncIn(cl.ID) + } + return pk, nil +} + +// OnPacketSent 在 broker 把报文写出后累计出站计数（服务器 → 客户端）。 +func (h *meshtasticFilterHook) OnPacketSent(cl *mqtt.Client, pk packets.Packet, b []byte) { + if cl != nil { + h.clientStats.IncOut(cl.ID) + } +} + // OnSubscribed 客户端订阅成功后打印订阅的 topic filter 列表。 func (h *meshtasticFilterHook) OnSubscribed(cl *mqtt.Client, pk packets.Packet, reasonCodes []byte) { if !h.consoleLog { @@ -152,8 +175,8 @@ func (h *meshtasticFilterHook) OnPublish(cl *mqtt.Client, pk packets.Packet) (pa if h.autoAcker != nil { h.autoAcker(record) } - if record["type"] != "empty_packet" { - printJSON(record) + if h.packetConsoleLog && record["type"] != "empty_packet" { + printMeshtasticRecord(record) } return pk, nil } @@ -296,7 +319,8 @@ func run(cfg *configpkg.Config) error { } messageStats := &mqttforwardpkg.Stats{} - server, mqttHook, mqttAddr, err := startMQTTServer(cfg, store, dbQueue, messageStats, blocking, settings) + clientStats := mqttforwardpkg.NewClientStats() + server, mqttHook, mqttAddr, err := startMQTTServer(cfg, store, dbQueue, messageStats, clientStats, blocking, settings) if err != nil { return err } @@ -385,7 +409,7 @@ func run(cfg *configpkg.Config) error { if err != nil { return err } - mqttStatus := webpkg.MQTTRuntimeStatus{Server: server, Address: mqttAddr, TLS: cfg.MQTT.TLS.Enabled, Stats: messageStats, DBQueue: dbQueue} + mqttStatus := webpkg.MQTTRuntimeStatus{Server: server, Address: mqttAddr, TLS: cfg.MQTT.TLS.Enabled, Stats: messageStats, ClientStats: clientStats, DBQueue: dbQueue} handler := webpkg.NewRouter(cfg.Web, cfg.ConsoleLog.Web, store, sessions, mqttStatus, blocking, forwardManager, settings, botSender) webAddresses := []string{} if cfg.Web.PortEnabled { @@ -440,19 +464,21 @@ func run(cfg *configpkg.Config) error { return runErr } -func startMQTTServer(cfg *configpkg.Config, store *storepkg.Store, dbQueue *storepkg.WriteQueue, stats *mqttforwardpkg.Stats, blocking *blockingpkg.Cache, settings *rspkg.Cache) (*mqtt.Server, *meshtasticFilterHook, string, error) { +func startMQTTServer(cfg *configpkg.Config, store *storepkg.Store, dbQueue *storepkg.WriteQueue, stats *mqttforwardpkg.Stats, clientStats *mqttforwardpkg.ClientStats, blocking *blockingpkg.Cache, settings *rspkg.Cache) (*mqtt.Server, *meshtasticFilterHook, string, error) { server := mqtt.New(&mqtt.Options{InlineClient: true}) if err := server.AddHook(new(mqttauth.AllowHook), nil); err != nil { return nil, nil, "", err } hook := &meshtasticFilterHook{ - key: cfg.Key, - dbQueue: dbQueue, - stats: stats, - blocking: blocking, - settings: settings, - pkiResolver: botpkg.NewPKIKeyResolver(store), - consoleLog: cfg.ConsoleLog.MQTT, + key: cfg.Key, + dbQueue: dbQueue, + stats: stats, + clientStats: clientStats, + blocking: blocking, + settings: settings, + pkiResolver: botpkg.NewPKIKeyResolver(store), + consoleLog: cfg.ConsoleLog.MQTT, + packetConsoleLog: cfg.ConsoleLog.Meshtastic, } if err := server.AddHook(hook, nil); err != nil { return nil, nil, "", err @@ -476,7 +502,76 @@ func startMQTTServer(cfg *configpkg.Config, store *storepkg.Store, dbQueue *stor // printJSON 将记录编码为 JSON 后按数据包类型着色输出。 func printJSON(record map[string]any) { - //printJSONBytes(record, mqtpp.MustJSON(record)) + printJSONBytes(record, mqtpp.MustJSON(record)) +} + +// printMeshtasticRecord 把 Meshtastic 解码后的 record 按 type 拼成可读的彩色单行， +// 不输出原始 JSON。保留与 printJSONBytes 一致的色码方案。 +func printMeshtasticRecord(record map[string]any) { + if record == nil { + return + } + typ, _ := record["type"].(string) + from := stringField(record, "from") + channel := stringField(record, "channel_id") + gateway := stringField(record, "gateway_id") + + var color string + var body string + switch typ { + case "nodeinfo": + color = ansiGreenBGWhiteText + body = fmt.Sprintf("nodeinfo from=%s long=%q short=%q hw=%s role=%s", + from, stringField(record, "long_name"), stringField(record, "short_name"), + stringField(record, "hw_model"), stringField(record, "role")) + case "map_report": + color = ansiBlueBGWhiteText + body = fmt.Sprintf("map_report from=%s long=%q lat=%v lon=%v alt=%v fw=%s region=%s", + from, stringField(record, "long_name"), + record["latitude"], record["longitude"], record["altitude"], + stringField(record, "firmware_version"), stringField(record, "region")) + case "text_message": + color = ansiPurpleBGWhiteText + body = fmt.Sprintf("text from=%s channel=%s text=%q", + from, channel, stringField(record, "text")) + case "position": + color = ansiCyanBGBlackText + body = fmt.Sprintf("position from=%s lat=%v lon=%v alt=%v", + from, record["latitude"], record["longitude"], record["altitude"]) + case "telemetry": + color = ansiYellowBGBlackText + body = fmt.Sprintf("telemetry from=%s tt=%v metrics=%v", + from, record["telemetry_type"], record["metrics"]) + case "routing": + color = ansiGrayBGWhiteText + body = fmt.Sprintf("routing from=%s pkt_id=%v", from, record["packet_id"]) + case "traceroute": + color = ansiGrayBGWhiteText + body = fmt.Sprintf("traceroute from=%s pkt_id=%v", from, record["packet_id"]) + default: + if record["error"] != nil { + color = ansiRedBGWhiteText + body = fmt.Sprintf("%-10s from=%s error=%v topic=%s", typ, from, + record["error"], stringField(record, "topic")) + } else { + body = fmt.Sprintf("%-10s from=%s", typ, from) + } + } + if gateway != "" { + body += " gw=" + gateway + } + if color != "" { + fmt.Printf("%s%s%s\n", color, body, ansiReset) + return + } + fmt.Println(body) +} + +func stringField(record map[string]any, key string) string { + if v, ok := record[key].(string); ok { + return v + } + return "" } // printJSONBytes 使用已编码好的 JSON 文本，并根据记录 type 选择控制台颜色。 diff --git a/meshmap_frontend/src/components/AdminDashboard.vue b/meshmap_frontend/src/components/AdminDashboard.vue index 182b093..3ddea21 100644 --- a/meshmap_frontend/src/components/AdminDashboard.vue +++ b/meshmap_frontend/src/components/AdminDashboard.vue @@ -162,8 +162,8 @@ onBeforeUnmount(() => { Username Listener Remote Addr - Remote Host - Remote Port + 客户端→服务器 + 服务器→客户端 @@ -172,8 +172,8 @@ onBeforeUnmount(() => { {{ client.username || '-' }} {{ client.listener || '-' }} {{ client.remote_addr || '-' }} - {{ client.remote_host || '-' }} - {{ client.remote_port || '-' }} + {{ client.packets_in ?? 0 }} + {{ client.packets_out ?? 0 }} diff --git a/meshmap_frontend/src/types.ts b/meshmap_frontend/src/types.ts index 915d4be..00580c3 100644 --- a/meshmap_frontend/src/types.ts +++ b/meshmap_frontend/src/types.ts @@ -327,8 +327,8 @@ export interface AdminMqttClient { username: string listener: string remote_addr: string - remote_host: string - remote_port: string + packets_in: number + packets_out: number } export interface AdminRuntimeSettings {