From 58e4ae66b75d5aaf3918346cb51ec397cdd38df6 Mon Sep 17 00:00:00 2001 From: nfebe Date: Mon, 10 Aug 2026 20:32:25 +0100 Subject: [PATCH 01/10] chore: Apply gofmt to misaligned struct fields Formatting only. No behaviour changes. --- internal/auth/models.go | 45 ++++++++++++++++++------------------ internal/docker/resources.go | 8 +++---- internal/system/network.go | 8 +++---- 3 files changed, 30 insertions(+), 31 deletions(-) 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/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/system/network.go b/internal/system/network.go index bd312eb..29f819a 100644 --- a/internal/system/network.go +++ b/internal/system/network.go @@ -12,14 +12,14 @@ import ( ) type NetworkHealth struct { - ExternalAccess bool `json:"external_access"` - DNS DNSHealth `json:"dns"` + ExternalAccess bool `json:"external_access"` + DNS DNSHealth `json:"dns"` Interfaces []NetworkInterface `json:"interfaces"` - CheckedAt time.Time `json:"checked_at"` + CheckedAt time.Time `json:"checked_at"` } type DNSHealth struct { - Healthy bool `json:"healthy"` + Healthy bool `json:"healthy"` Resolvers []ResolverCheck `json:"resolvers"` } From db270846dd989e9d1200b0f220e3321290d6596b Mon Sep 17 00:00:00 2001 From: nfebe Date: Mon, 10 Aug 2026 20:32:51 +0100 Subject: [PATCH 02/10] fix(logs): Restore filtering a deployment's logs by service Reading or following a deployment's logs can again be narrowed to a single container instead of always returning every service interleaved. The requested name is checked against the deployment's own compose services first, so an unknown one is rejected with the list of what is available rather than reaching the container runtime. A file log source stays whole, since it is one file the deployment writes rather than per service output. --- internal/api/log_service_filter_test.go | 111 ++++++++++++++++++ internal/api/log_stream.go | 10 +- internal/api/logfile.go | 19 +++ internal/api/server.go | 11 +- internal/docker/compose.go | 12 +- .../docker/log_service_integration_test.go | 75 ++++++++++++ internal/docker/manager.go | 8 +- internal/docker/manager_test.go | 8 +- 8 files changed, 240 insertions(+), 14 deletions(-) create mode 100644 internal/api/log_service_filter_test.go create mode 100644 internal/docker/log_service_integration_test.go 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/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..9e7b9c3 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -2505,6 +2505,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 +2526,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 +2538,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/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..7e46362 100644 --- a/internal/docker/manager.go +++ b/internal/docker/manager.go @@ -275,8 +275,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 +856,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 +865,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", From cf188b981cf67b6455faa80775043bff3721e496 Mon Sep 17 00:00:00 2001 From: nfebe Date: Mon, 10 Aug 2026 20:33:08 +0100 Subject: [PATCH 03/10] feat(logs): Expose the proxy's own logs, readable per deployment The reverse proxy's access and error logs are now readable and followable from the panel, so a request that never reached a deployment is visible instead of being invisible everywhere. Access lines now start with the host they were served for, which makes the shared log filterable down to a single deployment by matching its domains. Access and error are offered separately because they are separate outputs, and reading them needs infrastructure access rather than access to any one deployment. --- internal/api/server.go | 3 + internal/api/system_logs.go | 337 ++++++++++++++++++ internal/api/system_logs_integration_test.go | 150 ++++++++ internal/api/system_logs_test.go | 113 ++++++ internal/infra/log_stream_integration_test.go | 130 +++++++ internal/infra/manager.go | 112 +++++- templates/infra/nginx/nginx.conf | 4 +- templates/infra/nginx/nginx.lua.conf | 4 +- 8 files changed, 843 insertions(+), 10 deletions(-) create mode 100644 internal/api/system_logs.go create mode 100644 internal/api/system_logs_integration_test.go create mode 100644 internal/api/system_logs_test.go create mode 100644 internal/infra/log_stream_integration_test.go diff --git a/internal/api/server.go b/internal/api/server.go index 9e7b9c3..c194d75 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -426,6 +426,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") @@ -678,6 +679,8 @@ 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.POST("/infrastructure/migrate/:name", s.authMiddleware.RequirePermission(auth.PermInfrastructureWrite), s.migrateToInfrastructure) // Registry endpoints 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..91881d0 --- /dev/null +++ b/internal/api/system_logs_integration_test.go @@ -0,0 +1,150 @@ +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) + } + }) + + const domain = "shop.example.test" + req := exec.Command("docker", "exec", container, + "wget", "-qO-", "--header", "Host: "+domain, "http://127.0.0.1/") + if out, err := req.CombinedOutput(); err != nil { + logs, _ := exec.Command("docker", "logs", container).CombinedOutput() + t.Fatalf("request through nginx failed: %v: %s (container: %s)", err, out, 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/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/templates/infra/nginx/nginx.conf b/templates/infra/nginx/nginx.conf index 8045be0..b91d7a5 100644 --- a/templates/infra/nginx/nginx.conf +++ b/templates/infra/nginx/nginx.conf @@ -10,7 +10,9 @@ http { include mime.types; default_type application/octet-stream; - log_format main '$remote_addr - $remote_user [$time_local] "$request" ' + # The host leads every line so an access line says which site it belongs to; one proxy + # serves every deployment, and without it the log cannot be read per deployment. + log_format main '$host $remote_addr - $remote_user [$time_local] "$request" ' '$status $body_bytes_sent "$http_referer" ' '"$http_user_agent" "$http_x_forwarded_for"'; diff --git a/templates/infra/nginx/nginx.lua.conf b/templates/infra/nginx/nginx.lua.conf index 5642ea3..b786c73 100644 --- a/templates/infra/nginx/nginx.lua.conf +++ b/templates/infra/nginx/nginx.lua.conf @@ -10,7 +10,9 @@ http { include mime.types; default_type application/octet-stream; - log_format main '$remote_addr - $remote_user [$time_local] "$request" ' + # The host leads every line so an access line says which site it belongs to; one proxy + # serves every deployment, and without it the log cannot be read per deployment. + log_format main '$host $remote_addr - $remote_user [$time_local] "$request" ' '$status $body_bytes_sent "$http_referer" ' '"$http_user_agent" "$http_x_forwarded_for"'; From 1e62e180ac0c2259ad73cca23722464c48996169 Mon Sep 17 00:00:00 2001 From: nfebe Date: Mon, 10 Aug 2026 20:33:16 +0100 Subject: [PATCH 04/10] feat(logs): Delete a log from the panel A log can now be emptied from the panel, which frees the disk it was using. A file source is emptied in place so the application keeps writing to the same file, and container output is emptied where the container runtime stores it. A source the runtime does not keep on disk is refused with the reason rather than appearing to succeed. Emptying a deployment's log needs write access to that deployment, and emptying a system log needs infrastructure write access, since neither is recoverable. --- internal/api/log_truncate.go | 142 ++++++++++++++++++++++++++++++ internal/api/log_truncate_test.go | 91 +++++++++++++++++++ internal/api/server.go | 2 + 3 files changed, 235 insertions(+) create mode 100644 internal/api/log_truncate.go create mode 100644 internal/api/log_truncate_test.go diff --git a/internal/api/log_truncate.go b/internal/api/log_truncate.go new file mode 100644 index 0000000..6b181e9 --- /dev/null +++ b/internal/api/log_truncate.go @@ -0,0 +1,142 @@ +package api + +import ( + "fmt" + "net/http" + "os" + "os/exec" + "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 truncateContainerLog(container string) error { + if strings.TrimSpace(container) == "" { + return fmt.Errorf("no container to clear") + } + + out, err := exec.Command("docker", "inspect", "--format", "{{.LogPath}}", container).Output() + if err != nil { + return fmt.Errorf("could not find the log for %s: %w", container, err) + } + + path := strings.TrimSpace(string(out)) + 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 + } + + 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 + } + + wanted := c.Query("service") + var failures []string + cleared := 0 + for _, svc := range services { + if wanted != "" && wanted != "all" && svc.Name != wanted { + continue + } + if svc.ContainerID == "" { + continue + } + if truncErr := 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 := 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..09769a8 --- /dev/null +++ b/internal/api/log_truncate_test.go @@ -0,0 +1,91 @@ +package api + +import ( + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "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") + } +} + +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/server.go b/internal/api/server.go index c194d75..dbca888 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -470,6 +470,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) @@ -681,6 +682,7 @@ func (s *Server) setupRoutes() { 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 From da86e183db8f254122ba6ebbe87c438908b5e546 Mon Sep 17 00:00:00 2001 From: nfebe Date: Mon, 10 Aug 2026 20:33:52 +0100 Subject: [PATCH 05/10] feat(observability): Watch logs against rules and triage what fires Logs can now be watched for patterns that matter, so a repeated error raises an incident on its own instead of waiting for someone to be looking at the right screen at the right moment. A rule sets the level and pattern that count, how many times within how long, and how long to stay quiet after firing. Only an incident reaches the assistant, and only once for each distinct fault: identical lines are collapsed by shape before anything is counted, the context handed over is capped, and a daily ceiling bounds the spend. Incidents carry a short explanation and suggested next step where the assistant is configured, and the plain incident where it is not. Acting on an incident goes through a registry, so notifying is the first responder rather than the only possible one. --- config.example.yml | 3 + internal/api/internal_logs.go | 82 ++++++ internal/api/internal_logs_test.go | 53 ++++ internal/api/internal_triage.go | 218 +++++++++++++++ internal/api/internal_triage_test.go | 202 ++++++++++++++ internal/api/server.go | 6 + internal/observ/api.go | 58 +++- internal/observ/config.go | 16 ++ internal/observ/fingerprint.go | 53 ++++ internal/observ/logengine.go | 380 +++++++++++++++++++++++++++ internal/observ/logengine_test.go | 277 +++++++++++++++++++ internal/observ/logrule.go | 124 +++++++++ internal/observ/logrule_store.go | 71 +++++ internal/observ/logrule_test.go | 163 ++++++++++++ internal/observ/logwatch.go | 252 ++++++++++++++++++ internal/observ/logwatch_test.go | 186 +++++++++++++ internal/observ/plugin.go | 37 ++- internal/observ/responder.go | 122 +++++++++ internal/observ/triage_client.go | 146 ++++++++++ pkg/config/config.go | 3 + 20 files changed, 2449 insertions(+), 3 deletions(-) create mode 100644 internal/api/internal_logs.go create mode 100644 internal/api/internal_logs_test.go create mode 100644 internal/api/internal_triage.go create mode 100644 internal/api/internal_triage_test.go create mode 100644 internal/observ/fingerprint.go create mode 100644 internal/observ/logengine.go create mode 100644 internal/observ/logengine_test.go create mode 100644 internal/observ/logrule.go create mode 100644 internal/observ/logrule_store.go create mode 100644 internal/observ/logrule_test.go create mode 100644 internal/observ/logwatch.go create mode 100644 internal/observ/logwatch_test.go create mode 100644 internal/observ/responder.go create mode 100644 internal/observ/triage_client.go 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..5714f12 --- /dev/null +++ b/internal/api/internal_logs.go @@ -0,0 +1,82 @@ +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 + } + } + + 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 { + path, pathErr := resolveLogFilePath(deployment.Path, source.Path) + if pathErr != nil { + return + } + _ = streamFileLogs(ctx, path, tail, sink) + return + } + _ = s.manager.StreamDeploymentLogs(ctx, name, deployment.Path, tail, sink, services...) +} 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..a2f5dbe --- /dev/null +++ b/internal/api/internal_triage.go @@ -0,0 +1,218 @@ +package api + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "strings" + "sync" + "time" + + "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. + text = text[:maxTriageContextChars] + "\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..f81fb05 --- /dev/null +++ b/internal/api/internal_triage_test.go @@ -0,0 +1,202 @@ +package api + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "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)) + } +} + +// 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/server.go b/internal/api/server.go index dbca888..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 { @@ -874,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) 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, "