[SDTEST-3832] Add target time parallelism setting

README.md

@@ -99,6 +99,7 @@ parallelism details, see [Running DDTest](docs/running.md).
 | `--command` | Override the default test command used by Ruby frameworks. For pytest, use `PYTEST_ADDOPTS` for pytest flags. |
 | `--min-parallelism` | Minimum CI node or worker count DDTest considers when planning. |
 | `--max-parallelism` | Maximum CI node or worker count DDTest considers when planning. |
+| `--target-time` | Target wall time DDTest tries to satisfy when selecting parallelism. |
 | `--ci-node` | Run only the files assigned to CI node **N**. |
 | `--tests-location` | Override the default test file discovery glob. |
 | `--tests-exclude-pattern` | Exclude matching test files from discovery. |

docs/running.md

@@ -133,3 +133,10 @@ decrease it to prefer faster wall time. Use duration values such as `25s`, `1m`,
 or `1500ms`; set `0s` to disable this overhead bias. When scores tie, DDTest
 prefers fewer CI nodes or workers, then lower wall time, then lower imbalance
 between workers.
+
+Set `--target-time` to make DDTest first choose among splits whose expected
+wall time is at or below that target. Use the same duration format, such as
+`10m`, `300s`, or `1500ms`; the default `0s` disables the target. If no split
+within `--min-parallelism` and `--max-parallelism` can meet the target, DDTest
+logs a warning and selects the split with the lowest expected wall time,
+ignoring CI job overhead, to get as close as possible to the target.

docs/settings.md

@@ -11,6 +11,7 @@ CLI flags take precedence over environment variables.
 | `--min-parallelism` | `DD_TEST_OPTIMIZATION_RUNNER_MIN_PARALLELISM` | | physical CPU count | Minimum count DDTest considers when planning. Interpret it as CI nodes in CI-node mode, or workers in a single-node run. |
 | `--max-parallelism` | `DD_TEST_OPTIMIZATION_RUNNER_MAX_PARALLELISM` | | physical CPU count | Maximum count DDTest considers when planning. Interpret it as CI nodes in CI-node mode, or workers in a single-node run. |
 | `--ci-job-overhead` | `DD_TEST_OPTIMIZATION_RUNNER_CI_JOB_OVERHEAD` | | `25s` | Modeled overhead for adding one more CI node. Accepts durations such as `25s`, `1m`, `1500ms`, or `0s` to disable this bias. Increase it to use fewer CI nodes; decrease it to prefer faster wall time. |
+| `--target-time` | `DD_TEST_OPTIMIZATION_RUNNER_TARGET_TIME` | | `0s` | Target wall time for the selected split. Accepts durations such as `10m`, `300s`, `1500ms`, or `0s` to disable the target. DDTest first considers splits at or below this wall time; if none are possible within the min/max parallelism range, it warns and selects the split with the lowest expected wall time, ignoring CI job overhead, to get as close as possible to the target. |
 | `--ci-node` | `DD_TEST_OPTIMIZATION_RUNNER_CI_NODE` | | `-1` (off) | Restrict this run to files assigned to CI node **N** (0-indexed). |
 | `--ci-node-workers` | `DD_TEST_OPTIMIZATION_RUNNER_CI_NODE_WORKERS` | | `1` | Number of workers to start on this CI node. Use a positive integer, or `ncpu` to use the node's available physical CPU cores. |
 | `--worker-env` | `DD_TEST_OPTIMIZATION_RUNNER_WORKER_ENV` | | `""` | Template env vars per worker: `--worker-env "DATABASE_NAME_TEST=app_test{{nodeIndex}}_{{workerIndex}}"`. `{{nodeIndex}}` is the CI node index (`0` for single-node runs); `{{workerIndex}}` is the worker process index within that CI node. |

internal/cmd/cmd.go

