diff options
6 files changed, 85 insertions, 24 deletions
diff --git a/opendc-web/opendc-web-api/src/main/resources/application-dev.properties b/opendc-web/opendc-web-api/src/main/resources/application-dev.properties index 08d11609..98e53ee7 100644 --- a/opendc-web/opendc-web-api/src/main/resources/application-dev.properties +++ b/opendc-web/opendc-web-api/src/main/resources/application-dev.properties @@ -37,4 +37,4 @@ quarkus.opendc-ui.path=/ quarkus.resteasy.path=/api opendc.security.enabled=false - +quarkus.opendc-runner.auth.enabled=false diff --git a/opendc-web/opendc-web-runner-quarkus-deployment/build.gradle.kts b/opendc-web/opendc-web-runner-quarkus-deployment/build.gradle.kts index d31c4839..b3f1ec3b 100644 --- a/opendc-web/opendc-web-runner-quarkus-deployment/build.gradle.kts +++ b/opendc-web/opendc-web-runner-quarkus-deployment/build.gradle.kts @@ -33,4 +33,5 @@ dependencies { implementation(platform(libs.quarkus.bom)) implementation(libs.quarkus.core.deployment) + implementation(libs.quarkus.arc.deployment) } diff --git a/opendc-web/opendc-web-runner-quarkus-deployment/src/main/java/org/opendc/web/runner/deployment/OpenDCRunnerProcessor.java b/opendc-web/opendc-web-runner-quarkus-deployment/src/main/java/org/opendc/web/runner/deployment/OpenDCRunnerProcessor.java index 1d01aabc..94921454 100644 --- a/opendc-web/opendc-web-runner-quarkus-deployment/src/main/java/org/opendc/web/runner/deployment/OpenDCRunnerProcessor.java +++ b/opendc-web/opendc-web-runner-quarkus-deployment/src/main/java/org/opendc/web/runner/deployment/OpenDCRunnerProcessor.java @@ -22,6 +22,7 @@ package org.opendc.web.runner.deployment; +import io.quarkus.arc.deployment.UnremovableBeanBuildItem; import io.quarkus.deployment.annotations.BuildProducer; import io.quarkus.deployment.annotations.BuildStep; import io.quarkus.deployment.annotations.Record; @@ -30,6 +31,7 @@ import io.quarkus.deployment.builditem.nativeimage.ServiceProviderBuildItem; import io.quarkus.deployment.util.ServiceUtil; import io.quarkus.runtime.RuntimeValue; import org.opendc.trace.spi.TraceFormat; +import org.opendc.web.runner.JobManager; import org.opendc.web.runner.OpenDCRunner; import org.opendc.web.runner.runtime.OpenDCRunnerRecorder; import org.opendc.web.runner.runtime.OpenDCRunnerRuntimeConfig; @@ -72,6 +74,14 @@ public class OpenDCRunnerProcessor { } /** + * Mark {@link JobManager} as unremoveable, since we look up this service dynamically in {@link OpenDCRunnerRecorder}. + */ + @BuildStep + UnremovableBeanBuildItem unremovableBeans() { + return UnremovableBeanBuildItem.beanTypes(JobManager.class); + } + + /** * Build step to create the runner service. */ @BuildStep(onlyIf = IsIncluded.class) diff --git a/opendc-web/opendc-web-runner-quarkus/src/main/java/org/opendc/web/runner/runtime/OpenDCRunnerRecorder.java b/opendc-web/opendc-web-runner-quarkus/src/main/java/org/opendc/web/runner/runtime/OpenDCRunnerRecorder.java index eccc8dcf..f5c056ef 100644 --- a/opendc-web/opendc-web-runner-quarkus/src/main/java/org/opendc/web/runner/runtime/OpenDCRunnerRecorder.java +++ b/opendc-web/opendc-web-runner-quarkus/src/main/java/org/opendc/web/runner/runtime/OpenDCRunnerRecorder.java @@ -26,12 +26,11 @@ import io.quarkus.runtime.RuntimeValue; import io.quarkus.runtime.ShutdownContext; import io.quarkus.runtime.annotations.Recorder; import org.jboss.logging.Logger; -import org.opendc.web.client.runner.OpenDCRunnerClient; import org.opendc.web.runner.JobManager; import org.opendc.web.runner.OpenDCRunner; +import javax.enterprise.inject.spi.CDI; import java.io.File; -import java.net.URI; /** * Helper class for starting the OpenDC web runner. @@ -44,8 +43,6 @@ public class OpenDCRunnerRecorder { * Helper method to create an {@link OpenDCRunner} instance. */ public RuntimeValue<OpenDCRunner> createRunner(OpenDCRunnerRuntimeConfig config) { - URI apiUrl = URI.create(config.apiUrl); - int parallelism = config.parallelism; if (parallelism < 0) { throw new IllegalArgumentException("Parallelism must be non-negative"); @@ -53,9 +50,9 @@ public class OpenDCRunnerRecorder { parallelism = Math.min(1, Runtime.getRuntime().availableProcessors() - 1); } - OpenDCRunnerClient client = new OpenDCRunnerClient(apiUrl, null); + JobManager manager = CDI.current().select(JobManager.class).get(); OpenDCRunner runner = new OpenDCRunner( - JobManager.create(client), + manager, new File(config.tracePath), parallelism, config.jobTimeout, @@ -73,18 +70,10 @@ public class OpenDCRunnerRecorder { OpenDCRunnerRuntimeConfig config, ShutdownContext shutdownContext) { if (config.enable) { - LOGGER.info("Starting OpenDC Runner in background (polling " + config.apiUrl + ")"); - - Thread thread = new Thread(() -> { - try { - // Wait for some time to allow Vert.x to bind to the local port - Thread.sleep(4000); - } catch (InterruptedException e) { - throw new RuntimeException(e); - } + LOGGER.info("Starting OpenDC Runner in background (polling every " + config.pollInterval + ")"); - runner.getValue().run(); - }); + Thread thread = new Thread(runner.getValue()); + thread.setName("opendc-runner"); thread.start(); shutdownContext.addShutdownTask(thread::interrupt); diff --git a/opendc-web/opendc-web-runner-quarkus/src/main/java/org/opendc/web/runner/runtime/OpenDCRunnerRuntimeConfig.java b/opendc-web/opendc-web-runner-quarkus/src/main/java/org/opendc/web/runner/runtime/OpenDCRunnerRuntimeConfig.java index e1dbf0a8..e9258f06 100644 --- a/opendc-web/opendc-web-runner-quarkus/src/main/java/org/opendc/web/runner/runtime/OpenDCRunnerRuntimeConfig.java +++ b/opendc-web/opendc-web-runner-quarkus/src/main/java/org/opendc/web/runner/runtime/OpenDCRunnerRuntimeConfig.java @@ -40,12 +40,6 @@ public class OpenDCRunnerRuntimeConfig { public boolean enable; /** - * The URI to the (local) API to communicate with. - */ - @ConfigItem(defaultValue = "http://${quarkus.http.host}:${quarkus.http.port}${quarkus.resteasy.path:}") - public String apiUrl; - - /** * The path where the workload traces are located. */ @ConfigItem(defaultValue = "traces") diff --git a/opendc-web/opendc-web-server/src/main/kotlin/org/opendc/web/server/util/runner/QuarkusJobManager.kt b/opendc-web/opendc-web-server/src/main/kotlin/org/opendc/web/server/util/runner/QuarkusJobManager.kt new file mode 100644 index 00000000..4704c5ae --- /dev/null +++ b/opendc-web/opendc-web-server/src/main/kotlin/org/opendc/web/server/util/runner/QuarkusJobManager.kt @@ -0,0 +1,67 @@ +/* + * 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.web.server.util.runner + +import org.opendc.web.proto.JobState +import org.opendc.web.proto.runner.Job +import org.opendc.web.runner.JobManager +import org.opendc.web.server.service.JobService +import javax.enterprise.context.ApplicationScoped +import javax.inject.Inject +import javax.transaction.Transactional + +/** + * Implementation of [JobManager] that interfaces directly with [JobService] without overhead of the REST API. + */ +@ApplicationScoped +class QuarkusJobManager @Inject constructor(private val service: JobService) : JobManager { + @Transactional + override fun findNext(): Job? { + return service.queryPending().firstOrNull() + } + + @Transactional + override fun claim(id: Long): Boolean { + return try { + service.updateState(id, JobState.CLAIMED, null) + true + } catch (e: IllegalStateException) { + false + } + } + + @Transactional + override fun heartbeat(id: Long) { + service.updateState(id, JobState.RUNNING, null) + } + + @Transactional + override fun fail(id: Long) { + service.updateState(id, JobState.FAILED, null) + } + + @Transactional + override fun finish(id: Long, results: Map<String, Any>) { + service.updateState(id, JobState.FINISHED, results) + } +} |