243 lines
7.5 KiB
Go
243 lines
7.5 KiB
Go
package collector
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"log"
|
|
"net/http"
|
|
"sort"
|
|
"strconv"
|
|
"strings"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"git.tornberg.me/go-otel/internal/issues"
|
|
"git.tornberg.me/go-otel/internal/storage"
|
|
)
|
|
|
|
// occurrenceBuffer bounds queued occurrence writes; beyond it, writes are
|
|
// dropped (and counted in the log) rather than slowing ingest.
|
|
const occurrenceBuffer = 4096
|
|
|
|
// initOptionalStores wires the store's optional capabilities: trace lookups
|
|
// and persistent issue occurrences.
|
|
func (c *Collector) initOptionalStores(store storage.Store) {
|
|
if tq, ok := store.(storage.TraceQuerier); ok {
|
|
c.traceQuerier = tq
|
|
}
|
|
occ, ok := store.(storage.OccurrenceStore)
|
|
if !ok {
|
|
return
|
|
}
|
|
c.occStore = occ
|
|
c.occChan = make(chan storage.Occurrence, occurrenceBuffer)
|
|
go func() {
|
|
for o := range c.occChan {
|
|
if err := occ.AppendOccurrence(context.Background(), o); err != nil {
|
|
log.Printf("failed to persist issue occurrence: %v", err)
|
|
}
|
|
}
|
|
}()
|
|
var dropped atomic.Int64
|
|
c.issues.SetOccurrenceSink(func(issueID, kind, service string, sample issues.Sample) {
|
|
raw, err := json.Marshal(sample)
|
|
if err != nil {
|
|
return
|
|
}
|
|
select {
|
|
case c.occChan <- storage.Occurrence{
|
|
IssueID: issueID, Kind: kind, Service: service, Timestamp: sample.Timestamp,
|
|
TraceID: sample.TraceID, SpanID: sample.SpanID, Sample: raw,
|
|
}:
|
|
default:
|
|
if n := dropped.Add(1); n%1000 == 1 {
|
|
log.Printf("issue occurrence buffer full; dropped %d so far", n)
|
|
}
|
|
}
|
|
})
|
|
}
|
|
|
|
// issueTraceJSON is one trace in the issue overview.
|
|
type issueTraceJSON struct {
|
|
TraceID string `json:"traceId"`
|
|
Hits int `json:"hits"`
|
|
FirstSeen int64 `json:"firstSeen"`
|
|
LastSeen int64 `json:"lastSeen"`
|
|
Summary *storage.TraceSummary `json:"summary,omitempty"`
|
|
Samples []issues.Sample `json:"samples"`
|
|
}
|
|
|
|
type issueTracesResponse struct {
|
|
IssueID string `json:"issueId"`
|
|
TotalOccurrences int `json:"totalOccurrences"`
|
|
Traces []issueTraceJSON `json:"traces"`
|
|
Untraced []issues.Sample `json:"untraced"`
|
|
HasMore bool `json:"hasMore"`
|
|
// NextBefore is the LastSeen boundary to pass as ?before= for the next page.
|
|
NextBefore int64 `json:"nextBefore,omitempty"`
|
|
// Complete is false when only the in-memory samples were available.
|
|
Complete bool `json:"complete"`
|
|
}
|
|
|
|
func decodeSamples(raw []json.RawMessage) []issues.Sample {
|
|
out := make([]issues.Sample, 0, len(raw))
|
|
for _, r := range raw {
|
|
var s issues.Sample
|
|
if json.Unmarshal(r, &s) == nil {
|
|
out = append(out, s)
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
// HandleIssueTraces returns every trace an issue occurred in with per-trace
|
|
// summaries (duration, spans, services, errors) and the occurrences in each,
|
|
// in a single request. Spans themselves stay on GET /api/traces/{id}.
|
|
func (c *Collector) HandleIssueTraces(w http.ResponseWriter, r *http.Request) {
|
|
id := r.PathValue("id")
|
|
issue, known := c.GetIssue(id)
|
|
|
|
var resp issueTracesResponse
|
|
resp.IssueID = id
|
|
if c.occStore != nil {
|
|
opts := storage.IssueTracesOptions{Limit: parseLimit(r, 50)}
|
|
if b, err := strconv.ParseInt(r.URL.Query().Get("before"), 10, 64); err == nil {
|
|
opts.Before = b
|
|
}
|
|
page, err := c.occStore.IssueTraces(r.Context(), id, opts)
|
|
if err != nil {
|
|
log.Printf("failed to load issue traces %s: %v", id, err)
|
|
http.Error(w, "failed to load issue traces", http.StatusInternalServerError)
|
|
return
|
|
}
|
|
if page.TotalOccurrences == 0 && !known {
|
|
http.Error(w, "issue not found", http.StatusNotFound)
|
|
return
|
|
}
|
|
resp.TotalOccurrences, resp.HasMore, resp.Complete = page.TotalOccurrences, page.HasMore, true
|
|
resp.Untraced = decodeSamples(page.Untraced)
|
|
for _, t := range page.Traces {
|
|
resp.Traces = append(resp.Traces, issueTraceJSON{
|
|
TraceID: t.TraceID, Hits: t.Hits, FirstSeen: t.FirstSeen, LastSeen: t.LastSeen,
|
|
Summary: t.Summary, Samples: decodeSamples(t.Samples),
|
|
})
|
|
}
|
|
if resp.HasMore && len(resp.Traces) > 0 {
|
|
resp.NextBefore = resp.Traces[len(resp.Traces)-1].LastSeen
|
|
}
|
|
} else {
|
|
if !known {
|
|
http.Error(w, "issue not found", http.StatusNotFound)
|
|
return
|
|
}
|
|
resp.TotalOccurrences = issue.Count
|
|
resp.Traces, resp.Untraced = groupInMemory(issue.Samples)
|
|
}
|
|
|
|
// Source-map resolution applies to stored samples exactly as it does for
|
|
// the in-memory ones.
|
|
if c.sourceMaps != nil {
|
|
for i := range resp.Traces {
|
|
resolveSamples(resp.Traces[i].Samples, c.sourceMaps)
|
|
}
|
|
resolveSamples(resp.Untraced, c.sourceMaps)
|
|
}
|
|
if resp.Traces == nil {
|
|
resp.Traces = []issueTraceJSON{}
|
|
}
|
|
if resp.Untraced == nil {
|
|
resp.Untraced = []issues.Sample{}
|
|
}
|
|
|
|
// Summaries for traces the store didn't summarise (fallback path, or
|
|
// traces older than the summary table): fill from the store when possible.
|
|
if c.traceQuerier != nil {
|
|
var missing []string
|
|
for _, t := range resp.Traces {
|
|
if t.Summary == nil {
|
|
missing = append(missing, t.TraceID)
|
|
}
|
|
}
|
|
if len(missing) > 0 {
|
|
if sums, err := c.traceQuerier.TraceSummaries(r.Context(), missing); err == nil {
|
|
for i := range resp.Traces {
|
|
if s, ok := sums[resp.Traces[i].TraceID]; ok {
|
|
resp.Traces[i].Summary = &s
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
w.Header().Set("Content-Type", "application/json")
|
|
if err := json.NewEncoder(w).Encode(resp); err != nil {
|
|
log.Printf("failed to encode issue traces: %v", err)
|
|
}
|
|
}
|
|
|
|
// groupInMemory builds the same shape from the tracker's bounded samples,
|
|
// newest trace first.
|
|
func groupInMemory(samples []issues.Sample) ([]issueTraceJSON, []issues.Sample) {
|
|
byTrace := map[string]*issueTraceJSON{}
|
|
var traces []*issueTraceJSON
|
|
var untraced []issues.Sample
|
|
for _, s := range samples {
|
|
if s.TraceID == "" {
|
|
untraced = append(untraced, s)
|
|
continue
|
|
}
|
|
t, ok := byTrace[s.TraceID]
|
|
if !ok {
|
|
t = &issueTraceJSON{TraceID: s.TraceID, FirstSeen: s.Timestamp, LastSeen: s.Timestamp}
|
|
byTrace[s.TraceID] = t
|
|
traces = append(traces, t)
|
|
}
|
|
t.Hits++
|
|
t.Samples = append(t.Samples, s)
|
|
if s.Timestamp < t.FirstSeen {
|
|
t.FirstSeen = s.Timestamp
|
|
}
|
|
if s.Timestamp > t.LastSeen {
|
|
t.LastSeen = s.Timestamp
|
|
}
|
|
}
|
|
sort.Slice(traces, func(i, j int) bool { return traces[i].LastSeen > traces[j].LastSeen })
|
|
out := make([]issueTraceJSON, len(traces))
|
|
for i, t := range traces {
|
|
// Newest first, matching the store-backed path.
|
|
sort.SliceStable(t.Samples, func(a, b int) bool { return t.Samples[a].Timestamp > t.Samples[b].Timestamp })
|
|
out[i] = *t
|
|
}
|
|
sort.Slice(untraced, func(i, j int) bool { return untraced[i].Timestamp > untraced[j].Timestamp })
|
|
return out, untraced
|
|
}
|
|
|
|
// HandleTraceSummaries returns per-trace summaries for ?ids=a,b,c (max 100).
|
|
// Unknown ids are omitted; stores without summaries answer an empty object.
|
|
func (c *Collector) HandleTraceSummaries(w http.ResponseWriter, r *http.Request) {
|
|
var ids []string
|
|
for _, id := range strings.Split(r.URL.Query().Get("ids"), ",") {
|
|
if id = strings.TrimSpace(id); id != "" {
|
|
ids = append(ids, id)
|
|
}
|
|
}
|
|
if len(ids) > 100 {
|
|
http.Error(w, "at most 100 ids", http.StatusBadRequest)
|
|
return
|
|
}
|
|
out := map[string]storage.TraceSummary{}
|
|
if c.traceQuerier != nil && len(ids) > 0 {
|
|
ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
|
|
defer cancel()
|
|
sums, err := c.traceQuerier.TraceSummaries(ctx, ids)
|
|
if err != nil {
|
|
log.Printf("failed to load trace summaries: %v", err)
|
|
http.Error(w, "failed to load trace summaries", http.StatusInternalServerError)
|
|
return
|
|
}
|
|
out = sums
|
|
}
|
|
w.Header().Set("Content-Type", "application/json")
|
|
_ = json.NewEncoder(w).Encode(out)
|
|
}
|