diff --git a/internal/komari/client.go b/internal/komari/client.go index fd3416d..a85627c 100644 --- a/internal/komari/client.go +++ b/internal/komari/client.go @@ -4,12 +4,15 @@ import ( "bytes" "encoding/json" "fmt" + "io" "net/http" "net/http/cookiejar" "net/url" + "strings" "sync" "time" + "nukumizu-backend/config" "nukumizu-backend/postLog" ) @@ -102,6 +105,9 @@ type Client struct { // NewClient creates a new Komari API client. func NewClient(baseURL string) *Client { + // Normalize away a trailing slash so paths are appended as "/api/..." and + // never become "//api/..." (which Komari's router rejects with 404). + baseURL = strings.TrimRight(baseURL, "/") jar, _ := cookiejar.New(nil) return &Client{ baseURL: baseURL, @@ -135,6 +141,10 @@ func (c *Client) Login(username, password string) error { var kr KomariResponse if err := json.NewDecoder(resp.Body).Decode(&kr); err != nil { + if config.GetConfig().System.DebugMode { + respBody, _ := io.ReadAll(resp.Body) + return fmt.Errorf("failed to parse komari login response: %w.\nResponse: %s", err, respBody) + } return fmt.Errorf("failed to parse komari login response: %w", err) } @@ -156,8 +166,9 @@ func (c *Client) Login(username, password string) error { return fmt.Errorf("komari login response missing session_token cookie") } -// getSessionToken returns the current session token. -func (c *Client) getSessionToken() string { +// GetSessionToken returns the current session token (empty string if not logged in). +// The WebSocket client uses it to authenticate the /api/rpc2 connection. +func (c *Client) GetSessionToken() string { c.mu.RLock() defer c.mu.RUnlock() return c.sessionToken diff --git a/internal/komari/ws.go b/internal/komari/ws.go index aaf9f0c..c830a69 100644 --- a/internal/komari/ws.go +++ b/internal/komari/ws.go @@ -3,7 +3,10 @@ package komari import ( "encoding/json" "fmt" + "net" "net/http" + "net/url" + "strings" "sync" "sync/atomic" "time" @@ -15,24 +18,56 @@ import ( "github.com/gorilla/websocket" ) -// WSMessage is the structure of messages received from Komari WebSocket. -type WSMessage struct { - Status string `json:"status"` - Data struct { - Online []string `json:"online"` - Data map[string]node.Report `json:"data"` - } `json:"data"` +// rpcMethodLatestStatus asks Komari for the live status of every node. +// It returns a map keyed by node UUID; each entry carries an `online` flag and +// the flattened latest report fields. +const rpcMethodLatestStatus = "common:getNodesLatestStatus" + +// jsonRpcResponse mirrors Komari's JSON-RPC 2.0 response envelope. +type jsonRpcResponse struct { + Version string `json:"jsonrpc"` + ID any `json:"id,omitempty"` + Result json.RawMessage `json:"result,omitempty"` + Error *struct { + Code int `json:"code"` + Message string `json:"message"` + } `json:"error,omitempty"` +} + +// nodeLatestStatus is one entry returned by common:getNodesLatestStatus. +// Komari flattens the live report into these fields. +type nodeLatestStatus struct { + Time string `json:"time"` + CPU float64 `json:"cpu"` + RAM int64 `json:"ram"` + RAMTotal int64 `json:"ram_total"` + Swap int64 `json:"swap"` + SwapTotal int64 `json:"swap_total"` + Load float64 `json:"load"` + Load5 float64 `json:"load5"` + Load15 float64 `json:"load15"` + Disk int64 `json:"disk"` + DiskTotal int64 `json:"disk_total"` + NetIn int64 `json:"net_in"` + NetOut int64 `json:"net_out"` + NetTotalUp int64 `json:"net_total_up"` + NetTotalDown int64 `json:"net_total_down"` + Process int `json:"process"` + Connections int `json:"connections"` + ConnectionsUDP int `json:"connections_udp"` + Online bool `json:"online"` + Uptime int64 `json:"uptime"` } // WSClient manages a WebSocket connection to Komari for real-time status updates. type WSClient struct { - komariURL string - conn *websocket.Conn - connMu sync.Mutex - reconnectCount atomic.Int32 - maxRetries int - stopCh chan struct{} - running atomic.Bool + komariURL string + conn *websocket.Conn + connMu sync.Mutex + reconnectCount atomic.Int32 + maxRetries int + stopCh chan struct{} + running atomic.Bool onReconnectFail func() // Callback when all reconnection attempts fail. } @@ -56,7 +91,6 @@ func (w *WSClient) Connect() error { conn, err := w.dial() if err != nil { - postLog.Error("Failed to connect to Komari WebSocket: " + err.Error()) return err } @@ -67,9 +101,9 @@ func (w *WSClient) Connect() error { w.reconnectCount.Store(0) w.running.Store(true) - // Request initial state. - if err := conn.WriteMessage(websocket.TextMessage, []byte("get")); err != nil { - postLog.Error("Failed to send 'get' message: " + err.Error()) + // Request the initial node status snapshot. + if err := w.requestStatus(conn); err != nil { + postLog.Error("Failed to send initial node status request: " + err.Error()) return err } @@ -79,17 +113,24 @@ func (w *WSClient) Connect() error { return nil } +// dial opens the WebSocket connection to Komari's JSON-RPC endpoint (/api/rpc2). +// Komari's CheckWebSocketOrigin rejects handshakes that carry no Origin header +// (HTTP 403 → "bad handshake"), and without the session_token cookie the +// connection is treated as an anonymous guest, so both are sent. func (w *WSClient) dial() (*websocket.Conn, error) { - // Build the WebSocket URL from the Komari HTTP URL. - wsURL := w.komariURL - if len(wsURL) > 7 && wsURL[:7] == "http://" { - wsURL = "ws://" + wsURL[7:] - } else if len(wsURL) > 8 && wsURL[:8] == "https://" { - wsURL = "wss://" + wsURL[8:] + wsURL, origin, err := w.buildWSURL() + if err != nil { + return nil, err } - wsURL = wsURL + "/api/clients" header := http.Header{} + header.Set("Origin", origin) + if client := GetClient(); client != nil { + if token := client.GetSessionToken(); token != "" { + header.Set("Cookie", "session_token="+token) + } + } + dialer := websocket.Dialer{ HandshakeTimeout: 10 * time.Second, } @@ -98,6 +139,50 @@ func (w *WSClient) dial() (*websocket.Conn, error) { return conn, err } +// buildWSURL converts the configured HTTP(S) base URL into the WebSocket URL for +// /api/rpc2 and derives the Origin header value that matches its host. +func (w *WSClient) buildWSURL() (wsURL, origin string, err error) { + base := strings.TrimRight(w.komariURL, "/") + if base == "" { + return "", "", fmt.Errorf("empty Komari URL") + } + + origin = base + switch { + case strings.HasPrefix(base, "http://"): + wsURL = "ws://" + strings.TrimPrefix(base, "http://") + case strings.HasPrefix(base, "https://"): + wsURL = "wss://" + strings.TrimPrefix(base, "https://") + default: + return "", "", fmt.Errorf("unsupported Komari URL scheme: %s", base) + } + + wsURL += "/api/rpc2" + + // Clean the Origin to "scheme://host" (no path/trailing slash). + if u, perr := url.Parse(origin); perr == nil { + origin = u.Scheme + "://" + u.Host + } + return wsURL, origin, nil +} + +// requestStatus asks Komari for a fresh snapshot of all node statuses. +func (w *WSClient) requestStatus(conn *websocket.Conn) error { + payload, err := json.Marshal(map[string]any{ + "jsonrpc": "2.0", + "method": rpcMethodLatestStatus, + "params": map[string]any{}, + "id": 1, + }) + if err != nil { + return err + } + return conn.WriteMessage(websocket.TextMessage, payload) +} + +// readLoop reads JSON-RPC responses and feeds the node tracker. +// Komari never pushes unsolicited messages on /api/rpc2, so the loop re-requests +// a fresh snapshot whenever the connection stays quiet for a refresh interval. func (w *WSClient) readLoop() { defer func() { if r := recover(); r != nil { @@ -106,6 +191,11 @@ func (w *WSClient) readLoop() { w.running.Store(false) }() + const ( + refreshInterval = 5 * time.Second + readTimeout = refreshInterval + 5*time.Second + ) + for { select { case <-w.stopCh: @@ -121,33 +211,94 @@ func (w *WSClient) readLoop() { return } + if err := conn.SetReadDeadline(time.Now().Add(readTimeout)); err != nil { + postLog.Warning("Komari WebSocket set read deadline error: " + err.Error()) + w.tryReconnect() + return + } + _, data, err := conn.ReadMessage() if err != nil { + if netErr, ok := err.(net.Error); ok && netErr.Timeout() { + // No push within the window — pull a fresh snapshot. + if rerr := w.requestStatus(conn); rerr != nil { + postLog.Warning("Failed to request node status refresh: " + rerr.Error()) + } + continue + } postLog.Warning("Komari WebSocket read error: " + err.Error()) w.tryReconnect() return } - var msg WSMessage - if err := json.Unmarshal(data, &msg); err != nil { + var rpcResp jsonRpcResponse + if err := json.Unmarshal(data, &rpcResp); err != nil { postLog.Warning("Failed to parse Komari WebSocket message: " + err.Error()) continue } - - if msg.Status != "success" { - postLog.Warning("Komari WebSocket message with non-success status") + if rpcResp.Error != nil { + postLog.Warning("Komari WebSocket RPC error: " + rpcResp.Error.Message) + continue + } + if len(rpcResp.Result) == 0 { continue } - // Update the node tracker with the received data. - tracker := node.GetTracker() - if tracker != nil { - tracker.UpdateStatus(msg.Data.Online, msg.Data.Data) + online, reports := parseLatestStatus(rpcResp.Result) + if tracker := node.GetTracker(); tracker != nil { + tracker.UpdateStatus(online, reports) } } } +// parseLatestStatus converts a common:getNodesLatestStatus result into the +// online UUID list and per-node report map consumed by the node tracker. +func parseLatestStatus(raw json.RawMessage) ([]string, map[string]node.Report) { + var statuses map[string]nodeLatestStatus + if err := json.Unmarshal(raw, &statuses); err != nil { + postLog.Warning("Failed to parse node latest status: " + err.Error()) + return nil, map[string]node.Report{} + } + + online := make([]string, 0, len(statuses)) + reports := make(map[string]node.Report, len(statuses)) + for uuid, s := range statuses { + if s.Online { + online = append(online, uuid) + } + + var r node.Report + r.CPU.Usage = s.CPU + r.RAM.Total = s.RAMTotal + r.RAM.Used = s.RAM + r.Swap.Total = s.SwapTotal + r.Swap.Used = s.Swap + r.Load.Load1 = s.Load + r.Load.Load5 = s.Load5 + r.Load.Load15 = s.Load15 + r.Disk.Total = s.DiskTotal + r.Disk.Used = s.Disk + r.Network.Up = s.NetOut + r.Network.Down = s.NetIn + r.Network.TotalUp = s.NetTotalUp + r.Network.TotalDown = s.NetTotalDown + r.Connections.TCP = s.Connections - s.ConnectionsUDP + r.Connections.UDP = s.ConnectionsUDP + r.Uptime = int(s.Uptime) + r.Process = s.Process + r.UpdatedAt = s.Time + reports[uuid] = r + } + return online, reports +} + func (w *WSClient) tryReconnect() { + select { + case <-w.stopCh: + return + default: + } + count := w.reconnectCount.Add(1) if int(count) > w.maxRetries { postLog.Error(fmt.Sprintf("Komari WebSocket reconnection failed after %d attempts", w.maxRetries)) @@ -188,9 +339,9 @@ func (w *WSClient) tryReconnect() { w.reconnectCount.Store(0) - // Request full state after reconnect. - if err := conn.WriteMessage(websocket.TextMessage, []byte("get")); err != nil { - postLog.Error("Failed to send 'get' after reconnect: " + err.Error()) + // Request a fresh status snapshot after reconnect. + if err := w.requestStatus(conn); err != nil { + postLog.Error("Failed to send status request after reconnect: " + err.Error()) return } diff --git a/internal/node/tracker.go b/internal/node/tracker.go index 01b39ce..a352129 100644 --- a/internal/node/tracker.go +++ b/internal/node/tracker.go @@ -137,7 +137,7 @@ func (t *Tracker) UpdateNodeList(nodeNames map[string]string) { } } - postLog.Debug(fmt.Sprintf("Node list updated: %d nodes", len(nodeNames))) + // postLog.Debug(fmt.Sprintf("Node list updated: %d nodes", len(nodeNames))) } // UpdateStatus processes a WebSocket status update from Komari. diff --git a/main.go b/main.go index 156f41f..1cfb0c7 100644 --- a/main.go +++ b/main.go @@ -44,9 +44,7 @@ func main() { flag.Parse() // Log startup banner. - postLog.Info(fmt.Sprintf("%s Ver.%s.%d.%s", softwareInfo.Name, softwareInfo.Version, softwareInfo.BuildVer, softwareInfo.BuildType)) - postLog.Info(fmt.Sprintf("Developed by %s", softwareInfo.Developer)) - postLog.Info(softwareInfo.Description) + postLog.Info(fmt.Sprintf("%s Ver.%s.%d.%s Developed by %s", softwareInfo.Name, softwareInfo.Version, softwareInfo.BuildVer, softwareInfo.BuildType, softwareInfo.Developer)) // Load configuration. cfg, err := config.LoadConfig(*configPath) @@ -59,9 +57,6 @@ func main() { postLog.InitLogBroadcaster() dbPath := cfg.DBPath - if dbPath == "" { - dbPath = "./db" - } if err := postLog.InitLogsDatabase(fmt.Sprintf("%s/log.db", dbPath)); err != nil { postLog.Fatal("Failed to initialize logs database: " + err.Error()) @@ -92,6 +87,7 @@ func main() { if err := komari.LoginAndStart(); err != nil { postLog.Fatal("Failed to login to Komari: " + err.Error()) + return } // --- Start Komari WebSocket connection ---