nano_metrics/server/main.go

package main

import (
	"bufio"
	"fmt"
	"log"
	"net/http"
	"os"
	"strconv"
	"strings"
	"time"

	"github.com/prometheus/client_golang/prometheus/promhttp"
	"github.com/user/nanometrics/server/config"
	"github.com/user/nanometrics/server/db"
	"github.com/user/nanometrics/server/handlers"
	"github.com/user/nanometrics/server/models"
	prom "github.com/user/nanometrics/server/prometheus"
	"github.com/user/nanometrics/server/transport"
	ws "github.com/user/nanometrics/server/websocket"
)

const serverVersion = "0.1.1"

func main() {
	cfg := config.Load()

	database, err := db.Open(cfg.DBPath)
	if err != nil {
		log.Fatalf("DB: %v", err)
	}

	hub := ws.NewHub()

	onMetrics := func(m *models.AgentMetrics) {
		if err := database.UpsertAgent(m); err != nil {
			log.Printf("[ingest] upsert agent: %v", err)
		}
		if err := database.InsertMetrics(m); err != nil {
			log.Printf("[ingest] insert metrics: %v", err)
		}
		prom.Update(m)
		hub.Broadcast(models.WSMessage{
			Type:    "metrics_update",
			AgentID: m.Hostname,
			Data:    m,
		})
	}

	if err := transport.StartUDP(cfg.UDPAddr, onMetrics); err != nil {
		log.Fatalf("UDP: %v", err)
	}

	var mqttClient *transport.MQTTClient
	if mc, err := transport.StartMQTT(cfg.MQTTBroker, cfg.MQTTTopicBase, onMetrics); err != nil {
		log.Printf("[mqtt] non disponible: %v", err)
	} else {
		mqttClient = mc
	}

	pushConfig := func(agentID string, agentCfg *models.AgentConfig) {
		if mqttClient != nil {
			if err := mqttClient.PushConfig(agentID, agentCfg); err != nil {
				log.Printf("[mqtt] push config to %s: %v", agentID, err)
			}
		}
	}

	// Maintenance : nettoyage + détection offline avec notification WS
	go func() {
		ticker := time.NewTicker(time.Minute)
		defer ticker.Stop()
		for range ticker.C {
			srvCfg, _ := database.GetServerConfig()
			_ = database.PruneOldMetrics(srvCfg.RetentionDays)
			ids, _ := database.MarkOfflineAndGetIDs(30)
			for _, id := range ids {
				hub.Broadcast(models.WSMessage{
					Type:    "status_update",
					AgentID: id,
					Data:    map[string]string{"status": "offline"},
				})
			}
		}
	}()

	// Métriques du serveur lui-même → footer du dashboard
	go func() {
		ticker := time.NewTicker(5 * time.Second)
		defer ticker.Stop()
		var prevIdle, prevTotal uint64
		for range ticker.C {
			stats, err := collectServerStats(&prevIdle, &prevTotal)
			if err != nil {
				continue
			}
			stats.Version = serverVersion
			hub.Broadcast(models.WSMessage{Type: "server_stats", Data: stats})
		}
	}()

	mux := http.NewServeMux()
	mux.Handle("/metrics", promhttp.Handler())
	mux.Handle("/ws", ws.Handler(hub))
	mux.HandleFunc("/api/agents", handlers.AgentsHandler(database))
	mux.HandleFunc("/api/agents/", func(w http.ResponseWriter, r *http.Request) {
		switch {
		case endsWith(r.URL.Path, "/history"):
			handlers.MetricsHistoryHandler(database)(w, r)
		case endsWith(r.URL.Path, "/config"):
			handlers.AgentConfigHandler(database, pushConfig)(w, r)
		case endsWith(r.URL.Path, "/icon") && r.Method == http.MethodPost:
			handlers.IconUploadHandler(database)(w, r)
		case endsWith(r.URL.Path, "/icon") && r.Method == http.MethodGet:
			handlers.IconGetHandler(database)(w, r)
		case r.Method == http.MethodDelete:
			handlers.DeleteAgentHandler(database, hub.Broadcast)(w, r)
		case r.Method == http.MethodGet:
			handlers.AgentDetailHandler(database)(w, r)
		default:
			http.NotFound(w, r)
		}
	})
	mux.HandleFunc("/api/config", handlers.ServerConfigHandler(database))

	if cfg.DashboardDir != "" {
		log.Printf("[http] dashboard servi depuis %s", cfg.DashboardDir)
		mux.Handle("/", http.FileServer(http.Dir(cfg.DashboardDir)))
	}

	log.Printf("[http] écoute sur %s", cfg.HTTPAddr)
	log.Fatal(http.ListenAndServe(cfg.HTTPAddr, mux))
}

func endsWith(path, suffix string) bool {
	return len(path) >= len(suffix) && path[len(path)-len(suffix):] == suffix
}

func collectServerStats(prevIdle, prevTotal *uint64) (*models.ServerStats, error) {
	idle, total, err := readCPUStat()
	if err != nil {
		return nil, err
	}
	var cpuPct float64
	if *prevTotal > 0 && total > *prevTotal {
		deltaIdle := float64(idle - *prevIdle)
		deltaTotal := float64(total - *prevTotal)
		cpuPct = 100.0 * (1.0 - deltaIdle/deltaTotal)
		if cpuPct < 0 {
			cpuPct = 0
		}
	}
	*prevIdle = idle
	*prevTotal = total

	memTotal, memAvail, err := readMemInfo()
	if err != nil {
		return nil, err
	}
	return &models.ServerStats{
		CPUPercent: cpuPct,
		MemUsed:    memTotal - memAvail,
		MemTotal:   memTotal,
	}, nil
}

func readCPUStat() (idle, total uint64, err error) {
	f, err := os.Open("/proc/stat")
	if err != nil {
		return 0, 0, err
	}
	defer f.Close()
	scanner := bufio.NewScanner(f)
	for scanner.Scan() {
		line := scanner.Text()
		if !strings.HasPrefix(line, "cpu ") {
			continue
		}
		fields := strings.Fields(line)[1:]
		var vals [10]uint64
		for i, v := range fields {
			if i >= 10 {
				break
			}
			vals[i], _ = strconv.ParseUint(v, 10, 64)
		}
		// idle = idle + iowait
		idle = vals[3] + vals[4]
		for _, v := range vals {
			total += v
		}
		return idle, total, nil
	}
	return 0, 0, fmt.Errorf("cpu line not found in /proc/stat")
}

func readMemInfo() (totalBytes, availBytes int64, err error) {
	f, err := os.Open("/proc/meminfo")
	if err != nil {
		return 0, 0, err
	}
	defer f.Close()
	scanner := bufio.NewScanner(f)
	var total, avail int64
	for scanner.Scan() {
		line := scanner.Text()
		fields := strings.Fields(line)
		if len(fields) < 2 {
			continue
		}
		val, _ := strconv.ParseInt(fields[1], 10, 64)
		switch fields[0] {
		case "MemTotal:":
			total = val * 1024
		case "MemAvailable:":
			avail = val * 1024
		}
		if total > 0 && avail > 0 {
			break
		}
	}
	if total == 0 {
		return 0, 0, fmt.Errorf("MemTotal not found in /proc/meminfo")
	}
	return total, avail, nil
}