kapitan/IXWebSocket
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

cobra to statsd bot ported to windows + add unittest

Browse Source
This commit is contained in:
Benjamin Sergeant
2020-03-22 19:36:29 -07:00
parent 5ad54a8904
commit a0ffb2ba53
27 changed files with 653 additions and 689 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
ixbots/CMakeLists.txt
View File

@ -7,12 +7,14 @@ set (IXBOTS_SOURCES
ixbots/IXCobraToSentryBot.cpp
ixbots/IXCobraToStatsdBot.cpp
ixbots/IXQueueManager.cpp
ixbots/IXStatsdClient.cpp
)
set (IXBOTS_HEADERS
ixbots/IXCobraToSentryBot.h
ixbots/IXCobraToStatsdBot.h
ixbots/IXQueueManager.h
ixbots/IXStatsdClient.h
)
add_library(ixbots STATIC
@ -30,8 +32,6 @@ if (NOT SPDLOG_FOUND)
set(SPDLOG_INCLUDE_DIRS ../third_party/spdlog/include)
endif()
set(STATSD_CLIENT_INCLUDE_DIRS ../third_party/statsd-client-cpp/src)
set(IXBOTS_INCLUDE_DIRS
.
..
@ -39,7 +39,6 @@ set(IXBOTS_INCLUDE_DIRS
../ixcobra
../ixsentry
${JSONCPP_INCLUDE_DIRS}
${SPDLOG_INCLUDE_DIRS}
${STATSD_CLIENT_INCLUDE_DIRS})
${SPDLOG_INCLUDE_DIRS})
target_include_directories( ixbots PUBLIC ${IXBOTS_INCLUDE_DIRS} )

2
ixbots/ixbots/IXCobraToSentryBot.h
View File

@ -1,7 +1,7 @@
/*
* IXCobraToSentryBot.h
* Author: Benjamin Sergeant
* Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
* Copyright (c) 2019-2020 Machine Zone, Inc. All rights reserved.
*/
#pragma once

74
ixbots/ixbots/IXCobraToStatsdBot.cpp
View File

@ -6,6 +6,7 @@
#include "IXCobraToStatsdBot.h"
#include "IXQueueManager.h"
#include "IXStatsdClient.h"
#include <atomic>
#include <chrono>
@ -16,10 +17,6 @@
#include <thread>
#include <vector>
#ifndef _WIN32
#include <statsd_client.h>
#endif
namespace ix
{
// fields are command line argument that can be specified multiple times
@ -63,11 +60,12 @@ namespace ix
const std::string& channel,
const std::string& filter,
const std::string& position,
const std::string& host,
int port,
const std::string& prefix,
StatsdClient& statsdClient,
const std::string& fields,
bool verbose)
bool verbose,
size_t maxQueueSize,
bool enableHeartbeat,
int runtime)
{
ix::CobraConnection conn;
conn.configure(config);
@ -80,11 +78,10 @@ namespace ix
std::atomic<uint64_t> receivedCount(0);
std::atomic<bool> stop(false);
size_t maxQueueSize = 1000;
QueueManager queueManager(maxQueueSize);
auto timer = [&sentCount, &receivedCount] {
while (true)
auto timer = [&sentCount, &receivedCount, &stop] {
while (!stop)
{
spdlog::info("messages received {} sent {}", receivedCount, sentCount);
@ -95,9 +92,11 @@ namespace ix
std::thread t1(timer);
auto heartbeat = [&sentCount, &receivedCount] {
auto heartbeat = [&sentCount, &receivedCount, &enableHeartbeat] {
std::string state("na");
if (!enableHeartbeat) return;
while (true)
{
std::stringstream ss;
@ -120,21 +119,13 @@ namespace ix
std::thread t2(heartbeat);
auto statsdSender = [&queueManager, &host, &port, &sentCount, &tokens, &prefix, &stop] {
// statsd client
// test with netcat as a server: `nc -ul 8125`
bool statsdBatch = true;
#ifndef _WIN32
statsd::StatsdClient statsdClient(host, port, prefix, statsdBatch);
#else
int statsdClient;
#endif
auto statsdSender = [&statsdClient, &queueManager, &sentCount, &tokens, &stop] {
while (true)
{
Json::Value msg = queueManager.pop();
if (msg.isNull()) continue;
if (stop) return;
if (msg.isNull()) continue;
std::string id;
for (auto&& attr : tokens)
@ -143,11 +134,8 @@ namespace ix
id += extractAttr(attr, msg);
}
sentCount += 1;
#ifndef _WIN32
statsdClient.count(id, 1);
#endif
sentCount += 1;
}
};
@ -214,12 +202,38 @@ namespace ix
}
});
while (true)
// Run forever
if (runtime == -1)
{
std::chrono::duration<double, std::milli> duration(1000);
std::this_thread::sleep_for(duration);
while (true)
{
auto duration = std::chrono::seconds(1);
std::this_thread::sleep_for(duration);
}
}
// Run for a duration, used by unittesting now
else
{
for (int i = 0 ; i < runtime; ++i)
{
auto duration = std::chrono::seconds(1);
std::this_thread::sleep_for(duration);
}
}
return 0;
//
// Cleanup.
// join all the bg threads and stop them.
//
conn.disconnect();
stop = true;
t1.join();
if (t2.joinable()) t2.join();
spdlog::info("heartbeat thread done");
t3.join();
return (int) sentCount;
}
} // namespace ix

12
ixbots/ixbots/IXCobraToStatsdBot.h
View File

@ -1,11 +1,12 @@
/*
* IXCobraToStatsdBot.h
* Author: Benjamin Sergeant
* Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
* Copyright (c) 2019-2020 Machine Zone, Inc. All rights reserved.
*/
#pragma once
#include <ixcobra/IXCobraConfig.h>
#include <ixbots/IXStatsdClient.h>
#include <string>
#include <stddef.h>
@ -15,9 +16,10 @@ namespace ix
const std::string& channel,
const std::string& filter,
const std::string& position,
const std::string& host,
int port,
const std::string& prefix,
StatsdClient& statsdClient,
const std::string& fields,
bool verbose);
bool verbose,
size_t maxQueueSize,
bool enableHeartbeat,
int runtime);
} // namespace ix

