feat: add repetition functionality for experiments

experiment.yml

@@ -16,6 +16,7 @@ runtimes:
 experiments:
 
   http-readiness:
+    repetitions: 10
     workloads:
       default:
         image: docker.io/library/nginx:latest
@@ -79,6 +80,7 @@ experiments:
         runtime: urunc
 
   memory:
+    repetitions: 10
     workloads:
       default:
         image: docker.io/library/nginx:alpine

internal/manifest/types.go

@@ -11,7 +11,8 @@ type Runtime struct {
 }
 
 type Experiment struct {
-	Workloads Workloads `yaml:"workloads"`
+	Workloads   Workloads `yaml:"workloads"`
+	Repetitions int       `yaml:"repetitions,omitempty"`
 }
 
 type Workloads struct {

internal/orchestrator/orchestrator.go

@@ -139,20 +139,49 @@ func (o *Orchestrator) runTrial(
 		}...)
 	}
 
-	for _, stage := range stages {
-		// if the stage's experiment does not match the trial's experiment, skip it
-		if runtimeTC.Trial.ExperimentName != stage.experiment {
-			continue
-		}
-		stageResult, err := stage.fn(ctx, runtimeTC)
-		if err != nil {
+	var repetitions int
+	if trial.Repetitions > 0 {
+		repetitions = trial.Repetitions
+	} else {
+		repetitions = 1
+	}
+
+	for i := 0; i < repetitions; i++ {
+
+		for _, stage := range stages {
+			// if the stage's experiment does not match the trial's experiment, skip it
+			if runtimeTC.Trial.ExperimentName != stage.experiment {
+				continue
+			}
+			stageResult, err := stage.fn(ctx, runtimeTC)
+			if err != nil {
+				result.RuntimeStages = append(result.RuntimeStages, stageResult)
+				return failTrial(result, stage.name, err)
+			}
+
 			result.RuntimeStages = append(result.RuntimeStages, stageResult)
-			return failTrial(result, stage.name, err)
 		}
+	}
+
+	// get adapter for the trial's experiment to generate the result
+	var resultAdapter harnessruntime.Adapter
+	for _, adapter := range adapters {
+		if adapter.ExperimentName() == trial.ExperimentName {
+			resultAdapter = adapter
+			break
+		}
+	}
 
-		result.RuntimeStages = append(result.RuntimeStages, stageResult)
+	if resultAdapter != nil {
+		generatedResult, err := resultAdapter.GenerateResult(ctx, runtimeTC, result.RuntimeStages)
+		if err != nil {
+			return failTrial(result, "", fmt.Errorf("generate result: %w", err))
+		}
+		result.Results = generatedResult
 	}
 
+	result.RuntimeStages = nil // Clear runtime stages to save space in the final result
+
 	result.EndedAt = time.Now()
 	result.Duration = result.EndedAt.Sub(result.StartedAt)
 

internal/orchestrator/types.go

@@ -33,4 +33,5 @@ type TrialResult struct {
 	EndedAt       time.Time                    `json:"endedAt"`
 	Duration      time.Duration                `json:"duration"`
 	RuntimeStages []harnessruntime.StageResult `json:"runtimeStages"`
+	Results       any                          `json:"results"`
 }

internal/plan/generate.go

@@ -57,6 +57,7 @@ func Generate(m *manifest.Manifest) (*Plan, error) {
 				runtimeName,
 				rt.Handler,
 				defaultWorkload,
+				exp.Repetitions,
 			)
 
 			trials = append(trials, trial)
@@ -82,6 +83,7 @@ func Generate(m *manifest.Manifest) (*Plan, error) {
 					rt.Name,
 					rt.Handler,
 					workload,
+					exp.Repetitions,
 				)
 
 				trials = append(trials, trial)
@@ -97,6 +99,7 @@ func Generate(m *manifest.Manifest) (*Plan, error) {
 					runtimeName,
 					rt.Handler,
 					workload,
+					exp.Repetitions,
 				)
 
 				trials = append(trials, trial)
@@ -116,6 +119,7 @@ func buildTrial(
 	runtimeName string,
 	runtimeHandler string,
 	workload manifest.Workload,
+	repetitions int,
 ) Trial {
 	id := makeTrialID(experimentName, workloadName, runtimeName)
 
@@ -133,6 +137,7 @@ func buildTrial(
 		CPUMethod:      workload.CPUMethod,
 		Timeout:        workload.Timeout,
 		MetricsBrief:   workload.MetricsBrief,
+		Repetitions:    repetitions,
 	}
 }
 

internal/plan/types.go

@@ -21,4 +21,5 @@ type Trial struct {
 	CPUMethod      string            `json:"cpuMethod,omitempty"`
 	Timeout        string            `json:"timeout,omitempty"`
 	MetricsBrief   bool              `json:"metricsBrief,omitempty"`
+	Repetitions    int               `json:"repetitions,omitempty"`
 }

internal/runtime/adapter.go

@@ -41,4 +41,5 @@ type Adapter interface {
 	Stop(ctx context.Context, tc TrialContext) (StageResult, error)
 	DeleteTask(ctx context.Context, tc TrialContext) (StageResult, error)
 	Cleanup(ctx context.Context, tc TrialContext) (StageResult, error)
+	GenerateResult(ctx context.Context, tc TrialContext, result []StageResult) (any, error)
 }

internal/runtime/cpu/adapter.go

@@ -191,3 +191,13 @@ func noOpStage(ctx context.Context, stage harnessruntime.Stage, tc harnessruntim
 		Description: fmt.Sprintf("No-op for CLI CPU benchmark: trial=%s", tc.Trial.ID),
 	}, nil
 }
+
+func (a *Adapter) GenerateResult(ctx context.Context, tc harnessruntime.TrialContext, result []harnessruntime.StageResult) (any, error) {
+	// For this adapter, we can return the metrics collected during the StartTask stage.
+	for _, stageResult := range result {
+		if stageResult.Stage == harnessruntime.StageStart {
+			return stageResult.Data, nil
+		}
+	}
+	return nil, errors.New("no StartTask stage result found")
+}

internal/runtime/httpreadiness/adapter.go

@@ -15,6 +15,7 @@ import (
 	"time"
 
 	harnessruntime "github.com/urunc-dev/evaluation_suite/internal/runtime"
+	"github.com/urunc-dev/evaluation_suite/internal/utils"
 )
 
 const (
@@ -410,3 +411,36 @@ func stageResult(
 		Data:        data,
 	}
 }
+
+func (a *Adapter) GenerateResult(ctx context.Context, tc harnessruntime.TrialContext, results []harnessruntime.StageResult) (any, error) {
+	// Find the WaitReady stages result to extract the readiness latency. there can be multiple instances. so we need to find the interquartile range of the readiness latency and then find the mean of the interquartile range.
+
+	var readinessLatencies []float64
+	for _, result := range results {
+		if result.Stage == harnessruntime.StageWaitReady {
+			if data, ok := result.Data.(map[string]interface{}); ok {
+				if latency, ok := data["readiness_latency_ms"].(float64); ok {
+					readinessLatencies = append(readinessLatencies, latency)
+				}
+			}
+		}
+	}
+
+	if len(readinessLatencies) == 0 {
+		return nil, fmt.Errorf("no readiness latency data found in trial %s", tc.Trial.ID)
+	}
+
+	lower, upper := utils.InterquartileRange(readinessLatencies)
+	iqr := make([]float64, 0)
+	for _, latency := range readinessLatencies {
+		if latency >= lower && latency <= upper {
+			iqr = append(iqr, latency)
+		}
+	}
+
+	meanLatency := utils.Mean(iqr)
+
+	return map[string]interface{}{
+		"readiness_latency_mean_ms": meanLatency,
+	}, nil
+}

internal/runtime/lifecycle/adapter.go

@@ -479,3 +479,32 @@ func fakeStage(
 		),
 	}, nil
 }
