diff --git a/docs/design/ETL_TODO.md b/docs/design/ETL_TODO.md index aa36c1498..28ed97a9b 100644 --- a/docs/design/ETL_TODO.md +++ b/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) diff --git a/pkg/channels/http.go b/pkg/channels/http.go index b4d5ef6b5..0e7ce4bb0 100644 --- a/pkg/channels/http.go +++ b/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, diff --git a/pkg/logger/logger.go b/pkg/logger/logger.go index df83ad263..cca454a3a 100644 --- a/pkg/logger/logger.go +++ b/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) }