diff --git a/config.example.yml b/config.example.yml index 46ca8d2..01dc3b2 100644 --- a/config.example.yml +++ b/config.example.yml @@ -126,6 +126,9 @@ ai: api_key: your-gemini-api-key-here model: gemini-2.5-flash timeout: 1m0s + # Ceiling on log incidents the assistant may be asked to explain in a day. + # Unattended work is the only kind that can run up a bill unwatched. + triage_daily_cap: 25 mcp: # Exposes the assistant's tool set to external MCP clients at /api/mcp. Every # call is authenticated and permission-gated exactly as the assistant is. diff --git a/internal/api/internal_logs.go b/internal/api/internal_logs.go new file mode 100644 index 0000000..e118aff --- /dev/null +++ b/internal/api/internal_logs.go @@ -0,0 +1,89 @@ +package api + +import ( + "encoding/json" + "net/http" + "strconv" + + "github.com/flatrun/agent/pkg/models" + "github.com/gin-gonic/gin" +) + +// streamInternalLogs hands a deployment's log lines to a built-in app as newline-delimited +// JSON. The user-facing stream is a websocket because a browser cannot set headers; an app +// can, so it gets the simpler transport and the same reader, which keeps log sources, the +// service filter and level parsing in one implementation. +func (s *Server) streamInternalLogs(c *gin.Context) { + if s.pluginToken == "" || c.GetHeader("X-Plugin-Token") != s.pluginToken { + c.JSON(http.StatusUnauthorized, gin.H{"error": "unauthorized"}) + return + } + + name := c.Query("deployment") + if name == "" { + c.JSON(http.StatusBadRequest, gin.H{"error": "deployment required"}) + return + } + + deployment, err := s.manager.GetDeployment(name) + if err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "Deployment not found"}) + return + } + + source, ok := resolveLogSource(deployment.Metadata, c.Query("source")) + if !ok { + c.JSON(http.StatusBadRequest, gin.H{"error": "unknown log source"}) + return + } + + services, err := s.resolveLogServices(name, c.Query("service")) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + + // Replaying a day of history on every reconnect would re-raise handled incidents. + tail := 0 + if v := c.Query("tail"); v != "" { + if n, parseErr := strconv.Atoi(v); parseErr == nil && n >= 0 { + tail = n + } + } + + // Everything that can fail is resolved before the status goes out, since a 200 followed by + // silence is indistinguishable from a stream that has nothing to say yet. + var filePath string + if source.Type == models.LogSourceFile { + filePath, err = resolveLogFilePath(deployment.Path, source.Path) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + } + + c.Writer.Header().Set("Content-Type", "application/x-ndjson") + c.Writer.Header().Set("Cache-Control", "no-cache") + c.Writer.WriteHeader(http.StatusOK) + c.Writer.Flush() + + ctx := c.Request.Context() + encoder := json.NewEncoder(c.Writer) + + sink := func(line string) { + record := parseLogRecord(line) + if source.Type == models.LogSourceFile && record.Service == "" { + record.Service = source.Name + } + if err := encoder.Encode(logLine{Type: "log", Line: line, Record: record}); err != nil { + return + } + c.Writer.Flush() + } + + if source.Type == models.LogSourceFile { + _ = streamFileLogs(ctx, filePath, tail, sink) + return + } + _ = s.manager.StreamDeploymentLogs(ctx, name, deployment.Path, tail, sink, services...) +} diff --git a/internal/api/internal_logs_stream_test.go b/internal/api/internal_logs_stream_test.go new file mode 100644 index 0000000..c06dc46 --- /dev/null +++ b/internal/api/internal_logs_stream_test.go @@ -0,0 +1,37 @@ +package api + +import ( + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "testing" + + "github.com/flatrun/agent/internal/docker" + "github.com/gin-gonic/gin" +) + +// A stream that answers 200 and then says nothing looks identical to one that has nothing to +// report yet, so a source that cannot be read has to fail before the status goes out. +func TestInternalLogStreamFailsBeforeAnsweringOK(t *testing.T) { + gin.SetMode(gin.TestMode) + base, name := writeLogFilterDeployment(t) + + metadata := "log_sources:\n - id: escape\n name: Escape\n type: file\n path: ../outside.log\n" + if err := os.WriteFile(filepath.Join(base, name, "service.yml"), []byte(metadata), 0644); err != nil { + t.Fatal(err) + } + + server := &Server{manager: docker.NewManager(base), pluginToken: "plugin-secret"} + router := gin.New() + router.GET("/internal/logs/stream", server.streamInternalLogs) + + req := httptest.NewRequest(http.MethodGet, "/internal/logs/stream?deployment="+name+"&source=escape", nil) + req.Header.Set("X-Plugin-Token", "plugin-secret") + w := httptest.NewRecorder() + router.ServeHTTP(w, req) + + if w.Code != http.StatusBadRequest { + t.Fatalf("expected 400, got %d: %s", w.Code, w.Body.String()) + } +} diff --git a/internal/api/internal_logs_test.go b/internal/api/internal_logs_test.go new file mode 100644 index 0000000..3bafbe6 --- /dev/null +++ b/internal/api/internal_logs_test.go @@ -0,0 +1,53 @@ +package api + +import ( + "encoding/json" + "testing" +) + +// The observability app reads this envelope to decide what is an incident. It is a wire +// contract between two packages that are compiled together but talk over HTTP, so the shape +// is pinned here and the app's watcher test decodes the same literal from the other side. +func TestInternalLogEnvelopeShape(t *testing.T) { + raw := "web-1 | 2026-08-06T12:00:31.123456Z ERROR connection refused talking to redis" + + encoded, err := json.Marshal(logLine{Type: "log", Line: raw, Record: parseLogRecord(raw)}) + if err != nil { + t.Fatal(err) + } + + var decoded struct { + Type string `json:"type"` + Line string `json:"line"` + Record struct { + Timestamp string `json:"timestamp"` + Service string `json:"service"` + Level string `json:"level"` + Message string `json:"message"` + } `json:"record"` + } + if err := json.Unmarshal(encoded, &decoded); err != nil { + t.Fatalf("the envelope must decode into the shape the watcher expects: %v", err) + } + + if decoded.Type != "log" { + t.Errorf("type = %q, want log", decoded.Type) + } + if decoded.Line != raw { + t.Errorf("line should be the untouched original, got %q", decoded.Line) + } + if decoded.Record.Service != "web-1" { + t.Errorf("service should come from the compose prefix, got %q", decoded.Record.Service) + } + if decoded.Record.Level != "error" { + t.Errorf("level should be parsed to a canonical name, got %q", decoded.Record.Level) + } + // The compose prefix and the leading timestamp are stripped; the level word stays in the + // message, which is what the app fingerprints on. + if decoded.Record.Message != "ERROR connection refused talking to redis" { + t.Errorf("message should be the line without the compose prefix or timestamp, got %q", decoded.Record.Message) + } + if decoded.Record.Timestamp != "2026-08-06T12:00:31.123456Z" { + t.Errorf("timestamp should be lifted out of the line, got %q", decoded.Record.Timestamp) + } +} diff --git a/internal/api/internal_triage.go b/internal/api/internal_triage.go new file mode 100644 index 0000000..36c1575 --- /dev/null +++ b/internal/api/internal_triage.go @@ -0,0 +1,224 @@ +package api + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "strings" + "sync" + "time" + "unicode/utf8" + + "github.com/flatrun/agent/internal/ai" + "github.com/gin-gonic/gin" +) + +// What one triage may cost. The app's funnel should keep it far below these; they exist so a +// bug in the funnel cannot become a bill. +const ( + maxTriageContextLines = 40 + maxTriageContextChars = 8000 + maxTriageOutputTokens = 400 + defaultTriageDailyCap = 25 +) + +// triageBudget counts triages against a daily ceiling. In-memory on purpose: this bounds a +// runaway, and a restart clearing the count costs less than persisting a counter for it. +type triageBudget struct { + mu sync.Mutex + day string + spent int + cap int +} + +func newTriageBudget(dailyCap int) *triageBudget { + if dailyCap <= 0 { + dailyCap = defaultTriageDailyCap + } + return &triageBudget{cap: dailyCap} +} + +func (b *triageBudget) take(now time.Time) (int, int, bool) { + b.mu.Lock() + defer b.mu.Unlock() + + today := now.UTC().Format("2006-01-02") + if b.day != today { + b.day = today + b.spent = 0 + } + if b.spent >= b.cap { + return b.spent, b.cap, false + } + b.spent++ + return b.spent, b.cap, true +} + +type triageRequest struct { + RuleName string `json:"rule_name"` + Deployment string `json:"deployment"` + Service string `json:"service"` + Level string `json:"level"` + Count int `json:"count"` + Sample string `json:"sample"` + Context []string `json:"context"` +} + +type triageResponse struct { + Summary string `json:"summary,omitempty"` + Cause string `json:"cause,omitempty"` + NextStep string `json:"next_step,omitempty"` + Severity string `json:"severity,omitempty"` + Confidence string `json:"confidence,omitempty"` +} + +const triageSystemPrompt = `You triage one recurring error from a container's logs on a self-hosted server. + +You are given the failing line, a little of the output around it, and how often it has just occurred. You are not given the application's source, and you cannot run anything. Say what the evidence supports and no more. + +Reply with JSON only, no prose and no code fence, with these keys: + summary one sentence an operator can read at 3am + cause the most likely cause, or "" if the lines do not say + next_step the single most useful thing to do next + severity one of: low, medium, high + confidence one of: low, medium, high + +If the lines are not enough to tell what is wrong, say so in summary, leave cause empty, and set confidence to low. A wrong confident answer is worse than an honest empty one.` + +// triageLogIncident explains one log incident for a built-in app. The app's funnel decides +// what is worth asking; this decides what the asking may cost. +func (s *Server) triageLogIncident(c *gin.Context) { + if s.pluginToken == "" || c.GetHeader("X-Plugin-Token") != s.pluginToken { + c.JSON(http.StatusUnauthorized, gin.H{"error": "unauthorized"}) + return + } + if s.aiProvider == nil { + c.JSON(http.StatusServiceUnavailable, gin.H{"error": "the assistant is not configured"}) + return + } + + var req triageRequest + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid request"}) + return + } + if strings.TrimSpace(req.Sample) == "" { + c.JSON(http.StatusBadRequest, gin.H{"error": "nothing to triage"}) + return + } + + spent, cap, ok := s.triageBudget.take(time.Now()) + if !ok { + // The ceiling lives here so it holds whatever asks for a triage. + c.JSON(http.StatusTooManyRequests, gin.H{ + "error": fmt.Sprintf("daily triage budget spent (%d of %d)", spent, cap), + }) + return + } + + prompt := buildTriagePrompt(req, s.redactorFor(req.Deployment)) + + ctx, cancel := context.WithTimeout(c.Request.Context(), 40*time.Second) + defer cancel() + + resp, err := s.aiProvider.Complete(ctx, ai.Request{ + Messages: []ai.Message{ + {Role: "system", Content: triageSystemPrompt}, + {Role: "user", Content: prompt}, + }, + MaxTokens: maxTriageOutputTokens, + Temperature: 0, + }) + if err != nil { + c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()}) + return + } + + verdict, err := parseTriageVerdict(resp.Content) + if err != nil { + c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()}) + return + } + + c.JSON(http.StatusOK, gin.H{ + "triage": verdict, + "usage": resp.Usage, + "budget": gin.H{"spent": spent, "cap": cap}, + }) +} + +func (s *Server) redactorFor(deployment string) *ai.Redactor { + secrets := s.systemSecretValues() + if deployment != "" { + secrets = s.deploymentSecretValues(deployment) + } + return ai.NewRedactor(secrets) +} + +// buildTriagePrompt caps by line count and by total size, since one line of a base64 payload +// can be larger than forty ordinary ones. +func buildTriagePrompt(req triageRequest, redactor *ai.Redactor) string { + lines := req.Context + if len(lines) > maxTriageContextLines { + lines = lines[len(lines)-maxTriageContextLines:] + } + + var b strings.Builder + fmt.Fprintf(&b, "Deployment: %s\n", req.Deployment) + if req.Service != "" { + fmt.Fprintf(&b, "Service: %s\n", req.Service) + } + if req.RuleName != "" { + fmt.Fprintf(&b, "Rule: %s\n", req.RuleName) + } + fmt.Fprintf(&b, "Level: %s\nOccurrences just now: %d\n\nFailing line:\n%s\n", req.Level, req.Count, req.Sample) + if len(lines) > 0 { + b.WriteString("\nSurrounding output:\n") + b.WriteString(strings.Join(lines, "\n")) + } + + text := b.String() + if len(text) > maxTriageContextChars { + // The head carries the deployment, the rule and the failing line. The cut walks back to + // a rune boundary so a log line in any language cannot end in half a character. + cut := maxTriageContextChars + for cut > 0 && !utf8.RuneStart(text[cut]) { + cut-- + } + text = text[:cut] + "\n[truncated]" + } + if redactor != nil { + text, _ = redactor.Redact(text) + } + return text +} + +// parseTriageVerdict tolerates a code fence, the usual way a model ignores "JSON only". +func parseTriageVerdict(content string) (triageResponse, error) { + text := strings.TrimSpace(content) + if fence := strings.Index(text, "```"); fence >= 0 { + rest := text[fence+3:] + if nl := strings.IndexByte(rest, '\n'); nl >= 0 { + rest = rest[nl+1:] + } + if end := strings.Index(rest, "```"); end >= 0 { + rest = rest[:end] + } + text = strings.TrimSpace(rest) + } + start := strings.IndexByte(text, '{') + end := strings.LastIndexByte(text, '}') + if start < 0 || end <= start { + return triageResponse{}, fmt.Errorf("the assistant did not answer with a verdict") + } + + var verdict triageResponse + if err := json.Unmarshal([]byte(text[start:end+1]), &verdict); err != nil { + return triageResponse{}, fmt.Errorf("the assistant's verdict could not be read: %w", err) + } + if strings.TrimSpace(verdict.Summary) == "" { + return triageResponse{}, fmt.Errorf("the assistant's verdict had no summary") + } + return verdict, nil +} diff --git a/internal/api/internal_triage_test.go b/internal/api/internal_triage_test.go new file mode 100644 index 0000000..bdd254f --- /dev/null +++ b/internal/api/internal_triage_test.go @@ -0,0 +1,227 @@ +package api + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + "unicode/utf8" + + "github.com/flatrun/agent/internal/ai" + "github.com/flatrun/agent/pkg/config" + "github.com/gin-gonic/gin" +) + +// countingProvider stands in for the model so the tests exercise the budget, the caps and the +// parsing rather than a network call. The package already has a stub for request shape; this +// one counts calls, which is what the budget assertions turn on. +type countingProvider struct { + calls int + lastUser string + reply string + err error +} + +func (p *countingProvider) Name() string { return "counting" } + +func (p *countingProvider) Complete(_ context.Context, req ai.Request) (*ai.Response, error) { + p.calls++ + for _, m := range req.Messages { + if m.Role == "user" { + p.lastUser = m.Content + } + } + if p.err != nil { + return nil, p.err + } + return &ai.Response{Content: p.reply}, nil +} + +func triageServer(t *testing.T, provider ai.Provider, dailyCap int) (*gin.Engine, *Server) { + t.Helper() + gin.SetMode(gin.TestMode) + + server := &Server{ + pluginToken: "plugin-secret", + aiProvider: provider, + triageBudget: newTriageBudget(dailyCap), + config: &config.Config{}, + } + router := gin.New() + router.POST("/internal/ai/triage", server.triageLogIncident) + return router, server +} + +func triagePost(t *testing.T, router *gin.Engine, token string, body map[string]any) *httptest.ResponseRecorder { + t.Helper() + raw, err := json.Marshal(body) + if err != nil { + t.Fatal(err) + } + req := httptest.NewRequest(http.MethodPost, "/internal/ai/triage", bytes.NewReader(raw)) + req.Header.Set("Content-Type", "application/json") + if token != "" { + req.Header.Set("X-Plugin-Token", token) + } + w := httptest.NewRecorder() + router.ServeHTTP(w, req) + return w +} + +// The daily ceiling is what stops a bug in the funnel becoming a bill, so it has to hold at +// the endpoint rather than only in the app that calls it. +func TestTriageRefusesOnceTheDailyBudgetIsSpent(t *testing.T) { + provider := &countingProvider{reply: `{"summary":"redis is unreachable","severity":"high","confidence":"medium"}`} + router, _ := triageServer(t, provider, 2) + + body := map[string]any{"deployment": "shop", "sample": "connection refused", "count": 5} + + for i := 0; i < 2; i++ { + if w := triagePost(t, router, "plugin-secret", body); w.Code != http.StatusOK { + t.Fatalf("call %d should be allowed, got %d: %s", i+1, w.Code, w.Body.String()) + } + } + + w := triagePost(t, router, "plugin-secret", body) + if w.Code != http.StatusTooManyRequests { + t.Fatalf("the third call should be refused, got %d", w.Code) + } + if !strings.Contains(w.Body.String(), "budget") { + t.Errorf("the refusal should say why, got %s", w.Body.String()) + } + if provider.calls != 2 { + t.Errorf("a refused triage must not reach the model, got %d calls", provider.calls) + } +} + +func TestTriageRequiresThePluginToken(t *testing.T) { + provider := &countingProvider{reply: `{"summary":"x"}`} + router, _ := triageServer(t, provider, 10) + + if w := triagePost(t, router, "", map[string]any{"sample": "x"}); w.Code != http.StatusUnauthorized { + t.Fatalf("expected 401 without a token, got %d", w.Code) + } + if w := triagePost(t, router, "wrong", map[string]any{"sample": "x"}); w.Code != http.StatusUnauthorized { + t.Fatalf("expected 401 with the wrong token, got %d", w.Code) + } + if provider.calls != 0 { + t.Errorf("an unauthorized call must not reach the model") + } +} + +// One line of a base64 payload can be bigger than forty ordinary ones, so the prompt is +// capped by size as well as by line count. +func TestTriagePromptIsCappedBySizeAndLines(t *testing.T) { + provider := &countingProvider{reply: `{"summary":"ok"}`} + router, _ := triageServer(t, provider, 10) + + context := make([]string, 500) + for i := range context { + context[i] = strings.Repeat("x", 500) + } + + w := triagePost(t, router, "plugin-secret", map[string]any{ + "deployment": "shop", + "sample": "boom", + "count": 3, + "context": context, + }) + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String()) + } + if len(provider.lastUser) > maxTriageContextChars+64 { + t.Errorf("prompt should be capped near %d chars, got %d", maxTriageContextChars, len(provider.lastUser)) + } +} + +// Logs are not all ASCII, and a prompt cut mid-character is not valid UTF-8. +func TestTriagePromptCutsOnACharacterBoundary(t *testing.T) { + provider := &countingProvider{reply: `{"summary":"ok"}`} + router, _ := triageServer(t, provider, 10) + + context := make([]string, 40) + for i := range context { + context[i] = strings.Repeat("é", 500) + } + + w := triagePost(t, router, "plugin-secret", map[string]any{ + "deployment": "shop", + "sample": "boom", + "count": 3, + "context": context, + }) + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String()) + } + if !utf8.ValidString(provider.lastUser) { + t.Error("the truncated prompt is not valid UTF-8") + } +} + +// A model that wraps its JSON in a code fence is the common case, not an error. +func TestTriageReadsAFencedVerdict(t *testing.T) { + provider := &countingProvider{reply: "```json\n{\"summary\":\"disk is full\",\"next_step\":\"free space\"}\n```"} + router, _ := triageServer(t, provider, 10) + + w := triagePost(t, router, "plugin-secret", map[string]any{"deployment": "shop", "sample": "no space left"}) + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String()) + } + var resp struct { + Triage triageResponse `json:"triage"` + } + if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil { + t.Fatal(err) + } + if resp.Triage.Summary != "disk is full" { + t.Errorf("expected the fenced verdict to be read, got %+v", resp.Triage) + } +} + +// A reply that is not a verdict is an error, not an incident annotated with prose. +func TestTriageRejectsANonVerdictReply(t *testing.T) { + provider := &countingProvider{reply: "I would need to see the source code to say."} + router, _ := triageServer(t, provider, 10) + + w := triagePost(t, router, "plugin-secret", map[string]any{"deployment": "shop", "sample": "boom"}) + if w.Code != http.StatusBadGateway { + t.Fatalf("expected 502 for an unusable reply, got %d: %s", w.Code, w.Body.String()) + } +} + +func TestTriageWithoutAProviderIsUnavailable(t *testing.T) { + router, _ := triageServer(t, nil, 10) + if w := triagePost(t, router, "plugin-secret", map[string]any{"sample": "x"}); w.Code != http.StatusServiceUnavailable { + t.Fatalf("expected 503 when the assistant is not configured, got %d", w.Code) + } +} + +// The budget rolls over rather than being spent forever. +func TestTriageBudgetResetsDaily(t *testing.T) { + b := newTriageBudget(1) + day1 := time.Date(2026, 8, 6, 23, 59, 0, 0, time.UTC) + if _, _, ok := b.take(day1); !ok { + t.Fatal("the first call of the day should be allowed") + } + if _, _, ok := b.take(day1); ok { + t.Fatal("the second call should be refused") + } + day2 := time.Date(2026, 8, 7, 0, 1, 0, 0, time.UTC) + if _, _, ok := b.take(day2); !ok { + t.Fatal("the next day should start fresh") + } +} + +func TestParseTriageVerdictRequiresASummary(t *testing.T) { + if _, err := parseTriageVerdict(`{"cause":"something"}`); err == nil { + t.Fatal("a verdict with no summary is not usable") + } + if _, err := parseTriageVerdict(fmt.Sprintf(`{"summary":%q}`, "ok")); err != nil { + t.Fatalf("a verdict with a summary should parse, got %v", err) + } +} diff --git a/internal/api/log_service_filter_test.go b/internal/api/log_service_filter_test.go new file mode 100644 index 0000000..5ab7901 --- /dev/null +++ b/internal/api/log_service_filter_test.go @@ -0,0 +1,111 @@ +package api + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "testing" + + "github.com/flatrun/agent/internal/docker" + "github.com/gin-gonic/gin" +) + +func writeLogFilterDeployment(t *testing.T) (base string, name string) { + t.Helper() + + base = t.TempDir() + name = "log-filter-app" + dir := filepath.Join(base, name) + if err := os.MkdirAll(dir, 0755); err != nil { + t.Fatal(err) + } + + compose := "name: " + name + ` +services: + web: + image: nginx:alpine + worker: + image: alpine:latest +` + if err := os.WriteFile(filepath.Join(dir, "docker-compose.yml"), []byte(compose), 0644); err != nil { + t.Fatal(err) + } + return base, name +} + +// A service name reaches compose as an argument, so it is checked against the compose file +// before it gets there rather than passed through. +func TestDeploymentLogsRejectUnknownService(t *testing.T) { + gin.SetMode(gin.TestMode) + + base, name := writeLogFilterDeployment(t) + server := &Server{manager: docker.NewManager(base)} + router := gin.New() + router.GET("/deployments/:name/logs", server.getDeploymentLogs) + + req := httptest.NewRequest(http.MethodGet, "/deployments/"+name+"/logs?service=--rm", nil) + w := httptest.NewRecorder() + router.ServeHTTP(w, req) + + if w.Code != http.StatusBadRequest { + t.Fatalf("expected 400 for an unknown service, got %d: %s", w.Code, w.Body.String()) + } + + var resp struct { + Error string `json:"error"` + } + if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil { + t.Fatalf("failed to decode response: %v", err) + } + for _, want := range []string{"web", "worker"} { + if !strings.Contains(resp.Error, want) { + t.Errorf("error should name the available service %q, got %q", want, resp.Error) + } + } +} + +// A file source is one file the deployment writes, so asking for a service alongside it is +// not an error: there is simply nothing per-service to narrow. +func TestDeploymentLogsAcceptServiceAlongsideFileSource(t *testing.T) { + gin.SetMode(gin.TestMode) + + base, name := writeLogFilterDeployment(t) + logPath := filepath.Join(base, name, "app.log") + if err := os.WriteFile(logPath, []byte("first line\nsecond line\n"), 0644); err != nil { + t.Fatal(err) + } + metadata := `log_sources: + - id: app-file + name: App file + type: file + path: app.log +` + if err := os.WriteFile(filepath.Join(base, name, "service.yml"), []byte(metadata), 0644); err != nil { + t.Fatal(err) + } + + server := &Server{manager: docker.NewManager(base)} + router := gin.New() + router.GET("/deployments/:name/logs", server.getDeploymentLogs) + + req := httptest.NewRequest(http.MethodGet, "/deployments/"+name+"/logs?source=app-file&service=web", nil) + w := httptest.NewRecorder() + router.ServeHTTP(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String()) + } + + var resp struct { + Logs string `json:"logs"` + } + if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil { + t.Fatalf("failed to decode response: %v", err) + } + if !strings.Contains(resp.Logs, "second line") { + t.Errorf("expected the file's contents, got %q", resp.Logs) + } +} diff --git a/internal/api/log_stream.go b/internal/api/log_stream.go index 028a604..8eccd26 100644 --- a/internal/api/log_stream.go +++ b/internal/api/log_stream.go @@ -104,6 +104,14 @@ func (s *Server) streamDeploymentLogs(c *gin.Context) { return } + // A file source is one file the deployment writes, so there is nothing per-service to + // narrow it to; the filter only applies to container output. + services, err := s.resolveLogServices(name, c.Query("service")) + if err != nil { + sendError(conn, err.Error()) + return + } + // The stream ends when the viewer disconnects, which is what stops the compose process // rather than leaving it attached for the life of the agent. ctx := c.Request.Context() @@ -145,7 +153,7 @@ func (s *Server) streamDeploymentLogs(c *gin.Context) { return } } else { - err = s.manager.StreamDeploymentLogs(ctx, name, deployment.Path, tail, sink) + err = s.manager.StreamDeploymentLogs(ctx, name, deployment.Path, tail, sink, services...) if err != nil && ctx.Err() == nil { sendError(conn, err.Error()) return diff --git a/internal/api/log_truncate.go b/internal/api/log_truncate.go new file mode 100644 index 0000000..76ffa59 --- /dev/null +++ b/internal/api/log_truncate.go @@ -0,0 +1,148 @@ +package api + +import ( + "fmt" + "net/http" + "os" + "strings" + + "github.com/flatrun/agent/internal/infra" + "github.com/flatrun/agent/pkg/models" + "github.com/gin-gonic/gin" +) + +// truncateContainerLog empties what Docker has stored for a container. +// +// Docker keeps the log itself, so there is nothing to delete through the API: the file behind +// LogPath is truncated in place. That keeps the container running and its file descriptor +// valid, which deleting the file would not. +func (s *Server) truncateContainerLog(container string) error { + if strings.TrimSpace(container) == "" { + return fmt.Errorf("no container to clear") + } + + path, err := s.manager.ContainerLogPath(container) + if err != nil { + return fmt.Errorf("could not find the log for %s: %w", container, err) + } + + path = strings.TrimSpace(path) + if path == "" { + // A container on journald, syslog or a remote driver has no file here to empty, and + // clearing whatever it does write is that system's business, not FlatRun's. + return fmt.Errorf("%s does not log to a file Docker owns, so there is nothing here to clear", container) + } + if err := os.Truncate(path, 0); err != nil { + if os.IsPermission(err) { + return fmt.Errorf("the agent is not allowed to clear %s: %w", path, err) + } + return err + } + return nil +} + +// deleteDeploymentLogs empties the log the viewer is reading: the file for a file source, or +// what Docker holds for the containers behind container output. +func (s *Server) deleteDeploymentLogs(c *gin.Context) { + name := c.Param("name") + + deployment, err := s.manager.GetDeployment(name) + if err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "Deployment not found"}) + return + } + + source, ok := resolveLogSource(deployment.Metadata, c.Query("source")) + if !ok { + c.JSON(http.StatusBadRequest, gin.H{"error": "unknown log source"}) + return + } + + // Checked against the compose file for the same reason reading is: a name that matches + // nothing would otherwise report success while emptying nothing. + wantedServices, err := s.resolveLogServices(name, c.Query("service")) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + + if source.Type == models.LogSourceFile { + path, pathErr := resolveLogFilePath(deployment.Path, source.Path) + if pathErr != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": pathErr.Error()}) + return + } + if truncErr := os.Truncate(path, 0); truncErr != nil && !os.IsNotExist(truncErr) { + c.JSON(http.StatusInternalServerError, gin.H{"error": truncErr.Error()}) + return + } + c.JSON(http.StatusOK, gin.H{"message": "Log cleared", "source": source.ID}) + return + } + + services, err := s.manager.GetComposeServices(name) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + + var failures []string + cleared := 0 + for _, svc := range services { + if len(wantedServices) > 0 && wantedServices[0] != svc.Name { + continue + } + if svc.ContainerID == "" { + continue + } + if truncErr := s.truncateContainerLog(svc.ContainerID); truncErr != nil { + failures = append(failures, truncErr.Error()) + continue + } + cleared++ + } + + if cleared == 0 && len(failures) > 0 { + c.JSON(http.StatusInternalServerError, gin.H{"error": strings.Join(failures, "; ")}) + return + } + c.JSON(http.StatusOK, gin.H{ + "message": "Log cleared", + "source": source.ID, + "cleared": cleared, + "warnings": failures, + }) +} + +// deleteSystemLogs empties what Docker holds for a system service's container. +func (s *Server) deleteSystemLogs(c *gin.Context) { + if s.infraManager == nil { + c.JSON(http.StatusServiceUnavailable, gin.H{"error": "Infrastructure not available"}) + return + } + + src, ok := s.resolveSystemLogSource(c.Query("source")) + if !ok { + c.JSON(http.StatusBadRequest, gin.H{"error": "unknown log source"}) + return + } + + container := s.infraManager.ContainerName(src.Service) + if container == "" { + c.JSON(http.StatusBadRequest, gin.H{"error": "unknown service"}) + return + } + + // Access and error share one container, so clearing either clears both. Saying so is + // better than quietly emptying more than was asked for. + if err := s.truncateContainerLog(container); err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + + message := "Log cleared" + if src.Stream != infra.LogStreamAll { + message = "Log cleared, including the other stream from the same container" + } + c.JSON(http.StatusOK, gin.H{"message": message, "source": src.ID}) +} diff --git a/internal/api/log_truncate_test.go b/internal/api/log_truncate_test.go new file mode 100644 index 0000000..c6582ed --- /dev/null +++ b/internal/api/log_truncate_test.go @@ -0,0 +1,109 @@ +package api + +import ( + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "testing" + + "github.com/flatrun/agent/internal/docker" + "github.com/gin-gonic/gin" +) + +func truncateRouter(t *testing.T, base string) *gin.Engine { + t.Helper() + gin.SetMode(gin.TestMode) + server := &Server{manager: docker.NewManager(base)} + router := gin.New() + router.DELETE("/deployments/:name/logs", server.deleteDeploymentLogs) + return router +} + +func TestDeleteDeploymentLogsEmptiesTheFile(t *testing.T) { + base, name := writeLogFilterDeployment(t) + logPath := filepath.Join(base, name, "app.log") + if err := os.WriteFile(logPath, []byte("first line\nsecond line\n"), 0644); err != nil { + t.Fatal(err) + } + metadata := `log_sources: + - id: app-file + name: App file + type: file + path: app.log +` + if err := os.WriteFile(filepath.Join(base, name, "service.yml"), []byte(metadata), 0644); err != nil { + t.Fatal(err) + } + + req := httptest.NewRequest(http.MethodDelete, "/deployments/"+name+"/logs?source=app-file", nil) + w := httptest.NewRecorder() + truncateRouter(t, base).ServeHTTP(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String()) + } + + info, err := os.Stat(logPath) + if err != nil { + t.Fatalf("the file should still exist so the application keeps writing to it: %v", err) + } + if info.Size() != 0 { + t.Errorf("expected an empty file, got %d bytes", info.Size()) + } +} + +// A source pointing outside the deployment is refused here as it is on read, so a crafted +// source cannot turn this into "truncate any file on the host". +func TestDeleteDeploymentLogsRefusesAnEscapingPath(t *testing.T) { + base, name := writeLogFilterDeployment(t) + outside := filepath.Join(base, "outside.log") + if err := os.WriteFile(outside, []byte("not yours\n"), 0644); err != nil { + t.Fatal(err) + } + metadata := "log_sources:\n - id: escape\n name: Escape\n type: file\n path: ../outside.log\n" + if err := os.WriteFile(filepath.Join(base, name, "service.yml"), []byte(metadata), 0644); err != nil { + t.Fatal(err) + } + + req := httptest.NewRequest(http.MethodDelete, "/deployments/"+name+"/logs?source=escape", nil) + w := httptest.NewRecorder() + truncateRouter(t, base).ServeHTTP(w, req) + + if w.Code != http.StatusBadRequest { + t.Fatalf("expected 400, got %d: %s", w.Code, w.Body.String()) + } + if info, err := os.Stat(outside); err != nil || info.Size() == 0 { + t.Errorf("the file outside the deployment must be untouched") + } +} + +// Emptying nothing at all should not read as success, which is what a mistyped service used to +// get: 200 with a count of zero. +func TestDeleteDeploymentLogsRejectsAnUnknownService(t *testing.T) { + base, name := writeLogFilterDeployment(t) + + req := httptest.NewRequest(http.MethodDelete, "/deployments/"+name+"/logs?source=stdout&service=nope", nil) + w := httptest.NewRecorder() + truncateRouter(t, base).ServeHTTP(w, req) + + if w.Code != http.StatusBadRequest { + t.Fatalf("expected 400, got %d: %s", w.Code, w.Body.String()) + } + if !strings.Contains(w.Body.String(), "web") { + t.Errorf("the error should name the services that do exist, got %s", w.Body.String()) + } +} + +func TestDeleteDeploymentLogsRejectsAnUnknownSource(t *testing.T) { + base, name := writeLogFilterDeployment(t) + + req := httptest.NewRequest(http.MethodDelete, "/deployments/"+name+"/logs?source=nope", nil) + w := httptest.NewRecorder() + truncateRouter(t, base).ServeHTTP(w, req) + + if w.Code != http.StatusBadRequest { + t.Fatalf("expected 400, got %d: %s", w.Code, w.Body.String()) + } +} diff --git a/internal/api/logfile.go b/internal/api/logfile.go index 10f9b8c..932ab03 100644 --- a/internal/api/logfile.go +++ b/internal/api/logfile.go @@ -25,6 +25,25 @@ func resolveLogSource(meta *models.ServiceMetadata, id string) (models.LogSource return meta.FindLogSource(id) } +// resolveLogServices turns the requested service into the compose service list to read logs +// from: none for "" or "all", meaning every service. The name is checked against the compose +// file rather than passed through, so a caller cannot smuggle arguments into compose. +func (s *Server) resolveLogServices(name, service string) ([]string, error) { + if service == "" || service == "all" { + return nil, nil + } + names, err := s.manager.GetComposeServiceNames(name) + if err != nil { + return nil, err + } + for _, sn := range names { + if sn == service { + return []string{service}, nil + } + } + return nil, fmt.Errorf("service '%s' not found in compose file, available: %s", service, strings.Join(names, ", ")) +} + var errLogPathEscapes = errors.New("log source path escapes the deployment directory") // readAllCap bounds how much a tail<=0 ("all") read pulls into memory. diff --git a/internal/api/server.go b/internal/api/server.go index 235d533..44f2bb3 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -104,6 +104,7 @@ type Server struct { certRenewer *ssl.Renewer planStore *plan.Store aiProvider ai.Provider + triageBudget *triageBudget aiSessions *ai.SessionStore aiAgents *ai.AgentStore mcpHandler http.Handler @@ -390,6 +391,7 @@ func New(cfg *config.Config, configPath string) *Server { } else if aiErr != ai.ErrDisabled { log.Printf("Warning: failed to initialize AI provider: %v", aiErr) } + s.triageBudget = newTriageBudget(cfg.AI.TriageDailyCap) if backupManager != nil { if err := s.applyBackupDestinations(); err != nil { @@ -426,6 +428,7 @@ func (s *Server) setupRoutes() { api.GET("/system/terminal/interactive", s.systemTerminalInteractive) api.GET("/deployments/:name/jobs/:jobId/stream", s.streamDeploymentJob) api.GET("/deployments/:name/logs/stream", s.streamDeploymentLogs) + api.GET("/system/logs/stream", s.streamSystemLogs) // Setup endpoints (public, gated by setup state) setupGroup := api.Group("/setup") @@ -469,6 +472,7 @@ func (s *Server) setupRoutes() { protected.GET("/deployments/:name/jobs/:jobId", s.authMiddleware.RequirePermission(auth.PermDeploymentsRead), s.authMiddleware.RequireDeploymentAccess(auth.AccessLevelRead), s.getDeploymentJob) protected.POST("/deployments/:name/actions/:actionId", s.authMiddleware.RequirePermission(auth.PermDeploymentsWrite), s.authMiddleware.RequireDeploymentAccess(auth.AccessLevelWrite), s.executeQuickAction) protected.GET("/deployments/:name/logs", s.authMiddleware.RequirePermission(auth.PermDeploymentsRead), s.authMiddleware.RequireDeploymentAccess(auth.AccessLevelRead), s.getDeploymentLogs) + protected.DELETE("/deployments/:name/logs", s.authMiddleware.RequirePermission(auth.PermDeploymentsWrite), s.authMiddleware.RequireDeploymentAccess(auth.AccessLevelWrite), s.deleteDeploymentLogs) protected.GET("/deployments/:name/log-sources", s.authMiddleware.RequirePermission(auth.PermDeploymentsRead), s.authMiddleware.RequireDeploymentAccess(auth.AccessLevelRead), s.getDeploymentLogSources) protected.PUT("/deployments/:name/log-sources", s.authMiddleware.RequirePermission(auth.PermDeploymentsWrite), s.authMiddleware.RequireDeploymentAccess(auth.AccessLevelWrite), s.updateDeploymentLogSources) protected.GET("/deployments/:name/compose", s.authMiddleware.RequirePermission(auth.PermDeploymentsRead), s.authMiddleware.RequireDeploymentAccess(auth.AccessLevelRead), s.getDeploymentCompose) @@ -678,6 +682,9 @@ func (s *Server) setupRoutes() { protected.POST("/infrastructure/:name/stop", s.authMiddleware.RequirePermission(auth.PermInfrastructureWrite), s.stopInfraService) protected.POST("/infrastructure/:name/restart", s.authMiddleware.RequirePermission(auth.PermInfrastructureWrite), s.restartInfraService) protected.GET("/infrastructure/:name/logs", s.authMiddleware.RequirePermission(auth.PermInfrastructureRead), s.getInfraServiceLogs) + protected.GET("/system/logs/sources", s.authMiddleware.RequirePermission(auth.PermInfrastructureRead), s.listSystemLogSources) + protected.GET("/system/logs", s.authMiddleware.RequirePermission(auth.PermInfrastructureRead), s.getSystemLogs) + protected.DELETE("/system/logs", s.authMiddleware.RequirePermission(auth.PermInfrastructureWrite), s.deleteSystemLogs) protected.POST("/infrastructure/migrate/:name", s.authMiddleware.RequirePermission(auth.PermInfrastructureWrite), s.migrateToInfrastructure) // Registry endpoints @@ -869,6 +876,10 @@ func (s *Server) setupRoutes() { // Plugin-emitted notifications (authenticated by the per-run plugin token). api.POST("/internal/notify/emit", s.emitNotification) + // Log lines and triage for built-in apps, on the same plugin token. Both keep one + // implementation in the agent rather than a second one inside every app. + api.GET("/internal/logs/stream", s.streamInternalLogs) + api.POST("/internal/ai/triage", s.triageLogIncident) // Ingest endpoints (no auth - called by nginx Lua) api.POST("/security/events/ingest", s.ingestSecurityEvent) @@ -2505,6 +2516,14 @@ func (s *Server) getDeploymentLogs(c *gin.Context) { return } + // A file source is one file the deployment writes, so there is nothing per-service to + // narrow it to; the filter only applies to container output. + services, err := s.resolveLogServices(name, c.Query("service")) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + var logs string if source.Type == models.LogSourceFile { path, err := resolveLogFilePath(deployment.Path, source.Path) @@ -2518,7 +2537,7 @@ func (s *Server) getDeploymentLogs(c *gin.Context) { return } } else { - logs, err = s.manager.GetDeploymentLogs(name, tail) + logs, err = s.manager.GetDeploymentLogs(name, tail, services...) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return @@ -2530,6 +2549,7 @@ func (s *Server) getDeploymentLogs(c *gin.Context) { c.JSON(http.StatusOK, gin.H{ "name": name, "source": source.ID, + "service": strings.Join(services, ","), "logs": logs, "records": parseLogRecords(logs), }) diff --git a/internal/api/system_logs.go b/internal/api/system_logs.go new file mode 100644 index 0000000..686b830 --- /dev/null +++ b/internal/api/system_logs.go @@ -0,0 +1,337 @@ +package api + +import ( + "encoding/json" + "log" + "net/http" + "strconv" + "strings" + "time" + + "github.com/flatrun/agent/internal/auth" + "github.com/flatrun/agent/internal/contextkeys" + "github.com/flatrun/agent/internal/infra" + "github.com/flatrun/agent/pkg/models" + "github.com/gin-gonic/gin" + "github.com/gorilla/websocket" +) + +// systemLogSource is one place the host itself writes logs, as opposed to a deployment: the +// proxy's access and error logs, and whatever shared infrastructure is running. +type systemLogSource struct { + ID string `json:"id"` + Name string `json:"name"` + Service string `json:"service"` + Stream string `json:"stream"` + // ByDeployment says whether lines carry which deployment they belong to, which only the + // proxy's access log does. + ByDeployment bool `json:"by_deployment"` +} + +func systemLogSourcesFor(services []models.InfraService) []systemLogSource { + var sources []systemLogSource + for _, svc := range services { + switch svc.Type { + case models.InfraTypeNginx: + sources = append(sources, + systemLogSource{ID: "nginx-access", Name: "nginx access", Service: svc.Name, Stream: infra.LogStreamStdout, ByDeployment: true}, + systemLogSource{ID: "nginx-error", Name: "nginx error", Service: svc.Name, Stream: infra.LogStreamStderr}, + ) + default: + sources = append(sources, systemLogSource{ + ID: svc.Name, + Name: svc.Name, + Service: svc.Name, + Stream: infra.LogStreamAll, + }) + } + } + return sources +} + +func (s *Server) systemLogSources() ([]systemLogSource, error) { + services, err := s.infraManager.ListServices() + if err != nil { + return nil, err + } + return systemLogSourcesFor(services), nil +} + +func (s *Server) resolveSystemLogSource(id string) (systemLogSource, bool) { + sources, err := s.systemLogSources() + if err != nil { + return systemLogSource{}, false + } + if id == "" && len(sources) > 0 { + return sources[0], true + } + for _, src := range sources { + if src.ID == id { + return src, true + } + } + return systemLogSource{}, false +} + +func (s *Server) listSystemLogSources(c *gin.Context) { + sources, err := s.systemLogSources() + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + if sources == nil { + sources = []systemLogSource{} + } + c.JSON(http.StatusOK, gin.H{"sources": sources}) +} + +// deploymentHostMatcher keeps only the access lines belonging to one deployment, by matching +// the host the request asked for against the domains that deployment serves. The host is a +// field of its own in the access log, so this never matches a domain that merely appears in a +// referer or a user agent. +func (s *Server) deploymentHostMatcher(name string) (func(string) bool, error) { + deployment, err := s.manager.GetDeployment(name) + if err != nil { + return nil, err + } + + hosts := map[string]struct{}{} + if deployment.Metadata != nil { + for _, d := range deployment.Metadata.GetDomains() { + if d.Domain != "" { + hosts[strings.ToLower(d.Domain)] = struct{}{} + } + } + } + if len(hosts) == 0 { + // Nothing is served for this deployment, so no access line can belong to it. + return func(string) bool { return false }, nil + } + + return func(line string) bool { + // Lines arrive with docker's timestamp ahead of nginx's own fields, so the host is + // one of the first two fields depending on whether that prefix is present. + fields := strings.Fields(line) + for i := 0; i < len(fields) && i < 2; i++ { + host := strings.ToLower(strings.TrimSuffix(fields[i], ":443")) + host = strings.TrimSuffix(host, ":80") + if _, ok := hosts[host]; ok { + return true + } + } + return false + }, nil +} + +// systemLogLineFilter combines every reason to drop a line into one test. +func (s *Server) systemLogLineFilter(c *gin.Context, src systemLogSource) (func(string) bool, bool) { + filter := strings.ToLower(c.Query("filter")) + deployment := c.Query("deployment") + + var matchesDeployment func(string) bool + if deployment != "" { + if !src.ByDeployment { + c.JSON(http.StatusBadRequest, gin.H{"error": "this log source does not say which deployment a line belongs to"}) + return nil, false + } + if !s.requireDeploymentAccess(c, deployment, auth.AccessLevelRead) { + return nil, false + } + matcher, err := s.deploymentHostMatcher(deployment) + if err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "Deployment not found"}) + return nil, false + } + matchesDeployment = matcher + } + + return func(line string) bool { + if filter != "" && !strings.Contains(strings.ToLower(line), filter) { + return false + } + if matchesDeployment != nil && !matchesDeployment(line) { + return false + } + return true + }, true +} + +func (s *Server) getSystemLogs(c *gin.Context) { + if s.infraManager == nil { + c.JSON(http.StatusServiceUnavailable, gin.H{"error": "Infrastructure not available"}) + return + } + + src, ok := s.resolveSystemLogSource(c.Query("source")) + if !ok { + c.JSON(http.StatusBadRequest, gin.H{"error": "unknown log source"}) + return + } + + tail, err := strconv.Atoi(c.DefaultQuery("tail", "100")) + if err != nil { + tail = 100 + } + + keep, ok := s.systemLogLineFilter(c, src) + if !ok { + return + } + + logs, err := s.infraManager.ServiceLogs(src.Service, tail, src.Stream) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + + kept := make([]string, 0) + for _, line := range strings.Split(strings.TrimRight(logs, "\n"), "\n") { + if line == "" || !keep(line) { + continue + } + kept = append(kept, line) + } + text := strings.Join(kept, "\n") + + records := parseLogRecords(text) + for i := range records { + if records[i].Service == "" { + records[i].Service = src.Name + } + } + + c.JSON(http.StatusOK, gin.H{ + "source": src.ID, + "logs": text, + "records": records, + }) +} + +// streamSystemLogs follows a system log source over a websocket until the viewer goes away, +// the same way a deployment's logs are followed. +func (s *Server) streamSystemLogs(c *gin.Context) { + conn, err := upgrader.Upgrade(c.Writer, c.Request, nil) + if err != nil { + log.Printf("system log stream: websocket upgrade failed: %v", err) + return + } + defer conn.Close() + + // The browser cannot set headers on a websocket, so the token arrives as the first + // message, the same way the terminal does it. + if s.authMiddleware.IsAuthEnabled() { + _ = conn.SetReadDeadline(time.Now().Add(10 * time.Second)) + + _, message, err := conn.ReadMessage() + if err != nil { + sendError(conn, "Authentication timeout") + return + } + + var incoming authMessage + if err := json.Unmarshal(message, &incoming); err != nil || incoming.Type != "auth" { + sendError(conn, "Invalid auth message format") + return + } + + actor, err := s.authMiddleware.ActorForTokenString(incoming.Token, c.ClientIP()) + if err != nil { + sendError(conn, "Invalid or expired token") + return + } + c.Set(contextkeys.Actor, actor) + _ = conn.SetReadDeadline(time.Time{}) + } else { + c.Set(contextkeys.Actor, &auth.ActorContext{Type: "anonymous", Role: auth.RoleAdmin}) + } + + actor := auth.GetActorFromContext(c) + if actor == nil || !actor.HasPermission(auth.PermInfrastructureRead) { + sendError(conn, "Permission denied: infrastructure:read required") + return + } + + if s.authMiddleware.IsAuthEnabled() { + if err := conn.WriteMessage(websocket.TextMessage, []byte(`{"type":"auth_success"}`)); err != nil { + return + } + } + + if s.infraManager == nil { + sendError(conn, "Infrastructure not available") + return + } + + src, ok := s.resolveSystemLogSource(c.Query("source")) + if !ok { + sendError(conn, "Unknown log source") + return + } + + tail := 100 + if v := c.Query("tail"); v != "" { + if n, err := strconv.Atoi(v); err == nil && n >= 0 { + tail = n + } + } + + // A deployment's own domains decide which access lines it may see, so the same check the + // snapshot makes applies here; the socket carries no body to report it in. + deployment := c.Query("deployment") + var matchesDeployment func(string) bool + if deployment != "" { + if !src.ByDeployment { + sendError(conn, "This log source does not say which deployment a line belongs to") + return + } + if actor.Role != auth.RoleAdmin && !actor.CanAccessDeployment(deployment, auth.AccessLevelRead) { + sendError(conn, "No access to this deployment") + return + } + matcher, err := s.deploymentHostMatcher(deployment) + if err != nil { + sendError(conn, "Deployment not found") + return + } + matchesDeployment = matcher + } + + filter := strings.ToLower(c.Query("filter")) + + ctx := c.Request.Context() + + // Nothing is expected from the viewer, but reading is how a closed socket is noticed. + go func() { + for { + if _, _, err := conn.ReadMessage(); err != nil { + conn.Close() + return + } + } + }() + + sink := func(line string) { + if filter != "" && !strings.Contains(strings.ToLower(line), filter) { + return + } + if matchesDeployment != nil && !matchesDeployment(line) { + return + } + record := parseLogRecord(line) + if record.Service == "" { + record.Service = src.Name + } + payload, err := json.Marshal(logLine{Type: "log", Line: line, Record: record}) + if err != nil { + return + } + _ = conn.WriteMessage(websocket.TextMessage, payload) + } + + if err := s.infraManager.StreamServiceLogs(ctx, src.Service, tail, src.Stream, sink); err != nil && ctx.Err() == nil { + sendError(conn, err.Error()) + return + } + + _ = conn.WriteMessage(websocket.TextMessage, []byte(`{"type":"end"}`)) +} diff --git a/internal/api/system_logs_integration_test.go b/internal/api/system_logs_integration_test.go new file mode 100644 index 0000000..e75b966 --- /dev/null +++ b/internal/api/system_logs_integration_test.go @@ -0,0 +1,160 @@ +package api + +import ( + "os" + "os/exec" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/flatrun/agent/internal/docker" + "github.com/flatrun/agent/templates" +) + +// TestNginxAccessLineNamesItsDeployment runs the real base config in nginx, makes a real +// request through it, and matches the line it wrote. Reading the proxy's log per deployment +// only works if the host survives into the line, which is a property of nginx's own +// formatting rather than anything this package can assert on its own. +func TestNginxAccessLineNamesItsDeployment(t *testing.T) { + if testing.Short() { + t.Skip("starts a real container") + } + if _, err := exec.LookPath("docker"); err != nil { + t.Skip("docker unavailable") + } + if err := exec.Command("docker", "info").Run(); err != nil { + t.Skipf("docker daemon unreachable: %v", err) + } + + conf, err := templates.GetNginxConfigWithData(false, templates.NginxConfigData{}) + if err != nil { + t.Fatalf("rendering the base config failed: %v", err) + } + + confDir := t.TempDir() + if err := os.WriteFile(filepath.Join(confDir, "nginx.conf"), conf, 0644); err != nil { + t.Fatal(err) + } + siteDir := filepath.Join(confDir, "conf.d") + if err := os.MkdirAll(siteDir, 0755); err != nil { + t.Fatal(err) + } + // The base config includes a rate limit file and whatever vhosts exist, so both have to + // be there for nginx to start at all. + if err := os.WriteFile(filepath.Join(siteDir, "rate_limits.conf"), []byte("# none\n"), 0644); err != nil { + t.Fatal(err) + } + site := `server { + listen 80 default_server; + server_name _; + location / { return 200 "ok"; } +} +` + if err := os.WriteFile(filepath.Join(siteDir, "site.conf"), []byte(site), 0644); err != nil { + t.Fatal(err) + } + + const container = "flatrun-accesslog-integration" + _ = exec.Command("docker", "rm", "-f", container).Run() + run := exec.Command("docker", "run", "-d", "--name", container, + "-v", filepath.Join(confDir, "nginx.conf")+":/etc/nginx/nginx.conf:ro", + "-v", siteDir+":/etc/nginx/conf.d:ro", + "nginx:alpine") + if out, err := run.CombinedOutput(); err != nil { + t.Fatalf("starting nginx failed: %v: %s", err, out) + } + t.Cleanup(func() { + if out, err := exec.Command("docker", "rm", "-f", container).CombinedOutput(); err != nil { + t.Logf("cleanup: %v: %s", err, out) + } + }) + + // docker run returns as soon as the container is created, which is before nginx has + // finished its entrypoint and bound the port, so the request is retried until it answers. + const domain = "shop.example.test" + var reqOut []byte + var reqErr error + for attempt := 0; attempt < 30; attempt++ { + req := exec.Command("docker", "exec", container, + "wget", "-qO-", "--header", "Host: "+domain, "http://127.0.0.1/") + if reqOut, reqErr = req.CombinedOutput(); reqErr == nil { + break + } + time.Sleep(500 * time.Millisecond) + } + if reqErr != nil { + logs, _ := exec.Command("docker", "logs", container).CombinedOutput() + t.Fatalf("request through nginx failed: %v: %s (container: %s)", reqErr, reqOut, logs) + } + + // The access log is buffered and flushed on a timer, so the line is not there the moment + // the request returns. The agent reads these logs with docker's timestamps in front, so + // that is the shape the matcher has to cope with. + readAccessLine := func(args ...string) string { + var last []byte + for attempt := 0; attempt < 20; attempt++ { + out, err := exec.Command("docker", append([]string{"logs"}, append(args, container)...)...).CombinedOutput() + if err != nil { + t.Fatalf("reading nginx logs failed: %v: %s", err, out) + } + last = out + for _, line := range strings.Split(string(out), "\n") { + if strings.Contains(line, "GET /") { + return line + } + } + time.Sleep(time.Second) + } + t.Fatalf("nginx wrote no access line: %s", last) + return "" + } + + stamped := readAccessLine("--timestamps") + accessLine := readAccessLine() + + // A deployment serving that domain, as the agent reads it off disk. + base := t.TempDir() + dir := filepath.Join(base, "shop") + if err := os.MkdirAll(dir, 0755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(dir, "docker-compose.yml"), []byte("name: shop\nservices:\n web:\n image: nginx:alpine\n"), 0644); err != nil { + t.Fatal(err) + } + metadata := "domains:\n - domain: " + domain + "\n" + if err := os.WriteFile(filepath.Join(dir, "service.yml"), []byte(metadata), 0644); err != nil { + t.Fatal(err) + } + + server := &Server{manager: docker.NewManager(base)} + matches, err := server.deploymentHostMatcher("shop") + if err != nil { + t.Fatalf("building the matcher failed: %v", err) + } + if !matches(accessLine) { + t.Errorf("the deployment's own access line was filtered out: %q", accessLine) + } + if !matches(stamped) { + t.Errorf("the deployment's own access line was filtered out once timestamped: %q", stamped) + } + + // A second deployment on another domain must not see it. + other := filepath.Join(base, "blog") + if err := os.MkdirAll(other, 0755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(other, "docker-compose.yml"), []byte("name: blog\nservices:\n web:\n image: nginx:alpine\n"), 0644); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(other, "service.yml"), []byte("domains:\n - domain: blog.example.test\n"), 0644); err != nil { + t.Fatal(err) + } + blogMatches, err := server.deploymentHostMatcher("blog") + if err != nil { + t.Fatalf("building the matcher failed: %v", err) + } + if blogMatches(accessLine) || blogMatches(stamped) { + t.Errorf("another deployment's access line was let through: %q", accessLine) + } +} diff --git a/internal/api/system_logs_test.go b/internal/api/system_logs_test.go new file mode 100644 index 0000000..a6da43d --- /dev/null +++ b/internal/api/system_logs_test.go @@ -0,0 +1,113 @@ +package api + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "github.com/flatrun/agent/internal/infra" + "github.com/flatrun/agent/pkg/config" + "github.com/gin-gonic/gin" +) + +func systemLogsRouter(cfg *config.Config, register func(*gin.Engine, *Server)) *gin.Engine { + gin.SetMode(gin.TestMode) + server := &Server{infraManager: infra.NewManager(cfg)} + router := gin.New() + register(router, server) + return router +} + +// The proxy writes its access log to stdout and its error log to stderr, so the two are +// offered apart; only the access log carries the host a request asked for, which is the one +// thing that can be matched back to a deployment. +func TestSystemLogSourcesSplitNginxAccessFromError(t *testing.T) { + cfg := &config.Config{} + cfg.Nginx.Enabled = true + cfg.Nginx.ContainerName = "flatrun-nginx" + cfg.Infrastructure.Redis.Enabled = true + cfg.Infrastructure.Redis.Container = "flatrun-redis" + + router := systemLogsRouter(cfg, func(r *gin.Engine, s *Server) { + r.GET("/system/logs/sources", s.listSystemLogSources) + }) + + req := httptest.NewRequest(http.MethodGet, "/system/logs/sources", nil) + w := httptest.NewRecorder() + router.ServeHTTP(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String()) + } + + var resp struct { + Sources []systemLogSource `json:"sources"` + } + if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil { + t.Fatalf("failed to decode response: %v", err) + } + + byID := map[string]systemLogSource{} + for _, src := range resp.Sources { + byID[src.ID] = src + } + + access, ok := byID["nginx-access"] + if !ok { + t.Fatalf("no access log source offered: %+v", resp.Sources) + } + if access.Stream != infra.LogStreamStdout || !access.ByDeployment { + t.Errorf("the access log should be readable per deployment from stdout, got %+v", access) + } + + errSrc, ok := byID["nginx-error"] + if !ok { + t.Fatalf("no error log source offered: %+v", resp.Sources) + } + if errSrc.Stream != infra.LogStreamStderr || errSrc.ByDeployment { + t.Errorf("the error log says nothing about which deployment a line is from, got %+v", errSrc) + } + + if _, ok := byID["redis"]; !ok { + t.Errorf("shared infrastructure should be readable too: %+v", resp.Sources) + } +} + +// Asking to see one deployment's requests in a log that never records the host would hand +// back an arbitrary subset, so it is refused rather than silently ignored. +func TestSystemLogsRejectDeploymentFilterOnASourceThatCannotAnswerIt(t *testing.T) { + cfg := &config.Config{} + cfg.Nginx.Enabled = true + cfg.Nginx.ContainerName = "flatrun-nginx" + + router := systemLogsRouter(cfg, func(r *gin.Engine, s *Server) { + r.GET("/system/logs", s.getSystemLogs) + }) + + req := httptest.NewRequest(http.MethodGet, "/system/logs?source=nginx-error&deployment=shop", nil) + w := httptest.NewRecorder() + router.ServeHTTP(w, req) + + if w.Code != http.StatusBadRequest { + t.Fatalf("expected 400, got %d: %s", w.Code, w.Body.String()) + } +} + +func TestSystemLogsRejectUnknownSource(t *testing.T) { + cfg := &config.Config{} + cfg.Nginx.Enabled = true + cfg.Nginx.ContainerName = "flatrun-nginx" + + router := systemLogsRouter(cfg, func(r *gin.Engine, s *Server) { + r.GET("/system/logs", s.getSystemLogs) + }) + + req := httptest.NewRequest(http.MethodGet, "/system/logs?source=not-a-source", nil) + w := httptest.NewRecorder() + router.ServeHTTP(w, req) + + if w.Code != http.StatusBadRequest { + t.Fatalf("expected 400, got %d: %s", w.Code, w.Body.String()) + } +} diff --git a/internal/auth/models.go b/internal/auth/models.go index 13b874b..7ae4be4 100644 --- a/internal/auth/models.go +++ b/internal/auth/models.go @@ -21,7 +21,6 @@ const ( AccessLevelAdmin = "admin" ) - func ValidAccessLevel(level string) bool { return level == AccessLevelRead || level == AccessLevelWrite || level == AccessLevelAdmin } @@ -88,21 +87,21 @@ func (d *DeploymentAccess) UnmarshalJSON(data []byte) error { } type APIKey struct { - ID int64 `json:"id"` - KeyID string `json:"key_id"` - UserID int64 `json:"user_id"` - Name string `json:"name"` - Description string `json:"description,omitempty"` - KeyHash string `json:"-"` - KeyPrefix string `json:"key_prefix"` - Role Role `json:"role,omitempty"` - Permissions []string `json:"permissions,omitempty"` - Deployments DeploymentAccess `json:"deployments,omitempty"` - ExpiresAt time.Time `json:"expires_at,omitempty"` - LastUsedAt time.Time `json:"last_used_at,omitempty"` - LastUsedIP string `json:"last_used_ip,omitempty"` - IsActive bool `json:"is_active"` - CreatedAt time.Time `json:"created_at"` + ID int64 `json:"id"` + KeyID string `json:"key_id"` + UserID int64 `json:"user_id"` + Name string `json:"name"` + Description string `json:"description,omitempty"` + KeyHash string `json:"-"` + KeyPrefix string `json:"key_prefix"` + Role Role `json:"role,omitempty"` + Permissions []string `json:"permissions,omitempty"` + Deployments DeploymentAccess `json:"deployments,omitempty"` + ExpiresAt time.Time `json:"expires_at,omitempty"` + LastUsedAt time.Time `json:"last_used_at,omitempty"` + LastUsedIP string `json:"last_used_ip,omitempty"` + IsActive bool `json:"is_active"` + CreatedAt time.Time `json:"created_at"` } type Session struct { @@ -127,13 +126,13 @@ type UserDeployment struct { } type ActorContext struct { - Type string `json:"type"` - UserID int64 `json:"user_id,omitempty"` - User *User `json:"user,omitempty"` - APIKey *APIKey `json:"api_key,omitempty"` - Role Role `json:"role"` - Permissions []string `json:"permissions,omitempty"` - Deployments map[string]string `json:"deployments,omitempty"` + Type string `json:"type"` + UserID int64 `json:"user_id,omitempty"` + User *User `json:"user,omitempty"` + APIKey *APIKey `json:"api_key,omitempty"` + Role Role `json:"role"` + Permissions []string `json:"permissions,omitempty"` + Deployments map[string]string `json:"deployments,omitempty"` } func (a *ActorContext) HasPermission(p Permission) bool { diff --git a/internal/docker/api.go b/internal/docker/api.go index d28cd6e..a26f571 100644 --- a/internal/docker/api.go +++ b/internal/docker/api.go @@ -58,6 +58,16 @@ func (a *APIClient) FindContainer(ctx context.Context, project, service string) return containers[0].ID, nil } +// ContainerLogPath is the file Docker keeps a container's output in, or "" when its logging +// driver keeps it somewhere Docker does not own, such as journald or a remote collector. +func (a *APIClient) ContainerLogPath(ctx context.Context, ref string) (string, error) { + info, err := a.cli.ContainerInspect(ctx, ref) + if err != nil { + return "", err + } + return info.LogPath, nil +} + // ContainerPrimaryIP returns the IP address of a project's first running // container on the given docker network. The agent runs on the host, so a // service that only exposes ports on an internal compose network (a self-hosted diff --git a/internal/docker/compose.go b/internal/docker/compose.go index 96ef16e..2a24dff 100644 --- a/internal/docker/compose.go +++ b/internal/docker/compose.go @@ -192,8 +192,11 @@ func (c *ComposeExecutor) PullService(deploymentPath, service string, opts ...Ru return c.runCompose(deploymentPath, opts, "pull", "--ignore-buildable", "--policy", "always", service) } -func (c *ComposeExecutor) Logs(deploymentPath string, tail int) (string, error) { - return c.runCompose(deploymentPath, nil, "logs", "--no-color", "--timestamps", "--tail", tailArg(tail)) +// Logs returns a tail of the deployment's output. Naming services narrows it to those +// containers; naming none returns every service, which is what compose does by default. +func (c *ComposeExecutor) Logs(deploymentPath string, tail int, services ...string) (string, error) { + args := append([]string{"logs", "--no-color", "--timestamps", "--tail", tailArg(tail)}, services...) + return c.runCompose(deploymentPath, nil, args...) } // tailArg is the compose --tail value: a count, or "all" for tail <= 0, since "0" shows no lines. @@ -461,8 +464,9 @@ func (c *ComposeExecutor) runCompose(deploymentPath string, opts []RunOption, ar // a user watching a container start reloads to see the next line. Following gives them the // line when the container writes it, and cancelling ctx stops the process rather than // leaving it attached for the life of the agent. -func (c *ComposeExecutor) StreamLogs(ctx context.Context, deploymentPath string, tail int, sink func(string)) error { - cmd, err := c.composeCommand(ctx, deploymentPath, "logs", "--follow", "--no-color", "--timestamps", "--tail", tailArg(tail)) +func (c *ComposeExecutor) StreamLogs(ctx context.Context, deploymentPath string, tail int, sink func(string), services ...string) error { + args := append([]string{"logs", "--follow", "--no-color", "--timestamps", "--tail", tailArg(tail)}, services...) + cmd, err := c.composeCommand(ctx, deploymentPath, args...) if err != nil { return err } diff --git a/internal/docker/log_service_integration_test.go b/internal/docker/log_service_integration_test.go new file mode 100644 index 0000000..4cfb429 --- /dev/null +++ b/internal/docker/log_service_integration_test.go @@ -0,0 +1,75 @@ +package docker + +import ( + "context" + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +// TestLogsNarrowToOneService is the point of the service filter: a deployment with two noisy +// containers hands back only the one the viewer picked. +func TestLogsNarrowToOneService(t *testing.T) { + if testing.Short() { + t.Skip("starts real containers") + } + + const name = "flatrun-logservice-integration" + base := t.TempDir() + dir := filepath.Join(base, name) + if err := os.MkdirAll(dir, 0755); err != nil { + t.Fatal(err) + } + + compose := "name: " + name + ` +services: + web: + image: alpine:latest + entrypoint: ["/bin/sh", "-c", "echo web-speaking; sleep 300"] + worker: + image: alpine:latest + entrypoint: ["/bin/sh", "-c", "echo worker-speaking; sleep 300"] +` + if err := os.WriteFile(filepath.Join(dir, "docker-compose.yml"), []byte(compose), 0644); err != nil { + t.Fatal(err) + } + + m := NewManager(base) + if m.apiClient == nil { + t.Skip("docker api client unavailable") + } + probe, cancelProbe := context.WithTimeout(context.Background(), 30*time.Second) + defer cancelProbe() + if _, err := m.apiClient.ListLiveComposeContainers(probe); err != nil { + t.Skipf("docker daemon unreachable: %v", err) + } + t.Cleanup(func() { + if _, err := m.executor.Down(dir); err != nil { + t.Logf("cleanup: %v", err) + } + }) + if out, err := m.StartDeployment(name); err != nil { + t.Fatalf("start failed: %v (%s)", err, out) + } + + all, err := m.GetDeploymentLogs(name, 100) + if err != nil { + t.Fatalf("reading every service failed: %v", err) + } + if !strings.Contains(all, "web-speaking") || !strings.Contains(all, "worker-speaking") { + t.Fatalf("expected both services without a filter, got: %s", all) + } + + only, err := m.GetDeploymentLogs(name, 100, "worker") + if err != nil { + t.Fatalf("reading one service failed: %v", err) + } + if !strings.Contains(only, "worker-speaking") { + t.Errorf("expected the picked service's output, got: %s", only) + } + if strings.Contains(only, "web-speaking") { + t.Errorf("expected no output from the other service, got: %s", only) + } +} diff --git a/internal/docker/manager.go b/internal/docker/manager.go index 579a00c..29692bf 100644 --- a/internal/docker/manager.go +++ b/internal/docker/manager.go @@ -54,6 +54,16 @@ func (m *Manager) indexContainersByProject(ctx context.Context) (containerIndex, return index, nil } +// ContainerLogPath is the file Docker keeps a container's output in, by container id or name. +func (m *Manager) ContainerLogPath(ref string) (string, error) { + if m.apiClient == nil { + return "", fmt.Errorf("docker api client unavailable") + } + ctx, cancel := context.WithTimeout(context.Background(), statusReadTimeout) + defer cancel() + return m.apiClient.ContainerLogPath(ctx, ref) +} + // ContainerPrimaryIP returns the first running container's address for a // deployment on the given docker network. A flatrun deploy names its compose // project after the deployment, so the project name is the deployment name. @@ -275,8 +285,8 @@ func (m *Manager) ListDeployments() ([]models.Deployment, error) { // The deployment path is passed in rather than looked up again: the caller has already read // the deployment, and following holds for as long as someone is watching, which is far too // long to hold the manager's lock. -func (m *Manager) StreamDeploymentLogs(ctx context.Context, name, path string, tail int, sink func(string)) error { - return m.executor.StreamLogs(ctx, path, tail, sink) +func (m *Manager) StreamDeploymentLogs(ctx context.Context, name, path string, tail int, sink func(string), services ...string) error { + return m.executor.StreamLogs(ctx, path, tail, sink, services...) } // FindDeployments returns deployments built from their on-disk metadata alone, @@ -856,7 +866,7 @@ func (m *Manager) ComposeExec(ctx context.Context, name string, service string, return m.apiClient.ExecInService(ctx, project, service, command) } -func (m *Manager) GetDeploymentLogs(name string, tail int) (string, error) { +func (m *Manager) GetDeploymentLogs(name string, tail int, services ...string) (string, error) { m.mu.RLock() deployment, err := m.discovery.GetDeployment(name) m.mu.RUnlock() @@ -865,7 +875,7 @@ func (m *Manager) GetDeploymentLogs(name string, tail int) (string, error) { return "", err } - return m.executor.Logs(deployment.Path, tail) + return m.executor.Logs(deployment.Path, tail, services...) } func (m *Manager) UpdateDeployment(name string, composeContent string) error { diff --git a/internal/docker/manager_test.go b/internal/docker/manager_test.go index 339dac1..fe6ba8a 100644 --- a/internal/docker/manager_test.go +++ b/internal/docker/manager_test.go @@ -407,10 +407,10 @@ func TestResolveServiceNotFound(t *testing.T) { func TestGetComposeServiceNames(t *testing.T) { tests := []struct { - name string - compose string - wantNames []string - wantErr bool + name string + compose string + wantNames []string + wantErr bool }{ { name: "single service", diff --git a/internal/docker/resources.go b/internal/docker/resources.go index fb5b890..69fd428 100644 --- a/internal/docker/resources.go +++ b/internal/docker/resources.go @@ -24,10 +24,10 @@ type ResourceUpdate struct { } type hostConfig struct { - Memory int64 `json:"Memory"` - MemorySwap int64 `json:"MemorySwap"` - NanoCpus int64 `json:"NanoCpus"` - CpuShares int64 `json:"CpuShares"` + Memory int64 `json:"Memory"` + MemorySwap int64 `json:"MemorySwap"` + NanoCpus int64 `json:"NanoCpus"` + CpuShares int64 `json:"CpuShares"` RestartPolicy restartPolicyInspect `json:"RestartPolicy"` } diff --git a/internal/infra/log_stream_integration_test.go b/internal/infra/log_stream_integration_test.go new file mode 100644 index 0000000..3163474 --- /dev/null +++ b/internal/infra/log_stream_integration_test.go @@ -0,0 +1,130 @@ +package infra + +import ( + "context" + "os/exec" + "strings" + "sync" + "testing" + "time" + + "github.com/flatrun/agent/pkg/config" +) + +func startTalkingContainer(t *testing.T, name, command string) { + t.Helper() + + if _, err := exec.LookPath("docker"); err != nil { + t.Skip("docker unavailable") + } + if err := exec.Command("docker", "info").Run(); err != nil { + t.Skipf("docker daemon unreachable: %v", err) + } + + _ = exec.Command("docker", "rm", "-f", name).Run() + out, err := exec.Command("docker", "run", "-d", "--name", name, "alpine:latest", "/bin/sh", "-c", command).CombinedOutput() + if err != nil { + t.Fatalf("starting the container failed: %v: %s", err, out) + } + t.Cleanup(func() { + if out, err := exec.Command("docker", "rm", "-f", name).CombinedOutput(); err != nil { + t.Logf("cleanup: %v: %s", err, out) + } + }) +} + +// Splitting a container's two outputs is what lets the proxy's access log and error log be +// read apart, so it has to hold against a real container rather than in principle. +func TestServiceLogsReadEachOutputApart(t *testing.T) { + if testing.Short() { + t.Skip("starts a real container") + } + + const name = "flatrun-infra-logsplit" + startTalkingContainer(t, name, "echo an-access-line; echo an-error-line >&2; sleep 60") + + m := NewManager(&config.Config{}) + + var all string + for attempt := 0; attempt < 15; attempt++ { + var err error + all, err = m.ServiceLogs(name, 100, LogStreamAll) + if err != nil { + t.Fatalf("reading logs failed: %v", err) + } + if strings.Contains(all, "an-access-line") && strings.Contains(all, "an-error-line") { + break + } + time.Sleep(time.Second) + } + if !strings.Contains(all, "an-access-line") || !strings.Contains(all, "an-error-line") { + t.Fatalf("expected both outputs together, got: %s", all) + } + + stdout, err := m.ServiceLogs(name, 100, LogStreamStdout) + if err != nil { + t.Fatalf("reading stdout failed: %v", err) + } + if !strings.Contains(stdout, "an-access-line") || strings.Contains(stdout, "an-error-line") { + t.Errorf("stdout should carry only what the container printed there, got: %s", stdout) + } + + stderr, err := m.ServiceLogs(name, 100, LogStreamStderr) + if err != nil { + t.Fatalf("reading stderr failed: %v", err) + } + if !strings.Contains(stderr, "an-error-line") || strings.Contains(stderr, "an-access-line") { + t.Errorf("stderr should carry only what the container printed there, got: %s", stderr) + } +} + +// Following one output has to keep delivering, which is what the viewer's Follow depends on. +func TestStreamServiceLogsFollowsOneOutput(t *testing.T) { + if testing.Short() { + t.Skip("starts a real container") + } + + const name = "flatrun-infra-logfollow" + startTalkingContainer(t, name, "i=0; while true; do i=$((i+1)); echo out-$i; echo err-$i >&2; sleep 1; done") + + m := NewManager(&config.Config{}) + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + var ( + mu sync.Mutex + lines []string + ) + got := make(chan struct{}) + var once sync.Once + + go func() { + _ = m.StreamServiceLogs(ctx, name, 10, LogStreamStdout, func(line string) { + mu.Lock() + lines = append(lines, line) + n := len(lines) + mu.Unlock() + // Wait for a few, so this cannot pass on the backlog alone. + if n >= 3 { + once.Do(func() { close(got) }) + } + }) + }() + + select { + case <-got: + case <-ctx.Done(): + mu.Lock() + defer mu.Unlock() + t.Fatalf("only received %d lines while following: %v", len(lines), lines) + } + + mu.Lock() + defer mu.Unlock() + for _, line := range lines { + if strings.Contains(line, "err-") { + t.Errorf("following stdout delivered a line from the other output: %q", line) + } + } +} diff --git a/internal/infra/manager.go b/internal/infra/manager.go index a8e7b0d..f951dcb 100644 --- a/internal/infra/manager.go +++ b/internal/infra/manager.go @@ -1,9 +1,12 @@ package infra import ( + "bufio" "bytes" + "context" "encoding/json" "fmt" + "io" "os" "os/exec" "path/filepath" @@ -113,24 +116,117 @@ func (m *Manager) RestartService(name string) error { return nil } +// A container writes two separate outputs and nginx uses both: the access log goes to stdout +// and the error log to stderr, so the two are readable apart without configuring a file. +const ( + LogStreamAll = "all" + LogStreamStdout = "stdout" + LogStreamStderr = "stderr" +) + func (m *Manager) GetServiceLogs(name string, tail int) (string, error) { + return m.ServiceLogs(name, tail, LogStreamAll) +} + +func (m *Manager) ServiceLogs(name string, tail int, stream string) (string, error) { containerName := m.resolveContainerName(name) if containerName == "" { return "", fmt.Errorf("unknown service: %s", name) } - args := []string{"logs"} - if tail > 0 { - args = append(args, "--tail", fmt.Sprintf("%d", tail)) + cmd := exec.Command("docker", dockerLogArgs(containerName, tail, false)...) + var out, errOut bytes.Buffer + switch stream { + case LogStreamStdout, LogStreamStderr: + cmd.Stdout, cmd.Stderr = &out, &errOut + default: + // One buffer for both keeps the two outputs in the order they were written. + cmd.Stdout, cmd.Stderr = &out, &out } - args = append(args, containerName) - cmd := exec.Command("docker", args...) - output, err := cmd.CombinedOutput() + if err := cmd.Run(); err != nil { + return "", fmt.Errorf("failed to get logs for %s: %w: %s", name, err, strings.TrimSpace(errOut.String()+out.String())) + } + + if stream == LogStreamStderr { + return errOut.String(), nil + } + return out.String(), nil +} + +// StreamServiceLogs follows an infrastructure container's output until ctx is done, handing +// each line to sink as it is written. +func (m *Manager) StreamServiceLogs(ctx context.Context, name string, tail int, stream string, sink func(string)) error { + containerName := m.resolveContainerName(name) + if containerName == "" { + return fmt.Errorf("unknown service: %s", name) + } + + cmd := exec.CommandContext(ctx, "docker", dockerLogArgs(containerName, tail, true)...) + stdout, err := cmd.StdoutPipe() + if err != nil { + return err + } + stderr, err := cmd.StderrPipe() if err != nil { - return "", fmt.Errorf("failed to get logs for %s: %w", name, err) + return err } - return string(output), nil + if err := cmd.Start(); err != nil { + return err + } + + var wg sync.WaitGroup + var mu sync.Mutex + follow := func(r io.Reader) { + defer wg.Done() + scanner := bufio.NewScanner(r) + scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024) + for scanner.Scan() { + line := scanner.Text() + // Both outputs feed one viewer, so a line is handed over whole. + mu.Lock() + sink(line) + mu.Unlock() + } + } + + if stream != LogStreamStderr { + wg.Add(1) + go follow(stdout) + } else { + go io.Copy(io.Discard, stdout) + } + if stream != LogStreamStdout { + wg.Add(1) + go follow(stderr) + } else { + go io.Copy(io.Discard, stderr) + } + wg.Wait() + + // A cancelled follow is the viewer leaving, not a failure. + if err := cmd.Wait(); err != nil && ctx.Err() == nil { + return err + } + return nil +} + +func dockerLogArgs(containerName string, tail int, follow bool) []string { + args := []string{"logs", "--timestamps"} + if follow { + args = append(args, "--follow") + } + if tail > 0 { + args = append(args, "--tail", fmt.Sprintf("%d", tail)) + } + return append(args, containerName) +} + +// ContainerName is the container behind an infrastructure service, or "" if there is none. +func (m *Manager) ContainerName(name string) string { + m.mu.RLock() + defer m.mu.RUnlock() + return m.resolveContainerName(name) } func (m *Manager) resolveContainerName(name string) string { diff --git a/internal/observ/api.go b/internal/observ/api.go index 6f7a3d1..6e5f2b4 100644 --- a/internal/observ/api.go +++ b/internal/observ/api.go @@ -39,10 +39,24 @@ type alertPersistence interface { Save([]AlertRule) error } +// logRuleAccess is the slice of the log engine the API needs; nil-safe. +type logRuleAccess interface { + Rules() []LogRule + SetRules([]LogRule) + Incidents() []Incident +} + +type logRulePersistence interface { + Load() []LogRule + Save([]LogRule) error +} + // alerts is optional wiring for the rule endpoints. type alerts struct { - engine alertAccess - store alertPersistence + engine alertAccess + store alertPersistence + logEngine logRuleAccess + logStore logRulePersistence } func Handler(store *Store, history *MetricsDB, health healthReporter, cfg configAccess, apply func(Config)) http.Handler { @@ -151,6 +165,46 @@ func HandlerWithAlerts(store *Store, history *MetricsDB, health healthReporter, } writeJSON(w, al.engine.Rules()) }) + mux.HandleFunc("/alerts/log-rules", func(w http.ResponseWriter, r *http.Request) { + if al.logEngine == nil || al.logStore == nil { + writeJSON(w, []LogRule{}) + return + } + if r.Method == http.MethodPut { + var incoming []LogRule + if err := json.NewDecoder(r.Body).Decode(&incoming); err != nil { + http.Error(w, "invalid rules", http.StatusBadRequest) + return + } + // Saving assigns ids and rejects a rule that would match everything. + if err := al.logStore.Save(incoming); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + al.logEngine.SetRules(al.logStore.Load()) + } + writeJSON(w, al.logEngine.Rules()) + }) + mux.HandleFunc("/alerts/incidents", func(w http.ResponseWriter, r *http.Request) { + if al.logEngine == nil { + writeJSON(w, []Incident{}) + return + } + incidents := al.logEngine.Incidents() + if deployment := r.URL.Query().Get("deployment"); deployment != "" { + filtered := make([]Incident, 0, len(incidents)) + for _, in := range incidents { + if in.Deployment == deployment { + filtered = append(filtered, in) + } + } + incidents = filtered + } + writeJSON(w, incidents) + }) + mux.HandleFunc("/alerts/responders", func(w http.ResponseWriter, _ *http.Request) { + writeJSON(w, KnownResponders()) + }) mux.HandleFunc("/alerts/firing", func(w http.ResponseWriter, _ *http.Request) { if al.engine == nil { writeJSON(w, []AlertEvent{}) diff --git a/internal/observ/config.go b/internal/observ/config.go index 08984b3..2805aed 100644 --- a/internal/observ/config.go +++ b/internal/observ/config.go @@ -23,6 +23,20 @@ type Config struct { // OTEL_EXPORTER_OTLP_ENDPOINT environment variable is honoured instead, and with // neither set nothing is pushed and the metrics are still there to scrape. OTLPEndpoint string `yaml:"otlp_endpoint,omitempty" json:"otlp_endpoint,omitempty"` + // Off unless turned on here, and still opt-in per rule after that. + LogTriage bool `yaml:"log_triage" json:"log_triage"` + // Bounds what an incident carries, and so the most a triage can be asked to read. + TriageContextLines int `yaml:"triage_context_lines,omitempty" json:"triage_context_lines,omitempty"` +} + +func (c Config) triageContextLines() int { + if c.TriageContextLines <= 0 { + return 12 + } + if c.TriageContextLines > maxLogContextLines { + return maxLogContextLines + } + return c.TriageContextLines } // DefaultConfig returns the built-in defaults. @@ -93,4 +107,6 @@ var ConfigSchema = map[string]any{ "restart_cooldown_seconds": map[string]any{"type": "number", "label": "Restart cooldown (seconds)", "default": 120, "min": 10}, "retention_days": map[string]any{"type": "number", "label": "Keep history for (days)", "default": 7, "min": 1}, "otlp_endpoint": map[string]any{"type": "string", "label": "OTLP endpoint", "placeholder": "http://localhost:4318", "help": "Push metrics to an OpenTelemetry backend. Leave empty to only serve them for scraping."}, + "log_triage": map[string]any{"type": "boolean", "label": "Let log rules ask the assistant", "default": false, "help": "Log rules that opt in can have the assistant explain an incident. Bounded by the agent's daily triage cap."}, + "triage_context_lines": map[string]any{"type": "number", "label": "Lines of context per incident", "default": 12, "min": 1, "max": 40}, } diff --git a/internal/observ/fingerprint.go b/internal/observ/fingerprint.go new file mode 100644 index 0000000..fa5f00b --- /dev/null +++ b/internal/observ/fingerprint.go @@ -0,0 +1,53 @@ +package observ + +import ( + "crypto/sha256" + "encoding/hex" + "regexp" + "strings" +) + +var ( + fpHex = regexp.MustCompile(`\b(?:0x)?[0-9a-fA-F]{8,}\b`) + fpUUID = regexp.MustCompile(`\b[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}\b`) + // Case-insensitive: normalization lowercases first, turning the ISO "T" into a "t". + fpTimestamp = regexp.MustCompile(`(?i)\b\d{4}-\d{2}-\d{2}[t ]\d{2}:\d{2}:\d{2}(?:\.\d+)?z?\b`) + fpDuration = regexp.MustCompile(`\b\d+(?:\.\d+)?(?:ms|s|m|h|us|µs|ns)\b`) + fpIPPort = regexp.MustCompile(`\b\d{1,3}(?:\.\d{1,3}){3}(?::\d+)?\b`) + fpQuoted = regexp.MustCompile(`"[^"]*"|'[^']*'`) + fpPath = regexp.MustCompile(`(?:/[\w.\-@+]+){2,}`) + fpNumber = regexp.MustCompile(`\b\d+\b`) + fpSpace = regexp.MustCompile(`\s+`) +) + +// fingerprint reduces a message to what stays the same across occurrences of one fault. +// +// It errs toward collapsing: two faults sharing a fingerprint costs one missed notification, +// while one fault spread across many costs a notification and a triage per line written. +func fingerprint(message string) string { + normalized := normalizeMessage(message) + sum := sha256.Sum256([]byte(normalized)) + return hex.EncodeToString(sum[:8]) +} + +func normalizeMessage(message string) string { + s := strings.ToLower(message) + // Specific shapes first, or the hex rule eats UUIDs and the number rule eats timestamps. + s = fpUUID.ReplaceAllString(s, "") + s = fpTimestamp.ReplaceAllString(s, "