From f77b89a77915a8436f33e1edc8e647d3b7ed7970 Mon Sep 17 00:00:00 2001 From: developer Date: Thu, 17 Sep 2026 09:30:18 +0000 Subject: [PATCH] Improve agent-facing history endpoints (Phase 1) - Fix /api/history/{type} to search the full persistent store instead of only the bounded in-memory ring buffer, so filtered history isn't silently truncated. - Add structured type/since/until filters to /api/history/search and /api/history/wait, alongside the existing free-text query. - Add GET /api/traces/{traceId} to fetch all entries for a trace in one call instead of relying on substring search. - Rate-limit /api/history/wait per client IP. - Add docs/PLAN.md tracking the phased improvement plan; update README and agents_example.md for the new/changed endpoints. Co-Authored-By: Claude Sonnet 5 --- README.md | 10 ++ agents_example.md | 18 ++-- cmd/server/main.go | 1 + docs/PLAN.md | 44 ++++++++ go.mod | 3 +- go.sum | 2 + internal/collector/collector.go | 38 ++++--- internal/collector/collector_test.go | 90 +++++++++++++++- internal/collector/handlers.go | 149 +++++++++++++++++---------- internal/collector/ratelimit.go | 81 +++++++++++++++ internal/storage/file_store.go | 53 ++++++++-- internal/storage/file_store_test.go | 55 +++++++++- internal/storage/store.go | 17 ++- 13 files changed, 473 insertions(+), 88 deletions(-) create mode 100644 docs/PLAN.md create mode 100644 internal/collector/ratelimit.go diff --git a/README.md b/README.md index 01d021c..553bf31 100644 --- a/README.md +++ b/README.md @@ -95,6 +95,16 @@ gRPC services for traces, metrics, and logs using standard OTLP protobuf definit - `GET /api/history/traces` - Retrieve only trace history - `GET /api/history/metrics` - Retrieve only metrics history - `GET /api/history/logs` - Retrieve only logs history +- `GET /api/history/search` - Search history. Supports `q` (free-text + substring), `type`, `since`/`until` (unix nanoseconds), and `limit`. +- `GET /api/history/wait` - Block until a matching entry arrives. Same + filters as `search`, plus `timeout` (seconds, default 30). Rate-limited + per client IP. +- `GET /api/traces/{traceId}` - Retrieve all stored entries (spans and + correlated logs) referencing a given trace id. + +All endpoints above search the full persistent store (when `OTELLOG_PATH` +storage is enabled), not just the in-memory ring buffer. ### Web Interface diff --git a/agents_example.md b/agents_example.md index c4175c4..9d653e2 100644 --- a/agents_example.md +++ b/agents_example.md @@ -5,14 +5,18 @@ This document provides instructions for AI agents on how to utilize the OTEL Col ## Base Configuration - **Base URL**: `https://otel.k6n.net` - **Endpoints**: - - `GET /api/history/search?q={query}&limit={limit}`: Search all historical telemetry data. + - `GET /api/history/search?q={query}&type={type}&since={ns}&until={ns}&limit={limit}`: Search all historical telemetry data. - **Default Limit**: 50 (max 100 recommended for context). - **Ordering**: Latest entries first. - - **Match Scope**: Case-insensitive substring match on raw JSON (covers service name, span body, attributes, etc.). - - `GET /api/history/wait?q={query}&timeout={seconds}`: **Wait** for a specific signal to arrive. + - **Match Scope**: Case-insensitive substring match on raw JSON (covers service name, span body, attributes, etc.), combinable with structured filters. + - **type**: optional, one of `traces`, `metrics`, `logs`. + - **since / until**: optional, unix nanosecond timestamps to bound the time range. + - `GET /api/history/wait?q={query}&type={type}&timeout={seconds}`: **Wait** for a specific signal to arrive. - **Blocking**: This call blocks until a match is found or the timeout (default 30s) is reached. - **Ideal for Automation**: Use this after triggering an action to wait for the exact result. - - `GET /api/history`: Fetch the recent in-memory history of all telemetry. + - **Rate limited**: capped per client IP; back off if you receive `429`. + - `GET /api/traces/{traceId}`: Fetch every stored entry (spans and correlated logs) referencing a specific trace id in one call. Prefer this over substring-searching for a trace id. + - `GET /api/history`: Fetch recent history of all telemetry (searches the full persistent store, not just recent in-memory entries). - `GET /api/history/traces`: Fetch recent traces. - `GET /api/history/logs`: Fetch recent logs. - `GET /api/history/metrics`: Fetch recent metrics. @@ -38,12 +42,12 @@ curl "https://otel.k6n.net/api/history/wait?q=5682138e4a9b4629852899d45e542456&t ``` ### 3. Identifying Trace Chains -If you find a log entry with a `traceId`, use the search endpoint to find all spans and logs associated with that specific trace to understand the full request flow. +If you find a log entry with a `traceId`, use the dedicated trace endpoint to fetch all spans and logs associated with that specific trace and understand the full request flow. **Example Request:** ```bash -# Search for all activity related to a specific trace -curl "https://otel.k6n.net/api/history/search?q=5682138e4a9b4629852899d45e542456&limit=100" +# Fetch all activity related to a specific trace +curl "https://otel.k6n.net/api/traces/5682138e4a9b4629852899d45e542456" ``` ### 3. Fetching Recent Telemetry diff --git a/cmd/server/main.go b/cmd/server/main.go index 81a2760..feb1bc2 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -205,6 +205,7 @@ func main() { uiMux.HandleFunc("/api/history/logs", coll.HandleHistoryByType("logs")) uiMux.HandleFunc("/api/history/search", coll.HandleSearch) uiMux.HandleFunc("/api/history/wait", coll.HandleWait) + uiMux.HandleFunc("GET /api/traces/{traceId}", coll.HandleTraceByID) uiMux.HandleFunc("/ws", wsServer.HandleConnections) uiMux.Handle("/", spaHandler(http.Dir("./dist"))) diff --git a/docs/PLAN.md b/docs/PLAN.md new file mode 100644 index 0000000..167d0e8 --- /dev/null +++ b/docs/PLAN.md @@ -0,0 +1,44 @@ +# Improvement plan + +Priority order: (1) agent endpoints, (2) recent issues for the web UI, (3) a +Sentry-like frontend error collector. + +## Phase 1 — Agent endpoints (highest priority) + +- [x] Fix `HandleHistoryByType`: it only filtered the in-memory ring buffer + (capped at `historySize`, default 100) instead of the persistent store, + so `/api/history/traces|metrics|logs` silently returned far less than + `/api/history/search` could see. Now routed through `SearchHistory`. +- [x] Structured filters on `/api/history/search` and `/api/history/wait`: + `type=`, `since=`, `until=` (unix nanoseconds), combinable with the + existing free-text `q=`. +- [x] `GET /api/traces/{traceId}` — assemble all entries referencing one + trace id in a single call, instead of relying on substring search + matching the id inside raw JSON. +- [x] Basic per-IP rate limiting on `/api/history/wait` (a blocking + long-poll, cheap to abuse). +- [ ] Cursor-based pagination for `/api/history/search` (deferred). +- [ ] Machine-readable tool manifest (`/.well-known/agent-tools.json` or an + MCP server) replacing `agents_example.md` (deferred). + +## Phase 2 — Recent issues for the web UI + +- [ ] Fingerprint spans with `status=ERROR`/exception events and logs with + `severity >= ERROR` into deduped Issues (count, firstSeen, lastSeen, + sample entries, affected services), updated incrementally on ingest. +- [ ] `GET /api/issues`, `GET /api/issues/{id}`. +- [ ] UI "Issues" tab (lives in the separate frontend/`dist` project). + +## Phase 3 — Frontend error collector (Sentry-style) + +- [ ] `static/otel-rum.js`: captures `window.onerror`, + `unhandledrejection`, resource load failures; translates them to OTLP + log records client-side and POSTs to the existing `/v1/logs` endpoint + (so Phase 2's issue derivation picks them up for free via a + `source: frontend` attribute). +- [ ] CORS support on the OTLP HTTP endpoints (currently absent — required + for any browser-origin POST). +- [ ] Lightweight ingest key for scoping/rate-limiting noisy pages (not + real auth). +- [ ] Explicitly out of scope for v1: source map resolution, session + replay, PII scrubbing. diff --git a/go.mod b/go.mod index 27b207a..321300b 100644 --- a/go.mod +++ b/go.mod @@ -1,6 +1,6 @@ module git.tornberg.me/go-otel -go 1.26 +go 1.26.0 require ( github.com/gorilla/websocket v1.5.3 @@ -26,6 +26,7 @@ require ( golang.org/x/net v0.52.0 // indirect golang.org/x/sys v0.42.0 // indirect golang.org/x/text v0.35.0 // indirect + golang.org/x/time v0.16.0 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20260401024825-9d38bb4040a9 // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20260401024825-9d38bb4040a9 // indirect gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect diff --git a/go.sum b/go.sum index d098e66..49c737d 100644 --- a/go.sum +++ b/go.sum @@ -76,6 +76,8 @@ golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo= golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= golang.org/x/text v0.35.0 h1:JOVx6vVDFokkpaq1AEptVzLTpDe9KGpj5tR4/X+ybL8= golang.org/x/text v0.35.0/go.mod h1:khi/HExzZJ2pGnjenulevKNX1W67CUy0AsXcNubPGCA= +golang.org/x/time v0.16.0 h1:vMb6ptszcQMkcwiRTAuNNU50gom6++Q/6gY2hDM6VDE= +golang.org/x/time v0.16.0/go.mod h1:rVKOqvZeKvrDKTQiAHJ7wmwP0RzleSphoEA9RcdLA0s= gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= google.golang.org/genproto/googleapis/api v0.0.0-20260401024825-9d38bb4040a9 h1:VPWxll4HlMw1Vs/qXtN7BvhZqsS9cdAittCNvVENElA= diff --git a/internal/collector/collector.go b/internal/collector/collector.go index f930944..6aa8d5e 100644 --- a/internal/collector/collector.go +++ b/internal/collector/collector.go @@ -29,6 +29,7 @@ type Collector struct { store storage.Store subscribers map[chan websocket.HistoryEntry]func(websocket.HistoryEntry) bool subMutex sync.Mutex + waitLimiter *ipRateLimiter } // New creates a new Collector instance. @@ -39,6 +40,8 @@ func New(broadcaster websocket.Broadcaster, historySize int, store storage.Store history: ring.New(historySize), store: store, subscribers: make(map[chan websocket.HistoryEntry]func(websocket.HistoryEntry) bool), + // Allow bursts of a few immediate waits, then steady-state ~1 req/sec per IP. + waitLimiter: newIPRateLimiter(1, 5), } } @@ -129,32 +132,39 @@ func (c *Collector) GetHistory(limit int) []websocket.HistoryEntry { return history } -// SearchHistory searches through stored telemetry data using the provided query. -func (c *Collector) SearchHistory(ctx context.Context, query string, limit int) ([]websocket.HistoryEntry, error) { +// SearchHistory searches through stored telemetry data using the provided options. +// When a persistent store is configured, the full stored history is searched; +// otherwise it falls back to the in-memory ring buffer. +func (c *Collector) SearchHistory(ctx context.Context, opts storage.SearchOptions, limit int) ([]websocket.HistoryEntry, error) { if limit <= 0 { limit = 50 } if c.store != nil { - return c.store.Search(ctx, query, limit) + return c.store.Search(ctx, opts, limit) } - // Fallback to in-memory if no store - history := c.GetHistory(limit) + // Fallback to in-memory if no store. Use the full ring buffer, not just + // 'limit' entries, so type/time filters see everything available. + history := c.GetHistory(0) var filtered []websocket.HistoryEntry // Lowercase query for case-insensitive search - lowerQuery := strings.ToLower(query) + lowerQuery := strings.ToLower(opts.Query) for _, entry := range history { - if query == "" { - filtered = append(filtered, entry) - } else { - // Basic string filter for in-memory, no need to re-marshal RawMessage - if strings.Contains(strings.ToLower(string(entry.Data)), lowerQuery) { - filtered = append(filtered, entry) - } + if opts.Query != "" && !strings.Contains(strings.ToLower(string(entry.Data)), lowerQuery) { + continue } - // Keep it simple and just filter all then limit, since historySize is usually small + if opts.Type != "" && entry.Type != opts.Type { + continue + } + if opts.Since != 0 && entry.Timestamp < opts.Since { + continue + } + if opts.Until != 0 && entry.Timestamp > opts.Until { + continue + } + filtered = append(filtered, entry) } // Respect limit and return latest messages first diff --git a/internal/collector/collector_test.go b/internal/collector/collector_test.go index eccdbcd..8214e69 100644 --- a/internal/collector/collector_test.go +++ b/internal/collector/collector_test.go @@ -3,9 +3,13 @@ package collector import ( "context" "encoding/json" + "net/http" + "net/http/httptest" + "strings" "testing" "time" + "git.tornberg.me/go-otel/internal/storage" "git.tornberg.me/go-otel/internal/websocket" ) @@ -22,8 +26,18 @@ func (m *mockStore) GetHistory(ctx context.Context, limit int) ([]websocket.Hist return m.entries, nil } -func (m *mockStore) Search(ctx context.Context, query string, limit int) ([]websocket.HistoryEntry, error) { - return m.entries, nil +func (m *mockStore) Search(ctx context.Context, opts storage.SearchOptions, limit int) ([]websocket.HistoryEntry, error) { + var filtered []websocket.HistoryEntry + for _, e := range m.entries { + if opts.Type != "" && e.Type != opts.Type { + continue + } + if opts.Query != "" && !strings.Contains(string(e.Data), opts.Query) { + continue + } + filtered = append(filtered, e) + } + return filtered, nil } func TestCollectorWithStore(t *testing.T) { @@ -81,3 +95,75 @@ func TestCollectorHistoryLimit(t *testing.T) { t.Errorf("expected 10 entries even with limit 100, got %d", len(history)) } } + +func TestHandleHistoryByTypeUsesStore(t *testing.T) { + store := &mockStore{ + entries: []websocket.HistoryEntry{ + {Type: "traces", Timestamp: 1, Data: json.RawMessage(`{"id":1}`)}, + {Type: "traces", Timestamp: 2, Data: json.RawMessage(`{"id":2}`)}, + {Type: "logs", Timestamp: 3, Data: json.RawMessage(`{"id":3}`)}, + }, + } + // Ring buffer is much smaller than the store's content, so a correct + // implementation must route through the store rather than the ring buffer. + coll := New(nil, 1, store) + + req := httptest.NewRequest(http.MethodGet, "/api/history/traces", nil) + rec := httptest.NewRecorder() + coll.HandleHistoryByType("traces")(rec, req) + + var results []websocket.HistoryEntry + if err := json.Unmarshal(rec.Body.Bytes(), &results); err != nil { + t.Fatalf("failed to decode response: %v", err) + } + if len(results) != 2 { + t.Fatalf("expected 2 trace entries from the store, got %d", len(results)) + } +} + +func TestHandleTraceByID(t *testing.T) { + store := &mockStore{ + entries: []websocket.HistoryEntry{ + {Type: "traces", Timestamp: 1, Data: json.RawMessage(`{"traceId":"abc123"}`)}, + {Type: "logs", Timestamp: 2, Data: json.RawMessage(`{"traceId":"abc123","msg":"related"}`)}, + {Type: "traces", Timestamp: 3, Data: json.RawMessage(`{"traceId":"other"}`)}, + }, + } + coll := New(nil, 10, store) + + req := httptest.NewRequest(http.MethodGet, "/api/traces/abc123", nil) + req.SetPathValue("traceId", "abc123") + rec := httptest.NewRecorder() + coll.HandleTraceByID(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("expected 200, got %d: %s", rec.Code, rec.Body.String()) + } + + var results []websocket.HistoryEntry + if err := json.Unmarshal(rec.Body.Bytes(), &results); err != nil { + t.Fatalf("failed to decode response: %v", err) + } + if len(results) != 2 { + t.Fatalf("expected 2 entries referencing trace abc123, got %d", len(results)) + } +} + +func TestHandleWaitRateLimited(t *testing.T) { + coll := New(nil, 10, nil) + + var lastCode int + for i := 0; i < 10; i++ { + req := httptest.NewRequest(http.MethodGet, "/api/history/wait?q=nope&timeout=0", nil) + req.RemoteAddr = "203.0.113.1:12345" + rec := httptest.NewRecorder() + coll.HandleWait(rec, req) + lastCode = rec.Code + if lastCode == http.StatusTooManyRequests { + break + } + } + if lastCode != http.StatusTooManyRequests { + t.Fatalf("expected rate limiting to eventually return 429, last code was %d", lastCode) + } +} diff --git a/internal/collector/handlers.go b/internal/collector/handlers.go index b26c0ca..18d15e4 100644 --- a/internal/collector/handlers.go +++ b/internal/collector/handlers.go @@ -12,6 +12,7 @@ import ( "strings" "time" + "git.tornberg.me/go-otel/internal/storage" "git.tornberg.me/go-otel/internal/websocket" "go.opentelemetry.io/collector/pdata/plog" "go.opentelemetry.io/collector/pdata/pmetric" @@ -146,23 +147,52 @@ func (c *Collector) HandleLogs(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) } -// HandleHistory returns the telemetry history as JSON. +// parseSearchOptions builds a storage.SearchOptions from common query +// parameters shared by the search/wait/history endpoints: q, type, since, until. +func parseSearchOptions(r *http.Request) storage.SearchOptions { + q := r.URL.Query() + opts := storage.SearchOptions{ + Query: q.Get("q"), + Type: q.Get("type"), + } + if since := q.Get("since"); since != "" { + if v, err := strconv.ParseInt(since, 10, 64); err == nil { + opts.Since = v + } + } + if until := q.Get("until"); until != "" { + if v, err := strconv.ParseInt(until, 10, 64); err == nil { + opts.Until = v + } + } + return opts +} + +func parseLimit(r *http.Request, def int) int { + if limitStr := r.URL.Query().Get("limit"); limitStr != "" { + if l, err := strconv.Atoi(limitStr); err == nil { + return l + } + } + return def +} + +// HandleHistory returns the telemetry history as JSON. It searches the full +// persistent store when one is configured, so results are not bounded by the +// in-memory ring buffer size. func (c *Collector) HandleHistory(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) return } - limitStr := r.URL.Query().Get("limit") - limit := 100 - if limitStr != "" { - if l, err := strconv.Atoi(limitStr); err == nil { - limit = l - } + history, err := c.SearchHistory(r.Context(), storage.SearchOptions{}, parseLimit(r, 100)) + if err != nil { + log.Printf("failed to fetch history: %v", err) + http.Error(w, "failed to fetch history", http.StatusInternalServerError) + return } - history := c.GetHistory(limit) - w.Header().Set("Content-Type", "application/json") if err := json.NewEncoder(w).Encode(history); err != nil { log.Printf("failed to encode history: %v", err) @@ -171,23 +201,16 @@ func (c *Collector) HandleHistory(w http.ResponseWriter, r *http.Request) { } } -// HandleSearch searches the telemetry history. +// HandleSearch searches the telemetry history. Supports the free-text 'q' +// parameter plus structured filters: type, since, until (unix nanoseconds). func (c *Collector) HandleSearch(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) return } - query := r.URL.Query().Get("q") - limitStr := r.URL.Query().Get("limit") - limit := 100 - if limitStr != "" { - if l, err := strconv.Atoi(limitStr); err == nil { - limit = l - } - } - - history, err := c.SearchHistory(r.Context(), query, limit) + opts := parseSearchOptions(r) + history, err := c.SearchHistory(r.Context(), opts, parseLimit(r, 100)) if err != nil { log.Printf("failed to search history: %v", err) http.Error(w, "failed to search history", http.StatusInternalServerError) @@ -202,14 +225,20 @@ func (c *Collector) HandleSearch(w http.ResponseWriter, r *http.Request) { } } -// HandleWait waits for a specific telemetry signal. +// HandleWait waits for a specific telemetry signal, optionally restricted by +// the same structured filters as HandleSearch (type, since, until). func (c *Collector) HandleWait(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) return } - query := r.URL.Query().Get("q") + if !c.waitLimiter.Allow(clientIP(r)) { + http.Error(w, "too many wait requests, slow down", http.StatusTooManyRequests) + return + } + + opts := parseSearchOptions(r) timeoutSec := 30 if t := r.URL.Query().Get("timeout"); t != "" { if ts, err := strconv.Atoi(t); err == nil { @@ -221,7 +250,7 @@ func (c *Collector) HandleWait(w http.ResponseWriter, r *http.Request) { defer cancel() // First, check if result already exists in recent history - matches, err := c.SearchHistory(ctx, query, 1) + matches, err := c.SearchHistory(ctx, opts, 1) if err == nil && len(matches) > 0 { w.Header().Set("Content-Type", "application/json") if err := json.NewEncoder(w).Encode(matches[0]); err != nil { @@ -231,13 +260,15 @@ func (c *Collector) HandleWait(w http.ResponseWriter, r *http.Request) { } // Otherwise, subscribe and wait - lowerQuery := strings.ToLower(query) + lowerQuery := strings.ToLower(opts.Query) ch, cleanup := c.Subscribe(func(entry websocket.HistoryEntry) bool { - if query == "" { + if opts.Type != "" && entry.Type != opts.Type { + return false + } + if opts.Query == "" { return true } - data, _ := json.Marshal(entry.Data) - return strings.Contains(strings.ToLower(string(data)), lowerQuery) + return strings.Contains(strings.ToLower(string(entry.Data)), lowerQuery) }) defer cleanup() @@ -252,7 +283,8 @@ func (c *Collector) HandleWait(w http.ResponseWriter, r *http.Request) { } } -// HandleHistoryByType returns telemetry history filtered by type. +// HandleHistoryByType returns telemetry history filtered by type, searching +// the full persistent store when one is configured. func (c *Collector) HandleHistoryByType(dataType string) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { @@ -260,33 +292,16 @@ func (c *Collector) HandleHistoryByType(dataType string) http.HandlerFunc { return } - limitStr := r.URL.Query().Get("limit") - limit := 100 - if limitStr != "" { - if l, err := strconv.Atoi(limitStr); err == nil { - limit = l - } - } - - // Fetch a larger set to allow for filtering, then slice - // Retrieve up to 1000 from memory then filter - history := c.GetHistory(5000) - - // Filter by type - var filteredHistory []websocket.HistoryEntry - for _, entry := range history { - if string(entry.Type) == dataType { - filteredHistory = append(filteredHistory, entry) - } - } - - // Limit the results - if len(filteredHistory) > limit { - filteredHistory = filteredHistory[len(filteredHistory)-limit:] + opts := storage.SearchOptions{Type: dataType} + history, err := c.SearchHistory(r.Context(), opts, parseLimit(r, 100)) + if err != nil { + log.Printf("failed to fetch history for type %s: %v", dataType, err) + http.Error(w, "failed to fetch history", http.StatusInternalServerError) + return } w.Header().Set("Content-Type", "application/json") - if err := json.NewEncoder(w).Encode(filteredHistory); err != nil { + if err := json.NewEncoder(w).Encode(history); err != nil { log.Printf("failed to encode filtered history: %v", err) http.Error(w, "failed to encode history", http.StatusInternalServerError) return @@ -294,6 +309,36 @@ func (c *Collector) HandleHistoryByType(dataType string) http.HandlerFunc { } } +// HandleTraceByID returns all stored entries referencing the given trace id, +// so agents don't have to rely on the trace id happening to substring-match +// inside a search result. +func (c *Collector) HandleTraceByID(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + + traceID := r.PathValue("traceId") + if traceID == "" { + http.Error(w, "traceId is required", http.StatusBadRequest) + return + } + + entries, err := c.SearchHistory(r.Context(), storage.SearchOptions{Query: traceID}, parseLimit(r, 1000)) + if err != nil { + log.Printf("failed to fetch trace %s: %v", traceID, err) + http.Error(w, "failed to fetch trace", http.StatusInternalServerError) + return + } + + w.Header().Set("Content-Type", "application/json") + if err := json.NewEncoder(w).Encode(entries); err != nil { + log.Printf("failed to encode trace: %v", err) + http.Error(w, "failed to encode trace", http.StatusInternalServerError) + return + } +} + // HandleHealth returns the health status of the collector. func (c *Collector) HandleHealth(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") diff --git a/internal/collector/ratelimit.go b/internal/collector/ratelimit.go new file mode 100644 index 0000000..03f464d --- /dev/null +++ b/internal/collector/ratelimit.go @@ -0,0 +1,81 @@ +package collector + +import ( + "net" + "net/http" + "strings" + "sync" + "time" + + "golang.org/x/time/rate" +) + +// ipRateLimiter tracks a token-bucket rate limiter per client IP, so a single +// caller can't monopolize a blocking endpoint like /api/history/wait while +// leaving other clients unaffected. Entries are evicted after being idle. +type ipRateLimiter struct { + rps rate.Limit + burst int + + mu sync.Mutex + limiters map[string]*rate.Limiter + lastSeen map[string]time.Time +} + +func newIPRateLimiter(rps rate.Limit, burst int) *ipRateLimiter { + l := &ipRateLimiter{ + rps: rps, + burst: burst, + limiters: make(map[string]*rate.Limiter), + lastSeen: make(map[string]time.Time), + } + go l.evictLoop() + return l +} + +// Allow reports whether a request from ip may proceed. +func (l *ipRateLimiter) Allow(ip string) bool { + l.mu.Lock() + limiter, ok := l.limiters[ip] + if !ok { + limiter = rate.NewLimiter(l.rps, l.burst) + l.limiters[ip] = limiter + } + l.lastSeen[ip] = time.Now() + l.mu.Unlock() + + return limiter.Allow() +} + +func (l *ipRateLimiter) evictLoop() { + ticker := time.NewTicker(10 * time.Minute) + defer ticker.Stop() + for range ticker.C { + cutoff := time.Now().Add(-10 * time.Minute) + l.mu.Lock() + for ip, seen := range l.lastSeen { + if seen.Before(cutoff) { + delete(l.limiters, ip) + delete(l.lastSeen, ip) + } + } + l.mu.Unlock() + } +} + +// clientIP extracts the caller's IP for rate limiting purposes, preferring +// X-Forwarded-For (set by the ingress/proxy in front of this service) and +// falling back to the raw connection address. +func clientIP(r *http.Request) string { + if fwd := r.Header.Get("X-Forwarded-For"); fwd != "" { + if idx := strings.IndexByte(fwd, ','); idx != -1 { + return fwd[:idx] + } + return fwd + } + host, _, err := net.SplitHostPort(r.RemoteAddr) + if err != nil { + return r.RemoteAddr + } + return host +} diff --git a/internal/storage/file_store.go b/internal/storage/file_store.go index 3497dbd..e943bdf 100644 --- a/internal/storage/file_store.go +++ b/internal/storage/file_store.go @@ -42,11 +42,40 @@ func (fs *FileStore) Append(ctx context.Context, entry websocket.HistoryEntry) e // GetHistory retrieves historical entries from the file store. func (fs *FileStore) GetHistory(ctx context.Context, limit int) ([]websocket.HistoryEntry, error) { - return fs.Search(ctx, "", limit) + return fs.Search(ctx, SearchOptions{}, limit) } -// Search searches for historical entries in the file store matching the query. -func (fs *FileStore) Search(ctx context.Context, query string, limit int) ([]websocket.HistoryEntry, error) { +// entryHead is used to cheaply inspect an entry's type/timestamp without +// unmarshaling the (potentially large) Data payload. +type entryHead struct { + Type string `json:"type"` + Timestamp int64 `json:"timestamp"` +} + +// matches reports whether the given line satisfies the type/time filters in opts. +// The free-text Query filter is applied separately by the caller against the raw line. +func (opts SearchOptions) matches(line string) bool { + if opts.Type == "" && opts.Since == 0 && opts.Until == 0 { + return true + } + var head entryHead + if err := json.Unmarshal([]byte(line), &head); err != nil { + return false + } + if opts.Type != "" && head.Type != opts.Type { + return false + } + if opts.Since != 0 && head.Timestamp < opts.Since { + return false + } + if opts.Until != 0 && head.Timestamp > opts.Until { + return false + } + return true +} + +// Search searches for historical entries in the file store matching opts. +func (fs *FileStore) Search(ctx context.Context, opts SearchOptions, limit int) ([]websocket.HistoryEntry, error) { if limit <= 0 { limit = 50 // Default to a manageable number } @@ -70,16 +99,20 @@ func (fs *FileStore) Search(ctx context.Context, query string, limit int) ([]web scanner.Buffer(buf, maxCapacity) // Lowercase query for case-insensitive search - lowerQuery := strings.ToLower(query) + lowerQuery := strings.ToLower(opts.Query) for scanner.Scan() { line := scanner.Text() - if query == "" || strings.Contains(strings.ToLower(line), lowerQuery) { - matchingLines = append(matchingLines, line) - // Only keep the most recent 'limit' matches to save memory - if len(matchingLines) > limit { - matchingLines = matchingLines[1:] - } + if opts.Query != "" && !strings.Contains(strings.ToLower(line), lowerQuery) { + continue + } + if !opts.matches(line) { + continue + } + matchingLines = append(matchingLines, line) + // Only keep the most recent 'limit' matches to save memory + if len(matchingLines) > limit { + matchingLines = matchingLines[1:] } } diff --git a/internal/storage/file_store_test.go b/internal/storage/file_store_test.go index e39961a..4e27522 100644 --- a/internal/storage/file_store_test.go +++ b/internal/storage/file_store_test.go @@ -50,7 +50,7 @@ func TestFileStore(t *testing.T) { } // Test Search - results, err := fs.Search(ctx, "hello", 10) + results, err := fs.Search(ctx, SearchOptions{Query: "hello"}, 10) if err != nil { t.Fatalf("Search failed: %v", err) } @@ -64,3 +64,56 @@ func TestFileStore(t *testing.T) { t.Errorf("unexpected data content: %s", string(data)) } } + +func TestFileStoreSearchFilters(t *testing.T) { + filename := "test-otel-filters.log" + defer func() { + _ = os.Remove(filename) + }() + + fs := NewFileStore(filename) + ctx := context.Background() + + entries := []websocket.HistoryEntry{ + {Type: "logs", Timestamp: 100, Data: json.RawMessage(`{"msg":"first log"}`)}, + {Type: "traces", Timestamp: 200, Data: json.RawMessage(`{"msg":"a trace"}`)}, + {Type: "logs", Timestamp: 300, Data: json.RawMessage(`{"msg":"second log"}`)}, + } + for _, e := range entries { + if err := fs.Append(ctx, e); err != nil { + t.Fatalf("failed to append entry: %v", err) + } + } + + // Filter by type only + results, err := fs.Search(ctx, SearchOptions{Type: "logs"}, 10) + if err != nil { + t.Fatalf("Search failed: %v", err) + } + if len(results) != 2 { + t.Fatalf("expected 2 log entries, got %d", len(results)) + } + for _, r := range results { + if r.Type != "logs" { + t.Errorf("expected type 'logs', got %s", r.Type) + } + } + + // Filter by time range: only the entry at Timestamp=200 should match + results, err = fs.Search(ctx, SearchOptions{Since: 150, Until: 250}, 10) + if err != nil { + t.Fatalf("Search failed: %v", err) + } + if len(results) != 1 || results[0].Timestamp != 200 { + t.Fatalf("expected single entry at timestamp 200, got %+v", results) + } + + // Combine type and query filters + results, err = fs.Search(ctx, SearchOptions{Type: "logs", Query: "second"}, 10) + if err != nil { + t.Fatalf("Search failed: %v", err) + } + if len(results) != 1 || results[0].Timestamp != 300 { + t.Fatalf("expected single entry at timestamp 300, got %+v", results) + } +} diff --git a/internal/storage/store.go b/internal/storage/store.go index 67f0bd5..be1526d 100644 --- a/internal/storage/store.go +++ b/internal/storage/store.go @@ -5,9 +5,24 @@ import ( "git.tornberg.me/go-otel/internal/websocket" ) +// SearchOptions narrows a Search call beyond the free-text query. +type SearchOptions struct { + // Query is a case-insensitive substring match against the raw entry JSON. + Query string + // Type restricts results to a single telemetry type (traces, metrics, logs). + // Empty matches any type. + Type string + // Since restricts results to entries with Timestamp >= Since (unix nanoseconds). + // Zero means no lower bound. + Since int64 + // Until restricts results to entries with Timestamp <= Until (unix nanoseconds). + // Zero means no upper bound. + Until int64 +} + // Store interface for telemetry history persistence. type Store interface { Append(ctx context.Context, entry websocket.HistoryEntry) error GetHistory(ctx context.Context, limit int) ([]websocket.HistoryEntry, error) - Search(ctx context.Context, query string, limit int) ([]websocket.HistoryEntry, error) + Search(ctx context.Context, opts SearchOptions, limit int) ([]websocket.HistoryEntry, error) }