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

(cobra bots) add a utility class to factor out the common bots features (heartbeat) and move cobra to sentry bot to use it

Browse Source
This commit is contained in:
Benjamin Sergeant 2020-04-16 14:49:49 -07:00
parent ccfd196863
commit 0f5d15aa11
6 changed files with 364 additions and 257 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
docs/CHANGELOG.md
View File

@ -1,6 +1,10 @@
# Changelog
All changes to this project will be documented in this file.
## [9.3.1] - 2020-04-16
(cobra bots) add a utility class to factor out the common bots features (heartbeat) and move cobra to sentry bot to use it
## [9.3.0] - 2020-04-15
(websocket) add a positive number to the heartbeat message sent, incremented each time the heartbeat is sent

2
ixbots/CMakeLists.txt
View File

@ -4,6 +4,7 @@
#
set (IXBOTS_SOURCES
ixbots/IXCobraBot.cpp
ixbots/IXCobraToSentryBot.cpp
ixbots/IXCobraToStatsdBot.cpp
ixbots/IXQueueManager.cpp
@ -11,6 +12,7 @@ set (IXBOTS_SOURCES
)
set (IXBOTS_HEADERS
ixbots/IXCobraBot.h
ixbots/IXCobraToSentryBot.h
ixbots/IXCobraToStatsdBot.h
ixbots/IXQueueManager.h

247
ixbots/ixbots/IXCobraBot.cpp Normal file
View File

@ -0,0 +1,247 @@
/*
* IXCobraBot.cpp
* Author: Benjamin Sergeant
* Copyright (c) 2020 Machine Zone, Inc. All rights reserved.
*/
#include "IXCobraBot.h"
#include "IXQueueManager.h"
#include <algorithm>
#include <chrono>
#include <ixcobra/IXCobraConnection.h>
#include <spdlog/spdlog.h>
#include <sstream>
#include <thread>
#include <vector>
namespace ix
{
int64_t CobraBot::run(const CobraConfig& config,
const std::string& channel,
const std::string& filter,
const std::string& position,
bool verbose,
size_t maxQueueSize,
bool enableHeartbeat,
int runtime)
{
ix::CobraConnection conn;
conn.configure(config);
conn.connect();
Json::FastWriter jsonWriter;
std::atomic<uint64_t> sentCount(0);
std::atomic<uint64_t> receivedCount(0);
std::atomic<bool> stop(false);
std::atomic<bool> throttled(false);
std::atomic<bool> fatalCobraError(false);
QueueManager queueManager(maxQueueSize);
auto timer = [&sentCount, &receivedCount, &stop] {
while (!stop)
{
spdlog::info("messages received {} sent {}", receivedCount, sentCount);
auto duration = std::chrono::seconds(1);
std::this_thread::sleep_for(duration);
}
spdlog::info("timer thread done");
};
std::thread t1(timer);
auto heartbeat = [&sentCount, &receivedCount, &stop, &enableHeartbeat] {
std::string state("na");
if (!enableHeartbeat) return;
while (!stop)
{
std::stringstream ss;
ss << "messages received " << receivedCount;
ss << "messages sent " << sentCount;
std::string currentState = ss.str();
if (currentState == state)
{
spdlog::error("no messages received or sent for 1 minute, exiting");
exit(1);
}
state = currentState;
auto duration = std::chrono::minutes(1);
std::this_thread::sleep_for(duration);
}
spdlog::info("heartbeat thread done");
};
std::thread t2(heartbeat);
auto sender =
[this, &queueManager, verbose, &sentCount, &stop, &throttled] {
while (true)
{
Json::Value msg = queueManager.pop();
if (stop) break;
if (msg.isNull()) continue;
if (_onBotMessageCallback && _onBotMessageCallback(msg, verbose, throttled))
{
// That might be too noisy
if (verbose)
{
spdlog::info("cobra bot: sending succesfull");
}
++sentCount;
}
else
{
spdlog::error("cobra bot: error sending");
}
if (stop) break;
}
spdlog::info("sender thread done");
};
std::thread t3(sender);
conn.setEventCallback([&conn,
&channel,
&filter,
&position,
&jsonWriter,
verbose,
&throttled,
&receivedCount,
&fatalCobraError,
&queueManager](const CobraEventPtr& event)
{
if (event->type == ix::CobraEventType::Open)
{
spdlog::info("Subscriber connected");
for (auto&& it : event->headers)
{
spdlog::info("{}: {}", it.first, it.second);
}
}
else if (event->type == ix::CobraEventType::Closed)
{
spdlog::info("Subscriber closed: {}", event->errMsg);
}
else if (event->type == ix::CobraEventType::Authenticated)
{
spdlog::info("Subscriber authenticated");
conn.subscribe(channel,
filter,
position,
[&jsonWriter, verbose, &throttled, &receivedCount, &queueManager](
const Json::Value& msg, const std::string& position) {
if (verbose)
{
spdlog::info("Subscriber received message {} -> {}", position, jsonWriter.write(msg));
}
// If we cannot send to sentry fast enough, drop the message
if (throttled)
{
return;
}
++receivedCount;
queueManager.add(msg);
});
}
else if (event->type == ix::CobraEventType::Subscribed)
{
spdlog::info("Subscriber: subscribed to channel {}", event->subscriptionId);
}
else if (event->type == ix::CobraEventType::UnSubscribed)
{
spdlog::info("Subscriber: unsubscribed from channel {}", event->subscriptionId);
}
else if (event->type == ix::CobraEventType::Error)
{
spdlog::error("Subscriber: error {}", event->errMsg);
}
else if (event->type == ix::CobraEventType::Published)
{
spdlog::error("Published message hacked: {}", event->msgId);
}
else if (event->type == ix::CobraEventType::Pong)
{
spdlog::info("Received websocket pong: {}", event->errMsg);
}
else if (event->type == ix::CobraEventType::HandshakeError)
{
spdlog::error("Subscriber: Handshake error: {}", event->errMsg);
fatalCobraError = true;
}
else if (event->type == ix::CobraEventType::AuthenticationError)
{
spdlog::error("Subscriber: Authentication error: {}", event->errMsg);
fatalCobraError = true;
}
else if (event->type == ix::CobraEventType::SubscriptionError)
{
spdlog::error("Subscriber: Subscription error: {}", event->errMsg);
fatalCobraError = true;
}
});
// Run forever
if (runtime == -1)
{
while (true)
{
auto duration = std::chrono::seconds(1);
std::this_thread::sleep_for(duration);
if (fatalCobraError) break;
}
}
// 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);
if (fatalCobraError) break;
}
}
//
// Cleanup.
// join all the bg threads and stop them.
//
conn.disconnect();
stop = true;
// progress thread
t1.join();
// heartbeat thread
if (t2.joinable()) t2.join();
// sentry sender thread
t3.join();
return fatalCobraError ? -1 : (int64_t) sentCount;
}
void CobraBot::setOnBotMessageCallback(const OnBotMessageCallback& callback)
{
_onBotMessageCallback = callback;
}
}

