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 all bots to used it + convert cobra_subscribe to be a bot and add a unittest for it

Browse Source
This commit is contained in:
Benjamin Sergeant 2020-04-16 21:58:10 -07:00
parent 0f5d15aa11
commit a2abe861d3
20 changed files with 478 additions and 558 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
docs/CHANGELOG.md
View File

@ -3,7 +3,7 @@ All changes to this project will be documented in this file.
## [9.3.1] - 2020-04-16 ## [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 (cobra bots) add a utility class to factor out the common bots features (heartbeat) and move all bots to used it + convert cobra_subscribe to be a bot and add a unittest for it
## [9.3.0] - 2020-04-15 ## [9.3.0] - 2020-04-15

2
ixbots/CMakeLists.txt
View File

@ -7,6 +7,7 @@ set (IXBOTS_SOURCES
ixbots/IXCobraBot.cpp ixbots/IXCobraBot.cpp
ixbots/IXCobraToSentryBot.cpp ixbots/IXCobraToSentryBot.cpp
ixbots/IXCobraToStatsdBot.cpp ixbots/IXCobraToStatsdBot.cpp
ixbots/IXCobraToStdoutBot.cpp
ixbots/IXQueueManager.cpp ixbots/IXQueueManager.cpp
ixbots/IXStatsdClient.cpp ixbots/IXStatsdClient.cpp
) )
@ -15,6 +16,7 @@ set (IXBOTS_HEADERS
ixbots/IXCobraBot.h ixbots/IXCobraBot.h
ixbots/IXCobraToSentryBot.h ixbots/IXCobraToSentryBot.h
ixbots/IXCobraToStatsdBot.h ixbots/IXCobraToStatsdBot.h
ixbots/IXCobraToStdoutBot.h
ixbots/IXQueueManager.h ixbots/IXQueueManager.h
ixbots/IXStatsdClient.h ixbots/IXStatsdClient.h
) )

50
ixbots/ixbots/IXCobraBot.cpp
View File

