Advanced Observability Guide¶

Monitor your AI agents, RAG pipelines, and integration workflows in production. This guide covers SLA metrics, distributed tracing, backlog debugging, audit logging, Grafana dashboards, and Prometheus alerting — giving you full visibility into agent behavior, retrieval quality, and pipeline performance.

Agent-Level Tracing¶

Helix provides built-in execution tracing for AI agents via TraceCollector. Every LLM call, tool invocation, approval decision, and error is captured with timing:

collector := adk.NewTraceCollector()
agent := adk.New(adk.AgentConfig{
    TraceCollector: collector,
    // ... model, tools, etc.
})

agent.Run(ctx, "query")
trace := collector.Last()
// trace.Steps: [{Type: "llm_call", Duration: 1.9s}, {Type: "tool_call", ToolName: "search", Duration: 5ms}, ...]
// trace.TotalTokens, trace.TotalCost, trace.Duration

Token usage and cost tracking are built into the LLM providers (OpenAI, Anthropic) and available on every Message.Usage.

1. SLA Metrics¶

The SLAInterceptor (in pkg/metrics/sla.go) monitors per-workflow SLA compliance by tracking error rates, latency percentiles, and threshold violations. It implements the engine.Interceptor interface and plugs into the exchange processing pipeline.

SLAConfig¶

Each workflow can have its own SLA thresholds defined through SLAConfig:

type SLAConfig struct {
    // MaxLatency is the maximum acceptable latency. Exchanges exceeding this
    // are counted as SLA violations.
    MaxLatency time.Duration

    // MaxErrorRate is the maximum acceptable error rate (0.0-1.0).
    // When exceeded, an SLA violation event is triggered.
    MaxErrorRate float64
}

Setting Up the SLAInterceptor¶

package main

import (
    "context"
    "fmt"
    "time"

    "github.com/arvenza/helix/pkg/engine"
    "github.com/arvenza/helix/pkg/metrics"
)

func main() {
    collector := metrics.NewCollector()
    slaInterceptor := metrics.NewSLAInterceptor(collector)

    // Define SLA thresholds per workflow
    slaInterceptor.SetSLA("order-processing", metrics.SLAConfig{
        MaxLatency:   500 * time.Millisecond,
        MaxErrorRate: 0.02, // 2%
    })
    slaInterceptor.SetSLA("payment-gateway", metrics.SLAConfig{
        MaxLatency:   1 * time.Second,
        MaxErrorRate: 0.01, // 1%
    })

    // Register a violation callback
    slaInterceptor.OnViolation = func(workflow string, violation string) {
        fmt.Printf("SLA VIOLATION [%s]: %s\n", workflow, violation)
        // Send to PagerDuty, Slack, etc.
    }

    // Start periodic gauge updates (every 10 seconds by default)
    ctx := context.Background()
    slaInterceptor.Start(ctx)
    defer slaInterceptor.Stop(ctx)

    // Register the interceptor with the engine
    hc := engine.NewContext("my-app")
    hc.AddInterceptor(slaInterceptor)

    // ... build and start workflows ...
}

Violation Callbacks¶

The OnViolation callback fires in two cases:

Latency violation: When an individual exchange exceeds MaxLatency.
Error rate violation: When the cumulative error rate exceeds MaxErrorRate (only evaluated after at least 100 exchanges to avoid false positives).

The callback receives the workflow ID and a human-readable description:

slaInterceptor.OnViolation = func(workflow string, violation string) {
    // violation examples:
    //   "latency 623ms exceeded threshold 500ms"
    //   "error rate 3.45% exceeded threshold 2.00%"
    alertManager.Fire(workflow, violation)
}

SLA Summary API¶

Retrieve a summary for any workflow programmatically:

