kevin/meshtastic_mqtt_server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: console_log.meshtastic 控制 packet 控制台输出；新增每客户端报文统计

Browse Source 
- ConsoleLogConfig 增加 Meshtastic 字段（默认 true）；旧配置自动补齐
- meshtasticFilterHook 增加 packetConsoleLog；OnPublish 中根据开关调用
  新的 printMeshtasticRecord 输出可读单行（key=value、按 type 着色），
  替代原来的 JSON dump；事件型 printJSON 调用保持不变
- 新增 internal/mqttforward/ClientStats，按 client_id 累计 in/out 报文计数
- meshtasticFilterHook 多挂 OnPacketRead / OnPacketSent / OnDisconnect
  事件，用于增减计数；Provides() 同步声明
- AdminMQTTClient JSON 视图删除 RemoteHost/RemotePort，新增
  packets_in / packets_out；前端 types.ts、AdminDashboard.vue 同步更新

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
kevin
2026-06-19 17:09:00 +08:00
co-authored by Claude
parent 937580e24f
commit 49a287e6e7
7 changed files with 237 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
install.sh
+1
View File
@@ -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"
internal/config/config.go
+20 -12
View File
@@ -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
internal/mqttforward/client_stats.go
+74
View File
@@ -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()
}
internal/web/mqtt_status.go
+16 -37
View File
@@ -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, ""
}
main.go
+120 -25
View File
@@ -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 选择控制台颜色。
meshmap_frontend/src/components/AdminDashboard.vue
+4 -4
View File
@@ -162,8 +162,8 @@ onBeforeUnmount(() => {
<th>Username</th>
<th>Listener</th>
<th>Remote Addr</th>
<th>Remote Host</th>
<th>Remote Port</th>
<th>客户端→服务器</th>
<th>服务器→客户端</th>
</tr>
</thead>
<tbody>
@@ -172,8 +172,8 @@ onBeforeUnmount(() => {
<td>{{ client.username || '-' }}</td>
<td>{{ client.listener || '-' }}</td>
<td>{{ client.remote_addr || '-' }}</td>
<td>{{ client.remote_host || '-' }}</td>
<td>{{ client.remote_port || '-' }}</td>
<td>{{ client.packets_in ?? 0 }}</td>
<td>{{ client.packets_out ?? 0 }}</td>
</tr>
</tbody>
</table>
meshmap_frontend/src/types.ts
+2 -2
View File
@@ -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 {