154
ixbots/ixbots/IXStatsdClient.cpp Normal file
View File

@ -0,0 +1,154 @@
/*
* Copyright (c) 2014, Rex
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are met:
*
* * Redistributions of source code must retain the above copyright notice, this
* list of conditions and the following disclaimer.
*
* * Redistributions in binary form must reproduce the above copyright notice,
* this list of conditions and the following disclaimer in the documentation
* and/or other materials provided with the distribution.
*
* * Neither the name of the {organization} nor the names of its
* contributors may be used to endorse or promote products derived from
* this software without specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
* DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
* FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
* DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
* SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
* CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
* OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*/
/*
* IXStatsdClient.cpp
* Author: Benjamin Sergeant
* Copyright (c) 2020 Machine Zone, Inc. All rights reserved.
*/
// Adapted from statsd-client-cpp
// test with netcat as a server: `nc -ul 8125`
#include "IXStatsdClient.h"
#include <stdlib.h>
#include <string.h>
#include <stdio.h>
namespace ix
{
StatsdClient::StatsdClient(const string& host,
int port,
const string& prefix)
: _host(host)
, _port(port)
, _prefix(prefix)
, _stop(false)
{
_thread = std::thread([this] {
while (!_stop)
{
std::deque<std::string> staged_message_queue;
{
std::lock_guard<std::mutex> lock(_mutex);
batching_message_queue_.swap(staged_message_queue);
}
while (!staged_message_queue.empty())
{
auto message = staged_message_queue.front();
_socket.sendto(message);
staged_message_queue.pop_front();
}
std::this_thread::sleep_for(std::chrono::seconds(1));
}
});
}
StatsdClient::~StatsdClient()
{
_stop = true;
if (_thread.joinable()) _thread.join();
_socket.close();
}
bool StatsdClient::init(std::string& errMsg)
{
return _socket.init(_host, _port, errMsg);
}
/* will change the original string */
void StatsdClient::cleanup(string& key)
{
size_t pos = key.find_first_of(":|@");
while (pos != string::npos)
{
key[pos] = '_';
pos = key.find_first_of(":|@");
}
}
int StatsdClient::dec(const string& key)
{
return count(key, -1);
}
int StatsdClient::inc(const string& key)
{
return count(key, 1);
}
int StatsdClient::count(const string& key, size_t value)
{
return send(key, value, "c");
}
int StatsdClient::gauge(const string& key, size_t value)
{
return send(key, value, "g");
}
int StatsdClient::timing(const string& key, size_t ms)
{
return send(key, ms, "ms");
}
int StatsdClient::send(string key, size_t value, const string &type)
{
cleanup(key);
char buf[256];
snprintf(buf, sizeof(buf), "%s%s:%zd|%s",
_prefix.c_str(), key.c_str(), value, type.c_str());
return send(buf);
}
int StatsdClient::send(const string &message)
{
std::lock_guard<std::mutex> lock(_mutex);
if (batching_message_queue_.empty() ||
batching_message_queue_.back().length() > max_batching_size)
{
batching_message_queue_.push_back(message);
}
else
{
(*batching_message_queue_.rbegin()).append("\n").append(message);
}
return 0;
}
} // end namespace ix

62
ixbots/ixbots/IXStatsdClient.h Normal file
View File

@ -0,0 +1,62 @@
/*
* IXStatsdClient.h
* Author: Benjamin Sergeant
* Copyright (c) 2020 Machine Zone, Inc. All rights reserved.
*/
#pragma once
#include <ixwebsocket/IXNetSystem.h>
#include <ixwebsocket/IXUdpSocket.h>
#include <string>
#include <thread>
#include <deque>
#include <mutex>
#include <atomic>
namespace ix {
class StatsdClient {
public:
StatsdClient(const std::string& host="127.0.0.1",
int port=8125,
const std::string& prefix = "");
~StatsdClient();
bool init(std::string& errMsg);
int inc(const std::string& key);
int dec(const std::string& key);
int count(const std::string& key, size_t value);
int gauge(const std::string& key, size_t value);
int timing(const std::string& key, size_t ms);
private:
/**
* (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"
*/
int send(std::string key, size_t value, const std::string& type);
void cleanup(std::string& key);
UdpSocket _socket;
std::string _host;
int _port;
std::string _prefix;
std::atomic<bool> _stop;
std::thread _thread;
std::mutex _mutex; // for the queue
std::deque<std::string> batching_message_queue_;
const uint64_t max_batching_size = 32768;
};
} // end namespace ix