/* * ZeroTier One - Network Virtualization Everywhere * Copyright (C) 2011-2018 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 . */ #ifdef ZT_CONTROLLER_USE_LIBPQ #include "PostgreSQL.hpp" #include "EmbeddedNetworkController.hpp" #include "../version.h" #include using json = nlohmann::json; namespace { static const char *_timestr() { time_t t = time(0); char *ts = ctime(&t); char *p = ts; if (!p) return ""; while (*p) { if (*p == '\n') { *p = (char)0; break; } ++p; } return ts; } } using namespace ZeroTier; PostgreSQL::PostgreSQL(EmbeddedNetworkController *const nc, const Identity &myId, const char *path) : DB(nc, myId, path) , _ready(0) , _connected(1) , _run(1) , _waitNoticePrinted(false) { _connString = std::string(path); _readyLock.lock(); _heartbeatThread = std::thread(&PostgreSQL::heartbeat, this); _membersDbWatcher = std::thread(&PostgreSQL::membersDbWatcher, this); _networksDbWatcher = std::thread(&PostgreSQL::networksDbWatcher, this); for (int i = 0; i < ZT_CONTROLLER_RETHINKDB_COMMIT_THREADS; ++i) { _commitThread[i] = std::thread(&PostgreSQL::commitThread, this); } _onlineNotificationThread = std::thread(&PostgreSQL::onlineNotificationThread, this); } PostgreSQL::~PostgreSQL() { _run = 0; std::this_thread::sleep_for(std::chrono::milliseconds(100)); _heartbeatThread.join(); _membersDbWatcher.join(); _networksDbWatcher.join(); for (int i = 0; i < ZT_CONTROLLER_RETHINKDB_COMMIT_THREADS; ++i) { _commitThread[i].join(); } _onlineNotificationThread.join(); } bool PostgreSQL::waitForReady() { while (_ready < 2) { if (!_waitNoticePrinted) { _waitNoticePrinted = true; fprintf(stderr, "[%s] NOTICE: %.10llx controller PostgreSQL waiting for initial data download..." ZT_EOL_S, ::_timestr(), (unsigned long long)_myAddress.toInt()); } _readyLock.lock(); _readyLock.unlock(); } return true; } bool PostgreSQL::isReady() { return ((_ready == 2)&&(_connected)); } void PostgreSQL::save(nlohmann::json *orig, nlohmann::json &record) { if (!record.is_object()) { return; } waitForReady(); if (orig) { if (*orig != record) { record["revision"] = OSUtils::jsonInt(record["revision"],0ULL) + 1; _commitQueue.post(new nlohmann::json(record)); } } else { record["revision"] = 1; _commitQueue.post(new nlohmann::json(record)); } } void PostgreSQL::eraseNetwork(const uint64_t networkId) { char tmp2[24]; waitForReady(); Utils::hex(networkId, tmp2); json *tmp = new json(); (*tmp)["id"] = tmp2; (*tmp)["objtype"] = "_delete_network"; _commitQueue.post(tmp); } void PostgreSQL::eraseMember(const uint64_t networkId, const uint64_t memberId) { char tmp2[24]; json *tmp = new json(); Utils::hex(networkId, tmp2); (*tmp)["nwid"] = tmp2; Utils::hex(memberId, tmp2); (*tmp)["id"] = tmp2; (*tmp)["objtype"] = "_delete_member"; _commitQueue.post(tmp); } void PostgreSQL::nodeIsOnline(const uint64_t networkId, const uint64_t memberId, const InetAddress &physicalAddress) { std::lock_guard l(_lastOnline_l); std::pair &i = _lastOnline[std::pair(networkId, memberId)]; i.first = OSUtils::now(); if (physicalAddress) { i.second = physicalAddress; } } void PostgreSQL::initializeNetworks(PGconn *conn) { try { if (PQstatus(conn) != CONNECTION_OK) { fprintf(stderr, "Bad Database Connection: %s", PQerrorMessage(conn)); exit(1); } const char *params[1] = { _myAddressStr.c_str() }; PGresult *res = PQexecParams(conn, "SELECT id, EXTRACT(EPOCH FROM creation_time AT TIME ZONE 'UTC')*1000, capabilities, " "enable_broadcast, EXTRACT(EPOCH FROM last_modified AT TIME ZONE 'UTC')*1000, mtu, multicast_limit, name, private, remote_trace_level, " "remote_trace_target, revision, rules, tags, v4_assign_mode, v6_assign_mode FROM ztc_network " "WHERE deleted = false AND controller_id = $1", 1, NULL, params, NULL, NULL, 0); if (PQresultStatus(res) != PGRES_TUPLES_OK) { fprintf(stderr, "Networks Initialization Failed: %s", PQerrorMessage(conn)); PQclear(res); exit(1); } int numRows = PQntuples(res); for (int i = 0; i < numRows; ++i) { json empty; json config; config["nwid"] = PQgetvalue(res, i, 0); config["creationTime"] = std::stoull(PQgetvalue(res, i, 1)); config["capabilities"] = json::parse(PQgetvalue(res, i, 2)); config["enableBroadcast"] = (strcmp(PQgetvalue(res, i, 3),"true")==0); config["lastModified"] = std::stoull(PQgetvalue(res, i, 4)); config["mtu"] = std::stoi(PQgetvalue(res, i, 5)); config["multicastLimit"] = std::stoi(PQgetvalue(res, i, 6)); config["name"] = PQgetvalue(res, i, 7); config["private"] = (strcmp(PQgetvalue(res, i, 8),"true")==0); config["remoteTraceLevel"] = std::stoi(PQgetvalue(res, i, 9)); config["remoteTraceTarget"] = PQgetvalue(res, i, 10); config["revision"] = std::stoull(PQgetvalue(res, i, 11)); config["rules"] = json::parse(PQgetvalue(res, i, 12)); config["tags"] = json::parse(PQgetvalue(res, i, 13)); config["v4AssignMode"] = json::parse(PQgetvalue(res, i, 14)); config["v6AssignMode"] = json::parse(PQgetvalue(res, i, 15)); config["objtype"] = "network"; config["ipAssignmentPools"] = json::array(); config["routes"] = json::array(); PGresult *r2 = PQexecParams(conn, "SELECT host(ip_range_start), host(ip_range_end) FROM ztc_network_assignment_pool WHERE network_id = $1", 1, NULL, params, NULL, NULL, 0); if (PQresultStatus(r2) != PGRES_TUPLES_OK) { fprintf(stderr, "ERROR: Error retreiving IP pools for network: %s\n", PQresultErrorMessage(r2)); PQclear(r2); PQclear(res); exit(1); } int n = PQntuples(r2); for (int j = 0; j < n; ++j) { json ip; ip["ipRangeStart"] = PQgetvalue(r2, j, 0); ip["ipRangeEnd"] = PQgetvalue(r2, j, 1); config["ipAssignmentPools"].push_back(ip); } PQclear(r2); r2 = PQexecParams(conn, "SELECT host(address), bits, host(via) FROM ztc_network_route WHERE network_id = $1", 1, NULL, params, NULL, NULL, 0); if (PQresultStatus(r2) != PGRES_TUPLES_OK) { fprintf(stderr, "ERROR: Error retreiving routes for network: %s\n", PQresultErrorMessage(r2)); PQclear(r2); PQclear(res); exit(1); } n = PQntuples(r2); for (int j = 0; j < n; ++j) { std::string addr = PQgetvalue(r2, j, 0); std::string bits = PQgetvalue(r2, j, 1); std::string via = PQgetvalue(r2, j, 2); json route; route["target"] = addr + "/" + bits; if (via == "NULL") { route["via"] = nullptr; } else { route["via"] = via; } config["routes"].push_back(route); } PQclear(r2); _networkChanged(empty, config, false); } PQclear(res); if (++this->_ready == 2) { if (_waitNoticePrinted) { fprintf(stderr,"[%s] NOTICE: %.10llx controller PostgreSQL data download complete." ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt()); } _readyLock.unlock(); } } catch (std::exception &e) { fprintf(stderr, "ERROR: Error initializing networks: %s", e.what()); exit(-1); } } void PostgreSQL::initializeMembers(PGconn *conn) { try { if (PQstatus(conn) != CONNECTION_OK) { fprintf(stderr, "Bad Database Connection: %s", PQerrorMessage(conn)); exit(1); } const char *params[1] = { _myAddressStr.c_str() }; PGresult *res = PQexecParams(conn, "SELECT m.id, m.network_id, m.active_bridge, m.authorized, m.capabilities, EXTRACT(EPOCH FROM m.creation_time AT TIME ZONE 'UTC')*1000, m.identity, " " EXTRACT(EPOCH FROM m.last_authorized_time AT TIME ZONE 'UTC')*1000, " " EXTRACT(EPOCH FROM m.last_deauthorized_time AT TIME ZONE 'UTC')*1000, " " m.remote_trace_level, m.remote_trace_target, m.tags, m.v_major, m.v_minor, m.v_rev, m.v_proto, " " m.no_auto_assign_ips, m.revision " "FROM ztc_member m " "INNER JOIN ztc_network n " " ON n.id = m.network_id " "WHERE n.controller_id = $1 AND m.deleted = false", 1, NULL, params, NULL, NULL, 0); if (PQresultStatus(res) != PGRES_TUPLES_OK) { fprintf(stderr, "Member Initialization Failed: %s", PQerrorMessage(conn)); PQclear(res); exit(1); } int numRows = PQntuples(res); for (int i = 0; i < numRows; ++i) { json empty; json config; std::string memberId(PQgetvalue(res, i, 0)); std::string networkId(PQgetvalue(res, i, 1)); config["id"] = memberId; config["nwid"] = networkId; config["activeBridge"] = (strcmp(PQgetvalue(res, i, 3), "true") == 0); config["authorized"] = (strcmp(PQgetvalue(res, i, 4), "true") == 0); config["capabilities"] = json::parse(PQgetvalue(res, i, 5)); config["creationTime"] = std::stoull(PQgetvalue(res, i, 6)); config["identity"] = PQgetvalue(res, i, 7); config["lastAuthorizedTime"] = std::stoull(PQgetvalue(res, i, 8)); config["lastDeauthorizedTime"] = std::stoull(PQgetvalue(res, i, 9)); config["remoteTraceLevel"] = std::stoi(PQgetvalue(res, i, 10)); config["remoteTraceTarget"] = PQgetvalue(res, i, 11); config["tags"] = json::parse(PQgetvalue(res, i, 12)); config["vMajor"] = std::stoi(PQgetvalue(res, i, 13)); config["vMinor"] = std::stoi(PQgetvalue(res, i, 14)); config["vRev"] = std::stoi(PQgetvalue(res, i, 15)); config["vProto"] = std::stoi(PQgetvalue(res, i, 16)); config["noAutoAssignIps"] = (strcmp(PQgetvalue(res, i, 17), "true") == 0); config["revision"] = std::stoull(PQgetvalue(res, i, 18)); config["objtype"] = "member"; config["ipAssignments"] = json::array(); const char *p2[2] = { memberId.c_str(), networkId.c_str() }; PGresult *r2 = PQexecParams(conn, "SELECT address FROM ztc_member_ip_assignment WHERE member_id = $1 AND network_id = $2", 2, NULL, p2, NULL, NULL, 0); if (PQresultStatus(r2) != PGRES_TUPLES_OK) { fprintf(stderr, "Member Initialization Failed: %s", PQerrorMessage(conn)); PQclear(r2); PQclear(res); exit(1); } int n = PQntuples(r2); for (int j = 0; j < n; ++j) { config["ipAssignments"].push_back(PQgetvalue(r2, j, 0)); } _memberChanged(empty, config, false); } PQclear(res); if (++this->_ready == 2) { if (_waitNoticePrinted) { fprintf(stderr,"[%s] NOTICE: %.10llx controller PostgreSQL data download complete." ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt()); } _readyLock.unlock(); } } catch (std::exception &e) { fprintf(stderr, "ERROR: Error initializing members: %s\n", e.what()); exit(-1); } } void PostgreSQL::heartbeat() { char publicId[1024]; char hostnameTmp[1024]; _myId.toString(false,publicId); if (gethostname(hostnameTmp, sizeof(hostnameTmp))!= 0) { hostnameTmp[0] = (char)0; } else { for (int i = 0; i < sizeof(hostnameTmp); ++i) { if ((hostnameTmp[i] == '.')||(hostnameTmp[i] == 0)) { hostnameTmp[i] = (char)0; break; } } } const char *controllerId = _myAddressStr.c_str(); const char *publicIdentity = publicId; const char *hostname = hostnameTmp; PGconn *conn = PQconnectdb(_path.c_str()); if (PQstatus(conn) == CONNECTION_BAD) { fprintf(stderr, "Connection to database failed: %s\n", PQerrorMessage(conn)); PQfinish(conn); exit(1); } while (_run == 1) { if(PQstatus(conn) != CONNECTION_OK) { PQfinish(conn); conn = PQconnectdb(_path.c_str()); } if (conn) { const char *values[8] = { controllerId, hostname, std::to_string(OSUtils::now()).c_str(), publicIdentity, std::to_string(ZEROTIER_ONE_VERSION_MAJOR).c_str(), std::to_string(ZEROTIER_ONE_VERSION_MINOR).c_str(), std::to_string(ZEROTIER_ONE_VERSION_REVISION).c_str(), std::to_string(ZEROTIER_ONE_VERSION_BUILD).c_str() }; int lengths[8] = { (int)strlen(values[0]), (int)strlen(values[1]), (int)strlen(values[2]), (int)strlen(values[3]), (int)strlen(values[4]), (int)strlen(values[5]), (int)strlen(values[6]), (int)strlen(values[7]) }; int binary[8] = {0,0,0,0,0,0,0,0}; PGresult *res = PQexecParams(conn, "INSERT INTO ztc_controller (id, cluster_host, last_alive, public_identity, v_major, v_minor, v_rev, v_build) " "VALUES ($1, $2, TO_TIMESTAMP($3::double precision/1000), $4, $5, $6, $7, $8) " "ON CONFLICT (id) DO UPDATE SET cluster_host = EXCLUDED.cluster_host, last_alive = EXCLUDED.last_alive, " "public_identity = EXCLUDED.public_identity, v_major = EXCLUDED.v_major, v_minor = EXCLUDED.v_minor, " "v_rev = EXCLUDED.v_rev, v_build = EXCLUDED.v_rev", 8, // number of parameters NULL, // oid field. ignore values, // values for substitution lengths, // lengths in bytes of each value binary, // binary? 0); if (PQresultStatus(res) != PGRES_COMMAND_OK) { fprintf(stderr, "Heartbeat Update Failed: %s\n", PQresultErrorMessage(res)); } PQclear(res); } std::this_thread::sleep_for(std::chrono::milliseconds(1000)); } PQfinish(conn); conn = NULL; } void PostgreSQL::membersDbWatcher() { PGconn *conn = PQconnectdb(_path.c_str()); if (PQstatus(conn) == CONNECTION_BAD) { fprintf(stderr, "Connection to database failed: %s\n", PQerrorMessage(conn)); PQfinish(conn); exit(1); } initializeMembers(conn); char buf[11] = {0}; std::string cmd = "LISTEN member_" + std::string(_myAddress.toString(buf)); PGresult *res = PQexec(conn, cmd.c_str()); if (!res || PQresultStatus(res) != PGRES_COMMAND_OK) { fprintf(stderr, "LISTEN command failed: %s\n", PQresultErrorMessage(res)); PQclear(res); PQfinish(conn); exit(1); } PQclear(res); res = NULL; while(_run == 1) { if (PQstatus(conn) != CONNECTION_OK) { fprintf(stderr, "ERROR: Member Watcher lost connection to Postgres."); exit(-1); } PGnotify *notify = NULL; PQconsumeInput(conn); while ((notify = PQnotifies(conn)) != NULL) { fprintf(stderr, "ASYNC NOTIFY of '%s' id:%s received\n", notify->relname, notify->extra); try { json tmp(json::parse(notify->extra)); json &ov = tmp["old_val"]; json &nv = tmp["new_val"]; json oldConfig, newConfig; if (ov.is_object()) oldConfig = ov; if (nv.is_object()) newConfig = nv; if (oldConfig.is_object() || newConfig.is_object()) { _memberChanged(oldConfig,newConfig,(this->_ready>=2)); } } catch (...) {} // ignore bad records free(notify); } std::this_thread::sleep_for(std::chrono::milliseconds(10)); } PQfinish(conn); conn = NULL; } void PostgreSQL::networksDbWatcher() { PGconn *conn = PQconnectdb(_path.c_str()); if (PQstatus(conn) == CONNECTION_BAD) { fprintf(stderr, "Connection to database failed: %s\n", PQerrorMessage(conn)); PQfinish(conn); exit(1); } initializeNetworks(conn); char buf[11] = {0}; std::string cmd = "LISTEN network_" + std::string(_myAddress.toString(buf)); PGresult *res = PQexec(conn, cmd.c_str()); if (!res || PQresultStatus(res) != PGRES_COMMAND_OK) { fprintf(stderr, "LISTEN command failed: %s\n", PQresultErrorMessage(res)); PQclear(res); PQfinish(conn); exit(1); } PQclear(res); res = NULL; while(_run == 1) { if (PQstatus(conn) != CONNECTION_OK) { fprintf(stderr, "ERROR: Network Watcher lost connection to Postgres."); exit(-1); } PGnotify *notify = NULL; PQconsumeInput(conn); while ((notify = PQnotifies(conn)) != NULL) { fprintf(stderr, "ASYNC NOTIFY of '%s' id:%s received\n", notify->relname, notify->extra); try { json tmp(json::parse(notify->extra)); json &ov = tmp["old_val"]; json &nv = tmp["new_val"]; json oldConfig, newConfig; if (ov.is_object()) oldConfig = ov; if (nv.is_object()) newConfig = nv; if (oldConfig.is_object()||newConfig.is_object()) { _networkChanged(oldConfig,newConfig,(this->_ready >= 2)); } } catch (...) {} // ignore bad records free(notify); } std::this_thread::sleep_for(std::chrono::milliseconds(10)); } PQfinish(conn); conn = NULL; } void PostgreSQL::commitThread() { PGconn *conn = PQconnectdb(_path.c_str()); if (PQstatus(conn) == CONNECTION_BAD) { fprintf(stderr, "ERROR: Connection to database failed: %s\n", PQerrorMessage(conn)); PQfinish(conn); exit(1); } json *config = nullptr; while(_commitQueue.get(config)&(_run == 1)) { if (!config) { continue; } if (PQstatus(conn) == CONNECTION_BAD) { fprintf(stderr, "ERROR: Connection to database failed: %s\n", PQerrorMessage(conn)); PQfinish(conn); exit(1); } try { const std::string objtype = (*config)["objtype"]; if (objtype == "member") { try { std::string memberId = (*config)["id"]; std::string networkId = (*config)["nwid"]; std::string name = (*config)["name"]; std::string description = (*config)["description"]; std::string identity = (*config)["identity"]; std::string target = "NULL"; if (!(*config)["remoteTraceTarget"].is_null()) { target = (*config)["remoteTraceTarget"]; } const char *values[19] = { memberId.c_str(), networkId.c_str(), ((*config)["activeBridge"] ? "true" : "false"), ((*config)["authorized"] ? "true" : "false"), OSUtils::jsonDump((*config)["capabilities"], -1).c_str(), name.c_str(), description.c_str(), identity.c_str(), std::to_string((long long)(*config)["lastAuthorizedTime"]).c_str(), std::to_string((long long)(*config)["lastDeauthorizedTime"]).c_str(), ((*config)["noAutoAssignIps"] ? "true" : "false"), std::to_string((int)(*config)["remoteTraceLevel"]).c_str(), (target == "NULL") ? NULL : target.c_str(), std::to_string((unsigned long long)(*config)["revision"]).c_str(), OSUtils::jsonDump((*config)["tags"], -1).c_str(), std::to_string((int)(*config)["vMajor"]).c_str(), std::to_string((int)(*config)["vMinor"]).c_str(), std::to_string((int)(*config)["vRev"]).c_str(), std::to_string((int)(*config)["vProto"]).c_str() }; PGresult *res = PQexecParams(conn, "INSERT INTO ztc_member (id, network_id, active_bridge, authorized, capabilities, " "name, description, identity, last_authorized_time, last_deauthorized_time, no_auto_assign_ips, " "remote_trace_level, remote_trace_target, revision, tags, v_major, v_minor, v_rev, v_proto) " "VALUES ($1, $2, $3, $4, $5, $6, $7, $8, " "TO_TIMESTAMP($9::double precision/1000), TO_TIMESTAMP($10::double precision/1000), " "$11, $12, $13, $14, $15, $16, $17, $18, $19) ON CONFLICT (network_id, id) DO UPDATE SET " "active_bridge = EXCLUDED.active_bridge, authorized = EXCDLUDED.authorized, capabilities = EXCLUDED.capabilities, " "name = EXCLUDED.name, description = EXCLUDED.description, " "identity = EXCLUDED.identity, last_authorized_time = EXCLUDED.last_authorized_time, " "last_deauthorized_time = EXCLUDED.last_deauthorized_time, no_auto_assign_ips = EXCLUDED.no_auto_assign_ips, " "remote_trace_level = EXCLUDED.remote_trace_level, remote_trace_target = EXCLUDED.remote_trace_target, " "revision = EXCLUDED.revision+1, tags = EXCLUDED.tags, v_major = EXCLUDED.v_major, " "v_minor = EXCLUDED.v_minor, v_rev = EXCLUDED.v_rev, v_proto = EXCLUDED.v_proto", 20, NULL, values, NULL, NULL, 0); if (PQresultStatus(res) != PGRES_COMMAND_OK) { fprintf(stderr, "ERROR: Error updating member: %s\n", PQresultErrorMessage(res)); PQclear(res); continue; } PQclear(res); res = PQexec(conn, "BEGIN"); if (PQresultStatus(res) != PGRES_COMMAND_OK) { fprintf(stderr, "ERROR: Error beginning transaction: %s\n", PQresultErrorMessage(res)); PQclear(res); continue; } PQclear(res); const char *v2[2] = { memberId.c_str(), networkId.c_str() }; res = PQexecParams(conn, "DELETE FROM ztc_member_ip_assignment WHERE member_id = $1 AND network_id = $2", 2, NULL, v2, NULL, NULL, 0); if (PQresultStatus(res) != PGRES_COMMAND_OK) { fprintf(stderr, "ERROR: Error updating IP address assignments: %s\n", PQresultErrorMessage(res)); PQclear(res); PQclear(PQexec(conn, "ROLLBACK"));; continue; } PQclear(res); for (auto i = (*config)["ipAssignments"].begin(); i != (*config)["ipAssignments"].end(); ++i) { std::string addr = *i; const char *v3[3] = { memberId.c_str(), networkId.c_str(), addr.c_str() }; res = PQexecParams(conn, "INSERT INTO ztc_member_ip_assignment (member_id, network_id, address) VALUES ($1, $2, $3)", 3, NULL, v3, NULL, NULL, 0); if (PQresultStatus(res) != PGRES_COMMAND_OK) { fprintf(stderr, "ERROR: Error setting IP addresses for member: %s\n", PQresultErrorMessage(res)); PQclear(res); PQclear(PQexec(conn, "ROLLBACK")); continue; } } res = PQexec(conn, "COMMIT"); if (PQresultStatus(res) != PGRES_COMMAND_OK) { fprintf(stderr, "ERROR: Error committing ip address data: %s\n", PQresultErrorMessage(res)); } PQclear(res); } catch (std::exception &e) { fprintf(stderr, "ERROR: Error updating member: %s\n", e.what()); } } else if (objtype == "network") { try { std::string id = (*config)["id"]; std::string controllerId = _myAddressStr.c_str(); std::string name = (*config)["name"]; std::string remoteTraceTarget("NULL"); if (!(*config)["remoteTraceTarget"].is_null()) { remoteTraceTarget = (*config)["remoteTraceTarget"]; } std::string rulesSource = (*config)["rulesSource"]; const char *values[16] = { id.c_str(), controllerId.c_str(), OSUtils::jsonDump((*config)["capabilitles"], -1).c_str(), ((*config)["enableBroadcast"] ? "true" : "false"), std::to_string(OSUtils::now()).c_str(), std::to_string((int)(*config)["mtu"]).c_str(), std::to_string((int)(*config)["multicastLimit"]).c_str(), name.c_str(), ((*config)["private"] ? "true" : "false"), std::to_string((int)(*config)["remoteTraceLevel"]).c_str(), (remoteTraceTarget == "NULL" ? NULL : remoteTraceTarget.c_str()), OSUtils::jsonDump((*config)["rules"], -1).c_str(), rulesSource.c_str(), OSUtils::jsonDump((*config)["tags"], -1).c_str(), OSUtils::jsonDump((*config)["v4AssignMode"],-1).c_str(), OSUtils::jsonDump((*config)["v6AssignMode"], -1).c_str(), }; PGresult *res = PQexecParams(conn, "UPDATE ztc_network SET controller_id = $2, capabilities = $3, enable_broadcast = $4, " "last_updated = $5, mtu = $6, multicast_limit = $7, name = $8, private = $9, " "remote_trace_level = $10, remote_trace_target = $11, rules = $12, rules_source = $13, " "tags = $14, v4_assign_mode = $15, v6_assign_mode = $16 " "WHERE id = $1", 16, NULL, values, NULL, NULL, 0); if (PQresultStatus(res) != PGRES_COMMAND_OK) { fprintf(stderr, "ERROR: Error updating network record: %s\n", PQresultErrorMessage(res)); PQclear(res); continue; } PQclear(res); res = PQexec(conn, "BEGIN"); if (PQresultStatus(res) != PGRES_COMMAND_OK) { fprintf(stderr, "ERROR: Error beginnning transaction: %s\n", PQresultErrorMessage(res)); PQclear(res); continue; } PQclear(res); const char *params[1] = { id.c_str() }; res = PQexecParams(conn, "DELETE FROM ztc_network_assignment_pool WHERE network_id = $1", 1, NULL, params, NULL, NULL, 0); if (PQresultStatus(res) != PGRES_COMMAND_OK) { fprintf(stderr, "ERROR: Error updating assignment pool: %s\n", PQresultErrorMessage(res)); PQclear(res); PQclear(PQexec(conn, "ROLLBACK")); continue; } PQclear(res); auto pool = (*config)["ipAssignmentPools"]; bool err = false; for (auto i = pool.begin(); i != pool.end(); ++i) { std::string start = (*i)["ipRangeStart"]; std::string end = (*i)["ipRangeEnd"]; const char *p[3] = { id.c_str(), start.c_str(), end.c_str() }; res = PQexecParams(conn, "INSERT INTO ztc_network_assignment_pool (network_id, ip_range_start, ip_range_end) " "VALUES ($1, $2, $3)", 3, NULL, p, NULL, NULL, 0); if (PQresultStatus(res) != PGRES_COMMAND_OK) { fprintf(stderr, "ERROR: Error updating assignment pool: %s\n", PQresultErrorMessage(res)); PQclear(res); err = true; break; } PQclear(res); } if (err) { PQclear(PQexec(conn, "ROLLBACK")); continue; } res = PQexecParams(conn, "DELETE FROM ztc_network_route WHERE network_id = $1", 1, NULL, params, NULL, NULL, 0); if (PQresultStatus(res) != PGRES_COMMAND_OK) { fprintf(stderr, "ERROR: Error updating routes: %s\n", PQresultErrorMessage(res)); PQclear(res); PQclear(PQexec(conn, "ROLLBACK")); continue; } auto routes = (*config)["routes"]; err = false; for (auto i = routes.begin(); i != routes.end(); ++i) { std::string t = (*i)["target"]; std::vector target; std::istringstream f(t); std::string s; while(std::getline(f, s, '/')) { target.push_back(s); } if (target.empty() || target.size() != 2) { continue; } std::string targetAddr = target[0]; std::string targetBits = target[1]; std::string via = "NULL"; if (!(*i)["via"].is_null()) { via = (*i)["via"]; } const char *p[4] = { id.c_str(), targetAddr.c_str(), targetBits.c_str(), (via == "NULL" ? NULL : via.c_str()), }; res = PQexecParams(conn, "INSERT INTO ztc_network_route (network_id, address, bits, via) VALUES ($1, $2, $3, $4)", 4, NULL, p, NULL, NULL, 0); if (PQresultStatus(res) != PGRES_COMMAND_OK) { fprintf(stderr, "ERROR: Error updating routes: %s\n", PQresultErrorMessage(res)); PQclear(res); err = true; break; } PQclear(res); } if (err) { PQclear(PQexec(conn, "ROLLBAcK")); continue; } res = PQexec(conn, "COMMIT"); if (PQresultStatus(res) != PGRES_COMMAND_OK) { fprintf(stderr, "ERROR: Error committing network update: %s\n", PQresultErrorMessage(res)); } PQclear(res); } catch (std::exception &e) { fprintf(stderr, "ERROR: Error updating member: %s\n", e.what()); } } else if (objtype == "trace") { fprintf(stderr, "ERROR: Trace not yet implemented"); } else if (objtype == "_delete_network") { try { std::string networkId = (*config)["nwid"]; const char *values[1] = { networkId.c_str() }; PGresult * res = PQexecParams(conn, "UPDATE ztc_network SET deleted = true WHERE id = $1", 1, NULL, values, NULL, NULL, 0); if (PQresultStatus(res) != PGRES_COMMAND_OK) { fprintf(stderr, "ERROR: Error deleting network: %s\n", PQresultErrorMessage(res)); } PQclear(res); } catch (std::exception &e) { fprintf(stderr, "ERROR: Error deleting network: %s\n", e.what()); } } else if (objtype == "_delete_member") { try { std::string memberId = (*config)["id"]; std::string networkId = (*config)["nwid"]; const char *values[2] = { memberId.c_str(), networkId.c_str() }; PGresult *res = PQexecParams(conn, "UPDATE ztc_member SET hidden = true, deleted = true WHERE id = $1 AND network_id = $2", 2, NULL, values, NULL, NULL, 0); if (PQresultStatus(res) != PGRES_COMMAND_OK) { fprintf(stderr, "ERROR: Error deleting member: %s\n", PQresultErrorMessage(res)); } PQclear(res); } catch (std::exception &e) { fprintf(stderr, "ERROR: Error deleting member: %s\n", e.what()); } } else { fprintf(stderr, "ERROR: unknown objtype"); } } catch (std::exception &e) { fprintf(stderr, "ERROR: Error getting objtype: %s\n", e.what()); } delete config; std::this_thread::sleep_for(std::chrono::milliseconds(10)); } PQfinish(conn); } void PostgreSQL::onlineNotificationThread() { PGconn *conn = PQconnectdb(_path.c_str()); if (PQstatus(conn) == CONNECTION_BAD) { fprintf(stderr, "Connection to database failed: %s\n", PQerrorMessage(conn)); PQfinish(conn); exit(1); } _connected = 1; int64_t lastUpdatedNetworkStatus = 0; std::unordered_map< std::pair,int64_t,_PairHasher > lastOnlineCumulative; while (_run == 1) { if (PQstatus(conn) != CONNECTION_OK) { fprintf(stderr, "ERROR: Online Notification thread lost connection to Postgres."); exit(-1); } std::unordered_map< std::pair,std::pair,_PairHasher > lastOnline; { std::lock_guard l(_lastOnline_l); lastOnline.swap(_lastOnline); } PGresult *res = NULL; int qCount = 0; if (!lastOnline.empty()) { fprintf(stderr, "Last Online Update\n"); res = PQexec(conn, "BEGIN"); if (PQresultStatus(res) != PGRES_COMMAND_OK) { fprintf(stderr, "ERROR: Error on BEGIN command (onlineNotificationThread): %s\n", PQresultErrorMessage(res)); PQclear(res); exit(1); } PQclear(res); } for (auto i=lastOnline.begin(); i != lastOnline.end(); ++i) { lastOnlineCumulative[i->first] = i->second.first; char nwidTmp[64]; char memTmp[64]; char ipTmp[64]; OSUtils::ztsnprintf(nwidTmp,sizeof(nwidTmp), "%.16llx", i->first.first); OSUtils::ztsnprintf(memTmp,sizeof(memTmp), "%.10llx", i->first.second); std::string networkId(nwidTmp); std::string memberId(memTmp); int64_t ts = i->second.first; std::string ipAddr = i->second.second.toIpString(ipTmp); const char *values[4] = { networkId.c_str(), memberId.c_str(), std::to_string(ts).c_str(), ipAddr.c_str() }; res = PQexecParams(conn, "INSERT INTO ztc_member_status (network_id, member_id, address, last_updated) VALUES ($1, $2, $3, $4)" "ON CONFLICT (network_id, member_id) DO UPDATE SET address = EXCLUDED.address, last_updated = EXCLUDED.last_updated", 8, // number of parameters NULL, // oid field. ignore values, // values for substitution NULL, // lengths in bytes of each value NULL, 0); if (PQresultStatus(res) != PGRES_COMMAND_OK) { fprintf(stderr, "Error on Member Status upsert: %s\n", PQresultErrorMessage(res)); PQclear(res); PQexec(conn, "ROLLBACK"); exit(1); } PQclear(res); if ((++qCount) == 1024) { res = PQexec(conn, "COMMIT"); if (PQresultStatus(res) != PGRES_COMMAND_OK) { fprintf(stderr, "ERROR: Error on commit (onlineNotificationThread): %s\n", PQresultErrorMessage(res)); PQclear(res); PQexec(conn, "ROLLBACK"); exit(1); } PQclear(res); res = PQexec(conn, "BEGIN"); if (PQresultStatus(res) != PGRES_COMMAND_OK) { fprintf(stderr, "ERROR: Error on BEGIN (onlineNotificationThread): %s\n", PQresultErrorMessage(res)); PQclear(res); exit(1); } PQclear(res); qCount = 0; } } if (qCount > 0) { fprintf(stderr, "qCount is %d\n", qCount); res = PQexec(conn, "COMMIT"); if (PQresultStatus(res) != PGRES_COMMAND_OK) { fprintf(stderr, "ERROR: Error on commit (onlineNotificationThread): %s\n", PQresultErrorMessage(res)); PQclear(res); PQexec(conn, "ROLLBACK"); exit(1); } PQclear(res); } const int64_t now = OSUtils::now(); if ((now - lastUpdatedNetworkStatus) > 10000) { lastUpdatedNetworkStatus = now; std::vector>> networks; { std::lock_guard l(_networks_l); for (auto i = _networks.begin(); i != _networks.end(); ++i) { networks.push_back(*i); } } int nCount = 0; if (!networks.empty()) { fprintf(stderr, "Network update"); res = PQexec(conn, "BEGIN"); if (PQresultStatus(res) != PGRES_COMMAND_OK) { fprintf(stderr, "ERROR: Error on BEGIN command (onlineNotificationThread): %s\n", PQresultErrorMessage(res)); PQclear(res); exit(1); } PQclear(res); } for (auto i = networks.begin(); i != networks.end(); ++i) { char tmp[64]; Utils::hex(i->first, tmp); std::string networkId(tmp); uint64_t authMemberCount = 0; uint64_t totalMemberCount = 0; uint64_t onlineMemberCount = 0; uint64_t bridgeCount = 0; uint64_t ts = now; { std::lock_guard l2(i->second->lock); authMemberCount = i->second->authorizedMembers.size(); totalMemberCount = i->second->members.size(); bridgeCount = i->second->activeBridgeMembers.size(); for (auto m=i->second->members.begin(); m != i->second->members.end(); ++m) { auto lo = lastOnlineCumulative.find(std::pair(i->first, m->first)); if (lo != lastOnlineCumulative.end()) { if ((now - lo->second) <= (ZT_NETWORK_AUTOCONF_DELAY * 2)) { ++onlineMemberCount; } else { lastOnlineCumulative.erase(lo); } } } } const char *values[6] = { networkId.c_str(), std::to_string(bridgeCount).c_str(), std::to_string(authMemberCount).c_str(), std::to_string(onlineMemberCount).c_str(), std::to_string(totalMemberCount).c_str(), std::to_string(ts).c_str() }; res = PQexecParams(conn, "INSERT INTO ztc_network_status (network_id, bridge_count, authorized_member_count, " "online_member_count, total_member_count, last_modified) VALUES ($1, $2, $3, $4, $5, $6) " "ON CONFLICT (network_id) DO UPDATE SET bridge_count = EXCLUDED.bridge_count, " "authorized_member_count = EXCLUDED.authorized_member_count, online_member_count = EXCDLUDED.online_member_count, " "total_member_count = EXCLUDED.total_member_count, last_modified = EXCLUDED.last_modified", 6, NULL, values, NULL, NULL, 0); if (PQresultStatus(res) != PGRES_COMMAND_OK) { fprintf(stderr, "ERROR: Error on Network Satus upsert (onlineNotificationThread): %s\n", PQresultErrorMessage(res)); PQclear(res); PQexec(conn, "ROLLBACK"); exit(1); } if ((++nCount) == 1024) { res = PQexec(conn, "COMMIT"); if (PQresultStatus(res) != PGRES_COMMAND_OK) { fprintf(stderr, "ERROR: Error on COMMIT (onlineNotificationThread): %s\n" , PQresultErrorMessage(res)); PQclear(res); PQexec(conn, "ROLLBACK"); exit(1); } res = PQexec(conn, "BEGIN"); if (PQresultStatus(res) != PGRES_COMMAND_OK) { fprintf(stderr, "ERROR: Error on BEGIN command (onlineNotificationThread): %s\n", PQresultErrorMessage(res)); PQclear(res); exit(1); } nCount = 0; } } if (nCount > 0) { fprintf(stderr, "nCount is %d\n", nCount); res = PQexec(conn, "COMMIT"); if (PQresultStatus(res) != PGRES_COMMAND_OK) { fprintf(stderr, "ERROR: Error on COMMIT (onlineNotificationThread): %s\n", PQresultErrorMessage(res)); PQclear(res); PQexec(conn, "ROLLBACK"); exit(1); } } } std::this_thread::sleep_for(std::chrono::milliseconds(250)); } PQfinish(conn); } #endif //ZT_CONTROLLER_USE_LIBPQ