This is an automated email from the ASF dual-hosted git repository. bossenti pushed a commit to branch abstract-data-explorer-from-influx in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit 2e34625cde754467f4f9a1174b4f54da7068f003 Author: bossenti <[email protected]> AuthorDate: Tue Apr 30 11:26:48 2024 +0200 refactor: adapt usages to new APIs --- .../influx/{migrate => }/DeleteDataQuery.java | 2 +- .../sinks/internal/jvm/datalake/DataLakeSink.java | 53 +++++++++++---------- .../apache/streampipes/ps/DataLakeResourceV4.java | 54 +++++++++------------- .../apache/streampipes/rest/ResetManagement.java | 11 ++--- 4 files changed, 57 insertions(+), 63 deletions(-) diff --git a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/DeleteDataQuery.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DeleteDataQuery.java similarity index 96% rename from streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/DeleteDataQuery.java rename to streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DeleteDataQuery.java index 4885cd1ed9..63eb528d25 100644 --- a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/DeleteDataQuery.java +++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DeleteDataQuery.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.dataexplorer.influx.migrate; +package org.apache.streampipes.dataexplorer.influx; import org.apache.streampipes.commons.environment.Environment; import org.apache.streampipes.commons.environment.Environments; diff --git a/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/DataLakeSink.java b/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/DataLakeSink.java index 2a8250cb82..4a2930b7d5 100644 --- a/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/DataLakeSink.java +++ b/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/DataLakeSink.java @@ -20,7 +20,8 @@ package org.apache.streampipes.sinks.internal.jvm.datalake; import org.apache.streampipes.commons.environment.Environments; import org.apache.streampipes.commons.exceptions.SpRuntimeException; -import org.apache.streampipes.dataexplorer.influx.migrate.TimeSeriesStore; +import org.apache.streampipes.dataexplorer.TimeSeriesStore; +import org.apache.streampipes.dataexplorer.management.DataExplorerDispatcher; import org.apache.streampipes.extensions.api.pe.context.EventSinkRuntimeContext; import org.apache.streampipes.model.DataSinkType; import org.apache.streampipes.model.datalake.DataLakeMeasure; @@ -54,25 +55,25 @@ public class DataLakeSink extends StreamPipesDataSink { @Override public DataSinkDescription declareModel() { return DataSinkBuilder - .create("org.apache.streampipes.sinks.internal.jvm.datalake", 1) - .withLocales(Locales.EN) - .withAssets(Assets.DOCUMENTATION, Assets.ICON) - .category(DataSinkType.INTERNAL) - .requiredStream(StreamRequirementsBuilder - .create() - .requiredPropertyWithUnaryMapping( - EpRequirements.timestampReq(), - Labels.withId(TIMESTAMP_MAPPING_KEY), - PropertyScope.NONE - ) - .build()) - .requiredTextParameter(Labels.withId(DATABASE_MEASUREMENT_KEY)) - .requiredSingleValueSelection( - Labels.withId(SCHEMA_UPDATE_KEY), - Options.from(SCHEMA_UPDATE_OPTION, EXTEND_EXISTING_SCHEMA_OPTION) - ) - - .build(); + .create("org.apache.streampipes.sinks.internal.jvm.datalake", 1) + .withLocales(Locales.EN) + .withAssets(Assets.DOCUMENTATION, Assets.ICON) + .category(DataSinkType.INTERNAL) + .requiredStream(StreamRequirementsBuilder + .create() + .requiredPropertyWithUnaryMapping( + EpRequirements.timestampReq(), + Labels.withId(TIMESTAMP_MAPPING_KEY), + PropertyScope.NONE + ) + .build()) + .requiredTextParameter(Labels.withId(DATABASE_MEASUREMENT_KEY)) + .requiredSingleValueSelection( + Labels.withId(SCHEMA_UPDATE_KEY), + Options.from(SCHEMA_UPDATE_OPTION, EXTEND_EXISTING_SCHEMA_OPTION) + ) + + .build(); } @Override @@ -95,11 +96,15 @@ public class DataLakeSink extends StreamPipesDataSink { measure.setSchemaUpdateStrategy(DataLakeMeasureSchemaUpdateStrategy.UPDATE_SCHEMA); } + measure = DataExplorerDispatcher.INSTANCE.getDataExplorerManager() + .getMeasurementSanitizer(runtimeContext.getStreamPipesClient(), measure) + .sanitizeAndRegister(); + this.timeSeriesStore = new TimeSeriesStore( - Environments.getEnvironment(), - runtimeContext.getStreamPipesClient(), - measure, - true + DataExplorerDispatcher.INSTANCE.getDataExplorerManager().getTimeseriesStorage(measure), + measure, + Environments.getEnvironment(), + true ); } diff --git a/streampipes-platform-services/src/main/java/org/apache/streampipes/ps/DataLakeResourceV4.java b/streampipes-platform-services/src/main/java/org/apache/streampipes/ps/DataLakeResourceV4.java index 4716a9ec06..34a195fd73 100644 --- a/streampipes-platform-services/src/main/java/org/apache/streampipes/ps/DataLakeResourceV4.java +++ b/streampipes-platform-services/src/main/java/org/apache/streampipes/ps/DataLakeResourceV4.java @@ -18,9 +18,10 @@ package org.apache.streampipes.ps; -import org.apache.streampipes.dataexplorer.DataExplorerSchemaManagement; +import org.apache.streampipes.dataexplorer.api.IDataExplorerQueryManagement; +import org.apache.streampipes.dataexplorer.api.IDataExplorerSchemaManagement; import org.apache.streampipes.dataexplorer.export.OutputFormat; -import org.apache.streampipes.dataexplorer.influx.migrate.DataExplorerQueryManagement; +import org.apache.streampipes.dataexplorer.management.DataExplorerDispatcher; import org.apache.streampipes.model.datalake.DataLakeMeasure; import org.apache.streampipes.model.datalake.DataSeries; import org.apache.streampipes.model.datalake.SpQueryResult; @@ -29,7 +30,6 @@ import org.apache.streampipes.model.message.Notifications; import org.apache.streampipes.model.monitoring.SpLogMessage; import org.apache.streampipes.rest.core.base.impl.AbstractRestResource; import org.apache.streampipes.rest.shared.exception.SpMessageException; -import org.apache.streampipes.storage.management.StorageDispatcher; import io.swagger.v3.oas.annotations.Operation; import io.swagger.v3.oas.annotations.Parameter; @@ -80,23 +80,22 @@ import static org.apache.streampipes.model.datalake.param.SupportedRestQueryPara @RequestMapping("/api/v4/datalake") public class DataLakeResourceV4 extends AbstractRestResource { - private final DataExplorerQueryManagement dataLakeManagement; - private final DataExplorerSchemaManagement dataExplorerSchemaManagement; + private final IDataExplorerQueryManagement dataExplorerQueryManagement; + private final IDataExplorerSchemaManagement dataExplorerSchemaManagement; public DataLakeResourceV4() { - var dataLakeStorage = StorageDispatcher.INSTANCE - .getNoSqlStore() - .getDataLakeStorage(); - this.dataExplorerSchemaManagement = new DataExplorerSchemaManagement(dataLakeStorage); - this.dataLakeManagement = new DataExplorerQueryManagement(dataExplorerSchemaManagement); + this.dataExplorerSchemaManagement = DataExplorerDispatcher.INSTANCE + .getDataExplorerManager() + .getSchemaManagement(); + this.dataExplorerQueryManagement = DataExplorerDispatcher.INSTANCE + .getDataExplorerManager() + .getQueryManagement(this.dataExplorerSchemaManagement); } - public DataLakeResourceV4(DataExplorerQueryManagement dataLakeManagement) { - var dataLakeStorage = StorageDispatcher.INSTANCE - .getNoSqlStore() - .getDataLakeStorage(); - this.dataLakeManagement = dataLakeManagement; - this.dataExplorerSchemaManagement = new DataExplorerSchemaManagement(dataLakeStorage); + public DataLakeResourceV4(IDataExplorerQueryManagement dataExplorerQueryManagement) { + this.dataExplorerQueryManagement = dataExplorerQueryManagement; + this.dataExplorerSchemaManagement = DataExplorerDispatcher.INSTANCE.getDataExplorerManager() + .getSchemaManagement(); } @DeleteMapping(path = "/measurements/{measurementID}") @@ -112,7 +111,7 @@ public class DataLakeResourceV4 extends AbstractRestResource { , @Parameter(in = ParameterIn.QUERY, description = "end date for slicing operation") @RequestParam(value = "endDate", required = false) Long endDate) { - SpQueryResult result = this.dataLakeManagement.deleteData(measurementID, startDate, endDate); + SpQueryResult result = this.dataExplorerQueryManagement.deleteData(measurementID, startDate, endDate); return ok(); } @@ -132,7 +131,7 @@ public class DataLakeResourceV4 extends AbstractRestResource { @Parameter(in = ParameterIn.PATH, description = "the id of the measurement series", required = true) @PathVariable("measurementID") String measurementID) { - boolean isSuccessDataLake = this.dataLakeManagement.deleteData(measurementID); + boolean isSuccessDataLake = this.dataExplorerQueryManagement.deleteData(measurementID); if (isSuccessDataLake) { boolean isSuccessEventProperty = this.dataExplorerSchemaManagement.deleteMeasurementByName(measurementID); @@ -165,7 +164,7 @@ public class DataLakeResourceV4 extends AbstractRestResource { @GetMapping(path = "/measurements/{measurementId}/tags", produces = MediaType.APPLICATION_JSON_VALUE) public ResponseEntity<Map<String, Object>> getTagValues(@PathVariable("measurementId") String measurementId, @RequestParam("fields") String fields) { - Map<String, Object> tagValues = dataLakeManagement.getTagValues(measurementId, fields); + Map<String, Object> tagValues = dataExplorerQueryManagement.getTagValues(measurementId, fields); return ok(tagValues); } @@ -235,7 +234,7 @@ public class DataLakeResourceV4 extends AbstractRestResource { ProvidedRestQueryParams sanitizedParams = populate(measurementID, queryParams); try { SpQueryResult result = - this.dataLakeManagement.getData(sanitizedParams, isIgnoreMissingValues(missingValueBehaviour)); + this.dataExplorerQueryManagement.getData(sanitizedParams, isIgnoreMissingValues(missingValueBehaviour)); return ok(result); } catch (RuntimeException e) { return badRequest(SpLogMessage.from(e)); @@ -251,7 +250,7 @@ public class DataLakeResourceV4 extends AbstractRestResource { var results = queryParams .stream() .map(qp -> new ProvidedRestQueryParams(qp.get("measureName"), qp)) - .map(params -> this.dataLakeManagement.getData(params, true)) + .map(params -> this.dataExplorerQueryManagement.getData(params, true)) .collect(Collectors.toList()); return ok(results); @@ -322,7 +321,7 @@ public class DataLakeResourceV4 extends AbstractRestResource { } OutputFormat outputFormat = format.equals("csv") ? OutputFormat.CSV : OutputFormat.JSON; - StreamingResponseBody streamingOutput = output -> dataLakeManagement.getDataAsStream( + StreamingResponseBody streamingOutput = output -> dataExplorerQueryManagement.getDataAsStream( sanitizedParams, outputFormat, isIgnoreMissingValues(missingValueBehaviour), @@ -343,7 +342,7 @@ public class DataLakeResourceV4 extends AbstractRestResource { responses = { @ApiResponse(responseCode = "200", description = "All measurement series successfully removed")}) public ResponseEntity<?> removeAll() { - boolean isSuccess = this.dataLakeManagement.deleteAllData(); + boolean isSuccess = this.dataExplorerQueryManagement.deleteAllData(); return ResponseEntity.ok(isSuccess); } @@ -360,13 +359,6 @@ public class DataLakeResourceV4 extends AbstractRestResource { // Checks if the parameter for missing value behaviour is set private boolean isIgnoreMissingValues(String missingValueBehaviour) { - boolean ignoreMissingValues; - if ("ignore".equals(missingValueBehaviour)) { - ignoreMissingValues = true; - } else { - ignoreMissingValues = false; - } - return ignoreMissingValues; + return "ignore".equals(missingValueBehaviour); } - } diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/ResetManagement.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/ResetManagement.java index 3c68acf0b9..c72b605982 100644 --- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/ResetManagement.java +++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/ResetManagement.java @@ -22,8 +22,7 @@ import org.apache.streampipes.commons.exceptions.SpRuntimeException; import org.apache.streampipes.commons.exceptions.connect.AdapterException; import org.apache.streampipes.commons.prometheus.adapter.AdapterMetricsManager; import org.apache.streampipes.connect.management.management.AdapterMasterManagement; -import org.apache.streampipes.dataexplorer.DataExplorerSchemaManagement; -import org.apache.streampipes.dataexplorer.influx.migrate.DataExplorerQueryManagement; +import org.apache.streampipes.dataexplorer.management.DataExplorerDispatcher; import org.apache.streampipes.manager.file.FileManager; import org.apache.streampipes.manager.pipeline.PipelineCacheManager; import org.apache.streampipes.manager.pipeline.PipelineCanvasMetadataCacheManager; @@ -137,12 +136,10 @@ public class ResetManagement { } private static void removeAllDataInDataLake() { - var dataLakeStorage = StorageDispatcher.INSTANCE - .getNoSqlStore() - .getDataLakeStorage(); - var dataLakeMeasureManagement = new DataExplorerSchemaManagement(dataLakeStorage); + var dataLakeMeasureManagement = DataExplorerDispatcher.INSTANCE.getDataExplorerManager() + .getSchemaManagement(); var dataExplorerQueryManagement = - new DataExplorerQueryManagement(dataLakeMeasureManagement); + DataExplorerDispatcher.INSTANCE.getDataExplorerManager().getQueryManagement(dataLakeMeasureManagement); List<DataLakeMeasure> allMeasurements = dataLakeMeasureManagement.getAllMeasurements(); allMeasurements.forEach(measurement -> { boolean isSuccessDataLake = dataExplorerQueryManagement.deleteData(measurement.getMeasureName());
