fix(loadgen): filter spamoor counters by spammer name prefix

Counter-based progress tracking summed all spammer metrics globally,
causing overshoot when concurrent workloads (e.g. baseline + burst)
ran on the same spamoor instance. Filter by spammer_name label prefix
so each workload only tracks its own spammers.

Author: chatton

apps/loadgen/internal/runner.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"log"
 	"math/rand/v2"
+	"strings"
 	"time"
 
 	"github.com/celestiaorg/tastora/framework/docker/evstack/spamoor"
@@ -142,7 +143,8 @@ func runCheck(ctx context.Context, api SpamoorClient, timeout time.Duration) err
 		return fmt.Errorf("waiting for spamoor sync: %w", err)
 	}
 
-	baselineSent, baselineFailed, err := getSpamoorCounters(api)
+	checkPrefix := "check-"
+	baselineSent, baselineFailed, err := getSpamoorCountersForPrefix(api, checkPrefix)
 	if err != nil {
 		return fmt.Errorf("get baseline metrics: %w", err)
 	}
@@ -173,7 +175,7 @@ func runCheck(ctx context.Context, api SpamoorClient, timeout time.Duration) err
 	checkCtx, cancel := context.WithTimeout(ctx, timeout)
 	defer cancel()
 
-	sent, failed, err := waitForSpamoorDone(checkCtx, api, 1, baselineSent, baselineFailed)
+	sent, failed, err := waitForSpamoorDone(checkCtx, api, 1, baselineSent, baselineFailed, checkPrefix)
 	if err != nil {
 		return fmt.Errorf("tx not confirmed: %w", err)
 	}
@@ -186,7 +188,7 @@ func runCheck(ctx context.Context, api SpamoorClient, timeout time.Duration) err
 	return nil
 }
 
