496 lines
15 KiB
Go
496 lines
15 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"io"
|
|
"log"
|
|
"net"
|
|
"net/http"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"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/status"
|
|
"google.golang.org/grpc/stats"
|
|
"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", handleTraces(coll))
|
|
otelMux.HandleFunc("/v1/metrics", handleMetrics(coll))
|
|
otelMux.HandleFunc("/v1/logs", handleLogs(coll))
|
|
|
|
// UI and API mux (port 8080)
|
|
uiMux := http.NewServeMux()
|
|
// Also register OTLP endpoints on 8080 for backwards compatibility
|
|
uiMux.HandleFunc("/v1/traces", handleTraces(coll))
|
|
uiMux.HandleFunc("/v1/metrics", handleMetrics(coll))
|
|
uiMux.HandleFunc("/v1/logs", handleLogs(coll))
|
|
|
|
uiMux.HandleFunc("/api/history", handleHistory(coll))
|
|
uiMux.HandleFunc("/api/history/traces", handleHistoryByType(coll, "traces"))
|
|
uiMux.HandleFunc("/api/history/metrics", handleHistoryByType(coll, "metrics"))
|
|
uiMux.HandleFunc("/api/history/logs", handleHistoryByType(coll, "logs"))
|
|
uiMux.HandleFunc("/api/history/search", handleSearch(coll))
|
|
uiMux.HandleFunc("/api/history/wait", handleWait(coll))
|
|
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 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 handleSearch(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
|
|
}
|
|
|
|
query := r.URL.Query().Get("q")
|
|
limitStr := r.URL.Query().Get("limit")
|
|
limit := 100
|
|
if limitStr != "" {
|
|
if l, err := strconv.Atoi(limitStr); err == nil {
|
|
limit = l
|
|
}
|
|
}
|
|
|
|
history, err := coll.SearchHistory(r.Context(), query, limit)
|
|
if err != nil {
|
|
log.Printf("failed to search history: %v", err)
|
|
http.Error(w, "failed to search history", http.StatusInternalServerError)
|
|
return
|
|
}
|
|
|
|
w.Header().Set("Content-Type", "application/json")
|
|
if err := json.NewEncoder(w).Encode(history); err != nil {
|
|
log.Printf("failed to encode search results: %v", err)
|
|
http.Error(w, "failed to encode history", http.StatusInternalServerError)
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
func handleWait(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
|
|
}
|
|
|
|
query := r.URL.Query().Get("q")
|
|
timeoutSec := 30
|
|
if t := r.URL.Query().Get("timeout"); t != "" {
|
|
if ts, err := strconv.Atoi(t); err == nil {
|
|
timeoutSec = ts
|
|
}
|
|
}
|
|
|
|
ctx, cancel := context.WithTimeout(r.Context(), time.Duration(timeoutSec)*time.Second)
|
|
defer cancel()
|
|
|
|
// First, check if result already exists in recent history
|
|
matches, err := coll.SearchHistory(ctx, query, 1)
|
|
if err == nil && len(matches) > 0 {
|
|
w.Header().Set("Content-Type", "application/json")
|
|
json.NewEncoder(w).Encode(matches[0])
|
|
return
|
|
}
|
|
|
|
// Otherwise, subscribe and wait
|
|
lowerQuery := strings.ToLower(query)
|
|
ch, cleanup := coll.Subscribe(func(entry websocket.HistoryEntry) bool {
|
|
if query == "" {
|
|
return true
|
|
}
|
|
data, _ := json.Marshal(entry.Data)
|
|
return strings.Contains(strings.ToLower(string(data)), lowerQuery)
|
|
})
|
|
defer cleanup()
|
|
|
|
select {
|
|
case entry := <-ch:
|
|
w.Header().Set("Content-Type", "application/json")
|
|
if err := json.NewEncoder(w).Encode(entry); err != nil {
|
|
log.Printf("failed to encode signal: %v", err)
|
|
}
|
|
case <-ctx.Done():
|
|
http.Error(w, "timeout waiting for signal", http.StatusGatewayTimeout)
|
|
}
|
|
}
|
|
}
|
|
|
|
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
|
|
}
|
|
}
|
|
}
|
|
|
|
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)
|
|
})
|
|
}
|