Merge pull request #157 from tobiasschuerg/feat/buckets

Feat: buckets management

Author: vlastahajek

examples/Buckets/Buckets.ino

@@ -0,0 +1,136 @@
+/**
+ * Buckets management Example code for InfluxDBClient library for Arduino
+ * Enter WiFi and InfluxDB parameters below
+ *
+ * This example supports only InfluxDB running from unsecure (http://...)
+ * For secure (https://...) or Influx Cloud 2 connection check SecureWrite example to 
+ * see how connect using secured connection (https)
+ **/
+
+#if defined(ESP32)
+#include <WiFiMulti.h>
+WiFiMulti wifiMulti;
+#define DEVICE "ESP32"
+#elif defined(ESP8266)
+#include <ESP8266WiFiMulti.h>
+ESP8266WiFiMulti wifiMulti;
+#define DEVICE "ESP8266"
+#endif
+
+#include <InfluxDbClient.h>
+
+// WiFi AP SSID
+#define WIFI_SSID "ssid"
+// WiFi password
+#define WIFI_PASSWORD "password"
+// InfluxDB  server url. Don't use localhost, always server name or ip address.
+// E.g. http://192.168.1.48:8086 (In InfluxDB 2 UI -> Load Data -> Client Libraries), 
+#define INFLUXDB_URL "influxdb-url"
+// InfluxDB 2 server or cloud API authentication token (Use: InfluxDB UI -> Load Data -> Tokens -> <select token>)
+// This token must have all buckets permission
+#define INFLUXDB_TOKEN "toked-id"
+// InfluxDB 2 organization id (Use: InfluxDB UI -> Settings -> Profile -> <name under tile> )
+#define INFLUXDB_ORG "org"
+// Bucket name that doesn't exist in the db yet
+#define INFLUXDB_BUCKET "test-bucket"
+
+void setup() {
+  Serial.begin(74880);
+
+  // Connect WiFi
+  Serial.println("Connecting to " WIFI_SSID);
+  WiFi.mode(WIFI_STA);
+  wifiMulti.addAP(WIFI_SSID, WIFI_PASSWORD);
+  while (wifiMulti.run() != WL_CONNECTED) {
+    Serial.print(".");
+    delay(500);
+  }
+  Serial.println();
+}
+
+// Creates client, bucket, writes data, verifies data and deletes bucket
+void testClient() {
+  // InfluxDB client instance
+  InfluxDBClient client(INFLUXDB_URL, INFLUXDB_ORG, INFLUXDB_BUCKET, INFLUXDB_TOKEN);
+
+  // Check server connection
+  if (client.validateConnection()) {
+    Serial.print("Connected to InfluxDB: ");
+    Serial.println(client.getServerUrl());
+  } else {
+    Serial.print("InfluxDB connection failed: ");
+    Serial.println(client.getLastErrorMessage());
+    return;
+  }
+
+  // Get dedicated client for buckets management
+  BucketsClient buckets = client.getBucketsClient();
+  
+  // Verify bucket does not exist, or delete it
+  if(buckets.checkBucketExists(INFLUXDB_BUCKET)) {
+    Serial.println("Bucket " INFLUXDB_BUCKET " already exists, deleting" );
+    // get reference
+    Bucket b = buckets.findBucket(INFLUXDB_BUCKET);
+    // Delete bucket
+    buckets.deleteBucket(b.getID());
+  } 
+
+  // create a bucket with retention policy one month. Leave out or set zero to infinity
+  uint32_t monthSec = 30*24*3600;
+  Bucket b = buckets.createBucket(INFLUXDB_BUCKET, monthSec);
+  if(!b) {
+    // some error occurred
+    Serial.print("Bucket creating error: ");
+    Serial.println(buckets.getLastErrorMessage());
+    return;
+  }
+  Serial.print("Created bucket: ");
+  Serial.println(b.toString());
+  
+  int numPoints = 10;
+  // Write some points
+  for(int i=0;i<numPoints;i++) {
+    Point point("test");
+    point.addTag("device_name", DEVICE);
+    point.addField("temperature", random(-20, 40) * 1.1f);
+    point.addField("humidity", random(10, 90));
+    if(!client.writePoint(point)) {
+      Serial.print("Write error: ");
+      Serial.println(client.getLastErrorMessage());
+    }
+  }
+  // verify written points
+  String query= "from(bucket: \"" INFLUXDB_BUCKET "\") |> range(start: -1h) |> pivot(rowKey:[\"_time\"],columnKey: [\"_field\"],valueColumn: \"_value\") |> count(column: \"humidity\")";
+  FluxQueryResult result = client.query(query);
+  // We expect one row
+  if(result.next()) { 
+    // Get count value
+    FluxValue val = result.getValueByName("humidity");
+    if(val.getLong() != numPoints) {
+      Serial.print("Test failure, expected ");
+      Serial.print(numPoints);
+      Serial.print(" got ");
+      Serial.println(val.getLong());
+    } else {
+      Serial.println("Test successfull");
+    }
+    // Advance to the end
+    result.next();
+  } else {
+    Serial.print("Query error: ");
+    Serial.println(result.getError());
+  };
+  result.close();
+      
+  buckets.deleteBucket(b.getID());
+}
+
+void loop() {
+  // Lets do an E2E test
+  // call a client test
+  testClient();
+  
+  Serial.println("Stopping");
+  // Stop here, don't loop
+  while(1) delay(1);
+}

