From 2cf563d88e8a2582f1e2f8b9bc268d05599b9e70 Mon Sep 17 00:00:00 2001 From: cabbage Date: Fri, 11 Sep 2026 09:09:52 +0000 Subject: [PATCH] feat: add bounded operational metrics --- cmd/rvbox-server/main.go | 2 + cmd/rvbox/main.go | 5 +- docs/operations-runbook.md | 18 +++ internal/client/agent/executor.go | 25 +++- internal/client/agent/runner.go | 45 ++++++ internal/client/agent/runner_test.go | 12 +- internal/observability/health.go | 143 +++++++++++++++++-- internal/observability/health_test.go | 29 ++++ internal/server/control/service.go | 14 +- internal/server/session/agent_server.go | 40 ++++++ internal/server/session/agent_server_test.go | 14 ++ test/coverage.toml | 9 ++ 12 files changed, 342 insertions(+), 14 deletions(-) diff --git a/cmd/rvbox-server/main.go b/cmd/rvbox-server/main.go index 6bd02b4..c76b4cb 100644 --- a/cmd/rvbox-server/main.go +++ b/cmd/rvbox-server/main.go @@ -134,6 +134,7 @@ func run(configPath string) error { controlService, err := control.NewService(control.Options{ Store: persistence, DefaultQueueTTL: configured.Queue.DefaultTTL, DefaultQueueTTLSet: true, TakeoverTTL: configured.Protocol.TakeoverTTL, WakeClient: registry.Wake, + Metrics: health, Limits: agentproto.Limits{MaxEnvelopeBytes: configured.Protocol.MaxAgentEnvelopeBytes, MaxExecutionSpecBytes: configured.Protocol.MaxExecutionSpecBytes, MaxRawChunkBytes: configured.Protocol.MaxRawChunkBytes, MaxScriptBytes: configured.Protocol.MaxScriptBytes, MaxDetailBytes: configured.Storage.ProtocolDetailMaxBytes}, }) if err != nil { @@ -156,6 +157,7 @@ func run(configPath string) error { } agent := &session.AgentServer{ Store: persistence, Registry: registry, Path: configured.Server.AgentPath, + Metrics: health, Limits: agentproto.Limits{ MaxEnvelopeBytes: configured.Protocol.MaxAgentEnvelopeBytes, MaxExecutionSpecBytes: configured.Protocol.MaxExecutionSpecBytes, diff --git a/cmd/rvbox/main.go b/cmd/rvbox/main.go index e9fe1b5..b6602b8 100644 --- a/cmd/rvbox/main.go +++ b/cmd/rvbox/main.go @@ -190,7 +190,7 @@ func runClientDaemon(ctx context.Context, configPath string, diagnostics io.Writ if err != nil { return fmt.Errorf("configure command supervisor: %w", err) } - executor, err := agent.NewExecutor(agent.ExecutorOptions{Store: state, Supervisor: supervised, WorkDir: configured.Client.DaemonCWD, Notify: func(issue domain.UUID) { + executor, err := agent.NewExecutor(agent.ExecutorOptions{Store: state, Supervisor: supervised, WorkDir: configured.Client.DaemonCWD, Metrics: health, Notify: func(issue domain.UUID) { select { case eventReady <- issue: default: @@ -216,7 +216,6 @@ func runClientDaemon(ctx context.Context, configPath string, diagnostics io.Writ } else if len(recovered) > 0 && diagnostics != nil { _, _ = fmt.Fprintf(diagnostics, "rvbox recovered %d uncertain launch(es)\n", len(recovered)) } - health.SetReady(true) if runErr := agent.Run(ctx, agent.RunnerOptions{ Store: state, Dial: func(dialContext context.Context) (agent.Transport, error) { @@ -227,6 +226,8 @@ func runClientDaemon(ctx context.Context, configPath string, diagnostics io.Writ Jitter: agent.CryptoJitter, Now: func() time.Time { return time.Now().UTC() }, OnDispatch: executor.Dispatch, OnScriptReady: executor.ScriptReady, OnStdin: executor.Stdin, OnCloseStdin: executor.CloseStdin, OnSignal: executor.Signal, OnTerminate: executor.Terminate, EventReady: eventReady, + Metrics: health, + OnSessionActive: health.SetReady, OnSessionError: func(sessionErr error) { if diagnostics != nil { _, _ = fmt.Fprintf(diagnostics, "rvbox client session retry: %v\n", sessionErr) diff --git a/docs/operations-runbook.md b/docs/operations-runbook.md index 2c1c0ad..b79d30b 100644 --- a/docs/operations-runbook.md +++ b/docs/operations-runbook.md @@ -25,6 +25,24 @@ the process is up; `/readyz` becomes successful only after durable recovery. The public endpoint accepts only `wss://HOST/v1/agent`. Do not publish port 6900 or add a proxy route for JSON-RPC. +## Health, metrics, and logs + +Keep the configured observability listener private to the host or monitoring +network. `/metrics` exposes only bounded, aggregate Prometheus samples: health +state; registration/takeover and protocol failures; session/reconnect and +heartbeat timing; dispatch/event/transition counts and latency; and client +output accounting. It never includes a command body, stdin, output, client ID, +or UUID label. Duration histograms use fixed buckets, so monitoring traffic +cannot create unbounded series. + +For a Windows client, readiness is false while its durable spool is recovering +or it has no reconciled server session; it becomes true only while the active +WSS session can exchange command data. Server liveness starts before +asynchronous recovery, while server readiness remains false until that recovery +completes. The configured rotating JSON/text service logs are diagnostics, not +a command-output store; use `rvc` history and the audited storage for command +evidence. + ## Backup and restore Stop dispatch before copying data: stop the server gracefully, confirm it is diff --git a/internal/client/agent/executor.go b/internal/client/agent/executor.go index 2a82a8d..b9664d7 100644 --- a/internal/client/agent/executor.go +++ b/internal/client/agent/executor.go @@ -12,6 +12,7 @@ import ( "github.com/rvbox/rvbox/internal/client/spool" "github.com/rvbox/rvbox/internal/client/supervisor" "github.com/rvbox/rvbox/internal/domain" + "github.com/rvbox/rvbox/internal/observability" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/timestamppb" ) @@ -25,6 +26,7 @@ type Executor struct { WorkDir string Now func() time.Time Notify func(domain.UUID) + Metrics *observability.Health mu sync.Mutex active map[domain.UUID]supervisor.Process @@ -38,6 +40,7 @@ type ExecutorOptions struct { WorkDir string Now func() time.Time Notify func(domain.UUID) + Metrics *observability.Health } func NewExecutor(options ExecutorOptions) (*Executor, error) { @@ -47,7 +50,7 @@ func NewExecutor(options ExecutorOptions) (*Executor, error) { if options.Now == nil { options.Now = func() time.Time { return time.Now().UTC() } } - return &Executor{Store: options.Store, Supervisor: options.Supervisor, WorkDir: options.WorkDir, Now: options.Now, Notify: options.Notify, active: make(map[domain.UUID]supervisor.Process), cancel: make(map[domain.UUID]context.CancelFunc), forced: make(map[domain.UUID]bool)}, nil + return &Executor{Store: options.Store, Supervisor: options.Supervisor, WorkDir: options.WorkDir, Now: options.Now, Notify: options.Notify, Metrics: options.Metrics, active: make(map[domain.UUID]supervisor.Process), cancel: make(map[domain.UUID]context.CancelFunc), forced: make(map[domain.UUID]bool)}, nil } // Dispatch is safe to invoke after CommandAccepted has been sent. Script @@ -158,6 +161,7 @@ func (executor *Executor) launch(ctx context.Context, issue domain.UUID, revisio executor.remove(issue) return err } + executor.incMetric("command_running") executor.notify(issue) go executor.watch(runContext, issue, revision, process, cancel) return nil @@ -171,6 +175,7 @@ func (executor *Executor) watch(ctx context.Context, issue domain.UUID, revision break } if err != nil { + executor.incMetric("output_loss") _ = executor.appendIncomplete(context.Background(), issue, revision, err.Error()) break } @@ -178,10 +183,13 @@ func (executor *Executor) watch(ctx context.Context, issue domain.UUID, revision continue } if _, err := executor.Store.AppendOutput(context.Background(), issue, spool.OutputInput{Stream: chunk.Stream, Raw: chunk.Data, ObservedAt: executor.Now()}); err != nil { + executor.incMetric("output_loss") _ = executor.appendIncomplete(context.Background(), issue, revision, err.Error()) _, _ = executor.Supervisor.Signal(context.Background(), process, supervisor.SignalKill) break } + executor.incMetric("output_chunk") + executor.incMetricBy("output_bytes", uint64(len(chunk.Data))) executor.notify(issue) } status, waitErr := process.Wait(context.Background()) @@ -210,9 +218,24 @@ func (executor *Executor) appendLifecycle(ctx context.Context, issue domain.UUID func (executor *Executor) appendLifecycleWithIdentity(ctx context.Context, issue domain.UUID, revision uint64, phase rvboxv1.CommandLifecycle, detail string, identity *rvboxv1.WindowsExecutionIdentity) error { _, err := executor.Store.AppendLifecycleWithIdentity(ctx, issue, uint32(phase), revision, detail, executor.Now(), identity) + if err == nil { + executor.incMetric("command_transition") + } return err } +func (executor *Executor) incMetric(name string) { + if executor != nil && executor.Metrics != nil { + executor.Metrics.Inc(name) + } +} + +func (executor *Executor) incMetricBy(name string, delta uint64) { + if executor != nil && executor.Metrics != nil { + executor.Metrics.IncBy(name, delta) + } +} + func (executor *Executor) reject(ctx context.Context, issue domain.UUID, revision uint64, cause error) error { return executor.rejectWithIdentity(ctx, issue, revision, cause, nil) } diff --git a/internal/client/agent/runner.go b/internal/client/agent/runner.go index 1b5ac69..d39cc00 100644 --- a/internal/client/agent/runner.go +++ b/internal/client/agent/runner.go @@ -12,6 +12,7 @@ import ( "github.com/rvbox/rvbox/internal/agentproto" "github.com/rvbox/rvbox/internal/client/spool" "github.com/rvbox/rvbox/internal/domain" + "github.com/rvbox/rvbox/internal/observability" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/timestamppb" ) @@ -39,6 +40,12 @@ type RunnerOptions struct { // transport session before normal reconnect backoff. It must not block; the // durable spool and retry policy remain owned by Run. OnSessionError func(error) + // OnSessionActive reports whether a reconciled transport is currently able + // to exchange command data. It lets daemon readiness represent connection + // state without making reconnect behavior depend on the observer. + OnSessionActive func(bool) + // Metrics is optional. It only receives static, bounded metric names. + Metrics *observability.Health // EventReady wakes the active session after a supervisor worker appends a // durable event. The network loop remains the sole writer; a reconnect can // safely ignore a stale notification because replay reads the spool again. @@ -98,6 +105,10 @@ func Run(ctx context.Context, options RunnerOptions) error { if err := waitUntil(ctx, delay); err != nil { return nil } + if failures > 0 { + options.incMetric("reconnect_attempt") + } + options.incMetric("connection_attempt") sessionStarted := options.Now() sessionErr := runOnce(ctx, options) if ctx.Err() != nil { @@ -106,6 +117,9 @@ func Run(ctx context.Context, options RunnerOptions) error { if sessionErr != nil && options.OnSessionError != nil { options.OnSessionError(sessionErr) } + if sessionErr != nil { + options.incMetric("connection_failure") + } if options.Now().Sub(sessionStarted) >= options.Backoff.StableReset { failures = 0 } @@ -173,6 +187,7 @@ func runOnce(ctx context.Context, options RunnerOptions) (resultErr error) { } session, err := Handshake(ctx, transport, options.Hello, limits) if err != nil { + options.incMetric("protocol_error") return fmt.Errorf("agent handshake: %w", err) } snapshot, err := options.Store.ReconcileSnapshot(ctx) @@ -201,6 +216,17 @@ func runOnce(ctx context.Context, options RunnerOptions) (resultErr error) { if err := sendCapacity(ctx, transport, session, options.Hello, snapshot, limits); err != nil { return fmt.Errorf("advertise client capacity: %w", err) } + activeStarted := options.Now() + options.incMetric("session_established") + if options.OnSessionActive != nil { + options.OnSessionActive(true) + } + defer func() { + if options.OnSessionActive != nil { + options.OnSessionActive(false) + } + options.observeMetric("session_duration", options.Now().Sub(activeStarted)) + }() return serveActive(ctx, transport, options, session, sentEvents) } @@ -343,9 +369,11 @@ func serveActive(ctx context.Context, transport Transport, options RunnerOptions } envelope, err := agentproto.DecodeEnvelope(encoded, limits, rvboxv1.Platform_PLATFORM_WINDOWS) if err != nil { + options.incMetric("protocol_error") return fmt.Errorf("decode active agent frame: %w", err) } if envelope.GetSessionId() != session.ID || envelope.GetSessionGeneration() != session.Generation { + options.incMetric("stale_message") return ErrProtocolHandshake } switch { @@ -452,6 +480,11 @@ func flushEvents(ctx context.Context, transport Transport, store *spool.Store, s func handleDispatch(ctx context.Context, transport Transport, options RunnerOptions, session Session, dispatch *rvboxv1.CommandDispatch, limits agentproto.Limits) error { acceptance, err := PersistDispatch(ctx, options.Store, session, dispatch, options.Now(), limits) accepted := err == nil + if accepted { + options.incMetric("command_dispatch_accepted") + } else { + options.incMetric("command_dispatch_rejected") + } ack := &rvboxv1.CommandAccepted{IssueUuid: dispatch.GetIssueUuid(), CommandRevision: dispatch.GetCommandRevision(), Accepted: accepted} if err != nil { ack.Rejection = rejectionForError(err, dispatch.GetIssueUuid()) @@ -474,6 +507,18 @@ func handleDispatch(ctx context.Context, transport Transport, options RunnerOpti return nil } +func (options RunnerOptions) incMetric(name string) { + if options.Metrics != nil { + options.Metrics.Inc(name) + } +} + +func (options RunnerOptions) observeMetric(name string, duration time.Duration) { + if options.Metrics != nil { + options.Metrics.ObserveDuration(name, duration) + } +} + func rejectionForError(err error, issue string) *rvboxv1.ControlError { code := rvboxv1.ControlError_TRANSIENT if errors.Is(err, spool.ErrCommandConflict) || errors.Is(err, spool.ErrAlreadyExecuted) { diff --git a/internal/client/agent/runner_test.go b/internal/client/agent/runner_test.go index 1cdd243..01cae5c 100644 --- a/internal/client/agent/runner_test.go +++ b/internal/client/agent/runner_test.go @@ -12,6 +12,7 @@ import ( "github.com/rvbox/rvbox/internal/agentproto" "github.com/rvbox/rvbox/internal/client/spool" "github.com/rvbox/rvbox/internal/domain" + "github.com/rvbox/rvbox/internal/observability" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/timestamppb" ) @@ -46,7 +47,9 @@ func TestRunOnceReconcilesReplaysAndAdvertisesCapacity_HP_RUNTIME_01(t *testing. reconcileResult, _ := proto.Marshal(&rvboxv1.AgentEnvelope{SessionId: "session", SessionGeneration: 1, Payload: &rvboxv1.AgentEnvelope_ReconcileResult{ReconcileResult: &rvboxv1.ReconcileResult{}}}) transport := &runnerTransport{reads: [][]byte{welcome, reconcileRequest, reconcileResult}, terminal: errors.New("transport closed")} hello := &rvboxv1.ClientHello{ClientId: "runner-client", SupportedProtocol: &rvboxv1.ProtocolRange{Major: 1, MinMinor: 0, MaxMinor: 0}, DaemonVersion: "test", Platform: rvboxv1.Platform_PLATFORM_WINDOWS, Architecture: "amd64", DaemonCwd: `C:\`, SupportedShells: []rvboxv1.ShellType{rvboxv1.ShellType_SHELL_POWERSHELL}, ClientInstanceId: store.ClientInstanceID().String(), MaxRunningCommands: 1, MaxQueuedCommands: 1, SentAt: timestamppb.Now()} - err = RunOnce(ctx, RunnerOptions{Store: store, Dial: func(context.Context) (Transport, error) { return transport, nil }, Hello: hello, Limits: agentproto.DefaultLimits(), Backoff: BackoffOptions{Initial: time.Millisecond, Maximum: time.Millisecond, StableReset: time.Second}, Jitter: func(value time.Duration) time.Duration { return 0 }, Now: func() time.Time { return time.Now().UTC() }}) + metrics := observability.New() + var active []bool + err = RunOnce(ctx, RunnerOptions{Store: store, Dial: func(context.Context) (Transport, error) { return transport, nil }, Hello: hello, Limits: agentproto.DefaultLimits(), Backoff: BackoffOptions{Initial: time.Millisecond, Maximum: time.Millisecond, StableReset: time.Second}, Jitter: func(value time.Duration) time.Duration { return 0 }, Now: func() time.Time { return time.Now().UTC() }, Metrics: metrics, OnSessionActive: func(value bool) { active = append(active, value) }}) if !errors.Is(err, transport.terminal) { t.Fatalf("RunOnce error = %v, want transport close", err) } @@ -64,6 +67,13 @@ func TestRunOnceReconcilesReplaysAndAdvertisesCapacity_HP_RUNTIME_01(t *testing. if err != nil || eventEnvelope.GetCommandEvent() == nil || eventEnvelope.GetCommandEvent().GetEventSeq() != 1 || eventEnvelope.GetCommandEvent().GetIssueUuid() != issue.String() { t.Fatalf("replayed event = %#v, %v", eventEnvelope, err) } + _, _, counters := metrics.Snapshot() + if counters["session_established"] != 1 { + t.Fatalf("session metrics = %#v", counters) + } + if len(active) != 2 || !active[0] || active[1] { + t.Fatalf("active transitions = %#v", active) + } } func TestRunnerOptionsRejectMissingJitter_HP_RUNTIME_02(t *testing.T) { diff --git a/internal/observability/health.go b/internal/observability/health.go index feb4e86..2cb2085 100644 --- a/internal/observability/health.go +++ b/internal/observability/health.go @@ -6,6 +6,7 @@ package observability import ( "context" "fmt" + "math" "net" "net/http" "sort" @@ -27,9 +28,28 @@ type Health struct { dirty atomic.Bool mu sync.Mutex count map[string]uint64 + gauge map[string]float64 + hist map[string]*durationHistogram } -func New() *Health { return &Health{count: make(map[string]uint64)} } +type durationHistogram struct { + count uint64 + sum float64 + buckets [len(durationBuckets)]uint64 +} + +// The fixed buckets are intentionally shared by every duration metric. This +// keeps the Prometheus surface bounded and makes comparable operational +// latencies available without allowing callers to create arbitrary labels. +var durationBuckets = [...]float64{0.001, 0.005, 0.01, 0.05, 0.1, 0.25, 0.5, 1, 5, 15, 60} + +func New() *Health { + return &Health{ + count: make(map[string]uint64), + gauge: make(map[string]float64), + hist: make(map[string]*durationHistogram), + } +} func (health *Health) SetReady(value bool) { if health != nil { @@ -44,11 +64,52 @@ func (health *Health) SetDirty(value bool) { } func (health *Health) Inc(name string) { - if health == nil || !validMetricName(name) { + health.IncBy(name, 1) +} + +// IncBy records a non-negative integral counter delta. Metric names are +// deliberately restricted to a small identifier grammar; callers cannot turn +// a client ID, request UUID, or other untrusted data into a metric name. +func (health *Health) IncBy(name string, delta uint64) { + if health == nil || delta == 0 || !validMetricName(name) { return } health.mu.Lock() - health.count[name]++ + health.count[name] += delta + health.mu.Unlock() +} + +// SetGauge records a bounded-cardinality instantaneous value. NaN and +// infinities are discarded because they are not safe Prometheus samples. +func (health *Health) SetGauge(name string, value float64) { + if health == nil || !validMetricName(name) || math.IsNaN(value) || math.IsInf(value, 0) { + return + } + health.mu.Lock() + health.gauge[name] = value + health.mu.Unlock() +} + +// ObserveDuration records a duration in a fixed-bucket histogram. Negative +// durations are invalid (normally a caller clock error) and are ignored. +func (health *Health) ObserveDuration(name string, duration time.Duration) { + if health == nil || !validMetricName(name) || duration < 0 { + return + } + seconds := duration.Seconds() + health.mu.Lock() + histogram := health.hist[name] + if histogram == nil { + histogram = &durationHistogram{} + health.hist[name] = histogram + } + histogram.count++ + histogram.sum += seconds + for index, upperBound := range durationBuckets { + if seconds <= upperBound { + histogram.buckets[index]++ + } + } health.mu.Unlock() } @@ -65,6 +126,46 @@ func (health *Health) Snapshot() (ready, dirty bool, counters map[string]uint64) return health.ready.Load(), health.dirty.Load(), copyCounters } +type MetricsSnapshot struct { + Ready bool + Dirty bool + Counters map[string]uint64 + Gauges map[string]float64 + Histograms map[string]DurationHistogramSnapshot +} + +type DurationHistogramSnapshot struct { + Count uint64 + Sum float64 + Buckets [len(durationBuckets)]uint64 +} + +// MetricsSnapshot returns a copy suitable for rendering or testing. It never +// includes application identifiers or payloads. +func (health *Health) MetricsSnapshot() MetricsSnapshot { + if health == nil { + return MetricsSnapshot{Dirty: true} + } + health.mu.Lock() + defer health.mu.Unlock() + result := MetricsSnapshot{ + Ready: health.ready.Load(), Dirty: health.dirty.Load(), + Counters: make(map[string]uint64, len(health.count)), + Gauges: make(map[string]float64, len(health.gauge)), + Histograms: make(map[string]DurationHistogramSnapshot, len(health.hist)), + } + for name, value := range health.count { + result.Counters[name] = value + } + for name, value := range health.gauge { + result.Gauges[name] = value + } + for name, value := range health.hist { + result.Histograms[name] = DurationHistogramSnapshot{Count: value.count, Sum: value.sum, Buckets: value.buckets} + } + return result +} + func (health *Health) Handler(paths Paths) http.Handler { if paths.Liveness == "" { paths.Liveness = "/livez" @@ -92,17 +193,38 @@ func (health *Health) Handler(paths Paths) http.Handler { _, _ = fmt.Fprintf(response, "ready=%s dirty=%s\n", strconv.FormatBool(ready), strconv.FormatBool(dirty)) }) mux.HandleFunc(paths.Metrics, func(response http.ResponseWriter, _ *http.Request) { - ready, dirty, counters := health.Snapshot() + snapshot := health.MetricsSnapshot() response.Header().Set("Content-Type", "text/plain; version=0.0.4") response.WriteHeader(http.StatusOK) - _, _ = fmt.Fprintf(response, "rvbox_health_ready %d\nrvbox_health_dirty %d\n", boolMetric(ready), boolMetric(dirty)) - keys := make([]string, 0, len(counters)) - for key := range counters { + _, _ = fmt.Fprintf(response, "rvbox_health_ready %d\nrvbox_health_dirty %d\n", boolMetric(snapshot.Ready), boolMetric(snapshot.Dirty)) + keys := make([]string, 0, len(snapshot.Counters)) + for key := range snapshot.Counters { keys = append(keys, key) } sort.Strings(keys) for _, key := range keys { - _, _ = fmt.Fprintf(response, "rvbox_%s_total %d\n", key, counters[key]) + _, _ = fmt.Fprintf(response, "rvbox_%s_total %d\n", key, snapshot.Counters[key]) + } + keys = keys[:0] + for key := range snapshot.Gauges { + keys = append(keys, key) + } + sort.Strings(keys) + for _, key := range keys { + _, _ = fmt.Fprintf(response, "rvbox_%s %g\n", key, snapshot.Gauges[key]) + } + keys = keys[:0] + for key := range snapshot.Histograms { + keys = append(keys, key) + } + sort.Strings(keys) + for _, key := range keys { + histogram := snapshot.Histograms[key] + for index, upperBound := range durationBuckets { + _, _ = fmt.Fprintf(response, "rvbox_%s_seconds_bucket{le=\"%g\"} %d\n", key, upperBound, histogram.Buckets[index]) + } + _, _ = fmt.Fprintf(response, "rvbox_%s_seconds_bucket{le=\"+Inf\"} %d\n", key, histogram.Count) + _, _ = fmt.Fprintf(response, "rvbox_%s_seconds_sum %g\nrvbox_%s_seconds_count %d\n", key, histogram.Sum, key, histogram.Count) } }) return mux @@ -142,7 +264,10 @@ func validMetricName(name string) bool { if name == "" { return false } - for _, character := range name { + for index, character := range name { + if index == 0 && !(character == '_' || character >= 'a' && character <= 'z' || character >= 'A' && character <= 'Z') { + return false + } if !(character == '_' || character >= 'a' && character <= 'z' || character >= 'A' && character <= 'Z' || character >= '0' && character <= '9') { return false } diff --git a/internal/observability/health_test.go b/internal/observability/health_test.go index 0862475..a98cdae 100644 --- a/internal/observability/health_test.go +++ b/internal/observability/health_test.go @@ -5,6 +5,7 @@ import ( "net/http/httptest" "strings" "testing" + "time" ) func TestHealthHandlerReadinessAndMetrics_HP_OPS_01(t *testing.T) { @@ -35,8 +36,36 @@ func TestHealthMetricNamesAreBounded_BH_OPS_01(t *testing.T) { t.Parallel() health := New() health.Inc("bad name") + health.Inc("1bad_name") _, _, counters := health.Snapshot() if len(counters) != 0 { t.Fatalf("invalid metric name recorded: %#v", counters) } } + +func TestHealthRendersBoundedGaugesAndDurationHistograms_HP_OPS_02(t *testing.T) { + t.Parallel() + health := New() + health.SetGauge("client_spool_bytes", 4096) + health.SetGauge("bad gauge", 1) + health.ObserveDuration("dispatch_latency", 12*time.Millisecond) + health.ObserveDuration("bad latency", time.Second) + + metrics := httptest.NewRecorder() + health.Handler(Paths{}).ServeHTTP(metrics, httptest.NewRequest(http.MethodGet, "/metrics", nil)) + body := metrics.Body.String() + for _, want := range []string{ + "rvbox_client_spool_bytes 4096", + "rvbox_dispatch_latency_seconds_bucket{le=\"0.05\"} 1", + "rvbox_dispatch_latency_seconds_bucket{le=\"+Inf\"} 1", + "rvbox_dispatch_latency_seconds_sum 0.012", + "rvbox_dispatch_latency_seconds_count 1", + } { + if !strings.Contains(body, want) { + t.Fatalf("metrics missing %q:\n%s", want, body) + } + } + if strings.Contains(body, "bad gauge") || strings.Contains(body, "bad latency") { + t.Fatalf("unbounded metric name rendered: %s", body) + } +} diff --git a/internal/server/control/service.go b/internal/server/control/service.go index ff7eb16..fb22cf3 100644 --- a/internal/server/control/service.go +++ b/internal/server/control/service.go @@ -18,6 +18,7 @@ import ( rvboxv1 "github.com/rvbox/rvbox/gen/go/rvbox/v1" "github.com/rvbox/rvbox/internal/agentproto" "github.com/rvbox/rvbox/internal/domain" + "github.com/rvbox/rvbox/internal/observability" "github.com/rvbox/rvbox/internal/server/store" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" @@ -48,6 +49,8 @@ type Options struct { Now func() time.Time CursorKey []byte WakeClient func(string) bool + // Metrics is optional and receives only static metric names. + Metrics *observability.Health } type Service struct { @@ -59,6 +62,7 @@ type Service struct { now func() time.Time cursors *domain.CursorCodec wakeClient func(string) bool + metrics *observability.Health } func NewService(options Options) (*Service, error) { @@ -94,7 +98,7 @@ func NewService(options Options) (*Service, error) { if err != nil { return nil, err } - return &Service{store: options.Store, defaultQueueTTL: options.DefaultQueueTTL, takeoverTTL: options.TakeoverTTL, limits: options.Limits, now: options.Now, cursors: codec, wakeClient: options.WakeClient}, nil + return &Service{store: options.Store, defaultQueueTTL: options.DefaultQueueTTL, takeoverTTL: options.TakeoverTTL, limits: options.Limits, now: options.Now, cursors: codec, wakeClient: options.WakeClient, metrics: options.Metrics}, nil } func (service *Service) ListClients(ctx context.Context, request *rvboxv1.ListClientsRequest) (*rvboxv1.ListClientsResponse, error) { @@ -676,11 +680,19 @@ func (service *Service) AuthorizeClientTakeover(ctx context.Context, request *rv expires := service.now().UTC().Add(service.takeoverTTL) value, _, err := service.store.AuthorizeClientTakeover(ctx, store.TakeoverAuthorization{ClientID: request.GetClientId(), ClientInstanceID: [16]byte(instance), RequestID: [16]byte(requestID), ExpiresAt: expires}) if err != nil { + service.incMetric("client_takeover_rejected") return nil, mapStoreError(err) } + service.incMetric("client_takeover_authorized") return &rvboxv1.AuthorizeClientTakeoverResponse{ExpiresAt: timestamppb.New(value)}, nil } +func (service *Service) incMetric(name string) { + if service != nil && service.metrics != nil { + service.metrics.Inc(name) + } +} + func incidentRecord(view store.IncidentView) *rvboxv1.StorageIncident { result := &rvboxv1.StorageIncident{IncidentId: domain.UUID(view.IncidentUUID).String(), DetectedAt: timestamppb.New(view.DetectedAt), State: rvboxv1.StorageIncidentState(view.State), Scope: string(view.Scope), ClientId: view.ClientID, Summary: view.Summary, DataLoss: view.DataLoss, AutomaticallyRepairable: view.AutomaticallyRepairable} if view.ResolvedAt != nil { diff --git a/internal/server/session/agent_server.go b/internal/server/session/agent_server.go index fa28ee7..496612e 100644 --- a/internal/server/session/agent_server.go +++ b/internal/server/session/agent_server.go @@ -16,6 +16,7 @@ import ( rvboxv1 "github.com/rvbox/rvbox/gen/go/rvbox/v1" "github.com/rvbox/rvbox/internal/agentproto" "github.com/rvbox/rvbox/internal/domain" + "github.com/rvbox/rvbox/internal/observability" "github.com/rvbox/rvbox/internal/server/store" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/timestamppb" @@ -45,6 +46,9 @@ type AgentServer struct { HeartbeatIdle time.Duration LivenessTimeout time.Duration Now func() time.Time + // Metrics is optional so transport tests and embeddings do not need an + // observability endpoint. Every emitted name is a static bounded value. + Metrics *observability.Health } func (server *AgentServer) ServeHTTP(response http.ResponseWriter, request *http.Request) { @@ -53,6 +57,7 @@ func (server *AgentServer) ServeHTTP(response http.ResponseWriter, request *http return } if request.Header.Get("Origin") != "" { + server.incMetric("protocol_error") http.Error(response, ErrUnexpectedOrigin.Error(), http.StatusForbidden) return } @@ -83,6 +88,7 @@ func (server *AgentServer) ServeHTTP(response http.ResponseWriter, request *http if err != nil { return } + server.incMetric("agent_connection") defer connection.CloseNow() server.serveConnection(request.Context(), connection, heartbeat, started) } @@ -98,6 +104,7 @@ func (server *AgentServer) serveConnection(parent context.Context, connection *w } helloEnvelope, err := agentproto.DecodeEnvelope(payload, server.limits(), rvboxv1.Platform_PLATFORM_UNSPECIFIED) if err != nil || helloEnvelope.GetClientHello() == nil { + server.incMetric("protocol_error") server.close(connection, websocket.StatusPolicyViolation, "invalid ClientHello") return } @@ -108,11 +115,13 @@ func (server *AgentServer) serveConnection(parent context.Context, connection *w reconcileBoundary := server.now() instanceID, err := domain.ParseUUIDv7(hello.GetClientInstanceId()) if err != nil { + server.incMetric("protocol_error") server.close(connection, websocket.StatusPolicyViolation, "invalid client instance ID") return } selected, err := domain.SelectProtocol(server.protocolRange(), hello.GetSupportedProtocol()) if err != nil { + server.incMetric("protocol_error") server.close(connection, websocket.StatusPolicyViolation, "unsupported protocol") return } @@ -126,15 +135,22 @@ func (server *AgentServer) serveConnection(parent context.Context, connection *w server.close(connection, websocket.StatusInternalError, "could not persist client capabilities") return } + registrationStarted := time.Now() registration, err := server.Store.RegisterClientSession(parent, store.ClientRegistration{ ClientID: hello.GetClientId(), Platform: uint32(hello.GetPlatform()), Architecture: hello.GetArchitecture(), DaemonVersion: hello.GetDaemonVersion(), DaemonCWD: hello.GetDaemonCwd(), SupportedShells: shells, ClientInstanceID: [16]byte(instanceID), SessionID: sessionID, ConnectedAt: server.now(), }) + server.observeMetric("sqlite_write_latency", time.Since(registrationStarted)) if err != nil { + server.incMetric("client_registration_failure") + if errors.Is(err, store.ErrTakeoverRequired) { + server.incMetric("client_takeover_required") + } server.close(connection, websocket.StatusPolicyViolation, sessionCloseReason(err)) return } + server.incMetric("client_registration") registry := server.Registry handle, err := registry.Install(hello.GetClientId(), sessionID, registration.Generation) if err != nil { @@ -227,16 +243,19 @@ func (server *AgentServer) serveConnection(parent context.Context, connection *w // to determine liveness. heartbeat.Observe(time.Since(started)) if messageType != websocket.MessageBinary { + server.incMetric("protocol_error") server.close(connection, websocket.StatusUnsupportedData, ErrUnexpectedMessage.Error()) return } envelope, err := agentproto.DecodeEnvelope(payload, server.limits(), hello.GetPlatform()) if err != nil || envelope.GetClientHello() != nil || envelope.GetSessionId() != encodedSessionID || envelope.GetSessionGeneration() != registration.Generation { + server.incMetric("stale_message") server.close(connection, websocket.StatusPolicyViolation, ErrStaleSession.Error()) return } clientID, err := server.Store.ValidateLiveSession(sessionContext, sessionID, registration.Generation) if err != nil || clientID != hello.GetClientId() { + server.incMetric("stale_message") server.close(connection, websocket.StatusPolicyViolation, ErrStaleSession.Error()) return } @@ -317,14 +336,20 @@ func (server *AgentServer) serveConnection(parent context.Context, connection *w } appendEvent, eventErr := eventAppendFromWire(event, hello.GetClientId(), registration.Generation, server.now()) if eventErr != nil { + server.incMetric("protocol_error") server.close(connection, websocket.StatusPolicyViolation, "invalid command event") return } appended, eventErr := server.Store.AppendCommandEvent(sessionContext, appendEvent) if eventErr != nil { + server.incMetric("command_event_rejected") server.close(connection, websocket.StatusPolicyViolation, "command event was not accepted") return } + server.incMetric("command_event") + if event.GetLifecycle() != nil { + server.incMetric("command_transition") + } ack, eventErr := proto.Marshal(&rvboxv1.AgentEnvelope{SessionId: encodedSessionID, SessionGeneration: registration.Generation, Payload: &rvboxv1.AgentEnvelope_EventAck{EventAck: &rvboxv1.EventAck{IssueUuid: event.GetIssueUuid(), ThroughEventSeq: appended.ThroughEventSeq}}}) if eventErr != nil || queue.EnqueueControl(Frame{Kind: FrameControl, Payload: ack}) != nil { server.close(connection, websocket.StatusInternalError, "could not acknowledge command event") @@ -452,6 +477,8 @@ func (server *AgentServer) enqueueNextDispatch(ctx context.Context, queue *Write beforeEnqueue(candidate) } if !queue.EnqueueData(Frame{Kind: FrameData, Payload: encoded, OnWritten: func() { + server.incMetric("command_dispatch") + server.observeMetric("dispatch_latency", server.now().Sub(candidate.IssueTime)) if onWritten != nil { onWritten(candidate.IssueUUID) } @@ -756,11 +783,24 @@ func (server *AgentServer) writeLoop(ctx context.Context, connection *websocket. return } case HeartbeatClose: + server.incMetric("heartbeat_timeout") return } } } +func (server *AgentServer) incMetric(name string) { + if server != nil && server.Metrics != nil { + server.Metrics.Inc(name) + } +} + +func (server *AgentServer) observeMetric(name string, duration time.Duration) { + if server != nil && server.Metrics != nil { + server.Metrics.ObserveDuration(name, duration) + } +} + func (server *AgentServer) close(connection *websocket.Conn, status websocket.StatusCode, reason string) { _ = connection.Close(status, reason) } diff --git a/internal/server/session/agent_server_test.go b/internal/server/session/agent_server_test.go index 42f176d..a68c38e 100644 --- a/internal/server/session/agent_server_test.go +++ b/internal/server/session/agent_server_test.go @@ -18,6 +18,7 @@ import ( rvboxv1 "github.com/rvbox/rvbox/gen/go/rvbox/v1" "github.com/rvbox/rvbox/internal/agentproto" "github.com/rvbox/rvbox/internal/domain" + "github.com/rvbox/rvbox/internal/observability" "github.com/rvbox/rvbox/internal/server/store" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/timestamppb" @@ -98,6 +99,19 @@ func TestAgentServerRegistrationAndReplacement_HP_SES_05(t *testing.T) { } } +func TestAgentServerEmitsBoundedRegistrationMetrics_HP_OPS_05(t *testing.T) { + server, agent, cleanup := newTestAgentServerWithStore(t) + defer cleanup() + metrics := observability.New() + agent.Metrics = metrics + connection, _ := dialAndHello(t, server, "metric-client", "019c46f1-1d02-7000-8000-000000000073") + defer connection.CloseNow() + _, _, counters := metrics.Snapshot() + if counters["agent_connection"] != 1 || counters["client_registration"] != 1 { + t.Fatalf("registration metrics = %#v", counters) + } +} + func TestAgentServerDispatchesCommandAdmittedDuringReconcile_HP_SES_11(t *testing.T) { server, agent, cleanup := newTestAgentServerWithStore(t) defer cleanup() diff --git a/test/coverage.toml b/test/coverage.toml index 372e8e2..7a7d41e 100644 --- a/test/coverage.toml +++ b/test/coverage.toml @@ -637,6 +637,15 @@ layer = "unit" status = "implemented" tests = ["internal/observability/rotate_test.go:TestFormatLogWritesStructuredBoundedRecords_HP_OPS_04"] +[[requirements]] +id = "HP-OPS-05" +layer = "unit" +status = "implemented" +tests = [ + "internal/observability/health_test.go:TestHealthRendersBoundedGaugesAndDurationHistograms_HP_OPS_02", + "internal/server/session/agent_server_test.go:TestAgentServerEmitsBoundedRegistrationMetrics_HP_OPS_05", +] + [[requirements]] id = "HP-DISPATCH-10" layer = "unit"