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 958a834991187707c7eb518a53bd0e0c6fdb0811 Author: bossenti <[email protected]> AuthorDate: Tue Apr 30 11:26:30 2024 +0200 refactor: several adaptions --- .../influx/TimeSeriesStorageInflux.java | 3 - .../influx/migrate/DataExplorerUtils.java | 86 ---------------------- .../dataexplorer}/AutoAggregationHandler.java | 16 +--- .../streampipes/dataexplorer/ImageStore.java | 2 - .../streampipes/dataexplorer/ImageStoreUtils.java | 3 +- .../streampipes/dataexplorer/TimeSeriesStore.java | 1 - .../streampipes/ps/DataLakeMeasureResourceV4.java | 17 +++-- 7 files changed, 15 insertions(+), 113 deletions(-) diff --git a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/TimeSeriesStorageInflux.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/TimeSeriesStorageInflux.java index 9680d7920d..26093e69dc 100644 --- a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/TimeSeriesStorageInflux.java +++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/TimeSeriesStorageInflux.java @@ -18,13 +18,10 @@ package org.apache.streampipes.dataexplorer.influx; -import org.apache.streampipes.client.api.IStreamPipesClient; import org.apache.streampipes.commons.environment.Environment; import org.apache.streampipes.commons.exceptions.SpRuntimeException; import org.apache.streampipes.dataexplorer.TimeSeriesStorage; -import org.apache.streampipes.dataexplorer.api.IDataLakeMeasurementSanitizer; import org.apache.streampipes.dataexplorer.influx.client.InfluxClientProvider; -import org.apache.streampipes.dataexplorer.influx.sanitize.DataLakeMeasurementSanitizerInflux; import org.apache.streampipes.dataexplorer.influx.sanitize.InfluxNameSanitizer; import org.apache.streampipes.model.datalake.DataLakeMeasure; import org.apache.streampipes.model.runtime.Event; diff --git a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/DataExplorerUtils.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/DataExplorerUtils.java deleted file mode 100644 index 97d8b9089d..0000000000 --- a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/DataExplorerUtils.java +++ /dev/null @@ -1,86 +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.dataexplorer.influx.migrate; - -import org.apache.streampipes.client.api.IStreamPipesClient; -import org.apache.streampipes.commons.exceptions.SpRuntimeException; -import org.apache.streampipes.dataexplorer.influx.sanitize.InfluxNameSanitizer; -import org.apache.streampipes.dataexplorer.influx.sanitize.MeasureNameSanitizer; -import org.apache.streampipes.model.datalake.DataLakeMeasure; -import org.apache.streampipes.model.schema.EventProperty; - -import java.util.List; -import java.util.stream.Collectors; - -public class DataExplorerUtils { - /** - * Sanitizes the event schema and stores the DataLakeMeasurement to the couchDB - * - * @param client StreamPipes client to store measure - * @param measure DataLakeMeasurement - */ - public static DataLakeMeasure sanitizeAndRegisterAtDataLake(IStreamPipesClient client, - DataLakeMeasure measure) throws SpRuntimeException { - sanitizeDataLakeMeasure(measure); - registerAtDataLake(client, measure); - - return measure; - } - - public static DataLakeMeasure sanitizeAndUpdateAtDataLake(IStreamPipesClient client, - DataLakeMeasure measure) throws SpRuntimeException { - sanitizeDataLakeMeasure(measure); - updateAtDataLake(client, measure); - return measure; - } - - private static void registerAtDataLake(IStreamPipesClient client, - DataLakeMeasure measure) throws SpRuntimeException { - client.dataLakeMeasureApi().create(measure); - } - - public static void updateAtDataLake(IStreamPipesClient client, - DataLakeMeasure measure) throws SpRuntimeException { - client.dataLakeMeasureApi().update(measure); - } - - - private static void sanitizeDataLakeMeasure(DataLakeMeasure measure) throws SpRuntimeException { - - // Removes selected timestamp from event schema - removeTimestampsFromEventSchema(measure); - - // Sanitize the data lake measure name - measure.setMeasureName(new MeasureNameSanitizer().sanitize(measure.getMeasureName())); - - // Removes all spaces with _ and validates that no special terms are used as runtime names - measure.getEventSchema() - .getEventProperties() - .forEach(ep -> ep.setRuntimeName(InfluxNameSanitizer.renameReservedKeywords(ep.getRuntimeName()))); - - } - - private static void removeTimestampsFromEventSchema(DataLakeMeasure measure) { - List<EventProperty> eventPropertiesWithoutTimestamp = measure.getEventSchema().getEventProperties() - .stream() - .filter(eventProperty -> !measure.getTimestampField().endsWith(eventProperty.getRuntimeName())) - .collect(Collectors.toList()); - measure.getEventSchema().setEventProperties(eventPropertiesWithoutTimestamp); - } - -} diff --git a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/AutoAggregationHandler.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/AutoAggregationHandler.java similarity index 91% rename from streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/AutoAggregationHandler.java rename to streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/AutoAggregationHandler.java index 692e7412be..45a20a036a 100644 --- a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/AutoAggregationHandler.java +++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/AutoAggregationHandler.java @@ -15,15 +15,14 @@ * limitations under the License. * */ -package org.apache.streampipes.dataexplorer.influx.migrate; +package org.apache.streampipes.dataexplorer; -import org.apache.streampipes.dataexplorer.DataExplorerSchemaManagement; +import org.apache.streampipes.dataexplorer.api.IDataExplorerQueryManagement; import org.apache.streampipes.model.datalake.param.ProvidedRestQueryParams; import org.apache.streampipes.model.datalake.param.SupportedRestQueryParams; import org.apache.streampipes.dataexplorer.param.model.SelectColumn; import org.apache.streampipes.model.datalake.DataLakeQueryOrdering; import org.apache.streampipes.model.datalake.SpQueryResult; -import org.apache.streampipes.storage.management.StorageDispatcher; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -50,16 +49,9 @@ public class AutoAggregationHandler { private final IDataExplorerQueryManagement dataLakeQueryManagement; private final ProvidedRestQueryParams queryParams; - public AutoAggregationHandler(ProvidedRestQueryParams params) { + public AutoAggregationHandler(ProvidedRestQueryParams params, IDataExplorerQueryManagement dataExplorerQueryManagement) { this.queryParams = params; - this.dataLakeQueryManagement = getDataLakeQueryManagement(); - } - - private IDataExplorerQueryManagement getDataLakeQueryManagement() { - var dataLakeStorage = StorageDispatcher.INSTANCE - .getNoSqlStore() - .getDataLakeStorage(); - return new DataExplorerQueryManagement(new DataExplorerSchemaManagement(dataLakeStorage)); + this.dataLakeQueryManagement = dataExplorerQueryManagement; } public ProvidedRestQueryParams makeAutoAggregationQueryParams() throws IllegalArgumentException { diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/ImageStore.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/ImageStore.java index 67af69225d..dbd3bbd860 100644 --- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/ImageStore.java +++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/ImageStore.java @@ -36,8 +36,6 @@ import java.util.List; import java.util.UUID; public class ImageStore { - - private static final Logger LOG = LoggerFactory.getLogger(ImageStore.class); private static final String DB_NAME = "images"; private List<EventProperty> imageProperties; diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/ImageStoreUtils.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/ImageStoreUtils.java index 65e777f874..59d4719a7d 100644 --- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/ImageStoreUtils.java +++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/ImageStoreUtils.java @@ -30,7 +30,8 @@ public class ImageStoreUtils { public static List<EventProperty> getImageProperties(DataLakeMeasure measure) { return measure.getEventSchema().getEventProperties().stream() .filter(eventProperty -> eventProperty.getDomainProperties() != null - && eventProperty.getDomainProperties().size() > 0 + && !eventProperty.getDomainProperties() + .isEmpty() && eventProperty.getDomainProperties().get(0).toString().equals(SPSensor.IMAGE)) .collect(Collectors.toList()); } diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/TimeSeriesStore.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/TimeSeriesStore.java index f97847585f..78887b52d2 100644 --- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/TimeSeriesStore.java +++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/TimeSeriesStore.java @@ -39,7 +39,6 @@ public class TimeSeriesStore { public TimeSeriesStore( ITimeSeriesStorage timeSeriesStorage, - IStreamPipesClient client, DataLakeMeasure measure, Environment environment, boolean enableImageStore diff --git a/streampipes-platform-services/src/main/java/org/apache/streampipes/ps/DataLakeMeasureResourceV4.java b/streampipes-platform-services/src/main/java/org/apache/streampipes/ps/DataLakeMeasureResourceV4.java index 64c009e7a8..fb14765e21 100644 --- a/streampipes-platform-services/src/main/java/org/apache/streampipes/ps/DataLakeMeasureResourceV4.java +++ b/streampipes-platform-services/src/main/java/org/apache/streampipes/ps/DataLakeMeasureResourceV4.java @@ -18,12 +18,10 @@ package org.apache.streampipes.ps; -import org.apache.streampipes.dataexplorer.DataExplorerSchemaManagement; import org.apache.streampipes.dataexplorer.api.IDataExplorerSchemaManagement; -import org.apache.streampipes.dataexplorer.influx.migrate.DataLakeMeasurementCount; +import org.apache.streampipes.dataexplorer.management.DataExplorerDispatcher; import org.apache.streampipes.model.datalake.DataLakeMeasure; import org.apache.streampipes.rest.core.base.impl.AbstractAuthGuardedRestResource; -import org.apache.streampipes.storage.management.StorageDispatcher; import org.springframework.http.MediaType; import org.springframework.http.ResponseEntity; @@ -48,10 +46,8 @@ public class DataLakeMeasureResourceV4 extends AbstractAuthGuardedRestResource { private final IDataExplorerSchemaManagement dataLakeMeasureManagement; public DataLakeMeasureResourceV4() { - var dataLakeStorage = StorageDispatcher.INSTANCE - .getNoSqlStore() - .getDataLakeStorage(); - this.dataLakeMeasureManagement = new DataExplorerSchemaManagement(dataLakeStorage); + this.dataLakeMeasureManagement = DataExplorerDispatcher.INSTANCE.getDataExplorerManager() + .getSchemaManagement(); } @PostMapping( @@ -67,7 +63,12 @@ public class DataLakeMeasureResourceV4 extends AbstractAuthGuardedRestResource { public ResponseEntity<Map<String, Integer>> getDataLakeInfos( @RequestParam(value = "filter", required = false) List<String> measurementNames) { var allMeasurements = this.dataLakeMeasureManagement.getAllMeasurements(); - return ok(new DataLakeMeasurementCount(allMeasurements, measurementNames).countMeasurementSizes()); + return ok(DataExplorerDispatcher.INSTANCE.getDataExplorerManager() + .getMeasurementCounter( + allMeasurements, + measurementNames + ) + .countMeasurementSizes()); } @GetMapping(path = "{id}", produces = MediaType.APPLICATION_JSON_VALUE)
