developerandClaude Sonnet 5
c20a0b77f3
- Add static/otel-rum.js: a framework-free browser script that captures
uncaught errors, unhandled promise rejections and failed resource loads,
translating them into OTLP log records POSTed to the existing /v1/logs
endpoint (source: frontend attribute) so they flow through the same
history/search/issue pipeline as backend telemetry.
- Add correlation-id support: traceHeaders() (manual) and same-origin
fetch auto-instrumentation inject a W3C traceparent header, so a
frontend error and the backend spans it triggered share a trace id and
can be found together via GET /api/traces/{traceId}.
- Add CORS handling (withCORS in cmd/server/main.go) on the OTLP HTTP
endpoints, required for any browser-origin POST.
- Add an optional shared ingest key (OTEL_INGEST_KEY, via '?key=' or
X-Otel-Ingest-Key) and a per-IP rate limit (20 rps/burst 50) on OTLP
HTTP ingestion, to scope/protect a public endpoint fed by a browser
script. Verified end-to-end (CORS preflight, cross-origin POST, ingest
key enforcement, rate limiting) against a running server.
- Serve static/otel-rum.js and agents_example.md as static HTTP routes so
both the script and the agent instructions doc are fetchable directly
from the base URL.
- Ignore the separately-cloned otel-ui/ frontend repo (has its own .git;
not meant to be nested into this repo's history).
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
275 lines
9.1 KiB
Go
275 lines
9.1 KiB
Go
// Package main implements the Go OTel Debugger server.
|
|
package main
|
|
|
|
import (
|
|
"context"
|
|
"log"
|
|
"net"
|
|
"net/http"
|
|
"os"
|
|
"strings"
|
|
|
|
"git.tornberg.me/go-otel/internal/collector"
|
|
"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"
|
|
logsgrpc "go.opentelemetry.io/proto/otlp/collector/logs/v1"
|
|
metricsgrpc "go.opentelemetry.io/proto/otlp/collector/metrics/v1"
|
|
tracegrpc "go.opentelemetry.io/proto/otlp/collector/trace/v1"
|
|
"google.golang.org/grpc"
|
|
"google.golang.org/grpc/stats"
|
|
"google.golang.org/grpc/status"
|
|
"google.golang.org/protobuf/proto"
|
|
)
|
|
|
|
// gRPC Logging Interceptor
|
|
func loggingInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
|
|
resp, err := handler(ctx, req)
|
|
if err != nil {
|
|
st, _ := status.FromError(err)
|
|
log.Printf("gRPC ERROR [%s] code=%s message=%s", info.FullMethod, st.Code(), st.Message())
|
|
}
|
|
//else {
|
|
// log.Printf("gRPC SUCCESS [%s]", info.FullMethod)
|
|
//}
|
|
return resp, err
|
|
}
|
|
|
|
// gRPC Connection Stats Handler
|
|
type connStatsHandler struct {
|
|
stats.Handler
|
|
}
|
|
|
|
func (h *connStatsHandler) TagConn(ctx context.Context, info *stats.ConnTagInfo) context.Context {
|
|
log.Printf("gRPC CONN [OPEN] from=%s", info.RemoteAddr)
|
|
return ctx
|
|
}
|
|
|
|
func (h *connStatsHandler) HandleConn(ctx context.Context, s stats.ConnStats) {
|
|
if _, ok := s.(*stats.ConnEnd); ok {
|
|
log.Printf("gRPC CONN [CLOSED]")
|
|
}
|
|
}
|
|
|
|
func (h *connStatsHandler) TagRPC(ctx context.Context, info *stats.RPCTagInfo) context.Context {
|
|
return ctx
|
|
}
|
|
|
|
func (h *connStatsHandler) HandleRPC(ctx context.Context, s stats.RPCStats) {}
|
|
|
|
type traceService struct {
|
|
tracegrpc.UnimplementedTraceServiceServer
|
|
collector *collector.Collector
|
|
}
|
|
|
|
func (s *traceService) Export(ctx context.Context, req *tracegrpc.ExportTraceServiceRequest) (*tracegrpc.ExportTraceServiceResponse, error) {
|
|
// log.Printf("Received gRPC traces %d", len(req.ResourceSpans))
|
|
|
|
// Convert protobuf to bytes then to pdata
|
|
data, err := proto.Marshal(req)
|
|
if err != nil {
|
|
log.Printf("failed to marshal trace request: %v", err)
|
|
return nil, err
|
|
}
|
|
|
|
unmarshaler := &ptrace.ProtoUnmarshaler{}
|
|
traces, err := unmarshaler.UnmarshalTraces(data)
|
|
if err != nil {
|
|
log.Printf("failed to unmarshal traces: %v", err)
|
|
return nil, err
|
|
}
|
|
|
|
if err := s.collector.ConsumeTraces(ctx, traces); err != nil {
|
|
log.Printf("failed to consume traces: %v", err)
|
|
return nil, err
|
|
}
|
|
|
|
return &tracegrpc.ExportTraceServiceResponse{}, nil
|
|
}
|
|
|
|
type metricsService struct {
|
|
metricsgrpc.UnimplementedMetricsServiceServer
|
|
collector *collector.Collector
|
|
}
|
|
|
|
func (s *metricsService) Export(ctx context.Context, req *metricsgrpc.ExportMetricsServiceRequest) (*metricsgrpc.ExportMetricsServiceResponse, error) {
|
|
// log.Printf("Received gRPC metrics %d", len(req.ResourceMetrics))
|
|
|
|
// Convert protobuf to bytes then to pdata
|
|
data, err := proto.Marshal(req)
|
|
if err != nil {
|
|
log.Printf("failed to marshal metrics request: %v", err)
|
|
return nil, err
|
|
}
|
|
|
|
unmarshaler := &pmetric.ProtoUnmarshaler{}
|
|
metrics, err := unmarshaler.UnmarshalMetrics(data)
|
|
if err != nil {
|
|
log.Printf("failed to unmarshal metrics: %v", err)
|
|
return nil, err
|
|
}
|
|
|
|
if err := s.collector.ConsumeMetrics(ctx, metrics); err != nil {
|
|
log.Printf("failed to consume metrics: %v", err)
|
|
return nil, err
|
|
}
|
|
|
|
return &metricsgrpc.ExportMetricsServiceResponse{}, nil
|
|
}
|
|
|
|
type logsService struct {
|
|
logsgrpc.UnimplementedLogsServiceServer
|
|
collector *collector.Collector
|
|
}
|
|
|
|
func (s *logsService) Export(ctx context.Context, req *logsgrpc.ExportLogsServiceRequest) (*logsgrpc.ExportLogsServiceResponse, error) {
|
|
// log.Printf("Received gRPC logs %d", len(req.ResourceLogs))
|
|
|
|
// Convert protobuf to bytes then to pdata
|
|
data, err := proto.Marshal(req)
|
|
if err != nil {
|
|
log.Printf("failed to marshal logs request: %v", err)
|
|
return nil, err
|
|
}
|
|
|
|
unmarshaler := &plog.ProtoUnmarshaler{}
|
|
logs, err := unmarshaler.UnmarshalLogs(data)
|
|
if err != nil {
|
|
log.Printf("failed to unmarshal logs: %v", err)
|
|
return nil, err
|
|
}
|
|
|
|
if err := s.collector.ConsumeLogs(ctx, logs); err != nil {
|
|
log.Printf("failed to consume logs: %v", err)
|
|
return nil, err
|
|
}
|
|
|
|
return &logsgrpc.ExportLogsServiceResponse{}, nil
|
|
}
|
|
|
|
// withCORS allows cross-origin ingestion (e.g. from the otel-rum.js browser
|
|
// script running on a third-party site) into the OTLP HTTP endpoints. It's
|
|
// deliberately permissive on origin, since this is a public ingest endpoint
|
|
// scoped by the ingest key/rate limiting instead of same-origin policy.
|
|
func withCORS(next http.HandlerFunc) http.HandlerFunc {
|
|
return func(w http.ResponseWriter, r *http.Request) {
|
|
w.Header().Set("Access-Control-Allow-Origin", "*")
|
|
w.Header().Set("Access-Control-Allow-Methods", "POST, OPTIONS")
|
|
w.Header().Set("Access-Control-Allow-Headers", "Content-Type, X-Otel-Ingest-Key")
|
|
if r.Method == http.MethodOptions {
|
|
w.WriteHeader(http.StatusNoContent)
|
|
return
|
|
}
|
|
next(w, r)
|
|
}
|
|
}
|
|
|
|
func main() {
|
|
|
|
// Configure history size (default 100, can be made configurable)
|
|
historySize := 100
|
|
logPath := os.Getenv("OTELLOG_PATH")
|
|
if logPath == "" {
|
|
logPath = "otel-agent.log"
|
|
}
|
|
store := storage.NewFileStore(logPath)
|
|
coll := collector.New(nil, historySize, store) // Temporary nil for wsServer
|
|
wsServer := websocket.New(coll)
|
|
// Update collector with wsServer
|
|
coll.SetBroadcaster(wsServer)
|
|
coll.SetIngestKey(os.Getenv("OTEL_INGEST_KEY"))
|
|
|
|
// Start gRPC server
|
|
go func() {
|
|
lis, err := net.Listen("tcp", ":4317")
|
|
if err != nil {
|
|
log.Fatal("failed to listen for gRPC", err)
|
|
}
|
|
|
|
s := grpc.NewServer(
|
|
grpc.UnaryInterceptor(loggingInterceptor),
|
|
grpc.StatsHandler(&connStatsHandler{}),
|
|
)
|
|
tracegrpc.RegisterTraceServiceServer(s, &traceService{collector: coll})
|
|
metricsgrpc.RegisterMetricsServiceServer(s, &metricsService{collector: coll})
|
|
logsgrpc.RegisterLogsServiceServer(s, &logsService{collector: coll})
|
|
|
|
log.Print("Starting gRPC server on port 4317")
|
|
if err := s.Serve(lis); err != nil {
|
|
log.Fatalf("failed to serve gRPC: %v", err)
|
|
}
|
|
}()
|
|
|
|
// OTLP HTTP mux (standard OTel port 4318)
|
|
otelMux := http.NewServeMux()
|
|
otelMux.HandleFunc("/v1/traces", withCORS(coll.HandleTraces))
|
|
otelMux.HandleFunc("/v1/metrics", withCORS(coll.HandleMetrics))
|
|
otelMux.HandleFunc("/v1/logs", withCORS(coll.HandleLogs))
|
|
otelMux.HandleFunc("/health", coll.HandleHealth)
|
|
|
|
// UI and API mux (port 8080)
|
|
uiMux := http.NewServeMux()
|
|
// Also register OTLP endpoints on 8080 for backwards compatibility.
|
|
// CORS is enabled here since this is the publicly exposed port (via the
|
|
// ingress) that a browser script such as static/otel-rum.js posts to.
|
|
uiMux.HandleFunc("/v1/traces", withCORS(coll.HandleTraces))
|
|
uiMux.HandleFunc("/v1/metrics", withCORS(coll.HandleMetrics))
|
|
uiMux.HandleFunc("/v1/logs", withCORS(coll.HandleLogs))
|
|
uiMux.HandleFunc("/health", coll.HandleHealth)
|
|
|
|
uiMux.HandleFunc("/api/history", coll.HandleHistory)
|
|
uiMux.HandleFunc("/api/history/traces", coll.HandleHistoryByType("traces"))
|
|
uiMux.HandleFunc("/api/history/metrics", coll.HandleHistoryByType("metrics"))
|
|
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("/api/issues", coll.HandleIssues)
|
|
uiMux.HandleFunc("GET /api/issues/{id}", coll.HandleIssueByID)
|
|
uiMux.HandleFunc("/ws", wsServer.HandleConnections)
|
|
uiMux.Handle("/static/", http.StripPrefix("/static/", http.FileServer(http.Dir("./static"))))
|
|
// Serve the agent instructions doc as a static route so an AI agent can
|
|
// fetch it directly from the base URL before making any API calls.
|
|
uiMux.HandleFunc("GET /agents_example.md", func(w http.ResponseWriter, r *http.Request) {
|
|
http.ServeFile(w, r, "agents_example.md")
|
|
})
|
|
uiMux.Handle("/", spaHandler(http.Dir("./dist")))
|
|
|
|
log.Printf("Starting HTTP server on port 8080 (UI/API)")
|
|
log.Printf("OTLP HTTP endpoints (also on port 8080):")
|
|
log.Printf(" Traces: http://localhost:8080/v1/traces")
|
|
log.Printf(" Metrics: http://localhost:8080/v1/metrics")
|
|
log.Printf(" Logs: http://localhost:8080/v1/logs")
|
|
log.Printf("OTLP gRPC endpoint:")
|
|
log.Printf(" Traces/Metrics/Logs: localhost:4317")
|
|
log.Printf("WebSocket: ws://localhost:8080/ws")
|
|
|
|
// Start OTLP HTTP server on port 4318
|
|
go func() {
|
|
log.Printf("Starting OTLP HTTP server on port 4318")
|
|
if err := http.ListenAndServe(":4318", otelMux); err != nil {
|
|
log.Printf("failed to start OTLP HTTP server: %v", err)
|
|
}
|
|
}()
|
|
|
|
if err := http.ListenAndServe(":8080", uiMux); err != nil {
|
|
log.Fatalf("failed to start HTTP server: %v", err)
|
|
}
|
|
}
|
|
func spaHandler(root http.FileSystem) http.Handler {
|
|
fileServer := http.FileServer(root)
|
|
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
path := r.URL.Path
|
|
// Skip API, WebSocket, and OTLP endpoints
|
|
if !strings.HasPrefix(path, "/api") && !strings.HasPrefix(path, "/ws") && !strings.HasPrefix(path, "/v1") {
|
|
if _, err := root.Open(path); err != nil {
|
|
// File does not exist, serve index.html
|
|
r.URL.Path = "/"
|
|
}
|
|
}
|
|
fileServer.ServeHTTP(w, r)
|
|
})
|
|
}
|