philipp/uncloud
1
0
Fork 0
mirror of https://github.com/psviderski/uncloud.git synced 2026-08-26 11:03:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(rtt): use proto Duration, add tests

Browse Source
This commit is contained in:
Pasha Sviderski
2026-04-20 11:36:44 +10:00
parent 8d023f5c53
commit 84990ad692
10 changed files with 387 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/corrosion/admin.go
+51 -41
View File
@@ -261,8 +261,8 @@ func (c *AdminClient) ClusterMembershipStates(latest bool) ([]ClusterMembershipS
type MemberRTTStats struct {
Addr netip.AddrPort
Median float64
StdDev float64
Median time.Duration
StdDev time.Duration
}
// ClusterMemberRTTs returns the median and standard deviation of round-trip times to each cluster member.
@@ -290,44 +290,54 @@ func (c *AdminClient) ClusterMemberRTTs() ([]MemberRTTStats, error) {
continue
}
sort.Float64s(rtts)
n := len(rtts)
var median float64
if n%2 == 0 {
median = (rtts[n/2-1] + rtts[n/2]) / 2
} else {
median = rtts[n/2]
}
var sum float64
for _, rtt := range rtts {
sum += rtt
}
avg := sum / float64(n)
var varianceSum float64
for _, rtt := range rtts {
diff := rtt - avg
varianceSum += diff * diff
}
stdDev := math.Sqrt(varianceSum / float64(n))
// Corrosion reports RTT samples as floating-point milliseconds.
medianMs, stdDevMs := computeRTTStatsMs(rtts)
stats = append(stats, MemberRTTStats{
Addr: addr,
Median: median,
StdDev: stdDev,
Median: time.Duration(medianMs * float64(time.Millisecond)),
StdDev: time.Duration(stdDevMs * float64(time.Millisecond)),
})
}
return stats, parseErr
}
// computeRTTStatsMs returns the median and population standard deviation (ms) of the given samples.
func computeRTTStatsMs(rtts []float64) (median, stdDev float64) {
if len(rtts) == 0 {
return 0, 0
}
sort.Float64s(rtts)
n := len(rtts)
if n%2 == 0 {
median = (rtts[n/2-1] + rtts[n/2]) / 2
} else {
median = rtts[n/2]
}
var sum float64
for _, rtt := range rtts {
sum += rtt
}
avg := sum / float64(n)
var varianceSum float64
for _, rtt := range rtts {
diff := rtt - avg
varianceSum += diff * diff
}
stdDev = math.Sqrt(varianceSum / float64(n))
return median, stdDev
}
func parseClusterMemberRTT(json map[string]any) (netip.AddrPort, []float64, error) {
var addr netip.AddrPort
var rtts []float64
var err error
// Parse state to get Addr
// Parse state to get Addr.
stateObj, ok := json["state"].(map[string]any)
if !ok {
return addr, nil, fmt.Errorf("missing or invalid 'state' field")
@@ -342,21 +352,21 @@ func parseClusterMemberRTT(json map[string]any) (netip.AddrPort, []float64, erro
return addr, nil, fmt.Errorf("missing or invalid 'addr' field in 'state'")
}
// Parse RTTs
if rttsVal, ok := json["rtts"]; ok {
if rttsSlice, ok := rttsVal.([]any); ok {
for _, v := range rttsSlice {
if f, ok := v.(float64); ok {
rtts = append(rtts, f)
} else {
return addr, nil, fmt.Errorf("invalid rtt value type: %T", v)
}
}
} else {
return addr, nil, fmt.Errorf("invalid 'rtts' field type: %T", rttsVal)
// The absent 'rtts' key or equal to null are treated as no samples yet.
rttsVal, ok := json["rtts"]
if !ok || rttsVal == nil {
return addr, nil, nil
}
rttsSlice, ok := rttsVal.([]any)
if !ok {
return addr, nil, fmt.Errorf("invalid 'rtts' field type: %T", rttsVal)
}
for _, v := range rttsSlice {
f, ok := v.(float64)
if !ok {
return addr, nil, fmt.Errorf("invalid rtt value type: %T", v)
}
} else {
return addr, nil, fmt.Errorf("missing 'rtts' field")
rtts = append(rtts, f)
}
return addr, rtts, nil
internal/corrosion/admin_test.go
+174
View File
@@ -0,0 +1,174 @@
package corrosion
import (
"math"
"net/netip"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestComputeRTTStatsMs(t *testing.T) {
t.Parallel()
tests := []struct {
name string
samples []float64
wantMedian float64
wantStdDev float64
}{
{
name: "single sample",
samples: []float64{42},
wantMedian: 42,
wantStdDev: 0,
},
{
name: "two samples (even, averaged)",
samples: []float64{10, 20},
wantMedian: 15,
// Population stddev: variance = ((10-15)^2 + (20-15)^2)/2 = 25; sqrt = 5.
wantStdDev: 5,
},
{
name: "three samples (odd, middle)",
samples: []float64{3, 1, 2},
wantMedian: 2,
// Mean = 2; variance = (1+0+1)/3 = 0.666...; stddev = sqrt(2/3).
wantStdDev: math.Sqrt(2.0 / 3.0),
},
{
name: "four samples (even, averaged)",
samples: []float64{1, 2, 3, 4},
wantMedian: 2.5,
// Mean = 2.5; variance = (2.25+0.25+0.25+2.25)/4 = 1.25; stddev = sqrt(1.25).
wantStdDev: math.Sqrt(1.25),
},
{
// Verifies that unsorted input is sorted before picking the median.
name: "unsorted input",
samples: []float64{100, 1, 50},
wantMedian: 50,
// Mean = 151/3. Variance = ((149^2 + 148^2 + 1^2) / 9) / 3.
wantStdDev: math.Sqrt(float64(149*149+148*148+1) / 9.0 / 3.0),
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
median, stdDev := computeRTTStatsMs(tt.samples)
assert.InDelta(t, tt.wantMedian, median, 1e-9, "median")
assert.InDelta(t, tt.wantStdDev, stdDev, 1e-9, "stdDev")
})
}
}
func TestParseClusterMemberRTT(t *testing.T) {
validState := map[string]any{"addr": "[fdcc:b618:5034:7afa:172a:1452:f2de:3c99]:51001"}
tests := []struct {
name string
input map[string]any
wantAddr string
wantRTTs []float64
wantErr bool
errSubstr string
}{
{
name: "valid state and rtts",
input: map[string]any{
"state": validState,
"rtts": []any{float64(10), float64(20), float64(30)},
},
wantAddr: "[fdcc:b618:5034:7afa:172a:1452:f2de:3c99]:51001",
wantRTTs: []float64{10, 20, 30},
},
{
name: "missing rtts key is not an error",
input: map[string]any{
"state": validState,
},
wantAddr: "[fdcc:b618:5034:7afa:172a:1452:f2de:3c99]:51001",
wantRTTs: nil,
},
{
name: "null rtts value is not an error",
input: map[string]any{
"state": validState,
"rtts": nil,
},
wantAddr: "[fdcc:b618:5034:7afa:172a:1452:f2de:3c99]:51001",
wantRTTs: nil,
},
{
name: "empty rtts array is not an error",
input: map[string]any{
"state": validState,
"rtts": []any{},
},
wantAddr: "[fdcc:b618:5034:7afa:172a:1452:f2de:3c99]:51001",
wantRTTs: nil,
},
{
name: "non-array rtts",
input: map[string]any{
"state": validState,
"rtts": "not-an-array",
},
wantErr: true,
errSubstr: "invalid 'rtts' field type",
},
{
name: "non-number element in rtts",
input: map[string]any{
"state": validState,
"rtts": []any{float64(10), "bad"},
},
wantErr: true,
errSubstr: "invalid rtt value type",
},
{
name: "missing state",
input: map[string]any{"rtts": []any{float64(1)}},
wantErr: true,
errSubstr: "missing or invalid 'state' field",
},
{
name: "missing addr in state",
input: map[string]any{
"state": map[string]any{},
"rtts": []any{float64(1)},
},
wantErr: true,
errSubstr: "missing or invalid 'addr' field",
},
{
name: "invalid addr format",
input: map[string]any{
"state": map[string]any{"addr": "not-an-addr"},
"rtts": []any{float64(1)},
},
wantErr: true,
errSubstr: "parse 'addr' field",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
addr, rtts, err := parseClusterMemberRTT(tt.input)
if tt.wantErr {
require.Error(t, err)
assert.Contains(t, err.Error(), tt.errSubstr)
return
}
require.NoError(t, err)
assert.Equal(t, tt.wantAddr, addr.String())
assert.Equal(t, tt.wantRTTs, rtts)
// Sanity: the parsed addr is a valid AddrPort.
_, perr := netip.ParseAddrPort(addr.String())
assert.NoError(t, perr)
})
}
}