Added support for carbon traces (#218)

* Started with the carbon trace implementation

* Moved the carbon trace system to the proper folders

12 files changed, 434 insertions, 10 deletions

diff --git a/opendc-compute/opendc-compute-carbon/build.gradle.kts b/opendc-compute/opendc-compute-carbon/build.gradle.kts
new file mode 100644
index 00000000..58b7bc86
--- /dev/null
+++ b/opendc-compute/opendc-compute-carbon/build.gradle.kts
@@ -0,0 +1,37 @@
+/*
+ * 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.
+ */
+
+description = "OpenDC Compute Service implementation"
+
+// Build configuration
+plugins {
+    `kotlin-library-conventions`
+}
+
+dependencies {
+    api(projects.opendcCompute.opendcComputeApi)
+    implementation(projects.opendcCommon)
+    implementation(project(mapOf("path" to ":opendc-trace:opendc-trace-api")))
+    implementation(project(mapOf("path" to ":opendc-simulator:opendc-simulator-compute")))
+
+    implementation(libs.kotlin.logging)
+}
diff --git a/opendc-compute/opendc-compute-carbon/src/main/kotlin/org/opendc/compute/carbon/CarbonTrace.kt b/opendc-compute/opendc-compute-carbon/src/main/kotlin/org/opendc/compute/carbon/CarbonTrace.kt
new file mode 100644
index 00000000..2ba3e4e3
--- /dev/null
+++ b/opendc-compute/opendc-compute-carbon/src/main/kotlin/org/opendc/compute/carbon/CarbonTrace.kt
@@ -0,0 +1,113 @@
+/*
+ * 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.compute.carbon
+
+import java.time.Instant
+
+/**
+ * A virtual machine workload.
+ *
+ * @param uid The unique identifier of the virtual machine.
+ * @param name The name of the virtual machine.
+ * @param cpuCapacity The required CPU capacity for the VM in MHz.
+ * @param cpuCount The number of vCPUs in the VM.
+ * @param memCapacity The provisioned memory for the VM in MB.
+ * @param startTime The start time of the VM.
+ * @param stopTime The stop time of the VM.
+ * @param trace The trace that belong to this VM.
+ * @param interferenceProfile The interference profile of this virtual machine.
+ */
+public data class CarbonFragment(
+    var startTime: Long,
+    var endTime: Long,
+    var carbonIntensity: Double,
+) {
+    init {
+        require(endTime > startTime) {
+            "The end time of a report should be higher than the start time -> start time: $startTime, end time: $endTime"
+        }
+        require(carbonIntensity >= 0.0) { "carbon intensity cannot be negative" }
+    }
+}
+
+public class CarbonTrace(reports: List<CarbonFragment>? = null) {
+    private var index: Int = 0
+    private val numberOfReports = reports?.size
+    private val reports = reports?.sortedBy { it.startTime }
+
+    private fun hasPreviousReport(): Boolean {
+        return index > 0
+    }
+
+    private fun hasNextReport(): Boolean {
+        if (numberOfReports == null) {
+            return false
+        }
+
+        return index < numberOfReports
+    }
+
+    public fun getCarbonIntensity(timestamp: Instant): Double {
+        return getCarbonIntensity(timestamp.toEpochMilli())
+    }
+
+    /**
+     * Get the carbon intensity of the energy at a given timestamp
+     * Returns the carbon intensity of the first or last [CarbonFragment]
+     * if the given timestamp is outside the information
+     *
+     * @param timestamp
+     * @return The carbon intensity at the given timestamp in gCO2/kWh
+     */
+    public fun getCarbonIntensity(timestamp: Long): Double {
+        if (reports == null) {
+            return 0.0
+        }
+
+        var currentFragment: CarbonFragment
+
+        while (true) {
+            currentFragment = reports[index]
+
+            if (currentFragment.startTime > timestamp) {
+                if (hasPreviousReport()) {
+                    index--
+                    continue
+                }
+                break
+            }
+
+            if (currentFragment.endTime <= timestamp) {
+                if (hasNextReport()) {
+                    index++
+                    continue
+                }
+                break
+            }
+
+            break
+        }
+
+        return currentFragment.carbonIntensity
+    }
+}
diff --git a/opendc-compute/opendc-compute-carbon/src/main/kotlin/org/opendc/compute/carbon/CarbonTraceLoader.kt b/opendc-compute/opendc-compute-carbon/src/main/kotlin/org/opendc/compute/carbon/CarbonTraceLoader.kt
new file mode 100644
index 00000000..685a1fb3
--- /dev/null
+++ b/opendc-compute/opendc-compute-carbon/src/main/kotlin/org/opendc/compute/carbon/CarbonTraceLoader.kt
@@ -0,0 +1,136 @@
+/*
+ * 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.compute.carbon
+
+import mu.KotlinLogging
+import org.opendc.trace.Trace
+import org.opendc.trace.conv.CARBON_INTENSITY_TIMESTAMP
+import org.opendc.trace.conv.CARBON_INTENSITY_VALUE
+import org.opendc.trace.conv.TABLE_CARBON_INTENSITY
+import java.io.File
+import java.lang.ref.SoftReference
+import java.time.Instant
+import java.util.concurrent.ConcurrentHashMap
+
+/**
+ * A helper class for loading compute workload traces into memory.
+ *
+ * @param baseDir The directory containing the traces.
+ */
+public class CarbonTraceLoader {
+    /**
+     * The logger for this instance.
+     */
+    private val logger = KotlinLogging.logger {}
+
+    /**
+     * The cache of workloads.
+     */
+    private val cache = ConcurrentHashMap<String, SoftReference<List<CarbonFragment>>>()
+
+    private val builder = CarbonFragmentBuilder()
+
+    /**
+     * Read the metadata into a workload.
+     */
+    private fun parseCarbon(trace: Trace): List<CarbonFragment> {
+        val reader = checkNotNull(trace.getTable(TABLE_CARBON_INTENSITY)).newReader()
+
+        val startTimeCol = reader.resolve(CARBON_INTENSITY_TIMESTAMP)
+        val carbonIntensityCol = reader.resolve(CARBON_INTENSITY_VALUE)
+
+        val entries = mutableListOf<CarbonFragment>()
+
+        try {
+            while (reader.nextRow()) {
+                val startTime = reader.getInstant(startTimeCol)!!
+                val carbonIntensity = reader.getDouble(carbonIntensityCol)
+
+                builder.add(startTime, carbonIntensity)
+            }
+
+            // Make sure the virtual machines are ordered by start time
+            builder.fixReportTimes()
+
+            return builder.fragments
+        } catch (e: Exception) {
+            e.printStackTrace()
+            throw e
+        } finally {
+            reader.close()
+        }
+    }
+
+    /**
+     * Load the trace with the specified [name] and [format].
+     */
+    public fun get(pathToFile: File): List<CarbonFragment> {
+        val trace = Trace.open(pathToFile, "carbon")
+
+        return parseCarbon(trace)
+    }
+
+    /**
+     * Clear the workload cache.
+     */
+    public fun reset() {
+        cache.clear()
+    }
+
+    /**
+     * A builder for a VM trace.
+     */
+    private class CarbonFragmentBuilder {
+        /**
+         * The total load of the trace.
+         */
+        public val fragments: MutableList<CarbonFragment> = mutableListOf<CarbonFragment>()
+
+        /**
+         * Add a fragment to the trace.
+         *
+         * @param startTime Timestamp at which the fragment starts (in epoch millis).
+         * @param carbonIntensity The carbon intensity during this fragment
+         */
+        fun add(
+            startTime: Instant,
+            carbonIntensity: Double,
+        ) {
+            fragments.add(
+                CarbonFragment(startTime.toEpochMilli(), Long.MAX_VALUE, carbonIntensity),
+            )
+        }
+
+        fun fixReportTimes() {
+            fragments.sortBy { it.startTime }
+
+            // For each report, set the end time to the start time of the next report
+            for (i in 0..fragments.size - 2) {
+                fragments[i].endTime = fragments[i + 1].startTime
+            }
+
+            // Set the start time of each report to the minimum value
+            fragments[0].startTime = Long.MIN_VALUE
+        }
+    }
+}
diff --git a/opendc-compute/opendc-compute-carbon/src/main/kotlin/org/opendc/compute/carbon/CarbonTraceReader.kt b/opendc-compute/opendc-compute-carbon/src/main/kotlin/org/opendc/compute/carbon/CarbonTraceReader.kt
new file mode 100644
index 00000000..3e0269f8
--- /dev/null
+++ b/opendc-compute/opendc-compute-carbon/src/main/kotlin/org/opendc/compute/carbon/CarbonTraceReader.kt
@@ -0,0 +1,52 @@
+/*
+ * 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.
+ */
+
+@file:JvmName("ComputeWorkloads")
+
+package org.opendc.compute.carbon
+
+import java.io.File
+import javax.management.InvalidAttributeValueException
+
+/**
+ * Construct a workload from a trace.
+ */
+public fun getCarbonTrace(pathToFile: String?): CarbonTrace {
+    if (pathToFile == null) {
+        return CarbonTrace(null)
+    }
+
+    return getCarbonTrace(File(pathToFile))
+}
+
+/**
+ * Construct a workload from a trace.
+ */
+public fun getCarbonTrace(file: File): CarbonTrace {
+    if (!file.exists()) {
+        throw InvalidAttributeValueException("The carbon trace cannot be found")
+    }
+
+    val fragments = CarbonTraceLoader().get(file)
+
+    return CarbonTrace(fragments)
+}
diff --git a/opendc-compute/opendc-compute-carbon/src/test/resources/log4j2.xml b/opendc-compute/opendc-compute-carbon/src/test/resources/log4j2.xml
new file mode 100644
index 00000000..0dfb75f2
--- /dev/null
+++ b/opendc-compute/opendc-compute-carbon/src/test/resources/log4j2.xml
@@ -0,0 +1,38 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ 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.
+  -->
+
+<Configuration status="WARN" packages="org.apache.logging.log4j.core">
+    <Appenders>
+        <Console name="Console" target="SYSTEM_OUT">
+            <PatternLayout pattern="%d{HH:mm:ss.SSS} [%highlight{%-5level}] %logger{36} - %msg%n" disableAnsi="false"/>
+        </Console>
+    </Appenders>
+    <Loggers>
+        <Logger name="org.opendc" level="trace" additivity="false">
+            <AppenderRef ref="Console"/>
+        </Logger>
+        <Root level="info">
+            <AppenderRef ref="Console"/>
+        </Root>
+    </Loggers>
+</Configuration>
diff --git a/opendc-compute/opendc-compute-simulator/build.gradle.kts b/opendc-compute/opendc-compute-simulator/build.gradle.kts
index 9692f6ba..0cddd296 100644
--- a/opendc-compute/opendc-compute-simulator/build.gradle.kts
+++ b/opendc-compute/opendc-compute-simulator/build.gradle.kts
@@ -37,6 +37,7 @@ dependencies {
     api(libs.microprofile.config)
     implementation(project(mapOf("path" to ":opendc-compute:opendc-compute-topology")))
     implementation(project(mapOf("path" to ":opendc-compute:opendc-compute-telemetry")))
+    implementation(project(mapOf("path" to ":opendc-compute:opendc-compute-carbon")))
 
     testImplementation(projects.opendcSimulator.opendcSimulatorCore)
     testRuntimeOnly(libs.slf4j.simple)
diff --git a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/provisioner/ComputeMonitorProvisioningStep.kt b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/provisioner/ComputeMonitorProvisioningStep.kt
index 09bc375d..f1123742 100644
--- a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/provisioner/ComputeMonitorProvisioningStep.kt
+++ b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/provisioner/ComputeMonitorProvisioningStep.kt
@@ -22,6 +22,7 @@
 
 package org.opendc.compute.simulator.provisioner
 
+import org.opendc.compute.carbon.CarbonTrace
 import org.opendc.compute.service.ComputeService
 import org.opendc.compute.telemetry.ComputeMetricReader
 import org.opendc.compute.telemetry.ComputeMonitor
@@ -36,13 +37,14 @@ public class ComputeMonitorProvisioningStep(
     private val monitor: ComputeMonitor,
     private val exportInterval: Duration,
     private val startTime: Duration = Duration.ofMillis(0),
+    private val carbonTrace: CarbonTrace = CarbonTrace(null),
 ) : ProvisioningStep {
     override fun apply(ctx: ProvisioningContext): AutoCloseable {
         val service =
             requireNotNull(
                 ctx.registry.resolve(serviceDomain, ComputeService::class.java),
             ) { "Compute service $serviceDomain does not exist" }
-        val metricReader = ComputeMetricReader(ctx.dispatcher, service, monitor, exportInterval, startTime)
+        val metricReader = ComputeMetricReader(ctx.dispatcher, service, monitor, exportInterval, startTime, carbonTrace)
         return AutoCloseable { metricReader.close() }
     }
 }
diff --git a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/provisioner/ComputeSteps.kt b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/provisioner/ComputeSteps.kt
index 8597b6f4..7a6b6927 100644
--- a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/provisioner/ComputeSteps.kt
+++ b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/provisioner/ComputeSteps.kt
@@ -24,6 +24,7 @@
 
 package org.opendc.compute.simulator.provisioner
 
+import org.opendc.compute.carbon.CarbonTrace
 import org.opendc.compute.service.ComputeService
 import org.opendc.compute.service.scheduler.ComputeScheduler
 import org.opendc.compute.telemetry.ComputeMonitor
@@ -58,8 +59,9 @@ public fun registerComputeMonitor(
     monitor: ComputeMonitor,
     exportInterval: Duration = Duration.ofMinutes(5),
     startTime: Duration = Duration.ofMillis(0),
+    carbonTrace: CarbonTrace = CarbonTrace(null),
 ): ProvisioningStep {
-    return ComputeMonitorProvisioningStep(serviceDomain, monitor, exportInterval, startTime)
+    return ComputeMonitorProvisioningStep(serviceDomain, monitor, exportInterval, startTime, carbonTrace)
 }
 
 /**
diff --git a/opendc-compute/opendc-compute-telemetry/build.gradle.kts b/opendc-compute/opendc-compute-telemetry/build.gradle.kts
index f7af3877..10f7c610 100644
--- a/opendc-compute/opendc-compute-telemetry/build.gradle.kts
+++ b/opendc-compute/opendc-compute-telemetry/build.gradle.kts
@@ -33,6 +33,7 @@ dependencies {
     implementation(libs.kotlin.logging)
     implementation(project(mapOf("path" to ":opendc-trace:opendc-trace-parquet")))
     implementation(project(mapOf("path" to ":opendc-compute:opendc-compute-service")))
+    implementation(project(mapOf("path" to ":opendc-compute:opendc-compute-carbon")))
 
     testImplementation(projects.opendcSimulator.opendcSimulatorCore)
     testRuntimeOnly(libs.log4j.core)
diff --git a/opendc-compute/opendc-compute-telemetry/src/main/kotlin/org/opendc/compute/telemetry/ComputeMetricReader.kt b/opendc-compute/opendc-compute-telemetry/src/main/kotlin/org/opendc/compute/telemetry/ComputeMetricReader.kt
index 91f7cecf..46759ed1 100644
--- a/opendc-compute/opendc-compute-telemetry/src/main/kotlin/org/opendc/compute/telemetry/ComputeMetricReader.kt
+++ b/opendc-compute/opendc-compute-telemetry/src/main/kotlin/org/opendc/compute/telemetry/ComputeMetricReader.kt
@@ -30,6 +30,7 @@ import mu.KotlinLogging
 import org.opendc.common.Dispatcher
 import org.opendc.common.asCoroutineDispatcher
 import org.opendc.compute.api.Server
+import org.opendc.compute.carbon.CarbonTrace
 import org.opendc.compute.service.ComputeService
 import org.opendc.compute.service.driver.Host
 import org.opendc.compute.telemetry.table.HostInfo
@@ -55,6 +56,7 @@ public class ComputeMetricReader(
     private val monitor: ComputeMonitor,
     private val exportInterval: Duration = Duration.ofMinutes(5),
     private val startTime: Duration = Duration.ofMillis(0),
+    private val carbonTrace: CarbonTrace = CarbonTrace(null),
 ) : AutoCloseable {
     private val logger = KotlinLogging.logger {}
     private val scope = CoroutineScope(dispatcher.asCoroutineDispatcher())
@@ -101,7 +103,7 @@ public class ComputeMetricReader(
             val now = this.clock.instant()
 
             for (host in this.service.hosts) {
-                val reader = this.hostTableReaders.computeIfAbsent(host) { HostTableReaderImpl(it, startTime) }
+                val reader = this.hostTableReaders.computeIfAbsent(host) { HostTableReaderImpl(it, startTime, carbonTrace) }
                 reader.record(now)
                 this.monitor.record(reader.copy())
                 reader.reset()
@@ -218,6 +220,7 @@ public class ComputeMetricReader(
     private class HostTableReaderImpl(
         host: Host,
         private val startTime: Duration = Duration.ofMillis(0),
+        private val carbonTrace: CarbonTrace = CarbonTrace(null),
     ) : HostTableReader {
         override fun copy(): HostTableReader {
             val newHostTable = HostTableReaderImpl(_host)
@@ -244,6 +247,8 @@ public class ComputeMetricReader(
             _cpuLostTime = table.cpuLostTime
             _powerDraw = table.powerDraw
             _energyUsage = table.energyUsage
+            _carbonIntensity = table.carbonIntensity
+            _carbonEmission = table.carbonEmission
             _uptime = table.uptime
             _downtime = table.downtime
             _bootTime = table.bootTime
@@ -322,6 +327,14 @@ public class ComputeMetricReader(
         private var _energyUsage = 0.0
         private var previousPowerTotal = 0.0
 
+        override val carbonIntensity: Double
+            get() = _carbonIntensity
+        private var _carbonIntensity = 0.0
+
+        override val carbonEmission: Double
+            get() = _carbonEmission
+        private var _carbonEmission = 0.0
+
         override val uptime: Long
             get() = _uptime - previousUptime
         private var _uptime = 0L
@@ -360,6 +373,9 @@ public class ComputeMetricReader(
             _cpuLostTime = hostCpuStats.lostTime
             _powerDraw = hostSysStats.powerDraw
             _energyUsage = hostSysStats.energyUsage
+            _carbonIntensity = carbonTrace.getCarbonIntensity(absoluteTimestamp)
+
+            _carbonEmission = carbonIntensity * (energyUsage / 3600000.0) // convert energy usage from J to kWh
             _uptime = hostSysStats.uptime.toMillis()
             _downtime = hostSysStats.downtime.toMillis()
             _bootTime = hostSysStats.bootTime
@@ -389,6 +405,9 @@ public class ComputeMetricReader(
             _cpuUtilization = 0.0
 
             _powerDraw = 0.0
+            _energyUsage = 0.0
+            _carbonIntensity = 0.0
+            _carbonEmission = 0.0
         }
     }
 
diff --git a/opendc-compute/opendc-compute-telemetry/src/main/kotlin/org/opendc/compute/telemetry/export/parquet/ParquetHostDataWriter.kt b/opendc-compute/opendc-compute-telemetry/src/main/kotlin/org/opendc/compute/telemetry/export/parquet/ParquetHostDataWriter.kt
index 4fc5a078..dc2d39c2 100644
--- a/opendc-compute/opendc-compute-telemetry/src/main/kotlin/org/opendc/compute/telemetry/export/parquet/ParquetHostDataWriter.kt
+++ b/opendc-compute/opendc-compute-telemetry/src/main/kotlin/org/opendc/compute/telemetry/export/parquet/ParquetHostDataWriter.kt
@@ -148,19 +148,27 @@ public class ParquetHostDataWriter(path: File, bufferSize: Int) :
             consumer.addDouble(data.energyUsage)
             consumer.endField("energy_usage", 18)
 
-            consumer.startField("uptime", 19)
+            consumer.startField("carbon_intensity", 19)
+            consumer.addDouble(data.carbonIntensity)
+            consumer.endField("carbon_intensity", 19)
+
+            consumer.startField("carbon_emission", 20)
+            consumer.addDouble(data.carbonEmission)
+            consumer.endField("carbon_emission", 20)
+
+            consumer.startField("uptime", 21)
             consumer.addLong(data.uptime)
-            consumer.endField("uptime", 19)
+            consumer.endField("uptime", 21)
 
-            consumer.startField("downtime", 20)
+            consumer.startField("downtime", 22)
             consumer.addLong(data.downtime)
-            consumer.endField("downtime", 20)
+            consumer.endField("downtime", 22)
 
             val bootTime = data.bootTime
             if (bootTime != null) {
-                consumer.startField("boot_time", 21)
+                consumer.startField("boot_time", 23)
                 consumer.addLong(bootTime.toEpochMilli())
-                consumer.endField("boot_time", 21)
+                consumer.endField("boot_time", 23)
             }
 
             consumer.endMessage()
@@ -234,6 +242,12 @@ public class ParquetHostDataWriter(path: File, bufferSize: Int) :
                         .required(PrimitiveType.PrimitiveTypeName.DOUBLE)
                         .named("energy_usage"),
                     Types
+                        .required(PrimitiveType.PrimitiveTypeName.DOUBLE)
+                        .named("carbon_intensity"),
+                    Types
+                        .required(PrimitiveType.PrimitiveTypeName.DOUBLE)
+                        .named("carbon_emission"),
+                    Types
                         .required(PrimitiveType.PrimitiveTypeName.INT64)
                         .named("uptime"),
                     Types
@@ -241,7 +255,6 @@ public class ParquetHostDataWriter(path: File, bufferSize: Int) :
                         .named("downtime"),
                     Types
                         .optional(PrimitiveType.PrimitiveTypeName.INT64)
-//                    .`as`(LogicalTypeAnnotation.timestampType(true, LogicalTypeAnnotation.TimeUnit.MILLIS))
                         .named("boot_time"),
                 )
                 .named("host")
diff --git a/opendc-compute/opendc-compute-telemetry/src/main/kotlin/org/opendc/compute/telemetry/table/HostTableReader.kt b/opendc-compute/opendc-compute-telemetry/src/main/kotlin/org/opendc/compute/telemetry/table/HostTableReader.kt
index 7246ef55..e6b19c11 100644
--- a/opendc-compute/opendc-compute-telemetry/src/main/kotlin/org/opendc/compute/telemetry/table/HostTableReader.kt
+++ b/opendc-compute/opendc-compute-telemetry/src/main/kotlin/org/opendc/compute/telemetry/table/HostTableReader.kt
@@ -118,6 +118,16 @@ public interface HostTableReader {
     public val energyUsage: Double
 
     /**
+     * The current carbon intensity of the host in gCO2 / kW.
+     */
+    public val carbonIntensity: Double
+
+    /**
+     * The current carbon emission since the last deadline in g.
+     */
+    public val carbonEmission: Double
+
+    /**
      * The uptime of the host since last time in ms.
      */
     public val uptime: Long