Adds load shifting over time (#319)

* Start time shifting

* Existing experiments work with new columns

* Remove unused traces dir

* Update java to 21 LTS and jacoco to be compatible

* Minimal working timeshifting

* Timeshift scheduler linked as carbon receiver

* Add basic tests for timeshift scheduler

* Run spotless apply

* Modify tarce format tests to support new fields

* Change all mentions of java 19 to 21

* Add a deferAll option to workload to make all tasks deferrable

* Run spotless apply

* Copy traces from resources in web dockerfile

3 files changed, 34 insertions, 2 deletions

diff --git a/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/experiment/specs/WorkloadSpec.kt b/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/experiment/specs/WorkloadSpec.kt
index cf40d88d..4d2e932f 100644
--- a/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/experiment/specs/WorkloadSpec.kt
+++ b/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/experiment/specs/WorkloadSpec.kt
@@ -44,6 +44,7 @@ public data class WorkloadSpec(
     val type: WorkloadTypes,
     val sampleFraction: Double = 1.0,
     val submissionTime: String? = null,
+    val deferAll: Boolean = false,
     val scalingPolicy: ScalingPolicyEnum = ScalingPolicyEnum.NoDelay,
 ) {
     public val name: String = File(pathToFile).nameWithoutExtension
@@ -74,6 +75,7 @@ public fun getWorkloadLoader(
     checkpointDuration: Long,
     checkpointIntervalScaling: Double,
     scalingPolicy: ScalingPolicy,
+    deferAll: Boolean,
 ): WorkloadLoader {
     return when (type) {
         WorkloadTypes.ComputeWorkload ->
@@ -84,6 +86,7 @@ public fun getWorkloadLoader(
                 checkpointDuration,
                 checkpointIntervalScaling,
                 scalingPolicy,
+                deferAll,
             )
     }
 }
diff --git a/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/runner/ScenarioReplayer.kt b/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/runner/ScenarioReplayer.kt
index 5664e0cb..d56e4e4b 100644
--- a/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/runner/ScenarioReplayer.kt
+++ b/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/runner/ScenarioReplayer.kt
@@ -34,9 +34,11 @@ import org.opendc.compute.failure.models.FailureModel
 import org.opendc.compute.simulator.TaskWatcher
 import org.opendc.compute.simulator.service.ComputeService
 import org.opendc.compute.simulator.service.ServiceTask
+import org.opendc.compute.simulator.service.TaskNature
 import org.opendc.compute.workload.Task
 import org.opendc.experiments.base.experiment.specs.FailureModelSpec
 import org.opendc.experiments.base.experiment.specs.createFailureModel
+import java.time.Duration
 import java.time.InstantSource
 import java.util.Random
 import kotlin.coroutines.coroutineContext
@@ -114,15 +116,26 @@ public suspend fun ComputeService.replay(
                 // Delay the task based on the startTime given by the trace.
                 if (!submitImmediately) {
                     delay(max(0, (start - now - simulationOffset)))
+                    entry.deadline -= simulationOffset
                 }
 
                 val workload = entry.trace
                 val meta = mutableMapOf<String, Any>("workload" to workload)
 
+                val nature =
+                    if (entry.nature == "deferrable") {
+                        TaskNature(true)
+                    } else {
+                        TaskNature(false)
+                    }
+
                 launch {
                     val task =
                         client.newTask(
                             entry.name,
+                            nature,
+                            Duration.ofMillis(entry.duration),
+                            entry.deadline,
                             client.newFlavor(
                                 entry.name,
                                 entry.cpuCount,
diff --git a/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/runner/ScenarioRunner.kt b/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/runner/ScenarioRunner.kt
index 4da035ec..f8cbb4fd 100644
--- a/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/runner/ScenarioRunner.kt
+++ b/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/runner/ScenarioRunner.kt
@@ -35,6 +35,7 @@ import org.opendc.compute.topology.clusterTopology
 import org.opendc.experiments.base.experiment.Scenario
 import org.opendc.experiments.base.experiment.specs.getScalingPolicy
 import org.opendc.experiments.base.experiment.specs.getWorkloadLoader
+import org.opendc.simulator.compute.power.CarbonReceiver
 import org.opendc.simulator.kotlin.runSimulation
 import java.io.File
 import java.time.Duration
@@ -92,20 +93,35 @@ public fun runScenario(
                     checkpointDuration,
                     checkpointIntervalScaling,
                     scalingPolicy,
+                    scenario.workloadSpec.deferAll,
                 )
             var workload = workloadLoader.sampleByLoad(scenario.workloadSpec.sampleFraction)
 
             val startTimeLong = workload.minOf { it.submissionTime }
             val startTime = Duration.ofMillis(startTimeLong)
 
+            val carbonReceivers = mutableListOf<CarbonReceiver>()
             val topology = clusterTopology(scenario.topologySpec.pathToFile)
             provisioner.runSteps(
                 setupComputeService(
                     serviceDomain,
-                    { createComputeScheduler(scenario.allocationPolicySpec.policyType, Random(it.seeder.nextLong())) },
+                    {
+                        val computeScheduler =
+                            createComputeScheduler(
+                                scenario.allocationPolicySpec.policyType,
+                                Random(it.seeder.nextLong()),
+                                timeSource,
+                            )
+
+                        if (computeScheduler is CarbonReceiver) {
+                            carbonReceivers.add(computeScheduler)
+                        }
+
+                        return@setupComputeService computeScheduler
+                    },
                     maxNumFailures = scenario.maxNumFailures,
                 ),
-                setupHosts(serviceDomain, topology, startTimeLong),
+                setupHosts(serviceDomain, topology, carbonReceivers, startTimeLong),
             )
 
             addExportModel(provisioner, serviceDomain, scenario, seed, startTime, scenario.id)