gsadmin/pulse
1
0
Fork 0
mirror of https://github.com/rcourtman/Pulse.git synced 2026-09-10 10:35:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

Remove deprecated aidiscovery package

Browse Source 
The aidiscovery package has been superseded by the consolidated
tools approach in internal/ai/tools/. Discovery functionality is
now handled through:

- pulse_query tool for resource search and discovery
- pulse_discovery tool for infrastructure scanning
- Better integration with the main AI chat pipeline

Removing:
- commands.go and related tests
- deep_scanner.go and tests
- formatters.go and tests
- service.go and tests
- store.go and tests
- tools_adapter.go
- types.go and tests
This commit is contained in:
rcourtman
2026-01-28 16:52:17 +00:00
parent e194e17159
commit c92811f3b2
13 changed files with 0 additions and 4420 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/aidiscovery/commands.go
-442
View File
@@ -1,442 +0,0 @@
package aidiscovery
import (
"fmt"
"strings"
)
// DiscoveryCommand represents a command to run during discovery.
type DiscoveryCommand struct {
Name string // Human-readable name
Command string // The command template
Description string // What this discovers
Categories []string // What categories of info this provides
Timeout int // Timeout in seconds (0 = default)
Optional bool // If true, don't fail if command fails
}
// CommandSet represents a set of commands for a resource type.
type CommandSet struct {
ResourceType ResourceType
Commands []DiscoveryCommand
}
// GetCommandsForResource returns the commands to run for a given resource type.
func GetCommandsForResource(resourceType ResourceType) []DiscoveryCommand {
switch resourceType {
case ResourceTypeLXC:
return getLXCCommands()
case ResourceTypeVM:
return getVMCommands()
case ResourceTypeDocker:
return getDockerCommands()
case ResourceTypeDockerVM, ResourceTypeDockerLXC:
return getNestedDockerCommands()
case ResourceTypeK8s:
return getK8sCommands()
case ResourceTypeHost:
return getHostCommands()
default:
return []DiscoveryCommand{}
}
}
// getLXCCommands returns commands for discovering LXC containers.
func getLXCCommands() []DiscoveryCommand {
return []DiscoveryCommand{
{
Name: "os_release",
Command: "cat /etc/os-release",
Description: "Operating system identification",
Categories: []string{"version", "config"},
Optional: true,
},
{
Name: "hostname",
Command: "hostname",
Description: "Container hostname",
Categories: []string{"config"},
Optional: true,
},
{
Name: "running_services",
Command: "systemctl list-units --type=service --state=running --no-pager 2>/dev/null | head -30 || service --status-all 2>/dev/null | grep '+' | head -30",
Description: "Running services and daemons",
Categories: []string{"service"},
Optional: true,
},
{
Name: "listening_ports",
Command: "ss -tlnp 2>/dev/null | head -25 || netstat -tlnp 2>/dev/null | head -25",
Description: "Network ports listening",
Categories: []string{"port", "network"},
Optional: true,
},
{
Name: "top_processes",
Command: "ps aux --sort=-rss 2>/dev/null | head -15 || ps aux | head -15",
Description: "Top processes by memory",
Categories: []string{"service"},
Optional: true,
},
{
Name: "disk_usage",
Command: "df -h 2>/dev/null | head -15",
Description: "Disk usage and mount points",
Categories: []string{"storage"},
Optional: true,
},
{
Name: "docker_check",
Command: "docker ps --format '{{.Names}}: {{.Image}} ({{.Status}})' 2>/dev/null | head -20 || echo 'no_docker'",
Description: "Docker containers if running",
Categories: []string{"service", "container"},
Optional: true,
},
{
Name: "installed_packages",
Command: "dpkg -l 2>/dev/null | grep -E '^ii' | awk '{print $2}' | head -50 || rpm -qa 2>/dev/null | head -50 || apk list --installed 2>/dev/null | head -50",
Description: "Installed packages",
Categories: []string{"version", "service"},
Optional: true,
},
{
Name: "config_files",
Command: "find /etc -name '*.conf' -o -name '*.yml' -o -name '*.yaml' -o -name '*.json' 2>/dev/null | head -30",
Description: "Configuration files",
Categories: []string{"config"},
Optional: true,
},
{
Name: "cron_jobs",
Command: "crontab -l 2>/dev/null | grep -v '^#' | head -10 || ls -la /etc/cron.d/ 2>/dev/null | head -10",
Description: "Scheduled jobs",
Categories: []string{"service"},
Optional: true,
},
{
Name: "hardware_info",
Command: "lspci 2>/dev/null | head -20 || echo 'no_lspci'",
Description: "Hardware devices (e.g., Coral TPU)",
Categories: []string{"hardware"},
Optional: true,
},
{
Name: "gpu_devices",
Command: "ls -la /dev/dri/ 2>/dev/null; ls -la /dev/apex* 2>/dev/null; nvidia-smi -L 2>/dev/null || echo 'no_gpu'",
Description: "GPU and TPU devices",
Categories: []string{"hardware"},
Optional: true,
},
}
}
// getVMCommands returns commands for discovering VMs (via QEMU guest agent).
func getVMCommands() []DiscoveryCommand {
return []DiscoveryCommand{
{
Name: "os_release",
Command: "cat /etc/os-release",
Description: "Operating system identification",
Categories: []string{"version", "config"},
Optional: true,
},
{
Name: "hostname",
Command: "hostname",
Description: "VM hostname",
Categories: []string{"config"},
Optional: true,
},
{
Name: "running_services",
Command: "systemctl list-units --type=service --state=running --no-pager 2>/dev/null | head -30",
Description: "Running services and daemons",
Categories: []string{"service"},
Optional: true,
},
{
Name: "listening_ports",
Command: "ss -tlnp 2>/dev/null | head -25 || netstat -tlnp 2>/dev/null | head -25",
Description: "Network ports listening",
Categories: []string{"port", "network"},
Optional: true,
},
{
Name: "top_processes",
Command: "ps aux --sort=-rss 2>/dev/null | head -15",
Description: "Top processes by memory",
Categories: []string{"service"},
Optional: true,
},
{
Name: "disk_usage",
Command: "df -h 2>/dev/null | head -15",
Description: "Disk usage and mount points",
Categories: []string{"storage"},
Optional: true,
},
{
Name: "docker_check",
Command: "docker ps --format '{{.Names}}: {{.Image}} ({{.Status}})' 2>/dev/null | head -20 || echo 'no_docker'",
Description: "Docker containers if running",
Categories: []string{"service", "container"},
Optional: true,
},
{
Name: "hardware_info",
Command: "lspci 2>/dev/null | head -20",
Description: "PCI hardware devices",
Categories: []string{"hardware"},
Optional: true,
},
{
Name: "gpu_devices",
Command: "ls -la /dev/dri/ 2>/dev/null; nvidia-smi -L 2>/dev/null || echo 'no_gpu'",
Description: "GPU devices",
Categories: []string{"hardware"},
Optional: true,
},
}
}
// getDockerCommands returns commands for discovering Docker containers.
// These are run inside the container via docker exec.
func getDockerCommands() []DiscoveryCommand {
return []DiscoveryCommand{
{
Name: "os_release",
Command: "cat /etc/os-release 2>/dev/null || cat /etc/alpine-release 2>/dev/null || echo 'unknown'",
Description: "Container OS",
Categories: []string{"version"},
Optional: true,
},
{
Name: "processes",
Command: "ps aux 2>/dev/null || echo 'no_ps'",
Description: "Running processes",
Categories: []string{"service"},
Optional: true,
},
{
Name: "listening_ports",
Command: "ss -tlnp 2>/dev/null || netstat -tlnp 2>/dev/null || echo 'no_ss'",
Description: "Listening ports inside container",
Categories: []string{"port"},
Optional: true,
},
{
Name: "env_vars",
Command: "env 2>/dev/null | grep -vE '(PASSWORD|SECRET|KEY|TOKEN|CREDENTIAL)' | head -30",
Description: "Environment variables (filtered)",
Categories: []string{"config"},
Optional: true,
},
{
Name: "config_files",
Command: "find /config /data /app /etc -maxdepth 2 -name '*.conf' -o -name '*.yml' -o -name '*.yaml' -o -name '*.json' 2>/dev/null | head -20",
Description: "Configuration files",
Categories: []string{"config"},
Optional: true,
},
}
}
// getNestedDockerCommands returns commands for Docker inside VMs or LXCs.
func getNestedDockerCommands() []DiscoveryCommand {
return []DiscoveryCommand{
{
Name: "docker_containers",
Command: "docker ps -a --format '{{.Names}}|{{.Image}}|{{.Status}}|{{.Ports}}'",
Description: "All Docker containers",
Categories: []string{"container", "service"},
Optional: false,
},
{
Name: "docker_images",
Command: "docker images --format '{{.Repository}}:{{.Tag}}' | head -20",
Description: "Docker images",
Categories: []string{"version"},
Optional: true,
},
{
Name: "docker_compose",
Command: "find /opt /home /root -name 'docker-compose*.yml' -o -name 'compose*.yml' 2>/dev/null | head -10",
Description: "Docker compose files",
Categories: []string{"config"},
Optional: true,
},
}
}
// getK8sCommands returns commands for discovering Kubernetes pods.
func getK8sCommands() []DiscoveryCommand {
return []DiscoveryCommand{
{
Name: "processes",
Command: "ps aux 2>/dev/null || echo 'no_ps'",
Description: "Running processes in pod",
Categories: []string{"service"},
Optional: true,
},
{
Name: "listening_ports",
Command: "ss -tlnp 2>/dev/null || netstat -tlnp 2>/dev/null || echo 'no_ss'",
Description: "Listening ports",
Categories: []string{"port"},
Optional: true,
},
{
Name: "env_vars",
Command: "env 2>/dev/null | grep -vE '(PASSWORD|SECRET|KEY|TOKEN|CREDENTIAL)' | head -30",
Description: "Environment variables (filtered)",
Categories: []string{"config"},
Optional: true,
},
}
}
// getHostCommands returns commands for discovering host systems.
func getHostCommands() []DiscoveryCommand {
return []DiscoveryCommand{
{
Name: "os_release",
Command: "cat /etc/os-release",
Description: "Operating system",
Categories: []string{"version", "config"},
Optional: true,
},
{
Name: "hostname",
Command: "hostname -f 2>/dev/null || hostname",
Description: "Full hostname",
Categories: []string{"config"},
Optional: true,
},
{
Name: "running_services",
Command: "systemctl list-units --type=service --state=running --no-pager 2>/dev/null | head -40",
Description: "Running services",
Categories: []string{"service"},
Optional: true,
},
{
Name: "listening_ports",
Command: "ss -tlnp 2>/dev/null | head -30",
Description: "Listening network ports",
Categories: []string{"port", "network"},
Optional: true,
},
{
Name: "docker_containers",
Command: "docker ps --format '{{.Names}}: {{.Image}} ({{.Status}})' 2>/dev/null | head -30 || echo 'no_docker'",
Description: "Docker containers on host",
Categories: []string{"container", "service"},
Optional: true,
},
{
Name: "proxmox_version",
Command: "pveversion 2>/dev/null || echo 'not_proxmox'",
Description: "Proxmox version if applicable",
Categories: []string{"version"},
Optional: true,
},
{
Name: "zfs_pools",
Command: "zpool list 2>/dev/null | head -10 || echo 'no_zfs'",
Description: "ZFS pools",
Categories: []string{"storage"},
Optional: true,
},
{
Name: "disk_usage",
Command: "df -h | head -20",
Description: "Disk usage",
Categories: []string{"storage"},
Optional: true,
},
{
Name: "hardware_info",
Command: "lscpu | head -20",
Description: "CPU information",
Categories: []string{"hardware"},
Optional: true,
},
{
Name: "memory_info",
Command: "free -h",
Description: "Memory information",
Categories: []string{"hardware"},
Optional: true,
},
}
}
// BuildLXCCommand wraps a command for execution in an LXC container.
func BuildLXCCommand(vmid string, cmd string) string {
return fmt.Sprintf("pct exec %s -- sh -c %q", vmid, cmd)
}
// BuildVMCommand wraps a command for execution in a VM via QEMU guest agent.
// Note: This requires the guest agent to be running.
func BuildVMCommand(vmid string, cmd string) string {
// For VMs, we use qm guest exec which requires the guest agent
return fmt.Sprintf("qm guest exec %s -- sh -c %q", vmid, cmd)
}
// BuildDockerCommand wraps a command for execution in a Docker container.
func BuildDockerCommand(containerName string, cmd string) string {
return fmt.Sprintf("docker exec %s sh -c %q", containerName, cmd)
}
// BuildNestedDockerCommand builds a command to run inside Docker on a VM/LXC.
func BuildNestedDockerCommand(vmid string, isLXC bool, containerName string, cmd string) string {
dockerCmd := BuildDockerCommand(containerName, cmd)
if isLXC {
return BuildLXCCommand(vmid, dockerCmd)
}
return BuildVMCommand(vmid, dockerCmd)
}
// BuildK8sCommand builds a command to run in a Kubernetes pod.
func BuildK8sCommand(namespace, podName, containerName, cmd string) string {
if containerName != "" {
return fmt.Sprintf("kubectl exec -n %s %s -c %s -- sh -c %q", namespace, podName, containerName, cmd)
}
return fmt.Sprintf("kubectl exec -n %s %s -- sh -c %q", namespace, podName, cmd)
}
// GetCLIAccessTemplate returns a CLI access template for a resource type.
func GetCLIAccessTemplate(resourceType ResourceType) string {
switch resourceType {
case ResourceTypeLXC:
return "pct exec {vmid} -- {command}"
case ResourceTypeVM:
return "qm guest exec {vmid} -- {command}"
case ResourceTypeDocker:
return "docker exec {container} {command}"
case ResourceTypeDockerLXC:
return "pct exec {vmid} -- docker exec {container} {command}"
case ResourceTypeDockerVM:
return "qm guest exec {vmid} -- docker exec {container} {command}"
case ResourceTypeK8s:
return "kubectl exec -n {namespace} {pod} -- {command}"
case ResourceTypeHost:
return "{command}"
default:
return "{command}"
}
}
// FormatCLIAccess formats a CLI access string with actual values.
func FormatCLIAccess(resourceType ResourceType, vmid, containerName, namespace, podName string) string {
template := GetCLIAccessTemplate(resourceType)
result := template
result = strings.ReplaceAll(result, "{vmid}", vmid)
result = strings.ReplaceAll(result, "{container}", containerName)
result = strings.ReplaceAll(result, "{namespace}", namespace)
result = strings.ReplaceAll(result, "{pod}", podName)
return result
}
internal/aidiscovery/commands_test.go
-78
View File
@@ -1,78 +0,0 @@
package aidiscovery
import (
"strings"
"testing"
)
func TestCommandsAndTemplates(t *testing.T) {
resourceTypes := []ResourceType{
ResourceTypeLXC,
ResourceTypeVM,
ResourceTypeDocker,
ResourceTypeDockerVM,
ResourceTypeDockerLXC,
ResourceTypeK8s,
ResourceTypeHost,
}
for _, rt := range resourceTypes {
cmds := GetCommandsForResource(rt)
if len(cmds) == 0 {
t.Fatalf("expected commands for %s", rt)
}
}
if len(GetCommandsForResource(ResourceType("unknown"))) != 0 {
t.Fatalf("expected no commands for unknown resource type")
}
if !strings.Contains(BuildLXCCommand("101", "echo hi"), "pct exec 101") {
t.Fatalf("unexpected LXC command")
}
if !strings.Contains(BuildVMCommand("101", "echo hi"), "qm guest exec 101") {
t.Fatalf("unexpected VM command")
}
if !strings.Contains(BuildDockerCommand("web", "echo hi"), "docker exec web") {
t.Fatalf("unexpected docker command")
}
nestedLXC := BuildNestedDockerCommand("201", true, "web", "echo hi")
if !strings.Contains(nestedLXC, "pct exec 201") || !strings.Contains(nestedLXC, "docker exec web") {
t.Fatalf("unexpected nested LXC command: %s", nestedLXC)
}
nestedVM := BuildNestedDockerCommand("301", false, "web", "echo hi")
if !strings.Contains(nestedVM, "qm guest exec 301") || !strings.Contains(nestedVM, "docker exec web") {
t.Fatalf("unexpected nested VM command: %s", nestedVM)
}
withContainer := BuildK8sCommand("default", "pod", "app", "echo hi")
if !strings.Contains(withContainer, "-c app") || !strings.Contains(withContainer, "kubectl exec") {
t.Fatalf("unexpected k8s command: %s", withContainer)
}
withoutContainer := BuildK8sCommand("default", "pod", "", "echo hi")
if strings.Contains(withoutContainer, "-c app") {
t.Fatalf("unexpected container selector: %s", withoutContainer)
}
template := GetCLIAccessTemplate(ResourceTypeK8s)
if !strings.Contains(template, "{namespace}") || !strings.Contains(template, "{pod}") {
t.Fatalf("unexpected template: %s", template)
}
for _, rt := range resourceTypes {
if tmpl := GetCLIAccessTemplate(rt); tmpl == "" {
t.Fatalf("expected template for %s", rt)
}
}
if tmpl := GetCLIAccessTemplate(ResourceType("unknown")); tmpl != "{command}" {
t.Fatalf("unexpected default template: %s", tmpl)
}
formatted := FormatCLIAccess(ResourceTypeK8s, "101", "container", "default", "pod")
if !strings.Contains(formatted, "default") || !strings.Contains(formatted, "pod") {
t.Fatalf("unexpected formatted access: %s", formatted)
}
}
internal/aidiscovery/deep_scanner.go
-378
View File
@@ -1,378 +0,0 @@
package aidiscovery
import (
"context"
"fmt"
"sync"
"time"
"github.com/google/uuid"
"github.com/rs/zerolog/log"
)
// CommandExecutor executes commands on infrastructure.
type CommandExecutor interface {
ExecuteCommand(ctx context.Context, agentID string, cmd ExecuteCommandPayload) (*CommandResultPayload, error)
GetConnectedAgents() []ConnectedAgent
IsAgentConnected(agentID string) bool
}
// ExecuteCommandPayload mirrors agentexec.ExecuteCommandPayload
type ExecuteCommandPayload struct {
RequestID string `json:"request_id"`
Command string `json:"command"`
TargetType string `json:"target_type"` // "host", "container", "vm"
TargetID string `json:"target_id,omitempty"` // VMID for container/VM
Timeout int `json:"timeout,omitempty"`
}
// CommandResultPayload mirrors agentexec.CommandResultPayload
type CommandResultPayload struct {
RequestID string `json:"request_id"`
Success bool `json:"success"`
Stdout string `json:"stdout,omitempty"`
Stderr string `json:"stderr,omitempty"`
ExitCode int `json:"exit_code"`
Error string `json:"error,omitempty"`
Duration int64 `json:"duration_ms"`
}
// ConnectedAgent mirrors agentexec.ConnectedAgent
type ConnectedAgent struct {
AgentID string
Hostname string
Version string
Platform string
Tags []string
ConnectedAt time.Time
}
// DeepScanner runs discovery commands on resources.
type DeepScanner struct {
executor CommandExecutor
mu sync.RWMutex
progress map[string]*DiscoveryProgress // resourceID -> progress
maxParallel int
timeout time.Duration
}
// NewDeepScanner creates a new deep scanner.
func NewDeepScanner(executor CommandExecutor) *DeepScanner {
return &DeepScanner{
executor: executor,
progress: make(map[string]*DiscoveryProgress),
maxParallel: 3, // Run up to 3 commands in parallel per resource
timeout: 30 * time.Second,
}
}
// ScanResult contains the results of a deep scan.
type ScanResult struct {
ResourceType ResourceType
ResourceID string
HostID string
Hostname string
CommandOutputs map[string]string
Errors map[string]string
StartedAt time.Time
CompletedAt time.Time
}
// Scan runs discovery commands on a resource and returns the outputs.
func (s *DeepScanner) Scan(ctx context.Context, req DiscoveryRequest) (*ScanResult, error) {
resourceID := MakeResourceID(req.ResourceType, req.HostID, req.ResourceID)
// Initialize progress
s.mu.Lock()
s.progress[resourceID] = &DiscoveryProgress{
ResourceID: resourceID,
Status: DiscoveryStatusRunning,
CurrentStep: "initializing",
StartedAt: time.Now(),
}
s.mu.Unlock()
defer func() {
s.mu.Lock()
delete(s.progress, resourceID)
s.mu.Unlock()
}()
result := &ScanResult{
ResourceType: req.ResourceType,
ResourceID: req.ResourceID,
HostID: req.HostID,
Hostname: req.Hostname,
CommandOutputs: make(map[string]string),
Errors: make(map[string]string),
StartedAt: time.Now(),
}
// Check if we have an agent for this host
if s.executor == nil {
return nil, fmt.Errorf("no command executor available")
}
// Find the agent for this host
agentID := s.findAgentForHost(req.HostID, req.Hostname)
if agentID == "" {
return nil, fmt.Errorf("no connected agent for host %s (%s)", req.HostID, req.Hostname)
}
// Get commands for this resource type
commands := GetCommandsForResource(req.ResourceType)
if len(commands) == 0 {
return nil, fmt.Errorf("no commands defined for resource type %s", req.ResourceType)
}
// Update progress
s.mu.Lock()
if prog, ok := s.progress[resourceID]; ok {
prog.TotalSteps = len(commands)
prog.CurrentStep = "running commands"
}
s.mu.Unlock()
// Run commands with limited parallelism
semaphore := make(chan struct{}, s.maxParallel)
var wg sync.WaitGroup
var mu sync.Mutex
for _, cmd := range commands {
wg.Add(1)
go func(cmd DiscoveryCommand) {
defer wg.Done()
select {
case semaphore <- struct{}{}:
defer func() { <-semaphore }()
case <-ctx.Done():
return
}
// Build the actual command to run
actualCmd := s.buildCommand(req.ResourceType, req.ResourceID, cmd.Command)
// Execute the command
cmdCtx, cancel := context.WithTimeout(ctx, s.timeout)
defer cancel()
cmdResult, err := s.executor.ExecuteCommand(cmdCtx, agentID, ExecuteCommandPayload{
RequestID: uuid.New().String(),
Command: actualCmd,
TargetType: s.getTargetType(req.ResourceType),
TargetID: req.ResourceID,
Timeout: cmd.Timeout,
})
mu.Lock()
defer mu.Unlock()
if err != nil {
if !cmd.Optional {
result.Errors[cmd.Name] = err.Error()
}
log.Debug().
Err(err).
Str("command", cmd.Name).
Str("resource", resourceID).
Msg("Command failed during discovery")
return
}
if cmdResult != nil {
output := cmdResult.Stdout
if cmdResult.Stderr != "" && output != "" {
output += "\n--- stderr ---\n" + cmdResult.Stderr
} else if cmdResult.Stderr != "" {
output = cmdResult.Stderr
}
if output != "" {
result.CommandOutputs[cmd.Name] = output
}
if !cmdResult.Success && cmdResult.Error != "" && !cmd.Optional {
result.Errors[cmd.Name] = cmdResult.Error
}
}
// Update progress
s.mu.Lock()
if prog, ok := s.progress[resourceID]; ok {
prog.CompletedSteps++
}
s.mu.Unlock()
}(cmd)
}
wg.Wait()
result.CompletedAt = time.Now()
log.Info().
Str("resource", resourceID).
Int("outputs", len(result.CommandOutputs)).
Int("errors", len(result.Errors)).
Dur("duration", result.CompletedAt.Sub(result.StartedAt)).
Msg("Deep scan completed")
return result, nil
}
// buildCommand wraps the command appropriately for the resource type.
// NOTE: For LXC/VM, the agent handles wrapping via pct exec / qm guest exec
// based on TargetType, so we don't wrap here. We only wrap for Docker containers
// since Docker isn't a recognized TargetType in the agent.
func (s *DeepScanner) buildCommand(resourceType ResourceType, resourceID string, cmd string) string {
switch resourceType {
case ResourceTypeLXC:
// Agent wraps with pct exec based on TargetType="container"
return cmd
case ResourceTypeVM:
// Agent wraps with qm guest exec based on TargetType="vm"
return cmd
case ResourceTypeDocker:
// Docker needs wrapping here since agent doesn't handle it
return BuildDockerCommand(resourceID, cmd)
case ResourceTypeHost:
// Commands run directly on host
return cmd
case ResourceTypeDockerLXC:
// Docker inside LXC - agent wraps with pct exec, we just add docker exec
// resourceID format: "vmid:container_name"
parts := splitResourceID(resourceID)
if len(parts) >= 2 {
return BuildDockerCommand(parts[1], cmd)
}
return cmd
case ResourceTypeDockerVM:
// Docker inside VM - agent wraps with qm guest exec, we just add docker exec
parts := splitResourceID(resourceID)
if len(parts) >= 2 {
return BuildDockerCommand(parts[1], cmd)
}
return cmd
default:
return cmd
}
}
// getTargetType returns the target type for the agent execution payload.
func (s *DeepScanner) getTargetType(resourceType ResourceType) string {
switch resourceType {
case ResourceTypeLXC:
return "container"
case ResourceTypeVM:
return "vm"
case ResourceTypeDocker:
return "host" // Docker commands run on host via docker exec
case ResourceTypeHost:
return "host"
default:
return "host"
}
}
// findAgentForHost finds the agent ID for a given host.
func (s *DeepScanner) findAgentForHost(hostID, hostname string) string {
agents := s.executor.GetConnectedAgents()
// First try exact match on agent ID
for _, agent := range agents {
if agent.AgentID == hostID {
return agent.AgentID
}
}
// Then try hostname match
for _, agent := range agents {
if agent.Hostname == hostname || agent.Hostname == hostID {
return agent.AgentID
}
}
// If only one agent connected, use it
if len(agents) == 1 {
return agents[0].AgentID
}
return ""
}
// GetProgress returns the current progress of a scan.
func (s *DeepScanner) GetProgress(resourceID string) *DiscoveryProgress {
s.mu.RLock()
defer s.mu.RUnlock()
if prog, ok := s.progress[resourceID]; ok {
return prog
}
return nil
}
// IsScanning returns whether a resource is currently being scanned.
func (s *DeepScanner) IsScanning(resourceID string) bool {
s.mu.RLock()
defer s.mu.RUnlock()
_, ok := s.progress[resourceID]
return ok
}
// splitResourceID splits a compound resource ID (e.g., "101:container_name").
func splitResourceID(id string) []string {
var parts []string
start := 0
for i, c := range id {
if c == ':' {
parts = append(parts, id[start:i])
start = i + 1
}
}
if start < len(id) {
parts = append(parts, id[start:])
}
return parts
}
// ScanDocker runs discovery on Docker containers via the host.
func (s *DeepScanner) ScanDocker(ctx context.Context, hostID, hostname, containerName string) (*ScanResult, error) {
req := DiscoveryRequest{
ResourceType: ResourceTypeDocker,
ResourceID: containerName,
HostID: hostID,
Hostname: hostname,
}
return s.Scan(ctx, req)
}
// ScanLXC runs discovery on an LXC container.
func (s *DeepScanner) ScanLXC(ctx context.Context, hostID, hostname, vmid string) (*ScanResult, error) {
req := DiscoveryRequest{
ResourceType: ResourceTypeLXC,
ResourceID: vmid,
HostID: hostID,
Hostname: hostname,
}
return s.Scan(ctx, req)
}
// ScanVM runs discovery on a VM via QEMU guest agent.
func (s *DeepScanner) ScanVM(ctx context.Context, hostID, hostname, vmid string) (*ScanResult, error) {
req := DiscoveryRequest{
ResourceType: ResourceTypeVM,
ResourceID: vmid,
HostID: hostID,
Hostname: hostname,
}
return s.Scan(ctx, req)
}
// ScanHost runs discovery on a host system.
func (s *DeepScanner) ScanHost(ctx context.Context, hostID, hostname string) (*ScanResult, error) {
req := DiscoveryRequest{
ResourceType: ResourceTypeHost,
ResourceID: hostID,
HostID: hostID,
Hostname: hostname,
}
return s.Scan(ctx, req)
}
internal/aidiscovery/deep_scanner_test.go
-325
View File
@@ -1,325 +0,0 @@
package aidiscovery
import (
"context"
"strings"
"sync"
"testing"
"time"
)
type stubExecutor struct {
mu sync.Mutex
commands []string
agents []ConnectedAgent
}
func (s *stubExecutor) ExecuteCommand(ctx context.Context, agentID string, cmd ExecuteCommandPayload) (*CommandResultPayload, error) {
s.mu.Lock()
s.commands = append(s.commands, cmd.Command)
s.mu.Unlock()
if err := ctx.Err(); err != nil {
return nil, err
}
if strings.Contains(cmd.Command, "docker ps -a") {
return &CommandResultPayload{
RequestID: cmd.RequestID,
Success: false,
Error: "boom",
}, nil
}
return &CommandResultPayload{
RequestID: cmd.RequestID,
Success: true,
Stdout: cmd.Command,
Duration: 5,
}, nil
}
func (s *stubExecutor) GetConnectedAgents() []ConnectedAgent {
return s.agents
}
func (s *stubExecutor) IsAgentConnected(agentID string) bool {
for _, agent := range s.agents {
if agent.AgentID == agentID {
return true
}
}
return false
}
type outputExecutor struct{}
func (outputExecutor) ExecuteCommand(ctx context.Context, agentID string, cmd ExecuteCommandPayload) (*CommandResultPayload, error) {
switch {
case strings.Contains(cmd.Command, "docker ps -a"):
return &CommandResultPayload{Success: true, Stdout: "out", Stderr: "err"}, nil
case strings.Contains(cmd.Command, "docker images"):
return &CommandResultPayload{Success: true, Stderr: "err-only"}, nil
default:
return &CommandResultPayload{Success: true}, nil
}
}
func (outputExecutor) GetConnectedAgents() []ConnectedAgent {
return []ConnectedAgent{{AgentID: "host1", Hostname: "host1"}}
}
func (outputExecutor) IsAgentConnected(string) bool { return true }
type errorExecutor struct{}
func (errorExecutor) ExecuteCommand(ctx context.Context, agentID string, cmd ExecuteCommandPayload) (*CommandResultPayload, error) {
return nil, context.DeadlineExceeded
}
func (errorExecutor) GetConnectedAgents() []ConnectedAgent {
return []ConnectedAgent{{AgentID: "host1", Hostname: "host1"}}
}
func (errorExecutor) IsAgentConnected(string) bool { return true }
func TestDeepScanner_Scan_NestedDockerCommands(t *testing.T) {
exec := &stubExecutor{
agents: []ConnectedAgent{
{AgentID: "host1", Hostname: "host1", ConnectedAt: time.Now()},
},
}
scanner := NewDeepScanner(exec)
result, err := scanner.Scan(context.Background(), DiscoveryRequest{
ResourceType: ResourceTypeDockerVM,
ResourceID: "101:web",
HostID: "host1",
Hostname: "host1",
})
if err != nil {
t.Fatalf("Scan error: %v", err)
}
if len(result.CommandOutputs) == 0 {
t.Fatalf("expected command outputs")
}
if _, ok := result.Errors["docker_containers"]; !ok {
t.Fatalf("expected docker_containers error, got %#v", result.Errors)
}
exec.mu.Lock()
defer exec.mu.Unlock()
foundWrapped := false
for _, cmd := range exec.commands {
if strings.Contains(cmd, "qm guest exec 101") && strings.Contains(cmd, "docker exec web") {
foundWrapped = true
break
}
}
if !foundWrapped {
t.Fatalf("expected nested docker command, got %#v", exec.commands)
}
}
func TestDeepScanner_FindAgentAndTargetType(t *testing.T) {
exec := &stubExecutor{
agents: []ConnectedAgent{
{AgentID: "a1", Hostname: "node1"},
{AgentID: "a2", Hostname: "node2"},
},
}
scanner := NewDeepScanner(exec)
if got := scanner.findAgentForHost("a2", ""); got != "a2" {
t.Fatalf("expected direct agent match, got %s", got)
}
if got := scanner.findAgentForHost("node1", "node1"); got != "a1" {
t.Fatalf("expected hostname match, got %s", got)
}
exec.agents = []ConnectedAgent{{AgentID: "solo", Hostname: "only"}}
if got := scanner.findAgentForHost("missing", "missing"); got != "solo" {
t.Fatalf("expected single agent fallback, got %s", got)
}
exec.agents = nil
if got := scanner.findAgentForHost("missing", "missing"); got != "" {
t.Fatalf("expected no agent, got %s", got)
}
if scanner.getTargetType(ResourceTypeLXC) != "container" {
t.Fatalf("unexpected target type for lxc")
}
if scanner.getTargetType(ResourceTypeVM) != "vm" {
t.Fatalf("unexpected target type for vm")
}
if scanner.getTargetType(ResourceTypeDocker) != "host" {
t.Fatalf("unexpected target type for docker")
}
if scanner.getTargetType(ResourceTypeHost) != "host" {
t.Fatalf("unexpected target type for host")
}
}
func TestSplitResourceID(t *testing.T) {
parts := splitResourceID("101:web:extra")
if len(parts) != 3 || parts[0] != "101" || parts[1] != "web" || parts[2] != "extra" {
t.Fatalf("unexpected parts: %#v", parts)
}
}
func TestDeepScanner_BuildCommandAndProgress(t *testing.T) {
scanner := NewDeepScanner(&stubExecutor{})
if cmd := scanner.buildCommand(ResourceTypeLXC, "101", "echo hi"); !strings.Contains(cmd, "pct exec 101") {
t.Fatalf("unexpected lxc command: %s", cmd)
}
if cmd := scanner.buildCommand(ResourceTypeVM, "101", "echo hi"); cmd != "echo hi" {
t.Fatalf("unexpected vm command: %s", cmd)
}
if cmd := scanner.buildCommand(ResourceTypeDocker, "web", "echo hi"); !strings.Contains(cmd, "docker exec web") {
t.Fatalf("unexpected docker command: %s", cmd)
}
if cmd := scanner.buildCommand(ResourceTypeHost, "host", "echo hi"); cmd != "echo hi" {
t.Fatalf("unexpected host command: %s", cmd)
}
dockerLXC := scanner.buildCommand(ResourceTypeDockerLXC, "201:web", "echo hi")
if !strings.Contains(dockerLXC, "pct exec 201") || !strings.Contains(dockerLXC, "docker exec web") {
t.Fatalf("unexpected docker lxc command: %s", dockerLXC)
}
if cmd := scanner.buildCommand(ResourceTypeDockerLXC, "bad", "echo hi"); cmd != "echo hi" {
t.Fatalf("expected fallback lxc command, got %s", cmd)
}
dockerVM := scanner.buildCommand(ResourceTypeDockerVM, "301:web", "echo hi")
if !strings.Contains(dockerVM, "qm guest exec 301") || !strings.Contains(dockerVM, "docker exec web") {
t.Fatalf("unexpected docker vm command: %s", dockerVM)
}
if cmd := scanner.buildCommand(ResourceTypeDockerVM, "bad", "echo hi"); cmd != "echo hi" {
t.Fatalf("expected fallback command, got %s", cmd)
}
if cmd := scanner.buildCommand(ResourceType("unknown"), "id", "echo hi"); cmd != "echo hi" {
t.Fatalf("expected default command, got %s", cmd)
}
scanner.progress["id"] = &DiscoveryProgress{ResourceID: "id"}
if scanner.GetProgress("id") == nil {
t.Fatalf("expected progress")
}
if !scanner.IsScanning("id") {
t.Fatalf("expected IsScanning true")
}
if scanner.GetProgress("missing") != nil {
t.Fatalf("expected nil progress")
}
if scanner.IsScanning("missing") {
t.Fatalf("expected IsScanning false")
}
noExec := NewDeepScanner(nil)
if _, err := noExec.ScanHost(context.Background(), "host1", "host1"); err == nil {
t.Fatalf("expected error without executor")
}
}
func TestDeepScanner_ScanWrappers(t *testing.T) {
exec := &stubExecutor{
agents: []ConnectedAgent{{AgentID: "host1", Hostname: "host1"}},
}
scanner := NewDeepScanner(exec)
scanner.maxParallel = 1
if _, err := scanner.ScanDocker(context.Background(), "host1", "host1", "web"); err != nil {
t.Fatalf("ScanDocker error: %v", err)
}
if _, err := scanner.ScanLXC(context.Background(), "host1", "host1", "101"); err != nil {
t.Fatalf("ScanLXC error: %v", err)
}
if _, err := scanner.ScanVM(context.Background(), "host1", "host1", "102"); err != nil {
t.Fatalf("ScanVM error: %v", err)
}
}
func TestDeepScanner_ScanErrors(t *testing.T) {
exec := &stubExecutor{
agents: []ConnectedAgent{{AgentID: "host1", Hostname: "host1"}},
}
scanner := NewDeepScanner(exec)
if _, err := scanner.Scan(context.Background(), DiscoveryRequest{
ResourceType: ResourceType("unknown"),
ResourceID: "id",
HostID: "host1",
Hostname: "host1",
}); err == nil {
t.Fatalf("expected error for unknown resource type")
}
exec.agents = nil
if _, err := scanner.Scan(context.Background(), DiscoveryRequest{
ResourceType: ResourceTypeDocker,
ResourceID: "web",
HostID: "host1",
Hostname: "host1",
}); err == nil {
t.Fatalf("expected error for missing agent")
}
}
func TestDeepScanner_OutputHandling(t *testing.T) {
exec := outputExecutor{}
scanner := NewDeepScanner(exec)
scanner.maxParallel = 1
result, err := scanner.Scan(context.Background(), DiscoveryRequest{
ResourceType: ResourceTypeDockerVM,
ResourceID: "101:web",
HostID: "host1",
Hostname: "host1",
})
if err != nil {
t.Fatalf("Scan error: %v", err)
}
if out := result.CommandOutputs["docker_containers"]; !strings.Contains(out, "--- stderr ---") {
t.Fatalf("expected combined stderr output, got %s", out)
}
if out := result.CommandOutputs["docker_images"]; out != "err-only" {
t.Fatalf("expected stderr-only output, got %s", out)
}
}
func TestDeepScanner_CommandErrorHandling(t *testing.T) {
scanner := NewDeepScanner(errorExecutor{})
scanner.maxParallel = 1
result, err := scanner.Scan(context.Background(), DiscoveryRequest{
ResourceType: ResourceTypeDockerVM,
ResourceID: "101:web",
HostID: "host1",
Hostname: "host1",
})
if err != nil {
t.Fatalf("Scan error: %v", err)
}
if _, ok := result.Errors["docker_containers"]; !ok {
t.Fatalf("expected error for non-optional command")
}
}
func TestDeepScanner_ScanCanceledContext(t *testing.T) {
exec := &stubExecutor{
agents: []ConnectedAgent{{AgentID: "host1", Hostname: "host1"}},
}
scanner := NewDeepScanner(exec)
scanner.maxParallel = 0
ctx, cancel := context.WithCancel(context.Background())
cancel()
if _, err := scanner.Scan(ctx, DiscoveryRequest{
ResourceType: ResourceTypeDockerVM,
ResourceID: "101:web",
HostID: "host1",
Hostname: "host1",
}); err != nil {
t.Fatalf("Scan error: %v", err)
}
}
internal/aidiscovery/formatters.go
-337
View File
@@ -1,337 +0,0 @@
package aidiscovery
import (
"fmt"
"strings"
"time"
)
// FormatForAIContext formats discoveries for inclusion in AI prompts.
// This provides context about resources for Patrol, Investigation, and Chat.
func FormatForAIContext(discoveries []*ResourceDiscovery) string {
if len(discoveries) == 0 {
return ""
}
var sb strings.Builder
sb.WriteString("## Infrastructure Discovery\n\n")
sb.WriteString("The following has been discovered about the affected resources:\n\n")
for _, d := range discoveries {
sb.WriteString(formatSingleDiscovery(d))
sb.WriteString("\n")
}
sb.WriteString("\n**IMPORTANT:** Use the CLI access methods shown above. For example:\n")
sb.WriteString("- For LXC containers, use `pct exec <vmid> -- <command>`\n")
sb.WriteString("- For VMs with guest agent, use `qm guest exec <vmid> -- <command>`\n")
sb.WriteString("- For Docker containers, use `docker exec <container> <command>`\n")
return sb.String()
}
// FormatSingleForAIContext formats a single discovery for AI context.
func FormatSingleForAIContext(d *ResourceDiscovery) string {
if d == nil {
return ""
}
return formatSingleDiscovery(d)
}
// formatSingleDiscovery formats a single discovery entry.
func formatSingleDiscovery(d *ResourceDiscovery) string {
var sb strings.Builder
// Header with service info
sb.WriteString(fmt.Sprintf("### %s (%s)\n", d.ServiceName, d.ID))
sb.WriteString(fmt.Sprintf("- **Type:** %s\n", d.ResourceType))
sb.WriteString(fmt.Sprintf("- **Host:** %s\n", d.Hostname))
if d.ServiceVersion != "" {
sb.WriteString(fmt.Sprintf("- **Version:** %s\n", d.ServiceVersion))
}
if d.Category != "" && d.Category != CategoryUnknown {
sb.WriteString(fmt.Sprintf("- **Category:** %s\n", d.Category))
}
// CLI access (most important for remediation)
if d.CLIAccess != "" {
sb.WriteString(fmt.Sprintf("- **CLI Access:** `%s`\n", d.CLIAccess))
}
// Config and data paths
if len(d.ConfigPaths) > 0 {
sb.WriteString(fmt.Sprintf("- **Config Paths:** %s\n", strings.Join(d.ConfigPaths, ", ")))
}
if len(d.DataPaths) > 0 {
sb.WriteString(fmt.Sprintf("- **Data Paths:** %s\n", strings.Join(d.DataPaths, ", ")))
}
// Ports
if len(d.Ports) > 0 {
var ports []string
for _, p := range d.Ports {
ports = append(ports, fmt.Sprintf("%d/%s", p.Port, p.Protocol))
}
sb.WriteString(fmt.Sprintf("- **Ports:** %s\n", strings.Join(ports, ", ")))
}
// Important facts
importantFacts := filterImportantFacts(d.Facts)
if len(importantFacts) > 0 {
sb.WriteString("- **Key Facts:**\n")
for _, f := range importantFacts {
sb.WriteString(fmt.Sprintf(" - %s: %s\n", f.Key, f.Value))
}
}
// User notes (critical for context)
if d.UserNotes != "" {
sb.WriteString(fmt.Sprintf("- **User Notes:** %s\n", d.UserNotes))
}
return sb.String()
}
// filterImportantFacts returns the most relevant facts for AI context.
func filterImportantFacts(facts []DiscoveryFact) []DiscoveryFact {
var important []DiscoveryFact
// Priority categories
priorityCategories := map[FactCategory]bool{
FactCategoryHardware: true, // GPU, TPU
FactCategoryDependency: true, // MQTT, database connections
FactCategorySecurity: true, // Auth info
FactCategoryVersion: true, // Version info
}
for _, f := range facts {
if priorityCategories[f.Category] && f.Confidence >= 0.7 {
important = append(important, f)
}
}
// Limit to top 5 facts
if len(important) > 5 {
important = important[:5]
}
return important
}
// FormatDiscoverySummary formats a summary of all discoveries.
func FormatDiscoverySummary(discoveries []*ResourceDiscovery) string {
if len(discoveries) == 0 {
return "No infrastructure discovery data available."
}
var sb strings.Builder
sb.WriteString(fmt.Sprintf("Infrastructure Discovery Summary (%d resources):\n\n", len(discoveries)))
// Group by resource type
byType := make(map[ResourceType][]*ResourceDiscovery)
for _, d := range discoveries {
byType[d.ResourceType] = append(byType[d.ResourceType], d)
}
for rt, ds := range byType {
sb.WriteString(fmt.Sprintf("**%s** (%d):\n", rt, len(ds)))
for _, d := range ds {
confidence := ""
if d.Confidence >= 0.9 {
confidence = " [high confidence]"
} else if d.Confidence >= 0.7 {
confidence = " [medium confidence]"
}
sb.WriteString(fmt.Sprintf(" - %s: %s%s\n", d.ResourceID, d.ServiceName, confidence))
}
sb.WriteString("\n")
}
return sb.String()
}
// FormatForRemediation formats discovery specifically for remediation context.
func FormatForRemediation(d *ResourceDiscovery) string {
if d == nil {
return ""
}
var sb strings.Builder
sb.WriteString("## Resource Context for Remediation\n\n")
sb.WriteString(fmt.Sprintf("**Resource:** %s (%s)\n", d.ServiceName, d.ID))
sb.WriteString(fmt.Sprintf("**Type:** %s on %s\n\n", d.ResourceType, d.Hostname))
// CLI access is most critical
if d.CLIAccess != "" {
sb.WriteString("### How to Execute Commands\n")
sb.WriteString(fmt.Sprintf("```\n%s\n```\n\n", d.CLIAccess))
}
// Service-specific info
if d.ServiceType != "" {
sb.WriteString(fmt.Sprintf("**Service:** %s", d.ServiceType))
if d.ServiceVersion != "" {
sb.WriteString(fmt.Sprintf(" v%s", d.ServiceVersion))
}
sb.WriteString("\n\n")
}
// Config paths for potential fixes
if len(d.ConfigPaths) > 0 {
sb.WriteString("### Configuration Files\n")
for _, p := range d.ConfigPaths {
sb.WriteString(fmt.Sprintf("- `%s`\n", p))
}
sb.WriteString("\n")
}
// User notes may contain important context
if d.UserNotes != "" {
sb.WriteString("### User Notes\n")
sb.WriteString(d.UserNotes)
sb.WriteString("\n\n")
}
// Hardware info for special considerations
for _, f := range d.Facts {
if f.Category == FactCategoryHardware {
sb.WriteString(fmt.Sprintf("**Hardware:** %s = %s\n", f.Key, f.Value))
}
}
return sb.String()
}
// FormatDiscoveryAge returns a human-readable age string.
func FormatDiscoveryAge(d *ResourceDiscovery) string {
if d == nil || d.UpdatedAt.IsZero() {
return "unknown"
}
age := time.Since(d.UpdatedAt)
switch {
case age < time.Minute:
return "just now"
case age < time.Hour:
mins := int(age.Minutes())
if mins == 1 {
return "1 minute ago"
}
return fmt.Sprintf("%d minutes ago", mins)
case age < 24*time.Hour:
hours := int(age.Hours())
if hours == 1 {
return "1 hour ago"
}
return fmt.Sprintf("%d hours ago", hours)
default:
days := int(age.Hours() / 24)
if days == 1 {
return "1 day ago"
}
return fmt.Sprintf("%d days ago", days)
}
}
// GetCLIExample returns an example CLI command for the resource.
func GetCLIExample(d *ResourceDiscovery, exampleCmd string) string {
if d == nil || d.CLIAccess == "" {
return ""
}
// Replace the placeholder with the example command
cli := d.CLIAccess
cli = strings.ReplaceAll(cli, "...", exampleCmd)
cli = strings.ReplaceAll(cli, "{command}", exampleCmd)
return cli
}
// FormatFactsTable formats facts as a simple table.
func FormatFactsTable(facts []DiscoveryFact) string {
if len(facts) == 0 {
return ""
}
var sb strings.Builder
sb.WriteString("| Category | Key | Value |\n")
sb.WriteString("|----------|-----|-------|\n")
for _, f := range facts {
value := f.Value
if len(value) > 50 {
value = value[:47] + "..."
}
sb.WriteString(fmt.Sprintf("| %s | %s | %s |\n", f.Category, f.Key, value))
}
return sb.String()
}
// BuildResourceContextForPatrol builds context for Patrol findings.
func BuildResourceContextForPatrol(store *Store, resourceIDs []string) string {
if store == nil || len(resourceIDs) == 0 {
return ""
}
discoveries, err := store.GetMultiple(resourceIDs)
if err != nil || len(discoveries) == 0 {
return ""
}
return FormatForAIContext(discoveries)
}
// ToJSON converts a discovery to a JSON-friendly map.
func ToJSON(d *ResourceDiscovery) map[string]any {
if d == nil {
return nil
}
facts := make([]map[string]any, 0, len(d.Facts))
for _, f := range d.Facts {
facts = append(facts, map[string]any{
"category": f.Category,
"key": f.Key,
"value": f.Value,
"source": f.Source,
"confidence": f.Confidence,
})
}
ports := make([]map[string]any, 0, len(d.Ports))
for _, p := range d.Ports {
ports = append(ports, map[string]any{
"port": p.Port,
"protocol": p.Protocol,
"process": p.Process,
"address": p.Address,
})
}
return map[string]any{
"id": d.ID,
"resource_type": d.ResourceType,
"resource_id": d.ResourceID,
"host_id": d.HostID,
"hostname": d.Hostname,
"service_type": d.ServiceType,
"service_name": d.ServiceName,
"service_version": d.ServiceVersion,
"category": d.Category,
"cli_access": d.CLIAccess,
"facts": facts,
"config_paths": d.ConfigPaths,
"data_paths": d.DataPaths,
"ports": ports,
"user_notes": d.UserNotes,
"confidence": d.Confidence,
"ai_reasoning": d.AIReasoning,
"discovered_at": d.DiscoveredAt,
"updated_at": d.UpdatedAt,
"scan_duration": d.ScanDuration,
}
}
internal/aidiscovery/formatters_test.go
-195
View File
@@ -1,195 +0,0 @@
package aidiscovery
import (
"strings"
"testing"
"time"
)
func TestFormattersAndTables(t *testing.T) {
if FormatForAIContext(nil) != "" {
t.Fatalf("expected empty context for nil discoveries")
}
discovery := &ResourceDiscovery{
ID: MakeResourceID(ResourceTypeDocker, "host1", "app"),
ResourceType: ResourceTypeDocker,
ResourceID: "app",
HostID: "host1",
Hostname: "host1",
ServiceType: "app",
ServiceName: "App Service",
ServiceVersion: "1.0",
Category: CategoryWebServer,
CLIAccess: "docker exec app ...",
ConfigPaths: []string{"/etc/app/config.yml"},
DataPaths: []string{"/var/lib/app"},
Ports: []PortInfo{{Port: 80, Protocol: "tcp"}},
UserNotes: "keepalive enabled",
Facts: []DiscoveryFact{
{Category: FactCategoryHardware, Key: "gpu", Value: "nvidia", Confidence: 0.9},
{Category: FactCategoryService, Key: "worker", Value: "enabled", Confidence: 0.9},
},
}
ctx := FormatForAIContext([]*ResourceDiscovery{discovery})
if !strings.Contains(ctx, "Infrastructure Discovery") || !strings.Contains(ctx, "App Service") {
t.Fatalf("unexpected context: %s", ctx)
}
if !strings.Contains(ctx, "docker exec") || !strings.Contains(ctx, "User Notes") {
t.Fatalf("missing expected fields in context")
}
if FormatSingleForAIContext(nil) != "" {
t.Fatalf("expected empty string for nil discovery")
}
if !strings.Contains(FormatSingleForAIContext(discovery), "App Service") {
t.Fatalf("expected single discovery output")
}
remediation := FormatForRemediation(discovery)
if !strings.Contains(remediation, "How to Execute Commands") || !strings.Contains(remediation, "Hardware") {
t.Fatalf("unexpected remediation output: %s", remediation)
}
if FormatForRemediation(nil) != "" {
t.Fatalf("expected empty remediation output for nil")
}
example := GetCLIExample(discovery, "ls /")
if !strings.Contains(example, "ls /") {
t.Fatalf("unexpected cli example: %s", example)
}
if GetCLIExample(&ResourceDiscovery{}, "ls /") != "" {
t.Fatalf("expected empty example when cli access missing")
}
table := FormatFactsTable([]DiscoveryFact{
{Category: FactCategoryVersion, Key: "app", Value: strings.Repeat("x", 60)},
})
if !strings.Contains(table, "...") {
t.Fatalf("expected truncated table value: %s", table)
}
if FormatFactsTable(nil) != "" {
t.Fatalf("expected empty facts table for nil")
}
jsonMap := ToJSON(discovery)
if jsonMap["service_name"] != "App Service" || jsonMap["resource_id"] != "app" {
t.Fatalf("unexpected json map: %#v", jsonMap)
}
if ToJSON(nil) != nil {
t.Fatalf("expected nil json map for nil discovery")
}
}
func TestFormatDiscoverySummaryAndAge(t *testing.T) {
now := time.Now()
if FormatDiscoverySummary(nil) == "" {
t.Fatalf("expected summary text for empty list")
}
if FormatDiscoveryAge(nil) != "unknown" {
t.Fatalf("expected unknown age for nil")
}
if FormatDiscoveryAge(&ResourceDiscovery{}) != "unknown" {
t.Fatalf("expected unknown age for zero timestamp")
}
discoveries := []*ResourceDiscovery{
{
ID: MakeResourceID(ResourceTypeVM, "node1", "101"),
ResourceType: ResourceTypeVM,
ResourceID: "101",
HostID: "node1",
ServiceName: "VM One",
Confidence: 0.95,
UpdatedAt: now.Add(-2 * time.Hour),
},
{
ID: MakeResourceID(ResourceTypeDocker, "host1", "app"),
ResourceType: ResourceTypeDocker,
ResourceID: "app",
HostID: "host1",
ServiceName: "App",
Confidence: 0.75,
UpdatedAt: now.Add(-2 * 24 * time.Hour),
},
}
summary := FormatDiscoverySummary(discoveries)
if !strings.Contains(summary, "[high confidence]") || !strings.Contains(summary, "[medium confidence]") {
t.Fatalf("unexpected summary: %s", summary)
}
tests := []struct {
name string
updated time.Time
expected string
}{
{name: "just-now", updated: now.Add(-30 * time.Second), expected: "just now"},
{name: "one-minute", updated: now.Add(-1 * time.Minute), expected: "1 minute ago"},
{name: "minutes", updated: now.Add(-10 * time.Minute), expected: "10 minutes ago"},
{name: "one-hour", updated: now.Add(-1 * time.Hour), expected: "1 hour ago"},
{name: "hours", updated: now.Add(-2 * time.Hour), expected: "2 hours ago"},
{name: "one-day", updated: now.Add(-24 * time.Hour), expected: "1 day ago"},
{name: "days", updated: now.Add(-3 * 24 * time.Hour), expected: "3 days ago"},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := FormatDiscoveryAge(&ResourceDiscovery{UpdatedAt: tt.updated})
if got != tt.expected {
t.Fatalf("expected %s, got %s", tt.expected, got)
}
})
}
}
func TestBuildResourceContextForPatrol(t *testing.T) {
store, err := NewStore(t.TempDir())
if err != nil {
t.Fatalf("NewStore error: %v", err)
}
store.crypto = nil
discovery := &ResourceDiscovery{
ID: MakeResourceID(ResourceTypeDocker, "host1", "app"),
ResourceType: ResourceTypeDocker,
ResourceID: "app",
HostID: "host1",
ServiceName: "App Service",
}
if err := store.Save(discovery); err != nil {
t.Fatalf("Save error: %v", err)
}
ctx := BuildResourceContextForPatrol(store, []string{discovery.ID})
if !strings.Contains(ctx, "App Service") {
t.Fatalf("unexpected patrol context: %s", ctx)
}
if BuildResourceContextForPatrol(nil, []string{discovery.ID}) != "" {
t.Fatalf("expected empty context for nil store")
}
if BuildResourceContextForPatrol(store, nil) != "" {
t.Fatalf("expected empty context for empty ids")
}
if BuildResourceContextForPatrol(store, []string{"missing"}) != "" {
t.Fatalf("expected empty context for missing discoveries")
}
}
func TestFilterImportantFactsLimit(t *testing.T) {
var facts []DiscoveryFact
for i := 0; i < 7; i++ {
facts = append(facts, DiscoveryFact{
Category: FactCategoryVersion,
Key: "k",
Value: "v",
Confidence: 0.9,
})
}
important := filterImportantFacts(facts)
if len(important) != 5 {
t.Fatalf("expected 5 facts, got %d", len(important))
}
}
internal/aidiscovery/service.go
-778
View File
@@ -1,778 +0,0 @@
// Package aidiscovery provides AI-powered infrastructure discovery capabilities.
// It discovers services, versions, configurations, and CLI access methods
// for VMs, LXCs, Docker containers, Kubernetes pods, and hosts.
package aidiscovery
import (
"context"
"encoding/json"
"fmt"
"strings"
"sync"
"time"
"github.com/rs/zerolog/log"
)
// StateProvider provides access to the current infrastructure state.
type StateProvider interface {
GetState() StateSnapshot
}
// StateSnapshot represents the infrastructure state. This mirrors models.StateSnapshot
// to avoid circular dependencies.
type StateSnapshot struct {
VMs []VM
Containers []Container
DockerHosts []DockerHost
}
// VM represents a virtual machine.
type VM struct {
VMID int
Name string
Node string
Status string
Instance string
}
// Container represents an LXC container.
type Container struct {
VMID int
Name string
Node string
Status string
Instance string
}
// DockerHost represents a Docker host.
type DockerHost struct {
AgentID string
Hostname string
Containers []DockerContainer
}
// DockerContainer represents a Docker container.
type DockerContainer struct {
ID string
Name string
Image string
Status string
Ports []DockerPort
Labels map[string]string
Mounts []DockerMount
}
// DockerPort represents a port mapping.
type DockerPort struct {
PublicPort int
PrivatePort int
Protocol string
}
// DockerMount represents a mount point.
type DockerMount struct {
Source string
Destination string
}
// AIAnalyzer provides AI analysis capabilities for discovery.
type AIAnalyzer interface {
AnalyzeForDiscovery(ctx context.Context, prompt string) (string, error)
}
// Service manages AI-powered infrastructure discovery.
type Service struct {
store *Store
scanner *DeepScanner
stateProvider StateProvider
aiAnalyzer AIAnalyzer
mu sync.RWMutex
running bool
stopCh chan struct{}
interval time.Duration
initialDelay time.Duration
lastRun time.Time
// Cache for AI analysis results (by image name)
analysisCache map[string]*AIAnalysisResponse
cacheMu sync.RWMutex
cacheExpiry time.Duration
lastCacheUpdate time.Time
}
// Config holds discovery service configuration.
type Config struct {
DataDir string
Interval time.Duration // How often to run background discovery
CacheExpiry time.Duration // How long to cache AI analysis results
}
// DefaultConfig returns the default discovery configuration.
func DefaultConfig() Config {
return Config{
Interval: 10 * time.Minute,
CacheExpiry: 1 * time.Hour,
}
}
// NewService creates a new discovery service.
func NewService(store *Store, scanner *DeepScanner, stateProvider StateProvider, cfg Config) *Service {
if cfg.Interval == 0 {
cfg.Interval = 10 * time.Minute
}
if cfg.CacheExpiry == 0 {
cfg.CacheExpiry = 1 * time.Hour
}
return &Service{
store: store,
scanner: scanner,
stateProvider: stateProvider,
interval: cfg.Interval,
initialDelay: 30 * time.Second,
cacheExpiry: cfg.CacheExpiry,
stopCh: make(chan struct{}),
analysisCache: make(map[string]*AIAnalysisResponse),
}
}
// SetAIAnalyzer sets the AI analyzer for discovery.
func (s *Service) SetAIAnalyzer(analyzer AIAnalyzer) {
s.mu.Lock()
defer s.mu.Unlock()
s.aiAnalyzer = analyzer
}
// Start begins the background discovery service.
func (s *Service) Start(ctx context.Context) {
s.mu.Lock()
if s.running {
s.mu.Unlock()
return
}
s.running = true
s.stopCh = make(chan struct{})
s.mu.Unlock()
log.Info().
Dur("interval", s.interval).
Msg("Starting AI-powered infrastructure discovery service")
go s.discoveryLoop(ctx)
}
// Stop stops the background discovery service.
func (s *Service) Stop() {
s.mu.Lock()
defer s.mu.Unlock()
if s.running {
close(s.stopCh)
s.running = false
}
}
// SetInterval updates the scan interval. Takes effect on next Start().
func (s *Service) SetInterval(interval time.Duration) {
s.mu.Lock()
defer s.mu.Unlock()
s.interval = interval
}
// IsRunning returns whether the background discovery loop is active.
func (s *Service) IsRunning() bool {
s.mu.RLock()
defer s.mu.RUnlock()
return s.running
}
// discoveryLoop runs periodic discovery.
func (s *Service) discoveryLoop(ctx context.Context) {
delay := s.initialDelay
if delay <= 0 {
delay = 30 * time.Second
}
// Run initial discovery after a short delay
select {
case <-time.After(delay):
case <-s.stopCh:
return
case <-ctx.Done():
return
}
s.runBackgroundDiscovery(ctx)
ticker := time.NewTicker(s.interval)
defer ticker.Stop()
for {
select {
case <-ticker.C:
s.runBackgroundDiscovery(ctx)
case <-s.stopCh:
log.Info().Msg("Stopping AI discovery service")
return
case <-ctx.Done():
log.Info().Msg("AI discovery context cancelled")
return
}
}
}
// runBackgroundDiscovery runs discovery on all resources in the background.
func (s *Service) runBackgroundDiscovery(ctx context.Context) {
defer func() {
if r := recover(); r != nil {
log.Error().Interface("panic", r).Stack().Msg("Recovered from panic in background AI discovery")
}
}()
s.mu.Lock()
s.lastRun = time.Now()
s.mu.Unlock()
// For background discovery, we only do shallow analysis based on metadata
// Deep scanning is triggered on-demand via DiscoverResource
if s.stateProvider == nil {
return
}
state := s.stateProvider.GetState()
s.discoverDockerContainers(ctx, state.DockerHosts)
}
// discoverDockerContainers runs discovery on Docker containers using metadata.
func (s *Service) discoverDockerContainers(ctx context.Context, hosts []DockerHost) {
s.mu.RLock()
analyzer := s.aiAnalyzer
s.mu.RUnlock()
if analyzer == nil {
log.Debug().Msg("AI analyzer not set, skipping Docker discovery")
return
}
for _, host := range hosts {
for _, container := range host.Containers {
select {
case <-ctx.Done():
return
default:
}
// Build resource ID
id := MakeResourceID(ResourceTypeDocker, host.AgentID, container.Name)
// Check if we already have a recent discovery
if !s.store.NeedsRefresh(id, s.cacheExpiry) {
continue
}
// Analyze using metadata (shallow discovery)
discovery := s.analyzeDockerContainer(ctx, analyzer, container, host)
if discovery != nil {
if err := s.store.Save(discovery); err != nil {
log.Warn().Err(err).Str("id", id).Msg("Failed to save discovery")
}
}
}
}
}
// analyzeDockerContainer analyzes a Docker container using AI.
func (s *Service) analyzeDockerContainer(ctx context.Context, analyzer AIAnalyzer, c DockerContainer, host DockerHost) *ResourceDiscovery {
// Check cache first
s.cacheMu.RLock()
cached, found := s.analysisCache[c.Image]
cacheValid := time.Since(s.lastCacheUpdate) < s.cacheExpiry
s.cacheMu.RUnlock()
var result *AIAnalysisResponse
if found && cacheValid {
result = cached
} else {
// Build prompt for AI analysis
prompt := s.buildMetadataAnalysisPrompt(c, host)
response, err := analyzer.AnalyzeForDiscovery(ctx, prompt)
if err != nil {
log.Warn().Err(err).Str("container", c.Name).Msg("AI analysis failed")
return nil
}
result = s.parseAIResponse(response)
if result == nil {
log.Warn().Str("container", c.Name).Msg("Failed to parse AI response")
return nil
}
// Cache the result
s.cacheMu.Lock()
s.analysisCache[c.Image] = result
s.lastCacheUpdate = time.Now()
s.cacheMu.Unlock()
}
// Skip unknown/low-confidence results
if result.ServiceType == "unknown" || result.Confidence < 0.5 {
return nil
}
// Build CLI access string
cliAccess := result.CLIAccess
if cliAccess != "" {
cliAccess = strings.ReplaceAll(cliAccess, "{container}", c.Name)
}
// Extract ports
var ports []PortInfo
for _, p := range c.Ports {
ports = append(ports, PortInfo{
Port: p.PrivatePort,
Protocol: p.Protocol,
Address: fmt.Sprintf(":%d", p.PublicPort),
})
}
return &ResourceDiscovery{
ID: MakeResourceID(ResourceTypeDocker, host.AgentID, c.Name),
ResourceType: ResourceTypeDocker,
ResourceID: c.Name,
HostID: host.AgentID,
Hostname: host.Hostname,
ServiceType: result.ServiceType,
ServiceName: result.ServiceName,
ServiceVersion: result.ServiceVersion,
Category: result.Category,
CLIAccess: cliAccess,
Facts: result.Facts,
ConfigPaths: result.ConfigPaths,
DataPaths: result.DataPaths,
Ports: ports,
Confidence: result.Confidence,
AIReasoning: result.Reasoning,
DiscoveredAt: time.Now(),
UpdatedAt: time.Now(),
}
}
// DiscoverResource performs deep discovery on a specific resource.
func (s *Service) DiscoverResource(ctx context.Context, req DiscoveryRequest) (*ResourceDiscovery, error) {
resourceID := MakeResourceID(req.ResourceType, req.HostID, req.ResourceID)
// Check if we have a recent discovery and force isn't set
if !req.Force {
existing, err := s.store.Get(resourceID)
if err == nil && existing != nil {
age := time.Since(existing.UpdatedAt)
if age < 5*time.Minute {
log.Debug().Str("id", resourceID).Dur("age", age).Msg("Using recent discovery")
return existing, nil
}
}
}
s.mu.RLock()
analyzer := s.aiAnalyzer
s.mu.RUnlock()
if analyzer == nil {
return nil, fmt.Errorf("AI analyzer not configured")
}
// Run deep scan if scanner is available
var scanResult *ScanResult
if s.scanner != nil {
var err error
scanResult, err = s.scanner.Scan(ctx, req)
if err != nil {
log.Warn().Err(err).Str("id", resourceID).Msg("Deep scan failed, using metadata only")
}
}
// Build analysis request
analysisReq := AIAnalysisRequest{
ResourceType: req.ResourceType,
ResourceID: req.ResourceID,
HostID: req.HostID,
Hostname: req.Hostname,
}
if scanResult != nil {
analysisReq.CommandOutputs = scanResult.CommandOutputs
}
// Add metadata if available
if s.stateProvider != nil {
analysisReq.Metadata = s.getResourceMetadata(req)
}
// Build prompt and analyze
prompt := s.buildDeepAnalysisPrompt(analysisReq)
response, err := analyzer.AnalyzeForDiscovery(ctx, prompt)
if err != nil {
return nil, fmt.Errorf("AI analysis failed: %w", err)
}
result := s.parseAIResponse(response)
if result == nil {
// Truncate response for error message
truncated := response
if len(truncated) > 500 {
truncated = truncated[:500] + "..."
}
return nil, fmt.Errorf("failed to parse AI response: %s", truncated)
}
// Build discovery result
discovery := &ResourceDiscovery{
ID: resourceID,
ResourceType: req.ResourceType,
ResourceID: req.ResourceID,
HostID: req.HostID,
Hostname: req.Hostname,
ServiceType: result.ServiceType,
ServiceName: result.ServiceName,
ServiceVersion: result.ServiceVersion,
Category: result.Category,
CLIAccess: s.formatCLIAccess(req.ResourceType, req.ResourceID, result.CLIAccess),
Facts: result.Facts,
ConfigPaths: result.ConfigPaths,
DataPaths: result.DataPaths,
Ports: result.Ports,
Confidence: result.Confidence,
AIReasoning: result.Reasoning,
DiscoveredAt: time.Now(),
UpdatedAt: time.Now(),
}
if scanResult != nil {
discovery.RawCommandOutput = scanResult.CommandOutputs
discovery.ScanDuration = scanResult.CompletedAt.Sub(scanResult.StartedAt).Milliseconds()
}
// Preserve user notes from existing discovery
existing, _ := s.store.Get(resourceID)
if existing != nil {
discovery.UserNotes = existing.UserNotes
discovery.UserSecrets = existing.UserSecrets
if discovery.DiscoveredAt.IsZero() || existing.DiscoveredAt.Before(discovery.DiscoveredAt) {
discovery.DiscoveredAt = existing.DiscoveredAt
}
}
// Save discovery
if err := s.store.Save(discovery); err != nil {
return nil, fmt.Errorf("failed to save discovery: %w", err)
}
return discovery, nil
}
// getResourceMetadata retrieves metadata for a resource from the state.
func (s *Service) getResourceMetadata(req DiscoveryRequest) map[string]any {
if s.stateProvider == nil {
return nil
}
state := s.stateProvider.GetState()
metadata := make(map[string]any)
switch req.ResourceType {
case ResourceTypeLXC:
for _, c := range state.Containers {
if fmt.Sprintf("%d", c.VMID) == req.ResourceID && c.Node == req.HostID {
metadata["name"] = c.Name
metadata["status"] = c.Status
metadata["vmid"] = c.VMID
break
}
}
case ResourceTypeVM:
for _, vm := range state.VMs {
if fmt.Sprintf("%d", vm.VMID) == req.ResourceID && vm.Node == req.HostID {
metadata["name"] = vm.Name
metadata["status"] = vm.Status
metadata["vmid"] = vm.VMID
break
}
}
case ResourceTypeDocker:
for _, host := range state.DockerHosts {
if host.AgentID == req.HostID || host.Hostname == req.HostID {
for _, c := range host.Containers {
if c.Name == req.ResourceID {
metadata["image"] = c.Image
metadata["status"] = c.Status
metadata["labels"] = c.Labels
break
}
}
break
}
}
}
return metadata
}
// formatCLIAccess formats the CLI access string with actual values.
func (s *Service) formatCLIAccess(resourceType ResourceType, resourceID, cliTemplate string) string {
if cliTemplate == "" {
// Use default template
cliTemplate = GetCLIAccessTemplate(resourceType)
}
result := cliTemplate
result = strings.ReplaceAll(result, "{vmid}", resourceID)
result = strings.ReplaceAll(result, "{container}", resourceID)
result = strings.ReplaceAll(result, "{command}", "...")
return result
}
// buildMetadataAnalysisPrompt builds a prompt for shallow metadata-based analysis.
func (s *Service) buildMetadataAnalysisPrompt(c DockerContainer, host DockerHost) string {
info := map[string]any{
"name": c.Name,
"image": c.Image,
"status": c.Status,
"host": host.Hostname,
}
if len(c.Ports) > 0 {
var ports []map[string]any
for _, p := range c.Ports {
ports = append(ports, map[string]any{
"public": p.PublicPort,
"private": p.PrivatePort,
"protocol": p.Protocol,
})
}
info["ports"] = ports
}
if len(c.Labels) > 0 {
info["labels"] = c.Labels
}
if len(c.Mounts) > 0 {
var mounts []string
for _, m := range c.Mounts {
mounts = append(mounts, m.Destination)
}
info["mounts"] = mounts
}
infoJSON, _ := json.MarshalIndent(info, "", " ")
return fmt.Sprintf(`Analyze this Docker container and identify what service it's running.
Container Information:
%s
Based on the image name, ports, labels, and mounts, determine:
1. What service/application is this?
2. What category does it belong to?
3. How should CLI commands be executed?
Respond in this exact JSON format:
{
"service_type": "lowercase_type",
"service_name": "Human Readable Name",
"service_version": "version if detectable from image tag",
"category": "database|web_server|cache|monitoring|backup|nvr|storage|container|network|security|media|home_automation|unknown",
"cli_access": "docker exec {container} <cli-tool>",
"facts": [],
"config_paths": [],
"data_paths": [],
"ports": [],
"confidence": 0.0-1.0,
"reasoning": "Brief explanation"
}
Respond with ONLY valid JSON.`, string(infoJSON))
}
// buildDeepAnalysisPrompt builds a prompt for deep analysis with command outputs.
func (s *Service) buildDeepAnalysisPrompt(req AIAnalysisRequest) string {
var sections []string
sections = append(sections, fmt.Sprintf(`Resource Type: %s
Resource ID: %s
Host: %s (%s)`, req.ResourceType, req.ResourceID, req.Hostname, req.HostID))
if len(req.Metadata) > 0 {
metaJSON, _ := json.MarshalIndent(req.Metadata, "", " ")
sections = append(sections, fmt.Sprintf("Metadata:\n%s", string(metaJSON)))
}
if len(req.CommandOutputs) > 0 {
sections = append(sections, "Command Outputs:")
for name, output := range req.CommandOutputs {
// Truncate long outputs
if len(output) > 2000 {
output = output[:2000] + "\n... (truncated)"
}
sections = append(sections, fmt.Sprintf("--- %s ---\n%s", name, output))
}
}
return fmt.Sprintf(`Analyze this infrastructure resource and provide detailed discovery information.
%s
Based on all available information, determine:
1. What service/application is running?
2. What version is it?
3. What are the important configuration paths?
4. What data paths should be backed up?
5. What ports are in use?
6. Any special hardware (GPU, TPU, etc.)?
7. Any dependencies (databases, message queues, etc.)?
Respond in this exact JSON format:
{
"service_type": "lowercase_type (e.g., frigate, postgres, pbs)",
"service_name": "Human Readable Name",
"service_version": "version number if found",
"category": "database|web_server|cache|monitoring|backup|nvr|storage|container|virtualizer|network|security|media|home_automation|unknown",
"cli_access": "command to access this service's CLI",
"facts": [
{"category": "version|config|service|port|hardware|network|storage|dependency|security", "key": "fact_name", "value": "fact_value", "source": "command_name", "confidence": 0.9}
],
"config_paths": ["/path/to/config.yml"],
"data_paths": ["/path/to/data"],
"ports": [{"port": 8080, "protocol": "tcp", "process": "nginx", "address": "0.0.0.0"}],
"confidence": 0.0-1.0,
"reasoning": "Explanation of identification"
}
Important:
- Extract version numbers from package lists, process output, or config files
- Identify config and data paths from mount points and file listings
- Note any special hardware like Coral TPU, NVIDIA GPU
- For LXC/VM, the CLI access should use pct exec or qm guest exec
- For Docker, use docker exec
Respond with ONLY valid JSON.`, strings.Join(sections, "\n\n"))
}
// parseAIResponse parses the AI's JSON response.
func (s *Service) parseAIResponse(response string) *AIAnalysisResponse {
log.Debug().Str("raw_response", response).Msg("AI discovery raw response")
response = strings.TrimSpace(response)
// Handle markdown code blocks
if strings.HasPrefix(response, "```") {
lines := strings.Split(response, "\n")
var jsonLines []string
inBlock := false
for _, line := range lines {
if strings.HasPrefix(line, "```") {
inBlock = !inBlock
continue
}
if inBlock {
jsonLines = append(jsonLines, line)
}
}
response = strings.Join(jsonLines, "\n")
}
// Find JSON object
start := strings.Index(response, "{")
end := strings.LastIndex(response, "}")
if start >= 0 && end > start {
response = response[start : end+1]
}
var result AIAnalysisResponse
if err := json.Unmarshal([]byte(response), &result); err != nil {
log.Debug().Err(err).Str("response", response).Msg("Failed to parse AI response")
return nil
}
// Set discovered_at for facts
now := time.Now()
for i := range result.Facts {
result.Facts[i].DiscoveredAt = now
}
return &result
}
// GetDiscovery retrieves a discovery by ID.
func (s *Service) GetDiscovery(id string) (*ResourceDiscovery, error) {
return s.store.Get(id)
}
// GetDiscoveryByResource retrieves a discovery by resource type and ID.
func (s *Service) GetDiscoveryByResource(resourceType ResourceType, hostID, resourceID string) (*ResourceDiscovery, error) {
return s.store.GetByResource(resourceType, hostID, resourceID)
}
// ListDiscoveries returns all discoveries.
func (s *Service) ListDiscoveries() ([]*ResourceDiscovery, error) {
return s.store.List()
}
// ListDiscoveriesByType returns discoveries for a specific resource type.
func (s *Service) ListDiscoveriesByType(resourceType ResourceType) ([]*ResourceDiscovery, error) {
return s.store.ListByType(resourceType)
}
// ListDiscoveriesByHost returns discoveries for a specific host.
func (s *Service) ListDiscoveriesByHost(hostID string) ([]*ResourceDiscovery, error) {
return s.store.ListByHost(hostID)
}
// UpdateNotes updates user notes for a discovery.
func (s *Service) UpdateNotes(id string, notes string, secrets map[string]string) error {
return s.store.UpdateNotes(id, notes, secrets)
}
// DeleteDiscovery deletes a discovery.
func (s *Service) DeleteDiscovery(id string) error {
return s.store.Delete(id)
}
// GetProgress returns the progress of an ongoing discovery.
func (s *Service) GetProgress(resourceID string) *DiscoveryProgress {
if s.scanner == nil {
return nil
}
return s.scanner.GetProgress(resourceID)
}
// GetStatus returns the service status.
func (s *Service) GetStatus() map[string]any {
s.mu.RLock()
defer s.mu.RUnlock()
s.cacheMu.RLock()
cacheSize := len(s.analysisCache)
s.cacheMu.RUnlock()
return map[string]any{
"running": s.running,
"last_run": s.lastRun,
"interval": s.interval.String(),
"cache_size": cacheSize,
"ai_analyzer_set": s.aiAnalyzer != nil,
"scanner_set": s.scanner != nil,
"store_set": s.store != nil,
}
}
// ClearCache clears the AI analysis cache.
func (s *Service) ClearCache() {
s.cacheMu.Lock()
defer s.cacheMu.Unlock()
s.analysisCache = make(map[string]*AIAnalysisResponse)
s.lastCacheUpdate = time.Time{}
}
internal/aidiscovery/service_test.go
-658
View File
@@ -1,658 +0,0 @@
package aidiscovery
import (
"context"
"os"
"path/filepath"
"strings"
"sync"
"testing"
"time"
)
type stubAnalyzer struct {
mu sync.Mutex
calls int
response string
}
func (s *stubAnalyzer) AnalyzeForDiscovery(ctx context.Context, prompt string) (string, error) {
s.mu.Lock()
s.calls++
s.mu.Unlock()
return s.response, nil
}
type errorAnalyzer struct{}
func (errorAnalyzer) AnalyzeForDiscovery(ctx context.Context, prompt string) (string, error) {
return "", context.Canceled
}
type stubStateProvider struct {
state StateSnapshot
}
func (s stubStateProvider) GetState() StateSnapshot {
return s.state
}
type panicStateProvider struct{}
func (panicStateProvider) GetState() StateSnapshot {
panic("boom")
}
func TestService_parseAIResponse_Markdown(t *testing.T) {
service := &Service{}
response := "```json\n{\n \"service_type\": \"nginx\",\n \"service_name\": \"Nginx\",\n \"service_version\": \"1.2\",\n \"category\": \"web_server\",\n \"cli_access\": \"docker exec {container} bash\",\n \"facts\": [{\"category\": \"version\", \"key\": \"nginx\", \"value\": \"1.2\", \"source\": \"cmd\", \"confidence\": 0.9}],\n \"config_paths\": [\"/etc/nginx/nginx.conf\"],\n \"data_paths\": [\"/var/www\"],\n \"ports\": [{\"port\": 80, \"protocol\": \"tcp\", \"process\": \"nginx\", \"address\": \"0.0.0.0\"}],\n \"confidence\": 0.9,\n \"reasoning\": \"image name\"\n}\n```"
parsed := service.parseAIResponse(response)
if parsed == nil {
t.Fatalf("expected parsed response")
}
if parsed.ServiceType != "nginx" || parsed.ServiceName != "Nginx" {
t.Fatalf("unexpected parsed result: %#v", parsed)
}
if len(parsed.Facts) != 1 || parsed.Facts[0].DiscoveredAt.IsZero() {
t.Fatalf("expected fact timestamp set: %#v", parsed.Facts)
}
if service.parseAIResponse("not json") != nil {
t.Fatalf("expected nil for invalid json")
}
}
func TestService_analyzeDockerContainer_CacheAndPorts(t *testing.T) {
store, err := NewStore(t.TempDir())
if err != nil {
t.Fatalf("NewStore error: %v", err)
}
store.crypto = nil
service := NewService(store, nil, nil, Config{CacheExpiry: time.Hour})
analyzer := &stubAnalyzer{
response: `{"service_type":"nginx","service_name":"Nginx","service_version":"1.2","category":"web_server","cli_access":"docker exec {container} nginx -v","facts":[],"config_paths":[],"data_paths":[],"ports":[],"confidence":0.9,"reasoning":"image"}`,
}
container := DockerContainer{
Name: "web",
Image: "nginx:latest",
Status: "running",
Ports: []DockerPort{
{PublicPort: 8080, PrivatePort: 80, Protocol: "tcp"},
},
}
host := DockerHost{
AgentID: "host1",
Hostname: "host1",
}
first := service.analyzeDockerContainer(context.Background(), analyzer, container, host)
if first == nil {
t.Fatalf("expected discovery")
}
if !strings.Contains(first.CLIAccess, "web") {
t.Fatalf("expected cli access to include container name, got %s", first.CLIAccess)
}
if len(first.Ports) != 1 || first.Ports[0].Port != 80 || first.Ports[0].Address != ":8080" {
t.Fatalf("unexpected ports: %#v", first.Ports)
}
second := service.analyzeDockerContainer(context.Background(), analyzer, container, host)
if second == nil {
t.Fatalf("expected cached discovery")
}
analyzer.mu.Lock()
calls := analyzer.calls
analyzer.mu.Unlock()
if calls != 1 {
t.Fatalf("expected analyzer called once, got %d", calls)
}
lowAnalyzer := &stubAnalyzer{
response: `{"service_type":"unknown","service_name":"","service_version":"","category":"unknown","cli_access":"","facts":[],"config_paths":[],"data_paths":[],"ports":[],"confidence":0.4,"reasoning":""}`,
}
lowContainer := DockerContainer{Name: "mystery", Image: "unknown:latest"}
if got := service.analyzeDockerContainer(context.Background(), lowAnalyzer, lowContainer, host); got != nil {
t.Fatalf("expected low confidence discovery to be skipped")
}
}
func TestService_DiscoverResource_RecentAndNoAnalyzer(t *testing.T) {
store, err := NewStore(t.TempDir())
if err != nil {
t.Fatalf("NewStore error: %v", err)
}
store.crypto = nil
service := NewService(store, nil, nil, DefaultConfig())
req := DiscoveryRequest{
ResourceType: ResourceTypeDocker,
ResourceID: "nginx",
HostID: "host1",
Hostname: "host1",
}
discovery := &ResourceDiscovery{
ID: MakeResourceID(req.ResourceType, req.HostID, req.ResourceID),
ResourceType: req.ResourceType,
ResourceID: req.ResourceID,
HostID: req.HostID,
Hostname: req.Hostname,
ServiceName: "Existing",
}
if err := store.Save(discovery); err != nil {
t.Fatalf("Save error: %v", err)
}
found, err := service.DiscoverResource(context.Background(), req)
if err != nil {
t.Fatalf("DiscoverResource error: %v", err)
}
if found == nil || found.ServiceName != "Existing" {
t.Fatalf("unexpected discovery: %#v", found)
}
_, err = service.DiscoverResource(context.Background(), DiscoveryRequest{
ResourceType: ResourceTypeVM,
ResourceID: "101",
HostID: "node1",
Hostname: "node1",
Force: true,
})
if err == nil || !strings.Contains(err.Error(), "AI analyzer") {
t.Fatalf("expected analyzer error, got %v", err)
}
service.SetAIAnalyzer(errorAnalyzer{})
_, err = service.DiscoverResource(context.Background(), DiscoveryRequest{
ResourceType: ResourceTypeVM,
ResourceID: "102",
HostID: "node1",
Hostname: "node1",
Force: true,
})
if err == nil || !strings.Contains(err.Error(), "AI analysis failed") {
t.Fatalf("expected analysis error, got %v", err)
}
service.SetAIAnalyzer(&stubAnalyzer{response: "not json"})
_, err = service.DiscoverResource(context.Background(), DiscoveryRequest{
ResourceType: ResourceTypeVM,
ResourceID: "103",
HostID: "node1",
Hostname: "node1",
Force: true,
})
if err == nil || !strings.Contains(err.Error(), "failed to parse") {
t.Fatalf("expected parse error, got %v", err)
}
}
func TestService_getResourceMetadata(t *testing.T) {
state := StateSnapshot{
VMs: []VM{
{VMID: 101, Name: "vm1", Node: "node1", Status: "running"},
},
Containers: []Container{
{VMID: 201, Name: "lxc1", Node: "node2", Status: "stopped"},
},
DockerHosts: []DockerHost{
{
AgentID: "agent1",
Hostname: "dock1",
Containers: []DockerContainer{
{Name: "redis", Image: "redis:latest", Status: "running", Labels: map[string]string{"tier": "cache"}},
},
},
},
}
service := NewService(nil, nil, stubStateProvider{state: state}, DefaultConfig())
vmMeta := service.getResourceMetadata(DiscoveryRequest{
ResourceType: ResourceTypeVM,
ResourceID: "101",
HostID: "node1",
})
if vmMeta["name"] != "vm1" || vmMeta["vmid"] != 101 {
t.Fatalf("unexpected vm metadata: %#v", vmMeta)
}
lxcMeta := service.getResourceMetadata(DiscoveryRequest{
ResourceType: ResourceTypeLXC,
ResourceID: "201",
HostID: "node2",
})
if lxcMeta["name"] != "lxc1" || lxcMeta["status"] != "stopped" {
t.Fatalf("unexpected lxc metadata: %#v", lxcMeta)
}
dockerMeta := service.getResourceMetadata(DiscoveryRequest{
ResourceType: ResourceTypeDocker,
ResourceID: "redis",
HostID: "agent1",
})
if dockerMeta["image"] != "redis:latest" || dockerMeta["status"] != "running" {
t.Fatalf("unexpected docker metadata: %#v", dockerMeta)
}
dockerByHost := service.getResourceMetadata(DiscoveryRequest{
ResourceType: ResourceTypeDocker,
ResourceID: "redis",
HostID: "dock1",
})
if dockerByHost["image"] != "redis:latest" {
t.Fatalf("unexpected docker hostname metadata: %#v", dockerByHost)
}
}
func TestService_formatCLIAccessAndStatus(t *testing.T) {
service := NewService(nil, nil, nil, DefaultConfig())
formatted := service.formatCLIAccess(ResourceTypeDocker, "redis", "")
if !strings.Contains(formatted, "redis") || !strings.Contains(formatted, "...") {
t.Fatalf("unexpected cli access: %s", formatted)
}
service.analysisCache = map[string]*AIAnalysisResponse{"nginx:latest": {ServiceType: "nginx"}}
service.running = true
status := service.GetStatus()
if status["running"] != true || status["cache_size"] != 1 {
t.Fatalf("unexpected status: %#v", status)
}
service.ClearCache()
if len(service.analysisCache) != 0 {
t.Fatalf("expected cache cleared")
}
}
func TestService_DefaultsAndSetAnalyzer(t *testing.T) {
service := NewService(nil, nil, nil, Config{})
if service.interval == 0 || service.cacheExpiry == 0 {
t.Fatalf("expected defaults for interval and cache expiry")
}
analyzer := &stubAnalyzer{response: `{}`}
service.SetAIAnalyzer(analyzer)
if service.aiAnalyzer == nil {
t.Fatalf("expected analyzer set")
}
if service.GetProgress("missing") != nil {
t.Fatalf("expected nil progress without scanner")
}
if service.getResourceMetadata(DiscoveryRequest{}) != nil {
t.Fatalf("expected nil metadata without state provider")
}
}
func TestService_RunBackgroundDiscoveryAndWrappers(t *testing.T) {
store, err := NewStore(t.TempDir())
if err != nil {
t.Fatalf("NewStore error: %v", err)
}
store.crypto = nil
state := StateSnapshot{
DockerHosts: []DockerHost{
{
AgentID: "host1",
Hostname: "host1",
Containers: []DockerContainer{
{Name: "web", Image: "nginx:latest", Status: "running"},
},
},
},
}
service := NewService(store, nil, stubStateProvider{state: state}, DefaultConfig())
service.SetAIAnalyzer(&stubAnalyzer{
response: `{"service_type":"nginx","service_name":"Nginx","service_version":"1.2","category":"web_server","cli_access":"docker exec {container} nginx -v","facts":[],"config_paths":[],"data_paths":[],"ports":[],"confidence":0.9,"reasoning":"image"}`,
})
service.runBackgroundDiscovery(context.Background())
id := MakeResourceID(ResourceTypeDocker, "host1", "web")
if got, err := service.GetDiscovery(id); err != nil || got == nil {
t.Fatalf("GetDiscovery error: %v", err)
}
if got, err := service.GetDiscoveryByResource(ResourceTypeDocker, "host1", "web"); err != nil || got == nil {
t.Fatalf("GetDiscoveryByResource error: %v", err)
}
if list, err := service.ListDiscoveries(); err != nil || len(list) != 1 {
t.Fatalf("ListDiscoveries unexpected: %v len=%d", err, len(list))
}
if list, err := service.ListDiscoveriesByType(ResourceTypeDocker); err != nil || len(list) != 1 {
t.Fatalf("ListDiscoveriesByType unexpected: %v len=%d", err, len(list))
}
if list, err := service.ListDiscoveriesByHost("host1"); err != nil || len(list) != 1 {
t.Fatalf("ListDiscoveriesByHost unexpected: %v len=%d", err, len(list))
}
if err := service.UpdateNotes(id, "note", map[string]string{"k": "v"}); err != nil {
t.Fatalf("UpdateNotes error: %v", err)
}
updated, err := service.GetDiscovery(id)
if err != nil || updated.UserNotes != "note" {
t.Fatalf("expected updated notes: %#v err=%v", updated, err)
}
scanner := NewDeepScanner(&stubExecutor{})
scanner.progress[id] = &DiscoveryProgress{ResourceID: id}
service.scanner = scanner
if service.GetProgress(id) == nil {
t.Fatalf("expected progress")
}
if err := service.DeleteDiscovery(id); err != nil {
t.Fatalf("DeleteDiscovery error: %v", err)
}
service.stateProvider = nil
service.runBackgroundDiscovery(context.Background())
}
func TestService_PromptsAndDiscoveryLoop(t *testing.T) {
service := NewService(nil, nil, nil, DefaultConfig())
container := DockerContainer{
Name: "web",
Image: "nginx:latest",
Status: "running",
Ports: []DockerPort{
{PublicPort: 8080, PrivatePort: 80, Protocol: "tcp"},
},
Labels: map[string]string{"app": "nginx"},
Mounts: []DockerMount{{Destination: "/etc/nginx"}},
}
host := DockerHost{Hostname: "host1"}
prompt := service.buildMetadataAnalysisPrompt(container, host)
if !strings.Contains(prompt, "\"ports\"") || !strings.Contains(prompt, "\"labels\"") || !strings.Contains(prompt, "\"mounts\"") {
t.Fatalf("unexpected metadata prompt: %s", prompt)
}
longOutput := strings.Repeat("a", 2100)
deepPrompt := service.buildDeepAnalysisPrompt(AIAnalysisRequest{
ResourceType: ResourceTypeDocker,
ResourceID: "web",
HostID: "host1",
Hostname: "host1",
Metadata: map[string]any{"image": "nginx"},
CommandOutputs: map[string]string{
"ps": longOutput,
},
})
if !strings.Contains(deepPrompt, "(truncated)") || !strings.Contains(deepPrompt, "Metadata:") {
t.Fatalf("unexpected deep prompt")
}
ctx, cancel := context.WithCancel(context.Background())
cancel()
service.initialDelay = time.Millisecond
service.Start(ctx)
service.Start(ctx)
service.Stop()
service.stopCh = make(chan struct{})
close(service.stopCh)
service.discoveryLoop(context.Background())
service.initialDelay = 0
service.stopCh = make(chan struct{})
close(service.stopCh)
service.discoveryLoop(context.Background())
}
func TestService_DiscoveryLoop_StopAndCancel(t *testing.T) {
state := StateSnapshot{
DockerHosts: []DockerHost{
{
AgentID: "host1",
Hostname: "host1",
Containers: []DockerContainer{
{Name: "web", Image: "nginx:latest", Status: "running"},
},
},
},
}
runLoop := func(stopWithCancel bool) {
store, err := NewStore(t.TempDir())
if err != nil {
t.Fatalf("NewStore error: %v", err)
}
store.crypto = nil
service := NewService(store, nil, stubStateProvider{state: state}, DefaultConfig())
analyzer := &stubAnalyzer{
response: `{"service_type":"nginx","service_name":"Nginx","service_version":"1.2","category":"web_server","cli_access":"docker exec {container} nginx -v","facts":[],"config_paths":[],"data_paths":[],"ports":[],"confidence":0.9,"reasoning":"image"}`,
}
service.SetAIAnalyzer(analyzer)
service.initialDelay = time.Millisecond
service.interval = time.Millisecond
service.cacheExpiry = time.Nanosecond
done := make(chan struct{})
ctx, cancel := context.WithCancel(context.Background())
go func() {
service.discoveryLoop(ctx)
close(done)
}()
time.Sleep(5 * time.Millisecond)
if stopWithCancel {
cancel()
} else {
close(service.stopCh)
}
select {
case <-done:
case <-time.After(50 * time.Millisecond):
t.Fatalf("discoveryLoop did not stop")
}
analyzer.mu.Lock()
calls := analyzer.calls
analyzer.mu.Unlock()
if calls < 2 {
t.Fatalf("expected multiple discoveries, got %d", calls)
}
}
runLoop(false)
runLoop(true)
}
func TestService_DiscoverDockerContainersSkips(t *testing.T) {
store, err := NewStore(t.TempDir())
if err != nil {
t.Fatalf("NewStore error: %v", err)
}
store.crypto = nil
service := NewService(store, nil, nil, DefaultConfig())
service.discoverDockerContainers(context.Background(), []DockerHost{{AgentID: "host1"}})
service.SetAIAnalyzer(&stubAnalyzer{
response: `{"service_type":"nginx","service_name":"Nginx","service_version":"1.2","category":"web_server","cli_access":"docker exec {container} nginx -v","facts":[],"config_paths":[],"data_paths":[],"ports":[],"confidence":0.9,"reasoning":"image"}`,
})
id := MakeResourceID(ResourceTypeDocker, "host1", "web")
if err := store.Save(&ResourceDiscovery{ID: id, ResourceType: ResourceTypeDocker}); err != nil {
t.Fatalf("Save error: %v", err)
}
service.cacheExpiry = time.Hour
service.discoverDockerContainers(context.Background(), []DockerHost{
{AgentID: "host1", Containers: []DockerContainer{{Name: "web", Image: "nginx:latest"}}},
})
badAnalyzer := &stubAnalyzer{response: "not json"}
if got := service.analyzeDockerContainer(context.Background(), badAnalyzer, DockerContainer{Name: "bad", Image: "bad"}, DockerHost{AgentID: "host1"}); got != nil {
t.Fatalf("expected nil for bad analysis")
}
canceled, cancel := context.WithCancel(context.Background())
cancel()
analyzer := &stubAnalyzer{response: `{"service_type":"nginx","service_name":"Nginx","service_version":"1.2","category":"web_server","cli_access":"docker exec {container} nginx -v","facts":[],"config_paths":[],"data_paths":[],"ports":[],"confidence":0.9,"reasoning":"image"}`}
service.SetAIAnalyzer(analyzer)
service.discoverDockerContainers(canceled, []DockerHost{
{AgentID: "host1", Containers: []DockerContainer{{Name: "web2", Image: "nginx:latest"}}},
})
analyzer.mu.Lock()
calls := analyzer.calls
analyzer.mu.Unlock()
if calls != 0 {
t.Fatalf("expected analyzer not called on canceled context")
}
errAnalyzer := errorAnalyzer{}
if got := service.analyzeDockerContainer(context.Background(), errAnalyzer, DockerContainer{Name: "err", Image: "err"}, DockerHost{AgentID: "host1"}); got != nil {
t.Fatalf("expected nil when analyzer returns error")
}
storePath := filepath.Join(t.TempDir(), "file")
if err := os.WriteFile(storePath, []byte("x"), 0600); err != nil {
t.Fatalf("WriteFile error: %v", err)
}
service.store.dataDir = storePath
service.discoverDockerContainers(context.Background(), []DockerHost{
{AgentID: "host1", Containers: []DockerContainer{{Name: "web3", Image: "nginx:latest"}}},
})
}
func TestService_RunBackgroundDiscoveryRecover(t *testing.T) {
service := NewService(nil, nil, panicStateProvider{}, DefaultConfig())
service.runBackgroundDiscovery(context.Background())
}
func TestService_DiscoverResource_SaveError(t *testing.T) {
store, err := NewStore(t.TempDir())
if err != nil {
t.Fatalf("NewStore error: %v", err)
}
store.crypto = nil
badPath := filepath.Join(t.TempDir(), "file")
if err := os.WriteFile(badPath, []byte("x"), 0600); err != nil {
t.Fatalf("WriteFile error: %v", err)
}
store.dataDir = badPath
service := NewService(store, nil, nil, DefaultConfig())
service.SetAIAnalyzer(&stubAnalyzer{
response: `{"service_type":"nginx","service_name":"Nginx","service_version":"1.2","category":"web_server","cli_access":"docker exec {container} nginx -v","facts":[],"config_paths":[],"data_paths":[],"ports":[],"confidence":0.9,"reasoning":"image"}`,
})
_, err = service.DiscoverResource(context.Background(), DiscoveryRequest{
ResourceType: ResourceTypeDocker,
ResourceID: "web",
HostID: "host1",
Hostname: "host1",
Force: true,
})
if err == nil || !strings.Contains(err.Error(), "failed to save discovery") {
t.Fatalf("expected save error, got %v", err)
}
}
func TestService_DiscoverResource_ScanError(t *testing.T) {
store, err := NewStore(t.TempDir())
if err != nil {
t.Fatalf("NewStore error: %v", err)
}
store.crypto = nil
scanner := NewDeepScanner(nil)
service := NewService(store, scanner, nil, DefaultConfig())
service.SetAIAnalyzer(&stubAnalyzer{
response: `{"service_type":"nginx","service_name":"Nginx","service_version":"1.2","category":"web_server","cli_access":"docker exec {container} nginx -v","facts":[],"config_paths":[],"data_paths":[],"ports":[],"confidence":0.9,"reasoning":"image"}`,
})
_, err = service.DiscoverResource(context.Background(), DiscoveryRequest{
ResourceType: ResourceTypeDocker,
ResourceID: "web",
HostID: "host1",
Hostname: "host1",
Force: true,
})
if err != nil {
t.Fatalf("expected scan error to be tolerated, got %v", err)
}
}
func TestService_DiscoveryLoop_ContextDoneAtStart(t *testing.T) {
service := NewService(nil, nil, nil, DefaultConfig())
service.initialDelay = time.Hour
service.stopCh = make(chan struct{})
ctx, cancel := context.WithCancel(context.Background())
cancel()
service.discoveryLoop(ctx)
}
func TestService_DiscoverResource_WithScanResult(t *testing.T) {
store, err := NewStore(t.TempDir())
if err != nil {
t.Fatalf("NewStore error: %v", err)
}
store.crypto = nil
exec := &stubExecutor{
agents: []ConnectedAgent{{AgentID: "host1", Hostname: "host1"}},
}
scanner := NewDeepScanner(exec)
scanner.maxParallel = 1
state := StateSnapshot{
DockerHosts: []DockerHost{
{
AgentID: "host1",
Hostname: "host1",
Containers: []DockerContainer{
{Name: "web", Image: "nginx:latest", Status: "running"},
},
},
},
}
service := NewService(store, scanner, stubStateProvider{state: state}, DefaultConfig())
service.SetAIAnalyzer(&stubAnalyzer{
response: `{"service_type":"nginx","service_name":"Nginx","service_version":"1.2","category":"web_server","cli_access":"docker exec {container} nginx -v","facts":[],"config_paths":[],"data_paths":[],"ports":[{"port":80,"protocol":"tcp","process":"nginx","address":"0.0.0.0"}],"confidence":0.9,"reasoning":"image"}`,
})
existing := &ResourceDiscovery{
ID: MakeResourceID(ResourceTypeDocker, "host1", "web"),
ResourceType: ResourceTypeDocker,
ResourceID: "web",
HostID: "host1",
Hostname: "host1",
UserNotes: "keep",
UserSecrets: map[string]string{"token": "secret"},
DiscoveredAt: time.Now().Add(-2 * time.Hour),
}
if err := store.Save(existing); err != nil {
t.Fatalf("Save error: %v", err)
}
found, err := service.DiscoverResource(context.Background(), DiscoveryRequest{
ResourceType: ResourceTypeDocker,
ResourceID: "web",
HostID: "host1",
Hostname: "host1",
Force: true,
})
if err != nil {
t.Fatalf("DiscoverResource error: %v", err)
}
if found.UserNotes != "keep" || found.UserSecrets["token"] != "secret" {
t.Fatalf("expected user fields preserved: %#v", found)
}
if len(found.RawCommandOutput) == 0 {
t.Fatalf("expected raw command output")
}
if found.DiscoveredAt.After(existing.DiscoveredAt) {
t.Fatalf("expected older discovered_at preserved")
}
}
internal/aidiscovery/store.go
-347
View File
@@ -1,347 +0,0 @@
package aidiscovery
import (
"encoding/json"
"fmt"
"os"
"path/filepath"
"strings"
"sync"
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/crypto"
"github.com/rs/zerolog/log"
)
// CryptoManager interface for encryption/decryption.
type CryptoManager interface {
Encrypt(plaintext []byte) ([]byte, error)
Decrypt(ciphertext []byte) ([]byte, error)
}
// Store provides encrypted per-resource storage for discovery data.
type Store struct {
mu sync.RWMutex
dataDir string
crypto CryptoManager
cache map[string]*ResourceDiscovery // In-memory cache
cacheTime map[string]time.Time // Cache timestamps
cacheTTL time.Duration
}
// For testing - allows injecting a mock crypto manager
var newCryptoManagerAt = crypto.NewCryptoManagerAt
// For testing - allows injecting a mock marshaler.
var marshalDiscovery = json.Marshal
// NewStore creates a new discovery store with automatic encryption.
func NewStore(dataDir string) (*Store, error) {
discoveryDir := filepath.Join(dataDir, "discovery")
if err := os.MkdirAll(discoveryDir, 0700); err != nil {
return nil, fmt.Errorf("failed to create discovery directory: %w", err)
}
// Initialize crypto manager for encryption (uses same key as other Pulse secrets)
cryptoMgr, err := newCryptoManagerAt(dataDir)
if err != nil {
log.Warn().Err(err).Msg("Failed to initialize crypto for discovery store, data will be unencrypted")
}
return &Store{
dataDir: discoveryDir,
crypto: cryptoMgr,
cache: make(map[string]*ResourceDiscovery),
cacheTime: make(map[string]time.Time),
cacheTTL: 5 * time.Minute,
}, nil
}
// getFilePath returns the file path for a resource ID.
func (s *Store) getFilePath(id string) string {
// Sanitize ID for filename: replace : with _
safeID := strings.ReplaceAll(id, ":", "_")
safeID = strings.ReplaceAll(safeID, "/", "_")
return filepath.Join(s.dataDir, safeID+".enc")
}
// Save persists a discovery to encrypted storage.
func (s *Store) Save(d *ResourceDiscovery) error {
s.mu.Lock()
defer s.mu.Unlock()
if d.ID == "" {
return fmt.Errorf("discovery ID is required")
}
// Update timestamp
d.UpdatedAt = time.Now()
if d.DiscoveredAt.IsZero() {
d.DiscoveredAt = d.UpdatedAt
}
data, err := marshalDiscovery(d)
if err != nil {
return fmt.Errorf("failed to marshal discovery: %w", err)
}
// Encrypt if crypto is available
if s.crypto != nil {
encrypted, err := s.crypto.Encrypt(data)
if err != nil {
return fmt.Errorf("failed to encrypt discovery: %w", err)
}
data = encrypted
}
// Write atomically using tmp file + rename
filePath := s.getFilePath(d.ID)
tmpPath := filePath + ".tmp"
if err := os.WriteFile(tmpPath, data, 0600); err != nil {
return fmt.Errorf("failed to write discovery file: %w", err)
}
if err := os.Rename(tmpPath, filePath); err != nil {
_ = os.Remove(tmpPath)
return fmt.Errorf("failed to finalize discovery file: %w", err)
}
// Update cache
s.cache[d.ID] = d
s.cacheTime[d.ID] = time.Now()
log.Debug().Str("id", d.ID).Str("service", d.ServiceType).Msg("Discovery saved")
return nil
}
// Get retrieves a discovery from storage.
func (s *Store) Get(id string) (*ResourceDiscovery, error) {
s.mu.RLock()
// Check cache first
if cached, ok := s.cache[id]; ok {
if cacheTime, hasTime := s.cacheTime[id]; hasTime {
if time.Since(cacheTime) < s.cacheTTL {
s.mu.RUnlock()
return cached, nil
}
}
}
s.mu.RUnlock()
s.mu.Lock()
defer s.mu.Unlock()
filePath := s.getFilePath(id)
data, err := os.ReadFile(filePath)
if err != nil {
if os.IsNotExist(err) {
return nil, nil // Not found is not an error
}
return nil, fmt.Errorf("failed to read discovery file: %w", err)
}
// Decrypt if crypto is available
if s.crypto != nil {
decrypted, err := s.crypto.Decrypt(data)
if err != nil {
return nil, fmt.Errorf("failed to decrypt discovery: %w", err)
}
data = decrypted
}
var discovery ResourceDiscovery
if err := json.Unmarshal(data, &discovery); err != nil {
return nil, fmt.Errorf("failed to unmarshal discovery: %w", err)
}
// Update cache
s.cache[id] = &discovery
s.cacheTime[id] = time.Now()
return &discovery, nil
}
// GetByResource retrieves a discovery by resource type and ID.
func (s *Store) GetByResource(resourceType ResourceType, hostID, resourceID string) (*ResourceDiscovery, error) {
id := MakeResourceID(resourceType, hostID, resourceID)
return s.Get(id)
}
// Delete removes a discovery from storage.
func (s *Store) Delete(id string) error {
s.mu.Lock()
defer s.mu.Unlock()
filePath := s.getFilePath(id)
if err := os.Remove(filePath); err != nil {
if os.IsNotExist(err) {
return nil // Already deleted
}
return fmt.Errorf("failed to delete discovery file: %w", err)
}
// Remove from cache
delete(s.cache, id)
delete(s.cacheTime, id)
log.Debug().Str("id", id).Msg("Discovery deleted")
return nil
}
// List returns all discoveries.
func (s *Store) List() ([]*ResourceDiscovery, error) {
s.mu.RLock()
defer s.mu.RUnlock()
entries, err := os.ReadDir(s.dataDir)
if err != nil {
if os.IsNotExist(err) {
return []*ResourceDiscovery{}, nil
}
return nil, fmt.Errorf("failed to list discovery directory: %w", err)
}
var discoveries []*ResourceDiscovery
for _, entry := range entries {
// Skip tmp files first to avoid reading partial writes.
if strings.HasSuffix(entry.Name(), ".tmp") {
continue
}
if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".enc") {
continue
}
data, err := os.ReadFile(filepath.Join(s.dataDir, entry.Name()))
if err != nil {
log.Warn().Err(err).Str("file", entry.Name()).Msg("Failed to read discovery file")
continue
}
// Decrypt if crypto is available
if s.crypto != nil {
decrypted, err := s.crypto.Decrypt(data)
if err != nil {
log.Warn().Err(err).Str("file", entry.Name()).Msg("Failed to decrypt discovery")
continue
}
data = decrypted
}
var discovery ResourceDiscovery
if err := json.Unmarshal(data, &discovery); err != nil {
log.Warn().Err(err).Str("file", entry.Name()).Msg("Failed to unmarshal discovery")
continue
}
discoveries = append(discoveries, &discovery)
}
return discoveries, nil
}
// ListByType returns discoveries for a specific resource type.
func (s *Store) ListByType(resourceType ResourceType) ([]*ResourceDiscovery, error) {
all, err := s.List()
if err != nil {
return nil, err
}
var filtered []*ResourceDiscovery
for _, d := range all {
if d.ResourceType == resourceType {
filtered = append(filtered, d)
}
}
return filtered, nil
}
// ListByHost returns discoveries for a specific host.
func (s *Store) ListByHost(hostID string) ([]*ResourceDiscovery, error) {
all, err := s.List()
if err != nil {
return nil, err
}
var filtered []*ResourceDiscovery
for _, d := range all {
if d.HostID == hostID {
filtered = append(filtered, d)
}
}
return filtered, nil
}
// UpdateNotes updates just the user notes and secrets for a discovery.
func (s *Store) UpdateNotes(id string, notes string, secrets map[string]string) error {
discovery, err := s.Get(id)
if err != nil {
return err
}
if discovery == nil {
return fmt.Errorf("discovery not found: %s", id)
}
discovery.UserNotes = notes
if secrets != nil {
discovery.UserSecrets = secrets
}
return s.Save(discovery)
}
// GetMultiple retrieves multiple discoveries by ID.
func (s *Store) GetMultiple(ids []string) ([]*ResourceDiscovery, error) {
var discoveries []*ResourceDiscovery
for _, id := range ids {
d, err := s.Get(id)
if err != nil {
log.Warn().Err(err).Str("id", id).Msg("Failed to get discovery")
continue
}
if d != nil {
discoveries = append(discoveries, d)
}
}
return discoveries, nil
}
// ClearCache clears the in-memory cache.
func (s *Store) ClearCache() {
s.mu.Lock()
defer s.mu.Unlock()
s.cache = make(map[string]*ResourceDiscovery)
s.cacheTime = make(map[string]time.Time)
}
// Exists checks if a discovery exists for the given ID.
func (s *Store) Exists(id string) bool {
s.mu.RLock()
if _, ok := s.cache[id]; ok {
s.mu.RUnlock()
return true
}
s.mu.RUnlock()
filePath := s.getFilePath(id)
_, err := os.Stat(filePath)
return err == nil
}
// GetAge returns how old the discovery is, or -1 if not found.
func (s *Store) GetAge(id string) time.Duration {
d, err := s.Get(id)
if err != nil || d == nil {
return -1
}
return time.Since(d.UpdatedAt)
}
// NeedsRefresh checks if a discovery needs to be refreshed.
func (s *Store) NeedsRefresh(id string, maxAge time.Duration) bool {
age := s.GetAge(id)
if age < 0 {
return true // Not found, needs discovery
}
return age > maxAge
}
internal/aidiscovery/store_test.go
-469
View File
@@ -1,469 +0,0 @@
package aidiscovery
import (
"encoding/json"
"os"
"path/filepath"
"strings"
"testing"
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/crypto"
)
type fakeCrypto struct{}
func (fakeCrypto) Encrypt(plaintext []byte) ([]byte, error) {
out := make([]byte, len(plaintext))
for i := range plaintext {
out[i] = plaintext[len(plaintext)-1-i]
}
return out, nil
}
func (fakeCrypto) Decrypt(ciphertext []byte) ([]byte, error) {
return fakeCrypto{}.Encrypt(ciphertext)
}
type errorCrypto struct{}
func (errorCrypto) Encrypt(plaintext []byte) ([]byte, error) {
return nil, os.ErrInvalid
}
func (errorCrypto) Decrypt(ciphertext []byte) ([]byte, error) {
return nil, os.ErrInvalid
}
func TestStore_SaveGetListAndNotes(t *testing.T) {
store, err := NewStore(t.TempDir())
if err != nil {
t.Fatalf("NewStore error: %v", err)
}
store.crypto = nil
d1 := &ResourceDiscovery{
ID: MakeResourceID(ResourceTypeDocker, "host1", "nginx"),
ResourceType: ResourceTypeDocker,
ResourceID: "nginx",
HostID: "host1",
ServiceName: "Nginx",
}
if err := store.Save(d1); err != nil {
t.Fatalf("Save error: %v", err)
}
got, err := store.Get(d1.ID)
if err != nil {
t.Fatalf("Get error: %v", err)
}
if got == nil || got.ServiceName != "Nginx" {
t.Fatalf("unexpected discovery: %#v", got)
}
if !store.Exists(d1.ID) {
t.Fatalf("expected discovery to exist")
}
if err := store.UpdateNotes(d1.ID, "notes", map[string]string{"token": "abc"}); err != nil {
t.Fatalf("UpdateNotes error: %v", err)
}
updated, err := store.Get(d1.ID)
if err != nil {
t.Fatalf("Get updated error: %v", err)
}
if updated.UserNotes != "notes" || updated.UserSecrets["token"] != "abc" {
t.Fatalf("notes not updated: %#v", updated)
}
d2 := &ResourceDiscovery{
ID: MakeResourceID(ResourceTypeVM, "node1", "101"),
ResourceType: ResourceTypeVM,
ResourceID: "101",
HostID: "node1",
ServiceName: "VM",
}
if err := store.Save(d2); err != nil {
t.Fatalf("Save d2 error: %v", err)
}
list, err := store.List()
if err != nil {
t.Fatalf("List error: %v", err)
}
if len(list) != 2 {
t.Fatalf("expected 2 discoveries, got %d", len(list))
}
byType, err := store.ListByType(ResourceTypeVM)
if err != nil {
t.Fatalf("ListByType error: %v", err)
}
if len(byType) != 1 || byType[0].ID != d2.ID {
t.Fatalf("unexpected ListByType: %#v", byType)
}
byHost, err := store.ListByHost("host1")
if err != nil {
t.Fatalf("ListByHost error: %v", err)
}
if len(byHost) != 1 || byHost[0].ID != d1.ID {
t.Fatalf("unexpected ListByHost: %#v", byHost)
}
summary := updated.ToSummary()
if summary.ID != d1.ID || !summary.HasUserNotes {
t.Fatalf("unexpected summary: %#v", summary)
}
if err := store.Delete(d1.ID); err != nil {
t.Fatalf("Delete error: %v", err)
}
if store.Exists(d1.ID) {
t.Fatalf("expected discovery to be deleted")
}
}
func TestStore_CryptoRoundTripAndPaths(t *testing.T) {
store, err := NewStore(t.TempDir())
if err != nil {
t.Fatalf("NewStore error: %v", err)
}
store.crypto = fakeCrypto{}
id := "docker:host1:app/name"
d := &ResourceDiscovery{
ID: id,
ResourceType: ResourceTypeDocker,
ResourceID: "app/name",
HostID: "host1",
ServiceName: "App",
}
if err := store.Save(d); err != nil {
t.Fatalf("Save error: %v", err)
}
path := store.getFilePath(id)
base := filepath.Base(path)
if strings.Contains(base, ":") || strings.Contains(base, "/") {
t.Fatalf("expected sanitized base filename, got %s", base)
}
loaded, err := store.Get(id)
if err != nil {
t.Fatalf("Get error: %v", err)
}
if loaded == nil || loaded.ServiceName != "App" {
t.Fatalf("unexpected discovery: %#v", loaded)
}
store.ClearCache()
if _, err := store.Get(id); err != nil {
t.Fatalf("Get with decrypt error: %v", err)
}
list, err := store.List()
if err != nil || len(list) != 1 {
t.Fatalf("List with decrypt error: %v len=%d", err, len(list))
}
}
func TestStore_NeedsRefreshAndGetMultiple(t *testing.T) {
store, err := NewStore(t.TempDir())
if err != nil {
t.Fatalf("NewStore error: %v", err)
}
store.crypto = nil
if !store.NeedsRefresh("missing", time.Minute) {
t.Fatalf("expected missing discovery to need refresh")
}
d := &ResourceDiscovery{
ID: MakeResourceID(ResourceTypeHost, "host1", "host1"),
ResourceType: ResourceTypeHost,
ResourceID: "host1",
HostID: "host1",
ServiceName: "Host",
}
if err := store.Save(d); err != nil {
t.Fatalf("Save error: %v", err)
}
path := store.getFilePath(d.ID)
data, err := os.ReadFile(path)
if err != nil {
t.Fatalf("ReadFile error: %v", err)
}
var saved ResourceDiscovery
if err := json.Unmarshal(data, &saved); err != nil {
t.Fatalf("Unmarshal error: %v", err)
}
saved.UpdatedAt = time.Now().Add(-2 * time.Hour)
data, err = json.Marshal(&saved)
if err != nil {
t.Fatalf("Marshal error: %v", err)
}
if err := os.WriteFile(path, data, 0600); err != nil {
t.Fatalf("WriteFile error: %v", err)
}
store.ClearCache()
if !store.NeedsRefresh(d.ID, time.Minute) {
t.Fatalf("expected old discovery to need refresh")
}
ids := []string{d.ID, "missing"}
multi, err := store.GetMultiple(ids)
if err != nil {
t.Fatalf("GetMultiple error: %v", err)
}
if len(multi) != 1 || multi[0].ID != d.ID {
t.Fatalf("unexpected GetMultiple: %#v", multi)
}
}
func TestStore_ErrorsAndListSkips(t *testing.T) {
dir := t.TempDir()
store, err := NewStore(dir)
if err != nil {
t.Fatalf("NewStore error: %v", err)
}
store.crypto = nil
if err := store.Save(&ResourceDiscovery{}); err == nil {
t.Fatalf("expected error for empty ID")
}
store.crypto = errorCrypto{}
if err := store.Save(&ResourceDiscovery{ID: "bad"}); err == nil {
t.Fatalf("expected encrypt error")
}
store.crypto = nil
if _, err := store.Get("missing"); err != nil {
t.Fatalf("unexpected missing error: %v", err)
}
d := &ResourceDiscovery{
ID: MakeResourceID(ResourceTypeDocker, "host1", "web"),
ResourceType: ResourceTypeDocker,
ResourceID: "web",
HostID: "host1",
ServiceName: "Web",
UserSecrets: map[string]string{"token": "abc"},
}
if err := store.Save(d); err != nil {
t.Fatalf("Save error: %v", err)
}
// Corrupt file to force unmarshal error during List.
badPath := filepath.Join(store.dataDir, "bad.enc")
if err := os.WriteFile(badPath, []byte("{bad"), 0600); err != nil {
t.Fatalf("WriteFile error: %v", err)
}
if err := os.WriteFile(filepath.Join(store.dataDir, "note.txt"), []byte("skip"), 0600); err != nil {
t.Fatalf("WriteFile error: %v", err)
}
if err := os.WriteFile(filepath.Join(store.dataDir, "skip.enc.tmp"), []byte("skip"), 0600); err != nil {
t.Fatalf("WriteFile error: %v", err)
}
if err := os.MkdirAll(filepath.Join(store.dataDir, "dir"), 0700); err != nil {
t.Fatalf("MkdirAll error: %v", err)
}
unreadable := filepath.Join(store.dataDir, "unreadable.enc")
if err := os.WriteFile(unreadable, []byte("nope"), 0000); err != nil {
t.Fatalf("WriteFile error: %v", err)
}
list, err := store.List()
if err != nil {
t.Fatalf("List error: %v", err)
}
if len(list) != 1 {
t.Fatalf("expected 1 discovery, got %d", len(list))
}
store.crypto = errorCrypto{}
list, err = store.List()
if err != nil {
t.Fatalf("List with crypto error: %v", err)
}
if len(list) != 0 {
t.Fatalf("expected crypto errors to skip entries")
}
store.crypto = errorCrypto{}
store.ClearCache()
if _, err := store.Get(d.ID); err == nil {
t.Fatalf("expected decrypt error")
}
store.crypto = nil
if got, err := store.GetByResource(ResourceTypeDocker, "host1", "web"); err != nil || got == nil {
t.Fatalf("GetByResource error: %v", err)
}
if err := store.UpdateNotes(d.ID, "notes-only", nil); err != nil {
t.Fatalf("UpdateNotes error: %v", err)
}
updated, err := store.Get(d.ID)
if err != nil || updated.UserSecrets == nil {
t.Fatalf("expected secrets to be preserved: %#v err=%v", updated, err)
}
store.crypto = errorCrypto{}
store.ClearCache()
if err := store.UpdateNotes(d.ID, "notes", nil); err == nil {
t.Fatalf("expected update notes error with crypto failure")
}
if got, err := store.GetMultiple([]string{d.ID}); err != nil || len(got) != 0 {
t.Fatalf("expected GetMultiple to skip errors")
}
if err := store.UpdateNotes("missing", "notes", nil); err == nil {
t.Fatalf("expected error for missing discovery")
}
if err := store.Delete("missing"); err != nil {
t.Fatalf("unexpected delete error: %v", err)
}
}
func TestStore_NewStoreError(t *testing.T) {
dir := t.TempDir()
file := filepath.Join(dir, "file")
if err := os.WriteFile(file, []byte("x"), 0600); err != nil {
t.Fatalf("WriteFile error: %v", err)
}
if _, err := NewStore(file); err == nil {
t.Fatalf("expected error for file data dir")
}
}
func TestStore_NewStoreCryptoFailure(t *testing.T) {
orig := newCryptoManagerAt
newCryptoManagerAt = func(dataDir string) (*crypto.CryptoManager, error) {
manager, err := crypto.NewCryptoManagerAt(dataDir)
if err != nil {
return nil, err
}
return manager, os.ErrInvalid
}
t.Cleanup(func() {
newCryptoManagerAt = orig
})
store, err := NewStore(t.TempDir())
if err != nil {
t.Fatalf("NewStore error: %v", err)
}
if store.crypto == nil {
t.Fatalf("expected crypto manager despite init warning")
}
}
func TestStore_SaveMarshalError(t *testing.T) {
store, err := NewStore(t.TempDir())
if err != nil {
t.Fatalf("NewStore error: %v", err)
}
store.crypto = nil
orig := marshalDiscovery
marshalDiscovery = func(any) ([]byte, error) {
return nil, os.ErrInvalid
}
t.Cleanup(func() {
marshalDiscovery = orig
})
if err := store.Save(&ResourceDiscovery{ID: "marshal"}); err == nil {
t.Fatalf("expected marshal error")
}
}
func TestStore_SaveAndGetErrors(t *testing.T) {
store, err := NewStore(t.TempDir())
if err != nil {
t.Fatalf("NewStore error: %v", err)
}
store.crypto = nil
id := MakeResourceID(ResourceTypeDocker, "host1", "web")
filePath := store.getFilePath(id)
if err := os.MkdirAll(filePath, 0700); err != nil {
t.Fatalf("MkdirAll error: %v", err)
}
if err := store.Save(&ResourceDiscovery{ID: id}); err == nil {
t.Fatalf("expected rename error")
}
tmpFile := filepath.Join(t.TempDir(), "file")
if err := os.WriteFile(tmpFile, []byte("x"), 0600); err != nil {
t.Fatalf("WriteFile error: %v", err)
}
store.dataDir = tmpFile
if err := store.Save(&ResourceDiscovery{ID: "bad"}); err == nil {
t.Fatalf("expected write error")
}
store.dataDir = t.TempDir()
store.crypto = nil
badPath := store.getFilePath("bad")
if err := os.WriteFile(badPath, []byte("{bad"), 0600); err != nil {
t.Fatalf("WriteFile error: %v", err)
}
if _, err := store.Get("bad"); err == nil {
t.Fatalf("expected unmarshal error")
}
}
func TestStore_ListErrors(t *testing.T) {
store, err := NewStore(t.TempDir())
if err != nil {
t.Fatalf("NewStore error: %v", err)
}
store.crypto = nil
store.dataDir = filepath.Join(t.TempDir(), "missing")
list, err := store.List()
if err != nil || len(list) != 0 {
t.Fatalf("expected empty list for missing dir")
}
file := filepath.Join(t.TempDir(), "file")
if err := os.WriteFile(file, []byte("x"), 0600); err != nil {
t.Fatalf("WriteFile error: %v", err)
}
store.dataDir = file
if _, err := store.List(); err == nil {
t.Fatalf("expected list error for file path")
}
if _, err := store.ListByType(ResourceTypeDocker); err == nil {
t.Fatalf("expected list by type error")
}
if _, err := store.ListByHost("host1"); err == nil {
t.Fatalf("expected list by host error")
}
}
func TestStore_DeleteError(t *testing.T) {
store, err := NewStore(t.TempDir())
if err != nil {
t.Fatalf("NewStore error: %v", err)
}
store.crypto = nil
id := MakeResourceID(ResourceTypeDocker, "host1", "dir")
filePath := store.getFilePath(id)
if err := os.MkdirAll(filePath, 0700); err != nil {
t.Fatalf("MkdirAll error: %v", err)
}
nested := filepath.Join(filePath, "nested")
if err := os.WriteFile(nested, []byte("x"), 0600); err != nil {
t.Fatalf("WriteFile error: %v", err)
}
if err := store.Delete(id); err == nil {
t.Fatalf("expected delete error for non-empty dir")
}
}
internal/aidiscovery/tools_adapter.go
-155
View File
@@ -1,155 +0,0 @@
package aidiscovery
import (
"github.com/rcourtman/pulse-go-rewrite/internal/ai/tools"
)
// ToolsAdapter wraps Service to implement tools.DiscoverySource
type ToolsAdapter struct {
service *Service
}
// NewToolsAdapter creates a new adapter for the discovery service
func NewToolsAdapter(service *Service) *ToolsAdapter {
if service == nil {
return nil
}
return &ToolsAdapter{service: service}
}
// GetDiscovery implements tools.DiscoverySource
func (a *ToolsAdapter) GetDiscovery(id string) (tools.DiscoverySourceData, error) {
discovery, err := a.service.GetDiscovery(id)
if err != nil {
return tools.DiscoverySourceData{}, err
}
if discovery == nil {
return tools.DiscoverySourceData{}, nil
}
return a.convertToSourceData(discovery), nil
}
// GetDiscoveryByResource implements tools.DiscoverySource
func (a *ToolsAdapter) GetDiscoveryByResource(resourceType, hostID, resourceID string) (tools.DiscoverySourceData, error) {
discovery, err := a.service.GetDiscoveryByResource(ResourceType(resourceType), hostID, resourceID)
if err != nil {
return tools.DiscoverySourceData{}, err
}
if discovery == nil {
return tools.DiscoverySourceData{}, nil
}
return a.convertToSourceData(discovery), nil
}
// ListDiscoveries implements tools.DiscoverySource
func (a *ToolsAdapter) ListDiscoveries() ([]tools.DiscoverySourceData, error) {
discoveries, err := a.service.ListDiscoveries()
if err != nil {
return nil, err
}
return a.convertList(discoveries), nil
}
// ListDiscoveriesByType implements tools.DiscoverySource
func (a *ToolsAdapter) ListDiscoveriesByType(resourceType string) ([]tools.DiscoverySourceData, error) {
discoveries, err := a.service.ListDiscoveriesByType(ResourceType(resourceType))
if err != nil {
return nil, err
}
return a.convertList(discoveries), nil
}
// ListDiscoveriesByHost implements tools.DiscoverySource
func (a *ToolsAdapter) ListDiscoveriesByHost(hostID string) ([]tools.DiscoverySourceData, error) {
discoveries, err := a.service.ListDiscoveriesByHost(hostID)
if err != nil {
return nil, err
}
return a.convertList(discoveries), nil
}
// FormatForAIContext implements tools.DiscoverySource
func (a *ToolsAdapter) FormatForAIContext(sourceData []tools.DiscoverySourceData) string {
// Convert back to ResourceDiscovery for formatting
discoveries := make([]*ResourceDiscovery, 0, len(sourceData))
for _, sd := range sourceData {
discoveries = append(discoveries, a.convertFromSourceData(sd))
}
return FormatForAIContext(discoveries)
}
func (a *ToolsAdapter) convertToSourceData(d *ResourceDiscovery) tools.DiscoverySourceData {
facts := make([]tools.DiscoverySourceFact, 0, len(d.Facts))
for _, f := range d.Facts {
facts = append(facts, tools.DiscoverySourceFact{
Category: string(f.Category),
Key: f.Key,
Value: f.Value,
Source: f.Source,
})
}
return tools.DiscoverySourceData{
ID: d.ID,
ResourceType: string(d.ResourceType),
ResourceID: d.ResourceID,
HostID: d.HostID,
Hostname: d.Hostname,
ServiceType: d.ServiceType,
ServiceName: d.ServiceName,
ServiceVersion: d.ServiceVersion,
Category: string(d.Category),
CLIAccess: d.CLIAccess,
Facts: facts,
ConfigPaths: d.ConfigPaths,
DataPaths: d.DataPaths,
UserNotes: d.UserNotes,
Confidence: d.Confidence,
AIReasoning: d.AIReasoning,
DiscoveredAt: d.DiscoveredAt,
UpdatedAt: d.UpdatedAt,
}
}
func (a *ToolsAdapter) convertFromSourceData(sd tools.DiscoverySourceData) *ResourceDiscovery {
facts := make([]DiscoveryFact, 0, len(sd.Facts))
for _, f := range sd.Facts {
facts = append(facts, DiscoveryFact{
Category: FactCategory(f.Category),
Key: f.Key,
Value: f.Value,
Source: f.Source,
})
}
return &ResourceDiscovery{
ID: sd.ID,
ResourceType: ResourceType(sd.ResourceType),
ResourceID: sd.ResourceID,
HostID: sd.HostID,
Hostname: sd.Hostname,
ServiceType: sd.ServiceType,
ServiceName: sd.ServiceName,
ServiceVersion: sd.ServiceVersion,
Category: ServiceCategory(sd.Category),
CLIAccess: sd.CLIAccess,
Facts: facts,
ConfigPaths: sd.ConfigPaths,
DataPaths: sd.DataPaths,
UserNotes: sd.UserNotes,
Confidence: sd.Confidence,
AIReasoning: sd.AIReasoning,
DiscoveredAt: sd.DiscoveredAt,
UpdatedAt: sd.UpdatedAt,
}
}
func (a *ToolsAdapter) convertList(discoveries []*ResourceDiscovery) []tools.DiscoverySourceData {
result := make([]tools.DiscoverySourceData, 0, len(discoveries))
for _, d := range discoveries {
if d != nil {
result = append(result, a.convertToSourceData(d))
}
}
return result
}
internal/aidiscovery/types.go
-236
View File
@@ -1,236 +0,0 @@
// Package discovery provides AI-powered infrastructure discovery capabilities.
// It discovers services, versions, configurations, and CLI access methods
// for VMs, LXCs, Docker containers, Kubernetes pods, and hosts.
package aidiscovery
import (
"fmt"
"time"
)
// ResourceType identifies the type of infrastructure resource.
type ResourceType string
const (
ResourceTypeVM ResourceType = "vm"
ResourceTypeLXC ResourceType = "lxc"
ResourceTypeDocker ResourceType = "docker"
ResourceTypeK8s ResourceType = "k8s"
ResourceTypeHost ResourceType = "host"
ResourceTypeDockerVM ResourceType = "docker_vm" // Docker on a VM
ResourceTypeDockerLXC ResourceType = "docker_lxc" // Docker in an LXC
)
// FactCategory categorizes discovery facts.
type FactCategory string
const (
FactCategoryVersion FactCategory = "version"
FactCategoryConfig FactCategory = "config"
FactCategoryService FactCategory = "service"
FactCategoryPort FactCategory = "port"
FactCategoryHardware FactCategory = "hardware"
FactCategoryNetwork FactCategory = "network"
FactCategoryStorage FactCategory = "storage"
FactCategoryDependency FactCategory = "dependency"
FactCategorySecurity FactCategory = "security"
)
// ServiceCategory categorizes the type of service discovered.
type ServiceCategory string
const (
CategoryDatabase ServiceCategory = "database"
CategoryWebServer ServiceCategory = "web_server"
CategoryCache ServiceCategory = "cache"
CategoryMessageQueue ServiceCategory = "message_queue"
CategoryMonitoring ServiceCategory = "monitoring"
CategoryBackup ServiceCategory = "backup"
CategoryNVR ServiceCategory = "nvr"
CategoryStorage ServiceCategory = "storage"
CategoryContainer ServiceCategory = "container"
CategoryVirtualizer ServiceCategory = "virtualizer"
CategoryNetwork ServiceCategory = "network"
CategorySecurity ServiceCategory = "security"
CategoryMedia ServiceCategory = "media"
CategoryHomeAuto ServiceCategory = "home_automation"
CategoryUnknown ServiceCategory = "unknown"
)
// ResourceDiscovery is the main data model for discovered resource information.
type ResourceDiscovery struct {
// Identity
ID string `json:"id"` // Unique ID: "lxc:minipc:101"
ResourceType ResourceType `json:"resource_type"` // vm, lxc, docker, k8s, host
ResourceID string `json:"resource_id"` // 101, container-name, etc.
HostID string `json:"host_id"` // Proxmox node name or host agent ID
Hostname string `json:"hostname"` // Human-readable host name
// AI-discovered info
ServiceType string `json:"service_type"` // frigate, postgres, pbs
ServiceName string `json:"service_name"` // Human-readable name
ServiceVersion string `json:"service_version"` // v0.13.2
Category ServiceCategory `json:"category"` // nvr, database, backup
CLIAccess string `json:"cli_access"` // pct exec 101 -- ...
// Deep discovery facts
Facts []DiscoveryFact `json:"facts"`
ConfigPaths []string `json:"config_paths"`
DataPaths []string `json:"data_paths"`
Ports []PortInfo `json:"ports"`
// User-added (also encrypted)
UserNotes string `json:"user_notes"`
UserSecrets map[string]string `json:"user_secrets"` // tokens, creds
// Metadata
Confidence float64 `json:"confidence"` // 0-1 confidence score
AIReasoning string `json:"ai_reasoning"` // AI explanation
DiscoveredAt time.Time `json:"discovered_at"` // First discovery
UpdatedAt time.Time `json:"updated_at"` // Last update
ScanDuration int64 `json:"scan_duration"` // Scan duration in ms
// Raw data for debugging/re-analysis
RawCommandOutput map[string]string `json:"raw_command_output,omitempty"`
}
// DiscoveryFact represents a single discovered fact about a resource.
type DiscoveryFact struct {
Category FactCategory `json:"category"` // version, config, service, port
Key string `json:"key"` // e.g., "coral_tpu", "mqtt_broker"
Value string `json:"value"` // e.g., "/dev/apex_0", "mosquitto:1883"
Source string `json:"source"` // command that found this
Confidence float64 `json:"confidence"` // 0-1 confidence for this fact
DiscoveredAt time.Time `json:"discovered_at"`
}
// PortInfo represents information about a listening port.
type PortInfo struct {
Port int `json:"port"`
Protocol string `json:"protocol"` // tcp, udp
Process string `json:"process"` // process name
Address string `json:"address"` // bind address
}
// MakeResourceID creates a standardized resource ID.
func MakeResourceID(resourceType ResourceType, hostID, resourceID string) string {
return fmt.Sprintf("%s:%s:%s", resourceType, hostID, resourceID)
}
// ParseResourceID parses a resource ID into its components.
func ParseResourceID(id string) (resourceType ResourceType, hostID, resourceID string, err error) {
var parts [3]string
count := 0
start := 0
for i, c := range id {
if c == ':' {
if count < 2 {
parts[count] = id[start:i]
count++
start = i + 1
}
}
}
if count == 2 {
parts[2] = id[start:]
return ResourceType(parts[0]), parts[1], parts[2], nil
}
return "", "", "", fmt.Errorf("invalid resource ID format: %s", id)
}
// DiscoveryRequest represents a request to discover a resource.
type DiscoveryRequest struct {
ResourceType ResourceType `json:"resource_type"`
ResourceID string `json:"resource_id"`
HostID string `json:"host_id"`
Hostname string `json:"hostname"`
Force bool `json:"force"` // Force re-scan even if recent
}
// DiscoveryStatus represents the status of a discovery scan.
type DiscoveryStatus string
const (
DiscoveryStatusPending DiscoveryStatus = "pending"
DiscoveryStatusRunning DiscoveryStatus = "running"
DiscoveryStatusCompleted DiscoveryStatus = "completed"
DiscoveryStatusFailed DiscoveryStatus = "failed"
DiscoveryStatusNotStarted DiscoveryStatus = "not_started"
)
// DiscoveryProgress represents the progress of an ongoing discovery.
type DiscoveryProgress struct {
ResourceID string `json:"resource_id"`
Status DiscoveryStatus `json:"status"`
CurrentStep string `json:"current_step"`
TotalSteps int `json:"total_steps"`
CompletedSteps int `json:"completed_steps"`
StartedAt time.Time `json:"started_at"`
Error string `json:"error,omitempty"`
}
// UpdateNotesRequest represents a request to update user notes.
type UpdateNotesRequest struct {
UserNotes string `json:"user_notes"`
UserSecrets map[string]string `json:"user_secrets,omitempty"`
}
// DiscoverySummary provides a summary of discoveries for listing.
type DiscoverySummary struct {
ID string `json:"id"`
ResourceType ResourceType `json:"resource_type"`
ResourceID string `json:"resource_id"`
HostID string `json:"host_id"`
Hostname string `json:"hostname"`
ServiceType string `json:"service_type"`
ServiceName string `json:"service_name"`
ServiceVersion string `json:"service_version"`
Category ServiceCategory `json:"category"`
Confidence float64 `json:"confidence"`
HasUserNotes bool `json:"has_user_notes"`
UpdatedAt time.Time `json:"updated_at"`
}
// ToSummary converts a full discovery to a summary.
func (d *ResourceDiscovery) ToSummary() DiscoverySummary {
return DiscoverySummary{
ID: d.ID,
ResourceType: d.ResourceType,
ResourceID: d.ResourceID,
HostID: d.HostID,
Hostname: d.Hostname,
ServiceType: d.ServiceType,
ServiceName: d.ServiceName,
ServiceVersion: d.ServiceVersion,
Category: d.Category,
Confidence: d.Confidence,
HasUserNotes: d.UserNotes != "",
UpdatedAt: d.UpdatedAt,
}
}
// AIAnalysisRequest is sent to the AI for analysis.
type AIAnalysisRequest struct {
ResourceType ResourceType `json:"resource_type"`
ResourceID string `json:"resource_id"`
HostID string `json:"host_id"`
Hostname string `json:"hostname"`
CommandOutputs map[string]string `json:"command_outputs"`
ExistingFacts []DiscoveryFact `json:"existing_facts,omitempty"`
Metadata map[string]any `json:"metadata,omitempty"` // Image, labels, etc.
}
// AIAnalysisResponse is returned by the AI.
type AIAnalysisResponse struct {
ServiceType string `json:"service_type"`
ServiceName string `json:"service_name"`
ServiceVersion string `json:"service_version"`
Category ServiceCategory `json:"category"`
CLIAccess string `json:"cli_access"`
Facts []DiscoveryFact `json:"facts"`
ConfigPaths []string `json:"config_paths"`
DataPaths []string `json:"data_paths"`
Ports []PortInfo `json:"ports"`
Confidence float64 `json:"confidence"`
Reasoning string `json:"reasoning"`
}
internal/aidiscovery/types_test.go
-22
View File
@@ -1,22 +0,0 @@
package aidiscovery
import "testing"
func TestResourceIDHelpers(t *testing.T) {
id := MakeResourceID(ResourceTypeDocker, "host1", "app")
if id != "docker:host1:app" {
t.Fatalf("unexpected id: %s", id)
}
rt, host, res, err := ParseResourceID(id)
if err != nil {
t.Fatalf("ParseResourceID error: %v", err)
}
if rt != ResourceTypeDocker || host != "host1" || res != "app" {
t.Fatalf("unexpected parse result: %s %s %s", rt, host, res)
}
if _, _, _, err := ParseResourceID("invalid"); err == nil {
t.Fatalf("expected parse error for invalid id")
}
}