diff --git a/cmd/server/main.go b/cmd/server/main.go index aeb344d..3cddd3e 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -2,14 +2,10 @@ 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" @@ -21,8 +17,8 @@ import ( 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/grpc/status" "google.golang.org/protobuf/proto" ) @@ -32,9 +28,10 @@ func loggingInterceptor(ctx context.Context, req interface{}, info *grpc.UnarySe 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) } + //else { + // log.Printf("gRPC SUCCESS [%s]", info.FullMethod) + //} return resp, err } @@ -55,7 +52,9 @@ func (h *connStatsHandler) HandleConn(ctx context.Context, s stats.ConnStats) { } } -func (h *connStatsHandler) TagRPC(ctx context.Context, info *stats.RPCTagInfo) context.Context { return ctx } +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 { @@ -181,23 +180,25 @@ func main() { // 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)) + 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", handleTraces(coll)) - uiMux.HandleFunc("/v1/metrics", handleMetrics(coll)) - uiMux.HandleFunc("/v1/logs", handleLogs(coll)) + 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", 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("/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"))) @@ -222,263 +223,6 @@ func main() { 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) { diff --git a/internal/collector/collector.go b/internal/collector/collector.go index 80bcdd0..2fb8713 100644 --- a/internal/collector/collector.go +++ b/internal/collector/collector.go @@ -98,7 +98,11 @@ func (c *Collector) Subscribe(filter func(websocket.HistoryEntry) bool) (chan we return ch, cleanup } -func (c *Collector) GetHistory() []websocket.HistoryEntry { +func (c *Collector) GetHistory(limit int) []websocket.HistoryEntry { + if limit <= 0 { + limit = c.historySize + } + c.historyMutex.RLock() defer c.historyMutex.RUnlock() @@ -108,6 +112,12 @@ func (c *Collector) GetHistory() []websocket.HistoryEntry { history = append(history, entry) } }) + + // Return only the last 'limit' items if we have more + if len(history) > limit { + history = history[len(history)-limit:] + } + return history } @@ -121,7 +131,7 @@ func (c *Collector) SearchHistory(ctx context.Context, query string, limit int) } // Fallback to in-memory if no store - history := c.GetHistory() + history := c.GetHistory(limit) var filtered []websocket.HistoryEntry // Lowercase query for case-insensitive search lowerQuery := strings.ToLower(query) diff --git a/internal/collector/collector_test.go b/internal/collector/collector_test.go index eea31d2..6ed2df3 100644 --- a/internal/collector/collector_test.go +++ b/internal/collector/collector_test.go @@ -50,3 +50,33 @@ OK: t.Errorf("expected type 'logs', got %s", store.entries[0].Type) } } + +func TestCollectorHistoryLimit(t *testing.T) { + coll := New(nil, 10, nil) + + for i := 0; i < 10; i++ { + coll.addToHistory(websocket.DataTypeLogs, map[string]interface{}{"msg": i}) + } + + history := coll.GetHistory(10) + if len(history) != 10 { + t.Errorf("expected 10 entries, got %d", len(history)) + } + + history = coll.GetHistory(5) + if len(history) != 5 { + t.Errorf("expected 5 entries, got %d", len(history)) + } + + // Should default to historySize (10) + history = coll.GetHistory(0) + if len(history) != 10 { + t.Errorf("expected default limit to be historySize (10), got %d", len(history)) + } + + // Should not crash on large limit, just return what it has + history = coll.GetHistory(100) + if len(history) != 10 { + t.Errorf("expected 10 entries even with limit 100, got %d", len(history)) + } +} diff --git a/internal/collector/handlers.go b/internal/collector/handlers.go new file mode 100644 index 0000000..e6dc7ac --- /dev/null +++ b/internal/collector/handlers.go @@ -0,0 +1,290 @@ +package collector + +import ( + "context" + "encoding/json" + "io" + "log" + "net/http" + "strconv" + "strings" + "time" + + "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" +) + +func (c *Collector) HandleTraces(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 := c.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 (c *Collector) HandleMetrics(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 := c.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 (c *Collector) HandleLogs(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 := c.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 (c *Collector) HandleHistory(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + + limitStr := r.URL.Query().Get("limit") + limit := 100 + if limitStr != "" { + if l, err := strconv.Atoi(limitStr); err == nil { + limit = l + } + } + + history := c.GetHistory(limit) + + 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 (c *Collector) HandleSearch(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 := c.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 (c *Collector) HandleWait(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 := c.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 := c.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 (c *Collector) HandleHistoryByType(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 + } + + limitStr := r.URL.Query().Get("limit") + limit := 100 + if limitStr != "" { + if l, err := strconv.Atoi(limitStr); err == nil { + limit = l + } + } + + // Fetch a larger set to allow for filtering, then slice + // Retrieve up to 1000 from memory then filter + history := c.GetHistory(5000) + + // Filter by type + var filteredHistory []websocket.HistoryEntry + for _, entry := range history { + if string(entry.Type) == dataType { + filteredHistory = append(filteredHistory, entry) + } + } + + // Limit the results + if len(filteredHistory) > limit { + filteredHistory = filteredHistory[len(filteredHistory)-limit:] + } + + 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 (c *Collector) HandleHealth(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + json.NewEncoder(w).Encode(map[string]string{"status": "ok"}) +} diff --git a/internal/websocket/server.go b/internal/websocket/server.go index 2c770c0..e6f2437 100644 --- a/internal/websocket/server.go +++ b/internal/websocket/server.go @@ -18,7 +18,7 @@ const ( ) type HistoryProvider interface { - GetHistory() []HistoryEntry + GetHistory(limit int) []HistoryEntry } type HistoryEntry struct { @@ -72,7 +72,7 @@ func (s *Server) HandleConnections(w http.ResponseWriter, r *http.Request) { // // Send history to new client // if s.historyProvider != nil { - // history := s.historyProvider.GetHistory() + // history := s.historyProvider.GetHistory(50) // for _, entry := range history { // if data, err := json.Marshal(entry); err == nil { // if err := ws.WriteMessage(websocket.TextMessage, data); err != nil { @@ -117,7 +117,8 @@ func (s *Server) SendHistory(ws *websocket.Conn) { s.mutex.Lock() defer s.mutex.Unlock() - history := s.historyProvider.GetHistory() // Assuming GetHistory() returns the history data + // Default to 100 entries for new socket connections if not otherwise specified + history := s.historyProvider.GetHistory(100) for _, entry := range history { data, err := json.Marshal(entry) if err != nil {