diff options
14 files changed, 298 insertions, 88 deletions
diff --git a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/SimpleVirtProvisioningService.kt b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/SimpleVirtProvisioningService.kt index 3603ae69..c25834a7 100644 --- a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/SimpleVirtProvisioningService.kt +++ b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/SimpleVirtProvisioningService.kt @@ -1,6 +1,7 @@ package com.atlarge.opendc.compute.virt.service import com.atlarge.odcsim.SimulationContext +import com.atlarge.odcsim.flow.EventFlow import com.atlarge.odcsim.simulationContext import com.atlarge.opendc.compute.core.Flavor import com.atlarge.opendc.compute.core.Server @@ -19,6 +20,7 @@ import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.Job import kotlinx.coroutines.delay +import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.launchIn import kotlinx.coroutines.flow.onEach import kotlinx.coroutines.launch @@ -57,11 +59,11 @@ class SimpleVirtProvisioningService( */ private val activeImages: MutableSet<ImageView> = mutableSetOf() - public var submittedVms = 0L - public var queuedVms = 0L - public var runningVms = 0L - public var finishedVms = 0L - public var unscheduledVms = 0L + public var submittedVms = 0 + public var queuedVms = 0 + public var runningVms = 0 + public var finishedVms = 0 + public var unscheduledVms = 0 private var maxCores = 0 private var maxMemory = 0L @@ -71,6 +73,13 @@ class SimpleVirtProvisioningService( */ private val allocationLogic = allocationPolicy() + /** + * The [EventFlow] to emit the events. + */ + internal val eventFlow = EventFlow<VirtProvisioningEvent>() + + override val events: Flow<VirtProvisioningEvent> = eventFlow + init { launch { val provisionedNodes = provisioningService.nodes() @@ -96,8 +105,17 @@ class SimpleVirtProvisioningService( image: Image, flavor: Flavor ): Server = withContext(coroutineContext) { - submittedVms++ - queuedVms++ + eventFlow.emit(VirtProvisioningEvent.MetricsAvailable( + this@SimpleVirtProvisioningService, + hypervisors.size, + availableHypervisors.size, + ++submittedVms, + runningVms, + finishedVms, + ++queuedVms, + unscheduledVms + )) + suspendCancellableCoroutine<Server> { cont -> val vmInstance = ImageView(name, image, flavor, cont) incomingImages += vmInstance @@ -141,7 +159,17 @@ class SimpleVirtProvisioningService( if (selectedHv == null) { if (requiredMemory > maxMemory || imageInstance.flavor.cpuCount > maxCores) { - unscheduledVms++ + eventFlow.emit(VirtProvisioningEvent.MetricsAvailable( + this@SimpleVirtProvisioningService, + hypervisors.size, + availableHypervisors.size, + submittedVms, + runningVms, + finishedVms, + queuedVms, + ++unscheduledVms + )) + incomingImages -= imageInstance logger.warn("Failed to spawn ${imageInstance.image}: does not fit [${clock.millis()}]") @@ -168,8 +196,17 @@ class SimpleVirtProvisioningService( ) imageInstance.server = server imageInstance.continuation.resume(server) - queuedVms-- - runningVms++ + + eventFlow.emit(VirtProvisioningEvent.MetricsAvailable( + this@SimpleVirtProvisioningService, + hypervisors.size, + availableHypervisors.size, + submittedVms, + ++runningVms, + finishedVms, + --queuedVms, + unscheduledVms + )) activeImages += imageInstance server.events @@ -178,8 +215,17 @@ class SimpleVirtProvisioningService( is ServerEvent.StateChanged -> { if (event.server.state == ServerState.SHUTOFF) { logger.info { "Server ${event.server.uid} ${event.server.name} ${event.server.flavor} finished." } - runningVms-- - finishedVms++ + + eventFlow.emit(VirtProvisioningEvent.MetricsAvailable( + this@SimpleVirtProvisioningService, + hypervisors.size, + availableHypervisors.size, + submittedVms, + --runningVms, + ++finishedVms, + queuedVms, + unscheduledVms + )) activeImages -= imageInstance selectedHv.provisionedCores -= server.flavor.cpuCount @@ -211,6 +257,7 @@ class SimpleVirtProvisioningService( if (server in hypervisors) { // Corner case for when the hypervisor already exists availableHypervisors += hypervisors.getValue(server) + } else { val hv = HypervisorView( server.uid, @@ -223,11 +270,33 @@ class SimpleVirtProvisioningService( maxMemory = max(maxMemory, server.flavor.memorySize) hypervisors[server] = hv } + + eventFlow.emit(VirtProvisioningEvent.MetricsAvailable( + this@SimpleVirtProvisioningService, + hypervisors.size, + availableHypervisors.size, + submittedVms, + runningVms, + finishedVms, + queuedVms, + unscheduledVms + )) } ServerState.SHUTOFF, ServerState.ERROR -> { val hv = hypervisors[server] ?: return availableHypervisors -= hv + eventFlow.emit(VirtProvisioningEvent.MetricsAvailable( + this@SimpleVirtProvisioningService, + hypervisors.size, + availableHypervisors.size, + submittedVms, + runningVms, + finishedVms, + queuedVms, + unscheduledVms + )) + if (incomingImages.isNotEmpty()) { requestCycle() } @@ -242,6 +311,17 @@ class SimpleVirtProvisioningService( hv.driver = server.services[VirtDriver] availableHypervisors += hv + eventFlow.emit(VirtProvisioningEvent.MetricsAvailable( + this@SimpleVirtProvisioningService, + hypervisors.size, + availableHypervisors.size, + submittedVms, + runningVms, + finishedVms, + queuedVms, + unscheduledVms + )) + hv.driver.events .onEach { event -> if (event is HypervisorEvent.VmsUpdated) { diff --git a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/VirtProvisioningEvent.kt b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/VirtProvisioningEvent.kt new file mode 100644 index 00000000..39f75913 --- /dev/null +++ b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/VirtProvisioningEvent.kt @@ -0,0 +1,52 @@ +/* + * MIT License + * + * Copyright (c) 2020 atlarge-research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package com.atlarge.opendc.compute.virt.service + +import com.atlarge.opendc.compute.virt.driver.VirtDriver + + +/** + * An event that is emitted by the [VirtProvisioningService]. + */ +public sealed class VirtProvisioningEvent { + /** + * The service that has emitted the event. + */ + public abstract val provisioner: VirtProvisioningService + + /** + * An event emitted for writing metrics. + */ + data class MetricsAvailable( + override val provisioner: VirtProvisioningService, + public val totalHostCount: Int, + public val availableHostCount: Int, + public val totalVmCount: Int, + public val activeVmCount: Int, + public val inactiveVmCount: Int, + public val waitingVmCount: Int, + public val failedVmCount: Int + ) : VirtProvisioningEvent() +} diff --git a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/VirtProvisioningService.kt b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/VirtProvisioningService.kt index 2ad7df84..c4cbd711 100644 --- a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/VirtProvisioningService.kt +++ b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/VirtProvisioningService.kt @@ -5,6 +5,7 @@ import com.atlarge.opendc.compute.core.Server import com.atlarge.opendc.compute.core.image.Image import com.atlarge.opendc.compute.virt.driver.VirtDriver import com.atlarge.opendc.compute.virt.service.allocation.AllocationPolicy +import kotlinx.coroutines.flow.Flow /** * A service for VM provisioning on a cloud. @@ -16,6 +17,11 @@ interface VirtProvisioningService { val allocationPolicy: AllocationPolicy /** + * The events emitted by the service. + */ + public val events: Flow<VirtProvisioningEvent> + + /** * Obtain the active hypervisors for this provisioner. */ public suspend fun drivers(): Set<VirtDriver> diff --git a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/ExperimentHelpers.kt b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/ExperimentHelpers.kt index ac43b6ac..548400d6 100644 --- a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/ExperimentHelpers.kt +++ b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/ExperimentHelpers.kt @@ -36,6 +36,7 @@ import com.atlarge.opendc.compute.metal.service.ProvisioningService import com.atlarge.opendc.compute.virt.HypervisorEvent import com.atlarge.opendc.compute.virt.driver.SimpleVirtDriver import com.atlarge.opendc.compute.virt.service.SimpleVirtProvisioningService +import com.atlarge.opendc.compute.virt.service.VirtProvisioningEvent import com.atlarge.opendc.compute.virt.service.allocation.AllocationPolicy import com.atlarge.opendc.core.failure.CorrelatedFaultInjector import com.atlarge.opendc.core.failure.FailureDomain @@ -150,13 +151,13 @@ suspend fun attachMonitor(scheduler: SimpleVirtProvisioningService, reporter: Ex // Monitor hypervisor events for (hypervisor in hypervisors) { // TODO Do not expose VirtDriver directly but use Hypervisor class. - reporter.reportHostStateChange(clock.millis(), hypervisor, (hypervisor as SimpleVirtDriver).server, scheduler.submittedVms, scheduler.queuedVms, scheduler.runningVms, scheduler.finishedVms) + reporter.reportHostStateChange(clock.millis(), hypervisor, (hypervisor as SimpleVirtDriver).server) hypervisor.server.events .onEach { event -> val time = clock.millis() when (event) { is ServerEvent.StateChanged -> { - reporter.reportHostStateChange(time, hypervisor, event.server, scheduler.submittedVms, scheduler.queuedVms, scheduler.runningVms, scheduler.finishedVms) + reporter.reportHostStateChange(time, hypervisor, event.server) } } } @@ -173,11 +174,7 @@ suspend fun attachMonitor(scheduler: SimpleVirtProvisioningService, reporter: Ex event.cpuUsage, event.cpuDemand, event.numberOfDeployedImages, - event.hostServer, - scheduler.submittedVms, - scheduler.queuedVms, - scheduler.runningVms, - scheduler.finishedVms + event.hostServer ) } } @@ -188,6 +185,16 @@ suspend fun attachMonitor(scheduler: SimpleVirtProvisioningService, reporter: Ex .onEach { reporter.reportPowerConsumption(hypervisor.server, it) } .launchIn(domain) } + + scheduler.events + .onEach { event -> + when (event) { + is VirtProvisioningEvent.MetricsAvailable -> + reporter.reportProvisionerMetrics(clock.millis(), event) + } + + } + .launchIn(domain) } /** @@ -197,7 +204,7 @@ suspend fun processTrace(reader: TraceReader<VmWorkload>, scheduler: SimpleVirtP val domain = simulationContext.domain try { - var submitted = 0L + var submitted = 0 val finished = Channel<Unit>(Channel.CONFLATED) val hypervisors = TreeSet(scheduler.drivers().map { (it as SimpleVirtDriver).server.name }) diff --git a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/ExperimentRunner.kt b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/ExperimentRunner.kt index 9455cb9d..7d65930c 100644 --- a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/ExperimentRunner.kt +++ b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/ExperimentRunner.kt @@ -33,15 +33,11 @@ import com.atlarge.opendc.experiments.sc20.util.DatabaseHelper import com.atlarge.opendc.format.environment.EnvironmentReader import com.atlarge.opendc.format.environment.sc20.Sc20ClusterEnvironmentReader import com.atlarge.opendc.format.trace.TraceReader -import kotlinx.coroutines.CoroutineDispatcher -import kotlinx.coroutines.asCoroutineDispatcher -import kotlinx.coroutines.launch -import kotlinx.coroutines.runBlocking -import kotlinx.coroutines.withContext import mu.KotlinLogging import java.io.Closeable import java.io.File import java.util.concurrent.Executors +import java.util.concurrent.Future import java.util.concurrent.atomic.AtomicInteger import javax.sql.DataSource import kotlin.system.measureTimeMillis @@ -159,11 +155,19 @@ public class ExperimentRunner( val plan = createPlan() val total = plan.size val finished = AtomicInteger() - val dispatcher = Executors.newWorkStealingPool(parallelism).asCoroutineDispatcher() + val executorService = Executors.newCachedThreadPool() + val planIterator = plan.iterator() + val futures = mutableListOf<Future<*>>() - runBlocking { - val mainDispatcher = coroutineContext[CoroutineDispatcher.Key]!! - for (run in plan) { + while (planIterator.hasNext()) { + futures.clear() + + repeat(parallelism) { + if (!planIterator.hasNext()) { + return@repeat + } + + val run = planIterator.next() val scenarioId = scenarioIds[run.scenario]!! rawTraceReaders.computeIfAbsent(run.scenario.workload.name) { name -> @@ -171,15 +175,14 @@ public class ExperimentRunner( Sc20RawParquetTraceReader(File(tracePath, name)) } - launch(dispatcher) { - launch(mainDispatcher) { + val future = executorService.submit { + synchronized(helper) { helper.startRun(scenarioId, run.id) } logger.info { "[${finished.get()}/$total] Starting run ($scenarioId, ${run.id})" } try { - val duration = measureTimeMillis { val reporter = reporterProvider.createReporter(scenarioIds[run.scenario]!!, run.id) val traceReader = createTraceReader(run.scenario.workload.name, performanceInterferenceModel, run) @@ -187,6 +190,7 @@ public class ExperimentRunner( try { run.scenario(run, reporter, environmentReader, traceReader) + logger.info { "Done" } } finally { reporter.close() } @@ -195,17 +199,23 @@ public class ExperimentRunner( finished.incrementAndGet() logger.info { "[${finished.get()}/$total] Finished run ($scenarioId, ${run.id}) in $duration milliseconds" } - withContext(mainDispatcher) { + synchronized(helper) { helper.finishRun(scenarioId, run.id, hasFailed = false) } } catch (e: Throwable) { logger.error("A run has failed", e) finished.incrementAndGet() - withContext(mainDispatcher) { + synchronized(helper) { helper.finishRun(scenarioId, run.id, hasFailed = true) } } } + + futures += future + } + + for (future in futures) { + future.get() } } } diff --git a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/ExperimentRunnerCli.kt b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/ExperimentRunnerCli.kt index b072fb12..9cbfcdc1 100644 --- a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/ExperimentRunnerCli.kt +++ b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/ExperimentRunnerCli.kt @@ -29,6 +29,7 @@ import com.atlarge.opendc.experiments.sc20.reporter.ExperimentPostgresReporter import com.atlarge.opendc.experiments.sc20.reporter.ExperimentReporter import com.atlarge.opendc.experiments.sc20.reporter.ExperimentReporterProvider import com.atlarge.opendc.experiments.sc20.reporter.PostgresHostMetricsWriter +import com.atlarge.opendc.experiments.sc20.reporter.PostgresProvisionerMetricsWriter import com.atlarge.opendc.format.trace.sc20.Sc20PerformanceInterferenceReader import com.atlarge.opendc.format.trace.sc20.Sc20VmPlacementReader import com.github.ajalt.clikt.core.CliktCommand @@ -132,6 +133,7 @@ class ExperimentCli : CliktCommand(name = "sc20-experiment") { override fun run() { val ds = HikariDataSource() + ds.maximumPoolSize = Runtime.getRuntime().availableProcessors() * 3 ds.jdbcUrl = jdbcUrl ds.addDataSourceProperty("reWriteBatchedInserts", "true") @@ -178,10 +180,12 @@ internal sealed class Reporter(name: String) : OptionGroup(name), ExperimentRepo override fun createReporter(scenario: Long, run: Int): ExperimentReporter { val hostWriter = PostgresHostMetricsWriter(ds, batchSize) - val delegate = ExperimentPostgresReporter(scenario, run, hostWriter) + val provisionerWriter = PostgresProvisionerMetricsWriter(ds, batchSize) + val delegate = ExperimentPostgresReporter(scenario, run, hostWriter, provisionerWriter) return object : ExperimentReporter by delegate { override fun close() { hostWriter.close() + provisionerWriter.close() } } } diff --git a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/Portfolios.kt b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/Portfolios.kt index a4a42b6e..11b35bf9 100644 --- a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/Portfolios.kt +++ b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/Portfolios.kt @@ -69,10 +69,10 @@ object HorVerPortfolio : AbstractSc20Portfolio("horizontal_vs_vertical") { ) override val workloads = listOf( - // Workload("solvinity", 0.1), - // Workload("solvinity", 0.25), + Workload("solvinity", 0.1), + Workload("solvinity", 0.25), Workload("solvinity", 0.5), - Workload("solvinity", 0.10) + Workload("solvinity", 1.0) ) override val operationalPhenomena = listOf( diff --git a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/reporter/ExperimentReporter.kt b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/reporter/ExperimentReporter.kt index 1c752cb1..049035cc 100644 --- a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/reporter/ExperimentReporter.kt +++ b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/reporter/ExperimentReporter.kt @@ -26,6 +26,7 @@ package com.atlarge.opendc.experiments.sc20.reporter import com.atlarge.opendc.compute.core.Server import com.atlarge.opendc.compute.virt.driver.VirtDriver +import com.atlarge.opendc.compute.virt.service.VirtProvisioningEvent import java.io.Closeable /** @@ -43,17 +44,13 @@ interface ExperimentReporter : Closeable { fun reportHostStateChange( time: Long, driver: VirtDriver, - server: Server, - submittedVms: Long, - queuedVms: Long, - runningVms: Long, - finishedVms: Long + server: Server ) {} /** * Report the power consumption of a host. */ - fun reportPowerConsumption(host: Server, draw: Double) + fun reportPowerConsumption(host: Server, draw: Double) {} /** * This method is invoked for a host for each slice that is finishes. @@ -68,10 +65,11 @@ interface ExperimentReporter : Closeable { cpuDemand: Double, numberOfDeployedImages: Int, hostServer: Server, - submittedVms: Long, - queuedVms: Long, - runningVms: Long, - finishedVms: Long, duration: Long = 5 * 60 * 1000L ) {} + + /** + * This method is invoked for a provisioner event. + */ + fun reportProvisionerMetrics(time: Long, event: VirtProvisioningEvent.MetricsAvailable) {} } diff --git a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/reporter/ParquetExperimentReporter.kt b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/reporter/ParquetExperimentReporter.kt index 58d384e7..87bf524f 100644 --- a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/reporter/ParquetExperimentReporter.kt +++ b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/reporter/ParquetExperimentReporter.kt @@ -24,7 +24,6 @@ package com.atlarge.opendc.experiments.sc20.reporter -import com.atlarge.odcsim.simulationContext import com.atlarge.opendc.compute.core.Server import com.atlarge.opendc.compute.core.ServerState import com.atlarge.opendc.compute.virt.driver.VirtDriver @@ -90,11 +89,7 @@ class ExperimentParquetReporter(destination: File) : override fun reportHostStateChange( time: Long, driver: VirtDriver, - server: Server, - submittedVms: Long, - queuedVms: Long, - runningVms: Long, - finishedVms: Long + server: Server ) { logger.info("Host ${server.uid} changed state ${server.state} [$time]") @@ -111,10 +106,6 @@ class ExperimentParquetReporter(destination: File) : 0.0, 0, server, - submittedVms, - queuedVms, - runningVms, - finishedVms, duration ) @@ -141,10 +132,6 @@ class ExperimentParquetReporter(destination: File) : cpuDemand: Double, numberOfDeployedImages: Int, hostServer: Server, - submittedVms: Long, - queuedVms: Long, - runningVms: Long, - finishedVms: Long, duration: Long ) { val record = GenericData.Record(schema) @@ -161,10 +148,10 @@ class ExperimentParquetReporter(destination: File) : record.put("host_state", hostServer.state) record.put("host_usage", cpuUsage) record.put("power_draw", lastPowerConsumption[hostServer] ?: 200.0) - record.put("total_submitted_vms", submittedVms) - record.put("total_queued_vms", queuedVms) - record.put("total_running_vms", runningVms) - record.put("total_finished_vms", finishedVms) + record.put("total_submitted_vms", -1) + record.put("total_queued_vms", -1) + record.put("total_running_vms", -1) + record.put("total_finished_vms", -1) queue.put(record) } diff --git a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/reporter/PostgresExperimentReporter.kt b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/reporter/PostgresExperimentReporter.kt index a92278d8..6f220640 100644 --- a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/reporter/PostgresExperimentReporter.kt +++ b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/reporter/PostgresExperimentReporter.kt @@ -24,15 +24,20 @@ package com.atlarge.opendc.experiments.sc20.reporter -import com.atlarge.odcsim.simulationContext import com.atlarge.opendc.compute.core.Server import com.atlarge.opendc.compute.core.ServerState import com.atlarge.opendc.compute.virt.driver.VirtDriver +import com.atlarge.opendc.compute.virt.service.VirtProvisioningEvent import mu.KotlinLogging private val logger = KotlinLogging.logger {} -class ExperimentPostgresReporter(val scenario: Long, val run: Int, val writer: PostgresHostMetricsWriter) : ExperimentReporter { +class ExperimentPostgresReporter( + val scenario: Long, + val run: Int, + val hostWriter: PostgresHostMetricsWriter, + val provisionerWriter: PostgresProvisionerMetricsWriter +) : ExperimentReporter { private val lastServerStates = mutableMapOf<Server, Pair<ServerState, Long>>() override fun reportVmStateChange(time: Long, server: Server) {} @@ -40,11 +45,7 @@ class ExperimentPostgresReporter(val scenario: Long, val run: Int, val writer: P override fun reportHostStateChange( time: Long, driver: VirtDriver, - server: Server, - submittedVms: Long, - queuedVms: Long, - runningVms: Long, - finishedVms: Long + server: Server ) { val lastServerState = lastServerStates[server] logger.debug("Host ${server.uid} changed state ${server.state} [$time]") @@ -61,10 +62,6 @@ class ExperimentPostgresReporter(val scenario: Long, val run: Int, val writer: P 0.0, 0, server, - submittedVms, - queuedVms, - runningVms, - finishedVms, duration ) @@ -93,13 +90,9 @@ class ExperimentPostgresReporter(val scenario: Long, val run: Int, val writer: P cpuDemand: Double, numberOfDeployedImages: Int, hostServer: Server, - submittedVms: Long, - queuedVms: Long, - runningVms: Long, - finishedVms: Long, duration: Long ) { - writer.write( + hostWriter.write( scenario, run, HostMetrics( time, duration, @@ -116,5 +109,22 @@ class ExperimentPostgresReporter(val scenario: Long, val run: Int, val writer: P ) } + override fun reportProvisionerMetrics(time: Long, event: VirtProvisioningEvent.MetricsAvailable) { + provisionerWriter.write( + scenario, + run, + ProvisionerMetrics( + time, + event.totalHostCount, + event.availableHostCount, + event.totalVmCount, + event.activeVmCount, + event.inactiveVmCount, + event.waitingVmCount, + event.failedVmCount + ) + ) + } + override fun close() {} } diff --git a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/reporter/PostgresHostMetricsWriter.kt b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/reporter/PostgresHostMetricsWriter.kt index 55b80e4c..5eb55f20 100644 --- a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/reporter/PostgresHostMetricsWriter.kt +++ b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/reporter/PostgresHostMetricsWriter.kt @@ -54,4 +54,6 @@ public class PostgresHostMetricsWriter(ds: DataSource, batchSize: Int) : stmt.setDouble(13, action.metrics.cpuDemand) stmt.setDouble(14, action.metrics.powerDraw) } + + override fun toString(): String = "host-writer" } diff --git a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/reporter/PostgresMetricsWriter.kt b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/reporter/PostgresMetricsWriter.kt index 367a3a5a..33c2d40e 100644 --- a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/reporter/PostgresMetricsWriter.kt +++ b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/reporter/PostgresMetricsWriter.kt @@ -48,7 +48,7 @@ public abstract class PostgresMetricsWriter<T>( /** * The thread for the actual writer. */ - private val writerThread: Thread = thread(name = "host-metrics-writer") { run() } + private val writerThread: Thread = thread(name = "metrics-writer") { run() } /** * Write the specified metrics to the database. @@ -79,6 +79,7 @@ public abstract class PostgresMetricsWriter<T>( * Start the writer thread. */ override fun run() { + writerThread.name = toString() val conn = ds.connection var batch = 0 @@ -114,7 +115,9 @@ public abstract class PostgresMetricsWriter<T>( } } } + } finally { + stmt.executeBatch() conn.commit() conn.close() } diff --git a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/reporter/PostgresProvisionerMetricsWriter.kt b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/reporter/PostgresProvisionerMetricsWriter.kt new file mode 100644 index 00000000..7bc88959 --- /dev/null +++ b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/reporter/PostgresProvisionerMetricsWriter.kt @@ -0,0 +1,55 @@ +/* + * MIT License + * + * Copyright (c) 2020 atlarge-research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package com.atlarge.opendc.experiments.sc20.reporter + +import java.sql.Connection +import java.sql.PreparedStatement +import java.sql.Timestamp +import javax.sql.DataSource + +/** + * A [PostgresMetricsWriter] for persisting [ProvisionerMetrics]. + */ +public class PostgresProvisionerMetricsWriter(ds: DataSource, batchSize: Int) : + PostgresMetricsWriter<ProvisionerMetrics>(ds, batchSize) { + override fun createStatement(conn: Connection): PreparedStatement { + return conn.prepareStatement("INSERT INTO provisioner_metrics (scenario_id, run_id, timestamp, host_total_count, host_available_count, vm_total_count, vm_active_count, vm_inactive_count, vm_waiting_count, vm_failed_count) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)") + } + + override fun persist(action: Action.Write<ProvisionerMetrics>, stmt: PreparedStatement) { + stmt.setLong(1, action.scenario) + stmt.setInt(2, action.run) + stmt.setTimestamp(3, Timestamp(action.metrics.time)) + stmt.setInt(4, action.metrics.totalHostCount) + stmt.setInt(5, action.metrics.availableHostCount) + stmt.setInt(6, action.metrics.totalVmCount) + stmt.setInt(7, action.metrics.activeVmCount) + stmt.setInt(8, action.metrics.inactiveVmCount) + stmt.setInt(9, action.metrics.waitingVmCount) + stmt.setInt(10, action.metrics.failedVmCount) + } + + override fun toString(): String = "provisioner-writer" +} diff --git a/opendc/opendc-experiments-sc20/src/test/kotlin/com/atlarge/opendc/experiments/sc20/Sc20IntegrationTest.kt b/opendc/opendc-experiments-sc20/src/test/kotlin/com/atlarge/opendc/experiments/sc20/Sc20IntegrationTest.kt index 5177c04a..1edb7bc2 100644 --- a/opendc/opendc-experiments-sc20/src/test/kotlin/com/atlarge/opendc/experiments/sc20/Sc20IntegrationTest.kt +++ b/opendc/opendc-experiments-sc20/src/test/kotlin/com/atlarge/opendc/experiments/sc20/Sc20IntegrationTest.kt @@ -158,7 +158,7 @@ class Sc20IntegrationTest { var totalOvercommissionedBurst = 0L var totalInterferedBurst = 0L - override suspend fun reportHostSlice( + override fun reportHostSlice( time: Long, requestedBurst: Long, grantedBurst: Long, @@ -168,10 +168,6 @@ class Sc20IntegrationTest { cpuDemand: Double, numberOfDeployedImages: Int, hostServer: Server, - submittedVms: Long, - queuedVms: Long, - runningVms: Long, - finishedVms: Long, duration: Long ) { totalRequestedBurst += requestedBurst |