feat: Add parquet reporter

9 files changed, 339 insertions, 121 deletions

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 03995160..e6fd504e 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
@@ -39,7 +39,6 @@ import java.io.Closeable
 import java.io.File
 import java.util.concurrent.ExecutorCompletionService
 import java.util.concurrent.Executors
-import java.util.concurrent.Future
 import javax.sql.DataSource
 
 /**
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 0d60ce8c..631c1085 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
@@ -28,6 +28,8 @@ import com.atlarge.opendc.experiments.sc20.reporter.ExperimentParquetReporter
 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.ParquetHostMetricsWriter
+import com.atlarge.opendc.experiments.sc20.reporter.ParquetProvisionerMetricsWriter
 import com.atlarge.opendc.experiments.sc20.reporter.PostgresHostMetricsWriter
 import com.atlarge.opendc.experiments.sc20.reporter.PostgresProvisionerMetricsWriter
 import com.atlarge.opendc.format.trace.sc20.Sc20PerformanceInterferenceReader
@@ -182,8 +184,11 @@ internal sealed class Reporter(name: String) : OptionGroup(name), ExperimentRepo
             .file()
             .defaultLazy { File("data") }
 
-        override fun createReporter(scenario: Long, run: Int): ExperimentReporter =
-            ExperimentParquetReporter(File(path, "results-$scenario-$run.parquet"))
+        override fun createReporter(scenario: Long, run: Int): ExperimentReporter {
+            val hostWriter = ParquetHostMetricsWriter(File(path, "$scenario-$run-host.parquet"), bufferSize)
+            val provisionerWriter = ParquetProvisionerMetricsWriter(File(path, "$scenario-$run-provisioner.parquet"), bufferSize)
+            return ExperimentParquetReporter(scenario, run, hostWriter, provisionerWriter)
+        }
 
         override fun close() {}
     }
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 87bf524f..9426933e 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
@@ -27,62 +27,19 @@ package com.atlarge.opendc.experiments.sc20.reporter
 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
-import org.apache.avro.SchemaBuilder
-import org.apache.avro.generic.GenericData
-import org.apache.hadoop.fs.Path
-import org.apache.parquet.avro.AvroParquetWriter
-import org.apache.parquet.hadoop.metadata.CompressionCodecName
-import java.io.File
-import java.util.concurrent.ArrayBlockingQueue
-import kotlin.concurrent.thread
 
 private val logger = KotlinLogging.logger {}
 
-class ExperimentParquetReporter(destination: File) :
+class ExperimentParquetReporter(
+    val scenario: Long,
+    val run: Int,
+    val hostWriter: ParquetHostMetricsWriter,
+    val provisionerWriter: ParquetProvisionerMetricsWriter
+) :
     ExperimentReporter {
     private val lastServerStates = mutableMapOf<Server, Pair<ServerState, Long>>()
-    private val schema = SchemaBuilder
-        .record("slice")
-        .namespace("com.atlarge.opendc.experiments.sc20")
-        .fields()
-        .name("time").type().longType().noDefault()
-        .name("duration").type().longType().noDefault()
-        .name("requested_burst").type().longType().noDefault()
-        .name("granted_burst").type().longType().noDefault()
-        .name("overcommissioned_burst").type().longType().noDefault()
-        .name("interfered_burst").type().longType().noDefault()
-        .name("cpu_usage").type().doubleType().noDefault()
-        .name("cpu_demand").type().doubleType().noDefault()
-        .name("image_count").type().intType().noDefault()
-        .name("server").type().stringType().noDefault()
-        .name("host_state").type().stringType().noDefault()
-        .name("host_usage").type().doubleType().noDefault()
-        .name("power_draw").type().doubleType().noDefault()
-        .name("total_submitted_vms").type().longType().noDefault()
-        .name("total_queued_vms").type().longType().noDefault()
-        .name("total_running_vms").type().longType().noDefault()
-        .name("total_finished_vms").type().longType().noDefault()
-        .endRecord()
-    private val writer = AvroParquetWriter.builder<GenericData.Record>(Path(destination.absolutePath))
-        .withSchema(schema)
-        .withCompressionCodec(CompressionCodecName.SNAPPY)
-        .withPageSize(4 * 1024 * 1024) // For compression
-        .withRowGroupSize(16 * 1024 * 1024) // For write buffering (Page size)
-        .build()
-    private val queue = ArrayBlockingQueue<GenericData.Record>(2048)
-    private val writerThread = thread(start = true, name = "sc20-writer") {
-        try {
-            while (true) {
-                val record = queue.take()
-                writer.write(record)
-            }
-        } catch (e: InterruptedException) {
-            // Do not rethrow this
-        } finally {
-            writer.close()
-        }
-    }
 
     override fun reportVmStateChange(time: Long, server: Server) {}
 
@@ -91,7 +48,7 @@ class ExperimentParquetReporter(destination: File) :
         driver: VirtDriver,
         server: Server
     ) {
-        logger.info("Host ${server.uid} changed state ${server.state} [$time]")
+        logger.debug("Host ${server.uid} changed state ${server.state} [$time]")
 
         val lastServerState = lastServerStates[server]
         if (server.state == ServerState.SHUTOFF && lastServerState != null) {
@@ -134,33 +91,42 @@ class ExperimentParquetReporter(destination: File) :
         hostServer: Server,
         duration: Long
     ) {
-        val record = GenericData.Record(schema)
-        record.put("time", time)
-        record.put("duration", duration)
-        record.put("requested_burst", requestedBurst)
-        record.put("granted_burst", grantedBurst)
-        record.put("overcommissioned_burst", overcommissionedBurst)
-        record.put("interfered_burst", interferedBurst)
-        record.put("cpu_usage", cpuUsage)
-        record.put("cpu_demand", cpuDemand)
-        record.put("image_count", numberOfDeployedImages)
-        record.put("server", hostServer.uid)
-        record.put("host_state", hostServer.state)
-        record.put("host_usage", cpuUsage)
-        record.put("power_draw", lastPowerConsumption[hostServer] ?: 200.0)
-        record.put("total_submitted_vms", -1)
-        record.put("total_queued_vms", -1)
-        record.put("total_running_vms", -1)
-        record.put("total_finished_vms", -1)
+        hostWriter.write(
+            scenario, run, HostMetrics(
+                time,
+                duration,
+                hostServer,
+                numberOfDeployedImages,
+                requestedBurst,
+                grantedBurst,
+                overcommissionedBurst,
+                interferedBurst,
+                cpuUsage,
+                cpuDemand,
+                lastPowerConsumption[hostServer] ?: 200.0
+            )
+        )
+    }
 
-        queue.put(record)
+    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() {
-        // Busy loop to wait for writer thread to finish
-        while (queue.isNotEmpty()) {
-            Thread.sleep(500)
-        }
-        writerThread.interrupt()
+        hostWriter.close()
+        provisionerWriter.close()
     }
 }
diff --git a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/reporter/ParquetHostMetricsWriter.kt b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/reporter/ParquetHostMetricsWriter.kt
new file mode 100644
index 00000000..33839191
--- /dev/null
+++ b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/reporter/ParquetHostMetricsWriter.kt
@@ -0,0 +1,73 @@
+/*
+ * 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 org.apache.avro.Schema
+import org.apache.avro.SchemaBuilder
+import org.apache.avro.generic.GenericData
+import java.io.File
+
+private val schema: Schema = SchemaBuilder
+    .record("host_metrics")
+    .namespace("com.atlarge.opendc.experiments.sc20")
+    .fields()
+    .name("scenario_id").type().longType().noDefault()
+    .name("run_id").type().intType().noDefault()
+    .name("timestamp").type().longType().noDefault()
+    .name("duration").type().longType().noDefault()
+    .name("host_id").type().stringType().noDefault()
+    .name("state").type().stringType().noDefault()
+    .name("vm_count").type().intType().noDefault()
+    .name("requested_burst").type().longType().noDefault()
+    .name("granted_burst").type().longType().noDefault()
+    .name("overcommissioned_burst").type().longType().noDefault()
+    .name("interfered_burst").type().longType().noDefault()
+    .name("cpu_usage").type().doubleType().noDefault()
+    .name("cpu_demand").type().doubleType().noDefault()
+    .name("power_draw").type().doubleType().noDefault()
+    .endRecord()
+
+public class ParquetHostMetricsWriter(path: File, batchSize: Int) :
+    ParquetMetricsWriter<HostMetrics>(path, schema, batchSize) {
+
+    override fun persist(action: Action.Write<HostMetrics>, row: GenericData.Record) {
+        row.put("scenario_id", action.scenario)
+        row.put("run_id", action.run)
+        row.put("host_id", action.metrics.host.name)
+        row.put("state", action.metrics.host.state.name)
+        row.put("timestamp", action.metrics.time)
+        row.put("duration", action.metrics.duration)
+        row.put("vm_count", action.metrics.vmCount)
+        row.put("requested_burst", action.metrics.requestedBurst)
+        row.put("granted_burst", action.metrics.grantedBurst)
+        row.put("overcommissioned_burst", action.metrics.overcommissionedBurst)
+        row.put("interfered_burst", action.metrics.interferedBurst)
+        row.put("cpu_usage", action.metrics.cpuUsage)
+        row.put("cpu_demand", action.metrics.cpuDemand)
+        row.put("power_draw", 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/ParquetMetricsWriter.kt b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/reporter/ParquetMetricsWriter.kt
new file mode 100644
index 00000000..e82e9e47
--- /dev/null
+++ b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/reporter/ParquetMetricsWriter.kt
@@ -0,0 +1,114 @@
+/*
+ * 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 mu.KotlinLogging
+import org.apache.avro.Schema
+import org.apache.avro.generic.GenericData
+import org.apache.hadoop.fs.Path
+import org.apache.parquet.avro.AvroParquetWriter
+import org.apache.parquet.hadoop.metadata.CompressionCodecName
+import java.io.Closeable
+import java.io.File
+import java.util.concurrent.ArrayBlockingQueue
+import java.util.concurrent.BlockingQueue
+import kotlin.concurrent.thread
+
+private val logger = KotlinLogging.logger {}
+
+public abstract class ParquetMetricsWriter<T>(
+    private val path: File,
+    private val schema: Schema,
+    private val bufferSize: Int = 4096
+) : Runnable, Closeable {
+    /**
+     * The queue of commands to process.
+     */
+    private val queue: BlockingQueue<Action> = ArrayBlockingQueue(bufferSize)
+    private val writerThread = thread(start = true, name = "parquet-writer") { run() }
+
+    /**
+     * Write the specified metrics to the database.
+     */
+    public fun write(scenario: Long, run: Int, metrics: T) {
+        queue.put(Action.Write(scenario, run, metrics))
+    }
+
+    /**
+     * Signal the writer to stop.
+     */
+    public override fun close() {
+        queue.put(Action.Stop)
+        writerThread.join()
+    }
+
+    /**
+     * Persist the specified metrics to the given [row].
+     */
+    public abstract fun persist(action: Action.Write<T>, row: GenericData.Record)
+
+    /**
+     * Start the writer thread.
+     */
+    override fun run() {
+        val writer = AvroParquetWriter.builder<GenericData.Record>(Path(path.absolutePath))
+            .withSchema(schema)
+            .withCompressionCodec(CompressionCodecName.SNAPPY)
+            .withPageSize(4 * 1024 * 1024) // For compression
+            .withRowGroupSize(16 * 1024 * 1024) // For write buffering (Page size)
+            .build()
+
+        try {
+            loop@ while (true) {
+                val action = queue.take()
+                when (action) {
+                    is Action.Stop -> break@loop
+                    is Action.Write<*> -> {
+                        val record = GenericData.Record(schema)
+                        @Suppress("UNCHECKED_CAST")
+                        persist(action as Action.Write<T>, record)
+                        writer.write(record)
+                    }
+                }
+            }
+        } catch (e: Throwable) {
+            logger.error("Writer failed", e)
+        } finally {
+            writer.close()
+        }
+    }
+
+    sealed class Action {
+        /**
+         * A poison pill that will stop the writer thread.
+         */
+        object Stop : Action()
+
+        /**
+         * Write the specified metrics to the database.
+         */
+        data class Write<out T>(val scenario: Long, val run: Int, val metrics: T) : Action()
+    }
+}
diff --git a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/reporter/ParquetProvisionerMetricsWriter.kt b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/reporter/ParquetProvisionerMetricsWriter.kt
new file mode 100644
index 00000000..0c74b23e
--- /dev/null
+++ b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/reporter/ParquetProvisionerMetricsWriter.kt
@@ -0,0 +1,65 @@
+/*
+ * 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 org.apache.avro.Schema
+import org.apache.avro.SchemaBuilder
+import org.apache.avro.generic.GenericData
+import java.io.File
+
+private val schema: Schema = SchemaBuilder
+    .record("host_metrics")
+    .namespace("com.atlarge.opendc.experiments.sc20")
+    .fields()
+    .name("scenario_id").type().longType().noDefault()
+    .name("run_id").type().intType().noDefault()
+    .name("timestamp").type().longType().noDefault()
+    .name("host_total_count").type().intType().noDefault()
+    .name("host_available_count").type().intType().noDefault()
+    .name("vm_total_count").type().intType().noDefault()
+    .name("vm_active_count").type().intType().noDefault()
+    .name("vm_inactive_count").type().intType().noDefault()
+    .name("vm_waiting_count").type().intType().noDefault()
+    .name("vm_failed_count").type().intType().noDefault()
+    .endRecord()
+
+public class ParquetProvisionerMetricsWriter(path: File, batchSize: Int) :
+    ParquetMetricsWriter<ProvisionerMetrics>(path, schema, batchSize) {
+
+    override fun persist(action: Action.Write<ProvisionerMetrics>, row: GenericData.Record) {
+        row.put("scenario_id", action.scenario)
+        row.put("run_id", action.run)
+        row.put("timestamp", action.metrics.time)
+        row.put("host_total_count", action.metrics.totalHostCount)
+        row.put("host_available_count", action.metrics.availableHostCount)
+        row.put("vm_total_count", action.metrics.totalVmCount)
+        row.put("vm_active_count", action.metrics.activeVmCount)
+        row.put("vm_inactive_count", action.metrics.inactiveVmCount)
+        row.put("vm_waiting_count", action.metrics.waitingVmCount)
+        row.put("vm_failed_count", action.metrics.failedVmCount)
+    }
+
+    override fun toString(): String = "host-writer"
+}
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 43e4a7a6..57e665ae 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
@@ -28,31 +28,31 @@ import de.bytefish.pgbulkinsert.row.SimpleRow
 import de.bytefish.pgbulkinsert.row.SimpleRowWriter
 import javax.sql.DataSource
 
+private val table: SimpleRowWriter.Table = SimpleRowWriter.Table(
+    "host_metrics",
+    *arrayOf(
+        "scenario_id",
+        "run_id",
+        "host_id",
+        "state",
+        "timestamp",
+        "duration",
+        "vm_count",
+        "requested_burst",
+        "granted_burst",
+        "overcommissioned_burst",
+        "interfered_burst",
+        "cpu_usage",
+        "cpu_demand",
+        "power_draw"
+    )
+)
+
 /**
  * A [PostgresMetricsWriter] for persisting [HostMetrics].
  */
 public class PostgresHostMetricsWriter(ds: DataSource, parallelism: Int, batchSize: Int) :