+
+func (a *Adapter) GenerateResult(ctx context.Context, tc harnessruntime.TrialContext, results []harnessruntime.StageResult) (any, error) {
+	// For this adapter, we can return the first instances of the create, start, and delete stages. There should only be one of each stage per trial.
+
+	var createStage, startStage, deleteStage *harnessruntime.StageResult
+
+	for _, result := range results {
+		switch result.Stage {
+		case harnessruntime.StageCreate:
+			if createStage == nil {
+				createStage = &result
+			}
+		case harnessruntime.StageStart:
+			if startStage == nil {
+				startStage = &result
+			}
+		case harnessruntime.StageDelete:
+			if deleteStage == nil {
+				deleteStage = &result
+			}
+		}
+	}
+
+	return map[string]interface{}{
+		"create_stage": createStage.Data,
+		"start_stage":  startStage.Data,
+		"delete_stage": deleteStage.Data,
+	}, nil
+}

internal/runtime/memory/adapter.go

@@ -13,6 +13,7 @@ import (
 	"time"
 
 	harnessruntime "github.com/urunc-dev/evaluation_suite/internal/runtime"
+	"github.com/urunc-dev/evaluation_suite/internal/utils"
 )
 
 type commandRunner func(context.Context, ...string) ([]byte, error)
@@ -223,3 +224,78 @@ func stageResult(
 		Data:        data,
 	}
 }
