Merge pull request #109 from bonitoo-io/feat/retry_improvements

feat: retry strategy improvements

Author: vlastahajek

CHANGELOG.md

@@ -2,6 +2,11 @@
 ## 3.5.0 [in progress]
 ### Features
  - [#107](https://github.com/tobiasschuerg/InfluxDB-Client-for-Arduino/pull/107) - Added possibility to set default tags. Use `WriteOptions::addDefaultTag()` to add a tag that will be added to each written point using the `writePoint()` function.
+ - [#109](https://github.com/tobiasschuerg/InfluxDB-Client-for-Arduino/pull/109) - Retry strategy improvements:
+   - Added `canSendRequest()` function to check if retry strategy is applied
+   - Added `getRemaingRetryTime()` function to get wait time before another request (write/query) can be sent
+   - Removed applying retry wait time in case of network error
+   - Better explanatory error message when a request is about to be sent in the retry wait state
 
 ### Documentation
 

src/InfluxDbClient.cpp

@@ -45,7 +45,7 @@ static const char UserAgent[] PROGMEM = "influxdb-client-arduino/" INFLUXDB_CLIE
 #include "util/debug.h"
 
 static const char UnitialisedMessage[] PROGMEM = "Unconfigured instance"; 
-static const char TooEarlyMessage[] PROGMEM = "Too early request";
+static const char TooEarlyMessage[] PROGMEM = "Cannot send request yet because of applied retry strategy. Remaining ";
 // This cannot be put to PROGMEM due to the way how it is used
 static const char RetryAfter[] = "Retry-After";
 static const char TransferEnconding[] = "Transfer-Encoding";
@@ -398,10 +398,6 @@ bool InfluxDBClient::writeRecord(String &record) {
         if(_bufferPointer == _writeBufferSize) { // writeBuffer is full
             _bufferPointer = 0;
             INFLUXDB_CLIENT_DEBUG("[W] Reached write buffer size, old points will be overwritten\n");
-            // if(isBufferFull()) {
-            //     //if already isBufferFull
-            //     _batchPointer = 0;
-            // }
         } 
 
         if(_bufferCeiling < _writeBufferSize) {
@@ -443,16 +439,25 @@ bool InfluxDBClient::flushBuffer() {
     return flushBufferInternal(false);
 }
 
-bool InfluxDBClient::flushBufferInternal(bool flashOnlyFull) {
+uint32_t InfluxDBClient::getRemaingRetryTime() {
+    uint32_t rem = 0;
     if(_lastRetryAfter > 0) {
-        uint32_t diff = (millis()-_lastRequestTime)/1000 ;
-        if(diff < _lastRetryAfter) {
-            INFLUXDB_CLIENT_DEBUG("[W] Cannot write yet, pause %ds, so far %ds\n", _lastRetryAfter, diff);
-            // retry after period didn't run out yet
-            _lastStatusCode = 0;
-            _lastErrorResponse = FPSTR(TooEarlyMessage);
-            return false;
-        }
+        int32_t diff = _lastRetryAfter - (millis()-_lastRequestTime)/1000;
+        rem  =  diff<0?0:(uint32_t)diff;
+    }
+    return rem;
+}
+
+bool InfluxDBClient::flushBufferInternal(bool flashOnlyFull) {
+    uint32_t rwt = getRemaingRetryTime();
+    if(rwt > 0) {
+        INFLUXDB_CLIENT_DEBUG("[W] Cannot write yet, pause %ds, %ds yet\n", _lastRetryAfter, rwt);
+        // retry after period didn't run out yet
+        _lastStatusCode = 0;
+        _lastErrorResponse = FPSTR(TooEarlyMessage);
+        _lastErrorResponse += String(rwt);
+        _lastErrorResponse += "s";
+        return false;
     }
     char *data;
     bool success = true;
@@ -483,7 +488,7 @@ bool InfluxDBClient::flushBufferInternal(bool flashOnlyFull) {
                     INFLUXDB_CLIENT_DEBUG("[D] Reached max retry count, dropping batch\n");
                     dropCurrentBatch();
                 }
-                if(!_lastRetryAfter) {
+                if(!_lastRetryAfter && statusCode > 0) {
                     _lastRetryAfter = _writeOptions._retryInterval;
                     if(_writeBuffer[_batchPointer]) {
                         for(int i=1;i<_writeBuffer[_batchPointer]->retryCount;i++) {
@@ -534,7 +539,7 @@ bool InfluxDBClient::validateConnection() {
 
    _lastErrorResponse = "";
 
-    afterRequest(200);
+    afterRequest(200, false);
 
     _httpClient.end();
 
@@ -591,9 +596,14 @@ static const char QueryDialect[] PROGMEM = "\
 }}";
 
 FluxQueryResult InfluxDBClient::query(String fluxQuery) {
-    if(_lastRetryAfter > 0 && (millis()-_lastRequestTime)/1000 < _lastRetryAfter) {
+    uint32_t rwt = getRemaingRetryTime();
+    if(rwt > 0) {
+        INFLUXDB_CLIENT_DEBUG("[W] Cannot query yet, pause %ds, %ds yet\n", _lastRetryAfter, rwt);
         // retry after period didn't run out yet
-        return FluxQueryResult(FPSTR(TooEarlyMessage));
+        String mess = FPSTR(TooEarlyMessage);
+        mess += String(rwt);
+        mess += "s";
+        return FluxQueryResult(mess);
     }
     if(!_wifiClient && !init()) {
         _lastStatusCode = 0;
@@ -635,17 +645,17 @@ FluxQueryResult InfluxDBClient::query(String fluxQuery) {
     }
 }
 
-void InfluxDBClient::afterRequest(int expectedStatusCode) {
-    _lastRequestTime = millis();
-     INFLUXDB_CLIENT_DEBUG("[D] HTTP status code - %d\n", _lastStatusCode);
-    if(_lastStatusCode >= 429) { //retryable server errors
+void InfluxDBClient::afterRequest(int expectedStatusCode,  bool modifyLastConnStatus) {
+    if(modifyLastConnStatus) {
+        _lastRequestTime = millis();
+        INFLUXDB_CLIENT_DEBUG("[D] HTTP status code - %d\n", _lastStatusCode);
         _lastRetryAfter = 0;
-        if(_httpClient.hasHeader(RetryAfter)) {
-            _lastRetryAfter = _httpClient.header(RetryAfter).toInt();
-            INFLUXDB_CLIENT_DEBUG("[D] Reply after - %d\n", _lastRetryAfter);
+        if(_lastStatusCode >= 429) { //retryable server errors
+            if(_httpClient.hasHeader(RetryAfter)) {
+                _lastRetryAfter = _httpClient.header(RetryAfter).toInt();
+                INFLUXDB_CLIENT_DEBUG("[D] Reply after - %d\n", _lastRetryAfter);
+            }
         }
-    } else {
-        _lastRetryAfter = 0;
     }
     _lastErrorResponse = "";
     if(_lastStatusCode != expectedStatusCode) {

src/InfluxDbClient.h

@@ -148,14 +148,20 @@ class InfluxDBClient {
     String getLastErrorMessage() const { return _lastErrorResponse; }
     // Returns server url
     String getServerUrl() const { return _serverUrl; }
+    // Check if it is possible to send write/query request to server. 
+    // Returns true if write or query can be send, or false, if server is overloaded and retry strategy is applied.
+    // Use getRemaingRetryTime() to get wait time in such case.
+    bool canSendRequest() { return getRemaingRetryTime() == 0; }
+    // Returns remaining wait time in seconds when retry strategy is applied.
+    uint32_t getRemaingRetryTime();
   protected:
     // Checks params and sets up security, if needed.
     // Returns true in case of success, otherwise false
     bool init();
     // Sets request params
     void beforeRequest();
     // Handles response
-    void afterRequest(int expectedStatusCode);
+    void afterRequest(int expectedStatusCode, bool modifyLastConnStatus = true);
     // Cleans instances
     void clean();
   protected:

src/util/debug.h

@@ -30,7 +30,7 @@
 #include <Arduino.h>
 
 #ifdef INFLUXDB_CLIENT_DEBUG_ENABLE
-# define INFLUXDB_CLIENT_DEBUG(fmt, ...) Serial.printf_P( (PGM_P)PSTR(fmt), ## __VA_ARGS__ )
+# define INFLUXDB_CLIENT_DEBUG(fmt, ...) Serial.printf("%.03f ",millis()/1000.0f);Serial.printf_P( (PGM_P)PSTR(fmt), ## __VA_ARGS__ )
 #else
 # define INFLUXDB_CLIENT_DEBUG(fmt, ...)
 #endif //INFLUXDB_CLIENT_DEBUG

test/test.ino

@@ -51,7 +51,9 @@ void setup() {
 class Test {
 public:
     static void run();
-private:
+private: //helpers
+    static void setServerUrl(InfluxDBClient &client, String serverUrl);
+private: // tests
     static void testOptions();
     static void testEcaping();
     static void testPoint();
@@ -481,16 +483,29 @@ void Test::testRetryOnFailedConnection() {
 
     InfluxDBClient clientOk(INFLUXDB_CLIENT_TESTING_URL, INFLUXDB_CLIENT_TESTING_ORG, INFLUXDB_CLIENT_TESTING_BUC, INFLUXDB_CLIENT_TESTING_TOK);
     clientOk.setWriteOptions(WriteOptions().batchSize(1).bufferSize(5));
+    waitServer(clientOk, true);
+    TEST_ASSERT(clientOk.validateConnection());
+    Point *p = createPoint("test1");
+    TEST_ASSERT(clientOk.writePoint(*p));
+    delete p;
+    p = createPoint("test1");
+    TEST_ASSERT(clientOk.writePoint(*p));
+    delete p;
+    TEST_ASSERT(clientOk.isBufferEmpty());
+
     clientOk.setHTTPOptions(HTTPOptions().httpReadTimeout(500));
 
     Serial.println("Stop server!");
     waitServer(clientOk, false);
     TEST_ASSERT(!clientOk.validateConnection());
-    Point *p = createPoint("test1");
+    TEST_ASSERTM(clientOk._lastRetryAfter == 0, String(clientOk._lastRetryAfter));
+    p = createPoint("test1");
     TEST_ASSERT(!clientOk.writePoint(*p));
+    TEST_ASSERTM(clientOk._lastRetryAfter == 0, String(clientOk._lastRetryAfter));
     delete p;
     p = createPoint("test1");
     TEST_ASSERT(!clientOk.writePoint(*p));
+    TEST_ASSERTM(clientOk._lastRetryAfter == 0, String(clientOk._lastRetryAfter));
     delete p;
 
     Serial.println("Start server!");
@@ -499,6 +514,7 @@ void Test::testRetryOnFailedConnection() {
     TEST_ASSERT(clientOk.validateConnection());
     p = createPoint("test1");
     TEST_ASSERT(clientOk.writePoint(*p));
+    TEST_ASSERTM(clientOk._lastRetryAfter == 0, String(clientOk._lastRetryAfter));
     delete p;
     TEST_ASSERT(clientOk.isBufferEmpty());
     String query = "select";
@@ -525,8 +541,8 @@ void Test::testBufferOverwriteBatchsize1() {
     TEST_ASSERT(client.isBufferFull());
     TEST_ASSERTM(client._writeBuffer[0]->buffer[0].indexOf("index=10i") > 0, client._writeBuffer[0]->buffer[0]);
 
-    client._serverUrl = INFLUXDB_CLIENT_TESTING_URL;
-    client.setUrls();
+    setServerUrl(client,INFLUXDB_CLIENT_TESTING_URL );
+    
     waitServer(client, true);
     client.setHTTPOptions(HTTPOptions().httpReadTimeout(5000));
     Point *p = createPoint("test1");
@@ -566,8 +582,7 @@ void Test::testBufferOverwriteBatchsize5() {
     TEST_ASSERT(client.isBufferFull());
     TEST_ASSERTM(client._writeBuffer[0]->buffer[0].indexOf("index=20i") > 0, client._writeBuffer[0]->buffer[0]);
 
-    client._serverUrl = INFLUXDB_CLIENT_TESTING_URL;
-    client.setUrls();
+    setServerUrl(client,INFLUXDB_CLIENT_TESTING_URL );
 
     waitServer(client, true);
     client.setHTTPOptions(HTTPOptions().httpReadTimeout(5000));
@@ -1685,34 +1700,42 @@ void Test::testRetryInterval() {
 
     String rec = "test1,direction=permanent-set,x-code=502,SSID=bonitoo.io,device_name=ESP32,device_id=4272205360 temperature=28.60,humidity=86i,code=69i,door=false,status=\"failed\",index=0";
     TEST_ASSERT(!client.writeRecord(rec));
+    TEST_ASSERT(!client.canSendRequest());
     TEST_ASSERTM(client._lastRetryAfter == 2, String(client._lastRetryAfter));
     TEST_ASSERTM(client._writeBuffer[0]->retryCount == 1, String(client._writeBuffer[0]->retryCount));
     delay(2000);
     rec = "test1,direction=permanent-unset,SSID=bonitoo.io,device_name=ESP32,device_id=4272205360 temperature=28.60,humidity=86i,code=69i,door=false,status=\"failed\",index=2";
     TEST_ASSERT(!client.writeRecord(rec));
+    TEST_ASSERT(!client.canSendRequest());
     TEST_ASSERTM(client._lastRetryAfter == 4, String(client._lastRetryAfter));
     TEST_ASSERTM(client._writeBuffer[0]->retryCount == 2, String(client._writeBuffer[0]->retryCount));
     delay(4000);
     rec = "test1,SSID=bonitoo.io,device_name=ESP32,device_id=4272205360 temperature=28.60,humidity=86i,code=69i,door=false,status=\"failed\",index=3";
     TEST_ASSERT(!client.writeRecord(rec));
+    TEST_ASSERT(!client.canSendRequest());
     TEST_ASSERTM(client._lastRetryAfter == 8, String(client._lastRetryAfter));
     TEST_ASSERTM(client._writeBuffer[0]->retryCount == 3, String(client._writeBuffer[0]->retryCount));
     delay(8000);
     rec = "test1,SSID=bonitoo.io,device_name=ESP32,device_id=4272205360 temperature=28.60,humidity=86i,code=69i,door=false,status=\"failed\",index=4";
     TEST_ASSERT(!client.writeRecord(rec));
+    TEST_ASSERT(!client.canSendRequest());
     TEST_ASSERTM(client._lastRetryAfter == 2, String(client._lastRetryAfter));
     TEST_ASSERT(!client._writeBuffer[0]);
     TEST_ASSERTM(client._writeBuffer[1]->retryCount == 0, String(client._writeBuffer[1]->retryCount));
 
     delay(2000);
     rec = "test1,SSID=bonitoo.io,device_name=ESP32,device_id=4272205360 temperature=28.60,humidity=86i,code=69i,door=false,status=\"failed\",index=5";
     TEST_ASSERT(!client.writeRecord(rec));
+    TEST_ASSERT(!client.canSendRequest());
     TEST_ASSERTM(client._lastRetryAfter == 2, String(client._lastRetryAfter));
     TEST_ASSERT(!client._writeBuffer[0]);
     TEST_ASSERTM(client._writeBuffer[1]->retryCount == 1, String(client._writeBuffer[1]->retryCount));
 
     delay(2000);
+    TEST_ASSERT(client.canSendRequest());
     TEST_ASSERTM(client.flushBuffer(), client.getLastErrorMessage());
+    TEST_ASSERT(client.isBufferEmpty());
+    TEST_ASSERT(!client.isBufferFull());
     String query = "select";
     FluxQueryResult q = client.query(query);
     TEST_ASSERT(countLines(q) == 3); //point with the direction tag is skipped
@@ -1782,6 +1805,11 @@ void Test::testDefaultTags() {
     deleteAll(INFLUXDB_CLIENT_TESTING_URL);
 }
 
+void Test::setServerUrl(InfluxDBClient &client, String serverUrl) {
+    client._serverUrl = serverUrl;
+    client.setUrls();
+}
+
 Point *createPoint(String measurement) {
     Point *point = new Point(measurement);
     point->addTag("SSID", WiFi.SSID());