chore: apply spammoor fix

Author: chatton

apps/loadgen/internal/runner/runner.go

@@ -15,8 +15,9 @@ import (
 )
 
 const (
-	metricSentTotal   = "spamoor_transactions_sent_total"
-	metricFailedTotal = "spamoor_transactions_failed_total"
+	metricSentTotal      = "spamoor_transactions_sent_total"
+	metricFailedTotal    = "spamoor_transaction_failures_total"
+	metricSpammerRunning = "spamoor_spammer_running"
 )
 
 type matrixOpts struct {
@@ -201,41 +202,69 @@ func waitForSpamoorDoneWithInterval(ctx context.Context, api spamoor.Client, tar
 
 	start := time.Now()
 	var prevSent float64
+	var sawRunning bool
 
 	for {
 		select {
 		case <-ctx.Done():
 			return sent, failed, fmt.Errorf("timed out waiting for %d txs (sent %.0f): %w", targetCount, sent, ctx.Err())
 		case <-ticker.C:
-			currentSent, currentFailed, mErr := getSpamoorCountersForPrefix(api, namePrefix)
+			metrics, mErr := api.GetMetrics()
 			if mErr != nil {
 				log.Printf("warning: failed to get metrics: %v", mErr)
 				continue
 			}
 
-			sent = currentSent - baselineSent
-			failed = currentFailed - baselineFailed
+			sent = sumCounterWithPrefix(metrics[metricSentTotal], namePrefix) - baselineSent
+			failed = sumCounterWithPrefix(metrics[metricFailedTotal], namePrefix) - baselineFailed
 			if sent < 0 {
 				sent = 0
 			}
 			if failed < 0 {
 				failed = 0
 			}
 
+			running := countRunningWithPrefix(metrics[metricSpammerRunning], namePrefix)
+			if running > 0 {
+				sawRunning = true
+			}
+
 			delta := sent - prevSent
 			rate := delta / pollInterval.Seconds()
 			elapsed := time.Since(start)
-			log.Printf("  progress: %.0f/%d sent (%.0f tx/s instant, %.1f tx/s avg, %.0f failed) [%s]",
-				sent, targetCount, rate, sent/elapsed.Seconds(), failed, elapsed.Round(time.Second))
+			log.Printf("  progress: %.0f/%d sent (%.0f tx/s instant, %.1f tx/s avg, %.0f failed, %d running) [%s]",
+				sent, targetCount, rate, sent/elapsed.Seconds(), failed, running, elapsed.Round(time.Second))
 			prevSent = sent
 
-			if sent >= float64(targetCount) {
+			// spammers report running=0 once the scenario fully completes,
+			// which includes flushing pending txs. the sent>=target fallback
+			// covers scenarios that finish before the gauge is first observed.
+			if running == 0 && (sawRunning || sent >= float64(targetCount)) {
 				return sent, failed, nil
 			}
 		}
 	}
 }
 
+func countRunningWithPrefix(f *dto.MetricFamily, namePrefix string) int {
+	if f == nil || f.GetType() != dto.MetricType_GAUGE {
+		return 0
+	}
+	var n int
+	for _, m := range f.GetMetric() {
+		if m.GetGauge() == nil {
+			continue
+		}
+		if namePrefix != "" && !hasLabelPrefix(m, "spammer_name", namePrefix) {
+			continue
+		}
+		if m.GetGauge().GetValue() > 0 {
+			n++
+		}
+	}
+	return n
+}
+
 func getSpamoorCountersForPrefix(api spamoor.Client, namePrefix string) (sent, failed float64, err error) {
 	metrics, err := api.GetMetrics()
 	if err != nil {

apps/loadgen/internal/runner/runner_test.go

@@ -82,8 +82,8 @@ func TestRunEntryFailsWhenSpammerDoesNotStart(t *testing.T) {
 func TestWaitForSpamoorDoneUsesDeltas(t *testing.T) {
 	client := &fakeClient{
 		metricsSeq: []metricSnapshot{
-			{sent: 105, failed: 3},
-			{sent: 108, failed: 4},
+			{sent: 105, failed: 3, running: 1},
+			{sent: 108, failed: 4, running: 0},
 		},
 		spammerNames: []string{"bench-test-0"},
 	}
@@ -94,6 +94,22 @@ func TestWaitForSpamoorDoneUsesDeltas(t *testing.T) {
 	require.Equal(t, 2.0, failed)
 }
 
+func TestWaitForSpamoorDoneWaitsForSpammersToStop(t *testing.T) {
+	client := &fakeClient{
+		metricsSeq: []metricSnapshot{
+			{sent: 110, failed: 0, running: 1},
+			{sent: 110, failed: 0, running: 1},
+			{sent: 110, failed: 0, running: 0},
+		},
+		spammerNames: []string{"bench-test-0"},
+	}
+
+	sent, _, err := waitForSpamoorDoneWithInterval(context.Background(), client, 10, 100, 0, "", time.Millisecond)
+	require.NoError(t, err)
+	require.Equal(t, 10.0, sent)
+	require.Equal(t, 3, client.metricsIndex)
+}
+
 func TestWaitForSpamoorDoneHonorsContext(t *testing.T) {
 	client := &fakeClient{
 		metricsSeq: []metricSnapshot{
@@ -167,8 +183,9 @@ func TestSumCounterWithPrefixFiltersCorrectly(t *testing.T) {
 }
 
 type metricSnapshot struct {
-	sent   float64
-	failed float64
+	sent    float64
+	failed  float64
+	running float64
 }
 
 type fakeClient struct {
@@ -234,16 +251,38 @@ func (f *fakeClient) GetMetrics() (map[string]*dto.MetricFamily, error) {
 	perFailed := snapshot.failed / float64(len(f.spammerNames))
 	sentMap := make(map[string]float64, len(f.spammerNames))
 	failedMap := make(map[string]float64, len(f.spammerNames))
+	runningMap := make(map[string]float64, len(f.spammerNames))
 	for _, name := range f.spammerNames {
 		sentMap[name] = perSpammer
 		failedMap[name] = perFailed
+		runningMap[name] = snapshot.running
 	}
 	return map[string]*dto.MetricFamily{
-		"spamoor_transactions_sent_total":   labeledCounterFamily("spamoor_transactions_sent_total", sentMap),
-		"spamoor_transactions_failed_total": labeledCounterFamily("spamoor_transactions_failed_total", failedMap),
+		"spamoor_transactions_sent_total":    labeledCounterFamily("spamoor_transactions_sent_total", sentMap),
+		"spamoor_transaction_failures_total": labeledCounterFamily("spamoor_transaction_failures_total", failedMap),
+		"spamoor_spammer_running":            labeledGaugeFamily("spamoor_spammer_running", runningMap),
 	}, nil
 }
 
+func labeledGaugeFamily(name string, spammerValues map[string]float64) *dto.MetricFamily {
+	gaugeType := dto.MetricType_GAUGE
+	labelName := "spammer_name"
+	var metrics []*dto.Metric
+	for spammerName, value := range spammerValues {
+		metrics = append(metrics, &dto.Metric{
+			Label: []*dto.LabelPair{
+				{Name: &labelName, Value: &spammerName},
+			},
+			Gauge: &dto.Gauge{Value: &value},
+		})
+	}
+	return &dto.MetricFamily{
+		Name:   &name,
+		Type:   &gaugeType,
+		Metric: metrics,
+	}
+}
+
 func (f *fakeClient) GetClients() ([]spamoorapi.Client, error) {
 	if f.getClientsErr != nil {
 		return nil, f.getClientsErr