void/ZeroTierOne
2
0
Fork 0
mirror of https://github.com/zerotier/ZeroTierOne.git synced 2025-08-02 23:12:51 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

controller tracing

Browse source
This commit is contained in:
Grant Limberg 2025-07-12 13:29:07 -07:00
parent 0e87c21d87
commit b9de551e0c
7 changed files with 484 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

127
controller/CV1.cpp
View file

@ -21,6 +21,7 @@
#include "CtlUtil.hpp" #include "CtlUtil.hpp"
#include "EmbeddedNetworkController.hpp" #include "EmbeddedNetworkController.hpp"
#include "Redis.hpp" #include "Redis.hpp"
#include "opentelemetry/trace/provider.h"
#include <chrono> #include <chrono>
#include <climits> #include <climits>
@ -61,6 +62,11 @@ CV1::CV1(const Identity& myId, const char* path, int listenPort, RedisConfig* rc
, _redisMemberStatus(false) , _redisMemberStatus(false)
, _smee(NULL) , _smee(NULL)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("cv1");
auto span = tracer->StartSpan("cv1::CV1");
auto scope = tracer->WithActiveSpan(span);
char myAddress[64]; char myAddress[64];
_myAddressStr = myId.address().toString(myAddress); _myAddressStr = myId.address().toString(myAddress);
_connString = std::string(path); _connString = std::string(path);
@ -98,6 +104,9 @@ CV1::CV1(const Identity& myId, const char* path, int listenPort, RedisConfig* rc
_pool->unborrow(c); _pool->unborrow(c);
if (_rc != NULL) { if (_rc != NULL) {
auto innerspan = tracer->StartSpan("cv1::CV1::configureRedis");
auto innerscope = tracer->WithActiveSpan(innerspan);
sw::redis::ConnectionOptions opts; sw::redis::ConnectionOptions opts;
sw::redis::ConnectionPoolOptions poolOpts; sw::redis::ConnectionPoolOptions poolOpts;
opts.host = _rc->hostname; opts.host = _rc->hostname;
@ -111,10 +120,12 @@ CV1::CV1(const Identity& myId, const char* path, int listenPort, RedisConfig* rc
poolOpts.connection_lifetime = std::chrono::minutes(3); poolOpts.connection_lifetime = std::chrono::minutes(3);
poolOpts.connection_idle_time = std::chrono::minutes(1); poolOpts.connection_idle_time = std::chrono::minutes(1);
if (_rc->clusterMode) { if (_rc->clusterMode) {
innerspan->SetAttribute("cluster_mode", "true");
fprintf(stderr, "Using Redis in Cluster Mode\n"); fprintf(stderr, "Using Redis in Cluster Mode\n");
_cluster = std::make_shared<sw::redis::RedisCluster>(opts, poolOpts); _cluster = std::make_shared<sw::redis::RedisCluster>(opts, poolOpts);
} }
else { else {
innerspan->SetAttribute("cluster_mode", "false");
fprintf(stderr, "Using Redis in Standalone Mode\n"); fprintf(stderr, "Using Redis in Standalone Mode\n");
_redis = std::make_shared<sw::redis::Redis>(opts, poolOpts); _redis = std::make_shared<sw::redis::Redis>(opts, poolOpts);
} }
@ -161,6 +172,11 @@ CV1::~CV1()
void CV1::configureSmee() void CV1::configureSmee()
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("cv1");
auto span = tracer->StartSpan("cv1::configureSmee");
auto scope = tracer->WithActiveSpan(span);
const char* TEMPORAL_SCHEME = "ZT_TEMPORAL_SCHEME"; const char* TEMPORAL_SCHEME = "ZT_TEMPORAL_SCHEME";
const char* TEMPORAL_HOST = "ZT_TEMPORAL_HOST"; const char* TEMPORAL_HOST = "ZT_TEMPORAL_HOST";
const char* TEMPORAL_PORT = "ZT_TEMPORAL_PORT"; const char* TEMPORAL_PORT = "ZT_TEMPORAL_PORT";
@ -202,6 +218,11 @@ bool CV1::isReady()
bool CV1::save(nlohmann::json& record, bool notifyListeners) bool CV1::save(nlohmann::json& record, bool notifyListeners)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("cv1");
auto span = tracer->StartSpan("cv1::save");
auto scope = tracer->WithActiveSpan(span);
bool modified = false; bool modified = false;
try { try {
if (! record.is_object()) { if (! record.is_object()) {
@ -257,6 +278,13 @@ bool CV1::save(nlohmann::json& record, bool notifyListeners)
void CV1::eraseNetwork(const uint64_t networkId) void CV1::eraseNetwork(const uint64_t networkId)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("cv1");
auto span = tracer->StartSpan("cv1::eraseNetwork");
auto scope = tracer->WithActiveSpan(span);
char networkIdStr[17];
span->SetAttribute("network_id", Utils::hex(networkId, networkIdStr));
fprintf(stderr, "PostgreSQL::eraseNetwork\n"); fprintf(stderr, "PostgreSQL::eraseNetwork\n");
char tmp2[24]; char tmp2[24];
waitForReady(); waitForReady();
@ -272,6 +300,15 @@ void CV1::eraseNetwork(const uint64_t networkId)
void CV1::eraseMember(const uint64_t networkId, const uint64_t memberId) void CV1::eraseMember(const uint64_t networkId, const uint64_t memberId)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("cv1");
auto span = tracer->StartSpan("cv1::eraseMember");
auto scope = tracer->WithActiveSpan(span);
char networkIdStr[17];
char memberIdStr[11];
span->SetAttribute("network_id", Utils::hex(networkId, networkIdStr));
span->SetAttribute("member_id", Utils::hex10(memberId, memberIdStr));
fprintf(stderr, "PostgreSQL::eraseMember\n"); fprintf(stderr, "PostgreSQL::eraseMember\n");
char tmp2[24]; char tmp2[24];
waitForReady(); waitForReady();
@ -289,6 +326,18 @@ void CV1::eraseMember(const uint64_t networkId, const uint64_t memberId)
void CV1::nodeIsOnline(const uint64_t networkId, const uint64_t memberId, const InetAddress& physicalAddress, const char* osArch) void CV1::nodeIsOnline(const uint64_t networkId, const uint64_t memberId, const InetAddress& physicalAddress, const char* osArch)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("cv1");
auto span = tracer->StartSpan("cv1::nodeIsOnline");
auto scope = tracer->WithActiveSpan(span);
char networkIdStr[17];
char memberIdStr[11];
char ipStr[INET6_ADDRSTRLEN];
span->SetAttribute("network_id", Utils::hex(networkId, networkIdStr));
span->SetAttribute("member_id", Utils::hex10(memberId, memberIdStr));
span->SetAttribute("physical_address", physicalAddress.toString(ipStr));
span->SetAttribute("os_arch", osArch);
std::lock_guard<std::mutex> l(_lastOnline_l); std::lock_guard<std::mutex> l(_lastOnline_l);
NodeOnlineRecord& i = _lastOnline[std::pair<uint64_t, uint64_t>(networkId, memberId)]; NodeOnlineRecord& i = _lastOnline[std::pair<uint64_t, uint64_t>(networkId, memberId)];
i.lastSeen = OSUtils::now(); i.lastSeen = OSUtils::now();
@ -305,6 +354,11 @@ void CV1::nodeIsOnline(const uint64_t networkId, const uint64_t memberId, const
AuthInfo CV1::getSSOAuthInfo(const nlohmann::json& member, const std::string& redirectURL) AuthInfo CV1::getSSOAuthInfo(const nlohmann::json& member, const std::string& redirectURL)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("cv1");
auto span = tracer->StartSpan("cv1::getSSOAuthInfo");
auto scope = tracer->WithActiveSpan(span);
Metrics::db_get_sso_info++; Metrics::db_get_sso_info++;
// NONCE is just a random character string. no semantic meaning // NONCE is just a random character string. no semantic meaning
// state = HMAC SHA384 of Nonce based on shared sso key // state = HMAC SHA384 of Nonce based on shared sso key
@ -484,6 +538,11 @@ AuthInfo CV1::getSSOAuthInfo(const nlohmann::json& member, const std::string& re
void CV1::initializeNetworks() void CV1::initializeNetworks()
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("cv1");
auto span = tracer->StartSpan("cv1::initializeNetworks");
auto scope = tracer->WithActiveSpan(span);
try { try {
std::string setKey = "networks:{" + _myAddressStr + "}"; std::string setKey = "networks:{" + _myAddressStr + "}";
@ -769,6 +828,11 @@ void CV1::initializeNetworks()
void CV1::initializeMembers() void CV1::initializeMembers()
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("cv1");
auto span = tracer->StartSpan("cv1::initializeMembers");
auto scope = tracer->WithActiveSpan(span);
std::string memberId; std::string memberId;
std::string networkId; std::string networkId;
try { try {
@ -1054,6 +1118,11 @@ void CV1::heartbeat()
const char* hostname = hostnameTmp; const char* hostname = hostnameTmp;
while (_run == 1) { while (_run == 1) {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("cv1");
auto span = tracer->StartSpan("cv1::heartbeat");
auto scope = tracer->WithActiveSpan(span);
// fprintf(stderr, "%s: heartbeat\n", controllerId); // fprintf(stderr, "%s: heartbeat\n", controllerId);
auto c = _pool->borrow(); auto c = _pool->borrow();
int64_t ts = OSUtils::now(); int64_t ts = OSUtils::now();
@ -1136,6 +1205,11 @@ void CV1::_membersWatcher_Postgres()
MemberNotificationReceiver m(this, *c->c, stream); MemberNotificationReceiver m(this, *c->c, stream);
while (_run == 1) { while (_run == 1) {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("cv1");
auto span = tracer->StartSpan("cv1::_membersWatcher_Postgres");
auto scope = tracer->WithActiveSpan(span);
c->c->await_notification(5, 0); c->c->await_notification(5, 0);
} }
@ -1149,6 +1223,11 @@ void CV1::_membersWatcher_Redis()
std::string lastID = "0"; std::string lastID = "0";
fprintf(stderr, "Listening to member stream: %s\n", key.c_str()); fprintf(stderr, "Listening to member stream: %s\n", key.c_str());
while (_run == 1) { while (_run == 1) {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("cv1");
auto span = tracer->StartSpan("cv1::_membersWatcher_Redis");
auto scope = tracer->WithActiveSpan(span);
try { try {
json tmp; json tmp;
std::unordered_map<std::string, ItemStream> result; std::unordered_map<std::string, ItemStream> result;
@ -1237,6 +1316,11 @@ void CV1::_networksWatcher_Postgres()
NetworkNotificationReceiver n(this, *c->c, stream); NetworkNotificationReceiver n(this, *c->c, stream);
while (_run == 1) { while (_run == 1) {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("cv1");
auto span = tracer->StartSpan("cv1::_networksWatcher_Postgres");
auto scope = tracer->WithActiveSpan(span);
c->c->await_notification(5, 0); c->c->await_notification(5, 0);
} }
} }
@ -1247,6 +1331,11 @@ void CV1::_networksWatcher_Redis()
std::string key = "network-stream:{" + std::string(_myAddress.toString(buf)) + "}"; std::string key = "network-stream:{" + std::string(_myAddress.toString(buf)) + "}";
std::string lastID = "0"; std::string lastID = "0";
while (_run == 1) { while (_run == 1) {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("cv1");
auto span = tracer->StartSpan("cv1::_networksWatcher_Redis");
auto scope = tracer->WithActiveSpan(span);
try { try {
json tmp; json tmp;
std::unordered_map<std::string, ItemStream> result; std::unordered_map<std::string, ItemStream> result;
@ -1314,6 +1403,11 @@ void CV1::commitThread()
fprintf(stderr, "%s: commitThread start\n", _myAddressStr.c_str()); fprintf(stderr, "%s: commitThread start\n", _myAddressStr.c_str());
std::pair<nlohmann::json, bool> qitem; std::pair<nlohmann::json, bool> qitem;
while (_commitQueue.get(qitem) & (_run == 1)) { while (_commitQueue.get(qitem) & (_run == 1)) {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("cv1");
auto span = tracer->StartSpan("cv1::commitThread");
auto scope = tracer->WithActiveSpan(span);
// fprintf(stderr, "commitThread tick\n"); // fprintf(stderr, "commitThread tick\n");
if (! qitem.first.is_object()) { if (! qitem.first.is_object()) {
fprintf(stderr, "not an object\n"); fprintf(stderr, "not an object\n");
@ -1339,6 +1433,9 @@ void CV1::commitThread()
nlohmann::json& config = (qitem.first); nlohmann::json& config = (qitem.first);
const std::string objtype = config["objtype"]; const std::string objtype = config["objtype"];
if (objtype == "member") { if (objtype == "member") {
auto mspan = tracer->StartSpan("cv1::commitThread::member");
auto mscope = tracer->WithActiveSpan(mspan);
// fprintf(stderr, "%s: commitThread: member\n", _myAddressStr.c_str()); // fprintf(stderr, "%s: commitThread: member\n", _myAddressStr.c_str());
std::string memberId; std::string memberId;
std::string networkId; std::string networkId;
@ -1489,6 +1586,9 @@ void CV1::commitThread()
} }
} }
else if (objtype == "network") { else if (objtype == "network") {
auto nspan = tracer->StartSpan("cv1::commitThread::network");
auto nscope = tracer->WithActiveSpan(nspan);
try { try {
// fprintf(stderr, "%s: commitThread: network\n", _myAddressStr.c_str()); // fprintf(stderr, "%s: commitThread: network\n", _myAddressStr.c_str());
pqxx::work w(*c->c); pqxx::work w(*c->c);
@ -1644,6 +1744,9 @@ void CV1::commitThread()
} }
} }
else if (objtype == "_delete_network") { else if (objtype == "_delete_network") {
auto dspan = tracer->StartSpan("cv1::commitThread::_delete_network");
auto dscope = tracer->WithActiveSpan(dspan);
// fprintf(stderr, "%s: commitThread: delete network\n", _myAddressStr.c_str()); // fprintf(stderr, "%s: commitThread: delete network\n", _myAddressStr.c_str());
try { try {
pqxx::work w(*c->c); pqxx::work w(*c->c);
@ -1677,6 +1780,9 @@ void CV1::commitThread()
} }
} }
else if (objtype == "_delete_member") { else if (objtype == "_delete_member") {
auto mspan = tracer->StartSpan("cv1::commitThread::_delete_member");
auto mscope = tracer->WithActiveSpan(mspan);
// fprintf(stderr, "%s commitThread: delete member\n", _myAddressStr.c_str()); // fprintf(stderr, "%s commitThread: delete member\n", _myAddressStr.c_str());
try { try {
pqxx::work w(*c->c); pqxx::work w(*c->c);
@ -1727,6 +1833,11 @@ void CV1::commitThread()
void CV1::notifyNewMember(const std::string& networkID, const std::string& memberID) void CV1::notifyNewMember(const std::string& networkID, const std::string& memberID)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("cv1");
auto span = tracer->StartSpan("cv1::notifyNewMember");
auto scope = tracer->WithActiveSpan(span);
smeeclient::smee_client_notify_network_joined(_smee, networkID.c_str(), memberID.c_str()); smeeclient::smee_client_notify_network_joined(_smee, networkID.c_str(), memberID.c_str());
} }
@ -1756,6 +1867,11 @@ void CV1::onlineNotification_Postgres()
nlohmann::json jtmp1, jtmp2; nlohmann::json jtmp1, jtmp2;
while (_run == 1) { while (_run == 1) {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("cv1");
auto span = tracer->StartSpan("cv1::onlineNotification_Postgres");
auto scope = tracer->WithActiveSpan(span);
auto c = _pool->borrow(); auto c = _pool->borrow();
auto c2 = _pool->borrow(); auto c2 = _pool->borrow();
try { try {
@ -1856,6 +1972,11 @@ void CV1::onlineNotification_Redis()
std::string controllerId = std::string(_myAddress.toString(buf)); std::string controllerId = std::string(_myAddress.toString(buf));
while (_run == 1) { while (_run == 1) {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("cv1");
auto span = tracer->StartSpan("cv1::onlineNotification_Redis");
auto scope = tracer->WithActiveSpan(span);
fprintf(stderr, "onlineNotification tick\n"); fprintf(stderr, "onlineNotification tick\n");
auto start = std::chrono::high_resolution_clock::now(); auto start = std::chrono::high_resolution_clock::now();
uint64_t count = 0; uint64_t count = 0;
@ -1892,8 +2013,12 @@ void CV1::onlineNotification_Redis()
} }
uint64_t CV1::_doRedisUpdate(sw::redis::Transaction& tx, std::string& controllerId, std::unordered_map<std::pair<uint64_t, uint64_t>, NodeOnlineRecord, _PairHasher>& lastOnline) uint64_t CV1::_doRedisUpdate(sw::redis::Transaction& tx, std::string& controllerId, std::unordered_map<std::pair<uint64_t, uint64_t>, NodeOnlineRecord, _PairHasher>& lastOnline)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("cv1");
auto span = tracer->StartSpan("cv1::_doRedisUpdate");
auto scope = tracer->WithActiveSpan(span);
nlohmann::json jtmp1, jtmp2; nlohmann::json jtmp1, jtmp2;
uint64_t count = 0; uint64_t count = 0;
for (auto i = lastOnline.begin(); i != lastOnline.end(); ++i) { for (auto i = lastOnline.begin(); i != lastOnline.end(); ++i) {

96
controller/CV2.cpp
View file

@ -20,6 +20,7 @@
#include "../version.h" #include "../version.h"
#include "CtlUtil.hpp" #include "CtlUtil.hpp"
#include "EmbeddedNetworkController.hpp" #include "EmbeddedNetworkController.hpp"
#include "opentelemetry/trace/provider.h"
#include <chrono> #include <chrono>
#include <climits> #include <climits>
@ -37,6 +38,11 @@ using namespace ZeroTier;
CV2::CV2(const Identity& myId, const char* path, int listenPort) : DB(), _pool(), _myId(myId), _myAddress(myId.address()), _ready(0), _connected(1), _run(1), _waitNoticePrinted(false), _listenPort(listenPort) CV2::CV2(const Identity& myId, const char* path, int listenPort) : DB(), _pool(), _myId(myId), _myAddress(myId.address()), _ready(0), _connected(1), _run(1), _waitNoticePrinted(false), _listenPort(listenPort)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("cv2");
auto span = tracer->StartSpan("cv2::CV2");
auto scope = tracer->WithActiveSpan(span);
fprintf(stderr, "CV2::CV2\n"); fprintf(stderr, "CV2::CV2\n");
char myAddress[64]; char myAddress[64];
_myAddressStr = myId.address().toString(myAddress); _myAddressStr = myId.address().toString(myAddress);
@ -106,6 +112,11 @@ bool CV2::isReady()
bool CV2::save(nlohmann::json& record, bool notifyListeners) bool CV2::save(nlohmann::json& record, bool notifyListeners)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("cv2");
auto span = tracer->StartSpan("cv2::save");
auto scope = tracer->WithActiveSpan(span);
bool modified = false; bool modified = false;
try { try {
if (! record.is_object()) { if (! record.is_object()) {
@ -114,6 +125,9 @@ bool CV2::save(nlohmann::json& record, bool notifyListeners)
} }
const std::string objtype = record["objtype"]; const std::string objtype = record["objtype"];
if (objtype == "network") { if (objtype == "network") {
auto nspan = tracer->StartSpan("cv2::save::network");
auto nscope = tracer->WithActiveSpan(nspan);
// fprintf(stderr, "network save\n"); // fprintf(stderr, "network save\n");
const uint64_t nwid = OSUtils::jsonIntHex(record["id"], 0ULL); const uint64_t nwid = OSUtils::jsonIntHex(record["id"], 0ULL);
if (nwid) { if (nwid) {
@ -127,6 +141,9 @@ bool CV2::save(nlohmann::json& record, bool notifyListeners)
} }
} }
else if (objtype == "member") { else if (objtype == "member") {
auto mspan = tracer->StartSpan("cv2::save::member");
auto mscope = tracer->WithActiveSpan(mspan);
std::string networkId = record["nwid"]; std::string networkId = record["nwid"];
std::string memberId = record["id"]; std::string memberId = record["id"];
const uint64_t nwid = OSUtils::jsonIntHex(record["nwid"], 0ULL); const uint64_t nwid = OSUtils::jsonIntHex(record["nwid"], 0ULL);
@ -161,6 +178,13 @@ bool CV2::save(nlohmann::json& record, bool notifyListeners)
void CV2::eraseNetwork(const uint64_t networkId) void CV2::eraseNetwork(const uint64_t networkId)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("cv2");
auto span = tracer->StartSpan("cv2::eraseNetwork");
auto scope = tracer->WithActiveSpan(span);
char networkIdStr[17];
span->SetAttribute("network_id", Utils::hex(networkId, networkIdStr));
fprintf(stderr, "PostgreSQL::eraseNetwork\n"); fprintf(stderr, "PostgreSQL::eraseNetwork\n");
char tmp2[24]; char tmp2[24];
waitForReady(); waitForReady();
@ -176,6 +200,15 @@ void CV2::eraseNetwork(const uint64_t networkId)
void CV2::eraseMember(const uint64_t networkId, const uint64_t memberId) void CV2::eraseMember(const uint64_t networkId, const uint64_t memberId)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("cv2");
auto span = tracer->StartSpan("cv2::eraseMember");
auto scope = tracer->WithActiveSpan(span);
char networkIdStr[17];
char memberIdStr[11];
span->SetAttribute("network_id", Utils::hex(networkId, networkIdStr));
span->SetAttribute("member_id", Utils::hex10(memberId, memberIdStr));
fprintf(stderr, "PostgreSQL::eraseMember\n"); fprintf(stderr, "PostgreSQL::eraseMember\n");
char tmp2[24]; char tmp2[24];
waitForReady(); waitForReady();
@ -193,6 +226,18 @@ void CV2::eraseMember(const uint64_t networkId, const uint64_t memberId)
void CV2::nodeIsOnline(const uint64_t networkId, const uint64_t memberId, const InetAddress& physicalAddress, const char* osArch) void CV2::nodeIsOnline(const uint64_t networkId, const uint64_t memberId, const InetAddress& physicalAddress, const char* osArch)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("cv2");
auto span = tracer->StartSpan("cv2::nodeIsOnline");
auto scope = tracer->WithActiveSpan(span);
char networkIdStr[17];
char memberIdStr[11];
char ipAddressStr[INET6_ADDRSTRLEN];
span->SetAttribute("network_id", Utils::hex(networkId, networkIdStr));
span->SetAttribute("member_id", Utils::hex10(memberId, memberIdStr));
span->SetAttribute("physical_address", ipAddressStr);
span->SetAttribute("os_arch", osArch);
std::lock_guard<std::mutex> l(_lastOnline_l); std::lock_guard<std::mutex> l(_lastOnline_l);
NodeOnlineRecord& i = _lastOnline[std::pair<uint64_t, uint64_t>(networkId, memberId)]; NodeOnlineRecord& i = _lastOnline[std::pair<uint64_t, uint64_t>(networkId, memberId)];
i.lastSeen = OSUtils::now(); i.lastSeen = OSUtils::now();
@ -210,6 +255,10 @@ void CV2::nodeIsOnline(const uint64_t networkId, const uint64_t memberId, const
AuthInfo CV2::getSSOAuthInfo(const nlohmann::json& member, const std::string& redirectURL) AuthInfo CV2::getSSOAuthInfo(const nlohmann::json& member, const std::string& redirectURL)
{ {
// TODO: Redo this for CV2 // TODO: Redo this for CV2
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("cv2");
auto span = tracer->StartSpan("cv2::getSSOAuthInfo");
auto scope = tracer->WithActiveSpan(span);
Metrics::db_get_sso_info++; Metrics::db_get_sso_info++;
// NONCE is just a random character string. no semantic meaning // NONCE is just a random character string. no semantic meaning
@ -372,6 +421,11 @@ AuthInfo CV2::getSSOAuthInfo(const nlohmann::json& member, const std::string& re
void CV2::initializeNetworks() void CV2::initializeNetworks()
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("cv2");
auto span = tracer->StartSpan("cv2::initializeNetworks");
auto scope = tracer->WithActiveSpan(span);
fprintf(stderr, "Initializing networks...\n"); fprintf(stderr, "Initializing networks...\n");
try { try {
@ -501,6 +555,11 @@ void CV2::initializeNetworks()
void CV2::initializeMembers() void CV2::initializeMembers()
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("cv2");
auto span = tracer->StartSpan("cv2::initializeMembers");
auto scope = tracer->WithActiveSpan(span);
std::string memberId; std::string memberId;
std::string networkId; std::string networkId;
try { try {
@ -672,6 +731,11 @@ void CV2::heartbeat()
const char* hostname = hostnameTmp; const char* hostname = hostnameTmp;
while (_run == 1) { while (_run == 1) {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("cv2");
auto span = tracer->StartSpan("cv2::heartbeat");
auto scope = tracer->WithActiveSpan(span);
auto c = _pool->borrow(); auto c = _pool->borrow();
int64_t ts = OSUtils::now(); int64_t ts = OSUtils::now();
@ -723,6 +787,11 @@ void CV2::membersDbWatcher()
MemberNotificationReceiver m(this, *c->c, stream); MemberNotificationReceiver m(this, *c->c, stream);
while (_run == 1) { while (_run == 1) {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("cv2");
auto span = tracer->StartSpan("cv2::membersDbWatcher");
auto scope = tracer->WithActiveSpan(span);
c->c->await_notification(5, 0); c->c->await_notification(5, 0);
} }
@ -742,6 +811,11 @@ void CV2::networksDbWatcher()
NetworkNotificationReceiver n(this, *c->c, stream); NetworkNotificationReceiver n(this, *c->c, stream);
while (_run == 1) { while (_run == 1) {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("cv2");
auto span = tracer->StartSpan("cv2::networksDbWatcher");
auto scope = tracer->WithActiveSpan(span);
c->c->await_notification(5, 0); c->c->await_notification(5, 0);
} }
@ -754,6 +828,11 @@ void CV2::commitThread()
fprintf(stderr, "%s: commitThread start\n", _myAddressStr.c_str()); fprintf(stderr, "%s: commitThread start\n", _myAddressStr.c_str());
std::pair<nlohmann::json, bool> qitem; std::pair<nlohmann::json, bool> qitem;
while (_commitQueue.get(qitem) && (_run == 1)) { while (_commitQueue.get(qitem) && (_run == 1)) {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("cv2");
auto span = tracer->StartSpan("cv2::commitThread");
auto scope = tracer->WithActiveSpan(span);
// fprintf(stderr, "commitThread tick\n"); // fprintf(stderr, "commitThread tick\n");
if (! qitem.first.is_object()) { if (! qitem.first.is_object()) {
fprintf(stderr, "not an object\n"); fprintf(stderr, "not an object\n");
@ -779,6 +858,9 @@ void CV2::commitThread()
nlohmann::json& config = (qitem.first); nlohmann::json& config = (qitem.first);
const std::string objtype = config["objtype"]; const std::string objtype = config["objtype"];
if (objtype == "member") { if (objtype == "member") {
auto span = tracer->StartSpan("cv2::commitThread::member");
auto scope = tracer->WithActiveSpan(span);
// fprintf(stderr, "%s: commitThread: member\n", _myAddressStr.c_str()); // fprintf(stderr, "%s: commitThread: member\n", _myAddressStr.c_str());
std::string memberId; std::string memberId;
std::string networkId; std::string networkId;
@ -903,6 +985,9 @@ void CV2::commitThread()
} }
} }
else if (objtype == "network") { else if (objtype == "network") {
auto span = tracer->StartSpan("cv2::commitThread::network");
auto scope = tracer->WithActiveSpan(span);
try { try {
// fprintf(stderr, "%s: commitThread: network\n", _myAddressStr.c_str()); // fprintf(stderr, "%s: commitThread: network\n", _myAddressStr.c_str());
pqxx::work w(*c->c); pqxx::work w(*c->c);
@ -948,6 +1033,9 @@ void CV2::commitThread()
} }
} }
else if (objtype == "_delete_network") { else if (objtype == "_delete_network") {
auto span = tracer->StartSpan("cv2::commitThread::delete_network");
auto scope = tracer->WithActiveSpan(span);
// fprintf(stderr, "%s: commitThread: delete network\n", _myAddressStr.c_str()); // fprintf(stderr, "%s: commitThread: delete network\n", _myAddressStr.c_str());
try { try {
// don't think we need this. Deletion handled by CV2 API // don't think we need this. Deletion handled by CV2 API
@ -965,6 +1053,9 @@ void CV2::commitThread()
} }
} }
else if (objtype == "_delete_member") { else if (objtype == "_delete_member") {
auto span = tracer->StartSpan("cv2::commitThread::delete_member");
auto scope = tracer->WithActiveSpan(span);
// fprintf(stderr, "%s commitThread: delete member\n", _myAddressStr.c_str()); // fprintf(stderr, "%s commitThread: delete member\n", _myAddressStr.c_str());
try { try {
pqxx::work w(*c->c); pqxx::work w(*c->c);
@ -1002,6 +1093,11 @@ void CV2::onlineNotificationThread()
nlohmann::json jtmp1, jtmp2; nlohmann::json jtmp1, jtmp2;
while (_run == 1) { while (_run == 1) {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("cv2");
auto span = tracer->StartSpan("cv2::onlineNotificationThread");
auto scope = tracer->WithActiveSpan(span);
auto c = _pool->borrow(); auto c = _pool->borrow();
auto c2 = _pool->borrow(); auto c2 = _pool->borrow();

15
controller/ConnectionPool.hpp
View file

@ -19,6 +19,7 @@
#endif #endif
#include "../node/Metrics.hpp" #include "../node/Metrics.hpp"
#include "opentelemetry/trace/provider.h"
#include <deque> #include <deque>
#include <exception> #include <exception>
@ -87,6 +88,11 @@ template <class T> class ConnectionPool {
*/ */
std::shared_ptr<T> borrow() std::shared_ptr<T> borrow()
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("connection_pool");
auto span = tracer->StartSpan("connection_pool::borrow");
auto scope = tracer->WithActiveSpan(span);
std::unique_lock<std::mutex> l(m_poolMutex); std::unique_lock<std::mutex> l(m_poolMutex);
while ((m_pool.size() + m_borrowed.size()) < m_minPoolSize) { while ((m_pool.size() + m_borrowed.size()) < m_minPoolSize) {
@ -104,6 +110,7 @@ template <class T> class ConnectionPool {
return std::static_pointer_cast<T>(conn); return std::static_pointer_cast<T>(conn);
} }
catch (std::exception& e) { catch (std::exception& e) {
span->SetStatus(opentelemetry::trace::StatusCode::kError, e.what());
Metrics::pool_errors++; Metrics::pool_errors++;
throw ConnectionUnavailable(); throw ConnectionUnavailable();
} }
@ -121,12 +128,15 @@ template <class T> class ConnectionPool {
return std::static_pointer_cast<T>(conn); return std::static_pointer_cast<T>(conn);
} }
catch (std::exception& e) { catch (std::exception& e) {
span->SetStatus(opentelemetry::trace::StatusCode::kError, e.what());
// Error creating a replacement connection // Error creating a replacement connection
Metrics::pool_errors++; Metrics::pool_errors++;
throw ConnectionUnavailable(); throw ConnectionUnavailable();
} }
} }
} }
span->SetStatus(opentelemetry::trace::StatusCode::kError, "No available connections in pool");
// Nothing available // Nothing available
Metrics::pool_errors++; Metrics::pool_errors++;
throw ConnectionUnavailable(); throw ConnectionUnavailable();
@ -151,6 +161,11 @@ template <class T> class ConnectionPool {
*/ */
void unborrow(std::shared_ptr<T> conn) void unborrow(std::shared_ptr<T> conn)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("connection_pool");
auto span = tracer->StartSpan("connection_pool::unborrow");
auto scope = tracer->WithActiveSpan(span);
// Lock // Lock
std::unique_lock<std::mutex> lock(m_poolMutex); std::unique_lock<std::mutex> lock(m_poolMutex);
m_borrowed.erase(conn); m_borrowed.erase(conn);

79
controller/DB.cpp
View file

@ -15,6 +15,7 @@
#include "../node/Metrics.hpp" #include "../node/Metrics.hpp"
#include "EmbeddedNetworkController.hpp" #include "EmbeddedNetworkController.hpp"
#include "opentelemetry/trace/provider.h"
#include <algorithm> #include <algorithm>
#include <chrono> #include <chrono>
@ -26,6 +27,11 @@ namespace ZeroTier {
void DB::initNetwork(nlohmann::json& network) void DB::initNetwork(nlohmann::json& network)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("db");
auto span = tracer->StartSpan("db::initNetwork");
auto scope = tracer->WithActiveSpan(span);
if (! network.count("private")) if (! network.count("private"))
network["private"] = true; network["private"] = true;
if (! network.count("creationTime")) if (! network.count("creationTime"))
@ -76,6 +82,11 @@ void DB::initNetwork(nlohmann::json& network)
void DB::initMember(nlohmann::json& member) void DB::initMember(nlohmann::json& member)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("db");
auto span = tracer->StartSpan("db::initMember");
auto scope = tracer->WithActiveSpan(span);
if (! member.count("authorized")) if (! member.count("authorized"))
member["authorized"] = false; member["authorized"] = false;
if (! member.count("ssoExempt")) if (! member.count("ssoExempt"))
@ -121,6 +132,11 @@ void DB::initMember(nlohmann::json& member)
void DB::cleanNetwork(nlohmann::json& network) void DB::cleanNetwork(nlohmann::json& network)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("db");
auto span = tracer->StartSpan("db::cleanNetwork");
auto scope = tracer->WithActiveSpan(span);
network.erase("clock"); network.erase("clock");
network.erase("authorizedMemberCount"); network.erase("authorizedMemberCount");
network.erase("activeMemberCount"); network.erase("activeMemberCount");
@ -130,6 +146,11 @@ void DB::cleanNetwork(nlohmann::json& network)
void DB::cleanMember(nlohmann::json& member) void DB::cleanMember(nlohmann::json& member)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("db");
auto span = tracer->StartSpan("db::cleanMember");
auto scope = tracer->WithActiveSpan(span);
member.erase("clock"); member.erase("clock");
member.erase("physicalAddr"); member.erase("physicalAddr");
member.erase("recentLog"); member.erase("recentLog");
@ -148,6 +169,13 @@ DB::~DB()
bool DB::get(const uint64_t networkId, nlohmann::json& network) bool DB::get(const uint64_t networkId, nlohmann::json& network)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("db");
auto span = tracer->StartSpan("db::getNetwork");
auto scope = tracer->WithActiveSpan(span);
char networkIdStr[17];
span->SetAttribute("network_id", Utils::hex(networkId, networkIdStr));
waitForReady(); waitForReady();
Metrics::db_get_network++; Metrics::db_get_network++;
std::shared_ptr<_Network> nw; std::shared_ptr<_Network> nw;
@ -167,6 +195,15 @@ bool DB::get(const uint64_t networkId, nlohmann::json& network)
bool DB::get(const uint64_t networkId, nlohmann::json& network, const uint64_t memberId, nlohmann::json& member) bool DB::get(const uint64_t networkId, nlohmann::json& network, const uint64_t memberId, nlohmann::json& member)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("db");
auto span = tracer->StartSpan("db::getNetworkAndMember");
auto scope = tracer->WithActiveSpan(span);
char networkIdStr[17];
char memberIdStr[11];
span->SetAttribute("network_id", Utils::hex(networkId, networkIdStr));
span->SetAttribute("member_id", Utils::hex(networkId, memberIdStr));
waitForReady(); waitForReady();
Metrics::db_get_network_and_member++; Metrics::db_get_network_and_member++;
std::shared_ptr<_Network> nw; std::shared_ptr<_Network> nw;
@ -190,6 +227,15 @@ bool DB::get(const uint64_t networkId, nlohmann::json& network, const uint64_t m
bool DB::get(const uint64_t networkId, nlohmann::json& network, const uint64_t memberId, nlohmann::json& member, NetworkSummaryInfo& info) bool DB::get(const uint64_t networkId, nlohmann::json& network, const uint64_t memberId, nlohmann::json& member, NetworkSummaryInfo& info)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("db");
auto span = tracer->StartSpan("db::getNetworkAndMemberAndSummary");
auto scope = tracer->WithActiveSpan(span);
char networkIdStr[17];
char memberIdStr[11];
span->SetAttribute("network_id", Utils::hex(networkId, networkIdStr));
span->SetAttribute("member_id", Utils::hex(memberId, memberIdStr));
waitForReady(); waitForReady();
Metrics::db_get_network_and_member_and_summary++; Metrics::db_get_network_and_member_and_summary++;
std::shared_ptr<_Network> nw; std::shared_ptr<_Network> nw;
@ -209,11 +255,24 @@ bool DB::get(const uint64_t networkId, nlohmann::json& network, const uint64_t m
return false; return false;
member = m->second; member = m->second;
} }
span->SetAttribute("authorized_member_count", info.authorizedMemberCount);
span->SetAttribute("allocated_ips", info.allocatedIps.size());
span->SetAttribute("total_member_count", info.totalMemberCount);
span->SetAttribute("most_recent_deauth_time", info.mostRecentDeauthTime);
span->SetAttribute("active_bridges", info.activeBridges.size());
return true; return true;
} }
bool DB::get(const uint64_t networkId, nlohmann::json& network, std::vector<nlohmann::json>& members) bool DB::get(const uint64_t networkId, nlohmann::json& network, std::vector<nlohmann::json>& members)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("db");
auto span = tracer->StartSpan("db::getNetworkAndMembers");
auto scope = tracer->WithActiveSpan(span);
char networkIdStr[17];
span->SetAttribute("network_id", Utils::hex(networkId, networkIdStr));
waitForReady(); waitForReady();
Metrics::db_get_member_list++; Metrics::db_get_member_list++;
std::shared_ptr<_Network> nw; std::shared_ptr<_Network> nw;
@ -236,6 +295,11 @@ bool DB::get(const uint64_t networkId, nlohmann::json& network, std::vector<nloh
void DB::networks(std::set<uint64_t>& networks) void DB::networks(std::set<uint64_t>& networks)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("db");
auto span = tracer->StartSpan("db::networks");
auto scope = tracer->WithActiveSpan(span);
waitForReady(); waitForReady();
Metrics::db_get_network_list++; Metrics::db_get_network_list++;
std::shared_lock<std::shared_mutex> l(_networks_l); std::shared_lock<std::shared_mutex> l(_networks_l);
@ -245,6 +309,11 @@ void DB::networks(std::set<uint64_t>& networks)
void DB::_memberChanged(nlohmann::json& old, nlohmann::json& memberConfig, bool notifyListeners) void DB::_memberChanged(nlohmann::json& old, nlohmann::json& memberConfig, bool notifyListeners)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("db");
auto span = tracer->StartSpan("db::_memberChanged");
auto scope = tracer->WithActiveSpan(span);
Metrics::db_member_change++; Metrics::db_member_change++;
uint64_t memberId = 0; uint64_t memberId = 0;
uint64_t networkId = 0; uint64_t networkId = 0;
@ -389,6 +458,11 @@ void DB::_memberChanged(nlohmann::json& old, nlohmann::json& memberConfig, bool
void DB::_networkChanged(nlohmann::json& old, nlohmann::json& networkConfig, bool notifyListeners) void DB::_networkChanged(nlohmann::json& old, nlohmann::json& networkConfig, bool notifyListeners)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("db");
auto span = tracer->StartSpan("db::_networkChanged");
auto scope = tracer->WithActiveSpan(span);
Metrics::db_network_change++; Metrics::db_network_change++;
if (notifyListeners) { if (notifyListeners) {
if (old.is_object() && old.contains("id") && networkConfig.is_object() && networkConfig.contains("id")) { if (old.is_object() && old.contains("id") && networkConfig.is_object() && networkConfig.contains("id")) {
@ -457,6 +531,11 @@ void DB::_networkChanged(nlohmann::json& old, nlohmann::json& networkConfig, boo
void DB::_fillSummaryInfo(const std::shared_ptr<_Network>& nw, NetworkSummaryInfo& info) void DB::_fillSummaryInfo(const std::shared_ptr<_Network>& nw, NetworkSummaryInfo& info)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("db");
auto span = tracer->StartSpan("db::_fillSummaryInfo");
auto scope = tracer->WithActiveSpan(span);
for (auto ab = nw->activeBridgeMembers.begin(); ab != nw->activeBridgeMembers.end(); ++ab) for (auto ab = nw->activeBridgeMembers.begin(); ab != nw->activeBridgeMembers.end(); ++ab)
info.activeBridges.push_back(Address(*ab)); info.activeBridges.push_back(Address(*ab));
std::sort(info.activeBridges.begin(), info.activeBridges.end()); std::sort(info.activeBridges.begin(), info.activeBridges.end());

122
controller/DBMirrorSet.cpp
View file

@ -13,12 +13,19 @@
#include "DBMirrorSet.hpp" #include "DBMirrorSet.hpp"
#include "opentelemetry/trace/provider.h"
namespace ZeroTier { namespace ZeroTier {
DBMirrorSet::DBMirrorSet(DB::ChangeListener* listener) : _listener(listener), _running(true), _syncCheckerThread(), _dbs(), _dbs_l() DBMirrorSet::DBMirrorSet(DB::ChangeListener* listener) : _listener(listener), _running(true), _syncCheckerThread(), _dbs(), _dbs_l()
{ {
_syncCheckerThread = std::thread([this]() { _syncCheckerThread = std::thread([this]() {
for (;;) { for (;;) {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("db_mirror_set");
auto span = tracer->StartSpan("db::syncChecker");
auto scope = tracer->WithActiveSpan(span);
for (int i = 0; i < 120; ++i) { // 1 minute delay between checks for (int i = 0; i < 120; ++i) { // 1 minute delay between checks
if (! _running) if (! _running)
return; return;
@ -77,6 +84,11 @@ DBMirrorSet::~DBMirrorSet()
bool DBMirrorSet::hasNetwork(const uint64_t networkId) const bool DBMirrorSet::hasNetwork(const uint64_t networkId) const
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("db_mirror_set");
auto span = tracer->StartSpan("db_mirror_set::hasNetwork");
auto scope = tracer->WithActiveSpan(span);
std::shared_lock<std::shared_mutex> l(_dbs_l); std::shared_lock<std::shared_mutex> l(_dbs_l);
for (auto d = _dbs.begin(); d != _dbs.end(); ++d) { for (auto d = _dbs.begin(); d != _dbs.end(); ++d) {
if ((*d)->hasNetwork(networkId)) if ((*d)->hasNetwork(networkId))
@ -87,6 +99,13 @@ bool DBMirrorSet::hasNetwork(const uint64_t networkId) const
bool DBMirrorSet::get(const uint64_t networkId, nlohmann::json& network) bool DBMirrorSet::get(const uint64_t networkId, nlohmann::json& network)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("db_mirror_set");
auto span = tracer->StartSpan("db_mirror_set::getNetwork");
auto scope = tracer->WithActiveSpan(span);
char networkIdStr[17];
span->SetAttribute("network_id", Utils::hex(networkId, networkIdStr));
std::shared_lock<std::shared_mutex> l(_dbs_l); std::shared_lock<std::shared_mutex> l(_dbs_l);
for (auto d = _dbs.begin(); d != _dbs.end(); ++d) { for (auto d = _dbs.begin(); d != _dbs.end(); ++d) {
if ((*d)->get(networkId, network)) { if ((*d)->get(networkId, network)) {
@ -98,6 +117,15 @@ bool DBMirrorSet::get(const uint64_t networkId, nlohmann::json& network)
bool DBMirrorSet::get(const uint64_t networkId, nlohmann::json& network, const uint64_t memberId, nlohmann::json& member) bool DBMirrorSet::get(const uint64_t networkId, nlohmann::json& network, const uint64_t memberId, nlohmann::json& member)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("db_mirror_set");
auto span = tracer->StartSpan("db_mirror_set::getNetworkAndMember");
auto scope = tracer->WithActiveSpan(span);
char networkIdStr[17];
char memberIdStr[11];
span->SetAttribute("network_id", Utils::hex(networkId, networkIdStr));
span->SetAttribute("member_id", Utils::hex10(memberId, memberIdStr));
std::shared_lock<std::shared_mutex> l(_dbs_l); std::shared_lock<std::shared_mutex> l(_dbs_l);
for (auto d = _dbs.begin(); d != _dbs.end(); ++d) { for (auto d = _dbs.begin(); d != _dbs.end(); ++d) {
if ((*d)->get(networkId, network, memberId, member)) if ((*d)->get(networkId, network, memberId, member))
@ -108,6 +136,15 @@ bool DBMirrorSet::get(const uint64_t networkId, nlohmann::json& network, const u
bool DBMirrorSet::get(const uint64_t networkId, nlohmann::json& network, const uint64_t memberId, nlohmann::json& member, DB::NetworkSummaryInfo& info) bool DBMirrorSet::get(const uint64_t networkId, nlohmann::json& network, const uint64_t memberId, nlohmann::json& member, DB::NetworkSummaryInfo& info)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("db_mirror_set");
auto span = tracer->StartSpan("db_mirror_set::getNetworkAndMemberWithSummary");
auto scope = tracer->WithActiveSpan(span);
char networkIdStr[17];
char memberIdStr[11];
span->SetAttribute("network_id", Utils::hex(networkId, networkIdStr));
span->SetAttribute("member_id", Utils::hex10(memberId, memberIdStr));
std::shared_lock<std::shared_mutex> l(_dbs_l); std::shared_lock<std::shared_mutex> l(_dbs_l);
for (auto d = _dbs.begin(); d != _dbs.end(); ++d) { for (auto d = _dbs.begin(); d != _dbs.end(); ++d) {
if ((*d)->get(networkId, network, memberId, member, info)) if ((*d)->get(networkId, network, memberId, member, info))
@ -118,6 +155,13 @@ bool DBMirrorSet::get(const uint64_t networkId, nlohmann::json& network, const u
bool DBMirrorSet::get(const uint64_t networkId, nlohmann::json& network, std::vector<nlohmann::json>& members) bool DBMirrorSet::get(const uint64_t networkId, nlohmann::json& network, std::vector<nlohmann::json>& members)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("db_mirror_set");
auto span = tracer->StartSpan("db_mirror_set::getNetworkAndMembers");
auto scope = tracer->WithActiveSpan(span);
char networkIdStr[17];
span->SetAttribute("network_id", Utils::hex(networkId, networkIdStr));
std::shared_lock<std::shared_mutex> l(_dbs_l); std::shared_lock<std::shared_mutex> l(_dbs_l);
for (auto d = _dbs.begin(); d != _dbs.end(); ++d) { for (auto d = _dbs.begin(); d != _dbs.end(); ++d) {
if ((*d)->get(networkId, network, members)) if ((*d)->get(networkId, network, members))
@ -128,6 +172,11 @@ bool DBMirrorSet::get(const uint64_t networkId, nlohmann::json& network, std::ve
AuthInfo DBMirrorSet::getSSOAuthInfo(const nlohmann::json& member, const std::string& redirectURL) AuthInfo DBMirrorSet::getSSOAuthInfo(const nlohmann::json& member, const std::string& redirectURL)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("db_mirror_set");
auto span = tracer->StartSpan("db_mirror_set::getSSOAuthInfo");
auto scope = tracer->WithActiveSpan(span);
std::shared_lock<std::shared_mutex> l(_dbs_l); std::shared_lock<std::shared_mutex> l(_dbs_l);
for (auto d = _dbs.begin(); d != _dbs.end(); ++d) { for (auto d = _dbs.begin(); d != _dbs.end(); ++d) {
AuthInfo info = (*d)->getSSOAuthInfo(member, redirectURL); AuthInfo info = (*d)->getSSOAuthInfo(member, redirectURL);
@ -140,6 +189,11 @@ AuthInfo DBMirrorSet::getSSOAuthInfo(const nlohmann::json& member, const std::st
void DBMirrorSet::networks(std::set<uint64_t>& networks) void DBMirrorSet::networks(std::set<uint64_t>& networks)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("db_mirror_set");
auto span = tracer->StartSpan("db_mirror_set::networks");
auto scope = tracer->WithActiveSpan(span);
std::shared_lock<std::shared_mutex> l(_dbs_l); std::shared_lock<std::shared_mutex> l(_dbs_l);
for (auto d = _dbs.begin(); d != _dbs.end(); ++d) { for (auto d = _dbs.begin(); d != _dbs.end(); ++d) {
(*d)->networks(networks); (*d)->networks(networks);
@ -148,6 +202,11 @@ void DBMirrorSet::networks(std::set<uint64_t>& networks)
bool DBMirrorSet::waitForReady() bool DBMirrorSet::waitForReady()
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("db_mirror_set");
auto span = tracer->StartSpan("db_mirror_set::waitForReady");
auto scope = tracer->WithActiveSpan(span);
bool r = false; bool r = false;
std::shared_lock<std::shared_mutex> l(_dbs_l); std::shared_lock<std::shared_mutex> l(_dbs_l);
for (auto d = _dbs.begin(); d != _dbs.end(); ++d) { for (auto d = _dbs.begin(); d != _dbs.end(); ++d) {
@ -158,6 +217,11 @@ bool DBMirrorSet::waitForReady()
bool DBMirrorSet::isReady() bool DBMirrorSet::isReady()
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("db_mirror_set");
auto span = tracer->StartSpan("db_mirror_set::isReady");
auto scope = tracer->WithActiveSpan(span);
std::shared_lock<std::shared_mutex> l(_dbs_l); std::shared_lock<std::shared_mutex> l(_dbs_l);
for (auto d = _dbs.begin(); d != _dbs.end(); ++d) { for (auto d = _dbs.begin(); d != _dbs.end(); ++d) {
if (! (*d)->isReady()) if (! (*d)->isReady())
@ -168,6 +232,11 @@ bool DBMirrorSet::isReady()
bool DBMirrorSet::save(nlohmann::json& record, bool notifyListeners) bool DBMirrorSet::save(nlohmann::json& record, bool notifyListeners)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("db_mirror_set");
auto span = tracer->StartSpan("db_mirror_set::save");
auto scope = tracer->WithActiveSpan(span);
std::vector<std::shared_ptr<DB> > dbs; std::vector<std::shared_ptr<DB> > dbs;
{ {
std::unique_lock<std::shared_mutex> l(_dbs_l); std::unique_lock<std::shared_mutex> l(_dbs_l);
@ -191,6 +260,13 @@ bool DBMirrorSet::save(nlohmann::json& record, bool notifyListeners)
void DBMirrorSet::eraseNetwork(const uint64_t networkId) void DBMirrorSet::eraseNetwork(const uint64_t networkId)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("db_mirror_set");
auto span = tracer->StartSpan("db_mirror_set::eraseNetwork");
auto scope = tracer->WithActiveSpan(span);
char networkIdStr[17];
span->SetAttribute("network_id", Utils::hex(networkId, networkIdStr));
std::unique_lock<std::shared_mutex> l(_dbs_l); std::unique_lock<std::shared_mutex> l(_dbs_l);
for (auto d = _dbs.begin(); d != _dbs.end(); ++d) { for (auto d = _dbs.begin(); d != _dbs.end(); ++d) {
(*d)->eraseNetwork(networkId); (*d)->eraseNetwork(networkId);
@ -199,6 +275,15 @@ void DBMirrorSet::eraseNetwork(const uint64_t networkId)
void DBMirrorSet::eraseMember(const uint64_t networkId, const uint64_t memberId) void DBMirrorSet::eraseMember(const uint64_t networkId, const uint64_t memberId)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("db_mirror_set");
auto span = tracer->StartSpan("db_mirror_set::eraseMember");
auto scope = tracer->WithActiveSpan(span);
char networkIdStr[17];
char memberIdStr[11];
span->SetAttribute("network_id", Utils::hex(networkId, networkIdStr));
span->SetAttribute("member_id", Utils::hex10(memberId, memberIdStr));
std::unique_lock<std::shared_mutex> l(_dbs_l); std::unique_lock<std::shared_mutex> l(_dbs_l);
for (auto d = _dbs.begin(); d != _dbs.end(); ++d) { for (auto d = _dbs.begin(); d != _dbs.end(); ++d) {
(*d)->eraseMember(networkId, memberId); (*d)->eraseMember(networkId, memberId);
@ -207,6 +292,18 @@ void DBMirrorSet::eraseMember(const uint64_t networkId, const uint64_t memberId)
void DBMirrorSet::nodeIsOnline(const uint64_t networkId, const uint64_t memberId, const InetAddress& physicalAddress, const char* osArch) void DBMirrorSet::nodeIsOnline(const uint64_t networkId, const uint64_t memberId, const InetAddress& physicalAddress, const char* osArch)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("db_mirror_set");
auto span = tracer->StartSpan("db_mirror_set::nodeIsOnline");
auto scope = tracer->WithActiveSpan(span);
char networkIdStr[17];
char memberIdStr[11];
char phyAddressStr[INET6_ADDRSTRLEN];
span->SetAttribute("network_id", Utils::hex(networkId, networkIdStr));
span->SetAttribute("member_id", Utils::hex10(memberId, memberIdStr));
span->SetAttribute("physical_address", physicalAddress.toString(phyAddressStr));
span->SetAttribute("os_arch", osArch);
std::shared_lock<std::shared_mutex> l(_dbs_l); std::shared_lock<std::shared_mutex> l(_dbs_l);
for (auto d = _dbs.begin(); d != _dbs.end(); ++d) { for (auto d = _dbs.begin(); d != _dbs.end(); ++d) {
(*d)->nodeIsOnline(networkId, memberId, physicalAddress, osArch); (*d)->nodeIsOnline(networkId, memberId, physicalAddress, osArch);
@ -220,6 +317,13 @@ void DBMirrorSet::nodeIsOnline(const uint64_t networkId, const uint64_t memberId
void DBMirrorSet::onNetworkUpdate(const void* db, uint64_t networkId, const nlohmann::json& network) void DBMirrorSet::onNetworkUpdate(const void* db, uint64_t networkId, const nlohmann::json& network)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("db_mirror_set");
auto span = tracer->StartSpan("db_mirror_set::onNetworkUpdate");
auto scope = tracer->WithActiveSpan(span);
char networkIdStr[17];
span->SetAttribute("network_id", Utils::hex(networkId, networkIdStr));
nlohmann::json record(network); nlohmann::json record(network);
std::unique_lock<std::shared_mutex> l(_dbs_l); std::unique_lock<std::shared_mutex> l(_dbs_l);
for (auto d = _dbs.begin(); d != _dbs.end(); ++d) { for (auto d = _dbs.begin(); d != _dbs.end(); ++d) {
@ -232,6 +336,15 @@ void DBMirrorSet::onNetworkUpdate(const void* db, uint64_t networkId, const nloh
void DBMirrorSet::onNetworkMemberUpdate(const void* db, uint64_t networkId, uint64_t memberId, const nlohmann::json& member) void DBMirrorSet::onNetworkMemberUpdate(const void* db, uint64_t networkId, uint64_t memberId, const nlohmann::json& member)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("db_mirror_set");
auto span = tracer->StartSpan("db_mirror_set::onNetworkMemberUpdate");
auto scope = tracer->WithActiveSpan(span);
char networkIdStr[17];
char memberIdStr[11];
span->SetAttribute("network_id", Utils::hex(networkId, networkIdStr));
span->SetAttribute("member_id", Utils::hex10(memberId, memberIdStr));
nlohmann::json record(member); nlohmann::json record(member);
std::unique_lock<std::shared_mutex> l(_dbs_l); std::unique_lock<std::shared_mutex> l(_dbs_l);
for (auto d = _dbs.begin(); d != _dbs.end(); ++d) { for (auto d = _dbs.begin(); d != _dbs.end(); ++d) {
@ -244,6 +357,15 @@ void DBMirrorSet::onNetworkMemberUpdate(const void* db, uint64_t networkId, uint
void DBMirrorSet::onNetworkMemberDeauthorize(const void* db, uint64_t networkId, uint64_t memberId) void DBMirrorSet::onNetworkMemberDeauthorize(const void* db, uint64_t networkId, uint64_t memberId)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("db_mirror_set");
auto span = tracer->StartSpan("db_mirror_set::onNetworkMemberDeauthorize");
auto scope = tracer->WithActiveSpan(span);
char networkIdStr[17];
char memberIdStr[11];
span->SetAttribute("network_id", Utils::hex(networkId, networkIdStr));
span->SetAttribute("member_id", Utils::hex10(memberId, memberIdStr));
_listener->onNetworkMemberDeauthorize(this, networkId, memberId); _listener->onNetworkMemberDeauthorize(this, networkId, memberId);
} }

32
controller/FileDB.cpp
View file

@ -14,11 +14,17 @@
#include "FileDB.hpp" #include "FileDB.hpp"
#include "../node/Metrics.hpp" #include "../node/Metrics.hpp"
#include "opentelemetry/trace/provider.h"
namespace ZeroTier { namespace ZeroTier {
FileDB::FileDB(const char* path) : DB(), _path(path), _networksPath(_path + ZT_PATH_SEPARATOR_S + "network"), _tracePath(_path + ZT_PATH_SEPARATOR_S + "trace"), _running(true) FileDB::FileDB(const char* path) : DB(), _path(path), _networksPath(_path + ZT_PATH_SEPARATOR_S + "network"), _tracePath(_path + ZT_PATH_SEPARATOR_S + "trace"), _running(true)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("filedb");
auto span = tracer->StartSpan("filedb::FileDB");
auto scope = tracer->WithActiveSpan(span);
OSUtils::mkdir(_path.c_str()); OSUtils::mkdir(_path.c_str());
OSUtils::lockDownFile(_path.c_str(), true); OSUtils::lockDownFile(_path.c_str(), true);
OSUtils::mkdir(_networksPath.c_str()); OSUtils::mkdir(_networksPath.c_str());
@ -85,11 +91,19 @@ bool FileDB::isReady()
bool FileDB::save(nlohmann::json& record, bool notifyListeners) bool FileDB::save(nlohmann::json& record, bool notifyListeners)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("filedb");
auto span = tracer->StartSpan("filedb::save");
auto scope = tracer->WithActiveSpan(span);
char p1[4096], p2[4096], pb[4096]; char p1[4096], p2[4096], pb[4096];
bool modified = false; bool modified = false;
try { try {
const std::string objtype = record["objtype"]; const std::string objtype = record["objtype"];
if (objtype == "network") { if (objtype == "network") {
auto span = tracer->StartSpan("filedb::save::network");
auto scope = tracer->WithActiveSpan(span);
const uint64_t nwid = OSUtils::jsonIntHex(record["id"], 0ULL); const uint64_t nwid = OSUtils::jsonIntHex(record["id"], 0ULL);
if (nwid) { if (nwid) {
nlohmann::json old; nlohmann::json old;
@ -106,6 +120,9 @@ bool FileDB::save(nlohmann::json& record, bool notifyListeners)
} }
} }
else if (objtype == "member") { else if (objtype == "member") {
auto span = tracer->StartSpan("filedb::save::member");
auto scope = tracer->WithActiveSpan(span);
const uint64_t id = OSUtils::jsonIntHex(record["id"], 0ULL); const uint64_t id = OSUtils::jsonIntHex(record["id"], 0ULL);
const uint64_t nwid = OSUtils::jsonIntHex(record["nwid"], 0ULL); const uint64_t nwid = OSUtils::jsonIntHex(record["nwid"], 0ULL);
if ((id) && (nwid)) { if ((id) && (nwid)) {
@ -136,6 +153,11 @@ bool FileDB::save(nlohmann::json& record, bool notifyListeners)
void FileDB::eraseNetwork(const uint64_t networkId) void FileDB::eraseNetwork(const uint64_t networkId)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("filedb");
auto span = tracer->StartSpan("filedb::eraseNetwork");
auto scope = tracer->WithActiveSpan(span);
nlohmann::json network, nullJson; nlohmann::json network, nullJson;
get(networkId, network); get(networkId, network);
char p[16384]; char p[16384];
@ -150,6 +172,11 @@ void FileDB::eraseNetwork(const uint64_t networkId)
void FileDB::eraseMember(const uint64_t networkId, const uint64_t memberId) void FileDB::eraseMember(const uint64_t networkId, const uint64_t memberId)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("filedb");
auto span = tracer->StartSpan("filedb::eraseMember");
auto scope = tracer->WithActiveSpan(span);
nlohmann::json network, member, nullJson; nlohmann::json network, member, nullJson;
get(networkId, network, memberId, member); get(networkId, network, memberId, member);
char p[4096]; char p[4096];
@ -162,6 +189,11 @@ void FileDB::eraseMember(const uint64_t networkId, const uint64_t memberId)
void FileDB::nodeIsOnline(const uint64_t networkId, const uint64_t memberId, const InetAddress& physicalAddress, const char* osArch) void FileDB::nodeIsOnline(const uint64_t networkId, const uint64_t memberId, const InetAddress& physicalAddress, const char* osArch)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("filedb");
auto span = tracer->StartSpan("filedb::nodeIsOnline");
auto scope = tracer->WithActiveSpan(span);
char mid[32], atmp[64]; char mid[32], atmp[64];
OSUtils::ztsnprintf(mid, sizeof(mid), "%.10llx", (unsigned long long)memberId); OSUtils::ztsnprintf(mid, sizeof(mid), "%.10llx", (unsigned long long)memberId);
physicalAddress.toString(atmp); physicalAddress.toString(atmp);

14
controller/PostgreSQL.cpp
View file

@ -2,6 +2,8 @@
#include "PostgreSQL.hpp" #include "PostgreSQL.hpp"
#include "opentelemetry/trace/provider.h"
#include <nlohmann/json.hpp> #include <nlohmann/json.hpp>
using namespace nlohmann; using namespace nlohmann;
@ -15,6 +17,12 @@ MemberNotificationReceiver::MemberNotificationReceiver(DB* p, pqxx::connection&
void MemberNotificationReceiver::operator()(const std::string& payload, int packend_pid) void MemberNotificationReceiver::operator()(const std::string& payload, int packend_pid)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("db_member_notification");
auto span = tracer->StartSpan("db_member_notification::operator()");
auto scope = tracer->WithActiveSpan(span);
span->SetAttribute("payload", payload);
fprintf(stderr, "Member Notification received: %s\n", payload.c_str()); fprintf(stderr, "Member Notification received: %s\n", payload.c_str());
Metrics::pgsql_mem_notification++; Metrics::pgsql_mem_notification++;
json tmp(json::parse(payload)); json tmp(json::parse(payload));
@ -38,6 +46,12 @@ NetworkNotificationReceiver::NetworkNotificationReceiver(DB* p, pqxx::connection
void NetworkNotificationReceiver::operator()(const std::string& payload, int packend_pid) void NetworkNotificationReceiver::operator()(const std::string& payload, int packend_pid)
{ {
auto provider = opentelemetry::trace::Provider::GetTracerProvider();
auto tracer = provider->GetTracer("db_network_notification");
auto span = tracer->StartSpan("db_network_notification::operator()");
auto scope = tracer->WithActiveSpan(span);
span->SetAttribute("payload", payload);
fprintf(stderr, "Network Notification received: %s\n", payload.c_str()); fprintf(stderr, "Network Notification received: %s\n", payload.c_str());
Metrics::pgsql_net_notification++; Metrics::pgsql_net_notification++;
json tmp(json::parse(payload)); json tmp(json::parse(payload));