summary := slaInterceptor.SLASummary("order-processing")
fmt.Printf("Workflow:    %s\n", summary.Workflow)
fmt.Printf("Errors:      %d\n", summary.TotalErrors)
fmt.Printf("Successes:   %d\n", summary.TotalSuccess)
fmt.Printf("Error Rate:  %.2f%%\n", summary.ErrorRate*100)
fmt.Printf("P50 Latency: %.3fs\n", summary.P50Latency)
fmt.Printf("P95 Latency: %.3fs\n", summary.P95Latency)
fmt.Printf("P99 Latency: %.3fs\n", summary.P99Latency)
fmt.Printf("Violations:  %d\n", summary.Violations)

The RouteSLASummary struct is JSON-serializable, making it suitable for the management REST API:

type RouteSLASummary struct {
    Workflow     string  `json:"workflow"`
    TotalErrors  int64   `json:"total_errors"`
    TotalSuccess int64   `json:"total_successes"`
    ErrorRate    float64 `json:"error_rate"`
    P50Latency   float64 `json:"p50_latency_seconds"`
    P95Latency   float64 `json:"p95_latency_seconds"`
    P99Latency   float64 `json:"p99_latency_seconds"`
    Violations   int64   `json:"sla_violations"`
}

Prometheus Metric Names¶

The SLA interceptor emits the following metrics:

Metric	Type	Labels	Description
`helix_sla_violations_total`	counter	`workflow`, `type`	Total SLA violations (`type` is `latency` or `error_rate`)
`helix_route_errors_total`	counter	`workflow`	Total errors per workflow
`helix_route_successes_total`	counter	`workflow`	Total successes per workflow
`helix_route_p99_latency_seconds`	gauge	`workflow`	Estimated p99 latency
`helix_route_p95_latency_seconds`	gauge	`workflow`	Estimated p95 latency
`helix_route_p50_latency_seconds`	gauge	`workflow`	Estimated p50 latency

2. Per-Workflow Latency Percentiles¶

PercentileEstimator¶

The PercentileEstimator is a streaming percentile calculator backed by a fixed-size circular buffer. It keeps the most recent N samples (default: 1000) and computes percentiles on demand by sorting a copy of the active window.

est := metrics.NewPercentileEstimator(2000) // keep last 2000 samples

// Record observations (typically done by SLAInterceptor automatically)
est.Observe(0.045)  // 45ms
est.Observe(0.120)  // 120ms
est.Observe(0.003)  // 3ms

// Query percentiles
p50 := est.Percentile(0.50) // median
p95 := est.Percentile(0.95)
p99 := est.Percentile(0.99)

// Check sample count
count := est.Count() // up to buffer size

The estimator is thread-safe (guarded by sync.Mutex) and is automatically created per workflow by the SLAInterceptor using sync.Map for lock-free workflow-level lookups.

How Gauge Updates Work¶

The SLAInterceptor runs a background goroutine (started by Start()) that periodically iterates over all per-workflow estimators and pushes the latest p50/p95/p99 values into Prometheus gauges:

Every 10 seconds (default updateInterval):
  for each workflow estimator:
    helix_workflow_p50_latency_seconds{workflow="..."} = estimator.Percentile(0.50)
    helix_workflow_p95_latency_seconds{workflow="..."} = estimator.Percentile(0.95)
    helix_workflow_p99_latency_seconds{workflow="..."} = estimator.Percentile(0.99)

This approach avoids per-exchange gauge writes while keeping Prometheus data reasonably fresh.

Combining with MetricsInterceptor¶

For full observability, stack both interceptors. They share the same start-time property on the exchange:

collector := metrics.NewCollector()

metricsInt := metrics.NewMetricsInterceptor(collector)
slaInt := metrics.NewSLAInterceptor(collector)
slaInt.SetSLA("my-workflow", metrics.SLAConfig{
    MaxLatency:   200 * time.Millisecond,
    MaxErrorRate: 0.05,
})

hc := engine.NewContext("my-app")
hc.AddInterceptor(metricsInt) // counters, histograms, inflight
hc.AddInterceptor(slaInt)     // SLA checks, percentile gauges

// Expose /metrics
http.Handle("/metrics", collector.Handler())

3. Trace Sampling¶

AlwaysSample¶

