Add advanced config options for InfluxDB

CHANGELOG.md

@@ -1,5 +1,9 @@
 # Changelog
 
+### v0.2.3
+
+ - Added advanced configuration options for InfluxDB
+
 ### v0.2.2
 
  - Fixed a rare crash when hcidump was returning invalid data

ruuvi-collector.properties.example

@@ -44,9 +44,27 @@
 #            NOTE: the influxdb_legacy storage method does not support extended values
 #storage.values=extended
 
+
+#
+# Advanced configuration - Change these only if you know what you are doing
+#
+
 # Commands for starting the BLE scanning and the dump.
 # The scan command is only executed, output and exit is ignored. Setting this blank will disable this command.
 # The dump command is executed and it's output is parsed, the collector will exit when/if this command ends
 # In general you want to change these only if you intend to use wrapper scripts or you need to use absolute paths to the executables
 #command.scan=hcitool lescan --duplicates --passive
 #command.dump=hcidump --raw
+
+# Retention policy to use (note: you must create it yourself)
+#influxRetentionPolicy=autogen
+
+# Use gzip, saves bandwidth at minimal CPU cost
+#influxGzip=true
+
+# Use batch mode, improved performance at the cost of increased delay for measurements to show up. Does not affect the timestamps.
+#influxBatch=true
+
+# Maximum number of datapoints and maximum time waited in milliseconds before sending a batch. Has no effect if batch mode disabled.
+#influxBatchMaxSize=2000
+#influxBatchMaxTime=100

src/main/java/fi/tkgwf/ruuvi/config/Config.java

@@ -28,6 +28,11 @@ public abstract class Config {
     private static String influxMeasurement = "ruuvi_measurements";
     private static String influxUser = "ruuvi";
     private static String influxPassword = "ruuvi";
+    private static String influxRetentionPolicy = "autogen";
+    private static boolean influxGzip = true;
+    private static boolean influxBatch = true;
+    private static int influxBatchMaxSize = 2000;
+    private static int influxBatchMaxTimeMs = 100;
     private static long measurementUpdateLimit = 9900;
     private static String storageMethod = "influxdb";
     private static String storageValues = "extended";
@@ -110,6 +115,29 @@ private static void readConfig() {
                         case "command.dump":
                             dumpCommand = value.split(" ");
                             break;
+                        case "influxRetentionPolicy":
+                            influxRetentionPolicy = value;
+                            break;
+                        case "influxGzip":
+                            influxGzip = Boolean.parseBoolean(value);
+                            break;
+                        case "influxBatch":
+                            influxBatch = Boolean.parseBoolean(value);
+                            break;
+                        case "influxBatchMaxSize":
+                            try {
+                                influxBatchMaxSize = Integer.parseInt(value);
+                            } catch (NumberFormatException ex) {
+                                LOG.warn("Malformed number format for influxBatchMaxSize: '" + value + '\'');
+                            }
+                            break;
+                        case "influxBatchMaxTime":
+                            try {
+                                influxBatchMaxTimeMs = Integer.parseInt(value);
+                            } catch (NumberFormatException ex) {
+                                LOG.warn("Malformed number format for influxBatchMaxTime: '" + value + '\'');
+                            }
+                            break;
                     }
                 }
             }
@@ -182,6 +210,26 @@ public static String getInfluxPassword() {
         return influxPassword;
     }
 
+    public static String getInfluxRetentionPolicy() {
+        return influxRetentionPolicy;
+    }
+
+    public static boolean isInfluxGzip() {
+        return influxGzip;
+    }
+
+    public static boolean isInfluxBatch() {
+        return influxBatch;
+    }
+
+    public static int getInfluxBatchMaxSize() {
+        return influxBatchMaxSize;
+    }
+
+    public static int getInfluxBatchMaxTimeMs() {
+        return influxBatchMaxTimeMs;
+    }
+
     public static long getMeasurementUpdateLimit() {
         return measurementUpdateLimit;
     }

src/main/java/fi/tkgwf/ruuvi/db/InfluxDBConnection.java

@@ -13,10 +13,41 @@ public class InfluxDBConnection implements DBConnection {
     private final InfluxDB influxDB;
 
     public InfluxDBConnection() {
-        influxDB = InfluxDBFactory.connect(Config.getInfluxUrl(), Config.getInfluxUser(), Config.getInfluxPassword());
-        influxDB.setDatabase(Config.getInfluxDatabase());
-        influxDB.enableGzip();
-        influxDB.enableBatch(2000, 100, TimeUnit.MILLISECONDS); // TODO: make these configurable
+        this(
+                Config.getInfluxUrl(),
+                Config.getInfluxUser(),
+                Config.getInfluxPassword(),
+                Config.getInfluxDatabase(),
+                Config.getInfluxRetentionPolicy(),
+                Config.isInfluxGzip(),
+                Config.isInfluxBatch(),
+                Config.getInfluxBatchMaxSize(),
+                Config.getInfluxBatchMaxTimeMs()
+        );
+    }
+
+    public InfluxDBConnection(
+            String url,
+            String user,
+            String password,
+            String database,
+            String retentionPolicy,
+            boolean gzip,
+            boolean batch,
+            int batchSize,
+            int batchTime
+    ) {
+        influxDB = InfluxDBFactory.connect(url, user, password).setDatabase(database).setRetentionPolicy(retentionPolicy);
+        if (gzip) {
+            influxDB.enableGzip();
+        } else {
+            influxDB.disableGzip();
+        }
+        if (batch) {
+            influxDB.enableBatch(batchSize, batchTime, TimeUnit.MILLISECONDS);
+        } else {
+            influxDB.disableBatch();
+        }
     }
 
     @Override

src/main/java/fi/tkgwf/ruuvi/utils/InfluxDataMigrator.java

@@ -51,7 +51,7 @@ public class InfluxDataMigrator {
     public synchronized void migrate() {
         LOG.info("Starting migration...");
         long start = System.currentTimeMillis();
-        DBConnection db = new InfluxDBConnection();
+        DBConnection db = Config.getDBConnection();
         // Theres a hard limit (?) of 5 concurrent queries per instance
         InfluxDB influx1 = createInfluxDB();
         InfluxDB influx2 = createInfluxDB();