@ -23,6 +23,7 @@ namespace ix
const std::string& position, const std::string& position,
bool verbose, bool verbose,
size_t maxQueueSize, size_t maxQueueSize,
bool useQueue,
bool enableHeartbeat, bool enableHeartbeat,
int runtime) int runtime)
{ {
@ -83,16 +84,18 @@ namespace ix
std::thread t2(heartbeat); std::thread t2(heartbeat);
auto sender = auto sender =
[this, &queueManager, verbose, &sentCount, &stop, &throttled] { [this, &queueManager, verbose, &sentCount, &stop, &throttled, &fatalCobraError] {
while (true) while (true)
{ {
Json::Value msg = queueManager.pop(); auto data = queueManager.pop();
Json::Value msg = data.first;
std::string position = data.second;
if (stop) break; if (stop) break;
if (msg.isNull()) continue; if (msg.isNull()) continue;
if (_onBotMessageCallback && _onBotMessageCallback(msg, verbose, throttled)) if (_onBotMessageCallback && _onBotMessageCallback(msg, position, verbose, throttled, fatalCobraError))
{ {
// That might be too noisy // That might be too noisy
if (verbose) if (verbose)
@ -114,16 +117,21 @@ namespace ix
std::thread t3(sender); std::thread t3(sender);
conn.setEventCallback([&conn, std::string subscriptionPosition(position);
conn.setEventCallback([this,
&conn,
&channel, &channel,
&filter, &filter,
&position, &subscriptionPosition,
&jsonWriter, &jsonWriter,
verbose, verbose,
&throttled, &throttled,
&receivedCount, &receivedCount,
&fatalCobraError, &fatalCobraError,
&queueManager](const CobraEventPtr& event) &useQueue,
&queueManager,
&sentCount](const CobraEventPtr& event)
{ {
if (event->type == ix::CobraEventType::Open) if (event->type == ix::CobraEventType::Open)
{ {
@ -141,16 +149,20 @@ namespace ix
else if (event->type == ix::CobraEventType::Authenticated) else if (event->type == ix::CobraEventType::Authenticated)
{ {
spdlog::info("Subscriber authenticated"); spdlog::info("Subscriber authenticated");
spdlog::info("Subscribing to {} at position {}", channel, subscriptionPosition);
spdlog::info("Using filter: {}", filter);
conn.subscribe(channel, conn.subscribe(channel,
filter, filter,
position, subscriptionPosition,
[&jsonWriter, verbose, &throttled, &receivedCount, &queueManager]( [this, &jsonWriter, verbose, &throttled, &receivedCount, &queueManager, &useQueue, &subscriptionPosition, &fatalCobraError, &sentCount](
const Json::Value& msg, const std::string& position) { const Json::Value& msg, const std::string& position) {
if (verbose) if (verbose)
{ {
spdlog::info("Subscriber received message {} -> {}", position, jsonWriter.write(msg)); spdlog::info("Subscriber received message {} -> {}", position, jsonWriter.write(msg));
} }
subscriptionPosition = position;
// If we cannot send to sentry fast enough, drop the message // If we cannot send to sentry fast enough, drop the message
if (throttled) if (throttled)
{ {
@ -158,7 +170,27 @@ namespace ix
} }
++receivedCount; ++receivedCount;
queueManager.add(msg);
if (useQueue)
{
queueManager.add(msg, position);
}
else
{
if (_onBotMessageCallback && _onBotMessageCallback(msg, position, verbose, throttled, fatalCobraError))
{
// That might be too noisy
if (verbose)
{
spdlog::info("cobra bot: sending succesfull");
}
++sentCount;
}
else
{
spdlog::error("cobra bot: error sending");
}
}
}); });
} }
else if (event->type == ix::CobraEventType::Subscribed) else if (event->type == ix::CobraEventType::Subscribed)

8
ixbots/ixbots/IXCobraBot.h
View File

@ -9,15 +9,14 @@
#include <ixcobra/IXCobraConfig.h> #include <ixcobra/IXCobraConfig.h>
#include <stddef.h> #include <stddef.h>
#include <json/json.h> #include <json/json.h>
#include <mutex> #include <functional>
#include <condition_variable>
#include <queue>
#include <map>
namespace ix namespace ix
{ {
using OnBotMessageCallback = std::function<bool(const Json::Value&, using OnBotMessageCallback = std::function<bool(const Json::Value&,
const std::string&,
const bool verbose, const bool verbose,
std::atomic<bool>&,
std::atomic<bool>&)>; std::atomic<bool>&)>;
class CobraBot class CobraBot
@ -31,6 +30,7 @@ namespace ix
const std::string& position, const std::string& position,
bool verbose, bool verbose,
size_t maxQueueSize, size_t maxQueueSize,
bool useQueue,
bool enableHeartbeat, bool enableHeartbeat,
int runtime); int runtime);

27
ixbots/ixbots/IXCobraToSentryBot.cpp
View File

@ -12,26 +12,26 @@
#include <ixcobra/IXCobraConnection.h> #include <ixcobra/IXCobraConnection.h>
#include <spdlog/spdlog.h> #include <spdlog/spdlog.h>
#include <sstream> #include <sstream>
#include <thread>
#include <vector> #include <vector>
namespace ix namespace ix
{ {
int cobra_to_sentry_bot(const CobraConfig& config, int64_t cobra_to_sentry_bot(const CobraConfig& config,
const std::string& channel, const std::string& channel,
const std::string& filter, const std::string& filter,
const std::string& position, const std::string& position,
SentryClient& sentryClient, SentryClient& sentryClient,
bool verbose, bool verbose,
bool strict, size_t maxQueueSize,
size_t maxQueueSize, bool enableHeartbeat,
bool enableHeartbeat, int runtime)
int runtime)
{ {
CobraBot bot; CobraBot bot;
bot.setOnBotMessageCallback([&sentryClient](const Json::Value& msg, bot.setOnBotMessageCallback([&sentryClient](const Json::Value& msg,
const std::string& /*position*/,
const bool verbose, const bool verbose,
std::atomic<bool>& throttled) -> bool { std::atomic<bool>& throttled,
std::atomic<bool>& /*fatalCobraError*/) -> bool {
auto ret = sentryClient.send(msg, verbose); auto ret = sentryClient.send(msg, verbose);
HttpResponsePtr response = ret.first; HttpResponsePtr response = ret.first;
@ -102,12 +102,15 @@ namespace ix
return success; return success;
}); });
bool useQueue = true;
return bot.run(config, return bot.run(config,
channel, channel,
filter, filter,
position, position,
verbose, verbose,
maxQueueSize, maxQueueSize,
useQueue,
enableHeartbeat, enableHeartbeat,
runtime); runtime);
} }

20
ixbots/ixbots/IXCobraToSentryBot.h
View File

@ -8,17 +8,17 @@
#include <ixcobra/IXCobraConfig.h> #include <ixcobra/IXCobraConfig.h>
#include <ixsentry/IXSentryClient.h> #include <ixsentry/IXSentryClient.h>
#include <string> #include <string>
#include <cstdint>
namespace ix namespace ix
{ {
int cobra_to_sentry_bot(const CobraConfig& config, int64_t cobra_to_sentry_bot(const CobraConfig& config,
const std::string& channel, const std::string& channel,
const std::string& filter, const std::string& filter,
const std::string& position, const std::string& position,
SentryClient& sentryClient, SentryClient& sentryClient,
bool verbose, bool verbose,
bool strict, size_t maxQueueSize,
size_t maxQueueSize, bool enableHeartbeat,
bool enableHeartbeat, int runtime);
int runtime);
} // namespace ix } // namespace ix

299
ixbots/ixbots/IXCobraToStatsdBot.cpp
View File

@ -7,14 +7,12 @@
#include "IXCobraToStatsdBot.h" #include "IXCobraToStatsdBot.h"
#include "IXQueueManager.h" #include "IXQueueManager.h"
#include "IXStatsdClient.h" #include "IXStatsdClient.h"
#include "IXCobraBot.h"
#include <atomic>
#include <chrono> #include <chrono>
#include <condition_variable>
#include <ixcobra/IXCobraConnection.h> #include <ixcobra/IXCobraConnection.h>
#include <spdlog/spdlog.h> #include <spdlog/spdlog.h>
#include <sstream> #include <sstream>
#include <thread>
#include <vector> #include <vector>
namespace ix namespace ix
@ -56,18 +54,18 @@ namespace ix
return val; return val;
} }
int cobra_to_statsd_bot(const ix::CobraConfig& config, int64_t cobra_to_statsd_bot(const ix::CobraConfig& config,
const std::string& channel, const std::string& channel,
const std::string& filter, const std::string& filter,
const std::string& position, const std::string& position,
StatsdClient& statsdClient, StatsdClient& statsdClient,
const std::string& fields, const std::string& fields,
const std::string& gauge, const std::string& gauge,
const std::string& timer, const std::string& timer,
bool verbose, bool verbose,
size_t maxQueueSize, size_t maxQueueSize,
bool enableHeartbeat, bool enableHeartbeat,
int runtime) int runtime)
{ {
ix::CobraConnection conn; ix::CobraConnection conn;
conn.configure(config); conn.configure(config);
@ -75,242 +73,85 @@ namespace ix
auto tokens = parseFields(fields); auto tokens = parseFields(fields);
Json::FastWriter jsonWriter; CobraBot bot;
std::atomic<uint64_t> sentCount(0); bot.setOnBotMessageCallback([&statsdClient, &tokens, &gauge, &timer](const Json::Value& msg,
std::atomic<uint64_t> receivedCount(0); const std::string& /*position*/,
std::atomic<bool> stop(false); const bool verbose,
std::atomic<bool> fatalCobraError(false); std::atomic<bool>& /*throttled*/,
std::atomic<bool>& fatalCobraError) -> bool {
QueueManager queueManager(maxQueueSize); std::string id;
for (auto&& attr : tokens)
auto progress = [&sentCount, &receivedCount, &stop] {
while (!stop)
{ {
spdlog::info("messages received {} sent {}", receivedCount, sentCount); id += ".";
auto val = extractAttr(attr, msg);
auto duration = std::chrono::seconds(1); id += val.asString();
std::this_thread::sleep_for(duration);
} }
spdlog::info("timer thread done"); if (gauge.empty() && timer.empty())
};
std::thread t1(progress);
auto heartbeat = [&sentCount, &receivedCount, &stop, &enableHeartbeat] {
std::string state("na");
if (!enableHeartbeat) return;
while (!stop)
{ {
std::stringstream ss; statsdClient.count(id, 1);
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);
} }
else
spdlog::info("heartbeat thread done");
};
std::thread t2(heartbeat);
auto statsdSender = [&statsdClient, &queueManager, &sentCount, &tokens, &stop, &gauge, &timer, &fatalCobraError, &verbose] {
while (true)
{ {
Json::Value msg = queueManager.pop(); std::string attrName = (!gauge.empty()) ? gauge : timer;
auto val = extractAttr(attrName, msg);
size_t x;
if (stop) return; if (val.isInt())
if (msg.isNull()) continue;
std::string id;
for (auto&& attr : tokens)
{ {
id += "."; x = (size_t) val.asInt();
auto val = extractAttr(attr, msg);
id += val.asString();
} }
else if (val.isInt64())
if (gauge.empty() && timer.empty())
{ {
statsdClient.count(id, 1); x = (size_t) val.asInt64();
}
else if (val.isUInt())
{
x = (size_t) val.asUInt();
}
else if (val.isUInt64())
{
x = (size_t) val.asUInt64();
}
else if (val.isDouble())
{
x = (size_t) val.asUInt64();
} }
else else
{ {
std::string attrName = (!gauge.empty()) ? gauge : timer; spdlog::error("Gauge {} is not a numeric type", gauge);
auto val = extractAttr(attrName, msg);
size_t x;
if (val.isInt())
{
x = (size_t) val.asInt();
}
else if (val.isInt64())
{
x = (size_t) val.asInt64();
}
else if (val.isUInt())
{
x = (size_t) val.asUInt();
}
else if (val.isUInt64())
{
x = (size_t) val.asUInt64();
}
else if (val.isDouble())
{
x = (size_t) val.asUInt64();
}
else
{
spdlog::error("Gauge {} is not a numberic type", gauge);
fatalCobraError = true;
break;
}
if (verbose)
{
spdlog::info("{} - {} -> {}", id, attrName, x);
}
if (!gauge.empty())
{
statsdClient.gauge(id, x);
}
else
{
statsdClient.timing(id, x);
}
}
sentCount += 1;
}
};
std::thread t3(statsdSender);
conn.setEventCallback(
[&conn, &channel, &filter, &position, &jsonWriter, verbose, &queueManager, &receivedCount, &fatalCobraError](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, &queueManager, verbose, &receivedCount](
const Json::Value& msg, const std::string& position) {
if (verbose)
{
spdlog::info("Subscriber received message {} -> {}", position, jsonWriter.write(msg));
}
receivedCount++;
++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");
}
else if (event->type == ix::CobraEventType::HandshakeError)
{
spdlog::error("Subscriber: Handshake error: {}", event->errMsg);
fatalCobraError = true; fatalCobraError = true;
return false;
} }
else if (event->type == ix::CobraEventType::AuthenticationError)
if (verbose)
{ {
spdlog::error("Subscriber: Authentication error: {}", event->errMsg); spdlog::info("{} - {} -> {}", id, attrName, x);
fatalCobraError = true;
} }
else if (event->type == ix::CobraEventType::SubscriptionError)
if (!gauge.empty())
{ {
spdlog::error("Subscriber: Subscription error: {}", event->errMsg); statsdClient.gauge(id, x);
fatalCobraError = true; }
else
{
statsdClient.timing(id, x);
} }
});
// 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; return true;
} });
}
// bool useQueue = true;
// Cleanup.
// join all the bg threads and stop them.
//
conn.disconnect();
stop = true;
// progress thread return bot.run(config,
t1.join(); channel,
filter,
// heartbeat thread position,
if (t2.joinable()) t2.join(); verbose,
maxQueueSize,
// statsd sender thread useQueue,
t3.join(); enableHeartbeat,
runtime);
return fatalCobraError ? -1 : (int) sentCount;
} }
} // namespace ix } // namespace ix

25
ixbots/ixbots/IXCobraToStatsdBot.h
View File

@ -9,19 +9,20 @@
#include <ixbots/IXStatsdClient.h> #include <ixbots/IXStatsdClient.h>
#include <string> #include <string>
#include <stddef.h> #include <stddef.h>
#include <cstdint>
namespace ix namespace ix
{ {
int cobra_to_statsd_bot(const ix::CobraConfig& config, int64_t cobra_to_statsd_bot(const ix::CobraConfig& config,
const std::string& channel, const std::string& channel,
const std::string& filter, const std::string& filter,
const std::string& position, const std::string& position,
StatsdClient& statsdClient, StatsdClient& statsdClient,
const std::string& fields, const std::string& fields,
const std::string& gauge, const std::string& gauge,
const std::string& timer, const std::string& timer,
bool verbose, bool verbose,
size_t maxQueueSize, size_t maxQueueSize,
bool enableHeartbeat, bool enableHeartbeat,
int runtime); int runtime);
} // namespace ix } // namespace ix

106
ixbots/ixbots/IXCobraToStdoutBot.cpp Normal file
View File

@ -0,0 +1,106 @@
/*
* IXCobraToStdoutBot.cpp
* Author: Benjamin Sergeant
* Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
*/
#include "IXCobraToStdoutBot.h"
#include "IXCobraBot.h"
#include "IXQueueManager.h"
#include <chrono>
#include <spdlog/spdlog.h>
#include <sstream>
#include <iostream>
namespace ix
{
using StreamWriterPtr = std::unique_ptr<Json::StreamWriter>;
StreamWriterPtr makeStreamWriter()
{
Json::StreamWriterBuilder builder;
builder["commentStyle"] = "None";
builder["indentation"] = ""; // will make the JSON object compact
std::unique_ptr<Json::StreamWriter> jsonWriter(builder.newStreamWriter());
return jsonWriter;
}
std::string timeSinceEpoch()
{
std::chrono::system_clock::time_point tp = std::chrono::system_clock::now();
std::chrono::system_clock::duration dtn = tp.time_since_epoch();
std::stringstream ss;
ss << dtn.count() * std::chrono::system_clock::period::num /
std::chrono::system_clock::period::den;
return ss.str();
}
void writeToStdout(bool fluentd,
const StreamWriterPtr& jsonWriter,
const Json::Value& msg,
const std::string& position)
{
Json::Value enveloppe;
if (fluentd)
{
enveloppe["producer"] = "cobra";
enveloppe["consumer"] = "fluentd";
Json::Value nestedMessage(msg);
nestedMessage["position"] = position;
nestedMessage["created_at"] = timeSinceEpoch();
enveloppe["message"] = nestedMessage;
jsonWriter->write(enveloppe, &std::cout);
std::cout << std::endl; // add lf and flush
}
else
{
enveloppe = msg;
std::cout << position << " ";
jsonWriter->write(enveloppe, &std::cout);
std::cout << std::endl;
}
}
int64_t cobra_to_stdout_bot(const CobraConfig& config,
const std::string& channel,
const std::string& filter,
const std::string& position,
bool fluentd,
bool quiet,
bool verbose,
size_t maxQueueSize,
bool enableHeartbeat,
int runtime)
{
CobraBot bot;
auto jsonWriter = makeStreamWriter();
bot.setOnBotMessageCallback([&fluentd, &quiet, &jsonWriter](const Json::Value& msg,
const std::string& position,
const bool /*verbose*/,
std::atomic<bool>& /*throttled*/,
std::atomic<bool>& /*fatalCobraError*/) -> bool {
if (!quiet)
{
writeToStdout(fluentd, jsonWriter, msg, position);
}
return true;
});
bool useQueue = false;
return bot.run(config,
channel,
filter,
position,
verbose,
maxQueueSize,
useQueue,
enableHeartbeat,
runtime);
}
} // namespace ix

25
ixbots/ixbots/IXCobraToStdoutBot.h Normal file
View File

@ -0,0 +1,25 @@
/*
* IXCobraToStdoutBot.h
* Author: Benjamin Sergeant
* Copyright (c) 2019-2020 Machine Zone, Inc. All rights reserved.
*/
#pragma once
#include <ixcobra/IXCobraConfig.h>
#include <string>
#include <stddef.h>
#include <cstdint>
namespace ix
{
int64_t cobra_to_stdout_bot(const ix::CobraConfig& config,
const std::string& channel,
const std::string& filter,
const std::string& position,
bool fluentd,
bool quiet,
bool verbose,
size_t maxQueueSize,
bool enableHeartbeat,
int runtime);
} // namespace ix

10
ixbots/ixbots/IXQueueManager.cpp
View File

@ -10,14 +10,14 @@
namespace ix namespace ix
{ {
Json::Value QueueManager::pop() std::pair<Json::Value, std::string> QueueManager::pop()
{ {
std::unique_lock<std::mutex> lock(_mutex); std::unique_lock<std::mutex> lock(_mutex);
if (_queues.empty()) if (_queues.empty())
{ {
Json::Value val; Json::Value val;
return val; return std::make_pair(val, std::string());
} }
std::vector<std::string> games; std::vector<std::string> games;
@ -35,7 +35,7 @@ namespace ix
if (_queues[game].empty()) if (_queues[game].empty())
{ {
Json::Value val; Json::Value val;
return val; return std::make_pair(val, std::string());
} }
auto msg = _queues[game].front(); auto msg = _queues[game].front();
@ -43,7 +43,7 @@ namespace ix
return msg; return msg;
} }
void QueueManager::add(Json::Value msg) void QueueManager::add(const Json::Value& msg, const std::string& position)
{ {
std::unique_lock<std::mutex> lock(_mutex); std::unique_lock<std::mutex> lock(_mutex);
@ -59,7 +59,7 @@ namespace ix
// in queuing too many events. // in queuing too many events.
if (_queues[game].size() < _maxQueueSize) if (_queues[game].size() < _maxQueueSize)
{ {
_queues[game].push(msg); _queues[game].push(std::make_pair(msg, position));
_condition.notify_one(); _condition.notify_one();
} }
} }

6
ixbots/ixbots/IXQueueManager.h
View File

@ -23,11 +23,11 @@ namespace ix
{ {
} }
Json::Value pop(); std::pair<Json::Value, std::string> pop();
void add(Json::Value msg); void add(const Json::Value& msg, const std::string& position);
private: private:
std::map<std::string, std::queue<Json::Value>> _queues; std::map<std::string, std::queue<std::pair<Json::Value, std::string>>> _queues;
std::mutex _mutex; std::mutex _mutex;
std::condition_variable _condition; std::condition_variable _condition;
size_t _maxQueueSize; size_t _maxQueueSize;

1
test/CMakeLists.txt
View File

@ -66,6 +66,7 @@ if (UNIX)
IXCobraMetricsPublisherTest.cpp IXCobraMetricsPublisherTest.cpp
IXCobraToSentryBotTest.cpp IXCobraToSentryBotTest.cpp
IXCobraToStatsdBotTest.cpp IXCobraToStatsdBotTest.cpp
IXCobraToStdoutBotTest.cpp
) )
endif() endif()

20
test/IXCobraToSentryBotTest.cpp
View File

@ -141,7 +141,6 @@ TEST_CASE("Cobra_to_sentry_bot", "[cobra_bots]")
std::string filter; std::string filter;
std::string position("$"); std::string position("$");
bool verbose = true; bool verbose = true;
bool strict = true;
size_t maxQueueSize = 10; size_t maxQueueSize = 10;
bool enableHeartbeat = false; bool enableHeartbeat = false;
@ -161,16 +160,15 @@ TEST_CASE("Cobra_to_sentry_bot", "[cobra_bots]")
// Only run the bot for 3 seconds // Only run the bot for 3 seconds
int runtime = 3; int runtime = 3;
int sentCount = cobra_to_sentry_bot(config, int64_t sentCount = cobra_to_sentry_bot(config,
channel, channel,
filter, filter,
position, position,
sentryClient, sentryClient,
verbose, verbose,
strict, maxQueueSize,
maxQueueSize, enableHeartbeat,
enableHeartbeat, runtime);
runtime);
// //
// We want at least 2 messages to be sent // We want at least 2 messages to be sent
// //

24
test/IXCobraToStatsdBotTest.cpp
View File

@ -114,18 +114,18 @@ TEST_CASE("Cobra_to_statsd_bot", "[cobra_bots]")
std::string gauge; std::string gauge;
std::string timer; std::string timer;
int sentCount = ix::cobra_to_statsd_bot(config, int64_t sentCount = ix::cobra_to_statsd_bot(config,
channel, channel,
filter, filter,
position, position,
statsdClient, statsdClient,
fields, fields,
gauge, gauge,
timer, timer,
verbose, verbose,
maxQueueSize, maxQueueSize,
enableHeartbeat, enableHeartbeat,
runtime); runtime);
// //
// We want at least 2 messages to be sent // We want at least 2 messages to be sent
// //

127
test/IXCobraToStdoutBotTest.cpp Normal file
View File

@ -0,0 +1,127 @@
/*
* IXCobraToStdoutTest.cpp
* Author: Benjamin Sergeant
* Copyright (c) 2020 Machine Zone. All rights reserved.
*/
#include "IXTest.h"
#include "catch.hpp"
#include <chrono>
#include <iostream>
#include <ixbots/IXCobraToStdoutBot.h>
#include <ixcobra/IXCobraConnection.h>
#include <ixcobra/IXCobraMetricsPublisher.h>
#include <ixcrypto/IXUuid.h>
#include <ixsentry/IXSentryClient.h>
#include <ixsnake/IXRedisServer.h>
#include <ixsnake/IXSnakeServer.h>
#include <ixwebsocket/IXHttpServer.h>
#include <ixwebsocket/IXUserAgent.h>
using namespace ix;
namespace
{
void runPublisher(const ix::CobraConfig& config, const std::string& channel)
{
ix::CobraMetricsPublisher cobraMetricsPublisher;
cobraMetricsPublisher.configure(config, channel);
cobraMetricsPublisher.setSession(uuid4());
cobraMetricsPublisher.enable(true);
Json::Value msg;
msg["fps"] = 60;
cobraMetricsPublisher.setGenericAttributes("game", "ody");
// Wait a bit
ix::msleep(500);
// publish some messages
cobraMetricsPublisher.push("sms_metric_A_id", msg); // (msg #1)
cobraMetricsPublisher.push("sms_metric_B_id", msg); // (msg #2)
ix::msleep(500);
cobraMetricsPublisher.push("sms_metric_A_id", msg); // (msg #3)
cobraMetricsPublisher.push("sms_metric_D_id", msg); // (msg #4)
ix::msleep(500);
cobraMetricsPublisher.push("sms_metric_A_id", msg); // (msg #4)
cobraMetricsPublisher.push("sms_metric_F_id", msg); // (msg #5)
ix::msleep(500);
}
} // namespace
TEST_CASE("Cobra_to_stdout_bot", "[cobra_bots]")
{
SECTION("Exchange and count sent/received messages.")
{
int port = getFreePort();
snake::AppConfig appConfig = makeSnakeServerConfig(port, true);
// Start a redis server
ix::RedisServer redisServer(appConfig.redisPort);
auto res = redisServer.listen();
REQUIRE(res.first);
redisServer.start();
// Start a snake server
snake::SnakeServer snakeServer(appConfig);
snakeServer.run();
// Run the bot for a small amount of time
std::string channel = ix::generateSessionId();
std::string appkey("FC2F10139A2BAc53BB72D9db967b024f");
std::string role = "_sub";
std::string secret = "66B1dA3ED5fA074EB5AE84Dd8CE3b5ba";
std::string endpoint = makeCobraEndpoint(port, true);
ix::CobraConfig config;
config.endpoint = endpoint;
config.appkey = appkey;
config.rolename = role;
config.rolesecret = secret;
config.socketTLSOptions = makeClientTLSOptions();
std::thread publisherThread(runPublisher, config, channel);
std::string filter;
std::string position("$");
bool verbose = true;
bool quiet = false;
size_t maxQueueSize = 10;
bool enableHeartbeat = false;
// Only run the bot for 3 seconds
int runtime = 3;
// We could try to capture the output ... not sure how.
bool fluentd = true;
int64_t sentCount = ix::cobra_to_stdout_bot(config,
channel,
filter,
position,
fluentd,
quiet,
verbose,
maxQueueSize,
enableHeartbeat,
runtime);
//
// We want at least 2 messages to be sent
//
REQUIRE(sentCount >= 2);
// Give us 1s for all messages to be received
ix::msleep(1000);
spdlog::info("Stopping snake server...");
snakeServer.stop();
spdlog::info("Stopping redis server...");
redisServer.stop();
publisherThread.join();
}
}

1
ws/CMakeLists.txt
View File

@ -49,7 +49,6 @@ add_executable(ws
ws_redis_subscribe.cpp ws_redis_subscribe.cpp
ws_redis_server.cpp ws_redis_server.cpp
ws_snake.cpp ws_snake.cpp
ws_cobra_subscribe.cpp
ws_cobra_metrics_publish.cpp ws_cobra_metrics_publish.cpp
ws_cobra_publish.cpp ws_cobra_publish.cpp
ws_cobra_metrics_to_redis.cpp ws_cobra_metrics_to_redis.cpp

60
ws/ws.cpp
View File

@ -13,6 +13,7 @@
#include <fstream> #include <fstream>
#include <ixbots/IXCobraToSentryBot.h> #include <ixbots/IXCobraToSentryBot.h>
#include <ixbots/IXCobraToStatsdBot.h> #include <ixbots/IXCobraToStatsdBot.h>
#include <ixbots/IXCobraToStdoutBot.h>
#include <ixcore/utils/IXCoreLogger.h> #include <ixcore/utils/IXCoreLogger.h>
#include <ixsentry/IXSentryClient.h> #include <ixsentry/IXSentryClient.h>
#include <ixwebsocket/IXNetSystem.h> #include <ixwebsocket/IXNetSystem.h>
@ -93,7 +94,6 @@ int main(int argc, char** argv)
bool quiet = false; bool quiet = false;
bool fluentd = false; bool fluentd = false;
bool compress = false; bool compress = false;
bool strict = false;
bool stress = false; bool stress = false;
bool disableAutomaticReconnection = false; bool disableAutomaticReconnection = false;
bool disablePerMessageDeflate = false; bool disablePerMessageDeflate = false;
@ -291,7 +291,6 @@ int main(int argc, char** argv)
"Size of the queue to hold messages before they are sent to Sentry"); "Size of the queue to hold messages before they are sent to Sentry");
cobra2sentry->add_option("channel", channel, "Channel")->required(); cobra2sentry->add_option("channel", channel, "Channel")->required();
cobra2sentry->add_flag("-v", verbose, "Verbose"); cobra2sentry->add_flag("-v", verbose, "Verbose");
cobra2sentry->add_flag("-s", strict, "Strict mode. Error out when sending to sentry fails");
cobra2sentry->add_option("--pidfile", pidfile, "Pid file"); cobra2sentry->add_option("--pidfile", pidfile, "Pid file");
cobra2sentry->add_option("--filter", filter, "Stream SQL Filter"); cobra2sentry->add_option("--filter", filter, "Stream SQL Filter");
cobra2sentry->add_option("--position", position, "Stream position"); cobra2sentry->add_option("--position", position, "Stream position");
@ -451,8 +450,18 @@ int main(int argc, char** argv)
} }
else if (app.got_subcommand("cobra_subscribe")) else if (app.got_subcommand("cobra_subscribe"))
{ {
ret = ix::ws_cobra_subscribe_main( bool enableHeartbeat = true;
cobraConfig, channel, filter, position, quiet, fluentd, runtime); int64_t sentCount = ix::cobra_to_stdout_bot(cobraConfig,
channel,
filter,
position,
fluentd,
quiet,
verbose,
maxQueueSize,
enableHeartbeat,
runtime);
ret = (int) sentCount;
} }
else if (app.got_subcommand("cobra_publish")) else if (app.got_subcommand("cobra_publish"))
{ {
@ -484,18 +493,18 @@ int main(int argc, char** argv)
} }
else else
{ {
ret = ix::cobra_to_statsd_bot(cobraConfig, ret = (int) ix::cobra_to_statsd_bot(cobraConfig,
channel, channel,
filter, filter,
position, position,
statsdClient, statsdClient,
fields, fields,
gauge, gauge,
timer, timer,
verbose, verbose,
maxQueueSize, maxQueueSize,
enableHeartbeat, enableHeartbeat,
runtime); runtime);
} }
} }
} }
@ -505,16 +514,15 @@ int main(int argc, char** argv)
ix::SentryClient sentryClient(dsn); ix::SentryClient sentryClient(dsn);
sentryClient.setTLSOptions(tlsOptions); sentryClient.setTLSOptions(tlsOptions);
ret = ix::cobra_to_sentry_bot(cobraConfig, ret = (int) ix::cobra_to_sentry_bot(cobraConfig,
channel, channel,
filter, filter,
position, position,
sentryClient, sentryClient,
verbose, verbose,
strict, maxQueueSize,
maxQueueSize, enableHeartbeat,
enableHeartbeat, runtime);
runtime);
} }
else if (app.got_subcommand("cobra_metrics_to_redis")) else if (app.got_subcommand("cobra_metrics_to_redis"))
{ {

8
ws/ws.h
View File

@ -77,14 +77,6 @@ namespace ix
const std::string& channel, const std::string& channel,
bool verbose); bool verbose);
int ws_cobra_subscribe_main(const ix::CobraConfig& config,
const std::string& channel,
const std::string& filter,
const std::string& position,
bool quiet,
bool fluentd,
int runtime);
int ws_cobra_publish_main(const ix::CobraConfig& appkey, int ws_cobra_publish_main(const ix::CobraConfig& appkey,
const std::string& channel, const std::string& channel,
const std::string& path); const std::string& path);

215
ws/ws_cobra_subscribe.cpp
View File

@ -1,215 +0,0 @@
/*
* ws_cobra_subscribe.cpp
* Author: Benjamin Sergeant
* Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
*/
#include <atomic>
#include <chrono>
#include <iostream>
#include <ixcobra/IXCobraConnection.h>
#include <spdlog/spdlog.h>
#include <sstream>
#include <thread>
namespace ix
{
using StreamWriterPtr = std::unique_ptr<Json::StreamWriter>;
StreamWriterPtr makeStreamWriter()
{
Json::StreamWriterBuilder builder;
builder["commentStyle"] = "None";
builder["indentation"] = ""; // will make the JSON object compact
std::unique_ptr<Json::StreamWriter> jsonWriter(builder.newStreamWriter());
return jsonWriter;
}
std::string timeSinceEpoch()
{
std::chrono::system_clock::time_point tp = std::chrono::system_clock::now();
std::chrono::system_clock::duration dtn = tp.time_since_epoch();
std::stringstream ss;
ss << dtn.count() * std::chrono::system_clock::period::num /
std::chrono::system_clock::period::den;
return ss.str();
}
void writeToStdout(bool fluentd,
const StreamWriterPtr& jsonWriter,
const Json::Value& msg,
const std::string& position)
{
Json::Value enveloppe;
if (fluentd)
{
enveloppe["producer"] = "cobra";
enveloppe["consumer"] = "fluentd";
Json::Value nestedMessage(msg);
nestedMessage["position"] = position;
nestedMessage["created_at"] = timeSinceEpoch();
enveloppe["message"] = nestedMessage;
jsonWriter->write(enveloppe, &std::cout);
std::cout << std::endl; // add lf and flush
}
else
{
enveloppe = msg;
std::cout << position << " ";
jsonWriter->write(enveloppe, &std::cout);
std::cout << std::endl;
}
}
int ws_cobra_subscribe_main(const ix::CobraConfig& config,
const std::string& channel,
const std::string& filter,
const std::string& position,
bool quiet,
bool fluentd,
int runtime)
{
ix::CobraConnection conn;
conn.configure(config);
conn.connect();
std::atomic<int> msgPerSeconds(0);
std::atomic<int> msgCount(0);
std::atomic<bool> stop(false);
std::atomic<bool> fatalCobraError(false);
auto jsonWriter = makeStreamWriter();
auto timer = [&msgPerSeconds, &msgCount, &stop] {
while (!stop)
{
spdlog::info("#messages {} msg/s {}", msgCount, msgPerSeconds);
msgPerSeconds = 0;
auto duration = std::chrono::seconds(1);
std::this_thread::sleep_for(duration);
}
};
std::thread t(timer);
std::string subscriptionPosition(position);
conn.setEventCallback([&conn,
&channel,
&jsonWriter,
&filter,
&subscriptionPosition,
&msgCount,
&msgPerSeconds,
&quiet,
&fluentd,
&fatalCobraError](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");
spdlog::info("Subscribing to {} at position {}", channel, subscriptionPosition);
conn.subscribe(
channel,
filter,
subscriptionPosition,
[&jsonWriter,
&quiet,
&msgPerSeconds,
&msgCount,
&fluentd,
&subscriptionPosition](const Json::Value& msg, const std::string& position) {
if (!quiet)
{
writeToStdout(fluentd, jsonWriter, msg, position);
}
msgPerSeconds++;
msgCount++;
subscriptionPosition = position;
});
}
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;
}
}
stop = true;
conn.disconnect();
t.join();
return fatalCobraError ? 1 : 0;
}
} // namespace ix