Records every span. Suitable for development and low-traffic environments:

tracer := tracing.NewTracer("my-service", tracing.AlwaysSample{}, exporter)

NeverSample¶

Drops all spans. Useful for disabling tracing without removing instrumentation:

tracer := tracing.NewTracer("my-service", tracing.NeverSample{})

RateSampler¶

Limits trace collection to N traces per second using a token-bucket approach with atomic operations. This prevents observability overhead from dominating high-throughput workflows:

// Allow at most 50 traces per second
sampler := tracing.NewRateSampler(50)
tracer := tracing.NewTracer("my-service", sampler, exporter)

How it works internally:

Each call to ShouldSample() atomically increments a per-second counter.
If the current second has changed since the last reset, the counter resets to zero (via CompareAndSwap).
If the counter exceeds maxPerSecond, the sample is rejected.
Rejected spans still propagate the trace ID through the context so downstream services can correlate, but the span itself is not recorded or exported.

Implementing a Custom Sampler¶

Any type satisfying the Sampler interface can be used:

type Sampler interface {
    ShouldSample(traceID tracing.TraceID, name string) bool
}

Example -- sample only spans whose name matches a pattern:

type PatternSampler struct {
    patterns []string
}

func (s *PatternSampler) ShouldSample(_ tracing.TraceID, name string) bool {
    for _, p := range s.patterns {
        if strings.Contains(name, p) {
            return true
        }
    }
    return false
}

sampler := &PatternSampler{patterns: []string{"payment", "order"}}
tracer := tracing.NewTracer("my-service", sampler, exporter)

4. Component-Level Spans¶

Beyond workflow-level spans created automatically by TracingInterceptor, the Tracer provides helpers for fine-grained component-level tracing. These create child spans under the current span in the context.

StartProducerSpan¶

Use when sending a message to an endpoint:

func (p *MyProducer) Process(ctx context.Context, ex *exchange.Exchange) error {
    ctx, span := tracer.StartProducerSpan(ctx, "kafka", "topic:orders")
    defer span.End()

    // span.Kind is SpanKindProducer
    // span.Attributes: helix.component="kafka", helix.endpoint="topic:orders"

    err := p.sendToKafka(ctx, ex)
    if err != nil {
        span.SetStatus(tracing.StatusError, err.Error())
        return err
    }
    return nil
}

StartConsumerSpan¶

Use when receiving a message from an endpoint:

func (c *MyConsumer) poll(ctx context.Context) {
    ctx, span := tracer.StartConsumerSpan(ctx, "amqp", "queue:events")
    defer span.End()

    // span.Kind is SpanKindConsumer
    // span.Attributes: helix.component="amqp", helix.endpoint="queue:events"

    msg, err := c.channel.Consume(ctx)
    if err != nil {
        span.SetStatus(tracing.StatusError, err.Error())
        return
    }
    span.SetAttribute("message.size", fmt.Sprintf("%d", len(msg.Body)))
}

StartProcessorSpan¶

Use within orchestration pattern processors to trace individual processing steps:

func (p *EnrichProcessor) Process(ctx context.Context, ex *exchange.Exchange) error {
    ctx, span := tracer.StartProcessorSpan(ctx, "content-enricher")
    defer span.End()

    // span.Kind is SpanKindInternal
    // span.Attributes: helix.processor="content-enricher"

    enriched, err := p.enrich(ctx, ex)
    if err != nil {
        span.SetStatus(tracing.StatusError, err.Error())
        return err
    }
    span.SetAttribute("enrichment.fields", "3")
    return nil
}

Span Attributes and Events¶

All spans support arbitrary attributes and timestamped events:

span.SetAttribute("db.statement", "SELECT * FROM orders WHERE id = ?")
span.SetAttribute("db.system", "postgresql")

span.AddEvent("cache.miss", map[string]string{
    "cache.key": "order:12345",
})

5. Custom Span Exporters¶

SpanExporter Interface¶

type SpanExporter interface {
    Export(ctx context.Context, spans []*tracing.Span) error
}