-type waitForDoneFunc func(context.Context, SpamoorClient, int, float64, float64) (float64, float64, error)
+type waitForDoneFunc func(ctx context.Context, api SpamoorClient, targetCount int, baselineSent, baselineFailed float64, namePrefix string) (float64, float64, error)
 
 func runEntry(ctx context.Context, api SpamoorClient, entry Entry) error {
 	return runEntryWithWait(ctx, api, entry, waitForSpamoorDone)
@@ -198,7 +200,9 @@ func runEntryWithWait(ctx context.Context, api SpamoorClient, entry Entry, wait
 	log.Printf("[%s] scenario=%s spammers=%d count_per=%d total=%d",
 		entry.TestName, entry.Scenario, entry.NumSpammers, entry.CountPerSpammer, totalCount)
 
-	baselineSent, baselineFailed, err := getSpamoorCounters(api)
+	namePrefix := spammerPrefix(entry.TestName)
+
+	baselineSent, baselineFailed, err := getSpamoorCountersForPrefix(api, namePrefix)
 	if err != nil {
 		return fmt.Errorf("get baseline metrics: %w", err)
 	}
@@ -207,7 +211,7 @@ func runEntryWithWait(ctx context.Context, api SpamoorClient, entry Entry, wait
 
 	var spammerIDs []int
 	for i := range entry.NumSpammers {
-		name := fmt.Sprintf("bench-%s-%d", entry.TestName, i)
+		name := fmt.Sprintf("%s%d", namePrefix, i)
 		id, err := api.CreateSpammer(name, entry.Scenario, scenarioCfg, true)
 		if err != nil {
 			return fmt.Errorf("create spammer %s: %w", name, err)
@@ -234,7 +238,7 @@ func runEntryWithWait(ctx context.Context, api SpamoorClient, entry Entry, wait
 	}
 
 	start := time.Now()
-	sent, failed, err := wait(ctx, api, totalCount, baselineSent, baselineFailed)
+	sent, failed, err := wait(ctx, api, totalCount, baselineSent, baselineFailed, namePrefix)
 	elapsed := time.Since(start)
 
 	if err != nil {
@@ -263,11 +267,11 @@ func DeleteAllSpammers(api SpamoorClient) error {
 	return nil
 }
 
-func waitForSpamoorDone(ctx context.Context, api SpamoorClient, targetCount int, baselineSent, baselineFailed float64) (sent, failed float64, err error) {
-	return waitForSpamoorDoneWithInterval(ctx, api, targetCount, baselineSent, baselineFailed, 2*time.Second)
+func waitForSpamoorDone(ctx context.Context, api SpamoorClient, targetCount int, baselineSent, baselineFailed float64, namePrefix string) (sent, failed float64, err error) {
+	return waitForSpamoorDoneWithInterval(ctx, api, targetCount, baselineSent, baselineFailed, namePrefix, 2*time.Second)
 }
 
-func waitForSpamoorDoneWithInterval(ctx context.Context, api SpamoorClient, targetCount int, baselineSent, baselineFailed float64, pollInterval time.Duration) (sent, failed float64, err error) {
+func waitForSpamoorDoneWithInterval(ctx context.Context, api SpamoorClient, targetCount int, baselineSent, baselineFailed float64, namePrefix string, pollInterval time.Duration) (sent, failed float64, err error) {
 	ticker := time.NewTicker(pollInterval)
 	defer ticker.Stop()
 
@@ -279,7 +283,7 @@ func waitForSpamoorDoneWithInterval(ctx context.Context, api SpamoorClient, targ
 		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 := getSpamoorCounters(api)
+			currentSent, currentFailed, mErr := getSpamoorCountersForPrefix(api, namePrefix)
 			if mErr != nil {
 				log.Printf("warning: failed to get metrics: %v", mErr)
 				continue
@@ -308,13 +312,13 @@ func waitForSpamoorDoneWithInterval(ctx context.Context, api SpamoorClient, targ
 	}
 }
 
-func getSpamoorCounters(api SpamoorClient) (sent, failed float64, err error) {
+func getSpamoorCountersForPrefix(api SpamoorClient, namePrefix string) (sent, failed float64, err error) {
 	metrics, err := api.GetMetrics()
 	if err != nil {
 		return 0, 0, err
 	}
-	return sumCounter(metrics["spamoor_transactions_sent_total"]),
-		sumCounter(metrics["spamoor_transactions_failed_total"]),
+	return sumCounterWithPrefix(metrics["spamoor_transactions_sent_total"], namePrefix),
+		sumCounterWithPrefix(metrics["spamoor_transactions_failed_total"], namePrefix),
 		nil
 }
 
@@ -371,15 +375,32 @@ func getSpamoorHeight(api SpamoorClient) (uint64, error) {
 	return clients[0].Height, nil
 }
 
-func sumCounter(f *dto.MetricFamily) float64 {
+func sumCounterWithPrefix(f *dto.MetricFamily, namePrefix string) float64 {
 	if f == nil || f.GetType() != dto.MetricType_COUNTER {
 		return 0
 	}
 	var sum float64
 	for _, m := range f.GetMetric() {
-		if m.GetCounter() != nil && m.GetCounter().Value != nil {
-			sum += m.GetCounter().GetValue()
+		if m.GetCounter() == nil || m.GetCounter().Value == nil {
+			continue
+		}
+		if namePrefix != "" && !hasLabelPrefix(m, "spammer_name", namePrefix) {
+			continue
 		}
+		sum += m.GetCounter().GetValue()
 	}
 	return sum
 }
+
+func spammerPrefix(testName string) string {
+	return fmt.Sprintf("bench-%s-", testName)
+}
+
+func hasLabelPrefix(m *dto.Metric, name, prefix string) bool {
+	for _, lp := range m.GetLabel() {
+		if lp.GetName() == name {
+			return strings.HasPrefix(lp.GetValue(), prefix)
+		}
+	}
+	return false
+}

apps/loadgen/internal/runner_test.go

@@ -23,6 +23,7 @@ func TestRunEntryUsesBaselineCounters(t *testing.T) {
 		metricsSeq: []metricSnapshot{
 			{sent: 100, failed: 7},
 		},
+		spammerNames: []string{"bench-baseline-0", "bench-baseline-1"},
 	}
 
 	var gotTarget int
@@ -35,7 +36,7 @@ func TestRunEntryUsesBaselineCounters(t *testing.T) {
 		Env:             map[string]string{"BENCH_COUNT_PER_SPAMMER": "5"},
 		NumSpammers:     2,
 		CountPerSpammer: 5,
-	}, func(ctx context.Context, api SpamoorClient, targetCount int, baselineSent, baselineFailed float64) (float64, float64, error) {
+	}, func(ctx context.Context, api SpamoorClient, targetCount int, baselineSent, baselineFailed float64, namePrefix string) (float64, float64, error) {
 		gotTarget = targetCount
 		gotBaselineSent = baselineSent
 		gotBaselineFailed = baselineFailed
@@ -58,6 +59,7 @@ func TestRunEntryFailsWhenSpammerDoesNotStart(t *testing.T) {
 		metricsSeq: []metricSnapshot{
 			{sent: 0, failed: 0},
 		},
+		spammerNames: []string{"bench-fail-0"},
 	}
 
 	err := runEntryWithWait(context.Background(), client, Entry{
@@ -66,7 +68,7 @@ func TestRunEntryFailsWhenSpammerDoesNotStart(t *testing.T) {
 		Env:             map[string]string{"BENCH_COUNT_PER_SPAMMER": "1"},
 		NumSpammers:     1,
 		CountPerSpammer: 1,
-	}, func(ctx context.Context, api SpamoorClient, targetCount int, baselineSent, baselineFailed float64) (float64, float64, error) {
+	}, func(ctx context.Context, api SpamoorClient, targetCount int, baselineSent, baselineFailed float64, namePrefix string) (float64, float64, error) {
 		t.Fatal("wait function should not be called when spammer startup fails")
 		return 0, 0, nil
 	})
@@ -80,9 +82,10 @@ func TestWaitForSpamoorDoneUsesDeltas(t *testing.T) {
 			{sent: 105, failed: 3},
 			{sent: 108, failed: 4},
 		},
+		spammerNames: []string{"bench-test-0"},
 	}
 
-	sent, failed, err := waitForSpamoorDoneWithInterval(context.Background(), client, 8, 100, 2, time.Millisecond)
+	sent, failed, err := waitForSpamoorDoneWithInterval(context.Background(), client, 8, 100, 2, "", time.Millisecond)
 	require.NoError(t, err)
 	require.Equal(t, 8.0, sent)
 	require.Equal(t, 2.0, failed)
@@ -93,12 +96,13 @@ func TestWaitForSpamoorDoneHonorsContext(t *testing.T) {
 		metricsSeq: []metricSnapshot{
 			{sent: 100, failed: 0},
 		},
+		spammerNames: []string{"bench-test-0"},
 	}
 
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Millisecond)
 	defer cancel()
 
-	_, _, err := waitForSpamoorDoneWithInterval(ctx, client, 2, 100, 0, 10*time.Millisecond)
+	_, _, err := waitForSpamoorDoneWithInterval(ctx, client, 2, 100, 0, "", 10*time.Millisecond)
 	require.ErrorContains(t, err, "timed out waiting for 2 txs")
 }
 
@@ -126,6 +130,41 @@ func TestWaitForSyncHonorsContext(t *testing.T) {
 	require.ErrorContains(t, err, "cancelled waiting for sync")
 }
 
+func TestSumCounterWithPrefixFiltersCorrectly(t *testing.T) {
+	family := labeledCounterFamily("spamoor_transactions_sent_total", map[string]float64{
+		"bench-EOATransferBurst-0": 1000,
+		"bench-EOATransferBurst-1": 1000,
+		"bench-EOATransfer-0":      500,
+		"bench-EOATransfer-1":      500,
+	})
+
+	t.Run("no prefix sums all", func(t *testing.T) {
+		total := sumCounterWithPrefix(family, "")
+		require.Equal(t, 3000.0, total)
+	})
+
+	t.Run("burst prefix sums only burst spammers", func(t *testing.T) {
+		total := sumCounterWithPrefix(family, "bench-EOATransferBurst-")
+		require.Equal(t, 2000.0, total)
+	})
+
+	t.Run("baseline prefix does not match burst spammers", func(t *testing.T) {
+		// "bench-EOATransfer-" does not match "bench-EOATransferBurst-"
+		// because the dash after "Transfer" differs from "B" in "Burst".
+		total := sumCounterWithPrefix(family, "bench-EOATransfer-")
+		require.Equal(t, 1000.0, total)
+	})
+
+	t.Run("nil family returns zero", func(t *testing.T) {
+		require.Equal(t, 0.0, sumCounterWithPrefix(nil, "anything"))
+	})
+
+	t.Run("unmatched prefix returns zero", func(t *testing.T) {
+		total := sumCounterWithPrefix(family, "bench-Nonexistent-")
+		require.Equal(t, 0.0, total)
+	})
+}
+
 type metricSnapshot struct {
 	sent   float64
 	failed float64
@@ -141,6 +180,7 @@ type fakeSpamoorClient struct {
 	clientsSeq     [][]spamoor.Client
 	clientsIndex   int
 	getClientsErr  error
+	spammerNames []string
 }
 
 type createCall struct {
@@ -189,9 +229,17 @@ func (f *fakeSpamoorClient) GetMetrics() (map[string]*dto.MetricFamily, error) {
 		}
 	}
 
+	perSpammer := snapshot.sent / float64(len(f.spammerNames))
+	perFailed := snapshot.failed / float64(len(f.spammerNames))
+	sentMap := make(map[string]float64, len(f.spammerNames))
+	failedMap := make(map[string]float64, len(f.spammerNames))
+	for _, name := range f.spammerNames {
+		sentMap[name] = perSpammer
+		failedMap[name] = perFailed
+	}
 	return map[string]*dto.MetricFamily{
-		"spamoor_transactions_sent_total":   counterFamily("spamoor_transactions_sent_total", snapshot.sent),
-		"spamoor_transactions_failed_total": counterFamily("spamoor_transactions_failed_total", snapshot.failed),
+		"spamoor_transactions_sent_total":   labeledCounterFamily("spamoor_transactions_sent_total", sentMap),
+		"spamoor_transactions_failed_total": labeledCounterFamily("spamoor_transactions_failed_total", failedMap),
 	}, nil
 }
 
@@ -210,17 +258,23 @@ func (f *fakeSpamoorClient) GetClients() ([]spamoor.Client, error) {
 	return clients, nil
 }
 
-func counterFamily(name string, value float64) *dto.MetricFamily {
+func labeledCounterFamily(name string, spammerValues map[string]float64) *dto.MetricFamily {
 	counterType := dto.MetricType_COUNTER
-	return &dto.MetricFamily{
-		Name: &name,
-		Type: &counterType,
-		Metric: []*dto.Metric{
-			{
-				Counter: &dto.Counter{
-					Value: &value,
-				},
+	labelName := "spammer_name"
+	var metrics []*dto.Metric
+	for spammerName, value := range spammerValues {
+		n := spammerName
+		v := value
+		metrics = append(metrics, &dto.Metric{
+			Label: []*dto.LabelPair{
+				{Name: &labelName, Value: &n},
 			},
-		},
+			Counter: &dto.Counter{Value: &v},
+		})
+	}
+	return &dto.MetricFamily{
+		Name:   &name,
+		Type:   &counterType,
+		Metric: metrics,
 	}
 }