Add read-only guest monitoring agents

2026-08-06 20:24:13 +03:00
parent 5401b5b0f2
commit f1ff15e344
8 changed files with 715 additions and 5 deletions
--- a/agent.go
+++ b/agent.go
@@ -0,0 +1,457 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"crypto/hmac"
+	"crypto/rand"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"os"
+	"path/filepath"
+	"runtime"
+	"strconv"
+	"strings"
+	"syscall"
+	"time"
+)
+
+const agentOfflineAfter = 45 * time.Second
+
+type agentOptions struct {
+	Server, EnrollmentToken, Name, ConfigPath string
+	VMID                                      int
+}
+
+type agentConfig struct {
+	Server string `json:"server"`
+	ID     string `json:"id"`
+	Secret string `json:"secret"`
+	Name   string `json:"name"`
+	VMID   int    `json:"vmid"`
+}
+
+type AgentContainer struct {
+	ID          string   `json:"id"`
+	Name        string   `json:"name"`
+	Image       string   `json:"image"`
+	State       string   `json:"state"`
+	Status      string   `json:"status"`
+	Health      string   `json:"health,omitempty"`
+	Restarts    int64    `json:"restarts"`
+	CPUPercent  float64  `json:"cpuPercent"`
+	MemoryUsage uint64   `json:"memoryUsage"`
+	Ports       []string `json:"ports,omitempty"`
+}
+
+type AgentReport struct {
+	Hostname        string           `json:"hostname"`
+	MachineID       string           `json:"machineId"`
+	Version         string           `json:"version"`
+	OS              string           `json:"os"`
+	Arch            string           `json:"arch"`
+	UptimeSeconds   int64            `json:"uptimeSeconds"`
+	Load1           float64          `json:"load1"`
+	CPUPercent      float64          `json:"cpuPercent"`
+	MemoryTotal     uint64           `json:"memoryTotal"`
+	MemoryUsed      uint64           `json:"memoryUsed"`
+	RootTotal       uint64           `json:"rootTotal"`
+	RootUsed        uint64           `json:"rootUsed"`
+	DockerAvailable bool             `json:"dockerAvailable"`
+	DockerError     string           `json:"dockerError,omitempty"`
+	Containers      []AgentContainer `json:"containers"`
+	CollectedAt     int64            `json:"collectedAt"`
+}
+
+type ManagedAgent struct {
+	ID            string      `json:"id"`
+	Name          string      `json:"name"`
+	VMID          int         `json:"vmid"`
+	Hostname      string      `json:"hostname"`
+	Version       string      `json:"version"`
+	LastSeen      int64       `json:"lastSeen"`
+	EnrolledAt    int64       `json:"enrolledAt"`
+	Online        bool        `json:"online"`
+	RemoteAddress string      `json:"remoteAddress"`
+	Report        AgentReport `json:"report"`
+}
+
+type agentEnrollmentRequest struct {
+	Token, Name, Hostname, MachineID string
+	VMID                             int
+}
+
+func randomHex(bytesCount int) (string, error) {
+	b := make([]byte, bytesCount)
+	if _, err := rand.Read(b); err != nil {
+		return "", err
+	}
+	return hex.EncodeToString(b), nil
+}
+
+func tokenHash(value string) string {
+	sum := sha256.Sum256([]byte(value))
+	return hex.EncodeToString(sum[:])
+}
+
+func (s *Store) CreateAgentEnrollmentToken() (string, error) {
+	token, err := randomHex(24)
+	if err != nil {
+		return "", err
+	}
+	_, err = s.db.Exec(`INSERT INTO agent_enrollment_tokens(token_hash,created_at,expires_at) VALUES(?,?,?)`, tokenHash(token), time.Now().Unix(), time.Now().Add(15*time.Minute).Unix())
+	return token, err
+}
+
+func (s *Store) EnrollAgent(req agentEnrollmentRequest) (agentConfig, error) {
+	tx, err := s.db.Begin()
+	if err != nil {
+		return agentConfig{}, err
+	}
+	defer tx.Rollback()
+	var expires int64
+	if err = tx.QueryRow(`SELECT expires_at FROM agent_enrollment_tokens WHERE token_hash=? AND used_at IS NULL`, tokenHash(req.Token)).Scan(&expires); err != nil || expires < time.Now().Unix() {
+		return agentConfig{}, errors.New("одноразовый токен недействителен или истёк")
+	}
+	id, err := randomHex(12)
+	if err != nil {
+		return agentConfig{}, err
+	}
+	secret, err := randomHex(32)
+	if err != nil {
+		return agentConfig{}, err
+	}
+	name := strings.TrimSpace(req.Name)
+	if name == "" {
+		name = req.Hostname
+	}
+	now := time.Now().Unix()
+	if _, err = tx.Exec(`INSERT INTO agents(id,name,vmid,hostname,machine_id,secret_hash,enrolled_at,last_seen,report_json,remote_address) VALUES(?,?,?,?,?,?,?,?,?,?)`, id, name, req.VMID, req.Hostname, req.MachineID, tokenHash(secret), now, now, `{}`, ""); err != nil {
+		return agentConfig{}, err
+	}
+	if _, err = tx.Exec(`UPDATE agent_enrollment_tokens SET used_at=? WHERE token_hash=?`, now, tokenHash(req.Token)); err != nil {
+		return agentConfig{}, err
+	}
+	if err = tx.Commit(); err != nil {
+		return agentConfig{}, err
+	}
+	return agentConfig{ID: id, Secret: secret, Name: name, VMID: req.VMID}, nil
+}
+
+func (s *Store) SaveAgentReport(id, secret, remote string, report AgentReport) error {
+	var expected string
+	if s.db.QueryRow(`SELECT secret_hash FROM agents WHERE id=?`, id).Scan(&expected) != nil || !hmac.Equal([]byte(expected), []byte(tokenHash(secret))) {
+		return errors.New("агент не авторизован")
+	}
+	data, _ := json.Marshal(report)
+	_, err := s.db.Exec(`UPDATE agents SET hostname=?,version=?,last_seen=?,report_json=?,remote_address=? WHERE id=?`, report.Hostname, report.Version, time.Now().Unix(), string(data), remote, id)
+	return err
+}
+
+func (s *Store) Agents() ([]ManagedAgent, error) {
+	rows, err := s.db.Query(`SELECT id,name,vmid,hostname,version,enrolled_at,last_seen,report_json,remote_address FROM agents ORDER BY name`)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	result := []ManagedAgent{}
+	for rows.Next() {
+		var a ManagedAgent
+		var raw string
+		if err = rows.Scan(&a.ID, &a.Name, &a.VMID, &a.Hostname, &a.Version, &a.EnrolledAt, &a.LastSeen, &raw, &a.RemoteAddress); err != nil {
+			return nil, err
+		}
+		_ = json.Unmarshal([]byte(raw), &a.Report)
+		a.Online = time.Since(time.Unix(a.LastSeen, 0)) < agentOfflineAfter
+		result = append(result, a)
+	}
+	return result, rows.Err()
+}
+
+func (s *Store) DeleteAgent(id string) error {
+	_, err := s.db.Exec(`DELETE FROM agents WHERE id=?`, id)
+	return err
+}
+
+func runGuestAgent(opts agentOptions) error {
+	config, err := loadAgentConfig(opts.ConfigPath)
+	if err != nil && !os.IsNotExist(err) {
+		return err
+	}
+	if opts.Server != "" {
+		config.Server = strings.TrimRight(opts.Server, "/")
+	}
+	if opts.Name != "" {
+		config.Name = opts.Name
+	}
+	if opts.VMID > 0 {
+		config.VMID = opts.VMID
+	}
+	if config.Server == "" {
+		return errors.New("укажите --agent-server")
+	}
+	if config.ID == "" {
+		if opts.EnrollmentToken == "" {
+			return errors.New("для первого запуска нужен --agent-enroll-token")
+		}
+		config, err = enrollRemote(config, opts.EnrollmentToken)
+		if err != nil {
+			return err
+		}
+		if err = saveAgentConfig(opts.ConfigPath, config); err != nil {
+			return err
+		}
+	}
+	client := &http.Client{Timeout: 12 * time.Second}
+	for {
+		report := collectAgentReport()
+		data, _ := json.Marshal(report)
+		req, _ := http.NewRequest(http.MethodPost, config.Server+"/api/agent/report", bytes.NewReader(data))
+		req.Header.Set("Content-Type", "application/json")
+		req.Header.Set("Authorization", "Bearer "+config.ID+"."+config.Secret)
+		response, requestErr := client.Do(req)
+		if requestErr == nil {
+			io.Copy(io.Discard, response.Body)
+			response.Body.Close()
+			if response.StatusCode >= 300 {
+				requestErr = fmt.Errorf("сервер вернул %s", response.Status)
+			}
+		}
+		if requestErr != nil {
+			fmt.Fprintf(os.Stderr, "agent: %v\n", requestErr)
+		}
+		time.Sleep(10 * time.Second)
+	}
+}
+
+func enrollRemote(config agentConfig, token string) (agentConfig, error) {
+	hostname, _ := os.Hostname()
+	machineID, _ := os.ReadFile("/etc/machine-id")
+	payload := agentEnrollmentRequest{Token: token, Name: config.Name, VMID: config.VMID, Hostname: hostname, MachineID: strings.TrimSpace(string(machineID))}
+	data, _ := json.Marshal(payload)
+	response, err := http.Post(config.Server+"/api/agent/enroll", "application/json", bytes.NewReader(data))
+	if err != nil {
+		return config, err
+	}
+	defer response.Body.Close()
+	if response.StatusCode != http.StatusOK {
+		message, _ := io.ReadAll(io.LimitReader(response.Body, 4096))
+		return config, fmt.Errorf("регистрация: %s", strings.TrimSpace(string(message)))
+	}
+	var credentials agentConfig
+	if err = json.NewDecoder(response.Body).Decode(&credentials); err != nil {
+		return config, err
+	}
+	credentials.Server = config.Server
+	return credentials, nil
+}
+
+func loadAgentConfig(path string) (agentConfig, error) {
+	data, err := os.ReadFile(path)
+	if err != nil {
+		return agentConfig{}, err
+	}
+	var c agentConfig
+	err = json.Unmarshal(data, &c)
+	return c, err
+}
+func saveAgentConfig(path string, c agentConfig) error {
+	if err := os.MkdirAll(filepath.Dir(path), 0700); err != nil {
+		return err
+	}
+	data, _ := json.MarshalIndent(c, "", "  ")
+	return os.WriteFile(path, data, 0600)
+}
+
+func collectAgentReport() AgentReport {
+	hostname, _ := os.Hostname()
+	machineID, _ := os.ReadFile("/etc/machine-id")
+	report := AgentReport{Hostname: hostname, MachineID: strings.TrimSpace(string(machineID)), Version: version, OS: runtime.GOOS, Arch: runtime.GOARCH, CollectedAt: time.Now().Unix()}
+	if data, err := os.ReadFile("/proc/uptime"); err == nil {
+		fields := strings.Fields(string(data))
+		if len(fields) > 0 {
+			value, _ := strconv.ParseFloat(fields[0], 64)
+			report.UptimeSeconds = int64(value)
+		}
+	}
+	if data, err := os.ReadFile("/proc/loadavg"); err == nil {
+		fields := strings.Fields(string(data))
+		if len(fields) > 0 {
+			report.Load1, _ = strconv.ParseFloat(fields[0], 64)
+		}
+	}
+	report.CPUPercent = sampleAgentCPU()
+	if data, err := os.ReadFile("/proc/meminfo"); err == nil {
+		values := map[string]uint64{}
+		for _, line := range strings.Split(string(data), "\n") {
+			f := strings.Fields(line)
+			if len(f) >= 2 {
+				v, _ := strconv.ParseUint(f[1], 10, 64)
+				values[strings.TrimSuffix(f[0], ":")] = v * 1024
+			}
+		}
+		report.MemoryTotal = values["MemTotal"]
+		available := values["MemAvailable"]
+		if report.MemoryTotal > available {
+			report.MemoryUsed = report.MemoryTotal - available
+		}
+	}
+	var stat syscall.Statfs_t
+	if syscall.Statfs("/", &stat) == nil {
+		report.RootTotal = stat.Blocks * uint64(stat.Bsize)
+		free := stat.Bavail * uint64(stat.Bsize)
+		if report.RootTotal > free {
+			report.RootUsed = report.RootTotal - free
+		}
+	}
+	report.Containers, report.DockerError = collectDockerContainers()
+	report.DockerAvailable = report.DockerError == ""
+	return report
+}
+
+func sampleAgentCPU() float64 {
+	total1, idle1 := readAgentCPU()
+	time.Sleep(100 * time.Millisecond)
+	total2, idle2 := readAgentCPU()
+	if total2 <= total1 {
+		return 0
+	}
+	totalDelta, idleDelta := total2-total1, idle2-idle1
+	return float64(totalDelta-idleDelta) / float64(totalDelta) * 100
+}
+
+func readAgentCPU() (uint64, uint64) {
+	data, err := os.ReadFile("/proc/stat")
+	if err != nil {
+		return 0, 0
+	}
+	line := strings.SplitN(string(data), "\n", 2)[0]
+	fields := strings.Fields(line)
+	if len(fields) < 5 {
+		return 0, 0
+	}
+	var total uint64
+	values := make([]uint64, 0, len(fields)-1)
+	for _, field := range fields[1:] {
+		v, _ := strconv.ParseUint(field, 10, 64)
+		values = append(values, v)
+		total += v
+	}
+	idle := values[3]
+	if len(values) > 4 {
+		idle += values[4]
+	}
+	return total, idle
+}
+
+func collectDockerContainers() ([]AgentContainer, string) {
+	if _, err := os.Stat("/var/run/docker.sock"); err != nil {
+		return []AgentContainer{}, "Docker socket недоступен"
+	}
+	transport := &http.Transport{DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
+		return (&net.Dialer{}).DialContext(ctx, "unix", "/var/run/docker.sock")
+	}}
+	client := &http.Client{Transport: transport, Timeout: 5 * time.Second}
+	response, err := client.Get("http://docker/containers/json?all=1")
+	if err != nil {
+		return nil, err.Error()
+	}
+	defer response.Body.Close()
+	if response.StatusCode != 200 {
+		return nil, response.Status
+	}
+	var raw []struct {
+		ID, Image, State, Status string
+		Names                    []string
+		Ports                    []struct {
+			IP                      string
+			PrivatePort, PublicPort int
+			Type                    string
+		}
+	}
+	if err = json.NewDecoder(response.Body).Decode(&raw); err != nil {
+		return nil, err.Error()
+	}
+	result := make([]AgentContainer, 0, len(raw))
+	for _, v := range raw {
+		name := strings.TrimPrefix(firstString(v.Names), "/")
+		item := AgentContainer{ID: shortID(v.ID), Name: name, Image: v.Image, State: v.State, Status: v.Status}
+		for _, p := range v.Ports {
+			if p.PublicPort > 0 {
+				item.Ports = append(item.Ports, fmt.Sprintf("%d:%d/%s", p.PublicPort, p.PrivatePort, p.Type))
+			}
+		}
+		inspect, inspectErr := client.Get("http://docker/containers/" + v.ID + "/json")
+		if inspectErr == nil {
+			var detail struct {
+				RestartCount int64
+				State        struct{ Health *struct{ Status string } }
+			}
+			if json.NewDecoder(inspect.Body).Decode(&detail) == nil {
+				item.Restarts = detail.RestartCount
+				if detail.State.Health != nil {
+					item.Health = detail.State.Health.Status
+				}
+			}
+			inspect.Body.Close()
+		}
+		statsResponse, statsErr := client.Get("http://docker/containers/" + v.ID + "/stats?stream=false")
+		if statsErr == nil {
+			var stats struct {
+				CPUStats struct {
+					CPUUsage struct {
+						TotalUsage uint64 `json:"total_usage"`
+					} `json:"cpu_usage"`
+					SystemCPUUsage uint64 `json:"system_cpu_usage"`
+					OnlineCPUs     uint64 `json:"online_cpus"`
+				} `json:"cpu_stats"`
+				PreCPUStats struct {
+					CPUUsage struct {
+						TotalUsage uint64 `json:"total_usage"`
+					} `json:"cpu_usage"`
+					SystemCPUUsage uint64 `json:"system_cpu_usage"`
+				} `json:"precpu_stats"`
+				MemoryStats struct {
+					Usage uint64            `json:"usage"`
+					Stats map[string]uint64 `json:"stats"`
+				} `json:"memory_stats"`
+			}
+			if json.NewDecoder(statsResponse.Body).Decode(&stats) == nil {
+				cpuDelta := stats.CPUStats.CPUUsage.TotalUsage - stats.PreCPUStats.CPUUsage.TotalUsage
+				systemDelta := stats.CPUStats.SystemCPUUsage - stats.PreCPUStats.SystemCPUUsage
+				cpus := stats.CPUStats.OnlineCPUs
+				if cpus == 0 {
+					cpus = 1
+				}
+				if systemDelta > 0 {
+					item.CPUPercent = float64(cpuDelta) / float64(systemDelta) * float64(cpus) * 100
+				}
+				item.MemoryUsage = stats.MemoryStats.Usage
+				if cache := stats.MemoryStats.Stats["inactive_file"]; item.MemoryUsage > cache {
+					item.MemoryUsage -= cache
+				}
+			}
+			statsResponse.Body.Close()
+		}
+		result = append(result, item)
+	}
+	return result, ""
+}
+func firstString(v []string) string {
+	if len(v) > 0 {
+		return v[0]
+	}
+	return ""
+}
+func shortID(v string) string {
+	if len(v) > 12 {
+		return v[:12]
+	}
+	return v
+}