Built-In Exporters¶

LogExporter -- writes spans to structured logs via slog:

exporter := tracing.NewLogExporter(slog.Default())
tracer := tracing.NewTracer("my-service", nil, exporter)

CollectorExporter -- sends spans to an OpenTelemetry Collector via OTLP/HTTP (JSON):

exporter := tracing.NewCollectorExporter(
    "http://otel-collector:4318",
    "helix-service",
)
tracer := tracing.NewTracer("my-service", nil, exporter)

The collector exporter builds a minimal OTLP JSON payload with proper resource attributes (service.name), scope info (helix-tracing/1.0.0), and span data including kind, status, attributes, and events.

SpanExporterFunc Adapter¶

For one-off or lightweight exporters, use the function adapter:

exporter := tracing.SpanExporterFunc(func(ctx context.Context, spans []*tracing.Span) error {
    for _, s := range spans {
        if s.Status == tracing.StatusError {
            alertService.NotifySlowSpan(s.Name, s.Duration())
        }
    }
    return nil
})

tracer := tracing.NewTracer("my-service", nil, exporter)

Writing a Custom Exporter (Jaeger Example)¶

type JaegerExporter struct {
    endpoint string
    client   *http.Client
}

func NewJaegerExporter(endpoint string) *JaegerExporter {
    return &JaegerExporter{
        endpoint: endpoint,
        client:   &http.Client{Timeout: 10 * time.Second},
    }
}

func (e *JaegerExporter) Export(ctx context.Context, spans []*tracing.Span) error {
    // Convert spans to Jaeger Thrift/JSON format and POST
    payload := convertToJaegerFormat(spans)
    req, _ := http.NewRequestWithContext(ctx, "POST",
        e.endpoint+"/api/traces", bytes.NewReader(payload))
    req.Header.Set("Content-Type", "application/json")

    resp, err := e.client.Do(req)
    if err != nil {
        return err
    }
    defer resp.Body.Close()
    if resp.StatusCode >= 400 {
        return fmt.Errorf("jaeger export failed: %d", resp.StatusCode)
    }
    return nil
}

Multiple Exporters¶

The Tracer supports multiple exporters. On each flush, every registered exporter receives the full batch:

tracer := tracing.NewTracer("my-service", nil,
    tracing.NewLogExporter(nil),
    tracing.NewCollectorExporter("http://otel-collector:4318", "my-service"),
)

// Add more exporters at runtime
tracer.AddExporter(myCustomExporter)

6. Backlog Tracer¶

The BacklogTracer (in pkg/engine/tracer.go) is a debugging tool that captures exchange snapshots in a ring buffer. Unlike distributed tracing (which is always-on in production), the backlog tracer is designed for on-demand diagnostic sessions.

Enable/Disable via API¶

The tracer starts disabled and can be toggled at runtime:

bt := engine.NewBacklogTracer(500) // keep last 500 events
hc.AddInterceptor(bt)

// Enable tracing when needed
bt.Enable()

// ... process exchanges ...

// Disable when done
bt.Disable()

Typical REST API integration:

// POST /api/debug/tracer/enable
http.HandleFunc("/api/debug/tracer/enable", func(w http.ResponseWriter, r *http.Request) {
    bt.Enable()
    w.WriteHeader(http.StatusOK)
})

// POST /api/debug/tracer/disable
http.HandleFunc("/api/debug/tracer/disable", func(w http.ResponseWriter, r *http.Request) {
    bt.Disable()
    w.WriteHeader(http.StatusOK)
})

// GET /api/debug/tracer/traces
http.HandleFunc("/api/debug/tracer/traces", func(w http.ResponseWriter, r *http.Request) {
    traces := bt.GetTraces()
    json.NewEncoder(w).Encode(traces)
})

// DELETE /api/debug/tracer/traces
http.HandleFunc("/api/debug/tracer/clear", func(w http.ResponseWriter, r *http.Request) {
    bt.Clear()
    w.WriteHeader(http.StatusOK)
})

