From 26265e776c17b1b80ef95b23e475e29491cc17ed Mon Sep 17 00:00:00 2001 From: matst80 Date: Fri, 3 Apr 2026 11:19:00 +0200 Subject: [PATCH] better --- .gitignore | 2 + Dockerfile | 6 +- README.md | 4 +- agents_example.md | 77 ++++++++++++++++ build.sh | 8 ++ cmd/server/main.go | 128 ++++++++++++++++++++++++++- go.mod | 1 + go.sum | 2 + internal/collector/collector.go | 102 +++++++++++++++++++-- internal/collector/collector_test.go | 52 +++++++++++ internal/storage/file_store.go | 83 +++++++++++++++++ internal/storage/file_store_test.go | 64 ++++++++++++++ internal/storage/store.go | 12 +++ k8s/deployment.yaml | 3 + 14 files changed, 533 insertions(+), 11 deletions(-) create mode 100644 .gitignore create mode 100644 agents_example.md create mode 100755 build.sh create mode 100644 internal/collector/collector_test.go create mode 100644 internal/storage/file_store.go create mode 100644 internal/storage/file_store_test.go create mode 100644 internal/storage/store.go diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..66e8e43 --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +./server +.DS_Store \ No newline at end of file diff --git a/Dockerfile b/Dockerfile index 006ef57..583d5f9 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,4 +1,6 @@ # UI build stage +ARG TARGETOS +ARG TARGETARCH FROM node:alpine AS ui-builder WORKDIR /app @@ -8,7 +10,7 @@ COPY otel-ui/ . RUN npm install && npm run build # Go build stage -FROM golang:1.25-alpine AS go-builder +FROM --platform=$BUILDPLATFORM golang:1.25-alpine AS go-builder WORKDIR /app @@ -21,7 +23,7 @@ COPY internal/ ./internal # Copy built UI dist COPY --from=ui-builder /app/dist ./otel-ui/dist -RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -a -installsuffix cgo -o /go-otel ./cmd/server +RUN CGO_ENABLED=0 GOOS=${TARGETOS} GOARCH=${TARGETARCH} go build -a -installsuffix cgo -o /go-otel ./cmd/server # Final stage FROM alpine:latest diff --git a/README.md b/README.md index 2775b5d..01d021c 100644 --- a/README.md +++ b/README.md @@ -39,7 +39,7 @@ OTLP Client → HTTP/gRPC Server → Collector → WebSocket Broadcast → Conne 1. Clone the repository: ```sh - git clone https://git.tornberg.me/go-otel.git + git clone https://git.k6n.net/go-otel.git cd go-otel ``` @@ -148,4 +148,4 @@ To modify the application: 2. Run locally with `go run ./cmd/server` 3. For UI changes, edit `index.html` directly -The application uses standard Go modules and OpenTelemetry collector libraries for data processing. \ No newline at end of file +The application uses standard Go modules and OpenTelemetry collector libraries for data processing. diff --git a/agents_example.md b/agents_example.md new file mode 100644 index 0000000..c4175c4 --- /dev/null +++ b/agents_example.md @@ -0,0 +1,77 @@ +# AI Agent Debugging with OTEL Collector + +This document provides instructions for AI agents on how to utilize the OTEL Collector's debugging features to search and fetch telemetry data (traces, logs, and metrics). + +## Base Configuration +- **Base URL**: `https://otel.k6n.net` +- **Endpoints**: + - `GET /api/history/search?q={query}&limit={limit}`: Search all historical telemetry data. + - **Default Limit**: 50 (max 100 recommended for context). + - **Ordering**: Latest entries first. + - **Match Scope**: Case-insensitive substring match on raw JSON (covers service name, span body, attributes, etc.). + - `GET /api/history/wait?q={query}&timeout={seconds}`: **Wait** for a specific signal to arrive. + - **Blocking**: This call blocks until a match is found or the timeout (default 30s) is reached. + - **Ideal for Automation**: Use this after triggering an action to wait for the exact result. + - `GET /api/history`: Fetch the recent in-memory history of all telemetry. + - `GET /api/history/traces`: Fetch recent traces. + - `GET /api/history/logs`: Fetch recent logs. + - `GET /api/history/metrics`: Fetch recent metrics. + +## Agent Instructions + +### 1. Searching for Context +When debugging an issue, start by searching for relevant keywords in the logs and traces. This helps locate specific errors or trace IDs associated with a failure. Since results are **latest first**, you will always see the most recent events related to your search. + +**Example Request:** +```bash +# Search for recent "ERROR" occurrences (default limit 50, latest first) +curl "https://otel.k6n.net/api/history/search?q=error" +``` + +### 2. Waiting for a Signal +Use the `wait` endpoint to synchronize your debugging flow. This is perfect for verifying that a specific event occurred after you triggered an action. + +**Example Request:** +```bash +# Wait up to 60s for a specific traceId to show up in the collector +curl "https://otel.k6n.net/api/history/wait?q=5682138e4a9b4629852899d45e542456&timeout=60" +``` + +### 3. Identifying Trace Chains +If you find a log entry with a `traceId`, use the search endpoint to find all spans and logs associated with that specific trace to understand the full request flow. + +**Example Request:** +```bash +# Search for all activity related to a specific trace +curl "https://otel.k6n.net/api/history/search?q=5682138e4a9b4629852899d45e542456&limit=100" +``` + +### 3. Fetching Recent Telemetry +To see the most recent activity without specific keywords, use the search endpoint with an empty query. + +**Example Request:** +```bash +curl "https://otel.k6n.net/api/history/search?limit=20" +``` + +### 4. Data Format +The returned data is a JSON array of `HistoryEntry` objects: +```json +[ + { + "type": "logs", + "timestamp": 1712132978000000000, + "data": { ... OTEL Log Records ... } + }, + { + "type": "traces", + "timestamp": 1712132979000000000, + "data": { ... OTEL Resource Spans ... } + } +] +``` + +## Best Practices for AI Agents +- **Iterative Search**: Start with a broad search (e.g., service name) and narrow down using specific attributes or error messages found in the results. +- **Limit results**: Always use the `limit` parameter to manage context window size (default is 100). +- **Correlate Signals**: Look for timestamps that coincide across logs and metrics to identify resource constraints or timing issues. diff --git a/build.sh b/build.sh new file mode 100755 index 0000000..62366c6 --- /dev/null +++ b/build.sh @@ -0,0 +1,8 @@ +#!/bin/bash + +docker build -t matst80/otel-debug:latest . +docker push matst80/otel-debug:latest + +kubectl rollout restart deployment otel-debug-receiver -n monitoring + +kubectl logs -f deployment/otel-debug-receiver -n monitoring \ No newline at end of file diff --git a/cmd/server/main.go b/cmd/server/main.go index 3ff694d..5855a4c 100644 --- a/cmd/server/main.go +++ b/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 { diff --git a/go.mod b/go.mod index 59d55de..d6249f3 100644 --- a/go.mod +++ b/go.mod @@ -17,6 +17,7 @@ require ( github.com/json-iterator/go v1.1.12 // indirect github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee // indirect + github.com/natefinch/lumberjack v2.0.0+incompatible // indirect go.opentelemetry.io/collector/featuregate v1.45.0 // indirect go.opentelemetry.io/otel v1.38.0 // indirect go.opentelemetry.io/otel/sdk/metric v1.38.0 // indirect diff --git a/go.sum b/go.sum index ba16a2a..cd0708c 100644 --- a/go.sum +++ b/go.sum @@ -26,6 +26,8 @@ github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJ github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee h1:W5t00kpgFdJifH4BDsTlE89Zl93FEloxaWZfGcifgq8= github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= +github.com/natefinch/lumberjack v2.0.0+incompatible h1:4QJd3OLAMgj7ph+yZTuX13Ld4UpgHp07nNdFX7mqFfM= +github.com/natefinch/lumberjack v2.0.0+incompatible/go.mod h1:Wi9p2TTF5DG5oU+6YfsmYQpsTIOm0B1VNzQg9Mw6nPk= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= diff --git a/internal/collector/collector.go b/internal/collector/collector.go index c318c8a..80bcdd0 100644 --- a/internal/collector/collector.go +++ b/internal/collector/collector.go @@ -5,9 +5,11 @@ import ( "context" "encoding/json" "log" + "strings" "sync" "time" + "git.tornberg.me/go-otel/internal/storage" "git.tornberg.me/go-otel/internal/websocket" "go.opentelemetry.io/collector/consumer" @@ -21,13 +23,18 @@ type Collector struct { historySize int history *ring.Ring historyMutex sync.RWMutex + store storage.Store + subscribers map[chan websocket.HistoryEntry]func(websocket.HistoryEntry) bool + subMutex sync.Mutex } -func New(broadcaster websocket.Broadcaster, historySize int) *Collector { +func New(broadcaster websocket.Broadcaster, historySize int, store storage.Store) *Collector { return &Collector{ broadcaster: broadcaster, historySize: historySize, history: ring.New(historySize), + store: store, + subscribers: make(map[chan websocket.HistoryEntry]func(websocket.HistoryEntry) bool), } } @@ -36,15 +43,59 @@ func (c *Collector) SetBroadcaster(broadcaster websocket.Broadcaster) { } func (c *Collector) addToHistory(dataType websocket.DataType, data interface{}) { - c.historyMutex.Lock() - defer c.historyMutex.Unlock() - - c.history.Value = websocket.HistoryEntry{ + entry := websocket.HistoryEntry{ Type: string(dataType), Timestamp: time.Now().UnixNano(), Data: data, } + + c.historyMutex.Lock() + c.history.Value = entry c.history = c.history.Next() + c.historyMutex.Unlock() + + if c.store != nil { + // Asynchronously append to store to avoid blocking ingestion + go func() { + if err := c.store.Append(context.Background(), entry); err != nil { + log.Printf("failed to append to store: %v", err) + } + }() + } + + c.notifySubscribers(entry) +} + +func (c *Collector) notifySubscribers(entry websocket.HistoryEntry) { + c.subMutex.Lock() + defer c.subMutex.Unlock() + + for ch, filter := range c.subscribers { + if filter(entry) { + select { + case ch <- entry: + // Successfully sent + default: + // Channel blocked, skip + } + } + } +} + +func (c *Collector) Subscribe(filter func(websocket.HistoryEntry) bool) (chan websocket.HistoryEntry, func()) { + ch := make(chan websocket.HistoryEntry, 1) + c.subMutex.Lock() + c.subscribers[ch] = filter + c.subMutex.Unlock() + + cleanup := func() { + c.subMutex.Lock() + delete(c.subscribers, ch) + c.subMutex.Unlock() + close(ch) + } + + return ch, cleanup } func (c *Collector) GetHistory() []websocket.HistoryEntry { @@ -60,6 +111,47 @@ func (c *Collector) GetHistory() []websocket.HistoryEntry { return history } +func (c *Collector) SearchHistory(ctx context.Context, query string, limit int) ([]websocket.HistoryEntry, error) { + if limit <= 0 { + limit = 50 + } + + if c.store != nil { + return c.store.Search(ctx, query, limit) + } + + // Fallback to in-memory if no store + history := c.GetHistory() + var filtered []websocket.HistoryEntry + // Lowercase query for case-insensitive search + lowerQuery := strings.ToLower(query) + + for _, entry := range history { + if query == "" { + filtered = append(filtered, entry) + } else { + // Basic string filter for in-memory + data, _ := json.Marshal(entry.Data) + if strings.Contains(strings.ToLower(string(data)), lowerQuery) { + filtered = append(filtered, entry) + } + } + // Keep it simple and just filter all then limit, since historySize is usually small + } + + // Respect limit and return latest messages first + if len(filtered) > limit { + filtered = filtered[len(filtered)-limit:] + } + + // Reverse to match Store's Search behavior (latest first) + for i, j := 0, len(filtered)-1; i < j; i, j = i+1, j-1 { + filtered[i], filtered[j] = filtered[j], filtered[i] + } + + return filtered, nil +} + func (c *Collector) ConsumeTraces(ctx context.Context, td ptrace.Traces) error { marshaler := &ptrace.JSONMarshaler{} buf, err := marshaler.MarshalTraces(td) diff --git a/internal/collector/collector_test.go b/internal/collector/collector_test.go new file mode 100644 index 0000000..eea31d2 --- /dev/null +++ b/internal/collector/collector_test.go @@ -0,0 +1,52 @@ +package collector + +import ( + "context" + "testing" + "time" + + "git.tornberg.me/go-otel/internal/websocket" +) + +type mockStore struct { + entries []websocket.HistoryEntry +} + +func (m *mockStore) Append(ctx context.Context, entry websocket.HistoryEntry) error { + m.entries = append(m.entries, entry) + return nil +} + +func (m *mockStore) GetHistory(ctx context.Context, limit int) ([]websocket.HistoryEntry, error) { + return m.entries, nil +} + +func (m *mockStore) Search(ctx context.Context, query string, limit int) ([]websocket.HistoryEntry, error) { + return m.entries, nil +} + +func TestCollectorWithStore(t *testing.T) { + store := &mockStore{} + coll := New(nil, 10, store) + + coll.addToHistory(websocket.DataTypeLogs, map[string]interface{}{"msg": "test"}) + + // Give it a moment since it's asynchronous + timeout := time.After(100 * time.Millisecond) + for { + select { + case <-timeout: + t.Fatal("timed out waiting for entry to be appended to store") + default: + if len(store.entries) > 0 { + goto OK + } + time.Sleep(10 * time.Millisecond) + } + } + +OK: + if store.entries[0].Type != "logs" { + t.Errorf("expected type 'logs', got %s", store.entries[0].Type) + } +} diff --git a/internal/storage/file_store.go b/internal/storage/file_store.go new file mode 100644 index 0000000..ae76d88 --- /dev/null +++ b/internal/storage/file_store.go @@ -0,0 +1,83 @@ +package storage + +import ( + "bufio" + "context" + "encoding/json" + "os" + "strings" + + "git.tornberg.me/go-otel/internal/websocket" + "github.com/natefinch/lumberjack" +) + +type FileStore struct { + logger *lumberjack.Logger +} + +func NewFileStore(filename string) *FileStore { + return &FileStore{ + logger: &lumberjack.Logger{ + Filename: filename, + MaxSize: 100, // megabytes + MaxBackups: 3, + MaxAge: 28, // days + Compress: true, // disabled by default + }, + } +} + +func (fs *FileStore) Append(ctx context.Context, entry websocket.HistoryEntry) error { + data, err := json.Marshal(entry) + if err != nil { + return err + } + _, err = fs.logger.Write(append(data, '\n')) + return err +} + +func (fs *FileStore) GetHistory(ctx context.Context, limit int) ([]websocket.HistoryEntry, error) { + return fs.Search(ctx, "", limit) +} + +func (fs *FileStore) Search(ctx context.Context, query string, limit int) ([]websocket.HistoryEntry, error) { + if limit <= 0 { + limit = 50 // Default to a manageable number + } + + file, err := os.Open(fs.logger.Filename) + if err != nil { + if os.IsNotExist(err) { + return nil, nil // No logs yet + } + return nil, err + } + defer file.Close() + + var matchingLines []string + scanner := bufio.NewScanner(file) + // Lowercase query for case-insensitive search + lowerQuery := strings.ToLower(query) + + for scanner.Scan() { + line := scanner.Text() + if query == "" || strings.Contains(strings.ToLower(line), lowerQuery) { + matchingLines = append(matchingLines, line) + // Only keep the most recent 'limit' matches to save memory + if len(matchingLines) > limit { + matchingLines = matchingLines[1:] + } + } + } + + var results []websocket.HistoryEntry + // Reverse to get latest first (since matchingLines contains them in chronological order) + for i := len(matchingLines) - 1; i >= 0; i-- { + var entry websocket.HistoryEntry + if err := json.Unmarshal([]byte(matchingLines[i]), &entry); err == nil { + results = append(results, entry) + } + } + + return results, scanner.Err() +} diff --git a/internal/storage/file_store_test.go b/internal/storage/file_store_test.go new file mode 100644 index 0000000..fcccf69 --- /dev/null +++ b/internal/storage/file_store_test.go @@ -0,0 +1,64 @@ +package storage + +import ( + "context" + "encoding/json" + "os" + "testing" + + "git.tornberg.me/go-otel/internal/websocket" +) + +func TestFileStore(t *testing.T) { + filename := "test-otel.log" + defer os.Remove(filename) + + fs := NewFileStore(filename) + ctx := context.Background() + + entry1 := websocket.HistoryEntry{ + Type: "logs", + Timestamp: 100, + Data: map[string]interface{}{"msg": "hello world"}, + } + entry2 := websocket.HistoryEntry{ + Type: "traces", + Timestamp: 200, + Data: map[string]interface{}{"msg": "trace event"}, + } + + if err := fs.Append(ctx, entry1); err != nil { + t.Fatalf("failed to append entry1: %v", err) + } + if err := fs.Append(ctx, entry2); err != nil { + t.Fatalf("failed to append entry2: %v", err) + } + + // Test GetHistory (limit 1) + history, err := fs.GetHistory(ctx, 1) + if err != nil { + t.Fatalf("GetHistory failed: %v", err) + } + if len(history) != 1 { + t.Errorf("expected 1 entry, got %d", len(history)) + } + // Should be latest entry + if history[0].Type != "traces" { + t.Errorf("expected latest entry type 'traces', got %s", history[0].Type) + } + + // Test Search + results, err := fs.Search(ctx, "hello", 10) + if err != nil { + t.Fatalf("Search failed: %v", err) + } + if len(results) != 1 { + t.Errorf("expected 1 result for search 'hello', got %d", len(results)) + } + + // Check content of result + data, _ := json.Marshal(results[0].Data) + if string(data) != `{"msg":"hello world"}` { + t.Errorf("unexpected data content: %s", string(data)) + } +} diff --git a/internal/storage/store.go b/internal/storage/store.go new file mode 100644 index 0000000..d4456d5 --- /dev/null +++ b/internal/storage/store.go @@ -0,0 +1,12 @@ +package storage + +import ( + "context" + "git.tornberg.me/go-otel/internal/websocket" +) + +type Store interface { + Append(ctx context.Context, entry websocket.HistoryEntry) error + GetHistory(ctx context.Context, limit int) ([]websocket.HistoryEntry, error) + Search(ctx context.Context, query string, limit int) ([]websocket.HistoryEntry, error) +} diff --git a/k8s/deployment.yaml b/k8s/deployment.yaml index e92abaf..f708bae 100644 --- a/k8s/deployment.yaml +++ b/k8s/deployment.yaml @@ -20,6 +20,9 @@ spec: name: http - containerPort: 4317 name: grpc + env: + - name: GRPC_GO_LOG_SEVERITY_LEVEL + value: "info" resources: requests: memory: "128Mi"