diff options
Diffstat (limited to 'opendc-compute/opendc-compute-workload')
4 files changed, 227 insertions, 47 deletions
diff --git a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/ComputeServiceHelper.kt b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/ComputeServiceHelper.kt index 59203b66..a1a65da3 100644 --- a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/ComputeServiceHelper.kt +++ b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/ComputeServiceHelper.kt @@ -22,10 +22,6 @@ package org.opendc.compute.workload -import io.opentelemetry.sdk.metrics.SdkMeterProvider -import io.opentelemetry.sdk.metrics.export.MetricProducer -import io.opentelemetry.sdk.resources.Resource -import io.opentelemetry.semconv.resource.attributes.ResourceAttributes import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.delay import kotlinx.coroutines.launch @@ -33,12 +29,11 @@ import kotlinx.coroutines.yield import org.opendc.compute.service.ComputeService import org.opendc.compute.service.scheduler.ComputeScheduler import org.opendc.compute.simulator.SimHost +import org.opendc.compute.workload.telemetry.TelemetryManager import org.opendc.compute.workload.topology.HostSpec import org.opendc.simulator.compute.kernel.interference.VmInterferenceModel import org.opendc.simulator.compute.workload.SimTraceWorkload import org.opendc.simulator.flow.FlowEngine -import org.opendc.telemetry.compute.* -import org.opendc.telemetry.sdk.toOtelClock import java.time.Clock import java.time.Duration import java.util.* @@ -50,6 +45,7 @@ import kotlin.math.max * * @param context [CoroutineContext] to run the simulation in. * @param clock [Clock] instance tracking simulation time. + * @param telemetry Helper class for managing telemetry. * @param scheduler [ComputeScheduler] implementation to use for the service. * @param failureModel A failure model to use for injecting failures. * @param interferenceModel The model to use for performance interference. @@ -58,6 +54,7 @@ import kotlin.math.max public class ComputeServiceHelper( private val context: CoroutineContext, private val clock: Clock, + private val telemetry: TelemetryManager, scheduler: ComputeScheduler, private val failureModel: FailureModel? = null, private val interferenceModel: VmInterferenceModel? = null, @@ -69,25 +66,17 @@ public class ComputeServiceHelper( public val service: ComputeService /** - * The [MetricProducer] that are used by the [ComputeService] and the simulated hosts. - */ - public val producers: List<MetricProducer> - get() = _metricProducers - private val _metricProducers = mutableListOf<MetricProducer>() - - /** * The [FlowEngine] to simulate the hosts. */ - private val engine = FlowEngine(context, clock) + private val _engine = FlowEngine(context, clock) /** * The hosts that belong to this class. */ - private val hosts = mutableSetOf<SimHost>() + private val _hosts = mutableSetOf<SimHost>() init { - val (service, serviceMeterProvider) = createService(scheduler, schedulingQuantum) - this._metricProducers.add(serviceMeterProvider) + val service = createService(scheduler, schedulingQuantum) this.service = service } @@ -165,27 +154,14 @@ public class ComputeServiceHelper( * @return The [SimHost] that has been constructed by the runner. */ public fun registerHost(spec: HostSpec, optimize: Boolean = false): SimHost { - val resource = Resource.builder() - .put(HOST_ID, spec.uid.toString()) - .put(HOST_NAME, spec.name) - .put(HOST_ARCH, ResourceAttributes.HostArchValues.AMD64) - .put(HOST_NCPUS, spec.model.cpus.size) - .put(HOST_MEM_CAPACITY, spec.model.memory.sumOf { it.size }) - .build() - - val meterProvider = SdkMeterProvider.builder() - .setClock(clock.toOtelClock()) - .setResource(resource) - .build() - _metricProducers.add(meterProvider) - + val meterProvider = telemetry.createMeterProvider(spec) val host = SimHost( spec.uid, spec.name, spec.model, spec.meta, context, - engine, + _engine, meterProvider, spec.hypervisor, powerDriver = spec.powerDriver, @@ -193,7 +169,7 @@ public class ComputeServiceHelper( optimize = optimize ) - hosts.add(host) + _hosts.add(host) service.addHost(host) return host @@ -202,27 +178,18 @@ public class ComputeServiceHelper( override fun close() { service.close() - for (host in hosts) { + for (host in _hosts) { host.close() } - hosts.clear() + _hosts.clear() } /** * Construct a [ComputeService] instance. */ - private fun createService(scheduler: ComputeScheduler, schedulingQuantum: Duration): Pair<ComputeService, SdkMeterProvider> { - val resource = Resource.builder() - .put(ResourceAttributes.SERVICE_NAME, "opendc-compute") - .build() - - val meterProvider = SdkMeterProvider.builder() - .setClock(clock.toOtelClock()) - .setResource(resource) - .build() - - val service = ComputeService(context, clock, meterProvider, scheduler, schedulingQuantum) - return service to meterProvider + private fun createService(scheduler: ComputeScheduler, schedulingQuantum: Duration): ComputeService { + val meterProvider = telemetry.createMeterProvider(scheduler) + return ComputeService(context, clock, meterProvider, scheduler, schedulingQuantum) } } diff --git a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/telemetry/NoopTelemetryManager.kt b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/telemetry/NoopTelemetryManager.kt new file mode 100644 index 00000000..4e7d0b75 --- /dev/null +++ b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/telemetry/NoopTelemetryManager.kt @@ -0,0 +1,36 @@ +/* + * 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.compute.workload.telemetry + +import io.opentelemetry.api.metrics.MeterProvider +import org.opendc.compute.service.scheduler.ComputeScheduler +import org.opendc.compute.workload.topology.HostSpec + +/** + * A [TelemetryManager] that does nothing. + */ +public class NoopTelemetryManager : TelemetryManager { + override fun createMeterProvider(host: HostSpec): MeterProvider = MeterProvider.noop() + + override fun createMeterProvider(scheduler: ComputeScheduler): MeterProvider = MeterProvider.noop() +} diff --git a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/telemetry/SdkTelemetryManager.kt b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/telemetry/SdkTelemetryManager.kt new file mode 100644 index 00000000..478c0609 --- /dev/null +++ b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/telemetry/SdkTelemetryManager.kt @@ -0,0 +1,135 @@ +/* + * 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.compute.workload.telemetry + +import io.opentelemetry.api.metrics.MeterProvider +import io.opentelemetry.sdk.common.CompletableResultCode +import io.opentelemetry.sdk.metrics.SdkMeterProvider +import io.opentelemetry.sdk.metrics.data.AggregationTemporality +import io.opentelemetry.sdk.metrics.data.MetricData +import io.opentelemetry.sdk.metrics.export.MetricProducer +import io.opentelemetry.sdk.metrics.export.MetricReader +import io.opentelemetry.sdk.metrics.export.MetricReaderFactory +import io.opentelemetry.sdk.resources.Resource +import io.opentelemetry.semconv.resource.attributes.ResourceAttributes +import org.opendc.compute.service.scheduler.ComputeScheduler +import org.opendc.compute.workload.topology.HostSpec +import org.opendc.telemetry.compute.* +import org.opendc.telemetry.sdk.toOtelClock +import java.time.Clock + +/** + * A [TelemetryManager] using the OpenTelemetry Java SDK. + */ +public class SdkTelemetryManager(private val clock: Clock) : TelemetryManager, AutoCloseable { + /** + * The [SdkMeterProvider]s that belong to the workload runner. + */ + private val _meterProviders = mutableListOf<SdkMeterProvider>() + + /** + * The internal [MetricProducer] registered with the runner. + */ + private val _metricProducers = mutableListOf<MetricProducer>() + + /** + * The list of [MetricReader]s that have been registered with the runner. + */ + private val _metricReaders = mutableListOf<MetricReader>() + + /** + * A [MetricProducer] that combines all the other metric producers. + */ + public val metricProducer: MetricProducer = object : MetricProducer { + private val producers = _metricProducers + + override fun collectAllMetrics(): Collection<MetricData> = producers.flatMap(MetricProducer::collectAllMetrics) + + override fun toString(): String = "SdkTelemetryManager.AggregateMetricProducer" + } + + /** + * Register a [MetricReader] for this manager. + * + * @param factory The factory for the reader to register. + */ + public fun registerMetricReader(factory: MetricReaderFactory) { + val reader = factory.apply(metricProducer) + _metricReaders.add(reader) + } + + override fun createMeterProvider(scheduler: ComputeScheduler): MeterProvider { + val resource = Resource.builder() + .put(ResourceAttributes.SERVICE_NAME, "opendc-compute") + .build() + + return createMeterProvider(resource) + } + + override fun createMeterProvider(host: HostSpec): MeterProvider { + val resource = Resource.builder() + .put(HOST_ID, host.uid.toString()) + .put(HOST_NAME, host.name) + .put(HOST_ARCH, ResourceAttributes.HostArchValues.AMD64) + .put(HOST_NCPUS, host.model.cpus.size) + .put(HOST_MEM_CAPACITY, host.model.memory.sumOf { it.size }) + .build() + + return createMeterProvider(resource) + } + + /** + * Construct a [SdkMeterProvider] for the specified [resource]. + */ + private fun createMeterProvider(resource: Resource): SdkMeterProvider { + val meterProvider = SdkMeterProvider.builder() + .setClock(clock.toOtelClock()) + .setResource(resource) + .registerMetricReader { producer -> + _metricProducers.add(producer) + object : MetricReader { + override fun getPreferredTemporality(): AggregationTemporality = AggregationTemporality.CUMULATIVE + override fun flush(): CompletableResultCode = CompletableResultCode.ofSuccess() + override fun shutdown(): CompletableResultCode = CompletableResultCode.ofSuccess() + } + } + .build() + _meterProviders.add(meterProvider) + return meterProvider + } + + override fun close() { + for (meterProvider in _meterProviders) { + meterProvider.close() + } + + _meterProviders.clear() + + for (metricReader in _metricReaders) { + metricReader.shutdown() + } + + _metricReaders.clear() + _metricProducers.clear() + } +} diff --git a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/telemetry/TelemetryManager.kt b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/telemetry/TelemetryManager.kt new file mode 100644 index 00000000..b67050ce --- /dev/null +++ b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/telemetry/TelemetryManager.kt @@ -0,0 +1,42 @@ +/* + * 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.compute.workload.telemetry + +import io.opentelemetry.api.metrics.MeterProvider +import org.opendc.compute.service.scheduler.ComputeScheduler +import org.opendc.compute.workload.topology.HostSpec + +/** + * Helper class to manage the telemetry for a [ComputeServiceHelper] instance. + */ +public interface TelemetryManager { + /** + * Construct a [MeterProvider] for the specified [ComputeScheduler]. + */ + public fun createMeterProvider(scheduler: ComputeScheduler): MeterProvider + + /** + * Construct a [MeterProvider] for the specified [HostSpec]. + */ + public fun createMeterProvider(host: HostSpec): MeterProvider +} |