mirror
/
devilutionX-ZeroTierOne
mirror of https://github.com/diasurgical/ZeroTierOne.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Big cleanup of controller code, should help performance.

pull/1/head
Adam Ierymenko 9 years ago
parent
4e77365e8d
commit
1205578935
4 changed files with 391 additions and 251 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 177
      controller/EmbeddedNetworkController.cpp
  2. 26
      controller/EmbeddedNetworkController.hpp
  3. 276
      controller/JSONDB.cpp
  4. 163
      controller/JSONDB.hpp

177
controller/EmbeddedNetworkController.cpp
Unescape View File

@ -59,9 +59,6 @@ using json = nlohmann::json;
// Min duration between requests for an address/nwid combo to prevent floods
#define ZT_NETCONF_MIN_REQUEST_PERIOD 1000
// Nodes are considered active if they've queried in less than this long
#define ZT_NETCONF_NODE_ACTIVE_THRESHOLD (ZT_NETWORK_AUTOCONF_DELAY * 2)
namespace ZeroTier {
static json _renderRule(ZT_VirtualNetworkRule &rule)
@ -474,9 +471,11 @@ void EmbeddedNetworkController::request(
{
Mutex::Lock _l(_threads_m);
if (_threads.size() == 0) {
long hwc = (long)std::thread::hardware_concurrency();
if (hwc <= 0)
long hwc = (long)(std::thread::hardware_concurrency() / 2);
if (hwc < 1)
hwc = 1;
else if (hwc > 16)
hwc = 16;
for(long i=0;i<hwc;++i)
_threads.push_back(Thread::start(this));
}
@ -506,8 +505,8 @@ unsigned int EmbeddedNetworkController::handleControlPlaneHttpGET(
char nwids[24];
Utils::snprintf(nwids,sizeof(nwids),"%.16llx",(unsigned long long)nwid);
json network(_db.get("network",nwids));
if (!network.size())
json network;
if (!_db.getNetwork(nwid,network))
return 404;
if (path.size() >= 3) {
@ -516,22 +515,21 @@ unsigned int EmbeddedNetworkController::handleControlPlaneHttpGET(
if (path.size() >= 4) {
const uint64_t address = Utils::hexStrToU64(path[3].c_str());
json member(_db.get("network",nwids,"member",Address(address).toString()));
if (!member.size())
json member;
if (!_db.getNetworkMember(nwid,address,member))
return 404;
_addMemberNonPersistedFields(member,OSUtils::now());
responseBody = OSUtils::jsonDump(member);
responseContentType = "application/json";
} else {
responseBody = "{";
_db.filter((std::string("network/") + nwids + "/member/"),[&responseBody](const std::string &n,const json &member) {
_db.eachMember(nwid,[&responseBody](uint64_t networkId,uint64_t nodeId,const json &member) {
if ((member.is_object())&&(member.size() > 0)) {
responseBody.append((responseBody.length() == 1) ? "\"" : ",\"");
responseBody.append(OSUtils::jsonString(member["id"],"0"));
responseBody.append("\":");
responseBody.append(OSUtils::jsonString(member["revision"],"0"));
}
return true; // never delete
});
responseBody.push_back('}');
responseContentType = "application/json";
@ -543,9 +541,9 @@ unsigned int EmbeddedNetworkController::handleControlPlaneHttpGET(
} else {
const uint64_t now = OSUtils::now();
_NetworkMemberInfo nmi;
_getNetworkMemberInfo(now,nwid,nmi);
_addNetworkNonPersistedFields(network,now,nmi);
JSONDB::NetworkSummaryInfo ns;
_db.getNetworkSummaryInfo(nwid,ns);
_addNetworkNonPersistedFields(network,now,ns);
responseBody = OSUtils::jsonDump(network);
responseContentType = "application/json";
return 200;
@ -553,21 +551,20 @@ unsigned int EmbeddedNetworkController::handleControlPlaneHttpGET(
}
} else if (path.size() == 1) {
std::set<std::string> networkIds;
_db.filter("network/",[&networkIds](const std::string &n,const json &obj) {
if (n.length() == (16 + 8))
networkIds.insert(n.substr(8));
return true; // do not delete
});
std::vector<uint64_t> networkIds(_db.networkIds());
std::sort(networkIds.begin(),networkIds.end());
char tmp[64];
responseBody.push_back('[');
for(std::set<std::string>::iterator i(networkIds.begin());i!=networkIds.end();++i) {
responseBody.append((responseBody.length() == 1) ? "\"" : ",\"");
responseBody.append(*i);
responseBody.append("\"");
for(std::vector<uint64_t>::const_iterator i(networkIds.begin());i!=networkIds.end();++i) {
if (responseBody.length() > 1)
responseBody.push_back(',');
Utils::snprintf(tmp,sizeof(tmp),"\"%.16llx\"",(unsigned long long)*i);
responseBody.append(tmp);
}
responseBody.push_back(']');
responseContentType = "application/json";
return 200;
} // else 404
@ -625,7 +622,8 @@ unsigned int EmbeddedNetworkController::handleControlPlaneHttpPOST(
char addrs[24];
Utils::snprintf(addrs,sizeof(addrs),"%.10llx",(unsigned long long)address);
json member(_db.get("network",nwids,"member",Address(address).toString()));
json member;
_db.getNetworkMember(nwid,address,member);
json origMember(member); // for detecting changes
_initMember(member);
@ -649,7 +647,6 @@ unsigned int EmbeddedNetworkController::handleControlPlaneHttpPOST(
// Member is being de-authorized, so spray Revocation objects to all online members
if (!newAuth) {
_clearNetworkMemberInfoCache(nwid);
Revocation rev((uint32_t)_node->prng(),nwid,0,now,ZT_REVOCATION_FLAG_FAST_PROPAGATE,Address(address),Revocation::CREDENTIAL_TYPE_COM);
rev.sign(_signingId);
Mutex::Lock _l(_lastRequestTime_m);
@ -722,7 +719,7 @@ unsigned int EmbeddedNetworkController::handleControlPlaneHttpPOST(
member["lastModified"] = now;
json &revj = member["revision"];
member["revision"] = (revj.is_number() ? ((uint64_t)revj + 1ULL) : 1ULL);
_db.put("network",nwids,"member",Address(address).toString(),member);
_db.saveNetworkMember(nwid,address,member);
_pushMemberUpdate(now,nwid,member);
}
@ -799,8 +796,7 @@ unsigned int EmbeddedNetworkController::handleControlPlaneHttpPOST(
Utils::getSecureRandom(&nwidPostfix,sizeof(nwidPostfix));
uint64_t tryNwid = nwidPrefix | (nwidPostfix & 0xffffffULL);
if ((tryNwid & 0xffffffULL) == 0ULL) tryNwid |= 1ULL;
Utils::snprintf(nwids,sizeof(nwids),"%.16llx",(unsigned long long)tryNwid);
if (_db.get("network",nwids).size() <= 0) {
if (!_db.hasNetwork(tryNwid)) {
nwid = tryNwid;
break;
}
@ -808,8 +804,10 @@ unsigned int EmbeddedNetworkController::handleControlPlaneHttpPOST(
if (!nwid)
return 503;
}
json network(_db.get("network",nwids));
Utils::snprintf(nwids,sizeof(nwids),"%.16llx",(unsigned long long)nwid);
json network;
_db.getNetwork(nwid,network);
json origNetwork(network); // for detecting changes
_initNetwork(network);
@ -1023,18 +1021,17 @@ unsigned int EmbeddedNetworkController::handleControlPlaneHttpPOST(
json &revj = network["revision"];
network["revision"] = (revj.is_number() ? ((uint64_t)revj + 1ULL) : 1ULL);
network["lastModified"] = now;
_db.put("network",nwids,network);
_db.saveNetwork(nwid,network);
// Send an update to all members of the network
_db.filter((std::string("network/") + nwids + "/member/"),[this,&now,&nwid](const std::string &n,const json &obj) {
_pushMemberUpdate(now,nwid,obj);
return true; // do not delete
_db.eachMember(nwid,[this,&now,&nwid](uint64_t networkId,uint64_t nodeId,const json &obj) {
this->_pushMemberUpdate(now,nwid,obj);
});
}
_NetworkMemberInfo nmi;
_getNetworkMemberInfo(now,nwid,nmi);
_addNetworkNonPersistedFields(network,now,nmi);
JSONDB::NetworkSummaryInfo ns;
_db.getNetworkSummaryInfo(nwid,ns);
_addNetworkNonPersistedFields(network,now,ns);
responseBody = OSUtils::jsonDump(network);
responseContentType = "application/json";
@ -1074,20 +1071,11 @@ unsigned int EmbeddedNetworkController::handleControlPlaneHttpDELETE(
if (path[0] == "network") {
if ((path.size() >= 2)&&(path[1].length() == 16)) {
const uint64_t nwid = Utils::hexStrToU64(path[1].c_str());
char nwids[24];
Utils::snprintf(nwids,sizeof(nwids),"%.16llx",nwid);
json network(_db.get("network",nwids));
if (!network.size())
return 404;
if (path.size() >= 3) {
if ((path.size() == 4)&&(path[2] == "member")&&(path[3].length() == 10)) {
const uint64_t address = Utils::hexStrToU64(path[3].c_str());
json member = _db.get("network",nwids,"member",Address(address).toString());
_db.erase("network",nwids,"member",Address(address).toString());
json member = _db.eraseNetworkMember(nwid,address);
if (!member.size())
return 404;
responseBody = OSUtils::jsonDump(member);
@ -1095,15 +1083,9 @@ unsigned int EmbeddedNetworkController::handleControlPlaneHttpDELETE(
return 200;
}
} else {
std::string pfx("network/");
pfx.append(nwids);
_db.filter(pfx,[](const std::string &n,const json &obj) {
return false; // delete
});
Mutex::Lock _l2(_nmiCache_m);
_nmiCache.erase(nwid);
json network = _db.eraseNetwork(nwid);
if (!network.size())
return 404;
responseBody = OSUtils::jsonDump(network);
responseContentType = "application/json";
return 200;
@ -1143,7 +1125,7 @@ void EmbeddedNetworkController::threadMain()
void EmbeddedNetworkController::_circuitTestCallback(ZT_Node *node,ZT_CircuitTest *test,const ZT_CircuitTestReport *report)
{
char tmp[1024],id[128];
char tmp[2048],id[128];
EmbeddedNetworkController *const self = reinterpret_cast<EmbeddedNetworkController *>(test->ptr);
if ((!test)||(!report)||(!test->credentialNetworkId)) return; // sanity check
@ -1152,6 +1134,7 @@ void EmbeddedNetworkController::_circuitTestCallback(ZT_Node *node,ZT_CircuitTes
Utils::snprintf(id,sizeof(id),"network/%.16llx/test/%.16llx-%.16llx-%.10llx-%.10llx",test->credentialNetworkId,test->testId,now,report->upstream,report->current);
Utils::snprintf(tmp,sizeof(tmp),
"{\"id\": \"%s\","
"\"objtype\": \"circuit_test\","
"\"timestamp\": %llu,"
"\"networkId\": \"%.16llx\","
"\"testId\": \"%.16llx\","
@ -1219,16 +1202,14 @@ void EmbeddedNetworkController::_request(
char nwids[24];
Utils::snprintf(nwids,sizeof(nwids),"%.16llx",nwid);
json network(_db.get("network",nwids));
json member(_db.get("network",nwids,"member",identity.address().toString()));
if (!network.size()) {
json network,member;
JSONDB::NetworkSummaryInfo ns;
if (!_db.getNetworkAndMember(nwid,identity.address().toInt(),network,member,ns)) {
_sender->ncSendError(nwid,requestPacketId,identity.address(),NetworkController::NC_ERROR_OBJECT_NOT_FOUND);
return;
}
const bool newMember = (member.size() == 0);
json origMember(member); // for detecting modification later
_initMember(member);
@ -1365,7 +1346,7 @@ void EmbeddedNetworkController::_request(
if (!authorizedBy) {
if (origMember != member) {
member["lastModified"] = now;
_db.put("network",nwids,"member",identity.address().toString(),member);
_db.saveNetworkMember(nwid,identity.address().toInt(),member);
}
_sender->ncSendError(nwid,requestPacketId,identity.address(),NetworkController::NC_ERROR_ACCESS_DENIED);
return;
@ -1376,15 +1357,13 @@ void EmbeddedNetworkController::_request(
// -------------------------------------------------------------------------
NetworkConfig nc;
_NetworkMemberInfo nmi;
_getNetworkMemberInfo(now,nwid,nmi);
uint64_t credentialtmd = ZT_NETWORKCONFIG_DEFAULT_CREDENTIAL_TIME_MAX_MAX_DELTA;
if (now > nmi.mostRecentDeauthTime) {
if (now > ns.mostRecentDeauthTime) {
// If we recently de-authorized a member, shrink credential TTL/max delta to
// be below the threshold required to exclude it. Cap this to a min/max to
// prevent jitter or absurdly large values.
const uint64_t deauthWindow = now - nmi.mostRecentDeauthTime;
const uint64_t deauthWindow = now - ns.mostRecentDeauthTime;
if (deauthWindow < ZT_NETWORKCONFIG_DEFAULT_CREDENTIAL_TIME_MIN_MAX_DELTA) {
credentialtmd = ZT_NETWORKCONFIG_DEFAULT_CREDENTIAL_TIME_MIN_MAX_DELTA;
} else if (deauthWindow < (ZT_NETWORKCONFIG_DEFAULT_CREDENTIAL_TIME_MAX_MAX_DELTA + 5000ULL)) {
@ -1403,9 +1382,8 @@ void EmbeddedNetworkController::_request(
Utils::scopy(nc.name,sizeof(nc.name),OSUtils::jsonString(network["name"],"").c_str());
nc.multicastLimit = (unsigned int)OSUtils::jsonInt(network["multicastLimit"],32ULL);
for(std::set<Address>::const_iterator ab(nmi.activeBridges.begin());ab!=nmi.activeBridges.end();++ab) {
for(std::vector<Address>::const_iterator ab(ns.activeBridges.begin());ab!=ns.activeBridges.end();++ab)
nc.addSpecialist(*ab,ZT_NETWORKCONFIG_SPECIALIST_TYPE_ACTIVE_BRIDGE);
}
json &v4AssignMode = network["v4AssignMode"];
json &v6AssignMode = network["v6AssignMode"];
@ -1629,14 +1607,13 @@ void EmbeddedNetworkController::_request(
}
// If it's routed, then try to claim and assign it and if successful end loop
if ((routedNetmaskBits > 0)&&(!nmi.allocatedIps.count(ip6))) {
if ( (routedNetmaskBits > 0) && (!std::binary_search(ns.allocatedIps.begin(),ns.allocatedIps.end(),ip6)) ) {
ipAssignments.push_back(ip6.toIpString());
member["ipAssignments"] = ipAssignments;
ip6.setPort((unsigned int)routedNetmaskBits);
if (nc.staticIpCount < ZT_MAX_ZT_ASSIGNED_ADDRESSES)
nc.staticIps[nc.staticIpCount++] = ip6;
haveManagedIpv6AutoAssignment = true;
_clearNetworkMemberInfoCache(nwid); // clear cache to prevent IP assignment duplication on many rapid assigns
break;
}
}
@ -1682,7 +1659,7 @@ void EmbeddedNetworkController::_request(
// If it's routed, then try to claim and assign it and if successful end loop
const InetAddress ip4(Utils::hton(ip),0);
if ((routedNetmaskBits > 0)&&(!nmi.allocatedIps.count(ip4))) {
if ( (routedNetmaskBits > 0) && (!std::binary_search(ns.allocatedIps.begin(),ns.allocatedIps.end(),ip4)) ) {
ipAssignments.push_back(ip4.toIpString());
member["ipAssignments"] = ipAssignments;
if (nc.staticIpCount < ZT_MAX_ZT_ASSIGNED_ADDRESSES) {
@ -1692,7 +1669,6 @@ void EmbeddedNetworkController::_request(
v4ip->sin_addr.s_addr = Utils::hton(ip);
}
haveManagedIpv4AutoAssignment = true;
_clearNetworkMemberInfoCache(nwid); // clear cache to prevent IP assignment duplication on many rapid assigns
break;
}
}
@ -1720,65 +1696,12 @@ void EmbeddedNetworkController::_request(
if (member != origMember) {
member["lastModified"] = now;
_db.put("network",nwids,"member",identity.address().toString(),member);
_db.saveNetworkMember(nwid,identity.address().toInt(),member);
}
_sender->ncSendConfig(nwid,requestPacketId,identity.address(),nc,metaData.getUI(ZT_NETWORKCONFIG_REQUEST_METADATA_KEY_VERSION,0) < 6);
}
void EmbeddedNetworkController::_getNetworkMemberInfo(uint64_t now,uint64_t nwid,_NetworkMemberInfo &nmi)
{
char pfx[256];
Utils::snprintf(pfx,sizeof(pfx),"network/%.16llx/member",nwid);
Mutex::Lock _l(_nmiCache_m);
std::map<uint64_t,_NetworkMemberInfo>::iterator c(_nmiCache.find(nwid));
if ((c != _nmiCache.end())&&((now - c->second.nmiTimestamp) < 1000)) { // a short duration cache but limits CPU use on big networks
nmi = c->second;
return;
}
_db.filter(pfx,[&nmi,&now](const std::string &n,const json &member) {
try {
if (OSUtils::jsonBool(member["authorized"],false)) {
++nmi.authorizedMemberCount;
if (member.count("recentLog")) {
const json &mlog = member["recentLog"];
if ((mlog.is_array())&&(mlog.size() > 0)) {
const json &mlog1 = mlog[0];
if (mlog1.is_object()) {
if ((now - OSUtils::jsonInt(mlog1["ts"],0ULL)) < ZT_NETCONF_NODE_ACTIVE_THRESHOLD)
++nmi.activeMemberCount;
}
}
}
if (OSUtils::jsonBool(member["activeBridge"],false)) {
nmi.activeBridges.insert(Address(Utils::hexStrToU64(OSUtils::jsonString(member["id"],"0000000000").c_str())));
}
if (member.count("ipAssignments")) {
const json &mips = member["ipAssignments"];
if (mips.is_array()) {
for(unsigned long i=0;i<mips.size();++i) {
InetAddress mip(OSUtils::jsonString(mips[i],""));
if ((mip.ss_family == AF_INET)||(mip.ss_family == AF_INET6))
nmi.allocatedIps.insert(mip);
}
}
}
} else {
nmi.mostRecentDeauthTime = std::max(nmi.mostRecentDeauthTime,OSUtils::jsonInt(member["lastDeauthorizedTime"],0ULL));
}
} catch ( ... ) {}
return true;
});
nmi.nmiTimestamp = now;
_nmiCache[nwid] = nmi;
}
void EmbeddedNetworkController::_pushMemberUpdate(uint64_t now,uint64_t nwid,const nlohmann::json &member)
{
try {

26
controller/EmbeddedNetworkController.hpp
Unescape View File

@ -104,23 +104,8 @@ private:
Dictionary<ZT_NETWORKCONFIG_METADATA_DICT_CAPACITY> metaData;
};
// Gathers a bunch of statistics about members of a network, IP assignments, etc. that we need in various places
struct _NetworkMemberInfo
{
_NetworkMemberInfo() : authorizedMemberCount(0),activeMemberCount(0),totalMemberCount(0),mostRecentDeauthTime(0) {}
std::set<Address> activeBridges;
std::set<InetAddress> allocatedIps;
unsigned long authorizedMemberCount;
unsigned long activeMemberCount;
unsigned long totalMemberCount;
uint64_t mostRecentDeauthTime;
uint64_t nmiTimestamp; // time this NMI structure was computed
};
static void _circuitTestCallback(ZT_Node *node,ZT_CircuitTest *test,const ZT_CircuitTestReport *report);
void _request(uint64_t nwid,const InetAddress &fromAddr,uint64_t requestPacketId,const Identity &identity,const Dictionary<ZT_NETWORKCONFIG_METADATA_DICT_CAPACITY> &metaData);
void _getNetworkMemberInfo(uint64_t now,uint64_t nwid,_NetworkMemberInfo &nmi);
inline void _clearNetworkMemberInfoCache(const uint64_t nwid) { Mutex::Lock _l(_nmiCache_m); _nmiCache.erase(nwid); }
void _pushMemberUpdate(uint64_t now,uint64_t nwid,const nlohmann::json &member);
// These init objects with default and static/informational fields
@ -164,12 +149,12 @@ private:
}
network["objtype"] = "network";
}
inline void _addNetworkNonPersistedFields(nlohmann::json &network,uint64_t now,const _NetworkMemberInfo &nmi)
inline void _addNetworkNonPersistedFields(nlohmann::json &network,uint64_t now,const JSONDB::NetworkSummaryInfo &ns)
{
network["clock"] = now;
network["authorizedMemberCount"] = nmi.authorizedMemberCount;
network["activeMemberCount"] = nmi.activeMemberCount;
network["totalMemberCount"] = nmi.totalMemberCount;
network["authorizedMemberCount"] = ns.authorizedMemberCount;
network["activeMemberCount"] = ns.activeMemberCount;
network["totalMemberCount"] = ns.totalMemberCount;
}
inline void _addMemberNonPersistedFields(nlohmann::json &member,uint64_t now)
{
@ -183,9 +168,6 @@ private:
std::vector<Thread> _threads;
Mutex _threads_m;
std::map<uint64_t,_NetworkMemberInfo> _nmiCache;
Mutex _nmiCache_m;
JSONDB _db;
Node *const _node;

276
controller/JSONDB.cpp
Unescape View File

@ -26,12 +26,11 @@ static const nlohmann::json _EMPTY_JSON(nlohmann::json::object());
static const std::map<std::string,std::string> _ZT_JSONDB_GET_HEADERS;
JSONDB::JSONDB(const std::string &basePath) :
_basePath(basePath),
_ready(false)
_basePath(basePath)
{
if ((_basePath.length() > 7)&&(_basePath.substr(0,7) == "http://")) {
// TODO: this doesn't yet support IPv6 since bracketed address notiation isn't supported.
// Typically it's used with 127.0.0.1 anyway.
// Typically it's just used with 127.0.0.1 anyway.
std::string hn = _basePath.substr(7);
std::size_t hnend = hn.find_first_of('/');
if (hnend != std::string::npos)
@ -50,7 +49,32 @@ JSONDB::JSONDB(const std::string &basePath) :
OSUtils::mkdir(_basePath.c_str());
OSUtils::lockDownFile(_basePath.c_str(),true); // networks might contain auth tokens, etc., so restrict directory permissions
}
_reload(_basePath,std::string());
unsigned int cnt = 0;
while (!_load(_basePath)) {
if ((++cnt & 7) == 0)
fprintf(stderr,"WARNING: controller still waiting to read '%s'..." ZT_EOL_S,_basePath.c_str());
Thread::sleep(250);
}
for(std::unordered_map<uint64_t,_NW>::iterator n(_networks.begin());n!=_networks.end();++n)
_recomputeSummaryInfo(n->first);
}
JSONDB::~JSONDB()
{
{
Mutex::Lock _l(_networks_m);
_networks.clear();
}
{
Mutex::Lock _l(_summaryThread_m);
if (_summaryThread) {
_updateSummaryInfoQueue.post(0);
_updateSummaryInfoQueue.post(0);
Thread::join(_summaryThread);
}
}
}
bool JSONDB::writeRaw(const std::string &n,const std::string &obj)
@ -73,84 +97,173 @@ bool JSONDB::writeRaw(const std::string &n,const std::string &obj)
}
}
bool JSONDB::put(const std::string &n,const nlohmann::json &obj)
void JSONDB::saveNetwork(const uint64_t networkId,const nlohmann::json &networkConfig)
{
const bool r = writeRaw(n,OSUtils::jsonDump(obj));
char n[256];
Utils::snprintf(n,sizeof(n),"network/%.16llx",(unsigned long long)networkId);
writeRaw(n,OSUtils::jsonDump(networkConfig));
{
Mutex::Lock _l(_db_m);
_db[n].obj = obj;
Mutex::Lock _l(_networks_m);
_networks[networkId].config = networkConfig;
}
return r;
//_recomputeSummaryInfo(networkId);
}
nlohmann::json JSONDB::get(const std::string &n)
void JSONDB::saveNetworkMember(const uint64_t networkId,const uint64_t nodeId,const nlohmann::json &memberConfig)
{
while (!_ready) {
Thread::sleep(250);
_reload(_basePath,std::string());
char n[256];
Utils::snprintf(n,sizeof(n),"network/%.16llx/member/%.10llx",(unsigned long long)networkId,(unsigned long long)nodeId);
writeRaw(n,OSUtils::jsonDump(memberConfig));
{
Mutex::Lock _l(_networks_m);
_networks[networkId].members[nodeId] = memberConfig;
}
_recomputeSummaryInfo(networkId);
}
{
Mutex::Lock _l(_db_m);
std::map<std::string,_E>::iterator e(_db.find(n));
if (e != _db.end())
return e->second.obj;
nlohmann::json JSONDB::eraseNetwork(const uint64_t networkId)
{
if (!_httpAddr) { // Member deletion is done by Central in harnessed mode, and deleting the cache network entry also deletes all members
std::vector<uint64_t> memberIds;
{
Mutex::Lock _l(_networks_m);
std::unordered_map<uint64_t,_NW>::iterator i(_networks.find(networkId));
if (i == _networks.end())
return _EMPTY_JSON;
for(std::unordered_map<uint64_t,nlohmann::json>::iterator m(i->second.members.begin());m!=i->second.members.end();++m)
memberIds.push_back(m->first);
}
for(std::vector<uint64_t>::iterator m(memberIds.begin());m!=memberIds.end();++m)
eraseNetworkMember(networkId,*m,false);
}
std::string buf;
char n[256];
Utils::snprintf(n,sizeof(n),"network/%.16llx",(unsigned long long)networkId);
if (_httpAddr) {
std::map<std::string,std::string> headers;
const unsigned int sc = Http::GET(1048576,ZT_JSONDB_HTTP_TIMEOUT,reinterpret_cast<const struct sockaddr *>(&_httpAddr),(_basePath+"/"+n).c_str(),_ZT_JSONDB_GET_HEADERS,headers,buf);
if (sc != 200)
return _EMPTY_JSON;
// Deletion is currently done by Central in harnessed mode
//std::map<std::string,std::string> headers;
//std::string body;
//Http::DEL(1048576,ZT_JSONDB_HTTP_TIMEOUT,reinterpret_cast<const struct sockaddr *>(&_httpAddr),(_basePath+"/"+n).c_str(),_ZT_JSONDB_GET_HEADERS,headers,body);
} else {
const std::string path(_genPath(n,false));
if (!path.length())
return _EMPTY_JSON;
if (!OSUtils::readFile(path.c_str(),buf))
return _EMPTY_JSON;
if (path.length())
OSUtils::rm(path.c_str());
}
{
Mutex::Lock _l(_db_m);
try {
_E &e2 = _db[n];
e2.obj = OSUtils::jsonParse(buf);
return e2.obj;
} catch ( ... ) {
_db.erase(n);
return _EMPTY_JSON;
}
Mutex::Lock _l(_networks_m);
std::unordered_map<uint64_t,_NW>::iterator i(_networks.find(networkId));
if (i == _networks.end())
return _EMPTY_JSON; // sanity check, shouldn't happen
nlohmann::json tmp(i->second.config);
_networks.erase(i);
return tmp;
}
}
void JSONDB::erase(const std::string &n)
nlohmann::json JSONDB::eraseNetworkMember(const uint64_t networkId,const uint64_t nodeId,bool recomputeSummaryInfo)
{
_erase(n);
char n[256];
Utils::snprintf(n,sizeof(n),"network/%.16llx/member/%.10llx",(unsigned long long)networkId,(unsigned long long)nodeId);
if (_httpAddr) {
// Deletion is currently done by the caller in Central harnessed mode
//std::map<std::string,std::string> headers;
//std::string body;
//Http::DEL(1048576,ZT_JSONDB_HTTP_TIMEOUT,reinterpret_cast<const struct sockaddr *>(&_httpAddr),(_basePath+"/"+n).c_str(),_ZT_JSONDB_GET_HEADERS,headers,body);
} else {
const std::string path(_genPath(n,false));
if (path.length())
OSUtils::rm(path.c_str());
}
{
Mutex::Lock _l(_db_m);
_db.erase(n);
Mutex::Lock _l(_networks_m);
std::unordered_map<uint64_t,_NW>::iterator i(_networks.find(networkId));
if (i == _networks.end())
return _EMPTY_JSON;
std::unordered_map<uint64_t,nlohmann::json>::iterator j(i->second.members.find(nodeId));
if (j == i->second.members.end())
return _EMPTY_JSON;
nlohmann::json tmp(j->second);
i->second.members.erase(j);
if (recomputeSummaryInfo)
_recomputeSummaryInfo(networkId);
return tmp;
}
}
void JSONDB::_erase(const std::string &n)
void JSONDB::threadMain()
throw()
{
if (_httpAddr) {
std::string body;
std::map<std::string,std::string> headers;
Http::DEL(1048576,ZT_JSONDB_HTTP_TIMEOUT,reinterpret_cast<const struct sockaddr *>(&_httpAddr),(_basePath+"/"+n).c_str(),_ZT_JSONDB_GET_HEADERS,headers,body);
} else {
std::string path(_genPath(n,true));
if (!path.length())
return;
OSUtils::rm(path.c_str());
uint64_t networkId = 0;
while ((networkId = _updateSummaryInfoQueue.get()) != 0) {
const uint64_t now = OSUtils::now();
{
Mutex::Lock _l(_networks_m);
std::unordered_map<uint64_t,_NW>::iterator n(_networks.find(networkId));
if (n != _networks.end()) {
NetworkSummaryInfo &ns = n->second.summaryInfo;
ns.activeBridges.clear();
ns.allocatedIps.clear();
ns.authorizedMemberCount = 0;
ns.activeMemberCount = 0;
ns.totalMemberCount = 0;
ns.mostRecentDeauthTime = 0;
for(std::unordered_map<uint64_t,nlohmann::json>::const_iterator m(n->second.members.begin());m!=n->second.members.end();++m) {
try {
if (OSUtils::jsonBool(m->second["authorized"],false)) {
++ns.authorizedMemberCount;
try {
const nlohmann::json &mlog = m->second["recentLog"];
if ((mlog.is_array())&&(mlog.size() > 0)) {
const nlohmann::json &mlog1 = mlog[0];
if (mlog1.is_object()) {
if ((now - OSUtils::jsonInt(mlog1["ts"],0ULL)) < (ZT_NETWORK_AUTOCONF_DELAY * 2))
++ns.activeMemberCount;
}
}
} catch ( ... ) {}
try {
if (OSUtils::jsonBool(m->second["activeBridge"],false))
ns.activeBridges.push_back(Address(m->first));
} catch ( ... ) {}
try {
const nlohmann::json &mips = m->second["ipAssignments"];
if (mips.is_array()) {
for(unsigned long i=0;i<mips.size();++i) {
InetAddress mip(OSUtils::jsonString(mips[i],""));
if ((mip.ss_family == AF_INET)||(mip.ss_family == AF_INET6))
ns.allocatedIps.push_back(mip);
}
}
} catch ( ... ) {}
} else {
try {
ns.mostRecentDeauthTime = std::max(ns.mostRecentDeauthTime,OSUtils::jsonInt(m->second["lastDeauthorizedTime"],0ULL));
} catch ( ... ) {}
}
++ns.totalMemberCount;
} catch ( ... ) {}
}
std::sort(ns.activeBridges.begin(),ns.activeBridges.end());
std::sort(ns.allocatedIps.begin(),ns.allocatedIps.end());
n->second.summaryInfoLastComputed = now;
}
}
}
}
void JSONDB::_reload(const std::string &p,const std::string &b)
bool JSONDB::_load(const std::string &p)
{
if (_httpAddr) {
Mutex::Lock _l(_db_m);
std::string body;
std::map<std::string,std::string> headers;
const unsigned int sc = Http::GET(2147483647,ZT_JSONDB_HTTP_TIMEOUT,reinterpret_cast<const struct sockaddr *>(&_httpAddr),_basePath.c_str(),_ZT_JSONDB_GET_HEADERS,headers,body);
@ -159,30 +272,73 @@ void JSONDB::_reload(const std::string &p,const std::string &b)
nlohmann::json dbImg(OSUtils::jsonParse(body));
std::string tmp;
if (dbImg.is_object()) {
_db.clear();
Mutex::Lock _l(_networks_m);
for(nlohmann::json::iterator i(dbImg.begin());i!=dbImg.end();++i) {
if (i.value().is_object()) {
tmp = i.key();
_db[tmp].obj = i.value();
nlohmann::json &j = i.value();
if (j.is_object()) {
std::string id(OSUtils::jsonString(j["id"],"0"));
std::string objtype(OSUtils::jsonString(j["objtype"],""));
if ((id.length() == 16)&&(objtype == "network")) {
const uint64_t nwid = Utils::hexStrToU64(id.c_str());
if (nwid)
_networks[nwid].config = j;
} else if ((id.length() == 10)&&(objtype == "member")) {
const uint64_t mid = Utils::hexStrToU64(id.c_str());
const uint64_t nwid = Utils::hexStrToU64(OSUtils::jsonString(j["nwid"],"0").c_str());
if ((mid)&&(nwid))
_networks[nwid].members[mid] = j;
}
}
}
_ready = true;
return true;
}
} catch ( ... ) {} // invalid JSON, so maybe incomplete request
}
return false;
} else {
_ready = true;
std::vector<std::string> dl(OSUtils::listDirectory(p.c_str(),true));
for(std::vector<std::string>::const_iterator di(dl.begin());di!=dl.end();++di) {
if ((di->length() > 5)&&(di->substr(di->length() - 5) == ".json")) {
this->get(b + di->substr(0,di->length() - 5));
std::string buf;
if (OSUtils::readFile((p + ZT_PATH_SEPARATOR_S + *di).c_str(),buf)) {
try {
nlohmann::json j(OSUtils::jsonParse(buf));
std::string id(OSUtils::jsonString(j["id"],"0"));
std::string objtype(OSUtils::jsonString(j["objtype"],""));
if ((id.length() == 16)&&(objtype == "network")) {
const uint64_t nwid = Utils::strToU64(id.c_str());
if (nwid) {
Mutex::Lock _l(_networks_m);
_networks[nwid].config = j;
}
} else if ((id.length() == 10)&&(objtype == "member")) {
const uint64_t mid = Utils::strToU64(id.c_str());
const uint64_t nwid = Utils::strToU64(OSUtils::jsonString(j["nwid"],"0").c_str());
if ((mid)&&(nwid)) {
Mutex::Lock _l(_networks_m);
_networks[nwid].members[mid] = j;
}
}
} catch ( ... ) {}
}
} else {
this->_reload((p + ZT_PATH_SEPARATOR + *di),(b + *di + ZT_PATH_SEPARATOR));
this->_load((p + ZT_PATH_SEPARATOR_S + *di));
}
}
return true;
}
}
void JSONDB::_recomputeSummaryInfo(const uint64_t networkId)
{
Mutex::Lock _l(_summaryThread_m);
if (!_summaryThread)
_summaryThread = Thread::start(this);
_updateSummaryInfoQueue.post(networkId);
}
std::string JSONDB::_genPath(const std::string &n,bool create)
{
std::vector<std::string> pt(OSUtils::split(n.c_str(),"/","",""));

163
controller/JSONDB.hpp
Unescape View File

@ -28,6 +28,7 @@
#include <stdexcept>
#include <vector>
#include <algorithm>
#include <unordered_map>
#include "../node/Constants.hpp"
#include "../node/Utils.hpp"
@ -37,6 +38,7 @@
#include "../osdep/OSUtils.hpp"
#include "../osdep/Http.hpp"
#include "../osdep/Thread.hpp"
#include "../osdep/BlockingQueue.hpp"
namespace ZeroTier {
@ -46,68 +48,145 @@ namespace ZeroTier {
class JSONDB
{
public:
struct NetworkSummaryInfo
{
NetworkSummaryInfo() : authorizedMemberCount(0),activeMemberCount(0),totalMemberCount(0),mostRecentDeauthTime(0) {}
std::vector<Address> activeBridges;
std::vector<InetAddress> allocatedIps;
unsigned long authorizedMemberCount;
unsigned long activeMemberCount;
unsigned long totalMemberCount;
uint64_t mostRecentDeauthTime;
};
JSONDB(const std::string &basePath);
~JSONDB();
bool writeRaw(const std::string &n,const std::string &obj);
bool put(const std::string &n,const nlohmann::json &obj);
inline bool put(const std::string &n1,const std::string &n2,const nlohmann::json &obj) { return this->put((n1 + "/" + n2),obj); }
inline bool put(const std::string &n1,const std::string &n2,const std::string &n3,const nlohmann::json &obj) { return this->put((n1 + "/" + n2 + "/" + n3),obj); }
inline bool put(const std::string &n1,const std::string &n2,const std::string &n3,const std::string &n4,const nlohmann::json &obj) { return this->put((n1 + "/" + n2 + "/" + n3 + "/" + n4),obj); }
inline bool put(const std::string &n1,const std::string &n2,const std::string &n3,const std::string &n4,const std::string &n5,const nlohmann::json &obj) { return this->put((n1 + "/" + n2 + "/" + n3 + "/" + n4 + "/" + n5),obj); }
inline bool hasNetwork(const uint64_t networkId) const
{
Mutex::Lock _l(_networks_m);
std::unordered_map<uint64_t,_NW>::const_iterator i(_networks.find(networkId));
return (i != _networks.end());
}
inline bool getNetwork(const uint64_t networkId,nlohmann::json &config) const
{
Mutex::Lock _l(_networks_m);
std::unordered_map<uint64_t,_NW>::const_iterator i(_networks.find(networkId));
if (i == _networks.end())
return false;
config = i->second.config;
return true;
}
inline bool getNetworkSummaryInfo(const uint64_t networkId,NetworkSummaryInfo &ns) const
{
for(;;) {
{
Mutex::Lock _l(_networks_m);
std::unordered_map<uint64_t,_NW>::const_iterator i(_networks.find(networkId));
if (i == _networks.end())
return false;
if (i->second.summaryInfoLastComputed) {
ns = i->second.summaryInfo;
return true;
}
}
Thread::sleep(100); // wait for this to be done the first time, which happens when we start
}
}
/**
* @return Bit mask: 0 == none, 1 == network only, 3 == network and member
*/
inline int getNetworkAndMember(const uint64_t networkId,const uint64_t nodeId,nlohmann::json &networkConfig,nlohmann::json &memberConfig,NetworkSummaryInfo &ns) const
{
Mutex::Lock _l(_networks_m);
std::unordered_map<uint64_t,_NW>::const_iterator i(_networks.find(networkId));
if (i == _networks.end())
return 0;
networkConfig = i->second.config;
ns = i->second.summaryInfo;
std::unordered_map<uint64_t,nlohmann::json>::const_iterator j(i->second.members.find(nodeId));
if (j == i->second.members.end())
return 1;
memberConfig = j->second;
return 3;
}
inline bool getNetworkMember(const uint64_t networkId,const uint64_t nodeId,nlohmann::json &memberConfig) const
{
Mutex::Lock _l(_networks_m);
std::unordered_map<uint64_t,_NW>::const_iterator i(_networks.find(networkId));
if (i == _networks.end())
return false;
std::unordered_map<uint64_t,nlohmann::json>::const_iterator j(i->second.members.find(nodeId));
if (j == i->second.members.end())
return false;
memberConfig = j->second;
return true;
}
void saveNetwork(const uint64_t networkId,const nlohmann::json &networkConfig);
nlohmann::json get(const std::string &n);
inline nlohmann::json get(const std::string &n1,const std::string &n2) { return this->get((n1 + "/" + n2)); }
inline nlohmann::json get(const std::string &n1,const std::string &n2,const std::string &n3) { return this->get((n1 + "/" + n2 + "/" + n3)); }
inline nlohmann::json get(const std::string &n1,const std::string &n2,const std::string &n3,const std::string &n4) { return this->get((n1 + "/" + n2 + "/" + n3 + "/" + n4)); }
inline nlohmann::json get(const std::string &n1,const std::string &n2,const std::string &n3,const std::string &n4,const std::string &n5) { return this->get((n1 + "/" + n2 + "/" + n3 + "/" + n4 + "/" + n5)); }
void saveNetworkMember(const uint64_t networkId,const uint64_t nodeId,const nlohmann::json &memberConfig);
void erase(const std::string &n);
nlohmann::json eraseNetwork(const uint64_t networkId);
inline void erase(const std::string &n1,const std::string &n2) { this->erase(n1 + "/" + n2); }
inline void erase(const std::string &n1,const std::string &n2,const std::string &n3) { this->erase(n1 + "/" + n2 + "/" + n3); }
inline void erase(const std::string &n1,const std::string &n2,const std::string &n3,const std::string &n4) { this->erase(n1 + "/" + n2 + "/" + n3 + "/" + n4); }
inline void erase(const std::string &n1,const std::string &n2,const std::string &n3,const std::string &n4,const std::string &n5) { this->erase(n1 + "/" + n2 + "/" + n3 + "/" + n4 + "/" + n5); }
nlohmann::json eraseNetworkMember(const uint64_t networkId,const uint64_t nodeId,bool recomputeSummaryInfo = true);
std::vector<uint64_t> networkIds() const
{
std::vector<uint64_t> r;
Mutex::Lock _l(_networks_m);
for(std::unordered_map<uint64_t,_NW>::const_iterator n(_networks.begin());n!=_networks.end();++n)
r.push_back(n->first);
return r;
}
template<typename F>
inline void filter(const std::string &prefix,F func)
inline void eachMember(const uint64_t networkId,F func)
{
while (!_ready) {
Thread::sleep(250);
_reload(_basePath,std::string());
}
{
Mutex::Lock _l(_db_m);
for(std::map<std::string,_E>::iterator i(_db.lower_bound(prefix));i!=_db.end();) {
if ((i->first.length() >= prefix.length())&&(!memcmp(i->first.data(),prefix.data(),prefix.length()))) {
if (!func(i->first,i->second.obj)) {
this->_erase(i->first);
_db.erase(i++);
} else {
++i;
}
} else break;
Mutex::Lock _l(_networks_m);
std::unordered_map<uint64_t,_NW>::const_iterator i(_networks.find(networkId));
if (i != _networks.end()) {
for(std::unordered_map<uint64_t,nlohmann::json>::const_iterator m(i->second.members.begin());m!=i->second.members.end();++m) {
try {
func(networkId,m->first,m->second);
} catch ( ... ) {}
}
}
}
void threadMain()
throw();
private:
void _erase(const std::string &n);
void _reload(const std::string &p,const std::string &b);
bool _load(const std::string &p);
void _recomputeSummaryInfo(const uint64_t networkId);
std::string _genPath(const std::string &n,bool create);
struct _E
std::string _basePath;
InetAddress _httpAddr;
BlockingQueue<uint64_t> _updateSummaryInfoQueue;
Thread _summaryThread;
Mutex _summaryThread_m;
struct _NW
{
nlohmann::json obj;
inline bool operator==(const _E &e) const { return (obj == e.obj); }
inline bool operator!=(const _E &e) const { return (obj != e.obj); }
_NW() : summaryInfoLastComputed(0) {}
nlohmann::json config;
NetworkSummaryInfo summaryInfo;
uint64_t summaryInfoLastComputed;
std::unordered_map<uint64_t,nlohmann::json> members;
};
InetAddress _httpAddr;
std::string _basePath;
std::map<std::string,_E> _db;
Mutex _db_m;
volatile bool _ready;
std::unordered_map<uint64_t,_NW> _networks;
Mutex _networks_m;
};
} // namespace ZeroTier

Write Preview
Loading…
Cancel
Save