src/BucketsClient.cpp

@@ -0,0 +1,245 @@
+/**
+ * 
+ * BucketsClient.cpp: InfluxDB Buckets Client
+ * 
+ * MIT License
+ * 
+ * Copyright (c) 2020 InfluxData
+ * 
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ * 
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ * 
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+*/
+#include "BucketsClient.h"
+#include "util/helpers.h"
+
+//#define INFLUXDB_CLIENT_DEBUG_ENABLE
+#include "util/debug.h"
+
+static const char *propTemplate PROGMEM = "\"%s\":";
+// Finds first id property from JSON response
+enum class PropType {
+  String,
+  Number
+};
+
+static String findProperty(const char *prop,const String &json, PropType type = PropType::String);
+
+static String findProperty(const char *prop,const String &json, PropType type) {
+  INFLUXDB_CLIENT_DEBUG("[D] Searching for %s in %s\n", prop, json.c_str());
+  int propLen = strlen_P(propTemplate)+strlen(prop)-2;
+  char *propSearch = new char[propLen+1];
+  sprintf_P(propSearch, propTemplate, prop);
+  int i = json.indexOf(propSearch);
+  delete [] propSearch;
+  if(i>-1) {
+    INFLUXDB_CLIENT_DEBUG("[D]   Found at %d\n", i);
+    switch(type) {
+      case PropType::String:
+        i = json.indexOf("\"", i+propLen);
+        if(i>-1) {
+         INFLUXDB_CLIENT_DEBUG("[D]    Found starting \" at %d\n", i);
+          int e = json.indexOf("\"", i+1);
+          if(e>-1) {
+            INFLUXDB_CLIENT_DEBUG("[D]    Found ending \" at %d\n", e);
+            return json.substring(i+1, e);
+          }
+        }
+      break;
+      case PropType::Number:
+        i = i+propLen;
+        while(json[i] == ' ') {
+          i++;
+        }
+        INFLUXDB_CLIENT_DEBUG("[D]    Found beginning of number at %d\n", i);
+        int e = json.indexOf(",", i+1);
+        if(e>-1) {
+          INFLUXDB_CLIENT_DEBUG("[D]    Found , at %d\n", e);
+          return json.substring(i, e);
+        }
+      break;
+    }
+  }
+  return "";
+}
+
+char *copyChars(const char *str) {
+  char *ret = new char[strlen(str)+1];
+  strcpy(ret, str);
+  return ret;
+}
+
+Bucket::Bucket():_data(nullptr) {
+}
+
+Bucket::Bucket(const char *id, const char *name, const uint32_t expire) {
+  _data = std::make_shared<Data>(id, name, expire);
+}
+
+Bucket::Bucket(const Bucket &other) {
+    _data = other._data;
+}
+
+Bucket& Bucket::operator=(const Bucket& other) {
+    if(this != &other) {
+        _data = other._data;
+    }
+    return *this;
+}
+
+Bucket::~Bucket() {
+}
+
+
+Bucket::Data::Data(const char *id, const char *name, const uint32_t expire) {
+  this->id = copyChars(id);
+  this->name = copyChars(name);
+  this->expire = expire;
+}
+
+Bucket::Data::~Data() {
+  delete [] id;
+  delete [] name;
+}
+
+
+const char *toStringTmplt PROGMEM = "Bucket: ID %s, Name %s, expire %u";
+String Bucket::toString() const {
+  int len = strlen_P(toStringTmplt) + (_data?strlen(_data->name):0) + (_data?strlen(_data->id):0) + 10 + 1; //10 is maximum length of string representation of expire
+  char *buff = new char[len];
+  sprintf_P(buff, toStringTmplt, getID(), getName(), getExpire());
+  String ret = buff;
+  return ret;
+}
+
+BucketsClient::BucketsClient() {
+  _data = nullptr;
+}
+
+BucketsClient::BucketsClient(ConnectionInfo *pConnInfo, HTTPService *service) {
+  _data = std::make_shared<Data>(pConnInfo, service);
+}
+
+BucketsClient::BucketsClient(const BucketsClient &other) {
+  _data = other._data;
+}
+
+BucketsClient &BucketsClient::operator=(const BucketsClient &other) {
+  if(this != &other) {
+        _data = other._data;
+    }
+    return *this;
+}
+
+BucketsClient &BucketsClient::operator=(std::nullptr_t) {
+  _data = nullptr;
+  return *this;
+}
+
+String BucketsClient::getOrgID(const char *org) {
+  if(!_data) {
+    return "";
+  }
+  if(isValidID(org)) {
+    return org;
+  }
+  String url = _data->pService->getServerAPIURL();
+  url += "orgs?org=";
+  url += urlEncode(org);
+  String id;
+  INFLUXDB_CLIENT_DEBUG("[D] getOrgID: url %s\n", url.c_str());
+  _data->pService->doGET(url.c_str(), 200, [&id](HTTPClient *client){
+    id = findProperty("id",client->getString());
+    return true;
+  });
+  return id;
+}
+
+bool BucketsClient::checkBucketExists(const char *bucketName) {
+  Bucket b = findBucket(bucketName);
+  return !b.isNull();
+}
+
+static const char *CreateBucketTemplate PROGMEM = "{\"name\":\"%s\",\"orgID\":\"%s\",\"retentionRules\":[{\"everySeconds\":%u}]}";
+
+Bucket BucketsClient::createBucket(const char *bucketName, uint32_t expiresSec) {
+  Bucket b;
+  if(_data) {
+    String orgID = getOrgID(_data->pConnInfo->org.c_str());
+    
+    if(!orgID.length()) {
+      return b;
+    }
+    int expireLen = 0;
+    uint32_t e = expiresSec; 
+    do {
+      expireLen++;
+      e /=10;
+    } while(e > 0);
+    int len = strlen_P(CreateBucketTemplate) + strlen(bucketName) + orgID.length() + expireLen+1;
+    char *body = new char[len];
+    sprintf_P(body, CreateBucketTemplate, bucketName, orgID.c_str(), expiresSec);
+    String url = _data->pService->getServerAPIURL();
+    url += "buckets";
+    INFLUXDB_CLIENT_DEBUG("[D] CreateBucket: url %s, body %s\n", url.c_str(), body);
+    _data->pService->doPOST(url.c_str(), body, "application/json", 201, [&b](HTTPClient *client){
+      String resp = client->getString();
+      String id = findProperty("id", resp);
+      String name = findProperty("name", resp);
+      String expireStr = findProperty("everySeconds", resp, PropType::Number);
+      uint32_t expire = strtoul(expireStr.c_str(), nullptr, 10);
+      b = Bucket(id.c_str(), name.c_str(), expire);
+      return true;
+    });
+    delete [] body;
+  }
+  return b;
+}
+
+bool BucketsClient::deleteBucket(const char *id) {
+  if(!_data) {
+    
+    return false;
+  }
+  String url = _data->pService->getServerAPIURL();
+  url += "buckets/";
+  url += id;
+  INFLUXDB_CLIENT_DEBUG("[D] deleteBucket: url %s\n", url.c_str());
+  return _data->pService->doDELETE(url.c_str(), 204, nullptr);
+}
+
+Bucket BucketsClient::findBucket(const char *bucketName) {
+  Bucket b;
+  if(_data) {
+    String url = _data->pService->getServerAPIURL();
+    url += "buckets?name=";
+    url += urlEncode(bucketName);
+    INFLUXDB_CLIENT_DEBUG("[D] findBucket: url %s\n", url.c_str());
+    _data->pService->doGET(url.c_str(), 200, [&b](HTTPClient *client){
+      String resp = client->getString();
+      String id = findProperty("id", resp);
+      if(id.length()) {
+        String name = findProperty("name", resp);
+        String expireStr = findProperty("everySeconds", resp, PropType::Number);
+        uint32_t expire = strtoul(expireStr.c_str(), nullptr, 10);
+        b = Bucket(id.c_str(), name.c_str(), expire);
+      }
+      return true;
+    });
+  }
+  return b;
+}