mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-09-10 18:45:53 +00:00
926906b134
The Proxmox-side LXC Docker inventory poll occasionally produces an ok=true report with zero CONTAINER lines but a valid PS_OK marker: docker ps -a inside the LXC genuinely returned an empty list for that one call, typically while watchtower is briefly recreating a container or the dockerd is mid-restart. ApplyDockerReport then wipes the host's container list, every connected WebSocket client sees every row for that host flash blank, and the next successful poll 10 s later restores it. Before calling ApplyDockerReport, check whether the host already has containers tracked in state. If yes and the new report has zero containers, log at debug and skip the apply, counting the poll as skipped rather than collected. A host that is genuinely empty still applies on first sight (no previous containers means no previous state to protect). Add regression tests for both branches: a populated host receiving an empty report keeps its container list; a freshly-seen empty host still gets created.
1157 lines
34 KiB
Go
1157 lines
34 KiB
Go
package monitoring
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"math"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/rcourtman/pulse-go-rewrite/internal/models"
|
|
agentsdocker "github.com/rcourtman/pulse-go-rewrite/pkg/agents/docker"
|
|
"github.com/rs/zerolog/log"
|
|
)
|
|
|
|
const proxmoxGuestDockerInventoryMarker = "PULSE_DOCKER_GUEST_INVENTORY_V1"
|
|
|
|
// DockerChecker provides the ability to check for Docker inside LXC containers.
|
|
// This is typically implemented by wrapping the agentexec.Server.
|
|
type DockerChecker interface {
|
|
// CheckDockerInContainer checks if Docker is installed inside an LXC container.
|
|
// Returns true if Docker socket exists, false otherwise.
|
|
// The node parameter is the Proxmox node hostname where the container runs.
|
|
CheckDockerInContainer(ctx context.Context, node string, vmid int) (bool, error)
|
|
}
|
|
|
|
// DockerInventoryCollector provides explicitly opted-in Docker inventory from
|
|
// an LXC guest through the Proxmox node that owns it.
|
|
type DockerInventoryCollector interface {
|
|
// CollectDockerInventory returns a Docker agent-compatible report for the
|
|
// supplied Proxmox LXC container. The bool is false when collection was
|
|
// intentionally skipped, for example because the container is outside the
|
|
// configured VMID allowlist or no Docker runtime is present.
|
|
CollectDockerInventory(ctx context.Context, container models.Container) (agentsdocker.Report, bool, error)
|
|
}
|
|
|
|
// containerDockerCheck represents a container that needs Docker checking
|
|
type containerDockerCheck struct {
|
|
index int
|
|
container models.Container
|
|
reason string // "new", "restarted", "first_check"
|
|
}
|
|
|
|
// containerDockerResult holds the result of a Docker check
|
|
type containerDockerResult struct {
|
|
index int
|
|
hasDocker bool
|
|
checked bool
|
|
err error
|
|
}
|
|
|
|
// CheckContainersForDocker checks Docker presence for containers that need it.
|
|
// This is called during container polling to detect Docker in:
|
|
// - New containers that are running (first time seen)
|
|
// - Containers that have restarted (uptime reset)
|
|
// - Running containers that have never been checked
|
|
//
|
|
// Checks are performed in parallel for efficiency.
|
|
// Returns the containers with updated Docker status.
|
|
func (m *Monitor) CheckContainersForDocker(ctx context.Context, containers []models.Container) []models.Container {
|
|
m.mu.RLock()
|
|
checker := m.dockerChecker
|
|
m.mu.RUnlock()
|
|
|
|
if checker == nil {
|
|
return containers
|
|
}
|
|
|
|
// Get previous container state for comparison
|
|
previousContainers := make(map[string]models.Container)
|
|
for _, ct := range m.state.GetContainers() {
|
|
previousContainers[ct.ID] = ct
|
|
}
|
|
|
|
// Identify containers that need Docker checking
|
|
var needsCheck []containerDockerCheck
|
|
for i, ct := range containers {
|
|
if ct.Status != "running" {
|
|
// Not running - preserve previous Docker status if any
|
|
if prev, ok := previousContainers[ct.ID]; ok {
|
|
containers[i].HasDocker = prev.HasDocker
|
|
containers[i].DockerCheckedAt = prev.DockerCheckedAt
|
|
}
|
|
continue
|
|
}
|
|
|
|
// Check if this container needs Docker detection
|
|
reason := m.containerNeedsDockerCheck(ct, previousContainers)
|
|
if reason != "" {
|
|
needsCheck = append(needsCheck, containerDockerCheck{
|
|
index: i,
|
|
container: ct,
|
|
reason: reason,
|
|
})
|
|
} else {
|
|
// Preserve previous Docker status
|
|
if prev, ok := previousContainers[ct.ID]; ok {
|
|
containers[i].HasDocker = prev.HasDocker
|
|
containers[i].DockerCheckedAt = prev.DockerCheckedAt
|
|
}
|
|
}
|
|
}
|
|
|
|
if len(needsCheck) == 0 {
|
|
return containers
|
|
}
|
|
|
|
// Check Docker in parallel
|
|
results := m.checkDockerParallel(ctx, checker, needsCheck)
|
|
|
|
// Apply results
|
|
checkedCount := 0
|
|
dockerCount := 0
|
|
for _, result := range results {
|
|
if result.checked {
|
|
containers[result.index].HasDocker = result.hasDocker
|
|
containers[result.index].DockerCheckedAt = time.Now()
|
|
checkedCount++
|
|
if result.hasDocker {
|
|
dockerCount++
|
|
}
|
|
} else if result.err != nil {
|
|
// Check failed - preserve previous status if available
|
|
ct := containers[result.index]
|
|
if prev, ok := previousContainers[ct.ID]; ok {
|
|
containers[result.index].HasDocker = prev.HasDocker
|
|
containers[result.index].DockerCheckedAt = prev.DockerCheckedAt
|
|
}
|
|
}
|
|
}
|
|
|
|
if checkedCount > 0 {
|
|
log.Info().
|
|
Int("checked", checkedCount).
|
|
Int("with_docker", dockerCount).
|
|
Int("total_candidates", len(needsCheck)).
|
|
Msg("Docker detection completed for containers")
|
|
}
|
|
|
|
return containers
|
|
}
|
|
|
|
// containerNeedsDockerCheck determines if a running container needs Docker checking.
|
|
// Returns the reason for checking, or empty string if no check is needed.
|
|
func (m *Monitor) containerNeedsDockerCheck(ct models.Container, previousContainers map[string]models.Container) string {
|
|
prev, existed := previousContainers[ct.ID]
|
|
|
|
// New container - never seen before
|
|
if !existed {
|
|
return "new"
|
|
}
|
|
|
|
// Never been checked
|
|
if prev.DockerCheckedAt.IsZero() {
|
|
return "first_check"
|
|
}
|
|
|
|
// Container restarted - uptime is less than before
|
|
// (This catches containers that were stopped and started again)
|
|
if ct.Uptime < prev.Uptime && prev.Uptime > 0 {
|
|
return "restarted"
|
|
}
|
|
|
|
// Container was previously stopped and is now running
|
|
if prev.Status != "running" {
|
|
return "started"
|
|
}
|
|
|
|
// No check needed - use cached value
|
|
return ""
|
|
}
|
|
|
|
// checkDockerParallel checks Docker for multiple containers in parallel
|
|
func (m *Monitor) checkDockerParallel(ctx context.Context, checker DockerChecker, checks []containerDockerCheck) []containerDockerResult {
|
|
results := make([]containerDockerResult, len(checks))
|
|
|
|
// Use a semaphore to limit concurrent checks (avoid overwhelming the system)
|
|
const maxConcurrent = 5
|
|
sem := make(chan struct{}, maxConcurrent)
|
|
var wg sync.WaitGroup
|
|
|
|
for i, check := range checks {
|
|
wg.Add(1)
|
|
go func(idx int, chk containerDockerCheck) {
|
|
defer wg.Done()
|
|
|
|
// Acquire semaphore
|
|
select {
|
|
case sem <- struct{}{}:
|
|
defer func() { <-sem }()
|
|
case <-ctx.Done():
|
|
results[idx] = containerDockerResult{index: chk.index, err: ctx.Err()}
|
|
return
|
|
}
|
|
|
|
// Perform the check
|
|
hasDocker, err := checker.CheckDockerInContainer(ctx, chk.container.Node, chk.container.VMID)
|
|
if err != nil {
|
|
log.Debug().
|
|
Str("container", chk.container.Name).
|
|
Int("vmid", chk.container.VMID).
|
|
Str("reason", chk.reason).
|
|
Err(err).
|
|
Msg("Failed to check Docker in container")
|
|
results[idx] = containerDockerResult{index: chk.index, err: err}
|
|
return
|
|
}
|
|
|
|
log.Debug().
|
|
Str("container", chk.container.Name).
|
|
Int("vmid", chk.container.VMID).
|
|
Str("reason", chk.reason).
|
|
Bool("has_docker", hasDocker).
|
|
Msg("Docker check completed")
|
|
|
|
results[idx] = containerDockerResult{
|
|
index: chk.index,
|
|
hasDocker: hasDocker,
|
|
checked: true,
|
|
}
|
|
}(i, check)
|
|
}
|
|
|
|
wg.Wait()
|
|
return results
|
|
}
|
|
|
|
// CollectProxmoxGuestDockerInventory collects Docker container inventory for
|
|
// running LXC guests that have already opted into Proxmox-side Docker detection
|
|
// and were confirmed to expose a Docker socket.
|
|
func (m *Monitor) CollectProxmoxGuestDockerInventory(ctx context.Context, containers []models.Container) {
|
|
m.mu.RLock()
|
|
collector := m.dockerInventoryCollector
|
|
m.mu.RUnlock()
|
|
|
|
if collector == nil {
|
|
return
|
|
}
|
|
|
|
candidates := make([]models.Container, 0, len(containers))
|
|
for _, ct := range containers {
|
|
if ct.Status != "running" || !ct.HasDocker || ct.VMID <= 0 || strings.TrimSpace(ct.Node) == "" || ct.IsOCI {
|
|
continue
|
|
}
|
|
if m.hasOnlineHostAgentForContainer(ct.ID) {
|
|
log.Debug().
|
|
Str("container", ct.Name).
|
|
Str("containerID", ct.ID).
|
|
Int("vmid", ct.VMID).
|
|
Msg("Skipping Proxmox LXC Docker inventory because a guest-local host agent is linked")
|
|
continue
|
|
}
|
|
candidates = append(candidates, ct)
|
|
}
|
|
|
|
if len(candidates) == 0 {
|
|
return
|
|
}
|
|
|
|
const maxConcurrent = 3
|
|
sem := make(chan struct{}, maxConcurrent)
|
|
var wg sync.WaitGroup
|
|
var mu sync.Mutex
|
|
collected := 0
|
|
failed := 0
|
|
skipped := 0
|
|
|
|
for _, ct := range candidates {
|
|
wg.Add(1)
|
|
go func(container models.Container) {
|
|
defer wg.Done()
|
|
select {
|
|
case sem <- struct{}{}:
|
|
defer func() { <-sem }()
|
|
case <-ctx.Done():
|
|
mu.Lock()
|
|
failed++
|
|
mu.Unlock()
|
|
return
|
|
}
|
|
|
|
report, ok, err := collector.CollectDockerInventory(ctx, container)
|
|
if err != nil {
|
|
log.Debug().
|
|
Err(err).
|
|
Str("container", container.Name).
|
|
Int("vmid", container.VMID).
|
|
Msg("Failed to collect Proxmox LXC Docker inventory")
|
|
mu.Lock()
|
|
failed++
|
|
mu.Unlock()
|
|
return
|
|
}
|
|
if !ok {
|
|
mu.Lock()
|
|
skipped++
|
|
mu.Unlock()
|
|
return
|
|
}
|
|
|
|
enrichGuestDockerReportFromContainer(&report, container)
|
|
|
|
// Refuse to overwrite a known-populated host with an empty
|
|
// container list. docker ps -a inside an LXC occasionally
|
|
// returns zero containers for a single poll (during
|
|
// watchtower-driven container recreate cycles, or while the
|
|
// daemon is restarting), which would otherwise wipe every
|
|
// row for that host on every connected WebSocket client
|
|
// until the next successful poll restores it. A host that
|
|
// genuinely has no containers still applies on first sight,
|
|
// and a host transitioning to zero containers will catch
|
|
// up as soon as we hit the existing offline grace window.
|
|
if len(report.Containers) == 0 && m.dockerHostHasContainers(report.Agent.ID) {
|
|
log.Debug().
|
|
Str("container", container.Name).
|
|
Int("vmid", container.VMID).
|
|
Str("agentID", report.Agent.ID).
|
|
Msg("Skipping Proxmox LXC Docker inventory apply: report has zero containers but host previously had containers (transient docker ps state)")
|
|
mu.Lock()
|
|
skipped++
|
|
mu.Unlock()
|
|
return
|
|
}
|
|
|
|
if _, err := m.ApplyDockerReport(report, nil); err != nil {
|
|
log.Warn().
|
|
Err(err).
|
|
Str("container", container.Name).
|
|
Int("vmid", container.VMID).
|
|
Msg("Failed to apply Proxmox LXC Docker inventory report")
|
|
mu.Lock()
|
|
failed++
|
|
mu.Unlock()
|
|
return
|
|
}
|
|
|
|
mu.Lock()
|
|
collected++
|
|
mu.Unlock()
|
|
}(ct)
|
|
}
|
|
|
|
wg.Wait()
|
|
|
|
if collected > 0 || failed > 0 || skipped > 0 {
|
|
log.Info().
|
|
Int("collected", collected).
|
|
Int("failed", failed).
|
|
Int("skipped", skipped).
|
|
Int("candidates", len(candidates)).
|
|
Msg("Proxmox LXC Docker inventory collection completed")
|
|
}
|
|
}
|
|
|
|
func (m *Monitor) hasOnlineHostAgentForContainer(containerID string) bool {
|
|
containerID = strings.TrimSpace(containerID)
|
|
if m == nil || m.state == nil || containerID == "" {
|
|
return false
|
|
}
|
|
for _, host := range m.state.GetHosts() {
|
|
if strings.TrimSpace(host.LinkedContainerID) == containerID && strings.EqualFold(strings.TrimSpace(host.Status), "online") {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// dockerHostHasContainers reports whether the docker host identified by the
|
|
// given agent ID (or canonical host ID alias) currently has at least one
|
|
// container tracked in state. Used by CollectProxmoxGuestDockerInventory to
|
|
// detect the "previously populated, now reporting zero" pattern that comes
|
|
// from transient docker ps blips inside an LXC and would otherwise wipe
|
|
// container rows for every WebSocket client.
|
|
func (m *Monitor) dockerHostHasContainers(agentID string) bool {
|
|
agentID = strings.TrimSpace(agentID)
|
|
if m == nil || m.state == nil || agentID == "" {
|
|
return false
|
|
}
|
|
for _, host := range m.state.GetDockerHosts() {
|
|
if host.ID == agentID || strings.TrimSpace(host.AgentID) == agentID {
|
|
return len(host.Containers) > 0
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// AgentDockerChecker implements DockerChecker using the agent execution system.
|
|
// This wraps command execution to check for Docker inside LXC containers.
|
|
type AgentDockerChecker struct {
|
|
executeCommand func(ctx context.Context, hostname string, command string, timeout int) (string, int, error)
|
|
}
|
|
|
|
// NewAgentDockerChecker creates a new checker that uses agent command execution.
|
|
// The executeCommand function should execute a command on the given hostname and return
|
|
// (stdout, exitCode, error).
|
|
func NewAgentDockerChecker(executeCommand func(ctx context.Context, hostname string, command string, timeout int) (string, int, error)) *AgentDockerChecker {
|
|
return &AgentDockerChecker{
|
|
executeCommand: executeCommand,
|
|
}
|
|
}
|
|
|
|
// CheckDockerInContainer checks if Docker is installed inside an LXC container
|
|
// by running `pct exec <vmid> -- test -S /var/run/docker.sock` on the Proxmox node.
|
|
func (c *AgentDockerChecker) CheckDockerInContainer(ctx context.Context, node string, vmid int) (bool, error) {
|
|
if c.executeCommand == nil {
|
|
return false, fmt.Errorf("no command executor configured")
|
|
}
|
|
|
|
// Check for Docker socket - this is the most reliable indicator
|
|
// We use test -S which checks if the file exists and is a socket
|
|
cmd := fmt.Sprintf("pct exec %d -- test -S /var/run/docker.sock && echo yes || echo no", vmid)
|
|
|
|
stdout, exitCode, err := c.executeCommand(ctx, node, cmd, 10) // 10 second timeout
|
|
if err != nil {
|
|
return false, fmt.Errorf("command execution failed: %w", err)
|
|
}
|
|
|
|
// The test command itself might fail if container is not accessible
|
|
// but our echo fallback should always give us output
|
|
stdout = strings.TrimSpace(stdout)
|
|
|
|
// If we got "yes", Docker socket exists
|
|
if strings.Contains(stdout, "yes") {
|
|
return true, nil
|
|
}
|
|
|
|
// If exit code is non-zero and we didn't get "no", the container might not be accessible
|
|
if exitCode != 0 && !strings.Contains(stdout, "no") {
|
|
return false, fmt.Errorf("container not accessible (exit code %d): %s", exitCode, stdout)
|
|
}
|
|
|
|
return false, nil
|
|
}
|
|
|
|
// AgentDockerInventoryCollector implements DockerInventoryCollector using the
|
|
// agent execution system. It intentionally gathers only Docker page inventory
|
|
// fields from LXC guests: container ID, name, image, state/status, ports, and
|
|
// aggregate docker stats. It does not call docker inspect and does not collect
|
|
// labels, environment, mounts, commands, files, or process details.
|
|
type AgentDockerInventoryCollector struct {
|
|
executeCommand func(ctx context.Context, hostname string, command string, timeout int) (string, int, error)
|
|
allowedVMIDs map[int]struct{}
|
|
}
|
|
|
|
// AgentDockerInventoryCollectorOptions configures explicitly opted-in
|
|
// Proxmox-side LXC Docker inventory collection.
|
|
type AgentDockerInventoryCollectorOptions struct {
|
|
// AllowedVMIDs limits inventory collection to specific Proxmox VMIDs. An
|
|
// empty map means all running Docker-enabled LXC guests are eligible.
|
|
AllowedVMIDs map[int]struct{}
|
|
}
|
|
|
|
// NewAgentDockerInventoryCollector creates a collector that uses agent command
|
|
// execution on Proxmox nodes to run a minimal read-only Docker inventory script
|
|
// inside LXC guests.
|
|
func NewAgentDockerInventoryCollector(
|
|
executeCommand func(ctx context.Context, hostname string, command string, timeout int) (string, int, error),
|
|
options AgentDockerInventoryCollectorOptions,
|
|
) *AgentDockerInventoryCollector {
|
|
allowed := make(map[int]struct{}, len(options.AllowedVMIDs))
|
|
for vmid := range options.AllowedVMIDs {
|
|
if vmid > 0 {
|
|
allowed[vmid] = struct{}{}
|
|
}
|
|
}
|
|
return &AgentDockerInventoryCollector{
|
|
executeCommand: executeCommand,
|
|
allowedVMIDs: allowed,
|
|
}
|
|
}
|
|
|
|
// CollectDockerInventory collects minimal Docker inventory from an LXC guest.
|
|
func (c *AgentDockerInventoryCollector) CollectDockerInventory(ctx context.Context, container models.Container) (agentsdocker.Report, bool, error) {
|
|
if c == nil || c.executeCommand == nil {
|
|
return agentsdocker.Report{}, false, fmt.Errorf("no command executor configured")
|
|
}
|
|
if container.VMID <= 0 {
|
|
return agentsdocker.Report{}, false, fmt.Errorf("container VMID is required")
|
|
}
|
|
if len(c.allowedVMIDs) > 0 {
|
|
if _, ok := c.allowedVMIDs[container.VMID]; !ok {
|
|
return agentsdocker.Report{}, false, nil
|
|
}
|
|
}
|
|
|
|
command := buildProxmoxGuestDockerInventoryCommand(container.VMID)
|
|
stdout, exitCode, err := c.executeCommand(ctx, container.Node, command, 20)
|
|
if err != nil {
|
|
return agentsdocker.Report{}, false, fmt.Errorf("command execution failed: %w", err)
|
|
}
|
|
if exitCode != 0 && !strings.Contains(stdout, proxmoxGuestDockerInventoryMarker) {
|
|
return agentsdocker.Report{}, false, fmt.Errorf("container docker inventory failed (exit code %d): %s", exitCode, strings.TrimSpace(stdout))
|
|
}
|
|
|
|
report, ok, parseErr := parseProxmoxGuestDockerInventory(stdout, container, time.Now().UTC())
|
|
if parseErr != nil {
|
|
return agentsdocker.Report{}, false, parseErr
|
|
}
|
|
return report, ok, nil
|
|
}
|
|
|
|
func buildProxmoxGuestDockerInventoryCommand(vmid int) string {
|
|
script := strings.Join([]string{
|
|
fmt.Sprintf("printf '%s\\n'", proxmoxGuestDockerInventoryMarker),
|
|
"hn=\"$(hostname -f 2>/dev/null || hostname 2>/dev/null || true)\"",
|
|
"printf 'HOSTNAME\\t%s\\n' \"$hn\"",
|
|
"un=\"$(uname -srm 2>/dev/null || true)\"",
|
|
"printf 'UNAME\\t%s\\n' \"$un\"",
|
|
"cpus=\"$(getconf _NPROCESSORS_ONLN 2>/dev/null || true)\"",
|
|
"printf 'CPUS\\t%s\\n' \"$cpus\"",
|
|
"awk '/MemTotal:/ {printf \"MEMTOTAL\\t%.0f\\n\", $2 * 1024}' /proc/meminfo 2>/dev/null || true",
|
|
"if ! command -v docker >/dev/null 2>&1; then printf 'NO_DOCKER\\n'; exit 0; fi",
|
|
"if ! test -S /var/run/docker.sock; then printf 'NO_DOCKER\\n'; exit 0; fi",
|
|
"version=\"$(docker version --format '{{json .Server.Version}}' 2>/dev/null || true)\"",
|
|
"if [ -n \"$version\" ]; then printf 'VERSION\\t%s\\n' \"$version\"; fi",
|
|
"if docker ps -a --no-trunc --format 'CONTAINER\\t{{json .ID}}\\t{{json .Names}}\\t{{json .Image}}\\t{{json .State}}\\t{{json .Status}}\\t{{json .Ports}}\\t{{json .RunningFor}}' 2>/dev/null; then printf 'PS_OK\\n'; fi",
|
|
"ids=\"$(docker ps -aq --no-trunc 2>/dev/null)\"",
|
|
"if [ -n \"$ids\" ]; then docker stats --no-stream --no-trunc --format 'STAT\\t{{json .ID}}\\t{{json .Name}}\\t{{json .CPUPerc}}\\t{{json .MemUsage}}\\t{{json .MemPerc}}\\t{{json .NetIO}}\\t{{json .BlockIO}}' $ids 2>/dev/null || true; fi",
|
|
}, "\n")
|
|
|
|
return fmt.Sprintf("pct exec %d -- sh -c %s", vmid, shellSingleQuote(script))
|
|
}
|
|
|
|
func shellSingleQuote(value string) string {
|
|
return "'" + strings.ReplaceAll(value, "'", "'\"'\"'") + "'"
|
|
}
|
|
|
|
func parseProxmoxGuestDockerInventory(output string, container models.Container, timestamp time.Time) (agentsdocker.Report, bool, error) {
|
|
if !strings.Contains(output, proxmoxGuestDockerInventoryMarker) {
|
|
return agentsdocker.Report{}, false, fmt.Errorf("docker inventory output missing marker")
|
|
}
|
|
|
|
hostInfo := agentsdocker.HostInfo{
|
|
Hostname: strings.TrimSpace(container.Name),
|
|
Name: proxmoxGuestDockerDisplayName(container),
|
|
Runtime: "docker",
|
|
}
|
|
if hostInfo.Hostname == "" {
|
|
hostInfo.Hostname = fmt.Sprintf("lxc-%d", container.VMID)
|
|
}
|
|
hostInfo.Name = proxmoxGuestDockerDisplayName(container)
|
|
|
|
containers := make([]agentsdocker.Container, 0)
|
|
containerIndex := make(map[string]int)
|
|
noDocker := false
|
|
psOK := false
|
|
|
|
for _, rawLine := range strings.Split(output, "\n") {
|
|
line := strings.TrimRight(rawLine, "\r")
|
|
if line == "" || line == proxmoxGuestDockerInventoryMarker {
|
|
continue
|
|
}
|
|
if line == "NO_DOCKER" {
|
|
noDocker = true
|
|
continue
|
|
}
|
|
if line == "PS_OK" {
|
|
psOK = true
|
|
continue
|
|
}
|
|
|
|
fields := strings.Split(line, "\t")
|
|
if len(fields) == 1 && strings.HasPrefix(line, "VERSION\\t") {
|
|
fields = []string{"VERSION", strings.TrimPrefix(line, "VERSION\\t")}
|
|
}
|
|
switch fields[0] {
|
|
case "HOSTNAME":
|
|
if len(fields) >= 2 {
|
|
if hostname := strings.TrimSpace(fields[1]); hostname != "" {
|
|
hostInfo.Hostname = hostname
|
|
}
|
|
}
|
|
case "UNAME":
|
|
if len(fields) >= 2 {
|
|
applyUnameToDockerHostInfo(fields[1], &hostInfo)
|
|
}
|
|
case "CPUS":
|
|
if len(fields) >= 2 {
|
|
if cpus, err := strconv.Atoi(strings.TrimSpace(fields[1])); err == nil && cpus > 0 {
|
|
hostInfo.TotalCPU = cpus
|
|
}
|
|
}
|
|
case "MEMTOTAL":
|
|
if len(fields) >= 2 {
|
|
if total, err := strconv.ParseInt(strings.TrimSpace(fields[1]), 10, 64); err == nil && total > 0 {
|
|
hostInfo.TotalMemoryBytes = total
|
|
hostInfo.Memory.TotalBytes = total
|
|
}
|
|
}
|
|
case "VERSION":
|
|
if len(fields) >= 2 {
|
|
version := decodeInventoryJSONString(fields[1])
|
|
hostInfo.RuntimeVersion = version
|
|
hostInfo.DockerVersion = version
|
|
}
|
|
case "CONTAINER":
|
|
payload, ok := parseDockerInventoryContainerLine(fields)
|
|
if !ok {
|
|
continue
|
|
}
|
|
containerIndex[payload.ID] = len(containers)
|
|
if payload.Name != "" {
|
|
containerIndex[payload.Name] = len(containers)
|
|
}
|
|
containers = append(containers, payload)
|
|
case "STAT":
|
|
stat := parseDockerInventoryStatLine(fields)
|
|
if stat.id == "" && stat.name == "" {
|
|
continue
|
|
}
|
|
idx, ok := findDockerInventoryContainerIndex(containerIndex, containers, stat.id, stat.name)
|
|
if !ok {
|
|
continue
|
|
}
|
|
applyDockerInventoryStat(&containers[idx], stat)
|
|
}
|
|
}
|
|
|
|
if noDocker && len(containers) == 0 && hostInfo.DockerVersion == "" {
|
|
return agentsdocker.Report{}, false, nil
|
|
}
|
|
if !noDocker && !psOK {
|
|
return agentsdocker.Report{}, false, fmt.Errorf("docker ps did not complete (no PS_OK marker); refusing to apply partial inventory for %q", strings.TrimSpace(container.Name))
|
|
}
|
|
if timestamp.IsZero() {
|
|
timestamp = time.Now().UTC()
|
|
}
|
|
|
|
return agentsdocker.Report{
|
|
Agent: agentsdocker.AgentInfo{
|
|
ID: proxmoxGuestDockerAgentID(container),
|
|
Type: "unified",
|
|
IntervalSeconds: 30,
|
|
},
|
|
Host: hostInfo,
|
|
Containers: containers,
|
|
Timestamp: timestamp,
|
|
}, true, nil
|
|
}
|
|
|
|
func proxmoxGuestDockerAgentID(container models.Container) string {
|
|
id := strings.TrimSpace(container.ID)
|
|
if id == "" {
|
|
id = fmt.Sprintf("%s:%s:%d", strings.TrimSpace(container.Instance), strings.TrimSpace(container.Node), container.VMID)
|
|
}
|
|
id = strings.Trim(id, ":")
|
|
if id == "" {
|
|
id = fmt.Sprintf("lxc:%d", container.VMID)
|
|
}
|
|
return "proxmox-lxc-docker:" + id
|
|
}
|
|
|
|
func proxmoxGuestDockerDisplayName(container models.Container) string {
|
|
name := strings.TrimSpace(container.Name)
|
|
if name != "" {
|
|
return name
|
|
}
|
|
return fmt.Sprintf("LXC %d", container.VMID)
|
|
}
|
|
|
|
// enrichGuestDockerReportFromContainer copies CPU / memory / disk telemetry
|
|
// from the underlying Proxmox LXC into the Docker host info before the report
|
|
// is applied. The Docker inventory script collects only sizing metadata
|
|
// (CPU count, MEMTOTAL); the LXC's live resource usage is already polled via
|
|
// the PVE cluster/resources endpoint, so the Docker host row should show
|
|
// what the rest of Pulse already knows about that guest. Only zero-valued
|
|
// fields are populated so values reported by the inventory script itself
|
|
// (when present in future revisions) still take precedence.
|
|
func enrichGuestDockerReportFromContainer(report *agentsdocker.Report, container models.Container) {
|
|
if report == nil {
|
|
return
|
|
}
|
|
if report.Host.CPUUsagePercent == 0 && container.CPU > 0 {
|
|
report.Host.CPUUsagePercent = container.CPU * 100
|
|
}
|
|
if report.Host.Memory.TotalBytes == 0 && container.Memory.Total > 0 {
|
|
report.Host.Memory.TotalBytes = container.Memory.Total
|
|
}
|
|
if report.Host.Memory.UsedBytes == 0 && container.Memory.Used > 0 {
|
|
report.Host.Memory.UsedBytes = container.Memory.Used
|
|
}
|
|
if report.Host.Memory.FreeBytes == 0 && container.Memory.Free > 0 {
|
|
report.Host.Memory.FreeBytes = container.Memory.Free
|
|
}
|
|
if report.Host.Memory.Usage == 0 && container.Memory.Usage > 0 {
|
|
report.Host.Memory.Usage = container.Memory.Usage
|
|
}
|
|
if report.Host.TotalMemoryBytes == 0 && container.Memory.Total > 0 {
|
|
report.Host.TotalMemoryBytes = container.Memory.Total
|
|
}
|
|
if len(report.Host.Disks) == 0 && container.Disk.Total > 0 {
|
|
report.Host.Disks = []agentsdocker.Disk{{
|
|
Mountpoint: "/",
|
|
TotalBytes: container.Disk.Total,
|
|
UsedBytes: container.Disk.Used,
|
|
FreeBytes: container.Disk.Free,
|
|
Usage: container.Disk.Usage,
|
|
}}
|
|
}
|
|
if report.Host.UptimeSeconds == 0 && container.Uptime > 0 {
|
|
report.Host.UptimeSeconds = container.Uptime
|
|
}
|
|
}
|
|
|
|
func applyUnameToDockerHostInfo(uname string, host *agentsdocker.HostInfo) {
|
|
if host == nil {
|
|
return
|
|
}
|
|
parts := strings.Fields(uname)
|
|
if len(parts) == 0 {
|
|
return
|
|
}
|
|
host.OS = strings.ToLower(parts[0])
|
|
if len(parts) >= 2 {
|
|
host.KernelVersion = parts[1]
|
|
}
|
|
if len(parts) >= 3 {
|
|
host.Architecture = parts[len(parts)-1]
|
|
}
|
|
}
|
|
|
|
func parseDockerInventoryContainerLine(fields []string) (agentsdocker.Container, bool) {
|
|
if len(fields) < 8 {
|
|
return agentsdocker.Container{}, false
|
|
}
|
|
id := strings.TrimSpace(decodeInventoryJSONString(fields[1]))
|
|
name := strings.TrimSpace(decodeInventoryJSONString(fields[2]))
|
|
if id == "" && name == "" {
|
|
return agentsdocker.Container{}, false
|
|
}
|
|
state := strings.TrimSpace(decodeInventoryJSONString(fields[4]))
|
|
status := strings.TrimSpace(decodeInventoryJSONString(fields[5]))
|
|
if state == "" {
|
|
state = dockerStateFromStatus(status)
|
|
}
|
|
if id == "" {
|
|
id = name
|
|
}
|
|
|
|
return agentsdocker.Container{
|
|
ID: id,
|
|
Name: strings.TrimPrefix(name, "/"),
|
|
Image: strings.TrimSpace(decodeInventoryJSONString(fields[3])),
|
|
State: state,
|
|
Status: status,
|
|
UptimeSeconds: parseDockerStatusUptime(status),
|
|
Ports: parseDockerPorts(decodeInventoryJSONString(fields[6])),
|
|
}, true
|
|
}
|
|
|
|
type dockerInventoryStat struct {
|
|
id string
|
|
name string
|
|
cpuPercent float64
|
|
memUsage int64
|
|
memLimit int64
|
|
memPercent float64
|
|
networkRX uint64
|
|
networkTX uint64
|
|
blockRead uint64
|
|
blockWrite uint64
|
|
hasNetworkIO bool
|
|
hasBlockIO bool
|
|
}
|
|
|
|
func parseDockerInventoryStatLine(fields []string) dockerInventoryStat {
|
|
if len(fields) < 8 {
|
|
return dockerInventoryStat{}
|
|
}
|
|
stat := dockerInventoryStat{
|
|
id: strings.TrimSpace(decodeInventoryJSONString(fields[1])),
|
|
name: strings.TrimSpace(decodeInventoryJSONString(fields[2])),
|
|
cpuPercent: parsePercent(decodeInventoryJSONString(fields[3])),
|
|
memPercent: parsePercent(decodeInventoryJSONString(fields[5])),
|
|
}
|
|
stat.memUsage, stat.memLimit = parseDockerSizePair(decodeInventoryJSONString(fields[4]))
|
|
if rx, tx, ok := parseDockerUintPair(decodeInventoryJSONString(fields[6])); ok {
|
|
stat.networkRX = rx
|
|
stat.networkTX = tx
|
|
stat.hasNetworkIO = true
|
|
}
|
|
if read, write, ok := parseDockerUintPair(decodeInventoryJSONString(fields[7])); ok {
|
|
stat.blockRead = read
|
|
stat.blockWrite = write
|
|
stat.hasBlockIO = true
|
|
}
|
|
return stat
|
|
}
|
|
|
|
func applyDockerInventoryStat(container *agentsdocker.Container, stat dockerInventoryStat) {
|
|
if container == nil {
|
|
return
|
|
}
|
|
container.CPUPercent = stat.cpuPercent
|
|
container.MemoryUsageBytes = stat.memUsage
|
|
container.MemoryLimitBytes = stat.memLimit
|
|
container.MemoryPercent = stat.memPercent
|
|
if stat.hasNetworkIO {
|
|
container.NetworkRXBytes = stat.networkRX
|
|
container.NetworkTXBytes = stat.networkTX
|
|
}
|
|
if stat.hasBlockIO {
|
|
container.BlockIO = &agentsdocker.ContainerBlockIO{
|
|
ReadBytes: stat.blockRead,
|
|
WriteBytes: stat.blockWrite,
|
|
}
|
|
}
|
|
}
|
|
|
|
func findDockerInventoryContainerIndex(index map[string]int, containers []agentsdocker.Container, id, name string) (int, bool) {
|
|
for _, key := range []string{strings.TrimSpace(id), strings.TrimSpace(name)} {
|
|
if key == "" {
|
|
continue
|
|
}
|
|
if idx, ok := index[key]; ok {
|
|
return idx, true
|
|
}
|
|
}
|
|
|
|
if id != "" {
|
|
matched := -1
|
|
for i, container := range containers {
|
|
if strings.HasPrefix(container.ID, id) || strings.HasPrefix(id, container.ID) {
|
|
if matched >= 0 {
|
|
return 0, false
|
|
}
|
|
matched = i
|
|
}
|
|
}
|
|
if matched >= 0 {
|
|
return matched, true
|
|
}
|
|
}
|
|
|
|
return 0, false
|
|
}
|
|
|
|
func decodeInventoryJSONString(raw string) string {
|
|
raw = strings.TrimSpace(raw)
|
|
if raw == "" {
|
|
return ""
|
|
}
|
|
var value string
|
|
if err := json.Unmarshal([]byte(raw), &value); err == nil {
|
|
return value
|
|
}
|
|
return strings.Trim(raw, `"`)
|
|
}
|
|
|
|
func dockerStateFromStatus(status string) string {
|
|
normalized := strings.ToLower(strings.TrimSpace(status))
|
|
switch {
|
|
case strings.HasPrefix(normalized, "up "):
|
|
return "running"
|
|
case strings.HasPrefix(normalized, "exited"):
|
|
return "exited"
|
|
case strings.HasPrefix(normalized, "created"):
|
|
return "created"
|
|
default:
|
|
return normalized
|
|
}
|
|
}
|
|
|
|
func parseDockerStatusUptime(status string) int64 {
|
|
normalized := strings.ToLower(strings.TrimSpace(status))
|
|
if !strings.HasPrefix(normalized, "up ") {
|
|
return 0
|
|
}
|
|
normalized = strings.TrimSpace(strings.TrimPrefix(normalized, "up "))
|
|
normalized = strings.TrimPrefix(normalized, "about ")
|
|
if strings.HasPrefix(normalized, "less than a second") {
|
|
return 1
|
|
}
|
|
if strings.HasPrefix(normalized, "an ") {
|
|
normalized = "1 " + strings.TrimPrefix(normalized, "an ")
|
|
}
|
|
fields := strings.Fields(normalized)
|
|
if len(fields) < 2 {
|
|
return 0
|
|
}
|
|
amount, err := strconv.ParseInt(fields[0], 10, 64)
|
|
if err != nil || amount < 0 {
|
|
return 0
|
|
}
|
|
unit := strings.TrimSuffix(fields[1], "s")
|
|
switch unit {
|
|
case "second":
|
|
return amount
|
|
case "minute":
|
|
return amount * 60
|
|
case "hour":
|
|
return amount * 60 * 60
|
|
case "day":
|
|
return amount * 24 * 60 * 60
|
|
case "week":
|
|
return amount * 7 * 24 * 60 * 60
|
|
case "month":
|
|
return amount * 30 * 24 * 60 * 60
|
|
case "year":
|
|
return amount * 365 * 24 * 60 * 60
|
|
default:
|
|
return 0
|
|
}
|
|
}
|
|
|
|
func parsePercent(raw string) float64 {
|
|
raw = strings.TrimSpace(strings.TrimSuffix(raw, "%"))
|
|
if raw == "" || raw == "--" {
|
|
return 0
|
|
}
|
|
value, err := strconv.ParseFloat(raw, 64)
|
|
if err != nil || math.IsNaN(value) || math.IsInf(value, 0) {
|
|
return 0
|
|
}
|
|
return value
|
|
}
|
|
|
|
func parseDockerSizePair(raw string) (int64, int64) {
|
|
left, right, ok := splitSlashPair(raw)
|
|
if !ok {
|
|
return 0, 0
|
|
}
|
|
used, _ := parseDockerByteSize(left)
|
|
limit, _ := parseDockerByteSize(right)
|
|
return used, limit
|
|
}
|
|
|
|
func parseDockerUintPair(raw string) (uint64, uint64, bool) {
|
|
left, right, ok := splitSlashPair(raw)
|
|
if !ok {
|
|
return 0, 0, false
|
|
}
|
|
leftBytes, leftOK := parseDockerByteSize(left)
|
|
rightBytes, rightOK := parseDockerByteSize(right)
|
|
if !leftOK && !rightOK {
|
|
return 0, 0, false
|
|
}
|
|
return uint64(maxInt64(leftBytes, 0)), uint64(maxInt64(rightBytes, 0)), true
|
|
}
|
|
|
|
func splitSlashPair(raw string) (string, string, bool) {
|
|
parts := strings.Split(raw, "/")
|
|
if len(parts) != 2 {
|
|
return "", "", false
|
|
}
|
|
return strings.TrimSpace(parts[0]), strings.TrimSpace(parts[1]), true
|
|
}
|
|
|
|
func parseDockerByteSize(raw string) (int64, bool) {
|
|
raw = strings.TrimSpace(raw)
|
|
if raw == "" || raw == "--" {
|
|
return 0, false
|
|
}
|
|
raw = strings.ReplaceAll(raw, " ", "")
|
|
i := 0
|
|
for i < len(raw) {
|
|
ch := raw[i]
|
|
if (ch >= '0' && ch <= '9') || ch == '.' {
|
|
i++
|
|
continue
|
|
}
|
|
break
|
|
}
|
|
if i == 0 {
|
|
return 0, false
|
|
}
|
|
value, err := strconv.ParseFloat(raw[:i], 64)
|
|
if err != nil || math.IsNaN(value) || math.IsInf(value, 0) {
|
|
return 0, false
|
|
}
|
|
unit := strings.ToLower(strings.TrimSpace(raw[i:]))
|
|
multiplier := float64(1)
|
|
switch unit {
|
|
case "", "b":
|
|
multiplier = 1
|
|
case "kb":
|
|
multiplier = 1000
|
|
case "kib":
|
|
multiplier = 1024
|
|
case "mb":
|
|
multiplier = 1000 * 1000
|
|
case "mib":
|
|
multiplier = 1024 * 1024
|
|
case "gb":
|
|
multiplier = 1000 * 1000 * 1000
|
|
case "gib":
|
|
multiplier = 1024 * 1024 * 1024
|
|
case "tb":
|
|
multiplier = 1000 * 1000 * 1000 * 1000
|
|
case "tib":
|
|
multiplier = 1024 * 1024 * 1024 * 1024
|
|
case "pb":
|
|
multiplier = 1000 * 1000 * 1000 * 1000 * 1000
|
|
case "pib":
|
|
multiplier = 1024 * 1024 * 1024 * 1024 * 1024
|
|
default:
|
|
return 0, false
|
|
}
|
|
result := value * multiplier
|
|
if result < 0 {
|
|
return 0, false
|
|
}
|
|
if result > float64(math.MaxInt64) {
|
|
return math.MaxInt64, true
|
|
}
|
|
return int64(math.Round(result)), true
|
|
}
|
|
|
|
func parseDockerPorts(raw string) []agentsdocker.ContainerPort {
|
|
raw = strings.TrimSpace(raw)
|
|
if raw == "" {
|
|
return nil
|
|
}
|
|
segments := strings.Split(raw, ",")
|
|
ports := make([]agentsdocker.ContainerPort, 0, len(segments))
|
|
for _, segment := range segments {
|
|
segment = strings.TrimSpace(segment)
|
|
if segment == "" {
|
|
continue
|
|
}
|
|
if strings.Contains(segment, "->") {
|
|
parts := strings.SplitN(segment, "->", 2)
|
|
private, protocol, ok := parseDockerPortSpec(parts[1])
|
|
if !ok {
|
|
continue
|
|
}
|
|
ip, publicPort := parseDockerPublishedPort(parts[0])
|
|
ports = append(ports, agentsdocker.ContainerPort{
|
|
PrivatePort: private,
|
|
PublicPort: publicPort,
|
|
Protocol: protocol,
|
|
IP: ip,
|
|
})
|
|
continue
|
|
}
|
|
private, protocol, ok := parseDockerPortSpec(segment)
|
|
if !ok {
|
|
continue
|
|
}
|
|
ports = append(ports, agentsdocker.ContainerPort{
|
|
PrivatePort: private,
|
|
Protocol: protocol,
|
|
})
|
|
}
|
|
if len(ports) == 0 {
|
|
return nil
|
|
}
|
|
return ports
|
|
}
|
|
|
|
func parseDockerPortSpec(raw string) (int, string, bool) {
|
|
raw = strings.TrimSpace(raw)
|
|
if raw == "" {
|
|
return 0, "", false
|
|
}
|
|
protocol := "tcp"
|
|
if before, after, ok := strings.Cut(raw, "/"); ok {
|
|
raw = before
|
|
if strings.TrimSpace(after) != "" {
|
|
protocol = strings.TrimSpace(after)
|
|
}
|
|
}
|
|
if before, _, ok := strings.Cut(raw, "-"); ok {
|
|
raw = before
|
|
}
|
|
port, err := strconv.Atoi(strings.TrimSpace(raw))
|
|
if err != nil || port <= 0 {
|
|
return 0, "", false
|
|
}
|
|
return port, protocol, true
|
|
}
|
|
|
|
func parseDockerPublishedPort(raw string) (string, int) {
|
|
raw = strings.TrimSpace(raw)
|
|
if raw == "" {
|
|
return "", 0
|
|
}
|
|
portPart := raw
|
|
ip := ""
|
|
if idx := strings.LastIndex(raw, ":"); idx >= 0 {
|
|
ip = strings.Trim(raw[:idx], "[]")
|
|
portPart = raw[idx+1:]
|
|
}
|
|
if before, _, ok := strings.Cut(portPart, "-"); ok {
|
|
portPart = before
|
|
}
|
|
port, err := strconv.Atoi(strings.TrimSpace(portPart))
|
|
if err != nil || port <= 0 {
|
|
return ip, 0
|
|
}
|
|
return ip, port
|
|
}
|
|
|
|
// ParseProxmoxGuestDockerInventoryVMIDs parses a comma-separated VMID allowlist.
|
|
// Empty input means all running Docker-enabled LXC guests are eligible.
|
|
func ParseProxmoxGuestDockerInventoryVMIDs(raw string) (map[int]struct{}, []string) {
|
|
allowed := make(map[int]struct{})
|
|
invalid := make([]string, 0)
|
|
for _, part := range strings.Split(raw, ",") {
|
|
part = strings.TrimSpace(part)
|
|
if part == "" {
|
|
continue
|
|
}
|
|
vmid, err := strconv.Atoi(part)
|
|
if err != nil || vmid <= 0 {
|
|
invalid = append(invalid, part)
|
|
continue
|
|
}
|
|
allowed[vmid] = struct{}{}
|
|
}
|
|
return allowed, invalid
|
|
}
|
|
|
|
// SetDockerChecker configures Docker detection for LXC containers.
|
|
// When set, Docker presence will be automatically detected during container polling
|
|
// for new containers and containers that have restarted.
|
|
func (m *Monitor) SetDockerChecker(checker DockerChecker) {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
m.dockerChecker = checker
|
|
if checker != nil {
|
|
log.Info().Msg("Docker detection enabled for LXC containers")
|
|
} else {
|
|
log.Info().Msg("Docker detection disabled for LXC containers")
|
|
}
|
|
}
|
|
|
|
// GetDockerChecker returns the current Docker checker, if configured.
|
|
func (m *Monitor) GetDockerChecker() DockerChecker {
|
|
m.mu.RLock()
|
|
defer m.mu.RUnlock()
|
|
return m.dockerChecker
|
|
}
|
|
|
|
// SetDockerInventoryCollector configures explicitly opted-in Docker inventory
|
|
// collection for Docker-enabled LXC containers.
|
|
func (m *Monitor) SetDockerInventoryCollector(collector DockerInventoryCollector) {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
m.dockerInventoryCollector = collector
|
|
if collector != nil {
|
|
log.Info().Msg("Docker inventory collection enabled for LXC containers")
|
|
} else {
|
|
log.Info().Msg("Docker inventory collection disabled for LXC containers")
|
|
}
|
|
}
|
|
|
|
// GetDockerInventoryCollector returns the current Docker inventory collector,
|
|
// if configured.
|
|
func (m *Monitor) GetDockerInventoryCollector() DockerInventoryCollector {
|
|
m.mu.RLock()
|
|
defer m.mu.RUnlock()
|
|
return m.dockerInventoryCollector
|
|
}
|