2 * Copyright (c) 2010-2023 Contributors to the openHAB project
4 * See the NOTICE file(s) distributed with this work for additional
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
11 * SPDX-License-Identifier: EPL-2.0
13 package org.openhab.persistence.rrd4j.internal;
15 import java.io.IOException;
16 import java.nio.file.Files;
17 import java.nio.file.Path;
18 import java.time.Instant;
19 import java.time.ZoneId;
20 import java.time.ZonedDateTime;
21 import java.util.ArrayList;
22 import java.util.Collections;
23 import java.util.HashMap;
24 import java.util.Iterator;
25 import java.util.List;
26 import java.util.Locale;
29 import java.util.concurrent.ConcurrentHashMap;
30 import java.util.concurrent.Executors;
31 import java.util.concurrent.RejectedExecutionException;
32 import java.util.concurrent.ScheduledExecutorService;
33 import java.util.concurrent.ScheduledFuture;
34 import java.util.concurrent.TimeUnit;
36 import javax.measure.Quantity;
37 import javax.measure.Unit;
39 import org.eclipse.jdt.annotation.NonNullByDefault;
40 import org.eclipse.jdt.annotation.Nullable;
41 import org.openhab.core.OpenHAB;
42 import org.openhab.core.common.NamedThreadFactory;
43 import org.openhab.core.items.GroupItem;
44 import org.openhab.core.items.Item;
45 import org.openhab.core.items.ItemNotFoundException;
46 import org.openhab.core.items.ItemRegistry;
47 import org.openhab.core.items.ItemUtil;
48 import org.openhab.core.library.CoreItemFactory;
49 import org.openhab.core.library.items.ColorItem;
50 import org.openhab.core.library.items.ContactItem;
51 import org.openhab.core.library.items.DimmerItem;
52 import org.openhab.core.library.items.NumberItem;
53 import org.openhab.core.library.items.RollershutterItem;
54 import org.openhab.core.library.items.SwitchItem;
55 import org.openhab.core.library.types.DecimalType;
56 import org.openhab.core.library.types.OnOffType;
57 import org.openhab.core.library.types.OpenClosedType;
58 import org.openhab.core.library.types.PercentType;
59 import org.openhab.core.library.types.QuantityType;
60 import org.openhab.core.persistence.FilterCriteria;
61 import org.openhab.core.persistence.FilterCriteria.Ordering;
62 import org.openhab.core.persistence.HistoricItem;
63 import org.openhab.core.persistence.PersistenceItemInfo;
64 import org.openhab.core.persistence.PersistenceService;
65 import org.openhab.core.persistence.QueryablePersistenceService;
66 import org.openhab.core.persistence.strategy.PersistenceCronStrategy;
67 import org.openhab.core.persistence.strategy.PersistenceStrategy;
68 import org.openhab.core.types.State;
69 import org.osgi.service.component.annotations.Activate;
70 import org.osgi.service.component.annotations.Component;
71 import org.osgi.service.component.annotations.ConfigurationPolicy;
72 import org.osgi.service.component.annotations.Modified;
73 import org.osgi.service.component.annotations.Reference;
74 import org.rrd4j.ConsolFun;
75 import org.rrd4j.DsType;
76 import org.rrd4j.core.FetchData;
77 import org.rrd4j.core.FetchRequest;
78 import org.rrd4j.core.RrdDb;
79 import org.rrd4j.core.RrdDb.Builder;
80 import org.rrd4j.core.RrdDbPool;
81 import org.rrd4j.core.RrdDef;
82 import org.rrd4j.core.Sample;
83 import org.slf4j.Logger;
84 import org.slf4j.LoggerFactory;
87 * This is the implementation of the RRD4j {@link PersistenceService}. To learn
88 * more about RRD4j please visit their
89 * <a href="https://github.com/rrd4j/rrd4j">website</a>.
91 * @author Kai Kreuzer - Initial contribution
92 * @author Jan N. Klug - some improvements
93 * @author Karel Goderis - remove TimerThread dependency
96 @Component(service = { PersistenceService.class,
97 QueryablePersistenceService.class }, configurationPid = "org.openhab.rrd4j", configurationPolicy = ConfigurationPolicy.OPTIONAL)
98 public class RRD4jPersistenceService implements QueryablePersistenceService {
100 private static final String DEFAULT_OTHER = "default_other";
101 private static final String DEFAULT_NUMERIC = "default_numeric";
102 private static final String DEFAULT_QUANTIFIABLE = "default_quantifiable";
104 private static final Set<String> SUPPORTED_TYPES = Set.of(CoreItemFactory.SWITCH, CoreItemFactory.CONTACT,
105 CoreItemFactory.DIMMER, CoreItemFactory.NUMBER, CoreItemFactory.ROLLERSHUTTER, CoreItemFactory.COLOR);
107 private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(3,
108 new NamedThreadFactory("RRD4j"));
110 private final Map<String, RrdDefConfig> rrdDefs = new ConcurrentHashMap<>();
112 private static final String DATASOURCE_STATE = "state";
114 private static final Path DB_FOLDER = Path.of(OpenHAB.getUserDataFolder(), "persistence", "rrd4j").toAbsolutePath();
116 private static final RrdDbPool DATABASE_POOL = new RrdDbPool();
118 private final Logger logger = LoggerFactory.getLogger(RRD4jPersistenceService.class);
120 private final Map<String, ScheduledFuture<?>> scheduledJobs = new HashMap<>();
122 private final ItemRegistry itemRegistry;
124 public static Path getDatabasePath(String name) {
125 return DB_FOLDER.resolve(name + ".rrd");
128 public static RrdDbPool getDatabasePool() {
129 return DATABASE_POOL;
133 public RRD4jPersistenceService(final @Reference ItemRegistry itemRegistry) {
134 this.itemRegistry = itemRegistry;
138 public String getId() {
143 public String getLabel(@Nullable Locale locale) {
148 public synchronized void store(final Item item, @Nullable final String alias) {
149 if (!isSupportedItemType(item)) {
150 logger.trace("Ignoring item '{}' since its type {} is not supported", item.getName(), item.getType());
153 final String name = alias == null ? item.getName() : alias;
158 } catch (Exception e) {
159 logger.warn("Failed to open rrd4j database '{}' to store data ({})", name, e.toString());
165 ConsolFun function = getConsolidationFunction(db);
166 long now = System.currentTimeMillis() / 1000;
167 if (function != ConsolFun.AVERAGE) {
169 // we store the last value again, so that the value change
170 // in the database is not interpolated, but
171 // happens right at this spot
172 if (now - 1 > db.getLastUpdateTime()) {
173 // only do it if there is not already a value
174 double lastValue = db.getLastDatasourceValue(DATASOURCE_STATE);
175 if (!Double.isNaN(lastValue)) {
176 Sample sample = db.createSample();
177 sample.setTime(now - 1);
178 sample.setValue(DATASOURCE_STATE, lastValue);
180 logger.debug("Stored '{}' as value '{}' in rrd4j database (again)", name, lastValue);
183 } catch (IOException e) {
184 logger.debug("Error storing last value (again): {}", e.getMessage());
188 Sample sample = db.createSample();
193 if (item instanceof NumberItem && item.getState() instanceof QuantityType) {
194 NumberItem nItem = (NumberItem) item;
195 QuantityType<?> qState = (QuantityType<?>) item.getState();
196 Unit<? extends Quantity<?>> unit = nItem.getUnit();
198 QuantityType<?> convertedState = qState.toUnit(unit);
199 if (convertedState != null) {
200 value = convertedState.doubleValue();
203 "Failed to convert state '{}' to unit '{}'. Please check your item definition for correctness.",
207 value = qState.doubleValue();
210 DecimalType state = item.getStateAs(DecimalType.class);
212 value = state.toBigDecimal().doubleValue();
216 if (db.getDatasource(DATASOURCE_STATE).getType() == DsType.COUNTER) { // counter values must be
217 // adjusted by stepsize
218 value = value * db.getRrdDef().getStep();
220 sample.setValue(DATASOURCE_STATE, value);
222 logger.debug("Stored '{}' as value '{}' in rrd4j database", name, value);
224 } catch (IllegalArgumentException e) {
225 String message = e.getMessage();
226 if (message != null && message.contains("at least one second step is required")) {
227 // we try to store the value one second later
228 ScheduledFuture<?> job = scheduledJobs.get(name);
231 scheduledJobs.remove(name);
233 job = scheduler.schedule(() -> store(item, name), 1, TimeUnit.SECONDS);
234 scheduledJobs.put(name, job);
236 logger.warn("Could not persist '{}' to rrd4j database: {}", name, e.getMessage());
238 } catch (Exception e) {
239 logger.warn("Could not persist '{}' to rrd4j database: {}", name, e.getMessage());
243 } catch (IOException e) {
244 logger.debug("Error closing rrd4j database: {}", e.getMessage());
249 public void store(Item item) {
254 public Iterable<HistoricItem> query(FilterCriteria filter) {
255 String itemName = filter.getItemName();
259 db = getDB(itemName);
260 } catch (Exception e) {
261 logger.warn("Failed to open rrd4j database '{}' for querying ({})", itemName, e.toString());
265 logger.debug("Could not find item '{}' in rrd4j database", itemName);
272 item = itemRegistry.getItem(itemName);
273 if (item instanceof NumberItem) {
274 // we already retrieve the unit here once as it is a very costly operation,
275 // see https://github.com/openhab/openhab-addons/issues/8928
276 unit = ((NumberItem) item).getUnit();
278 } catch (ItemNotFoundException e) {
279 logger.debug("Could not find item '{}' in registry", itemName);
283 long end = filter.getEndDate() == null ? System.currentTimeMillis() / 1000
284 : filter.getEndDate().toInstant().getEpochSecond();
287 if (filter.getBeginDate() == null) {
288 // as rrd goes back for years and gets more and more
289 // inaccurate, we only support descending order
290 // and a single return value
291 // if there is no begin date is given - this case is
292 // required specifically for the historicState()
293 // query, which we want to support
294 if (filter.getOrdering() == Ordering.DESCENDING && filter.getPageSize() == 1
295 && filter.getPageNumber() == 0) {
296 if (filter.getEndDate() == null) {
297 // we are asked only for the most recent value!
298 double lastValue = db.getLastDatasourceValue(DATASOURCE_STATE);
299 if (!Double.isNaN(lastValue)) {
300 HistoricItem rrd4jItem = new RRD4jItem(itemName, mapToState(lastValue, item, unit),
301 ZonedDateTime.ofInstant(Instant.ofEpochMilli(db.getLastArchiveUpdateTime() * 1000),
302 ZoneId.systemDefault()));
303 return List.of(rrd4jItem);
311 throw new UnsupportedOperationException("rrd4j does not allow querys without a begin date, "
312 + "unless order is descending and a single value is requested");
315 start = filter.getBeginDate().toInstant().getEpochSecond();
318 FetchRequest request = db.createFetchRequest(getConsolidationFunction(db), start, end, 1);
319 FetchData result = request.fetchData();
321 List<HistoricItem> items = new ArrayList<>();
322 long ts = result.getFirstTimestamp();
323 long step = result.getRowCount() > 1 ? result.getStep() : 0;
324 for (double value : result.getValues(DATASOURCE_STATE)) {
325 if (!Double.isNaN(value) && (((ts >= start) && (ts <= end)) || (start == end))) {
326 RRD4jItem rrd4jItem = new RRD4jItem(itemName, mapToState(value, item, unit),
327 ZonedDateTime.ofInstant(Instant.ofEpochSecond(ts), ZoneId.systemDefault()));
328 items.add(rrd4jItem);
333 } catch (IOException e) {
334 logger.warn("Could not query rrd4j database for item '{}': {}", itemName, e.getMessage());
339 } catch (IOException e) {
340 logger.debug("Error closing rrd4j database: {}", e.getMessage());
346 public Set<PersistenceItemInfo> getItemInfo() {
350 protected synchronized @Nullable RrdDb getDB(String alias) {
352 Path path = getDatabasePath(alias);
354 Builder builder = RrdDb.getBuilder();
355 builder.setPool(DATABASE_POOL);
357 if (Files.exists(path)) {
358 // recreate the RrdDb instance from the file
359 builder.setPath(path.toString());
360 db = builder.build();
362 if (!Files.exists(DB_FOLDER)) {
363 Files.createDirectories(DB_FOLDER);
365 RrdDef rrdDef = getRrdDef(alias, path);
366 if (rrdDef != null) {
367 // create a new database file
368 builder.setRrdDef(rrdDef);
369 db = builder.build();
372 "Did not create rrd4j database for item '{}' since no rrd definition could be determined. This is likely due to an unsupported item type.",
376 } catch (IOException e) {
377 logger.error("Could not create rrd4j database file '{}': {}", path, e.getMessage());
378 } catch (RejectedExecutionException e) {
379 // this happens if the system is shut down
380 logger.debug("Could not create rrd4j database file '{}': {}", path, e.getMessage());
385 private @Nullable RrdDefConfig getRrdDefConfig(String itemName) {
386 RrdDefConfig useRdc = null;
387 for (Map.Entry<String, RrdDefConfig> e : rrdDefs.entrySet()) {
388 // try to find special config
389 RrdDefConfig rdc = e.getValue();
390 if (rdc.appliesTo(itemName)) {
395 if (useRdc == null) { // not defined, use defaults
397 Item item = itemRegistry.getItem(itemName);
398 if (!isSupportedItemType(item)) {
401 if (item instanceof NumberItem) {
402 NumberItem numberItem = (NumberItem) item;
403 useRdc = numberItem.getDimension() != null ? rrdDefs.get(DEFAULT_QUANTIFIABLE)
404 : rrdDefs.get(DEFAULT_NUMERIC);
406 useRdc = rrdDefs.get(DEFAULT_OTHER);
408 } catch (ItemNotFoundException e) {
409 logger.debug("Could not find item '{}' in registry", itemName);
413 logger.trace("Using rrd definition '{}' for item '{}'.", useRdc, itemName);
417 private @Nullable RrdDef getRrdDef(String itemName, Path path) {
418 RrdDef rrdDef = new RrdDef(path.toString());
419 RrdDefConfig useRdc = getRrdDefConfig(itemName);
420 if (useRdc != null) {
421 rrdDef.setStep(useRdc.step);
422 rrdDef.setStartTime(System.currentTimeMillis() / 1000 - 1);
423 rrdDef.addDatasource(DATASOURCE_STATE, useRdc.dsType, useRdc.heartbeat, useRdc.min, useRdc.max);
424 for (RrdArchiveDef rad : useRdc.archives) {
425 rrdDef.addArchive(rad.fcn, rad.xff, rad.steps, rad.rows);
433 public ConsolFun getConsolidationFunction(RrdDb db) {
435 return db.getRrdDef().getArcDefs()[0].getConsolFun();
436 } catch (IOException e) {
437 return ConsolFun.MAX;
441 @SuppressWarnings({ "unchecked", "rawtypes" })
442 private State mapToState(double value, @Nullable Item item, @Nullable Unit unit) {
443 if (item instanceof GroupItem) {
444 item = ((GroupItem) item).getBaseItem();
447 if (item instanceof SwitchItem && !(item instanceof DimmerItem)) {
448 return value == 0.0d ? OnOffType.OFF : OnOffType.ON;
449 } else if (item instanceof ContactItem) {
450 return value == 0.0d ? OpenClosedType.CLOSED : OpenClosedType.OPEN;
451 } else if (item instanceof DimmerItem || item instanceof RollershutterItem || item instanceof ColorItem) {
452 // make sure Items that need PercentTypes instead of DecimalTypes do receive the right information
453 return new PercentType((int) Math.round(value * 100));
454 } else if (item instanceof NumberItem) {
456 return new QuantityType(value, unit);
459 return new DecimalType(value);
462 private boolean isSupportedItemType(Item item) {
463 if (item instanceof GroupItem) {
464 final Item baseItem = ((GroupItem) item).getBaseItem();
465 if (baseItem != null) {
470 return SUPPORTED_TYPES.contains(ItemUtil.getMainItemType(item.getType()));
474 protected void activate(final Map<String, Object> config) {
479 protected void modified(final Map<String, Object> config) {
480 // clean existing definitions
483 // add default configurations
485 RrdDefConfig defaultNumeric = new RrdDefConfig(DEFAULT_NUMERIC);
486 // use 10 seconds as a step size for numeric values and allow a 10 minute silence between updates
487 defaultNumeric.setDef("GAUGE,600,U,U,10");
488 // define 5 different boxes:
489 // 1. granularity of 10s for the last hour
490 // 2. granularity of 1m for the last week
491 // 3. granularity of 15m for the last year
492 // 4. granularity of 1h for the last 5 years
493 // 5. granularity of 1d for the last 10 years
495 .addArchives("LAST,0.5,1,360:LAST,0.5,6,10080:LAST,0.5,90,36500:LAST,0.5,360,43800:LAST,0.5,8640,3650");
496 rrdDefs.put(DEFAULT_NUMERIC, defaultNumeric);
498 RrdDefConfig defaultQuantifiable = new RrdDefConfig(DEFAULT_QUANTIFIABLE);
499 // use 10 seconds as a step size for numeric values and allow a 10 minute silence between updates
500 defaultQuantifiable.setDef("GAUGE,600,U,U,10");
501 // define 5 different boxes:
502 // 1. granularity of 10s for the last hour
503 // 2. granularity of 1m for the last week
504 // 3. granularity of 15m for the last year
505 // 4. granularity of 1h for the last 5 years
506 // 5. granularity of 1d for the last 10 years
507 defaultQuantifiable.addArchives(
508 "AVERAGE,0.5,1,360:AVERAGE,0.5,6,10080:AVERAGE,0.5,90,36500:AVERAGE,0.5,360,43800:AVERAGE,0.5,8640,3650");
509 rrdDefs.put(DEFAULT_QUANTIFIABLE, defaultQuantifiable);
511 RrdDefConfig defaultOther = new RrdDefConfig(DEFAULT_OTHER);
512 // use 5 seconds as a step size for discrete values and allow a 1h silence between updates
513 defaultOther.setDef("GAUGE,3600,U,U,5");
514 // define 4 different boxes:
515 // 1. granularity of 5s for the last hour
516 // 2. granularity of 1m for the last week
517 // 3. granularity of 15m for the last year
518 // 4. granularity of 4h for the last 10 years
519 defaultOther.addArchives("LAST,0.5,1,720:LAST,0.5,12,10080:LAST,0.5,180,35040:LAST,0.5,2880,21900");
520 rrdDefs.put(DEFAULT_OTHER, defaultOther);
522 if (config.isEmpty()) {
523 logger.debug("using default configuration only");
527 Iterator<String> keys = config.keySet().iterator();
528 while (keys.hasNext()) {
529 String key = keys.next();
531 if ("service.pid".equals(key) || "component.name".equals(key)) {
532 // ignore service.pid and name
536 String[] subkeys = key.split("\\.");
537 if (subkeys.length != 2) {
538 logger.debug("config '{}' should have the format 'name.configkey'", key);
542 Object v = config.get(key);
543 if (v instanceof String) {
544 String value = (String) v;
545 String name = subkeys[0].toLowerCase();
546 String property = subkeys[1].toLowerCase();
548 if (value.isBlank()) {
549 logger.trace("Config is empty: {}", property);
552 logger.trace("Processing config: {} = {}", property, value);
555 RrdDefConfig rrdDef = rrdDefs.get(name);
556 if (rrdDef == null) {
557 rrdDef = new RrdDefConfig(name);
558 rrdDefs.put(name, rrdDef);
562 if ("def".equals(property)) {
563 rrdDef.setDef(value);
564 } else if ("archives".equals(property)) {
565 rrdDef.addArchives(value);
566 } else if ("items".equals(property)) {
567 rrdDef.addItems(value);
569 logger.debug("Unknown property {} : {}", property, value);
571 } catch (IllegalArgumentException e) {
572 logger.warn("Ignoring illegal configuration: {}", e.getMessage());
577 for (RrdDefConfig rrdDef : rrdDefs.values()) {
578 if (rrdDef.isValid()) {
579 logger.debug("Created {}", rrdDef);
581 logger.info("Removing invalid definition {}", rrdDef);
582 rrdDefs.remove(rrdDef.name);
587 private static class RrdArchiveDef {
588 public @Nullable ConsolFun fcn;
590 public int steps, rows;
593 public String toString() {
594 StringBuilder sb = new StringBuilder(" " + fcn);
595 sb.append(" xff = ").append(xff);
596 sb.append(" steps = ").append(steps);
597 sb.append(" rows = ").append(rows);
598 return sb.toString();
602 private class RrdDefConfig {
604 public @Nullable DsType dsType;
605 public int heartbeat, step;
606 public double min, max;
607 public List<RrdArchiveDef> archives;
608 public List<String> itemNames;
610 private boolean isInitialized;
612 public RrdDefConfig(String name) {
614 archives = new ArrayList<>();
615 itemNames = new ArrayList<>();
616 isInitialized = false;
619 public void setDef(String defString) {
620 String[] opts = defString.split(",");
621 if (opts.length != 5) { // check if correct number of parameters
622 logger.warn("invalid number of parameters {}: {}", name, defString);
626 if ("ABSOLUTE".equals(opts[0])) { // dsType
627 dsType = DsType.ABSOLUTE;
628 } else if ("COUNTER".equals(opts[0])) {
629 dsType = DsType.COUNTER;
630 } else if ("DERIVE".equals(opts[0])) {
631 dsType = DsType.DERIVE;
632 } else if ("GAUGE".equals(opts[0])) {
633 dsType = DsType.GAUGE;
635 logger.warn("{}: dsType {} not supported", name, opts[0]);
638 heartbeat = Integer.parseInt(opts[1]);
640 if ("U".equals(opts[2])) {
643 min = Double.parseDouble(opts[2]);
646 if ("U".equals(opts[3])) {
649 max = Double.parseDouble(opts[3]);
652 step = Integer.parseInt(opts[4]);
654 isInitialized = true; // successfully initialized
659 public void addArchives(String archivesString) {
660 String splitArchives[] = archivesString.split(":");
661 for (String archiveString : splitArchives) {
662 String[] opts = archiveString.split(",");
663 if (opts.length != 4) { // check if correct number of parameters
664 logger.warn("invalid number of parameters {}: {}", name, archiveString);
667 RrdArchiveDef arc = new RrdArchiveDef();
669 if ("AVERAGE".equals(opts[0])) {
670 arc.fcn = ConsolFun.AVERAGE;
671 } else if ("MIN".equals(opts[0])) {
672 arc.fcn = ConsolFun.MIN;
673 } else if ("MAX".equals(opts[0])) {
674 arc.fcn = ConsolFun.MAX;
675 } else if ("LAST".equals(opts[0])) {
676 arc.fcn = ConsolFun.LAST;
677 } else if ("FIRST".equals(opts[0])) {
678 arc.fcn = ConsolFun.FIRST;
679 } else if ("TOTAL".equals(opts[0])) {
680 arc.fcn = ConsolFun.TOTAL;
682 logger.warn("{}: consolidation function {} not supported", name, opts[0]);
684 arc.xff = Double.parseDouble(opts[1]);
685 arc.steps = Integer.parseInt(opts[2]);
686 arc.rows = Integer.parseInt(opts[3]);
691 public void addItems(String itemsString) {
692 Collections.addAll(itemNames, itemsString.split(","));
695 public boolean appliesTo(String item) {
696 return itemNames.contains(item);
699 public boolean isValid() { // a valid configuration must be initialized
700 // and contain at least one function
701 return isInitialized && !archives.isEmpty();
705 public String toString() {
706 StringBuilder sb = new StringBuilder(name);
707 sb.append(" = ").append(dsType);
708 sb.append(" heartbeat = ").append(heartbeat);
709 sb.append(" min/max = ").append(min).append("/").append(max);
710 sb.append(" step = ").append(step);
711 sb.append(" ").append(archives.size()).append(" archives(s) = [");
712 for (RrdArchiveDef arc : archives) {
713 sb.append(arc.toString());
716 sb.append(itemNames.size()).append(" items(s) = [");
717 for (String item : itemNames) {
718 sb.append(item).append(" ");
721 return sb.toString();
726 public List<PersistenceStrategy> getDefaultStrategies() {
727 return List.of(PersistenceStrategy.Globals.RESTORE, PersistenceStrategy.Globals.CHANGE,
728 new PersistenceCronStrategy("everyMinute", "0 * * * * ?"));