You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampipes.apache.org by ze...@apache.org on 2021/08/31 20:51:30 UTC
[incubator-streampipes] branch dev updated: [STREAMPIPES-79] Add
remove datalake index to clean up API
This is an automated email from the ASF dual-hosted git repository.
zehnder pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
The following commit(s) were added to refs/heads/dev by this push:
new db5c26d [STREAMPIPES-79] Add remove datalake index to clean up API
db5c26d is described below
commit db5c26dde8e809e59403b6bf10bac8bcaead2849
Author: Philipp Zehnder <ze...@fzi.de>
AuthorDate: Tue Aug 31 22:50:36 2021 +0200
[STREAMPIPES-79] Add remove datalake index to clean up API
---
streampipes-connect-container-master/pom.xml | 5 ++++
.../container/master/general/ResetManagement.java | 13 ++++++++
.../dataexplorer/DataLakeManagementV4.java | 4 +++
ui/cypress/support/utils/PipelineUtils.ts | 4 +--
ui/cypress/tests/adapter/fileStream.ts | 35 +++++++++++-----------
ui/cypress/tests/adapter/machineDataSimulator.ts | 24 +++++++--------
.../adapter/randomDataSimulatorStream.spec.ts | 23 +++++++-------
ui/cypress/tests/pipeline/pipelineTest.ts | 35 +++++++++++-----------
8 files changed, 84 insertions(+), 59 deletions(-)
diff --git a/streampipes-connect-container-master/pom.xml b/streampipes-connect-container-master/pom.xml
index 4936982..557da95 100644
--- a/streampipes-connect-container-master/pom.xml
+++ b/streampipes-connect-container-master/pom.xml
@@ -49,6 +49,11 @@
<artifactId>streampipes-user-management</artifactId>
<version>0.69.0-SNAPSHOT</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.streampipes</groupId>
+ <artifactId>streampipes-data-explorer</artifactId>
+ <version>0.69.0-SNAPSHOT</version>
+ </dependency>
<!-- External dependencies -->
<dependency>
diff --git a/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/general/ResetManagement.java b/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/general/ResetManagement.java
index 23ae843..fb33d62 100644
--- a/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/general/ResetManagement.java
+++ b/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/general/ResetManagement.java
@@ -20,6 +20,7 @@ package org.apache.streampipes.connect.container.master.general;
import org.apache.streampipes.connect.api.exception.AdapterException;
import org.apache.streampipes.connect.container.master.management.AdapterMasterManagement;
+import org.apache.streampipes.dataexplorer.DataLakeManagementV4;
import org.apache.streampipes.manager.file.FileManager;
import org.apache.streampipes.manager.pipeline.PipelineCacheManager;
import org.apache.streampipes.manager.pipeline.PipelineCanvasMetadataCacheManager;
@@ -27,6 +28,7 @@ import org.apache.streampipes.manager.pipeline.PipelineManager;
import org.apache.streampipes.manager.storage.UserManagementService;
import org.apache.streampipes.model.client.file.FileMetadata;
import org.apache.streampipes.model.connect.adapter.AdapterDescription;
+import org.apache.streampipes.model.datalake.DataLakeMeasure;
import org.apache.streampipes.model.pipeline.Pipeline;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -84,6 +86,17 @@ public class ResetManagement {
FileManager.deleteFile(fileMetadata.getFileId());
});
+ // Remove all data in data lake
+ DataLakeManagementV4 dataLakeManagementV4 = new DataLakeManagementV4();
+ List<DataLakeMeasure> allMeasurements = dataLakeManagementV4.getAllMeasurements();
+ allMeasurements.forEach(measurement -> {
+ boolean isSuccessDataLake = dataLakeManagementV4.removeMeasurement(measurement.getMeasureName());
+
+ if (isSuccessDataLake) {
+ dataLakeManagementV4.removeEventProperty(measurement.getMeasureName());
+ }
+ });
+
logger.info("Resetting the system was completed");
}
}
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java
index c4544a8..62cf11d 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java
@@ -205,6 +205,10 @@ public class DataLakeManagementV4 {
return false;
}
+ public DataResult deleteData(String measurementID) {
+ return this.deleteData(measurementID, null, null);
+ }
+
public DataResult deleteData(String measurementID, Long startDate, Long endDate) {
Map<String, QueryParamsV4> queryParts = DataLakeManagementUtils.getDeleteQueryParams(measurementID, startDate, endDate);
return new DataExplorerQueryV4(queryParts).executeQuery();
diff --git a/ui/cypress/support/utils/PipelineUtils.ts b/ui/cypress/support/utils/PipelineUtils.ts
index 28ed64b..93162c2 100644
--- a/ui/cypress/support/utils/PipelineUtils.ts
+++ b/ui/cypress/support/utils/PipelineUtils.ts
@@ -70,7 +70,7 @@ export class PipelineUtils {
cy.dataCy('sp-element-configuration-save').click();
// Select sink
- cy.dataCy('sp-possible-elements-' + pipelineInput.processingElement.name, { timeout : 10000 }).click();
+ cy.dataCy('sp-possible-elements-' + pipelineInput.processingElement.name, { timeout: 10000 }).click();
cy.dataCy('sp-compatible-elements-' + pipelineInput.dataSink.name).click();
StaticPropertyUtils.input(pipelineInput.dataSink.config);
@@ -94,7 +94,7 @@ export class PipelineUtils {
// Delete pipeline
cy.visit('#/pipelines');
cy.dataCy('delete').should('have.length', 1);
- cy.dataCy('delete').click();
+ cy.dataCy('delete').click({ force: true });
cy.dataCy('sp-pipeline-stop-and-delete').click();
cy.dataCy('delete', { timeout: 10000 }).should('have.length', 0);
}
diff --git a/ui/cypress/tests/adapter/fileStream.ts b/ui/cypress/tests/adapter/fileStream.ts
index 68f7c99..2c84ae3 100644
--- a/ui/cypress/tests/adapter/fileStream.ts
+++ b/ui/cypress/tests/adapter/fileStream.ts
@@ -21,25 +21,26 @@ import { FileManagementUtils } from '../../support/utils/FileManagementUtils';
import { GenericAdapterBuilder } from '../../support/builder/GenericAdapterBuilder';
describe('Test File Stream Adapter', () => {
- before('Setup Test', () => {
- cy.initStreamPipesTest();
- FileManagementUtils.addFile('fileTest/random.csv');
- });
+ before('Setup Test', () => {
+ cy.initStreamPipesTest();
+ FileManagementUtils.addFile('fileTest/random.csv');
+ });
- it('Perform Test', () => {
- const adapterInput = GenericAdapterBuilder
- .create('File_Stream')
- .setName('File Stream Adapter Test')
- .setTimestampProperty('timestamp')
- .addProtocolInput('input', 'speed', '1')
- .addProtocolInput('checkbox', 'replaceTimestamp', 'check')
- .setFormat('csv')
- .addFormatInput('input', 'delimiter', ';')
- .addFormatInput('checkbox', 'header', 'check')
- .build();
+ it('Perform Test', () => {
+ const adapterInput = GenericAdapterBuilder
+ .create('File_Stream')
+ .setName('File Stream Adapter Test')
+ .setTimestampProperty('timestamp')
+ .addProtocolInput('input', 'speed', '1')
+ .addProtocolInput('checkbox', 'replaceTimestamp', 'check')
+ .setFormat('csv')
+ .addFormatInput('input', 'delimiter', ';')
+ .addFormatInput('checkbox', 'header', 'check')
+ .build();
- AdapterUtils.testGenericStreamAdapter(adapterInput);
- });
+ AdapterUtils.testGenericStreamAdapter(adapterInput);
+ AdapterUtils.deleteAdapter();
+ });
});
diff --git a/ui/cypress/tests/adapter/machineDataSimulator.ts b/ui/cypress/tests/adapter/machineDataSimulator.ts
index bd617cd..8fbc1d7 100644
--- a/ui/cypress/tests/adapter/machineDataSimulator.ts
+++ b/ui/cypress/tests/adapter/machineDataSimulator.ts
@@ -18,21 +18,21 @@
import { AdapterUtils } from '../../support/utils/AdapterUtils';
import { SpecificAdapterBuilder } from '../../support/builder/SpecificAdapterBuilder';
-import { FileManagementUtils } from '../../support/utils/FileManagementUtils';
describe('Test Random Data Simulator Stream Adapter', () => {
- before('Setup Test', () => {
- cy.initStreamPipesTest();
- });
+ before('Setup Test', () => {
+ cy.initStreamPipesTest();
+ });
- it('Perform Test', () => {
- const adapterInput = SpecificAdapterBuilder
- .create('Machine_Data_Simulator')
- .setName('Machine Data Simulator Test')
- .addInput('input', 'wait-time-ms', '1000')
- .build();
+ it('Perform Test', () => {
+ const adapterInput = SpecificAdapterBuilder
+ .create('Machine_Data_Simulator')
+ .setName('Machine Data Simulator Test')
+ .addInput('input', 'wait-time-ms', '1000')
+ .build();
- AdapterUtils.testSpecificStreamAdapter(adapterInput);
- });
+ AdapterUtils.testSpecificStreamAdapter(adapterInput);
+ AdapterUtils.deleteAdapter();
+ });
});
diff --git a/ui/cypress/tests/adapter/randomDataSimulatorStream.spec.ts b/ui/cypress/tests/adapter/randomDataSimulatorStream.spec.ts
index ec4f95e..ed30a0c 100644
--- a/ui/cypress/tests/adapter/randomDataSimulatorStream.spec.ts
+++ b/ui/cypress/tests/adapter/randomDataSimulatorStream.spec.ts
@@ -20,17 +20,18 @@ import { AdapterUtils } from '../../support/utils/AdapterUtils';
import { SpecificAdapterBuilder } from '../../support/builder/SpecificAdapterBuilder';
describe('Test Random Data Simulator Stream Adapter', () => {
- before('Setup Test', () => {
- cy.initStreamPipesTest();
- });
+ before('Setup Test', () => {
+ cy.initStreamPipesTest();
+ });
- it('Perform Test', () => {
- const adapterInput = SpecificAdapterBuilder
- .create('Random_Data_Simulator_\\(Stream\\)')
- .setName('Random Data Simulator Adapter Test')
- .addInput('input', 'wait-time-ms', '1000')
- .build();
+ it('Perform Test', () => {
+ const adapterInput = SpecificAdapterBuilder
+ .create('Random_Data_Simulator_\\(Stream\\)')
+ .setName('Random Data Simulator Adapter Test')
+ .addInput('input', 'wait-time-ms', '1000')
+ .build();
- AdapterUtils.testSpecificStreamAdapter(adapterInput);
- });
+ AdapterUtils.testSpecificStreamAdapter(adapterInput);
+ AdapterUtils.deleteAdapter();
+ });
});
diff --git a/ui/cypress/tests/pipeline/pipelineTest.ts b/ui/cypress/tests/pipeline/pipelineTest.ts
index 6ce625e..946c492 100644
--- a/ui/cypress/tests/pipeline/pipelineTest.ts
+++ b/ui/cypress/tests/pipeline/pipelineTest.ts
@@ -25,26 +25,27 @@ const adapterName = 'simulator';
describe('Test Random Data Simulator Stream Adapter', () => {
- before('Setup Test', () => {
- cy.initStreamPipesTest();
- AdapterUtils.addMachineDataSimulator(adapterName);
+ before('Setup Test', () => {
+ cy.initStreamPipesTest();
+ AdapterUtils.addMachineDataSimulator(adapterName);
});
it('Perform Test', () => {
- const pipelineInput = PipelineBuilder.create('Pipeline Test')
- .addSource(adapterName)
- .addProcessingElement(
- PipelineElementBuilder.create('field_renamer')
- .addInput('drop-down', 'convert-property', 'timestamp')
- .addInput('input', 'field-name', 't')
- .build())
- .addSink(
- PipelineElementBuilder.create('dashboard_sink')
- .addInput('input', 'visualization-name', 'Demo')
- .build())
- .build();
+ const pipelineInput = PipelineBuilder.create('Pipeline Test')
+ .addSource(adapterName)
+ .addProcessingElement(
+ PipelineElementBuilder.create('field_renamer')
+ .addInput('drop-down', 'convert-property', 'timestamp')
+ .addInput('input', 'field-name', 't')
+ .build())
+ .addSink(
+ PipelineElementBuilder.create('dashboard_sink')
+ .addInput('input', 'visualization-name', 'Demo')
+ .build())
+ .build();
- PipelineUtils.testPipeline(pipelineInput);
+ PipelineUtils.testPipeline(pipelineInput);
+ PipelineUtils.deletePipeline();
});
- });
+});