You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampipes.apache.org by bo...@apache.org on 2024/04/30 09:34:43 UTC
(streampipes) 06/10: refactor: abstract time series store from Influx
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 <bo...@posteo.de>
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();
}
}