diff --git a/loadtest/im-pressure/agent.go b/loadtest/im-pressure/agent.go index a31d0d3..712f28d 100644 --- a/loadtest/im-pressure/agent.go +++ b/loadtest/im-pressure/agent.go @@ -173,7 +173,8 @@ func runAgent(ctx context.Context, cfg *Config, manifest *Manifest, machine, coo } base := AgentSummary{ RunID: start.RunID, Scenario: start.Scenario, Machine: machine, StartedAt: startedAt, - LoadEndedAt: loadEndedAt, FinishedAt: time.Now().UTC(), ReadyConnections: len(run.clients), + LoadEndedAt: loadEndedAt, FinishedAt: time.Now().UTC(), ExpectedConnections: cfg.UsersPerMachine, + ReadyConnections: len(run.clients), PublishAttempted: run.attempted.Load(), PublishAccepted: int64(len(local)), PublishRejected: run.rejected.Load(), PublishErrors: run.publishErrorCounts(), AckLatency: run.ack.Snapshot(), AcceptedLocal: local, diff --git a/loadtest/im-pressure/merge.go b/loadtest/im-pressure/merge.go index 45b1dbd..ddab5e9 100644 --- a/loadtest/im-pressure/merge.go +++ b/loadtest/im-pressure/merge.go @@ -10,33 +10,34 @@ import ( ) type CombinedReport struct { - RunID string `json:"run_id"` - Scenario string `json:"scenario"` - GeneratedAt time.Time `json:"generated_at"` - Agents []AgentSummary `json:"agents"` - ReadyConnections int `json:"ready_connections"` - PublishAttempted int64 `json:"publish_attempted"` - PublishAccepted int64 `json:"publish_accepted"` - PublishRejected int64 `json:"publish_rejected"` - PublishErrors map[string]int64 `json:"publish_errors,omitempty"` - PublishSuccessRate float64 `json:"publish_success_rate"` - AckP95MS float64 `json:"ack_p95_ms"` - DeliveryP95MS float64 `json:"delivery_p95_ms"` - ExpectedDeliveries int64 `json:"expected_deliveries"` - ReceivedDeliveries int64 `json:"received_deliveries"` - RealtimeLost int64 `json:"realtime_lost"` - RealtimeLostStored int64 `json:"realtime_lost_but_persisted"` - ArrivalRate float64 `json:"arrival_rate"` - LossRate float64 `json:"loss_rate"` - Duplicates int64 `json:"duplicates"` - OutOfOrder int64 `json:"out_of_order"` - Disconnects int64 `json:"disconnects"` - HistoryChecked int64 `json:"history_checked"` - HistoryFound int64 `json:"history_found"` - PersistenceLost int64 `json:"persistence_lost"` - Missing []MissingDelivery `json:"missing,omitempty"` - Passed bool `json:"passed"` - FailureReasons []string `json:"failure_reasons,omitempty"` + RunID string `json:"run_id"` + Scenario string `json:"scenario"` + GeneratedAt time.Time `json:"generated_at"` + Agents []AgentSummary `json:"agents"` + ExpectedConnections int `json:"expected_connections"` + ReadyConnections int `json:"ready_connections"` + PublishAttempted int64 `json:"publish_attempted"` + PublishAccepted int64 `json:"publish_accepted"` + PublishRejected int64 `json:"publish_rejected"` + PublishErrors map[string]int64 `json:"publish_errors,omitempty"` + PublishSuccessRate float64 `json:"publish_success_rate"` + AckP95MS float64 `json:"ack_p95_ms"` + DeliveryP95MS float64 `json:"delivery_p95_ms"` + ExpectedDeliveries int64 `json:"expected_deliveries"` + ReceivedDeliveries int64 `json:"received_deliveries"` + RealtimeLost int64 `json:"realtime_lost"` + RealtimeLostStored int64 `json:"realtime_lost_but_persisted"` + ArrivalRate float64 `json:"arrival_rate"` + LossRate float64 `json:"loss_rate"` + Duplicates int64 `json:"duplicates"` + OutOfOrder int64 `json:"out_of_order"` + Disconnects int64 `json:"disconnects"` + HistoryChecked int64 `json:"history_checked"` + HistoryFound int64 `json:"history_found"` + PersistenceLost int64 `json:"persistence_lost"` + Missing []MissingDelivery `json:"missing,omitempty"` + Passed bool `json:"passed"` + FailureReasons []string `json:"failure_reasons,omitempty"` } func runMerge(aPath, bPath, output string) error { @@ -54,6 +55,11 @@ func runMerge(aPath, bPath, output string) error { report := CombinedReport{RunID: a.RunID, Scenario: a.Scenario, GeneratedAt: time.Now().UTC(), Agents: []AgentSummary{a, b}, Passed: true, PublishErrors: make(map[string]int64)} for _, summary := range report.Agents { + expectedConnections := summary.ExpectedConnections + if expectedConnections == 0 { + expectedConnections = 1000 + } + report.ExpectedConnections += expectedConnections report.ReadyConnections += summary.ReadyConnections report.PublishAttempted += summary.PublishAttempted report.PublishAccepted += summary.PublishAccepted @@ -85,8 +91,9 @@ func runMerge(aPath, bPath, output string) error { report.ArrivalRate = float64(report.ReceivedDeliveries) / float64(report.ExpectedDeliveries) report.LossRate = float64(report.RealtimeLost) / float64(report.ExpectedDeliveries) } - if report.ReadyConnections != 2000 { - report.FailureReasons = append(report.FailureReasons, fmt.Sprintf("ready connections=%d expected=2000", report.ReadyConnections)) + if report.ReadyConnections != report.ExpectedConnections { + report.FailureReasons = append(report.FailureReasons, fmt.Sprintf("ready connections=%d expected=%d", + report.ReadyConnections, report.ExpectedConnections)) } if report.PublishAttempted > 0 && report.PublishSuccessRate < 0.9999 { report.FailureReasons = append(report.FailureReasons, diff --git a/loadtest/im-pressure/merge_test.go b/loadtest/im-pressure/merge_test.go index 4eb588b..33c3864 100644 --- a/loadtest/im-pressure/merge_test.go +++ b/loadtest/im-pressure/merge_test.go @@ -10,10 +10,12 @@ import ( func TestRunMergeComputesArrivalAndPersistenceGate(t *testing.T) { dir := t.TempDir() a := AgentSummary{RunID: "run1", Scenario: "p2p", Machine: "a", ReadyConnections: 1000, - PublishAttempted: 10, PublishAccepted: 10, ExpectedPeer: 10, ReceivedPeer: 9, + ExpectedConnections: 1000, + PublishAttempted: 10, PublishAccepted: 10, ExpectedPeer: 10, ReceivedPeer: 9, RealtimeLost: 1, HistoryChecked: 10, HistoryFound: 10} b := AgentSummary{RunID: "run1", Scenario: "p2p", Machine: "b", ReadyConnections: 1000, - PublishAttempted: 10, PublishAccepted: 10, ExpectedPeer: 10, ReceivedPeer: 10, + ExpectedConnections: 1000, + PublishAttempted: 10, PublishAccepted: 10, ExpectedPeer: 10, ReceivedPeer: 10, HistoryChecked: 10, HistoryFound: 9, PersistenceLost: 1} aPath, bPath, reportPath := filepath.Join(dir, "a.json"), filepath.Join(dir, "b.json"), filepath.Join(dir, "report.json") writeSummaryForTest(t, aPath, a) diff --git a/loadtest/im-pressure/model.go b/loadtest/im-pressure/model.go index d822511..f1fbc06 100644 --- a/loadtest/im-pressure/model.go +++ b/loadtest/im-pressure/model.go @@ -343,38 +343,39 @@ type AcceptedMessage struct { } type AgentSummary struct { - RunID string `json:"run_id"` - Scenario string `json:"scenario"` - Machine string `json:"machine"` - StartedAt time.Time `json:"started_at"` - LoadEndedAt time.Time `json:"load_ended_at"` - FinishedAt time.Time `json:"finished_at"` - ReadyConnections int `json:"ready_connections"` - PublishAttempted int64 `json:"publish_attempted"` - PublishAccepted int64 `json:"publish_accepted"` - PublishRejected int64 `json:"publish_rejected"` - PublishErrors map[string]int64 `json:"publish_errors,omitempty"` - ExpectedPeer int64 `json:"expected_peer"` - ReceivedPeer int64 `json:"received_peer"` - ExpectedEcho int64 `json:"expected_echo"` - ReceivedEcho int64 `json:"received_echo"` - ExpectedGroup int64 `json:"expected_group"` - ReceivedGroup int64 `json:"received_group"` - RealtimeLost int64 `json:"realtime_lost"` - RealtimeLostStored int64 `json:"realtime_lost_but_persisted"` - Duplicates int64 `json:"duplicates"` - OutOfOrder int64 `json:"out_of_order"` - Disconnects int64 `json:"disconnects"` - UnexpectedMessages int64 `json:"unexpected_messages"` - HistoryChecked int64 `json:"history_checked"` - HistoryFound int64 `json:"history_found"` - PersistenceLost int64 `json:"persistence_lost"` - AckLatency HistogramSnapshot `json:"ack_latency"` - DeliveryLatency HistogramSnapshot `json:"delivery_latency"` - Errors []string `json:"errors,omitempty"` - Missing []MissingDelivery `json:"missing,omitempty"` - PersistenceMissing []string `json:"persistence_missing,omitempty"` - AcceptedLocal []AcceptedMessage `json:"accepted_local,omitempty"` + RunID string `json:"run_id"` + Scenario string `json:"scenario"` + Machine string `json:"machine"` + StartedAt time.Time `json:"started_at"` + LoadEndedAt time.Time `json:"load_ended_at"` + FinishedAt time.Time `json:"finished_at"` + ExpectedConnections int `json:"expected_connections"` + ReadyConnections int `json:"ready_connections"` + PublishAttempted int64 `json:"publish_attempted"` + PublishAccepted int64 `json:"publish_accepted"` + PublishRejected int64 `json:"publish_rejected"` + PublishErrors map[string]int64 `json:"publish_errors,omitempty"` + ExpectedPeer int64 `json:"expected_peer"` + ReceivedPeer int64 `json:"received_peer"` + ExpectedEcho int64 `json:"expected_echo"` + ReceivedEcho int64 `json:"received_echo"` + ExpectedGroup int64 `json:"expected_group"` + ReceivedGroup int64 `json:"received_group"` + RealtimeLost int64 `json:"realtime_lost"` + RealtimeLostStored int64 `json:"realtime_lost_but_persisted"` + Duplicates int64 `json:"duplicates"` + OutOfOrder int64 `json:"out_of_order"` + Disconnects int64 `json:"disconnects"` + UnexpectedMessages int64 `json:"unexpected_messages"` + HistoryChecked int64 `json:"history_checked"` + HistoryFound int64 `json:"history_found"` + PersistenceLost int64 `json:"persistence_lost"` + AckLatency HistogramSnapshot `json:"ack_latency"` + DeliveryLatency HistogramSnapshot `json:"delivery_latency"` + Errors []string `json:"errors,omitempty"` + Missing []MissingDelivery `json:"missing,omitempty"` + PersistenceMissing []string `json:"persistence_missing,omitempty"` + AcceptedLocal []AcceptedMessage `json:"accepted_local,omitempty"` } type MissingDelivery struct {