diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..a2ad30b --- /dev/null +++ b/Makefile @@ -0,0 +1,8 @@ +.PHONY: lint install-lint + +lint: + @which golangci-lint > /dev/null || (echo "golangci-lint not found, installing..." && make install-lint) + golangci-lint run ./... + +install-lint: + go install github.com/golangci/golangci-lint/cmd/golangci-lint@latest diff --git a/cmd/server/main.go b/cmd/server/main.go index 570483b..d106b1c 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -1,3 +1,4 @@ +// Package main implements the Go OTel Debugger server. package main import ( @@ -46,8 +47,7 @@ func (h *connStatsHandler) TagConn(ctx context.Context, info *stats.ConnTagInfo) } func (h *connStatsHandler) HandleConn(ctx context.Context, s stats.ConnStats) { - switch s.(type) { - case *stats.ConnEnd: + if _, ok := s.(*stats.ConnEnd); ok { log.Printf("gRPC CONN [CLOSED]") } } @@ -55,6 +55,7 @@ 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) HandleRPC(ctx context.Context, s stats.RPCStats) {} type traceService struct { diff --git a/internal/collector/collector.go b/internal/collector/collector.go index 2fb8713..24254cf 100644 --- a/internal/collector/collector.go +++ b/internal/collector/collector.go @@ -1,3 +1,5 @@ +// Package collector implements an OpenTelemetry consumer that stores telemetry in history +// and broadcasts it to connected clients. package collector import ( @@ -18,6 +20,7 @@ import ( "go.opentelemetry.io/collector/pdata/ptrace" ) +// Collector handles the collection, storage and broadcasting of telemetry data. type Collector struct { broadcaster websocket.Broadcaster historySize int @@ -28,6 +31,7 @@ type Collector struct { subMutex sync.Mutex } +// New creates a new Collector instance. func New(broadcaster websocket.Broadcaster, historySize int, store storage.Store) *Collector { return &Collector{ broadcaster: broadcaster, @@ -38,6 +42,7 @@ func New(broadcaster websocket.Broadcaster, historySize int, store storage.Store } } +// SetBroadcaster updates the broadcaster used by the collector. func (c *Collector) SetBroadcaster(broadcaster websocket.Broadcaster) { c.broadcaster = broadcaster } @@ -82,6 +87,8 @@ func (c *Collector) notifySubscribers(entry websocket.HistoryEntry) { } } +// Subscribe adds a new history subscriber with the given filter. +// It returns a channel for receiving filtered history entries and a cleanup function. func (c *Collector) Subscribe(filter func(websocket.HistoryEntry) bool) (chan websocket.HistoryEntry, func()) { ch := make(chan websocket.HistoryEntry, 1) c.subMutex.Lock() @@ -98,6 +105,7 @@ func (c *Collector) Subscribe(filter func(websocket.HistoryEntry) bool) (chan we return ch, cleanup } +// GetHistory returns the last 'limit' history entries. func (c *Collector) GetHistory(limit int) []websocket.HistoryEntry { if limit <= 0 { limit = c.historySize @@ -121,6 +129,7 @@ func (c *Collector) GetHistory(limit int) []websocket.HistoryEntry { return history } +// SearchHistory searches through stored telemetry data using the provided query. func (c *Collector) SearchHistory(ctx context.Context, query string, limit int) ([]websocket.HistoryEntry, error) { if limit <= 0 { limit = 50 @@ -162,6 +171,7 @@ func (c *Collector) SearchHistory(ctx context.Context, query string, limit int) return filtered, nil } +// ConsumeTraces implements the consumer.Traces interface. func (c *Collector) ConsumeTraces(ctx context.Context, td ptrace.Traces) error { marshaler := &ptrace.JSONMarshaler{} buf, err := marshaler.MarshalTraces(td) @@ -187,6 +197,7 @@ func (c *Collector) ConsumeTraces(ctx context.Context, td ptrace.Traces) error { return nil } +// ConsumeMetrics implements the consumer.Metrics interface. func (c *Collector) ConsumeMetrics(ctx context.Context, md pmetric.Metrics) error { marshaler := &pmetric.JSONMarshaler{} buf, err := marshaler.MarshalMetrics(md) @@ -212,6 +223,7 @@ func (c *Collector) ConsumeMetrics(ctx context.Context, md pmetric.Metrics) erro return nil } +// ConsumeLogs implements the consumer.Logs interface. func (c *Collector) ConsumeLogs(ctx context.Context, ld plog.Logs) error { marshaler := &plog.JSONMarshaler{} buf, err := marshaler.MarshalLogs(ld) @@ -238,7 +250,7 @@ func (c *Collector) ConsumeLogs(ctx context.Context, ld plog.Logs) error { return nil } -// Implement consumer interfaces +// Capabilities returns the capabilities of the collector. func (c *Collector) Capabilities() consumer.Capabilities { return consumer.Capabilities{MutatesData: false} } diff --git a/internal/collector/handlers.go b/internal/collector/handlers.go index e6dc7ac..b26c0ca 100644 --- a/internal/collector/handlers.go +++ b/internal/collector/handlers.go @@ -1,3 +1,5 @@ +// Package collector implements an OpenTelemetry consumer that stores telemetry in history +// and broadcasts it to connected clients. package collector import ( @@ -16,6 +18,7 @@ import ( "go.opentelemetry.io/collector/pdata/ptrace" ) +// HandleTraces handles OTLP trace requests. 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 { @@ -61,6 +64,7 @@ func (c *Collector) HandleTraces(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) } +// HandleMetrics handles OTLP metric requests. func (c *Collector) HandleMetrics(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) @@ -101,6 +105,7 @@ func (c *Collector) HandleMetrics(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) } +// HandleLogs handles OTLP log requests. func (c *Collector) HandleLogs(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) @@ -141,6 +146,7 @@ func (c *Collector) HandleLogs(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) } +// HandleHistory returns the telemetry history as JSON. func (c *Collector) HandleHistory(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) @@ -165,6 +171,7 @@ func (c *Collector) HandleHistory(w http.ResponseWriter, r *http.Request) { } } +// HandleSearch searches the telemetry history. func (c *Collector) HandleSearch(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) @@ -195,6 +202,7 @@ func (c *Collector) HandleSearch(w http.ResponseWriter, r *http.Request) { } } +// HandleWait waits for a specific telemetry signal. func (c *Collector) HandleWait(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) @@ -216,7 +224,9 @@ func (c *Collector) HandleWait(w http.ResponseWriter, r *http.Request) { 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]) + if err := json.NewEncoder(w).Encode(matches[0]); err != nil { + log.Printf("failed to encode match: %v", err) + } return } @@ -242,6 +252,7 @@ func (c *Collector) HandleWait(w http.ResponseWriter, r *http.Request) { } } +// HandleHistoryByType returns telemetry history filtered by type. func (c *Collector) HandleHistoryByType(dataType string) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { @@ -283,8 +294,11 @@ func (c *Collector) HandleHistoryByType(dataType string) http.HandlerFunc { } } +// HandleHealth returns the health status of the collector. 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"}) + if err := json.NewEncoder(w).Encode(map[string]string{"status": "ok"}); err != nil { + log.Printf("failed to encode health check: %v", err) + } } diff --git a/internal/storage/file_store.go b/internal/storage/file_store.go index db7fee3..3497dbd 100644 --- a/internal/storage/file_store.go +++ b/internal/storage/file_store.go @@ -1,3 +1,4 @@ +// Package storage implements persistent storage for telemetry data. package storage import ( @@ -11,10 +12,12 @@ import ( "github.com/natefinch/lumberjack" ) +// FileStore implements the Store interface using a file-based logger. type FileStore struct { logger *lumberjack.Logger } +// NewFileStore creates a new FileStore instance that writes to filename. func NewFileStore(filename string) *FileStore { return &FileStore{ logger: &lumberjack.Logger{ @@ -27,6 +30,7 @@ func NewFileStore(filename string) *FileStore { } } +// Append saves a new history entry to the file store. func (fs *FileStore) Append(ctx context.Context, entry websocket.HistoryEntry) error { data, err := json.Marshal(entry) if err != nil { @@ -36,10 +40,12 @@ func (fs *FileStore) Append(ctx context.Context, entry websocket.HistoryEntry) e return err } +// GetHistory retrieves historical entries from the file store. func (fs *FileStore) GetHistory(ctx context.Context, limit int) ([]websocket.HistoryEntry, error) { return fs.Search(ctx, "", limit) } +// Search searches for historical entries in the file store matching the query. func (fs *FileStore) Search(ctx context.Context, query string, limit int) ([]websocket.HistoryEntry, error) { if limit <= 0 { limit = 50 // Default to a manageable number @@ -52,7 +58,9 @@ func (fs *FileStore) Search(ctx context.Context, query string, limit int) ([]web } return nil, err } - defer file.Close() + defer func() { + _ = file.Close() + }() var matchingLines []string scanner := bufio.NewScanner(file) diff --git a/internal/storage/file_store_test.go b/internal/storage/file_store_test.go index fcccf69..1cc1bf8 100644 --- a/internal/storage/file_store_test.go +++ b/internal/storage/file_store_test.go @@ -11,7 +11,9 @@ import ( func TestFileStore(t *testing.T) { filename := "test-otel.log" - defer os.Remove(filename) + defer func() { + _ = os.Remove(filename) + }() fs := NewFileStore(filename) ctx := context.Background() diff --git a/internal/storage/store.go b/internal/storage/store.go index d4456d5..67f0bd5 100644 --- a/internal/storage/store.go +++ b/internal/storage/store.go @@ -5,6 +5,7 @@ import ( "git.tornberg.me/go-otel/internal/websocket" ) +// Store interface for telemetry history persistence. type Store interface { Append(ctx context.Context, entry websocket.HistoryEntry) error GetHistory(ctx context.Context, limit int) ([]websocket.HistoryEntry, error) diff --git a/internal/websocket/server.go b/internal/websocket/server.go index e6f2437..da6168a 100644 --- a/internal/websocket/server.go +++ b/internal/websocket/server.go @@ -1,3 +1,4 @@ +// Package websocket handles WebSocket connections and broadcasting telemetry updates to clients. package websocket import ( @@ -9,29 +10,35 @@ import ( "github.com/gorilla/websocket" ) +// DataType represents the type of telemetry data. type DataType string +// Telemetry data types. const ( DataTypeTraces DataType = "traces" DataTypeMetrics DataType = "metrics" DataTypeLogs DataType = "logs" ) +// HistoryProvider provides access to historical telemetry entries. type HistoryProvider interface { GetHistory(limit int) []HistoryEntry } +// HistoryEntry represents a single archived telemetry entry. type HistoryEntry struct { Type string `json:"type"` Timestamp int64 `json:"timestamp"` Data interface{} `json:"data"` } +// Entry represents a telemetry update sent to clients. type Entry struct { Type string `json:"type"` Data interface{} `json:"data"` } +// Broadcaster interface for sending telemetry updates to all connected clients. type Broadcaster interface { Broadcast(data *Entry) } @@ -42,12 +49,14 @@ var upgrader = websocket.Upgrader{ }, } +// Server manages WebSocket client connections and broadcasts. type Server struct { clients map[*websocket.Conn]bool mutex sync.Mutex historyProvider HistoryProvider } +// New creates a new WebSocket Server instance. func New(historyProvider HistoryProvider) *Server { return &Server{ @@ -56,13 +65,18 @@ func New(historyProvider HistoryProvider) *Server { } } +// HandleConnections upgrades HTTP requests to WebSocket and manages client lifecycle. func (s *Server) HandleConnections(w http.ResponseWriter, r *http.Request) { ws, err := upgrader.Upgrade(w, r, nil) if err != nil { log.Printf("failed to upgrade connection: %v", err) return } - defer ws.Close() + defer func() { + if err := ws.Close(); err != nil { + log.Printf("failed to close connection: %v", err) + } + }() s.mutex.Lock() s.clients[ws] = true @@ -98,6 +112,7 @@ func (s *Server) HandleConnections(w http.ResponseWriter, r *http.Request) { log.Printf("client disconnected") } +// Broadcast sends a telemetry entry to all connected clients. func (s *Server) Broadcast(entry *Entry) { s.mutex.Lock() defer s.mutex.Unlock() @@ -113,6 +128,7 @@ func (s *Server) Broadcast(entry *Entry) { } } +// SendHistory sends historical entries to a specific WebSocket client. func (s *Server) SendHistory(ws *websocket.Conn) { s.mutex.Lock() defer s.mutex.Unlock() @@ -131,6 +147,7 @@ func (s *Server) SendHistory(ws *websocket.Conn) { } } +// HandleMessage processes incoming messages from a WebSocket client. func (s *Server) HandleMessage(ws *websocket.Conn, msg []byte) { var message map[string]interface{} if err := json.Unmarshal(msg, &message); err != nil { @@ -148,6 +165,7 @@ func (s *Server) HandleMessage(ws *websocket.Conn, msg []byte) { } } +// SendPong sends a pong message back to a client. func (s *Server) SendPong(ws *websocket.Conn) { if err := ws.WriteMessage(websocket.PongMessage, nil); err != nil { log.Printf("error sending pong: %v", err)