@@ -64,6 +64,7 @@ var rootPersistentFlagBindings = []persistentFlagBinding{
 	{configKey: "min_parallelism", flagName: "min-parallelism"},
 	{configKey: "max_parallelism", flagName: "max-parallelism"},
 	{configKey: "parallel_runner_overhead", flagName: "ci-job-overhead"},
+	{configKey: "target_time", flagName: "target-time"},
 	{configKey: "worker_env", flagName: "worker-env"},
 	{configKey: "ci_node", flagName: "ci-node"},
 	{configKey: "ci_node_workers", flagName: "ci-node-workers"},
@@ -85,6 +86,7 @@ func init() {
 	rootCmd.PersistentFlags().Int("min-parallelism", defaultParallelism, "Minimum number of parallel test processes (default: number of physical CPUs)")
 	rootCmd.PersistentFlags().Int("max-parallelism", defaultParallelism, "Maximum number of parallel test processes (default: number of physical CPUs)")
 	rootCmd.PersistentFlags().String("ci-job-overhead", settings.DefaultParallelRunnerOverhead().String(), "Modeled overhead for adding one more CI job / parallel runner (for example, 25s, 1m, 1500ms, or 0s to disable the bias). Increase it to use fewer CI jobs; decrease it to prefer faster wall time")
+	rootCmd.PersistentFlags().String("target-time", settings.DefaultTargetTime().String(), "Target wall time for selected CI job / parallel runner split (for example, 10m, 300s, 1500ms, or 0s to disable the target)")
 	rootCmd.PersistentFlags().String("worker-env", "", "Worker environment configuration")
 	rootCmd.PersistentFlags().Int("ci-node", -1, "CI node index to run (0-indexed; default: -1 disables CI-node mode)")
 	rootCmd.PersistentFlags().String("ci-node-workers", "1", `Number of parallel workers per CI node (positive integer or "ncpu"; default: 1)`)

internal/cmd/cmd_test.go

@@ -92,6 +92,12 @@ func TestRootCommandFlags(t *testing.T) {
 		return
 	}
 
+	targetTimeFlag := rootCmd.PersistentFlags().Lookup("target-time")
+	if targetTimeFlag == nil {
+		t.Error("target-time flag should be defined")
+		return
+	}
+
 	// Check default values
 	if platformFlag.DefValue != "ruby" {
 		t.Errorf("expected platform default to be 'ruby', got %q", platformFlag.DefValue)
@@ -141,6 +147,11 @@ func TestRootCommandFlags(t *testing.T) {
 	if parallelRunnerOverheadFlag.DefValue != expectedParallelRunnerOverhead {
 		t.Errorf("expected ci-job-overhead default to be %q, got %q", expectedParallelRunnerOverhead, parallelRunnerOverheadFlag.DefValue)
 	}
+
+	expectedTargetTime := settings.DefaultTargetTime().String()
+	if targetTimeFlag.DefValue != expectedTargetTime {
+		t.Errorf("expected target-time default to be %q, got %q", expectedTargetTime, targetTimeFlag.DefValue)
+	}
 }
 
 func TestCommandHierarchy(t *testing.T) {
@@ -380,6 +391,9 @@ func TestFlagBinding(t *testing.T) {
 	if err := rootCmd.PersistentFlags().Set("ci-job-overhead", "30s"); err != nil {
 		t.Fatalf("Error setting ci-job-overhead flag: %v", err)
 	}
+	if err := rootCmd.PersistentFlags().Set("target-time", "10m"); err != nil {
+		t.Fatalf("Error setting target-time flag: %v", err)
+	}
 
 	// Check that viper picks up the flag values
 	if viper.GetString("platform") != "python" {
@@ -418,6 +432,9 @@ func TestFlagBinding(t *testing.T) {
 	if viper.GetString("parallel_runner_overhead") != "30s" {
 		t.Errorf("expected viper parallel_runner_overhead to be '30s', got %q", viper.GetString("parallel_runner_overhead"))
 	}
+	if viper.GetString("target_time") != "10m" {
+		t.Errorf("expected viper target_time to be '10m', got %q", viper.GetString("target_time"))
+	}
 }
 
 func TestBindPersistentFlags(t *testing.T) {

internal/planner/parallelism.go

@@ -7,15 +7,23 @@ import (
 
 // calculateParallelRunnerSplit determines the selected runner split by
 // estimating candidates between the configured min and max parallelism.
-func calculateParallelRunnerSplit(testFileWeights map[string]int, minParallelism, maxParallelism int, parallelRunnerOverhead time.Duration) splitScore {
+func calculateParallelRunnerSplit(testFileWeights map[string]int, minParallelism, maxParallelism int, parallelRunnerOverhead, targetTime time.Duration) splitScore {
+	return calculateParallelRunnerSplitSelection(testFileWeights, minParallelism, maxParallelism, parallelRunnerOverhead, targetTime).selected
+}
+
+func calculateParallelRunnerSplitSelection(testFileWeights map[string]int, minParallelism, maxParallelism int, parallelRunnerOverhead, targetTime time.Duration) splitSelection {
 	files := sortedWeightedTestFiles(testFileWeights)
-	selector := splitSelector{parallelRunnerOverhead: parallelRunnerOverhead}
+	selector := splitSelector{
+		parallelRunnerOverhead: parallelRunnerOverhead,
+		targetTime:             targetTime,
+	}
 
 	// maxParallelism could be 0 or negative!
 	if maxParallelism <= 1 {
 		score := scoreSortedWeightedRunnerSplit(files, 1)
 		selector.logCandidate(score)
-		return score
+		selector.maybeWarnTargetTimeUnreachable(score, 1, 1)
+		return selector.selection(score, score, []splitScore{score})
 	}
 
 	if minParallelism < 1 {
@@ -32,22 +40,43 @@ func calculateParallelRunnerSplit(testFileWeights map[string]int, minParallelism
 	if len(files) == 0 {
 		score := scoreSortedWeightedRunnerSplit(files, minParallelism)
 		selector.logCandidate(score)
-		return score
+		selector.maybeWarnTargetTimeUnreachable(score, minParallelism, maxParallelism)
+		return selector.selection(score, score, []splitScore{score})
 	}
 
 	candidateMax := maxUsefulParallelism(minParallelism, maxParallelism, len(files))
 
-	best := scoreSortedWeightedRunnerSplit(files, minParallelism)
-	selector.logCandidate(best)
+	bestWithoutTarget := scoreSortedWeightedRunnerSplit(files, minParallelism)
+	lowestWallTime := bestWithoutTarget
+	bestWithinTarget := bestWithoutTarget
+	targetBestFound := selector.meetsTargetTime(bestWithoutTarget)
+	candidates := []splitScore{bestWithoutTarget}
+	selector.logCandidate(bestWithoutTarget)
 	for parallelRunners := minParallelism + 1; parallelRunners <= candidateMax; parallelRunners++ {
 		score := scoreSortedWeightedRunnerSplit(files, parallelRunners)
+		candidates = append(candidates, score)
 		selector.logCandidate(score)
-		if selector.better(score, best) {
-			best = score
+		if selector.better(score, bestWithoutTarget) {
+			bestWithoutTarget = score
+		}
+		if selector.betterWallTime(score, lowestWallTime) {
+			lowestWallTime = score
 		}
+		if selector.meetsTargetTime(score) && (!targetBestFound || selector.better(score, bestWithinTarget)) {
+			bestWithinTarget = score
+			targetBestFound = true
+		}
+	}
+
+	if targetBestFound {
+		return selector.selection(bestWithinTarget, bestWithoutTarget, candidates)
+	}
+	if selector.targetTime <= 0 {
+		return selector.selection(bestWithoutTarget, bestWithoutTarget, candidates)
 	}
 
-	return best
+	selector.maybeWarnTargetTimeUnreachable(lowestWallTime, minParallelism, maxParallelism)
+	return selector.selection(lowestWallTime, bestWithoutTarget, candidates)
 }
 
 func maxUsefulParallelism(minParallelism, maxParallelism, filesCount int) int {
@@ -62,6 +91,16 @@ func maxUsefulParallelism(minParallelism, maxParallelism, filesCount int) int {
 
 type splitSelector struct {
 	parallelRunnerOverhead time.Duration
+	targetTime             time.Duration
+}
+
+type splitSelection struct {
+	selected               splitScore
+	bestWithoutTarget      splitScore
+	candidates             []splitScore
+	parallelRunnerOverhead time.Duration
+	targetTime             time.Duration
+	available              bool
 }
 
 func (s splitSelector) better(candidate, currentBest splitScore) bool {
@@ -79,6 +118,16 @@ func (s splitSelector) better(candidate, currentBest splitScore) bool {
 	return candidate.imbalance < currentBest.imbalance
 }
 
+func (s splitSelector) betterWallTime(candidate, currentBest splitScore) bool {
+	if candidate.wallTime != currentBest.wallTime {
+		return candidate.wallTime < currentBest.wallTime
+	}
+	if candidate.parallelRunners != currentBest.parallelRunners {
+		return candidate.parallelRunners < currentBest.parallelRunners
+	}
+	return candidate.imbalance < currentBest.imbalance
+}
+
 // selectionScore models each candidate as wallTime + runners * overhead. When
 // scores tie, the selector intentionally prefers fewer runners before comparing
 // wall time and imbalance.
@@ -93,12 +142,42 @@ func (s splitSelector) parallelRunnerOverheadMillis() int {
 	return int(s.parallelRunnerOverhead / time.Millisecond)
 }
 
+func (s splitSelector) selection(selected, bestWithoutTarget splitScore, candidates []splitScore) splitSelection {
+	return splitSelection{
+		selected:               selected,
+		bestWithoutTarget:      bestWithoutTarget,
+		candidates:             candidates,
+		parallelRunnerOverhead: s.parallelRunnerOverhead,
+		targetTime:             s.targetTime,
+		available:              true,
+	}
+}
+
+func (s splitSelector) meetsTargetTime(score splitScore) bool {
+	return s.targetTime > 0 && score.wallTimeDuration() <= s.targetTime
+}
+
+func (s splitSelector) maybeWarnTargetTimeUnreachable(best splitScore, minParallelism, maxParallelism int) {
+	if s.targetTime <= 0 || s.meetsTargetTime(best) {
+		return
+	}
+
+	slog.Warn("No parallel runner split meets target time; selecting split with lowest expected wall time",
+		"targetTime", s.targetTime,
+		"minParallelism", minParallelism,
+		"maxParallelism", maxParallelism,
+		"selectedParallelRunners", best.parallelRunners,
+		"selectedExpectedWallTime", best.wallTimeDuration())
+}
+
 func (s splitSelector) logCandidate(score splitScore) {
 	slog.Debug("Considered parallel runner split",
 		"parallelRunners", score.parallelRunners,
 		"expectedWallTime", score.wallTimeDuration(),
 		"imbalance", score.imbalanceDuration(),
 		"expectedTotalRuntime", score.totalRuntimeDuration(),
 		"parallelRunnerOverhead", s.parallelRunnerOverhead,
+		"targetTime", s.targetTime,
+		"meetsTargetTime", s.meetsTargetTime(score),
 		"selectionScore", time.Duration(s.selectionScore(score))*time.Millisecond)
 }

internal/planner/parallelism_test.go

@@ -14,7 +14,7 @@ import (
 const testParallelRunnerOverhead = 25 * time.Second
 
 func testCalculateParallelRunners(testFileWeights map[string]int, minParallelism, maxParallelism int) int {
-	return calculateParallelRunnerSplit(testFileWeights, minParallelism, maxParallelism, testParallelRunnerOverhead).parallelRunners
+	return calculateParallelRunnerSplit(testFileWeights, minParallelism, maxParallelism, testParallelRunnerOverhead, 0).parallelRunners
 }
 
 func TestCalculateParallelRunners_MaxParallelismIsOne(t *testing.T) {
@@ -144,7 +144,7 @@ func TestCalculateParallelRunnerSplit_ReturnsSelectedScore(t *testing.T) {
 		"test5.rb": 6,
 	}
 
-	result := calculateParallelRunnerSplit(testFileWeights, 3, 4, testParallelRunnerOverhead)
+	result := calculateParallelRunnerSplit(testFileWeights, 3, 4, testParallelRunnerOverhead, 0)
 	expected := splitScore{
 		parallelRunners: 3,
 		wallTime:        12,
@@ -201,7 +201,7 @@ func TestCalculateParallelRunnerSplit_RealGitHubActionsArtifacts(t *testing.T) {
 				)
 			}
 
-			result := calculateParallelRunnerSplit(fixture.TestFileWeights, 1, 8, testParallelRunnerOverhead)
+			result := calculateParallelRunnerSplit(fixture.TestFileWeights, 1, 8, testParallelRunnerOverhead, 0)
 			if result.parallelRunners != tt.expectedParallelRunners {
 				t.Fatalf(
 					"calculateParallelRunnerSplit() = %d runners, expected %d; artifact selected %d before this fix",
@@ -236,14 +236,67 @@ func TestCalculateParallelRunnerSplit_ParallelRunnerOverheadTunesFanout(t *testi
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			result := calculateParallelRunnerSplit(fixture.TestFileWeights, 1, 8, tt.parallelRunnerOverhead)
+			result := calculateParallelRunnerSplit(fixture.TestFileWeights, 1, 8, tt.parallelRunnerOverhead, 0)
 			if result.parallelRunners != tt.expectedParallelRunners {
 				t.Errorf("calculateParallelRunnerSplit() = %d runners, expected %d", result.parallelRunners, tt.expectedParallelRunners)
 			}
 		})
 	}
 }
 
+func TestCalculateParallelRunnerSplit_TargetTimeFiltersCandidateSplits(t *testing.T) {
+	testFileWeights := map[string]int{
+		"test1.rb": 60_000,
+		"test2.rb": 60_000,
+		"test3.rb": 60_000,
+		"test4.rb": 60_000,
+	}
+	minParallelism := 1
+	maxParallelism := 4
+	parallelRunnerOverhead := 5 * time.Minute
+	targetTime := 2 * time.Minute
+
+	result := calculateParallelRunnerSplit(testFileWeights, minParallelism, maxParallelism, parallelRunnerOverhead, targetTime)
+	if result.parallelRunners != 2 {
+		t.Errorf("calculateParallelRunnerSplit() = %d runners, expected 2 to satisfy target time with the lowest selection score", result.parallelRunners)
+	}
+	if result.wallTimeDuration() > targetTime {
+		t.Errorf("calculateParallelRunnerSplit() wall time = %s, expected at or below %s", result.wallTimeDuration(), targetTime)
+	}
+}
+
+func TestCalculateParallelRunnerSplit_TargetTimeImpossibleWarnsAndSelectsLowestWallTime(t *testing.T) {
+	logs := captureLogs(t)
+	testFileWeights := map[string]int{
+		"test1.rb": 60_000,
+		"test2.rb": 60_000,
+		"test3.rb": 60_000,
+		"test4.rb": 60_000,
+	}
+	minParallelism := 1
+	maxParallelism := 4
+	parallelRunnerOverhead := 5 * time.Minute
+	targetTime := 59 * time.Second
+
+	result := calculateParallelRunnerSplitSelection(testFileWeights, minParallelism, maxParallelism, parallelRunnerOverhead, targetTime)
+	if result.selected.parallelRunners != 4 {
+		t.Errorf("calculateParallelRunnerSplit() = %d runners, expected 4 from fallback lowest wall time split", result.selected.parallelRunners)
+	}
+	if result.selected.wallTimeDuration() != time.Minute {
+		t.Errorf("calculateParallelRunnerSplit() wall time = %s, expected 1m0s", result.selected.wallTimeDuration())
+	}
+	if result.bestWithoutTarget.parallelRunners != 1 {
+		t.Errorf("calculateParallelRunnerSplit() without target = %d runners, expected 1 from best selection score", result.bestWithoutTarget.parallelRunners)
+	}
+
+	logOutput := logs.String()
+	if !strings.Contains(logOutput, "No parallel runner split meets target time") ||
+		!strings.Contains(logOutput, "targetTime=59s") ||
+		!strings.Contains(logOutput, "selectedExpectedWallTime=1m0s") {
+		t.Errorf("expected target-time warning with fallback details, got logs: %s", logOutput)
+	}
+}
+
 type splitSelectionFixture struct {
 	SkippablePercentage    float64                             `json:"skippablePercentage"`
 	CurrentParallelRunners int                                 `json:"currentParallelRunners"`
@@ -276,7 +329,7 @@ func TestCalculateParallelRunnerSplit_LogsCandidateSplits(t *testing.T) {
 		"test3.rb": 10,
 	}
 
-	_ = calculateParallelRunnerSplit(testFileWeights, 1, 3, testParallelRunnerOverhead)
+	_ = calculateParallelRunnerSplit(testFileWeights, 1, 3, testParallelRunnerOverhead, 0)
 
 	logOutput := logs.String()
 	if strings.Count(logOutput, "Considered parallel runner split") != 3 ||
@@ -298,6 +351,6 @@ func BenchmarkCalculateParallelRunners20000TestFiles(b *testing.B) {
 
 	b.ResetTimer()
 	for range b.N {
-		_ = calculateParallelRunnerSplit(testFileWeights, 1, 256, testParallelRunnerOverhead)
+		_ = calculateParallelRunnerSplit(testFileWeights, 1, 256, testParallelRunnerOverhead, 0)
 	}
 }

internal/planner/planner.go

@@ -209,12 +209,14 @@ func (tp *TestPlanner) Plan(ctx context.Context) error {
 		return fmt.Errorf("failed to write skippable percentage: %w", err)
 	}
 
-	parallelRunnerSplit := calculateParallelRunnerSplit(
+	parallelRunnerSelection := calculateParallelRunnerSplitSelection(
 		tp.testFileWeights,
 		settings.GetMinParallelism(),
 		settings.GetMaxParallelism(),
 		settings.GetParallelRunnerOverhead(),
+		settings.GetTargetTime(),
 	)
+	parallelRunnerSplit := parallelRunnerSelection.selected
 	parallelRunners := parallelRunnerSplit.parallelRunners
 	runnersContent := fmt.Sprintf("%d", parallelRunners)
 	if err := writePlanFile(constants.ParallelRunnersOutputPath, []byte(runnersContent)); err != nil {
@@ -237,7 +239,7 @@ func (tp *TestPlanner) Plan(ctx context.Context) error {
 	}
 
 	if settings.GetReportEnabled() {
-		printPlanReport(tp.reportWriter, tp, parallelRunnerSplit)
+		printPlanReport(tp.reportWriter, tp, parallelRunnerSelection)
 	}
 
 	tp.planLoaded = true

internal/planner/planner_test.go

@@ -749,9 +749,9 @@ func TestTestPlanner_Setup_WithParallelRunners(t *testing.T) {
 	report := reportOutput.String()
 	for _, expectedReportLine := range []string{
 		"Planning\n",
-		"  Runner split\n",
-		"    Estimated runtime:",
-		"    Runners: 1",
+		"  Split selection\n",
+		"    Estimated runtime with TIA:",
+		"    Selected: 1 runner",
 		"    Expected wall time:",
 		"    Imbalance:",
 	} {