+
+func (a *Adapter) GenerateResult(ctx context.Context, tc harnessruntime.TrialContext, results []harnessruntime.StageResult) (any, error) {
+	// For this adapter, we can return the metrics collected during the WaitReady stages. there can be multiple WaitReady stages, so we need the mean of cgroups and shims metrics.
+
+	var cgroupMetricsList []CgroupMetrics
+	var shimMetricsList []ShimMetrics
+
+	for _, result := range results {
+		if result.Stage == harnessruntime.StageWaitReady {
+			metrics, ok := result.Data.(Metrics)
+			if !ok {
+				return nil, fmt.Errorf("invalid data type for stage %s: expected Metrics, got %T", result.Stage, result.Data)
+			}
+			cgroupMetricsList = append(cgroupMetricsList, metrics.Cgroup)
+			shimMetricsList = append(shimMetricsList, metrics.Shim)
+		}
+	}
+
+	if len(cgroupMetricsList) == 0 || len(shimMetricsList) == 0 {
+		return nil, errors.New("no metrics collected during WaitReady stages")
+	}
+
+	// Calculate mean of cgroup metrics
+	meanCgroupMetrics := CgroupMetrics{
+		CurrentBytes: uint64(utils.Mean(
+			func() []float64 {
+				values := make([]float64, len(cgroupMetricsList))
+				for i, m := range cgroupMetricsList {
+					values[i] = float64(m.CurrentBytes)
+				}
+				return values
+			}())),
+		PeakBytes: uint64(utils.Mean(
+			func() []float64 {
+				values := make([]float64, len(cgroupMetricsList))
+				for i, m := range cgroupMetricsList {
+					values[i] = float64(m.PeakBytes)
+				}
+				return values
+			}())),
+	}
+
+	// Calculate mean of shim metrics
+	meanShimMetrics := ShimMetrics{
+		USSBytes: uint64(utils.Mean(
+			func() []float64 {
+				values := make([]float64, len(shimMetricsList))
+				for i, m := range shimMetricsList {
+					values[i] = float64(m.USSBytes)
+				}
+				return values
+			}())),
+		RSSBytes: uint64(utils.Mean(
+			func() []float64 {
+				values := make([]float64, len(shimMetricsList))
+				for i, m := range shimMetricsList {
+					values[i] = float64(m.RSSBytes)
+				}
+				return values
+			}())),
+		PSSBytes: uint64(utils.Mean(
+			func() []float64 {
+				values := make([]float64, len(shimMetricsList))
+				for i, m := range shimMetricsList {
+					values[i] = float64(m.PSSBytes)
+				}
+				return values
+			}())),
+	}
+
+	return Metrics{
+		Cgroup: meanCgroupMetrics,
+		Shim:   meanShimMetrics,
+	}, nil
+}

internal/runtime/network/adapter.go

@@ -393,3 +393,17 @@ func ExtractJSONObject(output []byte) ([]byte, error) {
 
 	return nil, fmt.Errorf("no valid JSON object found in output")
 }
+
+func (a *Adapter) GenerateResult(ctx context.Context, tc harnessruntime.TrialContext, results []harnessruntime.StageResult) (any, error) {
+	// For this adapter, we can return the metrics collected during the StartTask stage.
+	for _, result := range results {
+		if result.Stage == harnessruntime.StageStart {
+			if metrics, ok := result.Data.(map[string]any); ok {
+				return metrics, nil
+			}
+			return nil, fmt.Errorf("invalid data type for stage %s: expected map[string]any, got %T", result.Stage, result.Data)
+		}
+	}
+
+	return nil, fmt.Errorf("no metrics collected during StartTask stage")
+}

internal/runtime/storage/adapter.go

@@ -399,3 +399,14 @@ func extractFIOJSON(stdout string) ([]byte, error) {
 
 	return cleaned.Bytes(), nil
 }
+
+func (a *Adapter) GenerateResult(ctx context.Context, tc harnessruntime.TrialContext, results []harnessruntime.StageResult) (any, error) {
+	// return the first Start stage result that contains the fio JSON output
+	for _, result := range results {
+		if result.Stage == harnessruntime.StageStart {
+			return result.Data, nil
+		}
+	}
+
+	return nil, fmt.Errorf("no Start stage result found for trial %s", tc.Trial.ID)
+}