Ring Buffer Behavior¶

The backlog tracer uses a fixed-size ring buffer. Once the buffer fills, new events overwrite the oldest:

NewBacklogTracer(1000) -- allocates space for 1000 events (default if <= 0)
SetMaxBacklog(n) -- resizes the buffer (clears existing data)
GetTraces() -- returns all events in chronological order
Clear() -- empties the buffer

Workflow Filtering¶

Limit tracing to specific routes to reduce noise:

bt.AddRouteFilter("order-processing")
bt.AddRouteFilter("payment-gateway")

// Remove a filter
bt.RemoveRouteFilter("payment-gateway")

When the filter set is non-empty, only exchanges with matching workflow ID are recorded.

TraceEvent Structure¶

Each captured event contains:

type TraceEvent struct {
    Timestamp   time.Time
    ExchangeID  string
    RouteID     string
    ProcessorID string
    Body        string              // message body as string
    Headers     map[string]any
    Properties  map[string]any
    Elapsed     time.Duration       // only set in After events
    Error       string              // only set on failure
}

The interceptor captures two events per exchange: one in Before (with the incoming body/headers) and one in After (with the outgoing body, elapsed time, and any error).

Querying Traces¶

// All traces in chronological order
all := bt.GetTraces()

// Filter by workflow
orderTraces := bt.GetTracesByWorkflow("order-processing")

// Filter by exchange ID
exchangeTraces := bt.GetTracesByExchange("exchange-abc-123")

// Human-readable dump for logging
fmt.Println(bt.DumpTraces())

The DumpTraces() method produces formatted output:

BacklogTracer: 3 trace(s)
================================================================================
[0] 2026-03-10T14:23:01.123Z | exchange=abc-123 workflow=order-processing processor= elapsed=0s
     Body: {"orderId": 42}
     Headers: map[Content-Type:application/json]
--------------------------------------------------------------------------------
[1] 2026-03-10T14:23:01.145Z | exchange=abc-123 workflow=order-processing processor= elapsed=22ms
     Body: {"orderId": 42, "status": "confirmed"}
     Headers: map[Content-Type:application/json]
--------------------------------------------------------------------------------

8. Auto-Flush Lifecycle¶

The Tracer supports a managed lifecycle for periodic span flushing. This prevents spans from sitting in the buffer indefinitely and ensures a final flush on shutdown.

Start and Stop¶

tracer := tracing.NewTracer("my-service", nil, exporter)

// Begin periodic flushing (default: every 5 seconds)
ctx := context.Background()
tracer.Start(ctx)

// ... application runs ...

// Stop: cancels the periodic flush goroutine, performs a final flush
tracer.Stop(ctx)

Custom Flush Interval¶

tracer := tracing.NewTracerWithOptions(
    "my-service",
    tracing.NewRateSampler(100),
    []tracing.SpanExporter{exporter},
    tracing.WithFlushInterval(2 * time.Second),
)
tracer.Start(ctx)

Flush Behavior¶

Periodic flush: Every flushInterval (default 5s), all buffered spans are exported to every registered exporter, then the buffer is cleared.
Shutdown flush: When Stop() is called, the periodic goroutine is cancelled and a final flush is performed with context.Background() to ensure no spans are lost.
Manual flush: Call tracer.Flush(ctx) at any time to force an immediate export.
Error handling: Flush errors from periodic ticks are logged via slog.Warn. The first error from any exporter is returned from manual Flush() calls.

Pending Span Count¶

Check how many spans are waiting for export:

pending := tracer.PendingSpans()
fmt.Printf("%d spans buffered\n", pending)

9. Bounded Span Buffer¶

To prevent unbounded memory growth under high load, the tracer enforces a hard limit on its in-memory span buffer.

How It Works¶

Maximum buffer size: 10,000 spans (constant maxSpanBuffer).
Overflow behavior: When the buffer reaches capacity, the oldest 10% of spans (1,000) are dropped to make room.
Warning log: A slog.Warn message is emitted when overflow occurs, including the number of dropped spans.