42
ixbots/ixbots/IXCobraBot.h Normal file
View File

@ -0,0 +1,42 @@
/*
* IXCobraBot.h
* Author: Benjamin Sergeant
* Copyright (c) 2020 Machine Zone, Inc. All rights reserved.
*/
#pragma once
#include <ixcobra/IXCobraConfig.h>
#include <stddef.h>
#include <json/json.h>
#include <mutex>
#include <condition_variable>
#include <queue>
#include <map>
namespace ix
{
using OnBotMessageCallback = std::function<bool(const Json::Value&,
const bool verbose,
std::atomic<bool>&)>;
class CobraBot
{
public:
CobraBot() = default;
int64_t run(const CobraConfig& config,
const std::string& channel,
const std::string& filter,
const std::string& position,
bool verbose,
size_t maxQueueSize,
bool enableHeartbeat,
int runtime);
void setOnBotMessageCallback(const OnBotMessageCallback& callback);
private:
OnBotMessageCallback _onBotMessageCallback;
};
}

324
ixbots/ixbots/IXCobraToSentryBot.cpp
View File

@ -5,6 +5,7 @@
*/
#include "IXCobraToSentryBot.h"
#include "IXCobraBot.h"
#include "IXQueueManager.h"
#include <chrono>
@ -27,276 +28,87 @@ namespace ix
bool enableHeartbeat,
int runtime)
{
ix::CobraConnection conn;
conn.configure(config);
conn.connect();
CobraBot bot;
bot.setOnBotMessageCallback([&sentryClient](const Json::Value& msg,
const bool verbose,
std::atomic<bool>& throttled) -> bool {
auto ret = sentryClient.send(msg, verbose);
HttpResponsePtr response = ret.first;
Json::FastWriter jsonWriter;
std::atomic<uint64_t> sentCount(0);
std::atomic<uint64_t> receivedCount(0);
std::atomic<bool> errorSending(false);
std::atomic<bool> stop(false);
std::atomic<bool> throttled(false);
std::atomic<bool> fatalCobraError(false);
QueueManager queueManager(maxQueueSize);
auto timer = [&sentCount, &receivedCount, &stop] {
while (!stop)
if (!response)
{
spdlog::info("messages received {} sent {}", receivedCount, sentCount);
auto duration = std::chrono::seconds(1);
std::this_thread::sleep_for(duration);
spdlog::warn("Null HTTP Response");
return false;
}
spdlog::info("timer thread done");
};
std::thread t1(timer);
auto heartbeat = [&sentCount, &receivedCount, &stop, &enableHeartbeat] {
std::string state("na");
if (!enableHeartbeat) return;
while (!stop)
if (verbose)
{
std::stringstream ss;
ss << "messages received " << receivedCount;
ss << "messages sent " << sentCount;
std::string currentState = ss.str();
if (currentState == state)
{
spdlog::error("no messages received or sent for 1 minute, exiting");
exit(1);
}
state = currentState;
auto duration = std::chrono::minutes(1);
std::this_thread::sleep_for(duration);
}
spdlog::info("heartbeat thread done");
};
std::thread t2(heartbeat);
auto sentrySender =
[&queueManager, verbose, &errorSending, &sentCount, &stop, &throttled, &sentryClient] {
while (true)
{
Json::Value msg = queueManager.pop();
if (stop) break;
if (msg.isNull()) continue;
auto ret = sentryClient.send(msg, verbose);
HttpResponsePtr response = ret.first;
if (!response)
{
spdlog::warn("Null HTTP Response");
continue;
}
if (verbose)
{
for (auto it : response->headers)
{
spdlog::info("{}: {}", it.first, it.second);
}
spdlog::info("Upload size: {}", response->uploadSize);
spdlog::info("Download size: {}", response->downloadSize);
spdlog::info("Status: {}", response->statusCode);
if (response->errorCode != HttpErrorCode::Ok)
{
spdlog::info("error message: {}", response->errorMsg);
}
if (response->headers["Content-Type"] != "application/octet-stream")
{
spdlog::info("payload: {}", response->payload);
}
}
if (response->statusCode != 200)
{
spdlog::error("Error sending data to sentry: {}", response->statusCode);
spdlog::error("Body: {}", ret.second);
spdlog::error("Response: {}", response->payload);
errorSending = true;
// Error 429 Too Many Requests
if (response->statusCode == 429)
{
auto retryAfter = response->headers["Retry-After"];
std::stringstream ss;
ss << retryAfter;
int seconds;
ss >> seconds;
if (!ss.eof() || ss.fail())
{
seconds = 30;
spdlog::warn("Error parsing Retry-After header. "
"Using {} for the sleep duration",
seconds);
}
spdlog::warn("Error 429 - Too Many Requests. ws will sleep "
"and retry after {} seconds",
retryAfter);
throttled = true;
auto duration = std::chrono::seconds(seconds);
std::this_thread::sleep_for(duration);
throttled = false;
}
}
else
{
++sentCount;
}
if (stop) break;
}
spdlog::info("sentrySender thread done");
};
std::thread t3(sentrySender);
conn.setEventCallback([&conn,
&channel,
&filter,
&position,
&jsonWriter,
verbose,
&throttled,
&receivedCount,
&fatalCobraError,
&queueManager](const CobraEventPtr& event)
{
if (event->type == ix::CobraEventType::Open)
{
spdlog::info("Subscriber connected");
for (auto&& it : event->headers)
for (auto it : response->headers)
{
spdlog::info("{}: {}", it.first, it.second);
}
}
else if (event->type == ix::CobraEventType::Closed)
{
spdlog::info("Subscriber closed: {}", event->errMsg);
}
else if (event->type == ix::CobraEventType::Authenticated)
{
spdlog::info("Subscriber authenticated");
conn.subscribe(channel,
filter,
position,
[&jsonWriter, verbose, &throttled, &receivedCount, &queueManager](
const Json::Value& msg, const std::string& position) {
if (verbose)
{
spdlog::info("Subscriber received message {} -> {}", position, jsonWriter.write(msg));
}
// If we cannot send to sentry fast enough, drop the message
if (throttled)
{
return;
}
spdlog::info("Upload size: {}", response->uploadSize);
spdlog::info("Download size: {}", response->downloadSize);
++receivedCount;
queueManager.add(msg);
});
spdlog::info("Status: {}", response->statusCode);
if (response->errorCode != HttpErrorCode::Ok)
{
spdlog::info("error message: {}", response->errorMsg);
}
if (response->headers["Content-Type"] != "application/octet-stream")
{
spdlog::info("payload: {}", response->payload);
}
}
else if (event->type == ix::CobraEventType::Subscribed)
bool success = response->statusCode == 200;
if (!success)
{
spdlog::info("Subscriber: subscribed to channel {}", event->subscriptionId);
}
else if (event->type == ix::CobraEventType::UnSubscribed)
{
spdlog::info("Subscriber: unsubscribed from channel {}", event->subscriptionId);
}
else if (event->type == ix::CobraEventType::Error)
{
spdlog::error("Subscriber: error {}", event->errMsg);
}
else if (event->type == ix::CobraEventType::Published)
{
spdlog::error("Published message hacked: {}", event->msgId);
}
else if (event->type == ix::CobraEventType::Pong)
{
spdlog::info("Received websocket pong: {}", event->errMsg);
}
else if (event->type == ix::CobraEventType::HandshakeError)
{
spdlog::error("Subscriber: Handshake error: {}", event->errMsg);
fatalCobraError = true;
}
else if (event->type == ix::CobraEventType::AuthenticationError)
{
spdlog::error("Subscriber: Authentication error: {}", event->errMsg);
fatalCobraError = true;
}
else if (event->type == ix::CobraEventType::SubscriptionError)
{
spdlog::error("Subscriber: Subscription error: {}", event->errMsg);
fatalCobraError = true;
spdlog::error("Error sending data to sentry: {}", response->statusCode);
spdlog::error("Body: {}", ret.second);
spdlog::error("Response: {}", response->payload);
// Error 429 Too Many Requests
if (response->statusCode == 429)
{
auto retryAfter = response->headers["Retry-After"];
std::stringstream ss;
ss << retryAfter;
int seconds;
ss >> seconds;
if (!ss.eof() || ss.fail())
{
seconds = 30;
spdlog::warn("Error parsing Retry-After header. "
"Using {} for the sleep duration",
seconds);
}
spdlog::warn("Error 429 - Too Many Requests. ws will sleep "
"and retry after {} seconds",
retryAfter);
throttled = true;
auto duration = std::chrono::seconds(seconds);
std::this_thread::sleep_for(duration);
throttled = false;
}
}
return success;
});
// Run forever
if (runtime == -1)
{
while (true)
{
auto duration = std::chrono::seconds(1);
std::this_thread::sleep_for(duration);
if (strict && errorSending) break;
if (fatalCobraError) break;
}
}
// 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);
if (strict && errorSending) break;
if (fatalCobraError) break;
}
}
//
// Cleanup.
// join all the bg threads and stop them.
//
conn.disconnect();
stop = true;
// progress thread
t1.join();
// heartbeat thread
if (t2.joinable()) t2.join();
// sentry sender thread
t3.join();
return ((strict && errorSending) || fatalCobraError) ? -1 : (int) sentCount;
return bot.run(config,
channel,
filter,
position,
verbose,
maxQueueSize,
enableHeartbeat,
runtime);
}
} // namespace ix

2
ixwebsocket/IXWebSocketVersion.h
View File

@ -6,4 +6,4 @@
#pragma once
#define IX_WEBSOCKET_VERSION "9.3.0"
#define IX_WEBSOCKET_VERSION "9.3.1"