When receiving a message, the handler instantiates a `NebulousApp` + * object. If we already received the performance indicators from the + * utility evaluator, perform initial deployment; otherwise, wait. */ // Note that there is another, earlier app creation message sent via the // channel `eu.nebulouscloud.ui.application.new`, but its format is not @@ -191,21 +213,106 @@ public class ExnConnector { public class AppCreationMessageHandler extends Handler { @Override public void onMessage(String key, String address, Map body, Message message, Context context) { + NebulousApp app = null; try { Object app_id = message.property("application"); // might be null if (app_id == null) app_id = message.subject(); // if app_id is still null, the filename will look a bit funky but it's not a problem log.info("App creation message received", keyValue("appId", app_id)); JsonNode appMessage = mapper.valueToTree(body); - Main.logFile("app-message-" + app_id + ".json", appMessage); - NebulousApp app = NebulousApp.newFromAppMessage( + Main.logFile("app-message-" + app_id + ".json", appMessage.toPrettyString()); + app = NebulousApp.newFromAppMessage( // TODO create a new ExnConnector here? mapper.valueToTree(body), ExnConnector.this); NebulousApps.add(app); - app.sendAMPL(); - app.deployUnmodifiedApplication(); + String appIdFromMessage = app.getUUID(); + if (NebulousApps.relevantPerformanceIndicators.containsKey(appIdFromMessage)) { + // If the performance indicators haven't arrived yet, this + // will happen in PerformanceIndicatorMessageHandler below. + app.setStateReady(NebulousApps.relevantPerformanceIndicators.get(appIdFromMessage)); + app.deployUnmodifiedApplication(); + // Not strictly necessary to remove the performance + // indicators, but let's not leave unneeded data around + NebulousApps.relevantPerformanceIndicators.remove(appIdFromMessage); + } } catch (Exception e) { log.error("Error while receiving app creation message", e); + if (app != null) app.setStateFailed(); + } + } + } + + /** + * A handler that receives the performance indicators that the utility + * evaluator sends. If the application object already exists (usually the + * case), start initial deployment, otherwise store the performance + * indicators so the initial app creation message can pick them up. + */ + public class PerformanceIndicatorMessageHandler extends Handler { + @Override + public void onMessage(String key, String address, Map body, Message message, Context context) { + Object appIdObject = null; + try { + appIdObject = message.property("application"); + if (appIdObject == null) appIdObject = message.subject(); + } catch (ClientException e) { + log.error("Received performance indicator message without application property, aborting"); + return; + } + String appId = null; + if (appIdObject == null) { + log.error("Received performance indicator message without application property, aborting"); + } else { + appId = appIdObject.toString(); // should be a string already + log.info("Received performance indicator message", keyValue("appId", appId)); + } + JsonNode appMessage = mapper.valueToTree(body); + Main.logFile("performance-indicators-" + appIdObject + ".json", appMessage.toPrettyString()); + NebulousApp app = NebulousApps.get(appId); + if (app == null) { + NebulousApps.relevantPerformanceIndicators.put(appId, appMessage); + log.info("Received performance indicator message for unknown app object, storing", + keyValue("appId", appId)); + } else { + app.setStateReady(appMessage); + app.deployUnmodifiedApplication(); + } + } + } + + /** + * A handler that detects when the solver has started for a given + * application. + */ + public class SolverStatusMessageHandler extends Handler { + @Override + public void onMessage(String key, String address, Map body, Message message, Context context) { + Object appIdObject = null; + String appId = null; + try { + appIdObject = message.property("application"); + } catch (ClientException e) { + log.error("Received solver ready message {} without application property, aborting", body); + return; + } + if (appIdObject == null) { + log.error("Received solver ready message {} without application property, aborting", body); + return; + } else { + appId = appIdObject.toString(); // should be a string already + log.info("Received solver status message {}", body, keyValue("appId", appId)); + } + + JsonNode appMessage = mapper.valueToTree(body); + String status = appMessage.at("/state").textValue(); + if (status == null || !status.equals("started")) return; + + NebulousApp app = NebulousApps.get(appId); + if (app == null) { + log.info("Received solver status message for unknown app object, this should not happen", + keyValue("appId", appId)); + } else { + app.sendAMPL(); } } } @@ -227,7 +334,7 @@ public class ExnConnector { log.warn("Received solver solution without 'application' message property, discarding it"); return; } - Main.logFile("solver-solution-" + app_id + ".json", json_body); + Main.logFile("solver-solution-" + app_id + ".json", json_body.toPrettyString()); NebulousApp app = NebulousApps.get(app_id); if (app == null) { log.warn("Received solver solutions for non-existant application, discarding.", @@ -271,13 +378,13 @@ public class ExnConnector { String salRawResponse = response.at("/body").asText(); // it's already a string, asText() is for the type system JsonNode metadata = response.at("/metaData"); JsonNode salResponse = mapper.missingNode(); // the data coming from SAL - try { - salResponse = mapper.readTree(salRawResponse); - } catch (JsonProcessingException e) { + try { + salResponse = mapper.readTree(salRawResponse); + } catch (JsonProcessingException e) { log.error("Could not read message body as JSON: body = '{}'", salRawResponse, keyValue("appId", appID), keyValue("caller", caller), e); return mapper.missingNode(); - } + } if (!metadata.at("/status").asText().startsWith("2")) { // we only accept 200, 202, numbers of that nature log.error("exn-middleware-sal request failed with error code '{}' and message '{}'", @@ -391,14 +498,14 @@ public class ExnConnector { if (cpu1 != cpu2) return cpu1 - cpu2; else return Math.toIntExact(ram1 - ram2); }); - return candidates; + return candidates; } /** * Define a cluster with the given name and node list. - * + * *
The cluster is passed in a JSON of the following shape: - * + * *
{@code
* {
* "name":"485d7-1",
@@ -420,13 +527,13 @@ public class ExnConnector {
* }
* }
* }
- *
+ *
* Each value for {@code nodeName} has to be globally unique, must * start with a letter and contain numbers, letters and hyphens only. - * + * *
The values for {@code nodeCandidateId} and {@code cloudId} come from
* the return value of a call to {@link #findNodeCandidates()}.
- *
+ *
* @param appID The application's id, used only for logging.
* @param clusterName The cluster name, used only for logging.
* @param cluster A JSON object, as detailed above.
@@ -434,7 +541,7 @@ public class ExnConnector {
*/
public boolean defineCluster(String appID, String clusterName, ObjectNode cluster) {
// https://openproject.nebulouscloud.eu/projects/nebulous-collaboration-hub/wiki/deployment-manager-sal-1#specification-of-endpoints-being-developed
- Main.logFile("define-cluster-" + appID + ".json", cluster);
+ Main.logFile("define-cluster-" + appID + ".json", cluster.toPrettyString());
Map TODO: also send performance indicators to solver here
*/
public void sendAMPL() {
- String ampl = AMPLGenerator.generateAMPL(this);
+ String ampl_model = AMPLGenerator.generateAMPL(this);
+ String ampl_data = relevantPerformanceIndicators.at("/initialDataFile").textValue();
ObjectNode msg = jsonMapper.createObjectNode();
- msg.put("FileName", getUUID() + ".ampl");
- msg.put("FileContent", ampl);
+
+ msg.put("ModelFileName", getUUID() + ".ampl");
+ msg.put("ModelFileContent", ampl_model);
+ if (ampl_data != null && ampl_data.length() > 0) {
+ msg.put("DataFileName", getUUID() + ".dat");
+ msg.put("DataFileContent", ampl_data);
+ }
msg.put("ObjectiveFunction", getObjectiveFunction());
ObjectNode constants = msg.withObject("Constants");
// Define initial values for constant utility functions:
@@ -499,10 +522,28 @@ public class NebulousApp {
constant.put("Variable", variableName);
constant.set("Value", value);
}
- log.info("Sending AMPL file to solver", keyValue("amplMessage", msg), keyValue("appId", UUID));
+ log.info("Sending AMPL files to solver", keyValue("amplMessage", msg), keyValue("appId", UUID));
exnConnector.getAmplMessagePublisher().send(jsonMapper.convertValue(msg, Map.class), getUUID(), true);
- Main.logFile("to-solver-" + getUUID() + ".json", msg.toString());
- Main.logFile("to-solver-" + getUUID() + ".ampl", ampl);
+ Main.logFile("to-solver-" + getUUID() + ".json", msg.toPrettyString());
+ Main.logFile("to-solver-" + getUUID() + ".ampl", ampl_model);
+ }
+
+ /**
+ * Send the metric list for the given app. This is done two times: once
+ * before app cluster creation to initialize EMS, once after cluster app
+ * creation to initialize the solver.
+ *
+ * @param app The application under deployment.
+ */
+ public void sendMetricList() {
+ Publisher metricsChannel = exnConnector.getMetricListPublisher();
+ ObjectNode msg = jsonMapper.createObjectNode();
+ ArrayNode metricNames = msg.withArray("/metrics");
+ getRawMetrics().forEach((k, v) -> metricNames.add(k));
+ getCompositeMetrics().forEach((k, v) -> metricNames.add(k));
+ log.info("Sending metric list", keyValue("appId", UUID));
+ metricsChannel.send(jsonMapper.convertValue(msg, Map.class), getUUID(), true);
+ Main.logFile("metric-names-" + getUUID() + ".json", msg.toPrettyString());
}
/**
diff --git a/optimiser-controller/src/main/java/eu/nebulouscloud/optimiser/controller/NebulousAppDeployer.java b/optimiser-controller/src/main/java/eu/nebulouscloud/optimiser/controller/NebulousAppDeployer.java
index 8a3c394..3b797e7 100644
--- a/optimiser-controller/src/main/java/eu/nebulouscloud/optimiser/controller/NebulousAppDeployer.java
+++ b/optimiser-controller/src/main/java/eu/nebulouscloud/optimiser/controller/NebulousAppDeployer.java
@@ -8,6 +8,7 @@ import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
+import eu.nebulouscloud.exn.core.Publisher;
import eu.nebulouscloud.optimiser.kubevela.KubevelaAnalyzer;
import org.ow2.proactive.sal.model.AttributeRequirement;
import org.ow2.proactive.sal.model.NodeCandidate;
@@ -137,6 +138,9 @@ public class NebulousAppDeployer {
* {@code false} otherwise.
*/
private static boolean isClusterDeploymentFinished(JsonNode clusterStatus) {
+ if (clusterStatus == null || !clusterStatus.isObject())
+ // Catch various failure states, e.g., SAL spuriously returning null
+ return false;
return clusterStatus.withArray("/nodes")
.findParents("state")
.stream()
@@ -213,12 +217,15 @@ public class NebulousAppDeployer {
// controller.
// - Select node candidates, making sure to only select edge nodes
// once.
+ // - (Before deploying the cluster) send metric name list.
// - Create a SAL cluster.
// - Deploy the SAL cluster.
// - Add node affinity traits to the KubeVela file.
// - Deploy the SAL application.
// - Store cluster state (deployed KubeVela file, etc.) in
// NebulousApp object.
+ // - Asynchronously, triggered via solver readiness message: wait for
+ // solver to be ready, send AMPL and re-send metric name list.
// ------------------------------------------------------------
// Extract node requirements
@@ -355,8 +362,9 @@ public class NebulousAppDeployer {
return;
}
- // TODO: send performance indicators (for monitoring system, which
- // needs it before cluster creation)
+ // ------------------------------------------------------------
+ // Send metrics to EMS
+ app.sendMetricList();
// ------------------------------------------------------------
// Create cluster
diff --git a/optimiser-controller/src/main/java/eu/nebulouscloud/optimiser/controller/NebulousApps.java b/optimiser-controller/src/main/java/eu/nebulouscloud/optimiser/controller/NebulousApps.java
index 1c78416..3bb82c9 100644
--- a/optimiser-controller/src/main/java/eu/nebulouscloud/optimiser/controller/NebulousApps.java
+++ b/optimiser-controller/src/main/java/eu/nebulouscloud/optimiser/controller/NebulousApps.java
@@ -7,15 +7,24 @@ import java.util.Collection;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import com.fasterxml.jackson.databind.JsonNode;
+
/**
- * Class that manages a collection of NebulousApp instances.
+ * Class that manages a collection of NebulousApp instances. Also used to
+ * temporarily store the relevant performance indicators which are calculated
+ * by the utility evaluator from the same incoming app creation message.
*/
@Slf4j
public class NebulousApps {
/** The global app registry. */
- // (Putting this here until we find a better place.)
- private static final Map