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 599edf0c611a5917b88e57cf31a79d588777b85a Author: bossenti <[email protected]> AuthorDate: Tue Apr 30 11:22:49 2024 +0200 refactor: abstract time series store from Influx --- .../dataexplorer/api/ITimeSeriesStorage.java | 20 +-- ...fluxStore.java => TimeSeriesStorageInflux.java} | 125 ++---------------- ...eTest.java => TimeSeriesStorageInfluxTest.java} | 10 +- .../streampipes/dataexplorer}/ImageStore.java | 2 +- .../streampipes/dataexplorer}/ImageStoreUtils.java | 2 +- .../dataexplorer/TimeSeriesStorage.java | 145 +++++++++++++++++++++ .../streampipes/dataexplorer}/TimeSeriesStore.java | 28 ++-- 7 files changed, 181 insertions(+), 151 deletions(-) diff --git a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/ImageStoreUtils.java b/streampipes-data-explorer-api/src/main/java/org/apache/streampipes/dataexplorer/api/ITimeSeriesStorage.java similarity index 52% copy from streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/ImageStoreUtils.java copy to streampipes-data-explorer-api/src/main/java/org/apache/streampipes/dataexplorer/api/ITimeSeriesStorage.java index 6504303a5a..f3299db16a 100644 --- a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/ImageStoreUtils.java +++ b/streampipes-data-explorer-api/src/main/java/org/apache/streampipes/dataexplorer/api/ITimeSeriesStorage.java @@ -16,22 +16,14 @@ * */ -package org.apache.streampipes.dataexplorer.influx.migrate; +package org.apache.streampipes.dataexplorer.api; -import org.apache.streampipes.model.datalake.DataLakeMeasure; -import org.apache.streampipes.model.schema.EventProperty; -import org.apache.streampipes.vocabulary.SPSensor; +import org.apache.streampipes.commons.exceptions.SpRuntimeException; +import org.apache.streampipes.model.runtime.Event; -import java.util.List; -import java.util.stream.Collectors; +public interface ITimeSeriesStorage { -public class ImageStoreUtils { + void onEvent(Event event) throws SpRuntimeException; - public static List<EventProperty> getImageProperties(DataLakeMeasure measure) { - return measure.getEventSchema().getEventProperties().stream() - .filter(eventProperty -> eventProperty.getDomainProperties() != null - && eventProperty.getDomainProperties().size() > 0 - && eventProperty.getDomainProperties().get(0).toString().equals(SPSensor.IMAGE)) - .collect(Collectors.toList()); - } + void close() throws SpRuntimeException; } diff --git a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/InfluxStore.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/TimeSeriesStorageInflux.java similarity index 52% rename from streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/InfluxStore.java rename to streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/TimeSeriesStorageInflux.java index 0d10ac147f..9680d7920d 100644 --- a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/InfluxStore.java +++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/TimeSeriesStorageInflux.java @@ -18,115 +18,46 @@ 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; -import org.apache.streampipes.model.schema.EventProperty; import org.apache.streampipes.model.schema.EventPropertyPrimitive; import org.influxdb.InfluxDB; import org.influxdb.dto.Point; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import java.util.HashMap; -import java.util.List; -import java.util.Map; import java.util.concurrent.TimeUnit; -import java.util.stream.Collectors; -public class InfluxStore { +public class TimeSeriesStorageInflux extends TimeSeriesStorage { - private static final Logger LOG = LoggerFactory.getLogger(InfluxStore.class); - - private final DataLakeMeasure measure; - private final List<EventProperty> allEventProperties; - private final Map<String, String> sanitizedRuntimeNames = new HashMap<>(); private final InfluxDB influxDb; private final PropertyHandler propertyHandler; - public InfluxStore( + public TimeSeriesStorageInflux( DataLakeMeasure measure, Environment environment, InfluxClientProvider influxClientProvider ) throws SpRuntimeException { - this.measure = measure; - storeSanitizedRuntimeNames(measure); - allEventProperties = getAllEventPropertiesExceptTimestamp(measure); + super(measure); influxDb = influxClientProvider.getInitializedInfluxDBClient(environment); propertyHandler = new PropertyHandler(); } - /** - * Takes an StreamPipes event, transforms it to an InfluxDB point and writes it to the InfluxDB - * - * @param event The event which should be saved - * @throws SpRuntimeException If the column name (key-value of the event map) is not allowed - */ - public void onEvent(Event event) throws SpRuntimeException { - - validateInputEventAndLogMissingFields(event); - - sanitizeRuntimeNamesInEvent(event); - + protected void writeToTimeSeriesStorage(Event event) throws SpRuntimeException { var point = initializePointWithTimestamp(event); - iterateOverallEventProperties(event, point); - influxDb.write(point.build()); } - private void validateInputEventAndLogMissingFields(Event event) { - checkEventIsNotNull(event); - - logMissingFields(event); - - logNullFields(event); - } - - /** - * Logs all fields which are present in the schema, but not in the provided event - */ - private void logMissingFields(Event event) { - var missingFields = getMissingProperties(allEventProperties, event); - if (!missingFields.isEmpty()) { - LOG.debug( - "Ignored {} fields which were present in the schema, but not in the provided event: {}", - missingFields.size(), - String.join(", ", missingFields) - ); - } - } - - - /** - * Logs all fields that contain null values - */ - private void logNullFields(Event event) { - List<String> nullFields = allEventProperties - .stream() - .filter(EventPropertyPrimitive.class::isInstance) - .filter(ep -> { - var runtimeName = ep.getRuntimeName(); - var field = event.getOptionalFieldByRuntimeName(runtimeName); - - return field.isPresent() && field.get() - .getAsPrimitive() - .getRawValue() == null; - }) - .map(EventProperty::getRuntimeName) - .collect(Collectors.toList()); - - if (!nullFields.isEmpty()) { - LOG.warn("Ignored {} fields which had a value 'null': {}", nullFields.size(), String.join(", ", nullFields)); - } - } - private void iterateOverallEventProperties( Event event, Point.Builder point @@ -156,28 +87,6 @@ public class InfluxStore { }); } - - /** - * Returns a list of the runtime names that are missing within the event - */ - private List<String> getMissingProperties( - List<EventProperty> allEventProperties, - Event event - ) { - return allEventProperties.stream() - .map(EventProperty::getRuntimeName) - .filter(runtimeName -> event.getOptionalFieldByRuntimeName(runtimeName) - .isEmpty()) - .collect(Collectors.toList()); - } - - private void checkEventIsNotNull(Event event) { - if (event == null) { - throw new SpRuntimeException("event is null"); - } - } - - /** * Shuts down the connection to the InfluxDB server */ @@ -191,7 +100,6 @@ public class InfluxStore { influxDb.close(); } - /** * Creates a point object which is later written to the influxDB and adds the value of the timestamp field */ @@ -206,7 +114,7 @@ public class InfluxStore { /** * store sanitized target property runtime names in local variable */ - private void storeSanitizedRuntimeNames(DataLakeMeasure measure) { + protected void storeSanitizedRuntimeNames() { measure.getEventSchema() .getEventProperties() .forEach(ep -> sanitizedRuntimeNames.put( @@ -215,22 +123,10 @@ public class InfluxStore { )); } - /** - * Returns all measurements properties except the timestamp field - */ - private List<EventProperty> getAllEventPropertiesExceptTimestamp(DataLakeMeasure dataLakeMeasure) { - return dataLakeMeasure.getEventSchema() - .getEventProperties() - .stream() - .filter(ep -> !dataLakeMeasure.getTimestampField() - .endsWith(ep.getRuntimeName())) - .collect(Collectors.toList()); - } - /** * Iterates over all properties of the event and renames the key if it is a reserved keywords in InfluxDB */ - private void sanitizeRuntimeNamesInEvent(Event event) { + protected void sanitizeRuntimeNamesInEvent(Event event) { // sanitize event for (var key : event.getRaw() .keySet()) { @@ -240,5 +136,4 @@ public class InfluxStore { } } } - } diff --git a/streampipes-data-explorer-influx/src/test/java/org/apache/streampipes/dataexplorer/influx/InfluxStoreTest.java b/streampipes-data-explorer-influx/src/test/java/org/apache/streampipes/dataexplorer/influx/TimeSeriesStorageInfluxTest.java similarity index 97% rename from streampipes-data-explorer-influx/src/test/java/org/apache/streampipes/dataexplorer/influx/InfluxStoreTest.java rename to streampipes-data-explorer-influx/src/test/java/org/apache/streampipes/dataexplorer/influx/TimeSeriesStorageInfluxTest.java index aec50f849d..3ddd6f7007 100644 --- a/streampipes-data-explorer-influx/src/test/java/org/apache/streampipes/dataexplorer/influx/InfluxStoreTest.java +++ b/streampipes-data-explorer-influx/src/test/java/org/apache/streampipes/dataexplorer/influx/TimeSeriesStorageInfluxTest.java @@ -53,7 +53,7 @@ import java.util.concurrent.TimeUnit; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertThrows; -public class InfluxStoreTest { +public class TimeSeriesStorageInfluxTest { private InfluxDB influxDBMock; @@ -300,8 +300,8 @@ public class InfluxStoreTest { /** * Executes the onEvent method and returns the resuting data point using an argument captor */ - private Point executeOnEvent(InfluxStore influxStore, Event event) { - influxStore.onEvent(event); + private Point executeOnEvent(TimeSeriesStorageInflux timeSeriesStorageInflux, Event event) { + timeSeriesStorageInflux.onEvent(event); var pointArgumentCaptor = ArgumentCaptor.forClass(Point.class); Mockito.verify(influxDBMock).write(pointArgumentCaptor.capture()); @@ -374,7 +374,7 @@ public class InfluxStoreTest { /** * Initializes an influx store with the given event schema */ - private InfluxStore getInfluxStore(EventSchema eventSchema) { + private TimeSeriesStorageInflux getInfluxStore(EventSchema eventSchema) { DataLakeMeasure measure = new DataLakeMeasure( EXPECTED_MEASUREMENT, @@ -386,7 +386,7 @@ public class InfluxStoreTest { Mockito.when(influxClientProviderMock.getInitializedInfluxDBClient(ArgumentMatchers.any())) .thenReturn(influxDBMock); - return new InfluxStore(measure, null, influxClientProviderMock); + return new TimeSeriesStorageInflux(measure, null, influxClientProviderMock); } } \ No newline at end of file diff --git a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/ImageStore.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/ImageStore.java similarity index 98% rename from streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/ImageStore.java rename to streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/ImageStore.java index 6cfd2087be..67af69225d 100644 --- a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/ImageStore.java +++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/ImageStore.java @@ -16,7 +16,7 @@ * */ -package org.apache.streampipes.dataexplorer.influx.migrate; +package org.apache.streampipes.dataexplorer; import org.apache.streampipes.commons.environment.Environment; import org.apache.streampipes.commons.exceptions.SpRuntimeException; diff --git a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/ImageStoreUtils.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/ImageStoreUtils.java similarity index 96% rename from streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/ImageStoreUtils.java rename to streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/ImageStoreUtils.java index 6504303a5a..65e777f874 100644 --- a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/ImageStoreUtils.java +++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/ImageStoreUtils.java @@ -16,7 +16,7 @@ * */ -package org.apache.streampipes.dataexplorer.influx.migrate; +package org.apache.streampipes.dataexplorer; import org.apache.streampipes.model.datalake.DataLakeMeasure; import org.apache.streampipes.model.schema.EventProperty; diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/TimeSeriesStorage.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/TimeSeriesStorage.java new file mode 100644 index 0000000000..7273d55b9e --- /dev/null +++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/TimeSeriesStorage.java @@ -0,0 +1,145 @@ +/* + * 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; + +import org.apache.streampipes.commons.exceptions.SpRuntimeException; +import org.apache.streampipes.dataexplorer.api.ITimeSeriesStorage; +import org.apache.streampipes.model.datalake.DataLakeMeasure; +import org.apache.streampipes.model.runtime.Event; +import org.apache.streampipes.model.schema.EventProperty; +import org.apache.streampipes.model.schema.EventPropertyPrimitive; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +public abstract class TimeSeriesStorage implements ITimeSeriesStorage { + + private static final Logger LOG = LoggerFactory.getLogger(TimeSeriesStorage.class); + + protected final DataLakeMeasure measure; + protected final List<EventProperty> allEventProperties; + protected final Map<String, String> sanitizedRuntimeNames = new HashMap<>(); + + public TimeSeriesStorage(DataLakeMeasure measure) { + this.measure = measure; + storeSanitizedRuntimeNames(); + allEventProperties = getAllEventPropertiesExceptTimestamp(); + } + + @Override + public void onEvent(Event event) throws SpRuntimeException { + validateInputEventAndLogMissingFields(event); + sanitizeRuntimeNamesInEvent(event); + writeToTimeSeriesStorage(event); + } + + private void validateInputEventAndLogMissingFields(Event event) { + checkEventIsNotNull(event); + + logMissingFields(event); + + logNullFields(event); + } + + private void checkEventIsNotNull(Event event) { + if (event == null) { + throw new SpRuntimeException("Input event is null"); + } + } + + /** + * Logs all fields which are present in the schema, but not in the provided event + */ + private void logMissingFields(Event event) { + var missingFields = getMissingProperties(allEventProperties, event); + if (!missingFields.isEmpty()) { + LOG.debug( + "Ignored {} fields which were present in the schema, but not in the provided event: {}", + missingFields.size(), + String.join(", ", missingFields) + ); + } + } + + /** + * Returns a list of the runtime names that are missing within the event + */ + private List<String> getMissingProperties( + List<EventProperty> allEventProperties, + Event event + ) { + return allEventProperties.stream() + .map(EventProperty::getRuntimeName) + .filter(runtimeName -> event.getOptionalFieldByRuntimeName(runtimeName) + .isEmpty()) + .toList(); + } + + /** + * Logs all fields that contain null values + */ + private void logNullFields(Event event) { + List<String> nullFields = allEventProperties + .stream() + .filter(EventPropertyPrimitive.class::isInstance) + .filter(ep -> { + var runtimeName = ep.getRuntimeName(); + var field = event.getOptionalFieldByRuntimeName(runtimeName); + + return field.isPresent() && field.get() + .getAsPrimitive() + .getRawValue() == null; + }) + .map(EventProperty::getRuntimeName) + .collect(Collectors.toList()); + + if (!nullFields.isEmpty()) { + LOG.warn("Ignored {} fields which had a value 'null': {}", nullFields.size(), String.join(", ", nullFields)); + } + } + + /** + * Returns all measurements properties except the timestamp field + */ + private List<EventProperty> getAllEventPropertiesExceptTimestamp() { + return measure.getEventSchema() + .getEventProperties() + .stream() + .filter(ep -> !measure.getTimestampField() + .endsWith(ep.getRuntimeName())) + .toList(); + } + + /** + * store sanitized target property runtime names in variable `sanitizedRuntimeNames` + */ + protected abstract void storeSanitizedRuntimeNames(); + + /** + * Iterates over all properties of the event and renames the key if it is a reserved keywords in InfluxDB + */ + protected abstract void sanitizeRuntimeNamesInEvent(Event event); + + protected abstract void writeToTimeSeriesStorage(Event event) throws SpRuntimeException; +} diff --git a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/TimeSeriesStore.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/TimeSeriesStore.java similarity index 74% rename from streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/TimeSeriesStore.java rename to streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/TimeSeriesStore.java index ef8af9a24d..f97847585f 100644 --- a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/TimeSeriesStore.java +++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/TimeSeriesStore.java @@ -16,13 +16,12 @@ * */ -package org.apache.streampipes.dataexplorer.influx.migrate; +package org.apache.streampipes.dataexplorer; 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.influx.client.InfluxClientProvider; -import org.apache.streampipes.dataexplorer.influx.InfluxStore; +import org.apache.streampipes.dataexplorer.api.ITimeSeriesStorage; import org.apache.streampipes.model.datalake.DataLakeMeasure; import org.apache.streampipes.model.runtime.Event; @@ -34,23 +33,22 @@ import java.io.IOException; public class TimeSeriesStore { private static final Logger LOG = LoggerFactory.getLogger(TimeSeriesStore.class); - private final InfluxStore influxStore; + private final ITimeSeriesStorage timeSeriesStorage; private ImageStore imageStore; - public TimeSeriesStore(Environment environment, - IStreamPipesClient client, - DataLakeMeasure measure, - boolean enableImageStore) { - - DataExplorerUtils.sanitizeAndRegisterAtDataLake(client, measure); + public TimeSeriesStore( + ITimeSeriesStorage timeSeriesStorage, + IStreamPipesClient client, + DataLakeMeasure measure, + Environment environment, + boolean enableImageStore + ) { if (enableImageStore) { this.imageStore = new ImageStore(measure, environment); } - - this.influxStore = new InfluxStore(measure, environment, new InfluxClientProvider()); - + this.timeSeriesStorage = timeSeriesStorage; } public boolean onEvent(Event event) throws SpRuntimeException { @@ -60,7 +58,7 @@ public class TimeSeriesStore { } // Store event in time series database - this.influxStore.onEvent(event); + this.timeSeriesStorage.onEvent(event); return true; } @@ -75,6 +73,6 @@ public class TimeSeriesStore { } } - this.influxStore.close(); + this.timeSeriesStorage.close(); } }
