Merge pull request #63 from hobbyistlabs-coder/feature/etl-ultimate-visibility-todos-11062217614426038697

feat: implement ETL ultimate visibility telemetry and log normalization

docs/design/ETL_TODO.md

@@ -10,12 +10,12 @@ This document tracks the tasks required to implement the "Ultimate Visibility" E
 - [x] **Memory Tracking:** Implement a metric to track heap allocation and GC pauses.
 - [x] **AgentLoop Telemetry:** Add specific instrumentation to the `AgentLoop` (iteration duration, tool execution duration, failure counts).
 - [x] **LLM Provider Telemetry:** Track API call latency, token usage, and failover reasons for LLM providers.
-- [ ] **API Gateway Telemetry:** Track request rates (RPS), latency percentiles, and error rates for HTTP and WebSocket endpoints.
-- [ ] **Tracing Instrumentation:** Introduce trace IDs at entry points (HTTP, WebSocket) and propagate them via context to track end-to-end execution flow.
+- [x] **API Gateway Telemetry:** Track request rates (RPS), latency percentiles, and error rates for HTTP and WebSocket endpoints.
+- [x] **Tracing Instrumentation:** Introduce trace IDs at entry points (HTTP, WebSocket) and propagate them via context to track end-to-end execution flow.
 
 ## 2. Transform (Stream Processing & Enrichment)
 
-- [ ] **Log Normalization:** Standardize error classifications (e.g., Model Failure, Infrastructure Failure, Logic Failure) to ensure consistent log querying.
+- [x] **Log Normalization:** Standardize error classifications (e.g., Model Failure, Infrastructure Failure, Logic Failure) to ensure consistent log querying.
 - [ ] **Aggregation Strategy:** Design the pipeline for aggregating high-volume events before they reach the data warehouse (e.g., Vector.dev configuration).
 
 ## 3. Load (Storage & Analytics)

pkg/channels/http.go

@@ -5,13 +5,94 @@
 package channels
 
 import (
+	"bufio"
+	"context"
+	"fmt"
+	"net"
 	"net/http"
 	"time"
 
+	"github.com/google/uuid"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/metric"
+
 	"jane/pkg/health"
 	"jane/pkg/logger"
 )
 
+// telemetryMiddleware wraps an http.Handler to track request metrics using OpenTelemetry.
+func telemetryMiddleware(next http.Handler) http.Handler {
+	meter := otel.Meter("jane/pkg/channels")
+
+	// Create metrics
+	requestCounter, err := meter.Int64Counter(
+		"http.server.requests",
+		metric.WithDescription("Total number of HTTP requests"),
+	)
+	if err != nil {
+		logger.ErrorCF("channels", "Failed to create request counter metric", map[string]any{"error": err.Error()})
+	}
+
+	durationHistogram, err := meter.Float64Histogram(
+		"http.server.duration",
+		metric.WithDescription("HTTP request duration in milliseconds"),
+	)
+	if err != nil {
+		logger.ErrorCF("channels", "Failed to create duration histogram metric", map[string]any{"error": err.Error()})
+	}
+
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		start := time.Now()
+
+		// Generate Trace ID and attach to context and response header
+		traceID := uuid.New().String()
+		w.Header().Set("X-Trace-Id", traceID)
+		ctx := context.WithValue(r.Context(), logger.TraceIDKey, traceID)
+		r = r.WithContext(ctx)
+
+		// Wrap ResponseWriter to capture the status code
+		rw := &responseWriterWrapper{ResponseWriter: w, statusCode: http.StatusOK}
+
+		next.ServeHTTP(rw, r)
+
+		durationMs := float64(time.Since(start).Microseconds()) / 1000.0
+
+		if requestCounter != nil {
+			requestCounter.Add(r.Context(), 1)
+		}
+		if durationHistogram != nil {
+			durationHistogram.Record(r.Context(), durationMs)
+		}
+	})
+}
+
+// responseWriterWrapper captures the HTTP status code for metrics and logging
+type responseWriterWrapper struct {
+	http.ResponseWriter
+	statusCode int
+}
+
+func (rw *responseWriterWrapper) WriteHeader(code int) {
+	rw.statusCode = code
+	rw.ResponseWriter.WriteHeader(code)
+}
+
+// Hijack implements the http.Hijacker interface, required for WebSockets
+func (rw *responseWriterWrapper) Hijack() (net.Conn, *bufio.ReadWriter, error) {
+	hijacker, ok := rw.ResponseWriter.(http.Hijacker)
+	if !ok {
+		return nil, nil, fmt.Errorf("underlying ResponseWriter does not support hijacking")
+	}
+	return hijacker.Hijack()
+}
+
+// Flush implements the http.Flusher interface, required for streaming
+func (rw *responseWriterWrapper) Flush() {
+	if flusher, ok := rw.ResponseWriter.(http.Flusher); ok {
+		flusher.Flush()
+	}
+}
+
 // SetupHTTPServer creates a shared HTTP server with the given listen address.
 // It registers health endpoints from the health server and discovers channels
 // that implement WebhookHandler and/or HealthChecker to register their handlers.