-    PostgresMetricsWriter<HostMetrics>(ds, parallelism, batchSize) {
-
-    override val table: SimpleRowWriter.Table = SimpleRowWriter.Table(
-        "host_metrics",
-        *arrayOf(
-            "scenario_id",
-            "run_id",
-            "host_id",
-            "state",
-            "timestamp",
-            "duration",
-            "vm_count",
-            "requested_burst",
-            "granted_burst",
-            "overcommissioned_burst",
-            "interfered_burst",
-            "cpu_usage",
-            "cpu_demand",
-            "power_draw"
-        )
-    )
+    PostgresMetricsWriter<HostMetrics>(ds, table, parallelism, batchSize) {
 
     override fun persist(action: Action.Write<HostMetrics>, row: SimpleRow) {
         row.setLong("scenario_id", action.scenario)
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 715800a3..bee01e51 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
@@ -40,6 +40,7 @@ import javax.sql.DataSource
  */
 public abstract class PostgresMetricsWriter<T>(
     private val ds: DataSource,
+    private val table: SimpleRowWriter.Table,
     private val parallelism: Int = 8,
     private val bufferSize: Int = 4096
 ) : Runnable, Closeable {
@@ -72,11 +73,6 @@ public abstract class PostgresMetricsWriter<T>(
     }
 
     /**
-     * Create the table to which we write.
-     */
-    public abstract val table: SimpleRowWriter.Table
-
-    /**
      * Persist the specified metrics to the given [row].
      */
     public abstract fun persist(action: Action.Write<T>, row: SimpleRow)
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
index a7a86206..17788112 100644
--- 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
@@ -28,27 +28,27 @@ import de.bytefish.pgbulkinsert.row.SimpleRow
 import de.bytefish.pgbulkinsert.row.SimpleRowWriter
 import javax.sql.DataSource
 
+private val table: SimpleRowWriter.Table = SimpleRowWriter.Table(
+    "provisioner_metrics",
+    *arrayOf(
+        "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"
+    )
+)
+
 /**
  * A [PostgresMetricsWriter] for persisting [ProvisionerMetrics].
  */
 public class PostgresProvisionerMetricsWriter(ds: DataSource, parallelism: Int, batchSize: Int) :
-    PostgresMetricsWriter<ProvisionerMetrics>(ds, parallelism, batchSize) {
-
-    override val table: SimpleRowWriter.Table = SimpleRowWriter.Table(
-        "provisioner_metrics",
-        *arrayOf(
-            "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"
-        )
-    )
+    PostgresMetricsWriter<ProvisionerMetrics>(ds, table, parallelism, batchSize) {
 
     override fun persist(action: Action.Write<ProvisionerMetrics>, row: SimpleRow) {
         row.setLong("scenario_id", action.scenario)