mirror of
https://github.com/Kpa-clawbot/meshcore-analyzer.git
synced 2026-10-23 09:32:26 +00:00
perf(#1265): wire /api/observers/clock-skew + /api/nodes/clock-skew into analytics recomputer (#1266)
RED: 97f49a0c · CI:
https://github.com/Kpa-clawbot/CoreScope/actions/runs/26046530920
Fixes #1265.
## Problem
On staging two clock-skew endpoints serve compute-on-request:
- `/api/observers/clock-skew` — 3.3s
- `/api/nodes/clock-skew` — 8.9s
Both drive a full `clockSkew.Recompute` over 100k+ adverts while holding
`s.mu.RLock`, blocking under concurrent reader load.
## Fix
Wire both endpoints into the established `analytics_recomputer.go`
pattern (PRs #1248 / #1259 / #1263). Two new slots:
- `recompObserversClockSkew` — wraps `computeObserverCalibrations()`
- `recompNodesClockSkew` — wraps `computeFleetClockSkew()`
Accessors `GetObserverCalibrations` / `GetFleetClockSkew` now prefer the
atomic-pointer snapshot; on-request compute is fallback-only for the
brief window before initial sync compute lands (and for tests that skip
the recomputer).
Default interval **300s**, overridable via:
```json
"analytics": {
"recomputeIntervalSeconds": {
"observersClockSkew": 300,
"nodesClockSkew": 300
}
}
```
`config.example.json` + the `_comment_analytics` doc updated.
## TDD
- RED `97f49a0c` — `TestClockSkewRecomputersRegistered` +
`TestClockSkewHandlersSteadyStateLatency` (8 concurrent readers × 25
reqs per endpoint, p99 < 100ms gate). Fails on master: recomputer slots
nil.
- GREEN `19599375` — wire + accessor switch. p99 well under 5ms on the
test fixture.
## Verification
```
cd cmd/server && go test ./... -count=1 # ok 42s
bash ~/.openclaw/skills/pr-preflight/scripts/run-all.sh origin/master # all gates pass
```
---------
Co-authored-by: CoreScope Bot <bot@corescope.local>
This commit is contained in:
1 parent
ae17a2be12
commit
4cd8445233
6 files changed
+168
-14
No files matched your search
@@ -146,13 +146,15 @@ func (r *analyticsRecomputer) ComputeRuns() int64 {
|
||||
// per-endpoint recompute interval from config.json. Zero values fall
|
||||
// back to the defaultInterval passed to StartAnalyticsRecomputers.
|
||||
type AnalyticsRecomputeIntervals struct {
|
||||
Topology time.Duration
|
||||
RF time.Duration
|
||||
Distance time.Duration
|
||||
Channels time.Duration
|
||||
HashCollisions time.Duration
|
||||
HashSizes time.Duration
|
||||
Roles time.Duration
|
||||
Topology time.Duration
|
||||
RF time.Duration
|
||||
Distance time.Duration
|
||||
Channels time.Duration
|
||||
HashCollisions time.Duration
|
||||
HashSizes time.Duration
|
||||
Roles time.Duration
|
||||
ObserversClockSkew time.Duration
|
||||
NodesClockSkew time.Duration
|
||||
}
|
||||
|
||||
func pickInterval(override, def time.Duration) time.Duration {
|
||||
@@ -224,10 +226,19 @@ func (s *PacketStore) StartAnalyticsRecomputers(defaultInterval time.Duration, o
|
||||
"roles", pickInterval(ov.Roles, defaultInterval),
|
||||
func() interface{} { return s.computeAnalyticsRoles() },
|
||||
)
|
||||
s.recompObserversClockSkew = newAnalyticsRecomputer(
|
||||
"observers-clock-skew", pickInterval(ov.ObserversClockSkew, defaultInterval),
|
||||
func() interface{} { return s.computeObserverCalibrations() },
|
||||
)
|
||||
s.recompNodesClockSkew = newAnalyticsRecomputer(
|
||||
"nodes-clock-skew", pickInterval(ov.NodesClockSkew, defaultInterval),
|
||||
func() interface{} { return s.computeFleetClockSkew() },
|
||||
)
|
||||
all := []*analyticsRecomputer{
|
||||
s.recompTopology, s.recompRF, s.recompDistance,
|
||||
s.recompChannels, s.recompHashCollisions, s.recompHashSizes,
|
||||
s.recompRoles,
|
||||
s.recompObserversClockSkew, s.recompNodesClockSkew,
|
||||
}
|
||||
s.analyticsRecomputerMu.Unlock()
|
||||
|
||||
|
||||
@@ -694,9 +694,27 @@ func (s *PacketStore) getNodeClockSkewLocked(pubkey string) *NodeClockSkew {
|
||||
}
|
||||
}
|
||||
|
||||
// GetFleetClockSkew returns clock skew data for all nodes that have skew data.
|
||||
// Must NOT be called with s.mu held.
|
||||
// GetFleetClockSkew returns clock skew data for all nodes, preferring
|
||||
// the steady-state recomputer snapshot (issue #1265). Falls back to an
|
||||
// on-request compute if the recomputer is not yet running.
|
||||
func (s *PacketStore) GetFleetClockSkew() []*NodeClockSkew {
|
||||
s.analyticsRecomputerMu.RLock()
|
||||
rc := s.recompNodesClockSkew
|
||||
s.analyticsRecomputerMu.RUnlock()
|
||||
if rc != nil {
|
||||
if v := rc.Load(); v != nil {
|
||||
if r, ok := v.([]*NodeClockSkew); ok {
|
||||
return r
|
||||
}
|
||||
}
|
||||
}
|
||||
return s.computeFleetClockSkew()
|
||||
}
|
||||
|
||||
// computeFleetClockSkew is the underlying compute used by the
|
||||
// recomputer and the on-request fallback. Must NOT be called with
|
||||
// s.mu held.
|
||||
func (s *PacketStore) computeFleetClockSkew() []*NodeClockSkew {
|
||||
s.mu.RLock()
|
||||
defer s.mu.RUnlock()
|
||||
|
||||
@@ -707,7 +725,7 @@ func (s *PacketStore) GetFleetClockSkew() []*NodeClockSkew {
|
||||
nameMap[ni.PublicKey] = ni
|
||||
}
|
||||
|
||||
var results []*NodeClockSkew
|
||||
var results = []*NodeClockSkew{}
|
||||
for pubkey := range s.byNode {
|
||||
cs := s.getNodeClockSkewLocked(pubkey)
|
||||
if cs == nil {
|
||||
@@ -727,8 +745,26 @@ func (s *PacketStore) GetFleetClockSkew() []*NodeClockSkew {
|
||||
return results
|
||||
}
|
||||
|
||||
// GetObserverCalibrations returns the current observer clock offsets.
|
||||
// GetObserverCalibrations returns the current observer clock offsets,
|
||||
// preferring the steady-state recomputer snapshot (issue #1265). Falls
|
||||
// back to an on-request compute when the recomputer is not running.
|
||||
func (s *PacketStore) GetObserverCalibrations() []ObserverCalibration {
|
||||
s.analyticsRecomputerMu.RLock()
|
||||
rc := s.recompObserversClockSkew
|
||||
s.analyticsRecomputerMu.RUnlock()
|
||||
if rc != nil {
|
||||
if v := rc.Load(); v != nil {
|
||||
if r, ok := v.([]ObserverCalibration); ok {
|
||||
return r
|
||||
}
|
||||
}
|
||||
}
|
||||
return s.computeObserverCalibrations()
|
||||
}
|
||||
|
||||
// computeObserverCalibrations is the underlying compute used by the
|
||||
// recomputer and on-request fallback. Must NOT be called with s.mu held.
|
||||
func (s *PacketStore) computeObserverCalibrations() []ObserverCalibration {
|
||||
s.mu.RLock()
|
||||
defer s.mu.RUnlock()
|
||||
|
||||
|
||||
@@ -0,0 +1,101 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"sort"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Issue #1265: /api/observers/clock-skew (3.3s) and /api/nodes/clock-skew (8.9s)
|
||||
// must be wired into the steady-state analytics recomputer so reads serve
|
||||
// from an atomic-pointer snapshot in <100ms p99 under concurrent load.
|
||||
|
||||
func TestClockSkewRecomputersRegistered(t *testing.T) {
|
||||
db := setupTestDB(t)
|
||||
defer db.Close()
|
||||
store := NewPacketStore(db, nil)
|
||||
|
||||
stop := store.StartAnalyticsRecomputers(50 * time.Millisecond)
|
||||
defer stop()
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
|
||||
store.analyticsRecomputerMu.RLock()
|
||||
rcObs := store.recompObserversClockSkew
|
||||
rcNodes := store.recompNodesClockSkew
|
||||
store.analyticsRecomputerMu.RUnlock()
|
||||
|
||||
if rcObs == nil {
|
||||
t.Fatalf("recompObserversClockSkew not registered after StartAnalyticsRecomputers (issue #1265 not fixed)")
|
||||
}
|
||||
if rcNodes == nil {
|
||||
t.Fatalf("recompNodesClockSkew not registered after StartAnalyticsRecomputers (issue #1265 not fixed)")
|
||||
}
|
||||
if rcObs.Load() == nil {
|
||||
t.Fatalf("recompObserversClockSkew snapshot is nil after initial compute")
|
||||
}
|
||||
if rcNodes.Load() == nil {
|
||||
t.Fatalf("recompNodesClockSkew snapshot is nil after initial compute")
|
||||
}
|
||||
}
|
||||
|
||||
func TestClockSkewHandlersSteadyStateLatency(t *testing.T) {
|
||||
db := setupTestDB(t)
|
||||
defer db.Close()
|
||||
store := NewPacketStore(db, nil)
|
||||
stop := store.StartAnalyticsRecomputers(50 * time.Millisecond)
|
||||
defer stop()
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
|
||||
s := &Server{store: store}
|
||||
|
||||
endpoints := []struct {
|
||||
name string
|
||||
path string
|
||||
handler http.HandlerFunc
|
||||
}{
|
||||
{"observers", "/api/observers/clock-skew", s.handleObserverClockSkew},
|
||||
{"nodes", "/api/nodes/clock-skew", s.handleFleetClockSkew},
|
||||
}
|
||||
|
||||
for _, ep := range endpoints {
|
||||
ep := ep
|
||||
t.Run(ep.name, func(t *testing.T) {
|
||||
const readers = 8
|
||||
const perReader = 25
|
||||
var (
|
||||
mu sync.Mutex
|
||||
samples []time.Duration
|
||||
wg sync.WaitGroup
|
||||
)
|
||||
wg.Add(readers)
|
||||
for r := 0; r < readers; r++ {
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
for i := 0; i < perReader; i++ {
|
||||
rr := httptest.NewRecorder()
|
||||
req := httptest.NewRequest(http.MethodGet, ep.path, nil)
|
||||
t0 := time.Now()
|
||||
ep.handler(rr, req)
|
||||
dt := time.Since(t0)
|
||||
if rr.Code != http.StatusOK {
|
||||
t.Errorf("%s status = %d, want 200", ep.path, rr.Code)
|
||||
}
|
||||
mu.Lock()
|
||||
samples = append(samples, dt)
|
||||
mu.Unlock()
|
||||
}
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
|
||||
sort.Slice(samples, func(i, j int) bool { return samples[i] < samples[j] })
|
||||
p99 := samples[int(float64(len(samples))*0.99)]
|
||||
if p99 > 100*time.Millisecond {
|
||||
t.Fatalf("%s p99 latency = %v over %d reqs, want <100ms (recomputer snapshot)", ep.path, p99, len(samples))
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -483,7 +483,7 @@ func (c *Config) IsObserverBlacklisted(id string) bool {
|
||||
// data slowly." Lower values give fresher data at higher CPU cost.
|
||||
//
|
||||
// RecomputeIntervalSeconds keys (all optional):
|
||||
// topology, rf, distance, channels, hashCollisions, hashSizes, roles
|
||||
// topology, rf, distance, channels, hashCollisions, hashSizes, roles, observersClockSkew, nodesClockSkew
|
||||
type AnalyticsConfig struct {
|
||||
DefaultIntervalSeconds int `json:"defaultIntervalSeconds,omitempty"`
|
||||
RecomputeIntervalSeconds map[string]int `json:"recomputeIntervalSeconds,omitempty"`
|
||||
@@ -520,5 +520,7 @@ func (c *Config) AnalyticsRecomputeIntervals() AnalyticsRecomputeIntervals {
|
||||
out.HashCollisions = get("hashCollisions")
|
||||
out.HashSizes = get("hashSizes")
|
||||
out.Roles = get("roles")
|
||||
out.ObserversClockSkew = get("observersClockSkew")
|
||||
out.NodesClockSkew = get("nodesClockSkew")
|
||||
return out
|
||||
}
|
||||
@@ -165,6 +165,8 @@ type PacketStore struct {
|
||||
recompHashCollisions *analyticsRecomputer
|
||||
recompHashSizes *analyticsRecomputer
|
||||
recompRoles *analyticsRecomputer
|
||||
recompObserversClockSkew *analyticsRecomputer
|
||||
recompNodesClockSkew *analyticsRecomputer
|
||||
cacheHits int64
|
||||
cacheMisses int64
|
||||
// Rate-limited invalidation (fixes #533: caches cleared faster than hit)
|
||||
|
||||
+4
-2
@@ -259,8 +259,10 @@
|
||||
"channels": 300,
|
||||
"hashCollisions": 300,
|
||||
"hashSizes": 300,
|
||||
"roles": 300
|
||||
"roles": 300,
|
||||
"observersClockSkew": 300,
|
||||
"nodesClockSkew": 300
|
||||
}
|
||||
},
|
||||
"_comment_analytics": "Issue #1240 + #1256. Each analytics endpoint (topology, rf, distance, channels, hashCollisions, hashSizes, roles) is recomputed in the background on the configured interval and served from an atomic-pointer cache. Reads never block on compute. Default 300s (5 min) per endpoint reflects the operator principle: serving slightly stale data quickly beats real-time data slowly. Lower values = fresher data at higher CPU cost. Only the default query (no region/window) is precomputed; region- and window-filtered requests fall back to the legacy on-request compute + 60s TTL cache."
|
||||
"_comment_analytics": "Issue #1240 + #1256 + #1265. Each analytics endpoint (topology, rf, distance, channels, hashCollisions, hashSizes, roles, observersClockSkew, nodesClockSkew) is recomputed in the background on the configured interval and served from an atomic-pointer cache. Reads never block on compute. Default 300s (5 min) per endpoint reflects the operator principle: serving slightly stale data quickly beats real-time data slowly. Lower values = fresher data at higher CPU cost. Only the default query (no region/window) is precomputed; region- and window-filtered requests fall back to the legacy on-request compute + 60s TTL cache."
|
||||
}
|
||||
Reference in new issue
Block a user