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 <[email protected]> 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,
