RethinkDB.cpp 17 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480
  1. /*
  2. * ZeroTier One - Network Virtualization Everywhere
  3. * Copyright (C) 2011-2015 ZeroTier, Inc.
  4. *
  5. * This program is free software: you can redistribute it and/or modify
  6. * it under the terms of the GNU General Public License as published by
  7. * the Free Software Foundation, either version 3 of the License, or
  8. * (at your option) any later version.
  9. *
  10. * This program is distributed in the hope that it will be useful,
  11. * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12. * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  13. * GNU General Public License for more details.
  14. *
  15. * You should have received a copy of the GNU General Public License
  16. * along with this program. If not, see <http://www.gnu.org/licenses/>.
  17. */
  18. #ifdef ZT_CONTROLLER_USE_RETHINKDB
  19. #include <stdio.h>
  20. #include <stdlib.h>
  21. #include <unistd.h>
  22. #include <time.h>
  23. #include "RethinkDB.hpp"
  24. #include "EmbeddedNetworkController.hpp"
  25. #include "../version.h"
  26. #include <chrono>
  27. #include <algorithm>
  28. #include <stdexcept>
  29. #include "../ext/librethinkdbxx/build/include/rethinkdb.h"
  30. namespace R = RethinkDB;
  31. using json = nlohmann::json;
  32. namespace ZeroTier {
  33. static const char *_timestr()
  34. {
  35. time_t t = time(0);
  36. char *ts = ctime(&t);
  37. if (!ts)
  38. return "error";
  39. char *p = ts;
  40. while (*p) {
  41. if (*p == '\n') {
  42. *p = (char)0;
  43. break;
  44. }
  45. ++p;
  46. }
  47. return ts;
  48. }
  49. RethinkDB::RethinkDB(EmbeddedNetworkController *const nc,const Identity &myId,const char *path) :
  50. DB(nc,myId,path),
  51. _ready(2), // two tables need to be synchronized before we're ready, so this is ready when it reaches 0
  52. _run(1),
  53. _waitNoticePrinted(false)
  54. {
  55. // rethinkdb:host:port:db[:auth]
  56. std::vector<std::string> ps(OSUtils::split(path,":","",""));
  57. if ((ps.size() < 4)||(ps[0] != "rethinkdb"))
  58. throw std::runtime_error("invalid rethinkdb database url");
  59. _host = ps[1];
  60. _port = Utils::strToInt(ps[2].c_str());
  61. _db = ps[3];
  62. if (ps.size() > 4)
  63. _auth = ps[4];
  64. _readyLock.lock();
  65. _membersDbWatcher = std::thread([this]() {
  66. try {
  67. while (_run == 1) {
  68. try {
  69. std::unique_ptr<R::Connection> rdb(R::connect(this->_host,this->_port,this->_auth));
  70. if (rdb) {
  71. _membersDbWatcherConnection = (void *)rdb.get();
  72. auto cur = R::db(this->_db).table("Member",R::optargs("read_mode","outdated")).get_all(this->_myAddressStr,R::optargs("index","controllerId")).changes(R::optargs("squash",0.05,"include_initial",true,"include_types",true,"include_states",true)).run(*rdb);
  73. while (cur.has_next()) {
  74. if (_run != 1) break;
  75. json tmp(json::parse(cur.next().as_json()));
  76. if ((tmp["type"] == "state")&&(tmp["state"] == "ready")) {
  77. if (--this->_ready == 0) {
  78. if (_waitNoticePrinted)
  79. fprintf(stderr,"[%s] NOTICE: %.10llx controller RethinkDB data download complete." ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt());
  80. this->_readyLock.unlock();
  81. }
  82. } else {
  83. try {
  84. json &ov = tmp["old_val"];
  85. json &nv = tmp["new_val"];
  86. if (ov.is_object()||nv.is_object()) {
  87. //if (nv.is_object()) printf("MEMBER: %s" ZT_EOL_S,nv.dump().c_str());
  88. this->_memberChanged(ov,nv,(this->_ready <= 0));
  89. }
  90. } catch ( ... ) {} // ignore bad records
  91. }
  92. }
  93. }
  94. } catch (std::exception &e) {
  95. fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (member change stream): %s" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt(),e.what());
  96. } catch (R::Error &e) {
  97. fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (member change stream): %s" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt(),e.message.c_str());
  98. } catch ( ... ) {
  99. fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (member change stream): unknown exception" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt());
  100. }
  101. std::this_thread::sleep_for(std::chrono::milliseconds(250));
  102. }
  103. } catch ( ... ) {}
  104. });
  105. _networksDbWatcher = std::thread([this]() {
  106. try {
  107. while (_run == 1) {
  108. try {
  109. std::unique_ptr<R::Connection> rdb(R::connect(this->_host,this->_port,this->_auth));
  110. if (rdb) {
  111. _networksDbWatcherConnection = (void *)rdb.get();
  112. auto cur = R::db(this->_db).table("Network",R::optargs("read_mode","outdated")).get_all(this->_myAddressStr,R::optargs("index","controllerId")).changes(R::optargs("squash",0.05,"include_initial",true,"include_types",true,"include_states",true)).run(*rdb);
  113. while (cur.has_next()) {
  114. if (_run != 1) break;
  115. json tmp(json::parse(cur.next().as_json()));
  116. if ((tmp["type"] == "state")&&(tmp["state"] == "ready")) {
  117. if (--this->_ready == 0) {
  118. if (_waitNoticePrinted)
  119. fprintf(stderr,"[%s] NOTICE: %.10llx controller RethinkDB data download complete." ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt());
  120. this->_readyLock.unlock();
  121. }
  122. } else {
  123. try {
  124. json &ov = tmp["old_val"];
  125. json &nv = tmp["new_val"];
  126. if (ov.is_object()||nv.is_object()) {
  127. //if (nv.is_object()) printf("NETWORK: %s" ZT_EOL_S,nv.dump().c_str());
  128. this->_networkChanged(ov,nv,(this->_ready <= 0));
  129. }
  130. } catch ( ... ) {} // ignore bad records
  131. }
  132. }
  133. }
  134. } catch (std::exception &e) {
  135. fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (network change stream): %s" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt(),e.what());
  136. } catch (R::Error &e) {
  137. fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (network change stream): %s" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt(),e.message.c_str());
  138. } catch ( ... ) {
  139. fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (network change stream): unknown exception" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt());
  140. }
  141. std::this_thread::sleep_for(std::chrono::milliseconds(250));
  142. }
  143. } catch ( ... ) {}
  144. });
  145. for(int t=0;t<ZT_CONTROLLER_RETHINKDB_COMMIT_THREADS;++t) {
  146. _commitThread[t] = std::thread([this]() {
  147. try {
  148. std::unique_ptr<R::Connection> rdb;
  149. nlohmann::json *config = (nlohmann::json *)0;
  150. while ((this->_commitQueue.get(config))&&(_run == 1)) {
  151. if (!config)
  152. continue;
  153. json record;
  154. const std::string objtype = (*config)["objtype"];
  155. const char *table;
  156. std::string deleteId;
  157. try {
  158. if (objtype == "member") {
  159. const std::string nwid = (*config)["nwid"];
  160. const std::string id = (*config)["id"];
  161. record["id"] = nwid + "-" + id;
  162. record["controllerId"] = this->_myAddressStr;
  163. record["networkId"] = nwid;
  164. record["nodeId"] = id;
  165. record["config"] = *config;
  166. table = "Member";
  167. } else if (objtype == "network") {
  168. const std::string id = (*config)["id"];
  169. record["id"] = id;
  170. record["controllerId"] = this->_myAddressStr;
  171. record["config"] = *config;
  172. table = "Network";
  173. } else if (objtype == "trace") {
  174. record = *config;
  175. table = "RemoteTrace";
  176. } else if (objtype == "_delete_network") {
  177. deleteId = (*config)["id"];
  178. table = "Network";
  179. } else if (objtype == "_delete_member") {
  180. deleteId = (*config)["nwid"];
  181. deleteId.push_back('-');
  182. const std::string tmp = (*config)["id"];
  183. deleteId.append(tmp);
  184. table = "Member";
  185. } else {
  186. delete config;
  187. continue;
  188. }
  189. delete config;
  190. } catch ( ... ) {
  191. delete config;
  192. continue;
  193. }
  194. while (_run == 1) {
  195. try {
  196. if (!rdb)
  197. rdb = R::connect(this->_host,this->_port,this->_auth);
  198. if (rdb) {
  199. if (deleteId.length() > 0) {
  200. //printf("DELETE: %s" ZT_EOL_S,deleteId.c_str());
  201. R::db(this->_db).table(table).get(deleteId).delete_().run(*rdb);
  202. } else {
  203. //printf("UPSERT: %s" ZT_EOL_S,record.dump().c_str());
  204. R::db(this->_db).table(table).insert(R::Datum::from_json(record.dump()),R::optargs("conflict","update","return_changes",false)).run(*rdb);
  205. }
  206. break;
  207. } else {
  208. fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (insert/update): connect failed (will retry)" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt());
  209. }
  210. } catch (std::exception &e) {
  211. fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (insert/update): %s" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt(),e.what());
  212. rdb.reset();
  213. } catch (R::Error &e) {
  214. fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (insert/update): %s" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt(),e.message.c_str());
  215. rdb.reset();
  216. } catch ( ... ) {
  217. fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (insert/update): unknown exception" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt());
  218. rdb.reset();
  219. }
  220. std::this_thread::sleep_for(std::chrono::milliseconds(250));
  221. }
  222. }
  223. } catch ( ... ) {}
  224. });
  225. }
  226. _onlineNotificationThread = std::thread([this]() {
  227. int64_t lastUpdatedNetworkStatus = 0;
  228. std::unordered_map< std::pair<uint64_t,uint64_t>,int64_t,_PairHasher > lastOnlineCumulative;
  229. try {
  230. std::unique_ptr<R::Connection> rdb;
  231. while (_run == 1) {
  232. try {
  233. if (!rdb)
  234. rdb = R::connect(this->_host,this->_port,this->_auth);
  235. if (rdb) {
  236. R::Array batch;
  237. R::Object tmpobj;
  238. std::unordered_map< std::pair<uint64_t,uint64_t>,std::pair<int64_t,InetAddress>,_PairHasher > lastOnline;
  239. {
  240. std::lock_guard<std::mutex> l(_lastOnline_l);
  241. lastOnline.swap(_lastOnline);
  242. }
  243. for(auto i=lastOnline.begin();i!=lastOnline.end();++i) {
  244. lastOnlineCumulative[i->first] = i->second.first;
  245. char tmp[64],tmp2[64];
  246. OSUtils::ztsnprintf(tmp,sizeof(tmp),"%.16llx-%.10llx",i->first.first,i->first.second);
  247. tmpobj["id"] = tmp;
  248. tmpobj["ts"] = i->second.first;
  249. tmpobj["phy"] = i->second.second.toIpString(tmp2);
  250. batch.emplace_back(tmpobj);
  251. if (batch.size() >= 1024) {
  252. R::db(this->_db).table("MemberStatus",R::optargs("read_mode","outdated")).insert(batch,R::optargs("conflict","update")).run(*rdb);
  253. batch.clear();
  254. }
  255. }
  256. if (batch.size() > 0) {
  257. R::db(this->_db).table("MemberStatus",R::optargs("read_mode","outdated")).insert(batch,R::optargs("conflict","update")).run(*rdb);
  258. batch.clear();
  259. }
  260. tmpobj.clear();
  261. const int64_t now = OSUtils::now();
  262. if ((now - lastUpdatedNetworkStatus) > 10000) {
  263. lastUpdatedNetworkStatus = now;
  264. std::vector< std::pair< uint64_t,std::shared_ptr<_Network> > > networks;
  265. {
  266. std::lock_guard<std::mutex> l(_networks_l);
  267. networks.reserve(_networks.size() + 1);
  268. for(auto i=_networks.begin();i!=_networks.end();++i)
  269. networks.push_back(*i);
  270. }
  271. for(auto i=networks.begin();i!=networks.end();++i) {
  272. char tmp[64];
  273. Utils::hex(i->first,tmp);
  274. tmpobj["id"] = tmp;
  275. {
  276. std::lock_guard<std::mutex> l2(i->second->lock);
  277. tmpobj["authorizedMemberCount"] = i->second->authorizedMembers.size();
  278. tmpobj["totalMemberCount"] = i->second->members.size();
  279. unsigned long onlineMemberCount = 0;
  280. for(auto m=i->second->members.begin();m!=i->second->members.end();++m) {
  281. auto lo = lastOnlineCumulative.find(std::pair<uint64_t,uint64_t>(i->first,m->first));
  282. if (lo != lastOnlineCumulative.end()) {
  283. if ((now - lo->second) <= (ZT_NETWORK_AUTOCONF_DELAY * 2))
  284. ++onlineMemberCount;
  285. else lastOnlineCumulative.erase(lo);
  286. }
  287. }
  288. tmpobj["onlineMemberCount"] = onlineMemberCount;
  289. tmpobj["bridgeCount"] = i->second->activeBridgeMembers.size();
  290. tmpobj["ts"] = now;
  291. }
  292. batch.emplace_back(tmpobj);
  293. if (batch.size() >= 1024) {
  294. R::db(this->_db).table("NetworkStatus",R::optargs("read_mode","outdated")).insert(batch,R::optargs("conflict","update")).run(*rdb);
  295. batch.clear();
  296. }
  297. }
  298. if (batch.size() > 0) {
  299. R::db(this->_db).table("NetworkStatus",R::optargs("read_mode","outdated")).insert(batch,R::optargs("conflict","update")).run(*rdb);
  300. batch.clear();
  301. }
  302. }
  303. }
  304. } catch (std::exception &e) {
  305. fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (node status update): %s" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt(),e.what());
  306. rdb.reset();
  307. } catch (R::Error &e) {
  308. fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (node status update): %s" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt(),e.message.c_str());
  309. rdb.reset();
  310. } catch ( ... ) {
  311. fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (node status update): unknown exception" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt());
  312. rdb.reset();
  313. }
  314. std::this_thread::sleep_for(std::chrono::milliseconds(250));
  315. }
  316. } catch ( ... ) {}
  317. });
  318. _heartbeatThread = std::thread([this]() {
  319. try {
  320. R::Object controllerRecord;
  321. std::unique_ptr<R::Connection> rdb;
  322. {
  323. char publicId[1024];
  324. //char secretId[1024];
  325. char hostname[1024];
  326. this->_myId.toString(false,publicId);
  327. //this->_myId.toString(true,secretId);
  328. if (gethostname(hostname,sizeof(hostname)) != 0) {
  329. hostname[0] = (char)0;
  330. } else {
  331. for(int i=0;i<sizeof(hostname);++i) {
  332. if ((hostname[i] == '.')||(hostname[i] == 0)) {
  333. hostname[i] = (char)0;
  334. break;
  335. }
  336. }
  337. }
  338. controllerRecord["id"] = this->_myAddressStr.c_str();
  339. controllerRecord["publicIdentity"] = publicId;
  340. //controllerRecord["secretIdentity"] = secretId;
  341. if (hostname[0])
  342. controllerRecord["clusterHost"] = hostname;
  343. controllerRecord["vMajor"] = ZEROTIER_ONE_VERSION_MAJOR;
  344. controllerRecord["vMinor"] = ZEROTIER_ONE_VERSION_MINOR;
  345. controllerRecord["vRev"] = ZEROTIER_ONE_VERSION_REVISION;
  346. controllerRecord["vBuild"] = ZEROTIER_ONE_VERSION_BUILD;
  347. }
  348. while (_run == 1) {
  349. try {
  350. if (!rdb)
  351. rdb = R::connect(this->_host,this->_port,this->_auth);
  352. if (rdb) {
  353. controllerRecord["lastAlive"] = OSUtils::now();
  354. //printf("HEARTBEAT: %s" ZT_EOL_S,tmp);
  355. R::db(this->_db).table("Controller",R::optargs("read_mode","outdated")).insert(controllerRecord,R::optargs("conflict","update")).run(*rdb);
  356. }
  357. } catch ( ... ) {
  358. rdb.reset();
  359. }
  360. std::this_thread::sleep_for(std::chrono::milliseconds(1000));
  361. }
  362. } catch ( ... ) {}
  363. });
  364. }
  365. RethinkDB::~RethinkDB()
  366. {
  367. _run = 0;
  368. std::this_thread::sleep_for(std::chrono::milliseconds(100));
  369. _commitQueue.stop();
  370. for(int t=0;t<ZT_CONTROLLER_RETHINKDB_COMMIT_THREADS;++t)
  371. _commitThread[t].join();
  372. if (_membersDbWatcherConnection)
  373. ((R::Connection *)_membersDbWatcherConnection)->close();
  374. if (_networksDbWatcherConnection)
  375. ((R::Connection *)_networksDbWatcherConnection)->close();
  376. _membersDbWatcher.join();
  377. _networksDbWatcher.join();
  378. _heartbeatThread.join();
  379. _onlineNotificationThread.join();
  380. }
  381. bool RethinkDB::waitForReady()
  382. {
  383. while (_ready > 0) {
  384. if (!_waitNoticePrinted) {
  385. _waitNoticePrinted = true;
  386. fprintf(stderr,"[%s] NOTICE: %.10llx controller RethinkDB waiting for initial data download..." ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt());
  387. }
  388. _readyLock.lock();
  389. _readyLock.unlock();
  390. }
  391. return true;
  392. }
  393. void RethinkDB::save(nlohmann::json *orig,nlohmann::json &record)
  394. {
  395. if (!record.is_object()) // sanity check
  396. return;
  397. waitForReady();
  398. if (orig) {
  399. if (*orig != record) {
  400. nlohmann::json *q = new nlohmann::json();
  401. try {
  402. record["revision"] = OSUtils::jsonInt(record["revision"],0ULL) + 1;
  403. for(auto kv=record.begin();kv!=record.end();++kv) {
  404. if ((kv.key() == "id")||(kv.key() == "nwid")||(kv.key() == "objtype")||((*q)[kv.key()] != kv.value()))
  405. (*q)[kv.key()] = kv.value();
  406. }
  407. } catch ( ... ) {
  408. delete q;
  409. throw;
  410. }
  411. }
  412. } else {
  413. record["revision"] = 1;
  414. _commitQueue.post(new nlohmann::json(record));
  415. }
  416. }
  417. void RethinkDB::eraseNetwork(const uint64_t networkId)
  418. {
  419. char tmp2[24];
  420. waitForReady();
  421. Utils::hex(networkId,tmp2);
  422. json *tmp = new json();
  423. (*tmp)["id"] = tmp2;
  424. (*tmp)["objtype"] = "_delete_network"; // pseudo-type, tells thread to delete network
  425. _commitQueue.post(tmp);
  426. }
  427. void RethinkDB::eraseMember(const uint64_t networkId,const uint64_t memberId)
  428. {
  429. char tmp2[24];
  430. json *tmp = new json();
  431. waitForReady();
  432. Utils::hex(networkId,tmp2);
  433. (*tmp)["nwid"] = tmp2;
  434. Utils::hex10(memberId,tmp2);
  435. (*tmp)["id"] = tmp2;
  436. (*tmp)["objtype"] = "_delete_member"; // pseudo-type, tells thread to delete network
  437. _commitQueue.post(tmp);
  438. }
  439. void RethinkDB::nodeIsOnline(const uint64_t networkId,const uint64_t memberId,const InetAddress &physicalAddress)
  440. {
  441. std::lock_guard<std::mutex> l(_lastOnline_l);
  442. std::pair<int64_t,InetAddress> &i = _lastOnline[std::pair<uint64_t,uint64_t>(networkId,memberId)];
  443. i.first = OSUtils::now();
  444. if (physicalAddress)
  445. i.second = physicalAddress;
  446. }
  447. } // namespace ZeroTier
  448. #endif // ZT_CONTROLLER_USE_RETHINKDB