// Package collector implements an OpenTelemetry consumer that stores telemetry in history // and broadcasts it to connected clients. package collector import ( "context" "encoding/base64" "encoding/json" "fmt" "io" "log" "net/http" "strconv" "strings" "time" "git.tornberg.me/go-otel/internal/issues" "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" "go.opentelemetry.io/collector/pdata/ptrace" ) // checkIngest applies the shared ingest rate limit and, if configured, the // ingest key check. It writes an error response and returns false when the // request should be rejected. func (c *Collector) checkIngest(w http.ResponseWriter, r *http.Request) bool { if !c.ingestLimiter.Allow(clientIP(r)) { http.Error(w, "too many requests, slow down", http.StatusTooManyRequests) return false } if c.ingestKey == "" { return true } key := r.Header.Get("X-Otel-Ingest-Key") if key == "" { key = r.URL.Query().Get("key") } if key != c.ingestKey { http.Error(w, "invalid or missing ingest key", http.StatusUnauthorized) return false } return true } // HandleTraces handles OTLP trace requests. func (c *Collector) HandleTraces(w http.ResponseWriter, r *http.Request) { log.Printf("Received traces request: method=%s, content-type=%s", r.Method, r.Header.Get("Content-Type")) if r.Method != http.MethodPost { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) return } if !c.checkIngest(w, r) { return } body, err := io.ReadAll(r.Body) if err != nil { log.Printf("failed to read request body: %v", err) http.Error(w, "failed to read request body", http.StatusInternalServerError) return } log.Printf("Received traces body, size=%d", len(body)) var unmarshaler ptrace.Unmarshaler contentType := r.Header.Get("Content-Type") switch contentType { case "application/x-protobuf": unmarshaler = &ptrace.ProtoUnmarshaler{} case "application/json": unmarshaler = &ptrace.JSONUnmarshaler{} default: unmarshaler = &ptrace.JSONUnmarshaler{} // default to JSON } traces, err := unmarshaler.UnmarshalTraces(body) if err != nil { log.Printf("failed to unmarshal traces: %v", err) http.Error(w, "failed to unmarshal traces", http.StatusBadRequest) return } log.Printf("Successfully unmarshaled traces, span_count=%d", traces.SpanCount()) if err := c.ConsumeTraces(context.Background(), traces); err != nil { log.Printf("failed to consume traces: %v", err) http.Error(w, "failed to consume traces", http.StatusInternalServerError) return } w.WriteHeader(http.StatusOK) } // HandleMetrics handles OTLP metric requests. func (c *Collector) HandleMetrics(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) return } if !c.checkIngest(w, r) { return } body, err := io.ReadAll(r.Body) if err != nil { log.Printf("failed to read request body: %v", err) http.Error(w, "failed to read request body", http.StatusInternalServerError) return } var unmarshaler pmetric.Unmarshaler contentType := r.Header.Get("Content-Type") switch contentType { case "application/x-protobuf": unmarshaler = &pmetric.ProtoUnmarshaler{} case "application/json": unmarshaler = &pmetric.JSONUnmarshaler{} default: unmarshaler = &pmetric.JSONUnmarshaler{} // default to JSON } metrics, err := unmarshaler.UnmarshalMetrics(body) if err != nil { log.Printf("failed to unmarshal metrics: %v", err) http.Error(w, "failed to unmarshal metrics", http.StatusBadRequest) return } if err := c.ConsumeMetrics(context.Background(), metrics); err != nil { log.Printf("failed to consume metrics: %v", err) http.Error(w, "failed to consume metrics", http.StatusInternalServerError) return } w.WriteHeader(http.StatusOK) } // HandleLogs handles OTLP log requests. func (c *Collector) HandleLogs(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) return } if !c.checkIngest(w, r) { return } body, err := io.ReadAll(r.Body) if err != nil { log.Printf("failed to read request body: %v", err) http.Error(w, "failed to read request body", http.StatusInternalServerError) return } var unmarshaler plog.Unmarshaler contentType := r.Header.Get("Content-Type") switch contentType { case "application/x-protobuf": unmarshaler = &plog.ProtoUnmarshaler{} case "application/json": unmarshaler = &plog.JSONUnmarshaler{} default: unmarshaler = &plog.JSONUnmarshaler{} // default to JSON } logs, err := unmarshaler.UnmarshalLogs(body) if err != nil { log.Printf("failed to unmarshal logs: %v", err) http.Error(w, "failed to unmarshal logs", http.StatusBadRequest) return } if err := c.ConsumeLogs(context.Background(), logs); err != nil { log.Printf("failed to consume logs: %v", err) http.Error(w, "failed to consume logs", http.StatusInternalServerError) return } w.WriteHeader(http.StatusOK) } // 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 } 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 } w.Header().Set("Content-Type", "application/json") if err := json.NewEncoder(w).Encode(history); err != nil { log.Printf("failed to encode history: %v", err) http.Error(w, "failed to encode history", http.StatusInternalServerError) return } } // 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 } 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) return } w.Header().Set("Content-Type", "application/json") if err := json.NewEncoder(w).Encode(history); err != nil { log.Printf("failed to encode search results: %v", err) http.Error(w, "failed to encode history", http.StatusInternalServerError) return } } // 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 } 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 { timeoutSec = ts } } ctx, cancel := context.WithTimeout(r.Context(), time.Duration(timeoutSec)*time.Second) defer cancel() // First, check if result already exists in recent history 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 { log.Printf("failed to encode match: %v", err) } return } // Otherwise, subscribe and wait lowerQuery := strings.ToLower(opts.Query) ch, cleanup := c.Subscribe(func(entry websocket.HistoryEntry) bool { if opts.Type != "" && entry.Type != opts.Type { return false } if opts.Query == "" { return true } return strings.Contains(strings.ToLower(string(entry.Data)), lowerQuery) }) defer cleanup() select { case entry := <-ch: w.Header().Set("Content-Type", "application/json") if err := json.NewEncoder(w).Encode(entry); err != nil { log.Printf("failed to encode signal: %v", err) } case <-ctx.Done(): http.Error(w, "timeout waiting for signal", http.StatusGatewayTimeout) } } // 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 { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) return } 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(history); err != nil { log.Printf("failed to encode filtered history: %v", err) http.Error(w, "failed to encode history", http.StatusInternalServerError) return } } } // 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 } } // HandleIssues returns deduplicated error issues derived from ingested // traces and logs, most recently seen first. func (c *Collector) HandleIssues(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) return } opts, err := parseIssueListOptions(r) if err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } page := c.ListIssuesFiltered(opts) w.Header().Set("Content-Type", "application/json") if page.HasMore && len(page.Issues) > 0 { last := page.Issues[len(page.Issues)-1] next := encodeIssueCursor(last.LastSeen, last.ID) w.Header().Set("X-Next-Cursor", next) nextQuery := r.URL.Query() nextQuery.Set("cursor", next) w.Header().Set("Link", fmt.Sprintf("<%s?%s>; rel=next", r.URL.Path, nextQuery.Encode())) } // Keep the response as a JSON array for compatibility. Pagination metadata // is exposed in headers so existing UI clients do not need a migration. if err := json.NewEncoder(w).Encode(page.Issues); err != nil { log.Printf("failed to encode issues: %v", err) http.Error(w, "failed to encode issues", http.StatusInternalServerError) return } } const maxIssuePageSize = 200 type issueCursor struct { LastSeen int64 `json:"lastSeen"` ID string `json:"id"` } func parseIssueListOptions(r *http.Request) (issues.ListOptions, error) { q := r.URL.Query() opts := issues.ListOptions{ Service: q.Get("service"), Kind: q.Get("kind"), Status: q.Get("status"), Title: q.Get("q"), SourceIP: q.Get("sourceIp"), Route: q.Get("route"), Limit: parseLimit(r, 100), } if opts.Limit <= 0 { opts.Limit = 100 } if opts.Limit > maxIssuePageSize { opts.Limit = maxIssuePageSize } for name, dst := range map[string]*int64{"since": &opts.Since, "until": &opts.Until} { if value := q.Get(name); value != "" { parsed, err := strconv.ParseInt(value, 10, 64) if err != nil || parsed < 0 { return issues.ListOptions{}, fmt.Errorf("invalid %s; expected unix nanoseconds", name) } *dst = parsed } } if opts.Since != 0 && opts.Until != 0 && opts.Since > opts.Until { return issues.ListOptions{}, fmt.Errorf("since must be less than or equal to until") } if value := q.Get("cursor"); value != "" { decoded, err := base64.RawURLEncoding.DecodeString(value) if err != nil { return issues.ListOptions{}, fmt.Errorf("invalid cursor") } var cursor issueCursor if err := json.Unmarshal(decoded, &cursor); err != nil || cursor.LastSeen <= 0 || cursor.ID == "" { return issues.ListOptions{}, fmt.Errorf("invalid cursor") } opts.AfterLastSeen = cursor.LastSeen opts.AfterID = cursor.ID } return opts, nil } func encodeIssueCursor(lastSeen int64, id string) string { data, _ := json.Marshal(issueCursor{LastSeen: lastSeen, ID: id}) return base64.RawURLEncoding.EncodeToString(data) } // HandleSourceMap accepts a JavaScript source map for a deployed release. // Multipart fields: release, source, file. func (c *Collector) HandleSourceMap(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) return } if c.sourceMaps == nil { http.Error(w, "source maps are not configured", http.StatusNotImplemented) return } if err := r.ParseMultipartForm(50 << 20); err != nil { http.Error(w, "invalid multipart body", http.StatusBadRequest) return } release, source := r.FormValue("release"), r.FormValue("source") file, _, err := r.FormFile("file") if err != nil { http.Error(w, "file is required", http.StatusBadRequest) return } defer file.Close() if err := c.sourceMaps.Upload(release, source, file); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } w.WriteHeader(http.StatusCreated) } // HandleIssueByID returns a single issue by id, including its samples. func (c *Collector) HandleIssueByID(w http.ResponseWriter, r *http.Request) { id := r.PathValue("id") switch r.Method { case http.MethodGet: issue, ok := c.GetIssue(id) if !ok { http.Error(w, "issue not found", http.StatusNotFound) return } w.Header().Set("Content-Type", "application/json") if err := json.NewEncoder(w).Encode(issue); err != nil { log.Printf("failed to encode issue: %v", err) http.Error(w, "failed to encode issue", http.StatusInternalServerError) } case http.MethodPatch: c.HandleIssueStatus(w, r, id) default: http.Error(w, "method not allowed", http.StatusMethodNotAllowed) } } // HandlePatterns returns deterministic related-issue groups for visualization. func (c *Collector) HandlePatterns(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) return } w.Header().Set("Content-Type", "application/json") _ = json.NewEncoder(w).Encode(c.ListPatterns()) } // HandleRollouts returns normalized Kubernetes Deployment rollout history. func (c *Collector) HandleRollouts(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) return } if !c.KubernetesEnabled() { jsonError(w, http.StatusNotImplemented, c.KubernetesError()) return } w.Header().Set("Content-Type", "application/json") _ = json.NewEncoder(w).Encode(c.ListRollouts()) } func (c *Collector) HandleRolloutByID(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) return } if !c.KubernetesEnabled() { jsonError(w, http.StatusNotImplemented, c.KubernetesError()) return } rollout, ok := c.GetRollout(r.PathValue("id")) if !ok { http.Error(w, "rollout not found", http.StatusNotFound) return } w.Header().Set("Content-Type", "application/json") _ = json.NewEncoder(w).Encode(rollout) } func (c *Collector) HandleRolloutImpact(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) return } if !c.KubernetesEnabled() { jsonError(w, http.StatusNotImplemented, c.KubernetesError()) return } rollout, ok := c.GetRollout(r.PathValue("id")) if !ok { http.Error(w, "rollout not found", http.StatusNotFound) return } response := map[string]interface{}{"configured": c.prometheus != nil, "rollout": rollout, "points": []interface{}{}} if c.prometheus != nil { start := time.Unix(0, rollout.StartedAt).Add(-15 * time.Minute) end := time.Now() if start.Before(end.Add(-24 * time.Hour)) { start = end.Add(-24 * time.Hour) } points, err := c.prometheus.ErrorRate(r.Context(), rollout.Service, rollout.Release, start, end) if err != nil { jsonError(w, http.StatusBadGateway, err.Error()) return } response["points"] = points } w.Header().Set("Content-Type", "application/json") _ = json.NewEncoder(w).Encode(response) } func jsonError(w http.ResponseWriter, status int, message string) { w.Header().Set("Content-Type", "application/json") w.WriteHeader(status) _ = json.NewEncoder(w).Encode(map[string]string{"error": message}) } // HandleIssueSimilar returns explainable, non-merging related-issue suggestions. func (c *Collector) HandleIssueSimilar(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) return } limit := parseLimit(r, 10) result, ok := c.SimilarIssues(r.PathValue("id"), limit) if !ok { http.Error(w, "issue not found", http.StatusNotFound) return } w.Header().Set("Content-Type", "application/json") _ = json.NewEncoder(w).Encode(result) } // HandleSimilarityFeedback records a human review label for a pair of issues. // Body: {"issueId":"...", "decision":"same|different"}. func (c *Collector) HandleSimilarityFeedback(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) return } var req struct { IssueID string `json:"issueId"` Decision string `json:"decision"` } if err := json.NewDecoder(io.LimitReader(r.Body, 4096)).Decode(&req); err != nil { http.Error(w, "invalid JSON body", http.StatusBadRequest) return } if !c.SetSimilarityFeedback(r.PathValue("id"), req.IssueID, req.Decision) { http.Error(w, "invalid issue pair or decision", http.StatusBadRequest) return } w.Header().Set("Content-Type", "application/json") _ = json.NewEncoder(w).Encode(map[string]string{"status": "recorded", "decision": req.Decision}) } // HandleIssueStatus updates the triage state (open/resolved/ignored) of an // issue. Body: {"status": "resolved"}. func (c *Collector) HandleIssueStatus(w http.ResponseWriter, r *http.Request, id string) { var req struct { Status string `json:"status"` } body, err := io.ReadAll(io.LimitReader(r.Body, 4096)) if err != nil { http.Error(w, "failed to read request body", http.StatusBadRequest) return } if err := json.Unmarshal(body, &req); err != nil { http.Error(w, "invalid JSON body", http.StatusBadRequest) return } issue, ok := c.SetIssueStatus(id, req.Status) if !ok { http.Error(w, "issue not found or invalid status", http.StatusBadRequest) return } w.Header().Set("Content-Type", "application/json") if err := json.NewEncoder(w).Encode(issue); err != nil { log.Printf("failed to encode issue: %v", err) } } // 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") w.WriteHeader(http.StatusOK) if err := json.NewEncoder(w).Encode(map[string]string{"status": "ok"}); err != nil { log.Printf("failed to encode health check: %v", err) } } // statsEnabled reports whether an aggregate-capable store is configured. func (c *Collector) statsEnabled() bool { return c.statsQuerier != nil } // statsNotEnabled writes the response used when the file store is active: // a stable 501 the UI treats as "view not available". func statsNotEnabled(w http.ResponseWriter) { w.Header().Set("Content-Type", "application/json") w.WriteHeader(http.StatusNotImplemented) _ = json.NewEncoder(w).Encode(map[string]string{ "error": "stats require OTEL_CLICKHOUSE_ADDR (ClickHouse storage) to be configured", }) } // parseStatsCommon reads the shared service/hours params for stats endpoints. func parseStatsCommon(r *http.Request) (service string, hours int) { service = r.URL.Query().Get("service") hours = 72 if h, err := strconv.Atoi(r.URL.Query().Get("hours")); err == nil && h > 0 && h <= 24*90 { hours = h } return service, hours } // HandleStatsErrors returns hourly/daily total and error counts for the // trends graphs. Only available with ClickHouse storage. func (c *Collector) HandleStatsErrors(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) return } if !c.statsEnabled() { statsNotEnabled(w) return } service, hours := parseStatsCommon(r) daily := hours > 24*14 // beyond two weeks, daily buckets read better points, err := c.statsQuerier.ErrorStats(r.Context(), service, hours, daily) if err != nil { log.Printf("failed to compute error stats: %v", err) http.Error(w, "failed to compute error stats", http.StatusInternalServerError) return } w.Header().Set("Content-Type", "application/json") _ = json.NewEncoder(w).Encode(map[string]interface{}{ "granularity": map[bool]string{true: "day", false: "hour"}[daily], "points": points, }) } // HandleStatsReleases returns per-release error-count time series so two // deploys can be compared line-by-line. Only available with ClickHouse. func (c *Collector) HandleStatsReleases(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) return } if !c.statsEnabled() { statsNotEnabled(w) return } service, hours := parseStatsCommon(r) series, err := c.statsQuerier.ReleaseSeries(r.Context(), service, hours) if err != nil { log.Printf("failed to compute release stats: %v", err) http.Error(w, "failed to compute release stats", http.StatusInternalServerError) return } w.Header().Set("Content-Type", "application/json") _ = json.NewEncoder(w).Encode(map[string]interface{}{ "series": series, }) } // HandleStatsServices lists the distinct services present in the rollup. func (c *Collector) HandleStatsServices(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) return } if !c.statsEnabled() { statsNotEnabled(w) return } services, err := c.statsQuerier.Services(r.Context()) if err != nil { log.Printf("failed to list services: %v", err) http.Error(w, "failed to list services", http.StatusInternalServerError) return } w.Header().Set("Content-Type", "application/json") _ = json.NewEncoder(w).Encode(map[string]interface{}{"services": services}) }