MiFaZhan
111 lines
5.1 KiB
Go
111 lines
5.1 KiB
Go
package dailyusage
|
|
|
|
import (
|
|
"reflect"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/queue/zset"
|
|
)
|
|
|
|
func TestLocalSupplyObservationFirstEventsAndStages(t *testing.T) {
|
|
started := time.Now()
|
|
o := &supplyObservation{started: started, channels: map[string]*supplyChannel{"a": {stage: "initializing", changed: started, stageDuration: map[string]time.Duration{}}}}
|
|
o.event("a", "sql_query", 0, false, false, started.Add(10*time.Millisecond))
|
|
o.event("a", "page_ready", 2, false, false, started.Add(20*time.Millisecond))
|
|
o.event("a", "redis_write", 1, true, false, started.Add(30*time.Millisecond))
|
|
o.event("a", "channel_capacity", 1, false, false, started.Add(40*time.Millisecond))
|
|
o.event("a", "sql_query", 0, false, false, started.Add(50*time.Millisecond))
|
|
o.event("a", "redis_write", 0, true, false, started.Add(60*time.Millisecond))
|
|
o.event("a", "upper_bound", 0, false, true, started.Add(70*time.Millisecond))
|
|
o.event("unknown", "ignored", 0, false, false, started)
|
|
s := o.channels["a"]
|
|
if s.firstQuery == nil || *s.firstQuery != 10 || s.firstEnqueue == nil || *s.firstEnqueue != 30 || !s.ended || s.pending != 0 || s.version != 7 || s.stageDuration["channel_capacity"] != 10*time.Millisecond || s.stageDuration["sql_query"] != 20*time.Millisecond {
|
|
t.Fatalf("observation=%+v", s)
|
|
}
|
|
}
|
|
|
|
func TestSupplyObservationAccumulatesSubMillisecondStages(t *testing.T) {
|
|
started := time.Unix(0, 0)
|
|
o := &supplyObservation{started: started, channels: map[string]*supplyChannel{"a": {stage: "redis_read", changed: started, stageDuration: map[string]time.Duration{}}}}
|
|
for i := 1; i <= 100000; i++ {
|
|
stage := "redis_read"
|
|
if i%2 == 1 {
|
|
stage = "redis_write"
|
|
}
|
|
o.event("a", stage, 0, false, i == 100000, started.Add(time.Duration(i)*500*time.Microsecond))
|
|
}
|
|
if got := o.channels["a"].stageMilliseconds(started.Add(time.Hour)); got["redis_read"] != 25000 || got["redis_write"] != 25000 {
|
|
t.Fatalf("stageMs=%v, want 25000ms per stage", got)
|
|
}
|
|
}
|
|
|
|
func TestSupplyObservationFinalFlushPreservesFraction(t *testing.T) {
|
|
started := time.Unix(0, 0)
|
|
o := &supplyObservation{started: started, channels: map[string]*supplyChannel{"a": {stage: "redis_write", changed: started, stageDuration: map[string]time.Duration{}}}}
|
|
o.event("a", "redis_read", 0, false, false, started.Add(600*time.Microsecond))
|
|
o.event("a", "redis_write", 0, false, false, started.Add(1200*time.Microsecond))
|
|
s := o.channels["a"]
|
|
if got := s.stageMilliseconds(started.Add(1800 * time.Microsecond)); got["redis_write"] != 1 || got["redis_read"] != 0 {
|
|
t.Fatalf("final stageMs=%v", got)
|
|
}
|
|
if s.stageDuration["redis_write"] != 1200*time.Microsecond || s.stageDuration["redis_read"] != 600*time.Microsecond {
|
|
t.Fatalf("fractional duration lost: %v", s.stageDuration)
|
|
}
|
|
}
|
|
|
|
func TestSupplySamplesDistinguishReadyLeaseFutureAndClosedRound(t *testing.T) {
|
|
store := testStore(t)
|
|
ctx := contextForTest(t)
|
|
p := testProducer(t, scannerFor(nil, 0), store, Config{})
|
|
if err := store.CreateRun(ctx, zset.RunMeta{RunID: "supply"}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
active, _ := store.RoundActiveKey("supply", 1, "a")
|
|
lease, _ := store.RoundLeaseKey("supply", 1, "a")
|
|
_, _ = store.RedisClient().Do(ctx, "ZADD", active, time.Now().Add(-time.Minute).UnixMilli(), "ready", time.Now().Add(time.Hour).UnixMilli(), "future")
|
|
_, _ = store.RedisClient().Do(ctx, "HSET", lease, "leased", "token")
|
|
values, err := p.sampleSupply(ctx, "supply", "a")
|
|
if err != nil || !reflect.DeepEqual(values, []int64{1, 1, 1, 1, 1}) {
|
|
t.Fatalf("sample=%v err=%v", values, err)
|
|
}
|
|
meta, _ := store.RunMetaKey("supply")
|
|
for _, args := range [][]any{{"blocked:a", 1}, {"blocked:a", 0, "producerDone", 1}, {"producerDone", 0, "currentRound", 2}, {"currentRound", 1, "syncType", "replay"}} {
|
|
_, _ = store.RedisClient().Do(ctx, "HSET", append([]any{meta}, args...)...)
|
|
values, err = p.sampleSupply(ctx, "supply", "a")
|
|
if err != nil || values[0] != 1 || values[1] != 0 {
|
|
t.Fatalf("ineligible sample=%v err=%v", values, err)
|
|
}
|
|
}
|
|
values, err = p.sampleSupply(ctx, "missing", "a")
|
|
if err != nil || values[0] != 0 || values[1] != 0 {
|
|
t.Fatalf("missing sample=%v err=%v", values, err)
|
|
}
|
|
}
|
|
|
|
func TestSupplySamplerTracksWaitingChannelsAndStops(t *testing.T) {
|
|
store := testStore(t)
|
|
ctx := contextForTest(t)
|
|
p := testProducer(t, scannerFor(nil, 0), store, Config{})
|
|
if err := store.CreateRun(ctx, zset.RunMeta{RunID: "sampling"}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
finish := p.startObservation(ctx, "sampling", []string{"a", "b"}, time.Now())
|
|
defer finish()
|
|
o := p.observation
|
|
p.observe("a", "channel_capacity", 1, false, false)
|
|
p.observe("b", "sql_slot", 0, false, false)
|
|
waitUntil(t, ctx, func() bool {
|
|
o.mu.Lock()
|
|
defer o.mu.Unlock()
|
|
return o.channels["a"].validSamples > 0 && o.channels["b"].validSamples > 0
|
|
})
|
|
p.observe("a", "cancelled", 0, false, true)
|
|
p.observe("b", "cancelled", 0, false, true)
|
|
o.mu.Lock()
|
|
defer o.mu.Unlock()
|
|
if o.channels["a"].stageDuration["channel_capacity"] < 900*time.Millisecond || o.channels["b"].stageDuration["sql_slot"] < 900*time.Millisecond {
|
|
t.Fatalf("missing wait attribution: a=%+v b=%+v", o.channels["a"], o.channels["b"])
|
|
}
|
|
}
|