better

cmd/server/main.go

@@ -7,9 +7,12 @@ import (
 	"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"
@@ -18,9 +21,43 @@ 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/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
@@ -115,7 +152,8 @@ func main() {
 
 	// Configure history size (default 100, can be made configurable)
 	historySize := 100
-	coll := collector.New(nil, historySize) // Temporary nil for wsServer
+	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)
@@ -127,7 +165,10 @@ func main() {
 			log.Fatal("failed to listen for gRPC", err)
 		}
 
-		s := grpc.NewServer()
+		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})
@@ -148,6 +189,8 @@ func main() {
 	http.HandleFunc("/api/history/traces", handleHistoryByType(coll, "traces"))
 	http.HandleFunc("/api/history/metrics", handleHistoryByType(coll, "metrics"))
 	http.HandleFunc("/api/history/logs", handleHistoryByType(coll, "logs"))
+	http.HandleFunc("/api/history/search", handleSearch(coll))
+	http.HandleFunc("/api/history/wait", handleWait(coll))
 
 	// WebSocket server for visualization
 	http.HandleFunc("/ws", wsServer.HandleConnections)
@@ -320,6 +363,87 @@ func handleHistory(coll *collector.Collector) http.HandlerFunc {
 	}
 }
 
+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 {