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 a3d5c245b7ec3520e43612733d5f706889c6646a Author: bossenti <[email protected]> AuthorDate: Tue Apr 30 11:20:35 2024 +0200 refactor: abstract sanitation --- .../api/IDataLakeMeasurementSanitizer.java | 48 +++++++++ .../DataLakeMeasurementSanitizerInflux.java | 44 ++++++++ .../DataLakeMeasurementSanitizerInfluxTest.java | 93 +++++++++++++++++ .../dataexplorer/DataLakeMeasurementSanitizer.java | 115 +++++++++++++++++++++ 4 files changed, 300 insertions(+) diff --git a/streampipes-data-explorer-api/src/main/java/org/apache/streampipes/dataexplorer/api/IDataLakeMeasurementSanitizer.java b/streampipes-data-explorer-api/src/main/java/org/apache/streampipes/dataexplorer/api/IDataLakeMeasurementSanitizer.java new file mode 100644 index 0000000000..706f66ce38 --- /dev/null +++ b/streampipes-data-explorer-api/src/main/java/org/apache/streampipes/dataexplorer/api/IDataLakeMeasurementSanitizer.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.model.datalake.DataLakeMeasure; + +/** + * The IDataLakeMeasurementSanitizer interface defines methods for sanitizing and registering or + * updating data lake measures. + * Implementations of this interface provide functionality to ensure that the measurement complies to + * the requirements of the underlying time series storage, e.g., to not contain any reserved symbols. + */ +public interface IDataLakeMeasurementSanitizer { + + /** + * Sanitizes and registers a data lake measure. + * This method should perform any necessary data validation and cleanup operations + * before registering the measure in the data lake. + * + * @return The sanitized and registered data lake measure. + */ + DataLakeMeasure sanitizeAndRegister(); + + /** + * Sanitizes and updates a data lake measure. + * This method should perform any necessary data validation and cleanup operations + * before updating the measure in the data lake. + * + * @return The sanitized and updated data lake measure. + */ + DataLakeMeasure sanitizeAndUpdate(); + +} diff --git a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/sanitize/DataLakeMeasurementSanitizerInflux.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/sanitize/DataLakeMeasurementSanitizerInflux.java new file mode 100644 index 0000000000..29fd6f4ccc --- /dev/null +++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/sanitize/DataLakeMeasurementSanitizerInflux.java @@ -0,0 +1,44 @@ +/* + * 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.sanitize; + +import org.apache.streampipes.client.api.IStreamPipesClient; +import org.apache.streampipes.commons.exceptions.SpRuntimeException; +import org.apache.streampipes.dataexplorer.DataLakeMeasurementSanitizer; +import org.apache.streampipes.model.datalake.DataLakeMeasure; + +public class DataLakeMeasurementSanitizerInflux extends DataLakeMeasurementSanitizer { + public DataLakeMeasurementSanitizerInflux( + IStreamPipesClient client, + DataLakeMeasure measure + ) { + super(client, measure); + } + + @Override + protected void cleanDataLakeMeasure() throws SpRuntimeException { + // 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()))); + } +} diff --git a/streampipes-data-explorer-influx/src/test/java/org/apache/streampipes/dataexplorer/influx/sanitize/DataLakeMeasurementSanitizerInfluxTest.java b/streampipes-data-explorer-influx/src/test/java/org/apache/streampipes/dataexplorer/influx/sanitize/DataLakeMeasurementSanitizerInfluxTest.java new file mode 100644 index 0000000000..5d6cb0542d --- /dev/null +++ b/streampipes-data-explorer-influx/src/test/java/org/apache/streampipes/dataexplorer/influx/sanitize/DataLakeMeasurementSanitizerInfluxTest.java @@ -0,0 +1,93 @@ +/* + * 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.sanitize; + +import org.apache.streampipes.client.api.IDataLakeMeasureApi; +import org.apache.streampipes.client.api.IStreamPipesClient; +import org.apache.streampipes.commons.exceptions.SpRuntimeException; +import org.apache.streampipes.model.datalake.DataLakeMeasure; +import org.apache.streampipes.test.generator.EventPropertyPrimitiveTestBuilder; +import org.apache.streampipes.test.generator.EventSchemaTestBuilder; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +public class DataLakeMeasurementSanitizerInfluxTest { + + private IStreamPipesClient clientMock; + + @BeforeEach + public void setUp() { + clientMock = mock(IStreamPipesClient.class); + + var apiMock = mock(IDataLakeMeasureApi.class); + when(clientMock.dataLakeMeasureApi()).thenReturn(apiMock); + + } + + @Test + public void cleanDataLakeMeasure() { + var measure = new DataLakeMeasure( + "test?Measurement", + "timestamp", + EventSchemaTestBuilder.create() + .withEventProperties(List.of( + EventPropertyPrimitiveTestBuilder.create() + .withRuntimeName("timestamp") + .build(), + EventPropertyPrimitiveTestBuilder.create() + .withRuntimeName("value") + .build(), + EventPropertyPrimitiveTestBuilder.create() + .withRuntimeName("all") + .build() + )) + .build() + ); + + var result = new DataLakeMeasurementSanitizerInflux(clientMock, measure).sanitizeAndRegister(); + + assertEquals("test_Measurement", result.getMeasureName()); + assertEquals(2, result.getEventSchema().getEventProperties().size()); + assertEquals("value", result.getEventSchema().getEventProperties().get(0).getRuntimeName()); + assertEquals("all_", result.getEventSchema().getEventProperties().get(1).getRuntimeName()); + } + + @Test + public void cleanDataLakeMeasureNoTimestampField() { + var measure = new DataLakeMeasure("test", EventSchemaTestBuilder.create() + .withEventProperties(List.of( + EventPropertyPrimitiveTestBuilder.create() + .withRuntimeName( + "value") + .build() + )) + .build()); + + assertThrows(SpRuntimeException.class, + () -> new DataLakeMeasurementSanitizerInflux(clientMock, measure).sanitizeAndRegister()); + } +} diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeMeasurementSanitizer.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeMeasurementSanitizer.java new file mode 100644 index 0000000000..752653e045 --- /dev/null +++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeMeasurementSanitizer.java @@ -0,0 +1,115 @@ +/* + * 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.client.api.IStreamPipesClient; +import org.apache.streampipes.commons.exceptions.SpRuntimeException; +import org.apache.streampipes.dataexplorer.api.IDataLakeMeasurementSanitizer; +import org.apache.streampipes.model.datalake.DataLakeMeasure; +import org.apache.streampipes.model.schema.EventProperty; + +import java.util.List; + +/** + * Base class with shared implementation that is common for all time series storage backends. + * Leaves open the storage specific implementation + */ +public abstract class DataLakeMeasurementSanitizer implements IDataLakeMeasurementSanitizer { + + protected final DataLakeMeasure measure; + protected final IStreamPipesClient client; + + public DataLakeMeasurementSanitizer(IStreamPipesClient client, DataLakeMeasure measure){ + this.client = client; + this.measure = measure; + } + + /** + * Sanitizes the data lake measure and registers it with the data lake. + * <p> + * This method first sanitizes the data lake measure, + * then registers it at the data lake. + * + * @return The sanitized and registered data lake measure. + */ + @Override + public DataLakeMeasure sanitizeAndRegister(){ + sanitizeDataLakeMeasure(); + registerAtDataLake(); + + return measure; + } + + /** + * Sanitizes the data lake measure and updates it in the data lake. + * <p> + * This method first sanitizes the data lake measure, + * then updates it at the data lake. + * + * @return The sanitized and updated data lake measure. + */ + @Override + public DataLakeMeasure sanitizeAndUpdate(){ + sanitizeDataLakeMeasure(); + updateAtDataLake(); + + return measure; + } + + + + private void registerAtDataLake() throws SpRuntimeException { + client.dataLakeMeasureApi().create(measure); + } + + private void updateAtDataLake() throws SpRuntimeException { + client.dataLakeMeasureApi().update(measure); + } + + private void sanitizeDataLakeMeasure() throws SpRuntimeException { + removeTimestampsFromEventSchema(); + cleanDataLakeMeasure(); + } + + /** + * Cleans the data lake measure to ensure compliance with the requirements of the respective time series storage. + * <p> + * This method performs the following steps: + * <ol> + * <li>Sanitizes the name of the measure.</li> + * <li>Sanitizes all runtime names associated with the measure.</li> + * </ol> + * @throws SpRuntimeException if an error occurs during the cleaning process. + */ + protected abstract void cleanDataLakeMeasure() throws SpRuntimeException; + + protected void removeTimestampsFromEventSchema() throws SpRuntimeException{ + var timestampField = measure.getTimestampField(); + + if (timestampField == null){ + throw new SpRuntimeException("Data lake measurement does not have a timestamp field - timestamp field is null."); + } + + List<EventProperty> eventPropertiesWithoutTimestamp = measure.getEventSchema().getEventProperties() + .stream() + .filter(eventProperty -> !timestampField.endsWith(eventProperty.getRuntimeName())) + .toList(); + measure.getEventSchema().setEventProperties(eventPropertiesWithoutTimestamp); + } +}
