nebulous/optimiser-controller
Code Issues Proposed changes

Optionally save messages in files

Browse Source 
During testing, we might want to look at the content of app creation
messages, AMPL code etc.  Specify a directory where such files can be
created and stored.

Start with `--log-dir` or set `LOGDIR` environment variable to store
received app creation messages, sent AMPL files, etc.

Set LOGDIR to /tmp/nebulous for test deployment.

Refine logging levels: output only errors and warnings by default,
and increasing levels when started with `-v`, `-vv`, `-vvv`.

Change-Id: Ie5cc770a609a0abd978bc113fb1da4dfa46873f0
This commit is contained in:
Rudi Schlatte
2024-01-24 17:08:08 +01:00
parent 4a3ee57725
commit d80c1f5ea3
8 changed files with 112 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
charts/nebulous-optimiser-controller/templates/deployment.yaml
View File

@@ -37,7 +37,9 @@ spec:
- name: http
containerPort: 8080
protocol: TCP
env:
env:
- name: LOGDIR
value: "{{ .Values.debug.LOGDIR }}"
- name: ACTIVEMQ_HOST
value: "{{ .Values.activemq.ACTIVEMQ_HOST }}"
- name: ACTIVEMQ_PORT

7
charts/nebulous-optimiser-controller/values.yaml
View File

@@ -81,6 +81,9 @@ tolerations: []
affinity: {}
debug:
LOGDIR: /tmp/nebulous
sal:
SAL_URL: sal
SAL_USER: admin
@@ -89,7 +92,7 @@ activemq:
ACTIVEMQ_HOST: activemq
ACTIVEMQ_PORT: 5672
ACTIVEMQ_USER: admin
secrets:
ACTIVEMQ_PASSWORD: nebulous
SAL_PASSWORD: admin
SAL_PASSWORD: admin

11
optimiser-controller/src/main/java/eu/nebulouscloud/optimiser/controller/ExnConnector.java
View File

