(cobra to statsd bot) bot init was missing + capture socket error

docs/CHANGELOG.md

@@ -1,6 +1,10 @@
 # Changelog
 All changes to this project will be documented in this file.
 
+## [9.1.8] - 2020-03-29
+
+(cobra to statsd bot) bot init was missing + capture socket error
+
 ## [9.1.7] - 2020-03-29
 
 (cobra to statsd bot) add ability to extract a numerical value and send a gauge event to statsd

ixbots/ixbots/IXStatsdClient.cpp

@@ -43,11 +43,10 @@
 #include <stdlib.h>
 #include <string.h>
 #include <stdio.h>
+#include <iostream>
 
 namespace ix
 {
-    const uint64_t StatsdClient::_maxQueueSize = 32768;
-
     StatsdClient::StatsdClient(const std::string& host,
                                int port,
                                const std::string& prefix)
@@ -56,23 +55,11 @@ namespace ix
       , _prefix(prefix)
       , _stop(false)
     {
-        _thread = std::thread([this] {
+        _thread = std::thread([this]
+        {
             while (!_stop)
             {
-                std::deque<std::string> stagedQueue;
+                flushQueue();
-
-                {
-                    std::lock_guard<std::mutex> lock(_mutex);
-                    _queue.swap(stagedQueue);
-                }
-
-                while (!stagedQueue.empty())
-                {
-                    auto message = stagedQueue.front();
-                    _socket.sendto(message);
-                    stagedQueue.pop_front();
-                }
-
                 std::this_thread::sleep_for(std::chrono::seconds(1));
             }
         });
@@ -127,31 +114,39 @@ namespace ix
         return send(key, ms, "ms");
     }
 
-    int StatsdClient::send(std::string key, size_t value, const std::string &type)
+    int StatsdClient::send(std::string key, size_t value, const std::string& type)
     {
         cleanup(key);
 
         char buf[256];
-        snprintf(buf, sizeof(buf), "%s%s:%zd|%s",
+        snprintf(buf, sizeof(buf), "%s%s:%zd|%s\n",
                  _prefix.c_str(), key.c_str(), value, type.c_str());
 
-        return send(buf);
+        enqueue(buf);
+        return 0;
     }
 
-    int StatsdClient::send(const std::string &message)
+    void StatsdClient::enqueue(const std::string& message)
+    {
+        std::lock_guard<std::mutex> lock(_mutex);
+        _queue.push_back(message);
+    }
+
+    void StatsdClient::flushQueue()
     {
         std::lock_guard<std::mutex> lock(_mutex);
 
-        if (_queue.empty() ||
+        while (!_queue.empty())
-            _queue.back().length() > _maxQueueSize)
         {
-            _queue.push_back(message);
+            auto message = _queue.front();
+            auto ret = _socket.sendto(message);
+            if (ret != 0)
+            {
+                std::cerr << "error: "
+                          << strerror(UdpSocket::getErrno())
+                          << std::endl;
+            }
+            _queue.pop_front();
         }
-        else
-        {
-            (*_queue.rbegin()).append("\n").append(message);
-        }
-
-        return 0;
     }
 } // end namespace ix

ixbots/ixbots/IXStatsdClient.h

@@ -32,11 +32,7 @@ namespace ix
         int timing(const std::string& key, size_t ms);
 
     private:
-        /**
+        void enqueue(const std::string& message);
-         * (Low Level Api) manually send a message
-         * which might be composed of several lines.
-         */
-        int send(const std::string& message);
 
         /* (Low Level Api) manually send a message
          * type = "c", "g" or "ms"
@@ -44,6 +40,7 @@ namespace ix
         int send(std::string key, size_t value, const std::string& type);
 
         void cleanup(std::string& key);
+        void flushQueue();
 
         UdpSocket _socket;
 
@@ -56,7 +53,6 @@ namespace ix
         std::mutex _mutex; // for the queue
 
         std::deque<std::string> _queue;
-        static const uint64_t _maxQueueSize;
     };
 
 } // end namespace ix

ixwebsocket/IXWebSocketVersion.h

@@ -6,4 +6,4 @@
 
 #pragma once
 
-#define IX_WEBSOCKET_VERSION "9.1.7"
+#define IX_WEBSOCKET_VERSION "9.1.8"

ws/ws.cpp

@@ -459,17 +459,27 @@ int main(int argc, char** argv)
         int runtime = -1;
         ix::StatsdClient statsdClient(hostname, statsdPort, prefix);
 
-        ret = ix::cobra_to_statsd_bot(cobraConfig,
+        std::string errMsg;
-                                      channel,
+        bool initialized = statsdClient.init(errMsg);
-                                      filter,
+        if (!initialized)
-                                      position,
+        {
-                                      statsdClient,
+            spdlog::error(errMsg);
-                                      fields,
+            ret = 0;
-                                      gauge,
+        }
-                                      verbose,
+        else
-                                      maxQueueSize,
+        {
-                                      enableHeartbeat,
+            ret = ix::cobra_to_statsd_bot(cobraConfig,
-                                      runtime);
+                                          channel,
+                                          filter,
+                                          position,
+                                          statsdClient,
+                                          fields,
+                                          gauge,
+                                          verbose,
+                                          maxQueueSize,
+                                          enableHeartbeat,
+                                          runtime);
+        }
     }
     else if (app.got_subcommand("cobra_to_sentry"))
     {