This is an automated email from the ASF dual-hosted git repository. riemer pushed a commit to branch sp-1157-upgrade-third-party-services in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit 530bac0c30adb597c25d664ea513c5000c17d039 Author: Dominik Riemer <[email protected]> AuthorDate: Sun Jan 29 15:53:50 2023 +0100 Refactor code to support Influx 2 (#1157) --- .../deploy/standalone/influxdb/docker-compose.yml | 10 +++- .../apache/streampipes/commons/constants/Envs.java | 38 +++++++++++- .../commons/environment/DefaultEnvironment.java | 54 +++++++++++++++++ .../commons/environment/Environment.java | 32 ++++++++++- .../environment/variable/EnvironmentVariable.java | 12 +++- .../streampipes/config/backend/BackendConfig.java | 4 -- .../dataexplorer/commons/DataExplorerWriter.java | 67 ---------------------- .../dataexplorer/commons/TimeSeriesStore.java | 8 +-- .../dataexplorer/commons/auth/AuthInterceptor.java | 29 ++++++---- .../dataexplorer/commons/image/ImageStore.java | 20 ++++--- .../commons/influx/InfluxClientProvider.java | 66 +++++++++++++++++++++ .../commons/influx/InfluxConnectionSettings.java | 39 ++++++------- .../dataexplorer/commons/influx/InfluxStore.java | 14 ++--- streampipes-data-explorer/pom.xml | 5 ++ .../dataexplorer/DataLakeManagementV4.java | 3 +- .../dataexplorer/query/DataExplorerQuery.java | 4 +- .../dataexplorer/utils/DataExplorerUtils.java | 15 ----- .../dataexplorer/v4/query/DataExplorerQueryV4.java | 6 +- .../sinks/internal/jvm/datalake/DataLakeSink.java | 3 +- 19 files changed, 275 insertions(+), 154 deletions(-) diff --git a/installer/cli/deploy/standalone/influxdb/docker-compose.yml b/installer/cli/deploy/standalone/influxdb/docker-compose.yml index 43f8f6e1a..7ea38aca1 100644 --- a/installer/cli/deploy/standalone/influxdb/docker-compose.yml +++ b/installer/cli/deploy/standalone/influxdb/docker-compose.yml @@ -16,13 +16,20 @@ version: "3.4" services: influxdb: - image: fogsyio/influxdb:1.7 + image: influxdb:2.6 environment: - INFLUXDB_DATA_ENGINE=tsm1 - INFLUXDB_REPORTING_DISABLED=false - INFLUXDB_ADMIN_ENABLED=true + - DOCKER_INFLUXDB_INIT_USERNAME=admin + - DOCKER_INFLUXDB_INIT_PASSWORD=sp-admin + - DOCKER_INFLUXDB_INIT_ADMIN_TOKEN=sp-admin + - DOCKER_INFLUXDB_INIT_ORG=sp + - DOCKER_INFLUXDB_INIT_BUCKET=sp + - DOCKER_INFLUXDB_INIT_MODE=upgrade volumes: - influxdb:/var/lib/influxdb + - influxdb2:/var/lib/influxdb2 logging: driver: "json-file" options: @@ -33,6 +40,7 @@ services: volumes: influxdb: + influxdb2: networks: spnet: diff --git a/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java b/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java index 546050058..067df3dea 100644 --- a/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java +++ b/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java @@ -23,9 +23,9 @@ public enum Envs { SP_PORT("SP_PORT", null), @Deprecated(since = "0.90.0", forRemoval = true) - SP_CONSUL_LOCATION("CONSUL_LOCATION", "consul"), + SP_CONSUL_LOCATION("CONSUL_LOCATION", "consul", "localhost"), - SP_CONSUL_HOST("SP_CONSUL_HOST", "consul"), + SP_CONSUL_HOST("SP_CONSUL_HOST", "consul", "localhost"), SP_CONSUL_PORT("SP_CONSUL_PORT", "8500"), SP_KAFKA_RETENTION_MS("SP_KAFKA_RETENTION_MS", null), SP_JWT_SECRET("JWT_SECRET", null), @@ -42,14 +42,42 @@ public enum Envs { SP_CLIENT_SECRET("SP_CLIENT_SECRET", null), SP_ENCRYPTION_PASSCODE("SP_ENCRYPTION_PASSCODE", null), SP_DEBUG("SP_DEBUG", "false"), - SP_MAX_WAIT_TIME_AT_SHUTDOWN("SP_MAX_WAIT_TIME_AT_SHUTDOWN", null); + SP_MAX_WAIT_TIME_AT_SHUTDOWN("SP_MAX_WAIT_TIME_AT_SHUTDOWN", null), + + // CouchDB Storage + SP_COUCHDB_PROTOCOL("SP_COUCHDB_PROTOCOL", "http"), + SP_COUCHDB_HOST("SP_COUCHDB_HOST", "couchdb", "localhost"), + SP_COUCHDB_PORT("SP_COUCHDB_PORT", "5984"), + SP_COUCHDB_USER("SP_COUCHDB_USER", "admin"), + SP_COUCHDB_PASSWORD("SP_COUCHDB_PASSWORD", "admin"), + + + // Time Series Storage + SP_TS_STORAGE_PROTOCOL("SP_TS_STORAGE_PROTOCOL", "http"), + SP_TS_STORAGE_HOST("SP_TS_STORAGE_HOST", "influxdb", "localhost"), + SP_TS_STORAGE_PORT("SP_TS_STORAGE_PORT", "8086"), + + SP_TS_STORAGE_TOKEN("SP_TS_STORAGE_TOKEN", "sp-admin"), + + SP_TS_STORAGE_ORG("SP_TS_STORAGE_ORG", "sp"), + + SP_TS_STORAGE_BUCKET("SP_TS_STORAGE_BUCKET", "sp"); private final String envVariableName; private final String defaultValue; + private final String devDefaultValue; + + Envs(String envVariableName, String defaultValue, String devDefaultValue) { + this.envVariableName = envVariableName; + this.defaultValue = defaultValue; + this.devDefaultValue = devDefaultValue; + } + Envs(String envVariableName, String defaultValue) { this.envVariableName = envVariableName; this.defaultValue = defaultValue; + this.devDefaultValue = defaultValue; } public boolean exists() { @@ -87,4 +115,8 @@ public enum Envs { public String getDefaultValue() { return defaultValue; } + + public String getDevDefaultValue() { + return devDefaultValue; + } } 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 0c051ef16..8ccf01948 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 @@ -40,6 +40,60 @@ public class DefaultEnvironment implements Environment { return new BooleanEnvironmentVariable(Envs.SP_DEBUG); } + @Override + public StringEnvironmentVariable getTsStorageProtocol() { + return new StringEnvironmentVariable(Envs.SP_TS_STORAGE_PROTOCOL); + } + + @Override + public StringEnvironmentVariable getTsStorageHost() { + return new StringEnvironmentVariable(Envs.SP_TS_STORAGE_HOST); + } + + @Override + public IntEnvironmentVariable getTsStoragePort() { + return new IntEnvironmentVariable(Envs.SP_TS_STORAGE_PORT); + } + + @Override + public StringEnvironmentVariable getTsStorageToken() { + return new StringEnvironmentVariable(Envs.SP_TS_STORAGE_TOKEN); + } + + @Override + public StringEnvironmentVariable getTsStorageOrg() { + return new StringEnvironmentVariable(Envs.SP_TS_STORAGE_ORG); + } + + @Override + public StringEnvironmentVariable getTsStorageBucket() { + return new StringEnvironmentVariable(Envs.SP_TS_STORAGE_BUCKET); + } + + @Override + public StringEnvironmentVariable getCouchDbProtocol() { + return new StringEnvironmentVariable(Envs.SP_COUCHDB_PROTOCOL); + } + @Override + public StringEnvironmentVariable getCouchDbHost() { + return new StringEnvironmentVariable(Envs.SP_COUCHDB_HOST); + } + + @Override + public IntEnvironmentVariable getCouchDbPort() { + return new IntEnvironmentVariable(Envs.SP_COUCHDB_PORT); + } + + @Override + public StringEnvironmentVariable getCouchDbUsername() { + return new StringEnvironmentVariable(Envs.SP_COUCHDB_USER); + } + + @Override + public StringEnvironmentVariable getCouchDbPassword() { + return new StringEnvironmentVariable(Envs.SP_COUCHDB_PASSWORD); + } + @Override public StringEnvironmentVariable getConsulLocation() { return new StringEnvironmentVariable(Envs.SP_CONSUL_LOCATION); diff --git a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/Environment.java b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/Environment.java index b3a3256d2..63be042c3 100644 --- a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/Environment.java +++ b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/Environment.java @@ -24,12 +24,40 @@ import org.apache.streampipes.commons.environment.variable.StringEnvironmentVari public interface Environment { + // Consul env variables StringEnvironmentVariable getConsulHost(); IntEnvironmentVariable getConsulPort(); - BooleanEnvironmentVariable getSpDebug(); - @Deprecated(since = "0.90.0", forRemoval = true) StringEnvironmentVariable getConsulLocation(); + + BooleanEnvironmentVariable getSpDebug(); + + // Time series storage env variables + + StringEnvironmentVariable getTsStorageProtocol(); + + StringEnvironmentVariable getTsStorageHost(); + + IntEnvironmentVariable getTsStoragePort(); + + StringEnvironmentVariable getTsStorageToken(); + + StringEnvironmentVariable getTsStorageOrg(); + + StringEnvironmentVariable getTsStorageBucket(); + + // CouchDB env variables + + StringEnvironmentVariable getCouchDbProtocol(); + + StringEnvironmentVariable getCouchDbHost(); + + IntEnvironmentVariable getCouchDbPort(); + + StringEnvironmentVariable getCouchDbUsername(); + + StringEnvironmentVariable getCouchDbPassword(); + } diff --git a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/variable/EnvironmentVariable.java b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/variable/EnvironmentVariable.java index 769d5fea9..e02031df4 100644 --- a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/variable/EnvironmentVariable.java +++ b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/variable/EnvironmentVariable.java @@ -23,8 +23,9 @@ import org.apache.streampipes.commons.constants.Envs; public abstract class EnvironmentVariable<T> { - private T defaultValue; - private String envVariableName; + private final T defaultValue; + private final String envVariableName; + private boolean devModeActive; public EnvironmentVariable(String envVariableName, T defaultValue) { @@ -34,7 +35,8 @@ public abstract class EnvironmentVariable<T> { public EnvironmentVariable(Envs envVariable) { this.envVariableName = envVariable.getEnvVariableName(); - this.defaultValue = parse(envVariable.getDefaultValue()); + this.devModeActive = isDevModeActive(); + this.defaultValue = devModeActive ? parse(envVariable.getDevDefaultValue()) : parse(envVariable.getDefaultValue()); } public T getValue() { @@ -57,6 +59,10 @@ public abstract class EnvironmentVariable<T> { return resolver.resolve(); } + private boolean isDevModeActive() { + return CustomEnvs.getEnvAsBoolean(Envs.SP_DEBUG.getEnvVariableName()); + } + public abstract T parse(String value); } diff --git a/streampipes-config/src/main/java/org/apache/streampipes/config/backend/BackendConfig.java b/streampipes-config/src/main/java/org/apache/streampipes/config/backend/BackendConfig.java index 1c02d9883..43e602362 100644 --- a/streampipes-config/src/main/java/org/apache/streampipes/config/backend/BackendConfig.java +++ b/streampipes-config/src/main/java/org/apache/streampipes/config/backend/BackendConfig.java @@ -241,10 +241,6 @@ public enum BackendConfig { return config.getInteger(BackendConfigKeys.INFLUX_PORT); } - public String getInfluxUrl() { - return "http://"; + getInfluxHost() + ":" + getInfluxPort(); - } - public String getInfluxDatabaseName() { return config.getString(BackendConfigKeys.INFLUX_DATA_BASE); } diff --git a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/DataExplorerWriter.java b/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/DataExplorerWriter.java deleted file mode 100644 index 022bc75d8..000000000 --- a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/DataExplorerWriter.java +++ /dev/null @@ -1,67 +0,0 @@ -/* - * 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.commons; - -import org.apache.streampipes.dataexplorer.commons.configs.DataExplorerConfigurations; -import org.apache.streampipes.dataexplorer.commons.influx.InfluxConnectionSettings; - -import org.influxdb.InfluxDB; -import org.influxdb.InfluxDBFactory; -import org.influxdb.dto.Point; - -import java.util.Map; -import java.util.concurrent.TimeUnit; - -@Deprecated -public class DataExplorerWriter { - private InfluxDB influxDB; - - // TODO return a connection here - public void connect(InfluxConnectionSettings dataExplorerConnectionSettings) { - this.influxDB = InfluxDBFactory.connect( - dataExplorerConnectionSettings.getInfluxDbHost() + ":" + dataExplorerConnectionSettings.getInfluxDbPort(), - dataExplorerConnectionSettings.getUser(), dataExplorerConnectionSettings.getPassword()); - this.influxDB.setDatabase(DataExplorerConfigurations.DATA_LAKE_DATABASE_NAME); - } - - public void close() { - this.influxDB.close(); - } - - public void write(Map<String, Object> data, - String measurement) { - Point.Builder builder = Point.measurement(measurement) - .time((Long) data.get("timestamp"), TimeUnit.MILLISECONDS); - - data.remove("timestamp"); - - for (String key : data.keySet()) { - if (data.get(key) instanceof Double || data.get(key) == null) { - builder.addField(key, (Double) data.get(key)); - } else if (data.get(key) instanceof Integer) { - builder.addField(key, (Integer) data.get(key)); - } else { - builder.tag(key, (String) data.get(key)); - } - } - - this.influxDB.write(builder.build()); - } - -} diff --git a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/TimeSeriesStore.java b/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/TimeSeriesStore.java index 83b31be88..687f13961 100644 --- a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/TimeSeriesStore.java +++ b/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/TimeSeriesStore.java @@ -19,12 +19,12 @@ package org.apache.streampipes.dataexplorer.commons; import org.apache.streampipes.client.StreamPipesClient; +import org.apache.streampipes.commons.environment.Environment; import org.apache.streampipes.commons.exceptions.SpRuntimeException; import org.apache.streampipes.dataexplorer.commons.image.ImageStore; import org.apache.streampipes.dataexplorer.commons.influx.InfluxStore; import org.apache.streampipes.model.datalake.DataLakeMeasure; import org.apache.streampipes.model.runtime.Event; -import org.apache.streampipes.svcdiscovery.api.SpConfig; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -38,7 +38,7 @@ public class TimeSeriesStore { private ImageStore imageStore; - public TimeSeriesStore(SpConfig config, + public TimeSeriesStore(Environment environment, StreamPipesClient client, DataLakeMeasure measure, boolean enableImageStore) { @@ -47,10 +47,10 @@ public class TimeSeriesStore { if (enableImageStore) { // TODO check if event properties are replaces correctly - this.imageStore = new ImageStore(measure, config); + this.imageStore = new ImageStore(measure, environment); } - this.influxStore = new InfluxStore(measure, config); + this.influxStore = new InfluxStore(measure, environment); } diff --git a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/Environment.java b/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/auth/AuthInterceptor.java similarity index 57% copy from streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/Environment.java copy to streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/auth/AuthInterceptor.java index b3a3256d2..52c37c4b2 100644 --- a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/Environment.java +++ b/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/auth/AuthInterceptor.java @@ -16,20 +16,29 @@ * */ -package org.apache.streampipes.commons.environment; +package org.apache.streampipes.dataexplorer.commons.auth; -import org.apache.streampipes.commons.environment.variable.BooleanEnvironmentVariable; -import org.apache.streampipes.commons.environment.variable.IntEnvironmentVariable; -import org.apache.streampipes.commons.environment.variable.StringEnvironmentVariable; +import okhttp3.Interceptor; +import okhttp3.Response; +import org.jetbrains.annotations.NotNull; -public interface Environment { +import java.io.IOException; - StringEnvironmentVariable getConsulHost(); +public class AuthInterceptor implements Interceptor { - IntEnvironmentVariable getConsulPort(); + private String token; - BooleanEnvironmentVariable getSpDebug(); + public AuthInterceptor(String token) { + this.token = token; + } - @Deprecated(since = "0.90.0", forRemoval = true) - StringEnvironmentVariable getConsulLocation(); + @NotNull + @Override + public Response intercept(@NotNull Chain chain) throws IOException { + var req = chain.request(); + var authHeaderValue = "Token " + token; + req = req.newBuilder().addHeader("Authorization", authHeaderValue).build(); + + return chain.proceed(req); + } } diff --git a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/image/ImageStore.java b/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/image/ImageStore.java index 66be08389..e9323d111 100644 --- a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/image/ImageStore.java +++ b/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/image/ImageStore.java @@ -18,12 +18,11 @@ package org.apache.streampipes.dataexplorer.commons.image; +import org.apache.streampipes.commons.environment.Environment; import org.apache.streampipes.commons.exceptions.SpRuntimeException; -import org.apache.streampipes.dataexplorer.commons.configs.CouchDbEnvKeys; 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.svcdiscovery.api.SpConfig; import org.apache.commons.codec.binary.Base64; import org.lightcouch.CouchDbClient; @@ -44,18 +43,21 @@ public class ImageStore { private List<EventProperty> imageProperties; private CouchDbClient couchDbClient; - public ImageStore(DataLakeMeasure measure, SpConfig config) { - this.couchDbClient = new CouchDbClient(from(config)); + public ImageStore(DataLakeMeasure measure, + Environment environment) { + this.couchDbClient = new CouchDbClient(from(environment)); this.imageProperties = ImageStoreUtils.getImageProperties(measure); } - private static CouchDbProperties from(SpConfig config) { - String couchDbProtocol = config.getString(CouchDbEnvKeys.COUCHDB_PROTOCOL); - String couchDbHost = config.getString(CouchDbEnvKeys.COUCHDB_HOST); - int couchDbPort = config.getInteger(CouchDbEnvKeys.COUCHDB_PORT); + private static CouchDbProperties from(Environment env) { + String couchDbProtocol = env.getCouchDbProtocol().getValueOrDefault(); + String couchDbHost = env.getCouchDbHost().getValueOrDefault(); + int couchDbPort = env.getCouchDbPort().getValueOrDefault(); + String username = env.getCouchDbUsername().getValueOrDefault(); + String password = env.getCouchDbPassword().getValueOrDefault(); return new CouchDbProperties(DB_NAME, true, couchDbProtocol, - couchDbHost, couchDbPort, null, null); + couchDbHost, couchDbPort, username, password); } public void onEvent(Event event) throws SpRuntimeException { diff --git a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxClientProvider.java b/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxClientProvider.java new file mode 100644 index 000000000..455552bb1 --- /dev/null +++ b/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxClientProvider.java @@ -0,0 +1,66 @@ +/* + * 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.commons.influx; + +import org.apache.streampipes.commons.environment.Environment; +import org.apache.streampipes.commons.environment.Environments; +import org.apache.streampipes.dataexplorer.commons.auth.AuthInterceptor; + +import okhttp3.OkHttpClient; +import org.influxdb.InfluxDB; +import org.influxdb.InfluxDBFactory; + +import java.util.concurrent.TimeUnit; + +public class InfluxClientProvider { + + /** + * Create a new InfluxDB client from environment variables + * + * @return InfluxDB + */ + public static InfluxDB getInfluxDBClient() { + var env = getEnvironment(); + return getInfluxDBClient(InfluxConnectionSettings.from(env)); + } + + /** + * Create a new InfluxDB client from provided settings + * + * @param settings Connection settings + * @return InfluxDB + */ + public static InfluxDB getInfluxDBClient(InfluxConnectionSettings settings) { + OkHttpClient.Builder okHttpClientBuilder = getHttpClientBuilder(settings.getToken()); + + return InfluxDBFactory.connect(settings.getConnectionUrl(), okHttpClientBuilder); + } + + private static OkHttpClient.Builder getHttpClientBuilder(String authToken) { + return new OkHttpClient().newBuilder() + .addInterceptor(new AuthInterceptor(authToken)) + .connectTimeout(120, TimeUnit.SECONDS) + .readTimeout(120, TimeUnit.SECONDS) + .writeTimeout(120, TimeUnit.SECONDS); + } + + private static Environment getEnvironment() { + return Environments.getEnvironment(); + } +} diff --git a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxConnectionSettings.java b/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxConnectionSettings.java index 0716a9bf7..d94f41825 100644 --- a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxConnectionSettings.java +++ b/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxConnectionSettings.java @@ -18,38 +18,37 @@ package org.apache.streampipes.dataexplorer.commons.influx; -import org.apache.streampipes.dataexplorer.commons.configs.DataExplorerEnvKeys; -import org.apache.streampipes.svcdiscovery.api.SpConfig; +import org.apache.streampipes.commons.environment.Environment; public class InfluxConnectionSettings { private final Integer influxDbPort; + + private final String influxDbProtocol; private final String influxDbHost; private final String databaseName; - private final String user; - private final String password; + private final String token; - private InfluxConnectionSettings(String influxDbHost, + private InfluxConnectionSettings(String influxDbProtocol, + String influxDbHost, Integer influxDbPort, String databaseName, - String user, - String password) { + String token) { + this.influxDbProtocol = influxDbProtocol; this.influxDbHost = influxDbHost; this.influxDbPort = influxDbPort; this.databaseName = databaseName; - this.user = user; - this.password = password; + this.token = token; } - public static InfluxConnectionSettings from(SpConfig configStore) { + public static InfluxConnectionSettings from(Environment environment) { return new InfluxConnectionSettings( - configStore.getString(DataExplorerEnvKeys.DATA_LAKE_PROTOCOL) + "://" - + configStore.getString(DataExplorerEnvKeys.DATA_LAKE_HOST), - configStore.getInteger(DataExplorerEnvKeys.DATA_LAKE_PORT), - configStore.getString(DataExplorerEnvKeys.DATA_LAKE_DATABASE_NAME), - configStore.getString(DataExplorerEnvKeys.DATA_LAKE_USERNAME), - configStore.getString(DataExplorerEnvKeys.DATA_LAKE_PASSWORD)); + environment.getTsStorageProtocol().getValueOrDefault(), + environment.getTsStorageHost().getValueOrDefault(), + environment.getTsStoragePort().getValueOrDefault(), + environment.getTsStorageBucket().getValueOrDefault(), + environment.getTsStorageToken().getValueOrDefault()); } public Integer getInfluxDbPort() { @@ -64,11 +63,11 @@ public class InfluxConnectionSettings { return databaseName; } - public String getUser() { - return user; + public String getToken() { + return token; } - public String getPassword() { - return password; + public String getConnectionUrl() { + return influxDbProtocol + "://" + influxDbHost + ":" + influxDbPort; } } diff --git a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxStore.java b/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxStore.java index 7ef23ca0e..481aacc1c 100644 --- a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxStore.java +++ b/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxStore.java @@ -18,6 +18,7 @@ package org.apache.streampipes.dataexplorer.commons.influx; +import org.apache.streampipes.commons.environment.Environment; import org.apache.streampipes.commons.exceptions.SpRuntimeException; import org.apache.streampipes.model.datalake.DataLakeMeasure; import org.apache.streampipes.model.runtime.Event; @@ -25,11 +26,9 @@ import org.apache.streampipes.model.runtime.field.PrimitiveField; import org.apache.streampipes.model.schema.EventProperty; import org.apache.streampipes.model.schema.EventPropertyPrimitive; import org.apache.streampipes.model.schema.PropertyScope; -import org.apache.streampipes.svcdiscovery.api.SpConfig; import org.apache.streampipes.vocabulary.XSD; import org.influxdb.InfluxDB; -import org.influxdb.InfluxDBFactory; import org.influxdb.dto.Point; import org.influxdb.dto.Pong; import org.influxdb.dto.Query; @@ -64,8 +63,8 @@ public class InfluxStore { } public InfluxStore(DataLakeMeasure measure, - SpConfig configStore) throws SpRuntimeException { - this(measure, InfluxConnectionSettings.from(configStore)); + Environment environment) throws SpRuntimeException { + this(measure, InfluxConnectionSettings.from(environment)); } /** @@ -75,15 +74,12 @@ public class InfluxStore { * be found */ private void connect(InfluxConnectionSettings settings) throws SpRuntimeException { - // Connecting to the server - // "http://"; must be in front - String urlAndPort = settings.getInfluxDbHost() + ":" + settings.getInfluxDbPort(); - influxDb = InfluxDBFactory.connect(urlAndPort, settings.getUser(), settings.getPassword()); + influxDb = InfluxClientProvider.getInfluxDBClient(settings); // Checking, if server is available Pong response = influxDb.ping(); if (response.getVersion().equalsIgnoreCase("unknown")) { - throw new SpRuntimeException("Could not connect to InfluxDb Server: " + urlAndPort); + throw new SpRuntimeException("Could not connect to InfluxDb Server: " + settings.getConnectionUrl()); } String databaseName = settings.getDatabaseName(); diff --git a/streampipes-data-explorer/pom.xml b/streampipes-data-explorer/pom.xml index f03ffaad9..0f8c4aa37 100644 --- a/streampipes-data-explorer/pom.xml +++ b/streampipes-data-explorer/pom.xml @@ -34,6 +34,11 @@ <artifactId>streampipes-config</artifactId> <version>0.91.0-SNAPSHOT</version> </dependency> + <dependency> + <groupId>org.apache.streampipes</groupId> + <artifactId>streampipes-data-explorer-commons</artifactId> + <version>0.91.0-SNAPSHOT</version> + </dependency> <dependency> <groupId>org.apache.streampipes</groupId> <artifactId>streampipes-model</artifactId> diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java index 6928f940e..e02a2f84f 100644 --- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java +++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java @@ -19,6 +19,7 @@ package org.apache.streampipes.dataexplorer; import org.apache.streampipes.config.backend.BackendConfig; +import org.apache.streampipes.dataexplorer.commons.influx.InfluxClientProvider; import org.apache.streampipes.dataexplorer.param.RetentionPolicyQueryParams; import org.apache.streampipes.dataexplorer.query.DeleteDataQuery; import org.apache.streampipes.dataexplorer.query.EditRetentionPolicyQuery; @@ -182,7 +183,7 @@ public class DataLakeManagementV4 { public Map<String, Object> getTagValues(String measurementId, String fields) { - InfluxDB influxDB = DataExplorerUtils.getInfluxDBClient(); + InfluxDB influxDB = InfluxClientProvider.getInfluxDBClient(); Map<String, Object> tags = new HashMap<>(); if (fields != null && !("".equals(fields))) { List<String> fieldList = Arrays.asList(fields.split(",")); diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataExplorerQuery.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataExplorerQuery.java index d34b5bba4..3bd5140e0 100644 --- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataExplorerQuery.java +++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataExplorerQuery.java @@ -18,7 +18,7 @@ package org.apache.streampipes.dataexplorer.query; import org.apache.streampipes.config.backend.BackendConfig; -import org.apache.streampipes.dataexplorer.utils.DataExplorerUtils; +import org.apache.streampipes.dataexplorer.commons.influx.InfluxClientProvider; import org.apache.streampipes.model.datalake.DataSeries; import org.apache.streampipes.model.datalake.SpQueryResult; @@ -32,7 +32,7 @@ import java.util.List; public abstract class DataExplorerQuery<T> { public T executeQuery() throws RuntimeException { - InfluxDB influxDB = DataExplorerUtils.getInfluxDBClient(); + InfluxDB influxDB = InfluxClientProvider.getInfluxDBClient(); DataExplorerQueryBuilder queryBuilder = DataExplorerQueryBuilder.create(BackendConfig.INSTANCE.getInfluxDatabaseName()); getQuery(queryBuilder); diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/utils/DataExplorerUtils.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/utils/DataExplorerUtils.java index 31061ec66..a0b512565 100644 --- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/utils/DataExplorerUtils.java +++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/utils/DataExplorerUtils.java @@ -17,16 +17,10 @@ */ package org.apache.streampipes.dataexplorer.utils; -import org.apache.streampipes.config.backend.BackendConfig; import org.apache.streampipes.model.datalake.DataLakeMeasure; import org.apache.streampipes.storage.management.StorageDispatcher; -import okhttp3.OkHttpClient; -import org.influxdb.InfluxDB; -import org.influxdb.InfluxDBFactory; - import java.util.List; -import java.util.concurrent.TimeUnit; public class DataExplorerUtils { @@ -36,13 +30,4 @@ public class DataExplorerUtils { .getDataLakeStorage() .getAllDataLakeMeasures(); } - - public static InfluxDB getInfluxDBClient() { - OkHttpClient.Builder okHttpClientBuilder = new OkHttpClient().newBuilder() - .connectTimeout(120, TimeUnit.SECONDS) - .readTimeout(120, TimeUnit.SECONDS) - .writeTimeout(120, TimeUnit.SECONDS); - - return InfluxDBFactory.connect(BackendConfig.INSTANCE.getInfluxUrl(), okHttpClientBuilder); - } } diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/DataExplorerQueryV4.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/DataExplorerQueryV4.java index 7d4af4071..05d4f7987 100644 --- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/DataExplorerQueryV4.java +++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/DataExplorerQueryV4.java @@ -19,7 +19,7 @@ package org.apache.streampipes.dataexplorer.v4.query; import org.apache.streampipes.config.backend.BackendConfig; -import org.apache.streampipes.dataexplorer.utils.DataExplorerUtils; +import org.apache.streampipes.dataexplorer.commons.influx.InfluxClientProvider; import org.apache.streampipes.dataexplorer.v4.params.DeleteFromStatementParams; import org.apache.streampipes.dataexplorer.v4.params.FillParams; import org.apache.streampipes.dataexplorer.v4.params.GroupingByTagsParams; @@ -88,7 +88,7 @@ public class DataExplorerQueryV4 { } public SpQueryResult executeQuery(boolean ignoreMissingValues) throws RuntimeException { - InfluxDB influxDB = DataExplorerUtils.getInfluxDBClient(); + InfluxDB influxDB = InfluxClientProvider.getInfluxDBClient(); List<QueryElement<?>> queryElements = getQueryElements(); if (this.maximumAmountOfEvents != -1) { @@ -119,7 +119,7 @@ public class DataExplorerQueryV4 { } public SpQueryResult executeQuery(Query query, boolean ignoreMissingValues) { - InfluxDB influxDB = DataExplorerUtils.getInfluxDBClient(); + InfluxDB influxDB = InfluxClientProvider.getInfluxDBClient(); var dataResult = executeQuery(influxDB, query, ignoreMissingValues); influxDB.close(); diff --git a/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/DataLakeSink.java b/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/DataLakeSink.java index 0088f916e..e23dd1879 100644 --- a/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/DataLakeSink.java +++ b/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/DataLakeSink.java @@ -18,6 +18,7 @@ package org.apache.streampipes.sinks.internal.jvm.datalake; +import org.apache.streampipes.commons.environment.Environments; import org.apache.streampipes.commons.exceptions.SpRuntimeException; import org.apache.streampipes.dataexplorer.commons.TimeSeriesStore; import org.apache.streampipes.model.DataSinkType; @@ -69,7 +70,7 @@ public class DataLakeSink extends StreamPipesDataSink { DataLakeMeasure measure = new DataLakeMeasure(measureName, timestampField, eventSchema); - this.timeSeriesStore = new TimeSeriesStore(runtimeContext.getConfigStore().getConfig(), + this.timeSeriesStore = new TimeSeriesStore(Environments.getEnvironment(), runtimeContext.getStreamPipesClient(), measure, true);
