NannyAI/nannyagent
SHA256
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Integrate-with-supabase-backend

Browse Source
This commit is contained in:
Harshavardhan Musanalli
2025-10-28 07:53:14 +01:00
parent 8832450a1f
commit 8328f8d5b3
6 changed files with 267 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
agent.go
View File

@@ -128,13 +128,10 @@ func (a *LinuxDiagnosticAgent) DiagnoseIssue(issue string) error {
if len(diagnosticResp.Commands) > 0 {
fmt.Printf("🔧 Executing diagnostic commands...\n")
for _, cmd := range diagnosticResp.Commands {
fmt.Printf("⚙️ Executing command '%s': %s\n", cmd.ID, cmd.Command)
result := a.executor.Execute(cmd)
commandResults = append(commandResults, result)
if result.ExitCode == 0 {
fmt.Printf("✅ Command '%s' completed successfully\n", cmd.ID)
} else {
if result.ExitCode != 0 {
fmt.Printf("❌ Command '%s' failed with exit code %d\n", cmd.ID, result.ExitCode)
}
}
@@ -143,7 +140,6 @@ func (a *LinuxDiagnosticAgent) DiagnoseIssue(issue string) error {
// Execute eBPF programs if present
var ebpfResults []map[string]interface{}
if len(diagnosticResp.EBPFPrograms) > 0 {
fmt.Printf("🔬 Executing %d eBPF programs...\n", len(diagnosticResp.EBPFPrograms))
ebpfResults = a.executeEBPFPrograms(diagnosticResp.EBPFPrograms)
}
@@ -170,15 +166,6 @@ func (a *LinuxDiagnosticAgent) DiagnoseIssue(issue string) error {
evidenceSummary = append(evidenceSummary, summaryStr)
}
allResults["ebpf_evidence_summary"] = evidenceSummary
fmt.Printf("<22> Sending eBPF monitoring data to TensorZero:\n")
for _, summary := range evidenceSummary {
fmt.Printf(" - %s\n", summary)
}
fmt.Printf("✅ Executed %d commands, %d eBPF programs\n", len(commandResults), len(ebpfResults))
} else {
fmt.Printf("✅ Executed %d commands\n", len(commandResults))
}
resultsJSON, err := json.MarshalIndent(allResults, "", " ")
@@ -227,7 +214,7 @@ func (a *LinuxDiagnosticAgent) executeEBPFPrograms(ebpfPrograms []EBPFRequest) [
}
for _, prog := range ebpfPrograms {
fmt.Printf("🔬 Starting eBPF program [%s]: %s -> %s (%ds)\n", prog.Name, prog.Type, prog.Target, int(prog.Duration))
// eBPF program starting - only show in debug mode
// Actually start the eBPF program using the real manager
programID, err := a.ebpfManager.StartEBPFProgram(prog)
@@ -248,16 +235,11 @@ func (a *LinuxDiagnosticAgent) executeEBPFPrograms(ebpfPrograms []EBPFRequest) [
}
// Let the eBPF program run for the specified duration
fmt.Printf("⏰ Waiting %d seconds for eBPF program to collect data...\n", int(prog.Duration))
time.Sleep(time.Duration(prog.Duration) * time.Second)
// Give the collectEvents goroutine a moment to finish and store results
fmt.Printf("⏳ Allowing program to complete data collection...\n")
time.Sleep(500 * time.Millisecond)
// Get the results (should be in completedResults now)
fmt.Printf("📊 Getting results for eBPF program [%s]...\n", prog.Name)
// Use a channel to implement timeout for GetProgramResults
type resultPair struct {
trace *EBPFTrace
@@ -282,11 +264,9 @@ func (a *LinuxDiagnosticAgent) executeEBPFPrograms(ebpfPrograms []EBPFRequest) [
}
// Try to stop the program (may already be stopped by collectEvents)
fmt.Printf("🛑 Stopping eBPF program [%s]...\n", prog.Name)
stopErr := a.ebpfManager.StopProgram(programID)
if stopErr != nil {
fmt.Printf("⚠️ eBPF program [%s] cleanup: %v (may have already completed)\n", prog.Name, stopErr)
// Don't return here, we still want to process results if we got them
// Only show warning in debug mode - this is normal for completed programs
}
if resultErr != nil {
@@ -325,7 +305,6 @@ func (a *LinuxDiagnosticAgent) executeEBPFPrograms(ebpfPrograms []EBPFRequest) [
result["end_time"] = trace.EndTime.Format(time.RFC3339)
result["actual_duration"] = trace.EndTime.Sub(trace.StartTime).Seconds()
fmt.Printf("✅ eBPF program [%s] completed - collected %d real events\n", prog.Name, trace.EventCount)
} else {
result["data_points"] = 0
result["error"] = "No trace data returned"

90
internal/logging/logger.go Normal file
View File

@@ -0,0 +1,90 @@
package logging
import (
"fmt"
"log"
"log/syslog"
"os"
)
type Logger struct {
syslogWriter *syslog.Writer
debugMode bool
}
var defaultLogger *Logger
func init() {
defaultLogger = NewLogger()
}
func NewLogger() *Logger {
l := &Logger{
debugMode: os.Getenv("DEBUG") == "true",
}
// Try to connect to syslog
if writer, err := syslog.New(syslog.LOG_INFO|syslog.LOG_DAEMON, "nannyagentv2"); err == nil {
l.syslogWriter = writer
}
return l
}
func (l *Logger) Info(format string, args ...interface{}) {
msg := fmt.Sprintf(format, args...)
if l.syslogWriter != nil {
l.syslogWriter.Info(msg)
}
log.Printf("[INFO] %s", msg)
}
func (l *Logger) Debug(format string, args ...interface{}) {
if !l.debugMode {
return
}
msg := fmt.Sprintf(format, args...)
if l.syslogWriter != nil {
l.syslogWriter.Debug(msg)
}
log.Printf("[DEBUG] %s", msg)
}
func (l *Logger) Warning(format string, args ...interface{}) {
msg := fmt.Sprintf(format, args...)
if l.syslogWriter != nil {
l.syslogWriter.Warning(msg)
}
log.Printf("[WARNING] %s", msg)
}
func (l *Logger) Error(format string, args ...interface{}) {
msg := fmt.Sprintf(format, args...)
if l.syslogWriter != nil {
l.syslogWriter.Err(msg)
}
log.Printf("[ERROR] %s", msg)
}
func (l *Logger) Close() {
if l.syslogWriter != nil {
l.syslogWriter.Close()
}
}
// Global logging functions
func Info(format string, args ...interface{}) {
defaultLogger.Info(format, args...)
}
func Debug(format string, args ...interface{}) {
defaultLogger.Debug(format, args...)
}
func Warning(format string, args ...interface{}) {
defaultLogger.Warning(format, args...)
}
func Error(format string, args ...interface{}) {
defaultLogger.Error(format, args...)
}

167
internal/types/types.go
View File

@@ -168,3 +168,170 @@ type MetricsRequest struct {
BlockDevices []BlockDevice `json:"block_devices"`
NetworkStats map[string]uint64 `json:"network_stats"`
}
// eBPF related types
type EBPFEvent struct {
Timestamp int64 `json:"timestamp"`
EventType string `json:"event_type"`
ProcessID int `json:"process_id"`
ProcessName string `json:"process_name"`
UserID int `json:"user_id"`
Data map[string]interface{} `json:"data"`
}
type EBPFTrace struct {
TraceID string `json:"trace_id"`
StartTime time.Time `json:"start_time"`
EndTime time.Time `json:"end_time"`
Capability string `json:"capability"`
Events []EBPFEvent `json:"events"`
Summary string `json:"summary"`
EventCount int `json:"event_count"`
ProcessList []string `json:"process_list"`
}
type EBPFRequest struct {
Name string `json:"name"`
Type string `json:"type"` // "tracepoint", "kprobe", "kretprobe"
Target string `json:"target"` // tracepoint path or function name
Duration int `json:"duration"` // seconds
Filters map[string]string `json:"filters,omitempty"`
Description string `json:"description"`
}
type NetworkEvent struct {
Timestamp uint64 `json:"timestamp"`
PID uint32 `json:"pid"`
TID uint32 `json:"tid"`
UID uint32 `json:"uid"`
EventType string `json:"event_type"`
Comm [16]byte `json:"-"`
CommStr string `json:"comm"`
}
// Agent types
type DiagnosticResponse struct {
ResponseType string `json:"response_type"`
Reasoning string `json:"reasoning"`
Commands []Command `json:"commands"`
}
type ResolutionResponse struct {
ResponseType string `json:"response_type"`
RootCause string `json:"root_cause"`
ResolutionPlan string `json:"resolution_plan"`
Confidence string `json:"confidence"`
}
type Command struct {
ID string `json:"id"`
Command string `json:"command"`
Description string `json:"description"`
}
type CommandResult struct {
ID string `json:"id"`
Command string `json:"command"`
Description string `json:"description"`
Output string `json:"output"`
ExitCode int `json:"exit_code"`
Error string `json:"error,omitempty"`
}
type EBPFEnhancedDiagnosticResponse struct {
ResponseType string `json:"response_type"`
Reasoning string `json:"reasoning"`
Commands []Command `json:"commands"`
EBPFPrograms []EBPFRequest `json:"ebpf_programs"`
NextActions []string `json:"next_actions,omitempty"`
}
type TensorZeroRequest struct {
Model string `json:"model"`
Messages []map[string]interface{} `json:"messages"`
EpisodeID string `json:"tensorzero::episode_id,omitempty"`
}
type TensorZeroResponse struct {
Choices []map[string]interface{} `json:"choices"`
EpisodeID string `json:"episode_id"`
}
// WebSocket types
type WebSocketMessage struct {
Type string `json:"type"`
Data interface{} `json:"data"`
}
type InvestigationTask struct {
TaskID string `json:"task_id"`
InvestigationID string `json:"investigation_id"`
AgentID string `json:"agent_id"`
DiagnosticPayload map[string]interface{} `json:"diagnostic_payload"`
EpisodeID string `json:"episode_id,omitempty"`
}
type TaskResult struct {
TaskID string `json:"task_id"`
Success bool `json:"success"`
CommandResults map[string]interface{} `json:"command_results,omitempty"`
Error string `json:"error,omitempty"`
}
type HeartbeatData struct {
AgentID string `json:"agent_id"`
Timestamp time.Time `json:"timestamp"`
Version string `json:"version"`
}
// Investigation server types
type InvestigationRequest struct {
Issue string `json:"issue"`
AgentID string `json:"agent_id"`
EpisodeID string `json:"episode_id,omitempty"`
Timestamp string `json:"timestamp,omitempty"`
Priority string `json:"priority,omitempty"`
Description string `json:"description,omitempty"`
}
type InvestigationResponse struct {
Status string `json:"status"`
Message string `json:"message"`
Results map[string]interface{} `json:"results,omitempty"`
AgentID string `json:"agent_id"`
Timestamp string `json:"timestamp"`
EpisodeID string `json:"episode_id,omitempty"`
Investigation *PendingInvestigation `json:"investigation,omitempty"`
}
type PendingInvestigation struct {
ID string `json:"id"`
Issue string `json:"issue"`
AgentID string `json:"agent_id"`
Status string `json:"status"`
DiagnosticPayload map[string]interface{} `json:"diagnostic_payload"`
CommandResults map[string]interface{} `json:"command_results,omitempty"`
EpisodeID *string `json:"episode_id,omitempty"`
CreatedAt string `json:"created_at"`
StartedAt *string `json:"started_at,omitempty"`
CompletedAt *string `json:"completed_at,omitempty"`
ErrorMessage *string `json:"error_message,omitempty"`
}
// System types
type SystemInfo struct {
Hostname string `json:"hostname"`
Platform string `json:"platform"`
PlatformInfo map[string]string `json:"platform_info"`
KernelVersion string `json:"kernel_version"`
Uptime string `json:"uptime"`
LoadAverage []float64 `json:"load_average"`
CPUInfo map[string]string `json:"cpu_info"`
MemoryInfo map[string]string `json:"memory_info"`
DiskInfo []map[string]string `json:"disk_info"`
}
// Executor types
type CommandExecutor struct {
timeout time.Duration
}

8
investigation_server.go
View File

@@ -60,7 +60,7 @@ func NewInvestigationServer(agent *LinuxDiagnosticAgent, authManager *auth.AuthM
if authManager != nil {
if id, err := authManager.GetCurrentAgentID(); err == nil {
agentID = id
fmt.Printf("✅ Retrieved agent ID from auth manager: %s\n", agentID)
} else {
fmt.Printf("❌ Failed to get agent ID from auth manager: %v\n", err)
}
@@ -250,7 +250,7 @@ func (s *InvestigationServer) sendCommandResultsToTensorZero(diagnosticResp Diag
// Check if it's a resolution response
if err := json.Unmarshal([]byte(content), &resolutionResp); err == nil && resolutionResp.ResponseType == "resolution" {
fmt.Printf("✅ TensorZero provided final resolution\n")
return map[string]interface{}{
"type": "resolution",
"response": resolutionResp,
@@ -355,7 +355,7 @@ func (s *InvestigationServer) handleDiagnosticExecution(requestBody map[string]i
result := s.agent.executor.Execute(cmd)
commandResults = append(commandResults, result)
fmt.Printf("✅ Command '%s' completed with exit code %d\n", cmd.ID, result.ExitCode)
if result.Error != "" {
fmt.Printf("⚠️ Command '%s' had error: %s\n", cmd.ID, result.Error)
}
@@ -471,7 +471,7 @@ func (s *InvestigationServer) handlePendingInvestigation(investigation PendingIn
return
}
fmt.Printf("✅ Realtime investigation %s completed successfully\n", investigation.InvestigationID)
}
// updateInvestigationStatus updates the status of a pending investigation

4
main.go
View File

@@ -73,7 +73,7 @@ func checkKernelVersionCompatibility() {
os.Exit(1)
}
fmt.Printf("✅ Kernel version %s is compatible with eBPF\n", kernelVersion)
}
// checkEBPFSupport validates eBPF subsystem availability
@@ -97,7 +97,7 @@ func checkEBPFSupport() {
syscall.Close(int(fd))
}
fmt.Printf("✅ eBPF syscall is available\n")
}
// runInteractiveDiagnostics starts the interactive diagnostic session

26
websocket_client.go
View File

@@ -130,7 +130,6 @@ func (w *WebSocketClient) Start() error {
// Stop closes the WebSocket connection
func (c *WebSocketClient) Stop() {
fmt.Println("🛑 Stopping WebSocket client...")
c.cancel()
if c.conn != nil {
c.conn.Close()
@@ -313,8 +312,6 @@ func (c *WebSocketClient) handleInvestigationTask(data interface{}) {
// executeDiagnosticCommands executes the commands from a diagnostic response
func (c *WebSocketClient) executeDiagnosticCommands(diagnosticPayload map[string]interface{}) (map[string]interface{}, error) {
fmt.Println("🔧 Executing diagnostic commands...")
results := map[string]interface{}{
"agent_id": c.agentID,
"execution_time": time.Now().UTC().Format(time.RFC3339),
@@ -360,8 +357,6 @@ func (c *WebSocketClient) executeDiagnosticCommands(diagnosticPayload map[string
if err != nil {
result["error"] = err.Error()
fmt.Printf("❌ Command [%s] failed: %v (exit code: %d)\n", id, err, exitCode)
} else {
// Command completed successfully - output captured
}
commandResults = append(commandResults, result)
@@ -374,17 +369,11 @@ func (c *WebSocketClient) executeDiagnosticCommands(diagnosticPayload map[string
// Execute eBPF programs if present
ebpfPrograms, hasEBPF := diagnosticPayload["ebpf_programs"].([]interface{})
if hasEBPF && len(ebpfPrograms) > 0 {
fmt.Printf("🔬 Executing %d eBPF programs...\n", len(ebpfPrograms))
ebpfResults := c.executeEBPFPrograms(ebpfPrograms)
results["ebpf_results"] = ebpfResults
results["total_ebpf_programs"] = len(ebpfPrograms)
} else {
fmt.Printf("ℹ️ No eBPF programs in diagnostic payload\n")
}
fmt.Printf("✅ Executed %d commands, %d successful\n",
results["total_commands"], results["successful_commands"])
return results, nil
}
@@ -455,8 +444,6 @@ func (c *WebSocketClient) executeCommandsFromPayload(commands []interface{}) []m
if err != nil {
result["error"] = err.Error()
fmt.Printf("❌ Command [%s] failed: %v (exit code: %d)\n", id, err, exitCode)
} else {
fmt.Printf("✅ Command [%s] completed successfully\n", id)
}
commandResults = append(commandResults, result)
@@ -657,14 +644,10 @@ func (c *WebSocketClient) handlePendingInvestigation(investigation PendingInvest
},
}
fmt.Printf("🔄 Sending command results to TensorZero for continued analysis...\n")
fmt.Printf("📤 Command results payload size: %d bytes\n", len(commandsJSON))
// Use the episode ID from the investigation to maintain conversation continuity
episodeID := ""
if investigation.EpisodeID != nil {
episodeID = *investigation.EpisodeID
fmt.Printf("🔗 Using episode ID: %s\n", episodeID)
}
// Continue conversation until resolution (same as agent)
@@ -678,8 +661,6 @@ func (c *WebSocketClient) handlePendingInvestigation(investigation PendingInvest
return
}
fmt.Printf("✅ TensorZero responded successfully\n")
if len(tzResp.Choices) == 0 {
fmt.Printf("⚠️ No choices in TensorZero response\n")
c.updateInvestigationStatus(investigation.ID, "completed", resultsForDB, nil)
@@ -688,7 +669,7 @@ func (c *WebSocketClient) handlePendingInvestigation(investigation PendingInvest
aiContent := tzResp.Choices[0].Message.Content
if len(aiContent) > 300 {
fmt.Printf("🤖 AI Response preview: %s...\n", aiContent[:300])
// AI response received successfully
} else {
fmt.Printf("🤖 AI Response: %s\n", aiContent)
}
@@ -705,7 +686,6 @@ func (c *WebSocketClient) handlePendingInvestigation(investigation PendingInvest
if err := json.Unmarshal([]byte(aiContent), &resolutionResp); err == nil && resolutionResp.ResponseType == "resolution" {
// This is the final resolution - show summary and complete
fmt.Printf("✅ Detected RESOLUTION response - completing investigation\n")
fmt.Printf("\n=== DIAGNOSIS COMPLETE ===\n")
fmt.Printf("Root Cause: %s\n", resolutionResp.RootCause)
fmt.Printf("Resolution Plan: %s\n", resolutionResp.ResolutionPlan)
@@ -722,7 +702,6 @@ func (c *WebSocketClient) handlePendingInvestigation(investigation PendingInvest
}
if err := json.Unmarshal([]byte(aiContent), &diagnosticResp); err == nil && diagnosticResp.ResponseType == "diagnostic" {
fmt.Printf("✅ Detected DIAGNOSTIC response - continuing conversation\n")
fmt.Printf("🔄 AI requested additional diagnostics, executing...\n")
// Execute additional commands if any
@@ -738,7 +717,6 @@ func (c *WebSocketClient) handlePendingInvestigation(investigation PendingInvest
// Execute additional eBPF programs if any
if len(diagnosticResp.EBPFPrograms) > 0 {
fmt.Printf("🔬 Executing %d additional eBPF programs...\n", len(diagnosticResp.EBPFPrograms))
ebpfResults := c.executeEBPFPrograms(diagnosticResp.EBPFPrograms)
additionalResults["ebpf_results"] = ebpfResults
}
@@ -766,9 +744,7 @@ func (c *WebSocketClient) handlePendingInvestigation(investigation PendingInvest
// Attach final AI response to results for DB and mark as completed_with_analysis
resultsForDB["ai_response"] = finalAIContent
fmt.Printf("💾 Updating database with results and AI analysis...\n")
c.updateInvestigationStatus(investigation.ID, "completed_with_analysis", resultsForDB, nil)
fmt.Printf("✅ Investigation completed with AI analysis\n")
}
// updateInvestigationStatus updates the status of a pending investigation