cabbage/rvbox
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add bounded operational metrics

Browse Source
This commit is contained in:
cabbage
2026-09-11 09:09:52 +00:00
parent 46185f1f6c
commit 2cf563d88e
12 changed files with 342 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/client/agent/executor.go
+24 -1
View File
@@ -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)
}
internal/client/agent/runner.go
+45
View File
@@ -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) {
internal/client/agent/runner_test.go
+11 -1
View File
@@ -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) {
internal/observability/health.go
+134 -9
View File
@@ -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
}
internal/observability/health_test.go
+29
View File
@@ -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)
}
}
internal/server/control/service.go
+13 -1
View File
@@ -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 {
internal/server/session/agent_server.go
+40
View File
@@ -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)
}
internal/server/session/agent_server_test.go
+14
View File
@@ -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()