@@ -12,6 +12,7 @@ import lombok.extern.slf4j.Slf4j;
import org.apache.qpid.protonj2.client.Message;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
@@ -88,7 +89,7 @@ public class ExnConnector {
public synchronized void start(CountDownLatch synchronizer) {
this.synchronizer = synchronizer;
conn.start();
log.info("ExnConnector started.");
log.debug("ExnConnector started.");
}
/**
@@ -101,7 +102,7 @@ public class ExnConnector {
if (synchronizer != null) {
synchronizer.countDown();
}
log.info("ExnConnector stopped.");
log.debug("ExnConnector stopped.");
}
/**
@@ -122,6 +123,8 @@ public class ExnConnector {
try {
String app_id = message.subject();
log.info("App creation message received for app {}", app_id);
JsonNode appMessage = mapper.valueToTree(body);
Main.logFile("app-message-" + app_id + ".json", appMessage);
NebulousApp app = NebulousApp.newFromAppMessage(mapper.valueToTree(body), amplMessagePublisher);
NebulousApps.add(app);
app.sendAMPL();
@@ -145,10 +148,10 @@ public class ExnConnector {
String app_id = message.subject();
NebulousApp app = NebulousApps.get(app_id);
if (app == null) {
log.error("Received solver solutions for non-existant app " + app_id);
log.warn("Received solver solutions for non-existant app {}, discarding.", app_id);
return;
} else {
log.info("Received solver solutions for app {}", app_id);
log.debug("Received solver solutions for app {}", app_id);
ObjectNode json_body = mapper.convertValue(body, ObjectNode.class);
app.processSolution(json_body);
}

2
optimiser-controller/src/main/java/eu/nebulouscloud/optimiser/controller/LocalExecution.java
View File

@@ -49,7 +49,7 @@ public class LocalExecution implements Callable<Integer> {
}
NebulousApp app = NebulousApp.newFromAppMessage(msg, publisher);
if (connector != null) {
log.info("Sending AMPL to channel {}", publisher);
log.debug("Sending AMPL to channel {}", publisher);
app.sendAMPL();
}
System.out.println(AMPLGenerator.generateAMPL(app));

96
optimiser-controller/src/main/java/eu/nebulouscloud/optimiser/controller/Main.java
View File

@@ -1,8 +1,14 @@
package eu.nebulouscloud.optimiser.controller;
import java.io.FileWriter;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.time.LocalDateTime;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicReference;
import java.util.logging.Level;
import eu.nebulouscloud.exn.core.Context;
import eu.nebulouscloud.exn.handlers.ConnectorHandler;
@@ -75,18 +81,17 @@ public class Main implements Callable<Integer> {
defaultValue = "${ACTIVEMQ_PASSWORD}")
private String activemq_password;
@Option(names = {"--log-dir"},
description = "Directory where to log incoming and outgoing messages as files. Can also be set via the @|bold LOGDIR|@ variable.",
paramLabel = "LOGDIR",
defaultValue = "${LOGDIR}")
@Getter
private static Path logDirectory;
@Option(names = {"--verbose", "-v"},
description = "Turn on more verbose logging output.",
description = "Turn on more verbose logging output. Can be given multiple times. When not given, print only warnings and error messages. With @|underline -v|@, print status messages. With @|underline -vvv|@, print everything.",
scope = ScopeType.INHERIT)
public void setVerbose(boolean[] verbose) {
// java.util.logging wants to be configured with a configuration file.
// Convince it otherwise.
java.util.logging.Logger rootLogger = java.util.logging.Logger.getLogger("");
rootLogger.setLevel(java.util.logging.Level.FINER);
for (java.util.logging.Handler handler : rootLogger.getHandlers()) {
handler.setLevel(rootLogger.getLevel());
}
}
private boolean[] verbosity;
/**
* The connector to the SAL library.
@@ -118,22 +123,55 @@ public class Main implements Callable<Integer> {
* `activeMQConnector.start`.
*/
private void init() {
log.info("Beginning common startup of optimiser-controller");
log.debug("Beginning common startup of optimiser-controller");
// Set log level. java.util.logging wants to be configured with a
// configuration file, convince it otherwise.
java.util.logging.Logger rootLogger = java.util.logging.Logger.getLogger("");
Level level;
if (verbosity == null) level = Level.SEVERE;
else
switch (verbosity.length) {
case 0: level = Level.SEVERE; break; // can't happen
case 1: level = Level.INFO; break; // Skip warning, since I (rudi) want INFO with just `-v`
case 2: level = Level.FINE; break;
default: level = Level.ALL; break;
}
rootLogger.setLevel(level);
for (java.util.logging.Handler handler : rootLogger.getHandlers()) {
handler.setLevel(rootLogger.getLevel());
}
// Set up directory for file logs (dumps of contents of incoming or
// outgoing messages).
if (logDirectory != null) {
if (!Files.exists(logDirectory)) {
try {
Files.createDirectories(logDirectory);
} catch (IOException e) {
log.warn("Could not create log directory {}. Continuing without file logging.");
logDirectory = null;
}
} else if (!Files.isDirectory(logDirectory) || !Files.isWritable(logDirectory)) {
log.warn("Trying to use a file as log directory, or directory not writable: {}. Continuing without file logging.", logDirectory);
logDirectory = null;
} else {
log.debug("Logging all messages to directory {}", logDirectory);
}
}
// Start connection to SAL if possible.
if (sal_uri != null && sal_user != null && sal_password != null) {
salConnector = new SalConnector(sal_uri, sal_user, sal_password);
if (!salConnector.isConnected()) {
log.error("Connection to SAL unsuccessful");
log.warn("Connection to SAL unsuccessful, continuing without SAL");
} else {
log.info("Established connection to SAL");
NebulousApp.setSalConnector(salConnector);
}
} else {
log.info("SAL login information not specified, skipping");
log.debug("SAL login information not specified, skipping");
}
// Start connection to ActiveMQ if possible.
if (activemq_user != null && activemq_password != null) {
log.info("Preparing ActiveMQ connection: host={} port={}",
log.debug("Preparing ActiveMQ connection: host={} port={}",
activemq_host, activemq_port);
activeMQConnector
= new ExnConnector(activemq_host, activemq_port,
@@ -144,7 +182,7 @@ public class Main implements Callable<Integer> {
}
});
} else {
log.info("ActiveMQ login info not set, only operating locally.");
log.debug("ActiveMQ login info not set, only operating locally.");
}
}
@@ -157,7 +195,7 @@ public class Main implements Callable<Integer> {
public Integer call() {
CountDownLatch exn_synchronizer = new CountDownLatch(1);
if (activeMQConnector != null) {
log.info("Starting connection to ActiveMQ");
log.debug("Starting connection to ActiveMQ");
activeMQConnector.start(exn_synchronizer);
} else {
log.error("ActiveMQ connector not initialized so we're unresponsive. Will keep running to keep CI/CD happy but don't expect anything more from me.");
@@ -186,4 +224,26 @@ public class Main implements Callable<Integer> {
.execute(args);
System.exit(exitCode);
}
/**
* Log a file into the given log directory. Does nothing if {@link
* Main#logDirectory} is not set.
*
* @param name The filename. Note that the file that is written will have
* a longer name including a timestamp, so this argument does not need to
* be unique.
* @param contents The content of the file to be written. Will be
* converted to String via `toString`.
*/
public static void logFile(String name, Object contents) {
if (Main.logDirectory == null) return;
String prefix = LocalDateTime.now().toString();
Path path = logDirectory.resolve(prefix + "--" + name);
try (FileWriter out = new FileWriter(path.toFile())) {
out.write(contents.toString());
log.trace("Wrote log file {}", path);
} catch (IOException e) {
log.warn("Error while trying to create data file in log directory", e);
}
}
}

21
optimiser-controller/src/main/java/eu/nebulouscloud/optimiser/controller/NebulousApp.java
View File

@@ -171,7 +171,7 @@ public class NebulousApp {
// What's left is neither a raw nor composite metric.
performanceIndicators.put(m.get("key").asText(), m);
}
log.info("New App instantiated: Name='{}', UUID='{}'", name, UUID);
log.debug("New App instantiated: Name='{}', UUID='{}'", name, UUID);
}
/**
@@ -316,13 +316,14 @@ public class NebulousApp {
*/
public void sendAMPL() {
if (ampl_message_channel == null) {
log.error("AMPL publisher not set, cannot send AMPL file");
log.warn("AMPL publisher not set, cannot send AMPL file");
return;
}
String ampl = AMPLGenerator.generateAMPL(this);
ObjectNode msg = mapper.createObjectNode();
msg.put(getUUID() + ".ampl", ampl);
ampl_message_channel.send(mapper.convertValue(msg, Map.class), getUUID());
Main.logFile(getUUID() + ".ampl", ampl);
}
@@ -383,7 +384,7 @@ public class NebulousApp {
RequirementOperator.GEQ, Long.toString(sal_cores)));
} else {
// floatValue returns 0.0 if node is not numeric
log.error("CPU of component {} is 0 or not a number", c.get("name").asText());
log.warn("CPU of component {} is 0 or not a number", c.get("name").asText());
}
}
if (properties.has("memory")) {;
@@ -393,7 +394,7 @@ public class NebulousApp {
} else if (sal_memory.endsWith("Gi")) {
sal_memory = String.valueOf(Integer.parseInt(sal_memory.substring(0, sal_memory.length() - 2)) * 1024);
} else if (!properties.get("memory").isNumber()) {
log.error("Unsupported memory specification in component {} :{} (wanted 'Mi' or 'Gi') ",
log.warn("Unsupported memory specification in component {} :{} (wanted 'Mi' or 'Gi') ",
properties.get("name").asText(),
properties.get("memory").asText());
sal_memory = null;
@@ -425,7 +426,7 @@ public class NebulousApp {
public void startApplication(JsonNode kubevela) {
log.info("Starting application {} with KubeVela", UUID);
if (salConnector == null) {
log.error("Tried to submit job, but do not have a connection to SAL");
log.warn("Tried to submit job, but do not have a connection to SAL");
return;
}
// The overall flow:
@@ -442,7 +443,7 @@ public class NebulousApp {
// ------------------------------------------------------------
// 1. Create SAL job
log.info("Creating job info");
log.debug("Creating job info");
JobInformation jobinfo = new JobInformation(UUID, name);
// TODO: figure out what ports to specify here
List<Communication> communications = List.of();
@@ -468,7 +469,7 @@ public class NebulousApp {
if (!success) {
// This can happen if the job has already been submitted
log.error("Error trying to create the job; SAL createJob returned {}", success);
log.info("Check if a job with id {} already exists, run stopJobs if yes", UUID);
log.debug("Check if a job with id {} already exists, run stopJobs if yes", UUID);
return;
}
@@ -478,7 +479,7 @@ public class NebulousApp {
// ------------------------------------------------------------
// 3. Create coordinator node
log.info("Creating app coordinator node");
log.debug("Creating app coordinator node");
List<NodeCandidate> controller_candidates = salConnector.findNodeCandidates(controller_requirements);
if (controller_candidates.isEmpty()) {
log.error("Could not find node candidates for controller node; requirements: {}", controller_requirements);
@@ -497,7 +498,7 @@ public class NebulousApp {
// ------------------------------------------------------------
// 4. Submit job
log.info("Starting job");
log.debug("Starting job");
String return_job_id = salConnector.submitJob(UUID);
if (return_job_id.equals("-1")) {
log.error("Failed to add start job {}, SAL returned {}",
@@ -512,7 +513,7 @@ public class NebulousApp {
// ------------------------------------------------------------
// 6. Create worker nodes from requirements
log.info("Starting worker nodes");
log.debug("Starting worker nodes");
for (Map.Entry<String, List<Requirement>> e : requirements.entrySet()) {
List<NodeCandidate> candidates = salConnector.findNodeCandidates(e.getValue());
if (candidates.isEmpty()) {

4
optimiser-controller/src/main/java/eu/nebulouscloud/optimiser/controller/NebulousApps.java
View File

@@ -25,7 +25,7 @@ public class NebulousApps {
public static synchronized void add(NebulousApp app) {
String uuid = app.getUUID();
apps.put(uuid, app);
log.info("Added app {}", uuid);
log.debug("Added app {}", uuid);
}
/**
@@ -47,7 +47,7 @@ public class NebulousApps {
public static synchronized NebulousApp remove(String uuid) {
NebulousApp app = apps.remove(uuid);
if (app != null) {
log.info("Removed app {}", uuid);
log.debug("Removed app {}", uuid);
} else {
log.error("Trying to remove unknown app with uuid {}", uuid);
}

10
optimiser-controller/src/main/java/eu/nebulouscloud/optimiser/controller/SalConnector.java
View File

@@ -108,7 +108,7 @@ public class SalConnector {
*/
private boolean connect(String sal_username, String sal_password) {
URI endpoint_uri = sal_uri.resolve(connectStr);
log.info("Connecting to SAL as a service at uri {}", endpoint_uri);
log.trace("Connecting to SAL as a service at uri {}", endpoint_uri);
try {
this.session_id = HttpClient.create()
@@ -127,7 +127,7 @@ public class SalConnector {
log.error("Error while connecting to SAL", e);
return false;
}
log.info("Connected to SAL, sessionid {}...", session_id.substring(0, 10));
log.debug("Connected to SAL, sessionid {}...", session_id.substring(0, 10));
return true;
}
@@ -172,10 +172,10 @@ public class SalConnector {
} else {
return bytes.asString().mapNotNull(s -> {
try {
log.info("Received message: {}", s);
log.trace("Received message: {}", s);
return objectMapper.readValue(s, NodeCandidate[].class);
} catch (IOException e) {
log.error(e.getMessage(), e);;
log.error(e.getMessage(), e);
return null;
}
});
@@ -303,7 +303,7 @@ public class SalConnector {
} catch (JsonProcessingException e) {
log.error(e.getMessage(), e);;
}
log.info("Sending body json: {}", json);
log.trace("Sending body json: {}", json);
return ByteBufMono.fromString(Mono.just(json));
}