myxce/src/frontend/apiserver/store/concurrency_test.go
Rongsong Shen 86a65d640a test(apiserver): add integration tests for P0+P1 backend features
Covers ~30 integration tests across 8 sections from the
apiserver-backend-integration-test-plan:

- §1: Cluster config push (store population, merge semantics, zero-preserve)
- §2: Host control (OnHostUpdated merge, validateHost, preCheck, batch ops)
- §3: Resource value push (all 7 value types, nil/unknown discriminant, overwrite)
- §4: Heartbeat keep-alive (interval update, zero-preserve)
- §5: Cross-feature (config+heartbeat, config+host, worker+config field preservation)
- §6: RBAC (cluster_admin bypass, algorithm confusion, expired tokens, hostgroup bypass)
- §7: Store thread safety (concurrent config pushes, host control+config, resource values)
- §8: Input safety (hostname length, empty lists, malformed payloads)

All tests use CGO_ENABLED=0 with //go:build !cgo build tag.
§8.1 hostname-length skipped pending P1-2b implementation.
2026-04-18 11:56:31 +00:00

132 lines
No EOL
2.8 KiB
Go

package store
import (
"apiserver/models"
"sync"
"sync/atomic"
"testing"
)
// ========== §7. Store Thread Safety ==========
// All tests in this file MUST be run with -race:
// go test -race ./store/...
// resetStores reinitializes all global stores for test isolation.
func resetStores() {
InitStores()
}
// --- 7.1 P2: Concurrent config pushes don't corrupt data ---
func TestIntegration_ConcurrentConfigPushes(t *testing.T) {
resetStores()
var wg sync.WaitGroup
const goroutines = 10
for i := 0; i < goroutines; i++ {
wg.Add(1)
go func(idx int) {
defer wg.Done()
model := models.HostInfo{
HostName: "node01",
HostModel: "model_x",
HostType: "LINUX64",
MaxSlot: idx + 1,
}
Hosts.Put("node01", model)
}(i)
}
wg.Wait()
got, ok := Hosts.Get("node01")
if !ok {
t.Fatal("node01 should exist after concurrent pushes")
}
if got.HostName != "node01" {
t.Errorf("HostName = %q, want node01 (no zero-value corruption)", got.HostName)
}
}
// --- 7.2 P2: Concurrent host control + config push ---
func TestIntegration_ConcurrentHostControlAndConfigPush(t *testing.T) {
resetStores()
const iterations = 100
var wg sync.WaitGroup
wg.Add(iterations)
for i := 0; i < iterations; i++ {
go func() {
defer wg.Done()
Hosts.Put("node01", models.HostInfo{
HostName: "node01",
HostModel: "x86_64",
})
}()
}
wg.Add(iterations)
for i := 0; i < iterations; i++ {
go func() {
defer wg.Done()
Hosts.Has("node01")
}()
}
wg.Wait()
// No panics and store should be in a valid state
got, ok := Hosts.Get("node01")
if !ok {
t.Error("node01 should exist after concurrent access")
}
if got.HostName != "node01" {
t.Errorf("HostName = %q, want node01", got.HostName)
}
}
// --- 7.3 P2: Concurrent resource value pushes for same key ---
func TestIntegration_ConcurrentResourceValuePushes(t *testing.T) {
resetStores()
var wg sync.WaitGroup
const goroutines = 10
var count atomic.Int64
for i := 0; i < goroutines; i++ {
wg.Add(1)
go func(idx int) {
defer wg.Done()
ResourceLocations.Put("mem:node01", models.ResourceDistributionInfo{
ResourceName: "mem",
Location: "node01",
NumericValue: float64(idx + 1),
ValueType: 1, // NUMERIC
IsDynamic: true,
})
count.Add(1)
}(i)
}
wg.Wait()
got, ok := ResourceLocations.Get("mem:node01")
if !ok {
t.Fatal("mem:node01 should exist after concurrent pushes")
}
// Final value must be one of the 10 valid values (no corruption)
validValue := false
for i := 1; i <= goroutines; i++ {
if got.NumericValue == float64(i) {
validValue = true
break
}
}
if !validValue {
t.Errorf("NumericValue = %f, expected one of 1..10 (no corruption)", got.NumericValue)
}
if ResourceLocations.Count() != 1 {
t.Errorf("ResourceLocations count = %d, want 1 (key exists exactly once)", ResourceLocations.Count())
}
}