Documentation
¶
Overview ¶
Package observability provides metrics and instrumentation for the orchestrator.
Description ¶
This package implements Prometheus metrics for monitoring streaming chat operations. Metrics include:
- Request counters (by endpoint, status, error type)
- Token usage (input/output tokens by model)
- Latency histograms (time to first token, total duration)
- Active stream gauges
Integration ¶
Metrics are exposed via /metrics endpoint. Use with Prometheus + Grafana for dashboards and alerting.
Thread Safety ¶
All metric operations are thread-safe via Prometheus's internal locking.
Index ¶
- type Endpoint
- type ErrorCode
- type StreamingMetrics
- func (m *StreamingMetrics) RecordClientDisconnect(endpoint Endpoint)
- func (m *StreamingMetrics) RecordError(endpoint Endpoint, code ErrorCode)
- func (m *StreamingMetrics) RecordKeepAlive(endpoint Endpoint)
- func (m *StreamingMetrics) RecordRequest(endpoint Endpoint, success bool)
- func (m *StreamingMetrics) RecordStreamDuration(endpoint Endpoint, seconds float64, success bool)
- func (m *StreamingMetrics) RecordTimeToFirstToken(endpoint Endpoint, seconds float64)
- func (m *StreamingMetrics) RecordTokens(inputTokens, outputTokens int, model string)
- func (m *StreamingMetrics) StreamEnded(endpoint Endpoint)
- func (m *StreamingMetrics) StreamStarted(endpoint Endpoint)
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type ErrorCode ¶
type ErrorCode string
ErrorCode represents a categorized error type for metrics.
const ( // ErrorCodePolicyViolation indicates blocked due to policy scan. ErrorCodePolicyViolation ErrorCode = "policy_violation" // ErrorCodeValidation indicates request validation failure. ErrorCodeValidation ErrorCode = "validation" // ErrorCodeLLMError indicates LLM API failure. ErrorCodeLLMError ErrorCode = "llm_error" // ErrorCodeTimeout indicates operation timeout. ErrorCodeTimeout ErrorCode = "timeout" // ErrorCodeRAGError indicates RAG retrieval failure. ErrorCodeRAGError ErrorCode = "rag_error" // ErrorCodeInternal indicates internal server error. ErrorCodeInternal ErrorCode = "internal" // ErrorCodeClientDisconnect indicates client disconnected. ErrorCodeClientDisconnect ErrorCode = "client_disconnect" )
type StreamingMetrics ¶
type StreamingMetrics struct {
// RequestsTotal counts streaming requests by endpoint and status.
// Labels: endpoint (direct_stream, rag_stream), status (success, error)
RequestsTotal *prometheus.CounterVec
// TokensTotal counts tokens processed by direction and model.
// Labels: direction (input, output), model (claude-3-5-sonnet, etc.)
TokensTotal *prometheus.CounterVec
// TimeToFirstTokenSeconds measures latency to first token.
// Labels: endpoint (direct_stream, rag_stream)
TimeToFirstTokenSeconds *prometheus.HistogramVec
// StreamDurationSeconds measures total stream duration.
// Labels: endpoint (direct_stream, rag_stream), status (success, error)
StreamDurationSeconds *prometheus.HistogramVec
// ActiveStreams tracks currently active streaming connections.
// Labels: endpoint (direct_stream, rag_stream)
ActiveStreams *prometheus.GaugeVec
// ErrorsTotal counts errors by type and endpoint.
// Labels: endpoint, error_code (policy_violation, llm_error, timeout, etc.)
ErrorsTotal *prometheus.CounterVec
// KeepAlivesTotal counts keepalive pings sent.
// Labels: endpoint
KeepAlivesTotal *prometheus.CounterVec
// ClientDisconnectsTotal counts client disconnections during streaming.
// Labels: endpoint
ClientDisconnectsTotal *prometheus.CounterVec
}
StreamingMetrics holds all Prometheus metrics for streaming chat operations.
Description ¶
Provides counters, histograms, and gauges for monitoring streaming performance and resource usage. Initialize once at startup via NewStreamingMetrics().
Fields ¶
- RequestsTotal: Counter of streaming requests by endpoint and status
- TokensTotal: Counter of tokens processed (input/output by model)
- TimeToFirstTokenSeconds: Histogram of time to first token
- StreamDurationSeconds: Histogram of total stream duration
- ActiveStreams: Gauge of currently active streams
- ErrorsTotal: Counter of errors by type and endpoint
Thread Safety ¶
All operations are thread-safe.
var DefaultMetrics *StreamingMetrics
DefaultMetrics is the singleton instance of StreamingMetrics. Initialized by InitMetrics().
func InitMetrics ¶
func InitMetrics() *StreamingMetrics
InitMetrics initializes the default metrics instance.
Description ¶
Creates and registers all Prometheus metrics. Should be called once at application startup, after Prometheus registry is available.
Outputs ¶
- *StreamingMetrics: The initialized metrics instance.
Examples ¶
func main() {
observability.InitMetrics()
// ... start server ...
}
Limitations ¶
- Panics if called twice (duplicate registration).
Assumptions ¶
- Prometheus default registry is available.
func (*StreamingMetrics) RecordClientDisconnect ¶
func (m *StreamingMetrics) RecordClientDisconnect(endpoint Endpoint)
RecordClientDisconnect increments the client disconnect counter.
Inputs ¶
- endpoint: The endpoint where disconnect occurred.
func (*StreamingMetrics) RecordError ¶
func (m *StreamingMetrics) RecordError(endpoint Endpoint, code ErrorCode)
RecordError records a streaming error.
Inputs ¶
- endpoint: The endpoint where the error occurred.
- code: The error type code.
func (*StreamingMetrics) RecordKeepAlive ¶
func (m *StreamingMetrics) RecordKeepAlive(endpoint Endpoint)
RecordKeepAlive increments the keepalive counter.
Inputs ¶
- endpoint: The endpoint that sent the keepalive.
func (*StreamingMetrics) RecordRequest ¶
func (m *StreamingMetrics) RecordRequest(endpoint Endpoint, success bool)
RecordRequest records a completed streaming request.
Inputs ¶
- endpoint: The endpoint that handled the request.
- success: Whether the request completed successfully.
func (*StreamingMetrics) RecordStreamDuration ¶
func (m *StreamingMetrics) RecordStreamDuration(endpoint Endpoint, seconds float64, success bool)
RecordStreamDuration records the total stream duration.
Inputs ¶
- endpoint: The endpoint that handled the stream.
- seconds: Total duration in seconds.
- success: Whether the stream completed successfully.
func (*StreamingMetrics) RecordTimeToFirstToken ¶
func (m *StreamingMetrics) RecordTimeToFirstToken(endpoint Endpoint, seconds float64)
RecordTimeToFirstToken records the time to first token latency.
Inputs ¶
- endpoint: The endpoint handling the stream.
- seconds: Time to first token in seconds.
func (*StreamingMetrics) RecordTokens ¶
func (m *StreamingMetrics) RecordTokens(inputTokens, outputTokens int, model string)
RecordTokens records token usage.
Inputs ¶
- inputTokens: Number of input tokens.
- outputTokens: Number of output tokens.
- model: The model used.
func (*StreamingMetrics) StreamEnded ¶
func (m *StreamingMetrics) StreamEnded(endpoint Endpoint)
StreamEnded decrements the active streams gauge.
Inputs ¶
- endpoint: The endpoint that handled the stream.
func (*StreamingMetrics) StreamStarted ¶
func (m *StreamingMetrics) StreamStarted(endpoint Endpoint)
StreamStarted increments the active streams gauge.
Inputs ¶
- endpoint: The endpoint handling the stream.
Source Files