Enabled support for multiple collection workloads

Author: daniel-percona

README.md

@@ -99,6 +99,26 @@ Quick access:
 * Includes `executionStats` vs `queryPlanner` guidance, severity-based explain eligibility, filtering counters, and expected outcomes
 * Includes query traceability fields so each finding can be mapped to a concrete query source/definition
 * Includes concise findings cards + Explore drawer workflow with execution-plan stage flow and technical details tabs
+* Includes mixed sharded/non-sharded runtime support with per-collection auto detection and explicit sharding strategy controls in the UI
+
+### 6. Workload Lifecycle & Timing Semantics
+
+PLGM now exposes an explicit lifecycle so users can distinguish preparation from measured workload execution:
+
+* `initializing`: collection setup, index creation, sharding/runtime preparation, optional seeding
+* `running`: actual query workload execution
+* `completed` / `failed`: terminal states
+
+Important timing behavior:
+
+* Initialization time is tracked separately and is **not** treated as execution runtime
+* Execution timer and progress bar start from the execution phase
+* During initialization, the UI shows progress context (step name, progress %, and recent activity events)
+
+Current limitation:
+
+* Initialization progress is structured and user-visible, but still coarse-grained (step/event based)
+* It is not yet per-index/per-shard real-time streaming
 
 ## The Interactive UI
 
@@ -681,6 +701,24 @@ If you point to a folder, plgm will scan and merge **all** `.json` files found i
 export PLGM_COLLECTIONS_PATH="./resources/custom_collections/"
 ```
 
+#### Single-File Multi-Definition Format
+When using a single file, plgm supports both legacy array format and wrapped multi-definition format:
+
+- Collections:
+  - `[{...}]`
+  - `{"collections":[...]}`
+- Queries:
+  - `[{...}]`
+  - `{"queries":[...]}`
+
+For multi-collection workloads, each query must identify its target collection (and database when needed). plgm validates query-to-collection mapping before execution and fails early on unknown or ambiguous references.
+Using the same collection name in different databases is supported (for example `db1.orders` and `db2.orders`).
+
+Example files:
+
+- [`resources/collections/multi_collections.json`](./resources/collections/multi_collections.json)
+- [`resources/queries/multi_queries.json`](./resources/queries/multi_queries.json)
+
 #### Default Workload Filtering
 When using **Directory Mode**, the behavior depends on the `PLGM_DEFAULT_WORKLOAD` setting:
 
@@ -734,6 +772,9 @@ You can override any setting in `config.yaml` using environment variables. This
 | `insights_explain_workers` | `PLGM_INSIGHTS_EXPLAIN_WORKERS` | Post-run explain worker count | `1` |
 | `insights_explain_retries` | `PLGM_INSIGHTS_EXPLAIN_RETRIES` | Retry count for explain timeout/failure | `1` |
 | `insights_explain_backoff_ms` | `PLGM_INSIGHTS_EXPLAIN_BACKOFF_MS` | Backoff between explain retries (ms) | `150` |
+| **Sharding Strategy** | | | |
+| `sharding_mode` | `PLGM_SHARDING_MODE` | Per-collection sharding behavior: `auto`, `force_on`, `force_off` | `auto` |
+| `sharding_skip_generic_without_config` | `PLGM_SHARDING_SKIP_GENERIC_WITHOUT_CONFIG` | Skip generic sharding setup for collections without `shardConfig` (recommended for mixed workloads) | `true` |
 | **Workload Control** | | | |
 | `concurrency` | `PLGM_CONCURRENCY` | Number of active worker goroutines | `50` |
 | `duration` | `PLGM_DURATION` | Test duration (Go duration string) | `5m`, `60s` |
@@ -789,7 +830,7 @@ When executed, plgm performs the following steps:
     * Spawns the configured number of **Active Workers**.
     * Continuously generates and executes queries (Find, Insert, Update, Delete, Aggregate, Upsert) based on your configured ratios.
     * Generates realistic BSON data for Inserts and Updates (supports recursion and complex schemas).
-    * Workers pick a random collection from the provided list for every operation.
+    * Workers pick a random collection from the provided list for every operation, then route query execution using queries bound to that same database+collection.
 4.  **Reporting:**
     * Outputs a real-time status report every N seconds (configurable).
     * Prints a detailed summary table at the end of the run.

TESTING.md

@@ -22,6 +22,8 @@ Integration tests are behind the `integration` build tag and are skipped from de
 Current integration smoke test:
 
 - `internal/mongo/TestRunWorkloadIntegration_OneShotExecutesFindUpdateAndSkipsInsert`
+- `internal/mongo/TestRunWorkloadIntegration_OneShotMultiDBMixedShardConfig`
+- `internal/mongo/TestRunWorkloadIntegration_MongosMixedShardedAndUnshardedCollections` (mongos-only, gated)
 
 ### Option A: Start MongoDB with Docker Compose (recommended)
 
@@ -37,6 +39,12 @@ Then run the integration smoke test:
 go test -tags=integration ./internal/mongo -run TestRunWorkloadIntegration_OneShotExecutesFindUpdateAndSkipsInsert -v
 ```
 
