matsandClaude Sonnet 5.5
51a8412ae1
Needle scored ~chance on the golden set and never gated alerts, so drop the sidecar, proxy handlers, UI buttons, k8s container and build step. The labelled eval sets move to cmd/triage-eval/data. nginx 4xx (registry 401 challenges, /oauth2/auth, scanner 405s) is now dropped in FilterActionable instead of only being discounted, so it cannot ride along with a real 5xx. nginx 5xx still alerts. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
1036 lines
35 KiB
Go
1036 lines
35 KiB
Go
// Package collector implements an OpenTelemetry consumer that stores telemetry in history
|
|
// and broadcasts it to connected clients.
|
|
package collector
|
|
|
|
import (
|
|
"context"
|
|
"crypto/subtle"
|
|
"encoding/base64"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"fmt"
|
|
"git.tornberg.me/go-otel/internal/sourcemaps"
|
|
"io"
|
|
"log"
|
|
"net/http"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"git.tornberg.me/go-otel/internal/features"
|
|
"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/pcommon"
|
|
"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
|
|
}
|
|
|
|
stampFrontendClient(logs, r)
|
|
|
|
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
|
|
}
|
|
|
|
var entries []websocket.HistoryEntry
|
|
var err error
|
|
if c.traceQuerier != nil {
|
|
// Indexed lookup, payloads trimmed to this trace's spans/logs.
|
|
entries, err = c.traceQuerier.TraceEntries(r.Context(), traceID, parseLimit(r, 1000))
|
|
} else {
|
|
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) {
|
|
traceID, spanID := traceContextFromRequest(r)
|
|
logUpload := func(severity plog.SeverityNumber, body string, attrs map[string]any) {
|
|
if err := c.EmitInternalLog(r.Context(), severity, body, attrs, traceID, spanID); err != nil {
|
|
log.Printf("failed to emit sourcemap diagnostic: %v", err)
|
|
}
|
|
}
|
|
logUpload(plog.SeverityNumberInfo, "sourcemap upload received", map[string]any{
|
|
"event.name": "sourcemap.upload",
|
|
"http.route": "/api/sourcemaps",
|
|
"http.request.method": r.Method,
|
|
})
|
|
if r.Method != http.MethodPost {
|
|
logUpload(plog.SeverityNumberWarn, "sourcemap upload rejected: method not allowed", map[string]any{
|
|
"event.name": "sourcemap.upload.rejected",
|
|
"sourcemap.failure_stage": "method",
|
|
})
|
|
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
|
|
return
|
|
}
|
|
if !c.checkSourceMapKey(r) {
|
|
logUpload(plog.SeverityNumberWarn, "sourcemap upload rejected: invalid or missing key", map[string]any{
|
|
"event.name": "sourcemap.upload.rejected",
|
|
"sourcemap.failure_stage": "auth",
|
|
"client.address": clientIP(r),
|
|
})
|
|
http.Error(w, "invalid or missing sourcemap key", http.StatusUnauthorized)
|
|
return
|
|
}
|
|
if c.sourceMaps == nil {
|
|
logUpload(plog.SeverityNumberError, "sourcemap upload failed: resolver is not configured", map[string]any{
|
|
"event.name": "sourcemap.upload.failed",
|
|
"sourcemap.failure_stage": "configuration",
|
|
})
|
|
http.Error(w, "source maps are not configured", http.StatusNotImplemented)
|
|
return
|
|
}
|
|
if err := r.ParseMultipartForm(50 << 20); err != nil {
|
|
logUpload(plog.SeverityNumberError, "sourcemap upload failed: invalid multipart body", map[string]any{
|
|
"event.name": "sourcemap.upload.failed",
|
|
"sourcemap.failure_stage": "multipart",
|
|
"error.type": "multipart_parse",
|
|
})
|
|
http.Error(w, "invalid multipart body", http.StatusBadRequest)
|
|
return
|
|
}
|
|
release, source := r.FormValue("release"), r.FormValue("source")
|
|
file, fileHeader, err := r.FormFile("file")
|
|
if err != nil {
|
|
logUpload(plog.SeverityNumberError, "sourcemap upload failed: file is missing", map[string]any{
|
|
"event.name": "sourcemap.upload.failed",
|
|
"sourcemap.failure_stage": "file",
|
|
"sourcemap.release": release,
|
|
"sourcemap.source": source,
|
|
"error.type": "missing_file",
|
|
})
|
|
http.Error(w, "file is required", http.StatusBadRequest)
|
|
return
|
|
}
|
|
defer file.Close()
|
|
if err := c.sourceMaps.Upload(release, source, file); err != nil {
|
|
logUpload(plog.SeverityNumberError, "sourcemap upload failed", map[string]any{
|
|
"event.name": "sourcemap.upload.failed",
|
|
"sourcemap.failure_stage": "store",
|
|
"sourcemap.release": release,
|
|
"sourcemap.source": source,
|
|
"sourcemap.size_bytes": fileHeader.Size,
|
|
"error.type": "source_map_invalid_or_unwritable",
|
|
"error.message": err.Error(),
|
|
})
|
|
http.Error(w, err.Error(), http.StatusBadRequest)
|
|
return
|
|
}
|
|
logUpload(plog.SeverityNumberInfo, "sourcemap upload stored", map[string]any{
|
|
"event.name": "sourcemap.upload.stored",
|
|
"sourcemap.release": release,
|
|
"sourcemap.source": source,
|
|
"sourcemap.size_bytes": fileHeader.Size,
|
|
})
|
|
w.WriteHeader(http.StatusCreated)
|
|
}
|
|
|
|
// checkSourceMapKey authenticates a source map upload against the shared
|
|
// OTEL_SOURCEMAP_KEY (X-Sourcemap-Key header). Uploads are refused while no key
|
|
// is configured so the endpoint is never open by accident.
|
|
func (c *Collector) checkSourceMapKey(r *http.Request) bool {
|
|
if c.sourceMapKey == "" {
|
|
return false
|
|
}
|
|
got := r.Header.Get("X-Sourcemap-Key")
|
|
return subtle.ConstantTimeCompare([]byte(got), []byte(c.sourceMapKey)) == 1
|
|
}
|
|
|
|
// traceContextFromRequest preserves the caller's trace linkage for internal
|
|
// logs when the upload came through an instrumented client or proxy.
|
|
func traceContextFromRequest(r *http.Request) (pcommon.TraceID, pcommon.SpanID) {
|
|
var traceID pcommon.TraceID
|
|
var spanID pcommon.SpanID
|
|
parts := strings.Split(r.Header.Get("Traceparent"), "-")
|
|
if len(parts) != 4 || len(parts[1]) != 32 || len(parts[2]) != 16 {
|
|
return traceID, spanID
|
|
}
|
|
if _, err := hex.Decode(traceID[:], []byte(parts[1])); err != nil {
|
|
return pcommon.NewTraceIDEmpty(), pcommon.NewSpanIDEmpty()
|
|
}
|
|
if _, err := hex.Decode(spanID[:], []byte(parts[2])); err != nil {
|
|
return pcommon.NewTraceIDEmpty(), pcommon.NewSpanIDEmpty()
|
|
}
|
|
return traceID, spanID
|
|
}
|
|
|
|
// HandleListSourceMaps lists the release/source pairs currently loaded, so a
|
|
// deploy can verify its uploads (and that they survived a restart).
|
|
func (c *Collector) HandleListSourceMaps(w http.ResponseWriter, r *http.Request) {
|
|
list := []sourcemaps.Entry{}
|
|
if c.sourceMaps != nil {
|
|
list = c.sourceMaps.List()
|
|
}
|
|
w.Header().Set("Content-Type", "application/json")
|
|
_ = json.NewEncoder(w).Encode(list)
|
|
}
|
|
|
|
// HandleResolveStacks maps raw browser stacktraces back to source using the
|
|
// uploaded source maps. Body: {"items":[{"release":"...","stacktrace":"..."}]};
|
|
// the response has one resolved stacktrace per item, in order.
|
|
func (c *Collector) HandleResolveStacks(w http.ResponseWriter, r *http.Request) {
|
|
traceID, spanID := traceContextFromRequest(r)
|
|
logResolve := func(severity plog.SeverityNumber, body string, attrs map[string]any) {
|
|
if err := c.EmitInternalLog(r.Context(), severity, body, attrs, traceID, spanID); err != nil {
|
|
log.Printf("failed to emit sourcemap diagnostic: %v", err)
|
|
}
|
|
}
|
|
logResolve(plog.SeverityNumberInfo, "sourcemap resolution received", map[string]any{
|
|
"event.name": "sourcemap.resolve",
|
|
"http.route": "/api/sourcemaps/resolve",
|
|
"http.request.method": r.Method,
|
|
"sourcemap.resolver_ready": c.sourceMaps != nil,
|
|
})
|
|
var req struct {
|
|
Items []struct {
|
|
Release string `json:"release"`
|
|
Stacktrace string `json:"stacktrace"`
|
|
} `json:"items"`
|
|
}
|
|
if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 4<<20)).Decode(&req); err != nil {
|
|
logResolve(plog.SeverityNumberError, "sourcemap resolution failed: invalid body", map[string]any{
|
|
"event.name": "sourcemap.resolve.failed",
|
|
"sourcemap.failure_stage": "decode",
|
|
"error.type": "invalid_json",
|
|
})
|
|
http.Error(w, "invalid body", http.StatusBadRequest)
|
|
return
|
|
}
|
|
out := make([]string, len(req.Items))
|
|
frames := make([][]sourcemaps.Frame, len(req.Items))
|
|
statusCounts := map[string]int{}
|
|
for i, it := range req.Items {
|
|
out[i] = it.Stacktrace
|
|
frames[i] = []sourcemaps.Frame{}
|
|
if c.sourceMaps != nil {
|
|
out[i] = c.sourceMaps.ResolveStack(it.Release, it.Stacktrace)
|
|
frames[i] = c.sourceMaps.ResolveFrames(it.Release, it.Stacktrace)
|
|
}
|
|
for _, frame := range frames[i] {
|
|
statusCounts[frame.Status]++
|
|
}
|
|
}
|
|
logResolve(plog.SeverityNumberInfo, "sourcemap resolution completed", map[string]any{
|
|
"event.name": "sourcemap.resolve.completed",
|
|
"sourcemap.item_count": len(req.Items),
|
|
"sourcemap.frame_count": statusCounts["mapped"] + statusCounts["no-release"] + statusCounts["no-map"] + statusCounts["unmapped"],
|
|
"sourcemap.mapped": statusCounts["mapped"],
|
|
"sourcemap.no_release": statusCounts["no-release"],
|
|
"sourcemap.no_map": statusCounts["no-map"],
|
|
"sourcemap.unmapped": statusCounts["unmapped"],
|
|
"sourcemap.resolver_ready": c.sourceMaps != nil,
|
|
})
|
|
w.Header().Set("Content-Type", "application/json")
|
|
_ = json.NewEncoder(w).Encode(map[string]any{"resolved": out, "frames": frames})
|
|
}
|
|
|
|
// 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)
|
|
}
|
|
}
|
|
|
|
// HandleMLDetectorTest runs a deterministic, in-memory spike through the
|
|
// production feature and label builder. It stores no telemetry and exists so
|
|
// an operator can verify the detector wiring from the authenticated UI.
|
|
func (c *Collector) HandleMLDetectorTest(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodPost {
|
|
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
|
|
return
|
|
}
|
|
const service = "ml-verification"
|
|
base := time.Date(2026, 1, 8, 12, 0, 0, 0, time.UTC)
|
|
events := make([]features.Event, 0, 100)
|
|
for day := 1; day <= 7; day++ {
|
|
for i := 0; i < 10; i++ {
|
|
events = append(events, features.Event{Timestamp: base.AddDate(0, 0, -day).Add(time.Duration(i) * time.Second), Service: service, Kind: "span", IsError: i == 0, DurationMS: float64(i + 1)})
|
|
}
|
|
}
|
|
for i := 0; i < 10; i++ {
|
|
events = append(events, features.Event{Timestamp: base.Add(time.Duration(i) * time.Second), Service: service, Release: "verification", Kind: "span", DurationMS: float64(i + 1)})
|
|
events = append(events, features.Event{Timestamp: base.Add(5 * time.Minute).Add(time.Duration(i) * time.Second), Service: service, Release: "verification", Kind: "log", IsError: i < 5, IssueFingerprint: "verification-spike"})
|
|
}
|
|
rows := features.Build(events, features.Config{ObservedUntil: base.Add(20 * time.Minute)})
|
|
for _, row := range rows {
|
|
if row.WindowStart.Equal(base) && row.Service == service && row.Release == "verification" {
|
|
w.Header().Set("Content-Type", "application/json")
|
|
_ = json.NewEncoder(w).Encode(map[string]interface{}{
|
|
"passed": row.LabelKnown && row.Label,
|
|
"scenario": "five errors in ten future events against a seven-day 10% baseline",
|
|
"row": row,
|
|
})
|
|
return
|
|
}
|
|
}
|
|
jsonError(w, http.StatusInternalServerError, "detector test did not generate its verification row")
|
|
}
|
|
|
|
// 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})
|
|
}
|
|
|
|
// HandleStatsEndpoints returns normalized HTTP traffic groups. It is useful
|
|
// for diagnosing noisy hosts and proxy/instrumentation duplication without
|
|
// loading raw telemetry into the browser.
|
|
func (c *Collector) HandleStatsEndpoints(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)
|
|
host := strings.ToLower(strings.TrimSpace(r.URL.Query().Get("host")))
|
|
limit := parseLimit(r, 100)
|
|
if limit <= 0 {
|
|
limit = 100
|
|
}
|
|
if limit > 500 {
|
|
limit = 500
|
|
}
|
|
stats, err := c.statsQuerier.EndpointStats(r.Context(), host, service, hours, limit)
|
|
if err != nil {
|
|
log.Printf("failed to compute endpoint stats: %v", err)
|
|
http.Error(w, "failed to compute endpoint stats", http.StatusInternalServerError)
|
|
return
|
|
}
|
|
w.Header().Set("Content-Type", "application/json")
|
|
_ = json.NewEncoder(w).Encode(map[string]interface{}{
|
|
"hours": hours,
|
|
"host": host,
|
|
"endpoints": stats,
|
|
})
|
|
}
|
|
|
|
// stampFrontendClient fills in client.address and user_agent.original on
|
|
// browser-originated log records (source=frontend) from the ingest request,
|
|
// since the browser cannot know its own public IP. Values the SDK already
|
|
// sent are kept, and backend services are left alone.
|
|
func stampFrontendClient(logs plog.Logs, r *http.Request) {
|
|
ip := strings.TrimSpace(clientIP(r))
|
|
ua := r.Header.Get("User-Agent")
|
|
rls := logs.ResourceLogs()
|
|
for i := 0; i < rls.Len(); i++ {
|
|
sls := rls.At(i).ScopeLogs()
|
|
for j := 0; j < sls.Len(); j++ {
|
|
recs := sls.At(j).LogRecords()
|
|
for k := 0; k < recs.Len(); k++ {
|
|
a := recs.At(k).Attributes()
|
|
if v, ok := a.Get("source"); !ok || v.Str() != "frontend" {
|
|
continue
|
|
}
|
|
if _, ok := a.Get("client.address"); !ok && ip != "" {
|
|
a.PutStr("client.address", ip)
|
|
}
|
|
if _, ok := a.Get("browser.userAgent"); !ok && ua != "" {
|
|
a.PutStr("user_agent.original", ua)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|