sechmachine/VerseVDI-Data-Plane
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test(gateway): retain raw wire qualification evidence
Verify Data Plane / gateway (push) Successful in 4m40s
Details

Browse Source
This commit is contained in:
sechmachine
2026-08-10 02:42:54 +07:00
parent c0e362c028
commit ce3b307983
6 changed files with 1165 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gateway/qualification_contract_test.go
+964 -6
View File
File diff suppressed because it is too large Load Diff
gateway/qualification_harness_test.go
+159 -10
View File
@@ -10,6 +10,7 @@ import (
"crypto/sha256"
"crypto/tls"
"encoding/binary"
"encoding/csv"
"encoding/hex"
"encoding/json"
"errors"
@@ -38,7 +39,7 @@ import (
)
const (
qualificationToolVersion = "versevdi-gateway-qualification/v9"
qualificationToolVersion = "versevdi-gateway-qualification/v10"
qualificationImpairmentQueuePackets = nativeApolloVideoQueuePackets
qualificationImpairmentMaxPackets = 100_000
qualificationImpairmentPacketCount = 10_000
@@ -50,6 +51,7 @@ const (
qualificationApolloVideoRateBitsPerSecond = 1_000_000_000 * 80 / 100
qualificationApolloVideoBatchBytes = 64 * 1024
qualificationApolloVideoBatchPackets = 64
qualificationWireTimebase = "monotonic offsets from constrained run start"
)
type qualificationMediaProfile struct {
@@ -204,13 +206,22 @@ type qualificationImpairmentObservation struct {
RawSamples string `json:"raw_samples"`
RawSamplesSHA256 string `json:"raw_samples_sha256"`
RawSamplesBytes int64 `json:"raw_samples_bytes"`
RawWireSamples string `json:"raw_wire_samples,omitempty"`
RawWireSamplesSHA256 string `json:"raw_wire_samples_sha256,omitempty"`
RawWireSamplesBytes int64 `json:"raw_wire_samples_bytes,omitempty"`
RawWireRows int `json:"raw_wire_rows,omitempty"`
RawWireDeliveryRows int `json:"raw_wire_delivery_rows,omitempty"`
RawWireTransitionRows int `json:"raw_wire_transition_rows,omitempty"`
RawWireTimebase string `json:"raw_wire_timebase,omitempty"`
}
type qualificationCapacityStep struct {
ReductionPercent int `json:"reduction_percent"`
Convergence time.Duration `json:"convergence_ns"`
MaximumFiveSecond int64 `json:"maximum_five_second_bytes"`
FiveSecondCap int64 `json:"five_second_cap_bytes"`
ReductionPercent int `json:"reduction_percent"`
TransitionAfter time.Duration `json:"transition_after_ns"`
Convergence time.Duration `json:"convergence_ns"`
MaximumFiveSecond int64 `json:"maximum_five_second_bytes"`
FiveSecondCap int64 `json:"five_second_cap_bytes"`
RecomputationSource string `json:"recomputation_source"`
}
type qualificationResourceSample struct {
@@ -227,6 +238,15 @@ type qualificationDeliverySample struct {
Bytes int64
}
type qualificationWireFileEvidence struct {
Name string
SHA256 string
Bytes int64
Rows int
DeliveryRows int
TransitionRows int
}
type qualificationFairnessEvidence struct {
Evaluation time.Duration `json:"evaluation_ns"`
PerFlowBytes map[string]int64 `json:"per_flow_bytes"`
@@ -1507,7 +1527,10 @@ func runQualificationImpairment(t *testing.T, profile qualificationImpairmentPro
beforeProviderDrops := path.session.mediaDrops.Load()
beforeSourceUDP := path.sourceUDP.Load()
beforePacer := path.server.pacer.reservations.Load()
wireDeliveries := make([]qualificationDeliverySample, 0, len(jobs)*2)
wireFragmentsPerUnit := (media.PacketBytes + frameV2PayloadSize - 1) / frameV2PayloadSize
wireDeliveryLimit := len(jobs) * wireFragmentsPerUnit
wireDeliveries := make([]qualificationDeliverySample, 0, wireDeliveryLimit)
wireOverflow := false
grace := max(2*profile.RTT+2*profile.Jitter, 2*time.Second)
lastTarget := time.Duration(packetCount) * spacing
if len(jobs) > 0 {
@@ -1528,6 +1551,10 @@ func runQualificationImpairment(t *testing.T, profile qualificationImpairmentPro
seen := make([]bool, packetCount)
for len(result.packets) < len(jobs) {
recovered, receiveErr := path.receivePayloadObserved(receiveCtx, func(receivedAt time.Time, bytes int) {
if len(wireDeliveries) >= wireDeliveryLimit {
wireOverflow = true
return
}
wireDeliveries = append(wireDeliveries, qualificationDeliverySample{At: receivedAt, Bytes: int64(bytes)})
})
if receiveErr != nil {
@@ -1601,6 +1628,9 @@ func runQualificationImpairment(t *testing.T, profile qualificationImpairmentPro
if received.err != nil {
return qualificationImpairmentObservation{}, received.err
}
if wireOverflow {
return qualificationImpairmentObservation{}, errors.New("qualification public-wire observation bound exceeded")
}
var totalLatency, totalJitter, previousLatency time.Duration
previousDelivered := -1
@@ -1713,12 +1743,33 @@ func runQualificationImpairment(t *testing.T, profile qualificationImpairmentPro
if err != nil {
return qualificationImpairmentObservation{}, err
}
wirePath := strings.TrimSuffix(rawPath, ".csv.gz") + "-wire.csv.gz"
for _, reduction := range profile.CapacitySteps {
step := qualificationCapacityStepObservation(wireDeliveries, stepAt[reduction], media, reduction)
step.TransitionAfter = stepAt[reduction].Sub(started)
step.RecomputationSource = filepath.Base(wirePath)
observation.CapacityStepObservations = append(observation.CapacityStepObservations, step)
}
if len(profile.CapacitySteps) > 0 {
wireEvidence, writeErr := writeQualificationWireSamples(
wirePath, started, profile.CapacitySteps, stepAt, wireDeliveries,
wireDeliveryLimit+len(profile.CapacitySteps),
)
if writeErr != nil {
return qualificationImpairmentObservation{}, writeErr
}
observation.RawWireSamples = wireEvidence.Name
observation.RawWireSamplesSHA256 = wireEvidence.SHA256
observation.RawWireSamplesBytes = wireEvidence.Bytes
observation.RawWireRows = wireEvidence.Rows
observation.RawWireDeliveryRows = wireEvidence.DeliveryRows
observation.RawWireTransitionRows = wireEvidence.TransitionRows
observation.RawWireTimebase = qualificationWireTimebase
}
for _, step := range observation.CapacityStepObservations {
if packetCount >= qualificationImpairmentPacketCount &&
(step.Convergence > 10*time.Second || step.MaximumFiveSecond > step.FiveSecondCap*105/100) {
return qualificationImpairmentObservation{}, fmt.Errorf("capacity step %d failed measured convergence=%s five-second=%d", reduction, step.Convergence, step.MaximumFiveSecond)
return qualificationImpairmentObservation{}, fmt.Errorf("capacity step %d failed measured convergence=%s five-second=%d", step.ReductionPercent, step.Convergence, step.MaximumFiveSecond)
}
}
if observation.Delivered+observation.Dropped != observation.Sent ||
@@ -2128,6 +2179,100 @@ func qualificationFileSHA256(path string) (string, int64, error) {
return hex.EncodeToString(hash.Sum(nil)), size, nil
}
func writeQualificationWireSamples(
path string,
epoch time.Time,
reductions []int,
transitions map[int]time.Time,
deliveries []qualificationDeliverySample,
maximumRows int,
) (qualificationWireFileEvidence, error) {
type wireRecord struct {
kind string
reduction int
after time.Duration
bytes int64
}
if maximumRows < len(reductions)+len(deliveries) {
return qualificationWireFileEvidence{}, errors.New("qualification public-wire row bound exceeded")
}
records := make([]wireRecord, 0, len(reductions)+len(deliveries))
for _, reduction := range reductions {
at := transitions[reduction]
if at.IsZero() || at.Before(epoch) {
return qualificationWireFileEvidence{}, fmt.Errorf("qualification capacity transition %d missing or before epoch", reduction)
}
records = append(records, wireRecord{kind: "transition", reduction: reduction, after: at.Sub(epoch)})
}
for _, delivery := range deliveries {
if delivery.At.Before(epoch) || delivery.Bytes <= 0 {
return qualificationWireFileEvidence{}, errors.New("qualification public-wire delivery invalid")
}
records = append(records, wireRecord{kind: "delivery", after: delivery.At.Sub(epoch), bytes: delivery.Bytes})
}
sort.SliceStable(records, func(first, second int) bool {
if records[first].after != records[second].after {
return records[first].after < records[second].after
}
return records[first].kind == "transition" && records[second].kind != "transition"
})
file, err := os.OpenFile(path, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o640)
if err != nil {
return qualificationWireFileEvidence{}, err
}
compressed := gzip.NewWriter(file)
buffered := bufio.NewWriter(compressed)
writer := csv.NewWriter(buffered)
closeAll := func() error {
writer.Flush()
if err := writer.Error(); err != nil {
_ = compressed.Close()
_ = file.Close()
return err
}
if err := buffered.Flush(); err != nil {
_ = compressed.Close()
_ = file.Close()
return err
}
if err := compressed.Close(); err != nil {
_ = file.Close()
return err
}
return file.Close()
}
if err := writer.Write([]string{"record_type", "reduction_percent", "transition_after_ns", "received_after_ns", "encoded_bytes"}); err != nil {
_ = closeAll()
return qualificationWireFileEvidence{}, err
}
evidence := qualificationWireFileEvidence{Name: filepath.Base(path)}
for _, record := range records {
var row []string
switch record.kind {
case "transition":
row = []string{"transition", strconv.Itoa(record.reduction), strconv.FormatInt(record.after.Nanoseconds(), 10), "", ""}
evidence.TransitionRows++
case "delivery":
row = []string{"delivery", "", "", strconv.FormatInt(record.after.Nanoseconds(), 10), strconv.FormatInt(record.bytes, 10)}
evidence.DeliveryRows++
default:
_ = closeAll()
return qualificationWireFileEvidence{}, errors.New("qualification public-wire record type invalid")
}
if err := writer.Write(row); err != nil {
_ = closeAll()
return qualificationWireFileEvidence{}, err
}
}
if err := closeAll(); err != nil {
return qualificationWireFileEvidence{}, err
}
evidence.Rows = evidence.TransitionRows + evidence.DeliveryRows
evidence.SHA256, evidence.Bytes, err = qualificationFileSHA256(path)
return evidence, err
}
type qualificationFlowDelivery struct {
at time.Time
flow string
@@ -2189,7 +2334,7 @@ func runQualificationFleetStage(t *testing.T, fleet *qualificationFleet, profile
func qualificationPacerEvidence(t *testing.T, rawPath string, baselineDuration, stepDuration time.Duration) (qualificationFairnessEvidence, error) {
t.Helper()
flows := []string{"one", "two", "three", "four", "five", "six", "seven", "eight"}
flows := qualificationFairnessFlows()
profile := qualificationMediaProfile{Name: "fairness-h264", Codec: "h264", BitrateKbps: 8000, PacketBytes: 1000}
fleet := newQualificationFleet(t, len(flows), profile, 8000)
defer fleet.Close()
@@ -2246,8 +2391,8 @@ func qualificationPacerEvidence(t *testing.T, rawPath string, baselineDuration,
return qualificationFairnessEvidence{}, fmt.Errorf("capacity step %d failed convergence=%s five-second=%d", step.reduction, convergence, maximum)
}
evidence.CapacitySteps = append(evidence.CapacitySteps, qualificationCapacityStep{
ReductionPercent: step.reduction, Convergence: convergence,
MaximumFiveSecond: maximum, FiveSecondCap: step.cap * 5,
ReductionPercent: step.reduction, TransitionAfter: stepStart.Sub(start), Convergence: convergence,
MaximumFiveSecond: maximum, FiveSecondCap: step.cap * 5, RecomputationSource: filepath.Base(rawPath),
})
}
end := start.Add(baselineDuration + 2*stepDuration)
@@ -2263,6 +2408,10 @@ func qualificationPacerEvidence(t *testing.T, rawPath string, baselineDuration,
return evidence, nil
}
func qualificationFairnessFlows() []string {
return []string{"one", "two", "three", "four", "five", "six", "seven", "eight"}
}
func qualificationPacerConvergence(deliveries []qualificationFlowDelivery, start time.Time, flows []string, targetBytesPerSecond int64) time.Duration {
consecutive := 0
for second := time.Duration(0); second < 10*time.Second; second += time.Second {