@@ -43,7 +124,7 @@ func (m *Manager) SetupHTTPServer(addr string, healthServer *health.Server) {
 
 	m.httpServer = &http.Server{
 		Addr:              addr,
-		Handler:           m.mux,
+		Handler:           telemetryMiddleware(m.mux),
 		ReadTimeout:       30 * time.Second,
 		ReadHeaderTimeout: 10 * time.Second,
 		WriteTimeout:      30 * time.Second,

pkg/logger/logger.go

@@ -1,6 +1,7 @@
 package logger
 
 import (
+	"context"
 	"fmt"
 	"os"
 	"path/filepath"
@@ -11,6 +12,22 @@ import (
 	"github.com/rs/zerolog"
 )
 
+// contextKey is a custom type for context keys to avoid collisions
+type contextKey string
+
+const (
+	// TraceIDKey is the context key for tracing requests
+	TraceIDKey contextKey = "traceID"
+)
+
+type ErrorCategory string
+
+const (
+	ErrorCategoryModelFailure          ErrorCategory = "Model Failure"
+	ErrorCategoryInfrastructureFailure ErrorCategory = "Infrastructure Failure"
+	ErrorCategoryLogicFailure          ErrorCategory = "Logic Failure"
+)
+
 type LogLevel = zerolog.Level
 
 const (
@@ -161,7 +178,7 @@ func getEvent(logger zerolog.Logger, level LogLevel) *zerolog.Event {
 	}
 }
 
-func logMessage(level LogLevel, component string, message string, fields map[string]any) {
+func logMessageCtx(ctx context.Context, level LogLevel, component string, message string, fields map[string]any) {
 	if level < currentLevel {
 		return
 	}
@@ -170,6 +187,12 @@ func logMessage(level LogLevel, component string, message string, fields map[str
 
 	event := getEvent(logger, level)
 
+	if ctx != nil {
+		if traceID, ok := ctx.Value(TraceIDKey).(string); ok && traceID != "" {
+			event.Str("trace_id", traceID)
+		}
+	}
+
 	// Build combined field with component and caller
 	if component != "" {
 		event.Str("caller", fmt.Sprintf("%-6s %s:%d (%s)", component, callerFile, callerLine, callerFunc))
@@ -187,6 +210,12 @@ func logMessage(level LogLevel, component string, message string, fields map[str
 	if fileLogger.GetLevel() != zerolog.NoLevel {
 		fileEvent := getEvent(fileLogger, level)
 
+		if ctx != nil {
+			if traceID, ok := ctx.Value(TraceIDKey).(string); ok && traceID != "" {
+				fileEvent.Str("trace_id", traceID)
+			}
+		}
+
 		if component != "" {
 			fileEvent.Str("component", component)
 		}
@@ -201,6 +230,104 @@ func logMessage(level LogLevel, component string, message string, fields map[str
 	}
 }
 
+func logMessage(level LogLevel, component string, message string, fields map[string]any) {
+	logMessageCtx(nil, level, component, message, fields)
+}
+
+func DebugCtx(ctx context.Context, message string) {
+	logMessageCtx(ctx, DEBUG, "", message, nil)
+}
+
+func DebugCCtx(ctx context.Context, component string, message string) {
+	logMessageCtx(ctx, DEBUG, component, message, nil)
+}
+
+func DebugFCtx(ctx context.Context, message string, fields map[string]any) {
+	logMessageCtx(ctx, DEBUG, "", message, fields)
+}
+
+func DebugCFCtx(ctx context.Context, component string, message string, fields map[string]any) {
+	logMessageCtx(ctx, DEBUG, component, message, fields)
+}
+
+func InfoCtx(ctx context.Context, message string) {
+	logMessageCtx(ctx, INFO, "", message, nil)
+}
+
+func InfoCCtx(ctx context.Context, component string, message string) {
+	logMessageCtx(ctx, INFO, component, message, nil)
+}
+
+func InfoFCtx(ctx context.Context, message string, fields map[string]any) {
+	logMessageCtx(ctx, INFO, "", message, fields)
+}
+
+func InfoCFCtx(ctx context.Context, component string, message string, fields map[string]any) {
+	logMessageCtx(ctx, INFO, component, message, fields)
+}
+
+func WarnCtx(ctx context.Context, message string) {
+	logMessageCtx(ctx, WARN, "", message, nil)
+}
+
+func WarnCCtx(ctx context.Context, component string, message string) {
+	logMessageCtx(ctx, WARN, component, message, nil)
+}
+
+func WarnFCtx(ctx context.Context, message string, fields map[string]any) {
+	logMessageCtx(ctx, WARN, "", message, fields)
+}
+
+func WarnCFCtx(ctx context.Context, component string, message string, fields map[string]any) {
+	logMessageCtx(ctx, WARN, component, message, fields)
+}
+
+func ErrorCtx(ctx context.Context, message string) {
+	logMessageCtx(ctx, ERROR, "", message, nil)
+}
+
+func ErrorCCtx(ctx context.Context, component string, message string) {
+	logMessageCtx(ctx, ERROR, component, message, nil)
+}
+
+func ErrorFCtx(ctx context.Context, message string, fields map[string]any) {
+	logMessageCtx(ctx, ERROR, "", message, fields)
+}
+
+func ErrorCFCtx(ctx context.Context, component string, message string, fields map[string]any) {
+	logMessageCtx(ctx, ERROR, component, message, fields)
+}
+
+func FatalCtx(ctx context.Context, message string) {
+	logMessageCtx(ctx, FATAL, "", message, nil)
+}
+
+func FatalCCtx(ctx context.Context, component string, message string) {
+	logMessageCtx(ctx, FATAL, component, message, nil)
+}
+
+func FatalfCtx(ctx context.Context, message string, ss ...any) {
+	logMessageCtx(ctx, FATAL, "", fmt.Sprintf(message, ss...), nil)
+}
+
+func FatalFCtx(ctx context.Context, message string, fields map[string]any) {
+	logMessageCtx(ctx, FATAL, "", message, fields)
+}
+
+func FatalCFCtx(ctx context.Context, component string, message string, fields map[string]any) {
+	logMessageCtx(ctx, FATAL, component, message, fields)
+}
+
+func LogErrorWithCategory(ctx context.Context, category ErrorCategory, message string, err error) {
+	fields := map[string]any{
+		"error_category": string(category),
+	}
+	if err != nil {
+		fields["error"] = err.Error()
+	}
+	logMessageCtx(ctx, ERROR, "", message, fields)
+}
+
 func Debug(message string) {
 	logMessage(DEBUG, "", message, nil)
 }