git.basschouten.com Git - openhab-addons.git/blob

   1 /**
   2  * Copyright (c) 2010-2023 Contributors to the openHAB project
   3  *
   4  * See the NOTICE file(s) distributed with this work for additional
   5  * information.
   6  *
   7  * This program and the accompanying materials are made available under the
   8  * terms of the Eclipse Public License 2.0 which is available at
   9  * http://www.eclipse.org/legal/epl-2.0
  10  *
  11  * SPDX-License-Identifier: EPL-2.0
  12  */
  13 package org.openhab.persistence.influxdb.internal.influx1;
  14
  15 import static org.openhab.persistence.influxdb.internal.InfluxDBConstants.COLUMN_TIME_NAME_V1;
  16 import static org.openhab.persistence.influxdb.internal.InfluxDBConstants.COLUMN_VALUE_NAME_V1;
  17 import static org.openhab.persistence.influxdb.internal.InfluxDBConstants.FIELD_VALUE_NAME;
  18 import static org.openhab.persistence.influxdb.internal.InfluxDBConstants.TAG_ITEM_NAME;
  19
  20 import java.time.Instant;
  21 import java.util.ArrayList;
  22 import java.util.Collections;
  23 import java.util.List;
  24 import java.util.Map;
  25 import java.util.Objects;
  26 import java.util.Optional;
  27 import java.util.concurrent.TimeUnit;
  28
  29 import org.eclipse.jdt.annotation.NonNullByDefault;
  30 import org.eclipse.jdt.annotation.Nullable;
  31 import org.influxdb.InfluxDB;
  32 import org.influxdb.InfluxDBException;
  33 import org.influxdb.InfluxDBFactory;
  34 import org.influxdb.dto.BatchPoints;
  35 import org.influxdb.dto.Point;
  36 import org.influxdb.dto.Pong;
  37 import org.influxdb.dto.Query;
  38 import org.influxdb.dto.QueryResult;
  39 import org.openhab.core.persistence.FilterCriteria;
  40 import org.openhab.persistence.influxdb.internal.FilterCriteriaQueryCreator;
  41 import org.openhab.persistence.influxdb.internal.InfluxDBConfiguration;
  42 import org.openhab.persistence.influxdb.internal.InfluxDBMetadataService;
  43 import org.openhab.persistence.influxdb.internal.InfluxDBRepository;
  44 import org.openhab.persistence.influxdb.internal.InfluxPoint;
  45 import org.slf4j.Logger;
  46 import org.slf4j.LoggerFactory;
  47
  48 import com.influxdb.exceptions.InfluxException;
  49
  50 /**
  51  * Implementation of {@link InfluxDBRepository} for InfluxDB 1.0
  52  *
  53  * @author Joan Pujol Espinar - Initial contribution. Most code has been moved
  54  *         from
  55  *         {@link org.openhab.persistence.influxdb.InfluxDBPersistenceService}
  56  *         where it was in previous version
  57  */
  58 @NonNullByDefault
  59 public class InfluxDB1RepositoryImpl implements InfluxDBRepository {
  60     private final Logger logger = LoggerFactory.getLogger(InfluxDB1RepositoryImpl.class);
  61     private final InfluxDBConfiguration configuration;
  62     private final FilterCriteriaQueryCreator queryCreator;
  63     private @Nullable InfluxDB client;
  64
  65     public InfluxDB1RepositoryImpl(InfluxDBConfiguration configuration,
  66             InfluxDBMetadataService influxDBMetadataService) {
  67         this.configuration = configuration;
  68         this.queryCreator = new InfluxDB1FilterCriteriaQueryCreatorImpl(configuration, influxDBMetadataService);
  69     }
  70
  71     @Override
  72     public boolean isConnected() {
  73         return client != null;
  74     }
  75
  76     @Override
  77     public boolean connect() {
  78         final InfluxDB createdClient = InfluxDBFactory.connect(configuration.getUrl(), configuration.getUser(),
  79                 configuration.getPassword());
  80         createdClient.setDatabase(configuration.getDatabaseName());
  81         createdClient.setRetentionPolicy(configuration.getRetentionPolicy());
  82         createdClient.enableBatch(200, 100, TimeUnit.MILLISECONDS);
  83         this.client = createdClient;
  84         return checkConnectionStatus();
  85     }
  86
  87     @Override
  88     public void disconnect() {
  89         final InfluxDB currentClient = client;
  90         if (currentClient != null) {
  91             currentClient.close();
  92         }
  93         this.client = null;
  94     }
  95
  96     @Override
  97     public boolean checkConnectionStatus() {
  98         final InfluxDB currentClient = client;
  99         if (currentClient != null) {
 100             try {
 101                 Pong pong = currentClient.ping();
 102                 String version = pong.getVersion();
 103                 // may be check for version >= 0.9
 104                 if (version != null && !version.contains("unknown")) {
 105                     logger.debug("database status is OK, version is {}", version);
 106                     return true;
 107                 } else {
 108                     logger.warn("database ping error, version is: \"{}\" response time was \"{}\"", version,
 109                             pong.getResponseTime());
 110                 }
 111             } catch (RuntimeException e) {
 112                 logger.warn("database error: {}", e.getMessage(), e);
 113             }
 114         } else {
 115             logger.warn("checkConnection: database is not connected");
 116         }
 117         return false;
 118     }
 119
 120     @Override
 121     public boolean write(List<InfluxPoint> influxPoints) {
 122         final InfluxDB currentClient = this.client;
 123         if (currentClient == null) {
 124             return false;
 125         }
 126         try {
 127             List<Point> points = influxPoints.stream().map(this::convertPointToClientFormat).filter(Optional::isPresent)
 128                     .map(Optional::get).toList();
 129             BatchPoints batchPoints = BatchPoints.database(configuration.getDatabaseName())
 130                     .retentionPolicy(configuration.getRetentionPolicy()).points(points).build();
 131             currentClient.write(batchPoints);
 132         } catch (InfluxException | InfluxDBException e) {
 133             logger.debug("Writing to database failed", e);
 134             return false;
 135         }
 136         return true;
 137     }
 138
 139     @Override
 140     public boolean remove(FilterCriteria filter) {
 141         logger.warn("Removing data is not supported in InfluxDB v1.");
 142         return false;
 143     }
 144
 145     private Optional<Point> convertPointToClientFormat(InfluxPoint point) {
 146         Point.Builder clientPoint = Point.measurement(point.getMeasurementName()).time(point.getTime().toEpochMilli(),
 147                 TimeUnit.MILLISECONDS);
 148         Object value = point.getValue();
 149         if (value instanceof String string) {
 150             clientPoint.addField(FIELD_VALUE_NAME, string);
 151         } else if (value instanceof Number number) {
 152             clientPoint.addField(FIELD_VALUE_NAME, number);
 153         } else if (value instanceof Boolean boolean1) {
 154             clientPoint.addField(FIELD_VALUE_NAME, boolean1);
 155         } else if (value == null) {
 156             clientPoint.addField(FIELD_VALUE_NAME, "null");
 157         } else {
 158             logger.warn("Could not convert {}, discarding this datapoint", point);
 159             return Optional.empty();
 160         }
 161         point.getTags().forEach(clientPoint::tag);
 162         return Optional.of(clientPoint.build());
 163     }
 164
 165     @Override
 166     public List<InfluxRow> query(FilterCriteria filter, String retentionPolicy) {
 167         try {
 168             final InfluxDB currentClient = client;
 169             if (currentClient != null) {
 170                 String query = queryCreator.createQuery(filter, retentionPolicy);
 171                 logger.trace("Query {}", query);
 172                 Query parsedQuery = new Query(query, configuration.getDatabaseName());
 173                 List<QueryResult.Result> results = currentClient.query(parsedQuery, TimeUnit.MILLISECONDS).getResults();
 174                 return convertClientResultToRepository(results);
 175             } else {
 176                 throw new InfluxException("API not present");
 177             }
 178         } catch (InfluxException | InfluxDBException e) {
 179             logger.warn("Failed to execute query '{}': {}", filter, e.getMessage());
 180             return List.of();
 181         }
 182     }
 183
 184     private List<InfluxRow> convertClientResultToRepository(List<QueryResult.Result> results) {
 185         List<InfluxRow> rows = new ArrayList<>();
 186         for (QueryResult.Result result : results) {
 187             List<QueryResult.Series> allSeries = result.getSeries();
 188             if (result.getError() != null) {
 189                 logger.warn("{}", result.getError());
 190                 continue;
 191             }
 192             if (allSeries == null) {
 193                 logger.debug("query returned no series");
 194             } else {
 195                 for (QueryResult.Series series : allSeries) {
 196                     logger.trace("series {}", series);
 197                     String defaultItemName = series.getName();
 198                     List<List<Object>> allValues = series.getValues();
 199                     if (allValues == null) {
 200                         logger.debug("query returned no values");
 201                     } else {
 202                         List<String> columns = series.getColumns();
 203                         logger.trace("columns {}", columns);
 204                         if (columns != null) {
 205                             int timestampColumn = columns.indexOf(COLUMN_TIME_NAME_V1);
 206                             int valueColumn = columns.indexOf(COLUMN_VALUE_NAME_V1);
 207                             int itemNameColumn = columns.indexOf(TAG_ITEM_NAME);
 208                             if (valueColumn == -1 || timestampColumn == -1) {
 209                                 throw new IllegalStateException("missing column");
 210                             }
 211                             for (List<Object> valueObject : allValues) {
 212                                 Double rawTime = (Double) valueObject.get(timestampColumn);
 213                                 Instant time = Instant.ofEpochMilli(rawTime.longValue());
 214                                 Object value = valueObject.get(valueColumn);
 215                                 String itemName = itemNameColumn == -1 ? defaultItemName
 216                                         : Objects.requireNonNullElse((String) valueObject.get(itemNameColumn),
 217                                                 defaultItemName);
 218                                 logger.trace("adding historic item {}: time {} value {}", itemName, time, value);
 219                                 rows.add(new InfluxRow(time, itemName, value));
 220                             }
 221                         }
 222                     }
 223                 }
 224             }
 225         }
 226         return rows;
 227     }
 228
 229     @Override
 230     public Map<String, Integer> getStoredItemsCount() {
 231         return Collections.emptyMap();
 232     }
 233 }