(ws cobra to sentry) handle null events for empty queues

docs/CHANGELOG.md

@@ -1,6 +1,22 @@
 # Changelog
 All changes to this project will be documented in this file.
 
+## [7.8.5] - 2019-12-28
+
+(ws cobra to sentry) handle null events for empty queues
+
+## [7.8.4] - 2019-12-27
+
+(ws cobra to sentry) game is picked in a fair manner, so that all games get the same share of sent events
+
+## [7.8.3] - 2019-12-27
+
+(ws cobra to sentry) refactor queue related code into a class
+
+## [7.8.2] - 2019-12-25
+
+(ws cobra to sentry) bound the queue size used to hold up cobra messages before they are sent to sentry. Default queue size is a 100 messages. Without such limit the program runs out of memory when a subscriber receive a lot of messages that cannot make it to sentry
+
 ## [7.8.1] - 2019-12-25
 
 (ws client) use correct compilation defines so that spdlog is not used as a header only library (reduce binary size and increase compilation speed)

ixwebsocket/IXWebSocketVersion.h

@@ -6,4 +6,4 @@
 
 #pragma once
 
-#define IX_WEBSOCKET_VERSION "7.7.1"
+#define IX_WEBSOCKET_VERSION "7.8.5"

ws/ws.cpp

@@ -105,6 +105,7 @@ int main(int argc, char** argv)
     int count = 1;
     int jobs = 4;
     uint32_t maxWaitBetweenReconnectionRetries;
+    size_t maxQueueSize = 100;
 
     auto addTLSOptions = [&tlsOptions, &verifyNone](CLI::App* app) {
         app->add_option(
@@ -268,6 +269,7 @@ int main(int argc, char** argv)
     cobra2sentry->add_option("--rolesecret", rolesecret, "Role secret")->required();
     cobra2sentry->add_option("--dsn", dsn, "Sentry DSN");
     cobra2sentry->add_option("--jobs", jobs, "Number of thread sending events to Sentry");
+    cobra2sentry->add_option("--queue_size", maxQueueSize, "Size of the queue to hold messages before they are sent to Sentry");
     cobra2sentry->add_option("channel", channel, "Channel")->required();
     cobra2sentry->add_flag("-v", verbose, "Verbose");
     cobra2sentry->add_flag("-s", strict, "Strict mode. Error out when sending to sentry fails");
@@ -455,6 +457,7 @@ int main(int argc, char** argv)
                                           verbose,
                                           strict,
                                           jobs,
+                                          maxQueueSize,
                                           tlsOptions);
     }
     else if (app.got_subcommand("cobra_metrics_to_redis"))

ws/ws.h

@@ -119,6 +119,7 @@ namespace ix
                                 bool verbose,
                                 bool strict,
                                 int jobs,