+Run all integration tests available for your endpoint:
+
+```bash
+go test -tags=integration ./internal/mongo -v
+```
+
 Stop MongoDB when done:
 
 ```bash
@@ -54,14 +62,44 @@ go test -tags=integration ./internal/mongo -run TestRunWorkloadIntegration_OneSh
 
 If your MongoDB requires authentication, include credentials in `PLGM_IT_MONGO_URI`.
 
+### Mongos Mixed Sharding Integration Test (CI/profiled env)
+
+This test validates a real mongos path with one sharded and one unsharded collection in the same run:
+
+- `TestRunWorkloadIntegration_MongosMixedShardedAndUnshardedCollections`
+
+It is intentionally gated and will skip unless explicitly enabled:
+
+```bash
+PLGM_IT_MONGO_URI='mongodb://<mongos-host>:27017' \
+PLGM_IT_ENABLE_MONGOS_SHARD_TEST=true \
+go test -tags=integration ./internal/mongo \
+  -run TestRunWorkloadIntegration_MongosMixedShardedAndUnshardedCollections \
+  -v -count=1
+```
+
+Recommended CI/profiled environment requirements:
+
+- URI must point to a real `mongos` router
+- test user must have privileges for:
+  - `listShards`
+  - `enableSharding`
+  - `shardCollection`
+  - normal read/write on test databases
+- ideally run in a dedicated test cluster/profile to avoid interference with production workloads
+
 ## Run Everything Available
 
 ```bash
 # 1) Unit tests
 go test ./...
 
 # 2) Integration smoke test(s)
-go test -tags=integration ./internal/mongo -run TestRunWorkloadIntegration_OneShotExecutesFindUpdateAndSkipsInsert -v
+go test -tags=integration ./internal/mongo -v
+
+# 3) Optional: mongos-only mixed sharding coverage
+PLGM_IT_ENABLE_MONGOS_SHARD_TEST=true \
+go test -tags=integration ./internal/mongo -run TestRunWorkloadIntegration_MongosMixedShardedAndUnshardedCollections -v -count=1
 ```
 
 ## Insights Feature Test Focus (Unit)
@@ -74,3 +112,15 @@ go test ./internal/webui -run Insights -v
 ```
 
 These cover filtering/severity behavior, `/api/insights` lifecycle, and export payload parity with insights output.
+
+## Workload Lifecycle Observability
+
+`/api/stats` now includes explicit lifecycle state and structured initialization progress fields:
+
+- `lifecyclePhase` (`initializing`, `running`, `completed`, `failed`)
+- `lifecycleStep`, `lifecycleStepIndex`, `lifecycleStepTotal`
+- `lifecycleStepDone`, `lifecycleStepWork`, `lifecycleStepProgressPct`
+- `lifecycleRecentEvents` (bounded recent activity list)
+- `initializationDurationSec` and `executionElapsedSec`
+
+This enables UI/automation to distinguish preparation time from measured execution time.

