From 77e8afa2e4874ba2ce644a7408dfa8a4fdeb301c Mon Sep 17 00:00:00 2001 From: "google-labs-jules[bot]" <161369871+google-labs-jules[bot]@users.noreply.github.com> Date: Sun, 15 Mar 2026 05:52:46 +0000 Subject: [PATCH] Add ETL Visibility framework documentation and tracking - Create docs/design/ETL_VISIBILITY.md with proposed observability framework - Create docs/design/ETL_TODO.md to track ETL implementation tasks - Add ResourceTracker to pkg/health to track memory, goroutines, and GC stats - Hook ResourceTracker into gateway startup in cmd/picoclaw/internal/gateway Co-authored-by: hobbyistlabs-coder <267281733+hobbyistlabs-coder@users.noreply.github.com> --- cmd/picoclaw/internal/gateway/helpers.go | 6 ++ docs/design/ETL_TODO.md | 26 ++++++++ docs/design/ETL_VISIBILITY.md | 52 ++++++++++++++++ go.mod | 1 - pkg/health/resource_tracker.go | 77 ++++++++++++++++++++++++ pkg/health/resource_tracker_test.go | 47 +++++++++++++++ 6 files changed, 208 insertions(+), 1 deletion(-) create mode 100644 docs/design/ETL_TODO.md create mode 100644 docs/design/ETL_VISIBILITY.md create mode 100644 pkg/health/resource_tracker.go create mode 100644 pkg/health/resource_tracker_test.go diff --git a/cmd/picoclaw/internal/gateway/helpers.go b/cmd/picoclaw/internal/gateway/helpers.go index 88f905edb..19f5fb236 100644 --- a/cmd/picoclaw/internal/gateway/helpers.go +++ b/cmd/picoclaw/internal/gateway/helpers.go @@ -185,6 +185,11 @@ func gatewayCmd(debug bool) error { addr := fmt.Sprintf("%s:%d", cfg.Gateway.Host, cfg.Gateway.Port) channelManager.SetupHTTPServer(addr, healthServer) + // Start resource tracker for ETL Visibility + resourceTracker := health.NewResourceTracker(1 * time.Minute) + resourceTracker.Start(ctx) + fmt.Println("✓ Resource tracker started") + if err := channelManager.StartAll(ctx); err != nil { fmt.Printf("Error starting channels: %v\n", err) return err @@ -214,6 +219,7 @@ func gatewayCmd(debug bool) error { deviceService.Stop() heartbeatService.Stop() cronService.Stop() + resourceTracker.Stop() mediaStore.Stop() agentLoop.Stop() agentLoop.Close() diff --git a/docs/design/ETL_TODO.md b/docs/design/ETL_TODO.md new file mode 100644 index 000000000..c1c2178bc --- /dev/null +++ b/docs/design/ETL_TODO.md @@ -0,0 +1,26 @@ +# ETL Visibility TODO List + +This document tracks the tasks required to implement the "Ultimate Visibility" ETL framework. + +## 1. Extract (Ingestion & Telemetry Collection) + +- [ ] **Structured Logging:** Ensure `zerolog` is used consistently across the codebase for structured JSON logging. Add context to logs where missing (session IDs, tool inputs/outputs). +- [ ] **Basic Metrics Implementation:** Introduce a metrics package (e.g., using `expvar` or a Prometheus client) to expose basic application metrics. +- [x] **Goroutine Tracking:** Implement a metric to track the number of active Goroutines. +- [x] **Memory Tracking:** Implement a metric to track heap allocation and GC pauses. +- [ ] **AgentLoop Telemetry:** Add specific instrumentation to the `AgentLoop` (iteration duration, tool execution duration, failure counts). +- [ ] **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. + +## 2. Transform (Stream Processing & Enrichment) + +- [ ] **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) + +- [ ] **Time-Series Database:** Set up or integrate with a time-series database (e.g., Prometheus) for metrics storage. +- [ ] **Log Warehouse:** Set up or integrate with an OLAP database (e.g., ClickHouse, Elasticsearch) for log and trace storage. +- [ ] **Dashboards:** Create initial Grafana dashboards visualizing the Four Golden Signals (Latency, Traffic, Errors, Saturation). +- [ ] **Alerting:** Configure alerts based on metric thresholds (e.g., GC pauses > 20ms, Goroutine counts continuously rising). diff --git a/docs/design/ETL_VISIBILITY.md b/docs/design/ETL_VISIBILITY.md new file mode 100644 index 000000000..595b6c789 --- /dev/null +++ b/docs/design/ETL_VISIBILITY.md @@ -0,0 +1,52 @@ +# Ultimate Visibility ETL Framework + +## 1. The "Ultimate Visibility" ETL Framework + +To build a robust observability pipeline, we must separate data extraction (telemetry generation) from transformation (aggregation/enrichment) and loading (storage/visualization). + +* **Extract (Ingestion & Telemetry Collection):** + * **Logs:** Capture structured JSON logs natively (e.g., using `zerolog`). This includes session replay data, Chain of Thought (CoT), and tool-call inputs/outputs. + * **Metrics:** Instrument the codebase (especially the Go backend) with OpenTelemetry or Prometheus clients to capture real-time gauges and counters. + * **Traces:** Inject trace IDs at the edge (HTTP/API layer) and pass them down through the context to track requests across the message bus, LLM provider calls, and background asynchronous jobs. +* **Transform (Stream Processing & Enrichment):** + * Use a stream processing engine (like Apache Kafka + Flink, or Vector.dev for lightweight log routing). + * **Normalization:** Convert raw unstructured errors into categorized dimensions (e.g., standardizing errors into `Model Failure`, `Infrastructure Failure`, or `Logic Failure`). + * **Aggregation:** Pre-calculate rolling aggregates, such as 1-minute tumbling windows for request throughput or API failover counts. +* **Load (Storage & Analytics):** + * **Time-Series Data:** Push metric aggregations to a time-series database (e.g., Prometheus, VictoriaMetrics) for low-latency alerting. + * **Log/Event Warehouse:** Push structured JSON logs and traces to an OLAP database (e.g., ClickHouse or Elasticsearch) for deep-dive querying, session replays, and CoT debugging. + * **Presentation:** Layer Grafana or Apache Superset on top to provide single-pane-of-glass dashboards. + +--- + +## 2. Top KPIs for System Health + +For a unified pulse on system health, we track the "Four Golden Signals" tailored to an AI-agent architecture: + +1. **Latency (Response Times):** p50, p90, and p99 durations for API endpoints, LLM provider response times, and full AgentLoop iterations. +2. **Error Rates & Categorization:** Percentage of failed requests, specifically tracking HTTP 5xx errors vs. internal categorizations (e.g., `FailoverTimeout`, `FailoverContextLength`). +3. **Throughput (Traffic):** Requests per second (RPS) at the API gateway, active WebSocket connections, and parallel tool execution batches processed per minute. +4. **Resource Utilization:** CPU saturation, memory allocation (Heap vs. In-use), and concurrent execution primitives (e.g., Goroutine counts). + +--- + +## 3. Deep Dive: Resource Utilization as a Go/No-Go Signal + +Let's focus on **Resource Utilization**, specifically tracking memory footprint and concurrency overhead (Goroutines) in our Go-based backend. + +In highly concurrent systems, it is incredibly easy to introduce silent performance degradation—such as memory leaks from unclosed response bodies or Goroutine leaks from blocked channels (e.g., a blocked `summaryJobs` worker). + +**How Tracking This Provides a 'Go/No-Go' Signal:** + +Imagine we are proposing **[Feature X]: "Real-time Omnichannel Continuous Summarization"**—a feature that spawns a new background process for every active chat session to continuously summarize context using an external Model Context Protocol (MCP) server. + +Before and during the rollout of Feature X, our ETL pipeline focuses on these precise metrics: +* **Goroutine Count:** Is the baseline number of Goroutines stable, or does it climb linearly with time? +* **Heap Memory (In-Use vs. Idle):** Are we seeing aggressive memory spikes that trigger frequent Garbage Collection (GC) pauses? +* **GC Pause Duration:** Are GC pauses exceeding 10-20ms, stealing CPU time from the main event loop? + +**The Decision Matrix:** +* **The "Go" Signal:** We deploy Feature X to a staging/canary environment. The ETL pipeline shows a predictable, flat-line increase in Goroutines (e.g., +1 per active session) that properly scale down when the session terminates. GC pauses remain under 5ms, and overall CPU utilization increases by less than 15%. *Verdict: Safe to merge and roll out to production.* +* **The "No-Go" Signal:** Upon enabling Feature X, our time-series dashboard reveals the Goroutine count continuously climbing even after sessions end (indicating a leak). We see a corresponding sawtooth pattern in Heap Memory, leading to prolonged GC pauses (>50ms) that cause the main `AgentLoop` to miss its SLAs and trigger `FailoverTimeout` errors to the LLM providers. *Verdict: Hard No-Go. The feature introduces severe performance overhead and must be refactored (e.g., by utilizing a fixed-size worker pool instead of unbounded Goroutines) before it can be shipped.* + +By relying on this hard data extracted from the application layer, transformed into actionable percentiles, and loaded into real-time dashboards, engineering leadership no longer has to guess about system impact. The metrics make the deployment decisions for us. \ No newline at end of file diff --git a/go.mod b/go.mod index e209afd9c..055bc6948 100644 --- a/go.mod +++ b/go.mod @@ -4,7 +4,6 @@ go 1.25.7 require ( github.com/adhocore/gronx v1.19.6 - github.com/alecthomas/kong v1.14.0 github.com/alpacahq/alpaca-trade-api-go/v3 v3.9.1 github.com/anthropics/anthropic-sdk-go v1.22.1 github.com/bwmarrin/discordgo v0.29.0 diff --git a/pkg/health/resource_tracker.go b/pkg/health/resource_tracker.go new file mode 100644 index 000000000..2ce4291a5 --- /dev/null +++ b/pkg/health/resource_tracker.go @@ -0,0 +1,77 @@ +package health + +import ( + "context" + "runtime" + "sync" + "time" + + "jane/pkg/logger" +) + +// ResourceTracker tracks and logs basic system resource usage over time. +// This is part of the ETL Ultimate Visibility framework to monitor Go/No-Go signals +// such as Goroutine leaks and memory spikes. +type ResourceTracker struct { + interval time.Duration + stopCh chan struct{} + stopOnce sync.Once +} + +// NewResourceTracker creates a new ResourceTracker that logs metrics every `interval`. +func NewResourceTracker(interval time.Duration) *ResourceTracker { + if interval == 0 { + interval = 60 * time.Second // Default to 1 minute + } + return &ResourceTracker{ + interval: interval, + stopCh: make(chan struct{}), + } +} + +// Start begins tracking resources in a background goroutine. +func (rt *ResourceTracker) Start(ctx context.Context) { + ticker := time.NewTicker(rt.interval) + go func() { + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-rt.stopCh: + return + case <-ticker.C: + rt.logResources() + } + } + }() +} + +// Stop gracefully stops the resource tracker. +func (rt *ResourceTracker) Stop() { + rt.stopOnce.Do(func() { + close(rt.stopCh) + }) +} + +func (rt *ResourceTracker) logResources() { + var m runtime.MemStats + runtime.ReadMemStats(&m) + + goroutines := runtime.NumGoroutine() + + // Convert bytes to megabytes for readability in logs + allocMB := float64(m.Alloc) / 1024 / 1024 + totalAllocMB := float64(m.TotalAlloc) / 1024 / 1024 + sysMB := float64(m.Sys) / 1024 / 1024 + + logger.InfoCF("SystemHealth", "Resource tracking telemetry", map[string]any{ + "goroutines": goroutines, + "memory_alloc_mb": allocMB, + "memory_total_mb": totalAllocMB, + "memory_sys_mb": sysMB, + "num_gc": m.NumGC, + "gc_pause_ns": m.PauseNs[(m.NumGC+255)%256], // Latest GC pause time + "gc_pause_total_ns": m.PauseTotalNs, + }) +} diff --git a/pkg/health/resource_tracker_test.go b/pkg/health/resource_tracker_test.go new file mode 100644 index 000000000..9f1717d54 --- /dev/null +++ b/pkg/health/resource_tracker_test.go @@ -0,0 +1,47 @@ +package health + +import ( + "context" + "runtime" + "testing" + "time" + + "github.com/stretchr/testify/assert" +) + +func TestResourceTracker_StartStop(t *testing.T) { + interval := 10 * time.Millisecond + rt := NewResourceTracker(interval) + assert.NotNil(t, rt) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + rt.Start(ctx) + + // Wait on stop channel to simulate proper lifecycle management + // without flakiness + rt.Stop() + + // Test idempotent Stop + rt.Stop() + assert.NotPanics(t, func() { + rt.Stop() + }) +} + +func TestResourceTracker_logResources(t *testing.T) { + rt := NewResourceTracker(1 * time.Second) + + // Since we can't easily assert on the stdout without mocking zerolog globally, + // we just ensure the function runs without panicking. + assert.NotPanics(t, func() { + rt.logResources() + }) + + // Verify reasonable values are accessible + var m runtime.MemStats + runtime.ReadMemStats(&m) + assert.GreaterOrEqual(t, m.Alloc, uint64(0)) + assert.GreaterOrEqual(t, runtime.NumGoroutine(), 1) +}