This is an automated email from the ASF dual-hosted git repository.
gomm 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 a4d0bad Initial setup for auto migration
a4d0bad is described below
commit a4d0bad2a417022c2e430007273c1bf04c8e5cd6
Author: Daniel Gomm <[email protected]>
AuthorDate: Fri Apr 16 13:14:56 2021 +0200
Initial setup for auto migration
---
.../management/pe/InvocableElementManager.java | 23 +++++++
.../management/rebalance/MigrationManager.java | 26 ++++++++
.../management/rebalance/triggers/Comparator.java | 6 ++
.../rebalance/triggers/ThresholdTriggerPolicy.java | 7 ++
.../rebalance/triggers/TrendTriggerPolicy.java | 77 ++++++++++++++++++++++
.../rebalance/triggers/TriggerPolicy.java | 7 ++
.../management/resource/ResourceManager.java | 16 ++++-
.../migration/MigrationPipelineGenerator.java | 61 +++++++++++++++++
.../manager/node/StreamPipesClusterManager.java | 8 +--
.../org/apache/streampipes/rest/impl/Node.java | 30 ++++++++-
10 files changed, 255 insertions(+), 6 deletions(-)
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 abc4841..76f406b 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,6 +38,7 @@ 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;
@@ -163,6 +164,19 @@ public class InvocableElementManager implements
IPipelineElementLifeCycle {
return response;
}
+ public void postMigrationRequest(InvocableStreamPipesEntity
instanceToMigrate){
+ try {
+ String url = generateBackendMigrationEndpoint();
+ String desc = toJson(instanceToMigrate);
+ org.apache.http.client.fluent.Response resp = Request.Post(url)
+ .bodyString(desc, ContentType.APPLICATION_JSON)
+ .execute();
+ resp.returnContent();
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ }
+
private void adaptPipelineDescription(InvocableStreamPipesEntity graph,
PipelineElementReconfigurationEntity reconfigurationEntity){
List<StaticProperty> staticProperties = new ArrayList<>();
graph.getStaticProperties().forEach(sp -> {
@@ -292,6 +306,15 @@ public class InvocableElementManager implements
IPipelineElementLifeCycle {
+ NodeControllerConfig.INSTANCE.getNodeControllerId();
}
+ private String generateBackendMigrationEndpoint() {
+ return HTTP_PROTOCOL
+ + NodeControllerConfig.INSTANCE.backendLocation()
+ + COLON
+ + NodeControllerConfig.INSTANCE.backendPort()
+ + SLASH
+ +
"streampipes-backend/api/v2/users/[email protected]/nodes/rebalance";
+ }
+
private void registerAtConsul(InvocableRegistration registration) {
try {
String endpoint = consulURL().toString() + SLASH +
CONSUL_REGISTRATION_ROUTE;
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
new file mode 100644
index 0000000..28c98a6
--- /dev/null
+++
b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/rebalance/MigrationManager.java
@@ -0,0 +1,26 @@
+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
new file mode 100644
index 0000000..efceb8f
--- /dev/null
+++
b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/rebalance/triggers/Comparator.java
@@ -0,0 +1,6 @@
+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
new file mode 100644
index 0000000..e1a349c
--- /dev/null
+++
b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/rebalance/triggers/ThresholdTriggerPolicy.java
@@ -0,0 +1,7 @@
+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/TrendTriggerPolicy.java
b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/rebalance/triggers/TrendTriggerPolicy.java
new file mode 100644
index 0000000..b5329eb
--- /dev/null
+++
b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/rebalance/triggers/TrendTriggerPolicy.java
@@ -0,0 +1,77 @@
+package org.apache.streampipes.node.controller.management.rebalance.triggers;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Queue;
+import java.util.concurrent.ArrayBlockingQueue;
+
+public class TrendTriggerPolicy<T extends Number> implements TriggerPolicy<T>{
+
+ private Queue<T> history;
+ private T threshold;
+ private Comparator comparator;
+ private int numberOfThresholdViolations;
+ private static final Logger LOG =
+
LoggerFactory.getLogger(TrendTriggerPolicy.class.getCanonicalName());
+
+ public TrendTriggerPolicy(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){
+ this(length, comparator, threshold, length);
+ }
+
+ @Override
+ public void addValue(T value) {
+ if(!this.history.offer(value)) {
+ this.history.poll();
+ this.history.offer(value);
+ }
+ }
+
+ @Override
+ public boolean isTriggered() {
+ int numViolations = 0;
+ switch (this.comparator){
+ case GREATER:
+ for(T value : this.history){
+ //TODO: Replace comparison with sth more robust than
conversion into double values (either
+ // compare double or long values? Or compare BigDecimal
values instead?)
+ if(value.doubleValue() > this.threshold.doubleValue()){
+ numViolations++;
+ }
+ }
+ break;
+ case SMALLER:
+ for(T value : this.history){
+ if(value.doubleValue() < this.threshold.doubleValue()){
+ numViolations++;
+ }
+ }
+ break;
+ }
+ if(numViolations >= this.numberOfThresholdViolations){
+ LOG.info("Trend above threshold detected.");
+ return true;
+ }
+ return false;
+ }
+
+ @Override
+ public void reset() {
+ this.history = new ArrayBlockingQueue<>(this.history.size());
+ }
+
+ //TODO: Remove -- Only for debugging purposes
+ @Override
+ public String toString() {
+ return "TrendTriggerPolicy{" +
+ "history=" + history +
+ '}';
+ }
+}
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
new file mode 100644
index 0000000..61103ea
--- /dev/null
+++
b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/rebalance/triggers/TriggerPolicy.java
@@ -0,0 +1,7 @@
+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 d274fba..edc2b97 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
@@ -19,6 +19,11 @@ package
org.apache.streampipes.node.controller.management.resource;
import com.fasterxml.jackson.core.JsonProcessingException;
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.resource.model.ResourceMetrics;
import org.apache.streampipes.serializers.json.JacksonSerializer;
import org.slf4j.Logger;
@@ -40,7 +45,7 @@ public class ResourceManager {
private static final Logger LOG =
LoggerFactory.getLogger(ResourceManager.class.getCanonicalName());
- // OSHI to retreive system information
+ // OSHI to retrieve system information
private final SystemInfo si = new SystemInfo();
private final HardwareAbstractionLayer hal = si.getHardware();
private final OperatingSystem os = si.getOperatingSystem();
@@ -68,6 +73,8 @@ 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
@@ -96,6 +103,13 @@ public class ResourceManager {
resourceMetrics.setFreeDiskSpaceInBytes(k.getValue().get("usableDiskSpace"));
}
+ tp.addValue(resourceMetrics.getCpuLoadInPercent());
+ if(tp.isTriggered()){
+ LOG.info("Auto migration triggered.");
+ if(MigrationManager.migrateRandomPE())
+ tp.reset();
+ }
+
} catch (InterruptedException e) {
LOG.error("Thread interrupted. {}", e.toString());
}
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
new file mode 100644
index 0000000..83bff42
--- /dev/null
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/MigrationPipelineGenerator.java
@@ -0,0 +1,61 @@
+package org.apache.streampipes.manager.migration;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import org.apache.streampipes.manager.node.StreamPipesClusterManager;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+import org.apache.streampipes.model.graph.DataProcessorInvocation;
+import org.apache.streampipes.model.node.NodeInfoDescription;
+import org.apache.streampipes.model.pipeline.Pipeline;
+
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import java.util.Random;
+
+public class MigrationPipelineGenerator {
+
+ public static Pipeline
generateMigrationPipeline(InvocableStreamPipesEntity entityToMigrate, Pipeline
correspondingPipeline){
+
+ List<NodeInfoDescription> possibleTargetNodes = new ArrayList<>();
+ List<NodeInfoDescription> nodeInfo =
StreamPipesClusterManager.getAllActiveAndHealthyNodes();
+ nodeInfo.forEach(desc ->{
+ if(desc.getSupportedElements().stream().anyMatch(element ->
element.equals(entityToMigrate.getAppId()))
+ &&
!desc.getNodeControllerId().equals(entityToMigrate.getDeploymentTargetNodeId()))
+ possibleTargetNodes.add(desc);
+ });
+
+ if(possibleTargetNodes.isEmpty())
+ return null;
+
+ //Choose random node; should be adjusted to seek for a proper node to
migrate to (e.g. based on user e.g.
+ // selected labels, locality, free resources,...)
+ NodeInfoDescription targetNode = possibleTargetNodes.get(new
Random().nextInt(possibleTargetNodes.size()));
+
+
entityToMigrate.setDeploymentTargetNodeHostname(targetNode.getHostname());
+
entityToMigrate.setDeploymentTargetNodeId(targetNode.getNodeControllerId());
+ entityToMigrate.setDeploymentTargetNodePort(targetNode.getPort());
+ entityToMigrate.setElementEndpointHostname(targetNode.getHostname());
+ entityToMigrate.setElementEndpointPort(targetNode.getPort());
+
+ Optional<DataProcessorInvocation> originalInvocation =
+ correspondingPipeline.getSepas().stream().filter(dp ->
+
dp.getDeploymentRunningInstanceId().equals(entityToMigrate.getDeploymentRunningInstanceId()))
+ .findFirst();
+ int index =
correspondingPipeline.getSepas().indexOf(originalInvocation.get());
+
+ Pipeline targetPipeline;
+ try {
+ targetPipeline =
PipelineElementMigrationHandler.deepCopyPipeline(correspondingPipeline);
+ } catch (JsonProcessingException e) {
+ e.printStackTrace();
+ return null;
+ }
+
+ targetPipeline.getSepas().remove(index);
+ targetPipeline.getSepas().add(index, (DataProcessorInvocation)
entityToMigrate);
+
+ return targetPipeline;
+ }
+
+}
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 b3c29d3..2e49f8e 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
@@ -38,14 +38,14 @@ import java.util.Optional;
public class StreamPipesClusterManager extends AbstractClusterManager {
private static final Logger LOG =
LoggerFactory.getLogger(StreamPipesClusterManager.class.getCanonicalName());
- public static List<NodeInfoDescription> getAvailableNodes() {
- List<NodeInfoDescription> availableAndHealthyNodes = new ArrayList<>();
+ public static List<NodeInfoDescription> getAllActiveAndHealthyNodes() {
+ List<NodeInfoDescription> activeAndHealthyNodes = new ArrayList<>();
getNodeStorageApi().getAllActiveNodes().forEach(node -> {
if (ClusterHealthCheck.check(node)) {
- availableAndHealthyNodes.add(node);
+ activeAndHealthyNodes.add(node);
}
});
- return availableAndHealthyNodes;
+ return activeAndHealthyNodes;
}
public static List<NodeInfoDescription> getAllNodes() {
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 31f138c..2a434e1 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
@@ -17,10 +17,15 @@
*/
package org.apache.streampipes.rest.impl;
+import org.apache.streampipes.manager.migration.MigrationPipelineGenerator;
import org.apache.streampipes.manager.node.StreamPipesClusterManager;
+import org.apache.streampipes.manager.operations.Operations;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
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.rest.api.INode;
import org.apache.streampipes.rest.shared.annotation.JacksonSerialized;
@@ -103,7 +108,7 @@ public class Node extends AbstractRestResource implements
INode {
@Produces(MediaType.APPLICATION_JSON)
@Override
public Response getAvailableNodes() {
- return ok(StreamPipesClusterManager.getAvailableNodes());
+ return ok(StreamPipesClusterManager.getAllActiveAndHealthyNodes());
}
@GET
@@ -113,4 +118,27 @@ public class Node extends AbstractRestResource implements
INode {
public Response getNodes() {
return ok(StreamPipesClusterManager.getAllNodes());
}
+
+
+ @POST
+ @Path("/rebalance")
+ @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));
+ }
+ }
}