This is implemented inside StartSpan():

if len(spans) >= 10000:
    drop oldest 1000 spans
    log warning: "tracing: span buffer full, dropped oldest spans"
    append new span

Recommendations¶

In production, set a flush interval shorter than the time it takes to fill 10,000 spans. For example, at 1,000 spans/sec, use a flush interval of 5 seconds or less.
Use RateSampler to limit the rate of span creation if throughput is very high.
Monitor the tracing: span buffer full log message as a signal that you need either faster export or more aggressive sampling.
Note: Only sampled spans consume buffer space. Unsampled spans (rejected by the Sampler) are still created for context propagation but are never added to the buffer.

10. Advanced Audit Patterns¶

The audit system (in pkg/engine/audit.go) records exchange lifecycle events for compliance, debugging, and forensics.

Audit Store Implementations¶

InMemoryAuditStore -- fixed-size ring buffer, no external dependencies:

store := engine.NewInMemoryAuditStore(50000) // keep last 50K entries

SlogAuditStore -- writes every entry as a structured log record AND keeps a ring buffer for API queries:

logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{
    Level: slog.LevelInfo,
}))
store := engine.NewSlogAuditStore(logger, 50000)

Each entry logged by SlogAuditStore produces a structured JSON line:

{"time":"2026-03-10T14:23:01Z","level":"INFO","msg":"audit","exchange_id":"abc-123","route_id":"order-processing","from_uri":"kafka://orders","operation":"completed","duration":"22ms","error":""}

AuditInterceptor Configuration¶

store := engine.NewSlogAuditStore(nil, 10000)

// Capture up to 512 bytes of message body in audit entries
audit := engine.NewAuditInterceptor(store, 512)

hc := engine.NewContext("my-app")
hc.AddInterceptor(audit)

Set bodyPreview to 0 to disable body capture entirely (recommended for sensitive data).

AuditEntry Fields¶

type AuditEntry struct {
    Timestamp   time.Time      `json:"timestamp"`
    ExchangeID  string         `json:"exchange_id"`
    RouteID     string         `json:"route_id"`
    FromURI     string         `json:"from_uri"`
    Operation   string         `json:"operation"`     // "completed" or "failed"
    Duration    time.Duration  `json:"duration_ns"`
    Error       string         `json:"error,omitempty"`
    Headers     map[string]any `json:"headers,omitempty"`
    BodyPreview string         `json:"body_preview,omitempty"`
}

Exposing the Audit API¶

auditStore := audit.Store()

http.HandleFunc("/api/audit", func(w http.ResponseWriter, r *http.Request) {
    entries := auditStore.Entries()
    json.NewEncoder(w).Encode(entries)
})

Custom AuditStore¶

Implement the AuditStore interface to write to a database, message queue, or external service:

type AuditStore interface {
    Record(entry AuditEntry)
    Entries() []AuditEntry
}

Example -- PostgreSQL audit store:

type PgAuditStore struct {
    pool *pgxpool.Pool
}

func (s *PgAuditStore) Record(entry engine.AuditEntry) {
    _, _ = s.pool.Exec(context.Background(),
        `INSERT INTO audit_log (timestamp, exchange_id, route_id, from_uri, operation, duration_ns, error, body_preview)
         VALUES ($1, $2, $3, $4, $5, $6, $7, $8)`,
        entry.Timestamp, entry.ExchangeID, entry.RouteID, entry.FromURI,
        entry.Operation, entry.Duration.Nanoseconds(), entry.Error, entry.BodyPreview,
    )
}

func (s *PgAuditStore) Entries() []engine.AuditEntry {
    // Query from database
    rows, _ := s.pool.Query(context.Background(),
        `SELECT timestamp, exchange_id, route_id, from_uri, operation, duration_ns, error, body_preview
         FROM audit_log ORDER BY timestamp DESC LIMIT 10000`)
    // ... scan rows ...
}

Full Interceptor Stack¶

Combine all interceptors for complete observability:

collector := metrics.NewCollector()

hc := engine.NewContext("my-app")
hc.AddInterceptor(metrics.NewMetricsInterceptor(collector))    // Prometheus metrics
hc.AddInterceptor(slaInterceptor)                              // SLA monitoring
hc.AddInterceptor(tracing.NewTracingInterceptor(tracer))       // Distributed tracing
hc.AddInterceptor(engine.NewAuditInterceptor(auditStore, 256)) // Audit logging
hc.AddInterceptor(backlogTracer)                               // Debug tracer (off by default)

11. Grafana Dashboards¶

Helix ships two pre-built Grafana dashboards in deploy/grafana/.

Helix Overview Dashboard¶

File: deploy/grafana/helix-overview.json UID: helix-overview

A high-level operational dashboard showing:

Panel	Query	Description
Exchanges / sec	`sum(rate(helix_exchanges_total[5m])) by (route)`	Exchange throughput per workflow
Error Rate	Error exchanges as percentage of total	Per-workflow error ratio
In-Flight Exchanges	`helix_exchanges_inflight`	Current backpressure level
Exchange Duration	`helix_exchange_duration_seconds` histogram	Latency distribution
Workflow Uptime	`helix_workflow_uptime_seconds`	How long each workflow has been running

Helix Routes Dashboard¶

File: deploy/grafana/helix-workflows.json UID: helix-workflows

A per-workflow drill-down dashboard with a $route template variable (auto-populated from label_values(helix_exchanges_total, route)).

Panel	Description
Workflow Throughput	Exchange rate for the selected workflow, split by status
Latency Percentiles	p50/p95/p99 from SLA gauges
Error Breakdown	Error vs success counts
In-Flight	Current exchange backlog

Importing Dashboards¶

Open Grafana and go to Dashboards > Import.
Upload the JSON files or paste the contents.
Select your Prometheus data source when prompted for ${DS_PROMETHEUS}.
Both dashboards auto-refresh every 10 seconds.

Adding SLA Panels¶

To visualize SLA violations alongside latency, add custom panels:

SLA Violations Rate:

sum(rate(helix_sla_violations_total[5m])) by (route, type)

P99 Latency vs SLA Threshold:

helix_workflow_p99_latency_seconds{workflow="$route"}

Add a threshold line at your SLA target (e.g., 0.5 for 500ms).

Error Rate vs SLA Threshold:

sum(rate(helix_route_errors_total[5m])) by (route)
/
(sum(rate(helix_route_errors_total[5m])) by (route) + sum(rate(helix_route_successes_total[5m])) by (route))

12. Prometheus Alerting Rules¶

Helix provides a set of production-ready alerting rules in deploy/prometheus/alerts.yaml.

Included Alerts¶

HelixHighErrorRate (critical)¶

Fires when the error rate exceeds 5% of total exchanges for 5 minutes:

- alert: HelixHighErrorRate
  expr: |
    (
      sum(rate(helix_exchanges_total{status="error"}[5m])) by (route)
      /
      sum(rate(helix_exchanges_total[5m])) by (route)
    ) > 0.05
  for: 5m
  labels:
    severity: critical
  annotations:
    summary: "High error rate on route {{ $labels.route }}"
    description: >-
      Workflow {{ $labels.workflow }} has an error rate of
      {{ $value | humanizePercentage }} over the last 5 minutes,
      exceeding the 5% threshold.

HelixHighLatency (warning)¶

Fires when p99 latency exceeds 5 seconds for 5 minutes:

- alert: HelixHighLatency
  expr: |
    histogram_quantile(0.99,
      sum(rate(helix_exchange_duration_seconds_bucket[5m])) by (route, le)
    ) > 5
  for: 5m
  labels:
    severity: warning
  annotations:
    summary: "High p99 latency on route {{ $labels.route }}"
    description: >-
      Workflow {{ $labels.workflow }} has a p99 latency of
      {{ $value | humanizeDuration }}, exceeding the 5-second threshold
      for the last 5 minutes.

