Commit 1c0d65e2 authored by Rita Sousa's avatar Rita Sousa

Updated Monitors to listen "global" topic and updated fakeWorker Docker images

parent cc8b4eb4
......@@ -96,4 +96,5 @@ app/.idea/
**/.idea/
app/*.txt
app/pidsToMonitor.txt
dataclay
......@@ -31,10 +31,15 @@ In this demo, a fake ElasticSystem is created. To have several fake Workers (Doc
cd app/fakeWorkers/
docker build -t fake_docker_workers .
```
or
```
docker pull rita09sousa/fake_docker_workers:2.0
docker image tag rita09sousa/fake_docker_workers:2.0 fake_docker_workers
```
Then, start some Workers with constant resource usage:
```
docker run -d --rm -it --name fakeworker<NUM_FAKEWORKER> fake_docker_workers fakeworker.py <NUM_FAKEWORKER>
docker run -d --rm -it -e FAKE_WORKER=<NUM_FAKEWORKER> --name fakeworker<NUM_FAKEWORKER> fake_docker_workers
```
where <NUM_FAKEWORKER> must be a number between 1 and 6.
......
FROM python:3.7-alpine
ENV fakeWorker 3
ENV FAKE_WORKER 3
WORKDIR /dockerFakeWorkers/
COPY ./fakeworker.py .
RUN chmod +x ./fakeworker.py
ENTRYPOINT ["./fakeworker.py"]
CMD ["${FAKE_WORKER}"]
CMD ["fakeworker.py ${fakeWorker}"]
ENTRYPOINT ["python3"]
########################## OLD AND HEAVY FAKEWORKERS ##########################
......
#!/usr/bin/python 3
import time
import sys
import threading
......@@ -30,13 +32,14 @@ def thread_function(name):
time.sleep(1)
if len(sys.argv) != 2:
print('Number of arguments wrong. Usage: {} <IdOfFakeworker>'.format(sys.argv[0]))
print("Number of arguments wrong. Usage: {} <IdOfFakeworker>".format(sys.argv[0]))
exit(1)
try:
idFakeWorker = int(sys.argv[1])
if idFakeWorker < 0 or idFakeWorker > 6:
print('Fakeworker ID wrong. It should be number between 1 and 6.')
print("Fakeworker ID wrong. It should be number between 1 and 6.")
exit(1)
except ValueError:
print("Argument is not an int")
......@@ -55,7 +58,7 @@ elif idFakeWorker == 5:
elif idFakeWorker == 6:
defineValues(500000, 2)
print(N, NUM_THREADS)
#print(N, NUM_THREADS)
if NUM_THREADS == 1:
thread_function(1)
......
/*
* Copyright 2020 Instituto Superior de Engenharia do Porto
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package app;
public class ActivationWorkersManager implements Runnable {
private ResourceManager rm;
private ActiveWorkersMap activeWorkers;
private float cpuReactivationThreshold;
private final int PERIOD = 5000;
ActivationWorkersManager(ResourceManager rm, ActiveWorkersMap activeWorkers, float cpuThreshold) {
this.rm = rm;
this.activeWorkers = activeWorkers;
this.cpuReactivationThreshold = cpuThreshold / 2;
}
public void run() {
while (!Thread.interrupted()) {
// This sleep is here temporarily (or not) because of the time of system initialization
try {
Thread.sleep(PERIOD);
} catch (InterruptedException e) {
System.out.println("Thread " + Thread.currentThread().getName() + " was interrupted. Aborting...");
break;
}
if (activeWorkers.size() == 0){
System.out.println("No active Workers. Let's activate another Worker.");
if(activeWorkers.reactivateWorker("NaW")){ // No active Workers
//rm.updateActiveWorkersEverywhere();
} else {
System.out.println("1- There are no healthy workers to reactivate! No workers were activated!");
}
continue;
}
activeWorkers.isHistoryUpdated(); // which also means that all metrics are filled
float actualCPUUsage = activeWorkers.getTotalCPUUsage();
if(actualCPUUsage < cpuReactivationThreshold && !activeWorkers.hasNan()){
System.out.printf("Current CPU usage of Node is %.2f. Let's activate another Worker.\n",actualCPUUsage);
if(activeWorkers.reactivateWorker("time")){
//rm.updateActiveWorkersEverywhere();
} else {
System.out.println("2- There are no healthy workers to reactivate! No workers were activated!");
}
}
}
}
}
This diff is collapsed.
......@@ -38,14 +38,12 @@ import java.net.HttpURLConnection;
public class DataclayWritingManager implements Runnable{
private List<Worker> workerList;
private ActiveWorkersMap activeWorkers;
private List<String> idContainers;
private static final String BASE_GET_URL = "http://localhost/containers/";
private static final String USER_AGENT = "Mozilla/5.0";
DataclayWritingManager(List<Worker> workerList, ActiveWorkersMap activeWorkers) {
DataclayWritingManager(List<Worker> workerList) {
this.workerList = workerList;
this.activeWorkers = activeWorkers;
}
@Override
......@@ -260,7 +258,7 @@ public class DataclayWritingManager implements Runnable{
}
private void insertEnergyMetrics(JSONObject metricsJson){
// FIXME: Check Slack
}
private boolean isJSONValid(String test) {
......
......@@ -21,13 +21,14 @@ import app.mqtt_callback.MqttCallBack;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttMessage;
public class EnergyMonitor implements Runnable {
public class Monitor implements Runnable {
private static final String TOPIC = "nuvlabox-status";
private ResourceManager rm;
EnergyMonitor(ResourceManager rm) {
Monitor(ResourceManager rm) {
this.rm = rm;
}
......@@ -39,12 +40,9 @@ public class EnergyMonitor implements Runnable {
MqttClient subscriber = null;
try{
subscriber = new MqttClient("tcp://datagateway", MqttClient.generateClientId());
subscriber.setCallback( new MqttCallBack("energy",rm) );
subscriber.setCallback( new MqttCallBack(rm) );
subscriber.connect();
// TODO: Get right topic for energy dimension
String[] topics = new String[] { "power-consumption" };
subscriber.subscribe(topics);
subscriber.subscribe(TOPIC);
} catch(MqttException e){
System.out.println("[ERROR] Creation of MQTT Energy Subscriber failed!!!");
e.printStackTrace();
......@@ -54,4 +52,4 @@ public class EnergyMonitor implements Runnable {
System.out.println(threadName + " Probe is connected!");
}
}
}
\ No newline at end of file
......@@ -49,7 +49,7 @@ public class NFRTool {
// FIXME: Threshold values for demo purposes only
private static final float CPU_THRESHOLD = 0.35f; // 0.8f;
private static final float ENERGY_THRESHOLD = 20.0f; // Watts
private static final float ENERGY_THRESHOLD = 0.5f; // Watts
private static float cpuThreshold;
private static float energyThreshold;
......@@ -182,33 +182,18 @@ public class NFRTool {
+" IP address=" + worker.getIp()
+" state=" + (worker.isActive() ? "active" : "inactive")));
ActiveWorkersMap metricsActiveWorkersList = new ActiveWorkersMap(workerList);
ResourceManager resourceManager = new ResourceManager(node);
//Thread rmThread = new Thread(resourceManager, "Resource Manager");
ActivationWorkersManager activationWorkersManager = new ActivationWorkersManager(resourceManager,
metricsActiveWorkersList, cpuThreshold);
Thread awmThread = new Thread(activationWorkersManager, "Activation Workers Manager");
TimeMonitor timeMonitor = new TimeMonitor(resourceManager);
Thread tmThread = new Thread(timeMonitor, "Time Monitor");
EnergyMonitor energyMonitor = new EnergyMonitor(resourceManager);
Thread emThread = new Thread(energyMonitor, "Energy Monitor");
DataclayWritingManager dataClayWritingManager = new DataclayWritingManager(/*mp,*/ workerList,
metricsActiveWorkersList);
DataclayWritingManager dataClayWritingManager = new DataclayWritingManager(workerList);
Thread dwmThread = new Thread(dataClayWritingManager, "Dataclay Writing Manager");
tmThread.start();
emThread.start();
//awmThread.start();
Monitor monitor = new Monitor(resourceManager);
Thread monitorThread = new Thread(monitor, "Time & Energy Monitor");
monitorThread.start();
dwmThread.start();
tmThread.join();
emThread.join();
//awmThread.join();
monitorThread.join();
dwmThread.join();
exitGracefully();
......
......@@ -18,6 +18,11 @@
package app;
import org.json.JSONObject;
import java.util.HashMap;
import java.util.Map;
import org.json.JSONArray;
import org.json.JSONException;
import es.bsc.compss.nfr.model.Node;
......@@ -55,65 +60,100 @@ public class ResourceManager {
node.getEnergyThreshold();
}
public void checkWorkersConsumption(String topic, String message){
JSONObject nodeMetrics = null;
public void checkWorkersConsumption(String message){
JSONObject nodeInfo = null;
try{
nodeMetrics = new JSONObject(message);
nodeInfo = new JSONObject(message);
} catch(JSONException e){
System.out.println("[ERROR] Something wrong with the received MQTT message: \"" + message + "\"");
e.printStackTrace();
return;
}
switch (topic) {
case "cpu":
// To update the Node's cpuThreshold variable
getNodeCpuThreshold();
JSONObject nodeMetrics = null;
if(nodeInfo.has("resources")){
nodeMetrics = nodeInfo.getJSONObject("resources");
String dimension = null;
// TODO: Implement Fork/Join to process JSON's dimension in parallel??
// See: https://docs.oracle.com/javase/tutorial/essential/concurrency/forkjoin.html
if(nodeMetrics.has("load")){
float thresholdFloat = nodeMetrics.getInt("capacity") * cpuThreshold;
if(nodeMetrics.getInt("load") > thresholdFloat){
String violationMessageStr = Utils.getViolationMessage(node, topic);
if(nodeMetrics.has("cpu")){
dimension = "cpu";
getNodeCpuThreshold(); // To update the Node's cpuThreshold variable
JSONObject nodeCpuMetrics = nodeMetrics.getJSONObject("cpu");
// System.out.println(nodeCpuMetrics.toString());
if(nodeCpuMetrics.has("load")){
float thresholdFloat = nodeCpuMetrics.getInt("capacity") * cpuThreshold;
if(nodeCpuMetrics.getInt("load") > thresholdFloat){
String violationMessageStr = Utils.getViolationMessage(node, dimension);
try {
sender.send(violationMessageStr);
} catch (Exception e) {
System.out.println("[ERROR] Resource Manager failed notificating Global Resource Manager");
e.printStackTrace();
e.printStackTrace();
}
} else {
System.out.printf("Resource %s is fine...\n", topic);
System.out.printf("Resource %s is fine...\n", dimension);
}
}
break;
}
case "ram":
if(nodeMetrics.has("ram")){
dimension = "ram";
// TODO: Define Node's ramThreshold variable
if(nodeMetrics.has("used")){
float thresholdFloat = (float) nodeMetrics.getInt("capacity");
if(nodeMetrics.getInt("used") > thresholdFloat){
String violationMessageStr = Utils.getViolationMessage(node, topic);
JSONObject nodeRamMetrics = nodeMetrics.getJSONObject("ram");
// System.out.println(nodeRamMetrics.toString());
if(nodeRamMetrics.has("used")){
float thresholdFloat = (float) nodeRamMetrics.getInt("capacity");
if(nodeRamMetrics.getInt("used") > thresholdFloat){
String violationMessageStr = Utils.getViolationMessage(node, dimension);
try {
sender.send(violationMessageStr);
} catch (Exception e) {
System.out.println("[ERROR] Resource Manager failed notificating Global Resource Manager");
e.printStackTrace();
e.printStackTrace();
}
} else {
System.out.printf("Resource %s is fine...\n", topic);
System.out.printf("Resource %s is fine...\n", dimension);
}
}
break;
// TODO: Get right topic for energy dimension
case "power-consumption":
// To update the Node's energyThreshold variable
getNodeEnergyCapacity();
// TODO: Check metrics
break;
default:
break;
}
if(nodeMetrics.has("power-consumption")){
dimension = "energy";
getNodeEnergyCapacity(); // To update the Node's energyThreshold variable
JSONArray nodeEnergyMetrics = (JSONArray) nodeMetrics.get("power-consumption");
// System.out.println(nodeEnergyMetrics.toString());
float totalPowerConsumed = 0.0f;
for (int i = 0; i < nodeEnergyMetrics.length(); i++) {
JSONObject joPower = nodeEnergyMetrics.getJSONObject(i);
if (((String) joPower.get("metric-name")).contains("_power")) {
totalPowerConsumed += joPower.getFloat("energy-consumption");
}
}
// TODO: Improve Node's energyThreshold variable
float thresholdFloat = energyThreshold;
if(totalPowerConsumed > thresholdFloat){
String violationMessageStr = Utils.getViolationMessage(node, dimension);
try {
sender.send(violationMessageStr);
} catch (Exception e) {
System.out.println("[ERROR] Resource Manager failed notificating Global Resource Manager");
e.printStackTrace();
}
} else {
System.out.printf("Resource %s is fine...\n", dimension);
}
}
} else {
System.out.println("[ERROR] NuvlaBox is NOT returning resources information!!!");
return;
}
}
}
/*
* Copyright 2020 Instituto Superior de Engenharia do Porto
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package app;
import app.mqtt_callback.MqttCallBack;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttMessage;
public class TimeMonitor implements Runnable {
private ResourceManager rm;
TimeMonitor(ResourceManager rm) {
this.rm = rm;
}
public void run() {
String threadName = Thread.currentThread().getName();
// Connect to NuvlaBox DataGateway as a subscriber
MqttClient subscriber = null;
try{
subscriber = new MqttClient("tcp://datagateway", MqttClient.generateClientId());
subscriber.setCallback( new MqttCallBack("time",rm) );
subscriber.connect();
String[] topics = new String[] { "cpu", "ram" };
subscriber.subscribe(topics);
} catch(MqttException e){
System.out.println("[ERROR] Creation of MQTT Time Subscriber failed!!!");
e.printStackTrace();
System.exit(1);
}
System.out.println(threadName + " Probe is connected!");
}
}
package app.mqtt_callback;
package app.mqtt_callback;
import app.ResourceManager;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttException;
public class MqttCallBack implements MqttCallback {
private String dimensionMonitor = "";
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttException;
public class MqttCallBack implements MqttCallback {
private ResourceManager resourceManager;
//private DataclayWritingManager dataclayWritingManager;
public MqttCallBack(String monitor, ResourceManager rm/*, DataclayWritingManager dwm*/) {
dimensionMonitor = monitor;
public MqttCallBack(ResourceManager rm) {
resourceManager = rm;
//dataclayWritingManager = dwm;
}
public void connectionLost(Throwable cause) {
System.out.println("Connection lost on instance \"" + dimensionMonitor + "\"!"
public void connectionLost(Throwable cause) {
System.out.println("Connection lost!"
// + " with cause \"" + cause.getMessage() + "\""
// + " Reason code " + ((MqttException)cause).getReasonCode()
+ "\n Cause -> " + ((MqttException)cause).getCause());
cause.printStackTrace();
// TODO: Reconnect
}
public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
// + " Reason code " + ((MqttException)cause).getReasonCode()
+ "\n Cause -> " + ((MqttException) cause).getCause());
cause.printStackTrace();
// TODO: Reconnect
}
public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
System.out.println("Message from topic " + topic + " received!");
resourceManager.checkWorkersConsumption(topic, new String(mqttMessage.getPayload()));
}
public void deliveryComplete(IMqttDeliveryToken token) {
resourceManager.checkWorkersConsumption(new String(mqttMessage.getPayload()));
}
public void deliveryComplete(IMqttDeliveryToken token) {
try {
System.out.println("Delivery token \"" + token.hashCode()
+ "\" received by instance \"" + dimensionMonitor + "\"");
System.out.println("Delivery token \"" + token.hashCode() + "\"");
} catch (Exception e) {
e.printStackTrace();
}
}
}
}
}
\ No newline at end of file
package app.service;
/*
* Copyright 2020 Instituto Superior de Engenharia do Porto
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package app.service;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
......@@ -16,6 +33,9 @@ public class SendService {
public SendService(String uri, String exchangeName) throws Exception{
factory = new ConnectionFactory();
factory.setUri(uri);
factory.setRequestedHeartbeat(60);
connection = factory.newConnection();
channel = connection.createChannel();
channel.exchangeDeclare(exchangeName,"fanout");
......@@ -23,6 +43,8 @@ public class SendService {
}
public void send(String message) throws Exception{
// TODO: Add something that ensures that in case the GRM goes down, e.g. the NFR sends suggestions
// See: https://www.rabbitmq.com/reliability.html#publisher-side
channel.basicPublish(exchangeName, "", null, message.getBytes("UTF-8"));
System.out.println("Resource Manager sent violation '" + message + "'");
}
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment