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 bf01f5885a0532935f91ce191d66e8b9d2372196 Author: Daniel Gomm <[email protected]> AuthorDate: Wed Apr 28 21:02:33 2021 +0200 Refactored and reorganized offloading --- .../offloading/OffloadingPolicyManager.java | 35 ++++++++++++++++++ .../offloading/model/OffloadingStrategy.java | 43 ++++++++++++++++++++++ .../{ => model}/policies/Comparator.java | 2 +- .../{ => model}/policies/OffloadingPolicy.java | 2 +- .../ThresholdViolationOffloadingPolicy.java} | 22 ++++++----- .../model/property/CPULoadResourceProperty.java | 11 ++++++ .../property/FreeDiskSpaceResourceProperty.java | 11 ++++++ .../model/property/FreeMemoryResourceProperty.java | 10 +++++ .../model/property/ResourceProperty.java | 7 ++++ .../model/selection/CPULoadSelectionStrategy.java | 15 ++++++++ .../model/selection/RandomSelectionStrategy.java | 18 +++++++++ .../model/selection/SelectionStrategy.java | 7 ++++ .../SingleOccurrenceThresholdViolationPolicy.java | 24 ------------ .../management/resource/ResourceManager.java | 30 +++++++-------- 14 files changed, 184 insertions(+), 53 deletions(-) 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 new file mode 100644 index 0000000..07e896b --- /dev/null +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/OffloadingPolicyManager.java @@ -0,0 +1,35 @@ +package org.apache.streampipes.node.controller.management.offloading; + +import org.apache.streampipes.node.controller.management.offloading.model.OffloadingStrategy; +import org.apache.streampipes.node.controller.management.pe.InvocableElementManager; +import org.apache.streampipes.node.controller.management.resource.model.ResourceMetrics; + +import java.util.ArrayList; +import java.util.List; + +public class OffloadingPolicyManager { + + private final List<OffloadingStrategy<?>> offloadingStrategies = new ArrayList<>(); + private static OffloadingPolicyManager instance; + + public static OffloadingPolicyManager getInstance(){ + if(instance == null){ + instance = new OffloadingPolicyManager(); + } + return instance; + } + + public void checkPolicies(ResourceMetrics rm){ + for(OffloadingStrategy strategy:offloadingStrategies){ + strategy.getOffloadingPolicy().addValue(strategy.getResourceProperty().getProperty(rm)); + if(strategy.getOffloadingPolicy().isViolated()){ + InvocableElementManager.getInstance().postOffloadRequest(strategy.getSelectionStrategy().selectEntity()); + } + } + } + + public void addOffloadingStrategy(OffloadingStrategy<?> offloadingStrategy){ + this.offloadingStrategies.add(offloadingStrategy); + } + +} diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/OffloadingStrategy.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/OffloadingStrategy.java new file mode 100644 index 0000000..433362a --- /dev/null +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/OffloadingStrategy.java @@ -0,0 +1,43 @@ +package org.apache.streampipes.node.controller.management.offloading.model; + +import org.apache.streampipes.node.controller.management.offloading.model.policies.OffloadingPolicy; +import org.apache.streampipes.node.controller.management.offloading.model.property.ResourceProperty; +import org.apache.streampipes.node.controller.management.offloading.model.selection.SelectionStrategy; + +public class OffloadingStrategy<T> { + private SelectionStrategy selectionStrategy; + private OffloadingPolicy<T> offloadingPolicy; + private ResourceProperty<T> resourceProperty; + + public OffloadingStrategy(OffloadingPolicy offloadingPolicy, ResourceProperty resourceProperty, + SelectionStrategy selectionStrategy){ + this.offloadingPolicy = offloadingPolicy; + this.resourceProperty = resourceProperty; + this.selectionStrategy = selectionStrategy; + } + + public SelectionStrategy getSelectionStrategy() { + return selectionStrategy; + } + + public void setSelectionStrategy(SelectionStrategy selectionStrategy) { + this.selectionStrategy = selectionStrategy; + } + + public OffloadingPolicy<?> getOffloadingPolicy() { + return offloadingPolicy; + } + + public void setOffloadingPolicy(OffloadingPolicy<T> offloadingPolicy) { + this.offloadingPolicy = offloadingPolicy; + } + + public ResourceProperty<?> getResourceProperty() { + return resourceProperty; + } + + public void setResourceProperty(ResourceProperty<T> resourceProperty) { + this.resourceProperty = resourceProperty; + } + +} 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/model/policies/Comparator.java similarity index 97% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/policies/Comparator.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/policies/Comparator.java index eb2a898..ec31bfc 100644 --- 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/model/policies/Comparator.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.management.offloading.policies; +package org.apache.streampipes.node.controller.management.offloading.model.policies; public enum Comparator { SMALLER, 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/model/policies/OffloadingPolicy.java similarity index 97% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/policies/OffloadingPolicy.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/policies/OffloadingPolicy.java index 1f0c802..429d4c9 100644 --- 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/model/policies/OffloadingPolicy.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.management.offloading.policies; +package org.apache.streampipes.node.controller.management.offloading.model.policies; public interface OffloadingPolicy<T> { void addValue(T value); diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/policies/MultiOccurrenceThresholdViolationPolicy.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/policies/ThresholdViolationOffloadingPolicy.java similarity index 75% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/policies/MultiOccurrenceThresholdViolationPolicy.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/policies/ThresholdViolationOffloadingPolicy.java index 2e225a4..e91fcda 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/policies/MultiOccurrenceThresholdViolationPolicy.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/policies/ThresholdViolationOffloadingPolicy.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.management.offloading.policies; +package org.apache.streampipes.node.controller.management.offloading.model.policies; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -23,27 +23,31 @@ import org.slf4j.LoggerFactory; import java.util.Queue; import java.util.concurrent.ArrayBlockingQueue; -public class MultiOccurrenceThresholdViolationPolicy<T extends Number> implements OffloadingPolicy<T> { +public class ThresholdViolationOffloadingPolicy<T extends Comparable<T>> implements OffloadingPolicy<T> { private static final Logger LOG = - LoggerFactory.getLogger(MultiOccurrenceThresholdViolationPolicy.class.getCanonicalName()); + LoggerFactory.getLogger(ThresholdViolationOffloadingPolicy.class.getCanonicalName()); private Queue<T> history; private final T threshold; private final Comparator comparator; private final int numberOfThresholdViolations; - public MultiOccurrenceThresholdViolationPolicy(int length, Comparator comparator, T threshold, - int numberOfThresholdViolations){ + public ThresholdViolationOffloadingPolicy(int length, Comparator comparator, T threshold, + int numberOfThresholdViolations){ this.history = new ArrayBlockingQueue<>(length); this.comparator = comparator; this.threshold = threshold; this.numberOfThresholdViolations = numberOfThresholdViolations; } - public MultiOccurrenceThresholdViolationPolicy(int length, Comparator comparator, T threshold){ + public ThresholdViolationOffloadingPolicy(int length, Comparator comparator, T threshold){ this(length, comparator, threshold, length); } + public ThresholdViolationOffloadingPolicy(Comparator comparator, T threshold) { + this(1, comparator, threshold); + } + @Override public void addValue(T value) { if(!this.history.offer(value)) { @@ -58,16 +62,14 @@ public class MultiOccurrenceThresholdViolationPolicy<T extends Number> implement 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()){ + if(value.compareTo(this.threshold) > 0){ numViolations++; } } break; case SMALLER: for(T value : this.history){ - if(value.doubleValue() < this.threshold.doubleValue()){ + if(value.compareTo(this.threshold) < 0){ numViolations++; } } 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 new file mode 100644 index 0000000..169a755 --- /dev/null +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/property/CPULoadResourceProperty.java @@ -0,0 +1,11 @@ +package org.apache.streampipes.node.controller.management.offloading.model.property; + +import org.apache.streampipes.node.controller.management.resource.model.ResourceMetrics; + +public class CPULoadResourceProperty implements ResourceProperty<Float>{ + + @Override + public Float getProperty(ResourceMetrics resourceMetrics) { + return resourceMetrics.getCpuLoadInPercent(); + } +} \ No newline at end of file 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 new file mode 100644 index 0000000..9cf4d00 --- /dev/null +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/property/FreeDiskSpaceResourceProperty.java @@ -0,0 +1,11 @@ +package org.apache.streampipes.node.controller.management.offloading.model.property; + +import org.apache.streampipes.node.controller.management.resource.model.ResourceMetrics; + +public class FreeDiskSpaceResourceProperty implements ResourceProperty<Long>{ + + @Override + public Long getProperty(ResourceMetrics resourceMetrics) { + return resourceMetrics.getFreeDiskSpaceInBytes(); + } +} 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 new file mode 100644 index 0000000..fcabeb7 --- /dev/null +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/property/FreeMemoryResourceProperty.java @@ -0,0 +1,10 @@ +package org.apache.streampipes.node.controller.management.offloading.model.property; + +import org.apache.streampipes.node.controller.management.resource.model.ResourceMetrics; + +public class FreeMemoryResourceProperty implements ResourceProperty<Long> { + @Override + public Long getProperty(ResourceMetrics resourceMetrics) { + return resourceMetrics.getFreeMemoryInBytes(); + } +} 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 new file mode 100644 index 0000000..3378ea3 --- /dev/null +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/property/ResourceProperty.java @@ -0,0 +1,7 @@ +package org.apache.streampipes.node.controller.management.offloading.model.property; + +import org.apache.streampipes.node.controller.management.resource.model.ResourceMetrics; + +public interface ResourceProperty<T> { + T getProperty(ResourceMetrics resourceMetrics); +} diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/selection/CPULoadSelectionStrategy.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/selection/CPULoadSelectionStrategy.java new file mode 100644 index 0000000..efd055a --- /dev/null +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/selection/CPULoadSelectionStrategy.java @@ -0,0 +1,15 @@ +package org.apache.streampipes.node.controller.management.offloading.model.selection; + +import org.apache.streampipes.model.base.InvocableStreamPipesEntity; + +import java.lang.management.ManagementFactory; +import java.lang.management.ThreadMXBean; + +public class CPULoadSelectionStrategy implements SelectionStrategy{ + @Override + public InvocableStreamPipesEntity selectEntity() { + //TODO: Migrate the PE with the highest load on CPU (or possibly other resource) + ThreadMXBean bean = ManagementFactory.getThreadMXBean(); + return null; + } +} diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/selection/RandomSelectionStrategy.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/selection/RandomSelectionStrategy.java new file mode 100644 index 0000000..0dcd551 --- /dev/null +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/selection/RandomSelectionStrategy.java @@ -0,0 +1,18 @@ +package org.apache.streampipes.node.controller.management.offloading.model.selection; + +import org.apache.streampipes.model.base.InvocableStreamPipesEntity; +import org.apache.streampipes.node.controller.management.pe.storage.RunningInvocableInstances; + +import java.util.List; +import java.util.Random; + +public class RandomSelectionStrategy implements SelectionStrategy{ + + @Override + public InvocableStreamPipesEntity selectEntity() { + List<InvocableStreamPipesEntity> instances = RunningInvocableInstances.INSTANCE.getAll(); + if(instances.size() == 0) + return null; + return instances.get(new Random().nextInt(instances.size())); + } +} diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/selection/SelectionStrategy.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/selection/SelectionStrategy.java new file mode 100644 index 0000000..54a1859 --- /dev/null +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/selection/SelectionStrategy.java @@ -0,0 +1,7 @@ +package org.apache.streampipes.node.controller.management.offloading.model.selection; + +import org.apache.streampipes.model.base.InvocableStreamPipesEntity; + +public interface SelectionStrategy { + InvocableStreamPipesEntity selectEntity(); +} 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 deleted file mode 100644 index f7f514b..0000000 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/policies/SingleOccurrenceThresholdViolationPolicy.java +++ /dev/null @@ -1,24 +0,0 @@ -/* - * 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/resource/ResourceManager.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/resource/ResourceManager.java index eab7d60..a57acba 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 @@ -20,10 +20,12 @@ 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.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.offloading.model.OffloadingStrategy; +import org.apache.streampipes.node.controller.management.offloading.OffloadingPolicyManager; +import org.apache.streampipes.node.controller.management.offloading.model.policies.Comparator; +import org.apache.streampipes.node.controller.management.offloading.model.policies.ThresholdViolationOffloadingPolicy; +import org.apache.streampipes.node.controller.management.offloading.model.property.CPULoadResourceProperty; +import org.apache.streampipes.node.controller.management.offloading.model.selection.RandomSelectionStrategy; 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; @@ -48,11 +50,13 @@ public class ResourceManager { private final OperatingSystem os = si.getOperatingSystem(); private final Calendar cal = Calendar.getInstance(); private final ResourceMetrics resourceMetrics = new ResourceMetrics(); - // Offloading policy - private final OffloadingPolicy<Float> multiStepThresholdPolicy = new MultiOccurrenceThresholdViolationPolicy<>(5, - Comparator.GREATER,99f, 5); - private ResourceManager() {} + private ResourceManager() { + //Offloading Policy + OffloadingPolicyManager.getInstance().addOffloadingStrategy(new OffloadingStrategy<Float>(new + ThresholdViolationOffloadingPolicy<>(5, Comparator.GREATER,0.6f, 5), + new CPULoadResourceProperty(), new RandomSelectionStrategy())); + } public static ResourceManager getInstance() { if (instance == null) { @@ -82,15 +86,7 @@ public class ResourceManager { }; 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"); - } - } + OffloadingPolicyManager.getInstance().checkPolicies(resourceMetrics); } private void retrieveResources() {
