package main import ( "context" "encoding/json" "io" "log" "net" "net/http" "git.tornberg.me/go-otel/internal/collector" "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/protobuf/proto" ) 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 } func main() { // Configure history size (default 100, can be made configurable) historySize := 100 coll := collector.New(nil, historySize) // Temporary nil for wsServer wsServer := websocket.New(coll) // Update collector with wsServer coll.SetBroadcaster(wsServer) // 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() 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 endpoints http.HandleFunc("/v1/traces", handleTraces(coll)) http.HandleFunc("/v1/metrics", handleMetrics(coll)) http.HandleFunc("/v1/logs", handleLogs(coll)) // History API endpoints http.HandleFunc("/api/history", handleHistory(coll)) http.HandleFunc("/api/history/traces", handleHistoryByType(coll, "traces")) http.HandleFunc("/api/history/metrics", handleHistoryByType(coll, "metrics")) http.HandleFunc("/api/history/logs", handleHistoryByType(coll, "logs")) // WebSocket server for visualization http.HandleFunc("/ws", wsServer.HandleConnections) http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) w.Write([]byte("OTel Debug Receiver is running")) }) log.Printf("Starting HTTP server on port 8080") log.Printf("OTLP HTTP endpoints:") 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 endpoints:") log.Printf(" Traces: localhost:4317") log.Printf(" Metrics: localhost:4317") log.Printf(" Logs: localhost:4317") log.Printf("WebSocket: ws://localhost:8080/ws") if err := http.ListenAndServe(":8080", nil); err != nil { log.Fatalf("failed to start HTTP server: %v", err) } } func handleTraces(coll *collector.Collector) http.HandlerFunc { return func(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 } 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 := coll.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) } } func handleMetrics(coll *collector.Collector) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) 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 := coll.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) } } func handleLogs(coll *collector.Collector) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) 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 := coll.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) } } func handleHistory(coll *collector.Collector) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) return } history := coll.GetHistory() 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 } } } func handleHistoryByType(coll *collector.Collector, 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 } history := coll.GetHistory() // Filter by type var filteredHistory []websocket.HistoryEntry for _, entry := range history { if string(entry.Type) == dataType { filteredHistory = append(filteredHistory, entry) } } w.Header().Set("Content-Type", "application/json") if err := json.NewEncoder(w).Encode(filteredHistory); err != nil { log.Printf("failed to encode filtered history: %v", err) http.Error(w, "failed to encode history", http.StatusInternalServerError) return } } }