InfluxDBConnector.java
package de.dlr.shepard.data.timeseries.migration.influxtimeseries;
import de.dlr.shepard.common.util.Constants;
import de.dlr.shepard.common.util.IConnector;
import io.quarkus.logging.Log;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.inject.Inject;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import okhttp3.OkHttpClient;
import org.eclipse.microprofile.config.ConfigProvider;
import org.influxdb.BatchOptions;
import org.influxdb.InfluxDB;
import org.influxdb.InfluxDBException;
import org.influxdb.InfluxDBFactory;
import org.influxdb.dto.Pong;
import org.influxdb.dto.Query;
import org.influxdb.dto.QueryResult;
/**
* Connector for read and write access to the Influx timeseries database. The
* class represents the lowest level of data access to the Influx database. The
* Influx database is accessed directly by using query strings.
*/
@ApplicationScoped
public class InfluxDBConnector implements IConnector {
private InfluxDB influxDB;
/* Use this constructor for testing purpose only.
* As soon as influxDB is provided via dependency injection, this constructor can be used for that purpose. */
InfluxDBConnector(InfluxDB influxDb) {
this.influxDB = influxDb;
}
@Inject
InfluxDBConnector() {}
/**
* Establishes a connection to the Influx server by using the URL saved in the
* config.properties file returned by the DatabaseHelper. In addition the
* logging is being configured.
*
*/
@Override
public boolean connect() {
if (this.influxDB == null) {
String host = ConfigProvider.getConfig().getValue("influx.host", String.class);
String username = ConfigProvider.getConfig().getValue("influx.username", String.class);
String password = ConfigProvider.getConfig().getValue("influx.password", String.class);
OkHttpClient.Builder okHttpClientBuilder = new OkHttpClient.Builder()
.connectTimeout(60, TimeUnit.SECONDS)
.readTimeout(600, TimeUnit.SECONDS);
influxDB = InfluxDBFactory.connect(String.format("http://%s", host), username, password, okHttpClientBuilder);
}
influxDB.enableBatch(
BatchOptions.DEFAULTS.exceptionHandler((failedPoints, throwable) ->
Log.errorf("Exception while writing the following points: %s, Exception: %s", failedPoints, throwable)
)
);
return true;
}
@Override
public boolean disconnect() {
if (influxDB != null) influxDB.close();
return true;
}
/**
* Tests whether a connection exists.
*
* @return True if connection exists, otherwise false.
*/
@Override
public boolean alive() {
Pong response;
try {
response = influxDB.ping();
} catch (InfluxDBException ex) {
return false;
}
return response != null && !response.getVersion().equalsIgnoreCase("unknown");
}
/**
* Creates a new database.
*
* @param databaseName Name of the new database to be created.
*/
public void createDatabase(String databaseName) {
String query = String.format("CREATE DATABASE \"%s\"", databaseName);
influxDB.query(new Query(query));
}
/**
* Deletes a database.
*
* @param databaseName Name of the database to be deleted.
*/
public void deleteDatabase(String databaseName) {
String query = String.format("DROP DATABASE \"%s\"", databaseName);
influxDB.query(new Query(query));
}
/**
* Writes all InfluxPoint objects, saved in the TimeseriesPayload object, into
* the influx database. Therefore the method uses the name of the database, the
* name of the measurement, the location, the device, the symbolic name and the
* name of the field provided by the given TimeseriesPayload object. The actual
* write operation is done by using the unix timestamp in nanoseconds and the
* value of every InfluxPoint object in the TimeseriesPayload object. All these
* variables have to be defined in the given Timeseries object for a successful
* write operation.
*
* @param database The database to store the payload in
* @param payload Combines the required attributes in a structured way.
* @return An error if there was a problem, empty string if all went well
*/
public String saveTimeseriesPayload(String database, InfluxTimeseriesPayload payload) {
var timeseries = payload.getTimeseries();
var expectedType = getExpectedDatatype(database, timeseries.getMeasurement(), timeseries.getField());
var batchPoints = InfluxUtil.createBatch(database, payload, expectedType);
try {
influxDB.write(batchPoints);
} catch (InfluxDBException e) {
Log.errorf("InfluxdbException while writing payload %s: %s", payload.getTimeseries(), e.getMessage());
return e.getMessage();
}
return "";
}
/**
* Returns a Timeseries object containing all the influx data points which match
* the given conditions consigned to the method. The returned InfluxPoint
* objects are in the given measurement of the specific database between the
* start and the end timestamp.
*
* @param startTimeStamp Start timestamp from which values should be returned
* @param endTimeStamp End timestamp to which values should be returned
* @param database Name of the database
* @param timeseries Specifies the data to load from database
* @param function The aggregate function
* @param groupBy The time interval measurements get grouped by
* @param fillOption The fill option for missing values
* @return A Timeseries object containing a list of all InfluxPoints in the
* given measurement of the specific database between the two given
* timestamps matching the "device"-tag, the "location"-tag and the
* "symbolic_name"-tag.
*/
public InfluxTimeseriesPayload getTimeseriesPayload(
long startTimeStamp,
long endTimeStamp,
String database,
InfluxTimeseries timeseries,
InfluxSingleValuedUnaryFunction function,
Long groupBy,
InfluxFillOption fillOption
) {
Query query = InfluxUtil.buildQuery(
startTimeStamp,
endTimeStamp,
database,
timeseries,
function,
groupBy,
fillOption
);
Log.debugf("Influx Query: %s", query.getCommand());
QueryResult queryResult;
queryResult = influxDB.query(query);
if (InfluxUtil.isQueryResultValid(queryResult)) {
return InfluxUtil.extractPayload(queryResult, timeseries);
}
return new InfluxTimeseriesPayload(timeseries, Collections.emptyList());
}
/**
* Returns a list of timeseries objects that are in the given database.
*
* @param database the given database
* @return a list of timeseries objects
*/
public List<InfluxTimeseries> getTimeseriesAvailable(String database) {
Query query = new Query(String.format("SHOW SERIES ON \"%s\"", database));
QueryResult queryResult = influxDB.query(query);
if (!InfluxUtil.isQueryResultValid(queryResult)) {
Log.warn("There was an error while querying the Influxdb for available timeseries");
return Collections.emptyList();
}
var values = queryResult.getResults().get(0).getSeries().get(0).getValues();
var fields = getFields(database);
var result = new ArrayList<InfluxTimeseries>(values.size());
for (var value : values) {
var series = ((String) value.get(0)).split(",");
// The first entry is the measurement
var meas = series[0];
var tags = extractTags(series);
var dev = tags.getOrDefault(Constants.DEVICE, "");
var loc = tags.getOrDefault(Constants.LOCATION, "");
var symName = tags.getOrDefault(Constants.SYMBOLICNAME, "");
for (var field : fields.getOrDefault(meas, Collections.emptyList())) {
result.add(new InfluxTimeseries(meas, dev, loc, symName, field));
}
}
return result;
}
private Map<String, String> extractTags(String[] series) {
var result = new HashMap<String, String>();
for (var tagString : series) {
var tags = tagString.split("=", 2);
// Skip entries with less than 2 tuples (most likely the measurement)
if (tags.length < 2) continue;
result.put(tags[0], tags[1]);
}
return result;
}
private Map<String, List<String>> getFields(String database) {
Query query = new Query(String.format("SHOW FIELD KEYS ON \"%s\"", database));
QueryResult queryResult = influxDB.query(query);
if (!InfluxUtil.isQueryResultValid(queryResult)) {
Log.warn("There was an error while querying the Influxdb for available fields");
return Collections.emptyMap();
}
var series = queryResult.getResults().get(0).getSeries();
var result = new HashMap<String, List<String>>();
for (var s : series) {
var fields = new ArrayList<String>();
for (var value : s.getValues()) {
fields.add((String) value.get(0));
}
result.put(s.getName(), fields);
}
return result;
}
public boolean databaseExist(String database) {
QueryResult queryResult = influxDB.query(new Query("SHOW DATABASES"));
if (!InfluxUtil.isQueryResultValid(queryResult)) {
Log.warn("There was an error while querying the Influxdb for databases");
return false;
}
var values = queryResult.getResults().get(0).getSeries().get(0).getValues();
for (var databaseName : values) {
if (databaseName.get(0).toString().trim().equals(database)) {
return true;
}
}
return false;
}
public InfluxTimeseriesDataType getTimeseriesDataType(String database, String measurement, String field)
throws Exception {
var dataTypeAsString = getExpectedDatatype(database, measurement, field);
try {
return InfluxTimeseriesDataType.valueOf(dataTypeAsString.toUpperCase());
} catch (IllegalArgumentException ex) {
var message = String.format("Timeseries in influxDB has unknown data type: %s", dataTypeAsString);
throw new Exception(message, ex);
}
}
/**
* Returns the expected data type of a field or an empty string according to
* https://docs.influxdata.com/influxdb/v1.8/write_protocols/line_protocol_reference/#data-types
*/
private String getExpectedDatatype(String database, String measurement, String field) {
String queryString = String.format("SHOW FIELD KEYS ON \"%s\" FROM \"%s\"", database, measurement);
QueryResult result = influxDB.query(new Query(queryString));
if (!InfluxUtil.isQueryResultValid(result)) {
Log.infof("Could not get expected datatype query string \"%s\"", queryString);
return "";
}
var values = result.getResults().get(0).getSeries().get(0).getValues();
for (var value : values) {
if (value.get(0).equals(field)) return (String) value.get(1);
}
return "";
}
}