+                                size_t maxQueueSize,
                                 const ix::SocketTLSOptions& tlsOptions);
 
     int ws_cobra_metrics_to_redis(const std::string& appkey,

ws/ws_cobra_to_sentry.cpp

@@ -15,9 +15,81 @@
 #include <sstream>
 #include <thread>
 #include <vector>
+#include <map>
 
 namespace ix
 {
+    class QueueManager
+    {
+    public:
+        QueueManager(size_t maxQueueSize, std::atomic<bool> &stop) : _maxQueueSize(maxQueueSize), _stop(stop) {}
+
+        Json::Value pop();
+        void add(Json::Value msg);
+
+    private:
+        std::map<std::string, std::queue<Json::Value>> _queues;
+        std::mutex _mutex;
+        std::condition_variable _condition;
+        size_t _maxQueueSize;
+        std::atomic<bool>& _stop;
+    };
+
+    Json::Value QueueManager::pop()
+    {
+        std::unique_lock<std::mutex> lock(_mutex);
+
+        if (_queues.empty())
+        {
+            Json::Value val;
+            return val;
+        }
+
+        std::vector<std::string> games;
+        for (auto it : _queues)
+        {
+            games.push_back(it.first);
+        }
+
+        std::random_shuffle(games.begin(), games.end());
+        std::string game = games[0];
+
+        spdlog::info("Sending event for game '{}'", game);
+
+        _condition.wait(lock, [this] { return !_stop; });
+
+        if (_queues[game].empty())
+        {
+            Json::Value val;
+            return val;
+        }
+
+        auto msg = _queues[game].front();
+        _queues[game].pop();
+        return msg;
+    }
+
+    void QueueManager::add(Json::Value msg)
+    {
+        std::unique_lock<std::mutex> lock(_mutex);
+
+        std::string game;
+        if (msg.isMember("device") && msg["device"].isMember("game"))
+        {
+            game = msg["device"]["game"].asString();
+        }
+
+        if (game.empty()) return;
+
+        // if the sending is not fast enough there is no point
+        // in queuing too many events.
+        if (_queues[game].size() < _maxQueueSize)
+        {
+            _queues[game].push(msg);
+            _condition.notify_one();
+        }
+    }
+
     int ws_cobra_to_sentry_main(const std::string& appkey,
                                 const std::string& endpoint,
                                 const std::string& rolename,
@@ -28,6 +100,7 @@ namespace ix
                                 bool verbose,
                                 bool strict,
                                 int jobs,
+                                size_t maxQueueSize,
                                 const ix::SocketTLSOptions& tlsOptions)
     {
         ix::CobraConnection conn;
@@ -46,9 +119,7 @@ namespace ix
         std::atomic<bool> stop(false);
         std::atomic<bool> throttled(false);
 
-        std::condition_variable condition;
-        std::mutex conditionVariableMutex;
-        std::queue<Json::Value> queue;
+        QueueManager queueManager(maxQueueSize, stop);
 
         auto timer = [&sentCount, &receivedCount] {
             while (true)
@@ -62,9 +133,7 @@ namespace ix
 
         std::thread t1(timer);
 
-        auto sentrySender = [&condition,
-                             &conditionVariableMutex,
-                             &queue,
+        auto sentrySender = [&queueManager,
                              verbose,
                              &errorSending,
                              &sentCount,
@@ -75,15 +144,14 @@ namespace ix
 
             while (true)
             {
-                Json::Value msg;
+                Json::Value msg = queueManager.pop();
 
+                while (msg.isNull())
                 {
-                    std::unique_lock<std::mutex> lock(conditionVariableMutex);
-                    condition.wait(lock, [&queue, &stop] { return !queue.empty() && !stop; });
-
-                    msg = queue.front();
-                    queue.pop();
+                    msg = queueManager.pop();
+                    if (stop) return;
                 }
+                if (stop) return;
 
                 auto ret = sentryClient.send(msg, verbose);
                 HttpResponsePtr response = ret.first;
@@ -174,13 +242,12 @@ namespace ix
                                verbose,
                                &throttled,
                                &receivedCount,
-                               &condition,
-                               &conditionVariableMutex,
-                               &queue](ix::CobraConnectionEventType eventType,
-                                       const std::string& errMsg,
-                                       const ix::WebSocketHttpHeaders& headers,
-                                       const std::string& subscriptionId,
-                                       CobraConnection::MsgId msgId) {
+                               &queueManager](
+                                        ix::CobraConnectionEventType eventType,
+                                        const std::string& errMsg,
+                                        const ix::WebSocketHttpHeaders& headers,
+                                        const std::string& subscriptionId,
+                                        CobraConnection::MsgId msgId) {
             if (eventType == ix::CobraConnection_EventType_Open)
             {
                 spdlog::info("Subscriber connected");
@@ -203,9 +270,7 @@ namespace ix
                                 verbose,
                                 &throttled,
                                 &receivedCount,
-                                &condition,
-                                &conditionVariableMutex,
-                                &queue](const Json::Value& msg) {
+                                &queueManager](const Json::Value& msg) {
                                    if (verbose)
                                    {
                                        spdlog::info(jsonWriter.write(msg));
@@ -214,18 +279,11 @@ namespace ix
                                    // If we cannot send to sentry fast enough, drop the message
                                    if (throttled)
                                    {
-                                       condition.notify_one();
                                        return;
                                    }
 
                                    ++receivedCount;
-
-                                   {
-                                       std::unique_lock<std::mutex> lock(conditionVariableMutex);
-                                       queue.push(msg);
-                                   }
-
-                                   condition.notify_one();
+                                   queueManager.add(msg);
                                });
             }
             else if (eventType == ix::CobraConnection_EventType_Subscribed)