more otel

internal/collector/collector.go

@@ -7,11 +7,16 @@ import (
 	"context"
 	"encoding/json"
 	"log"
+	"os"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
 
 	"git.tornberg.me/go-otel/internal/issues"
+	kube "git.tornberg.me/go-otel/internal/kubernetes"
+	"git.tornberg.me/go-otel/internal/prometheus"
+	"git.tornberg.me/go-otel/internal/sourcemaps"
 	"git.tornberg.me/go-otel/internal/storage"
 	"git.tornberg.me/go-otel/internal/websocket"
 
@@ -28,6 +33,7 @@ type Collector struct {
 	history      *ring.Ring
 	historyMutex sync.RWMutex
 	store        storage.Store
+	storeChan    chan websocket.HistoryEntry
 	subscribers  map[chan websocket.HistoryEntry]func(websocket.HistoryEntry) bool
 	subMutex     sync.Mutex
 	waitLimiter  *ipRateLimiter
@@ -35,15 +41,26 @@ type Collector struct {
 
 	ingestLimiter *ipRateLimiter
 	ingestKey     string
+
+	notifyIssuesFn func()
+
+	// statsQuerier is the ClickHouse store when configured; nil otherwise.
+	// The stats endpoints return 501 when it's nil so the UI can hide those
+	// views instead of erroring.
+	statsQuerier storage.StatsQuerier
+	sourceMaps   *sourcemaps.Resolver
+	kubeWatcher  *kube.Watcher
+	prometheus   *prometheus.Client
 }
 
 // New creates a new Collector instance.
 func New(broadcaster websocket.Broadcaster, historySize int, store storage.Store) *Collector {
-	return &Collector{
+	c := &Collector{
 		broadcaster: broadcaster,
 		historySize: historySize,
 		history:     ring.New(historySize),
 		store:       store,
+		storeChan:   make(chan websocket.HistoryEntry, 10000),
 		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),
@@ -53,6 +70,26 @@ func New(broadcaster websocket.Broadcaster, historySize int, store storage.Store
 		// looping) from hammering the collector.
 		ingestLimiter: newIPRateLimiter(20, 50),
 	}
+
+	if store != nil {
+		// Bounded worker pool prevents concurrent ClickHouse INSERTs from
+		// multiplying query and merge memory under bursty ingest traffic.
+		numWorkers := 4
+		if value, err := strconv.Atoi(os.Getenv("OTEL_STORE_WORKERS")); err == nil && value > 0 && value <= 32 {
+			numWorkers = value
+		}
+		for i := 0; i < numWorkers; i++ {
+			go func() {
+				for entry := range c.storeChan {
+					if err := c.store.Append(context.Background(), entry); err != nil {
+						log.Printf("failed to append to store: %v", err)
+					}
+				}
+			}()
+		}
+	}
+
+	return c
 }
 
 // SetIngestKey configures an optional shared key that HTTP OTLP ingestion
@@ -64,19 +101,102 @@ func (c *Collector) SetIngestKey(key string) {
 	c.ingestKey = key
 }
 
+// SetStatsQuerier enables the aggregate stats endpoints (error trends,
+// release comparisons) when a store supporting them is configured.
+func (c *Collector) SetStatsQuerier(q storage.StatsQuerier) {
+	c.statsQuerier = q
+}
+
+func (c *Collector) SetKubernetesWatcher(w *kube.Watcher)     { c.kubeWatcher = w }
+func (c *Collector) SetPrometheusClient(p *prometheus.Client) { c.prometheus = p }
+func (c *Collector) ListRollouts() []kube.Rollout {
+	if c.kubeWatcher == nil {
+		return nil
+	}
+	return c.kubeWatcher.List()
+}
+func (c *Collector) GetRollout(id string) (kube.Rollout, bool) {
+	if c.kubeWatcher == nil {
+		return kube.Rollout{}, false
+	}
+	return c.kubeWatcher.Get(id)
+}
+func (c *Collector) KubernetesEnabled() bool { return c.kubeWatcher != nil && c.kubeWatcher.Enabled() }
+func (c *Collector) KubernetesError() string {
+	if c.kubeWatcher == nil {
+		return "kubernetes integration is not configured"
+	}
+	return c.kubeWatcher.Error()
+}
+
 // ListIssues returns deduplicated error issues, most recently seen first.
 func (c *Collector) ListIssues(limit int) []issues.Issue {
-	return c.issues.List(limit)
+	list := c.issues.ListFiltered(issues.ListOptions{Limit: limit}).Issues
+	if c.sourceMaps != nil {
+		for i := range list {
+			resolveIssue(&list[i], c.sourceMaps)
+		}
+	}
+	return list
+}
+
+// ListIssuesFiltered returns a filtered page of deduplicated issues.
+func (c *Collector) ListIssuesFiltered(opts issues.ListOptions) issues.ListPage {
+	page := c.issues.ListFiltered(opts)
+	if c.sourceMaps != nil {
+		for i := range page.Issues {
+			resolveIssue(&page.Issues[i], c.sourceMaps)
+		}
+	}
+	return page
 }
 
 // GetIssue returns a single issue by id.
 func (c *Collector) GetIssue(id string) (issues.Issue, bool) {
-	return c.issues.Get(id)
+	issue, ok := c.issues.Get(id)
+	if ok && c.sourceMaps != nil {
+		resolveIssue(&issue, c.sourceMaps)
+	}
+	return issue, ok
+}
+
+func (c *Collector) SetSourceMaps(resolver *sourcemaps.Resolver) { c.sourceMaps = resolver }
+
+func resolveIssue(issue *issues.Issue, resolver *sourcemaps.Resolver) {
+	for i := range issue.Samples {
+		issue.Samples[i].ResolvedStacktrace = resolver.ResolveStack(issue.Samples[i].Release, issue.Samples[i].Stacktrace)
+	}
+}
+
+// SetIssueStatus updates the triage state of an issue.
+func (c *Collector) SetIssueStatus(id, status string) (issues.Issue, bool) {
+	return c.issues.SetStatus(id, status)
+}
+
+func (c *Collector) SimilarIssues(id string, limit int) ([]issues.SimilarIssue, bool) {
+	return c.issues.Similar(id, limit)
+}
+
+func (c *Collector) ListPatterns() []issues.Pattern { return c.issues.Patterns() }
+
+func (c *Collector) SetSimilarityFeedback(a, b, decision string) bool {
+	return c.issues.SetSimilarityFeedback(a, b, decision)
 }
 
 // SetBroadcaster updates the broadcaster used by the collector.
 func (c *Collector) SetBroadcaster(broadcaster websocket.Broadcaster) {
 	c.broadcaster = broadcaster
+	if notifier, ok := broadcaster.(interface{ NotifyIssues() }); ok {
+		c.notifyIssuesFn = notifier.NotifyIssues
+	}
+}
+
+// notifyIssues pings WebSocket clients that new error telemetry arrived, so
+// the UI can refresh its Issues view without waiting for its poll interval.
+func (c *Collector) notifyIssues() {
+	if c.notifyIssuesFn != nil {
+		c.notifyIssuesFn()
+	}
 }
 
 func (c *Collector) addToHistory(dataType websocket.DataType, data json.RawMessage) {
@@ -92,12 +212,11 @@ func (c *Collector) addToHistory(dataType websocket.DataType, data json.RawMessa
 	c.historyMutex.Unlock()
 
 	if c.store != nil {
-		// Asynchronously append to store to avoid blocking ingestion
-		go func() {
-			if err := c.store.Append(context.Background(), entry); err != nil {
-				log.Printf("failed to append to store: %v", err)
-			}
-		}()
+		select {
+		case c.storeChan <- entry:
+		default:
+			log.Printf("collector: store write queue full, dropping entry to preserve ingestion")
+		}
 	}
 
 	c.notifySubscribers(entry)
@@ -219,8 +338,10 @@ func (c *Collector) ConsumeTraces(ctx context.Context, td ptrace.Traces) error {
 	}
 
 	// Store and broadcast raw bytes as json.RawMessage to avoid unsafe interface{} usage
-		c.addToHistory(websocket.DataTypeTraces, buf)
-	extractSpanErrors(c.issues, td, time.Now().UnixNano())
+	c.addToHistory(websocket.DataTypeTraces, buf)
+	if extractSpanErrors(c.issues, td, time.Now().UnixNano()) {
+		c.notifyIssues()
+	}
 
 	entry := websocket.Entry{
 		Type: string(websocket.DataTypeTraces),
@@ -241,7 +362,7 @@ func (c *Collector) ConsumeMetrics(ctx context.Context, md pmetric.Metrics) erro
 	}
 
 	// Store and broadcast raw bytes as json.RawMessage to avoid unsafe interface{} usage
-		c.addToHistory(websocket.DataTypeMetrics, buf)
+	c.addToHistory(websocket.DataTypeMetrics, buf)
 
 	entry := websocket.Entry{
 		Type: string(websocket.DataTypeMetrics),
@@ -262,8 +383,10 @@ func (c *Collector) ConsumeLogs(ctx context.Context, ld plog.Logs) error {
 	}
 
 	// Store and broadcast raw bytes as json.RawMessage to avoid unsafe interface{} usage
-		c.addToHistory(websocket.DataTypeLogs, buf)
-	extractLogErrors(c.issues, ld, time.Now().UnixNano())
+	c.addToHistory(websocket.DataTypeLogs, buf)
+	if extractLogErrors(c.issues, ld, time.Now().UnixNano()) {
+		c.notifyIssues()
+	}
 
 	entry := websocket.Entry{
 		Type: string(websocket.DataTypeLogs),