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 4669941f50d0b1b8d2d12e00725318cb39427d88 Author: bossenti <[email protected]> AuthorDate: Tue Apr 30 11:27:05 2024 +0200 feat: data storage management --- streampipes-data-explorer-api/pom.xml | 5 ++ .../dataexplorer/api/IDataExplorerManager.java | 48 +++++++++++++++ .../influx/DataExplorerManagerInflux.java | 71 ++++++++++++++++++++++ streampipes-data-explorer-management/pom.xml | 5 ++ .../management/DataExplorerDispatcher.java | 39 ++++++++++++ .../management/SupportedDataExplorerStorages.java | 16 +++-- .../streampipes-sinks-internal-jvm/pom.xml | 12 +++- streampipes-platform-services/pom.xml | 5 ++ streampipes-rest/pom.xml | 5 ++ 9 files changed, 198 insertions(+), 8 deletions(-) diff --git a/streampipes-data-explorer-api/pom.xml b/streampipes-data-explorer-api/pom.xml index a433295db7..c8c573151d 100644 --- a/streampipes-data-explorer-api/pom.xml +++ b/streampipes-data-explorer-api/pom.xml @@ -34,6 +34,11 @@ <dependencies> <!-- StreamPipes dependencies --> + <dependency> + <groupId>org.apache.streampipes</groupId> + <artifactId>streampipes-client-api</artifactId> + <version>0.95.0-SNAPSHOT</version> + </dependency> <dependency> <groupId>org.apache.streampipes</groupId> <artifactId>streampipes-data-explorer-export</artifactId> diff --git a/streampipes-data-explorer-api/src/main/java/org/apache/streampipes/dataexplorer/api/IDataExplorerManager.java b/streampipes-data-explorer-api/src/main/java/org/apache/streampipes/dataexplorer/api/IDataExplorerManager.java new file mode 100644 index 0000000000..ef1b68a0ca --- /dev/null +++ b/streampipes-data-explorer-api/src/main/java/org/apache/streampipes/dataexplorer/api/IDataExplorerManager.java @@ -0,0 +1,48 @@ +/* + * 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.api; + +import org.apache.streampipes.client.api.IStreamPipesClient; +import org.apache.streampipes.model.datalake.DataLakeMeasure; + +import java.util.List; + +public interface IDataExplorerManager { + + /** + * Provide an instance of {@link IDataLakeMeasurementCounter} for counting the sizes of measurements within a data + * lake. + * + * @param allMeasurements A list of {@link DataLakeMeasure} objects representing all measurements in the data lake. + * @param measurementsToCount A list of measurement names for which the sizes should be counted. + * @return An instance of {@link IDataLakeMeasurementCounter} configured to count the sizes of the specified measurements. + */ + IDataLakeMeasurementCounter getMeasurementCounter( + List<DataLakeMeasure> allMeasurements, + List<String> measurementsToCount + ); + + IDataExplorerQueryManagement getQueryManagement(IDataExplorerSchemaManagement dataExplorerSchemaManagement); + + IDataExplorerSchemaManagement getSchemaManagement(); + + ITimeSeriesStorage getTimeseriesStorage(DataLakeMeasure measure); + + IDataLakeMeasurementSanitizer getMeasurementSanitizer(IStreamPipesClient client, DataLakeMeasure measure); +} diff --git a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataExplorerManagerInflux.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataExplorerManagerInflux.java new file mode 100644 index 0000000000..e79cb8d2bc --- /dev/null +++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataExplorerManagerInflux.java @@ -0,0 +1,71 @@ +/* + * 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; + +import org.apache.streampipes.client.api.IStreamPipesClient; +import org.apache.streampipes.commons.environment.Environments; +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.api.IDataExplorerManager; +import org.apache.streampipes.dataexplorer.api.IDataLakeMeasurementCounter; +import org.apache.streampipes.dataexplorer.api.IDataLakeMeasurementSanitizer; +import org.apache.streampipes.dataexplorer.api.ITimeSeriesStorage; +import org.apache.streampipes.dataexplorer.influx.client.InfluxClientProvider; +import org.apache.streampipes.dataexplorer.influx.sanitize.DataLakeMeasurementSanitizerInflux; +import org.apache.streampipes.model.datalake.DataLakeMeasure; +import org.apache.streampipes.storage.management.StorageDispatcher; + +import java.util.List; + +public enum DataExplorerManagerInflux implements IDataExplorerManager { + + INSTANCE; + + @Override + public IDataLakeMeasurementCounter getMeasurementCounter( + List<DataLakeMeasure> allMeasurements, + List<String> measurementsToCount) { + return new DataLakeMeasurementCounterInflux(allMeasurements, measurementsToCount); + } + + @Override + public IDataExplorerQueryManagement getQueryManagement( + IDataExplorerSchemaManagement dataExplorerSchemaManagement + ) { + return new DataExplorerQueryManagementInflux(dataExplorerSchemaManagement); + } + + @Override + public IDataExplorerSchemaManagement getSchemaManagement() { + return new DataExplorerSchemaManagement(StorageDispatcher.INSTANCE + .getNoSqlStore() + .getDataLakeStorage()); + } + + @Override + public ITimeSeriesStorage getTimeseriesStorage(DataLakeMeasure measure) { + return new TimeSeriesStorageInflux(measure, Environments.getEnvironment(), new InfluxClientProvider()); + } + + @Override + public IDataLakeMeasurementSanitizer getMeasurementSanitizer(IStreamPipesClient client, DataLakeMeasure measure) { + return new DataLakeMeasurementSanitizerInflux(client, measure); + } +} diff --git a/streampipes-data-explorer-management/pom.xml b/streampipes-data-explorer-management/pom.xml index 09c89dacff..56edc05e4e 100644 --- a/streampipes-data-explorer-management/pom.xml +++ b/streampipes-data-explorer-management/pom.xml @@ -44,6 +44,11 @@ <artifactId>streampipes-data-explorer-api</artifactId> <version>0.95.0-SNAPSHOT</version> </dependency> + <dependency> + <groupId>org.apache.streampipes</groupId> + <artifactId>streampipes-data-explorer-influx</artifactId> + <version>0.95.0-SNAPSHOT</version> + </dependency> </dependencies> <properties> diff --git a/streampipes-data-explorer-management/src/main/java/org/apache/streampipes/dataexplorer/management/DataExplorerDispatcher.java b/streampipes-data-explorer-management/src/main/java/org/apache/streampipes/dataexplorer/management/DataExplorerDispatcher.java new file mode 100644 index 0000000000..eb09e82492 --- /dev/null +++ b/streampipes-data-explorer-management/src/main/java/org/apache/streampipes/dataexplorer/management/DataExplorerDispatcher.java @@ -0,0 +1,39 @@ +/* + * 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.management; + +import org.apache.streampipes.commons.environment.Environments; +import org.apache.streampipes.dataexplorer.api.IDataExplorerManager; +import org.apache.streampipes.dataexplorer.influx.DataExplorerManagerInflux; + +public enum DataExplorerDispatcher { + INSTANCE; + + public IDataExplorerManager getDataExplorerManager() { + + // currently this SWITCH CASE statement is not necessary + // but aims to give an idea how to deal with multiple data explorer storages + return switch (Environments.getEnvironment() + .getTsStorage() + .getValueOrDefault()) { + case SupportedDataExplorerStorages.INFLUX_DB -> DataExplorerManagerInflux.INSTANCE; + default -> DataExplorerManagerInflux.INSTANCE; + }; + } +} diff --git a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/package-info.java b/streampipes-data-explorer-management/src/main/java/org/apache/streampipes/dataexplorer/management/SupportedDataExplorerStorages.java similarity index 65% rename from streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/package-info.java rename to streampipes-data-explorer-management/src/main/java/org/apache/streampipes/dataexplorer/management/SupportedDataExplorerStorages.java index c96a83d257..444d6e43be 100644 --- a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/package-info.java +++ b/streampipes-data-explorer-management/src/main/java/org/apache/streampipes/dataexplorer/management/SupportedDataExplorerStorages.java @@ -16,10 +16,16 @@ * */ +package org.apache.streampipes.dataexplorer.management; + /** - * This package contains classes that have direct dependencies on InfluxDB-specific implementations. - * To enhance code organization and promote agnostic design, these classes are slated for migration to other modules, - * such as 'streampipes-data-explorer' or 'streampipes-data-explorer-api'. - * Pending migration, they are temporarily housed within this package. + * Class containing constants for supported time series storage implementations. + * <p> + * Supported time series storage implementations: + * <ul> + * <li>{@link #INFLUX_DB}: Represents the InfluxDB time series storage implementation.</li> + * </ul> */ -package org.apache.streampipes.dataexplorer.influx.migrate; \ No newline at end of file +public class SupportedDataExplorerStorages { + public static final String INFLUX_DB = "influxDB"; +} diff --git a/streampipes-extensions/streampipes-sinks-internal-jvm/pom.xml b/streampipes-extensions/streampipes-sinks-internal-jvm/pom.xml index d9b9f7d216..239a09c3ac 100644 --- a/streampipes-extensions/streampipes-sinks-internal-jvm/pom.xml +++ b/streampipes-extensions/streampipes-sinks-internal-jvm/pom.xml @@ -41,19 +41,25 @@ <!-- StreamPipes dependencies --> <dependency> <groupId>org.apache.streampipes</groupId> - <artifactId>streampipes-extensions-management</artifactId> + <artifactId>streampipes-data-explorer-influx</artifactId> <version>0.95.0-SNAPSHOT</version> </dependency> <dependency> <groupId>org.apache.streampipes</groupId> - <artifactId>streampipes-wrapper-standalone</artifactId> + <artifactId>streampipes-data-explorer-management</artifactId> <version>0.95.0-SNAPSHOT</version> </dependency> <dependency> <groupId>org.apache.streampipes</groupId> - <artifactId>streampipes-data-explorer-influx</artifactId> + <artifactId>streampipes-extensions-management</artifactId> <version>0.95.0-SNAPSHOT</version> </dependency> + <dependency> + <groupId>org.apache.streampipes</groupId> + <artifactId>streampipes-wrapper-standalone</artifactId> + <version>0.95.0-SNAPSHOT</version> + </dependency> + <!-- External dependencies --> <!-- TODO remove after refactoring --> diff --git a/streampipes-platform-services/pom.xml b/streampipes-platform-services/pom.xml index 13f92da128..ffe439094a 100644 --- a/streampipes-platform-services/pom.xml +++ b/streampipes-platform-services/pom.xml @@ -38,6 +38,11 @@ <artifactId>streampipes-data-explorer-influx</artifactId> <version>0.95.0-SNAPSHOT</version> </dependency> + <dependency> + <groupId>org.apache.streampipes</groupId> + <artifactId>streampipes-data-explorer-management</artifactId> + <version>0.95.0-SNAPSHOT</version> + </dependency> <dependency> <groupId>org.apache.streampipes</groupId> <artifactId>streampipes-model</artifactId> diff --git a/streampipes-rest/pom.xml b/streampipes-rest/pom.xml index 48b45acddc..5a007f5d3d 100644 --- a/streampipes-rest/pom.xml +++ b/streampipes-rest/pom.xml @@ -45,6 +45,11 @@ <artifactId>streampipes-data-explorer-influx</artifactId> <version>0.95.0-SNAPSHOT</version> </dependency> + <dependency> + <groupId>org.apache.streampipes</groupId> + <artifactId>streampipes-data-explorer-management</artifactId> + <version>0.95.0-SNAPSHOT</version> + </dependency> <dependency> <groupId>org.apache.streampipes</groupId> <artifactId>streampipes-data-export</artifactId>
