Added cummulative latency dashboard and ability to configure workload iterations

Author: daniel-percona

README.md

@@ -244,6 +244,8 @@ Environment Variables (Overrides):
   PLGM_COLLECTIONS_PATH               Path to collection JSON
   PLGM_QUERIES_PATH                   Path to query JSON
   PLGM_DURATION                       Test duration (e.g. 60s, 5m)
+  PLGM_ITERATIONS                     Number of times to repeat the workload
+  PLGM_INTERVAL_DELAY                 Time to pause between iterations (e.g. 5s, 1m)
   PLGM_CONCURRENCY                    Number of active workers
   PLGM_DOCUMENTS_COUNT                Initial seed document count
   PLGM_DROP_COLLECTIONS               Drop collections on start (true/false)
@@ -642,6 +644,8 @@ You can override any setting in `config.yaml` using environment variables. This
 | **Workload Control** | | | |
 | `concurrency` | `PLGM_CONCURRENCY` | Number of active worker goroutines | `50` |
 | `duration` | `PLGM_DURATION` | Test duration (Go duration string) | `5m`, `60s` |
+| `iterations` | `PLGM_ITERATIONS` | Number of times to repeat the workload | `1` |
+| `interval_delay` | `PLGM_INTERVAL_DELAY` | Time to pause between iterations (e.g. 5s, 1m) | `0s` |
 | `default_workload` | `PLGM_DEFAULT_WORKLOAD` | Use built-in "Flights" workload (`true`/`false`) | `false` |
 | `collections_path` | `PLGM_COLLECTIONS_PATH` | Path to custom collection JSON files (supports directories for multi-collection load) | `./schemas` |
 | `queries_path` | `PLGM_QUERIES_PATH` | Path to custom query JSON files or directory. | `./queries` |
@@ -908,6 +912,18 @@ Control how plgm reacts to network lag or database pressure.
     * *Tip:* For stress testing, you might want to set `retry_attempts: 0` to see raw failure rates immediately.
     * *Default:* `2` attempts with `5ms` backoff.
 
+
+#### Workload Iterations & Scheduling
+Control how plgm repeats a given workload and schedules the time between runs.
+
+* **`iterations`**: The number of times to run the defined workload duration back-to-back.
+    * *Tip:* This is perfect for warming up the database cache during the first iteration and capturing true performance metrics on subsequent iterations, without having to manually restart the tool.
+    * *Default:* `1`
+* **`interval_delay`**: The amount of time to pause the workload between each iteration (e.g., `10s`, `1m`).
+    * *Tip:* Adding a delay allows database background processes (like log flushing, checkpointing, or compaction) to catch up, simulating real-world batch-processing patterns. 
+    * *Default:* `0s`
+
+
 ### 3. Custom Connection Parameters (`custom_params`)
 
 In the `config.yaml`, the `custom_params` section allows you to pass arbitrary options directly to the MongoDB driver's connection string. These parameters are appended as URI query options and are critical for tuning network throughput, security, and routing. All standard MongoDB connection parameters are supported.

cmd/plgm/main.go

