You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampipes.apache.org by wi...@apache.org on 2021/05/21 12:15:44 UTC
[incubator-streampipes] branch edge-extensions updated (6d017b7 ->
e76b976)
This is an automated email from the ASF dual-hosted git repository.
wiener pushed a change to branch edge-extensions
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git.
from 6d017b7 Fixed issues with auto offloading
new 95df602 fixed wrong offloading policy for default strategy
new e76b976 add migration, reconfiguration option to StreamPipes Client for performance tests
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.gitignore | 3 +
build-docker.sh => docker-build.sh | 31 ++++++-
docker-save.sh | 100 +++++++++++++++++++++
.../streampipes/client/api/AbstractClientApi.java | 6 ++
.../apache/streampipes/client/api/PipelineApi.java | 18 ++++
.../streampipes/client/http/HttpRequest.java | 6 ++
.../http/{PostRequest.java => PutRequest.java} | 20 ++---
...nse.java => PutRequestWithPayloadResponse.java} | 20 ++---
streampipes-node-controller/pom.xml | 2 +-
.../offloading/OffloadingPolicyManager.java | 10 ++-
.../strategies/OffloadingStrategyFactory.java | 2 +-
streampipes-performance-tests/pom.xml | 6 ++
.../EdgeExtensionsPerformanceTests.java | 96 ++++++++++++++++++++
13 files changed, 290 insertions(+), 30 deletions(-)
rename build-docker.sh => docker-build.sh (62%)
create mode 100755 docker-save.sh
copy streampipes-client/src/main/java/org/apache/streampipes/client/http/{PostRequest.java => PutRequest.java} (76%)
copy streampipes-client/src/main/java/org/apache/streampipes/client/http/{PostRequestWithPayloadResponse.java => PutRequestWithPayloadResponse.java} (67%)
create mode 100644 streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/EdgeExtensionsPerformanceTests.java
[incubator-streampipes] 02/02: add migration,
reconfiguration option to StreamPipes Client for performance tests
Posted by wi...@apache.org.
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
commit e76b976792d0f74931e015f97cbdbab0eab54913
Author: Patrick Wiener <wi...@fzi.de>
AuthorDate: Fri May 21 14:15:23 2021 +0200
add migration, reconfiguration option to StreamPipes Client for performance tests
---
.gitignore | 3 +
docker-save.sh | 100 +++++++++++++++++++++
.../streampipes/client/api/AbstractClientApi.java | 6 ++
.../apache/streampipes/client/api/PipelineApi.java | 18 ++++
.../streampipes/client/http/HttpRequest.java | 6 ++
.../apache/streampipes/client/http/PutRequest.java | 64 +++++++++++++
.../client/http/PutRequestWithPayloadResponse.java | 52 +++++++++++
streampipes-performance-tests/pom.xml | 6 ++
.../EdgeExtensionsPerformanceTests.java | 96 ++++++++++++++++++++
9 files changed, 351 insertions(+)
diff --git a/.gitignore b/.gitignore
index 3fec30b..5f629f2 100644
--- a/.gitignore
+++ b/.gitignore
@@ -16,6 +16,9 @@
# Covers JetBrains IDEs: IntelliJ, RubyMine, PhpStorm, AppCode, PyCharm, CLion, Android Studio and Webstorm
# Reference: https://intellij-support.jetbrains.com/hc/en-us/articles/206544839
+# docker images
+docker-images/
+
# User-specific stuff:
.idea/workspace.xml
.idea/tasks.xml
diff --git a/docker-save.sh b/docker-save.sh
new file mode 100755
index 0000000..45dc594
--- /dev/null
+++ b/docker-save.sh
@@ -0,0 +1,100 @@
+#!/bin/sh
+# 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.
+
+#!/usr/bin/env bash
+
+repo=apachestreampipes
+version=$(mvn help:evaluate -Dexpression=project.version -q -DforceStdout)
+
+dir=docker-images
+docker_bundled_edge_tar=bundled-edge-img.tar
+docker_bundled_edge_arm_tar=bundled-edge-img-armv7.tar
+docker_bundled_edge_aarch64_tar=bundled-edge-img-aarch64.tar
+docker_bundled_core_tar=bundled-core-img.tar
+
+docker_img_edge=(
+$repo/node-controller:$version \
+$repo/extensions-all-jvm:$version \
+eclipse-mosquitto:1.6.12 )
+
+
+docker_img_edge_arm=(
+$repo/node-controller:$version-armv7 \
+$repo/extensions-all-jvm:$version-armv7 \
+eclipse-mosquitto:1.6.12 )
+
+docker_img_edge_aarch64=(
+$repo/node-controller:$version-aarch64 \
+$repo/extensions-all-jvm:$version-aarch64 \
+eclipse-mosquitto:1.6.12 )
+
+docker_img_core=(
+$repo/ui:$version \
+$repo/backend:$version \
+$repo/pipeline-elements-all-jvm:$version \
+fogsyio/activemq:5.15.9 \
+fogsyio/consul:1.7.1 \
+fogsyio/couchdb:2.3.1 \
+fogsyio/kafka:2.2.0 \
+fogsyio/zookeeper:3.4.13 \
+fogsyio/influxdb:1.7 )
+
+docker_save_bundle(){
+ echo "Start saving Docker images to tar ..."
+ create_dir_if_not_exists
+ if [ "$1" == "edge" ]; then
+ if [ -z "$2" ]; then
+ echo "Save edge images (amd) to tar ..."
+ docker save ${docker_img_edge[@]} -o $dir/$docker_bundled_edge_tar
+ elif [ "$2" == "armv7" ]; then
+ echo "Save edge images (armv7) to tar ..."
+ docker save ${docker_img_edge_arm[@]} -o $dir/docker_bundled_edge_arm_tar
+ elif [ "$2" == "aarch64" ]; then
+ echo "Save edge images (aarch64) to tar ..."
+ docker save ${docker_img_edge_aarch64[@]} -o $dir/docker_bundled_edge_aarch64_tar
+ fi
+ elif [ "$1" == "core" ]; then
+ echo "Save core images to tar ..."
+ docker save ${docker_img_core[@]} -o $dir/$docker_bundled_core_tar
+ else
+ echo "Save all images to tar ..."
+ docker save ${docker_img_edge[@]} -o $dir/$docker_bundled_edge_tar
+ docker save ${docker_img_core[@]} -o $dir/$docker_bundled_core_tar
+ fi
+}
+
+create_dir_if_not_exists(){
+ if [[ ! -e $dir ]]; then
+ mkdir $dir
+ elif [[ ! -d $dir ]]; then
+ echo "$dir already exists but is not a directory" 1>&2
+ fi
+}
+
+usage() {
+ cat <<EOF
+Usage: ./bundle-docker.sh core
+ ./bundle-docker.sh edge
+ ./bundle-docker.sh edge arm
+ ./bundle-docker.sh edge aarch64
+EOF
+}
+
+if [ -z "$1" ]; then
+ usage
+else
+ docker_save_bundle $1 $2
+fi
\ No newline at end of file
diff --git a/streampipes-client/src/main/java/org/apache/streampipes/client/api/AbstractClientApi.java b/streampipes-client/src/main/java/org/apache/streampipes/client/api/AbstractClientApi.java
index 563bbca..e9fd627 100644
--- a/streampipes-client/src/main/java/org/apache/streampipes/client/api/AbstractClientApi.java
+++ b/streampipes-client/src/main/java/org/apache/streampipes/client/api/AbstractClientApi.java
@@ -68,6 +68,12 @@ public abstract class AbstractClientApi<T> {
new PostRequestWithoutPayloadResponse<>(clientConfig, apiPath, serializer, object).executeRequest();
}
+ protected <O> O put(StreamPipesApiPath apiPath, T object, Class<O> responseClass) {
+ ObjectSerializer<T, O> serializer = new ObjectSerializer<>();
+ return new PutRequestWithPayloadResponse<>(clientConfig, apiPath, serializer, object, responseClass)
+ .executeRequest();
+ }
+
protected <O> O delete(StreamPipesApiPath apiPath, Class<O> responseClass) {
Serializer<Void, O, O> serializer = new ObjectSerializer<>();
return new DeleteRequest<>(clientConfig, apiPath, responseClass, serializer).executeRequest();
diff --git a/streampipes-client/src/main/java/org/apache/streampipes/client/api/PipelineApi.java b/streampipes-client/src/main/java/org/apache/streampipes/client/api/PipelineApi.java
index 45a0959..f6d5dba 100644
--- a/streampipes-client/src/main/java/org/apache/streampipes/client/api/PipelineApi.java
+++ b/streampipes-client/src/main/java/org/apache/streampipes/client/api/PipelineApi.java
@@ -100,6 +100,24 @@ public class PipelineApi extends AbstractClientApi<Pipeline> implements CRUDApi<
return getSingle(getBaseResourcePath().addToPath(pipelineId).addToPath("stop"), PipelineOperationStatus.class);
}
+ public PipelineOperationStatus migrate(Pipeline pipeline) {
+ return migrate(pipeline.getPipelineId(), pipeline);
+ }
+
+ private PipelineOperationStatus migrate(String pipelineId, Pipeline pipeline) {
+ return post(getBaseResourcePath().addToPath("migrate").addToPath(pipelineId),
+ pipeline, PipelineOperationStatus.class);
+ }
+
+ public PipelineOperationStatus reconfigure(Pipeline pipeline) {
+ return reconfigure(pipeline.getPipelineId(), pipeline);
+ }
+
+ private PipelineOperationStatus reconfigure(String pipelineId, Pipeline pipeline) {
+ return put(getBaseResourcePath().addToPath("reconfigure").addToPath(pipelineId),
+ pipeline, PipelineOperationStatus.class);
+ }
+
@Override
protected StreamPipesApiPath getBaseResourcePath() {
return StreamPipesApiPath.fromUserApiPath(clientConfig.getCredentials())
diff --git a/streampipes-client/src/main/java/org/apache/streampipes/client/http/HttpRequest.java b/streampipes-client/src/main/java/org/apache/streampipes/client/http/HttpRequest.java
index 096d30e..485d751 100644
--- a/streampipes-client/src/main/java/org/apache/streampipes/client/http/HttpRequest.java
+++ b/streampipes-client/src/main/java/org/apache/streampipes/client/http/HttpRequest.java
@@ -62,6 +62,12 @@ public abstract class HttpRequest<SO, DSO, DT> {
return headers.toArray(new Header[0]);
}
+ protected Header[] standardPutHeaders() {
+ List<Header> headers = new ArrayList<>(Arrays.asList(standardHeaders()));
+ headers.add(Headers.contentTypeJson());
+ return headers.toArray(new Header[0]);
+ }
+
protected String makeUrl() {
return makeUrl(true);
}
diff --git a/streampipes-client/src/main/java/org/apache/streampipes/client/http/PutRequest.java b/streampipes-client/src/main/java/org/apache/streampipes/client/http/PutRequest.java
new file mode 100644
index 0000000..2fda50d
--- /dev/null
+++ b/streampipes-client/src/main/java/org/apache/streampipes/client/http/PutRequest.java
@@ -0,0 +1,64 @@
+/*
+ * 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.client.http;
+
+import org.apache.http.client.fluent.Request;
+import org.apache.http.entity.ContentType;
+import org.apache.streampipes.client.model.StreamPipesClientConfig;
+import org.apache.streampipes.client.serializer.Serializer;
+import org.apache.streampipes.client.util.StreamPipesApiPath;
+
+public abstract class PutRequest<SO, DSO, DT> extends HttpRequest<SO, DSO, DT> {
+
+ private SO body;
+ private boolean withBody;
+
+ public PutRequest(StreamPipesClientConfig clientConfig,
+ StreamPipesApiPath apiPath,
+ Serializer<SO, DSO, DT> serializer,
+ SO body) {
+ super(clientConfig, apiPath, serializer);
+ this.body = body;
+ this.withBody = true;
+ }
+
+ public PutRequest(StreamPipesClientConfig clientConfig,
+ StreamPipesApiPath apiPath,
+ Serializer<SO, DSO, DT> serializer) {
+ super(clientConfig, apiPath, serializer);
+ this.withBody = false;
+ }
+
+ @Override
+ protected Request makeRequest(Serializer<SO, DSO, DT> serializer) {
+ Request request = Request
+ .Put(makeUrl())
+ .setHeaders(standardPutHeaders());
+
+ if (withBody) {
+ addBody(request, serializer);
+ }
+
+ return request;
+ }
+
+ protected void addBody(Request request, Serializer<SO, DSO, DT> serializer) {
+ request.bodyString(serializer.serialize(body), ContentType.APPLICATION_JSON);
+ }
+
+}
diff --git a/streampipes-client/src/main/java/org/apache/streampipes/client/http/PutRequestWithPayloadResponse.java b/streampipes-client/src/main/java/org/apache/streampipes/client/http/PutRequestWithPayloadResponse.java
new file mode 100644
index 0000000..8cfef2a
--- /dev/null
+++ b/streampipes-client/src/main/java/org/apache/streampipes/client/http/PutRequestWithPayloadResponse.java
@@ -0,0 +1,52 @@
+/*
+ * 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.client.http;
+
+import org.apache.http.HttpEntity;
+import org.apache.streampipes.client.model.StreamPipesClientConfig;
+import org.apache.streampipes.client.serializer.Serializer;
+import org.apache.streampipes.client.util.StreamPipesApiPath;
+
+import java.io.IOException;
+
+public class PutRequestWithPayloadResponse<SO, DSO, DT> extends PutRequest<SO, DSO, DT> {
+
+ private Class<DSO> responseClass;
+
+ public PutRequestWithPayloadResponse(StreamPipesClientConfig clientConfig,
+ StreamPipesApiPath apiPath,
+ Serializer<SO, DSO, DT> serializer,
+ SO body,
+ Class<DSO> responseClass) {
+ super(clientConfig, apiPath, serializer, body);
+ this.responseClass = responseClass;
+ }
+
+ public PutRequestWithPayloadResponse(StreamPipesClientConfig clientConfig,
+ StreamPipesApiPath apiPath,
+ Serializer<SO, DSO, DT> serializer,
+ Class<DSO> responseClass) {
+ super(clientConfig, apiPath, serializer);
+ this.responseClass = responseClass;
+ }
+
+ @Override
+ protected DT afterRequest(Serializer<SO, DSO, DT> serializer, HttpEntity entity) throws IOException {
+ return serializer.deserialize(entityAsString(entity), responseClass);
+ }
+}
diff --git a/streampipes-performance-tests/pom.xml b/streampipes-performance-tests/pom.xml
index 24d1c88..2f5887f 100644
--- a/streampipes-performance-tests/pom.xml
+++ b/streampipes-performance-tests/pom.xml
@@ -49,6 +49,12 @@
<artifactId>streampipes-dataformat-json</artifactId>
<version>0.68.0-SNAPSHOT</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.streampipes</groupId>
+ <artifactId>streampipes-client</artifactId>
+ <version>0.68.0-SNAPSHOT</version>
+ <scope>compile</scope>
+ </dependency>
<!-- External dependencies -->
</dependencies>
diff --git a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/EdgeExtensionsPerformanceTests.java b/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/EdgeExtensionsPerformanceTests.java
new file mode 100644
index 0000000..c4723ad
--- /dev/null
+++ b/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/EdgeExtensionsPerformanceTests.java
@@ -0,0 +1,96 @@
+/*
+ * 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.performance;
+
+import org.apache.streampipes.client.StreamPipesClient;
+import org.apache.streampipes.client.StreamPipesCredentials;
+import org.apache.streampipes.model.pipeline.Pipeline;
+import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
+import org.apache.streampipes.model.staticproperty.FreeTextStaticProperty;
+
+import java.util.List;
+
+public class EdgeExtensionsPerformanceTests {
+
+ public static void main (String ... args) {
+ StreamPipesCredentials credentials = StreamPipesCredentials
+ .from(System.getenv("user"), System.getenv("apiKey"));
+
+ // Create an instance of the StreamPipes client
+ StreamPipesClient client = StreamPipesClient
+ .create("localhost", 8082, credentials, true);
+
+ boolean shouldBeStopped = true;
+ boolean shouldBeMigrated = false;
+ boolean shouldBeReconfigured = false;
+
+ // Get all pipelines
+ List<Pipeline> pipelines = client.pipelines().all();
+
+ Pipeline pipeline = pipelines.stream()
+ .filter(p -> p.getName().equals("reconfigure"))
+ .findFirst()
+ .orElseThrow(() -> new RuntimeException("Pipeline not found"));
+
+ if (!pipeline.isRunning()) {
+ // Start pipeline
+ PipelineOperationStatus message = client.pipelines().start(pipelines.get(0));
+ if (message.isSuccess()) {
+ System.out.println(message.getTitle());
+ }
+ } else {
+ // Stop pipeline
+
+ if (shouldBeStopped) {
+ PipelineOperationStatus message = client.pipelines().stop(pipelines.get(0));
+ if(message.isSuccess()) {
+ System.out.println("Pipeline successfully stopped");
+ }
+ } else if (shouldBeMigrated) {
+ // Migrate pipeline
+ pipeline.getSepas().forEach(p -> {
+ p.setDeploymentTargetNodeId("edge-02.node-controller");
+ p.setDeploymentTargetNodeHostname("edge02.example.de");
+ p.setDeploymentTargetNodePort(7078);
+ });
+
+ PipelineOperationStatus message = client.pipelines().migrate(pipeline);
+ if (message.isSuccess()) {
+ System.out.println(message.getTitle());
+ }
+ } else if (shouldBeReconfigured) {
+ // Reconfigure pipeline
+ pipeline.getSepas().forEach(p -> p.getStaticProperties().stream()
+ .filter(FreeTextStaticProperty.class::isInstance)
+ .map(FreeTextStaticProperty.class::cast)
+ .filter(FreeTextStaticProperty::isReconfigurable)
+ .forEach(sp -> {
+
+ if (sp.getInternalName().equals("i-am-reconfigurable")) {
+ sp.setValue("999");
+ }
+ }));
+
+ PipelineOperationStatus message = client.pipelines().reconfigure(pipeline);
+ if (message.isSuccess()) {
+ System.out.println(message.getTitle());
+ }
+ }
+ }
+ }
+}
[incubator-streampipes] 01/02: fixed wrong offloading policy for
default strategy
Posted by wi...@apache.org.
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
commit 95df60262865b717a45cd11193777c4896cfccab
Author: Patrick Wiener <wi...@fzi.de>
AuthorDate: Fri May 21 14:14:35 2021 +0200
fixed wrong offloading policy for default strategy
---
build-docker.sh => docker-build.sh | 31 +++++++++++++++++++---
streampipes-node-controller/pom.xml | 2 +-
.../offloading/OffloadingPolicyManager.java | 10 ++++---
.../strategies/OffloadingStrategyFactory.java | 2 +-
4 files changed, 35 insertions(+), 10 deletions(-)
diff --git a/build-docker.sh b/docker-build.sh
similarity index 62%
rename from build-docker.sh
rename to docker-build.sh
index 41ba9fd..6ee84b3 100755
--- a/build-docker.sh
+++ b/docker-build.sh
@@ -19,13 +19,36 @@
repo=apachestreampipes
version=$(mvn help:evaluate -Dexpression=project.version -q -DforceStdout)
-docker_build(){
+docker_build_amd(){
+ echo "Docker build for amd ..."
docker build --no-cache --pull \
-t $repo/$1:$version \
-f $2/Dockerfile $2
}
+docker_build_arm(){
+ echo "Docker build for arm ..."
+ docker buildx build \
+ --platform linux/arm/v7 \
+ -t $repo/$1:$version-armv7 \
+ -f $2/aarch64.Dockerfile $2 --load
+}
+
+docker_build_aarch64(){
+ echo "Docker build for aarch64 ..."
+ docker buildx build \
+ --platform linux/arm64 \
+ -t $repo/$1:$version-aarch64 \
+ -f $2/aarch64.Dockerfile $2 --load
+}
+
+docker_build_multiarch(){
+ docker_build_amd $1 $2
+ docker_build_arm $1 $2
+ docker_build_aarch64 $1 $2
+}
+
echo "Start Docker builds ..."
-docker_build backend streampipes-backend
-docker_build node-controller streampipes-node-controller
-docker_build ui ui
\ No newline at end of file
+docker_build_amd backend streampipes-backend
+docker_build_amd ui ui
+docker_build_multiarch node-controller streampipes-node-controller
\ No newline at end of file
diff --git a/streampipes-node-controller/pom.xml b/streampipes-node-controller/pom.xml
index 53aa908..04c7a92 100644
--- a/streampipes-node-controller/pom.xml
+++ b/streampipes-node-controller/pom.xml
@@ -118,7 +118,7 @@
<dependency>
<groupId>com.github.oshi</groupId>
<artifactId>oshi-core</artifactId>
- <version>5.3.6</version>
+ <version>5.7.3</version>
</dependency>
<dependency>
<groupId>org.eclipse.paho</groupId>
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 372c470..c9c51e1 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
@@ -73,12 +73,14 @@ public class OffloadingPolicyManager {
InvocableStreamPipesEntity offloadEntity = strategy.getSelectionStrategy().select(this.unsuccessfullyTriedEntities);
if(offloadEntity != null){
Response resp = PipelineElementManager.getInstance().offload(offloadEntity);
+
+ String appId = offloadEntity.getAppId();
+ String pipelineName = offloadEntity.getCorrespondingPipeline();
+
if(resp.isSuccess()){
- LOG.info("Successfully offloaded: " + offloadEntity.getAppId()
- + " of pipeline: " + offloadEntity.getCorrespondingPipeline());
+ LOG.info("Successfully offloaded: {} of pipeline: {}", appId, pipelineName);
} else{
- LOG.info("Failed to offload: " + offloadEntity.getAppId()
- + " of pipeline: " + offloadEntity.getCorrespondingPipeline());
+ LOG.warn("Failed to offload: {} of pipeline: {}", appId, pipelineName);
unsuccessfullyTriedEntities.add(offloadEntity);
}
} else LOG.info("No pipeline element found to offload");
diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/strategies/OffloadingStrategyFactory.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/strategies/OffloadingStrategyFactory.java
index 4def3fc..97f7244 100644
--- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/strategies/OffloadingStrategyFactory.java
+++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/strategies/OffloadingStrategyFactory.java
@@ -84,7 +84,7 @@ public class OffloadingStrategyFactory {
return new OffloadingStrategy<>(
new ThresholdViolationOffloadingPolicy<>(5, Comparator.SMALLER,
threshold, 4),
- new FreeMemoryResourceProperty(), new RandomSelectionStrategy());
+ new FreeDiskSpaceResourceProperty(), new RandomSelectionStrategy());
}
}