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 b6ddd4912fc144d6724983bfd94145697bb46d2b Author: bossenti <[email protected]> AuthorDate: Tue Apr 30 11:21:38 2024 +0200 refactor: abstract counter --- .../commons/environment/DefaultEnvironment.java | 5 ++++ .../api/IDataLakeMeasurementCounter.java | 35 ++++++++++++++++++++++ ....java => DataLakeMeasurementCounterInflux.java} | 14 ++++----- 3 files changed, 47 insertions(+), 7 deletions(-) diff --git a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/DefaultEnvironment.java b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/DefaultEnvironment.java index 5e613e0909..b7a48269f3 100644 --- a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/DefaultEnvironment.java +++ b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/DefaultEnvironment.java @@ -55,6 +55,11 @@ public class DefaultEnvironment implements Environment { return new IntEnvironmentVariable(Envs.SP_CORE_PORT); } + @Override + public StringEnvironmentVariable getTsStorage() { + return new StringEnvironmentVariable(Envs.SP_TS_STORAGE); + } + @Override public StringEnvironmentVariable getTsStorageProtocol() { return new StringEnvironmentVariable(Envs.SP_TS_STORAGE_PROTOCOL); diff --git a/streampipes-data-explorer-api/src/main/java/org/apache/streampipes/dataexplorer/api/IDataLakeMeasurementCounter.java b/streampipes-data-explorer-api/src/main/java/org/apache/streampipes/dataexplorer/api/IDataLakeMeasurementCounter.java new file mode 100644 index 0000000000..a5527f996a --- /dev/null +++ b/streampipes-data-explorer-api/src/main/java/org/apache/streampipes/dataexplorer/api/IDataLakeMeasurementCounter.java @@ -0,0 +1,35 @@ +/* + * 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 java.util.Map; + +/** + * Interface for counting the number of events per measurement within the StreamPipes data storage. + */ +public interface IDataLakeMeasurementCounter { + + /** + * Counts the sizes of measurements within the StreamPipes data storage. + * + * @return A map where each key represents a measurement name and its corresponding value represents + * the number of events contained by that measurement. + */ + Map<String, Integer> countMeasurementSizes(); +} \ No newline at end of file diff --git a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/DataLakeMeasurementCount.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataLakeMeasurementCounterInflux.java similarity index 88% rename from streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/DataLakeMeasurementCount.java rename to streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataLakeMeasurementCounterInflux.java index aa21d051fa..4f516f52c1 100644 --- a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/DataLakeMeasurementCount.java +++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataLakeMeasurementCounterInflux.java @@ -16,10 +16,9 @@ * */ -package org.apache.streampipes.dataexplorer.influx.migrate; +package org.apache.streampipes.dataexplorer.influx; -import org.apache.streampipes.dataexplorer.influx.DataExplorerInfluxQueryExecutor; -import org.apache.streampipes.dataexplorer.influx.DataLakeInfluxQueryBuilder; +import org.apache.streampipes.dataexplorer.api.IDataLakeMeasurementCounter; import org.apache.streampipes.model.datalake.AggregationFunction; import org.apache.streampipes.model.datalake.DataLakeMeasure; import org.apache.streampipes.model.datalake.SpQueryResult; @@ -29,19 +28,20 @@ import org.apache.streampipes.model.schema.PropertyScope; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.stream.Collectors; -public class DataLakeMeasurementCount { +public class DataLakeMeasurementCounterInflux implements IDataLakeMeasurementCounter { private final List<DataLakeMeasure> allMeasurements; private final List<String> measurementNames; private static final String COUNT_FIELD = "count"; - public DataLakeMeasurementCount(List<DataLakeMeasure> allMeasurements, - List<String> measurementNames) { + public DataLakeMeasurementCounterInflux(List<DataLakeMeasure> allMeasurements, + List<String> measurementNames) { this.allMeasurements = allMeasurements; this.measurementNames = measurementNames; } @@ -55,7 +55,7 @@ public class DataLakeMeasurementCount { var builder = DataLakeInfluxQueryBuilder .create(m.getMeasureName()).withEndTime(System.currentTimeMillis()) .withAggregatedColumn(firstColumn, AggregationFunction.COUNT); - var queryResult = new DataExplorerInfluxQueryExecutor().executeQuery(builder.build(), true); + var queryResult = new DataExplorerInfluxQueryExecutor().executeQuery(builder.build(), Optional.empty(), true); if (queryResult.getTotal() > 0) { var headers = queryResult.getHeaders(); return extractResult(queryResult, headers);
