go-otel/cmd/server/main.go

package main

import (
	"context"
	"log"
	"net"
	"net/http"
	"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) {
	switch s.(type) {
	case *stats.ConnEnd:
		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
}

func main() {

	// Configure history size (default 100, can be made configurable)
	historySize := 100
	store := storage.NewFileStore("otel-agent.log")
	coll := collector.New(nil, historySize, store) // 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(
			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", coll.HandleTraces)
	otelMux.HandleFunc("/v1/metrics", coll.HandleMetrics)
	otelMux.HandleFunc("/v1/logs", 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
	uiMux.HandleFunc("/v1/traces", coll.HandleTraces)
	uiMux.HandleFunc("/v1/metrics", coll.HandleMetrics)
	uiMux.HandleFunc("/v1/logs", 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("/ws", wsServer.HandleConnections)
	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)
	})
}