diff --git a/README.md b/README.md index 1a64ec3..0cb3dd3 100644 --- a/README.md +++ b/README.md @@ -336,6 +336,8 @@ to Docker or stored in the image. - `OTELLOG_PATH` - path to the persistent telemetry log file (default `otel-agent.log`). Only used when ClickHouse is not configured. - `OTEL_SOURCEMAP_KEY` - shared key required to upload source maps. +- `OTEL_ISSUE_STATUS_PATH` - file that keeps resolved/ignored issue decisions across restarts when ClickHouse is not used (default: next to `OTELLOG_PATH`). +- `OTEL_TODOS_PATH` - file for todos filed via `otel-todo.js` when ClickHouse is not used (default: next to `OTELLOG_PATH`). - `OTEL_SOURCEMAP_CACHE_MB` - size budget (on-disk MB, default 64) for maps kept parsed in memory; maps are indexed at start and parsed lazily on first use. - `OTEL_SOURCEMAP_DIR` - directory for uploaded release source maps (default diff --git a/cmd/server/main.go b/cmd/server/main.go index cffb4f6..a97cb3e 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -8,6 +8,7 @@ import ( "net/http" "net/http/pprof" "os" + "path/filepath" "strings" "git.tornberg.me/go-otel/internal/abuseipdb" @@ -18,6 +19,7 @@ import ( "git.tornberg.me/go-otel/internal/geoip" "git.tornberg.me/go-otel/internal/honeypot" "git.tornberg.me/go-otel/internal/investigation" + "git.tornberg.me/go-otel/internal/issues" kube "git.tornberg.me/go-otel/internal/kubernetes" "git.tornberg.me/go-otel/internal/prometheus" "git.tornberg.me/go-otel/internal/sourcemaps" @@ -185,6 +187,18 @@ func withCORS(next http.HandlerFunc) http.HandlerFunc { } } +// dataPath resolves a small state file: the env override, else next to the +// telemetry log (a mounted volume in the cluster), else the working directory. +func dataPath(env, name string) string { + if p := os.Getenv(env); p != "" { + return p + } + if lp := os.Getenv("OTELLOG_PATH"); lp != "" { + return filepath.Join(filepath.Dir(lp), name) + } + return name +} + // setupStore returns the telemetry store: ClickHouse when OTEL_CLICKHOUSE_ADDR // is set (host:port, native protocol), otherwise the JSONL file store. func setupStore(ctx context.Context) storage.Store { @@ -295,11 +309,7 @@ func main() { } } if todoStore == nil { - path := os.Getenv("OTEL_TODOS_PATH") - if path == "" { - path = "otel-todos.jsonl" - } - if st, err := todos.NewFileStore(path); err != nil { + if st, err := todos.NewFileStore(dataPath("OTEL_TODOS_PATH", "otel-todos.jsonl")); err != nil { log.Printf("todos disabled: %v", err) } else { todoStore = st @@ -308,6 +318,26 @@ func main() { if todoStore != nil { coll.SetTodoStore(todoStore) } + var statusStore issues.StatusStore + if chStore, ok := store.(*storage.ClickHouseStore); ok { + if st, err := issues.NewClickHouseStatusStore(context.Background(), chStore.Conn(), chStore.Database()); err != nil { + log.Printf("issue status: clickhouse unavailable, using file store: %v", err) + } else { + statusStore = st + } + } + if statusStore == nil { + if st, err := issues.NewFileStatusStore(dataPath("OTEL_ISSUE_STATUS_PATH", "otel-issue-status.json")); err != nil { + log.Printf("issue status will not be persisted: %v", err) + } else { + statusStore = st + } + } + if statusStore != nil { + if err := coll.SetIssueStatusStore(context.Background(), statusStore); err != nil { + log.Printf("failed to load issue status: %v", err) + } + } investigator := investigation.NewHandler(investigation.ConfigFromEnv(), investigationStore) abuse := abuseipdb.NewHandler(abuseipdb.ConfigFromEnv()) @@ -409,6 +439,7 @@ func main() { uiMux.Handle("GET /api/issues/{id}/similar", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleIssueSimilar))) uiMux.Handle("POST /api/issues/{id}/similar", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleSimilarityFeedback))) uiMux.Handle("GET /api/todos", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleTodoList))) + uiMux.Handle("PATCH /api/todos/{id}", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleTodoUpdate))) uiMux.Handle("GET /api/patterns", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandlePatterns))) uiMux.Handle("GET /api/topology", oidcAuth.RequireAuth(http.HandlerFunc(topo.HandleTopology))) uiMux.Handle("GET /api/rollouts", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleRollouts))) diff --git a/internal/collector/collector.go b/internal/collector/collector.go index ba2d043..e5aac47 100644 --- a/internal/collector/collector.go +++ b/internal/collector/collector.go @@ -222,6 +222,11 @@ func (c *Collector) SetIssueStatus(id, status string) (issues.Issue, bool) { return c.issues.SetStatus(id, status) } +// SetIssueStatusStore makes resolved/ignored decisions survive restarts. +func (c *Collector) SetIssueStatusStore(ctx context.Context, store issues.StatusStore) error { + return c.issues.SetStatusStore(ctx, store) +} + func (c *Collector) SimilarIssues(id string, limit int) ([]issues.SimilarIssue, bool) { return c.issues.Similar(id, limit) } diff --git a/internal/collector/todos.go b/internal/collector/todos.go index 2013b3c..c65157b 100644 --- a/internal/collector/todos.go +++ b/internal/collector/todos.go @@ -116,3 +116,30 @@ func (c *Collector) HandleTodoList(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") _ = json.NewEncoder(w).Encode(items) } + +// HandleTodoUpdate marks a todo done or open: PATCH /api/todos/{id} {"done": bool}. +func (c *Collector) HandleTodoUpdate(w http.ResponseWriter, r *http.Request) { + if c.todoStore == nil { + http.Error(w, "todos not enabled", http.StatusServiceUnavailable) + return + } + var req struct { + Done *bool `json:"done"` + } + if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1024)).Decode(&req); err != nil || req.Done == nil { + http.Error(w, "body must be {\"done\": true|false}", http.StatusBadRequest) + return + } + t, ok, err := c.todoStore.SetDone(r.Context(), r.PathValue("id"), *req.Done) + if err != nil { + log.Printf("failed to update todo: %v", err) + http.Error(w, "failed to update todo", http.StatusInternalServerError) + return + } + if !ok { + http.Error(w, "todo not found", http.StatusNotFound) + return + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(t) +} diff --git a/internal/issues/statusstore.go b/internal/issues/statusstore.go new file mode 100644 index 0000000..70dd639 --- /dev/null +++ b/internal/issues/statusstore.go @@ -0,0 +1,122 @@ +package issues + +import ( + "context" + "encoding/json" + "fmt" + "os" + "sync" + "time" + + "github.com/ClickHouse/clickhouse-go/v2/lib/driver" +) + +// TriageState is the part of an issue a person decided, kept across restarts. +// The tracker is in-memory, so without this every deploy would reopen every +// resolved and ignored issue. +type TriageState struct { + Status string `json:"status"` + ResolvedAt int64 `json:"resolvedAt,omitempty"` +} + +// StatusStore persists triage state by issue id. +type StatusStore interface { + Load(ctx context.Context) (map[string]TriageState, error) + Save(ctx context.Context, id string, s TriageState) error +} + +// FileStatusStore keeps all states in one JSON file, rewritten on each save. +type FileStatusStore struct { + mu sync.Mutex + path string + states map[string]TriageState +} + +func NewFileStatusStore(path string) (*FileStatusStore, error) { + fs := &FileStatusStore{path: path, states: map[string]TriageState{}} + b, err := os.ReadFile(path) + if os.IsNotExist(err) { + return fs, nil + } + if err != nil { + return nil, err + } + if err := json.Unmarshal(b, &fs.states); err != nil { + return nil, err + } + return fs, nil +} + +func (s *FileStatusStore) Load(context.Context) (map[string]TriageState, error) { + s.mu.Lock() + defer s.mu.Unlock() + out := make(map[string]TriageState, len(s.states)) + for k, v := range s.states { + out[k] = v + } + return out, nil +} + +func (s *FileStatusStore) Save(_ context.Context, id string, st TriageState) error { + s.mu.Lock() + defer s.mu.Unlock() + if st.Status == StatusOpen { + delete(s.states, id) + } else { + s.states[id] = st + } + b, err := json.Marshal(s.states) + if err != nil { + return err + } + tmp := s.path + ".tmp" + if err := os.WriteFile(tmp, b, 0o644); err != nil { + return err + } + return os.Rename(tmp, s.path) +} + +// ClickHouseStatusStore keeps the latest state per issue id. +type ClickHouseStatusStore struct { + conn driver.Conn + db string +} + +func NewClickHouseStatusStore(ctx context.Context, conn driver.Conn, db string) (*ClickHouseStatusStore, error) { + err := conn.Exec(ctx, fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s.issue_status ( + id String, + status LowCardinality(String), + resolved_at Int64, + updated_at DateTime64(3, 'UTC') + ) ENGINE = ReplacingMergeTree(updated_at) + ORDER BY id`, db)) + if err != nil { + return nil, err + } + return &ClickHouseStatusStore{conn: conn, db: db}, nil +} + +func (s *ClickHouseStatusStore) Load(ctx context.Context) (map[string]TriageState, error) { + rows, err := s.conn.Query(ctx, fmt.Sprintf(`SELECT id, status, resolved_at FROM %s.issue_status FINAL`, s.db)) + if err != nil { + return nil, err + } + defer rows.Close() + out := map[string]TriageState{} + for rows.Next() { + var id string + var st TriageState + if err := rows.Scan(&id, &st.Status, &st.ResolvedAt); err != nil { + return nil, err + } + if st.Status != StatusOpen { + out[id] = st + } + } + return out, rows.Err() +} + +func (s *ClickHouseStatusStore) Save(ctx context.Context, id string, st TriageState) error { + return s.conn.Exec(ctx, fmt.Sprintf(`INSERT INTO %s.issue_status (id, status, resolved_at, updated_at) VALUES (?, ?, ?, ?)`, s.db), + id, st.Status, st.ResolvedAt, time.Now().UTC()) +} diff --git a/internal/issues/statusstore_test.go b/internal/issues/statusstore_test.go new file mode 100644 index 0000000..9087c0d --- /dev/null +++ b/internal/issues/statusstore_test.go @@ -0,0 +1,50 @@ +package issues + +import ( + "context" + "path/filepath" + "testing" + "time" +) + +func TestTriagePersistsAcrossRestart(t *testing.T) { + path := filepath.Join(t.TempDir(), "status.json") + store, _ := NewFileStatusStore(path) + tr := NewTracker() + if err := tr.SetStatusStore(context.Background(), store); err != nil { + t.Fatal(err) + } + tr.Record("span", "svc", "boom", Sample{Timestamp: 1}) + id := Fingerprint("span", "svc", NormalizeTitle("boom")) + if _, ok := tr.SetStatus(id, StatusIgnored); !ok { + t.Fatal("set status") + } + time.Sleep(100 * time.Millisecond) // async save + + // "Restart": fresh tracker, same file. + store2, err := NewFileStatusStore(path) + if err != nil { + t.Fatal(err) + } + tr2 := NewTracker() + if err := tr2.SetStatusStore(context.Background(), store2); err != nil { + t.Fatal(err) + } + tr2.Record("span", "svc", "boom", Sample{Timestamp: 2}) + got, _ := tr2.Get(id) + if got.Status != StatusIgnored { + t.Fatalf("status after restart = %q, want ignored", got.Status) + } +} + +func TestResolvedIssueRegresses(t *testing.T) { + tr := NewTracker() + tr.Record("span", "svc", "boom", Sample{Timestamp: 1}) + id := Fingerprint("span", "svc", NormalizeTitle("boom")) + tr.SetStatus(id, StatusResolved) + tr.Record("span", "svc", "boom", Sample{Timestamp: 5}) + got, _ := tr.Get(id) + if got.Status != StatusOpen || got.RegressedAt != 5 { + t.Fatalf("got status=%q regressedAt=%d", got.Status, got.RegressedAt) + } +} diff --git a/internal/issues/tracker.go b/internal/issues/tracker.go index b5ac7b9..95c30af 100644 --- a/internal/issues/tracker.go +++ b/internal/issues/tracker.go @@ -3,8 +3,10 @@ package issues import ( + "context" "fmt" "hash/fnv" + "log" "regexp" "sort" "strings" @@ -86,6 +88,8 @@ type Issue struct { // Status is the triage state: open, resolved or ignored. Status string `json:"status"` ResolvedAt int64 `json:"resolvedAt,omitempty"` + // RegressedAt is when a resolved issue last fired again and was reopened. + RegressedAt int64 `json:"regressedAt,omitempty"` // NormalizedTitle is the deterministic, redacted form used for exact grouping. NormalizedTitle string `json:"normalizedTitle,omitempty"` // PatternID groups related exact issues without merging their operational IDs. @@ -117,10 +121,14 @@ type Pattern struct { // Tracker incrementally fingerprints and deduplicates errors as they're // recorded. It is safe for concurrent use. type Tracker struct { - mu sync.Mutex - issues map[string]*Issue - feedback map[string]string - sink OccurrenceSink + mu sync.Mutex + issues map[string]*Issue + // triage holds persisted resolved/ignored decisions by issue id, applied + // when an issue is (re)created after a restart. + triage map[string]TriageState + statusStore StatusStore + feedback map[string]string + sink OccurrenceSink } // OccurrenceSink receives every recorded occurrence (already normalized) so @@ -161,7 +169,43 @@ type ListPage struct { // NewTracker creates an empty Tracker. func NewTracker() *Tracker { - return &Tracker{issues: make(map[string]*Issue), feedback: make(map[string]string)} + return &Tracker{issues: make(map[string]*Issue), feedback: make(map[string]string), triage: make(map[string]TriageState)} +} + +// SetStatusStore loads persisted triage decisions and saves future ones. +func (t *Tracker) SetStatusStore(ctx context.Context, store StatusStore) error { + states, err := store.Load(ctx) + if err != nil { + return err + } + t.mu.Lock() + defer t.mu.Unlock() + t.statusStore = store + for id, st := range states { + t.triage[id] = st + if issue, ok := t.issues[id]; ok { + issue.Status, issue.ResolvedAt = st.Status, st.ResolvedAt + } + } + return nil +} + +// persistLocked records a decision. The write happens off the lock. +func (t *Tracker) persistLocked(id string, st TriageState) { + if st.Status == StatusOpen { + delete(t.triage, id) + } else { + t.triage[id] = st + } + if t.statusStore == nil { + return + } + store := t.statusStore + go func() { + if err := store.Save(context.Background(), id, st); err != nil { + log.Printf("failed to persist issue status: %v", err) + } + }() } // Record adds an occurrence of an error, creating a new issue or merging into @@ -196,6 +240,9 @@ func (t *Tracker) record(kind, service, title string, sample Sample) (string, Oc Status: StatusOpen, NormalizedTitle: NormalizeTitle(title), } + if st, ok := t.triage[id]; ok { + issue.Status, issue.ResolvedAt = st.Status, st.ResolvedAt + } issue.PatternSignature = patternSignature(kind, service, title, sample) issue.PatternID = Fingerprint("pattern", issue.PatternSignature) t.issues[id] = issue @@ -214,6 +261,8 @@ func (t *Tracker) record(kind, service, title string, sample Sample) (string, Oc if issue.Status == StatusResolved { issue.Status = StatusOpen issue.ResolvedAt = 0 + issue.RegressedAt = sample.Timestamp + t.persistLocked(id, TriageState{Status: StatusOpen}) } issue.Samples = append(issue.Samples, sample) if len(issue.Samples) > maxSamples { @@ -462,6 +511,8 @@ func (t *Tracker) SetStatus(id, status string) (Issue, bool) { } else { issue.ResolvedAt = time.Now().UnixNano() } + issue.RegressedAt = 0 + t.persistLocked(id, TriageState{Status: status, ResolvedAt: issue.ResolvedAt}) return cloneIssue(issue), true } diff --git a/internal/todos/todos.go b/internal/todos/todos.go index 8fa72d0..0c2ad70 100644 --- a/internal/todos/todos.go +++ b/internal/todos/todos.go @@ -27,12 +27,16 @@ type Todo struct { UserAgent string `json:"userAgent,omitempty"` Session string `json:"session,omitempty"` CreatedAt time.Time `json:"createdAt"` + Done bool `json:"done"` + DoneAt time.Time `json:"doneAt,omitzero"` } // Store persists todos. List returns newest first; an empty project means all. type Store interface { Add(ctx context.Context, t Todo) error List(ctx context.Context, project string, limit int) ([]Todo, error) + // SetDone marks a todo done or open; ok is false for an unknown id. + SetDone(ctx context.Context, id string, done bool) (t Todo, ok bool, err error) } // FileStore keeps todos in memory and appends them to a JSONL file. @@ -82,6 +86,35 @@ func (s *FileStore) Add(_ context.Context, t Todo) error { return nil } +func (s *FileStore) SetDone(_ context.Context, id string, done bool) (Todo, bool, error) { + s.mu.Lock() + defer s.mu.Unlock() + for i := range s.items { + if s.items[i].ID != id { + continue + } + s.items[i].Done = done + s.items[i].DoneAt = time.Time{} + if done { + s.items[i].DoneAt = time.Now().UTC() + } + var buf []byte + for _, t := range s.items { + line, err := json.Marshal(t) + if err != nil { + return Todo{}, false, err + } + buf = append(append(buf, line...), '\n') + } + tmp := s.path + ".tmp" + if err := os.WriteFile(tmp, buf, 0o644); err != nil { + return Todo{}, false, err + } + return s.items[i], true, os.Rename(tmp, s.path) + } + return Todo{}, false, nil +} + func (s *FileStore) List(_ context.Context, project string, limit int) ([]Todo, error) { s.mu.Lock() defer s.mu.Unlock() @@ -121,6 +154,11 @@ func NewClickHouseStore(ctx context.Context, conn driver.Conn, db string) (*Clic if err != nil { return nil, err } + for _, col := range []string{"done UInt8", "done_at DateTime64(3, 'UTC')"} { + if err := conn.Exec(ctx, fmt.Sprintf(`ALTER TABLE %s.todos ADD COLUMN IF NOT EXISTS %s`, db, col)); err != nil { + return nil, err + } + } return &ClickHouseStore{conn: conn, db: db}, nil } @@ -130,7 +168,7 @@ func (s *ClickHouseStore) Add(ctx context.Context, t Todo) error { } func (s *ClickHouseStore) List(ctx context.Context, project string, limit int) ([]Todo, error) { - q := fmt.Sprintf(`SELECT id, project, text, component, path, hash, url, user_agent, session, created_at FROM %s.todos`, s.db) + q := fmt.Sprintf(`SELECT id, project, text, component, path, hash, url, user_agent, session, created_at, done, done_at FROM %s.todos`, s.db) var args []any if project != "" { q += ` WHERE project = ?` @@ -146,10 +184,41 @@ func (s *ClickHouseStore) List(ctx context.Context, project string, limit int) ( var out []Todo for rows.Next() { var t Todo - if err := rows.Scan(&t.ID, &t.Project, &t.Text, &t.Component, &t.Path, &t.Hash, &t.URL, &t.UserAgent, &t.Session, &t.CreatedAt); err != nil { + var done uint8 + if err := rows.Scan(&t.ID, &t.Project, &t.Text, &t.Component, &t.Path, &t.Hash, &t.URL, &t.UserAgent, &t.Session, &t.CreatedAt, &done, &t.DoneAt); err != nil { return nil, err } + t.Done = done != 0 out = append(out, t) } return out, rows.Err() } + +func (s *ClickHouseStore) SetDone(ctx context.Context, id string, done bool) (Todo, bool, error) { + var d uint8 + doneAt := time.Unix(0, 0).UTC() + if done { + d, doneAt = 1, time.Now().UTC() + } + if err := s.conn.Exec(ctx, fmt.Sprintf(`ALTER TABLE %s.todos UPDATE done = ?, done_at = ? WHERE id = ? SETTINGS mutations_sync = 1`, s.db), d, doneAt, id); err != nil { + return Todo{}, false, err + } + rows, err := s.conn.Query(ctx, fmt.Sprintf(`SELECT id, project, text, component, path, hash, url, user_agent, session, created_at, done, done_at FROM %s.todos WHERE id = ?`, s.db), id) + if err != nil { + return Todo{}, false, err + } + defer rows.Close() + if !rows.Next() { + return Todo{}, false, nil + } + var t Todo + var dn uint8 + if err := rows.Scan(&t.ID, &t.Project, &t.Text, &t.Component, &t.Path, &t.Hash, &t.URL, &t.UserAgent, &t.Session, &t.CreatedAt, &dn, &t.DoneAt); err != nil { + return Todo{}, false, err + } + t.Done = dn != 0 + if !t.Done { + t.DoneAt = time.Time{} + } + return t, true, nil +} diff --git a/otel-ui/src/components/IssuesVisualizer.tsx b/otel-ui/src/components/IssuesVisualizer.tsx index 888dc0b..3bcee1a 100644 --- a/otel-ui/src/components/IssuesVisualizer.tsx +++ b/otel-ui/src/components/IssuesVisualizer.tsx @@ -373,6 +373,15 @@ const IssueCard = ({ {toneLabel[tone]} + {issue.regressedAt && issueStatusOf(issue) === "open" ? ( + + + ) : null}