You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampipes.apache.org by go...@apache.org on 2021/05/10 21:03:17 UTC
[incubator-streampipes] 06/07: resolve remaining merge conflicts
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
commit 5daf569a78b3c8394a26b96cd53c047c801a5aed
Author: Daniel Gomm <da...@outlook.de>
AuthorDate: Thu May 6 16:48:53 2021 +0200
resolve remaining merge conflicts
---
.../backend/StreamPipesBackendApplication.java | 2 --
.../offloading/OffloadingPolicyManager.java | 2 +-
.../model/property/CPULoadResourceProperty.java | 3 ++-
.../property/FreeDiskSpaceResourceProperty.java | 3 ++-
.../model/property/FreeMemoryResourceProperty.java | 3 ++-
.../offloading/model/property/ResourceProperty.java | 3 ++-
.../management/resource/ResourceManager.java | 5 -----
.../monitor/resource/ClusterResourceMonitor.java | 20 ++++++++++++++++++++
.../migration/MigrationPipelineGenerator.java | 7 +++----
.../migration/PipelineElementOffloadHandler.java | 5 ++---
10 files changed, 34 insertions(+), 19 deletions(-)
diff --git a/streampipes-backend/src/main/java/org/apache/streampipes/backend/StreamPipesBackendApplication.java b/streampipes-backend/src/main/java/org/apache/streampipes/backend/StreamPipesBackendApplication.java
index 3317e0d..ad13b82 100644
--- a/streampipes-backend/src/main/java/org/apache/streampipes/backend/StreamPipesBackendApplication.java
+++ b/streampipes-backend/src/main/java/org/apache/streampipes/backend/StreamPipesBackendApplication.java
@@ -20,12 +20,10 @@ package org.apache.streampipes.backend;
import org.apache.shiro.web.env.EnvironmentLoaderListener;
import org.apache.shiro.web.servlet.OncePerRequestFilter;
import org.apache.shiro.web.servlet.ShiroFilter;
-import org.apache.streampipes.manager.node.management.resources.ClusterResourceManager;
import org.apache.streampipes.manager.operations.Operations;
import org.apache.streampipes.model.pipeline.Pipeline;
import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
import org.apache.streampipes.node.management.NodeManagement;
-import org.apache.streampipes.node.management.operation.monitor.health.ClusterHealthCheckMonitor;
import org.apache.streampipes.rest.notifications.NotificationListener;
import org.apache.streampipes.storage.management.StorageDispatcher;
import org.slf4j.Logger;
diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/OffloadingPolicyManager.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/OffloadingPolicyManager.java
index ed0d922..e9a9c1a 100644
--- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/OffloadingPolicyManager.java
+++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/OffloadingPolicyManager.java
@@ -20,9 +20,9 @@ package org.apache.streampipes.node.controller.management.offloading;
import org.apache.streampipes.model.Response;
import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+import org.apache.streampipes.model.node.monitor.ResourceMetrics;
import org.apache.streampipes.node.controller.management.offloading.model.OffloadingStrategy;
import org.apache.streampipes.node.controller.management.pe.InvocableElementManager;
-import org.apache.streampipes.model.resource.ResourceMetrics;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/property/CPULoadResourceProperty.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/property/CPULoadResourceProperty.java
index b695c4f..3ded047 100644
--- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/property/CPULoadResourceProperty.java
+++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/property/CPULoadResourceProperty.java
@@ -18,7 +18,8 @@
package org.apache.streampipes.node.controller.management.offloading.model.property;
-import org.apache.streampipes.model.resource.ResourceMetrics;
+
+import org.apache.streampipes.model.node.monitor.ResourceMetrics;
public class CPULoadResourceProperty implements ResourceProperty<Float>{
diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/property/FreeDiskSpaceResourceProperty.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/property/FreeDiskSpaceResourceProperty.java
index 31e7ab8..7057ac9 100644
--- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/property/FreeDiskSpaceResourceProperty.java
+++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/property/FreeDiskSpaceResourceProperty.java
@@ -18,7 +18,8 @@
package org.apache.streampipes.node.controller.management.offloading.model.property;
-import org.apache.streampipes.model.resource.ResourceMetrics;
+
+import org.apache.streampipes.model.node.monitor.ResourceMetrics;
public class FreeDiskSpaceResourceProperty implements ResourceProperty<Long>{
diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/property/FreeMemoryResourceProperty.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/property/FreeMemoryResourceProperty.java
index e58c4a3..6189d53 100644
--- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/property/FreeMemoryResourceProperty.java
+++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/property/FreeMemoryResourceProperty.java
@@ -18,7 +18,8 @@
package org.apache.streampipes.node.controller.management.offloading.model.property;
-import org.apache.streampipes.model.resource.ResourceMetrics;
+
+import org.apache.streampipes.model.node.monitor.ResourceMetrics;
public class FreeMemoryResourceProperty implements ResourceProperty<Long> {
@Override
diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/property/ResourceProperty.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/property/ResourceProperty.java
index 13abc3f..d461330 100644
--- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/property/ResourceProperty.java
+++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/property/ResourceProperty.java
@@ -18,7 +18,8 @@
package org.apache.streampipes.node.controller.management.offloading.model.property;
-import org.apache.streampipes.model.resource.ResourceMetrics;
+
+import org.apache.streampipes.model.node.monitor.ResourceMetrics;
public interface ResourceProperty<T> {
T getProperty(ResourceMetrics resourceMetrics);
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 76eabfa..fb3b20b 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
@@ -22,11 +22,6 @@ import org.apache.streampipes.commons.exceptions.SpRuntimeException;
import org.apache.streampipes.node.controller.config.NodeConfiguration;
import org.apache.streampipes.node.controller.management.offloading.OffloadingPolicyManager;
import org.apache.streampipes.node.controller.management.offloading.model.OffloadingStrategyFactory;
-import org.apache.streampipes.model.resource.ResourceMetrics;
-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.model.node.monitor.ResourceMetrics;
import org.apache.streampipes.node.controller.management.resource.utils.DiskSpace;
import org.apache.streampipes.node.controller.management.resource.utils.ResourceUtils;
diff --git a/streampipes-node-management/src/main/java/org/apache/streampipes/node/management/operation/monitor/resource/ClusterResourceMonitor.java b/streampipes-node-management/src/main/java/org/apache/streampipes/node/management/operation/monitor/resource/ClusterResourceMonitor.java
index 008c3c4..47b7deb 100644
--- a/streampipes-node-management/src/main/java/org/apache/streampipes/node/management/operation/monitor/resource/ClusterResourceMonitor.java
+++ b/streampipes-node-management/src/main/java/org/apache/streampipes/node/management/operation/monitor/resource/ClusterResourceMonitor.java
@@ -18,12 +18,17 @@
package org.apache.streampipes.node.management.operation.monitor.resource;
+import org.apache.streampipes.model.node.NodeInfoDescription;
import org.apache.streampipes.model.node.monitor.ResourceMetrics;
import org.apache.streampipes.node.management.operation.monitor.health.ClusterHealthCheckMonitor;
import org.apache.streampipes.node.management.utils.HttpUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Queue;
+import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
@@ -34,6 +39,11 @@ public class ClusterResourceMonitor {
private static final String RESOURCE_COLLECTOR_ROUTE = "/api/v2/node/info/resources";
private static final int RESOURCE_COLLECTOR_INTERVAL_SECS = 60;
private static ClusterResourceMonitor instance = null;
+ private static Map<String, Queue<ResourceMetrics>> resourceMetricsMap = new HashMap<>();
+
+ public static Map<String, Queue<ResourceMetrics>> getResourceMetricsMap(){
+ return resourceMetricsMap;
+ }
private ClusterResourceMonitor() {}
@@ -60,6 +70,16 @@ public class ClusterResourceMonitor {
ResourceMetrics resourceMetrics = new NodeResourceCollector(endpoint).execute();
// TODO: implement
+ addResourceMetrics(node, resourceMetrics);
});
};
+
+ private void addResourceMetrics(NodeInfoDescription node, ResourceMetrics rm){
+ if(!resourceMetricsMap.containsKey(node.getNodeControllerId()))
+ resourceMetricsMap.put(node.getNodeControllerId(), new ArrayBlockingQueue<ResourceMetrics>(10));
+ if(!resourceMetricsMap.get(node.getNodeControllerId()).offer(rm)){
+ resourceMetricsMap.get(node.getNodeControllerId()).poll();
+ resourceMetricsMap.get(node.getNodeControllerId()).offer(rm);
+ }
+ }
}
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 f670341..057d8a9 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
@@ -18,15 +18,14 @@
package org.apache.streampipes.manager.migration;
import com.fasterxml.jackson.core.JsonProcessingException;
-import org.apache.streampipes.manager.node.StreamPipesClusterManager;
-import org.apache.streampipes.manager.node.management.resources.ClusterResourceManager;
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.node.monitor.ResourceMetrics;
import org.apache.streampipes.model.pipeline.Pipeline;
import org.apache.streampipes.model.resource.Hardware;
-import org.apache.streampipes.model.resource.ResourceMetrics;
import org.apache.streampipes.node.management.NodeManagement;
+import org.apache.streampipes.node.management.operation.monitor.resource.ClusterResourceMonitor;
import java.util.*;
@@ -96,7 +95,7 @@ public class MigrationPipelineGenerator {
//Currently only checking for free disk space and memory
List<NodeInfoDescription> filteredTargetNodes = new ArrayList<>();
for(NodeInfoDescription nodeInfo : possibleTargetNodes){
- Queue<ResourceMetrics> rmHistory = ClusterResourceManager.getResourceMetricsMap()
+ Queue<ResourceMetrics> rmHistory = ClusterResourceMonitor.getResourceMetricsMap()
.get(nodeInfo.getNodeControllerId());
if(rmHistory == null) return null;
Hardware hardware = entityToMigrate.getResourceRequirements().stream()
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/PipelineElementOffloadHandler.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/PipelineElementOffloadHandler.java
index 01ade43..c099274 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/PipelineElementOffloadHandler.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/PipelineElementOffloadHandler.java
@@ -37,10 +37,9 @@ public class PipelineElementOffloadHandler {
public Message handleOffloading() {
Pipeline currentPipeline = getPipelineById(graph.getCorrespondingPipeline());
- Pipeline offloadPipeline = MigrationPipelineGenerator.generateMigrationPipeline(graph, currentPipeline);
- //TODO: Handle this case properly
+ Pipeline offloadPipeline = new MigrationPipelineGenerator(graph, currentPipeline).generateMigrationPipeline();
if(offloadPipeline == null)
- return Notifications.error(NotificationType.UNKNOWN_ERROR);
+ return Notifications.error(NotificationType.NO_NODE_FOUND);
try {
PipelineOperationStatus status = Operations.handlePipelineElementMigration(offloadPipeline,