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"}) }