HelixInflightSpike (warning)¶

Fires when in-flight exchanges exceed 100 for 2 minutes:

- alert: HelixInflightSpike
  expr: |
    helix_exchanges_inflight > 100
  for: 2m
  labels:
    severity: warning
  annotations:
    summary: "In-flight exchange spike on route {{ $labels.route }}"
    description: >-
      Workflow {{ $labels.workflow }} has {{ $value }} in-flight exchanges,
      exceeding the threshold of 100 for the last 2 minutes.

HelixRouteDown (critical)¶

Fires when a workflow's uptime drops to 0 for 1 minute:

- alert: HelixRouteDown
  expr: |
    helix_route_uptime_seconds == 0
  for: 1m
  labels:
    severity: critical
  annotations:
    summary: "Workflow {{ $labels.workflow }} is down"
    description: >-
      Workflow {{ $labels.workflow }} has reported zero uptime for the last minute.

Loading the Alert Rules¶

Add to your prometheus.yml:

rule_files:
  - "/etc/prometheus/rules/helix-alerts.yaml"

Copy the rules file:

cp deploy/prometheus/alerts.yaml /etc/prometheus/rules/helix-alerts.yaml

Custom SLA-Based Alerts¶

Add alerts that use the SLA-specific metrics:

groups:
  - name: helix.sla.rules
    rules:
      # Alert on SLA violation spike
      - alert: HelixSLAViolationSpike
        expr: |
          sum(rate(helix_sla_violations_total[5m])) by (route) > 0.1
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "SLA violations detected on route {{ $labels.route }}"

      # Alert when p99 latency gauge exceeds threshold
      - alert: HelixP99LatencyHigh
        expr: |
          helix_route_p99_latency_seconds > 1.0
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "P99 latency above 1s on route {{ $labels.route }}"
          description: >-
            Workflow {{ $labels.workflow }} p99 latency is {{ $value }}s.

      # Alert on sustained high error rate from SLA counters
      - alert: HelixRouteErrorRateHigh
        expr: |
          (
            sum(rate(helix_route_errors_total[5m])) by (route)
            /
            (sum(rate(helix_route_errors_total[5m])) by (route)
             + sum(rate(helix_route_successes_total[5m])) by (route))
          ) > 0.03
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Error rate above 3% on route {{ $labels.route }}"

YAML Configuration Reference¶

Helix observability configuration in the application config file:

helix:
  metrics:
    enabled: true
    endpoint: "/metrics"

  tracing:
    enabled: true
    service-name: "my-helix-service"
    sampler: rate
    sample-rate: 100           # traces per second
    flush-interval: 5s
    exporter: otlp
    otlp-endpoint: "http://otel-collector:4318"

  audit:
    enabled: true
    body-preview: 256          # bytes of body to capture (0 to disable)
    max-entries: 50000         # ring buffer size
    store: slog                # "memory" or "slog"

  logging:
    format: json               # "text" or "json"
    level: info

  shutdown:
    timeout: 30s
    drain-timeout: 15s

Quick Reference: All Helix Metrics¶

Metric	Type	Labels	Source
`helix_exchanges_total`	counter	`workflow`, `status`	MetricsInterceptor
`helix_exchange_duration_seconds`	histogram	`workflow`	MetricsInterceptor
`helix_exchanges_inflight`	gauge	`workflow`	MetricsInterceptor
`helix_component_exchanges_total`	counter	`component`, `operation`	Component helpers
`helix_route_uptime_seconds`	gauge	`workflow`	Engine
`helix_sla_violations_total`	counter	`workflow`, `type`	SLAInterceptor
`helix_route_errors_total`	counter	`workflow`	SLAInterceptor
`helix_route_successes_total`	counter	`workflow`	SLAInterceptor
`helix_route_p50_latency_seconds`	gauge	`workflow`	SLAInterceptor
`helix_route_p95_latency_seconds`	gauge	`workflow`	SLAInterceptor
`helix_route_p99_latency_seconds`	gauge	`workflow`	SLAInterceptor