diff options
Diffstat (limited to 'opendc-trace/opendc-trace-api/src/main/kotlin')
55 files changed, 963 insertions, 5179 deletions
diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/conv/CarbonIntensityColumns.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/conv/CarbonColumns.kt index de74c4fd..32cdd78b 100644 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/conv/CarbonIntensityColumns.kt +++ b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/conv/CarbonColumns.kt @@ -20,16 +20,16 @@ * SOFTWARE. */ -@file:JvmName("CarbonIntensityColumns") +@file:JvmName("CarbonColumns") package org.opendc.trace.conv /** - * A column containing the task identifier. + * A column containing the timestamp of the carbon intensity measurement. */ -public const val CARBON_INTENSITY_TIMESTAMP: String = "timestamp" +public const val CARBON_TIMESTAMP: String = "timestamp" /** - * A column containing the task identifier. + * A column containing the intensity of the carbon when sampled. */ -public const val CARBON_INTENSITY_VALUE: String = "carbon_intensity" +public const val CARBON_INTENSITY: String = "carbon_intensity" diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/conv/InterferenceGroupColumns.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/conv/FragmentColumns.kt index fbbfdea9..e0d01ef2 100644 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/conv/InterferenceGroupColumns.kt +++ b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/conv/FragmentColumns.kt @@ -20,21 +20,26 @@ * SOFTWARE. */ -@file:JvmName("InterferenceGroupColumns") +@file:JvmName("FragmentColumns") package org.opendc.trace.conv /** - * Members of the interference group. + * Duration for the fragment. */ -public const val INTERFERENCE_GROUP_MEMBERS: String = "members" +public const val FRAGMENT_DURATION: String = "duration" /** - * Target load after which the interference occurs. + * Total CPU usage during the fragment in MHz. */ -public const val INTERFERENCE_GROUP_TARGET: String = "target" +public const val FRAGMENT_CPU_USAGE: String = "cpu_usage" /** - * Performance score when the interference occurs. + * Total GPU usage during the fragment in MHz. */ -public const val INTERFERENCE_GROUP_SCORE: String = "score" +public const val FRAGMENT_GPU_USAGE: String = "gpu_usage" + +/** + * Memory usage during the fragment in KB. + */ +public const val FRAGMENT_MEM_USAGE: String = "mem_usage" diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/conv/ResourceColumns.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/conv/ResourceColumns.kt deleted file mode 100644 index 3d0341b2..00000000 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/conv/ResourceColumns.kt +++ /dev/null @@ -1,115 +0,0 @@ -/* - * Copyright (c) 2022 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. - */ - -@file:JvmName("ResourceColumns") - -package org.opendc.trace.conv - -/** - * Identifier of the resource. - */ -@JvmField -public val resourceID: String = "id" - -/** - * The cluster to which the resource belongs. - */ -@JvmField -public val resourceClusterID: String = "cluster_id" - -/** - * Start time for the resource. - */ -@JvmField -public val resourceSubmissionTime: String = "submission_time" - -/** - * Carbon intensity of the resource. - */ -@JvmField -public val resourceCarbonIntensity: String = "carbon_intensity" - -/** - * End time for the resource. - */ -@JvmField -public val resourceDuration: String = "duration" - -/** - * Number of CPUs for the resource. - */ -@JvmField -public val resourceCpuCount: String = "cpu_count" - -/** - * Total CPU capacity of the resource in MHz. - */ -@JvmField -public val resourceCpuCapacity: String = "cpu_capacity" - -/** - * Memory capacity for the resource in KB. - */ -@JvmField -public val resourceMemCapacity: String = "mem_capacity" - -/** - * Number of GPU cores for the resource. - */ -@JvmField -public val resourceGpuCount: String = "gpu_count" - -/** - * Total GPU capacity of the resource in MHz. - */ -@JvmField -public val resourceGpuCapacity: String = "gpu_capacity" - -/** - * Total GPU memory capacity of the resource in MB. - */ -@JvmField -public val resourceGpuMemCapacity: String = "gpu_mem_capacity" - -/** - * The parents of the resource that need to be completed before this resource can be used. - */ -@JvmField -public val resourceParents: String = "parents" - -/** - * The children of the resource that cannot be started before this is completed. - */ -@JvmField -public val resourceChildren: String = "children" - -/** - * Nature of the task. Delayable, interruptible, etc. - */ -@JvmField -public val resourceNature: String = "nature" - -/** - * Deadline of the task. - */ -@JvmField -public val resourceDeadline: String = "deadline" diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/conv/ResourceStateColumns.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/conv/ResourceStateColumns.kt deleted file mode 100644 index f4ab7759..00000000 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/conv/ResourceStateColumns.kt +++ /dev/null @@ -1,103 +0,0 @@ -/* - * Copyright (c) 2022 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. - */ - -@file:JvmName("ResourceStateColumns") - -package org.opendc.trace.conv - -/** - * The timestamp at which the state was recorded. - */ -@JvmField -public val resourceStateTimestamp: String = "timestamp" - -/** - * Duration for the state. - */ -@JvmField -public val resourceStateDuration: String = "duration" - -/** - * A flag to indicate that the resource is powered on. - */ -@JvmField -public val resourceStatePoweredOn: String = "powered_on" - -/** - * Total CPU usage of the resource in MHz. - */ -@JvmField -public val resourceStateCpuUsage: String = "cpu_usage" - -/** - * Total CPU usage of the resource in percentage. - */ -@JvmField -public val resourceStateCpuUsagePct: String = "cpu_usage_pct" - -/** - * Total CPU demand of the resource in MHz. - */ -@JvmField -public val resourceStateCpuDemand: String = "cpu_demand" - -/** - * CPU ready percentage. - */ -@JvmField -public val resourceStateCpuReadyPct: String = "cpu_ready_pct" - -/** - * Memory usage of the resource in KB. - */ -@JvmField -public val resourceStateMemUsage: String = "mem_usage" - -/** - * Disk read throughput of the resource in KB/s. - */ -@JvmField -public val resourceStateDiskRead: String = "disk_read" - -/** - * Disk write throughput of the resource in KB/s. - */ -@JvmField -public val resourceStateDiskWrite: String = "disk_write" - -/** - * Network receive throughput of the resource in KB/s. - */ -@JvmField -public val resourceStateNetRx: String = "net_rx" - -/** - * Network transmit throughput of the resource in KB/s. - */ -@JvmField -public val resourceStateNetTx: String = "net_tx" - -/** - * Total GPU capacity of the resource in MHz. - */ -@JvmField -public val resourceStateGpuUsage: String = "gpu_usage" diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/conv/Tables.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/conv/Tables.kt index d4019f73..310d268a 100644 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/conv/Tables.kt +++ b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/conv/Tables.kt @@ -25,30 +25,21 @@ package org.opendc.trace.conv /** - * A table containing all workflows in a workload. - */ -public const val TABLE_WORKFLOWS: String = "workflows" - -/** * A table containing all tasks in a workload. */ public const val TABLE_TASKS: String = "tasks" /** - * A table containing all resources in a workload. + * A table containing all resource states in a workload. */ -public const val TABLE_RESOURCES: String = "resources" +public const val TABLE_FRAGMENTS: String = "fragments" /** - * A table containing all resource states in a workload. + * A table containing the carbon intensities of the region */ -public const val TABLE_RESOURCE_STATES: String = "resource_states" +public const val TABLE_CARBON: String = "carbon" /** - * A table containing the groups of resources that interfere when run on the same execution platform. + * A table containing failures that can be injected during simulation. */ -public const val TABLE_INTERFERENCE_GROUPS: String = "interference_groups" - -public const val TABLE_CARBON_INTENSITIES: String = "carbon_intensities" - public const val TABLE_FAILURES: String = "failures" diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/conv/TaskColumns.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/conv/TaskColumns.kt index 6ca87a60..0df52c71 100644 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/conv/TaskColumns.kt +++ b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/conv/TaskColumns.kt @@ -25,61 +25,71 @@ package org.opendc.trace.conv /** - * A column containing the task identifier. + * Identifier of the task. */ public const val TASK_ID: String = "id" /** - * A column containing the identifier of the workflow. + * Identifier of the task. */ -public const val TASK_WORKFLOW_ID: String = "workflow_id" +public const val TASK_NAME: String = "name" /** - * A column containing the submission time of the task. + * The time of submission of the task. */ -public const val TASK_SUBMIT_TIME: String = "submit_time" +public const val TASK_SUBMISSION_TIME: String = "submission_time" /** - * A column containing the wait time of the task. + * The duration of a task in ms */ -public const val TASK_WAIT_TIME: String = "wait_time" +public const val TASK_DURATION: String = "duration" /** - * A column containing the runtime time of the task. + * Number of CPUs for the task. */ -public const val TASK_RUNTIME: String = "runtime" +public const val TASK_CPU_COUNT: String = "cpu_count" /** - * A column containing the parents of a task. + * Total CPU capacity of the task in MHz. */ -public const val TASK_PARENTS: String = "parents" +public const val TASK_CPU_CAPACITY: String = "cpu_capacity" /** - * A column containing the children of a task. + * Memory capacity for the task in KB. */ -public const val TASK_CHILDREN: String = "children" +public const val TASK_MEM_CAPACITY: String = "mem_capacity" + +/** + * Number of GPU cores for the task. + */ +public const val TASK_GPU_COUNT: String = "gpu_count" /** - * A column containing the requested CPUs of a task. + * Total GPU capacity of the task in MHz. */ -public const val TASK_REQ_NCPUS: String = "req_ncpus" +public const val TASK_GPU_CAPACITY: String = "gpu_capacity" /** - * A column containing the allocated CPUs of a task. + * Total GPU memory capacity of the task in MB. */ -public const val TASK_ALLOC_NCPUS: String = "alloc_ncpus" +public const val TASK_GPU_MEM_CAPACITY: String = "gpu_mem_capacity" /** - * A column containing the status of a task. + * The parents of the task that need to be completed before this task can be used. */ -public const val TASK_STATUS: String = "status" +public const val TASK_PARENTS: String = "parents" + +/** + * The children of the task that cannot be started before this is completed. + */ +public const val TASK_CHILDREN: String = "children" /** - * A column containing the group id of a task. + * Nature of the task. Delayable, interruptible, etc. */ -public const val TASK_GROUP_ID: String = "group_id" +public const val TASK_NATURE: String = "nature" /** - * A column containing the user id of a task. + * Deadline of the task. */ -public const val TASK_USER_ID: String = "user_id" +public const val TASK_DEADLINE: String = "deadline" diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/azure/AzureResourceStateTableReader.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/azure/AzureResourceStateTableReader.kt deleted file mode 100644 index bcf6ff52..00000000 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/azure/AzureResourceStateTableReader.kt +++ /dev/null @@ -1,219 +0,0 @@ -/* - * Copyright (c) 2021 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 org.opendc.trace.azure - -import com.fasterxml.jackson.core.JsonToken -import com.fasterxml.jackson.dataformat.csv.CsvParser -import com.fasterxml.jackson.dataformat.csv.CsvSchema -import org.opendc.trace.TableReader -import org.opendc.trace.conv.resourceID -import org.opendc.trace.conv.resourceStateCpuUsagePct -import org.opendc.trace.conv.resourceStateTimestamp -import java.time.Duration -import java.time.Instant -import java.util.UUID - -/** - * A [TableReader] for the Azure v1 VM resource state table. - */ -internal class AzureResourceStateTableReader(private val parser: CsvParser) : TableReader { - /** - * A flag to indicate whether a single row has been read already. - */ - private var isStarted = false - - init { - parser.schema = schema - } - - override fun nextRow(): Boolean { - if (!isStarted) { - isStarted = true - } - - reset() - - if (!nextStart()) { - return false - } - - while (true) { - val token = parser.nextValue() - - if (token == null || token == JsonToken.END_OBJECT) { - break - } - - when (parser.currentName) { - "timestamp" -> timestamp = Instant.ofEpochSecond(parser.longValue) - "vm id" -> id = parser.text - "CPU avg cpu" -> cpuUsagePct = (parser.doubleValue / 100.0) // Convert from % to [0, 1] - } - } - - return true - } - - private val colID = 0 - private val colTimestamp = 1 - private val colCpuUsagePct = 2 - - override fun resolve(name: String): Int { - return when (name) { - resourceID -> colID - resourceStateTimestamp -> colTimestamp - resourceStateCpuUsagePct -> colCpuUsagePct - else -> -1 - } - } - - override fun isNull(index: Int): Boolean { - require(index in 0..colCpuUsagePct) { "Invalid column index" } - return false - } - - override fun getBoolean(index: Int): Boolean { - throw IllegalArgumentException("Invalid column") - } - - override fun getInt(index: Int): Int { - throw IllegalArgumentException("Invalid column") - } - - override fun getLong(index: Int): Long { - throw IllegalArgumentException("Invalid column") - } - - override fun getFloat(index: Int): Float { - throw IllegalArgumentException("Invalid column") - } - - override fun getDouble(index: Int): Double { - checkActive() - return when (index) { - colCpuUsagePct -> cpuUsagePct - else -> throw IllegalArgumentException("Invalid column") - } - } - - override fun getString(index: Int): String? { - checkActive() - return when (index) { - colID -> id - else -> throw IllegalArgumentException("Invalid column") - } - } - - override fun getUUID(index: Int): UUID? { - throw IllegalArgumentException("Invalid column") - } - - override fun getInstant(index: Int): Instant? { - checkActive() - return when (index) { - colTimestamp -> timestamp - else -> throw IllegalArgumentException("Invalid column") - } - } - - override fun getDuration(index: Int): Duration? { - throw IllegalArgumentException("Invalid column") - } - - override fun <T> getList( - index: Int, - elementType: Class<T>, - ): List<T>? { - throw IllegalArgumentException("Invalid column") - } - - override fun <K, V> getMap( - index: Int, - keyType: Class<K>, - valueType: Class<V>, - ): Map<K, V>? { - throw IllegalArgumentException("Invalid column") - } - - override fun <T> getSet( - index: Int, - elementType: Class<T>, - ): Set<T>? { - throw IllegalArgumentException("Invalid column") - } - - override fun close() { - parser.close() - } - - /** - * Helper method to check if the reader is active. - */ - private fun checkActive() { - check(isStarted && !parser.isClosed) { "No active row. Did you call nextRow()?" } - } - - /** - * Advance the parser until the next object start. - */ - private fun nextStart(): Boolean { - var token = parser.nextValue() - - while (token != null && token != JsonToken.START_OBJECT) { - token = parser.nextValue() - } - - return token != null - } - - /** - * State fields of the reader. - */ - private var id: String? = null - private var timestamp: Instant? = null - private var cpuUsagePct = Double.NaN - - /** - * Reset the state. - */ - private fun reset() { - id = null - timestamp = null - cpuUsagePct = Double.NaN - } - - companion object { - /** - * The [CsvSchema] that is used to parse the trace. - */ - private val schema = - CsvSchema.builder() - .addColumn("timestamp", CsvSchema.ColumnType.NUMBER) - .addColumn("vm id", CsvSchema.ColumnType.STRING) - .addColumn("CPU min cpu", CsvSchema.ColumnType.NUMBER) - .addColumn("CPU max cpu", CsvSchema.ColumnType.NUMBER) - .addColumn("CPU avg cpu", CsvSchema.ColumnType.NUMBER) - .setAllowComments(true) - .build() - } -} diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/azure/AzureResourceTableReader.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/azure/AzureResourceTableReader.kt deleted file mode 100644 index 55f26fa6..00000000 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/azure/AzureResourceTableReader.kt +++ /dev/null @@ -1,246 +0,0 @@ -/* - * Copyright (c) 2021 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 org.opendc.trace.azure - -import com.fasterxml.jackson.core.JsonToken -import com.fasterxml.jackson.dataformat.csv.CsvParser -import com.fasterxml.jackson.dataformat.csv.CsvSchema -import org.opendc.trace.TableReader -import org.opendc.trace.conv.resourceCpuCount -import org.opendc.trace.conv.resourceDuration -import org.opendc.trace.conv.resourceID -import org.opendc.trace.conv.resourceMemCapacity -import org.opendc.trace.conv.resourceSubmissionTime -import java.time.Duration -import java.time.Instant -import java.util.UUID - -/** - * A [TableReader] for the Azure v1 VM resources table. - */ -internal class AzureResourceTableReader(private val parser: CsvParser) : TableReader { - /** - * A flag to indicate whether a single row has been read already. - */ - private var isStarted = false - - init { - parser.schema = schema - } - - override fun nextRow(): Boolean { - if (!isStarted) { - isStarted = true - } - - reset() - - if (!nextStart()) { - return false - } - - while (true) { - val token = parser.nextValue() - - if (token == null || token == JsonToken.END_OBJECT) { - break - } - - when (parser.currentName) { - "vm id" -> id = parser.text - "timestamp vm created" -> startTime = Instant.ofEpochSecond(parser.longValue) - "timestamp vm deleted" -> stopTime = Instant.ofEpochSecond(parser.longValue) - "vm virtual core count" -> cpuCores = parser.intValue - "vm memory" -> memCapacity = parser.doubleValue * 1e6 // GB to KB - } - } - - return true - } - - private val colID = 0 - private val colStartTime = 1 - private val colStopTime = 2 - private val colCpuCount = 3 - private val colMemCapacity = 4 - - override fun resolve(name: String): Int { - return when (name) { - resourceID -> colID - resourceSubmissionTime -> colStartTime - resourceDuration -> colStopTime - resourceCpuCount -> colCpuCount - resourceMemCapacity -> colMemCapacity - else -> -1 - } - } - - override fun isNull(index: Int): Boolean { - require(index in 0..colMemCapacity) { "Invalid column index" } - return false - } - - override fun getBoolean(index: Int): Boolean { - throw IllegalArgumentException("Invalid column") - } - - override fun getInt(index: Int): Int { - checkActive() - return when (index) { - colCpuCount -> cpuCores - else -> throw IllegalArgumentException("Invalid column") - } - } - - override fun getLong(index: Int): Long { - checkActive() - return when (index) { - colCpuCount -> cpuCores.toLong() - else -> throw IllegalArgumentException("Invalid column") - } - } - - override fun getFloat(index: Int): Float { - throw IllegalArgumentException("Invalid column") - } - - override fun getDouble(index: Int): Double { - checkActive() - return when (index) { - colMemCapacity -> memCapacity - else -> throw IllegalArgumentException("Invalid column") - } - } - - override fun getString(index: Int): String? { - checkActive() - return when (index) { - colID -> id - else -> throw IllegalArgumentException("Invalid column") - } - } - - override fun getUUID(index: Int): UUID? { - throw IllegalArgumentException("Invalid column") - } - - override fun getInstant(index: Int): Instant? { - checkActive() - return when (index) { - colStartTime -> startTime - colStopTime -> stopTime - else -> throw IllegalArgumentException("Invalid column") - } - } - - override fun getDuration(index: Int): Duration? { - throw IllegalArgumentException("Invalid column") - } - - override fun <T> getList( - index: Int, - elementType: Class<T>, - ): List<T>? { - throw IllegalArgumentException("Invalid column") - } - - override fun <T> getSet( - index: Int, - elementType: Class<T>, - ): Set<T>? { - throw IllegalArgumentException("Invalid column") - } - - override fun <K, V> getMap( - index: Int, - keyType: Class<K>, - valueType: Class<V>, - ): Map<K, V>? { - throw IllegalArgumentException("Invalid column") - } - - override fun close() { - parser.close() - } - - /** - * Helper method to check if the reader is active. - */ - private fun checkActive() { - check(isStarted && !parser.isClosed) { "No active row. Did you call nextRow()?" } - } - - /** - * Advance the parser until the next object start. - */ - private fun nextStart(): Boolean { - var token = parser.nextValue() - - while (token != null && token != JsonToken.START_OBJECT) { - token = parser.nextValue() - } - - return token != null - } - - /** - * State fields of the reader. - */ - private var id: String? = null - private var startTime: Instant? = null - private var stopTime: Instant? = null - private var cpuCores = -1 - private var memCapacity = Double.NaN - - /** - * Reset the state. - */ - private fun reset() { - id = null - startTime = null - stopTime = null - cpuCores = -1 - memCapacity = Double.NaN - } - - companion object { - /** - * The [CsvSchema] that is used to parse the trace. - */ - private val schema = - CsvSchema.builder() - .addColumn("vm id", CsvSchema.ColumnType.NUMBER) - .addColumn("subscription id", CsvSchema.ColumnType.STRING) - .addColumn("deployment id", CsvSchema.ColumnType.NUMBER) - .addColumn("timestamp vm created", CsvSchema.ColumnType.NUMBER) - .addColumn("timestamp vm deleted", CsvSchema.ColumnType.NUMBER) - .addColumn("max cpu", CsvSchema.ColumnType.NUMBER) - .addColumn("avg cpu", CsvSchema.ColumnType.NUMBER) - .addColumn("p95 cpu", CsvSchema.ColumnType.NUMBER) - .addColumn("vm category", CsvSchema.ColumnType.NUMBER) - .addColumn("vm virtual core count", CsvSchema.ColumnType.NUMBER) - .addColumn("vm memory", CsvSchema.ColumnType.NUMBER) - .setAllowComments(true) - .build() - } -} diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/azure/AzureTraceFormat.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/azure/AzureTraceFormat.kt deleted file mode 100644 index 7ce1c11a..00000000 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/azure/AzureTraceFormat.kt +++ /dev/null @@ -1,147 +0,0 @@ -/* - * Copyright (c) 2021 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 org.opendc.trace.azure - -import com.fasterxml.jackson.dataformat.csv.CsvFactory -import com.fasterxml.jackson.dataformat.csv.CsvParser -import org.opendc.trace.TableColumn -import org.opendc.trace.TableColumnType -import org.opendc.trace.TableReader -import org.opendc.trace.TableWriter -import org.opendc.trace.conv.TABLE_RESOURCES -import org.opendc.trace.conv.TABLE_RESOURCE_STATES -import org.opendc.trace.conv.resourceCpuCount -import org.opendc.trace.conv.resourceDuration -import org.opendc.trace.conv.resourceID -import org.opendc.trace.conv.resourceMemCapacity -import org.opendc.trace.conv.resourceStateCpuUsagePct -import org.opendc.trace.conv.resourceStateTimestamp -import org.opendc.trace.conv.resourceSubmissionTime -import org.opendc.trace.spi.TableDetails -import org.opendc.trace.spi.TraceFormat -import org.opendc.trace.util.CompositeTableReader -import java.nio.file.Files -import java.nio.file.Path -import java.util.stream.Collectors -import java.util.zip.GZIPInputStream -import kotlin.io.path.inputStream -import kotlin.io.path.name - -/** - * A format implementation for the Azure v1 format. - */ -public class AzureTraceFormat : TraceFormat { - /** - * The name of this trace format. - */ - override val name: String = "azure" - - /** - * The [CsvFactory] used to create the parser. - */ - private val factory = - CsvFactory() - .enable(CsvParser.Feature.ALLOW_COMMENTS) - .enable(CsvParser.Feature.TRIM_SPACES) - - override fun create(path: Path) { - throw UnsupportedOperationException("Writing not supported for this format") - } - - override fun getTables(path: Path): List<String> = listOf(TABLE_RESOURCES, TABLE_RESOURCE_STATES) - - override fun getDetails( - path: Path, - table: String, - ): TableDetails { - return when (table) { - TABLE_RESOURCES -> - TableDetails( - listOf( - TableColumn(resourceID, TableColumnType.String), - TableColumn(resourceSubmissionTime, TableColumnType.Instant), - TableColumn(resourceDuration, TableColumnType.Instant), - TableColumn(resourceCpuCount, TableColumnType.Int), - TableColumn(resourceMemCapacity, TableColumnType.Double), - ), - ) - TABLE_RESOURCE_STATES -> - TableDetails( - listOf( - TableColumn(resourceID, TableColumnType.String), - TableColumn(resourceStateTimestamp, TableColumnType.Instant), - TableColumn(resourceStateCpuUsagePct, TableColumnType.Double), - ), - ) - else -> throw IllegalArgumentException("Table $table not supported") - } - } - - override fun newReader( - path: Path, - table: String, - projection: List<String>?, - ): TableReader { - return when (table) { - TABLE_RESOURCES -> { - val stream = GZIPInputStream(path.resolve("vmtable/vmtable.csv.gz").inputStream()) - AzureResourceTableReader(factory.createParser(stream)) - } - TABLE_RESOURCE_STATES -> newResourceStateReader(path) - else -> throw IllegalArgumentException("Table $table not supported") - } - } - - override fun newWriter( - path: Path, - table: String, - ): TableWriter { - throw UnsupportedOperationException("Writing not supported for this format") - } - - /** - * Construct a [TableReader] for reading over all VM CPU readings. - */ - private fun newResourceStateReader(path: Path): TableReader { - val partitions = - Files.walk(path.resolve("vm_cpu_readings"), 1) - .filter { !Files.isDirectory(it) && it.name.endsWith(".csv.gz") } - .collect(Collectors.toMap({ it.name.removeSuffix(".csv.gz") }, { it })) - .toSortedMap() - val it = partitions.iterator() - - return object : CompositeTableReader() { - override fun nextReader(): TableReader? { - return if (it.hasNext()) { - val (_, partPath) = it.next() - val stream = GZIPInputStream(partPath.inputStream()) - return AzureResourceStateTableReader(factory.createParser(stream)) - } else { - null - } - } - - override fun toString(): String = "AzureCompositeTableReader" - } - } -} diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/bitbrains/BitbrainsExResourceStateTableReader.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/bitbrains/BitbrainsExResourceStateTableReader.kt deleted file mode 100644 index 8387d1ed..00000000 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/bitbrains/BitbrainsExResourceStateTableReader.kt +++ /dev/null @@ -1,292 +0,0 @@ -/* - * Copyright (c) 2021 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 org.opendc.trace.bitbrains - -import org.opendc.trace.TableReader -import org.opendc.trace.conv.resourceClusterID -import org.opendc.trace.conv.resourceCpuCapacity -import org.opendc.trace.conv.resourceCpuCount -import org.opendc.trace.conv.resourceID -import org.opendc.trace.conv.resourceMemCapacity -import org.opendc.trace.conv.resourceStateCpuDemand -import org.opendc.trace.conv.resourceStateCpuReadyPct -import org.opendc.trace.conv.resourceStateCpuUsage -import org.opendc.trace.conv.resourceStateCpuUsagePct -import org.opendc.trace.conv.resourceStateDiskRead -import org.opendc.trace.conv.resourceStateDiskWrite -import org.opendc.trace.conv.resourceStateTimestamp -import java.io.BufferedReader -import java.time.Duration -import java.time.Instant -import java.util.UUID - -/** - * A [TableReader] for the Bitbrains resource state table. - */ -internal class BitbrainsExResourceStateTableReader(private val reader: BufferedReader) : TableReader { - private var state = State.Pending - - override fun nextRow(): Boolean { - val state = state - if (state == State.Closed) { - return false - } else if (state == State.Pending) { - this.state = State.Active - } - - reset() - - var line: String? - var num = 0 - - while (true) { - line = reader.readLine() - - if (line == null) { - this.state = State.Closed - return false - } - - num++ - - if (line[0] == '#' || line.isBlank()) { - // Ignore empty lines or comments - continue - } - - break - } - - line = line!!.trim() - - val length = line.length - var col = 0 - var start: Int - var end = 0 - - while (end < length) { - // Trim all whitespace before the field - start = end - while (start < length && line[start].isWhitespace()) { - start++ - } - - end = line.indexOf(' ', start) - - if (end < 0) { - end = length - } - - val field = line.subSequence(start, end) as String - when (col++) { - colTimestamp -> timestamp = Instant.ofEpochSecond(field.toLong(10)) - colCpuUsage -> cpuUsage = field.toDouble() - colCpuDemand -> cpuDemand = field.toDouble() - colDiskRead -> diskRead = field.toDouble() - colDiskWrite -> diskWrite = field.toDouble() - colClusterID -> cluster = field.trim() - colNcpus -> cpuCores = field.toInt(10) - colCpuReadyPct -> cpuReadyPct = field.toDouble() - colPoweredOn -> poweredOn = field.toInt(10) == 1 - colCpuCapacity -> cpuCapacity = field.toDouble() - colID -> id = field.trim() - colMemCapacity -> memCapacity = field.toDouble() * 1000 // Convert from MB to KB - } - } - - return true - } - - override fun resolve(name: String): Int { - return when (name) { - resourceID -> colID - resourceClusterID -> colClusterID - resourceStateTimestamp -> colTimestamp - resourceCpuCount -> colNcpus - resourceCpuCapacity -> colCpuCapacity - resourceStateCpuUsage -> colCpuUsage - resourceStateCpuUsagePct -> colCpuUsagePct - resourceStateCpuDemand -> colCpuDemand - resourceStateCpuReadyPct -> colCpuReadyPct - resourceMemCapacity -> colMemCapacity - resourceStateDiskRead -> colDiskRead - resourceStateDiskWrite -> colDiskWrite - else -> -1 - } - } - - override fun isNull(index: Int): Boolean { - require(index in 0 until colMax) { "Invalid column index" } - return false - } - - override fun getBoolean(index: Int): Boolean { - check(state == State.Active) { "No active row" } - return when (index) { - colPoweredOn -> poweredOn - else -> throw IllegalArgumentException("Invalid column") - } - } - - override fun getInt(index: Int): Int { - check(state == State.Active) { "No active row" } - return when (index) { - colNcpus -> cpuCores - else -> throw IllegalArgumentException("Invalid column") - } - } - - override fun getLong(index: Int): Long { - throw IllegalArgumentException("Invalid column") - } - - override fun getFloat(index: Int): Float { - throw IllegalArgumentException("Invalid column") - } - - override fun getDouble(index: Int): Double { - check(state == State.Active) { "No active row" } - return when (index) { - colCpuCapacity -> cpuCapacity - colCpuUsage -> cpuUsage - colCpuUsagePct -> cpuUsage / cpuCapacity - colCpuReadyPct -> cpuReadyPct - colCpuDemand -> cpuDemand - colMemCapacity -> memCapacity - colDiskRead -> diskRead - colDiskWrite -> diskWrite - else -> throw IllegalArgumentException("Invalid column") - } - } - - override fun getString(index: Int): String? { - check(state == State.Active) { "No active row" } - return when (index) { - colID -> id - colClusterID -> cluster - else -> throw IllegalArgumentException("Invalid column") - } - } - - override fun getUUID(index: Int): UUID? { - throw IllegalArgumentException("Invalid column") - } - - override fun getInstant(index: Int): Instant? { - check(state == State.Active) { "No active row" } - return when (index) { - colTimestamp -> timestamp - else -> throw IllegalArgumentException("Invalid column") - } - } - - override fun getDuration(index: Int): Duration? { - throw IllegalArgumentException("Invalid column") - } - - override fun <T> getList( - index: Int, - elementType: Class<T>, - ): List<T>? { - throw IllegalArgumentException("Invalid column") - } - - override fun <T> getSet( - index: Int, - elementType: Class<T>, - ): Set<T>? { - throw IllegalArgumentException("Invalid column") - } - - override fun <K, V> getMap( - index: Int, - keyType: Class<K>, - valueType: Class<V>, - ): Map<K, V>? { - throw IllegalArgumentException("Invalid column") - } - - override fun close() { - reader.close() - reset() - state = State.Closed - } - - /** - * State fields of the reader. - */ - private var id: String? = null - private var cluster: String? = null - private var timestamp: Instant? = null - private var cpuCores = -1 - private var cpuCapacity = Double.NaN - private var cpuUsage = Double.NaN - private var cpuDemand = Double.NaN - private var cpuReadyPct = Double.NaN - private var memCapacity = Double.NaN - private var diskRead = Double.NaN - private var diskWrite = Double.NaN - private var poweredOn: Boolean = false - - /** - * Reset the state of the reader. - */ - private fun reset() { - id = null - timestamp = null - cluster = null - cpuCores = -1 - cpuCapacity = Double.NaN - cpuUsage = Double.NaN - cpuDemand = Double.NaN - cpuReadyPct = Double.NaN - memCapacity = Double.NaN - diskRead = Double.NaN - diskWrite = Double.NaN - poweredOn = false - } - - /** - * Default column indices for the extended Bitbrains format. - */ - private val colTimestamp = 0 - private val colCpuUsage = 1 - private val colCpuDemand = 2 - private val colDiskRead = 4 - private val colDiskWrite = 6 - private val colClusterID = 10 - private val colNcpus = 12 - private val colCpuReadyPct = 13 - private val colPoweredOn = 14 - private val colCpuCapacity = 18 - private val colID = 19 - private val colMemCapacity = 20 - private val colCpuUsagePct = 21 - private val colMax = colCpuUsagePct + 1 - - private enum class State { - Pending, - Active, - Closed, - } -} diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/bitbrains/BitbrainsExTraceFormat.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/bitbrains/BitbrainsExTraceFormat.kt deleted file mode 100644 index 6115953f..00000000 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/bitbrains/BitbrainsExTraceFormat.kt +++ /dev/null @@ -1,135 +0,0 @@ -/* - * Copyright (c) 2021 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 org.opendc.trace.bitbrains - -import org.opendc.trace.TableColumn -import org.opendc.trace.TableColumnType -import org.opendc.trace.TableReader -import org.opendc.trace.TableWriter -import org.opendc.trace.conv.TABLE_RESOURCE_STATES -import org.opendc.trace.conv.resourceClusterID -import org.opendc.trace.conv.resourceCpuCapacity -import org.opendc.trace.conv.resourceCpuCount -import org.opendc.trace.conv.resourceID -import org.opendc.trace.conv.resourceMemCapacity -import org.opendc.trace.conv.resourceStateCpuDemand -import org.opendc.trace.conv.resourceStateCpuReadyPct -import org.opendc.trace.conv.resourceStateCpuUsage -import org.opendc.trace.conv.resourceStateCpuUsagePct -import org.opendc.trace.conv.resourceStateDiskRead -import org.opendc.trace.conv.resourceStateDiskWrite -import org.opendc.trace.conv.resourceStateTimestamp -import org.opendc.trace.spi.TableDetails -import org.opendc.trace.spi.TraceFormat -import org.opendc.trace.util.CompositeTableReader -import java.nio.file.Files -import java.nio.file.Path -import java.util.stream.Collectors -import kotlin.io.path.bufferedReader -import kotlin.io.path.extension -import kotlin.io.path.nameWithoutExtension - -/** - * A format implementation for the extended Bitbrains trace format. - */ -public class BitbrainsExTraceFormat : TraceFormat { - /** - * The name of this trace format. - */ - override val name: String = "bitbrains-ex" - - override fun create(path: Path) { - throw UnsupportedOperationException("Writing not supported for this format") - } - - override fun getTables(path: Path): List<String> = listOf(TABLE_RESOURCE_STATES) - - override fun getDetails( - path: Path, - table: String, - ): TableDetails { - return when (table) { - TABLE_RESOURCE_STATES -> - TableDetails( - listOf( - TableColumn(resourceID, TableColumnType.String), - TableColumn(resourceClusterID, TableColumnType.String), - TableColumn(resourceStateTimestamp, TableColumnType.Instant), - TableColumn(resourceCpuCount, TableColumnType.Int), - TableColumn(resourceCpuCapacity, TableColumnType.Double), - TableColumn(resourceStateCpuUsage, TableColumnType.Double), - TableColumn(resourceStateCpuUsagePct, TableColumnType.Double), - TableColumn(resourceStateCpuDemand, TableColumnType.Double), - TableColumn(resourceStateCpuReadyPct, TableColumnType.Double), - TableColumn(resourceMemCapacity, TableColumnType.Double), - TableColumn(resourceStateDiskRead, TableColumnType.Double), - TableColumn(resourceStateDiskWrite, TableColumnType.Double), - ), - ) - else -> throw IllegalArgumentException("Table $table not supported") - } - } - - override fun newReader( - path: Path, - table: String, - projection: List<String>?, - ): TableReader { - return when (table) { - TABLE_RESOURCE_STATES -> newResourceStateReader(path) - else -> throw IllegalArgumentException("Table $table not supported") - } - } - - override fun newWriter( - path: Path, - table: String, - ): TableWriter { - throw UnsupportedOperationException("Writing not supported for this format") - } - - /** - * Construct a [TableReader] for reading over all resource state partitions. - */ - private fun newResourceStateReader(path: Path): TableReader { - val partitions = - Files.walk(path, 1) - .filter { !Files.isDirectory(it) && it.extension == "txt" } - .collect(Collectors.toMap({ it.nameWithoutExtension }, { it })) - .toSortedMap() - val it = partitions.iterator() - - return object : CompositeTableReader() { - override fun nextReader(): TableReader? { - return if (it.hasNext()) { - val (_, partPath) = it.next() - return BitbrainsExResourceStateTableReader(partPath.bufferedReader()) - } else { - null - } - } - - override fun toString(): String = "BitbrainsExCompositeTableReader" - } - } -} diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/bitbrains/BitbrainsResourceStateTableReader.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/bitbrains/BitbrainsResourceStateTableReader.kt deleted file mode 100644 index e264fccb..00000000 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/bitbrains/BitbrainsResourceStateTableReader.kt +++ /dev/null @@ -1,365 +0,0 @@ -/* - * Copyright (c) 2021 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 org.opendc.trace.bitbrains - -import com.fasterxml.jackson.core.JsonParseException -import com.fasterxml.jackson.core.JsonToken -import com.fasterxml.jackson.dataformat.csv.CsvParser -import com.fasterxml.jackson.dataformat.csv.CsvSchema -import org.opendc.trace.TableReader -import org.opendc.trace.conv.resourceCpuCapacity -import org.opendc.trace.conv.resourceCpuCount -import org.opendc.trace.conv.resourceID -import org.opendc.trace.conv.resourceMemCapacity -import org.opendc.trace.conv.resourceStateCpuUsage -import org.opendc.trace.conv.resourceStateCpuUsagePct -import org.opendc.trace.conv.resourceStateDiskRead -import org.opendc.trace.conv.resourceStateDiskWrite -import org.opendc.trace.conv.resourceStateMemUsage -import org.opendc.trace.conv.resourceStateNetRx -import org.opendc.trace.conv.resourceStateNetTx -import org.opendc.trace.conv.resourceStateTimestamp -import java.text.NumberFormat -import java.time.Duration -import java.time.Instant -import java.time.LocalDateTime -import java.time.ZoneOffset -import java.time.format.DateTimeFormatter -import java.time.format.DateTimeParseException -import java.util.Locale -import java.util.UUID - -/** - * A [TableReader] for the Bitbrains resource state table. - */ -internal class BitbrainsResourceStateTableReader(private val partition: String, private val parser: CsvParser) : TableReader { - /** - * A flag to indicate whether a single row has been read already. - */ - private var isStarted = false - - /** - * The [DateTimeFormatter] used to parse the timestamps in case of the Materna trace. - */ - private val formatter = DateTimeFormatter.ofPattern("dd.MM.yyyy HH:mm:ss") - - /** - * The type of timestamps in the trace. - */ - private var timestampType: TimestampType = TimestampType.UNDECIDED - - /** - * The [NumberFormat] used to parse doubles containing a comma. - */ - private val nf = NumberFormat.getInstance(Locale.GERMAN) - - /** - * A flag to indicate that the trace contains decimals with a comma separator. - */ - private var usesCommaDecimalSeparator = false - - init { - parser.schema = schema - } - - override fun nextRow(): Boolean { - if (!isStarted) { - isStarted = true - } - - // Reset the row state - reset() - - if (!nextStart()) { - return false - } - - while (true) { - val token = parser.nextValue() - - if (token == null || token == JsonToken.END_OBJECT) { - break - } - - when (parser.currentName) { - "Timestamp [ms]" -> { - timestamp = - when (timestampType) { - TimestampType.UNDECIDED -> { - try { - val res = LocalDateTime.parse(parser.text, formatter).toInstant(ZoneOffset.UTC) - timestampType = TimestampType.DATE_TIME - res - } catch (e: DateTimeParseException) { - timestampType = TimestampType.EPOCH_MILLIS - Instant.ofEpochSecond(parser.longValue) - } - } - TimestampType.DATE_TIME -> LocalDateTime.parse(parser.text, formatter).toInstant(ZoneOffset.UTC) - TimestampType.EPOCH_MILLIS -> Instant.ofEpochSecond(parser.longValue) - } - } - "CPU cores" -> cpuCores = parser.intValue - "CPU capacity provisioned [MHZ]" -> cpuCapacity = parseSafeDouble() - "CPU usage [MHZ]" -> cpuUsage = parseSafeDouble() - "CPU usage [%]" -> cpuUsagePct = parseSafeDouble() / 100.0 // Convert to range [0, 1] - "Memory capacity provisioned [KB]" -> memCapacity = parseSafeDouble() - "Memory usage [KB]" -> memUsage = parseSafeDouble() - "Disk read throughput [KB/s]" -> diskRead = parseSafeDouble() - "Disk write throughput [KB/s]" -> diskWrite = parseSafeDouble() - "Network received throughput [KB/s]" -> netReceived = parseSafeDouble() - "Network transmitted throughput [KB/s]" -> netTransmitted = parseSafeDouble() - } - } - - return true - } - - private val colTimestamp = 0 - private val colCpuCount = 1 - private val colCpuCapacity = 2 - private val colCpuUsage = 3 - private val colCpuUsagePct = 4 - private val colMemCapacity = 5 - private val colMemUsage = 6 - private val colDiskRead = 7 - private val colDiskWrite = 8 - private val colNetRx = 9 - private val colNetTx = 10 - private val colID = 11 - - override fun resolve(name: String): Int { - return when (name) { - resourceID -> colID - resourceStateTimestamp -> colTimestamp - resourceCpuCount -> colCpuCount - resourceCpuCapacity -> colCpuCapacity - resourceStateCpuUsage -> colCpuUsage - resourceStateCpuUsagePct -> colCpuUsagePct - resourceMemCapacity -> colMemCapacity - resourceStateMemUsage -> colMemUsage - resourceStateDiskRead -> colDiskRead - resourceStateDiskWrite -> colDiskWrite - resourceStateNetRx -> colNetRx - resourceStateNetTx -> colNetTx - else -> -1 - } - } - - override fun isNull(index: Int): Boolean { - require(index in 0..colID) { "Invalid column index" } - return false - } - - override fun getBoolean(index: Int): Boolean { - throw IllegalArgumentException("Invalid column") - } - - override fun getInt(index: Int): Int { - checkActive() - return when (index) { - colCpuCount -> cpuCores - else -> throw IllegalArgumentException("Invalid column") - } - } - - override fun getLong(index: Int): Long { - throw IllegalArgumentException("Invalid column") - } - - override fun getFloat(index: Int): Float { - throw IllegalArgumentException("Invalid column") - } - - override fun getDouble(index: Int): Double { - checkActive() - return when (index) { - colCpuCapacity -> cpuCapacity - colCpuUsage -> cpuUsage - colCpuUsagePct -> cpuUsagePct - colMemCapacity -> memCapacity - colMemUsage -> memUsage - colDiskRead -> diskRead - colDiskWrite -> diskWrite - colNetRx -> netReceived - colNetTx -> netTransmitted - else -> throw IllegalArgumentException("Invalid column") - } - } - - override fun getString(index: Int): String { - checkActive() - return when (index) { - colID -> partition - else -> throw IllegalArgumentException("Invalid column") - } - } - - override fun getUUID(index: Int): UUID? { - throw IllegalArgumentException("Invalid column") - } - - override fun getInstant(index: Int): Instant? { - checkActive() - return when (index) { - colTimestamp -> timestamp - else -> throw IllegalArgumentException("Invalid column") - } - } - - override fun getDuration(index: Int): Duration? { - throw IllegalArgumentException("Invalid column") - } - - override fun <T> getList( - index: Int, - elementType: Class<T>, - ): List<T>? { - throw IllegalArgumentException("Invalid column") - } - - override fun <T> getSet( - index: Int, - elementType: Class<T>, - ): Set<T>? { - throw IllegalArgumentException("Invalid column") - } - - override fun <K, V> getMap( - index: Int, - keyType: Class<K>, - valueType: Class<V>, - ): Map<K, V>? { - throw IllegalArgumentException("Invalid column") - } - - override fun close() { - parser.close() - } - - /** - * Helper method to check if the reader is active. - */ - private fun checkActive() { - check(isStarted && !parser.isClosed) { "No active row. Did you call nextRow()?" } - } - - /** - * Advance the parser until the next object start. - */ - private fun nextStart(): Boolean { - var token = parser.nextValue() - - while (token != null && token != JsonToken.START_OBJECT) { - token = parser.nextValue() - } - - return token != null - } - - /** - * Try to parse the current value safely as double. - */ - private fun parseSafeDouble(): Double { - if (!usesCommaDecimalSeparator) { - try { - return parser.doubleValue - } catch (e: JsonParseException) { - usesCommaDecimalSeparator = true - } - } - - val text = parser.text - if (text.isBlank()) { - return 0.0 - } - - return nf.parse(text).toDouble() - } - - /** - * State fields of the reader. - */ - private var timestamp: Instant? = null - private var cpuCores = -1 - private var cpuCapacity = Double.NaN - private var cpuUsage = Double.NaN - private var cpuUsagePct = Double.NaN - private var memCapacity = Double.NaN - private var memUsage = Double.NaN - private var diskRead = Double.NaN - private var diskWrite = Double.NaN - private var netReceived = Double.NaN - private var netTransmitted = Double.NaN - - /** - * Reset the state. - */ - private fun reset() { - timestamp = null - cpuCores = -1 - cpuCapacity = Double.NaN - cpuUsage = Double.NaN - cpuUsagePct = Double.NaN - memCapacity = Double.NaN - memUsage = Double.NaN - diskRead = Double.NaN - diskWrite = Double.NaN - netReceived = Double.NaN - netTransmitted = Double.NaN - } - - /** - * The type of the timestamp in the trace. - */ - private enum class TimestampType { - UNDECIDED, - DATE_TIME, - EPOCH_MILLIS, - } - - companion object { - /** - * The [CsvSchema] that is used to parse the trace. - */ - private val schema = - CsvSchema.builder() - .addColumn("Timestamp [ms]", CsvSchema.ColumnType.NUMBER_OR_STRING) - .addColumn("CPU cores", CsvSchema.ColumnType.NUMBER) - .addColumn("CPU capacity provisioned [MHZ]", CsvSchema.ColumnType.NUMBER) - .addColumn("CPU usage [MHZ]", CsvSchema.ColumnType.NUMBER) - .addColumn("CPU usage [%]", CsvSchema.ColumnType.NUMBER) - .addColumn("Memory capacity provisioned [KB]", CsvSchema.ColumnType.NUMBER) - .addColumn("Memory usage [KB]", CsvSchema.ColumnType.NUMBER) - .addColumn("Memory usage [%]", CsvSchema.ColumnType.NUMBER) - .addColumn("Disk read throughput [KB/s]", CsvSchema.ColumnType.NUMBER) - .addColumn("Disk write throughput [KB/s]", CsvSchema.ColumnType.NUMBER) - .addColumn("Disk size [GB]", CsvSchema.ColumnType.NUMBER) - .addColumn("Network received throughput [KB/s]", CsvSchema.ColumnType.NUMBER) - .addColumn("Network transmitted throughput [KB/s]", CsvSchema.ColumnType.NUMBER) - .setAllowComments(true) - .setUseHeader(true) - .setColumnSeparator(';') - .build() - } -} diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/bitbrains/BitbrainsResourceTableReader.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/bitbrains/BitbrainsResourceTableReader.kt deleted file mode 100644 index a12785f0..00000000 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/bitbrains/BitbrainsResourceTableReader.kt +++ /dev/null @@ -1,175 +0,0 @@ -/* - * Copyright (c) 2021 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 org.opendc.trace.bitbrains - -import com.fasterxml.jackson.dataformat.csv.CsvFactory -import org.opendc.trace.TableReader -import org.opendc.trace.conv.resourceID -import java.nio.file.Path -import java.time.Duration -import java.time.Instant -import java.util.UUID - -/** - * A [TableReader] for the Bitbrains resource table. - */ -internal class BitbrainsResourceTableReader(private val factory: CsvFactory, vms: Map<String, Path>) : TableReader { - /** - * An iterator to iterate over the resource entries. - */ - private val it = vms.iterator() - - /** - * The state of the reader. - */ - private var state = State.Pending - - override fun nextRow(): Boolean { - if (state == State.Pending) { - state = State.Active - } - - reset() - - while (it.hasNext()) { - val (name, path) = it.next() - - val parser = factory.createParser(path.toFile()) - val reader = BitbrainsResourceStateTableReader(name, parser) - val idCol = reader.resolve(resourceID) - - try { - if (!reader.nextRow()) { - continue - } - - id = reader.getString(idCol) - return true - } finally { - reader.close() - } - } - - state = State.Closed - return false - } - - private val colID = 0 - - override fun resolve(name: String): Int { - return when (name) { - resourceID -> colID - else -> -1 - } - } - - override fun isNull(index: Int): Boolean { - require(index in 0..colID) { "Invalid column index" } - return false - } - - override fun getBoolean(index: Int): Boolean { - throw IllegalArgumentException("Invalid column") - } - - override fun getInt(index: Int): Int { - throw IllegalArgumentException("Invalid column") - } - - override fun getLong(index: Int): Long { - throw IllegalArgumentException("Invalid column") - } - - override fun getFloat(index: Int): Float { - throw IllegalArgumentException("Invalid column") - } - - override fun getDouble(index: Int): Double { - throw IllegalArgumentException("Invalid column") - } - - override fun getString(index: Int): String? { - check(state == State.Active) { "No active row" } - return when (index) { - colID -> id - else -> throw IllegalArgumentException("Invalid column") - } - } - - override fun getUUID(index: Int): UUID? { - throw IllegalArgumentException("Invalid column") - } - - override fun getInstant(index: Int): Instant? { - throw IllegalArgumentException("Invalid column") - } - - override fun getDuration(index: Int): Duration? { - throw IllegalArgumentException("Invalid column") - } - - override fun <T> getList( - index: Int, - elementType: Class<T>, - ): List<T>? { - throw IllegalArgumentException("Invalid column") - } - - override fun <T> getSet( - index: Int, - elementType: Class<T>, - ): Set<T>? { - throw IllegalArgumentException("Invalid column") - } - - override fun <K, V> getMap( - index: Int, - keyType: Class<K>, - valueType: Class<V>, - ): Map<K, V>? { - throw IllegalArgumentException("Invalid column") - } - - override fun close() { - reset() - state = State.Closed - } - - /** - * State fields of the reader. - */ - private var id: String? = null - - /** - * Reset the state of the reader. - */ - private fun reset() { - id = null - } - - private enum class State { - Pending, - Active, - Closed, - } -} diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/bitbrains/BitbrainsTraceFormat.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/bitbrains/BitbrainsTraceFormat.kt deleted file mode 100644 index 23853077..00000000 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/bitbrains/BitbrainsTraceFormat.kt +++ /dev/null @@ -1,159 +0,0 @@ -/* - * Copyright (c) 2021 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 org.opendc.trace.bitbrains - -import com.fasterxml.jackson.dataformat.csv.CsvFactory -import com.fasterxml.jackson.dataformat.csv.CsvParser -import org.opendc.trace.TableColumn -import org.opendc.trace.TableColumnType -import org.opendc.trace.TableReader -import org.opendc.trace.TableWriter -import org.opendc.trace.conv.TABLE_RESOURCES -import org.opendc.trace.conv.TABLE_RESOURCE_STATES -import org.opendc.trace.conv.resourceCpuCapacity -import org.opendc.trace.conv.resourceCpuCount -import org.opendc.trace.conv.resourceID -import org.opendc.trace.conv.resourceMemCapacity -import org.opendc.trace.conv.resourceStateCpuUsage -import org.opendc.trace.conv.resourceStateCpuUsagePct -import org.opendc.trace.conv.resourceStateDiskRead -import org.opendc.trace.conv.resourceStateDiskWrite -import org.opendc.trace.conv.resourceStateMemUsage -import org.opendc.trace.conv.resourceStateNetRx -import org.opendc.trace.conv.resourceStateNetTx -import org.opendc.trace.conv.resourceStateTimestamp -import org.opendc.trace.spi.TableDetails -import org.opendc.trace.spi.TraceFormat -import org.opendc.trace.util.CompositeTableReader -import java.nio.file.Files -import java.nio.file.Path -import java.util.stream.Collectors -import kotlin.io.path.extension -import kotlin.io.path.nameWithoutExtension - -/** - * A format implementation for the GWF trace format. - */ -public class BitbrainsTraceFormat : TraceFormat { - /** - * The name of this trace format. - */ - override val name: String = "bitbrains" - - /** - * The [CsvFactory] used to create the parser. - */ - private val factory = - CsvFactory() - .enable(CsvParser.Feature.ALLOW_COMMENTS) - .enable(CsvParser.Feature.TRIM_SPACES) - - override fun create(path: Path) { - throw UnsupportedOperationException("Writing not supported for this format") - } - - override fun getTables(path: Path): List<String> = listOf(TABLE_RESOURCES, TABLE_RESOURCE_STATES) - - override fun getDetails( - path: Path, - table: String, - ): TableDetails { - return when (table) { - TABLE_RESOURCES -> - TableDetails( - listOf( - TableColumn(resourceID, TableColumnType.String), - ), - ) - TABLE_RESOURCE_STATES -> - TableDetails( - listOf( - TableColumn(resourceID, TableColumnType.String), - TableColumn(resourceStateTimestamp, TableColumnType.Instant), - TableColumn(resourceCpuCount, TableColumnType.Int), - TableColumn(resourceCpuCapacity, TableColumnType.Double), - TableColumn(resourceStateCpuUsage, TableColumnType.Double), - TableColumn(resourceStateCpuUsagePct, TableColumnType.Double), - TableColumn(resourceMemCapacity, TableColumnType.Double), - TableColumn(resourceStateMemUsage, TableColumnType.Double), - TableColumn(resourceStateDiskRead, TableColumnType.Double), - TableColumn(resourceStateDiskWrite, TableColumnType.Double), - TableColumn(resourceStateNetRx, TableColumnType.Double), - TableColumn(resourceStateNetTx, TableColumnType.Double), - ), - ) - else -> throw IllegalArgumentException("Table $table not supported") - } - } - - override fun newReader( - path: Path, - table: String, - projection: List<String>?, - ): TableReader { - return when (table) { - TABLE_RESOURCES -> { - val vms = - Files.walk(path, 1) - .filter { !Files.isDirectory(it) && it.extension == "csv" } - .collect(Collectors.toMap({ it.nameWithoutExtension }, { it })) - .toSortedMap() - BitbrainsResourceTableReader(factory, vms) - } - TABLE_RESOURCE_STATES -> newResourceStateReader(path) - else -> throw IllegalArgumentException("Table $table not supported") - } - } - - override fun newWriter( - path: Path, - table: String, - ): TableWriter { - throw UnsupportedOperationException("Writing not supported for this format") - } - - /** - * Construct a [TableReader] for reading over all resource state partitions. - */ - private fun newResourceStateReader(path: Path): TableReader { - val partitions = - Files.walk(path, 1) - .filter { !Files.isDirectory(it) && it.extension == "csv" } - .collect(Collectors.toMap({ it.nameWithoutExtension }, { it })) - .toSortedMap() - val it = partitions.iterator() - - return object : CompositeTableReader() { - override fun nextReader(): TableReader? { - return if (it.hasNext()) { - val (partition, partPath) = it.next() - return BitbrainsResourceStateTableReader(partition, factory.createParser(partPath.toFile())) - } else { - null - } - } - - override fun toString(): String = "BitbrainsCompositeTableReader" - } - } -} diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/carbon/CarbonTableReader.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/carbon/CarbonTableReader.kt index 226c8806..c5face9f 100644 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/carbon/CarbonTableReader.kt +++ b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/carbon/CarbonTableReader.kt @@ -23,9 +23,9 @@ package org.opendc.trace.formats.carbon import org.opendc.trace.TableReader -import org.opendc.trace.conv.CARBON_INTENSITY_TIMESTAMP -import org.opendc.trace.conv.CARBON_INTENSITY_VALUE -import org.opendc.trace.formats.carbon.parquet.CarbonIntensityFragment +import org.opendc.trace.conv.CARBON_INTENSITY +import org.opendc.trace.conv.CARBON_TIMESTAMP +import org.opendc.trace.formats.carbon.parquet.CarbonFragment import org.opendc.trace.util.parquet.LocalParquetReader import java.time.Duration import java.time.Instant @@ -34,11 +34,11 @@ import java.util.UUID /** * A [TableReader] implementation for the WTF format. */ -internal class CarbonTableReader(private val reader: LocalParquetReader<CarbonIntensityFragment>) : TableReader { +internal class CarbonTableReader(private val reader: LocalParquetReader<CarbonFragment>) : TableReader { /** * The current record. */ - private var record: CarbonIntensityFragment? = null + private var record: CarbonFragment? = null override fun nextRow(): Boolean { try { @@ -57,8 +57,8 @@ internal class CarbonTableReader(private val reader: LocalParquetReader<CarbonIn override fun resolve(name: String): Int { return when (name) { - CARBON_INTENSITY_TIMESTAMP -> colTimestamp - CARBON_INTENSITY_VALUE -> colCarbonIntensity + CARBON_TIMESTAMP -> colTimestamp + CARBON_INTENSITY -> colCarbonIntensity else -> -1 } } diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/carbon/CarbonTraceFormat.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/carbon/CarbonTraceFormat.kt index d8adc739..764bb349 100644 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/carbon/CarbonTraceFormat.kt +++ b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/carbon/CarbonTraceFormat.kt @@ -26,10 +26,10 @@ import org.opendc.trace.TableColumn import org.opendc.trace.TableColumnType import org.opendc.trace.TableReader import org.opendc.trace.TableWriter -import org.opendc.trace.conv.CARBON_INTENSITY_TIMESTAMP -import org.opendc.trace.conv.CARBON_INTENSITY_VALUE -import org.opendc.trace.conv.TABLE_CARBON_INTENSITIES -import org.opendc.trace.formats.carbon.parquet.CarbonIntensityReadSupport +import org.opendc.trace.conv.CARBON_INTENSITY +import org.opendc.trace.conv.CARBON_TIMESTAMP +import org.opendc.trace.conv.TABLE_CARBON +import org.opendc.trace.formats.carbon.parquet.CarbonReadSupport import org.opendc.trace.spi.TableDetails import org.opendc.trace.spi.TraceFormat import org.opendc.trace.util.parquet.LocalParquetReader @@ -45,18 +45,18 @@ public class CarbonTraceFormat : TraceFormat { throw UnsupportedOperationException("Writing not supported for this format") } - override fun getTables(path: Path): List<String> = listOf(TABLE_CARBON_INTENSITIES) + override fun getTables(path: Path): List<String> = listOf(TABLE_CARBON) override fun getDetails( path: Path, table: String, ): TableDetails { return when (table) { - TABLE_CARBON_INTENSITIES -> + TABLE_CARBON -> TableDetails( listOf( - TableColumn(CARBON_INTENSITY_TIMESTAMP, TableColumnType.Instant), - TableColumn(CARBON_INTENSITY_VALUE, TableColumnType.Double), + TableColumn(CARBON_TIMESTAMP, TableColumnType.Instant), + TableColumn(CARBON_INTENSITY, TableColumnType.Double), ), ) else -> throw IllegalArgumentException("Table $table not supported") @@ -69,8 +69,8 @@ public class CarbonTraceFormat : TraceFormat { projection: List<String>?, ): TableReader { return when (table) { - TABLE_CARBON_INTENSITIES -> { - val reader = LocalParquetReader(path, CarbonIntensityReadSupport(projection)) + TABLE_CARBON -> { + val reader = LocalParquetReader(path, CarbonReadSupport(projection)) CarbonTableReader(reader) } else -> throw IllegalArgumentException("Table $table not supported") diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/carbon/parquet/CarbonIntensityFragment.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/carbon/parquet/CarbonFragment.kt index 3211cb6c..fe05876b 100644 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/carbon/parquet/CarbonIntensityFragment.kt +++ b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/carbon/parquet/CarbonFragment.kt @@ -27,7 +27,7 @@ import java.time.Instant /** * A task in the Workflow Trace Format. */ -internal data class CarbonIntensityFragment( +internal data class CarbonFragment( val timestamp: Instant, val carbonIntensity: Double, ) diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/carbon/parquet/CarbonIntensityReadSupport.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/carbon/parquet/CarbonReadSupport.kt index 2f4eac05..53087079 100644 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/carbon/parquet/CarbonIntensityReadSupport.kt +++ b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/carbon/parquet/CarbonReadSupport.kt @@ -26,26 +26,24 @@ import org.apache.hadoop.conf.Configuration import org.apache.parquet.hadoop.api.InitContext import org.apache.parquet.hadoop.api.ReadSupport import org.apache.parquet.io.api.RecordMaterializer -import org.apache.parquet.schema.LogicalTypeAnnotation import org.apache.parquet.schema.MessageType -import org.apache.parquet.schema.PrimitiveType import org.apache.parquet.schema.Types -import org.opendc.trace.conv.CARBON_INTENSITY_TIMESTAMP -import org.opendc.trace.conv.CARBON_INTENSITY_VALUE +import org.opendc.trace.conv.CARBON_INTENSITY +import org.opendc.trace.conv.CARBON_TIMESTAMP /** - * A [ReadSupport] instance for [Task] objects. + * A [ReadSupport] instance for [CarbonFragment] objects. * * @param projection The projection of the table to read. */ -internal class CarbonIntensityReadSupport(private val projection: List<String>?) : ReadSupport<CarbonIntensityFragment>() { +internal class CarbonReadSupport(private val projection: List<String>?) : ReadSupport<CarbonFragment>() { /** * Mapping of table columns to their Parquet column names. */ private val colMap = mapOf( - CARBON_INTENSITY_TIMESTAMP to "timestamp", - CARBON_INTENSITY_VALUE to "carbon_intensity", + CARBON_TIMESTAMP to "timestamp", + CARBON_INTENSITY to "carbon_intensity", ) override fun init(context: InitContext): ReadContext { @@ -53,16 +51,16 @@ internal class CarbonIntensityReadSupport(private val projection: List<String>?) if (projection != null) { Types.buildMessage() .apply { - val fieldByName = READ_SCHEMA.fields.associateBy { it.name } + val fieldByName = CARBON_SCHEMA.fields.associateBy { it.name } for (col in projection) { val fieldName = colMap[col] ?: continue addField(fieldByName.getValue(fieldName)) } } - .named(READ_SCHEMA.name) + .named(CARBON_SCHEMA.name) } else { - READ_SCHEMA + CARBON_SCHEMA } return ReadContext(projectedSchema) } @@ -72,24 +70,5 @@ internal class CarbonIntensityReadSupport(private val projection: List<String>?) keyValueMetaData: Map<String, String>, fileSchema: MessageType, readContext: ReadContext, - ): RecordMaterializer<CarbonIntensityFragment> = CarbonIntensityRecordMaterializer(readContext.requestedSchema) - - companion object { - /** - * Parquet read schema for the "tasks" table in the trace. - */ - @JvmStatic - val READ_SCHEMA: MessageType = - Types.buildMessage() - .addFields( - Types - .optional(PrimitiveType.PrimitiveTypeName.INT64) - .`as`(LogicalTypeAnnotation.timestampType(true, LogicalTypeAnnotation.TimeUnit.MILLIS)) - .named("timestamp"), - Types - .optional(PrimitiveType.PrimitiveTypeName.DOUBLE) - .named("carbon_intensity"), - ) - .named("carbon_intensity_fragment") - } + ): RecordMaterializer<CarbonFragment> = CarbonRecordMaterializer(readContext.requestedSchema) } diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/carbon/parquet/CarbonIntensityRecordMaterializer.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/carbon/parquet/CarbonRecordMaterializer.kt index f5d68f22..aa915a39 100644 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/carbon/parquet/CarbonIntensityRecordMaterializer.kt +++ b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/carbon/parquet/CarbonRecordMaterializer.kt @@ -30,9 +30,9 @@ import org.apache.parquet.schema.MessageType import java.time.Instant /** - * A [RecordMaterializer] for [Task] records. + * A [RecordMaterializer] for [CarbonFragment] records. */ -internal class CarbonIntensityRecordMaterializer(schema: MessageType) : RecordMaterializer<CarbonIntensityFragment>() { +internal class CarbonRecordMaterializer(schema: MessageType) : RecordMaterializer<CarbonFragment>() { /** * State of current record being read. */ @@ -76,8 +76,8 @@ internal class CarbonIntensityRecordMaterializer(schema: MessageType) : RecordMa override fun getConverter(fieldIndex: Int): Converter = converters[fieldIndex] } - override fun getCurrentRecord(): CarbonIntensityFragment = - CarbonIntensityFragment( + override fun getCurrentRecord(): CarbonFragment = + CarbonFragment( localTimestamp, localCarbonIntensity, ) diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/carbon/parquet/CarbonSchemas.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/carbon/parquet/CarbonSchemas.kt new file mode 100644 index 00000000..c8b11968 --- /dev/null +++ b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/carbon/parquet/CarbonSchemas.kt @@ -0,0 +1,43 @@ +/* + * Copyright (c) 2025 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 org.opendc.trace.formats.carbon.parquet + +import org.apache.parquet.schema.LogicalTypeAnnotation +import org.apache.parquet.schema.MessageType +import org.apache.parquet.schema.PrimitiveType +import org.apache.parquet.schema.Types + +private val CARBON_SCHEMA_v1: MessageType = + Types.buildMessage() + .addFields( + Types + .optional(PrimitiveType.PrimitiveTypeName.INT64) + .`as`(LogicalTypeAnnotation.timestampType(true, LogicalTypeAnnotation.TimeUnit.MILLIS)) + .named("timestamp"), + Types + .optional(PrimitiveType.PrimitiveTypeName.DOUBLE) + .named("carbon_intensity"), + ) + .named("carbon_intensity_fragment") + +public val CARBON_SCHEMA: MessageType = CARBON_SCHEMA_v1 diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/failure/parquet/FailureReadSupport.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/failure/parquet/FailureReadSupport.kt index d49f86c6..9bd8fd72 100644 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/failure/parquet/FailureReadSupport.kt +++ b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/failure/parquet/FailureReadSupport.kt @@ -27,14 +27,13 @@ import org.apache.parquet.hadoop.api.InitContext import org.apache.parquet.hadoop.api.ReadSupport import org.apache.parquet.io.api.RecordMaterializer import org.apache.parquet.schema.MessageType -import org.apache.parquet.schema.PrimitiveType import org.apache.parquet.schema.Types import org.opendc.trace.conv.FAILURE_DURATION import org.opendc.trace.conv.FAILURE_INTENSITY import org.opendc.trace.conv.FAILURE_INTERVAL /** - * A [ReadSupport] instance for [Task] objects. + * A [ReadSupport] instance for [FailureFragment] objects. * * @param projection The projection of the table to read. */ @@ -54,16 +53,16 @@ internal class FailureReadSupport(private val projection: List<String>?) : ReadS if (projection != null) { Types.buildMessage() .apply { - val fieldByName = READ_SCHEMA.fields.associateBy { it.name } + val fieldByName = FAILURE_SCHEMA.fields.associateBy { it.name } for (col in projection) { val fieldName = colMap[col] ?: continue addField(fieldByName.getValue(fieldName)) } } - .named(READ_SCHEMA.name) + .named(FAILURE_SCHEMA.name) } else { - READ_SCHEMA + FAILURE_SCHEMA } return ReadContext(projectedSchema) } @@ -74,25 +73,4 @@ internal class FailureReadSupport(private val projection: List<String>?) : ReadS fileSchema: MessageType, readContext: ReadContext, ): RecordMaterializer<FailureFragment> = FailureRecordMaterializer(readContext.requestedSchema) - - companion object { - /** - * Parquet read schema for the "tasks" table in the trace. - */ - @JvmStatic - val READ_SCHEMA: MessageType = - Types.buildMessage() - .addFields( - Types - .optional(PrimitiveType.PrimitiveTypeName.INT64) - .named("failure_interval"), - Types - .optional(PrimitiveType.PrimitiveTypeName.INT64) - .named("failure_duration"), - Types - .optional(PrimitiveType.PrimitiveTypeName.DOUBLE) - .named("failure_intensity"), - ) - .named("failure_fragment") - } } diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/failure/parquet/FailureRecordMaterializer.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/failure/parquet/FailureRecordMaterializer.kt index 5a00f8c9..83281984 100644 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/failure/parquet/FailureRecordMaterializer.kt +++ b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/failure/parquet/FailureRecordMaterializer.kt @@ -29,7 +29,7 @@ import org.apache.parquet.io.api.RecordMaterializer import org.apache.parquet.schema.MessageType /** - * A [RecordMaterializer] for [Task] records. + * A [RecordMaterializer] for [FailureFragment] records. */ internal class FailureRecordMaterializer(schema: MessageType) : RecordMaterializer<FailureFragment>() { /** diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/failure/parquet/FailureSchemas.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/failure/parquet/FailureSchemas.kt new file mode 100644 index 00000000..bafac387 --- /dev/null +++ b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/failure/parquet/FailureSchemas.kt @@ -0,0 +1,44 @@ +/* + * Copyright (c) 2025 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 org.opendc.trace.formats.failure.parquet + +import org.apache.parquet.schema.MessageType +import org.apache.parquet.schema.PrimitiveType +import org.apache.parquet.schema.Types + +private val FAILURE_SCHEMA_v1: MessageType = + Types.buildMessage() + .addFields( + Types + .optional(PrimitiveType.PrimitiveTypeName.INT64) + .named("failure_interval"), + Types + .optional(PrimitiveType.PrimitiveTypeName.INT64) + .named("failure_duration"), + Types + .optional(PrimitiveType.PrimitiveTypeName.DOUBLE) + .named("failure_intensity"), + ) + .named("failure_fragment") + +public val FAILURE_SCHEMA: MessageType = FAILURE_SCHEMA_v1 diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/gwf/GwfTaskTableReader.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/gwf/GwfTaskTableReader.kt deleted file mode 100644 index 8a2a99cb..00000000 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/gwf/GwfTaskTableReader.kt +++ /dev/null @@ -1,286 +0,0 @@ -/* - * Copyright (c) 2021 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 org.opendc.trace.gwf - -import com.fasterxml.jackson.core.JsonToken -import com.fasterxml.jackson.dataformat.csv.CsvParser -import com.fasterxml.jackson.dataformat.csv.CsvSchema -import org.opendc.trace.TableColumnType -import org.opendc.trace.TableReader -import org.opendc.trace.conv.TASK_ALLOC_NCPUS -import org.opendc.trace.conv.TASK_ID -import org.opendc.trace.conv.TASK_PARENTS -import org.opendc.trace.conv.TASK_REQ_NCPUS -import org.opendc.trace.conv.TASK_RUNTIME -import org.opendc.trace.conv.TASK_SUBMIT_TIME -import org.opendc.trace.conv.TASK_WORKFLOW_ID -import org.opendc.trace.util.convertTo -import java.time.Duration -import java.time.Instant -import java.util.UUID -import java.util.regex.Pattern - -/** - * A [TableReader] implementation for the GWF format. - */ -internal class GwfTaskTableReader(private val parser: CsvParser) : TableReader { - /** - * A flag to indicate whether a single row has been read already. - */ - private var isStarted = false - - init { - parser.schema = schema - } - - override fun nextRow(): Boolean { - if (!isStarted) { - isStarted = true - } - - // Reset the row state - reset() - - if (parser.isClosed || !nextStart()) { - return false - } - - while (true) { - val token = parser.nextValue() - - if (token == null || token == JsonToken.END_OBJECT) { - break - } - - when (parser.currentName) { - "WorkflowID" -> workflowId = parser.text - "JobID" -> jobId = parser.text - "SubmitTime" -> submitTime = Instant.ofEpochSecond(parser.longValue) - "RunTime" -> runtime = Duration.ofSeconds(parser.longValue) - "NProcs" -> nProcs = parser.intValue - "ReqNProcs" -> reqNProcs = parser.intValue - "Dependencies" -> dependencies = parseParents(parser.valueAsString) - } - } - - return true - } - - override fun resolve(name: String): Int { - return when (name) { - TASK_ID -> colJobID - TASK_WORKFLOW_ID -> colWorkflowID - TASK_SUBMIT_TIME -> colSubmitTime - TASK_RUNTIME -> colRuntime - TASK_ALLOC_NCPUS -> colNproc - TASK_REQ_NCPUS -> colReqNproc - TASK_PARENTS -> colDeps - else -> -1 - } - } - - override fun isNull(index: Int): Boolean { - require(index in 0..colDeps) { "Invalid column" } - return false - } - - override fun getBoolean(index: Int): Boolean { - throw IllegalArgumentException("Invalid column") - } - - override fun getInt(index: Int): Int { - checkActive() - return when (index) { - colReqNproc -> reqNProcs - colNproc -> nProcs - else -> throw IllegalArgumentException("Invalid column") - } - } - - override fun getLong(index: Int): Long { - throw IllegalArgumentException("Invalid column") - } - - override fun getFloat(index: Int): Float { - throw IllegalArgumentException("Invalid column") - } - - override fun getDouble(index: Int): Double { - throw IllegalArgumentException("Invalid column") - } - - override fun getString(index: Int): String? { - checkActive() - return when (index) { - colJobID -> jobId - colWorkflowID -> workflowId - else -> throw IllegalArgumentException("Invalid column") - } - } - - override fun getUUID(index: Int): UUID? { - throw IllegalArgumentException("Invalid column") - } - - override fun getInstant(index: Int): Instant? { - checkActive() - return when (index) { - colSubmitTime -> submitTime - else -> throw IllegalArgumentException("Invalid column") - } - } - - override fun getDuration(index: Int): Duration? { - checkActive() - return when (index) { - colRuntime -> runtime - else -> throw IllegalArgumentException("Invalid column") - } - } - - override fun <T> getList( - index: Int, - elementType: Class<T>, - ): List<T>? { - throw IllegalArgumentException("Invalid column") - } - - override fun <K, V> getMap( - index: Int, - keyType: Class<K>, - valueType: Class<V>, - ): Map<K, V>? { - throw IllegalArgumentException("Invalid column") - } - - override fun <T> getSet( - index: Int, - elementType: Class<T>, - ): Set<T>? { - checkActive() - return when (index) { - colDeps -> typeDeps.convertTo(dependencies, elementType) - else -> throw IllegalArgumentException("Invalid column") - } - } - - override fun close() { - parser.close() - } - - /** - * Helper method to check if the reader is active. - */ - private fun checkActive() { - check(isStarted && !parser.isClosed) { "No active row. Did you call nextRow()?" } - } - - /** - * The pattern used to parse the parents. - */ - private val pattern = Pattern.compile("\\s+") - - /** - * Parse the parents into a set of longs. - */ - private fun parseParents(value: String): Set<String> { - val result = mutableSetOf<String>() - val deps = value.split(pattern) - - for (dep in deps) { - if (dep.isBlank()) { - continue - } - - result.add(dep) - } - - return result - } - - /** - * Advance the parser until the next object start. - */ - private fun nextStart(): Boolean { - var token = parser.nextValue() - - while (token != null && token != JsonToken.START_OBJECT) { - token = parser.nextValue() - } - - return token != null - } - - /** - * Reader state fields. - */ - private var workflowId: String? = null - private var jobId: String? = null - private var submitTime: Instant? = null - private var runtime: Duration? = null - private var nProcs = -1 - private var reqNProcs = -1 - private var dependencies = emptySet<String>() - - /** - * Reset the state. - */ - private fun reset() { - workflowId = null - jobId = null - submitTime = null - runtime = null - nProcs = -1 - reqNProcs = -1 - dependencies = emptySet() - } - - private val colWorkflowID = 0 - private val colJobID = 1 - private val colSubmitTime = 2 - private val colRuntime = 3 - private val colNproc = 4 - private val colReqNproc = 5 - private val colDeps = 6 - - private val typeDeps = TableColumnType.Set(TableColumnType.String) - - companion object { - /** - * The [CsvSchema] that is used to parse the trace. - */ - private val schema = - CsvSchema.builder() - .addColumn("WorkflowID", CsvSchema.ColumnType.NUMBER) - .addColumn("JobID", CsvSchema.ColumnType.NUMBER) - .addColumn("SubmitTime", CsvSchema.ColumnType.NUMBER) - .addColumn("RunTime", CsvSchema.ColumnType.NUMBER) - .addColumn("NProcs", CsvSchema.ColumnType.NUMBER) - .addColumn("ReqNProcs", CsvSchema.ColumnType.NUMBER) - .addColumn("Dependencies", CsvSchema.ColumnType.STRING) - .setAllowComments(true) - .setUseHeader(true) - .setColumnSeparator(',') - .build() - } -} diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/gwf/GwfTraceFormat.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/gwf/GwfTraceFormat.kt deleted file mode 100644 index 097c5593..00000000 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/gwf/GwfTraceFormat.kt +++ /dev/null @@ -1,104 +0,0 @@ -/* - * Copyright (c) 2021 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 org.opendc.trace.gwf - -import com.fasterxml.jackson.dataformat.csv.CsvFactory -import com.fasterxml.jackson.dataformat.csv.CsvParser -import org.opendc.trace.TableColumn -import org.opendc.trace.TableColumnType -import org.opendc.trace.TableReader -import org.opendc.trace.TableWriter -import org.opendc.trace.conv.TABLE_TASKS -import org.opendc.trace.conv.TASK_ALLOC_NCPUS -import org.opendc.trace.conv.TASK_ID -import org.opendc.trace.conv.TASK_PARENTS -import org.opendc.trace.conv.TASK_REQ_NCPUS -import org.opendc.trace.conv.TASK_RUNTIME -import org.opendc.trace.conv.TASK_SUBMIT_TIME -import org.opendc.trace.conv.TASK_WORKFLOW_ID -import org.opendc.trace.spi.TableDetails -import org.opendc.trace.spi.TraceFormat -import java.nio.file.Path - -/** - * A [TraceFormat] implementation for the GWF trace format. - */ -public class GwfTraceFormat : TraceFormat { - /** - * The name of this trace format. - */ - override val name: String = "gwf" - - /** - * The [CsvFactory] used to create the parser. - */ - private val factory = - CsvFactory() - .enable(CsvParser.Feature.ALLOW_COMMENTS) - .enable(CsvParser.Feature.TRIM_SPACES) - - override fun create(path: Path) { - throw UnsupportedOperationException("Writing not supported for this format") - } - - override fun getTables(path: Path): List<String> = listOf(TABLE_TASKS) - - override fun getDetails( - path: Path, - table: String, - ): TableDetails { - return when (table) { - TABLE_TASKS -> - TableDetails( - listOf( - TableColumn(TASK_WORKFLOW_ID, TableColumnType.String), - TableColumn(TASK_ID, TableColumnType.String), - TableColumn(TASK_SUBMIT_TIME, TableColumnType.Instant), - TableColumn(TASK_RUNTIME, TableColumnType.Duration), - TableColumn(TASK_REQ_NCPUS, TableColumnType.Int), - TableColumn(TASK_ALLOC_NCPUS, TableColumnType.Int), - TableColumn(TASK_PARENTS, TableColumnType.Set(TableColumnType.String)), - ), - ) - else -> throw IllegalArgumentException("Table $table not supported") - } - } - - override fun newReader( - path: Path, - table: String, - projection: List<String>?, - ): TableReader { - return when (table) { - TABLE_TASKS -> GwfTaskTableReader(factory.createParser(path.toFile())) - else -> throw IllegalArgumentException("Table $table not supported") - } - } - - override fun newWriter( - path: Path, - table: String, - ): TableWriter { - throw UnsupportedOperationException("Writing not supported for this format") - } -} diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/opendc/OdcVmInterferenceJsonTableReader.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/opendc/OdcVmInterferenceJsonTableReader.kt deleted file mode 100644 index dba971d7..00000000 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/opendc/OdcVmInterferenceJsonTableReader.kt +++ /dev/null @@ -1,225 +0,0 @@ -/* - * Copyright (c) 2022 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 org.opendc.trace.formats.opendc - -import com.fasterxml.jackson.core.JsonParseException -import com.fasterxml.jackson.core.JsonParser -import com.fasterxml.jackson.core.JsonToken -import org.opendc.trace.TableColumnType -import org.opendc.trace.TableReader -import org.opendc.trace.conv.INTERFERENCE_GROUP_MEMBERS -import org.opendc.trace.conv.INTERFERENCE_GROUP_SCORE -import org.opendc.trace.conv.INTERFERENCE_GROUP_TARGET -import org.opendc.trace.util.convertTo -import java.time.Duration -import java.time.Instant -import java.util.UUID - -/** - * A [TableReader] implementation for the OpenDC VM interference JSON format. - */ -internal class OdcVmInterferenceJsonTableReader(private val parser: JsonParser) : TableReader { - /** - * A flag to indicate whether a single row has been read already. - */ - private var isStarted = false - - override fun nextRow(): Boolean { - if (!isStarted) { - isStarted = true - - parser.nextToken() - - if (!parser.isExpectedStartArrayToken) { - throw JsonParseException(parser, "Expected array at start, but got ${parser.currentToken()}") - } - } - - return if (parser.isClosed || parser.nextToken() == JsonToken.END_ARRAY) { - parser.close() - reset() - false - } else { - parseGroup(parser) - true - } - } - - private val colMembers = 0 - private val colTarget = 1 - private val colScore = 2 - - private val typeMembers = TableColumnType.Set(TableColumnType.String) - - override fun resolve(name: String): Int { - return when (name) { - INTERFERENCE_GROUP_MEMBERS -> colMembers - INTERFERENCE_GROUP_TARGET -> colTarget - INTERFERENCE_GROUP_SCORE -> colScore - else -> -1 - } - } - - override fun isNull(index: Int): Boolean { - return when (index) { - colMembers, colTarget, colScore -> false - else -> throw IllegalArgumentException("Invalid column index $index") - } - } - - override fun getBoolean(index: Int): Boolean { - throw IllegalArgumentException("Invalid column $index") - } - - override fun getInt(index: Int): Int { - throw IllegalArgumentException("Invalid column $index") - } - - override fun getLong(index: Int): Long { - throw IllegalArgumentException("Invalid column $index") - } - - override fun getFloat(index: Int): Float { - throw IllegalArgumentException("Invalid column $index") - } - - override fun getDouble(index: Int): Double { - checkActive() - return when (index) { - colTarget -> targetLoad - colScore -> score - else -> throw IllegalArgumentException("Invalid column $index") - } - } - - override fun getString(index: Int): String? { - throw IllegalArgumentException("Invalid column $index") - } - - override fun getUUID(index: Int): UUID? { - throw IllegalArgumentException("Invalid column $index") - } - - override fun getInstant(index: Int): Instant? { - throw IllegalArgumentException("Invalid column $index") - } - - override fun getDuration(index: Int): Duration? { - throw IllegalArgumentException("Invalid column $index") - } - - override fun <T> getList( - index: Int, - elementType: Class<T>, - ): List<T>? { - throw IllegalArgumentException("Invalid column $index") - } - - override fun <T> getSet( - index: Int, - elementType: Class<T>, - ): Set<T>? { - checkActive() - return when (index) { - colMembers -> typeMembers.convertTo(members, elementType) - else -> throw IllegalArgumentException("Invalid column $index") - } - } - - override fun <K, V> getMap( - index: Int, - keyType: Class<K>, - valueType: Class<V>, - ): Map<K, V>? { - throw IllegalArgumentException("Invalid column $index") - } - - override fun close() { - parser.close() - } - - private var members = emptySet<String>() - private var targetLoad = Double.POSITIVE_INFINITY - private var score = 1.0 - - /** - * Helper method to check if the reader is active. - */ - private fun checkActive() { - check(isStarted && !parser.isClosed) { "No active row. Did you call nextRow()?" } - } - - /** - * Reset the state. - */ - private fun reset() { - members = emptySet() - targetLoad = Double.POSITIVE_INFINITY - score = 1.0 - } - - /** - * Parse a group an interference JSON file. - */ - private fun parseGroup(parser: JsonParser) { - var targetLoad = Double.POSITIVE_INFINITY - var score = 1.0 - val members = mutableSetOf<String>() - - if (!parser.isExpectedStartObjectToken) { - throw JsonParseException(parser, "Expected object, but got ${parser.currentToken()}") - } - - while (parser.nextValue() != JsonToken.END_OBJECT) { - when (parser.currentName) { - "vms" -> parseGroupMembers(parser, members) - "minServerLoad" -> targetLoad = parser.doubleValue - "performanceScore" -> score = parser.doubleValue - } - } - - this.members = members - this.targetLoad = targetLoad - this.score = score - } - - /** - * Parse the members of a group. - */ - private fun parseGroupMembers( - parser: JsonParser, - members: MutableSet<String>, - ) { - if (!parser.isExpectedStartArrayToken) { - throw JsonParseException(parser, "Expected array for group members, but got ${parser.currentToken()}") - } - - while (parser.nextValue() != JsonToken.END_ARRAY) { - if (parser.currentToken() != JsonToken.VALUE_STRING) { - throw JsonParseException(parser, "Expected string value for group member") - } - - members.add(parser.text) - } - } -} diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/opendc/OdcVmInterferenceJsonTableWriter.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/opendc/OdcVmInterferenceJsonTableWriter.kt deleted file mode 100644 index b3286a1c..00000000 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/opendc/OdcVmInterferenceJsonTableWriter.kt +++ /dev/null @@ -1,192 +0,0 @@ -/* - * Copyright (c) 2022 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 org.opendc.trace.formats.opendc - -import com.fasterxml.jackson.core.JsonGenerator -import org.opendc.trace.TableWriter -import org.opendc.trace.conv.INTERFERENCE_GROUP_MEMBERS -import org.opendc.trace.conv.INTERFERENCE_GROUP_SCORE -import org.opendc.trace.conv.INTERFERENCE_GROUP_TARGET -import java.time.Duration -import java.time.Instant -import java.util.UUID - -/** - * A [TableWriter] implementation for the OpenDC VM interference JSON format. - */ -internal class OdcVmInterferenceJsonTableWriter(private val generator: JsonGenerator) : TableWriter { - /** - * A flag to indicate whether a row has been started. - */ - private var isRowActive = false - - init { - generator.writeStartArray() - } - - override fun startRow() { - // Reset state - members = emptySet() - targetLoad = Double.POSITIVE_INFINITY - score = 1.0 - - // Mark row as active - isRowActive = true - } - - override fun endRow() { - check(isRowActive) { "No active row" } - - generator.writeStartObject() - generator.writeArrayFieldStart("vms") - for (member in members) { - generator.writeString(member) - } - generator.writeEndArray() - generator.writeNumberField("minServerLoad", targetLoad) - generator.writeNumberField("performanceScore", score) - generator.writeEndObject() - } - - override fun resolve(name: String): Int { - return when (name) { - INTERFERENCE_GROUP_MEMBERS -> colMembers - INTERFERENCE_GROUP_TARGET -> colTarget - INTERFERENCE_GROUP_SCORE -> colScore - else -> -1 - } - } - - override fun setBoolean( - index: Int, - value: Boolean, - ) { - throw IllegalArgumentException("Invalid column $index") - } - - override fun setInt( - index: Int, - value: Int, - ) { - throw IllegalArgumentException("Invalid column $index") - } - - override fun setLong( - index: Int, - value: Long, - ) { - throw IllegalArgumentException("Invalid column $index") - } - - override fun setFloat( - index: Int, - value: Float, - ) { - throw IllegalArgumentException("Invalid column $index") - } - - override fun setDouble( - index: Int, - value: Double, - ) { - check(isRowActive) { "No active row" } - - when (index) { - colTarget -> targetLoad = (value as Number).toDouble() - colScore -> score = (value as Number).toDouble() - else -> throw IllegalArgumentException("Invalid column $index") - } - } - - override fun setString( - index: Int, - value: String, - ) { - throw IllegalArgumentException("Invalid column $index") - } - - override fun setUUID( - index: Int, - value: UUID, - ) { - throw IllegalArgumentException("Invalid column $index") - } - - override fun setInstant( - index: Int, - value: Instant, - ) { - throw IllegalArgumentException("Invalid column $index") - } - - override fun setDuration( - index: Int, - value: Duration, - ) { - throw IllegalArgumentException("Invalid column $index") - } - - override fun <T> setList( - index: Int, - value: List<T>, - ) { - throw IllegalArgumentException("Invalid column $index") - } - - override fun <T> setSet( - index: Int, - value: Set<T>, - ) { - check(isRowActive) { "No active row" } - - @Suppress("UNCHECKED_CAST") - when (index) { - colMembers -> members = value as Set<String> - else -> throw IllegalArgumentException("Invalid column index $index") - } - } - - override fun <K, V> setMap( - index: Int, - value: Map<K, V>, - ) { - throw IllegalArgumentException("Invalid column $index") - } - - override fun flush() { - generator.flush() - } - - override fun close() { - generator.writeEndArray() - generator.close() - } - - private val colMembers = 0 - private val colTarget = 1 - private val colScore = 2 - - private var members = emptySet<String>() - private var targetLoad = Double.POSITIVE_INFINITY - private var score = 1.0 -} diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/opendc/OdcVmTraceFormat.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/opendc/OdcVmTraceFormat.kt deleted file mode 100644 index 74e880be..00000000 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/opendc/OdcVmTraceFormat.kt +++ /dev/null @@ -1,202 +0,0 @@ -/* - * Copyright (c) 2021 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 org.opendc.trace.formats.opendc - -import com.fasterxml.jackson.core.JsonEncoding -import com.fasterxml.jackson.core.JsonFactory -import org.apache.parquet.column.ParquetProperties -import org.apache.parquet.hadoop.ParquetFileWriter -import org.apache.parquet.hadoop.metadata.CompressionCodecName -import org.opendc.trace.TableColumn -import org.opendc.trace.TableColumnType -import org.opendc.trace.TableReader -import org.opendc.trace.TableWriter -import org.opendc.trace.conv.INTERFERENCE_GROUP_MEMBERS -import org.opendc.trace.conv.INTERFERENCE_GROUP_SCORE -import org.opendc.trace.conv.INTERFERENCE_GROUP_TARGET -import org.opendc.trace.conv.TABLE_INTERFERENCE_GROUPS -import org.opendc.trace.conv.TABLE_RESOURCES -import org.opendc.trace.conv.TABLE_RESOURCE_STATES -import org.opendc.trace.conv.resourceChildren -import org.opendc.trace.conv.resourceCpuCapacity -import org.opendc.trace.conv.resourceCpuCount -import org.opendc.trace.conv.resourceDeadline -import org.opendc.trace.conv.resourceDuration -import org.opendc.trace.conv.resourceGpuCapacity -import org.opendc.trace.conv.resourceGpuCount -import org.opendc.trace.conv.resourceID -import org.opendc.trace.conv.resourceMemCapacity -import org.opendc.trace.conv.resourceNature -import org.opendc.trace.conv.resourceParents -import org.opendc.trace.conv.resourceStateCpuUsage -import org.opendc.trace.conv.resourceStateDuration -import org.opendc.trace.conv.resourceStateTimestamp -import org.opendc.trace.conv.resourceSubmissionTime -import org.opendc.trace.formats.opendc.parquet.ResourceReadSupport -import org.opendc.trace.formats.opendc.parquet.ResourceStateReadSupport -import org.opendc.trace.formats.opendc.parquet.ResourceStateWriteSupport -import org.opendc.trace.formats.opendc.parquet.ResourceWriteSupport -import org.opendc.trace.spi.TableDetails -import org.opendc.trace.spi.TraceFormat -import org.opendc.trace.util.parquet.LocalParquetReader -import org.opendc.trace.util.parquet.LocalParquetWriter -import java.nio.file.Files -import java.nio.file.Path -import kotlin.io.path.exists - -/** - * A [TraceFormat] implementation of the OpenDC virtual machine trace format. - */ -public class OdcVmTraceFormat : TraceFormat { - /** - * A [JsonFactory] that is used to parse the JSON-based interference model. - */ - private val jsonFactory = JsonFactory() - - /** - * The name of this trace format. - */ - override val name: String = "opendc-vm" - - override fun create(path: Path) { - // Construct directory containing the trace files - Files.createDirectories(path) - - val tables = getTables(path) - - for (table in tables) { - val writer = newWriter(path, table) - writer.close() - } - } - - override fun getTables(path: Path): List<String> = listOf(TABLE_RESOURCES, TABLE_RESOURCE_STATES, TABLE_INTERFERENCE_GROUPS) - - override fun getDetails( - path: Path, - table: String, - ): TableDetails { - return when (table) { - TABLE_RESOURCES -> - TableDetails( - listOf( - TableColumn(resourceID, TableColumnType.String), - TableColumn(resourceSubmissionTime, TableColumnType.Instant), - TableColumn(resourceDuration, TableColumnType.Long), - TableColumn(resourceCpuCount, TableColumnType.Int), - TableColumn(resourceCpuCapacity, TableColumnType.Double), - TableColumn(resourceMemCapacity, TableColumnType.Double), - TableColumn(resourceGpuCount, TableColumnType.Int), - TableColumn(resourceGpuCapacity, TableColumnType.Double), - TableColumn(resourceParents, TableColumnType.Set(TableColumnType.String)), - TableColumn(resourceChildren, TableColumnType.Set(TableColumnType.String)), - TableColumn(resourceNature, TableColumnType.String), - TableColumn(resourceDeadline, TableColumnType.Long), - ), - ) - TABLE_RESOURCE_STATES -> - TableDetails( - listOf( - TableColumn(resourceID, TableColumnType.String), - TableColumn(resourceStateTimestamp, TableColumnType.Instant), - TableColumn(resourceStateDuration, TableColumnType.Duration), - TableColumn(resourceCpuCount, TableColumnType.Int), - TableColumn(resourceStateCpuUsage, TableColumnType.Double), - ), - ) - TABLE_INTERFERENCE_GROUPS -> - TableDetails( - listOf( - TableColumn(INTERFERENCE_GROUP_MEMBERS, TableColumnType.Set(TableColumnType.String)), - TableColumn(INTERFERENCE_GROUP_TARGET, TableColumnType.Double), - TableColumn(INTERFERENCE_GROUP_SCORE, TableColumnType.Double), - ), - ) - else -> throw IllegalArgumentException("Table $table not supported") - } - } - - override fun newReader( - path: Path, - table: String, - projection: List<String>?, - ): TableReader { - return when (table) { - TABLE_RESOURCES -> { - val reader = LocalParquetReader(path.resolve("tasks.parquet"), ResourceReadSupport(projection)) - OdcVmResourceTableReader(reader) - } - TABLE_RESOURCE_STATES -> { - val reader = LocalParquetReader(path.resolve("fragments.parquet"), ResourceStateReadSupport(projection)) - OdcVmResourceStateTableReader(reader) - } - TABLE_INTERFERENCE_GROUPS -> { - val modelPath = path.resolve("interference-model.json") - val parser = - if (modelPath.exists()) { - jsonFactory.createParser(modelPath.toFile()) - } else { - jsonFactory.createParser("[]") // If model does not exist, return empty model - } - - OdcVmInterferenceJsonTableReader(parser) - } - else -> throw IllegalArgumentException("Table $table not supported") - } - } - - override fun newWriter( - path: Path, - table: String, - ): TableWriter { - return when (table) { - TABLE_RESOURCES -> { - val writer = - LocalParquetWriter.builder(path.resolve("tasks.parquet"), ResourceWriteSupport()) - .withCompressionCodec(CompressionCodecName.ZSTD) - .withPageWriteChecksumEnabled(true) - .withWriterVersion(ParquetProperties.WriterVersion.PARQUET_2_0) - .withWriteMode(ParquetFileWriter.Mode.OVERWRITE) - .build() - OdcVmResourceTableWriter(writer) - } - TABLE_RESOURCE_STATES -> { - val writer = - LocalParquetWriter.builder(path.resolve("fragments.parquet"), ResourceStateWriteSupport()) - .withCompressionCodec(CompressionCodecName.ZSTD) - .withDictionaryEncoding("id", true) - .withBloomFilterEnabled("id", true) - .withPageWriteChecksumEnabled(true) - .withWriterVersion(ParquetProperties.WriterVersion.PARQUET_2_0) - .withWriteMode(ParquetFileWriter.Mode.OVERWRITE) - .build() - OdcVmResourceStateTableWriter(writer) - } - TABLE_INTERFERENCE_GROUPS -> { - val generator = jsonFactory.createGenerator(path.resolve("interference-model.json").toFile(), JsonEncoding.UTF8) - OdcVmInterferenceJsonTableWriter(generator) - } - else -> throw IllegalArgumentException("Table $table not supported") - } - } -} diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/opendc/parquet/ResourceReadSupport.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/opendc/parquet/ResourceReadSupport.kt deleted file mode 100644 index cd2ccef7..00000000 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/opendc/parquet/ResourceReadSupport.kt +++ /dev/null @@ -1,214 +0,0 @@ -/* - * Copyright (c) 2022 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 org.opendc.trace.formats.opendc.parquet - -import org.apache.hadoop.conf.Configuration -import org.apache.parquet.hadoop.api.InitContext -import org.apache.parquet.hadoop.api.ReadSupport -import org.apache.parquet.io.api.RecordMaterializer -import org.apache.parquet.schema.LogicalTypeAnnotation -import org.apache.parquet.schema.MessageType -import org.apache.parquet.schema.PrimitiveType -import org.apache.parquet.schema.Type -import org.apache.parquet.schema.Types -import org.opendc.trace.TableColumn -import org.opendc.trace.conv.resourceChildren -import org.opendc.trace.conv.resourceCpuCapacity -import org.opendc.trace.conv.resourceCpuCount -import org.opendc.trace.conv.resourceDeadline -import org.opendc.trace.conv.resourceDuration -import org.opendc.trace.conv.resourceGpuCapacity -import org.opendc.trace.conv.resourceGpuCount -import org.opendc.trace.conv.resourceID -import org.opendc.trace.conv.resourceMemCapacity -import org.opendc.trace.conv.resourceNature -import org.opendc.trace.conv.resourceParents -import org.opendc.trace.conv.resourceSubmissionTime - -/** - * A [ReadSupport] instance for [Resource] objects. - */ -internal class ResourceReadSupport(private val projection: List<String>?) : ReadSupport<Resource>() { - /** - * Mapping from field names to [TableColumn]s. - */ - private val fieldMap = - mapOf( - "id" to resourceID, - "submissionTime" to resourceSubmissionTime, - "submission_time" to resourceSubmissionTime, - "duration" to resourceDuration, - "maxCores" to resourceCpuCount, - "cpu_count" to resourceCpuCount, - "cpu_capacity" to resourceCpuCapacity, - "requiredMemory" to resourceMemCapacity, - "mem_capacity" to resourceMemCapacity, - "gpu_count" to resourceGpuCount, - "gpu_capacity" to resourceGpuCapacity, - "parents" to resourceParents, - "children" to resourceChildren, - "nature" to resourceNature, - "deadline" to resourceDeadline, - ) - - override fun init(context: InitContext): ReadContext { - val projectedSchema = - if (projection != null) { - Types.buildMessage() - .apply { - val projectionSet = projection.toSet() - - for (field in READ_SCHEMA.fields) { - val col = fieldMap[field.name] ?: continue - if (col in projectionSet) { - addField(field) - } - } - } - .named(READ_SCHEMA.name) - } else { - READ_SCHEMA - } - - return ReadContext(projectedSchema) - } - - override fun prepareForRead( - configuration: Configuration, - keyValueMetaData: Map<String, String>, - fileSchema: MessageType, - readContext: ReadContext, - ): RecordMaterializer<Resource> = ResourceRecordMaterializer(readContext.requestedSchema) - - companion object { - /** - * Parquet read schema (version 2.0) for the "resources" table in the trace. - */ - @JvmStatic - val READ_SCHEMA_V2_0: MessageType = - Types.buildMessage() - .addFields( - Types - .required(PrimitiveType.PrimitiveTypeName.BINARY) - .`as`(LogicalTypeAnnotation.stringType()) - .named("id"), - Types - .required(PrimitiveType.PrimitiveTypeName.INT64) - .`as`(LogicalTypeAnnotation.timestampType(true, LogicalTypeAnnotation.TimeUnit.MILLIS)) - .named("submissionTime"), - Types - .required(PrimitiveType.PrimitiveTypeName.INT64) - .named("duration"), - Types - .required(PrimitiveType.PrimitiveTypeName.INT32) - .named("maxCores"), - Types - .required(PrimitiveType.PrimitiveTypeName.INT64) - .named("requiredMemory"), - Types - .optional(PrimitiveType.PrimitiveTypeName.BINARY) - .`as`(LogicalTypeAnnotation.stringType()) - .named("nature"), - Types - .optional(PrimitiveType.PrimitiveTypeName.INT64) - .named("deadline"), - ) - .named("resource") - - /** - * Parquet read schema (version 2.1) for the "resources" table in the trace. - */ - @JvmStatic - val READ_SCHEMA_V2_2: MessageType = - Types.buildMessage() - .addFields( - Types - .required(PrimitiveType.PrimitiveTypeName.BINARY) - .`as`(LogicalTypeAnnotation.stringType()) - .named("id"), - Types - .required(PrimitiveType.PrimitiveTypeName.INT64) - .`as`(LogicalTypeAnnotation.timestampType(true, LogicalTypeAnnotation.TimeUnit.MILLIS)) - .named("submission_time"), - Types - .required(PrimitiveType.PrimitiveTypeName.INT64) - .named("duration"), - Types - .required(PrimitiveType.PrimitiveTypeName.INT32) - .named("cpu_count"), - Types - .required(PrimitiveType.PrimitiveTypeName.DOUBLE) - .named("cpu_capacity"), - Types - .required(PrimitiveType.PrimitiveTypeName.INT64) - .named("mem_capacity"), - Types - .optional(PrimitiveType.PrimitiveTypeName.INT32) - .named("gpu_count"), - Types - .optional(PrimitiveType.PrimitiveTypeName.DOUBLE) - .named("gpu_capacity"), - Types - .buildGroup(Type.Repetition.OPTIONAL) - .addField( - Types.repeatedGroup() - .addField( - Types.optional(PrimitiveType.PrimitiveTypeName.BINARY) - .`as`(LogicalTypeAnnotation.stringType()) - .named("element"), - ) - .named("list"), - ) - .`as`(LogicalTypeAnnotation.listType()) - .named("parents"), - Types - .buildGroup(Type.Repetition.OPTIONAL) - .addField( - Types.repeatedGroup() - .addField( - Types.optional(PrimitiveType.PrimitiveTypeName.BINARY) - .`as`(LogicalTypeAnnotation.stringType()) - .named("element"), - ) - .named("list"), - ) - .`as`(LogicalTypeAnnotation.listType()) - .named("children"), - Types - .optional(PrimitiveType.PrimitiveTypeName.BINARY) - .`as`(LogicalTypeAnnotation.stringType()) - .named("nature"), - Types - .optional(PrimitiveType.PrimitiveTypeName.INT64) - .named("deadline"), - ) - .named("resource") - - /** - * Parquet read schema for the "resources" table in the trace. - */ - @JvmStatic - val READ_SCHEMA: MessageType = - READ_SCHEMA_V2_2 - } -} diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/opendc/parquet/ResourceStateReadSupport.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/opendc/parquet/ResourceStateReadSupport.kt deleted file mode 100644 index 53e594de..00000000 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/opendc/parquet/ResourceStateReadSupport.kt +++ /dev/null @@ -1,161 +0,0 @@ -/* - * Copyright (c) 2022 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 org.opendc.trace.formats.opendc.parquet - -import org.apache.hadoop.conf.Configuration -import org.apache.parquet.hadoop.api.InitContext -import org.apache.parquet.hadoop.api.ReadSupport -import org.apache.parquet.io.api.RecordMaterializer -import org.apache.parquet.schema.LogicalTypeAnnotation -import org.apache.parquet.schema.MessageType -import org.apache.parquet.schema.PrimitiveType -import org.apache.parquet.schema.Types -import org.opendc.trace.TableColumn -import org.opendc.trace.conv.resourceCpuCount -import org.opendc.trace.conv.resourceID -import org.opendc.trace.conv.resourceStateCpuUsage -import org.opendc.trace.conv.resourceStateDuration -import org.opendc.trace.conv.resourceStateTimestamp - -/** - * A [ReadSupport] instance for [ResourceState] objects. - */ -internal class ResourceStateReadSupport(private val projection: List<String>?) : ReadSupport<ResourceState>() { - /** - * Mapping from field names to [TableColumn]s. - */ - private val fieldMap = - mapOf( - "id" to resourceID, - "time" to resourceStateTimestamp, - "timestamp" to resourceStateTimestamp, - "duration" to resourceStateDuration, - "cores" to resourceCpuCount, - "cpu_count" to resourceCpuCount, - "cpuUsage" to resourceStateCpuUsage, - "cpu_usage" to resourceStateCpuUsage, - ) - - override fun init(context: InitContext): ReadContext { - val projectedSchema = - if (projection != null) { - Types.buildMessage() - .apply { - val projectionSet = projection.toSet() - - for (field in READ_SCHEMA.fields) { - val col = fieldMap[field.name] ?: continue - if (col in projectionSet) { - addField(field) - } - } - } - .named(READ_SCHEMA.name) - } else { - READ_SCHEMA - } - - return ReadContext(projectedSchema) - } - - override fun prepareForRead( - configuration: Configuration, - keyValueMetaData: Map<String, String>, - fileSchema: MessageType, - readContext: ReadContext, - ): RecordMaterializer<ResourceState> = ResourceStateRecordMaterializer(readContext.requestedSchema) - - companion object { - /** - * Parquet read schema (version 2.0) for the "resource states" table in the trace. - */ - @JvmStatic - val READ_SCHEMA_V2_0: MessageType = - Types.buildMessage() - .addFields( - Types - .required(PrimitiveType.PrimitiveTypeName.BINARY) - .`as`(LogicalTypeAnnotation.stringType()) - .named("id"), - Types - .required(PrimitiveType.PrimitiveTypeName.INT64) - .`as`(LogicalTypeAnnotation.timestampType(true, LogicalTypeAnnotation.TimeUnit.MILLIS)) - .named("time"), - Types - .required(PrimitiveType.PrimitiveTypeName.INT64) - .named("duration"), - Types - .required(PrimitiveType.PrimitiveTypeName.INT32) - .named("cores"), - Types - .required(PrimitiveType.PrimitiveTypeName.DOUBLE) - .named("cpuUsage"), - Types - .optional(PrimitiveType.PrimitiveTypeName.INT32) - .named("gpuCount"), - Types - .optional(PrimitiveType.PrimitiveTypeName.DOUBLE) - .named("gpuUsage"), - ) - .named("resource_state") - - /** - * Parquet read schema (version 2.1) for the "resource states" table in the trace. - */ - @JvmStatic - val READ_SCHEMA_V2_1: MessageType = - Types.buildMessage() - .addFields( - Types - .required(PrimitiveType.PrimitiveTypeName.BINARY) - .`as`(LogicalTypeAnnotation.stringType()) - .named("id"), - Types - .required(PrimitiveType.PrimitiveTypeName.INT64) - .`as`(LogicalTypeAnnotation.timestampType(true, LogicalTypeAnnotation.TimeUnit.MILLIS)) - .named("timestamp"), - Types - .required(PrimitiveType.PrimitiveTypeName.INT64) - .named("duration"), - Types - .required(PrimitiveType.PrimitiveTypeName.INT32) - .named("cpu_count"), - Types - .required(PrimitiveType.PrimitiveTypeName.DOUBLE) - .named("cpu_usage"), - Types - .optional(PrimitiveType.PrimitiveTypeName.INT32) - .named("gpu_count"), - Types - .optional(PrimitiveType.PrimitiveTypeName.DOUBLE) - .named("gpu_usage"), - ) - .named("resource_state") - - /** - * Parquet read schema for the "resource states" table in the trace. - */ - @JvmStatic - val READ_SCHEMA: MessageType = READ_SCHEMA_V2_0.union(READ_SCHEMA_V2_1) - } -} diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/swf/SwfTaskTableReader.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/swf/SwfTaskTableReader.kt deleted file mode 100644 index 5a79fd6f..00000000 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/swf/SwfTaskTableReader.kt +++ /dev/null @@ -1,236 +0,0 @@ -/* - * Copyright (c) 2021 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 org.opendc.trace.swf - -import org.opendc.trace.TableReader -import org.opendc.trace.conv.TASK_ALLOC_NCPUS -import org.opendc.trace.conv.TASK_GROUP_ID -import org.opendc.trace.conv.TASK_ID -import org.opendc.trace.conv.TASK_PARENTS -import org.opendc.trace.conv.TASK_REQ_NCPUS -import org.opendc.trace.conv.TASK_RUNTIME -import org.opendc.trace.conv.TASK_STATUS -import org.opendc.trace.conv.TASK_SUBMIT_TIME -import org.opendc.trace.conv.TASK_USER_ID -import org.opendc.trace.conv.TASK_WAIT_TIME -import java.io.BufferedReader -import java.time.Duration -import java.time.Instant -import java.util.UUID - -/** - * A [TableReader] implementation for the SWF format. - */ -internal class SwfTaskTableReader(private val reader: BufferedReader) : TableReader { - /** - * A flag to indicate the state of the reader - */ - private var state = State.Pending - - /** - * The current row. - */ - private var fields = emptyList<String>() - - /** - * A [Regex] object to match whitespace. - */ - private val whitespace = "\\s+".toRegex() - - override fun nextRow(): Boolean { - var line: String? - var num = 0 - - val state = state - if (state == State.Closed) { - return false - } else if (state == State.Pending) { - this.state = State.Active - } - - while (true) { - line = reader.readLine() - - if (line == null) { - this.state = State.Closed - return false - } - num++ - - if (line.isBlank()) { - // Ignore empty lines - continue - } else if (line.startsWith(";")) { - // Ignore comments for now - continue - } - - break - } - - fields = line!!.trim().split(whitespace) - - if (fields.size < 18) { - throw IllegalArgumentException("Invalid format at line $line") - } - - return true - } - - override fun resolve(name: String): Int { - return when (name) { - TASK_ID -> colJobID - TASK_SUBMIT_TIME -> colSubmitTime - TASK_WAIT_TIME -> colWaitTime - TASK_RUNTIME -> colRunTime - TASK_ALLOC_NCPUS -> colAllocNcpus - TASK_REQ_NCPUS -> colReqNcpus - TASK_STATUS -> colStatus - TASK_USER_ID -> colUserID - TASK_GROUP_ID -> colGroupID - TASK_PARENTS -> colParentJob - else -> -1 - } - } - - override fun isNull(index: Int): Boolean { - require(index in colJobID..colParentThinkTime) { "Invalid column index" } - return false - } - - override fun getBoolean(index: Int): Boolean { - throw IllegalArgumentException("Invalid column") - } - - override fun getInt(index: Int): Int { - check(state == State.Active) { "No active row" } - return when (index) { - colReqNcpus, colAllocNcpus, colStatus, colGroupID, colUserID -> fields[index].toInt(10) - else -> throw IllegalArgumentException("Invalid column") - } - } - - override fun getLong(index: Int): Long { - throw IllegalArgumentException("Invalid column") - } - - override fun getFloat(index: Int): Float { - throw IllegalArgumentException("Invalid column") - } - - override fun getDouble(index: Int): Double { - throw IllegalArgumentException("Invalid column") - } - - override fun getString(index: Int): String { - check(state == State.Active) { "No active row" } - return when (index) { - colJobID -> fields[index] - else -> throw IllegalArgumentException("Invalid column") - } - } - - override fun getUUID(index: Int): UUID? { - throw IllegalArgumentException("Invalid column") - } - - override fun getInstant(index: Int): Instant? { - check(state == State.Active) { "No active row" } - return when (index) { - colSubmitTime -> Instant.ofEpochSecond(fields[index].toLong(10)) - else -> throw IllegalArgumentException("Invalid column") - } - } - - override fun getDuration(index: Int): Duration? { - check(state == State.Active) { "No active row" } - return when (index) { - colWaitTime, colRunTime -> Duration.ofSeconds(fields[index].toLong(10)) - else -> throw IllegalArgumentException("Invalid column") - } - } - - override fun <T> getList( - index: Int, - elementType: Class<T>, - ): List<T>? { - throw IllegalArgumentException("Invalid column") - } - - override fun <T> getSet( - index: Int, - elementType: Class<T>, - ): Set<T>? { - check(state == State.Active) { "No active row" } - @Suppress("UNCHECKED_CAST") - return when (index) { - colParentJob -> { - require(elementType.isAssignableFrom(String::class.java)) - val parent = fields[index].toLong(10) - if (parent < 0) emptySet() else setOf(parent) - } - else -> throw IllegalArgumentException("Invalid column") - } as Set<T>? - } - - override fun <K, V> getMap( - index: Int, - keyType: Class<K>, - valueType: Class<V>, - ): Map<K, V>? { - throw IllegalArgumentException("Invalid column") - } - - override fun close() { - reader.close() - state = State.Closed - } - - /** - * Default column indices for the SWF format. - */ - private val colJobID = 0 - private val colSubmitTime = 1 - private val colWaitTime = 2 - private val colRunTime = 3 - private val colAllocNcpus = 4 - private val colAvgCpuTime = 5 - private val colUsedMem = 6 - private val colReqNcpus = 7 - private val colReqTime = 8 - private val colReqMem = 9 - private val colStatus = 10 - private val colUserID = 11 - private val colGroupID = 12 - private val colExecNum = 13 - private val colQueueNum = 14 - private val colPartNum = 15 - private val colParentJob = 16 - private val colParentThinkTime = 17 - - private enum class State { - Pending, - Active, - Closed, - } -} diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/swf/SwfTraceFormat.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/swf/SwfTraceFormat.kt deleted file mode 100644 index d59b07b4..00000000 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/swf/SwfTraceFormat.kt +++ /dev/null @@ -1,100 +0,0 @@ -/* - * 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 org.opendc.trace.swf - -import org.opendc.trace.TableColumn -import org.opendc.trace.TableColumnType -import org.opendc.trace.TableReader -import org.opendc.trace.TableWriter -import org.opendc.trace.conv.TABLE_TASKS -import org.opendc.trace.conv.TASK_ALLOC_NCPUS -import org.opendc.trace.conv.TASK_GROUP_ID -import org.opendc.trace.conv.TASK_ID -import org.opendc.trace.conv.TASK_PARENTS -import org.opendc.trace.conv.TASK_REQ_NCPUS -import org.opendc.trace.conv.TASK_RUNTIME -import org.opendc.trace.conv.TASK_STATUS -import org.opendc.trace.conv.TASK_SUBMIT_TIME -import org.opendc.trace.conv.TASK_USER_ID -import org.opendc.trace.conv.TASK_WAIT_TIME -import org.opendc.trace.spi.TableDetails -import org.opendc.trace.spi.TraceFormat -import java.nio.file.Path -import kotlin.io.path.bufferedReader - -/** - * Support for the Standard Workload Format (SWF) in OpenDC. - * - * The standard is defined by the PWA, see here: https://www.cse.huji.ac.il/labs/parallel/workload/swf.html - */ -public class SwfTraceFormat : TraceFormat { - override val name: String = "swf" - - override fun create(path: Path) { - throw UnsupportedOperationException("Writing not supported for this format") - } - - override fun getTables(path: Path): List<String> = listOf(TABLE_TASKS) - - override fun getDetails( - path: Path, - table: String, - ): TableDetails { - return when (table) { - TABLE_TASKS -> - TableDetails( - listOf( - TableColumn(TASK_ID, TableColumnType.String), - TableColumn(TASK_SUBMIT_TIME, TableColumnType.Instant), - TableColumn(TASK_WAIT_TIME, TableColumnType.Duration), - TableColumn(TASK_RUNTIME, TableColumnType.Duration), - TableColumn(TASK_REQ_NCPUS, TableColumnType.Int), - TableColumn(TASK_ALLOC_NCPUS, TableColumnType.Int), - TableColumn(TASK_PARENTS, TableColumnType.Set(TableColumnType.String)), - TableColumn(TASK_STATUS, TableColumnType.Int), - TableColumn(TASK_GROUP_ID, TableColumnType.Int), - TableColumn(TASK_USER_ID, TableColumnType.Int), - ), - ) - else -> throw IllegalArgumentException("Table $table not supported") - } - } - - override fun newReader( - path: Path, - table: String, - projection: List<String>?, - ): TableReader { - return when (table) { - TABLE_TASKS -> SwfTaskTableReader(path.bufferedReader()) - else -> throw IllegalArgumentException("Table $table not supported") - } - } - - override fun newWriter( - path: Path, - table: String, - ): TableWriter { - throw UnsupportedOperationException("Writing not supported for this format") - } -} diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/wfformat/WfFormatTaskTableReader.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/wfformat/WfFormatTaskTableReader.kt deleted file mode 100644 index 8f84e51f..00000000 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/wfformat/WfFormatTaskTableReader.kt +++ /dev/null @@ -1,314 +0,0 @@ -/* - * Copyright (c) 2021 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 org.opendc.trace.wfformat - -import com.fasterxml.jackson.core.JsonParseException -import com.fasterxml.jackson.core.JsonParser -import com.fasterxml.jackson.core.JsonToken -import org.opendc.trace.TableColumnType -import org.opendc.trace.TableReader -import org.opendc.trace.conv.TASK_CHILDREN -import org.opendc.trace.conv.TASK_ID -import org.opendc.trace.conv.TASK_PARENTS -import org.opendc.trace.conv.TASK_REQ_NCPUS -import org.opendc.trace.conv.TASK_RUNTIME -import org.opendc.trace.conv.TASK_WORKFLOW_ID -import org.opendc.trace.util.convertTo -import java.time.Duration -import java.time.Instant -import java.util.UUID -import kotlin.math.roundToInt - -/** - * A [TableReader] implementation for the WfCommons workload trace format. - */ -internal class WfFormatTaskTableReader(private val parser: JsonParser) : TableReader { - /** - * The current nesting of the parser. - */ - private var level: ParserLevel = ParserLevel.TOP - - override fun nextRow(): Boolean { - reset() - - var hasJob = false - - while (!hasJob) { - when (level) { - ParserLevel.TOP -> { - val token = parser.nextToken() - - // Check whether the document is not empty and starts with an object - if (token == null) { - parser.close() - break - } else if (token != JsonToken.START_OBJECT) { - throw JsonParseException(parser, "Expected object", parser.currentLocation) - } else { - level = ParserLevel.TRACE - } - } - ParserLevel.TRACE -> { - // Seek for the workflow object in the file - if (!seekWorkflow()) { - parser.close() - break - } else if (!parser.isExpectedStartObjectToken) { - throw JsonParseException(parser, "Expected object", parser.currentLocation) - } else { - level = ParserLevel.WORKFLOW - } - } - ParserLevel.WORKFLOW -> { - // Seek for the jobs object in the file - level = - if (!seekJobs()) { - ParserLevel.TRACE - } else if (!parser.isExpectedStartArrayToken) { - throw JsonParseException(parser, "Expected array", parser.currentLocation) - } else { - ParserLevel.JOB - } - } - ParserLevel.JOB -> { - when (parser.nextToken()) { - JsonToken.END_ARRAY -> level = ParserLevel.WORKFLOW - JsonToken.START_OBJECT -> { - parseJob() - hasJob = true - break - } - else -> throw JsonParseException(parser, "Unexpected token", parser.currentLocation) - } - } - } - } - - return hasJob - } - - override fun resolve(name: String): Int { - return when (name) { - TASK_ID -> colID - TASK_WORKFLOW_ID -> colWorkflowID - TASK_RUNTIME -> colRuntime - TASK_REQ_NCPUS -> colNproc - TASK_PARENTS -> colParents - TASK_CHILDREN -> colChildren - else -> -1 - } - } - - override fun isNull(index: Int): Boolean { - require(index in 0..colChildren) { "Invalid column value" } - return false - } - - override fun getBoolean(index: Int): Boolean { - throw IllegalArgumentException("Invalid column") - } - - override fun getInt(index: Int): Int { - checkActive() - return when (index) { - colNproc -> cores - else -> throw IllegalArgumentException("Invalid column") - } - } - - override fun getLong(index: Int): Long { - throw IllegalArgumentException("Invalid column") - } - - override fun getFloat(index: Int): Float { - throw IllegalArgumentException("Invalid column") - } - - override fun getDouble(index: Int): Double { - throw IllegalArgumentException("Invalid column") - } - - override fun getString(index: Int): String? { - checkActive() - return when (index) { - colID -> id - colWorkflowID -> workflowId - else -> throw IllegalArgumentException("Invalid column") - } - } - - override fun getUUID(index: Int): UUID? { - throw IllegalArgumentException("Invalid column") - } - - override fun getInstant(index: Int): Instant? { - throw IllegalArgumentException("Invalid column") - } - - override fun getDuration(index: Int): Duration? { - checkActive() - return when (index) { - colRuntime -> runtime - else -> throw IllegalArgumentException("Invalid column") - } - } - - override fun <T> getList( - index: Int, - elementType: Class<T>, - ): List<T>? { - throw IllegalArgumentException("Invalid column") - } - - override fun <T> getSet( - index: Int, - elementType: Class<T>, - ): Set<T>? { - checkActive() - return when (index) { - colParents -> typeParents.convertTo(parents, elementType) - colChildren -> typeChildren.convertTo(children, elementType) - else -> throw IllegalArgumentException("Invalid column") - } - } - - override fun <K, V> getMap( - index: Int, - keyType: Class<K>, - valueType: Class<V>, - ): Map<K, V>? { - throw IllegalArgumentException("Invalid column") - } - - override fun close() { - parser.close() - } - - /** - * Helper method to check if the reader is active. - */ - private fun checkActive() { - check(level != ParserLevel.TOP && !parser.isClosed) { "No active row. Did you call nextRow()?" } - } - - /** - * Parse the trace and seek until the workflow description. - */ - private fun seekWorkflow(): Boolean { - while (parser.nextValue() != JsonToken.END_OBJECT && !parser.isClosed) { - when (parser.currentName) { - "name" -> workflowId = parser.text - "workflow" -> return true - else -> parser.skipChildren() - } - } - - return false - } - - /** - * Parse the workflow description in the file and seek until the first job. - */ - private fun seekJobs(): Boolean { - while (parser.nextValue() != JsonToken.END_OBJECT) { - when (parser.currentName) { - "jobs" -> return true - else -> parser.skipChildren() - } - } - - return false - } - - /** - * Parse a single job in the file. - */ - private fun parseJob() { - while (parser.nextValue() != JsonToken.END_OBJECT) { - when (parser.currentName) { - "name" -> id = parser.text - "parents" -> parents = parseIds() - "children" -> children = parseIds() - "runtime" -> runtime = Duration.ofSeconds(parser.numberValue.toLong()) - "cores" -> cores = parser.floatValue.roundToInt() - else -> parser.skipChildren() - } - } - } - - /** - * Parse the parents/children of a job. - */ - private fun parseIds(): Set<String> { - if (!parser.isExpectedStartArrayToken) { - throw JsonParseException(parser, "Expected array", parser.currentLocation) - } - - val ids = mutableSetOf<String>() - - while (parser.nextToken() != JsonToken.END_ARRAY) { - if (parser.currentToken != JsonToken.VALUE_STRING) { - throw JsonParseException(parser, "Expected token", parser.currentLocation) - } - - ids.add(parser.valueAsString) - } - - return ids - } - - private enum class ParserLevel { - TOP, - TRACE, - WORKFLOW, - JOB, - } - - /** - * State fields for the parser. - */ - private var id: String? = null - private var workflowId: String? = null - private var runtime: Duration? = null - private var parents: Set<String>? = null - private var children: Set<String>? = null - private var cores = -1 - - private fun reset() { - id = null - runtime = null - parents = null - children = null - cores = -1 - } - - private val colID = 0 - private val colWorkflowID = 1 - private val colRuntime = 3 - private val colNproc = 4 - private val colParents = 5 - private val colChildren = 6 - - private val typeParents = TableColumnType.Set(TableColumnType.String) - private val typeChildren = TableColumnType.Set(TableColumnType.String) -} diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/wfformat/WfFormatTraceFormat.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/wfformat/WfFormatTraceFormat.kt deleted file mode 100644 index 2178fac6..00000000 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/wfformat/WfFormatTraceFormat.kt +++ /dev/null @@ -1,95 +0,0 @@ -/* - * Copyright (c) 2021 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 org.opendc.trace.wfformat - -import com.fasterxml.jackson.core.JsonFactory -import org.opendc.trace.TableColumn -import org.opendc.trace.TableColumnType -import org.opendc.trace.TableReader -import org.opendc.trace.TableWriter -import org.opendc.trace.conv.TABLE_TASKS -import org.opendc.trace.conv.TASK_CHILDREN -import org.opendc.trace.conv.TASK_ID -import org.opendc.trace.conv.TASK_PARENTS -import org.opendc.trace.conv.TASK_REQ_NCPUS -import org.opendc.trace.conv.TASK_RUNTIME -import org.opendc.trace.conv.TASK_WORKFLOW_ID -import org.opendc.trace.spi.TableDetails -import org.opendc.trace.spi.TraceFormat -import java.nio.file.Path - -/** - * A [TraceFormat] implementation for the WfCommons workload trace format. - */ -public class WfFormatTraceFormat : TraceFormat { - /** - * The [JsonFactory] that is used to created JSON parsers. - */ - private val factory = JsonFactory() - - override val name: String = "wfformat" - - override fun create(path: Path) { - throw UnsupportedOperationException("Writing not supported for this format") - } - - override fun getTables(path: Path): List<String> = listOf(TABLE_TASKS) - - override fun getDetails( - path: Path, - table: String, - ): TableDetails { - return when (table) { - TABLE_TASKS -> - TableDetails( - listOf( - TableColumn(TASK_ID, TableColumnType.String), - TableColumn(TASK_WORKFLOW_ID, TableColumnType.String), - TableColumn(TASK_RUNTIME, TableColumnType.Duration), - TableColumn(TASK_REQ_NCPUS, TableColumnType.Int), - TableColumn(TASK_PARENTS, TableColumnType.Set(TableColumnType.String)), - TableColumn(TASK_CHILDREN, TableColumnType.Set(TableColumnType.String)), - ), - ) - else -> throw IllegalArgumentException("Table $table not supported") - } - } - - override fun newReader( - path: Path, - table: String, - projection: List<String>?, - ): TableReader { - return when (table) { - TABLE_TASKS -> WfFormatTaskTableReader(factory.createParser(path.toFile())) - else -> throw IllegalArgumentException("Table $table not supported") - } - } - - override fun newWriter( - path: Path, - table: String, - ): TableWriter { - throw UnsupportedOperationException("Writing not supported for this format") - } -} diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/opendc/OdcVmResourceStateTableReader.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/FragmentTableReader.kt index d474e0ec..947746c6 100644 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/opendc/OdcVmResourceStateTableReader.kt +++ b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/FragmentTableReader.kt @@ -20,17 +20,14 @@ * SOFTWARE. */ -package org.opendc.trace.formats.opendc +package org.opendc.trace.formats.workload import org.opendc.trace.TableReader -import org.opendc.trace.conv.resourceCpuCount -import org.opendc.trace.conv.resourceGpuCount -import org.opendc.trace.conv.resourceID -import org.opendc.trace.conv.resourceStateCpuUsage -import org.opendc.trace.conv.resourceStateDuration -import org.opendc.trace.conv.resourceStateGpuUsage -import org.opendc.trace.conv.resourceStateTimestamp -import org.opendc.trace.formats.opendc.parquet.ResourceState +import org.opendc.trace.conv.FRAGMENT_CPU_USAGE +import org.opendc.trace.conv.FRAGMENT_DURATION +import org.opendc.trace.conv.FRAGMENT_GPU_USAGE +import org.opendc.trace.conv.TASK_ID +import org.opendc.trace.formats.workload.parquet.Fragment import org.opendc.trace.util.parquet.LocalParquetReader import java.time.Duration import java.time.Instant @@ -39,11 +36,11 @@ import java.util.UUID /** * A [TableReader] implementation for the OpenDC virtual machine trace format. */ -internal class OdcVmResourceStateTableReader(private val reader: LocalParquetReader<ResourceState>) : TableReader { +internal class FragmentTableReader(private val reader: LocalParquetReader<Fragment>) : TableReader { /** * The current record. */ - private var record: ResourceState? = null + private var record: Fragment? = null override fun nextRow(): Boolean { try { @@ -58,23 +55,16 @@ internal class OdcVmResourceStateTableReader(private val reader: LocalParquetRea } private val colID = 0 - private val colTimestamp = 1 - private val colDuration = 2 - private val colCpuCount = 3 - private val colCpuUsage = 4 - private val colGpuCount = 5 - private val colGpuUsage = 6 - private val colMemoryCapacity = 7 + private val colDuration = 1 + private val colCpuUsage = 2 + private val colGpuUsage = 3 override fun resolve(name: String): Int { return when (name) { - resourceID -> colID - resourceStateTimestamp -> colTimestamp - resourceStateDuration -> colDuration - resourceCpuCount -> colCpuCount - resourceStateCpuUsage -> colCpuUsage - resourceGpuCount -> colGpuCount - resourceStateGpuUsage -> colGpuUsage + TASK_ID -> colID + FRAGMENT_DURATION -> colDuration + FRAGMENT_CPU_USAGE -> colCpuUsage + FRAGMENT_GPU_USAGE -> colGpuUsage else -> -1 } } @@ -91,8 +81,7 @@ internal class OdcVmResourceStateTableReader(private val reader: LocalParquetRea override fun getInt(index: Int): Int { val record = checkNotNull(record) { "Reader in invalid state" } return when (index) { - colCpuCount -> record.cpuCount - colGpuCount -> record.gpuCount + colID -> record.id else -> throw IllegalArgumentException("Invalid column or type [index $index]") } } @@ -115,12 +104,7 @@ internal class OdcVmResourceStateTableReader(private val reader: LocalParquetRea } override fun getString(index: Int): String { - val record = checkNotNull(record) { "Reader in invalid state" } - - return when (index) { - colID -> record.id - else -> throw IllegalArgumentException("Invalid column index $index") - } + throw IllegalArgumentException("Invalid column index $index") } override fun getUUID(index: Int): UUID? { @@ -128,12 +112,7 @@ internal class OdcVmResourceStateTableReader(private val reader: LocalParquetRea } override fun getInstant(index: Int): Instant { - val record = checkNotNull(record) { "Reader in invalid state" } - - return when (index) { - colTimestamp -> record.timestamp - else -> throw IllegalArgumentException("Invalid column index $index") - } + throw IllegalArgumentException("Invalid column index $index") } override fun getDuration(index: Int): Duration { diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/opendc/OdcVmResourceStateTableWriter.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/FragmentTableWriter.kt index c6f117d2..33cd9e17 100644 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/opendc/OdcVmResourceStateTableWriter.kt +++ b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/FragmentTableWriter.kt @@ -20,18 +20,15 @@ * SOFTWARE. */ -package org.opendc.trace.formats.opendc +package org.opendc.trace.formats.workload import org.apache.parquet.hadoop.ParquetWriter import org.opendc.trace.TableWriter -import org.opendc.trace.conv.resourceCpuCount -import org.opendc.trace.conv.resourceGpuCount -import org.opendc.trace.conv.resourceID -import org.opendc.trace.conv.resourceStateCpuUsage -import org.opendc.trace.conv.resourceStateDuration -import org.opendc.trace.conv.resourceStateGpuUsage -import org.opendc.trace.conv.resourceStateTimestamp -import org.opendc.trace.formats.opendc.parquet.ResourceState +import org.opendc.trace.conv.FRAGMENT_CPU_USAGE +import org.opendc.trace.conv.FRAGMENT_DURATION +import org.opendc.trace.conv.FRAGMENT_GPU_USAGE +import org.opendc.trace.conv.TASK_ID +import org.opendc.trace.formats.workload.parquet.Fragment import java.time.Duration import java.time.Instant import java.util.UUID @@ -39,27 +36,21 @@ import java.util.UUID /** * A [TableWriter] implementation for the OpenDC virtual machine trace format. */ -internal class OdcVmResourceStateTableWriter(private val writer: ParquetWriter<ResourceState>) : TableWriter { +internal class FragmentTableWriter(private val writer: ParquetWriter<Fragment>) : TableWriter { /** * The current state for the record that is being written. */ private var localIsActive = false - private var localID: String = "" - private var localTimestamp: Instant = Instant.MIN + private var localID: Int = -99 private var localDuration: Duration = Duration.ZERO - private var localCpuCount: Int = 0 private var localCpuUsage: Double = Double.NaN - private var localGpuCount: Int = 0 private var localGpuUsage: Double = Double.NaN override fun startRow() { localIsActive = true - localID = "" - localTimestamp = Instant.MIN + localID = -99 localDuration = Duration.ZERO - localCpuCount = 0 localCpuUsage = Double.NaN - localGpuCount = 0 localGpuUsage = Double.NaN } @@ -67,23 +58,19 @@ internal class OdcVmResourceStateTableWriter(private val writer: ParquetWriter<R check(localIsActive) { "No active row" } localIsActive = false - check(lastId != localID || localTimestamp >= lastTimestamp) { "Records need to be ordered by (id, timestamp)" } + check(lastId != localID) { "Records need to be ordered by (id, timestamp)" } - writer.write(ResourceState(localID, localTimestamp, localDuration, localCpuCount, localCpuUsage, localGpuCount, localGpuUsage)) + writer.write(Fragment(localID, localDuration, localCpuUsage, localGpuUsage)) lastId = localID - lastTimestamp = localTimestamp } override fun resolve(name: String): Int { return when (name) { - resourceID -> colID - resourceStateTimestamp -> colTimestamp - resourceStateDuration -> colDuration - resourceCpuCount -> colCpuCount - resourceStateCpuUsage -> colCpuUsage - resourceGpuCount -> colGpuCount - resourceStateGpuUsage -> colGpuUsage + TASK_ID -> colID + FRAGMENT_DURATION -> colDuration + FRAGMENT_CPU_USAGE -> colCpuUsage + FRAGMENT_GPU_USAGE -> colGpuUsage else -> -1 } } @@ -101,8 +88,7 @@ internal class OdcVmResourceStateTableWriter(private val writer: ParquetWriter<R ) { check(localIsActive) { "No active row" } when (index) { - colCpuCount -> localCpuCount = value - colGpuCount -> localGpuCount = value + colID -> localID = value else -> throw IllegalArgumentException("Invalid column or type [index $index]") } } @@ -137,12 +123,7 @@ internal class OdcVmResourceStateTableWriter(private val writer: ParquetWriter<R index: Int, value: String, ) { - check(localIsActive) { "No active row" } - - when (index) { - colID -> localID = value - else -> throw IllegalArgumentException("Invalid column or type [index $index]") - } + throw IllegalArgumentException("Invalid column or type [index $index]") } override fun setUUID( @@ -156,12 +137,7 @@ internal class OdcVmResourceStateTableWriter(private val writer: ParquetWriter<R index: Int, value: Instant, ) { - check(localIsActive) { "No active row" } - - when (index) { - colTimestamp -> localTimestamp = value - else -> throw IllegalArgumentException("Invalid column or type [index $index]") - } + throw IllegalArgumentException("Invalid column or type [index $index]") } override fun setDuration( @@ -208,14 +184,10 @@ internal class OdcVmResourceStateTableWriter(private val writer: ParquetWriter<R /** * Last column values that are used to check for correct partitioning. */ - private var lastId: String? = null - private var lastTimestamp: Instant = Instant.MAX + private var lastId: Int? = null private val colID = 0 - private val colTimestamp = 1 - private val colDuration = 2 - private val colCpuCount = 3 - private val colCpuUsage = 4 - private val colGpuCount = 5 - private val colGpuUsage = 6 + private val colDuration = 1 + private val colCpuUsage = 2 + private val colGpuUsage = 3 } diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/opendc/OdcVmResourceTableReader.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/TaskTableReader.kt index 495a5d75..6c700b0c 100644 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/opendc/OdcVmResourceTableReader.kt +++ b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/TaskTableReader.kt @@ -20,23 +20,24 @@ * SOFTWARE. */ -package org.opendc.trace.formats.opendc +package org.opendc.trace.formats.workload import org.opendc.trace.TableColumnType import org.opendc.trace.TableReader -import org.opendc.trace.conv.resourceChildren -import org.opendc.trace.conv.resourceCpuCapacity -import org.opendc.trace.conv.resourceCpuCount -import org.opendc.trace.conv.resourceDeadline -import org.opendc.trace.conv.resourceDuration -import org.opendc.trace.conv.resourceGpuCapacity -import org.opendc.trace.conv.resourceGpuCount -import org.opendc.trace.conv.resourceID -import org.opendc.trace.conv.resourceMemCapacity -import org.opendc.trace.conv.resourceNature -import org.opendc.trace.conv.resourceParents -import org.opendc.trace.conv.resourceSubmissionTime -import org.opendc.trace.formats.opendc.parquet.Resource +import org.opendc.trace.conv.TASK_CHILDREN +import org.opendc.trace.conv.TASK_CPU_CAPACITY +import org.opendc.trace.conv.TASK_CPU_COUNT +import org.opendc.trace.conv.TASK_DEADLINE +import org.opendc.trace.conv.TASK_DURATION +import org.opendc.trace.conv.TASK_GPU_CAPACITY +import org.opendc.trace.conv.TASK_GPU_COUNT +import org.opendc.trace.conv.TASK_ID +import org.opendc.trace.conv.TASK_MEM_CAPACITY +import org.opendc.trace.conv.TASK_NAME +import org.opendc.trace.conv.TASK_NATURE +import org.opendc.trace.conv.TASK_PARENTS +import org.opendc.trace.conv.TASK_SUBMISSION_TIME +import org.opendc.trace.formats.workload.parquet.Task import org.opendc.trace.util.convertTo import org.opendc.trace.util.parquet.LocalParquetReader import java.time.Duration @@ -46,11 +47,11 @@ import java.util.UUID /** * A [TableReader] implementation for the "resources table" in the OpenDC virtual machine trace format. */ -internal class OdcVmResourceTableReader(private val reader: LocalParquetReader<Resource>) : TableReader { +internal class TaskTableReader(private val reader: LocalParquetReader<Task>) : TableReader { /** * The current record. */ - private var record: Resource? = null + private var record: Task? = null override fun nextRow(): Boolean { try { @@ -65,35 +66,37 @@ internal class OdcVmResourceTableReader(private val reader: LocalParquetReader<R } private val colID = 0 - private val colSubmissionTime = 1 - private val colDurationTime = 2 - private val colCpuCount = 3 - private val colCpuCapacity = 4 - private val colMemCapacity = 5 - private val colGpuCapacity = 6 - private val colGpuCount = 7 - private val colParents = 8 - private val colChildren = 9 - private val colNature = 10 - private val colDeadline = 11 - - private val typeParents = TableColumnType.Set(TableColumnType.String) - private val typeChildren = TableColumnType.Set(TableColumnType.String) + private val colName = 1 + private val colSubmissionTime = 2 + private val colDurationTime = 3 + private val colCpuCount = 4 + private val colCpuCapacity = 5 + private val colMemCapacity = 6 + private val colGpuCapacity = 7 + private val colGpuCount = 8 + private val colParents = 9 + private val colChildren = 10 + private val colNature = 11 + private val colDeadline = 12 + + private val typeParents = TableColumnType.Set(TableColumnType.Int) + private val typeChildren = TableColumnType.Set(TableColumnType.Int) override fun resolve(name: String): Int { return when (name) { - resourceID -> colID - resourceSubmissionTime -> colSubmissionTime - resourceDuration -> colDurationTime - resourceCpuCount -> colCpuCount - resourceCpuCapacity -> colCpuCapacity - resourceMemCapacity -> colMemCapacity - resourceGpuCount -> colGpuCount - resourceGpuCapacity -> colGpuCapacity - resourceParents -> colParents - resourceChildren -> colChildren - resourceNature -> colNature - resourceDeadline -> colDeadline + TASK_ID -> colID + TASK_NAME -> colName + TASK_SUBMISSION_TIME -> colSubmissionTime + TASK_DURATION -> colDurationTime + TASK_CPU_COUNT -> colCpuCount + TASK_CPU_CAPACITY -> colCpuCapacity + TASK_MEM_CAPACITY -> colMemCapacity + TASK_GPU_COUNT -> colGpuCount + TASK_GPU_CAPACITY -> colGpuCapacity + TASK_PARENTS -> colParents + TASK_CHILDREN -> colChildren + TASK_NATURE -> colNature + TASK_DEADLINE -> colDeadline else -> -1 } } @@ -117,6 +120,7 @@ internal class OdcVmResourceTableReader(private val reader: LocalParquetReader<R val record = checkNotNull(record) { "Reader in invalid state" } return when (index) { + colID -> record.id colCpuCount -> record.cpuCount colGpuCount -> record.gpuCount else -> throw IllegalArgumentException("Invalid column") @@ -151,7 +155,7 @@ internal class OdcVmResourceTableReader(private val reader: LocalParquetReader<R val record = checkNotNull(record) { "Reader in invalid state" } return when (index) { - colID -> record.id + colName -> record.name colNature -> record.nature else -> throw IllegalArgumentException("Invalid column") } diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/opendc/OdcVmResourceTableWriter.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/TaskTableWriter.kt index 022e288a..39be36c1 100644 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/opendc/OdcVmResourceTableWriter.kt +++ b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/TaskTableWriter.kt @@ -20,23 +20,24 @@ * SOFTWARE. */ -package org.opendc.trace.formats.opendc +package org.opendc.trace.formats.workload import org.apache.parquet.hadoop.ParquetWriter import org.opendc.trace.TableWriter -import org.opendc.trace.conv.resourceChildren -import org.opendc.trace.conv.resourceCpuCapacity -import org.opendc.trace.conv.resourceCpuCount -import org.opendc.trace.conv.resourceDeadline -import org.opendc.trace.conv.resourceDuration -import org.opendc.trace.conv.resourceGpuCapacity -import org.opendc.trace.conv.resourceGpuCount -import org.opendc.trace.conv.resourceID -import org.opendc.trace.conv.resourceMemCapacity -import org.opendc.trace.conv.resourceNature -import org.opendc.trace.conv.resourceParents -import org.opendc.trace.conv.resourceSubmissionTime -import org.opendc.trace.formats.opendc.parquet.Resource +import org.opendc.trace.conv.TASK_CHILDREN +import org.opendc.trace.conv.TASK_CPU_CAPACITY +import org.opendc.trace.conv.TASK_CPU_COUNT +import org.opendc.trace.conv.TASK_DEADLINE +import org.opendc.trace.conv.TASK_DURATION +import org.opendc.trace.conv.TASK_GPU_CAPACITY +import org.opendc.trace.conv.TASK_GPU_COUNT +import org.opendc.trace.conv.TASK_ID +import org.opendc.trace.conv.TASK_MEM_CAPACITY +import org.opendc.trace.conv.TASK_NAME +import org.opendc.trace.conv.TASK_NATURE +import org.opendc.trace.conv.TASK_PARENTS +import org.opendc.trace.conv.TASK_SUBMISSION_TIME +import org.opendc.trace.formats.workload.parquet.Task import java.time.Duration import java.time.Instant import java.util.UUID @@ -44,12 +45,13 @@ import java.util.UUID /** * A [TableWriter] implementation for the OpenDC virtual machine trace format. */ -internal class OdcVmResourceTableWriter(private val writer: ParquetWriter<Resource>) : TableWriter { +internal class TaskTableWriter(private val writer: ParquetWriter<Task>) : TableWriter { /** * The current state for the record that is being written. */ private var localIsActive = false - private var localId: String = "" + private var localId: Int = -99 + private var localName: String = "" private var localSubmissionTime: Instant = Instant.MIN private var localDuration: Long = 0L private var localCpuCount: Int = 0 @@ -57,14 +59,15 @@ internal class OdcVmResourceTableWriter(private val writer: ParquetWriter<Resour private var localMemCapacity: Double = Double.NaN private var localGpuCount: Int = 0 private var localGpuCapacity: Double = Double.NaN - private var localParents = mutableSetOf<String>() - private var localChildren = mutableSetOf<String>() + private var localParents = mutableSetOf<Int>() + private var localChildren = mutableSetOf<Int>() private var localNature: String? = null private var localDeadline: Long = -1 override fun startRow() { localIsActive = true - localId = "" + localId = -99 + localName = "" localSubmissionTime = Instant.MIN localDuration = 0L localCpuCount = 0 @@ -82,8 +85,9 @@ internal class OdcVmResourceTableWriter(private val writer: ParquetWriter<Resour check(localIsActive) { "No active row" } localIsActive = false writer.write( - Resource( + Task( localId, + localName, localSubmissionTime, localDuration, localCpuCount, @@ -101,18 +105,19 @@ internal class OdcVmResourceTableWriter(private val writer: ParquetWriter<Resour override fun resolve(name: String): Int { return when (name) { - resourceID -> colID - resourceSubmissionTime -> colSubmissionTime - resourceDuration -> colDuration - resourceCpuCount -> colCpuCount - resourceCpuCapacity -> colCpuCapacity - resourceMemCapacity -> colMemCapacity - resourceGpuCount -> colGpuCount - resourceGpuCapacity -> colGpuCapacity - resourceParents -> colParents - resourceChildren -> colChildren - resourceNature -> colNature - resourceDeadline -> colDeadline + TASK_ID -> colID + TASK_NAME -> colID + TASK_SUBMISSION_TIME -> colSubmissionTime + TASK_DURATION -> colDuration + TASK_CPU_COUNT -> colCpuCount + TASK_CPU_CAPACITY -> colCpuCapacity + TASK_MEM_CAPACITY -> colMemCapacity + TASK_GPU_COUNT -> colGpuCount + TASK_GPU_CAPACITY -> colGpuCapacity + TASK_PARENTS -> colParents + TASK_CHILDREN -> colChildren + TASK_NATURE -> colNature + TASK_DEADLINE -> colDeadline else -> -1 } } @@ -130,6 +135,7 @@ internal class OdcVmResourceTableWriter(private val writer: ParquetWriter<Resour ) { check(localIsActive) { "No active row" } when (index) { + colID -> localId = value colCpuCount -> localCpuCount = value colGpuCount -> localGpuCount = value else -> throw IllegalArgumentException("Invalid column or type [index $index]") @@ -174,7 +180,7 @@ internal class OdcVmResourceTableWriter(private val writer: ParquetWriter<Resour ) { check(localIsActive) { "No active row" } when (index) { - colID -> localId = value + colName -> localName = value colNature -> localNature = value else -> throw IllegalArgumentException("Invalid column index $index") } @@ -235,15 +241,16 @@ internal class OdcVmResourceTableWriter(private val writer: ParquetWriter<Resour } private val colID = 0 - private val colSubmissionTime = 1 - private val colDuration = 2 - private val colCpuCount = 3 - private val colCpuCapacity = 4 - private val colMemCapacity = 5 - private val colGpuCount = 6 - private val colGpuCapacity = 7 - private val colParents = 8 - private val colChildren = 9 - private val colNature = 10 - private val colDeadline = 11 + private val colName = 1 + private val colSubmissionTime = 2 + private val colDuration = 3 + private val colCpuCount = 4 + private val colCpuCapacity = 5 + private val colMemCapacity = 6 + private val colGpuCount = 7 + private val colGpuCapacity = 8 + private val colParents = 9 + private val colChildren = 10 + private val colNature = 11 + private val colDeadline = 12 } diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/WorkloadTraceFormat.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/WorkloadTraceFormat.kt new file mode 100644 index 00000000..7af0650e --- /dev/null +++ b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/WorkloadTraceFormat.kt @@ -0,0 +1,165 @@ +/* + * Copyright (c) 2021 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 org.opendc.trace.formats.workload + +import org.apache.parquet.column.ParquetProperties +import org.apache.parquet.hadoop.ParquetFileWriter +import org.apache.parquet.hadoop.metadata.CompressionCodecName +import org.opendc.trace.TableColumn +import org.opendc.trace.TableColumnType +import org.opendc.trace.TableReader +import org.opendc.trace.TableWriter +import org.opendc.trace.conv.FRAGMENT_CPU_USAGE +import org.opendc.trace.conv.FRAGMENT_DURATION +import org.opendc.trace.conv.TABLE_FRAGMENTS +import org.opendc.trace.conv.TABLE_TASKS +import org.opendc.trace.conv.TASK_CHILDREN +import org.opendc.trace.conv.TASK_CPU_CAPACITY +import org.opendc.trace.conv.TASK_CPU_COUNT +import org.opendc.trace.conv.TASK_DEADLINE +import org.opendc.trace.conv.TASK_DURATION +import org.opendc.trace.conv.TASK_GPU_CAPACITY +import org.opendc.trace.conv.TASK_GPU_COUNT +import org.opendc.trace.conv.TASK_ID +import org.opendc.trace.conv.TASK_MEM_CAPACITY +import org.opendc.trace.conv.TASK_NATURE +import org.opendc.trace.conv.TASK_PARENTS +import org.opendc.trace.conv.TASK_SUBMISSION_TIME +import org.opendc.trace.formats.workload.parquet.FragmentReadSupport +import org.opendc.trace.formats.workload.parquet.FragmentWriteSupport +import org.opendc.trace.formats.workload.parquet.TaskReadSupport +import org.opendc.trace.formats.workload.parquet.TaskWriteSupport +import org.opendc.trace.spi.TableDetails +import org.opendc.trace.spi.TraceFormat +import org.opendc.trace.util.parquet.LocalParquetReader +import org.opendc.trace.util.parquet.LocalParquetWriter +import java.nio.file.Files +import java.nio.file.Path + +/** + * A [TraceFormat] implementation of the OpenDC virtual machine trace format. + */ +public class WorkloadTraceFormat : TraceFormat { + /** + * The name of this trace format. + */ + override val name: String = "workload" + + override fun create(path: Path) { + // Construct directory containing the trace files + Files.createDirectories(path) + + val tables = getTables(path) + + for (table in tables) { + val writer = newWriter(path, table) + writer.close() + } + } + + override fun getTables(path: Path): List<String> = listOf(TABLE_TASKS, TABLE_FRAGMENTS) + + override fun getDetails( + path: Path, + table: String, + ): TableDetails { + return when (table) { + TABLE_TASKS -> + TableDetails( + listOf( + TableColumn(TASK_ID, TableColumnType.String), + TableColumn(TASK_SUBMISSION_TIME, TableColumnType.Instant), + TableColumn(TASK_DURATION, TableColumnType.Long), + TableColumn(TASK_CPU_COUNT, TableColumnType.Int), + TableColumn(TASK_CPU_CAPACITY, TableColumnType.Double), + TableColumn(TASK_MEM_CAPACITY, TableColumnType.Double), + TableColumn(TASK_GPU_COUNT, TableColumnType.Int), + TableColumn(TASK_GPU_CAPACITY, TableColumnType.Double), + TableColumn(TASK_PARENTS, TableColumnType.Set(TableColumnType.String)), + TableColumn(TASK_CHILDREN, TableColumnType.Set(TableColumnType.String)), + TableColumn(TASK_NATURE, TableColumnType.String), + TableColumn(TASK_DEADLINE, TableColumnType.Long), + ), + ) + TABLE_FRAGMENTS -> + TableDetails( + listOf( + TableColumn(TASK_ID, TableColumnType.String), + TableColumn(FRAGMENT_DURATION, TableColumnType.Duration), + TableColumn(TASK_CPU_COUNT, TableColumnType.Int), + TableColumn(FRAGMENT_CPU_USAGE, TableColumnType.Double), + ), + ) + else -> throw IllegalArgumentException("Table $table not supported") + } + } + + override fun newReader( + path: Path, + table: String, + projection: List<String>?, + ): TableReader { + return when (table) { + TABLE_TASKS -> { + val reader = LocalParquetReader(path.resolve("tasks.parquet"), TaskReadSupport(projection)) + TaskTableReader(reader) + } + TABLE_FRAGMENTS -> { + val reader = LocalParquetReader(path.resolve("fragments.parquet"), FragmentReadSupport(projection)) + FragmentTableReader(reader) + } + else -> throw IllegalArgumentException("Table $table not supported") + } + } + + override fun newWriter( + path: Path, + table: String, + ): TableWriter { + return when (table) { + TABLE_TASKS -> { + val writer = + LocalParquetWriter.builder(path.resolve("tasks.parquet"), TaskWriteSupport()) + .withCompressionCodec(CompressionCodecName.ZSTD) + .withPageWriteChecksumEnabled(true) + .withWriterVersion(ParquetProperties.WriterVersion.PARQUET_2_0) + .withWriteMode(ParquetFileWriter.Mode.OVERWRITE) + .build() + TaskTableWriter(writer) + } + TABLE_FRAGMENTS -> { + val writer = + LocalParquetWriter.builder(path.resolve("fragments.parquet"), FragmentWriteSupport()) + .withCompressionCodec(CompressionCodecName.ZSTD) + .withDictionaryEncoding("id", true) + .withBloomFilterEnabled("id", true) + .withPageWriteChecksumEnabled(true) + .withWriterVersion(ParquetProperties.WriterVersion.PARQUET_2_0) + .withWriteMode(ParquetFileWriter.Mode.OVERWRITE) + .build() + FragmentTableWriter(writer) + } + else -> throw IllegalArgumentException("Table $table not supported") + } + } +} diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/opendc/parquet/ResourceState.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/parquet/Fragment.kt index 10fc6be4..44385088 100644 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/opendc/parquet/ResourceState.kt +++ b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/parquet/Fragment.kt @@ -20,17 +20,13 @@ * SOFTWARE. */ -package org.opendc.trace.formats.opendc.parquet +package org.opendc.trace.formats.workload.parquet import java.time.Duration -import java.time.Instant -internal class ResourceState( - val id: String, - val timestamp: Instant, +internal class Fragment( + val id: Int, val duration: Duration, - val cpuCount: Int, val cpuUsage: Double, - val gpuCount: Int, val gpuUsage: Double, ) diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/parquet/FragmentReadSupport.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/parquet/FragmentReadSupport.kt new file mode 100644 index 00000000..3fa914bc --- /dev/null +++ b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/parquet/FragmentReadSupport.kt @@ -0,0 +1,79 @@ +/* + * Copyright (c) 2022 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 org.opendc.trace.formats.workload.parquet + +import org.apache.hadoop.conf.Configuration +import org.apache.parquet.hadoop.api.InitContext +import org.apache.parquet.hadoop.api.ReadSupport +import org.apache.parquet.io.api.RecordMaterializer +import org.apache.parquet.schema.MessageType +import org.apache.parquet.schema.Types +import org.opendc.trace.TableColumn +import org.opendc.trace.conv.FRAGMENT_CPU_USAGE +import org.opendc.trace.conv.FRAGMENT_DURATION +import org.opendc.trace.conv.TASK_ID + +/** + * A [ReadSupport] instance for [Fragment] objects. + */ +internal class FragmentReadSupport(private val projection: List<String>?) : ReadSupport<Fragment>() { + /** + * Mapping from field names to [TableColumn]s. + */ + private val fieldMap = + mapOf( + "id" to TASK_ID, + "duration" to FRAGMENT_DURATION, + "cpuUsage" to FRAGMENT_CPU_USAGE, + "cpu_usage" to FRAGMENT_CPU_USAGE, + ) + + override fun init(context: InitContext): ReadContext { + val projectedSchema = + if (projection != null) { + Types.buildMessage() + .apply { + val projectionSet = projection.toSet() + + for (field in FRAGMENT_SCHEMA.fields) { + val col = fieldMap[field.name] ?: continue + if (col in projectionSet) { + addField(field) + } + } + } + .named(FRAGMENT_SCHEMA.name) + } else { + FRAGMENT_SCHEMA + } + + return ReadContext(projectedSchema) + } + + override fun prepareForRead( + configuration: Configuration, + keyValueMetaData: Map<String, String>, + fileSchema: MessageType, + readContext: ReadContext, + ): RecordMaterializer<Fragment> = FragmentRecordMaterializer(readContext.requestedSchema) +} diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/opendc/parquet/ResourceStateRecordMaterializer.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/parquet/FragmentRecordMaterializer.kt index ee5e56aa..7902cab1 100644 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/opendc/parquet/ResourceStateRecordMaterializer.kt +++ b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/parquet/FragmentRecordMaterializer.kt @@ -20,9 +20,8 @@ * SOFTWARE. */ -package org.opendc.trace.formats.opendc.parquet +package org.opendc.trace.formats.workload.parquet -import org.apache.parquet.io.api.Binary import org.apache.parquet.io.api.Converter import org.apache.parquet.io.api.GroupConverter import org.apache.parquet.io.api.PrimitiveConverter @@ -32,13 +31,13 @@ import java.time.Duration import java.time.Instant /** - * A [RecordMaterializer] for [ResourceState] records. + * A [RecordMaterializer] for [Fragment] records. */ -internal class ResourceStateRecordMaterializer(schema: MessageType) : RecordMaterializer<ResourceState>() { +internal class FragmentRecordMaterializer(schema: MessageType) : RecordMaterializer<Fragment>() { /** * State of current record being read. */ - private var localId = "" + private var localId = -99 private var localTimestamp = Instant.MIN private var localDuration = Duration.ZERO private var localCpuCount = 0 @@ -59,8 +58,8 @@ internal class ResourceStateRecordMaterializer(schema: MessageType) : RecordMate when (type.name) { "id" -> object : PrimitiveConverter() { - override fun addBinary(value: Binary) { - localId = value.toStringUsingUTF8() + override fun addInt(value: Int) { + localId = value } } "timestamp", "time" -> @@ -104,8 +103,7 @@ internal class ResourceStateRecordMaterializer(schema: MessageType) : RecordMate } override fun start() { - localId = "" - localTimestamp = Instant.MIN + localId = -99 localDuration = Duration.ZERO localCpuCount = 0 localCpuUsage = 0.0 @@ -118,14 +116,11 @@ internal class ResourceStateRecordMaterializer(schema: MessageType) : RecordMate override fun getConverter(fieldIndex: Int): Converter = converters[fieldIndex] } - override fun getCurrentRecord(): ResourceState = - ResourceState( + override fun getCurrentRecord(): Fragment = + Fragment( localId, - localTimestamp, localDuration, - localCpuCount, localCpuUsage, - localGpuCount, localGpuUsage, ) diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/parquet/FragmentSchemas.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/parquet/FragmentSchemas.kt new file mode 100644 index 00000000..cd499e7e --- /dev/null +++ b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/parquet/FragmentSchemas.kt @@ -0,0 +1,80 @@ +/* + * Copyright (c) 2025 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 org.opendc.trace.formats.workload.parquet + +import org.apache.parquet.schema.LogicalTypeAnnotation +import org.apache.parquet.schema.MessageType +import org.apache.parquet.schema.PrimitiveType +import org.apache.parquet.schema.Types + +private val FRAGMENT_SCHEMA_v1: MessageType = + Types.buildMessage() + .addFields( + Types + .required(PrimitiveType.PrimitiveTypeName.BINARY) + .`as`(LogicalTypeAnnotation.stringType()) + .named("id"), + Types + .required(PrimitiveType.PrimitiveTypeName.INT64) + .`as`(LogicalTypeAnnotation.timestampType(true, LogicalTypeAnnotation.TimeUnit.MILLIS)) + .named("timestamp"), + Types + .required(PrimitiveType.PrimitiveTypeName.INT64) + .named("duration"), + Types + .required(PrimitiveType.PrimitiveTypeName.INT32) + .named("cpu_count"), + Types + .required(PrimitiveType.PrimitiveTypeName.DOUBLE) + .named("cpu_usage"), + Types + .optional(PrimitiveType.PrimitiveTypeName.INT32) + .named("gpu_count"), + Types + .optional(PrimitiveType.PrimitiveTypeName.DOUBLE) + .named("gpu_usage"), + ) + .named("resource_state") + +private val FRAGMENT_SCHEMA_v2: MessageType = + Types.buildMessage() + .addFields( + Types + .required(PrimitiveType.PrimitiveTypeName.INT32) + .named("id"), + Types + .required(PrimitiveType.PrimitiveTypeName.INT64) + .named("duration"), + Types + .required(PrimitiveType.PrimitiveTypeName.DOUBLE) + .named("cpu_usage"), + Types + .optional(PrimitiveType.PrimitiveTypeName.DOUBLE) + .named("gpu_usage"), + ) + .named("resource_state") + +/** + * Parquet read schema for the "resource states" table in the trace. + */ +public val FRAGMENT_SCHEMA: MessageType = FRAGMENT_SCHEMA_v2 diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/opendc/parquet/ResourceStateWriteSupport.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/parquet/FragmentWriteSupport.kt index 58c43916..e6b7ba4f 100644 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/opendc/parquet/ResourceStateWriteSupport.kt +++ b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/parquet/FragmentWriteSupport.kt @@ -20,11 +20,10 @@ * SOFTWARE. */ -package org.opendc.trace.formats.opendc.parquet +package org.opendc.trace.formats.workload.parquet import org.apache.hadoop.conf.Configuration import org.apache.parquet.hadoop.api.WriteSupport -import org.apache.parquet.io.api.Binary import org.apache.parquet.io.api.RecordConsumer import org.apache.parquet.schema.LogicalTypeAnnotation import org.apache.parquet.schema.MessageType @@ -32,9 +31,9 @@ import org.apache.parquet.schema.PrimitiveType import org.apache.parquet.schema.Types /** - * Support for writing [Resource] instances to Parquet format. + * Support for writing [Task] instances to Parquet format. */ -internal class ResourceStateWriteSupport : WriteSupport<ResourceState>() { +internal class FragmentWriteSupport : WriteSupport<Fragment>() { /** * The current active record consumer. */ @@ -48,32 +47,24 @@ internal class ResourceStateWriteSupport : WriteSupport<ResourceState>() { this.recordConsumer = recordConsumer } - override fun write(record: ResourceState) { + override fun write(record: Fragment) { write(recordConsumer, record) } private fun write( consumer: RecordConsumer, - record: ResourceState, + record: Fragment, ) { consumer.startMessage() consumer.startField("id", 0) - consumer.addBinary(Binary.fromCharSequence(record.id)) + consumer.addInteger(record.id) consumer.endField("id", 0) - consumer.startField("timestamp", 1) - consumer.addLong(record.timestamp.toEpochMilli()) - consumer.endField("timestamp", 1) - consumer.startField("duration", 2) consumer.addLong(record.duration.toMillis()) consumer.endField("duration", 2) - consumer.startField("cpu_count", 3) - consumer.addInteger(record.cpuCount) - consumer.endField("cpu_count", 3) - consumer.startField("cpu_usage", 4) consumer.addDouble(record.cpuUsage) consumer.endField("cpu_usage", 4) @@ -101,9 +92,6 @@ internal class ResourceStateWriteSupport : WriteSupport<ResourceState>() { .required(PrimitiveType.PrimitiveTypeName.INT64) .named("duration"), Types - .required(PrimitiveType.PrimitiveTypeName.INT32) - .named("cpu_count"), - Types .required(PrimitiveType.PrimitiveTypeName.DOUBLE) .named("cpu_usage"), ) diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/opendc/parquet/Resource.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/parquet/Task.kt index d727920a..f661d5a9 100644 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/opendc/parquet/Resource.kt +++ b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/parquet/Task.kt @@ -20,15 +20,16 @@ * SOFTWARE. */ -package org.opendc.trace.formats.opendc.parquet +package org.opendc.trace.formats.workload.parquet import java.time.Instant /** * A description of a resource in a trace. */ -internal data class Resource( - val id: String, +internal data class Task( + val id: Int, + val name: String, val submissionTime: Instant, val durationTime: Long, val cpuCount: Int, @@ -36,8 +37,8 @@ internal data class Resource( val memCapacity: Double, val gpuCount: Int = 0, val gpuCapacity: Double = 0.0, - val parents: Set<String> = emptySet(), - val children: Set<String> = emptySet(), + val parents: Set<Int> = emptySet(), + val children: Set<Int> = emptySet(), val nature: String? = null, val deadline: Long = -1, ) diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/parquet/TaskReadSupport.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/parquet/TaskReadSupport.kt new file mode 100644 index 00000000..4bbb18ac --- /dev/null +++ b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/parquet/TaskReadSupport.kt @@ -0,0 +1,101 @@ +/* + * Copyright (c) 2022 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 org.opendc.trace.formats.workload.parquet + +import org.apache.hadoop.conf.Configuration +import org.apache.parquet.hadoop.api.InitContext +import org.apache.parquet.hadoop.api.ReadSupport +import org.apache.parquet.io.api.RecordMaterializer +import org.apache.parquet.schema.MessageType +import org.apache.parquet.schema.Types +import org.opendc.trace.TableColumn +import org.opendc.trace.conv.TASK_CHILDREN +import org.opendc.trace.conv.TASK_CPU_CAPACITY +import org.opendc.trace.conv.TASK_CPU_COUNT +import org.opendc.trace.conv.TASK_DEADLINE +import org.opendc.trace.conv.TASK_DURATION +import org.opendc.trace.conv.TASK_GPU_CAPACITY +import org.opendc.trace.conv.TASK_GPU_COUNT +import org.opendc.trace.conv.TASK_ID +import org.opendc.trace.conv.TASK_MEM_CAPACITY +import org.opendc.trace.conv.TASK_NAME +import org.opendc.trace.conv.TASK_NATURE +import org.opendc.trace.conv.TASK_PARENTS +import org.opendc.trace.conv.TASK_SUBMISSION_TIME + +/** + * A [ReadSupport] instance for [Task] objects. + */ +internal class TaskReadSupport(private val projection: List<String>?) : ReadSupport<Task>() { + /** + * Mapping from field names to [TableColumn]s. + */ + private val fieldMap = + mapOf( + "id" to TASK_ID, + "name" to TASK_NAME, + "submissionTime" to TASK_SUBMISSION_TIME, + "submission_time" to TASK_SUBMISSION_TIME, + "duration" to TASK_DURATION, + "maxCores" to TASK_CPU_COUNT, + "cpu_count" to TASK_CPU_COUNT, + "cpu_capacity" to TASK_CPU_CAPACITY, + "requiredMemory" to TASK_MEM_CAPACITY, + "mem_capacity" to TASK_MEM_CAPACITY, + "gpu_count" to TASK_GPU_COUNT, + "gpu_capacity" to TASK_GPU_CAPACITY, + "parents" to TASK_PARENTS, + "children" to TASK_CHILDREN, + "nature" to TASK_NATURE, + "deadline" to TASK_DEADLINE, + ) + + override fun init(context: InitContext): ReadContext { + val projectedSchema = + if (projection != null) { + Types.buildMessage() + .apply { + val projectionSet = projection.toSet() + + for (field in TASK_SCHEMA.fields) { + val col = fieldMap[field.name] ?: continue + if (col in projectionSet) { + addField(field) + } + } + } + .named(TASK_SCHEMA.name) + } else { + TASK_SCHEMA + } + + return ReadContext(projectedSchema) + } + + override fun prepareForRead( + configuration: Configuration, + keyValueMetaData: Map<String, String>, + fileSchema: MessageType, + readContext: ReadContext, + ): RecordMaterializer<Task> = TaskRecordMaterializer(readContext.requestedSchema) +} diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/opendc/parquet/ResourceRecordMaterializer.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/parquet/TaskRecordMaterializer.kt index f9493721..12dc54b7 100644 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/opendc/parquet/ResourceRecordMaterializer.kt +++ b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/parquet/TaskRecordMaterializer.kt @@ -20,7 +20,7 @@ * SOFTWARE. */ -package org.opendc.trace.formats.opendc.parquet +package org.opendc.trace.formats.workload.parquet import org.apache.parquet.io.api.Binary import org.apache.parquet.io.api.Converter @@ -31,13 +31,14 @@ import org.apache.parquet.schema.MessageType import java.time.Instant /** - * A [RecordMaterializer] for [Resource] records. + * A [RecordMaterializer] for [Task] records. */ -internal class ResourceRecordMaterializer(schema: MessageType) : RecordMaterializer<Resource>() { +internal class TaskRecordMaterializer(schema: MessageType) : RecordMaterializer<Task>() { /** * State of current record being read. */ - private var localId = "" + private var localId = -99 + private var localName = "" private var localSubmissionTime = Instant.MIN private var localDuration = 0L private var localCpuCount = 0 @@ -45,8 +46,8 @@ internal class ResourceRecordMaterializer(schema: MessageType) : RecordMateriali private var localMemCapacity = 0.0 private var localGpuCount = 0 private var localGpuCapacity = 0.0 - private var localParents = mutableSetOf<String>() - private var localChildren = mutableSetOf<String>() + private var localParents = mutableSetOf<Int>() + private var localChildren = mutableSetOf<Int>() private var localNature: String? = null private var localDeadline = -1L @@ -63,8 +64,14 @@ internal class ResourceRecordMaterializer(schema: MessageType) : RecordMateriali when (type.name) { "id" -> object : PrimitiveConverter() { + override fun addInt(value: Int) { + localId = value + } + } + "name" -> + object : PrimitiveConverter() { override fun addBinary(value: Binary) { - localId = value.toStringUsingUTF8() + localName = value.toStringUsingUTF8() } } "submission_time", "submissionTime" -> @@ -132,7 +139,8 @@ internal class ResourceRecordMaterializer(schema: MessageType) : RecordMateriali } override fun start() { - localId = "" + localId = -99 + localName = "" localSubmissionTime = Instant.MIN localDuration = 0L localCpuCount = 0 @@ -151,9 +159,10 @@ internal class ResourceRecordMaterializer(schema: MessageType) : RecordMateriali override fun getConverter(fieldIndex: Int): Converter = converters[fieldIndex] } - override fun getCurrentRecord(): Resource = - Resource( + override fun getCurrentRecord(): Task = + Task( localId, + localName, localSubmissionTime, localDuration, localCpuCount, @@ -172,12 +181,11 @@ internal class ResourceRecordMaterializer(schema: MessageType) : RecordMateriali /** * Helper class to convert parent and child relations and add them to [relations]. */ - private class RelationConverter(private val relations: MutableSet<String>) : GroupConverter() { + private class RelationConverter(private val relations: MutableSet<Int>) : GroupConverter() { private val entryConverter = object : PrimitiveConverter() { - override fun addBinary(value: Binary) { - val str = value.toStringUsingUTF8() - relations.add(str) + override fun addInt(value: Int) { + relations.add(value) } } diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/parquet/TaskSchemas.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/parquet/TaskSchemas.kt new file mode 100644 index 00000000..f7f5e953 --- /dev/null +++ b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/parquet/TaskSchemas.kt @@ -0,0 +1,166 @@ +/* + * Copyright (c) 2025 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 org.opendc.trace.formats.workload.parquet + +import org.apache.parquet.schema.LogicalTypeAnnotation +import org.apache.parquet.schema.MessageType +import org.apache.parquet.schema.PrimitiveType +import org.apache.parquet.schema.Type +import org.apache.parquet.schema.Types + +private val TASK_SCHEMA_V1: MessageType = + Types.buildMessage() + .addFields( + Types + .required(PrimitiveType.PrimitiveTypeName.BINARY) + .`as`(LogicalTypeAnnotation.stringType()) + .named("id"), + Types + .required(PrimitiveType.PrimitiveTypeName.INT64) + .`as`(LogicalTypeAnnotation.timestampType(true, LogicalTypeAnnotation.TimeUnit.MILLIS)) + .named("submission_time"), + Types + .required(PrimitiveType.PrimitiveTypeName.INT64) + .named("duration"), + Types + .required(PrimitiveType.PrimitiveTypeName.INT32) + .named("cpu_count"), + Types + .required(PrimitiveType.PrimitiveTypeName.DOUBLE) + .named("cpu_capacity"), + Types + .required(PrimitiveType.PrimitiveTypeName.INT64) + .named("mem_capacity"), + Types + .optional(PrimitiveType.PrimitiveTypeName.INT32) + .named("gpu_count"), + Types + .optional(PrimitiveType.PrimitiveTypeName.DOUBLE) + .named("gpu_capacity"), + Types + .buildGroup(Type.Repetition.OPTIONAL) + .addField( + Types.repeatedGroup() + .addField( + Types.optional(PrimitiveType.PrimitiveTypeName.BINARY) + .`as`(LogicalTypeAnnotation.stringType()) + .named("element"), + ) + .named("list"), + ) + .`as`(LogicalTypeAnnotation.listType()) + .named("parents"), + Types + .buildGroup(Type.Repetition.OPTIONAL) + .addField( + Types.repeatedGroup() + .addField( + Types.optional(PrimitiveType.PrimitiveTypeName.BINARY) + .`as`(LogicalTypeAnnotation.stringType()) + .named("element"), + ) + .named("list"), + ) + .`as`(LogicalTypeAnnotation.listType()) + .named("children"), + Types + .optional(PrimitiveType.PrimitiveTypeName.BINARY) + .`as`(LogicalTypeAnnotation.stringType()) + .named("nature"), + Types + .optional(PrimitiveType.PrimitiveTypeName.INT64) + .named("deadline"), + ) + .named("resource") + +private val TASK_SCHEMA_V2: MessageType = + Types.buildMessage() + .addFields( + Types + .required(PrimitiveType.PrimitiveTypeName.INT32) + .named("id"), + Types + .optional(PrimitiveType.PrimitiveTypeName.BINARY) + .`as`(LogicalTypeAnnotation.stringType()) + .named("name"), + Types + .required(PrimitiveType.PrimitiveTypeName.INT64) + .`as`(LogicalTypeAnnotation.timestampType(true, LogicalTypeAnnotation.TimeUnit.MILLIS)) + .named("submission_time"), + Types + .required(PrimitiveType.PrimitiveTypeName.INT64) + .named("duration"), + Types + .required(PrimitiveType.PrimitiveTypeName.INT32) + .named("cpu_count"), + Types + .required(PrimitiveType.PrimitiveTypeName.DOUBLE) + .named("cpu_capacity"), + Types + .required(PrimitiveType.PrimitiveTypeName.INT64) + .named("mem_capacity"), + Types + .optional(PrimitiveType.PrimitiveTypeName.INT32) + .named("gpu_count"), + Types + .optional(PrimitiveType.PrimitiveTypeName.DOUBLE) + .named("gpu_capacity"), + Types + .buildGroup(Type.Repetition.OPTIONAL) + .addField( + Types.repeatedGroup() + .addField( + Types.optional( + PrimitiveType.PrimitiveTypeName.INT32, + ) + .named("element"), + ) + .named("list"), + ) + .`as`(LogicalTypeAnnotation.listType()) + .named("parents"), + Types + .buildGroup(Type.Repetition.OPTIONAL) + .addField( + Types.repeatedGroup() + .addField( + Types.optional( + PrimitiveType.PrimitiveTypeName.INT32, + ) + .named("element"), + ) + .named("list"), + ) + .`as`(LogicalTypeAnnotation.listType()) + .named("children"), + Types + .optional(PrimitiveType.PrimitiveTypeName.BINARY) + .`as`(LogicalTypeAnnotation.stringType()) + .named("nature"), + Types + .optional(PrimitiveType.PrimitiveTypeName.INT64) + .named("deadline"), + ) + .named("resource") + +public val TASK_SCHEMA: MessageType = TASK_SCHEMA_V2 diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/opendc/parquet/ResourceWriteSupport.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/parquet/TaskWriteSupport.kt index c3e984fb..a7ce62b8 100644 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/opendc/parquet/ResourceWriteSupport.kt +++ b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/parquet/TaskWriteSupport.kt @@ -20,47 +20,43 @@ * SOFTWARE. */ -package org.opendc.trace.formats.opendc.parquet +package org.opendc.trace.formats.workload.parquet import org.apache.hadoop.conf.Configuration import org.apache.parquet.hadoop.api.WriteSupport import org.apache.parquet.io.api.Binary import org.apache.parquet.io.api.RecordConsumer -import org.apache.parquet.schema.LogicalTypeAnnotation -import org.apache.parquet.schema.MessageType -import org.apache.parquet.schema.PrimitiveType -import org.apache.parquet.schema.Types import kotlin.math.roundToLong /** - * Support for writing [Resource] instances to Parquet format. + * Support for writing [Task] instances to Parquet format. */ -internal class ResourceWriteSupport : WriteSupport<Resource>() { +internal class TaskWriteSupport : WriteSupport<Task>() { /** * The current active record consumer. */ private lateinit var recordConsumer: RecordConsumer override fun init(configuration: Configuration): WriteContext { - return WriteContext(WRITE_SCHEMA, emptyMap()) + return WriteContext(TASK_SCHEMA, emptyMap()) } override fun prepareForWrite(recordConsumer: RecordConsumer) { this.recordConsumer = recordConsumer } - override fun write(record: Resource) { + override fun write(record: Task) { write(recordConsumer, record) } private fun write( consumer: RecordConsumer, - record: Resource, + record: Task, ) { consumer.startMessage() consumer.startField("id", 0) - consumer.addBinary(Binary.fromCharSequence(record.id)) + consumer.addInteger(record.id) consumer.endField("id", 0) consumer.startField("submission_time", 1) @@ -97,43 +93,4 @@ internal class ResourceWriteSupport : WriteSupport<Resource>() { consumer.endMessage() } - - companion object { - /** - * Parquet schema for the "resources" table in the trace. - */ - @JvmStatic - val WRITE_SCHEMA: MessageType = - Types.buildMessage() - .addFields( - Types - .required(PrimitiveType.PrimitiveTypeName.BINARY) - .`as`(LogicalTypeAnnotation.stringType()) - .named("id"), - Types - .required(PrimitiveType.PrimitiveTypeName.INT64) - .`as`(LogicalTypeAnnotation.timestampType(true, LogicalTypeAnnotation.TimeUnit.MILLIS)) - .named("submission_time"), - Types - .required(PrimitiveType.PrimitiveTypeName.INT64) - .named("duration"), - Types - .required(PrimitiveType.PrimitiveTypeName.INT32) - .named("cpu_count"), - Types - .required(PrimitiveType.PrimitiveTypeName.DOUBLE) - .named("cpu_capacity"), - Types - .required(PrimitiveType.PrimitiveTypeName.INT64) - .named("mem_capacity"), - Types - .optional(PrimitiveType.PrimitiveTypeName.BINARY) - .`as`(LogicalTypeAnnotation.stringType()) - .named("nature"), - Types - .optional(PrimitiveType.PrimitiveTypeName.INT64) - .named("deadline"), - ) - .named("resource") - } } diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/wtf/WtfTaskTableReader.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/wtf/WtfTaskTableReader.kt deleted file mode 100644 index 95582388..00000000 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/wtf/WtfTaskTableReader.kt +++ /dev/null @@ -1,187 +0,0 @@ -/* - * Copyright (c) 2021 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 org.opendc.trace.wtf - -import org.opendc.trace.TableColumnType -import org.opendc.trace.TableReader -import org.opendc.trace.conv.TASK_CHILDREN -import org.opendc.trace.conv.TASK_GROUP_ID -import org.opendc.trace.conv.TASK_ID -import org.opendc.trace.conv.TASK_PARENTS -import org.opendc.trace.conv.TASK_REQ_NCPUS -import org.opendc.trace.conv.TASK_RUNTIME -import org.opendc.trace.conv.TASK_SUBMIT_TIME -import org.opendc.trace.conv.TASK_USER_ID -import org.opendc.trace.conv.TASK_WAIT_TIME -import org.opendc.trace.conv.TASK_WORKFLOW_ID -import org.opendc.trace.util.convertTo -import org.opendc.trace.util.parquet.LocalParquetReader -import org.opendc.trace.wtf.parquet.Task -import java.time.Duration -import java.time.Instant -import java.util.UUID - -/** - * A [TableReader] implementation for the WTF format. - */ -internal class WtfTaskTableReader(private val reader: LocalParquetReader<Task>) : TableReader { - /** - * The current record. - */ - private var record: Task? = null - - override fun nextRow(): Boolean { - try { - val record = reader.read() - this.record = record - - return record != null - } catch (e: Throwable) { - this.record = null - throw e - } - } - - private val colID = 0 - private val colWorkflowID = 1 - private val colSubmitTime = 2 - private val colWaitTime = 3 - private val colRuntime = 4 - private val colReqNcpus = 5 - private val colParents = 6 - private val colChildren = 7 - private val colGroupID = 8 - private val colUserID = 9 - - private val typeParents = TableColumnType.Set(TableColumnType.String) - private val typeChildren = TableColumnType.Set(TableColumnType.String) - - override fun resolve(name: String): Int { - return when (name) { - TASK_ID -> colID - TASK_WORKFLOW_ID -> colWorkflowID - TASK_SUBMIT_TIME -> colSubmitTime - TASK_WAIT_TIME -> colWaitTime - TASK_RUNTIME -> colRuntime - TASK_REQ_NCPUS -> colReqNcpus - TASK_PARENTS -> colParents - TASK_CHILDREN -> colChildren - TASK_GROUP_ID -> colGroupID - TASK_USER_ID -> colUserID - else -> -1 - } - } - - override fun isNull(index: Int): Boolean { - require(index in colID..colUserID) { "Invalid column index" } - return false - } - - override fun getBoolean(index: Int): Boolean { - throw IllegalArgumentException("Invalid column") - } - - override fun getInt(index: Int): Int { - val record = checkNotNull(record) { "Reader in invalid state" } - - return when (index) { - colReqNcpus -> record.requestedCpus - colGroupID -> record.groupId - colUserID -> record.userId - else -> throw IllegalArgumentException("Invalid column") - } - } - - override fun getLong(index: Int): Long { - throw IllegalArgumentException("Invalid column") - } - - override fun getFloat(index: Int): Float { - throw IllegalArgumentException("Invalid column") - } - - override fun getDouble(index: Int): Double { - throw IllegalArgumentException("Invalid column") - } - - override fun getString(index: Int): String { - val record = checkNotNull(record) { "Reader in invalid state" } - return when (index) { - colID -> record.id - colWorkflowID -> record.workflowId - else -> throw IllegalArgumentException("Invalid column") - } - } - - override fun getUUID(index: Int): UUID? { - throw IllegalArgumentException("Invalid column") - } - - override fun getInstant(index: Int): Instant { - val record = checkNotNull(record) { "Reader in invalid state" } - return when (index) { - colSubmitTime -> record.submitTime - else -> throw IllegalArgumentException("Invalid column") - } - } - - override fun getDuration(index: Int): Duration { - val record = checkNotNull(record) { "Reader in invalid state" } - return when (index) { - colWaitTime -> record.waitTime - colRuntime -> record.runtime - else -> throw IllegalArgumentException("Invalid column") - } - } - - override fun <T> getList( - index: Int, - elementType: Class<T>, - ): List<T>? { - throw IllegalArgumentException("Invalid column") - } - - override fun <T> getSet( - index: Int, - elementType: Class<T>, - ): Set<T>? { - val record = checkNotNull(record) { "Reader in invalid state" } - return when (index) { - colParents -> typeParents.convertTo(record.parents, elementType) - colChildren -> typeChildren.convertTo(record.children, elementType) - else -> throw IllegalArgumentException("Invalid column") - } - } - - override fun <K, V> getMap( - index: Int, - keyType: Class<K>, - valueType: Class<V>, - ): Map<K, V>? { - throw IllegalArgumentException("Invalid column") - } - - override fun close() { - reader.close() - } -} diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/wtf/WtfTraceFormat.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/wtf/WtfTraceFormat.kt deleted file mode 100644 index 1386d2ef..00000000 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/wtf/WtfTraceFormat.kt +++ /dev/null @@ -1,102 +0,0 @@ -/* - * Copyright (c) 2021 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 org.opendc.trace.wtf - -import org.opendc.trace.TableColumn -import org.opendc.trace.TableColumnType -import org.opendc.trace.TableReader -import org.opendc.trace.TableWriter -import org.opendc.trace.conv.TABLE_TASKS -import org.opendc.trace.conv.TASK_CHILDREN -import org.opendc.trace.conv.TASK_GROUP_ID -import org.opendc.trace.conv.TASK_ID -import org.opendc.trace.conv.TASK_PARENTS -import org.opendc.trace.conv.TASK_REQ_NCPUS -import org.opendc.trace.conv.TASK_RUNTIME -import org.opendc.trace.conv.TASK_SUBMIT_TIME -import org.opendc.trace.conv.TASK_USER_ID -import org.opendc.trace.conv.TASK_WAIT_TIME -import org.opendc.trace.conv.TASK_WORKFLOW_ID -import org.opendc.trace.spi.TableDetails -import org.opendc.trace.spi.TraceFormat -import org.opendc.trace.util.parquet.LocalParquetReader -import org.opendc.trace.wtf.parquet.TaskReadSupport -import java.nio.file.Path - -/** - * A [TraceFormat] implementation for the Workflow Trace Format (WTF). - */ -public class WtfTraceFormat : TraceFormat { - override val name: String = "wtf" - - override fun create(path: Path) { - throw UnsupportedOperationException("Writing not supported for this format") - } - - override fun getTables(path: Path): List<String> = listOf(TABLE_TASKS) - - override fun getDetails( - path: Path, - table: String, - ): TableDetails { - return when (table) { - TABLE_TASKS -> - TableDetails( - listOf( - TableColumn(TASK_ID, TableColumnType.String), - TableColumn(TASK_WORKFLOW_ID, TableColumnType.String), - TableColumn(TASK_SUBMIT_TIME, TableColumnType.Instant), - TableColumn(TASK_WAIT_TIME, TableColumnType.Duration), - TableColumn(TASK_RUNTIME, TableColumnType.Duration), - TableColumn(TASK_REQ_NCPUS, TableColumnType.Int), - TableColumn(TASK_PARENTS, TableColumnType.Set(TableColumnType.String)), - TableColumn(TASK_CHILDREN, TableColumnType.Set(TableColumnType.String)), - TableColumn(TASK_GROUP_ID, TableColumnType.Int), - TableColumn(TASK_USER_ID, TableColumnType.Int), - ), - ) - else -> throw IllegalArgumentException("Table $table not supported") - } - } - - override fun newReader( - path: Path, - table: String, - projection: List<String>?, - ): TableReader { - return when (table) { - TABLE_TASKS -> { - val reader = LocalParquetReader(path.resolve("tasks/schema-1.0"), TaskReadSupport(projection), strictTyping = false) - WtfTaskTableReader(reader) - } - else -> throw IllegalArgumentException("Table $table not supported") - } - } - - override fun newWriter( - path: Path, - table: String, - ): TableWriter { - throw UnsupportedOperationException("Writing not supported for this format") - } -} diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/wtf/parquet/Task.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/wtf/parquet/Task.kt deleted file mode 100644 index a1db0cab..00000000 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/wtf/parquet/Task.kt +++ /dev/null @@ -1,42 +0,0 @@ -/* - * Copyright (c) 2022 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 org.opendc.trace.wtf.parquet - -import java.time.Duration -import java.time.Instant - -/** - * A task in the Workflow Trace Format. - */ -internal data class Task( - val id: String, - val workflowId: String, - val submitTime: Instant, - val waitTime: Duration, - val runtime: Duration, - val requestedCpus: Int, - val groupId: Int, - val userId: Int, - val parents: Set<String>, - val children: Set<String>, -) diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/wtf/parquet/TaskReadSupport.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/wtf/parquet/TaskReadSupport.kt deleted file mode 100644 index 1f9c506d..00000000 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/wtf/parquet/TaskReadSupport.kt +++ /dev/null @@ -1,148 +0,0 @@ -/* - * Copyright (c) 2022 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 org.opendc.trace.wtf.parquet - -import org.apache.hadoop.conf.Configuration -import org.apache.parquet.hadoop.api.InitContext -import org.apache.parquet.hadoop.api.ReadSupport -import org.apache.parquet.io.api.RecordMaterializer -import org.apache.parquet.schema.LogicalTypeAnnotation -import org.apache.parquet.schema.MessageType -import org.apache.parquet.schema.PrimitiveType -import org.apache.parquet.schema.Type -import org.apache.parquet.schema.Types -import org.opendc.trace.conv.TASK_CHILDREN -import org.opendc.trace.conv.TASK_GROUP_ID -import org.opendc.trace.conv.TASK_ID -import org.opendc.trace.conv.TASK_PARENTS -import org.opendc.trace.conv.TASK_REQ_NCPUS -import org.opendc.trace.conv.TASK_RUNTIME -import org.opendc.trace.conv.TASK_SUBMIT_TIME -import org.opendc.trace.conv.TASK_USER_ID -import org.opendc.trace.conv.TASK_WAIT_TIME -import org.opendc.trace.conv.TASK_WORKFLOW_ID - -/** - * A [ReadSupport] instance for [Task] objects. - * - * @param projection The projection of the table to read. - */ -internal class TaskReadSupport(private val projection: List<String>?) : ReadSupport<Task>() { - /** - * Mapping of table columns to their Parquet column names. - */ - private val colMap = - mapOf( - TASK_ID to "id", - TASK_WORKFLOW_ID to "workflow_id", - TASK_SUBMIT_TIME to "ts_submit", - TASK_WAIT_TIME to "wait_time", - TASK_RUNTIME to "runtime", - TASK_REQ_NCPUS to "resource_amount_requested", - TASK_PARENTS to "parents", - TASK_CHILDREN to "children", - TASK_GROUP_ID to "group_id", - TASK_USER_ID to "user_id", - ) - - override fun init(context: InitContext): ReadContext { - val projectedSchema = - if (projection != null) { - Types.buildMessage() - .apply { - val fieldByName = READ_SCHEMA.fields.associateBy { it.name } - - for (col in projection) { - val fieldName = colMap[col] ?: continue - addField(fieldByName.getValue(fieldName)) - } - } - .named(READ_SCHEMA.name) - } else { - READ_SCHEMA - } - return ReadContext(projectedSchema) - } - - override fun prepareForRead( - configuration: Configuration, - keyValueMetaData: Map<String, String>, - fileSchema: MessageType, - readContext: ReadContext, - ): RecordMaterializer<Task> = TaskRecordMaterializer(readContext.requestedSchema) - - companion object { - /** - * Parquet read schema for the "tasks" table in the trace. - */ - @JvmStatic - val READ_SCHEMA: MessageType = - Types.buildMessage() - .addFields( - Types - .optional(PrimitiveType.PrimitiveTypeName.INT64) - .named("id"), - Types - .optional(PrimitiveType.PrimitiveTypeName.INT64) - .named("workflow_id"), - Types - .optional(PrimitiveType.PrimitiveTypeName.INT64) - .`as`(LogicalTypeAnnotation.timestampType(true, LogicalTypeAnnotation.TimeUnit.MILLIS)) - .named("ts_submit"), - Types - .optional(PrimitiveType.PrimitiveTypeName.INT64) - .named("wait_time"), - Types - .optional(PrimitiveType.PrimitiveTypeName.INT64) - .named("runtime"), - Types - .optional(PrimitiveType.PrimitiveTypeName.DOUBLE) - .named("resource_amount_requested"), - Types - .optional(PrimitiveType.PrimitiveTypeName.INT32) - .named("user_id"), - Types - .optional(PrimitiveType.PrimitiveTypeName.INT32) - .named("group_id"), - Types - .buildGroup(Type.Repetition.OPTIONAL) - .addField( - Types.repeatedGroup() - .addField(Types.optional(PrimitiveType.PrimitiveTypeName.INT64).named("item")) - .named("list"), - ) - .`as`(LogicalTypeAnnotation.listType()) - .named("children"), - Types - .buildGroup(Type.Repetition.OPTIONAL) - .addField( - Types.repeatedGroup() - .addField(Types.optional(PrimitiveType.PrimitiveTypeName.INT64).named("item")) - .named("list"), - ) - .`as`(LogicalTypeAnnotation.listType()) - .named("parents"), - ) - .named("task") - } -} diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/wtf/parquet/TaskRecordMaterializer.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/wtf/parquet/TaskRecordMaterializer.kt deleted file mode 100644 index 412a4f8b..00000000 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/wtf/parquet/TaskRecordMaterializer.kt +++ /dev/null @@ -1,188 +0,0 @@ -/* - * Copyright (c) 2022 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 org.opendc.trace.wtf.parquet - -import org.apache.parquet.io.api.Converter -import org.apache.parquet.io.api.GroupConverter -import org.apache.parquet.io.api.PrimitiveConverter -import org.apache.parquet.io.api.RecordMaterializer -import org.apache.parquet.schema.MessageType -import java.time.Duration -import java.time.Instant -import kotlin.math.roundToInt -import kotlin.math.roundToLong - -/** - * A [RecordMaterializer] for [Task] records. - */ -internal class TaskRecordMaterializer(schema: MessageType) : RecordMaterializer<Task>() { - /** - * State of current record being read. - */ - private var localID = "" - private var localWorkflowID = "" - private var localSubmitTime = Instant.MIN - private var localWaitTime = Duration.ZERO - private var localRuntime = Duration.ZERO - private var localRequestedCpus = 0 - private var localGroupId = 0 - private var localUserId = 0 - private var localParents = mutableSetOf<String>() - private var localChildren = mutableSetOf<String>() - - /** - * Root converter for the record. - */ - private val root = - object : GroupConverter() { - /** - * The converters for the columns of the schema. - */ - private val converters = - schema.fields.map { type -> - when (type.name) { - "id" -> - object : PrimitiveConverter() { - override fun addLong(value: Long) { - localID = value.toString() - } - } - "workflow_id" -> - object : PrimitiveConverter() { - override fun addLong(value: Long) { - localWorkflowID = value.toString() - } - } - "ts_submit" -> - object : PrimitiveConverter() { - override fun addLong(value: Long) { - localSubmitTime = Instant.ofEpochMilli(value) - } - } - "wait_time" -> - object : PrimitiveConverter() { - override fun addLong(value: Long) { - localWaitTime = Duration.ofMillis(value) - } - } - "runtime" -> - object : PrimitiveConverter() { - override fun addLong(value: Long) { - localRuntime = Duration.ofMillis(value) - } - } - "resource_amount_requested" -> - object : PrimitiveConverter() { - override fun addDouble(value: Double) { - localRequestedCpus = value.roundToInt() - } - } - "group_id" -> - object : PrimitiveConverter() { - override fun addInt(value: Int) { - localGroupId = value - } - } - "user_id" -> - object : PrimitiveConverter() { - override fun addInt(value: Int) { - localUserId = value - } - } - "children" -> RelationConverter(localChildren) - "parents" -> RelationConverter(localParents) - else -> error("Unknown column $type") - } - } - - override fun start() { - localID = "" - localWorkflowID = "" - localSubmitTime = Instant.MIN - localWaitTime = Duration.ZERO - localRuntime = Duration.ZERO - localRequestedCpus = 0 - localGroupId = 0 - localUserId = 0 - localParents.clear() - localChildren.clear() - } - - override fun end() {} - - override fun getConverter(fieldIndex: Int): Converter = converters[fieldIndex] - } - - override fun getCurrentRecord(): Task = - Task( - localID, - localWorkflowID, - localSubmitTime, - localWaitTime, - localRuntime, - localRequestedCpus, - localGroupId, - localUserId, - localParents.toSet(), - localChildren.toSet(), - ) - - override fun getRootConverter(): GroupConverter = root - - /** - * Helper class to convert parent and child relations and add them to [relations]. - */ - private class RelationConverter(private val relations: MutableSet<String>) : GroupConverter() { - private val entryConverter = - object : PrimitiveConverter() { - override fun addLong(value: Long) { - relations.add(value.toString()) - } - - override fun addDouble(value: Double) { - relations.add(value.roundToLong().toString()) - } - } - - private val listConverter = - object : GroupConverter() { - override fun getConverter(fieldIndex: Int): Converter { - require(fieldIndex == 0) - return entryConverter - } - - override fun start() {} - - override fun end() {} - } - - override fun getConverter(fieldIndex: Int): Converter { - require(fieldIndex == 0) - return listConverter - } - - override fun start() {} - - override fun end() {} - } -} diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/spi/TraceFormat.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/spi/TraceFormat.kt index e586f90a..945d8f2f 100644 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/spi/TraceFormat.kt +++ b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/spi/TraceFormat.kt @@ -24,15 +24,9 @@ package org.opendc.trace.spi import org.opendc.trace.TableReader import org.opendc.trace.TableWriter -import org.opendc.trace.azure.AzureTraceFormat -import org.opendc.trace.bitbrains.BitbrainsTraceFormat import org.opendc.trace.formats.carbon.CarbonTraceFormat import org.opendc.trace.formats.failure.FailureTraceFormat -import org.opendc.trace.formats.opendc.OdcVmTraceFormat -import org.opendc.trace.gwf.GwfTraceFormat -import org.opendc.trace.swf.SwfTraceFormat -import org.opendc.trace.wfformat.WfFormatTraceFormat -import org.opendc.trace.wtf.WtfTraceFormat +import org.opendc.trace.formats.workload.WorkloadTraceFormat import java.nio.file.Path import java.util.ServiceLoader @@ -122,15 +116,9 @@ public interface TraceFormat { @JvmStatic public fun byName(name: String): TraceFormat? { return when (name) { - "azure" -> AzureTraceFormat() - "bitbrains" -> BitbrainsTraceFormat() "carbon" -> CarbonTraceFormat() "failure" -> FailureTraceFormat() - "gwf" -> GwfTraceFormat() - "opendc-vm" -> OdcVmTraceFormat() - "swf" -> SwfTraceFormat() - "wfformat" -> WfFormatTraceFormat() - "wtf" -> WtfTraceFormat() + "workload" -> WorkloadTraceFormat() else -> null } } |