Merge branch 'release/0.4.0'

Author: matus-tomlein

CHANGELOG

@@ -1,3 +1,10 @@
+Version 0.4.0 (2022-04-27)
+--------------------------
+Add back-off on 100% event sending failure (#9)
+Add event-level Subject (#61)
+Add IP address property to Subject (#62)
+Rename send_limit to batch_size in Emitter (#63)
+
 Version 0.3.0 (2022-04-19)
 --------------------------
 Add support for Linux in HTTP client (#5)

README.md

@@ -84,7 +84,7 @@ This will create two executables - the first is the test-suite which can be exec
 The other is an example program which will send one of every type of event to an endpoint of your choosing like so:
 
 ```bash
- host> cd build
+ host> cd build/example
  host> ./tracker_example {{ your collector uri }}
 ```
 
@@ -178,7 +178,7 @@ limitations under the License.
 [travis-image]: https://travis-ci.org/snowplow/snowplow-cpp-tracker.png?branch=master
 [travis]: https://travis-ci.org/snowplow/snowplow-cpp-tracker
 
-[release-image]: https://img.shields.io/badge/release-0.3.0-6ad7e5.svg?style=flat
+[release-image]: https://img.shields.io/badge/release-0.4.0-6ad7e5.svg?style=flat
 [releases]: https://github.com/snowplow/snowplow-cpp-tracker/releases
 
 [license-image]: https://img.shields.io/badge/license-Apache--2-blue.svg?style=flat

snowplow-cpp-tracker-example.vcxproj

@@ -102,7 +102,8 @@
     <ClCompile Include="include\sqlite3.c" />
     <ClCompile Include="src\client_session.cpp" />
     <ClCompile Include="src\cracked_url.cpp" />
-    <ClCompile Include="src\emitter.cpp" />
+    <ClCompile Include="src\emitter\emitter.cpp" />
+    <ClCompile Include="src\emitter\retry_delay.cpp" />
     <ClCompile Include="src\http\http_client_windows.cpp" />
     <ClCompile Include="src\http\http_request_result.cpp" />
     <ClCompile Include="src\payload\payload.cpp" />
@@ -125,7 +126,8 @@
     <ClInclude Include="src\client_session.hpp" />
     <ClInclude Include="src\constants.hpp" />
     <ClInclude Include="src\cracked_url.hpp" />
-    <ClInclude Include="src\emitter.hpp" />
+    <ClInclude Include="src\emitter\emitter.hpp" />
+    <ClInclude Include="src\emitter\retry_delay.hpp" />
     <ClInclude Include="src\http\http_client.hpp" />
     <ClInclude Include="src\http\http_client_windows.hpp" />
     <ClInclude Include="src\http\http_request_result.hpp" />

snowplow-cpp-tracker-example.vcxproj.filters

@@ -21,7 +21,10 @@
     <ClCompile Include="src\cracked_url.cpp">
       <Filter>Source Files</Filter>
     </ClCompile>
-    <ClCompile Include="src\emitter.cpp">
+    <ClCompile Include="src\emitter\emitter.cpp">
+      <Filter>Source Files</Filter>
+    </ClCompile>
+    <ClCompile Include="src\emitter\retry_delay.cpp">
       <Filter>Source Files</Filter>
     </ClCompile>
     <ClCompile Include="src\http\http_client_windows.cpp">
@@ -86,7 +89,10 @@
     <ClInclude Include="src\cracked_url.hpp">
       <Filter>Header Files</Filter>
     </ClInclude>
-    <ClInclude Include="src\emitter.hpp">
+    <ClInclude Include="src\emitter\emitter.hpp">
+      <Filter>Header Files</Filter>
+    </ClInclude>
+    <ClInclude Include="src\emitter\retry_delay.hpp">
       <Filter>Header Files</Filter>
     </ClInclude>
     <ClInclude Include="src\http\http_client.hpp">

snowplow-cpp-tracker.vcxproj

@@ -106,7 +106,8 @@
     <ClCompile Include="include\sqlite3.c" />
     <ClCompile Include="src\client_session.cpp" />
     <ClCompile Include="src\cracked_url.cpp" />
-    <ClCompile Include="src\emitter.cpp" />
+    <ClCompile Include="src\emitter\emitter.cpp" />
+    <ClCompile Include="src\emitter\retry_delay.cpp" />
     <ClCompile Include="src\http\http_client_windows.cpp" />
     <ClCompile Include="src\http\http_request_result.cpp" />
     <ClCompile Include="src\payload\payload.cpp" />
@@ -124,7 +125,8 @@
     <ClCompile Include="test\http\test_http_client.cpp" />
     <ClCompile Include="test\client_session_test.cpp" />
     <ClCompile Include="test\cracked_url_test.cpp" />
-    <ClCompile Include="test\emitter_test.cpp" />
+    <ClCompile Include="test\emitter\emitter_test.cpp" />
+    <ClCompile Include="test\emitter\retry_delay_test.cpp" />
     <ClCompile Include="test\http\http_client_test.cpp" />
     <ClCompile Include="test\http\http_request_result_test.cpp" />
     <ClCompile Include="test\main.cpp" />
@@ -143,7 +145,8 @@
     <ClInclude Include="src\client_session.hpp" />
     <ClInclude Include="src\constants.hpp" />
     <ClInclude Include="src\cracked_url.hpp" />
-    <ClInclude Include="src\emitter.hpp" />
+    <ClInclude Include="src\emitter\emitter.hpp" />
+    <ClInclude Include="src\emitter\retry_delay.hpp" />
     <ClInclude Include="src\http\http_client.hpp" />
     <ClInclude Include="src\http\http_client_windows.hpp" />
     <ClInclude Include="src\http\http_request_result.hpp" />

snowplow-cpp-tracker.vcxproj.filters

@@ -21,7 +21,10 @@
     <ClCompile Include="src\cracked_url.cpp">
       <Filter>Source Files</Filter>
     </ClCompile>
-    <ClCompile Include="src\emitter.cpp">
+    <ClCompile Include="src\emitter\emitter.cpp">
+      <Filter>Source Files</Filter>
+    </ClCompile>
+    <ClCompile Include="src\emitter\retry_delay.cpp">
       <Filter>Source Files</Filter>
     </ClCompile>
     <ClCompile Include="src\http\http_client_windows.cpp">
@@ -81,7 +84,10 @@
     <ClCompile Include="test\cracked_url_test.cpp">
       <Filter>Source Files</Filter>
     </ClCompile>
-    <ClCompile Include="test\emitter_test.cpp">
+    <ClCompile Include="test\emitter\emitter_test.cpp">
+      <Filter>Source Files</Filter>
+    </ClCompile>
+    <ClCompile Include="test\emitter\retry_delay_test.cpp">
       <Filter>Source Files</Filter>
     </ClCompile>
     <ClCompile Include="test\http\http_client_test.cpp">
@@ -125,7 +131,10 @@
     <ClInclude Include="src\cracked_url.hpp">
       <Filter>Header Files</Filter>
     </ClInclude>
-    <ClInclude Include="src\emitter.hpp">
+    <ClInclude Include="src\emitter\emitter.hpp">
+      <Filter>Header Files</Filter>
+    </ClInclude>
+    <ClInclude Include="src\emitter\retry_delay.hpp">
       <Filter>Header Files</Filter>
     </ClInclude>
     <ClInclude Include="src\http\http_client.hpp">

src/constants.hpp

@@ -21,7 +21,7 @@ using std::string;
 using std::set;
 
 namespace snowplow {
-const string SNOWPLOW_TRACKER_VERSION_LABEL = "cpp-0.3.0";
+const string SNOWPLOW_TRACKER_VERSION_LABEL = "cpp-0.4.0";
 
 // post requests
 const string SNOWPLOW_POST_PROTOCOL_VENDOR = "com.snowplowanalytics.snowplow";
@@ -70,6 +70,7 @@ const string SNOWPLOW_COLOR_DEPTH = "cd";
 const string SNOWPLOW_TIMEZONE = "tz";
 const string SNOWPLOW_LANGUAGE = "lang";
 const string SNOWPLOW_USERAGENT = "ua";
+const string SNOWPLOW_IP_ADDRESS = "ip";
 
 // structured event
 const string SNOWPLOW_SE_CATEGORY = "se_ca";

src/emitter.cpp src/emitter/emitter.cppsrc/emitter.cpp renamed to src/emitter/emitter.cpp

@@ -26,32 +26,33 @@ using std::transform;
 using std::equal;
 using std::move;
 using std::future;
+using std::this_thread::sleep_for;
 
 const int post_wrapper_bytes = 88; // "schema":"iglu:com.snowplowanalytics.snowplow/payload_data/jsonschema/1-0-4","data":[]
 const int post_stm_bytes = 22;     // "stm":"1443452851000"
 
 #if defined(__APPLE__)
-#include "http/http_client_apple.hpp"
+#include "../http/http_client_apple.hpp"
 unique_ptr<HttpClient> createDefaultHttpClient() {
   return unique_ptr<HttpClient>(new HttpClientApple());
 }
 #elif defined(WIN32) || defined(_WIN32) || defined(__WIN32) && !defined(__CYGWIN__)
-#include "http/http_client_windows.hpp"
+#include "../http/http_client_windows.hpp"
 unique_ptr<HttpClient> createDefaultHttpClient() {
   return unique_ptr<HttpClient>(new HttpClientWindows());
 }
 #else
-#include "http/http_client_curl.hpp"
+#include "../http/http_client_curl.hpp"
 unique_ptr<HttpClient> createDefaultHttpClient() {
   return unique_ptr<HttpClient>(new HttpClientCurl());
 }
 #endif
 
-Emitter::Emitter(const string &uri, Method method, Protocol protocol, int send_limit,
-                 int byte_limit_post, int byte_limit_get, shared_ptr<EventStore> event_store) : Emitter(uri, method, protocol, send_limit, byte_limit_post, byte_limit_get, std::move(event_store), createDefaultHttpClient()) {
+Emitter::Emitter(const string &uri, Method method, Protocol protocol, int batch_size,
+                 int byte_limit_post, int byte_limit_get, shared_ptr<EventStore> event_store) : Emitter(uri, method, protocol, batch_size, byte_limit_post, byte_limit_get, std::move(event_store), createDefaultHttpClient()) {
 }
 
-Emitter::Emitter(const string &uri, Method method, Protocol protocol, int send_limit,
+Emitter::Emitter(const string &uri, Method method, Protocol protocol, int batch_size,
                  int byte_limit_post, int byte_limit_get, shared_ptr<EventStore> event_store, unique_ptr<HttpClient> http_client) : m_url(this->get_collector_url(uri, protocol, method)) {
 
   if (uri == "") {
@@ -74,7 +75,7 @@ Emitter::Emitter(const string &uri, Method method, Protocol protocol, int send_l
 
   this->m_running = false;
   this->m_method = method;
-  this->m_send_limit = send_limit;
+  this->m_batch_size = batch_size;
   this->m_byte_limit_post = byte_limit_post;
   this->m_byte_limit_get = byte_limit_get;
   this->m_event_store = move(event_store);
@@ -133,7 +134,7 @@ void Emitter::flush() {
 void Emitter::run() {
   do {
     list<EventRow> event_rows;
-    m_event_store->get_event_rows_batch(&event_rows, m_send_limit);
+    m_event_store->get_event_rows_batch(&event_rows, m_batch_size);
 
     if (event_rows.size() > 0) {
       // emit the events
@@ -163,6 +164,19 @@ void Emitter::run() {
       delete_row_ids.splice(delete_row_ids.end(), success_row_ids);
       delete_row_ids.splice(delete_row_ids.end(), failed_wont_retry_row_ids);
       m_event_store->delete_event_rows_with_ids(delete_row_ids);
+
+      // update retry delay calculation based on whether the requests will be retried
+      if (!failed_will_retry_row_ids.empty()) {
+        m_retry_delay.will_retry_emit();
+      } else {
+        m_retry_delay.wont_retry_emit();
+      }
+
+      // sleep for the retry delay if there is one
+      auto retry_delay = m_retry_delay.get();
+      if (retry_delay.count() > 0) {
+        sleep_for(retry_delay);
+      }
     } else {
       m_check_fin.notify_all();
 

src/emitter.hpp src/emitter/emitter.hppsrc/emitter.hpp renamed to src/emitter/emitter.hpp

@@ -20,14 +20,15 @@ See the Apache License Version 2.0 for the specific language governing permissio
 #include <future>
 #include <thread>
 #include <algorithm>
-#include "constants.hpp"
-#include "utils.hpp"
-#include "storage/event_store.hpp"
-#include "payload/payload.hpp"
-#include "payload/self_describing_json.hpp"
-#include "cracked_url.hpp"
-#include "http/http_request_result.hpp"
-#include "http/http_client.hpp"
+#include "../constants.hpp"
+#include "../utils.hpp"
+#include "../storage/event_store.hpp"
+#include "../payload/payload.hpp"
+#include "../payload/self_describing_json.hpp"
+#include "../cracked_url.hpp"
+#include "../http/http_request_result.hpp"
+#include "../http/http_client.hpp"
+#include "retry_delay.hpp"
 
 using std::string;
 using std::thread;
@@ -91,12 +92,12 @@ class Emitter {
    * @param uri The URI to send events to
    * @param method The request type to use (GET or POST)
    * @param protocol The protocol to use (http or https)
-   * @param send_limit The maximum amount of events to send at a time
+   * @param batch_size The maximum amount of events to send at a time
    * @param byte_limit_post The byte limit when sending a POST request
    * @param byte_limit_get The byte limit when sending a GET request
    * @param event_store Defines the database to use for event queue
    */
-  Emitter(const string & uri, Method method, Protocol protocol, int send_limit, 
+  Emitter(const string & uri, Method method, Protocol protocol, int batch_size, 
     int byte_limit_post, int byte_limit_get, shared_ptr<EventStore> event_store);
 
   /**
@@ -105,13 +106,13 @@ class Emitter {
    * @param uri The URI to send events to
    * @param method The request type to use (GET or POST)
    * @param protocol The protocol to use (http or https)
-   * @param send_limit The maximum amount of events to send at a time
+   * @param batch_size The maximum amount of events to send at a time
    * @param byte_limit_post The byte limit when sending a POST request
    * @param byte_limit_get The byte limit when sending a GET request
    * @param event_store Defines the database to use for event queue
    * @param http_client Unique pointer to a custom HTTP client to send GET and POST requests with
    */
-  Emitter(const string & uri, Method method, Protocol protocol, int send_limit, 
+  Emitter(const string & uri, Method method, Protocol protocol, int batch_size, 
     int byte_limit_post, int byte_limit_get, shared_ptr<EventStore> event_store, unique_ptr<HttpClient> http_client);
   ~Emitter();
 
@@ -152,11 +153,11 @@ class Emitter {
   Method get_method() const { return m_method; }
 
   /**
-   * @brief Get the send limit.
+   * @brief Get the batch size.
    * 
    * @return unsigned int The maximum amount of events to send at a time
    */
-  unsigned int get_send_limit() const { return m_send_limit; }
+  unsigned int get_batch_size() const { return m_batch_size; }
 
   /**
    * @brief Get the byte limit for GET.
@@ -208,7 +209,7 @@ class Emitter {
   Method m_method;
   shared_ptr<EventStore> m_event_store;
   unique_ptr<HttpClient> m_http_client;
-  unsigned int m_send_limit;
+  unsigned int m_batch_size;
   unsigned int m_byte_limit_get;
   unsigned int m_byte_limit_post;
 
@@ -222,6 +223,7 @@ class Emitter {
   EmitterCallback m_callback;
   EmitStatus m_callback_emit_status;
   map<int, bool> m_custom_retry_for_status_codes;
+  RetryDelay m_retry_delay;
 
   void run();
   void do_send(const list<EventRow> &event_rows, list<HttpRequestResult> *results);

src/emitter/retry_delay.cpp

@@ -0,0 +1,58 @@
+/*
+Copyright (c) 2022 Snowplow Analytics Ltd. All rights reserved.
+
+This program is licensed to you under the Apache License Version 2.0,
+and you may not use this file except in compliance with the Apache License Version 2.0.
+You may obtain a copy of the Apache License Version 2.0 at http://www.apache.org/licenses/LICENSE-2.0.
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the Apache License Version 2.0 is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the Apache License Version 2.0 for the specific language governing permissions and limitations there under.
+*/
+
+#include "retry_delay.hpp"
+#include <random>
+
+using namespace snowplow;
+using std::min;
+
+RetryDelay::RetryDelay(double base, double factor, int retry_count_cap, double jitter) {
+  m_base = base;
+  m_factor = factor;
+  m_retry_count_cap = retry_count_cap;
+  m_jitter = jitter;
+  m_retry_count = 0;
+}
+
+void RetryDelay::will_retry_emit() {
+  m_retry_count++;
+}
+
+void RetryDelay::wont_retry_emit() {
+  m_retry_count = 0;
+}
+
+milliseconds RetryDelay::get() const {
+  if (m_retry_count == 0) {
+    return milliseconds(0);
+  }
+
+  double delay_ms = m_base * pow(m_factor, min(m_retry_count, m_retry_count_cap) - 1);
+
+  if (m_jitter != 0) {
+    std::random_device rd;
+    std::mt19937 mt(rd());
+    std::uniform_real_distribution<double> dist(0.0, 1.0);
+    double seed = dist(mt);
+    double deviation = floor(seed * m_jitter * delay_ms);
+
+    if (round(seed) == 1) {
+      delay_ms -= deviation;
+    } else {
+      delay_ms += deviation;
+    }
+  }
+
+  return milliseconds((unsigned long) delay_ms);
+}