This is an automated email from the ASF dual-hosted git repository.
wiener pushed a commit to branch edge-extensions
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
The following commit(s) were added to refs/heads/edge-extensions by this push:
new 1754994 some minor refactoring work on auto-offloading
1754994 is described below
commit 1754994e1c9b75f6795d7540301dbd0eea70b0d0
Author: Patrick Wiener <[email protected]>
AuthorDate: Sat Apr 17 01:30:25 2021 +0200
some minor refactoring work on auto-offloading
---
.../model/message/NotificationType.java | 7 +-
.../offloading/AutoOffloadingManager.java | 42 +++++
.../management/offloading/policies/Comparator.java | 23 +++
.../MultiOccurrenceThresholdViolationPolicy.java} | 40 +++--
.../offloading/policies/OffloadingPolicy.java | 24 +++
.../SingleOccurrenceThresholdViolationPolicy.java | 24 +++
.../management/pe/InvocableElementManager.java | 11 +-
.../management/rebalance/MigrationManager.java | 26 ---
.../management/rebalance/triggers/Comparator.java | 6 -
.../rebalance/triggers/ThresholdTriggerPolicy.java | 7 -
.../rebalance/triggers/TriggerPolicy.java | 7 -
.../management/resource/ResourceManager.java | 190 ++++++---------------
.../management/resource/utils/DiskSpace.java | 33 ++++
.../management/resource/utils/FileSystemType.java | 36 ++++
.../management/resource/utils/ResourceUtils.java | 97 +++++++++++
.../migration/MigrationPipelineGenerator.java | 17 ++
.../manager/node/StreamPipesClusterManager.java | 34 ++++
.../org/apache/streampipes/rest/impl/Node.java | 18 +-
.../node-configuration-details.component.ts | 4 -
19 files changed, 420 insertions(+), 226 deletions(-)
diff --git
a/streampipes-model/src/main/java/org/apache/streampipes/model/message/NotificationType.java
b/streampipes-model/src/main/java/org/apache/streampipes/model/message/NotificationType.java
index d7bf388..270d77d 100644
---
a/streampipes-model/src/main/java/org/apache/streampipes/model/message/NotificationType.java
+++
b/streampipes-model/src/main/java/org/apache/streampipes/model/message/NotificationType.java
@@ -71,10 +71,13 @@ public enum NotificationType {
ADMIN_USER_CREATED("Creating admin user...", ""),
NODE_JOIN_SUCCESS("Success", "Node successfully joined"),
- NODE_JOIN_ERROR("Error", "Node could not be joined");
+ NODE_JOIN_ERROR("Error", "Node could not be joined"),
+ OFFLOADING_SUCCESS("Offloading successful", "Pipeline element
successfully offloaded"),
+ OFFLOADING_ERROR("Offloading error", "Could not offload pipeline
element");
- private final String title;
+
+ private final String title;
private final String description;
NotificationType(String title, String description) {
diff --git
a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/AutoOffloadingManager.java
b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/AutoOffloadingManager.java
new file mode 100644
index 0000000..bc30ea2
--- /dev/null
+++
b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/AutoOffloadingManager.java
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.streampipes.node.controller.management.offloading;
+
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+import
org.apache.streampipes.node.controller.management.pe.InvocableElementManager;
+import
org.apache.streampipes.node.controller.management.pe.storage.RunningInvocableInstances;
+
+import java.util.List;
+import java.util.Random;
+
+public class AutoOffloadingManager {
+
+ public static boolean offloadRandom(){
+ List<InvocableStreamPipesEntity> instances =
RunningInvocableInstances.INSTANCE.getAll();
+ if(instances.size() == 0)
+ return false;
+ InvocableStreamPipesEntity randomlySelectedInstance =
instances.get(new Random().nextInt(instances.size()));
+
InvocableElementManager.getInstance().postOffloadRequest(randomlySelectedInstance);
+ return true;
+ }
+
+ public static void offloadHeaviest(){
+ //TODO: Migrate the PE with the highest load on CPU (or possibly other
resource)
+ }
+
+}
diff --git
a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/policies/Comparator.java
b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/policies/Comparator.java
new file mode 100644
index 0000000..eb2a898
--- /dev/null
+++
b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/policies/Comparator.java
@@ -0,0 +1,23 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.streampipes.node.controller.management.offloading.policies;
+
+public enum Comparator {
+ SMALLER,
+ GREATER
+}
diff --git
a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/rebalance/triggers/TrendTriggerPolicy.java
b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/policies/MultiOccurrenceThresholdViolationPolicy.java
similarity index 54%
rename from
streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/rebalance/triggers/TrendTriggerPolicy.java
rename to
streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/policies/MultiOccurrenceThresholdViolationPolicy.java
index b5329eb..2e225a4 100644
---
a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/rebalance/triggers/TrendTriggerPolicy.java
+++
b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/policies/MultiOccurrenceThresholdViolationPolicy.java
@@ -1,4 +1,21 @@
-package org.apache.streampipes.node.controller.management.rebalance.triggers;
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.streampipes.node.controller.management.offloading.policies;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -6,23 +23,24 @@ import org.slf4j.LoggerFactory;
import java.util.Queue;
import java.util.concurrent.ArrayBlockingQueue;
-public class TrendTriggerPolicy<T extends Number> implements TriggerPolicy<T>{
+public class MultiOccurrenceThresholdViolationPolicy<T extends Number>
implements OffloadingPolicy<T> {
+ private static final Logger LOG =
+
LoggerFactory.getLogger(MultiOccurrenceThresholdViolationPolicy.class.getCanonicalName());
private Queue<T> history;
- private T threshold;
- private Comparator comparator;
- private int numberOfThresholdViolations;
- private static final Logger LOG =
-
LoggerFactory.getLogger(TrendTriggerPolicy.class.getCanonicalName());
+ private final T threshold;
+ private final Comparator comparator;
+ private final int numberOfThresholdViolations;
- public TrendTriggerPolicy(int length, Comparator comparator,T threshold,
int numberOfThresholdViolations){
+ public MultiOccurrenceThresholdViolationPolicy(int length, Comparator
comparator, T threshold,
+ int
numberOfThresholdViolations){
this.history = new ArrayBlockingQueue<>(length);
this.comparator = comparator;
this.threshold = threshold;
this.numberOfThresholdViolations = numberOfThresholdViolations;
}
- public TrendTriggerPolicy(int length, Comparator comparator, T threshold){
+ public MultiOccurrenceThresholdViolationPolicy(int length, Comparator
comparator, T threshold){
this(length, comparator, threshold, length);
}
@@ -35,7 +53,7 @@ public class TrendTriggerPolicy<T extends Number> implements
TriggerPolicy<T>{
}
@Override
- public boolean isTriggered() {
+ public boolean isViolated() {
int numViolations = 0;
switch (this.comparator){
case GREATER:
@@ -56,7 +74,7 @@ public class TrendTriggerPolicy<T extends Number> implements
TriggerPolicy<T>{
break;
}
if(numViolations >= this.numberOfThresholdViolations){
- LOG.info("Trend above threshold detected.");
+ LOG.info("Threshold violation detected");
return true;
}
return false;
diff --git
a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/policies/OffloadingPolicy.java
b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/policies/OffloadingPolicy.java
new file mode 100644
index 0000000..1f0c802
--- /dev/null
+++
b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/policies/OffloadingPolicy.java
@@ -0,0 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.streampipes.node.controller.management.offloading.policies;
+
+public interface OffloadingPolicy<T> {
+ void addValue(T value);
+ boolean isViolated();
+ void reset();
+}
diff --git
a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/policies/SingleOccurrenceThresholdViolationPolicy.java
b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/policies/SingleOccurrenceThresholdViolationPolicy.java
new file mode 100644
index 0000000..f7f514b
--- /dev/null
+++
b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/policies/SingleOccurrenceThresholdViolationPolicy.java
@@ -0,0 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.streampipes.node.controller.management.offloading.policies;
+
+public class SingleOccurrenceThresholdViolationPolicy<T extends Number>
extends MultiOccurrenceThresholdViolationPolicy<T> {
+ public SingleOccurrenceThresholdViolationPolicy(Comparator comparator, T
threshold) {
+ super(1, comparator, threshold, 1);
+ }
+}
diff --git
a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/pe/InvocableElementManager.java
b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/pe/InvocableElementManager.java
index 76f406b..11d69c2 100644
---
a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/pe/InvocableElementManager.java
+++
b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/pe/InvocableElementManager.java
@@ -38,7 +38,6 @@ import
org.apache.streampipes.model.staticproperty.StaticProperty;
import org.apache.streampipes.node.controller.config.NodeControllerConfig;
import org.apache.streampipes.node.controller.management.node.NodeManager;
import
org.apache.streampipes.node.controller.management.pe.storage.RunningInvocableInstances;
-import
org.apache.streampipes.node.controller.management.resource.model.ResourceMetrics;
import org.apache.streampipes.serializers.json.JacksonSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -164,10 +163,10 @@ public class InvocableElementManager implements
IPipelineElementLifeCycle {
return response;
}
- public void postMigrationRequest(InvocableStreamPipesEntity
instanceToMigrate){
+ public void postOffloadRequest(InvocableStreamPipesEntity
instanceToOffload){
try {
- String url = generateBackendMigrationEndpoint();
- String desc = toJson(instanceToMigrate);
+ String url = generateBackendOffloadEndpoint();
+ String desc = toJson(instanceToOffload);
org.apache.http.client.fluent.Response resp = Request.Post(url)
.bodyString(desc, ContentType.APPLICATION_JSON)
.execute();
@@ -306,13 +305,13 @@ public class InvocableElementManager implements
IPipelineElementLifeCycle {
+ NodeControllerConfig.INSTANCE.getNodeControllerId();
}
- private String generateBackendMigrationEndpoint() {
+ private String generateBackendOffloadEndpoint() {
return HTTP_PROTOCOL
+ NodeControllerConfig.INSTANCE.backendLocation()
+ COLON
+ NodeControllerConfig.INSTANCE.backendPort()
+ SLASH
- +
"streampipes-backend/api/v2/users/[email protected]/nodes/rebalance";
+ +
"streampipes-backend/api/v2/users/[email protected]/nodes/offload";
}
private void registerAtConsul(InvocableRegistration registration) {
diff --git
a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/rebalance/MigrationManager.java
b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/rebalance/MigrationManager.java
deleted file mode 100644
index 28c98a6..0000000
---
a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/rebalance/MigrationManager.java
+++ /dev/null
@@ -1,26 +0,0 @@
-package org.apache.streampipes.node.controller.management.rebalance;
-
-import org.apache.http.client.fluent.Request;
-import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
-import
org.apache.streampipes.node.controller.management.pe.InvocableElementManager;
-import
org.apache.streampipes.node.controller.management.pe.storage.RunningInvocableInstances;
-
-import java.util.List;
-import java.util.Random;
-
-public class MigrationManager {
-
- public static boolean migrateRandomPE(){
- List<InvocableStreamPipesEntity> instances =
RunningInvocableInstances.INSTANCE.getAll();
- if(instances.size() == 0)
- return false;
- InvocableStreamPipesEntity randomlySelectedInstance =
instances.get(new Random().nextInt(instances.size()));
-
InvocableElementManager.getInstance().postMigrationRequest(randomlySelectedInstance);
- return true;
- }
-
- public static void migrateHeaviestPE(){
- //TODO: Migrate the PE with the highest load on CPU (or possibly other
resource)
- }
-
-}
diff --git
a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/rebalance/triggers/Comparator.java
b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/rebalance/triggers/Comparator.java
deleted file mode 100644
index efceb8f..0000000
---
a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/rebalance/triggers/Comparator.java
+++ /dev/null
@@ -1,6 +0,0 @@
-package org.apache.streampipes.node.controller.management.rebalance.triggers;
-
-public enum Comparator {
- SMALLER,
- GREATER
-}
diff --git
a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/rebalance/triggers/ThresholdTriggerPolicy.java
b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/rebalance/triggers/ThresholdTriggerPolicy.java
deleted file mode 100644
index e1a349c..0000000
---
a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/rebalance/triggers/ThresholdTriggerPolicy.java
+++ /dev/null
@@ -1,7 +0,0 @@
-package org.apache.streampipes.node.controller.management.rebalance.triggers;
-
-public class ThresholdTriggerPolicy<T extends Number> extends
TrendTriggerPolicy<T>{
- public ThresholdTriggerPolicy(Comparator comparator, T threshold) {
- super(1, comparator, threshold, 1);
- }
-}
diff --git
a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/rebalance/triggers/TriggerPolicy.java
b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/rebalance/triggers/TriggerPolicy.java
deleted file mode 100644
index 61103ea..0000000
---
a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/rebalance/triggers/TriggerPolicy.java
+++ /dev/null
@@ -1,7 +0,0 @@
-package org.apache.streampipes.node.controller.management.rebalance.triggers;
-
-public interface TriggerPolicy<T> {
- void addValue(T value);
- boolean isTriggered();
- void reset();
-}
diff --git
a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/resource/ResourceManager.java
b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/resource/ResourceManager.java
index edc2b97..eab7d60 100644
---
a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/resource/ResourceManager.java
+++
b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/resource/ResourceManager.java
@@ -18,42 +18,39 @@
package org.apache.streampipes.node.controller.management.resource;
import com.fasterxml.jackson.core.JsonProcessingException;
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
import org.apache.streampipes.node.controller.config.NodeControllerConfig;
-import
org.apache.streampipes.node.controller.management.pe.InvocableElementManager;
-import
org.apache.streampipes.node.controller.management.rebalance.MigrationManager;
-import
org.apache.streampipes.node.controller.management.rebalance.triggers.Comparator;
-import
org.apache.streampipes.node.controller.management.rebalance.triggers.TrendTriggerPolicy;
-import
org.apache.streampipes.node.controller.management.rebalance.triggers.TriggerPolicy;
+import
org.apache.streampipes.node.controller.management.offloading.AutoOffloadingManager;
+import
org.apache.streampipes.node.controller.management.offloading.policies.Comparator;
+import
org.apache.streampipes.node.controller.management.offloading.policies.MultiOccurrenceThresholdViolationPolicy;
+import
org.apache.streampipes.node.controller.management.offloading.policies.OffloadingPolicy;
import
org.apache.streampipes.node.controller.management.resource.model.ResourceMetrics;
+import
org.apache.streampipes.node.controller.management.resource.utils.DiskSpace;
+import
org.apache.streampipes.node.controller.management.resource.utils.ResourceUtils;
import org.apache.streampipes.serializers.json.JacksonSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import oshi.SystemInfo;
import oshi.hardware.*;
import oshi.software.os.*;
-import oshi.util.FormatUtil;
-import oshi.util.Util;
-
-import oshi.hardware.CentralProcessor;
import java.text.SimpleDateFormat;
-import java.time.Instant;
import java.util.*;
public class ResourceManager {
- private static final Logger LOG =
- LoggerFactory.getLogger(ResourceManager.class.getCanonicalName());
+ private static final Logger LOG =
LoggerFactory.getLogger(ResourceManager.class.getCanonicalName());
+ private static ResourceManager instance = null;
// OSHI to retrieve system information
private final SystemInfo si = new SystemInfo();
private final HardwareAbstractionLayer hal = si.getHardware();
private final OperatingSystem os = si.getOperatingSystem();
private final Calendar cal = Calendar.getInstance();
-
private final ResourceMetrics resourceMetrics = new ResourceMetrics();
-
- private static ResourceManager instance = null;
+ // Offloading policy
+ private final OffloadingPolicy<Float> multiStepThresholdPolicy = new
MultiOccurrenceThresholdViolationPolicy<>(5,
+ Comparator.GREATER,99f, 5);
private ResourceManager() {}
@@ -72,50 +69,52 @@ public class ResourceManager {
}
private final Runnable getCurrentResources = () -> {
-
- TriggerPolicy<Float> tp = new TrendTriggerPolicy<>(5,
Comparator.GREATER, 99f, 5);
-
while(true) {
try {
// get current node resource metrics
Thread.sleep(
NodeControllerConfig.INSTANCE.getNodeResourceUpdateFreqSecs() * 1000);
-
- cal.setTimeInMillis(System.currentTimeMillis());
- SimpleDateFormat dateFormat = new SimpleDateFormat("dd/MM/yyyy
hh:mm:ss a");
-
- /**
- * Monitors the node resources
- */
- Map<String, Map<String, Long>> diskUsage =
getDiskUsage(this.os.getFileSystem());
-
-
resourceMetrics.setSystemTime(dateFormat.format(cal.getTime()));
- resourceMetrics.setBooted(getBooted(this.os));
- resourceMetrics.setUptime(getUptime(this.os));
- resourceMetrics.setCpuLoad(String.format("%.1f%%",
getCpuLoad(this.hal.getProcessor())));
-
resourceMetrics.setCpuLoadInPercent(getCpuLoad(this.hal.getProcessor()));
- resourceMetrics.setCpuTemperature(String.format("%.2f°C",
getCpuTemperature(this.hal.getSensors())));
-
resourceMetrics.setCpuTemperatureCelcius(getCpuTemperature(this.hal.getSensors()));
-
resourceMetrics.setFreeMemoryInBytes(getAvailableMemory(this.hal.getMemory()));
-
resourceMetrics.setUsedMemoryInBytes(this.hal.getMemory().getTotal() -
getAvailableMemory(this.hal.getMemory()));
-
resourceMetrics.setTotalMemoryInBytes(this.hal.getMemory().getTotal());
-
- for (Map.Entry<String, Map<String, Long>> k :
diskUsage.entrySet()) {
-
resourceMetrics.setFreeDiskSpaceInBytes(k.getValue().get("usableDiskSpace"));
- }
-
- tp.addValue(resourceMetrics.getCpuLoadInPercent());
- if(tp.isTriggered()){
- LOG.info("Auto migration triggered.");
- if(MigrationManager.migrateRandomPE())
- tp.reset();
- }
-
+ retrieveResources();
+ checkOffloadingPolicy();
} catch (InterruptedException e) {
LOG.error("Thread interrupted. {}", e.toString());
}
}
};
+ private void checkOffloadingPolicy() {
+
multiStepThresholdPolicy.addValue(resourceMetrics.getCpuLoadInPercent());
+ if(multiStepThresholdPolicy.isViolated()){
+ LOG.info("Policy violated. Offloading procedure triggered");
+ if(AutoOffloadingManager.offloadRandom())
+ multiStepThresholdPolicy.reset();
+ else {
+ LOG.info("Offloading procedure aborted - No pipeline element
seems to be running");
+ }
+ }
+ }
+
+ private void retrieveResources() {
+ cal.setTimeInMillis(System.currentTimeMillis());
+ SimpleDateFormat dateFormat = new SimpleDateFormat("dd/MM/yyyy
hh:mm:ss a");
+
+ // Monitor node resources
+ resourceMetrics.setSystemTime(dateFormat.format(cal.getTime()));
+ resourceMetrics.setBooted(ResourceUtils.getBooted(this.os));
+ resourceMetrics.setUptime(ResourceUtils.getUptime(this.os));
+ resourceMetrics.setCpuLoad(String.format("%.1f%%",
ResourceUtils.getCpuLoad(this.hal.getProcessor())));
+
resourceMetrics.setCpuLoadInPercent(ResourceUtils.getCpuLoad(this.hal.getProcessor()));
+ resourceMetrics.setCpuTemperature(String.format("%.2f°C",
+ ResourceUtils.getCpuTemperature(this.hal.getSensors())));
+
resourceMetrics.setCpuTemperatureCelcius(ResourceUtils.getCpuTemperature(this.hal.getSensors()));
+
resourceMetrics.setFreeMemoryInBytes(ResourceUtils.getAvailableMemory(this.hal.getMemory()));
+ resourceMetrics.setUsedMemoryInBytes(this.hal.getMemory().getTotal() -
ResourceUtils.getAvailableMemory(this.hal.getMemory()));
+ resourceMetrics.setTotalMemoryInBytes(this.hal.getMemory().getTotal());
+
+ ResourceUtils.getDiskUsage(this.os.getFileSystem()).forEach((key,
value) -> {
+
resourceMetrics.setFreeDiskSpaceInBytes(value.get(DiskSpace.USABLE.getName()));
+ });
+ }
+
public String retrieveNodeResources() {
try {
return JacksonSerializer
@@ -123,98 +122,7 @@ public class ResourceManager {
.writerWithDefaultPrettyPrinter()
.writeValueAsString(resourceMetrics);
} catch (JsonProcessingException e) {
- LOG.error("Could not serialize node resources", e);
+ throw new SpRuntimeException("Could not serialize node resources "
+ e.getMessage(), e);
}
- return "{}";
- }
-
- private double getCpuTemperature(Sensors s) {
- return s.getCpuTemperature();
- }
-
- private Map<String, Map<String,Long>> getDiskUsage(FileSystem fs) {
- List<OSFileStore> fsArray = fs.getFileStores();
- Map<String, Map<String, Long>> m = new HashMap<>();
- for(OSFileStore f : fsArray) {
- Map<String, Long> i = new HashMap<>();
- // has SATA disk
- if (f.getVolume().contains("/dev/sda")){
- i.put("usableDiskSpace", f.getUsableSpace());
- i.put("totalDiskSpace", f.getTotalSpace());
- m.put(f.getVolume(), i);
- }
- else if (f.getVolume().contains("/dev/nvme")){
- i.put("usableDiskSpace", f.getUsableSpace());
- i.put("totalDiskSpace", f.getTotalSpace());
- m.put(f.getVolume(), i);
- }
- else if (f.getVolume().contains("/dev/disk")){
- i.put("usableDiskSpace", f.getUsableSpace());
- i.put("totalDiskSpace", f.getTotalSpace());
- m.put(f.getVolume(), i);
- }
- // Docker in RPi
- else if (f.getVolume().contains("/dev/root")){
- i.put("usableDiskSpace", f.getUsableSpace());
- i.put("totalDiskSpace", f.getTotalSpace());
- m.put(f.getVolume(), i);
- }
- // Docker in Jetson Nano
- else if (f.getVolume().contains("/dev/mmcblk0p1")){
- i.put("usableDiskSpace", f.getUsableSpace());
- i.put("totalDiskSpace", f.getTotalSpace());
- m.put(f.getVolume(), i);
- }
-// // has SATA disk
-// if (f.getVolume().contains("/dev/sda") && (
f.getMount().equals("/") || f.getMount().equals("/home"))){
-// i.put("usableDiskSpace", f.getUsableSpace());
-// i.put("totalDiskSpace", f.getTotalSpace());
-// m.put(f.getVolume(), i);
-// }
-// // has overlay disk (container setup)
-// else if (f.getVolume().equals("overlay") && (
f.getMount().equals("/") || f.getMount().equals("/home"))){
-// i.put("usableDiskSpace", f.getUsableSpace());
-// i.put("totalDiskSpace", f.getTotalSpace());
-// m.put(f.getVolume(), i);
-// }
-// // has NVME disk
-// else if(f.getVolume().contains("/dev/nvme") && (
f.getMount().equals("/") || f.getMount().equals("/home"))) {
-// i.put("usableDiskSpace", f.getUsableSpace());
-// i.put("totalDiskSpace", f.getTotalSpace());
-// m.put(f.getVolume(), i);
-// }
-// // disk on macOS
-// else if (f.getVolume().contains("/dev/disk") &&
f.getMount().equals("/")) {
-// i.put("usableDiskSpace", f.getUsableSpace());
-// i.put("totalDiskSpace", f.getTotalSpace());
-// m.put(f.getVolume(), i);
-// }
- }
- if (m.isEmpty()) {
- Map<String, Long> i = new HashMap<>();
- i.put("usableDiskSpace", 0L);
- i.put("totalDiskSpace", 0L);
- m.put("n/a", i);
- }
- return m;
- }
-
- private long getAvailableMemory(GlobalMemory m) {
- return m.getAvailable();
- }
-
- private float getCpuLoad(CentralProcessor c) {
- long[] prevTicks = c.getSystemCpuLoadTicks();
- // need to wait a second
- Util.sleep(1000);
- return (float) (c.getSystemCpuLoadBetweenTicks(prevTicks) * 100);
- }
-
- private String getUptime(OperatingSystem os) {
- return
String.valueOf(FormatUtil.formatElapsedSecs(os.getSystemUptime()));
- }
-
- private String getBooted(OperatingSystem os) {
- return String.valueOf(Instant.ofEpochSecond(os.getSystemBootTime()));
}
}
diff --git
a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/resource/utils/DiskSpace.java
b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/resource/utils/DiskSpace.java
new file mode 100644
index 0000000..326be6e
--- /dev/null
+++
b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/resource/utils/DiskSpace.java
@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.streampipes.node.controller.management.resource.utils;
+
+public enum DiskSpace {
+ USABLE("usableDiskSpace"),
+ TOTAL("totalDiskSpace");
+
+ private final String name;
+
+ DiskSpace(String name) {
+ this.name = name;
+ }
+
+ public String getName() {
+ return name;
+ }
+}
diff --git
a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/resource/utils/FileSystemType.java
b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/resource/utils/FileSystemType.java
new file mode 100644
index 0000000..44984db
--- /dev/null
+++
b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/resource/utils/FileSystemType.java
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.streampipes.node.controller.management.resource.utils;
+
+public enum FileSystemType {
+ SDA("/dev/sda"),
+ NVME("/dev/nvme"),
+ DISK("/dev/disk"),
+ ROOT("/dev/root"),
+ MMCBLK("/dev/mmcblk0p1");
+
+ private final String name;
+
+ FileSystemType(String name) {
+ this.name = name;
+ }
+
+ public String getName() {
+ return name;
+ }
+}
diff --git
a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/resource/utils/ResourceUtils.java
b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/resource/utils/ResourceUtils.java
new file mode 100644
index 0000000..603994f
--- /dev/null
+++
b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/resource/utils/ResourceUtils.java
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.streampipes.node.controller.management.resource.utils;
+
+import oshi.hardware.CentralProcessor;
+import oshi.hardware.GlobalMemory;
+import oshi.hardware.Sensors;
+import oshi.software.os.FileSystem;
+import oshi.software.os.OSFileStore;
+import oshi.software.os.OperatingSystem;
+import oshi.util.FormatUtil;
+import oshi.util.Util;
+
+import java.time.Instant;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class ResourceUtils {
+
+ public static double getCpuTemperature(Sensors s) {
+ return s.getCpuTemperature();
+ }
+
+ public static Map<String, Map<String,Long>> getDiskUsage(FileSystem fs) {
+ List<OSFileStore> fsArray = fs.getFileStores();
+ Map<String, Map<String, Long>> diskUsage = new HashMap<>();
+ for(OSFileStore f : fsArray) {
+ String volume = f.getVolume();
+ // has SATA disk
+ if (volume.contains(FileSystemType.SDA.getName())){
+ addDiskUsage(diskUsage, f);
+ } else if (volume.contains(FileSystemType.NVME.getName())){
+ addDiskUsage(diskUsage, f);
+ } else if (volume.contains(FileSystemType.DISK.getName())){
+ addDiskUsage(diskUsage, f);
+ } else if (volume.contains(FileSystemType.ROOT.getName())){
+ // Docker in RPi
+ addDiskUsage(diskUsage, f);
+ } else if (volume.contains(FileSystemType.MMCBLK.getName())){
+ // Docker in Jetson Nano
+ addDiskUsage(diskUsage, f);
+ }
+ }
+ return diskUsage.isEmpty() ? defaultDiskUsage() : diskUsage;
+ }
+
+ public static void addDiskUsage(Map<String, Map<String, Long>> m,
OSFileStore f) {
+ Map<String, Long> i = new HashMap<>();
+ i.put(DiskSpace.USABLE.getName(), f.getUsableSpace());
+ i.put(DiskSpace.TOTAL.getName(), f.getTotalSpace());
+ m.put(f.getVolume(), i);
+ }
+
+ public static Map<String, Map<String, Long>> defaultDiskUsage() {
+ return new HashMap<String, Map<String, Long>>() {{
+ put("n/a", new HashMap<String, Long>() {{
+ put(DiskSpace.USABLE.getName(), 0L);
+ put(DiskSpace.TOTAL.getName(), 0L);
+ }});
+ }};
+ }
+
+ public static long getAvailableMemory(GlobalMemory m) {
+ return m.getAvailable();
+ }
+
+ public static float getCpuLoad(CentralProcessor c) {
+ long[] prevTicks = c.getSystemCpuLoadTicks();
+ // need to wait a second
+ Util.sleep(1000);
+ return (float) (c.getSystemCpuLoadBetweenTicks(prevTicks) * 100);
+ }
+
+ public static String getUptime(OperatingSystem os) {
+ return
String.valueOf(FormatUtil.formatElapsedSecs(os.getSystemUptime()));
+ }
+
+ public static String getBooted(OperatingSystem os) {
+ return String.valueOf(Instant.ofEpochSecond(os.getSystemBootTime()));
+ }
+}
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/MigrationPipelineGenerator.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/MigrationPipelineGenerator.java
index 83bff42..b8a688e 100644
---
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/MigrationPipelineGenerator.java
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/MigrationPipelineGenerator.java
@@ -1,3 +1,20 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.streampipes.manager.migration;
import com.fasterxml.jackson.core.JsonProcessingException;
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/node/StreamPipesClusterManager.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/node/StreamPipesClusterManager.java
index 2e49f8e..3ab464b 100644
---
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/node/StreamPipesClusterManager.java
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/node/StreamPipesClusterManager.java
@@ -17,16 +17,23 @@
*/
package org.apache.streampipes.manager.node;
+import org.apache.streampipes.manager.migration.MigrationPipelineGenerator;
import
org.apache.streampipes.manager.node.management.cluster.AbstractClusterManager;
import org.apache.streampipes.manager.node.management.cluster.NodeSyncOptions;
import
org.apache.streampipes.manager.node.management.healthcheck.ClusterHealthCheck;
+import org.apache.streampipes.manager.operations.Operations;
+import org.apache.streampipes.model.Response;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
import org.apache.streampipes.model.eventrelay.SpDataStreamRelayContainer;
import org.apache.streampipes.model.message.Message;
import org.apache.streampipes.model.message.NotificationType;
import org.apache.streampipes.model.message.Notifications;
import org.apache.streampipes.model.node.NodeInfoDescription;
+import org.apache.streampipes.model.pipeline.Pipeline;
+import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
import org.apache.streampipes.storage.api.INodeDataStreamRelay;
import org.apache.streampipes.storage.api.INodeInfoStorage;
+import org.apache.streampipes.storage.api.IPipelineStorage;
import org.apache.streampipes.storage.management.StorageDispatcher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -161,6 +168,29 @@ public class StreamPipesClusterManager extends
AbstractClusterManager {
getNodeDataStreamRelayStorageApi().deleteRelayContainer(relayContainer);
}
+ public static Message handleOffloadRequest(InvocableStreamPipesEntity
elementToMigrate) {
+ Pipeline currentPipeline =
getPipelineStorageApi().getPipeline(elementToMigrate.getCorrespondingPipeline());
+ Pipeline offloadPipeline =
MigrationPipelineGenerator.generateMigrationPipeline(elementToMigrate,
+ currentPipeline);
+ //TODO: Handle this case properly
+ if(offloadPipeline == null)
+ return Notifications.error(NotificationType.UNKNOWN_ERROR);
+
+ try {
+ PipelineOperationStatus status =
Operations.handlePipelineElementMigration(offloadPipeline,
+ true, true, true);
+ if (status.isSuccess()) {
+ return
Notifications.success(NotificationType.OFFLOADING_SUCCESS);
+ } else {
+ return
Notifications.success(NotificationType.OFFLOADING_ERROR);
+ }
+
+ } catch (Exception e) {
+ e.printStackTrace();
+ return Notifications.error(NotificationType.UNKNOWN_ERROR);
+ }
+ }
+
// Helpers
private static INodeInfoStorage getNodeStorageApi() {
@@ -171,4 +201,8 @@ public class StreamPipesClusterManager extends
AbstractClusterManager {
return
StorageDispatcher.INSTANCE.getNoSqlStore().getNodeDataStreamRelayStorage();
}
+ private static IPipelineStorage getPipelineStorageApi() {
+ return
StorageDispatcher.INSTANCE.getNoSqlStore().getPipelineStorageAPI();
+ }
+
}
diff --git
a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/Node.java
b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/Node.java
index 2a434e1..5ccd75b 100644
--- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/Node.java
+++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/Node.java
@@ -121,24 +121,10 @@ public class Node extends AbstractRestResource implements
INode {
@POST
- @Path("/rebalance")
+ @Path("/offload")
@Produces(MediaType.APPLICATION_JSON)
@JacksonSerialized
public Response migrateProcessorToOtherNode(InvocableStreamPipesEntity
elementToMigrate) {
- Pipeline correspondingPipeline =
getPipelineStorage().getPipeline(elementToMigrate.getCorrespondingPipeline());
- Pipeline desiredPipeline =
MigrationPipelineGenerator.generateMigrationPipeline(elementToMigrate,
- correspondingPipeline);
- //TODO: Handle this case properly
- if(desiredPipeline == null)
- return
statusMessage(Notifications.error(NotificationType.UNKNOWN_ERROR));
-
- try {
- PipelineOperationStatus status =
Operations.handlePipelineElementMigration(desiredPipeline,
- true, true, true);
- return ok(status);
- } catch (Exception e) {
- e.printStackTrace();
- return
statusMessage(Notifications.error(NotificationType.UNKNOWN_ERROR));
- }
+ return
statusMessage(StreamPipesClusterManager.handleOffloadRequest(elementToMigrate));
}
}
diff --git
a/ui/src/app/configuration/node-configuration/node-configuration-details/node-configuration-details.component.ts
b/ui/src/app/configuration/node-configuration/node-configuration-details/node-configuration-details.component.ts
index bd9d555..86560e4 100644
---
a/ui/src/app/configuration/node-configuration/node-configuration-details/node-configuration-details.component.ts
+++
b/ui/src/app/configuration/node-configuration/node-configuration-details/node-configuration-details.component.ts
@@ -28,10 +28,6 @@ import {PanelType} from
"../../../core-ui/dialog/base-dialog/base-dialog.model";
import {DialogService} from
"../../../core-ui/dialog/base-dialog/base-dialog.service";
import {MatSnackBar} from "@angular/material/snack-bar";
-export interface Fruit {
- name: string;
-}
-
@Component({
selector: 'node-configuration-details',
templateUrl: './node-configuration-details.component.html',
