diff --git a/examples/streaming-daemon-go/README.md b/examples/streaming-daemon-go/README.md index 4c422b7..a2f91a0 100644 --- a/examples/streaming-daemon-go/README.md +++ b/examples/streaming-daemon-go/README.md @@ -14,7 +14,7 @@ This daemon receives client connections from Apache via `mod_socket_handoff` usi ## Features -- **Backend plugin architecture** - Extensible backends (langgraph, mock, openai, typing) with Caddy-style `init()` registration +- **Backend plugin architecture** - Extensible backends (langgraph, mock, noop-monitor, openai, typing) with Caddy-style `init()` registration - **YAML configuration** - Optional config file with flag overrides - **Goroutine-per-connection** - Lightweight concurrency for high throughput - **SCM_RIGHTS fd receiving** - Portable buffer sizing with `syscall.CmsgSpace` @@ -103,6 +103,7 @@ The daemon supports multiple streaming backends via a plugin architecture. Backe |---------|-------------|----------| | `langgraph` | LangGraph Platform API | Stateful agents with conversation threads | | `mock` | Fixed demo messages with configurable delay | Testing, benchmarking | +| `noop-monitor` | Holds the connection open, sends nothing | Connection volume/concurrency/duration measurement | | `openai` | OpenAI-compatible streaming API | GPT-4, Groq, Ollama, any OpenAI-compatible API | | `typing` | Character-by-character typewriter effect | Demos, fortune integration | @@ -224,6 +225,23 @@ Streams characters one at a time with realistic typing delays. Uses `/usr/games/ ./streaming-daemon -backend typing ``` +### Noop Monitor Backend + +Holds a handed-off connection open and sends only periodic `: ping` SSE keepalives — no LLM, no message data, no per-user logging. Its purpose is measurement: connection volume, concurrency, and duration are read straight from this daemon's Prometheus metrics. Intended to run as a dedicated monitor instance on its own socket and metrics port. + +```bash +./streaming-daemon -backend noop-monitor -socket /run/streaming-daemon/convos-monitor.sock +``` + +Config (`backend.noop_monitor.ping_interval_ms`, default `25000`) sets the keepalive interval; the keepalive doubles as client-disconnect detection since a write-only stream only learns the client left when a write fails. `server.max_stream_duration_ms` optionally caps connection lifetime (`0` = hold indefinitely). + +Besides the daemon-level metrics, it exposes two low-cardinality series: + +| Metric | Type | Labels | Meaning | +|--------|------|--------|---------| +| `noop_monitor_active` | gauge | `source` | Currently held connections by surface (`detail`, `message_list`, `other`) | +| `noop_monitor_closed_total` | counter | `source`, `reason` | Closed connections by surface and reason (`client_disconnect`, `max_lifetime`, `shutdown`) | + ### Adding New Backends To add a new backend (e.g., Anthropic): diff --git a/examples/streaming-daemon-go/backends/backend.go b/examples/streaming-daemon-go/backends/backend.go index 93897d6..b9ef0e7 100644 --- a/examples/streaming-daemon-go/backends/backend.go +++ b/examples/streaming-daemon-go/backends/backend.go @@ -70,18 +70,25 @@ type HandoffData struct { TestPattern string `json:"test_pattern,omitempty"` // For testing: passed to backend as X-Test-Pattern header // LangGraph-specific fields - ThreadID string `json:"thread_id,omitempty"` // For stateful runs (uses /threads/{id}/runs/stream) - AssistantID string `json:"assistant_id,omitempty"` // Override default assistant ID - StreamMode []string `json:"stream_mode,omitempty"` // Stream modes (e.g., ["messages", "updates", "custom"]) - LangGraphInput map[string]any `json:"langgraph_input,omitempty"` // Custom input fields (seller_id, shop_id, etc.) - Profile string `json:"profile,omitempty"` // Named LangGraph profile from config - LangGraphURL string `json:"langgraph_url,omitempty"` // Per-request API base URL override + ThreadID string `json:"thread_id,omitempty"` // For stateful runs (uses /threads/{id}/runs/stream) + AssistantID string `json:"assistant_id,omitempty"` // Override default assistant ID + StreamMode []string `json:"stream_mode,omitempty"` // Stream modes (e.g., ["messages", "updates", "custom"]) + LangGraphInput map[string]any `json:"langgraph_input,omitempty"` // Custom input fields (seller_id, shop_id, etc.) + Profile string `json:"profile,omitempty"` // Named LangGraph profile from config + LangGraphURL string `json:"langgraph_url,omitempty"` // Per-request API base URL override LangGraphKey string `json:"langgraph_api_key,omitempty"` // Per-request API key override - LG string `json:"lg,omitempty"` // Compact: "profile|url|key" (pipe-delimited, empty segment = no override for that position) + LG string `json:"lg,omitempty"` // Compact: "profile|url|key" (pipe-delimited, empty segment = no override for that position) // Per-request backend selection (overrides the daemon's default provider) Backend string `json:"backend,omitempty"` + // Convos connection-monitor fields (noop-monitor backend). + // Source labels the originating surface ("detail"|"message_list") for + // per-surface metrics. ConnectionID is a correlation id for optional + // per-connection debug logging only; neither has a functional consumer. + Source string `json:"source,omitempty"` + ConnectionID string `json:"connection_id,omitempty"` + // Image handoff fields (for multimodal requests) // Legacy single-image fields (deprecated, use Images/ImagePaths instead) ImagePath string `json:"image_path,omitempty"` // Deprecated: use ImagePaths diff --git a/examples/streaming-daemon-go/backends/noop_monitor.go b/examples/streaming-daemon-go/backends/noop_monitor.go new file mode 100644 index 0000000..90dc65f --- /dev/null +++ b/examples/streaming-daemon-go/backends/noop_monitor.go @@ -0,0 +1,110 @@ +package backends + +import ( + "context" + "errors" + "log/slog" + "net" + "time" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" + + "examples/config" +) + +// Backend-owned metrics for the connection monitor. Kept low-cardinality: +// source is clamped to {detail, message_list, other} and reason to a fixed +// set, so these are safe to expose per-label. +var ( + monitorActive = promauto.NewGaugeVec(prometheus.GaugeOpts{ + Name: "noop_monitor_active", + Help: "Currently held monitor connections, by originating surface", + }, []string{"source"}) + + monitorClosed = promauto.NewCounterVec(prometheus.CounterOpts{ + Name: "noop_monitor_closed_total", + Help: "Monitor connections closed, by surface and reason", + }, []string{"source", "reason"}) +) + +// NoopMonitor holds a handed-off client connection open, sending only periodic +// SSE keepalive comments, so connection volume/concurrency/duration can be read +// straight from this daemon's Prometheus metrics. It never streams message data. +type NoopMonitor struct { + pingInterval time.Duration +} + +func init() { + Register(&NoopMonitor{}) +} + +func (n *NoopMonitor) Name() string { + return "noop-monitor" +} + +func (n *NoopMonitor) Description() string { + return "Holds the connection open, sends nothing (convos connection monitor)" +} + +func (n *NoopMonitor) Init(cfg *config.BackendConfig) error { + n.pingInterval = 25 * time.Second + if cfg != nil && cfg.NoopMonitor.PingIntervalMs > 0 { + n.pingInterval = time.Duration(cfg.NoopMonitor.PingIntervalMs) * time.Millisecond + } + slog.Info("noop-monitor backend initialized", "ping_interval", n.pingInterval) + return nil +} + +// Stream holds the connection open, writing a `: ping` keepalive every +// pingInterval. A write-only stream over SOCK_SEQPACKET only learns the client +// left when a write fails, so the keepalive doubles as disconnect detection. +// A client disconnect is normal, not an error, so it returns (0, nil). +func (n *NoopMonitor) Stream(ctx context.Context, conn net.Conn, handoff HandoffData) (int64, error) { + source := normalizeSource(handoff.Source) + + RecordBackendRequest("noop-monitor") + monitorActive.WithLabelValues(source).Inc() + defer monitorActive.WithLabelValues(source).Dec() + + start := time.Now() + reason := "shutdown" + ticker := time.NewTicker(n.pingInterval) + defer ticker.Stop() + +loop: + for { + select { + case <-ctx.Done(): + if errors.Is(context.Cause(ctx), context.DeadlineExceeded) { + reason = "max_lifetime" + } + break loop + case <-ticker.C: + _ = conn.SetWriteDeadline(time.Now().Add(WriteTimeout)) + if _, err := conn.Write(pingMsg); err != nil { + reason = "client_disconnect" + break loop + } + } + } + + RecordBackendDuration("noop-monitor", time.Since(start).Seconds()) + monitorClosed.WithLabelValues(source, reason).Inc() + return 0, nil +} + +// pingMsg is an SSE comment line; clients ignore it, but the write attempt is +// how a write-only stream detects that the client has gone away. +var pingMsg = []byte(": ping\n\n") + +// normalizeSource clamps the client-influenced source label to a known set so a +// malformed or unexpected value can't blow up Prometheus label cardinality. +func normalizeSource(source string) string { + switch source { + case "detail", "message_list": + return source + default: + return "other" + } +} diff --git a/examples/streaming-daemon-go/backends/noop_monitor_test.go b/examples/streaming-daemon-go/backends/noop_monitor_test.go new file mode 100644 index 0000000..d886c9d --- /dev/null +++ b/examples/streaming-daemon-go/backends/noop_monitor_test.go @@ -0,0 +1,124 @@ +package backends + +import ( + "bytes" + "context" + "io" + "net" + "testing" + "time" + + "github.com/prometheus/client_golang/prometheus/testutil" +) + +// runMonitor runs Stream against a net.Pipe and returns once it exits. +// The drain callback receives the client end so a test can either read the +// pings or close the connection to simulate a disconnect. +func runMonitor(t *testing.T, ctx context.Context, n *NoopMonitor, h HandoffData, drain func(client net.Conn)) { + t.Helper() + client, server := net.Pipe() + defer client.Close() + defer server.Close() + + drain(client) + + done := make(chan struct{}) + go func() { + n.Stream(ctx, server, h) + close(done) + }() + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("Stream did not return within timeout") + } +} + +func TestNoopMonitorClientDisconnect(t *testing.T) { + before := testutil.ToFloat64(monitorClosed.WithLabelValues("detail", "client_disconnect")) + activeBefore := testutil.ToFloat64(monitorActive.WithLabelValues("detail")) + + n := &NoopMonitor{pingInterval: 5 * time.Millisecond} + runMonitor(t, context.Background(), n, HandoffData{Source: "detail"}, func(client net.Conn) { + client.Close() // client is gone; the first ping write fails + }) + + if got := testutil.ToFloat64(monitorClosed.WithLabelValues("detail", "client_disconnect")); got != before+1 { + t.Errorf("client_disconnect counter = %v, want %v", got, before+1) + } + if got := testutil.ToFloat64(monitorActive.WithLabelValues("detail")); got != activeBefore { + t.Errorf("active gauge = %v, want %v (should return to baseline)", got, activeBefore) + } +} + +func TestNoopMonitorShutdown(t *testing.T) { + before := testutil.ToFloat64(monitorClosed.WithLabelValues("detail", "shutdown")) + + ctx, cancel := context.WithCancel(context.Background()) + // Long ping interval so no write happens before the cancel is observed. + n := &NoopMonitor{pingInterval: 10 * time.Second} + go func() { + time.Sleep(10 * time.Millisecond) + cancel() + }() + runMonitor(t, ctx, n, HandoffData{Source: "detail"}, func(client net.Conn) { + go io.Copy(io.Discard, client) + }) + cancel() + + if got := testutil.ToFloat64(monitorClosed.WithLabelValues("detail", "shutdown")); got != before+1 { + t.Errorf("shutdown counter = %v, want %v", got, before+1) + } +} + +func TestNoopMonitorMaxLifetime(t *testing.T) { + before := testutil.ToFloat64(monitorClosed.WithLabelValues("detail", "max_lifetime")) + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond) + defer cancel() + n := &NoopMonitor{pingInterval: 10 * time.Second} + runMonitor(t, ctx, n, HandoffData{Source: "detail"}, func(client net.Conn) { + go io.Copy(io.Discard, client) + }) + + if got := testutil.ToFloat64(monitorClosed.WithLabelValues("detail", "max_lifetime")); got != before+1 { + t.Errorf("max_lifetime counter = %v, want %v", got, before+1) + } +} + +func TestNoopMonitorWritesPing(t *testing.T) { + n := &NoopMonitor{pingInterval: 5 * time.Millisecond} + client, server := net.Pipe() + defer client.Close() + defer server.Close() + + ctx, cancel := context.WithCancel(context.Background()) + go func() { + n.Stream(ctx, server, HandoffData{Source: "message_list"}) + }() + + got := make([]byte, len(pingMsg)) + _ = client.SetReadDeadline(time.Now().Add(2 * time.Second)) + if _, err := io.ReadFull(client, got); err != nil { + t.Fatalf("reading ping: %v", err) + } + if !bytes.Equal(got, pingMsg) { + t.Errorf("first write = %q, want %q", got, pingMsg) + } + cancel() +} + +func TestNormalizeSource(t *testing.T) { + tests := map[string]string{ + "detail": "detail", + "message_list": "message_list", + "": "other", + "bogus": "other", + } + for in, want := range tests { + if got := normalizeSource(in); got != want { + t.Errorf("normalizeSource(%q) = %q, want %q", in, got, want) + } + } +} diff --git a/examples/streaming-daemon-go/config/config.go b/examples/streaming-daemon-go/config/config.go index f57ade5..8a7d35d 100644 --- a/examples/streaming-daemon-go/config/config.go +++ b/examples/streaming-daemon-go/config/config.go @@ -27,7 +27,7 @@ type ServerConfig struct { MaxConnections int `yaml:"max_connections"` MaxStreamDurationMs int `yaml:"max_stream_duration_ms"` PprofAddr string `yaml:"pprof_addr"` - MemLimit string `yaml:"mem_limit"` // Soft memory limit, e.g. "768MiB", "1GiB" + MemLimit string `yaml:"mem_limit"` // Soft memory limit, e.g. "768MiB", "1GiB" GCPercent int `yaml:"gc_percent"` // GOGC value; 0 = not set (use -gc-percent flag for GOGC=0) DataDir string `yaml:"data_dir"` // Allowed directory for attachment file reads (default: /run/handoff-data) } @@ -38,10 +38,11 @@ type BackendConfig struct { DefaultModel string `yaml:"default_model"` // Backend-specific configuration sections - OpenAI OpenAIConfig `yaml:"openai"` - LangGraph LangGraphConfig `yaml:"langgraph"` - Mock MockConfig `yaml:"mock"` - Typing TypingConfig `yaml:"typing"` + OpenAI OpenAIConfig `yaml:"openai"` + LangGraph LangGraphConfig `yaml:"langgraph"` + Mock MockConfig `yaml:"mock"` + Typing TypingConfig `yaml:"typing"` + NoopMonitor NoopMonitorConfig `yaml:"noop_monitor"` } // OpenAIConfig contains OpenAI-compatible API settings. @@ -92,6 +93,11 @@ type TypingConfig struct { DefaultModel string `yaml:"default_model"` } +// NoopMonitorConfig contains noop-monitor backend settings. +type NoopMonitorConfig struct { + PingIntervalMs int `yaml:"ping_interval_ms"` // keepalive interval; 0 = default (25000) +} + // MetricsConfig contains Prometheus metrics settings. type MetricsConfig struct { Enabled bool `yaml:"enabled"` @@ -207,6 +213,11 @@ func (c *Config) Validate() error { return fmt.Errorf("backend.mock.message_delay_ms must be non-negative") } + // Validate noop-monitor backend + if c.Backend.NoopMonitor.PingIntervalMs < 0 { + return fmt.Errorf("backend.noop_monitor.ping_interval_ms must be non-negative") + } + return nil } @@ -297,6 +308,9 @@ func Default() *Config { Mock: MockConfig{ MessageDelayMs: 50, }, + NoopMonitor: NoopMonitorConfig{ + PingIntervalMs: 25000, + }, }, Metrics: MetricsConfig{ Enabled: true, diff --git a/examples/streaming-daemon-go/config/example.yaml b/examples/streaming-daemon-go/config/example.yaml index 2cef09e..ae39a87 100644 --- a/examples/streaming-daemon-go/config/example.yaml +++ b/examples/streaming-daemon-go/config/example.yaml @@ -64,6 +64,11 @@ backend: # Typing backend configuration typing: {} + # Noop-monitor backend configuration (connection monitor). + # Holds connections open sending only keepalives; measurement is via metrics. + noop_monitor: + ping_interval_ms: 25000 # Keepalive interval (also the disconnect-detection cadence) + metrics: enabled: true listen_addr: 127.0.0.1:9090