cmd/plgm/main.go

@@ -269,18 +269,18 @@ func main() {
 		log.Fatal(err)
 	}
 
-	validCollections := make(map[string]bool)
-	for _, col := range collectionsCfg.Collections {
-		validCollections[col.Name] = true
+	if err := config.ValidateCollectionDefinitions(collectionsCfg.Collections); err != nil {
+		log.Fatal(err)
 	}
 
-	var filteredQueries []config.QueryDefinition
-	for _, q := range queriesCfg.Queries {
-		if validCollections[q.Collection] {
-			filteredQueries = append(filteredQueries, q)
-		}
+	if err := config.NormalizeAndValidateQueries(queriesCfg.Queries); err != nil {
+		log.Fatal(err)
 	}
-	queriesCfg.Queries = filteredQueries
+	boundQueries, err := config.ValidateAndBindQueriesToCollections(queriesCfg.Queries, collectionsCfg.Collections)
+	if err != nil {
+		log.Fatal(err)
+	}
+	queriesCfg.Queries = boundQueries
 
 	dbName := collectionsCfg.Collections[0].DatabaseName
 	stats.PrintConfiguration(appCfg, collectionsCfg.Collections, version)
@@ -291,27 +291,41 @@ func main() {
 	}
 	defer conn.Disconnect(ctx)
 
+	initStart := time.Now()
+	log.Printf("[Lifecycle] Phase=initializing: preparing collections, indexes, sharding setup, and optional seed data")
+	log.Printf("[Lifecycle] Init Step 1/3: preparing collections")
 	if err := mongo.CreateCollectionsFromConfig(ctx, conn.Database, collectionsCfg, appCfg.DropCollections); err != nil {
 		log.Fatal(err)
 	}
+	totalIndexes := 0
+	for _, col := range collectionsCfg.Collections {
+		totalIndexes += len(col.Indexes)
+	}
+	log.Printf("[Lifecycle] Init Step 2/3: creating indexes (declared indexes=%d)", totalIndexes)
 	if err := mongo.CreateIndexesFromConfig(ctx, conn.Database, collectionsCfg); err != nil {
 		log.Fatal(err)
 	}
 
+	log.Printf("[Lifecycle] Init Step 3/3: optional seed (skip_seed=%v, documents_count=%d)", appCfg.SkipSeed, appCfg.DocumentsCount)
 	if !appCfg.SkipSeed && appCfg.DocumentsCount > 0 {
 		for _, col := range collectionsCfg.Collections {
 			if err := mongo.InsertRandomDocuments(ctx, conn.Database, col, appCfg.DocumentsCount, appCfg); err != nil {
 				log.Fatal(err)
 			}
 		}
 	}
+	initDuration := time.Since(initStart)
+	log.Printf("[Lifecycle] Initialization completed in %s", initDuration.Round(10*time.Millisecond))
+	log.Printf("[Lifecycle] Phase=running: execution timers now track query workload only")
 
 	intervalDuration, _ := time.ParseDuration(appCfg.IntervalDelay)
 	for i := 1; i <= appCfg.Iterations; i++ {
+		iterStart := time.Now()
 		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)
 		}
+		log.Printf("[Lifecycle] Iteration %d execution duration: %s", i, time.Since(iterStart).Round(10*time.Millisecond))
 		if i < appCfg.Iterations && intervalDuration > 0 {
 			log.Printf("Waiting %s before next iteration...", appCfg.IntervalDelay)
 			time.Sleep(intervalDuration)

config.yaml

@@ -79,6 +79,20 @@ insights_explain_workers: 1
 insights_explain_retries: 1
 insights_explain_backoff_ms: 150
 
+# ==============================================================================
+# Sharding Strategy
+# ==============================================================================
+# Controls how generic sharding setup is applied per collection.
+# Modes:
+# - auto: detect per collection (recommended)
+# - force_on: require mongos and apply sharding setup
+# - force_off: never apply sharding setup
+sharding_mode: "auto"
+
+# When true, collections without explicit shardConfig are skipped by generic sharding setup.
+# Recommended for mixed workloads where some collections are intentionally unsharded.
+sharding_skip_generic_without_config: true
+
 # ==============================================================================
 # Workload Definition
 # ==============================================================================

internal/config/collections.go

@@ -108,12 +108,8 @@ func LoadCollections(path string, loadDefault bool) (*CollectionsFile, error) {
 		allCollections = append(allCollections, loaded.Collections...)
 	}
 
-	// --- VALIDATION ---
-	// Ensure we didn't load an empty config due to JSON key mismatch
-	for i, col := range allCollections {
-		if col.DatabaseName == "" || col.Name == "" {
-			return nil, fmt.Errorf("loaded collection at index %d has empty 'database' or 'collection' name. Check your JSON keys: must be 'database' and 'collection' (lowercase)", i)
-		}
+	if err := ValidateCollectionDefinitions(allCollections); err != nil {
+		return nil, err
 	}
 
 	return &CollectionsFile{Collections: allCollections}, nil
@@ -149,3 +145,60 @@ func parseCollectionsBytes(b []byte) (*CollectionsFile, error) {
 
 	return nil, fmt.Errorf("invalid collections format")
 }
+
+func ParseCollectionsBytes(b []byte) (*CollectionsFile, error) {
+	return parseCollectionsBytes(b)
+}
+
+func ValidateCollectionDefinitions(cols []CollectionDefinition) error {
+	seenNamespaces := make(map[string]int, len(cols))
+
+	for i, col := range cols {
+		if strings.TrimSpace(col.DatabaseName) == "" || strings.TrimSpace(col.Name) == "" {
+			return fmt.Errorf("loaded collection at index %d has empty 'database' or 'collection' name. Check your JSON keys: must be 'database' and 'collection' (lowercase)", i)
+		}
+
+		nsKey := strings.ToLower(strings.TrimSpace(col.DatabaseName)) + "." + strings.ToLower(strings.TrimSpace(col.Name))
+		if prev, ok := seenNamespaces[nsKey]; ok {
+			return fmt.Errorf("duplicate namespace %q at index %d (already declared at index %d)", col.DatabaseName+"."+col.Name, i, prev)
+		}
+		seenNamespaces[nsKey] = i
+
+		for idxPos, idx := range col.Indexes {
+			if len(idx.Keys) == 0 {
+				return fmt.Errorf("collection %q has invalid index at position %d: index keys cannot be empty", col.Name, idxPos)
+			}
+		}
+		if err := validateFieldConstraints(col.Fields, ""); err != nil {
+			return fmt.Errorf("collection %q field validation failed: %w", col.DatabaseName+"."+col.Name, err)
+		}
+	}
+
+	return nil
+}
+
+func validateFieldConstraints(fields map[string]CollectionField, prefix string) error {
+	for name, field := range fields {
+		path := name
+		if prefix != "" {
+			path = prefix + "." + name
+		}
+		if field.Min != nil && field.Max != nil && *field.Min > *field.Max {
+			return fmt.Errorf("field %q has invalid min/max: min (%d) is greater than max (%d)", path, *field.Min, *field.Max)
+		}
+		if field.MinLength > 0 && field.MaxLength > 0 && field.MinLength > field.MaxLength {
+			return fmt.Errorf("field %q has invalid minLength/maxLength: minLength (%d) is greater than maxLength (%d)", path, field.MinLength, field.MaxLength)
+		}
+		if len(field.Fields) > 0 {
+			if err := validateFieldConstraints(field.Fields, path); err != nil {
+				return err
+			}
+		}
+		if field.Items != nil {
+			if err := validateFieldConstraints(map[string]CollectionField{"[]": *field.Items}, path); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}