diff --git a/README.md b/README.md index 77dbec7..1e325d9 100644 --- a/README.md +++ b/README.md @@ -161,6 +161,42 @@ daemon -f -p /var/run/rmm-agent.pid -o /usr/local/rmm/rmm-agent.log \ | GET | `/api/v1/agents/{id}/system` | Nur Systemdaten | | DELETE | `/api/v1/agents/{id}` | Agent loeschen | +### Agent-Status + +Status wird automatisch aus `last_heartbeat` berechnet (kein DB-Feld): + +| Status | Bedingung | +|--------|-----------| +| `online` | Heartbeat < 2 Minuten | +| `stale` | Heartbeat < 5 Minuten | +| `offline` | Heartbeat > 5 Minuten | +| `unknown` | Kein Heartbeat vorhanden | + +`GET /agents` liefert ein `status` Feld pro Agent, `GET /agents/{id}` ebenfalls. + +### Agent-Events + +| Methode | Endpoint | Beschreibung | +|---------|----------|-------------| +| GET | `/api/v1/agents/events` | Alle Events (`?limit=N&type=X`) | +| GET | `/api/v1/agents/{id}/events` | Events eines Agents (`?limit=N&type=X`) | + +Event-Typen: `online`, `offline`, `stale`, `connected`, `disconnected` + +Events werden automatisch geschrieben bei: +- WebSocket Connect/Disconnect (`connected`/`disconnected`) +- Inaktivitaets-Monitor erkennt Statuswechsel (`online`/`stale`/`offline`) + +```bash +# Alle Events (letzte 10) +curl -sk -H "X-API-Key: 01532e5a7c9e70bf2757df77a2f5b9b9" \ + "https://192.168.85.13:8443/api/v1/agents/events?limit=10" + +# Nur Offline-Events eines Agents +curl -sk -H "X-API-Key: 01532e5a7c9e70bf2757df77a2f5b9b9" \ + "https://192.168.85.13:8443/api/v1/agents/6beb8dfd38ecf03eab6c0f21b4ac7bee/events?type=offline" +``` + ### WebSocket | Methode | Endpoint | Beschreibung | @@ -518,8 +554,10 @@ rmm/ │ ├── models/ │ │ ├── agent.go # Agent, Register/Heartbeat Structs │ │ └── system.go # SystemData, alle Collector-Structs +│ ├── monitor/ +│ │ └── monitor.go # Inaktivitaets-Monitor (prueft Heartbeat, schreibt Events) │ ├── api/ -│ │ ├── handlers.go # REST API Handler +│ │ ├── handlers.go # REST API Handler (inkl. Agent-Status + Events) │ │ ├── tunnel_proxy.go # Tunnel + Exec + Update REST Endpoints │ │ ├── backup.go # Config-Backup Endpoints (trigger, list, download, diff) │ │ ├── wireguard.go # WireGuard Peer API Endpoints diff --git a/backend/api/handlers.go b/backend/api/handlers.go index 5e67e1e..3bbc637 100644 --- a/backend/api/handlers.go +++ b/backend/api/handlers.go @@ -6,6 +6,7 @@ import ( "encoding/json" "log" "net/http" + "strconv" "strings" "time" @@ -28,7 +29,9 @@ func (h *Handler) SetupRoutes(mux *http.ServeMux, hub *ws.Hub) { mux.HandleFunc("POST /api/v1/agent/register", h.registerAgent) mux.HandleFunc("POST /api/v1/agent/heartbeat", h.heartbeat) mux.HandleFunc("GET /api/v1/agents", h.listAgents) + mux.HandleFunc("GET /api/v1/agents/events", h.getAllAgentEvents) mux.HandleFunc("GET /api/v1/agents/{id}", h.getAgent) + mux.HandleFunc("GET /api/v1/agents/{id}/events", h.getAgentEvents) mux.HandleFunc("GET /api/v1/agents/{id}/system", h.getSystemData) mux.HandleFunc("DELETE /api/v1/agents/{id}", h.deleteAgent) @@ -151,7 +154,19 @@ func (h *Handler) listAgents(w http.ResponseWriter, r *http.Request) { if agents == nil { agents = []models.Agent{} } - writeJSON(w, http.StatusOK, agents) + + // Status fuer jeden Agent berechnen + var response []models.AgentResponse + for _, a := range agents { + response = append(response, models.AgentResponse{ + Agent: a, + Status: models.CalculateAgentStatus(a.LastHeartbeat), + }) + } + if response == nil { + response = []models.AgentResponse{} + } + writeJSON(w, http.StatusOK, response) } // GET /api/v1/agents/{id} @@ -170,6 +185,7 @@ func (h *Handler) getAgent(w http.ResponseWriter, r *http.Request) { result := map[string]interface{}{ "agent": agent, + "status": models.CalculateAgentStatus(agent.LastHeartbeat), "system_data": sysData, } writeJSON(w, http.StatusOK, result) @@ -207,3 +223,42 @@ func (h *Handler) deleteAgent(w http.ResponseWriter, r *http.Request) { log.Printf("Agent geloescht: %s", id) writeJSON(w, http.StatusOK, map[string]string{"message": "Agent geloescht"}) } + +// GET /api/v1/agents/{id}/events +func (h *Handler) getAgentEvents(w http.ResponseWriter, r *http.Request) { + agentID := r.PathValue("id") + eventType := r.URL.Query().Get("type") + limit := 100 + if l := r.URL.Query().Get("limit"); l != "" { + if n, err := strconv.Atoi(l); err == nil && n > 0 { + limit = n + } + } + + events, err := h.db.GetAgentEvents(agentID, eventType, limit) + if err != nil { + log.Printf("Fehler beim Laden der Events: %v", err) + writeError(w, http.StatusInternalServerError, "Fehler beim Laden") + return + } + writeJSON(w, http.StatusOK, events) +} + +// GET /api/v1/agents/events +func (h *Handler) getAllAgentEvents(w http.ResponseWriter, r *http.Request) { + eventType := r.URL.Query().Get("type") + limit := 100 + if l := r.URL.Query().Get("limit"); l != "" { + if n, err := strconv.Atoi(l); err == nil && n > 0 { + limit = n + } + } + + events, err := h.db.GetAllAgentEvents(eventType, limit) + if err != nil { + log.Printf("Fehler beim Laden der Events: %v", err) + writeError(w, http.StatusInternalServerError, "Fehler beim Laden") + return + } + writeJSON(w, http.StatusOK, events) +} diff --git a/backend/db/postgres.go b/backend/db/postgres.go index d402979..522f484 100644 --- a/backend/db/postgres.go +++ b/backend/db/postgres.go @@ -88,6 +88,15 @@ func (d *Database) migrate() error { CREATE INDEX IF NOT EXISTS idx_config_backups_agent ON config_backups(agent_id, created_at DESC); CREATE INDEX IF NOT EXISTS idx_config_backups_hash ON config_backups(agent_id, hash); + + CREATE TABLE IF NOT EXISTS agent_events ( + id SERIAL PRIMARY KEY, + agent_id TEXT NOT NULL REFERENCES agents(id) ON DELETE CASCADE, + event_type TEXT NOT NULL, + message TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() + ); + CREATE INDEX IF NOT EXISTS idx_agent_events_agent ON agent_events(agent_id, created_at DESC); ` if _, err := d.db.Exec(schema); err != nil { @@ -631,3 +640,101 @@ func (d *Database) GetMetricsSummary(agentID, metric, bucket string, from, to ti } return results, rows.Err() } + +// --- Agent Events Methoden --- + +// InsertAgentEvent schreibt ein Event in die agent_events Tabelle +func (d *Database) InsertAgentEvent(agentID, eventType, message string) error { + _, err := d.db.Exec( + "INSERT INTO agent_events (agent_id, event_type, message) VALUES ($1, $2, $3)", + agentID, eventType, message, + ) + if err != nil { + log.Printf("Event schreiben fehlgeschlagen: %v", err) + } + return err +} + +// GetAgentEvents holt Events fuer einen bestimmten Agent +func (d *Database) GetAgentEvents(agentID string, eventType string, limit int) ([]models.AgentEvent, error) { + if limit <= 0 { + limit = 100 + } + + query := "SELECT id, agent_id, event_type, COALESCE(message, ''), created_at FROM agent_events WHERE agent_id = $1" + args := []interface{}{agentID} + + if eventType != "" { + query += " AND event_type = $2" + args = append(args, eventType) + } + + query += fmt.Sprintf(" ORDER BY created_at DESC LIMIT $%d", len(args)+1) + args = append(args, limit) + + rows, err := d.db.Query(query, args...) + if err != nil { + return nil, err + } + defer rows.Close() + + var events []models.AgentEvent + for rows.Next() { + var e models.AgentEvent + if err := rows.Scan(&e.ID, &e.AgentID, &e.EventType, &e.Message, &e.CreatedAt); err != nil { + return nil, err + } + events = append(events, e) + } + if events == nil { + events = []models.AgentEvent{} + } + return events, rows.Err() +} + +// GetAllAgentEvents holt Events fuer alle Agents +func (d *Database) GetAllAgentEvents(eventType string, limit int) ([]models.AgentEvent, error) { + if limit <= 0 { + limit = 100 + } + + query := "SELECT id, agent_id, event_type, COALESCE(message, ''), created_at FROM agent_events" + args := []interface{}{} + + if eventType != "" { + query += " WHERE event_type = $1" + args = append(args, eventType) + } + + query += fmt.Sprintf(" ORDER BY created_at DESC LIMIT $%d", len(args)+1) + args = append(args, limit) + + rows, err := d.db.Query(query, args...) + if err != nil { + return nil, err + } + defer rows.Close() + + var events []models.AgentEvent + for rows.Next() { + var e models.AgentEvent + if err := rows.Scan(&e.ID, &e.AgentID, &e.EventType, &e.Message, &e.CreatedAt); err != nil { + return nil, err + } + events = append(events, e) + } + if events == nil { + events = []models.AgentEvent{} + } + return events, rows.Err() +} + +// GetAgentStatus berechnet den Status eines Agents aus last_heartbeat +func (d *Database) GetAgentStatus(agentID string) string { + var lastHB sql.NullTime + err := d.db.QueryRow("SELECT last_heartbeat FROM agents WHERE id = $1", agentID).Scan(&lastHB) + if err != nil || !lastHB.Valid { + return "unknown" + } + return models.CalculateAgentStatus(&lastHB.Time) +} diff --git a/backend/main.go b/backend/main.go index 5ad1284..b960da0 100644 --- a/backend/main.go +++ b/backend/main.go @@ -19,6 +19,7 @@ import ( "github.com/cynfo/rmm-backend/api" "github.com/cynfo/rmm-backend/config" "github.com/cynfo/rmm-backend/db" + "github.com/cynfo/rmm-backend/monitor" "github.com/cynfo/rmm-backend/ws" ) @@ -47,8 +48,15 @@ func main() { // WebSocket Hub starten hub := ws.NewHub() + hub.SetEventCallback(func(agentID, eventType, message string) { + database.InsertAgentEvent(agentID, eventType, message) + }) go hub.Run() + // Inaktivitaets-Monitor starten + mon := monitor.New(database) + go mon.Run() + // Router aufbauen mux := http.NewServeMux() handler := api.NewHandler(database) diff --git a/backend/models/agent.go b/backend/models/agent.go index 000cb6d..770e00e 100644 --- a/backend/models/agent.go +++ b/backend/models/agent.go @@ -13,6 +13,37 @@ type Agent struct { LastHeartbeat *time.Time `json:"last_heartbeat,omitempty"` } +// AgentResponse erweitert Agent um den berechneten Status +type AgentResponse struct { + Agent + Status string `json:"status"` +} + +// AgentEvent repraesentiert ein Event in der agent_events Tabelle +type AgentEvent struct { + ID int `json:"id"` + AgentID string `json:"agent_id"` + EventType string `json:"event_type"` + Message string `json:"message,omitempty"` + CreatedAt time.Time `json:"created_at"` +} + +// CalculateAgentStatus berechnet den Status aus dem letzten Heartbeat +func CalculateAgentStatus(lastHeartbeat *time.Time) string { + if lastHeartbeat == nil { + return "unknown" + } + age := time.Since(*lastHeartbeat) + switch { + case age < 2*time.Minute: + return "online" + case age < 5*time.Minute: + return "stale" + default: + return "offline" + } +} + // RegisterRequest - Agent-Registrierung type RegisterRequest struct { AgentID string `json:"agent_id,omitempty"` diff --git a/backend/monitor/monitor.go b/backend/monitor/monitor.go new file mode 100644 index 0000000..5a81384 --- /dev/null +++ b/backend/monitor/monitor.go @@ -0,0 +1,77 @@ +package monitor + +import ( + "log" + "sync" + "time" + + "github.com/cynfo/rmm-backend/db" + "github.com/cynfo/rmm-backend/models" +) + +// Monitor prueft periodisch den Agent-Status und schreibt Events bei Statuswechsel +type Monitor struct { + db *db.Database + lastStatus map[string]string + lastStatusMu sync.Mutex + interval time.Duration +} + +// New erstellt einen neuen Monitor +func New(database *db.Database) *Monitor { + return &Monitor{ + db: database, + lastStatus: make(map[string]string), + interval: 60 * time.Second, + } +} + +// Run startet den Monitor als Goroutine — blockiert, also mit go aufrufen +func (m *Monitor) Run() { + log.Println("Inaktivitaets-Monitor gestartet (Intervall: 60s)") + + // Initialen Status laden + m.checkAgents() + + ticker := time.NewTicker(m.interval) + defer ticker.Stop() + + for range ticker.C { + m.checkAgents() + } +} + +func (m *Monitor) checkAgents() { + agents, err := m.db.GetAgents() + if err != nil { + log.Printf("Monitor: Fehler beim Laden der Agents: %v", err) + return + } + + m.lastStatusMu.Lock() + defer m.lastStatusMu.Unlock() + + for _, agent := range agents { + newStatus := models.CalculateAgentStatus(agent.LastHeartbeat) + oldStatus, known := m.lastStatus[agent.ID] + + if !known { + // Erster Check — Status merken, kein Event + m.lastStatus[agent.ID] = newStatus + log.Printf("Monitor: Agent %s (%s) initialer Status: %s", agent.Name, agent.ID, newStatus) + continue + } + + if oldStatus == newStatus { + continue + } + + // Statuswechsel erkannt + log.Printf("Monitor: Agent %s (%s) Status-Wechsel: %s -> %s", agent.Name, agent.ID, oldStatus, newStatus) + m.lastStatus[agent.ID] = newStatus + + // Event schreiben + message := "Status-Wechsel: " + oldStatus + " -> " + newStatus + m.db.InsertAgentEvent(agent.ID, newStatus, message) + } +} diff --git a/backend/ws/hub.go b/backend/ws/hub.go index db7a840..7cfea7f 100644 --- a/backend/ws/hub.go +++ b/backend/ws/hub.go @@ -26,6 +26,9 @@ type Hub struct { // Response-Waiters: Command-ID -> Response-Channel responseWaiters map[string]chan map[string]interface{} waitersMux sync.RWMutex + + // Event-Callback fuer DB-Events bei Connect/Disconnect + eventCallback func(agentID, eventType, message string) } func NewHub() *Hub { @@ -41,6 +44,11 @@ func NewHub() *Hub { return hub } +// SetEventCallback setzt die Callback-Funktion fuer Connect/Disconnect Events +func (h *Hub) SetEventCallback(cb func(agentID, eventType, message string)) { + h.eventCallback = cb +} + func (h *Hub) Run() { log.Println("WebSocket Hub startet...") @@ -79,6 +87,11 @@ func (h *Hub) registerAgent(agent *AgentConnection) { agent.LastSeen = time.Now() log.Printf("Agent %s: WebSocket-Verbindung registriert", agent.AgentID) + + // Event schreiben + if h.eventCallback != nil { + go h.eventCallback(agent.AgentID, "connected", "WebSocket-Verbindung hergestellt") + } } func (h *Hub) unregisterAgent(agent *AgentConnection) { @@ -89,7 +102,12 @@ func (h *Hub) unregisterAgent(agent *AgentConnection) { delete(h.agents, agent.AgentID) existing.Connection.Close() log.Printf("Agent %s: WebSocket-Verbindung entfernt", agent.AgentID) - + + // Event schreiben + if h.eventCallback != nil { + go h.eventCallback(agent.AgentID, "disconnected", "WebSocket-Verbindung getrennt") + } + // Alle Tunnel für diesen Agent schließen h.tunnelManager.CloseAgentTunnels(agent.AgentID) }