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 f3dd443 Fixed connection issues with consul; updated performance tests
f3dd443 is described below
commit f3dd443963a335841770b895990bcc4c5ae68878
Author: daniel-gomm <[email protected]>
AuthorDate: Fri Jul 16 13:26:12 2021 +0200
Fixed connection issues with consul; updated performance tests
---
.../streampipes/container/util/ConsulUtil.java | 6 +++
.../streampipes/performance/TestFactory.java | 4 +-
.../performance/performancetest/GenericTest.java | 54 ++++++++++++++++------
.../pipeline/PipelineMigrationExecutor.java | 23 +++++++++
4 files changed, 71 insertions(+), 16 deletions(-)
diff --git
a/streampipes-container/src/main/java/org/apache/streampipes/container/util/ConsulUtil.java
b/streampipes-container/src/main/java/org/apache/streampipes/container/util/ConsulUtil.java
index e58e0b3..a1b2407 100644
---
a/streampipes-container/src/main/java/org/apache/streampipes/container/util/ConsulUtil.java
+++
b/streampipes-container/src/main/java/org/apache/streampipes/container/util/ConsulUtil.java
@@ -188,6 +188,7 @@ public class ConsulUtil {
peSvcs.put(svcId, new Tuple2<>(svcStatus, svcTag));
}
}
+ consul.destroy();
return peSvcs;
}
@@ -239,6 +240,7 @@ public class ConsulUtil {
endpoints.add(endpoint);
}
}
+ consul.destroy();
return endpoints;
}
@@ -266,6 +268,7 @@ public class ConsulUtil {
keyValues.put(key, v);
}
}
+ consul.destroy();
return keyValues;
}
@@ -312,6 +315,7 @@ public class ConsulUtil {
LOG.info("Updated config - key:" + key + " value: " + entry);
consul.keyValueClient().putValue(key, entry);
}
+ consul.destroy();
}
/**
@@ -323,6 +327,7 @@ public class ConsulUtil {
Consul consul = consulInstance();
LOG.info("Deregister service: " + svcId);
consul.agentClient().deregister(svcId);
+ consul.destroy();
}
/**
@@ -334,6 +339,7 @@ public class ConsulUtil {
Consul consul = consulInstance();
LOG.info("Delete config: {}", key);
consul.keyValueClient().deleteKeys(CONSUL_NAMESPACE + key);
+ consul.destroy();
}
// Helper methods
diff --git
a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/TestFactory.java
b/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/TestFactory.java
index c5f2a8c..0482870 100644
---
a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/TestFactory.java
+++
b/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/TestFactory.java
@@ -34,7 +34,7 @@ public class TestFactory {
case "Latency":
return getLatencyTest();
case "Migration":
- Object[] header_migration = {"timestampInMillis", "event",
"numberOfRuns", "durationInNanos", "durationInSecs"};
+ Object[] header_migration = {"timestampInMillis", "event",
"numberOfRuns", "durationInNanos", "durationInSecs", "originNode",
"targetNode"};
logger.logMQTT("Migration", header_migration);
return getMigrationTest();
case "Reconfiguration":
@@ -49,7 +49,7 @@ public class TestFactory {
public static Test getDeploymentTest(){
return new GenericTest(getPipelineName(), true, false,
- false, 0, 3000);
+ false, 0, 10000);
}
public static Test getLatencyTest(){
diff --git
a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/performancetest/GenericTest.java
b/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/performancetest/GenericTest.java
index ee9c78c..1d61957 100644
---
a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/performancetest/GenericTest.java
+++
b/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/performancetest/GenericTest.java
@@ -20,6 +20,7 @@ package org.apache.streampipes.performance.performancetest;
import org.apache.streampipes.client.StreamPipesClient;
import org.apache.streampipes.client.StreamPipesCredentials;
import org.apache.streampipes.logging.evaluation.EvaluationLogger;
+import org.apache.streampipes.model.Tuple2;
import org.apache.streampipes.model.graph.DataProcessorInvocation;
import org.apache.streampipes.model.pipeline.Pipeline;
import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
@@ -68,18 +69,17 @@ public class GenericTest implements Test{
public void execute(int nrRuns) {
String testType = System.getenv("TEST_TYPE");
-
+ Object[] line = null;
//Start Pipeline
if (!pipeline.isRunning()) {
long beforeStart = System.nanoTime();
PipelineOperationStatus startMessage =
client.pipelines().start(pipeline);
long deploymentDuration = System.nanoTime() - beforeStart;
if(testType.equals("Deployment")){
- Object[] line = {System.currentTimeMillis() ,"Deployment
duration", nrRuns, deploymentDuration, deploymentDuration/1000000000.0};
- evalLogger.logMQTT(testType, line);
+ line = new Object[]{System.currentTimeMillis(), "Deployment
duration", nrRuns, deploymentDuration, deploymentDuration / 1000000000.0, true};
}
+ System.out.println(startMessage.getTitle());
if (startMessage.isSuccess()) {
- System.out.println(startMessage.getTitle());
pipeline.setRunning(true);
}
}
@@ -93,16 +93,18 @@ public class GenericTest implements Test{
//Manipulate Pipeline
//Migration
if (shouldBeMigrated) {
- prepareMigration();
+ Tuple2<String, String> migrationNodes = prepareMigration();
long beforeMigration = System.nanoTime();
PipelineOperationStatus migrationMessage =
client.pipelines().migrate(pipeline);
long migrationDuration = System.nanoTime() - beforeMigration;
if(testType.equals("Migration")){
- Object[] line = {System.currentTimeMillis(), "Migration
duration", nrRuns, migrationDuration};
+ line = new Object[]{System.currentTimeMillis(), "Migration
duration", nrRuns, migrationDuration,
migrationDuration/1000000000.0,migrationNodes.a, migrationNodes.b, true};
evalLogger.logMQTT(testType, line);
}
- if (migrationMessage.isSuccess()) {
- System.out.println(migrationMessage.getTitle());
+ System.out.println(migrationMessage.getTitle());
+ if (!migrationMessage.isSuccess()) {
+ assert line != null;
+ line[line.length -1] = false;
}
}
//Reconfiguration
@@ -116,17 +118,23 @@ public class GenericTest implements Test{
sp.setValue(Float.toString(this.reconfigurableValue++));
}
}));
- Object[] line = {System.currentTimeMillis(), "Reconfiguration
triggered", nrRuns, (this.reconfigurableValue-1)};
- evalLogger.logMQTT(testType, line);
+ line = new Object[]{System.currentTimeMillis(), "Reconfiguration
triggered", nrRuns, (this.reconfigurableValue - 1), true};
System.out.println("Reconfiguration triggered with value " +
(this.reconfigurableValue-1));
PipelineOperationStatus message =
client.pipelines().reconfigure(pipeline);
- if (message.isSuccess()) {
- System.out.println(message.getTitle());
+ System.out.println(message.getTitle());
+ if (!message.isSuccess()) {
+ assert line != null;
+ line[line.length -1] = false;
}
}
try {
- Thread.sleep(timeToSleepAfterManipulation);
+ if(shouldBeMigrated){
+ long sleepTime = timeToSleepAfterManipulation +
(long)(Math.random()*10000);
+ Thread.sleep(sleepTime);
+ }else {
+ Thread.sleep(timeToSleepAfterManipulation);
+ }
} catch (InterruptedException e) {
e.printStackTrace();
}
@@ -137,25 +145,43 @@ public class GenericTest implements Test{
System.out.println("Pipeline successfully stopped");
pipeline.setRunning(false);
}
+ else{
+ System.out.println("Pipeline could not be stopped.");
+ assert line != null;
+ line[line.length -1] = false;
+ }
+ }
+ evalLogger.logMQTT(testType, line);
+ try {
+ Thread.sleep(5000);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
}
}
- private void prepareMigration(){
+ private Tuple2<String, String> prepareMigration(){
+ String nodeFrom = "";
+ String nodeTo = "";
String appId = System.getenv("MIGRATION_ENTITY_APP_ID");
String[] node1 = System.getenv("MIGRATION_NODE_1").split(";");
String[] node2 = System.getenv("MIGRATION_NODE_2").split(";");
Optional<DataProcessorInvocation> processor =
pipeline.getSepas().stream().filter(p ->
p.getAppId().equals(appId)).findFirst();
if (processor.isPresent()){
if (processor.get().getDeploymentTargetNodeId().equals(node1[0])){
+ nodeFrom = node1[0];
+ nodeTo = node2[0];
processor.get().setDeploymentTargetNodeId(node2[0]);
processor.get().setDeploymentTargetNodeHostname(node2[1]);
processor.get().setDeploymentTargetNodePort(Integer.parseInt(node2[2]));
} else{
+ nodeFrom = node2[0];
+ nodeTo = node1[0];
processor.get().setDeploymentTargetNodeId(node1[0]);
processor.get().setDeploymentTargetNodeHostname(node1[1]);
processor.get().setDeploymentTargetNodePort(Integer.parseInt(node1[2]));
}
}
+ return new Tuple2<>(nodeFrom, nodeTo);
}
}
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/PipelineMigrationExecutor.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/PipelineMigrationExecutor.java
index f5ead26..8b0631d 100644
---
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/PipelineMigrationExecutor.java
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/PipelineMigrationExecutor.java
@@ -17,6 +17,7 @@
*/
package org.apache.streampipes.manager.execution.pipeline;
+import org.apache.streampipes.logging.evaluation.EvaluationLogger;
import org.apache.streampipes.manager.data.PipelineGraph;
import org.apache.streampipes.manager.data.PipelineGraphBuilder;
import org.apache.streampipes.manager.data.PipelineGraphHelpers;
@@ -91,34 +92,56 @@ public class PipelineMigrationExecutor extends
AbstractPipelineExecutor {
// 5. stop origin relay
prepareMigration();
+ //TODO: Remove logging after evaluation
+ EvaluationLogger logger = EvaluationLogger.getInstance();
// Start target pipeline elements and relays on new target node
+ long before_start_target = System.nanoTime();
PipelineOperationStatus statusStartTarget =
startTargetPipelineElementsAndRelays(status);
if(!statusStartTarget.isSuccess()){
//Target PE could not be started; nothing to roll back
return status;
}
+ long start_target_duration = System.nanoTime() - before_start_target;
+ Object[] line_start_target = {System.currentTimeMillis(), "start
target element","",start_target_duration,start_target_duration/1000000000.0};
+ logger.logMQTT("Migration", line_start_target);
// Stop relays from origin predecessor
+ long downtime_beginning = System.nanoTime();
PipelineOperationStatus statusStopRelays =
stopRelaysFromPredecessorsBeforeMigration(status);
if(!statusStopRelays.isSuccess()){
rollbackToPreMigrationStepOne(statusStopRelays, status);
return status;
}
+ long stop_relays_origin_duration = System.nanoTime() -
downtime_beginning;
+ Object[] line_stop_relay = {System.currentTimeMillis(), "stop relay
from
origin","",stop_relays_origin_duration,stop_relays_origin_duration/1000000000.0};
+ logger.logMQTT("Migration", line_stop_relay);
// Start relays to target after migration
+ long before_start_relay_target = System.nanoTime();
PipelineOperationStatus statusStartRelays =
startRelaysFromPredecessorsAfterMigration(status);
if(!statusStartRelays.isSuccess()){
rollbackToPreMigrationStepTwo(statusStartRelays, status);
return status;
}
+ long start_relay_target_duration = System.nanoTime() -
before_start_relay_target;
+ Object[] line_start_relay = {System.currentTimeMillis(), "start relay
to
target","",start_relay_target_duration,start_relay_target_duration/1000000000.0};
+ logger.logMQTT("Migration", line_start_relay);
+
+ long downtime = System.nanoTime() - downtime_beginning;
+ Object[] line_downtime = {System.currentTimeMillis() ,"downtime", "",
downtime, downtime/1000000000.0};
+ logger.logMQTT("Migration", line_downtime);
//Stop origin and associated relay
+ long before_stop_origin = System.nanoTime();
PipelineOperationStatus statusStopOrigin =
stopOriginPipelineElementsAndRelays(status);
if(!statusStopOrigin.isSuccess()){
rollbackToPreMigrationStepThree(status);
return status;
}
+ long stop_origin_duration = System.nanoTime() - before_stop_origin;
+ Object[] line_stop_origin = {System.currentTimeMillis(), "stop origin
element","",stop_origin_duration,stop_origin_duration/1000000000.0};
+ logger.logMQTT("Migration", line_stop_origin);
List<InvocableStreamPipesEntity> graphs = new ArrayList<>();
graphs.addAll(pipeline.getActions());
