From 1613f42d0082cf6438ad0c62d89405ab82625f98 Mon Sep 17 00:00:00 2001 From: Adam Ierymenko Date: Tue, 7 Nov 2017 14:44:46 -0800 Subject: Re-integrate in-filesystem DB into new controller DB structure. --- attic/JSONDB.cpp | 525 +++++++++++++++++++++++++++++++++++++++++++++++++++++++ attic/JSONDB.hpp | 192 ++++++++++++++++++++ 2 files changed, 717 insertions(+) create mode 100644 attic/JSONDB.cpp create mode 100644 attic/JSONDB.hpp (limited to 'attic') diff --git a/attic/JSONDB.cpp b/attic/JSONDB.cpp new file mode 100644 index 00000000..67b13393 --- /dev/null +++ b/attic/JSONDB.cpp @@ -0,0 +1,525 @@ +/* + * ZeroTier One - Network Virtualization Everywhere + * Copyright (C) 2011-2015 ZeroTier, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ + +#include +#include +#include +#ifndef _WIN32 +#include +#include +#include +#include +#include +#include +#endif + +#include "JSONDB.hpp" +#include "EmbeddedNetworkController.hpp" + +namespace ZeroTier { + +static const nlohmann::json _EMPTY_JSON(nlohmann::json::object()); + +JSONDB::JSONDB(const std::string &basePath,EmbeddedNetworkController *parent) : + _parent(parent), + _basePath(basePath), + _rawInput(-1), + _rawOutput(-1), + _summaryThreadRun(true), + _dataReady(false) +{ +#ifndef __WINDOWS__ + if (_basePath == "-") { + // If base path is "-" we run in Central harnessed mode. We read pseudo-http-requests from stdin and write + // them to stdout. + _rawInput = STDIN_FILENO; + _rawOutput = STDOUT_FILENO; + fcntl(_rawInput,F_SETFL,O_NONBLOCK); + } else { +#endif + // Default mode of operation is to store files in the filesystem + OSUtils::mkdir(_basePath.c_str()); + OSUtils::lockDownFile(_basePath.c_str(),true); // networks might contain auth tokens, etc., so restrict directory permissions +#ifndef __WINDOWS__ + } +#endif + + _networks_m.lock(); // locked until data is loaded, etc. + + if (_rawInput < 0) { + _load(basePath); + _dataReady = true; + _networks_m.unlock(); + } else { + // In harnessed mode we leave the lock locked and wait for our initial DB from Central. + _summaryThread = Thread::start(this); + } +} + +JSONDB::~JSONDB() +{ + Thread t; + { + Mutex::Lock _l(_summaryThread_m); + _summaryThreadRun = false; + t = _summaryThread; + } + if (t) + Thread::join(t); +} + +bool JSONDB::writeRaw(const std::string &n,const std::string &obj) +{ + if (_rawOutput >= 0) { +#ifndef __WINDOWS__ + if (obj.length() > 0) { + Mutex::Lock _l(_rawLock); + //fprintf(stderr,"%s\n",obj.c_str()); + if ((long)write(_rawOutput,obj.data(),obj.length()) == (long)obj.length()) { + if (write(_rawOutput,"\n",1) == 1) + return true; + } + } else return true; +#endif + return false; + } else { + const std::string path(_genPath(n,true)); + if (!path.length()) + return false; + return OSUtils::writeFile(path.c_str(),obj); + } +} + +bool JSONDB::hasNetwork(const uint64_t networkId) const +{ + Mutex::Lock _l(_networks_m); + return (_networks.find(networkId) != _networks.end()); +} + +bool JSONDB::getNetwork(const uint64_t networkId,nlohmann::json &config) const +{ + Mutex::Lock _l(_networks_m); + const std::unordered_map::const_iterator i(_networks.find(networkId)); + if (i == _networks.end()) + return false; + config = nlohmann::json::from_msgpack(i->second.config); + return true; +} + +bool JSONDB::getNetworkSummaryInfo(const uint64_t networkId,NetworkSummaryInfo &ns) const +{ + Mutex::Lock _l(_networks_m); + const std::unordered_map::const_iterator i(_networks.find(networkId)); + if (i == _networks.end()) + return false; + ns = i->second.summaryInfo; + return true; +} + +int JSONDB::getNetworkAndMember(const uint64_t networkId,const uint64_t nodeId,nlohmann::json &networkConfig,nlohmann::json &memberConfig,NetworkSummaryInfo &ns) const +{ + Mutex::Lock _l(_networks_m); + const std::unordered_map::const_iterator i(_networks.find(networkId)); + if (i == _networks.end()) + return 0; + const std::unordered_map< uint64_t,std::vector >::const_iterator j(i->second.members.find(nodeId)); + if (j == i->second.members.end()) + return 1; + networkConfig = nlohmann::json::from_msgpack(i->second.config); + memberConfig = nlohmann::json::from_msgpack(j->second); + ns = i->second.summaryInfo; + return 3; +} + +bool JSONDB::getNetworkMember(const uint64_t networkId,const uint64_t nodeId,nlohmann::json &memberConfig) const +{ + Mutex::Lock _l(_networks_m); + const std::unordered_map::const_iterator i(_networks.find(networkId)); + if (i == _networks.end()) + return false; + const std::unordered_map< uint64_t,std::vector >::const_iterator j(i->second.members.find(nodeId)); + if (j == i->second.members.end()) + return false; + memberConfig = nlohmann::json::from_msgpack(j->second); + return true; +} + +void JSONDB::saveNetwork(const uint64_t networkId,const nlohmann::json &networkConfig) +{ + char n[64]; + OSUtils::ztsnprintf(n,sizeof(n),"network/%.16llx",(unsigned long long)networkId); + writeRaw(n,OSUtils::jsonDump(networkConfig,-1)); + { + Mutex::Lock _l(_networks_m); + _NW &nw = _networks[networkId]; + nw.config = nlohmann::json::to_msgpack(networkConfig); + } + _recomputeSummaryInfo(networkId); +} + +void JSONDB::saveNetworkMember(const uint64_t networkId,const uint64_t nodeId,const nlohmann::json &memberConfig) +{ + char n[256]; + OSUtils::ztsnprintf(n,sizeof(n),"network/%.16llx/member/%.10llx",(unsigned long long)networkId,(unsigned long long)nodeId); + writeRaw(n,OSUtils::jsonDump(memberConfig,-1)); + { + Mutex::Lock _l(_networks_m); + std::vector &m = _networks[networkId].members[nodeId]; + m = nlohmann::json::to_msgpack(memberConfig); + _members[nodeId].insert(networkId); + } + _recomputeSummaryInfo(networkId); +} + +nlohmann::json JSONDB::eraseNetwork(const uint64_t networkId) +{ + if (_rawOutput >= 0) { + // In harnessed mode, DB deletes occur in the Central database and we do + // not need to erase files. + } else { + std::vector memberIds; + { + Mutex::Lock _l(_networks_m); + const std::unordered_map::iterator i(_networks.find(networkId)); + if (i == _networks.end()) + return _EMPTY_JSON; + for(std::unordered_map< uint64_t,std::vector >::iterator m(i->second.members.begin());m!=i->second.members.end();++m) + memberIds.push_back(m->first); + } + for(std::vector::iterator m(memberIds.begin());m!=memberIds.end();++m) + eraseNetworkMember(networkId,*m,false); + + char n[256]; + OSUtils::ztsnprintf(n,sizeof(n),"network/%.16llx",(unsigned long long)networkId); + const std::string path(_genPath(n,false)); + if (path.length()) + OSUtils::rm(path.c_str()); + } + + // This also erases all members from the memory cache + { + Mutex::Lock _l(_networks_m); + std::unordered_map::iterator i(_networks.find(networkId)); + if (i == _networks.end()) + return _EMPTY_JSON; // sanity check, shouldn't happen + nlohmann::json tmp(nlohmann::json::from_msgpack(i->second.config)); + _networks.erase(i); + return tmp; + } +} + +nlohmann::json JSONDB::eraseNetworkMember(const uint64_t networkId,const uint64_t nodeId,bool recomputeSummaryInfo) +{ + if (_rawOutput >= 0) { + // In harnessed mode, DB deletes occur in Central and we do not remove files. + } else { + char n[256]; + OSUtils::ztsnprintf(n,sizeof(n),"network/%.16llx/member/%.10llx",(unsigned long long)networkId,(unsigned long long)nodeId); + const std::string path(_genPath(n,false)); + if (path.length()) + OSUtils::rm(path.c_str()); + } + + { + Mutex::Lock _l(_networks_m); + _members[nodeId].erase(networkId); + std::unordered_map::iterator i(_networks.find(networkId)); + if (i == _networks.end()) + return _EMPTY_JSON; + std::unordered_map< uint64_t,std::vector >::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::threadMain() + throw() +{ +#ifndef __WINDOWS__ + fd_set readfds,nullfds; + char *const readbuf = (_rawInput >= 0) ? (new char[1048576]) : (char *)0; + std::string rawInputBuf; + FD_ZERO(&readfds); + FD_ZERO(&nullfds); + struct timeval tv; +#endif + + std::vector todo; + + while (_summaryThreadRun) { +#ifndef __WINDOWS__ + if (_rawInput < 0) { + Thread::sleep(25); + } else { + // In IPC mode we wait but also select() on STDIN to read database updates + FD_SET(_rawInput,&readfds); + tv.tv_sec = 0; + tv.tv_usec = 25000; + select(_rawInput+1,&readfds,&nullfds,&nullfds,&tv); + if (FD_ISSET(_rawInput,&readfds)) { + const long rn = (long)read(_rawInput,readbuf,1048576); + bool gotMessage = false; + for(long i=0;i 0) { + try { + const nlohmann::json obj(OSUtils::jsonParse(rawInputBuf)); + gotMessage = true; + + if (!_dataReady) { + _dataReady = true; + _networks_m.unlock(); + } + + if (obj.is_array()) { + for(unsigned long i=0;i::iterator ii(todo.begin());ii!=todo.end();++ii) { + const uint64_t networkId = *ii; + std::unordered_map::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,std::vector >::const_iterator m(n->second.members.begin());m!=n->second.members.end();++m) { + try { + nlohmann::json member(nlohmann::json::from_msgpack(m->second)); + + if (OSUtils::jsonBool(member["authorized"],false)) { + ++ns.authorizedMemberCount; + + try { + const nlohmann::json &mlog = member["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(member["activeBridge"],false)) + ns.activeBridges.push_back(Address(m->first)); + } catch ( ... ) {} + + try { + const nlohmann::json &mips = member["ipAssignments"]; + if (mips.is_array()) { + for(unsigned long i=0;isecond.summaryInfoLastComputed = now; + } + } + } catch ( ... ) {} + + todo.clear(); + } + + if (!_dataReady) // sanity check + _networks_m.unlock(); + +#ifndef __WINDOWS__ + delete [] readbuf; +#endif +} + +bool JSONDB::_addOrUpdate(const nlohmann::json &j) +{ + try { + if (j.is_object()) { + std::string id(OSUtils::jsonString(j["id"],"0")); + const std::string objtype(OSUtils::jsonString(j["objtype"],"")); + if ((id.length() == 16)&&(objtype == "network")) { + + const uint64_t nwid = Utils::hexStrToU64(id.c_str()); + if (nwid) { + bool update; + { + Mutex::Lock _l(_networks_m); + _NW &nw = _networks[nwid]; + update = !nw.config.empty(); + nw.config = nlohmann::json::to_msgpack(j); + } + if (update) + _parent->onNetworkUpdate(nwid); + _recomputeSummaryInfo(nwid); + return true; + } + + } 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)) { + bool update = false; + bool deauth = false; + { + Mutex::Lock _l(_networks_m); + std::vector &m = _networks[nwid].members[mid]; + if (!m.empty()) { + update = true; + nlohmann::json oldm(nlohmann::json::from_msgpack(m)); + deauth = ((OSUtils::jsonBool(oldm["authorized"],false))&&(!OSUtils::jsonBool(j["authorized"],false))); + } + m = nlohmann::json::to_msgpack(j); + _members[mid].insert(nwid); + } + if (update) { + _parent->onNetworkMemberUpdate(nwid,mid); + if (deauth) + _parent->onNetworkMemberDeauthorize(nwid,mid); + } + _recomputeSummaryInfo(nwid); + return true; + } + + } else if (objtype == "_delete") { // pseudo-object-type, only used in Central harnessed mode + + const std::string deleteType(OSUtils::jsonString(j["deleteType"],"")); + id = OSUtils::jsonString(j["deleteId"],""); + if ((deleteType == "network")&&(id.length() == 16)) { + eraseNetwork(Utils::hexStrToU64(id.c_str())); + } else if ((deleteType == "member")&&(id.length() == 10)) { + const std::string networkId(OSUtils::jsonString(j["deleteNetworkId"],"")); + const uint64_t nwid = Utils::hexStrToU64(networkId.c_str()); + const uint64_t mid = Utils::hexStrToU64(id.c_str()); + if (networkId.length() == 16) + eraseNetworkMember(nwid,mid,true); + _parent->onNetworkMemberDeauthorize(nwid,mid); + } + + } + } + } catch ( ... ) {} + return false; +} + +bool JSONDB::_load(const std::string &p) +{ + // This is not used in stdin/stdout mode. Instead data is populated by + // sending it all to stdin. + + std::vector dl(OSUtils::listDirectory(p.c_str(),true)); + for(std::vector::const_iterator di(dl.begin());di!=dl.end();++di) { + if ((di->length() > 5)&&(di->substr(di->length() - 5) == ".json")) { + std::string buf; + if (OSUtils::readFile((p + ZT_PATH_SEPARATOR_S + *di).c_str(),buf)) { + try { + _addOrUpdate(OSUtils::jsonParse(buf)); + } catch ( ... ) {} + } + } else { + this->_load((p + ZT_PATH_SEPARATOR_S + *di)); + } + } + + return true; +} + +void JSONDB::_recomputeSummaryInfo(const uint64_t networkId) +{ + Mutex::Lock _l(_summaryThread_m); + if (std::find(_summaryThreadToDo.begin(),_summaryThreadToDo.end(),networkId) == _summaryThreadToDo.end()) + _summaryThreadToDo.push_back(networkId); + if (!_summaryThread) + _summaryThread = Thread::start(this); +} + +std::string JSONDB::_genPath(const std::string &n,bool create) +{ + std::vector pt(OSUtils::split(n.c_str(),"/","","")); + if (pt.size() == 0) + return std::string(); + + std::string p(_basePath); + if (create) OSUtils::mkdir(p.c_str()); + for(unsigned long i=0,j=(unsigned long)(pt.size()-1);i. + */ + +#ifndef ZT_JSONDB_HPP +#define ZT_JSONDB_HPP + +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include + +#include "../node/Constants.hpp" +#include "../node/Utils.hpp" +#include "../node/InetAddress.hpp" +#include "../node/Mutex.hpp" +#include "../ext/json/json.hpp" +#include "../osdep/OSUtils.hpp" +#include "../osdep/Thread.hpp" + +namespace ZeroTier { + +class EmbeddedNetworkController; + +/** + * Hierarchical JSON store that persists into the filesystem or via HTTP + */ +class JSONDB +{ +public: + struct NetworkSummaryInfo + { + NetworkSummaryInfo() : authorizedMemberCount(0),activeMemberCount(0),totalMemberCount(0),mostRecentDeauthTime(0) {} + std::vector
activeBridges; + std::vector allocatedIps; + unsigned long authorizedMemberCount; + unsigned long activeMemberCount; + unsigned long totalMemberCount; + int64_t mostRecentDeauthTime; + }; + + JSONDB(const std::string &basePath,EmbeddedNetworkController *parent); + ~JSONDB(); + + /** + * Write a JSON object to the data store + * + * It's important that obj contain a valid JSON object with no newlines (jsonDump with -1 + * for indentation), since newline-delimited JSON is what nodeJS's IPC speaks and this + * is important in Central-harnessed mode. + * + * @param n Path name of object + * @param obj Object in single-line no-CRs JSON object format (OSUtils::jsonDump(obj,-1)) + * @return True if write appears successful + */ + bool writeRaw(const std::string &n,const std::string &obj); + + bool hasNetwork(const uint64_t networkId) const; + + bool getNetwork(const uint64_t networkId,nlohmann::json &config) const; + + bool getNetworkSummaryInfo(const uint64_t networkId,NetworkSummaryInfo &ns) const; + + /** + * @return Bit mask: 0 == none, 1 == network only, 3 == network and member + */ + int getNetworkAndMember(const uint64_t networkId,const uint64_t nodeId,nlohmann::json &networkConfig,nlohmann::json &memberConfig,NetworkSummaryInfo &ns) const; + + bool getNetworkMember(const uint64_t networkId,const uint64_t nodeId,nlohmann::json &memberConfig) const; + + void saveNetwork(const uint64_t networkId,const nlohmann::json &networkConfig); + + void saveNetworkMember(const uint64_t networkId,const uint64_t nodeId,const nlohmann::json &memberConfig); + + nlohmann::json eraseNetwork(const uint64_t networkId); + + nlohmann::json eraseNetworkMember(const uint64_t networkId,const uint64_t nodeId,bool recomputeSummaryInfo = true); + + std::vector networkIds() const + { + std::vector r; + Mutex::Lock _l(_networks_m); + for(std::unordered_map::const_iterator n(_networks.begin());n!=_networks.end();++n) + r.push_back(n->first); + return r; + } + + inline unsigned long memberCount(const uint64_t networkId) + { + Mutex::Lock _l(_networks_m); + std::unordered_map::const_iterator i(_networks.find(networkId)); + if (i != _networks.end()) + return (unsigned long)i->second.members.size(); + return 0; + } + + template + inline void eachMember(const uint64_t networkId,F func) + { + Mutex::Lock _l(_networks_m); + std::unordered_map::const_iterator i(_networks.find(networkId)); + if (i != _networks.end()) { + for(std::unordered_map< uint64_t,std::vector >::const_iterator m(i->second.members.begin());m!=i->second.members.end();++m) { + try { + func(networkId,m->first,nlohmann::json::from_msgpack(m->second)); + } catch ( ... ) {} + } + } + } + + template + inline void eachId(F func) + { + Mutex::Lock _l(_networks_m); + for(std::unordered_map::const_iterator i(_networks.begin());i!=_networks.end();++i) { + for(std::unordered_map< uint64_t,std::vector >::const_iterator m(i->second.members.begin());m!=i->second.members.end();++m) { + try { + func(i->first,m->first); + } catch ( ... ) {} + } + } + } + + inline std::vector networksForMember(const uint64_t nodeId) + { + Mutex::Lock _l(_networks_m); + std::unordered_map< uint64_t,std::unordered_set< uint64_t > >::const_iterator m(_members.find(nodeId)); + if (m != _members.end()) { + return std::vector(m->second.begin(),m->second.end()); + } else { + return std::vector(); + } + } + + void threadMain() + throw(); + +private: + bool _addOrUpdate(const nlohmann::json &j); + bool _load(const std::string &p); + void _recomputeSummaryInfo(const uint64_t networkId); + std::string _genPath(const std::string &n,bool create); + + EmbeddedNetworkController *const _parent; + std::string _basePath; + int _rawInput,_rawOutput; + Mutex _rawLock; + + Thread _summaryThread; + std::vector _summaryThreadToDo; + volatile bool _summaryThreadRun; + Mutex _summaryThread_m; + + struct _NW + { + _NW() : summaryInfoLastComputed(0) {} + std::vector config; + NetworkSummaryInfo summaryInfo; + uint64_t summaryInfoLastComputed; + std::unordered_map< uint64_t,std::vector > members; + }; + + std::unordered_map< uint64_t,_NW > _networks; + std::unordered_map< uint64_t,std::unordered_set< uint64_t > > _members; + bool _dataReady; + Mutex _networks_m; +}; + +} // namespace ZeroTier + +#endif -- cgit v1.2.3