commit 89d5e27874c200c8815560805b920ce9f405bf1e Author: codex Date: Mon May 18 15:25:51 2026 +0000 💥 feat(agent)!: 添加主机监控代理首版 * 实现周期采集、快照缓存、本地 HTTP API 和主动上报 * 补齐 Linux 和 Windows 主机、网络、资源、进程和 TCP 采集 * 添加服务生命周期管理、配置加载、构建脚本和测试覆盖 BREAKING CHANGE: network.ips 拆分为 network.ipv4s 和 network.ipv6s。 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..522ee00 --- /dev/null +++ b/.gitignore @@ -0,0 +1,19 @@ +# Build outputs +dist/ +build/ +*.exe +*.test + +# Go temporary files +*.out +coverage.out + +# Local environment files +.env +*.local + +# Editor and OS metadata +.DS_Store +Thumbs.db +.idea/ +.vscode/ diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..8c5247b --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,115 @@ +# 仓库规则 + +## 项目要求 + +### 项目简介 + +- MFD Monitor Agent 是一个使用 Go 实现的单二进制主机状态监控 agent。 +- agent 需要支持 Windows 和 Linux，用于采集并暴露完整主机状态。 +- agent 默认通过本地 HTTP API 提供查询能力，默认监听 `127.0.0.1:8080`。 +- `GET /health` 用于返回进程存活状态、版本信息、最近快照时间和最近采集状态。 +- `GET /status` 用于返回缓存中的完整 `HostStatus` 主机状态快照。 +- agent 通过后台周期采集刷新内存缓存，HTTP 接口和主动上报都读取同一份缓存。 +- agent 支持可选的周期性 HTTP POST 主动上报，上报数据必须复用 `/status` 的 `HostStatus` JSON 结构。 +- agent 需要通过 CLI 管理服务生命周期，包括 `run`、`install`、`uninstall`、`start`、`stop` 和 `version`。 +- Linux 侧面向 systemd service；Windows 侧面向 Windows Service。 + +### 权限要求 + +- 安装阶段必须要求管理员/root 权限。 +- 首版运行阶段默认使用高权限服务账户，以稳定采集完整进程列表和 TCP 监听端口到进程的映射。 +- 接口层必须支持部分成功：某些字段或子采集器失败时，不应导致整个 `/status` 失败，应返回 `partial` 或具体错误原因。 +- 默认监听地址必须尽量收敛到本地地址 `127.0.0.1`，避免默认暴露敏感主机信息。 + +### 架构要求 + +- 使用单进程一体化架构。 +- 一个 agent 进程同时负责周期采集、快照缓存、HTTP 查询、主动上报，以及 Windows/Linux 服务生命周期管理。 +- 代码应按职责拆分模块： + - `collector/system`：主机名、OS、架构、启动时间、采集时间。 + - `collector/network`：网卡、IPv4/IPv6、MAC。 + - `collector/resource`：CPU、内存、磁盘。 + - `collector/process`：完整进程列表。 + - `collector/socket`：TCP 监听端口及其进程映射。 + - `snapshot`：聚合各采集器输出，形成统一 `HostStatus`。 + - `httpapi`：提供 `GET /health` 和 `GET /status`。 + - `reporter`：定时 POST `HostStatus` 到中心端。 + - `service`：封装 Windows/Linux 服务生命周期和安装命令。 + - `config`：统一加载监听地址、采集周期、上报周期、上报目标、机器标识等配置。 + +### HostStatus 要求 + +- `HostStatus` 应使用统一 JSON 结构，至少覆盖以下信息： + - `agent`：版本、启动时间、快照时间、快照状态。 + - `host`：机器标识、主机名、平台、平台版本、架构、运行时长。 + - `network`：网卡、IP、MAC、MTU、flags、地址列表。 + - `cpu`：逻辑核心数、物理核心数、使用率、负载、型号。 + - `memory`：总量、已用、空闲、可用、使用率。 + - `disks`：设备、挂载点、文件系统类型、总量、已用、空闲、使用率。 + - `processes`：状态和完整进程列表，包含 pid、ppid、名称、exe、cmdline、用户、状态、启动时间、CPU 和内存占用。 + - `tcp_listeners`：状态和 TCP 监听列表，包含 IP、端口、pid、进程名、进程路径、地址族。 + - `errors`：记录子采集器失败原因。 +- `processes.status` 和 `tcp_listeners.status` 必须支持 `ok`、`partial`、`error`。 + +### 采集要求 + +- 必须采用“周期采集 + 内存缓存”策略。 +- agent 启动后应立即采集一次，之后后台定时刷新缓存。 +- HTTP 接口不得在每次请求时实时扫描系统，应直接读取最近一次缓存。 +- 主动上报必须读取同一份缓存，避免重复采集。 +- 采集逻辑需要设置整体采集超时。 +- 单个子采集器失败不能拖垮整次快照。 +- 例如进程采集失败时，仍应返回 CPU、内存、磁盘等其他信息；TCP 监听采集不完整时，应将对应状态标记为 `partial`。 + +### 主动上报要求 + +- agent 除支持外部系统通过 HTTP 拉取状态外，还必须支持定时主动上报。 +- 上报目标使用配置中的固定中心服务 URL。 +- 上报方式为周期性 POST `HostStatus` JSON。 +- 配置项应包含 `report.enabled`、`report.endpoint`、`report.interval`、`report.timeout`。 +- 上报失败和超时应独立记录日志。 +- 上报失败可支持重试或简单退避，但不能影响本地 HTTP 服务，也不能导致进程退出。 + +### CLI 和服务要求 + +- CLI 至少提供以下命令： + - `agent run` + - `agent install` + - `agent uninstall` + - `agent start` + - `agent stop` + - `agent version` +- Linux 下，agent 进程以前台模式运行，并额外提供 systemd unit 模板，由安装流程写入 unit 并启停。 +- Windows 下，使用服务包装库实现 install、uninstall、start、stop、run，不依赖外部 NSSM。 + +### 技术和测试要求 + +- 优先使用 Go 实现，保持单二进制部署。 +- 系统信息采集优先使用成熟跨平台库，例如 `gopsutil`。 +- 如果跨平台库在端口到进程映射能力上不足，可以针对 Windows/Linux 做补充实现。 +- 单元测试应覆盖采集器到 `HostStatus` 的映射、空结果、字段缺失、权限错误、`partial/error` 状态和 snapshot 聚合逻辑。 +- 集成测试应覆盖 `GET /health`、`GET /status`、主动 POST 上报载荷、超时和失败重试行为。 +- 平台验证应覆盖 Windows 服务安装/启动/停止/卸载，以及 Linux systemd 安装、enable、start、restart。 +- 性能与稳定性验证应覆盖大量进程场景、采集周期与 HTTP 并发访问并存、较大 `/status` 响应体的内存占用。 + +### 首版边界 + +- 首版默认包含 IP、MAC、主机名、CPU、内存、磁盘、完整进程列表、TCP 监听端口及进程映射、本地 HTTP 查询、主动 HTTP 上报、Windows/Linux 服务化。 +- 首版默认不包含历史存储、趋势分析、分页查询、UDP 监听信息、线程列表、进程句柄、连接状态详情、强制 Token/mTLS。 + +## 注释规则 + +- 所有导出和内部的方法、函数、类结构、结构体、接口都必须编写清晰注释。 +- Go 代码中，每个包级函数、方法、类型、结构体和接口都需要使用简短注释说明其用途和行为。 +- 结构体注释需要说明该结构体代表什么。字段不够直观、包含单位、涉及安全影响，或会影响外部 JSON/API 行为时，需要为字段补充注释。 +- 方法和函数注释需要说明其职责、重要输入输出、副作用，以及相关错误行为。 +- 关键实现步骤需要添加简洁的行内注释说明逻辑，尤其是操作系统相关采集、权限敏感操作、缓存、上报、服务生命周期处理和错误降级路径。 +- 注释应解释代码为什么存在，或说明不直观逻辑如何工作。避免只重复代码本身含义的注释。 +- 修改代码行为时必须同步更新注释。不要留下过期注释、缺少上下文的 TODO，或容易误导的说明。 + +## 项目风格 + +- 优先使用直接、清晰的 Go 代码，并保持与现有单二进制 agent 结构一致。 +- 代码变更应聚焦在请求的行为范围内。 +- 保留采集器的部分成功行为：某个子系统失败时，应报告该子系统自身错误，而不是导致整个状态响应失败。 +- Linux 和 Windows 行为不同的代码路径，需要使用清晰命名和注释说明差异。 diff --git a/README.md b/README.md new file mode 100644 index 0000000..41772a7 --- /dev/null +++ b/README.md @@ -0,0 +1,79 @@ +# MFD Monitor Agent + +Single-binary host status agent implemented in Go. + +## Features + +- `GET /health` returns process liveness and latest snapshot metadata. +- `GET /status` returns the cached full host snapshot. +- Periodic background collection keeps HTTP responses fast. +- Optional periodic HTTP POST reporting reuses the same `HostStatus` JSON. +- Linux collectors read host, network, CPU, memory, disk, process, and TCP listener data from the OS. +- Windows and Linux service lifecycle commands are exposed through the CLI. + +## Commands + +```bash +agent run [-config config.json] +agent install [-config config.json] [-exe /path/to/agent] +agent uninstall [-config config.json] +agent start [-config config.json] +agent stop [-config config.json] +agent version +``` + +## Build + +Linux: + +```bash +./scripts/build-linux.sh +``` + +Windows PowerShell: + +```powershell +.\scripts\build-windows.ps1 +``` + +Cross-compile Windows from Linux: + +```bash +GOOS_TARGET=windows GOARCH_TARGET=amd64 OUTPUT_NAME=agent-windows-amd64.exe ./scripts/build-linux.sh +``` + +Build outputs are written to `dist/`, which is intentionally ignored by Git. + +## Configuration + +All settings have safe defaults. The HTTP API listens on `127.0.0.1:8080` by default. + +```json +{ + "host_id": "host-a", + "http": { + "listen_address": "127.0.0.1:8080" + }, + "collect": { + "interval": "30s", + "timeout": "10s" + }, + "report": { + "enabled": true, + "endpoint": "http://center.example.internal/agent/status", + "interval": "60s", + "timeout": "5s" + } +} +``` + +Environment overrides: + +- `MFD_AGENT_HOST_ID` +- `MFD_AGENT_LISTEN` +- `MFD_AGENT_COLLECT_INTERVAL` +- `MFD_AGENT_COLLECT_TIMEOUT` +- `MFD_AGENT_REPORT_ENABLED` +- `MFD_AGENT_REPORT_ENDPOINT` +- `MFD_AGENT_REPORT_INTERVAL` +- `MFD_AGENT_REPORT_TIMEOUT` diff --git a/cmd/agent/main.go b/cmd/agent/main.go new file mode 100644 index 0000000..55f8b96 --- /dev/null +++ b/cmd/agent/main.go @@ -0,0 +1,137 @@ +package main + +import ( + "context" + "errors" + "flag" + "fmt" + "log/slog" + "net/http" + "os" + "os/signal" + "syscall" + "time" + + "mfd-monitor-agent/internal/collector" + "mfd-monitor-agent/internal/config" + "mfd-monitor-agent/internal/httpapi" + "mfd-monitor-agent/internal/reporter" + "mfd-monitor-agent/internal/service" + "mfd-monitor-agent/internal/snapshot" +) + +const version = "0.1.0" + +// main runs the CLI and exits non-zero when a command fails. +func main() { + if err := runCLI(os.Args[1:]); err != nil { + _, _ = fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } +} + +// runCLI dispatches agent subcommands and loads config before invoking command handlers. +func runCLI(args []string) error { + if len(args) == 0 { + args = []string{"run"} + } + switch args[0] { + case "run": + fs := flag.NewFlagSet("run", flag.ExitOnError) + configPath := fs.String("config", "", "path to JSON config file") + if err := fs.Parse(args[1:]); err != nil { + return err + } + cfg, err := config.Load(*configPath) + if err != nil { + return err + } + return runAgent(cfg) + case "install": + cfg, executable, configPath, err := commandConfig(args[1:]) + if err != nil { + return err + } + return service.NewManager().Install(executable, cfg, service.InstallOptions{ConfigPath: configPath}) + case "uninstall": + cfg, _, _, err := commandConfig(args[1:]) + if err != nil { + return err + } + return service.NewManager().Uninstall(cfg) + case "start": + cfg, _, _, err := commandConfig(args[1:]) + if err != nil { + return err + } + return service.NewManager().Start(cfg) + case "stop": + cfg, _, _, err := commandConfig(args[1:]) + if err != nil { + return err + } + return service.NewManager().Stop(cfg) + case "version": + fmt.Println(version) + return nil + default: + return fmt.Errorf("unknown command %q; expected run, install, uninstall, start, stop, version", args[0]) + } +} + +// commandConfig parses service command flags and resolves the executable path. +func commandConfig(args []string) (config.Config, string, string, error) { + fs := flag.NewFlagSet("service", flag.ExitOnError) + configPath := fs.String("config", "", "path to JSON config file") + executableFlag := fs.String("exe", "", "path to agent executable") + if err := fs.Parse(args); err != nil { + return config.Config{}, "", "", err + } + cfg, err := config.Load(*configPath) + if err != nil { + return cfg, "", "", err + } + executable := *executableFlag + if executable == "" { + executable, err = os.Executable() + if err != nil { + return cfg, "", "", err + } + } + return cfg, executable, *configPath, nil +} + +// runAgent starts collection, optional reporting, and the local HTTP API until signaled. +func runAgent(cfg config.Config) error { + logger := slog.New(slog.NewJSONHandler(os.Stdout, nil)) + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + + startedAt := time.Now().UTC() + cache := snapshot.NewCache(collector.New(cfg.HostID), version, startedAt) + go cache.Run(ctx, cfg.Collect.Interval.Duration(), cfg.Collect.Timeout.Duration()) + + if cfg.Report.Enabled { + client := reporter.NewClient(cfg.Report.Endpoint, cfg.Report.Timeout.Duration()) + go reporter.Run(ctx, cache, client, cfg.Report.Interval.Duration(), logger) + } + + server := &http.Server{ + Addr: cfg.HTTP.ListenAddress, + Handler: httpapi.New(cache), + ReadHeaderTimeout: 5 * time.Second, + } + go func() { + <-ctx.Done() + shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _ = server.Shutdown(shutdownCtx) + }() + + logger.Info("agent listening", "address", cfg.HTTP.ListenAddress) + err := server.ListenAndServe() + if errors.Is(err, http.ErrServerClosed) { + return nil + } + return err +} diff --git a/docs/superpowers/plans/2026-05-18-mfd-monitor-agent-completion.md b/docs/superpowers/plans/2026-05-18-mfd-monitor-agent-completion.md new file mode 100644 index 0000000..7cb6f2c --- /dev/null +++ b/docs/superpowers/plans/2026-05-18-mfd-monitor-agent-completion.md @@ -0,0 +1,374 @@ +# MFD Monitor Agent Completion Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [x]`) syntax for tracking. + +**Goal:** Complete the host monitoring agent behavior required by `AGENTS.md` while preserving the current single-binary, cached-snapshot architecture. + +**Architecture:** Keep the current packages and add test seams only where needed. Linux collectors are completed with parser-level tests, service installation becomes testable through unit rendering and command injection, reporter retry/backoff stays isolated from HTTP and collection, and Windows service behavior remains build-tagged. + +**Tech Stack:** Go 1.22, standard library HTTP/config/process APIs, Linux `/proc`, systemd, Windows Service build tags, `go test`. + +--- + +## File Structure + +- Modify `cmd/agent/main.go`: preserve config path for service installation and add comments. +- Modify `internal/status/types.go`: add required comments for API-visible JSON model fields and types. +- Modify `internal/config/config.go`: add comments and preserve existing defaults/env behavior. +- Modify `internal/snapshot/snapshot.go`: add comments and tests for timeout/partial behavior. +- Modify `internal/httpapi/server.go`: add comments and tests for not-ready/method behavior. +- Modify `internal/reporter/reporter.go`: add retry/backoff and tests. +- Modify `internal/collector/collector.go`: add comments. +- Modify `internal/collector/collector_linux.go`: complete Linux CPU/process/TCP partial behavior and add parser seams. +- Modify `internal/collector/collector_other.go`: add comments and explicit unsupported behavior. +- Create `internal/collector/collector_linux_test.go`: parser and degradation tests. +- Modify `internal/service/service.go`: add comments and extend install options if needed. +- Modify `internal/service/service_linux.go`: root guard, systemd unit rendering, config path propagation, command seams. +- Create `internal/service/service_linux_test.go`: root guard and unit rendering tests. +- Modify `internal/service/service_windows.go`: document Windows service behavior and improve no-NSSM path where compile-testable. +- Modify `internal/service/service_other.go`: comments. + +## Task 1: Comment Coverage Baseline + +**Files:** +- Modify: `internal/status/types.go` +- Modify: `internal/config/config.go` +- Modify: `internal/snapshot/snapshot.go` +- Modify: `internal/httpapi/server.go` +- Modify: `internal/reporter/reporter.go` +- Modify: `internal/collector/collector.go` +- Modify: `internal/collector/collector_linux.go` +- Modify: `internal/collector/collector_other.go` +- Modify: `internal/service/service.go` +- Modify: `internal/service/service_linux.go` +- Modify: `internal/service/service_windows.go` +- Modify: `internal/service/service_other.go` + +- [x] **Step 1: Run comment lint proxy** + +Run: `go test ./...` + +Expected: PASS. This is the behavioral baseline before comments. + +- [x] **Step 2: Add comments** + +Add concise comments above every package-level type, struct, interface, function, and method. For API structs in `status`, include JSON/API meaning and units where relevant, for example: + +```go +// HostStatus is the full cached host snapshot returned by /status and posted by the reporter. +type HostStatus struct { + // Agent contains agent process metadata and the latest snapshot status. + Agent AgentInfo `json:"agent"` +} +``` + +- [x] **Step 3: Verify comments did not change behavior** + +Run: `go test ./...` + +Expected: PASS. + +## Task 2: HTTP Cache Edge Tests + +**Files:** +- Modify: `internal/httpapi/server_test.go` +- Modify: `internal/httpapi/server.go` + +- [x] **Step 1: Write failing tests** + +Add tests named: + +```go +func TestStatusReturnsServiceUnavailableBeforeSnapshot(t *testing.T) {} +func TestHealthReturnsStartingBeforeSnapshot(t *testing.T) {} +func TestStatusRejectsNonGET(t *testing.T) {} +func TestHealthRejectsNonGET(t *testing.T) {} +``` + +Assert `/status` returns `503` when `StaticProvider{OK:false}`, `/health` returns JSON status `starting`, and POST requests return `405`. + +- [x] **Step 2: Verify red or existing green** + +Run: `go test ./internal/httpapi -run 'Test(StatusReturnsServiceUnavailableBeforeSnapshot|HealthReturnsStartingBeforeSnapshot|StatusRejectsNonGET|HealthRejectsNonGET)'` + +Expected: PASS if behavior already exists, otherwise FAIL with the missing behavior. + +- [x] **Step 3: Implement only missing behavior** + +Keep handler behavior cache-only. Do not add live collection. + +- [x] **Step 4: Verify** + +Run: `go test ./internal/httpapi` + +Expected: PASS. + +## Task 3: Reporter Retry And Cancellation + +**Files:** +- Modify: `internal/reporter/reporter.go` +- Modify: `internal/reporter/reporter_test.go` + +- [x] **Step 1: Write failing tests** + +Add tests named: + +```go +func TestReportReturnsErrorForHTTPFailure(t *testing.T) {} +func TestRunRetriesFailedReportOnce(t *testing.T) {} +func TestRunStopsDuringRetryDelayWhenContextCanceled(t *testing.T) {} +``` + +Use `httptest.Server` and a short interval/delay. Count requests with an atomic counter. First request returns `500`, second returns `202`. + +- [x] **Step 2: Verify red** + +Run: `go test ./internal/reporter -run 'Test(ReportReturnsErrorForHTTPFailure|RunRetriesFailedReportOnce|RunStopsDuringRetryDelayWhenContextCanceled)'` + +Expected: retry tests FAIL because `Run` does not retry yet. + +- [x] **Step 3: Implement minimal retry** + +Add a retry delay parameter through an internal helper while preserving the public `Run` signature: + +```go +const defaultRetryDelay = 2 * time.Second + +func Run(ctx context.Context, provider Provider, client *Client, interval time.Duration, logger *slog.Logger) { + runWithRetryDelay(ctx, provider, client, interval, defaultRetryDelay, logger) +} +``` + +In `runWithRetryDelay`, after a failed report, wait for `retryDelay` or context cancellation, then retry once. + +- [x] **Step 4: Verify** + +Run: `go test ./internal/reporter` + +Expected: PASS. + +## Task 4: Linux Collector Parser Coverage + +**Files:** +- Create: `internal/collector/collector_linux_test.go` +- Modify: `internal/collector/collector_linux.go` + +- [x] **Step 1: Write parser tests** + +Add tests named: + +```go +func TestParseAddressIPv4(t *testing.T) {} +func TestParseAddressIPv6(t *testing.T) {} +func TestParseProcessStatStartTime(t *testing.T) {} +func TestCPUUsageFromProcStatSamples(t *testing.T) {} +``` + +Use sample strings from `/proc/net/tcp`, `/proc/net/tcp6`, `/proc//stat`, and `/proc/stat`. Expected IPv4 sample `0100007F:1F90` maps to `127.0.0.1:8080`. + +- [x] **Step 2: Verify red** + +Run: `go test ./internal/collector -run 'Test(ParseAddress|ParseProcessStatStartTime|CPUUsageFromProcStatSamples)'` + +Expected: new process stat and CPU usage parser tests FAIL until helpers exist. + +- [x] **Step 3: Implement parser helpers** + +Add helpers: + +```go +func parseProcessStatStartTime(raw string) (int64, error) +func parseCPUTimes(raw string) (idle uint64, total uint64, err error) +func cpuUsagePercent(prevIdle, prevTotal, nextIdle, nextTotal uint64) float64 +``` + +Keep helpers unexported and deterministic. + +- [x] **Step 4: Verify** + +Run: `go test ./internal/collector` + +Expected: PASS. + +## Task 5: Linux Collection Fields + +**Files:** +- Modify: `internal/collector/collector_linux.go` +- Modify: `internal/collector/collector_linux_test.go` + +- [x] **Step 1: Write failing tests** + +Add tests for reading process start time from a temporary stat file through a helper: + +```go +func TestReadProcessStatReturnsStartTime(t *testing.T) {} +``` + +Add a CPU usage test around the helper rather than sleeping on real `/proc`. + +- [x] **Step 2: Verify red** + +Run: `go test ./internal/collector -run 'Test(ReadProcessStatReturnsStartTime|CPUUsage)'` + +Expected: FAIL until `readProcess` and `collectResource` use the helpers. + +- [x] **Step 3: Implement collection** + +Update `readProcess` to set `StartTime`. Update `collectResource` to set `CPU.UsagePercent` through a bounded sample that respects context: + +```go +select { +case <-ctx.Done(): + return ctx.Err() +case <-time.After(100 * time.Millisecond): +} +``` + +- [x] **Step 4: Verify** + +Run: `go test ./internal/collector` + +Expected: PASS. + +## Task 6: TCP Listener Partial Mapping + +**Files:** +- Modify: `internal/collector/collector_linux.go` +- Modify: `internal/collector/collector_linux_test.go` + +- [x] **Step 1: Write failing test** + +Add: + +```go +func TestCollectSocketsMarksPartialWhenProcessFDsUnreadable(t *testing.T) {} +``` + +Use a helper seam around socket inode walking so the test can simulate an unreadable fd directory. + +- [x] **Step 2: Verify red** + +Run: `go test ./internal/collector -run TestCollectSocketsMarksPartialWhenProcessFDsUnreadable` + +Expected: FAIL because unreadable fd directories are currently ignored. + +- [x] **Step 3: Implement partial error propagation** + +Make inode collection return `(map[string]status.ProcessInfo, error)`. Keep best-effort mappings, but return an error count when pid fd directories cannot be read. `collectSockets` sets `tcp_listeners.status` to `partial` and returns the error while preserving listener items. + +- [x] **Step 4: Verify** + +Run: `go test ./internal/collector` + +Expected: PASS. + +## Task 7: Linux Service Install Guard And Unit Rendering + +**Files:** +- Modify: `cmd/agent/main.go` +- Modify: `internal/service/service.go` +- Modify: `internal/service/service_linux.go` +- Create: `internal/service/service_linux_test.go` + +- [x] **Step 1: Write failing tests** + +Add: + +```go +func TestRenderSystemdUnitIncludesConfigPath(t *testing.T) {} +func TestLinuxInstallRequiresRoot(t *testing.T) {} +``` + +Assert the rendered unit includes `ExecStart=/opt/agent run -config /etc/mfd-agent.json` when a config path is supplied. + +- [x] **Step 2: Verify red** + +Run: `go test ./internal/service -run 'Test(RenderSystemdUnitIncludesConfigPath|LinuxInstallRequiresRoot)'` + +Expected: FAIL because config path and root guard seams do not exist. + +- [x] **Step 3: Implement install options** + +Add a small options struct or extend `Install` input so `cmd/agent` can pass the config path. Add root guard through an injectable `currentUID` function and render the unit through a pure helper: + +```go +func renderSystemdUnit(executable string, configPath string, cfg config.Config) string +``` + +- [x] **Step 4: Verify** + +Run: `go test ./internal/service` + +Expected: PASS. + +## Task 8: Snapshot Timeout And Partial Tests + +**Files:** +- Modify: `internal/snapshot/snapshot_test.go` +- Modify: `internal/snapshot/snapshot.go` + +- [x] **Step 1: Write tests** + +Add: + +```go +func TestRunCollectsImmediately(t *testing.T) {} +func TestCollectWithTimeoutMarksCanceledCollectorPartial(t *testing.T) {} +``` + +Use a collector that waits on `ctx.Done()` and returns `ctx.Err()`. + +- [x] **Step 2: Verify** + +Run: `go test ./internal/snapshot` + +Expected: PASS if current timeout behavior already satisfies this; otherwise FAIL and then implement the minimal fix. + +## Task 9: Windows Build-Tagged Service Contracts + +**Files:** +- Modify: `internal/service/service_windows.go` +- Optional create: `internal/service/service_windows_test.go` + +- [x] **Step 1: Add comments and compile-safe structure** + +Ensure Windows service code is build-tagged, documented, and does not reference Linux-only APIs. + +- [x] **Step 2: Compile-check Windows** + +Run: `GOOS=windows GOARCH=amd64 go test ./...` + +Expected: PASS. If network/dependency constraints block a wrapper dependency, document the exact failure and keep Windows behavior explicit. + +## Task 10: Final Verification And Audit + +**Files:** +- Inspect: `AGENTS.md` +- Inspect: all modified Go files + +- [x] **Step 1: Run full tests** + +Run: `go test ./...` + +Expected: PASS. + +- [x] **Step 2: Run Windows compile-check** + +Run: `GOOS=windows GOARCH=amd64 go test ./...` + +Expected: PASS or documented blocker. + +- [x] **Step 3: Audit comments** + +Run: `grep -R "^func \\|^type " -n cmd internal` + +Expected: every listed package-level function/type has a preceding useful comment. + +- [x] **Step 4: Completion checklist** + +Map each explicit AGENTS.md requirement to implementation evidence, test evidence, or a documented first-version limitation. + +## Git Note + +This directory is not currently a Git repository, so plan steps cannot include real commits. Use the test checkpoints above as completion gates. + diff --git a/docs/superpowers/specs/2026-05-18-mfd-monitor-agent-completion-design.md b/docs/superpowers/specs/2026-05-18-mfd-monitor-agent-completion-design.md new file mode 100644 index 0000000..ef4cf46 --- /dev/null +++ b/docs/superpowers/specs/2026-05-18-mfd-monitor-agent-completion-design.md @@ -0,0 +1,77 @@ +# MFD Monitor Agent Completion Design + +## Objective + +Bring the current Go single-binary host monitoring agent into alignment with `AGENTS.md`: complete missing host status behavior, preserve cached snapshots, support Linux and Windows service lifecycle, add reporter retry/backoff, improve partial-success handling, and raise comments/tests to the repository standard. + +## Current Gaps + +- Package-level Go types, functions, methods, and interfaces are mostly undocumented. +- Linux collection misses CPU usage percent, process start time, and process CPU percent. +- TCP listener mapping does not clearly report permission-related partial mappings. +- Windows collector and service runtime are incomplete; current service install only shells out to `sc.exe`. +- Linux service install does not explicitly require root and does not preserve `-config` in the systemd unit. +- Reporter logs failures but has no retry or bounded backoff. +- Tests do not cover enough collector parsing, partial/error statuses, service guards, reporter retry, or HTTP edge cases. + +## Architecture + +Keep the existing single-process architecture and package boundaries: + +- `cmd/agent`: CLI and process lifecycle. +- `config`: defaults, JSON file loading, environment overrides. +- `collector`: platform collection into `status.HostStatus`. +- `snapshot`: collector aggregation and in-memory cache. +- `httpapi`: `GET /health` and `GET /status`, cache reads only. +- `reporter`: periodic POST of the cached `HostStatus`. +- `service`: Windows/Linux service install, uninstall, start, and stop. + +No historical storage, pagination, auth, UDP collection, thread lists, process handles, or trend analysis are included. + +## Behavior Design + +Collection remains immediate-on-start and periodic afterward. HTTP and reporter paths both read the same cached snapshot and never trigger a live scan. + +Each collector writes only its own section. If one collector fails, `snapshot.Cache` records a `StatusError`, marks `agent.snapshot_status` as `partial`, and keeps successful sections. + +Linux should be the first complete collection implementation: + +- CPU usage percent from `/proc/stat` with a bounded sample and context cancellation. +- Process start time from `/proc//stat`. +- Process CPU percent best-effort; if exact calculation is unavailable, keep the field stable and do not fail the process collector. +- TCP listeners from `/proc/net/tcp`, `/proc/net/tcp6`, and `/proc//fd`; unreadable process fd data should degrade listener status to `partial`. + +Windows should expose a real service runtime path without NSSM. If full Windows metric collection would require a larger dependency decision, platform files should keep unsupported subsections explicit with `partial`/`error` statuses and tests around those contracts. + +## Service Design + +Linux install should require root before writing `/etc/systemd/system`, render a systemd unit for the configured service name, include `agent run`, preserve `-config ` when supplied, run `daemon-reload`, and enable the service. Errors should include the failed operation. + +Windows install/start/stop/uninstall should remain behind Windows build tags and avoid NSSM. Administrator requirements should be checked where practical. + +## Reporter Design + +Reporter POST payload remains exactly the `HostStatus` JSON used by `/status`. Failed reports are logged and retried once after a short delay or handled with a simple bounded backoff. Retry/backoff must respect context cancellation and never terminate the process or block the local HTTP server. + +## Comments And Tests + +All Go package-level types, structs, interfaces, functions, and methods need concise comments. Comments should explain behavior, side effects, units, platform assumptions, and permission-sensitive degradation. + +Tests should be written before behavior changes and cover: + +- Linux parser helpers for CPU, memory, process stat/status, and TCP listeners. +- Snapshot partial aggregation and timeout behavior. +- HTTP not-ready and method-not-allowed behavior. +- Reporter HTTP errors, retry/backoff, payload structure, and cancellation. +- Linux service unit rendering, root guard, and config path propagation. +- Windows build-tagged service contracts where compile-testable. + +## Verification + +Completion requires: + +- `go test ./...` passes. +- Current Linux build passes. +- `GOOS=windows GOARCH=amd64 go test ./...` is attempted and either passes or any blocker is documented with the exact failure. +- A final audit maps each `AGENTS.md` requirement touched by this pass to code, tests, or a documented remaining platform limitation. + diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..a260ba9 --- /dev/null +++ b/go.mod @@ -0,0 +1,3 @@ +module mfd-monitor-agent + +go 1.22 diff --git a/internal/collector/collector.go b/internal/collector/collector.go new file mode 100644 index 0000000..3f58fd4 --- /dev/null +++ b/internal/collector/collector.go @@ -0,0 +1,37 @@ +package collector + +import ( + "context" + + "mfd-monitor-agent/internal/snapshot" + "mfd-monitor-agent/internal/status" +) + +// namedCollector binds a stable collector name to a collection function. +type namedCollector struct { + name string + fn func(context.Context, *status.HostStatus) error +} + +// Name returns the collector name used in snapshot error records. +func (c namedCollector) Name() string { + return c.name +} + +// Collect runs the wrapped collection function. +func (c namedCollector) Collect(ctx context.Context, out *status.HostStatus) error { + return c.fn(ctx, out) +} + +// New returns the default collector set for the current platform. +func New(hostID string) []snapshot.Collector { + return []snapshot.Collector{ + namedCollector{name: "system", fn: func(ctx context.Context, out *status.HostStatus) error { + return collectSystem(ctx, hostID, out) + }}, + namedCollector{name: "network", fn: collectNetwork}, + namedCollector{name: "resource", fn: collectResource}, + namedCollector{name: "process", fn: collectProcesses}, + namedCollector{name: "socket", fn: collectSockets}, + } +} diff --git a/internal/collector/collector_linux.go b/internal/collector/collector_linux.go new file mode 100644 index 0000000..de3b8f1 --- /dev/null +++ b/internal/collector/collector_linux.go @@ -0,0 +1,583 @@ +//go:build linux + +package collector + +import ( + "bufio" + "context" + "encoding/hex" + "errors" + "fmt" + "io/fs" + "net" + "os" + "os/user" + "path/filepath" + "runtime" + "sort" + "strconv" + "strings" + "syscall" + "time" + + "mfd-monitor-agent/internal/status" +) + +var socketInodesFunc = socketInodes +var readTCPListenersFunc = readTCPListeners + +// collectSystem fills Linux host identity, OS, architecture, and uptime fields. +func collectSystem(_ context.Context, hostID string, out *status.HostStatus) error { + hostname, err := os.Hostname() + if err != nil { + return err + } + out.Host.HostID = hostID + out.Host.Hostname = hostname + out.Host.Platform = runtime.GOOS + out.Host.PlatformVersion = readOSRelease() + out.Host.Arch = runtime.GOARCH + out.Host.UptimeSeconds = readUptime() + if out.Host.HostID == "" { + out.Host.HostID = hostname + } + return nil +} + +// collectNetwork fills Linux network interface, address, and MAC data. +func collectNetwork(_ context.Context, out *status.HostStatus) error { + ifaces, err := net.Interfaces() + if err != nil { + return err + } + ipSet := map[string]struct{}{} + macSet := map[string]struct{}{} + for _, iface := range ifaces { + addrs, _ := iface.Addrs() + item := status.NetworkInterface{ + Name: iface.Name, + MAC: iface.HardwareAddr.String(), + MTU: iface.MTU, + Flags: strings.Fields(iface.Flags.String()), + } + for _, addr := range addrs { + value := addr.String() + item.Addresses = append(item.Addresses, value) + ipSet[value] = struct{}{} + } + if item.MAC != "" { + macSet[item.MAC] = struct{}{} + } + out.Network.Interfaces = append(out.Network.Interfaces, item) + } + out.Network.IPv4s, out.Network.IPv6s = networkAddressSets(ipSet) + out.Network.MACs = sortedKeys(macSet) + return nil +} + +// collectResource fills CPU, memory, and disk resource data from Linux proc files. +func collectResource(ctx context.Context, out *status.HostStatus) error { + out.CPU.LogicalCores = runtime.NumCPU() + out.CPU.PhysicalCores, out.CPU.ModelName = readCPUInfo() + out.CPU.LoadAvg = readLoadAvg() + out.CPU.UsagePercent = readCPUUsagePercent(ctx) + out.Memory = readMemory() + disks, err := readDisks() + if err != nil { + return err + } + out.Disks = disks + return nil +} + +// collectProcesses reads /proc process directories and degrades to partial when some entries fail. +func collectProcesses(_ context.Context, out *status.HostStatus) error { + entries, err := os.ReadDir("/proc") + if err != nil { + out.Processes.Status = "error" + return err + } + var errs []error + for _, entry := range entries { + if !entry.IsDir() { + continue + } + pid, err := strconv.Atoi(entry.Name()) + if err != nil { + continue + } + process, err := readProcess(pid) + if err != nil { + errs = append(errs, err) + continue + } + out.Processes.Items = append(out.Processes.Items, process) + } + sort.Slice(out.Processes.Items, func(i, j int) bool { + return out.Processes.Items[i].PID < out.Processes.Items[j].PID + }) + if len(errs) > 0 { + out.Processes.Status = "partial" + return fmt.Errorf("%d process entries could not be read", len(errs)) + } + out.Processes.Status = "ok" + return nil +} + +// collectSockets reads Linux TCP listener tables and maps socket inodes to owning processes. +func collectSockets(_ context.Context, out *status.HostStatus) error { + inodes, inodeErr := socketInodesFunc() + var listeners []status.TCPListenerInfo + var errs []error + if inodeErr != nil { + errs = append(errs, inodeErr) + } + for _, file := range []struct { + path string + family string + }{ + {path: "/proc/net/tcp", family: "ipv4"}, + {path: "/proc/net/tcp6", family: "ipv6"}, + } { + items, err := readTCPListenersFunc(file.path, file.family, inodes) + if err != nil { + errs = append(errs, err) + continue + } + listeners = append(listeners, items...) + } + sort.Slice(listeners, func(i, j int) bool { + if listeners[i].IP == listeners[j].IP { + return listeners[i].Port < listeners[j].Port + } + return listeners[i].IP < listeners[j].IP + }) + out.TCPListeners.Items = listeners + if len(errs) > 0 { + out.TCPListeners.Status = "partial" + return errors.Join(errs...) + } + out.TCPListeners.Status = "ok" + return nil +} + +// readOSRelease returns the Linux PRETTY_NAME value when /etc/os-release is readable. +func readOSRelease() string { + file, err := os.Open("/etc/os-release") + if err != nil { + return "" + } + defer file.Close() + scanner := bufio.NewScanner(file) + for scanner.Scan() { + line := scanner.Text() + if strings.HasPrefix(line, "PRETTY_NAME=") { + return strings.Trim(strings.TrimPrefix(line, "PRETTY_NAME="), `"`) + } + } + return "" +} + +// readUptime returns host uptime in seconds from /proc/uptime. +func readUptime() uint64 { + data, err := os.ReadFile("/proc/uptime") + if err != nil { + return 0 + } + fields := strings.Fields(string(data)) + if len(fields) == 0 { + return 0 + } + value, _ := strconv.ParseFloat(fields[0], 64) + return uint64(value) +} + +// readCPUInfo returns physical CPU package count and the first model name from /proc/cpuinfo. +func readCPUInfo() (int, string) { + file, err := os.Open("/proc/cpuinfo") + if err != nil { + return 0, "" + } + defer file.Close() + physical := map[string]struct{}{} + model := "" + scanner := bufio.NewScanner(file) + for scanner.Scan() { + line := scanner.Text() + parts := strings.SplitN(line, ":", 2) + if len(parts) != 2 { + continue + } + key := strings.TrimSpace(parts[0]) + value := strings.TrimSpace(parts[1]) + switch key { + case "physical id": + physical[value] = struct{}{} + case "model name": + if model == "" { + model = value + } + } + } + return len(physical), model +} + +// readLoadAvg returns the 1, 5, and 15 minute Linux load averages when available. +func readLoadAvg() []float64 { + data, err := os.ReadFile("/proc/loadavg") + if err != nil { + return nil + } + fields := strings.Fields(string(data)) + var values []float64 + for i := 0; i < 3 && i < len(fields); i++ { + value, err := strconv.ParseFloat(fields[i], 64) + if err == nil { + values = append(values, value) + } + } + return values +} + +// readMemory returns Linux memory counters in bytes from /proc/meminfo. +func readMemory() status.MemoryInfo { + values := map[string]uint64{} + file, err := os.Open("/proc/meminfo") + if err != nil { + return status.MemoryInfo{} + } + defer file.Close() + scanner := bufio.NewScanner(file) + for scanner.Scan() { + fields := strings.Fields(scanner.Text()) + if len(fields) < 2 { + continue + } + value, _ := strconv.ParseUint(fields[1], 10, 64) + // /proc/meminfo reports memory in KiB; the API exposes bytes and + // includes memory.unit so clients can render the values explicitly. + values[strings.TrimSuffix(fields[0], ":")] = value * 1024 + } + total := values["MemTotal"] + free := values["MemFree"] + available := values["MemAvailable"] + used := total - available + percent := 0.0 + if total > 0 { + percent = float64(used) / float64(total) * 100 + } + return status.MemoryInfo{Unit: "byte", Total: total, Used: used, Free: free, Available: available, UsedPercent: percent} +} + +// readDisks returns physical and network mounted filesystem usage from /proc/mounts and statfs. +func readDisks() ([]status.DiskInfo, error) { + file, err := os.Open("/proc/mounts") + if err != nil { + return nil, err + } + defer file.Close() + var disks []status.DiskInfo + scanner := bufio.NewScanner(file) + for scanner.Scan() { + fields := strings.Fields(scanner.Text()) + if len(fields) < 3 { + continue + } + var stat syscall.Statfs_t + if err := syscall.Statfs(fields[1], &stat); err != nil { + continue + } + total := stat.Blocks * uint64(stat.Bsize) + if !shouldIncludeDiskMount(fields[0], fields[2], total) { + continue + } + free := stat.Bavail * uint64(stat.Bsize) + used := total - free + percent := 0.0 + if total > 0 { + percent = float64(used) / float64(total) * 100 + } + disks = append(disks, status.DiskInfo{ + Device: fields[0], MountPoint: fields[1], FSType: fields[2], + Unit: "byte", Total: total, Used: used, Free: free, UsedPercent: percent, + }) + } + return disks, scanner.Err() +} + +// shouldIncludeDiskMount reports whether a mount represents a user-relevant disk. +func shouldIncludeDiskMount(device string, fsType string, total uint64) bool { + if total == 0 { + return false + } + networkFSTypes := map[string]bool{ + "nfs": true, "nfs4": true, "cifs": true, "smb3": true, "sshfs": true, + } + if networkFSTypes[fsType] { + return true + } + return strings.HasPrefix(device, "/dev/") +} + +// readProcess returns one Linux process entry from /proc/. +func readProcess(pid int) (status.ProcessInfo, error) { + base := filepath.Join("/proc", strconv.Itoa(pid)) + statusValues, err := readProcessStatus(filepath.Join(base, "status")) + if err != nil { + return status.ProcessInfo{}, err + } + ppid, _ := strconv.Atoi(statusValues["PPid"]) + rss := parseKB(statusValues["VmRSS"]) + vms := parseKB(statusValues["VmSize"]) + username := "" + if uid := firstField(statusValues["Uid"]); uid != "" { + if u, err := user.LookupId(uid); err == nil { + username = u.Username + } + } + cmdline := "" + if data, err := os.ReadFile(filepath.Join(base, "cmdline")); err == nil { + cmdline = strings.Trim(strings.ReplaceAll(string(data), "\x00", " "), " ") + } + exe, _ := os.Readlink(filepath.Join(base, "exe")) + statValues, _ := readProcessStat(filepath.Join(base, "stat")) + return status.ProcessInfo{ + PID: pid, PPID: ppid, Name: statusValues["Name"], Exe: exe, Cmdline: cmdline, + Username: username, Status: statusValues["State"], StartTime: statValues.StartTime, + MemoryRSS: rss, MemoryVMS: vms, + }, nil +} + +// processStat contains process counters parsed from /proc//stat. +type processStat struct { + StartTime int64 +} + +// readProcessStat parses Linux process stat counters from path. +func readProcessStat(path string) (processStat, error) { + data, err := os.ReadFile(path) + if err != nil { + return processStat{}, err + } + startTime, err := parseProcessStatStartTime(string(data)) + if err != nil { + return processStat{}, err + } + return processStat{StartTime: startTime}, nil +} + +// readProcessStatus parses key-value fields from /proc//status. +func readProcessStatus(path string) (map[string]string, error) { + file, err := os.Open(path) + if err != nil { + return nil, err + } + defer file.Close() + values := map[string]string{} + scanner := bufio.NewScanner(file) + for scanner.Scan() { + parts := strings.SplitN(scanner.Text(), ":", 2) + if len(parts) == 2 { + values[parts[0]] = strings.TrimSpace(parts[1]) + } + } + return values, scanner.Err() +} + +// socketInodes maps Linux socket inode numbers to best-effort owning process data. +func socketInodes() (map[string]status.ProcessInfo, error) { + out := map[string]status.ProcessInfo{} + var readErrors int + _ = filepath.WalkDir("/proc", func(path string, d fs.DirEntry, err error) error { + if err != nil || !d.IsDir() || filepath.Dir(path) != "/proc" { + return nil + } + pid, err := strconv.Atoi(filepath.Base(path)) + if err != nil { + return nil + } + process, _ := readProcess(pid) + fdDir := filepath.Join(path, "fd") + entries, err := os.ReadDir(fdDir) + if err != nil { + readErrors++ + return nil + } + for _, entry := range entries { + target, err := os.Readlink(filepath.Join(fdDir, entry.Name())) + if err != nil || !strings.HasPrefix(target, "socket:[") { + continue + } + inode := strings.TrimSuffix(strings.TrimPrefix(target, "socket:["), "]") + out[inode] = process + } + return nil + }) + if readErrors > 0 { + return out, fmt.Errorf("%d process fd directories could not be read", readErrors) + } + return out, nil +} + +// readTCPListeners parses a Linux TCP table and returns sockets in LISTEN state. +func readTCPListeners(path string, family string, inodes map[string]status.ProcessInfo) ([]status.TCPListenerInfo, error) { + file, err := os.Open(path) + if err != nil { + return nil, err + } + defer file.Close() + var items []status.TCPListenerInfo + scanner := bufio.NewScanner(file) + if scanner.Scan() { + // Skip header. + } + for scanner.Scan() { + fields := strings.Fields(scanner.Text()) + if len(fields) < 10 || fields[3] != "0A" { + continue + } + ip, port, err := parseAddress(fields[1], family) + if err != nil { + continue + } + process := inodes[fields[9]] + items = append(items, status.TCPListenerInfo{ + IP: ip, Port: port, PID: process.PID, ProcessName: process.Name, + ProcessExe: process.Exe, Family: family, + }) + } + return items, scanner.Err() +} + +// readCPUUsagePercent samples /proc/stat over a short interval and returns best-effort usage. +func readCPUUsagePercent(ctx context.Context) float64 { + prevData, err := os.ReadFile("/proc/stat") + if err != nil { + return 0 + } + prevIdle, prevTotal, err := parseCPUTimes(string(prevData)) + if err != nil { + return 0 + } + select { + case <-ctx.Done(): + return 0 + case <-time.After(100 * time.Millisecond): + } + nextData, err := os.ReadFile("/proc/stat") + if err != nil { + return 0 + } + nextIdle, nextTotal, err := parseCPUTimes(string(nextData)) + if err != nil { + return 0 + } + return cpuUsagePercent(prevIdle, prevTotal, nextIdle, nextTotal) +} + +// parseAddress decodes a Linux /proc/net/tcp address into IP and port values. +func parseAddress(raw string, family string) (string, int, error) { + parts := strings.Split(raw, ":") + if len(parts) != 2 { + return "", 0, fmt.Errorf("invalid address %q", raw) + } + port64, err := strconv.ParseInt(parts[1], 16, 32) + if err != nil { + return "", 0, err + } + bytes, err := hex.DecodeString(parts[0]) + if err != nil { + return "", 0, err + } + if family == "ipv4" && len(bytes) == 4 { + return net.IPv4(bytes[3], bytes[2], bytes[1], bytes[0]).String(), int(port64), nil + } + if family == "ipv6" && len(bytes) == 16 { + reordered := make([]byte, 16) + for i := 0; i < 16; i += 4 { + reordered[i], reordered[i+1], reordered[i+2], reordered[i+3] = bytes[i+3], bytes[i+2], bytes[i+1], bytes[i] + } + return net.IP(reordered).String(), int(port64), nil + } + return "", 0, fmt.Errorf("invalid %s address", family) +} + +// parseProcessStatStartTime extracts field 22 from Linux /proc//stat content. +func parseProcessStatStartTime(raw string) (int64, error) { + end := strings.LastIndex(raw, ")") + if end < 0 || end+2 >= len(raw) { + return 0, fmt.Errorf("invalid process stat") + } + fields := strings.Fields(raw[end+2:]) + if len(fields) < 20 { + return 0, fmt.Errorf("process stat has %d fields after comm", len(fields)) + } + return strconv.ParseInt(fields[19], 10, 64) +} + +// parseCPUTimes extracts idle and total jiffy counters from the first cpu line in /proc/stat. +func parseCPUTimes(raw string) (idle uint64, total uint64, err error) { + scanner := bufio.NewScanner(strings.NewReader(raw)) + for scanner.Scan() { + fields := strings.Fields(scanner.Text()) + if len(fields) == 0 || fields[0] != "cpu" { + continue + } + if len(fields) < 5 { + return 0, 0, fmt.Errorf("cpu line has %d fields", len(fields)) + } + for i, field := range fields[1:] { + value, err := strconv.ParseUint(field, 10, 64) + if err != nil { + return 0, 0, err + } + total += value + if i == 3 || i == 4 { + idle += value + } + } + return idle, total, nil + } + if err := scanner.Err(); err != nil { + return 0, 0, err + } + return 0, 0, fmt.Errorf("cpu line not found") +} + +// cpuUsagePercent calculates non-idle CPU percentage between two /proc/stat samples. +func cpuUsagePercent(prevIdle, prevTotal, nextIdle, nextTotal uint64) float64 { + totalDelta := nextTotal - prevTotal + if totalDelta == 0 { + return 0 + } + idleDelta := nextIdle - prevIdle + if idleDelta >= totalDelta { + return 0 + } + return float64(totalDelta-idleDelta) / float64(totalDelta) * 100 +} + +// parseKB converts the first numeric kilobyte value in raw to bytes. +func parseKB(raw string) uint64 { + value, _ := strconv.ParseUint(firstField(raw), 10, 64) + return value * 1024 +} + +// firstField returns the first whitespace-separated token in raw. +func firstField(raw string) string { + fields := strings.Fields(raw) + if len(fields) == 0 { + return "" + } + return fields[0] +} + +// sortedKeys returns deterministic sorted keys from a string set. +func sortedKeys(values map[string]struct{}) []string { + keys := make([]string, 0, len(values)) + for key := range values { + keys = append(keys, key) + } + sort.Strings(keys) + return keys +} diff --git a/internal/collector/collector_linux_test.go b/internal/collector/collector_linux_test.go new file mode 100644 index 0000000..9dc853e --- /dev/null +++ b/internal/collector/collector_linux_test.go @@ -0,0 +1,169 @@ +//go:build linux + +package collector + +import ( + "context" + "fmt" + "os" + "path/filepath" + "strconv" + "strings" + "testing" + + "mfd-monitor-agent/internal/status" +) + +// TestParseAddressIPv4 verifies Linux IPv4 socket addresses are decoded. +func TestParseAddressIPv4(t *testing.T) { + ip, port, err := parseAddress("0100007F:1F90", "ipv4") + if err != nil { + t.Fatal(err) + } + if ip != "127.0.0.1" || port != 8080 { + t.Fatalf("address = %s:%d", ip, port) + } +} + +// TestParseAddressIPv6 verifies Linux IPv6 socket addresses are decoded. +func TestParseAddressIPv6(t *testing.T) { + ip, port, err := parseAddress("00000000000000000000000001000000:1F90", "ipv6") + if err != nil { + t.Fatal(err) + } + if ip != "::1" || port != 8080 { + t.Fatalf("address = %s:%d", ip, port) + } +} + +// TestParseProcessStatStartTime verifies field 22 parsing from /proc//stat. +func TestParseProcessStatStartTime(t *testing.T) { + raw := "1234 (agent worker) S 1 2 3 4 5 0 0 0 0 0 11 22 0 0 20 0 1 0 987654 0 0" + + startTime, err := parseProcessStatStartTime(raw) + if err != nil { + t.Fatal(err) + } + if startTime != 987654 { + t.Fatalf("start time = %d", startTime) + } +} + +// TestCPUUsageFromProcStatSamples verifies CPU usage from two /proc/stat samples. +func TestCPUUsageFromProcStatSamples(t *testing.T) { + prevIdle, prevTotal, err := parseCPUTimes("cpu 100 0 100 800 0 0 0 0 0 0\n") + if err != nil { + t.Fatal(err) + } + nextIdle, nextTotal, err := parseCPUTimes("cpu 150 0 150 900 0 0 0 0 0 0\n") + if err != nil { + t.Fatal(err) + } + + got := cpuUsagePercent(prevIdle, prevTotal, nextIdle, nextTotal) + if got != 50 { + t.Fatalf("usage percent = %f", got) + } +} + +// TestReadProcessStatReturnsStartTime verifies process start time is exposed. +func TestReadProcessStatReturnsStartTime(t *testing.T) { + path := filepath.Join(t.TempDir(), "stat") + raw := "1234 (agent worker) S 1 2 3 4 5 0 0 0 0 0 11 22 0 0 20 0 1 0 987654 0 0" + if err := os.WriteFile(path, []byte(raw), 0o600); err != nil { + t.Fatal(err) + } + + got, err := readProcessStat(path) + if err != nil { + t.Fatal(err) + } + if got.StartTime != 987654 { + t.Fatalf("start time = %d", got.StartTime) + } +} + +// TestReadMemoryReturnsBytesWithUnit verifies memory capacity fields are exposed in bytes. +func TestReadMemoryReturnsBytesWithUnit(t *testing.T) { + meminfo, err := os.ReadFile("/proc/meminfo") + if err != nil { + t.Fatal(err) + } + var totalKiB uint64 + for _, line := range strings.Split(string(meminfo), "\n") { + fields := strings.Fields(line) + if len(fields) >= 2 && fields[0] == "MemTotal:" { + totalKiB, err = strconv.ParseUint(fields[1], 10, 64) + if err != nil { + t.Fatal(err) + } + break + } + } + if totalKiB == 0 { + t.Fatal("MemTotal not found") + } + + got := readMemory() + + if got.Total != totalKiB*1024 { + t.Fatalf("total memory = %d bytes, want %d bytes", got.Total, totalKiB*1024) + } + if got.Unit != "byte" { + t.Fatalf("unit = %q, want byte", got.Unit) + } +} + +// TestShouldIncludeDiskMount keeps physical and network disks while dropping pseudo mounts. +func TestShouldIncludeDiskMount(t *testing.T) { + tests := []struct { + name string + device string + fsType string + total uint64 + want bool + }{ + {name: "physical block device", device: "/dev/mapper/pve-vm--103--disk--1", fsType: "ext4", total: 1024, want: true}, + {name: "nfs mount", device: "server:/export/data", fsType: "nfs4", total: 1024, want: true}, + {name: "lxcfs pseudo mount", device: "lxcfs", fsType: "fuse.lxcfs", total: 0, want: false}, + {name: "proc pseudo mount", device: "proc", fsType: "proc", total: 0, want: false}, + {name: "tmpfs memory mount", device: "tmpfs", fsType: "tmpfs", total: 1024, want: false}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + got := shouldIncludeDiskMount(test.device, test.fsType, test.total) + if got != test.want { + t.Fatalf("include = %t, want %t", got, test.want) + } + }) + } +} + +// TestCollectSocketsMarksPartialWhenProcessFDsUnreadable verifies socket mapping degradation. +func TestCollectSocketsMarksPartialWhenProcessFDsUnreadable(t *testing.T) { + oldSocketInodes := socketInodesFunc + oldReadTCPListeners := readTCPListenersFunc + t.Cleanup(func() { + socketInodesFunc = oldSocketInodes + readTCPListenersFunc = oldReadTCPListeners + }) + socketInodesFunc = func() (map[string]status.ProcessInfo, error) { + return map[string]status.ProcessInfo{}, fmt.Errorf("1 process fd directory could not be read") + } + readTCPListenersFunc = func(path string, family string, inodes map[string]status.ProcessInfo) ([]status.TCPListenerInfo, error) { + return []status.TCPListenerInfo{{IP: "127.0.0.1", Port: 8080, Family: family}}, nil + } + + var got status.HostStatus + err := collectSockets(context.Background(), &got) + if err == nil { + t.Fatal("expected partial mapping error") + } + if got.TCPListeners.Status != "partial" { + t.Fatalf("status = %q", got.TCPListeners.Status) + } + if len(got.TCPListeners.Items) != 2 { + t.Fatalf("listener count = %d", len(got.TCPListeners.Items)) + } +} diff --git a/internal/collector/collector_other.go b/internal/collector/collector_other.go new file mode 100644 index 0000000..a9aa5eb --- /dev/null +++ b/internal/collector/collector_other.go @@ -0,0 +1,75 @@ +//go:build !linux && !windows + +package collector + +import ( + "context" + "fmt" + "net" + "os" + "runtime" + "strings" + + "mfd-monitor-agent/internal/status" +) + +// collectSystem fills portable host identity fields for non-Linux platforms. +func collectSystem(_ context.Context, hostID string, out *status.HostStatus) error { + hostname, err := os.Hostname() + if err != nil { + return err + } + out.Host.HostID = hostID + if out.Host.HostID == "" { + out.Host.HostID = hostname + } + out.Host.Hostname = hostname + out.Host.Platform = runtime.GOOS + out.Host.Arch = runtime.GOARCH + return nil +} + +// collectNetwork fills interface, address, and MAC data using the portable net package. +func collectNetwork(_ context.Context, out *status.HostStatus) error { + ifaces, err := net.Interfaces() + if err != nil { + return err + } + seenIPs := map[string]struct{}{} + seenMACs := map[string]struct{}{} + for _, iface := range ifaces { + item := status.NetworkInterface{Name: iface.Name, MAC: iface.HardwareAddr.String(), MTU: iface.MTU, Flags: strings.Fields(iface.Flags.String())} + addrs, _ := iface.Addrs() + for _, addr := range addrs { + item.Addresses = append(item.Addresses, addr.String()) + seenIPs[addr.String()] = struct{}{} + } + if item.MAC != "" { + seenMACs[item.MAC] = struct{}{} + } + out.Network.Interfaces = append(out.Network.Interfaces, item) + } + out.Network.IPv4s, out.Network.IPv6s = networkAddressSets(seenIPs) + for mac := range seenMACs { + out.Network.MACs = append(out.Network.MACs, mac) + } + return nil +} + +// collectResource fills portable resource fields available without platform-specific APIs. +func collectResource(_ context.Context, out *status.HostStatus) error { + out.CPU.LogicalCores = runtime.NumCPU() + return nil +} + +// collectProcesses reports partial process collection until platform-specific support is added. +func collectProcesses(_ context.Context, out *status.HostStatus) error { + out.Processes.Status = "partial" + return fmt.Errorf("process collection is not implemented for %s without platform-specific support", runtime.GOOS) +} + +// collectSockets reports partial TCP listener collection until platform-specific support is added. +func collectSockets(_ context.Context, out *status.HostStatus) error { + out.TCPListeners.Status = "partial" + return fmt.Errorf("tcp listener process mapping is not implemented for %s without platform-specific support", runtime.GOOS) +} diff --git a/internal/collector/collector_windows.go b/internal/collector/collector_windows.go new file mode 100644 index 0000000..8f6e093 --- /dev/null +++ b/internal/collector/collector_windows.go @@ -0,0 +1,781 @@ +//go:build windows + +package collector + +import ( + "bytes" + "context" + "encoding/csv" + "errors" + "fmt" + "math/bits" + "net" + "os" + "os/exec" + "runtime" + "sort" + "strconv" + "strings" + "syscall" + "time" + "unsafe" + + "mfd-monitor-agent/internal/status" +) + +const ( + windowsAFInet = 2 + windowsAFInet6 = 23 + windowsNoError = 0 + windowsErrorInsufficientBuffer = 122 + windowsTCPTableOwnerPIDListen = 3 + windowsTCPStateListen = 2 + windowsDriveFixed = 3 + windowsDriveRemote = 4 + windowsProcessQueryInformation = 0x0400 + windowsProcessQueryLimitedInfo = 0x1000 + windowsProcessVMRead = 0x0010 + windowsTokenQuery = 0x0008 + windowsMaxPath = 260 + windowsRegSZ = 1 + windowsRelationProcessorCore = 0 +) + +var ( + windowsKernel32 = syscall.NewLazyDLL("kernel32.dll") + windowsGlobalMemoryStatusExProc = windowsKernel32.NewProc("GlobalMemoryStatusEx") + windowsGetTickCount64Proc = windowsKernel32.NewProc("GetTickCount64") + windowsGetLogicalDrivesProc = windowsKernel32.NewProc("GetLogicalDrives") + windowsGetDriveTypeWProc = windowsKernel32.NewProc("GetDriveTypeW") + windowsGetDiskFreeSpaceExWProc = windowsKernel32.NewProc("GetDiskFreeSpaceExW") + windowsGetSystemTimesProc = windowsKernel32.NewProc("GetSystemTimes") + windowsGetNativeSystemInfoProc = windowsKernel32.NewProc("GetNativeSystemInfo") + windowsGetLogicalProcessorInfoProc = windowsKernel32.NewProc("GetLogicalProcessorInformation") + windowsQueryFullProcessImageNameWProc = windowsKernel32.NewProc("QueryFullProcessImageNameW") + windowsIphlpapi = syscall.NewLazyDLL("iphlpapi.dll") + windowsGetExtendedTCPTableProc = windowsIphlpapi.NewProc("GetExtendedTcpTable") + windowsPsapi = syscall.NewLazyDLL("psapi.dll") + windowsGetProcessMemoryInfoProc = windowsPsapi.NewProc("GetProcessMemoryInfo") + windowsAdvapi32 = syscall.NewLazyDLL("advapi32.dll") + windowsOpenProcessTokenProc = windowsAdvapi32.NewProc("OpenProcessToken") + windowsGetTokenInformationProc = windowsAdvapi32.NewProc("GetTokenInformation") + windowsLookupAccountSidWProc = windowsAdvapi32.NewProc("LookupAccountSidW") +) + +// windowsMemoryStatus mirrors the Windows MEMORYSTATUSEX structure used by GlobalMemoryStatusEx. +type windowsMemoryStatus struct { + Length uint32 + MemoryLoad uint32 + TotalPhys uint64 + AvailPhys uint64 + TotalPageFile uint64 + AvailPageFile uint64 + TotalVirtual uint64 + AvailVirtual uint64 + AvailExtendedVirtual uint64 +} + +// windowsMIBTCPRowOwnerPID mirrors a MIB_TCPROW_OWNER_PID row returned by GetExtendedTcpTable. +type windowsMIBTCPRowOwnerPID struct { + State uint32 + LocalAddr uint32 + LocalPort uint32 + RemoteAddr uint32 + RemotePort uint32 + OwningPID uint32 +} + +// windowsMIBTCP6RowOwnerPID mirrors a MIB_TCP6ROW_OWNER_PID row returned by GetExtendedTcpTable. +type windowsMIBTCP6RowOwnerPID struct { + State uint32 + LocalAddr [16]byte + LocalScopeID uint32 + LocalPort uint32 + RemoteAddr [16]byte + RemoteScopeID uint32 + RemotePort uint32 + OwningPID uint32 +} + +// windowsSystemInfo mirrors the subset of SYSTEM_INFO needed for CPU package sizing. +type windowsSystemInfo struct { + ProcessorArchitecture uint16 + Reserved uint16 + PageSize uint32 + MinimumApplicationAddress uintptr + MaximumApplicationAddress uintptr + ActiveProcessorMask uintptr + NumberOfProcessors uint32 + ProcessorType uint32 + AllocationGranularity uint32 + ProcessorLevel uint16 + ProcessorRevision uint16 +} + +// windowsSystemTimes stores system CPU time counters as 100ns units. +type windowsSystemTimes struct { + Idle uint64 + Kernel uint64 + User uint64 +} + +// windowsLogicalProcessorInformation mirrors SYSTEM_LOGICAL_PROCESSOR_INFORMATION for core counting. +type windowsLogicalProcessorInformation struct { + ProcessorMask uintptr + Relationship uint32 + Data [16]byte +} + +// windowsProcessMemoryCounters mirrors PROCESS_MEMORY_COUNTERS. +type windowsProcessMemoryCounters struct { + CB uint32 + PageFaultCount uint32 + PeakWorkingSetSize uintptr + WorkingSetSize uintptr + QuotaPeakPagedPoolUsage uintptr + QuotaPagedPoolUsage uintptr + QuotaPeakNonPagedPoolUsage uintptr + QuotaNonPagedPoolUsage uintptr + PagefileUsage uintptr + PeakPagefileUsage uintptr +} + +// windowsProcessDetails contains best-effort per-process fields requiring extra handles or commands. +type windowsProcessDetails struct { + Exe string + Cmdline string + Username string + Status string + StartTime int64 + CPUPercent float64 + MemoryRSS uint64 + MemoryVMS uint64 +} + +// collectSystem fills Windows host identity, OS, and architecture fields. +func collectSystem(_ context.Context, hostID string, out *status.HostStatus) error { + hostname, err := os.Hostname() + if err != nil { + return err + } + out.Host.HostID = hostID + if out.Host.HostID == "" { + out.Host.HostID = hostname + } + out.Host.Hostname = hostname + out.Host.Platform = runtime.GOOS + out.Host.PlatformVersion = windowsVersion() + out.Host.Arch = runtime.GOARCH + out.Host.UptimeSeconds = windowsUptimeSeconds() + return nil +} + +// collectNetwork fills Windows interface, address, and MAC data using the portable net package. +func collectNetwork(_ context.Context, out *status.HostStatus) error { + ifaces, err := net.Interfaces() + if err != nil { + return err + } + seenIPs := map[string]struct{}{} + seenMACs := map[string]struct{}{} + for _, iface := range ifaces { + item := status.NetworkInterface{Name: iface.Name, MAC: iface.HardwareAddr.String(), MTU: iface.MTU, Flags: strings.Fields(iface.Flags.String())} + addrs, _ := iface.Addrs() + for _, addr := range addrs { + item.Addresses = append(item.Addresses, addr.String()) + seenIPs[addr.String()] = struct{}{} + } + if item.MAC != "" { + seenMACs[item.MAC] = struct{}{} + } + out.Network.Interfaces = append(out.Network.Interfaces, item) + } + out.Network.IPv4s, out.Network.IPv6s = networkAddressSets(seenIPs) + out.Network.MACs = windowsSortedKeys(seenMACs) + return nil +} + +// collectResource fills Windows CPU, memory, and disk resource data. +func collectResource(ctx context.Context, out *status.HostStatus) error { + var errs []error + out.CPU.LogicalCores = runtime.NumCPU() + out.CPU.PhysicalCores = windowsPhysicalCores() + out.CPU.ModelName = windowsCPUModelName() + out.CPU.UsagePercent = windowsReadCPUUsagePercent(ctx) + memory, err := windowsReadMemory() + if err != nil { + errs = append(errs, err) + } else { + out.Memory = memory + } + disks, err := windowsReadDisks() + if err != nil { + errs = append(errs, err) + } else { + out.Disks = disks + } + return errors.Join(errs...) +} + +// collectProcesses enumerates Windows processes with Toolhelp and returns best-effort names. +func collectProcesses(_ context.Context, out *status.HostStatus) error { + processes, err := windowsProcessList() + out.Processes.Items = processes + if err != nil { + out.Processes.Status = "partial" + return err + } + out.Processes.Status = "ok" + return nil +} + +// collectSockets reads Windows TCP listener tables and maps listener PIDs to process names. +func collectSockets(_ context.Context, out *status.HostStatus) error { + processes, processErr := windowsProcessMap() + var errs []error + if processErr != nil { + errs = append(errs, processErr) + } + listeners4, err := windowsTCPListeners(windowsAFInet, "ipv4", processes) + if err != nil { + errs = append(errs, err) + } + listeners6, err := windowsTCPListeners(windowsAFInet6, "ipv6", processes) + if err != nil { + errs = append(errs, err) + } + out.TCPListeners.Items = append(listeners4, listeners6...) + sort.Slice(out.TCPListeners.Items, func(i, j int) bool { + if out.TCPListeners.Items[i].IP == out.TCPListeners.Items[j].IP { + return out.TCPListeners.Items[i].Port < out.TCPListeners.Items[j].Port + } + return out.TCPListeners.Items[i].IP < out.TCPListeners.Items[j].IP + }) + if len(errs) > 0 { + out.TCPListeners.Status = "partial" + return errors.Join(errs...) + } + out.TCPListeners.Status = "ok" + return nil +} + +// windowsReadMemory calls GlobalMemoryStatusEx and converts physical memory counters to API fields. +func windowsReadMemory() (status.MemoryInfo, error) { + memory := windowsMemoryStatus{Length: uint32(unsafe.Sizeof(windowsMemoryStatus{}))} + ok, _, err := windowsGlobalMemoryStatusExProc.Call(uintptr(unsafe.Pointer(&memory))) + if ok == 0 { + return status.MemoryInfo{}, err + } + return windowsMemoryInfoFromStatus(memory), nil +} + +// windowsVersion returns the legacy Windows major.minor.build tuple exposed by GetVersion. +func windowsVersion() string { + version, err := syscall.GetVersion() + if err != nil { + return "" + } + major := byte(version) + minor := uint8(version >> 8) + build := uint16(version >> 16) + return fmt.Sprintf("%d.%d.%d", major, minor, build) +} + +// windowsUptimeSeconds returns host uptime in seconds from GetTickCount64. +func windowsUptimeSeconds() uint64 { + millis, _, _ := windowsGetTickCount64Proc.Call() + return uint64(millis) / 1000 +} + +// windowsPhysicalCores returns physical CPU core count from Windows processor topology. +func windowsPhysicalCores() int { + var size uint32 + result, _, _ := windowsGetLogicalProcessorInfoProc.Call(0, uintptr(unsafe.Pointer(&size))) + if result == 0 && size > 0 { + buf := make([]byte, size) + result, _, _ = windowsGetLogicalProcessorInfoProc.Call(uintptr(unsafe.Pointer(&buf[0])), uintptr(unsafe.Pointer(&size))) + if result != 0 { + entrySize := int(unsafe.Sizeof(windowsLogicalProcessorInformation{})) + cores := 0 + for offset := 0; offset+entrySize <= int(size); offset += entrySize { + entry := (*windowsLogicalProcessorInformation)(unsafe.Pointer(&buf[offset])) + if entry.Relationship == windowsRelationProcessorCore { + cores++ + } + } + if cores > 0 { + return cores + } + } + } + var info windowsSystemInfo + windowsGetNativeSystemInfoProc.Call(uintptr(unsafe.Pointer(&info))) + if info.NumberOfProcessors > 0 { + return int(info.NumberOfProcessors) + } + if info.ActiveProcessorMask != 0 { + return bits.OnesCount64(uint64(info.ActiveProcessorMask)) + } + return runtime.NumCPU() +} + +// windowsCPUModelName reads the first processor name from the registry. +func windowsCPUModelName() string { + keyPath, _ := syscall.UTF16PtrFromString(`HARDWARE\DESCRIPTION\System\CentralProcessor\0`) + var key syscall.Handle + if err := syscall.RegOpenKeyEx(syscall.HKEY_LOCAL_MACHINE, keyPath, 0, syscall.KEY_READ, &key); err != nil { + return "" + } + defer syscall.RegCloseKey(key) + valueName, _ := syscall.UTF16PtrFromString("ProcessorNameString") + var valueType uint32 + var size uint32 + if err := syscall.RegQueryValueEx(key, valueName, nil, &valueType, nil, &size); err != nil || valueType != windowsRegSZ || size == 0 { + return "" + } + buf := make([]byte, size) + if err := syscall.RegQueryValueEx(key, valueName, nil, &valueType, &buf[0], &size); err != nil { + return "" + } + utf16s := make([]uint16, 0, len(buf)/2) + for i := 0; i+1 < int(size); i += 2 { + utf16s = append(utf16s, uint16(buf[i])|uint16(buf[i+1])<<8) + } + return strings.TrimSpace(syscall.UTF16ToString(utf16s)) +} + +// windowsReadCPUUsagePercent samples GetSystemTimes and returns best-effort CPU usage. +func windowsReadCPUUsagePercent(ctx context.Context) float64 { + prev, err := windowsReadSystemTimes() + if err != nil { + return 0 + } + select { + case <-ctx.Done(): + return 0 + case <-time.After(100 * time.Millisecond): + } + next, err := windowsReadSystemTimes() + if err != nil { + return 0 + } + return windowsCPUUsagePercent(prev, next) +} + +// windowsReadSystemTimes returns idle, kernel, and user system counters from GetSystemTimes. +func windowsReadSystemTimes() (windowsSystemTimes, error) { + var idle, kernel, user syscall.Filetime + ok, _, err := windowsGetSystemTimesProc.Call( + uintptr(unsafe.Pointer(&idle)), + uintptr(unsafe.Pointer(&kernel)), + uintptr(unsafe.Pointer(&user)), + ) + if ok == 0 { + return windowsSystemTimes{}, err + } + return windowsSystemTimes{Idle: windowsFiletimeUint64(idle), Kernel: windowsFiletimeUint64(kernel), User: windowsFiletimeUint64(user)}, nil +} + +// windowsCPUUsagePercent calculates non-idle CPU percentage between two Windows CPU samples. +func windowsCPUUsagePercent(prev windowsSystemTimes, next windowsSystemTimes) float64 { + prevTotal := prev.Kernel + prev.User + nextTotal := next.Kernel + next.User + if nextTotal <= prevTotal || next.Idle < prev.Idle { + return 0 + } + totalDelta := nextTotal - prevTotal + idleDelta := next.Idle - prev.Idle + if totalDelta == 0 || idleDelta >= totalDelta { + return 0 + } + return float64(totalDelta-idleDelta) / float64(totalDelta) * 100 +} + +// windowsFiletimeUint64 converts a FILETIME to a monotonically comparable counter. +func windowsFiletimeUint64(value syscall.Filetime) uint64 { + return uint64(value.HighDateTime)<<32 | uint64(value.LowDateTime) +} + +// windowsReadDisks enumerates fixed and remote Windows drives and returns capacity data. +func windowsReadDisks() ([]status.DiskInfo, error) { + mask, _, _ := windowsGetLogicalDrivesProc.Call() + if mask == 0 { + return nil, syscall.GetLastError() + } + var disks []status.DiskInfo + var errs []error + for i := 0; i < 26; i++ { + if mask&(1<= free { + used = total - free + } + percent := 0.0 + if total > 0 { + percent = float64(used) / float64(total) * 100 + } + return status.DiskInfo{ + Device: root, MountPoint: root, FSType: windowsDriveTypeName(driveType), + Unit: "byte", Total: total, Used: used, Free: free, UsedPercent: percent, + } +} + +// windowsDriveTypeName maps Windows drive type constants to stable API labels. +func windowsDriveTypeName(driveType uint32) string { + switch driveType { + case windowsDriveFixed: + return "fixed" + case windowsDriveRemote: + return "remote" + default: + return "unknown" + } +} + +// windowsMemoryInfoFromStatus converts Windows physical memory counters into HostStatus memory data. +func windowsMemoryInfoFromStatus(memory windowsMemoryStatus) status.MemoryInfo { + used := uint64(0) + if memory.TotalPhys >= memory.AvailPhys { + used = memory.TotalPhys - memory.AvailPhys + } + percent := 0.0 + if memory.TotalPhys > 0 { + percent = float64(used) / float64(memory.TotalPhys) * 100 + } + return status.MemoryInfo{ + Unit: "byte", Total: memory.TotalPhys, Used: used, + Free: memory.AvailPhys, Available: memory.AvailPhys, UsedPercent: percent, + } +} + +// windowsTCPListeners returns IPv4 or IPv6 TCP listeners from GetExtendedTcpTable. +func windowsTCPListeners(family int, familyName string, processes map[int]status.ProcessInfo) ([]status.TCPListenerInfo, error) { + table, err := windowsExtendedTCPTable(family) + if err != nil { + return nil, err + } + if len(table) < 4 { + return nil, fmt.Errorf("windows tcp table for %s is too small", familyName) + } + count := *(*uint32)(unsafe.Pointer(&table[0])) + if family == windowsAFInet { + return windowsParseTCP4Rows(table[4:], count, processes), nil + } + return windowsParseTCP6Rows(table[4:], count, processes), nil +} + +// windowsExtendedTCPTable calls GetExtendedTcpTable and returns the raw table bytes. +func windowsExtendedTCPTable(family int) ([]byte, error) { + var size uint32 + result, _, _ := windowsGetExtendedTCPTableProc.Call( + 0, + uintptr(unsafe.Pointer(&size)), + 1, + uintptr(family), + uintptr(windowsTCPTableOwnerPIDListen), + 0, + ) + if result != windowsErrorInsufficientBuffer && result != windowsNoError { + return nil, fmt.Errorf("GetExtendedTcpTable size query failed: %d", result) + } + if size == 0 { + return nil, nil + } + table := make([]byte, size) + result, _, _ = windowsGetExtendedTCPTableProc.Call( + uintptr(unsafe.Pointer(&table[0])), + uintptr(unsafe.Pointer(&size)), + 1, + uintptr(family), + uintptr(windowsTCPTableOwnerPIDListen), + 0, + ) + if result != windowsNoError { + return nil, fmt.Errorf("GetExtendedTcpTable failed: %d", result) + } + return table, nil +} + +// windowsParseTCP4Rows converts raw MIB_TCPROW_OWNER_PID rows into listener records. +func windowsParseTCP4Rows(rows []byte, count uint32, processes map[int]status.ProcessInfo) []status.TCPListenerInfo { + rowSize := int(unsafe.Sizeof(windowsMIBTCPRowOwnerPID{})) + items := make([]status.TCPListenerInfo, 0, count) + for i := 0; i < int(count); i++ { + offset := i * rowSize + if offset+rowSize > len(rows) { + break + } + row := *(*windowsMIBTCPRowOwnerPID)(unsafe.Pointer(&rows[offset])) + if row.State != windowsTCPStateListen { + continue + } + process := processes[int(row.OwningPID)] + addr := row.LocalAddr + items = append(items, status.TCPListenerInfo{ + IP: net.IPv4(byte(addr), byte(addr>>8), byte(addr>>16), byte(addr>>24)).String(), + Port: windowsTCPPortFromNetworkOrder(row.LocalPort), PID: int(row.OwningPID), + ProcessName: process.Name, ProcessExe: process.Exe, Family: "ipv4", + }) + } + return items +} + +// windowsParseTCP6Rows converts raw MIB_TCP6ROW_OWNER_PID rows into listener records. +func windowsParseTCP6Rows(rows []byte, count uint32, processes map[int]status.ProcessInfo) []status.TCPListenerInfo { + rowSize := int(unsafe.Sizeof(windowsMIBTCP6RowOwnerPID{})) + items := make([]status.TCPListenerInfo, 0, count) + for i := 0; i < int(count); i++ { + offset := i * rowSize + if offset+rowSize > len(rows) { + break + } + row := *(*windowsMIBTCP6RowOwnerPID)(unsafe.Pointer(&rows[offset])) + if row.State != windowsTCPStateListen { + continue + } + process := processes[int(row.OwningPID)] + items = append(items, status.TCPListenerInfo{ + IP: net.IP(row.LocalAddr[:]).String(), Port: windowsTCPPortFromNetworkOrder(row.LocalPort), + PID: int(row.OwningPID), ProcessName: process.Name, ProcessExe: process.Exe, Family: "ipv6", + }) + } + return items +} + +// windowsTCPPortFromNetworkOrder decodes the low 16 bits Windows stores in network byte order. +func windowsTCPPortFromNetworkOrder(raw uint32) int { + port := uint16(raw) + return int(port>>8) | int(port&0xff)<<8 +} + +// windowsProcessList enumerates processes with the Windows Toolhelp snapshot API. +func windowsProcessList() ([]status.ProcessInfo, error) { + cmdlines := windowsProcessCommandLines() + snapshot, err := syscall.CreateToolhelp32Snapshot(syscall.TH32CS_SNAPPROCESS, 0) + if err != nil { + return nil, err + } + defer syscall.CloseHandle(snapshot) + var entry syscall.ProcessEntry32 + entry.Size = uint32(unsafe.Sizeof(entry)) + if err := syscall.Process32First(snapshot, &entry); err != nil { + return nil, err + } + var processes []status.ProcessInfo + for { + details := windowsReadProcessDetails(entry.ProcessID) + if cmdline := cmdlines[int(entry.ProcessID)]; cmdline != "" { + details.Cmdline = cmdline + } + processes = append(processes, windowsProcessInfoFromEntry(entry, details)) + err = syscall.Process32Next(snapshot, &entry) + if err != nil { + if errors.Is(err, syscall.ERROR_NO_MORE_FILES) { + return processes, nil + } + return processes, err + } + } +} + +// windowsProcessInfoFromEntry combines Toolhelp identity data with optional process details. +func windowsProcessInfoFromEntry(entry syscall.ProcessEntry32, details windowsProcessDetails) status.ProcessInfo { + name := syscall.UTF16ToString(entry.ExeFile[:]) + exe := details.Exe + if exe == "" { + exe = name + } + return status.ProcessInfo{ + PID: int(entry.ProcessID), PPID: int(entry.ParentProcessID), Name: name, + Exe: exe, Cmdline: details.Cmdline, Username: details.Username, Status: details.Status, + StartTime: details.StartTime, CPUPercent: details.CPUPercent, + MemoryRSS: details.MemoryRSS, MemoryVMS: details.MemoryVMS, + } +} + +// windowsReadProcessDetails opens a process and reads fields that may require elevated privileges. +func windowsReadProcessDetails(pid uint32) windowsProcessDetails { + handle, err := syscall.OpenProcess(windowsProcessQueryLimitedInfo|windowsProcessQueryInformation|windowsProcessVMRead, false, pid) + if err != nil { + return windowsProcessDetails{Status: "unknown"} + } + defer syscall.CloseHandle(handle) + details := windowsProcessDetails{Status: "running"} + details.Exe = windowsQueryProcessImageName(handle) + details.Username = windowsQueryProcessUsername(handle) + details.StartTime, details.CPUPercent = windowsQueryProcessTimes(handle) + details.MemoryRSS, details.MemoryVMS = windowsQueryProcessMemory(handle) + return details +} + +// windowsQueryProcessImageName returns the full process image path when access allows it. +func windowsQueryProcessImageName(handle syscall.Handle) string { + buf := make([]uint16, windowsMaxPath) + size := uint32(len(buf)) + ok, _, _ := windowsQueryFullProcessImageNameWProc.Call( + uintptr(handle), + 0, + uintptr(unsafe.Pointer(&buf[0])), + uintptr(unsafe.Pointer(&size)), + ) + if ok == 0 || size == 0 { + return "" + } + return syscall.UTF16ToString(buf[:size]) +} + +// windowsQueryProcessTimes returns start time and lifetime-average CPU percentage. +func windowsQueryProcessTimes(handle syscall.Handle) (int64, float64) { + var creation, exit, kernel, user syscall.Filetime + if err := syscall.GetProcessTimes(handle, &creation, &exit, &kernel, &user); err != nil { + return 0, 0 + } + startNanos := creation.Nanoseconds() + elapsedNanos := time.Now().UnixNano() - startNanos + if elapsedNanos <= 0 { + return startNanos / int64(time.Second), 0 + } + cpuNanos := int64(windowsFiletimeUint64(kernel)+windowsFiletimeUint64(user)) * 100 + percent := float64(cpuNanos) / float64(elapsedNanos) * 100 / float64(runtime.NumCPU()) + return startNanos / int64(time.Second), percent +} + +// windowsQueryProcessMemory returns working set and pagefile usage when access allows it. +func windowsQueryProcessMemory(handle syscall.Handle) (uint64, uint64) { + counters := windowsProcessMemoryCounters{CB: uint32(unsafe.Sizeof(windowsProcessMemoryCounters{}))} + ok, _, _ := windowsGetProcessMemoryInfoProc.Call( + uintptr(handle), + uintptr(unsafe.Pointer(&counters)), + uintptr(counters.CB), + ) + if ok == 0 { + return 0, 0 + } + return uint64(counters.WorkingSetSize), uint64(counters.PagefileUsage) +} + +// windowsQueryProcessUsername returns DOMAIN\user for a process token when access allows it. +func windowsQueryProcessUsername(handle syscall.Handle) string { + var token syscall.Handle + ok, _, _ := windowsOpenProcessTokenProc.Call(uintptr(handle), uintptr(windowsTokenQuery), uintptr(unsafe.Pointer(&token))) + if ok == 0 { + return "" + } + defer syscall.CloseHandle(token) + var size uint32 + windowsGetTokenInformationProc.Call(uintptr(token), 1, 0, 0, uintptr(unsafe.Pointer(&size))) + if size == 0 { + return "" + } + buf := make([]byte, size) + ok, _, _ = windowsGetTokenInformationProc.Call(uintptr(token), 1, uintptr(unsafe.Pointer(&buf[0])), uintptr(size), uintptr(unsafe.Pointer(&size))) + if ok == 0 || len(buf) < int(unsafe.Sizeof(uintptr(0)))+4 { + return "" + } + sid := *(*uintptr)(unsafe.Pointer(&buf[0])) + return windowsLookupAccountName(sid) +} + +// windowsLookupAccountName resolves a SID pointer to DOMAIN\name. +func windowsLookupAccountName(sid uintptr) string { + var nameSize, domainSize uint32 + var sidType uint32 + windowsLookupAccountSidWProc.Call(0, sid, 0, uintptr(unsafe.Pointer(&nameSize)), 0, uintptr(unsafe.Pointer(&domainSize)), uintptr(unsafe.Pointer(&sidType))) + if nameSize == 0 { + return "" + } + name := make([]uint16, nameSize) + domain := make([]uint16, domainSize) + ok, _, _ := windowsLookupAccountSidWProc.Call( + 0, + sid, + uintptr(unsafe.Pointer(&name[0])), + uintptr(unsafe.Pointer(&nameSize)), + uintptr(unsafe.Pointer(&domain[0])), + uintptr(unsafe.Pointer(&domainSize)), + uintptr(unsafe.Pointer(&sidType)), + ) + if ok == 0 { + return "" + } + account := syscall.UTF16ToString(name[:nameSize]) + domainName := syscall.UTF16ToString(domain[:domainSize]) + if domainName == "" { + return account + } + return domainName + `\` + account +} + +// windowsProcessCommandLines reads command lines in one best-effort WMIC call to avoid per-process shelling. +func windowsProcessCommandLines() map[int]string { + out := map[int]string{} + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + output, err := exec.CommandContext(ctx, "wmic", "process", "get", "ProcessId,CommandLine", "/format:csv").Output() + if err != nil { + return out + } + reader := csv.NewReader(bytes.NewReader(output)) + reader.FieldsPerRecord = -1 + records, err := reader.ReadAll() + if err != nil { + return out + } + for _, record := range records { + if len(record) < 3 || strings.EqualFold(record[1], "CommandLine") { + continue + } + pid, err := strconv.Atoi(strings.TrimSpace(record[2])) + if err != nil { + continue + } + out[pid] = strings.TrimSpace(record[1]) + } + return out +} + +// windowsProcessMap returns process metadata keyed by PID for TCP owner mapping. +func windowsProcessMap() (map[int]status.ProcessInfo, error) { + processes, err := windowsProcessList() + out := make(map[int]status.ProcessInfo, len(processes)) + for _, process := range processes { + out[process.PID] = process + } + return out, err +} + +// windowsSortedKeys returns deterministic sorted keys from a string set. +func windowsSortedKeys(values map[string]struct{}) []string { + keys := make([]string, 0, len(values)) + for key := range values { + keys = append(keys, key) + } + sort.Strings(keys) + return keys +} diff --git a/internal/collector/collector_windows_test.go b/internal/collector/collector_windows_test.go new file mode 100644 index 0000000..1974361 --- /dev/null +++ b/internal/collector/collector_windows_test.go @@ -0,0 +1,87 @@ +//go:build windows + +package collector + +import ( + "syscall" + "testing" +) + +// TestWindowsMemoryInfoFromStatus verifies Windows memory counters are exposed in bytes. +func TestWindowsMemoryInfoFromStatus(t *testing.T) { + got := windowsMemoryInfoFromStatus(windowsMemoryStatus{ + TotalPhys: 10, + AvailPhys: 4, + }) + + if got.Unit != "byte" { + t.Fatalf("unit = %q, want byte", got.Unit) + } + if got.Total != 10 || got.Available != 4 || got.Free != 4 || got.Used != 6 { + t.Fatalf("memory = %+v", got) + } + if got.UsedPercent != 60 { + t.Fatalf("used percent = %f, want 60", got.UsedPercent) + } +} + +// TestWindowsTCPPortFromNetworkOrder verifies Windows TCP port values are decoded. +func TestWindowsTCPPortFromNetworkOrder(t *testing.T) { + if got := windowsTCPPortFromNetworkOrder(0x0000901f); got != 8080 { + t.Fatalf("port = %d, want 8080", got) + } +} + +// TestWindowsDiskInfoFromSpace verifies Windows disk counters are mapped to API disk fields. +func TestWindowsDiskInfoFromSpace(t *testing.T) { + got := windowsDiskInfoFromSpace(`C:\`, windowsDriveFixed, 100, 40) + + if got.Device != `C:\` || got.MountPoint != `C:\` { + t.Fatalf("disk paths = %q/%q", got.Device, got.MountPoint) + } + if got.FSType != "fixed" || got.Unit != "byte" { + t.Fatalf("disk type/unit = %q/%q", got.FSType, got.Unit) + } + if got.Total != 100 || got.Free != 40 || got.Used != 60 { + t.Fatalf("disk capacity = %+v", got) + } + if got.UsedPercent != 60 { + t.Fatalf("used percent = %f, want 60", got.UsedPercent) + } +} + +// TestWindowsCPUUsagePercent verifies idle and kernel counters are converted to busy percentage. +func TestWindowsCPUUsagePercent(t *testing.T) { + prev := windowsSystemTimes{Idle: 100, Kernel: 500, User: 500} + next := windowsSystemTimes{Idle: 150, Kernel: 650, User: 700} + + got := windowsCPUUsagePercent(prev, next) + + if got != 75 { + t.Fatalf("cpu usage = %f, want 75", got) + } +} + +// TestWindowsProcessInfoFromEntry verifies Toolhelp process data is enriched with memory and command line data. +func TestWindowsProcessInfoFromEntry(t *testing.T) { + entry := syscall.ProcessEntry32{ProcessID: 123, ParentProcessID: 12} + copy(entry.ExeFile[:], syscall.StringToUTF16("agent.exe")) + + got := windowsProcessInfoFromEntry(entry, windowsProcessDetails{ + Exe: `C:\agent\agent.exe`, + Cmdline: `"C:\agent\agent.exe" run`, + Username: `NT AUTHORITY\SYSTEM`, + MemoryRSS: 2048, + MemoryVMS: 4096, + }) + + if got.PID != 123 || got.PPID != 12 || got.Name != "agent.exe" { + t.Fatalf("process identity = %+v", got) + } + if got.Exe != `C:\agent\agent.exe` || got.Cmdline != `"C:\agent\agent.exe" run` { + t.Fatalf("process command fields = %+v", got) + } + if got.Username != `NT AUTHORITY\SYSTEM` || got.MemoryRSS != 2048 || got.MemoryVMS != 4096 { + t.Fatalf("process details = %+v", got) + } +} diff --git a/internal/collector/network.go b/internal/collector/network.go new file mode 100644 index 0000000..76e034c --- /dev/null +++ b/internal/collector/network.go @@ -0,0 +1,27 @@ +package collector + +import ( + "net" + "sort" +) + +// networkAddressSets separates interface addresses into deterministic IPv4 and IPv6 lists. +func networkAddressSets(values map[string]struct{}) (ipv4s []string, ipv6s []string) { + for value := range values { + host, _, err := net.ParseCIDR(value) + if err != nil { + host = net.ParseIP(value) + } + if host == nil { + continue + } + if host.To4() != nil { + ipv4s = append(ipv4s, value) + continue + } + ipv6s = append(ipv6s, value) + } + sort.Strings(ipv4s) + sort.Strings(ipv6s) + return ipv4s, ipv6s +} diff --git a/internal/collector/network_test.go b/internal/collector/network_test.go new file mode 100644 index 0000000..bd21f85 --- /dev/null +++ b/internal/collector/network_test.go @@ -0,0 +1,23 @@ +package collector + +import ( + "reflect" + "testing" +) + +// TestNetworkAddressSetsSeparatesIPv4AndIPv6 verifies flattened IP lists are split by address family. +func TestNetworkAddressSetsSeparatesIPv4AndIPv6(t *testing.T) { + ipv4s, ipv6s := networkAddressSets(map[string]struct{}{ + "fe80::1/64": {}, + "192.168.1.10/24": {}, + "10.0.0.2/8": {}, + "not-an-ip": {}, + }) + + if want := []string{"10.0.0.2/8", "192.168.1.10/24"}; !reflect.DeepEqual(ipv4s, want) { + t.Fatalf("ipv4s = %#v, want %#v", ipv4s, want) + } + if want := []string{"fe80::1/64"}; !reflect.DeepEqual(ipv6s, want) { + t.Fatalf("ipv6s = %#v, want %#v", ipv6s, want) + } +} diff --git a/internal/config/config.go b/internal/config/config.go new file mode 100644 index 0000000..98fca36 --- /dev/null +++ b/internal/config/config.go @@ -0,0 +1,149 @@ +package config + +import ( + "encoding/json" + "errors" + "os" + "strconv" + "time" +) + +// Config contains all runtime, reporting, collection, and service settings. +type Config struct { + HostID string `json:"host_id"` + HTTP HTTPConfig `json:"http"` + Collect CollectConfig `json:"collect"` + Report ReportConfig `json:"report"` + Service ServiceConfig `json:"service"` +} + +// HTTPConfig controls the local HTTP API listener. +type HTTPConfig struct { + ListenAddress string `json:"listen_address"` +} + +// CollectConfig controls periodic host status collection. +type CollectConfig struct { + Interval Duration `json:"interval"` + Timeout Duration `json:"timeout"` +} + +// ReportConfig controls optional periodic status POST reporting. +type ReportConfig struct { + Enabled bool `json:"enabled"` + Endpoint string `json:"endpoint"` + Interval Duration `json:"interval"` + Timeout Duration `json:"timeout"` +} + +// ServiceConfig controls operating system service registration metadata. +type ServiceConfig struct { + Name string `json:"name"` + DisplayName string `json:"display_name"` + Description string `json:"description"` +} + +// Duration wraps time.Duration so JSON config can use Go duration strings. +type Duration time.Duration + +// Duration returns the standard library duration value. +func (d Duration) Duration() time.Duration { + return time.Duration(d) +} + +// MarshalJSON writes the duration as a Go duration string such as "30s". +func (d Duration) MarshalJSON() ([]byte, error) { + return json.Marshal(time.Duration(d).String()) +} + +// UnmarshalJSON parses a Go duration string from JSON. +func (d *Duration) UnmarshalJSON(data []byte) error { + var raw string + if err := json.Unmarshal(data, &raw); err != nil { + return err + } + parsed, err := time.ParseDuration(raw) + if err != nil { + return err + } + *d = Duration(parsed) + return nil +} + +// Default returns safe local defaults for running the agent. +func Default() Config { + return Config{ + HTTP: HTTPConfig{ListenAddress: "127.0.0.1:8080"}, + Collect: CollectConfig{ + Interval: Duration(30 * time.Second), + Timeout: Duration(10 * time.Second), + }, + Report: ReportConfig{ + Enabled: false, + Interval: Duration(60 * time.Second), + Timeout: Duration(5 * time.Second), + }, + Service: ServiceConfig{ + Name: "mfd-monitor-agent", + DisplayName: "MFD Monitor Agent", + Description: "Collects host status and exposes it over HTTP.", + }, + } +} + +// Load reads optional JSON config, applies environment overrides, and validates required settings. +func Load(path string) (Config, error) { + cfg := Default() + if path != "" { + data, err := os.ReadFile(path) + if err != nil { + return cfg, err + } + if err := json.Unmarshal(data, &cfg); err != nil { + return cfg, err + } + } + applyEnv(&cfg) + if cfg.Report.Enabled && cfg.Report.Endpoint == "" { + return cfg, errors.New("report endpoint is required when reporting is enabled") + } + return cfg, nil +} + +// applyEnv overlays supported MFD_AGENT_* environment variables onto cfg. +func applyEnv(cfg *Config) { + if value := os.Getenv("MFD_AGENT_HOST_ID"); value != "" { + cfg.HostID = value + } + if value := os.Getenv("MFD_AGENT_LISTEN"); value != "" { + cfg.HTTP.ListenAddress = value + } + if value := os.Getenv("MFD_AGENT_REPORT_ENABLED"); value != "" { + if parsed, err := strconv.ParseBool(value); err == nil { + cfg.Report.Enabled = parsed + } + } + if value := os.Getenv("MFD_AGENT_REPORT_ENDPOINT"); value != "" { + cfg.Report.Endpoint = value + } + if value := os.Getenv("MFD_AGENT_REPORT_INTERVAL"); value != "" { + if parsed, err := time.ParseDuration(value); err == nil { + cfg.Report.Interval = Duration(parsed) + } + } + if value := os.Getenv("MFD_AGENT_REPORT_TIMEOUT"); value != "" { + if parsed, err := time.ParseDuration(value); err == nil { + cfg.Report.Timeout = Duration(parsed) + } + } + if value := os.Getenv("MFD_AGENT_COLLECT_INTERVAL"); value != "" { + if parsed, err := time.ParseDuration(value); err == nil { + cfg.Collect.Interval = Duration(parsed) + } + } + if value := os.Getenv("MFD_AGENT_COLLECT_TIMEOUT"); value != "" { + if parsed, err := time.ParseDuration(value); err == nil { + cfg.Collect.Timeout = Duration(parsed) + } + } +} diff --git a/internal/config/config_test.go b/internal/config/config_test.go new file mode 100644 index 0000000..a1cb864 --- /dev/null +++ b/internal/config/config_test.go @@ -0,0 +1,71 @@ +package config + +import ( + "os" + "testing" + "time" +) + +// TestDefaultConfigUsesLocalhostAndSaneIntervals verifies safe local defaults. +func TestDefaultConfigUsesLocalhostAndSaneIntervals(t *testing.T) { + cfg := Default() + + if cfg.HTTP.ListenAddress != "127.0.0.1:8080" { + t.Fatalf("listen address = %q", cfg.HTTP.ListenAddress) + } + if cfg.Collect.Interval <= 0 || cfg.Collect.Timeout <= 0 { + t.Fatalf("collect intervals must be positive: %+v", cfg.Collect) + } + if cfg.Report.Enabled { + t.Fatal("reporting should be disabled by default") + } +} + +// TestLoadAppliesEnvironmentOverrides verifies environment variables override defaults. +func TestLoadAppliesEnvironmentOverrides(t *testing.T) { + t.Setenv("MFD_AGENT_LISTEN", "127.0.0.1:9090") + t.Setenv("MFD_AGENT_REPORT_ENABLED", "true") + t.Setenv("MFD_AGENT_REPORT_ENDPOINT", "http://center/status") + t.Setenv("MFD_AGENT_REPORT_INTERVAL", "2s") + t.Setenv("MFD_AGENT_HOST_ID", "host-a") + + cfg, err := Load("") + if err != nil { + t.Fatal(err) + } + + if cfg.HTTP.ListenAddress != "127.0.0.1:9090" { + t.Fatalf("listen address = %q", cfg.HTTP.ListenAddress) + } + if !cfg.Report.Enabled || cfg.Report.Endpoint != "http://center/status" { + t.Fatalf("report config not applied: %+v", cfg.Report) + } + if cfg.Report.Interval.Duration() != 2*time.Second { + t.Fatalf("report interval = %s", cfg.Report.Interval.Duration()) + } + if cfg.HostID != "host-a" { + t.Fatalf("host id = %q", cfg.HostID) + } +} + +// TestLoadJSONConfigFile verifies JSON config files are parsed into Config. +func TestLoadJSONConfigFile(t *testing.T) { + file, err := os.CreateTemp(t.TempDir(), "config-*.json") + if err != nil { + t.Fatal(err) + } + _, _ = file.WriteString(`{"http":{"listen_address":"127.0.0.1:7070"},"collect":{"interval":"3s","timeout":"1s"}}`) + _ = file.Close() + + cfg, err := Load(file.Name()) + if err != nil { + t.Fatal(err) + } + + if cfg.HTTP.ListenAddress != "127.0.0.1:7070" { + t.Fatalf("listen address = %q", cfg.HTTP.ListenAddress) + } + if cfg.Collect.Interval.Duration() != 3*time.Second || cfg.Collect.Timeout.Duration() != time.Second { + t.Fatalf("collect config = %+v", cfg.Collect) + } +} diff --git a/internal/httpapi/server.go b/internal/httpapi/server.go new file mode 100644 index 0000000..0985c78 --- /dev/null +++ b/internal/httpapi/server.go @@ -0,0 +1,69 @@ +package httpapi + +import ( + "encoding/json" + "net/http" + + "mfd-monitor-agent/internal/status" +) + +// Provider supplies the latest cached HostStatus to HTTP handlers. +type Provider interface { + // Get returns the latest snapshot and whether a snapshot is ready. + Get() (status.HostStatus, bool) +} + +// StaticProvider is a test provider with a fixed snapshot response. +type StaticProvider struct { + Snapshot status.HostStatus + OK bool +} + +// Get returns the configured static snapshot and readiness flag. +func (p StaticProvider) Get() (status.HostStatus, bool) { + return p.Snapshot, p.OK +} + +// New creates the HTTP API handler for health and status endpoints. +func New(provider Provider) http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + snapshot, ok := provider.Get() + if !ok { + writeJSON(w, http.StatusServiceUnavailable, map[string]any{ + "status": "starting", + }) + return + } + writeJSON(w, http.StatusOK, map[string]any{ + "status": "ok", + "version": snapshot.Agent.Version, + "snapshot_at": snapshot.Agent.SnapshotAt, + "snapshot_status": snapshot.Agent.SnapshotStatus, + }) + }) + mux.HandleFunc("/status", func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + snapshot, ok := provider.Get() + if !ok { + http.Error(w, "snapshot not ready", http.StatusServiceUnavailable) + return + } + writeJSON(w, http.StatusOK, snapshot) + }) + return mux +} + +// writeJSON serializes value as JSON with a response status code. +func writeJSON(w http.ResponseWriter, code int, value any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(code) + _ = json.NewEncoder(w).Encode(value) +} diff --git a/internal/httpapi/server_test.go b/internal/httpapi/server_test.go new file mode 100644 index 0000000..8866e60 --- /dev/null +++ b/internal/httpapi/server_test.go @@ -0,0 +1,117 @@ +package httpapi + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "time" + + "mfd-monitor-agent/internal/status" +) + +// TestStatusReturnsCachedSnapshot verifies /status reads the provider cache. +func TestStatusReturnsCachedSnapshot(t *testing.T) { + provider := StaticProvider{Snapshot: status.HostStatus{ + Agent: status.AgentInfo{Version: "test", SnapshotAt: time.Now(), SnapshotStatus: "ok"}, + Host: status.HostInfo{Hostname: "host-a"}, + }, OK: true} + server := New(provider) + + req := httptest.NewRequest(http.MethodGet, "/status", nil) + rec := httptest.NewRecorder() + server.ServeHTTP(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("status code = %d", rec.Code) + } + var got status.HostStatus + if err := json.NewDecoder(rec.Body).Decode(&got); err != nil { + t.Fatal(err) + } + if got.Host.Hostname != "host-a" { + t.Fatalf("hostname = %q", got.Host.Hostname) + } +} + +// TestHealthReportsSnapshotMetadata verifies /health returns snapshot metadata. +func TestHealthReportsSnapshotMetadata(t *testing.T) { + provider := StaticProvider{Snapshot: status.HostStatus{ + Agent: status.AgentInfo{Version: "test", SnapshotAt: time.Unix(10, 0), SnapshotStatus: "partial"}, + }, OK: true} + server := New(provider) + + req := httptest.NewRequest(http.MethodGet, "/health", nil) + rec := httptest.NewRecorder() + server.ServeHTTP(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("status code = %d", rec.Code) + } + var got map[string]any + if err := json.NewDecoder(rec.Body).Decode(&got); err != nil { + t.Fatal(err) + } + if got["snapshot_status"] != "partial" { + t.Fatalf("health = %+v", got) + } +} + +// TestStatusReturnsServiceUnavailableBeforeSnapshot verifies /status does not live collect. +func TestStatusReturnsServiceUnavailableBeforeSnapshot(t *testing.T) { + server := New(StaticProvider{OK: false}) + + req := httptest.NewRequest(http.MethodGet, "/status", nil) + rec := httptest.NewRecorder() + server.ServeHTTP(rec, req) + + if rec.Code != http.StatusServiceUnavailable { + t.Fatalf("status code = %d", rec.Code) + } +} + +// TestHealthReturnsStartingBeforeSnapshot verifies /health exposes startup state. +func TestHealthReturnsStartingBeforeSnapshot(t *testing.T) { + server := New(StaticProvider{OK: false}) + + req := httptest.NewRequest(http.MethodGet, "/health", nil) + rec := httptest.NewRecorder() + server.ServeHTTP(rec, req) + + if rec.Code != http.StatusServiceUnavailable { + t.Fatalf("status code = %d", rec.Code) + } + var got map[string]string + if err := json.NewDecoder(rec.Body).Decode(&got); err != nil { + t.Fatal(err) + } + if got["status"] != "starting" { + t.Fatalf("status = %q", got["status"]) + } +} + +// TestStatusRejectsNonGET verifies /status only accepts GET requests. +func TestStatusRejectsNonGET(t *testing.T) { + server := New(StaticProvider{OK: true}) + + req := httptest.NewRequest(http.MethodPost, "/status", nil) + rec := httptest.NewRecorder() + server.ServeHTTP(rec, req) + + if rec.Code != http.StatusMethodNotAllowed { + t.Fatalf("status code = %d", rec.Code) + } +} + +// TestHealthRejectsNonGET verifies /health only accepts GET requests. +func TestHealthRejectsNonGET(t *testing.T) { + server := New(StaticProvider{OK: true}) + + req := httptest.NewRequest(http.MethodPost, "/health", nil) + rec := httptest.NewRecorder() + server.ServeHTTP(rec, req) + + if rec.Code != http.StatusMethodNotAllowed { + t.Fatalf("status code = %d", rec.Code) + } +} diff --git a/internal/reporter/reporter.go b/internal/reporter/reporter.go new file mode 100644 index 0000000..47f3297 --- /dev/null +++ b/internal/reporter/reporter.go @@ -0,0 +1,99 @@ +package reporter + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "log/slog" + "net/http" + "time" + + "mfd-monitor-agent/internal/status" +) + +const defaultRetryDelay = 2 * time.Second + +// Provider supplies the cached HostStatus used for outbound reports. +type Provider interface { + // Get returns the latest snapshot and whether a snapshot is ready. + Get() (status.HostStatus, bool) +} + +// Client posts HostStatus snapshots to a configured center endpoint. +type Client struct { + endpoint string + client *http.Client +} + +// NewClient creates a reporting client with the provided endpoint and request timeout. +func NewClient(endpoint string, timeout time.Duration) *Client { + return &Client{ + endpoint: endpoint, + client: &http.Client{Timeout: timeout}, + } +} + +// Report posts one HostStatus snapshot as JSON and returns non-2xx responses as errors. +func (c *Client) Report(snapshot status.HostStatus) error { + body, err := json.Marshal(snapshot) + if err != nil { + return err + } + req, err := http.NewRequest(http.MethodPost, c.endpoint, bytes.NewReader(body)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + resp, err := c.client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return fmt.Errorf("report returned HTTP %d", resp.StatusCode) + } + return nil +} + +// Run periodically posts the latest cached snapshot until ctx is canceled. +func Run(ctx context.Context, provider Provider, client *Client, interval time.Duration, logger *slog.Logger) { + runWithRetryDelay(ctx, provider, client, interval, defaultRetryDelay, logger) +} + +// runWithRetryDelay runs the reporting loop and retries one failed report after retryDelay. +func runWithRetryDelay(ctx context.Context, provider Provider, client *Client, interval time.Duration, retryDelay time.Duration, logger *slog.Logger) { + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + snapshot, ok := provider.Get() + if !ok { + continue + } + if err := reportWithRetry(ctx, client, snapshot, retryDelay); err != nil { + logger.Warn("status report failed", "error", err) + } + } + } +} + +// reportWithRetry posts a snapshot and performs one cancellation-aware retry on failure. +func reportWithRetry(ctx context.Context, client *Client, snapshot status.HostStatus, retryDelay time.Duration) error { + err := client.Report(snapshot) + if err == nil { + return nil + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(retryDelay): + } + if retryErr := client.Report(snapshot); retryErr != nil { + return retryErr + } + return nil +} diff --git a/internal/reporter/reporter_test.go b/internal/reporter/reporter_test.go new file mode 100644 index 0000000..ccd50c7 --- /dev/null +++ b/internal/reporter/reporter_test.go @@ -0,0 +1,138 @@ +package reporter + +import ( + "context" + "encoding/json" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" + + "mfd-monitor-agent/internal/status" +) + +// StaticProvider is a test provider with a fixed cached snapshot. +type StaticProvider struct { + Snapshot status.HostStatus + OK bool +} + +// Get returns the configured static snapshot and readiness flag. +func (p StaticProvider) Get() (status.HostStatus, bool) { + return p.Snapshot, p.OK +} + +// TestReportPostsHostStatusJSON verifies reports reuse the HostStatus JSON payload. +func TestReportPostsHostStatusJSON(t *testing.T) { + var got status.HostStatus + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + t.Fatalf("method = %s", r.Method) + } + if r.Header.Get("Content-Type") != "application/json" { + t.Fatalf("content type = %q", r.Header.Get("Content-Type")) + } + if err := json.NewDecoder(r.Body).Decode(&got); err != nil { + t.Fatal(err) + } + w.WriteHeader(http.StatusAccepted) + })) + defer server.Close() + + client := NewClient(server.URL, time.Second) + err := client.Report(status.HostStatus{Host: status.HostInfo{Hostname: "host-a"}}) + if err != nil { + t.Fatal(err) + } + if got.Host.Hostname != "host-a" { + t.Fatalf("posted hostname = %q", got.Host.Hostname) + } +} + +// TestReportReturnsErrorForHTTPFailure verifies non-2xx responses fail one report. +func TestReportReturnsErrorForHTTPFailure(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + })) + defer server.Close() + + client := NewClient(server.URL, time.Second) + if err := client.Report(status.HostStatus{}); err == nil { + t.Fatal("expected error for HTTP 500") + } +} + +// TestRunRetriesFailedReportOnce verifies the loop retries one failed report. +func TestRunRetriesFailedReportOnce(t *testing.T) { + var requests atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + if requests.Add(1) == 1 { + w.WriteHeader(http.StatusInternalServerError) + return + } + w.WriteHeader(http.StatusAccepted) + })) + defer server.Close() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + provider := StaticProvider{Snapshot: status.HostStatus{Host: status.HostInfo{Hostname: "host-a"}}, OK: true} + client := NewClient(server.URL, time.Second) + done := make(chan struct{}) + go func() { + defer close(done) + runWithRetryDelay(ctx, provider, client, 10*time.Millisecond, time.Millisecond, slog.New(slog.NewTextHandler(io.Discard, nil))) + }() + + deadline := time.After(time.Second) + for requests.Load() < 2 { + select { + case <-deadline: + t.Fatalf("requests = %d", requests.Load()) + case <-time.After(time.Millisecond): + } + } + cancel() + <-done +} + +// TestRunStopsDuringRetryDelayWhenContextCanceled verifies retry waits honor cancellation. +func TestRunStopsDuringRetryDelayWhenContextCanceled(t *testing.T) { + var requests atomic.Int32 + firstRequest := make(chan struct{}) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + if requests.Add(1) == 1 { + close(firstRequest) + } + w.WriteHeader(http.StatusInternalServerError) + })) + defer server.Close() + + ctx, cancel := context.WithCancel(context.Background()) + provider := StaticProvider{Snapshot: status.HostStatus{}, OK: true} + client := NewClient(server.URL, time.Second) + done := make(chan struct{}) + go func() { + defer close(done) + runWithRetryDelay(ctx, provider, client, time.Millisecond, time.Hour, slog.New(slog.NewTextHandler(io.Discard, nil))) + }() + + select { + case <-firstRequest: + case <-time.After(time.Second): + t.Fatal("initial report did not run") + } + cancel() + + select { + case <-done: + case <-time.After(100 * time.Millisecond): + t.Fatal("run did not stop during retry delay") + } + if requests.Load() != 1 { + t.Fatalf("requests = %d", requests.Load()) + } +} diff --git a/internal/service/service.go b/internal/service/service.go new file mode 100644 index 0000000..bdfb947 --- /dev/null +++ b/internal/service/service.go @@ -0,0 +1,26 @@ +package service + +import "mfd-monitor-agent/internal/config" + +// InstallOptions carries platform-specific service installation inputs. +type InstallOptions struct { + // ConfigPath is preserved in the service command line when a config file was supplied. + ConfigPath string +} + +// Manager controls operating system service installation and lifecycle commands. +type Manager interface { + // Install registers the agent executable as an operating system service. + Install(executable string, cfg config.Config, opts InstallOptions) error + // Uninstall removes the operating system service registration. + Uninstall(cfg config.Config) error + // Start asks the operating system service manager to start the agent service. + Start(cfg config.Config) error + // Stop asks the operating system service manager to stop the agent service. + Stop(cfg config.Config) error +} + +// NewManager returns the platform-specific service manager. +func NewManager() Manager { + return newManager() +} diff --git a/internal/service/service_linux.go b/internal/service/service_linux.go new file mode 100644 index 0000000..9b9c3bf --- /dev/null +++ b/internal/service/service_linux.go @@ -0,0 +1,82 @@ +//go:build linux + +package service + +import ( + "fmt" + "os" + "os/exec" + "strings" + + "mfd-monitor-agent/internal/config" +) + +var currentUID = os.Geteuid + +// linuxManager manages a systemd service for the agent. +type linuxManager struct{} + +// newManager returns the Linux systemd-backed service manager. +func newManager() Manager { + return linuxManager{} +} + +// Install writes a systemd unit, reloads systemd, and enables the configured service. +func (linuxManager) Install(executable string, cfg config.Config, opts InstallOptions) error { + if currentUID() != 0 { + return fmt.Errorf("install requires root") + } + unit := renderSystemdUnit(executable, opts.ConfigPath, cfg) + path := "/etc/systemd/system/" + cfg.Service.Name + ".service" + if err := os.WriteFile(path, []byte(unit), 0644); err != nil { + return fmt.Errorf("write systemd unit: %w", err) + } + if err := exec.Command("systemctl", "daemon-reload").Run(); err != nil { + return fmt.Errorf("systemctl daemon-reload: %w", err) + } + if err := exec.Command("systemctl", "enable", cfg.Service.Name).Run(); err != nil { + return fmt.Errorf("systemctl enable: %w", err) + } + return nil +} + +// renderSystemdUnit builds the Linux systemd unit used by Install. +func renderSystemdUnit(executable string, configPath string, cfg config.Config) string { + args := []string{executable, "run"} + if configPath != "" { + args = append(args, "-config", configPath) + } + return fmt.Sprintf(`[Unit] +Description=%s +After=network-online.target +Wants=network-online.target + +[Service] +Type=simple +ExecStart=%s +Restart=always +RestartSec=5s + +[Install] +WantedBy=multi-user.target +`, cfg.Service.Description, strings.Join(args, " ")) +} + +// Uninstall disables the service, removes its unit file, and reloads systemd. +func (linuxManager) Uninstall(cfg config.Config) error { + _ = exec.Command("systemctl", "disable", "--now", cfg.Service.Name).Run() + if err := os.Remove("/etc/systemd/system/" + cfg.Service.Name + ".service"); err != nil && !os.IsNotExist(err) { + return err + } + return exec.Command("systemctl", "daemon-reload").Run() +} + +// Start asks systemd to start the configured service. +func (linuxManager) Start(cfg config.Config) error { + return exec.Command("systemctl", "start", cfg.Service.Name).Run() +} + +// Stop asks systemd to stop the configured service. +func (linuxManager) Stop(cfg config.Config) error { + return exec.Command("systemctl", "stop", cfg.Service.Name).Run() +} diff --git a/internal/service/service_linux_test.go b/internal/service/service_linux_test.go new file mode 100644 index 0000000..1d7ac95 --- /dev/null +++ b/internal/service/service_linux_test.go @@ -0,0 +1,37 @@ +//go:build linux + +package service + +import ( + "strings" + "testing" + + "mfd-monitor-agent/internal/config" +) + +// TestRenderSystemdUnitIncludesConfigPath verifies install preserves the config flag. +func TestRenderSystemdUnitIncludesConfigPath(t *testing.T) { + cfg := config.Default() + cfg.Service.Description = "MFD Agent" + + unit := renderSystemdUnit("/opt/agent", "/etc/mfd-agent.json", cfg) + + if !strings.Contains(unit, "ExecStart=/opt/agent run -config /etc/mfd-agent.json") { + t.Fatalf("unit missing config path:\n%s", unit) + } +} + +// TestLinuxInstallRequiresRoot verifies Linux install rejects non-root callers. +func TestLinuxInstallRequiresRoot(t *testing.T) { + oldUID := currentUID + t.Cleanup(func() { currentUID = oldUID }) + currentUID = func() int { return 1000 } + + err := linuxManager{}.Install("/opt/agent", config.Default(), InstallOptions{}) + if err == nil { + t.Fatal("expected root requirement error") + } + if !strings.Contains(err.Error(), "root") { + t.Fatalf("error = %v", err) + } +} diff --git a/internal/service/service_other.go b/internal/service/service_other.go new file mode 100644 index 0000000..4e65f16 --- /dev/null +++ b/internal/service/service_other.go @@ -0,0 +1,38 @@ +//go:build !linux && !windows + +package service + +import ( + "fmt" + "runtime" + + "mfd-monitor-agent/internal/config" +) + +// unsupportedManager reports service lifecycle operations as unsupported on this platform. +type unsupportedManager struct{} + +// newManager returns a manager that fails all service operations on unsupported platforms. +func newManager() Manager { + return unsupportedManager{} +} + +// Install reports that service installation is unsupported on this platform. +func (unsupportedManager) Install(string, config.Config, InstallOptions) error { + return fmt.Errorf("service install is not implemented for %s", runtime.GOOS) +} + +// Uninstall reports that service removal is unsupported on this platform. +func (unsupportedManager) Uninstall(config.Config) error { + return fmt.Errorf("service uninstall is not implemented for %s", runtime.GOOS) +} + +// Start reports that service start is unsupported on this platform. +func (unsupportedManager) Start(config.Config) error { + return fmt.Errorf("service start is not implemented for %s", runtime.GOOS) +} + +// Stop reports that service stop is unsupported on this platform. +func (unsupportedManager) Stop(config.Config) error { + return fmt.Errorf("service stop is not implemented for %s", runtime.GOOS) +} diff --git a/internal/service/service_windows.go b/internal/service/service_windows.go new file mode 100644 index 0000000..e6e52b6 --- /dev/null +++ b/internal/service/service_windows.go @@ -0,0 +1,43 @@ +//go:build windows + +package service + +import ( + "os/exec" + "strings" + + "mfd-monitor-agent/internal/config" +) + +// windowsManager manages the agent through the Windows Service Control Manager. +type windowsManager struct{} + +// newManager returns the Windows service manager. +func newManager() Manager { + return windowsManager{} +} + +// Install registers the agent executable as an auto-start Windows Service. +func (windowsManager) Install(executable string, cfg config.Config, opts InstallOptions) error { + args := []string{`"` + executable + `"`, "run"} + if opts.ConfigPath != "" { + args = append(args, "-config", `"`+opts.ConfigPath+`"`) + } + return exec.Command("sc.exe", "create", cfg.Service.Name, "binPath=", strings.Join(args, " "), "DisplayName=", cfg.Service.DisplayName, "start=", "auto").Run() +} + +// Uninstall stops and removes the configured Windows Service. +func (windowsManager) Uninstall(cfg config.Config) error { + _ = exec.Command("sc.exe", "stop", cfg.Service.Name).Run() + return exec.Command("sc.exe", "delete", cfg.Service.Name).Run() +} + +// Start asks the Windows Service Control Manager to start the service. +func (windowsManager) Start(cfg config.Config) error { + return exec.Command("sc.exe", "start", cfg.Service.Name).Run() +} + +// Stop asks the Windows Service Control Manager to stop the service. +func (windowsManager) Stop(cfg config.Config) error { + return exec.Command("sc.exe", "stop", cfg.Service.Name).Run() +} diff --git a/internal/snapshot/snapshot.go b/internal/snapshot/snapshot.go new file mode 100644 index 0000000..a9b47b4 --- /dev/null +++ b/internal/snapshot/snapshot.go @@ -0,0 +1,104 @@ +package snapshot + +import ( + "context" + "sync" + "time" + + "mfd-monitor-agent/internal/status" +) + +// Collector gathers one subsection of HostStatus into the provided snapshot. +type Collector interface { + // Name returns the stable collector name used in partial snapshot errors. + Name() string + // Collect fills its subsection of out and returns an error when that subsection is incomplete. + Collect(context.Context, *status.HostStatus) error +} + +// CollectorFunc adapts a function into a Collector for tests and small collectors. +type CollectorFunc func(context.Context, *status.HostStatus) error + +// Name returns a generic collector name for function-backed collectors. +func (fn CollectorFunc) Name() string { + return "collector" +} + +// Collect calls the wrapped function. +func (fn CollectorFunc) Collect(ctx context.Context, out *status.HostStatus) error { + return fn(ctx, out) +} + +// Cache stores the most recent HostStatus snapshot behind a concurrency-safe interface. +type Cache struct { + collectors []Collector + version string + startedAt time.Time + mu sync.RWMutex + last status.HostStatus + hasLast bool +} + +// NewCache creates an empty snapshot cache using the provided collectors and agent metadata. +func NewCache(collectors []Collector, version string, startedAt time.Time) *Cache { + return &Cache{collectors: collectors, version: version, startedAt: startedAt} +} + +// Collect runs all collectors once and stores the resulting snapshot even when some collectors fail. +func (c *Cache) Collect(ctx context.Context) status.HostStatus { + out := status.HostStatus{ + Agent: status.AgentInfo{ + Version: c.version, + StartedAt: c.startedAt, + SnapshotAt: time.Now().UTC(), + SnapshotStatus: "ok", + }, + Processes: status.ProcessResult{Status: "ok"}, + TCPListeners: status.TCPResult{Status: "ok"}, + } + + for _, collector := range c.collectors { + if err := collector.Collect(ctx, &out); err != nil { + out.Agent.SnapshotStatus = "partial" + out.Errors = append(out.Errors, status.StatusError{ + Collector: collector.Name(), + Message: err.Error(), + }) + } + } + + c.mu.Lock() + c.last = out + c.hasLast = true + c.mu.Unlock() + return out +} + +// Get returns the latest cached snapshot and whether collection has completed at least once. +func (c *Cache) Get() (status.HostStatus, bool) { + c.mu.RLock() + defer c.mu.RUnlock() + return c.last, c.hasLast +} + +// Run collects immediately and then refreshes the cache until ctx is canceled. +func (c *Cache) Run(ctx context.Context, interval time.Duration, timeout time.Duration) { + c.collectWithTimeout(ctx, timeout) + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + c.collectWithTimeout(ctx, timeout) + } + } +} + +// collectWithTimeout runs one collection cycle with an overall deadline. +func (c *Cache) collectWithTimeout(ctx context.Context, timeout time.Duration) { + collectCtx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + c.Collect(collectCtx) +} diff --git a/internal/snapshot/snapshot_test.go b/internal/snapshot/snapshot_test.go new file mode 100644 index 0000000..87222f8 --- /dev/null +++ b/internal/snapshot/snapshot_test.go @@ -0,0 +1,103 @@ +package snapshot + +import ( + "context" + "errors" + "testing" + "time" + + "mfd-monitor-agent/internal/status" +) + +// TestCollectorFailuresReturnPartialSnapshot verifies failed collectors do not fail snapshots. +func TestCollectorFailuresReturnPartialSnapshot(t *testing.T) { + cache := NewCache([]Collector{ + CollectorFunc(func(context.Context, *status.HostStatus) error { + return nil + }), + CollectorFunc(func(context.Context, *status.HostStatus) error { + return errors.New("permission denied") + }), + }, "test", time.Now()) + + got := cache.Collect(context.Background()) + + if got.Agent.SnapshotStatus != "partial" { + t.Fatalf("snapshot status = %q", got.Agent.SnapshotStatus) + } + if len(got.Errors) != 1 || got.Errors[0].Message != "permission denied" { + t.Fatalf("errors = %+v", got.Errors) + } +} + +// TestCacheStoresLastSnapshot verifies successful collection is cached. +func TestCacheStoresLastSnapshot(t *testing.T) { + cache := NewCache([]Collector{ + CollectorFunc(func(_ context.Context, out *status.HostStatus) error { + out.Host.Hostname = "host-a" + return nil + }), + }, "test", time.Now()) + + cache.Collect(context.Background()) + got, ok := cache.Get() + + if !ok { + t.Fatal("expected cached snapshot") + } + if got.Host.Hostname != "host-a" { + t.Fatalf("hostname = %q", got.Host.Hostname) + } +} + +// TestRunCollectsImmediately verifies Run collects before the first interval tick. +func TestRunCollectsImmediately(t *testing.T) { + collected := make(chan struct{}) + cache := NewCache([]Collector{ + CollectorFunc(func(_ context.Context, out *status.HostStatus) error { + out.Host.Hostname = "host-a" + close(collected) + return nil + }), + }, "test", time.Now()) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + go cache.Run(ctx, time.Hour, time.Second) + + select { + case <-collected: + case <-time.After(time.Second): + t.Fatal("initial collection did not run") + } + got, ok := cache.Get() + if !ok { + t.Fatal("expected cached snapshot") + } + if got.Host.Hostname != "host-a" { + t.Fatalf("hostname = %q", got.Host.Hostname) + } +} + +// TestCollectWithTimeoutMarksCanceledCollectorPartial verifies timeout errors are partial. +func TestCollectWithTimeoutMarksCanceledCollectorPartial(t *testing.T) { + cache := NewCache([]Collector{ + CollectorFunc(func(ctx context.Context, _ *status.HostStatus) error { + <-ctx.Done() + return ctx.Err() + }), + }, "test", time.Now()) + + cache.collectWithTimeout(context.Background(), time.Millisecond) + got, ok := cache.Get() + + if !ok { + t.Fatal("expected cached snapshot") + } + if got.Agent.SnapshotStatus != "partial" { + t.Fatalf("snapshot status = %q", got.Agent.SnapshotStatus) + } + if len(got.Errors) != 1 || got.Errors[0].Message != context.DeadlineExceeded.Error() { + t.Fatalf("errors = %+v", got.Errors) + } +} diff --git a/internal/status/types.go b/internal/status/types.go new file mode 100644 index 0000000..d7f91c9 --- /dev/null +++ b/internal/status/types.go @@ -0,0 +1,148 @@ +package status + +import "time" + +// HostStatus is the full cached host snapshot returned by /status and posted by the reporter. +type HostStatus struct { + // Agent describes the agent process and most recent snapshot metadata. + Agent AgentInfo `json:"agent"` + Host HostInfo `json:"host"` + Network NetworkInfo `json:"network"` + CPU CPUInfo `json:"cpu"` + Memory MemoryInfo `json:"memory"` + Disks []DiskInfo `json:"disks"` + Processes ProcessResult `json:"processes"` + TCPListeners TCPResult `json:"tcp_listeners"` + // Errors records collector-level failures that caused a partial snapshot. + Errors []StatusError `json:"errors"` +} + +// AgentInfo contains agent process metadata and the latest collection result. +type AgentInfo struct { + Version string `json:"version"` + StartedAt time.Time `json:"started_at"` + SnapshotAt time.Time `json:"snapshot_at"` + // SnapshotStatus is ok when all collectors succeed and partial when one or more fail. + SnapshotStatus string `json:"snapshot_status"` +} + +// HostInfo describes the monitored host identity and operating system. +type HostInfo struct { + HostID string `json:"host_id"` + Hostname string `json:"hostname"` + Platform string `json:"platform"` + PlatformVersion string `json:"platform_version"` + Arch string `json:"arch"` + // UptimeSeconds is the host uptime in seconds when available. + UptimeSeconds uint64 `json:"uptime_seconds"` +} + +// NetworkInfo contains all network interfaces plus flattened IPv4, IPv6, and MAC lists. +type NetworkInfo struct { + Interfaces []NetworkInterface `json:"interfaces"` + IPv4s []string `json:"ipv4s"` + IPv6s []string `json:"ipv6s"` + MACs []string `json:"macs"` +} + +// NetworkInterface describes one host network interface and its configured addresses. +type NetworkInterface struct { + Name string `json:"name"` + MAC string `json:"mac"` + // MTU is the interface maximum transmission unit in bytes. + MTU int `json:"mtu"` + Flags []string `json:"flags"` + Addresses []string `json:"addresses"` +} + +// CPUInfo describes host CPU capacity and current utilization. +type CPUInfo struct { + LogicalCores int `json:"logical_cores"` + PhysicalCores int `json:"physical_cores"` + // UsagePercent is the best-effort host CPU usage percentage for the latest collection. + UsagePercent float64 `json:"usage_percent"` + LoadAvg []float64 `json:"load_avg"` + ModelName string `json:"model_name"` +} + +// MemoryInfo describes host memory usage capacity fields in the declared unit. +type MemoryInfo struct { + // Unit declares the unit used by total, used, free, and available. + Unit string `json:"unit"` + // Total is total physical memory in the declared unit. + Total uint64 `json:"total"` + // Used is memory currently used in the declared unit. + Used uint64 `json:"used"` + // Free is memory currently free in the declared unit. + Free uint64 `json:"free"` + // Available is memory available to new workloads in the declared unit. + Available uint64 `json:"available"` + // UsedPercent is the percentage of total memory currently used. + UsedPercent float64 `json:"used_percent"` +} + +// DiskInfo describes one physical or network mounted filesystem and its capacity. +type DiskInfo struct { + Device string `json:"device"` + MountPoint string `json:"mount_point"` + FSType string `json:"fs_type"` + // Unit declares the unit used by total, used, and free. + Unit string `json:"unit"` + // Total is filesystem capacity in the declared unit. + Total uint64 `json:"total"` + // Used is consumed filesystem capacity in the declared unit. + Used uint64 `json:"used"` + // Free is available filesystem capacity in the declared unit. + Free uint64 `json:"free"` + // UsedPercent is the percentage of total filesystem capacity currently used. + UsedPercent float64 `json:"used_percent"` +} + +// ProcessResult contains process collection status and the collected process list. +type ProcessResult struct { + // Status is ok, partial, or error depending on process collection completeness. + Status string `json:"status"` + Items []ProcessInfo `json:"items"` +} + +// ProcessInfo describes one operating system process. +type ProcessInfo struct { + PID int `json:"pid"` + PPID int `json:"ppid"` + Name string `json:"name"` + Exe string `json:"exe"` + Cmdline string `json:"cmdline"` + Username string `json:"username"` + Status string `json:"status"` + // StartTime is the platform-specific process start timestamp when available. + StartTime int64 `json:"start_time"` + // CPUPercent is the best-effort process CPU usage percentage. + CPUPercent float64 `json:"cpu_percent"` + // MemoryRSS is resident memory size in bytes. + MemoryRSS uint64 `json:"memory_rss"` + // MemoryVMS is virtual memory size in bytes. + MemoryVMS uint64 `json:"memory_vms"` +} + +// TCPResult contains TCP listener collection status and listener entries. +type TCPResult struct { + // Status is ok, partial, or error depending on listener mapping completeness. + Status string `json:"status"` + Items []TCPListenerInfo `json:"items"` +} + +// TCPListenerInfo describes one listening TCP socket and its owning process when known. +type TCPListenerInfo struct { + IP string `json:"ip"` + Port int `json:"port"` + PID int `json:"pid"` + ProcessName string `json:"process_name"` + ProcessExe string `json:"process_exe"` + Family string `json:"family"` +} + +// StatusError records a collector failure without failing the whole snapshot. +type StatusError struct { + Collector string `json:"collector"` + Message string `json:"message"` +} diff --git a/mise.toml b/mise.toml new file mode 100644 index 0000000..b55f8c1 --- /dev/null +++ b/mise.toml @@ -0,0 +1,2 @@ +[tools] +go = "latest" diff --git a/plan.md b/plan.md new file mode 100644 index 0000000..63eb3f2 --- /dev/null +++ b/plan.md @@ -0,0 +1,349 @@ + 高权限主机状态 Agent 设计方案 + + 1. 项目简介 + + MFD Monitor Agent 是一个使用 Go 实现的单二进制主机状态监控 agent，用于在 Windows 和 Linux 主机上采集并暴露完整主机状态。 + + agent 提供本地 HTTP API，默认监听 127.0.0.1:8080。GET /health 用于返回进程存活状态和最近一次快照元数据；GET /status 用于返回缓存中的完整主机状态快照。 + + agent 通过后台周期采集刷新内存缓存，避免每次 HTTP 请求都实时扫描系统，从而提升响应速度并降低系统负载。 + + agent 支持可选的周期性 HTTP POST 主动上报，上报数据直接复用 HostStatus JSON 结构，避免维护两套数据模型。 + + agent 支持通过 CLI 管理服务生命周期，包括 run、install、uninstall、start、stop 和 version。Linux 侧面向 systemd service，Windows 侧面向 Windows Service。 + + 2. 目标 + + 开发一个独立的 Go 单二进制 agent，支持安装为 Windows Service 和 Linux systemd service。 + + 安装阶段要求管理员/root；首版运行阶段也默认使用高权限账户，以稳定采集完整主机信息。 + + agent 对外提供 HTTP 接口，核心接口为 GET /status，单次返回完整主机状态快照，包含： + - IP + - MAC + - 主机名 + - CPU + - 内存 + - 磁盘 + - 完整进程列表 + - 监听中的 TCP 端口及其进程映射 + + 同时 agent 支持按周期向中心端固定 HTTP 接口主动上报同一份状态 JSON。 + + + 3. 服务权限说明 + + 注册为服务这一步基本都需要管理员权限： + + - Windows：创建、删除、启动、停止 Windows Service 需要管理员权限 + - Linux：写入 systemd unit、systemctl enable/start 通常需要 root + + 服务运行阶段： + + - 如果只读取主机名、IP、MAC、内存、磁盘、CPU，通常不一定需要管理员权限 + - 但如果要稳定获取完整进程列表，以及监听 TCP 端口和进程映射，建议服务默认也以高权限运行 + + 建议首版权限策略： + + - 安装阶段：必须管理员/root + - 运行阶段：默认高权限运行 + - 接口层：即使某些字段采集失败，也不要让整个 /status 失败，应返回 partial 或错误原因 + + + 4. 架构方案 + + 推荐使用“单进程一体化”架构。 + + 一个 agent 进程中同时承担以下职责： + + - 周期采集主机状态 + - 缓存最近一次状态 + - 对外提供 HTTP 查询接口 + - 定时向中心端主动上报状态 + - 负责 Windows/Linux 服务生命周期管理 + + 虽然是单进程，但代码建议按模块拆分： + + - collector/system：主机名、OS、架构、启动时间、采集时间 + - collector/network：网卡、IPv4/IPv6、MAC + - collector/resource：CPU、内存、磁盘 + - collector/process：完整进程列表 + - collector/socket：TCP 监听端口及其进程映射 + - snapshot：聚合各采集器输出，形成统一 HostStatus + - httpapi：提供 GET /health、GET /status + - reporter：定时 POST HostStatus 到中心端 + - service：封装 Windows/Linux 服务生命周期和安装命令 + - config：统一加载监听地址、采集周期、上报周期、上报目标、机器标识等配置 + + + 5. HTTP 接口设计 + + 建议保留两个接口： + + - GET /health + 用于返回进程存活、版本、最近快照时间、最近采集状态 + + - GET /status + 返回完整 HostStatus + + 根据当前确认的需求，/status 采用单接口全量返回，不拆分接口，不使用 include 参数。 + + 默认监听地址建议为： + + - 127.0.0.1 + + 如果后续需要局域网访问，再通过配置改为指定内网地址。 + + 原因： + + - /status 会返回较多敏感主机信息 + - 服务以高权限运行时，应尽量缩小默认暴露面 + + + 6. HostStatus 建议字段 + + 建议返回统一 JSON 结构，主要包含以下字段： + + - agent + - version + - started_at + - snapshot_at + - snapshot_status + + - host + - host_id + - hostname + - platform + - platform_version + - arch + - uptime_seconds + + - network + - interfaces[] + - name + - mac + - mtu + - flags + - addresses[] + - ips[] + - macs[] + + - cpu + - logical_cores + - physical_cores + - usage_percent + - load_avg + - model_name + + - memory + - total + - used + - free + - available + - used_percent + + - disks[] + - device + - mount_point + - fs_type + - total + - used + - free + - used_percent + + - processes + - status: ok | partial | error + - items[] + - pid + - ppid + - name + - exe + - cmdline + - username + - status + - start_time + - cpu_percent + - memory_rss + - memory_vms + + - tcp_listeners + - status: ok | partial | error + - items[] + - ip + - port + - pid + - process_name + - process_exe + - family: ipv4 | ipv6 + + - errors[] + 用于记录子采集器失败原因，例如： + - process.cmdline permission denied + + + 7. 采集策略 + + 建议采用“周期采集 + 内存缓存”： + + - 启动后立即采集一次 + - 后台定时刷新缓存 + - HTTP 接口直接读取缓存，不在每次请求时实时扫描系统 + - 主动上报也读取同一份缓存，避免重复采集 + + 这样做的好处： + + - 响应更快 + - 降低系统扫描频率 + - 避免进程列表和端口列表每次请求都重新采集 + + 采集时建议设置： + + - 整体采集超时 + - 单个子采集器失败不拖垮整次快照 + + 例如： + + - 进程采集失败，仍返回 CPU、内存、磁盘等其他信息 + - tcp_listeners 采集不完整时，将状态标记为 partial + + + 8. 主动上报设计 + + agent 除了支持外部系统通过 HTTP 拉取状态，还支持定时主动上报。 + + 建议方式： + + - 配置一个固定中心服务 URL + - 周期性 POST HostStatus JSON + - 上报载荷直接复用 /status 的 JSON 结构 + + 这样可以避免维护两套数据模型。 + + 建议配置项： + + - report.enabled + - report.endpoint + - report.interval + - report.timeout + + 上报失败策略： + + - 超时和失败日志独立记录 + - 支持重试或简单退避 + - 上报失败不能影响本地 HTTP 服务 + - 上报失败不能导致进程退出 + + + 9. CLI 设计 + + 建议提供以下命令： + + - agent run + - agent install + - agent uninstall + - agent start + - agent stop + - agent version + + Linux： + + - agent 进程以前台模式运行 + - 额外提供 systemd unit 模板 + - 安装脚本负责写入 unit 并启停 + + Windows： + + - 使用服务包装库实现 install/uninstall/start/stop/run + - 不依赖外部 NSSM + + + 10. 技术建议 + + 推荐使用 Go 实现，原因： + + - 单二进制，部署简单 + - Windows 和 Linux 跨平台支持较好 + - 服务化实现相对成熟 + - 适合做常驻守护进程和 HTTP 服务 + + 系统信息采集建议优先使用成熟跨平台库，例如： + + - gopsutil + + 这样可以减少自己处理平台差异的工作量。 + + 如库在“端口到进程映射”能力上不足，可以针对 Windows/Linux 做补充实现。 + + + 11. 测试建议 + + 单元测试： + + - 各采集器输出能正确映射为 HostStatus + - 进程列表和 TCP 监听项在空结果、部分字段缺失、权限错误时能输出 partial/error + - snapshot 聚合逻辑在部分采集失败时仍返回可用快照 + + 集成测试： + + - 启动后 GET /health 返回正常 + - GET /status 返回完整 JSON，包含 cpu、processes、tcp_listeners + - 模拟中心端，验证 POST 的载荷、超时、失败重试行为 + + 平台验证： + + - Windows：服务安装、启动、停止、卸载 + - Linux：systemd 安装、enable、start、restart + + 性能与稳定性： + + - 大量进程场景下 /status 仍在可接受时间内返回 + - 采集周期与 HTTP 并发访问同时存在时无明显阻塞 + - /status 响应体较大时内存占用可控 + + + 12. 首版边界 + + 首版默认包括： + + - IP + - MAC + - 主机名 + - CPU + - 内存 + - 磁盘 + - 完整进程列表 + - TCP 监听端口及进程映射 + - 本地 HTTP 查询 + - 主动 HTTP 上报 + - Windows/Linux 服务化 + + 首版默认不包括： + + - 历史存储 + - 趋势分析 + - 分页查询 + - UDP 监听信息 + - 线程列表 + - 进程句柄 + - 连接状态详情 + - 强制 Token/mTLS + + + 13. 结论 + + 最务实的首版实现方案是： + + - 使用 Go + - 单进程一体化架构 + - 安装时管理员/root 权限 + - 运行时默认高权限服务账户 + - GET /status 单接口全量返回 + - 默认监听 127.0.0.1 + - 同时支持被动查询和主动上报 + + 如果进入实现阶段，下一步可以直接开始定义： + + - 配置文件结构 + - HostStatus Go struct + - collector 接口 + - Windows/Linux 服务安装与运行流程 diff --git a/scripts/build-linux.sh b/scripts/build-linux.sh new file mode 100755 index 0000000..066da61 --- /dev/null +++ b/scripts/build-linux.sh @@ -0,0 +1,22 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +OUT_DIR="${OUT_DIR:-${ROOT_DIR}/dist}" +VERSION="${VERSION:-0.1.0}" +GOOS_TARGET="${GOOS_TARGET:-linux}" +GOARCH_TARGET="${GOARCH_TARGET:-amd64}" +OUTPUT_NAME="${OUTPUT_NAME:-agent-${GOOS_TARGET}-${GOARCH_TARGET}}" + +mkdir -p "${OUT_DIR}" + +CGO_ENABLED="${CGO_ENABLED:-0}" \ +GOOS="${GOOS_TARGET}" \ +GOARCH="${GOARCH_TARGET}" \ +go build \ + -trimpath \ + -ldflags "-s -w -X main.version=${VERSION}" \ + -o "${OUT_DIR}/${OUTPUT_NAME}" \ + ./cmd/agent + +printf 'built %s\n' "${OUT_DIR}/${OUTPUT_NAME}" diff --git a/scripts/build-windows.ps1 b/scripts/build-windows.ps1 new file mode 100644 index 0000000..b928ab3 --- /dev/null +++ b/scripts/build-windows.ps1 @@ -0,0 +1,22 @@ +$ErrorActionPreference = "Stop" + +$RootDir = Resolve-Path (Join-Path $PSScriptRoot "..") +$OutDir = if ($env:OUT_DIR) { $env:OUT_DIR } else { Join-Path $RootDir "dist" } +$Version = if ($env:VERSION) { $env:VERSION } else { "0.1.0" } +$GoosTarget = if ($env:GOOS_TARGET) { $env:GOOS_TARGET } else { "windows" } +$GoarchTarget = if ($env:GOARCH_TARGET) { $env:GOARCH_TARGET } else { "amd64" } +$OutputName = if ($env:OUTPUT_NAME) { $env:OUTPUT_NAME } else { "agent-$GoosTarget-$GoarchTarget.exe" } + +New-Item -ItemType Directory -Force -Path $OutDir | Out-Null + +$env:CGO_ENABLED = if ($env:CGO_ENABLED) { $env:CGO_ENABLED } else { "0" } +$env:GOOS = $GoosTarget +$env:GOARCH = $GoarchTarget + +go build ` + -trimpath ` + -ldflags "-s -w -X main.version=$Version" ` + -o (Join-Path $OutDir $OutputName) ` + ./cmd/agent + +Write-Host "built $(Join-Path $OutDir $OutputName)"