test: cover daemon telemetry publication
This commit is contained in:
@@ -245,15 +245,7 @@ func publishServerTelemetry(ctx context.Context, persistence *store.Store, healt
|
||||
ticker := time.NewTicker(10 * time.Second)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
sampleContext, cancel := context.WithTimeout(ctx, time.Second)
|
||||
telemetry, err := persistence.Telemetry(sampleContext)
|
||||
cancel()
|
||||
if err != nil {
|
||||
health.Inc("telemetry_read_failure")
|
||||
} else {
|
||||
health.SetGauge("queue_depth", float64(telemetry.QueuedCommands))
|
||||
health.SetGauge("server_charged_bytes", float64(telemetry.ChargedBytes))
|
||||
}
|
||||
publishServerTelemetrySample(ctx, persistence.Telemetry, health)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
@@ -261,3 +253,18 @@ func publishServerTelemetry(ctx context.Context, persistence *store.Store, healt
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// publishServerTelemetrySample performs one intentionally bounded, best-effort
|
||||
// telemetry read. A failed metrics read must never affect server readiness or
|
||||
// durable command processing.
|
||||
func publishServerTelemetrySample(ctx context.Context, read func(context.Context) (store.Telemetry, error), health *observability.Health) {
|
||||
sampleContext, cancel := context.WithTimeout(ctx, time.Second)
|
||||
telemetry, err := read(sampleContext)
|
||||
cancel()
|
||||
if err != nil {
|
||||
health.Inc("telemetry_read_failure")
|
||||
return
|
||||
}
|
||||
health.SetGauge("queue_depth", float64(telemetry.QueuedCommands))
|
||||
health.SetGauge("server_charged_bytes", float64(telemetry.ChargedBytes))
|
||||
}
|
||||
|
||||
@@ -0,0 +1,34 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
|
||||
"github.com/rvbox/rvbox/internal/observability"
|
||||
"github.com/rvbox/rvbox/internal/server/store"
|
||||
)
|
||||
|
||||
func TestPublishServerTelemetrySample_HP_OPS_07(t *testing.T) {
|
||||
t.Parallel()
|
||||
health := observability.New()
|
||||
publishServerTelemetrySample(context.Background(), func(context.Context) (store.Telemetry, error) {
|
||||
return store.Telemetry{QueuedCommands: 3, ChargedBytes: 4096}, nil
|
||||
}, health)
|
||||
snapshot := health.MetricsSnapshot()
|
||||
if snapshot.Gauges["queue_depth"] != 3 || snapshot.Gauges["server_charged_bytes"] != 4096 {
|
||||
t.Fatalf("server telemetry gauges = %#v", snapshot.Gauges)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPublishServerTelemetrySampleFailure_BH_OPS_04(t *testing.T) {
|
||||
t.Parallel()
|
||||
health := observability.New()
|
||||
publishServerTelemetrySample(context.Background(), func(context.Context) (store.Telemetry, error) {
|
||||
return store.Telemetry{}, errors.New("store unavailable")
|
||||
}, health)
|
||||
_, _, counters := health.Snapshot()
|
||||
if counters["telemetry_read_failure"] != 1 {
|
||||
t.Fatalf("telemetry failure counters = %#v", counters)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user