@@ -10,6 +10,7 @@ import (
 	"net/url"
 	"os"
 	"strings"
+	"time"
 
 	"github.com/Percona-Lab/percona-load-generator-mongodb/internal/benchmark"
 	"github.com/Percona-Lab/percona-load-generator-mongodb/internal/config"
@@ -66,6 +67,8 @@ func main() {
 		fmt.Fprintf(os.Stderr, "  %-35s %s\n", "PLGM_COLLECTIONS_PATH", "Path to collection JSON")
 		fmt.Fprintf(os.Stderr, "  %-35s %s\n", "PLGM_QUERIES_PATH", "Path to query JSON")
 		fmt.Fprintf(os.Stderr, "  %-35s %s\n", "PLGM_DURATION", "Test duration (e.g. 60s, 5m)")
+		fmt.Fprintf(os.Stderr, "  %-35s %s\n", "PLGM_ITERATIONS", "Number of times to repeat the workload")
+		fmt.Fprintf(os.Stderr, "  %-35s %s\n", "PLGM_INTERVAL_DELAY", "Time to pause between iterations (e.g. 5s, 1m)")
 		fmt.Fprintf(os.Stderr, "  %-35s %s\n", "PLGM_CONCURRENCY", "Number of active workers")
 		fmt.Fprintf(os.Stderr, "  %-35s %s\n", "PLGM_DOCUMENTS_COUNT", "Initial seed document count")
 		fmt.Fprintf(os.Stderr, "  %-35s %s\n", "PLGM_DROP_COLLECTIONS", "Drop collections on start (true/false)")
@@ -235,8 +238,16 @@ func main() {
 		}
 		defer benchConn.Disconnect(ctx)
 
-		if err := benchmark.RunRawInjector(ctx, benchConn.Database, appCfg); err != nil {
-			log.Fatal(err)
+		intervalDuration, _ := time.ParseDuration(appCfg.IntervalDelay)
+		for i := 1; i <= appCfg.Iterations; i++ {
+			log.Printf("Starting Raw Injector iteration %d of %d", i, appCfg.Iterations)
+			if err := benchmark.RunRawInjector(ctx, benchConn.Database, appCfg); err != nil {
+				log.Fatal(err)
+			}
+			if i < appCfg.Iterations && intervalDuration > 0 {
+				log.Printf("Waiting %s before next iteration...", appCfg.IntervalDelay)
+				time.Sleep(intervalDuration)
+			}
 		}
 		return
 	}
@@ -295,7 +306,15 @@ func main() {
 		}
 	}
 
-	if err := mongo.RunWorkload(ctx, conn.Database, collectionsCfg.Collections, queriesCfg.Queries, appCfg); err != nil {
-		log.Fatal(err)
+	intervalDuration, _ := time.ParseDuration(appCfg.IntervalDelay)
+	for i := 1; i <= appCfg.Iterations; i++ {
+		log.Printf("Starting Standard Workload iteration %d of %d", i, appCfg.Iterations)
+		if err := mongo.RunWorkload(ctx, conn.Database, collectionsCfg.Collections, queriesCfg.Queries, appCfg); err != nil {
+			log.Fatal(err)
+		}
+		if i < appCfg.Iterations && intervalDuration > 0 {
+			log.Printf("Waiting %s before next iteration...", appCfg.IntervalDelay)
+			time.Sleep(intervalDuration)
+		}
 	}
 }

config.yaml

@@ -92,6 +92,12 @@ concurrency: 4
 # Set to "0" to run the queries exactly once and exit.
 duration: "10s"
 
+# Number of times to repeat the workload execution.
+iterations: 1
+
+# Time to pause between iterations (e.g. "5s", "1m").
+interval_delay: "0s"
+
 # Enable transactional workload
 use_transactions: false
 

images/dashboard_bottom.png

@@ -18,6 +18,8 @@ type AppConfig struct {
 	SkipSeed        bool   `yaml:"skip_seed"`
 	DocumentsCount  int    `yaml:"documents_count"`
 	Concurrency     int    `yaml:"concurrency"`
+	Iterations      int    `yaml:"iterations"`
+	IntervalDelay   string `yaml:"interval_delay"`
 
 	Duration           string `yaml:"duration"`
 	FindPercent        int    `yaml:"find_percent"`
@@ -132,6 +134,8 @@ func applyUIDefaults(cfg *AppConfig) {
 	cfg.DefaultWorkload = true
 	cfg.SkipSeed = true
 	cfg.DocumentsCount = 100000
+	cfg.Iterations = 1
+	cfg.IntervalDelay = "0s"
 
 	// --- MIX TAB ---
 	cfg.FindPercent = 50
@@ -336,6 +340,15 @@ func applyEnvOverrides(cfg *AppConfig) map[string]bool {
 		cfg.Duration = envDuration
 	}
 
+	if envIterations := os.Getenv("PLGM_ITERATIONS"); envIterations != "" {
+		if n, err := strconv.Atoi(envIterations); err == nil && n > 0 {
+			cfg.Iterations = n
+		}
+	}
+	if envInterval := os.Getenv("PLGM_INTERVAL_DELAY"); envInterval != "" {
+		cfg.IntervalDelay = envInterval
+	}
+
 	// Percentages
 	if p := os.Getenv("PLGM_FIND_PERCENT"); p != "" {
 		if n, err := strconv.Atoi(p); err == nil && n >= 0 {

images/dashboard_top.png

@@ -101,6 +101,41 @@ func (h *LatencyHistogram) GetPercentile(p float64) float64 {
 	return float64(MaxLatencyBin)
 }
 
+func (h *LatencyHistogram) GetStats() map[string]float64 {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+
+	if h.Count == 0 {
+		return map[string]float64{"avg": 0, "min": 0, "max": 0, "p95": 0, "p99": 0}
+	}
+
+	min := h.Min
+	if min == math.MaxFloat64 {
+		min = 0 // Sanity check to prevent rendering giant float limits
+	}
+
+	// Inline percentile calculation to avoid deadlocking
+	getPerc := func(p float64) float64 {
+		targetCount := int64(math.Ceil((p / 100.0) * float64(h.Count)))
+		var currentCount int64 = 0
+		for i, count := range h.Buckets {
+			currentCount += count
+			if currentCount >= targetCount {
+				return float64(i)
+			}
+		}
+		return float64(MaxLatencyBin)
+	}
+
+	return map[string]float64{
+		"avg": h.Sum / float64(h.Count),
+		"min": min,
+		"max": h.Max,
+		"p95": getPerc(95.0),
+		"p99": getPerc(99.0),
+	}
+}
+
 type Collector struct {
 	FindOps   uint64
 	InsertOps uint64
@@ -117,6 +152,9 @@ type Collector struct {
 	DeleteHist *LatencyHistogram
 	AggHist    *LatencyHistogram
 	TransHist  *LatencyHistogram
+	TotalHist  *LatencyHistogram
+
+	CurrentIteration int
 
 	startTime  time.Time
 	prevFind   uint64
@@ -137,12 +175,14 @@ func NewCollector() *Collector {
 		DeleteHist: &LatencyHistogram{Min: math.MaxFloat64},
 		AggHist:    &LatencyHistogram{Min: math.MaxFloat64},
 		TransHist:  &LatencyHistogram{Min: math.MaxFloat64},
+		TotalHist:  &LatencyHistogram{Min: math.MaxFloat64},
 		startTime:  time.Now(),
 	}
 }
 
 func (c *Collector) Track(opType string, duration time.Duration) {
 	ms := float64(duration.Nanoseconds()) / 1e6
+	c.TotalHist.Record(ms)
 	switch opType {
 	case "find":
 		atomic.AddUint64(&c.FindOps, 1)
@@ -170,6 +210,7 @@ func (c *Collector) Track(opType string, duration time.Duration) {
 
 func (c *Collector) Add(opType string, count int64, duration time.Duration) {
 	ms := float64(duration.Nanoseconds()) / 1e6
+	c.TotalHist.RecordBatch(ms, count)
 	switch opType {
 	case "find":
 		atomic.AddUint64(&c.FindOps, uint64(count))
@@ -236,7 +277,7 @@ func (c *Collector) Monitor(done <-chan struct{}, refreshRateSec int, concurrenc
 				csvWriter.Write([]string{
 					"Timestamp", "ElapsedSec",
 					"Select_OpsSec", "Insert_OpsSec", "Upsert_OpsSec",
-					"Update_OpsSec", "Delete_OpsSec", "Agg_OpsSec", "Trans_OpsSec",
+					"Update_OpsSec", "Delete_OpsSec", "Agg_OpsSec", "Trans_OpsSec", "Iteration",
 				})
 				csvWriter.Flush()
 			}
@@ -289,6 +330,11 @@ func (c *Collector) Monitor(done <-chan struct{}, refreshRateSec int, concurrenc
 				rateAgg := float64(currentAgg-lastAgg) / float64(refreshRateSec)
 				rateTrans := float64(currentTrans-lastTrans) / float64(refreshRateSec)
 
+				iter := c.CurrentIteration
+				if iter < 1 {
+					iter = 1
+				}
+
 				csvWriter.Write([]string{
 					time.Now().Format(time.RFC3339),
 					fmt.Sprintf("%.0f", elapsed),
@@ -299,8 +345,9 @@ func (c *Collector) Monitor(done <-chan struct{}, refreshRateSec int, concurrenc
 					fmt.Sprintf("%.2f", rateDelete),
 					fmt.Sprintf("%.2f", rateAgg),
 					fmt.Sprintf("%.2f", rateTrans),
+					strconv.Itoa(iter),
 				})
-				csvWriter.Flush() // Flush immediately so data is saved even if crashed
+				csvWriter.Flush()
 
 				// Update 'last' trackers for the next tick
 				lastFind = currentFind