summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--controller/PostgreSQL.cpp1368
-rw-r--r--controller/PostgreSQL.hpp11
-rw-r--r--docker/Dockerfile2
3 files changed, 474 insertions, 907 deletions
diff --git a/controller/PostgreSQL.cpp b/controller/PostgreSQL.cpp
index ea04f551..1b97042c 100644
--- a/controller/PostgreSQL.cpp
+++ b/controller/PostgreSQL.cpp
@@ -22,7 +22,7 @@
#include "EmbeddedNetworkController.hpp"
#include "../version.h"
-#include <libpq-fe.h>
+#include <pqxx/pqxx>
#include <sstream>
using json = nlohmann::json;
@@ -62,6 +62,67 @@ std::string join(const std::vector<std::string> &elements, const char * const se
}
+namespace ZeroTier {
+
+class _MemberNotificationReceiver : public pqxx::notification_receiver
+{
+private:
+ ZeroTier::PostgreSQL *_pgsql;
+
+public:
+ _MemberNotificationReceiver(pqxx::connection_base &c, const std::string &channel, ZeroTier::PostgreSQL *pgsql)
+ : pqxx::notification_receiver(c, channel)
+ , _pgsql(pgsql)
+ {}
+
+ virtual void operator()(const std::string &payload, int backend_pid)
+ {
+ try {
+ json tmp(json::parse(payload));
+ 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()) {
+ _pgsql->_memberChanged(oldConfig,newConfig,_pgsql->isReady());
+ }
+ } catch (std::exception &e) {
+ fprintf(stderr, "Exception parsing member notification: %s\n", e.what());
+ }
+ }
+};
+
+class _NetworkNotificationReceiver : public pqxx::notification_receiver
+{
+private:
+ ZeroTier::PostgreSQL *_pgsql;
+public:
+ _NetworkNotificationReceiver(pqxx::connection_base &c, const std::string &channel, ZeroTier::PostgreSQL *pgsql)
+ : pqxx::notification_receiver(c, channel)
+ , _pgsql(pgsql)
+ {}
+
+ virtual void operator()(const std::string &payload, int backend_pid)
+ {
+ try {
+ json tmp(json::parse(payload));
+ 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()) {
+ _pgsql->_networkChanged(oldConfig,newConfig,_pgsql->isReady());
+ }
+ } catch (std::exception &e) {
+ fprintf(stderr, "Exception parsing member notification: %s\n", e.what());
+ } // ignore bad records
+ }
+};
+
+} // namespace ZeroTier
+
using namespace ZeroTier;
PostgreSQL::PostgreSQL(EmbeddedNetworkController *const nc, const Identity &myId, const char *path)
@@ -71,6 +132,7 @@ PostgreSQL::PostgreSQL(EmbeddedNetworkController *const nc, const Identity &myId
, _run(1)
, _waitNoticePrinted(false)
{
+ fprintf(stderr, "PostgreSQL Constructed");
_connString = std::string(path);
_readyLock.lock();
@@ -173,157 +235,101 @@ void PostgreSQL::nodeIsOnline(const uint64_t networkId, const uint64_t memberId,
}
}
-void PostgreSQL::initializeNetworks(PGconn *conn)
+void PostgreSQL::initializeNetworks(pqxx::connection &conn)
{
try {
- if (PQstatus(conn) != CONNECTION_OK) {
- fprintf(stderr, "Bad Database Connection: %s", PQerrorMessage(conn));
+ if (!conn.is_open()) {
+ fprintf(stderr, "Bad Database Connection in initializeNetworks\n");
exit(1);
}
- const char *params[1] = {
- _myAddressStr.c_str()
- };
+ pqxx::work w(conn);
- 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, "
+ pqxx::result r = w.exec("SELECT id, EXTRACT(EPOCH FROM creation_time AT TIME ZONE 'UTC')*1000 AS creation_time, capabilities, "
+ "enable_broadcast, EXTRACT(EPOCH FROM last_modified AT TIME ZONE 'UTC')*1000 AS last_modified, 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);
- }
+ "WHERE deleted = false AND controller_id = '" + w.esc(_myAddressStr) + "'");
- int numRows = PQntuples(res);
- for (int i = 0; i < numRows; ++i) {
+
+ for (pqxx::result::const_iterator row = r.begin(); row != r.end(); ++row) {
json empty;
json config;
- const char *nwidparam[1] = {
- PQgetvalue(res, i, 0)
- };
-
- config["id"] = PQgetvalue(res, i, 0);
- config["nwid"] = PQgetvalue(res, i, 0);
+ std::string nwid = row["id"].as<std::string>();
+ config["id"] = nwid;
+ config["nwid"] = nwid;
try {
- config["creationTime"] = std::stoull(PQgetvalue(res, i, 1));
- } catch (std::exception &e) {
+ config["creationTime"] = row["creation_time"].as<uint64_t>();
+ } catch(std::exception &e) {
config["creationTime"] = 0ULL;
- //fprintf(stderr, "Error converting creation time: %s\n", PQgetvalue(res, i, 1));
}
- config["capabilities"] = json::parse(PQgetvalue(res, i, 2));
- config["enableBroadcast"] = (strcmp(PQgetvalue(res, i, 3),"t")==0);
+ config["capabilities"] = json::parse(row["capabilities"].as<std::string>());
+ config["enableBroadcast"] = row["enable_broadcast"].as<bool>();
try {
- config["lastModified"] = std::stoull(PQgetvalue(res, i, 4));
+ config["lastModified"] = row["last_modified"].as<uint64_t>();
} catch (std::exception &e) {
config["lastModified"] = 0ULL;
- //fprintf(stderr, "Error converting last modified: %s\n", PQgetvalue(res, i, 4));
}
try {
- config["mtu"] = std::stoi(PQgetvalue(res, i, 5));
+ config["mtu"] = row["mtu"].as<int>();
} catch (std::exception &e) {
config["mtu"] = 2800;
}
try {
- config["multicastLimit"] = std::stoi(PQgetvalue(res, i, 6));
+ config["multicastLimit"] = row["multicast_limit"].as<int>();
} catch (std::exception &e) {
config["multicastLimit"] = 64;
}
- config["name"] = PQgetvalue(res, i, 7);
- config["private"] = (strcmp(PQgetvalue(res, i, 8),"t")==0);
+ config["name"] = row["name"].as<std::string>();
+ config["private"] = row["private"].as<bool>();
try {
- config["remoteTraceLevel"] = std::stoi(PQgetvalue(res, i, 9));
+ config["remoteTraceLevel"] = row["remote_trace_level"].as<int>();
} catch (std::exception &e) {
config["remoteTraceLevel"] = 0;
}
- config["remoteTraceTarget"] = PQgetvalue(res, i, 10);
+ config["remoteTraceTarget"] = (row["remote_trace_target"].is_null() ? nullptr : row["remote_trace_target"].as<std::string>());
try {
- config["revision"] = std::stoull(PQgetvalue(res, i, 11));
+ config["revision"] = row["revision"].as<uint64_t>();
} catch (std::exception &e) {
config["revision"] = 0ULL;
//fprintf(stderr, "Error converting revision: %s\n", 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["rules"] = json::parse(row["rules"].as<std::string>());
+ config["tags"] = json::parse(row["tags"].as<std::string>());
+ config["v4AssignMode"] = json::parse(row["v4_assign_mode"].as<std::string>());
+ config["v6AssignMode"] = json::parse(row["v6_assign_mode"].as<std::string>());
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,
- nwidparam,
- 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) {
+ pqxx::work w2(conn);
+ pqxx::result res2 = w2.exec("SELECT host(ip_range_start) AS ip_range_start, host(ip_range_end) AS ip_range_end FROM ztc_network_assignment_pool WHERE network_id = '" + w2.esc(nwid) + "'");
+ for(pqxx::result::const_iterator it = res2.begin(); it != res2.end(); ++it) {
json ip;
- ip["ipRangeStart"] = PQgetvalue(r2, j, 0);
- ip["ipRangeEnd"] = PQgetvalue(r2, j, 1);
-
+ ip["ipRangeStart"] = it["ip_range_start"].as<std::string>();
+ ip["ipRangeEnd"] = it["ip_range_end"].as<std::string>();
config["ipAssignmentPools"].push_back(ip);
}
+ w2.commit();
- PQclear(r2);
-
- r2 = PQexecParams(conn,
- "SELECT host(address), bits, host(via) FROM ztc_network_route WHERE network_id = $1",
- 1,
- NULL,
- nwidparam,
- 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);
+ pqxx::work w3(conn);
+ pqxx::result res3 = w3.exec("SELECT host(address) AS address, bits, host(via) AS via FROM ztc_network_route WHERE network_id = '" + w3.esc(nwid) + "'");
+ for(pqxx::result::const_iterator it = res3.begin(); it != res3.end(); ++it) {
json route;
- route["target"] = addr + "/" + bits;
-
- if (via == "NULL") {
+ route["target"] = it["address"].as<std::string>() + "/" + it["bits"].as<std::string>();
+ if (route["via"].is_null()) {
route["via"] = nullptr;
} else {
- route["via"] = via;
+ route["via"] = it["via"].as<std::string>();
}
config["routes"].push_back(route);
}
+ w3.commit();
- PQclear(r2);
-
_networkChanged(empty, config, false);
}
- PQclear(res);
+ w.commit();
if (++this->_ready == 2) {
if (_waitNoticePrinted) {
@@ -337,147 +343,115 @@ void PostgreSQL::initializeNetworks(PGconn *conn)
}
}
-void PostgreSQL::initializeMembers(PGconn *conn)
+void PostgreSQL::initializeMembers(pqxx::connection &conn)
{
try {
- if (PQstatus(conn) != CONNECTION_OK) {
- fprintf(stderr, "Bad Database Connection: %s", PQerrorMessage(conn));
+ if (!conn.is_open()) {
+ fprintf(stderr, "Bad Database Connection in initializeMembers\n");
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 "
+ pqxx::work w(conn);
+ pqxx::result res = w.exec(
+ "SELECT m.id AS id, m.network_id AS network_id, m.active_bridge AS active_bridge, "
+ " m.authorized AS authorized, m.capabilities AS capabilities, "
+ " EXTRACT(EPOCH FROM m.creation_time AT TIME ZONE 'UTC')*1000 AS creation_time, m.identity AS identity, "
+ " EXTRACT(EPOCH FROM m.last_authorized_time AT TIME ZONE 'UTC')*1000 AS last_authorized_time, "
+ " EXTRACT(EPOCH FROM m.last_deauthorized_time AT TIME ZONE 'UTC')*1000 AS last_deauthorized_time, "
+ " m.remote_trace_level AS remote_trace_level, m.remote_trace_target AS remote_trace_target, "
+ " m.tags AS tags, m.v_major AS v_major, m.v_minor AS v_minor, m.v_rev AS v_rev, "
+ " m.v_proto AS v_proto, m.no_auto_assign_ips AS no_auto_assign_ips, m.revision AS 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);
- }
+ "WHERE n.controller_id = '" + w.esc(_myAddressStr) + "' AND m.deleted = false"
+ );
- int numRows = PQntuples(res);
- for (int i = 0; i < numRows; ++i) {
+ for(pqxx::result::const_iterator row = res.begin(); row != res.end(); ++row) {
json empty;
json config;
- std::string memberId(PQgetvalue(res, i, 0));
- std::string networkId(PQgetvalue(res, i, 1));
- std::string ctime = PQgetvalue(res, i, 5);
+ std::string networkId = row["network_id"].as<std::string>();
+ std::string memberId = row["id"].as<std::string>();
+
config["id"] = memberId;
config["nwid"] = networkId;
- config["activeBridge"] = (strcmp(PQgetvalue(res, i, 2), "t") == 0);
- config["authorized"] = (strcmp(PQgetvalue(res, i, 3), "t") == 0);
+ config["activeBridge"] = row["active_bridge"].as<bool>();
+ config["authorized"] = row["authorized"].as<bool>();
try {
- config["capabilities"] = json::parse(PQgetvalue(res, i, 4));
- } catch (std::exception &e) {
+ config["capabilities"] = json::parse(row["capabilities"].as<std::string>());
+ } catch(std::exception &e) {
config["capabilities"] = json::array();
}
try {
- config["creationTime"] = std::stoull(PQgetvalue(res, i, 5));
- } catch (std::exception &e) {
+ config["creationTime"] = row["creation_time"].as<uint64_t>();
+ } catch(std::exception &e) {
config["creationTime"] = 0ULL;
- //fprintf(stderr, "Error upding creation time (member): %s\n", PQgetvalue(res, i, 5));
}
- config["identity"] = PQgetvalue(res, i, 6);
+ config["identity"] = row["identity"].as<std::string>();
try {
- config["lastAuthorizedTime"] = std::stoull(PQgetvalue(res, i, 7));
+ config["lastAuthorizedTime"] = row["last_authorized_time"].as<uint64_t>();
} catch(std::exception &e) {
config["lastAuthorizedTime"] = 0ULL;
- //fprintf(stderr, "Error updating last auth time (member): %s\n", PQgetvalue(res, i, 7));
}
try {
- config["lastDeauthorizedTime"] = std::stoull(PQgetvalue(res, i, 8));
- } catch( std::exception &e) {
+ config["lastDeauthorizedTime"] = row["last_deauthorized_time"].as<uint64_t>();
+ } catch(std::exception &e) {
config["lastDeauthorizedTime"] = 0ULL;
- //fprintf(stderr, "Error updating last deauth time (member): %s\n", PQgetvalue(res, i, 8));
}
try {
- config["remoteTraceLevel"] = std::stoi(PQgetvalue(res, i, 9));
- } catch (std::exception &e) {
+ config["remoteTraceLevel"] = row["remote_trace_level"].as<int>();
+ } catch(std::exception &e) {
config["remoteTraceLevel"] = 0;
}
- config["remoteTraceTarget"] = PQgetvalue(res, i, 10);
+ config["remoteTraceTarget"] = (row["remote_trace_target"].is_null() ? nullptr : row["remote_trace_target"].as<std::string>());
try {
- config["tags"] = json::parse(PQgetvalue(res, i, 11));
- } catch (std::exception &e) {
+ config["tags"] = json::parse(row["tags"].as<std::string>());
+ } catch(std::exception &e) {
config["tags"] = json::array();
}
try {
- config["vMajor"] = std::stoi(PQgetvalue(res, i, 12));
+ config["vMajor"] = row["v_major"].as<int>();
} catch(std::exception &e) {
config["vMajor"] = -1;
}
try {
- config["vMinor"] = std::stoi(PQgetvalue(res, i, 13));
+ config["vMinor"] = row["v_minor"].as<int>();
} catch (std::exception &e) {
config["vMinor"] = -1;
}
try {
- config["vRev"] = std::stoi(PQgetvalue(res, i, 14));
+ config["vRev"] = row["v_rev"].as<int>();
} catch (std::exception &e) {
config["vRev"] = -1;
}
try {
- config["vProto"] = std::stoi(PQgetvalue(res, i, 15));
+ config["vProto"] = row["v_proto"].as<int>();
} catch (std::exception &e) {
config["vProto"] = -1;
}
- config["noAutoAssignIps"] = (strcmp(PQgetvalue(res, i, 16), "t") == 0);
+ config["noAutoAssignIps"] = row["no_auto_assign_ips"].as<bool>();
try {
- config["revision"] = std::stoull(PQgetvalue(res, i, 17));
+ config["revision"] = row["revision"].as<uint64_t>();
} catch (std::exception &e) {
config["revision"] = 0ULL;
//fprintf(stderr, "Error updating revision (member): %s\n", PQgetvalue(res, i, 17));
}
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));
+ pqxx::work w2(conn);
+ pqxx::result r2 = w2.exec(
+ "SELECT address FROM ztc_member_ip_assignment WHERE member_id = '"+w2.esc(memberId)+"' AND network_id = '"+w2.esc(networkId)+"'"
+ );
+ for(pqxx::result::const_iterator it = r2.begin(); it != r2.end(); ++it) {
+ config["ipAssignments"].push_back(it["address"].as<std::string>());
}
+ w2.commit();
_memberChanged(empty, config, false);
}
-
- PQclear(res);
+ w.commit();
if (++this->_ready == 2) {
if (_waitNoticePrinted) {
@@ -510,497 +484,237 @@ void PostgreSQL::heartbeat()
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);
+ fprintf(stderr, "Heartbeat connection opening");
+ pqxx::connection conn(_connString);
+ if (!conn.is_open()) {
+ fprintf(stderr, "Connection to database failed: heartbeat\n");
exit(1);
}
+ fprintf(stderr, "Heartbeat connection opened");
+ conn.prepare("heartbeat",
+ "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"
+ );
+
while (_run == 1) {
- if(PQstatus(conn) != CONNECTION_OK) {
- PQfinish(conn);
- conn = PQconnectdb(_path.c_str());
- }
- if (conn) {
- std::string major = std::to_string(ZEROTIER_ONE_VERSION_MAJOR);
- std::string minor = std::to_string(ZEROTIER_ONE_VERSION_MINOR);
- std::string rev = std::to_string(ZEROTIER_ONE_VERSION_REVISION);
- std::string build = std::to_string(ZEROTIER_ONE_VERSION_BUILD);
- std::string now = std::to_string(OSUtils::now());
- const char *values[8] = {
- controllerId,
- hostname,
- now.c_str(),
- publicIdentity,
- major.c_str(),
- minor.c_str(),
- rev.c_str(),
- build.c_str()
- };
-
- 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
- NULL, // lengths in bytes of each value
- NULL, // binary?
- 0);
-
- if (PQresultStatus(res) != PGRES_COMMAND_OK) {
- fprintf(stderr, "Heartbeat Update Failed: %s\n", PQresultErrorMessage(res));
- }
- PQclear(res);
+ try {
+ pqxx::work w(conn);
+ pqxx::result res = w.prepared("heartbeat")(controllerId)(hostname)
+ (OSUtils::now())(publicIdentity)(ZEROTIER_ONE_VERSION_MAJOR)
+ (ZEROTIER_ONE_VERSION_MINOR)(ZEROTIER_ONE_VERSION_REVISION)
+ (ZEROTIER_ONE_VERSION_BUILD).exec();
+ w.commit();
+ } catch (std::exception &e) {
+ fprintf(stderr, "Error inserting heartbeat: %s\n", e.what());
+ exit(1);
}
-
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);
+ try {
+ pqxx::connection conn(_connString);
+ if (!conn.is_open()) {
+ fprintf(stderr, "Connection to database failed: membersDbWatcher\n");
+ 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
+ initializeMembers(conn);
- free(notify);
+ char buf[11] = {0};
+ std::string cmd = "member_" + std::string(_myAddress.toString(buf));
+ _MemberNotificationReceiver receiver(conn, cmd, this);
+ while(_run == 1) {
+ conn.await_notification(5, 0);
+ std::this_thread::sleep_for(std::chrono::milliseconds(10));
}
- std::this_thread::sleep_for(std::chrono::milliseconds(10));
+ conn.disconnect();
+ } catch (std::exception &e) {
+ fprintf(stderr, "Exception in membersDbWatcher: %s\n", e.what());
+ exit(1);
}
- 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);
- }
+ try {
+ pqxx::connection conn(_connString);
+ if (!conn.is_open()) {
+ fprintf(stderr, "Connection to database failed: networksDbWatcher\n");
+ exit(1);
+ }
- PQclear(res); res = NULL;
+ initializeNetworks(conn);
- 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);
+ char buf[11] = {0};
+ std::string cmd = "network_" + std::string(_myAddress.toString(buf));
+ _NetworkNotificationReceiver receiver(conn, cmd, this);
+ while(_run == 1) {
+ conn.await_notification(5, 0);
+ std::this_thread::sleep_for(std::chrono::milliseconds(10));
}
- std::this_thread::sleep_for(std::chrono::milliseconds(10));
+ } catch(std::exception &e) {
+ fprintf(stderr, "Exception in networksDbWatcher: %s\n", e.what());
+ exit(1);
}
- 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);
+ pqxx::connection conn(_connString);
+ if (!conn.is_open()) {
+ fprintf(stderr, "ERROR: Connection to database failed: commitThread\n");
exit(1);
}
+ conn.prepare("insert_member",
+ "INSERT INTO ztc_member (id, network_id, active_bridge, authorized, capabilities, "
+ "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, "
+ "TO_TIMESTAMP($7::double precision/1000), TO_TIMESTAMP($8::double precision/1000), "
+ "$9, $10, (CASE WHEN $11='' THEN NULL ELSE $1 END), $12, $13, $14, $15, $16, $17) ON CONFLICT (network_id, id) DO UPDATE SET "
+ "active_bridge = EXCLUDED.active_bridge, authorized = EXCLUDED.authorized, capabilities = EXCLUDED.capabilities, "
+ "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");
+ conn.prepare("delete_ip_assignments",
+ "DELETE FROM ztc_member_ip_assignment WHERE member_id = $1 AND network_id = $2");
+ conn.prepare("insert_ip_assignments",
+ "INSERT INTO ztc_member_ip_assignment (member_id, network_id, address) VALUES ($1, $2, $3)");
+
+ conn.prepare("update_network",
+ "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");
+ conn.prepare("delete_network_ip_pool",
+ "DELETE FROM ztc_network_assignment_pool WHERE network_id = $1");
+ conn.prepare("insert_network_ip_pool",
+ "INSERT INTO ztc_network_assignment_pool (network_id, ip_range_start, ip_range_end) VALUES ($1, $2, $3)");
+ conn.prepare("delete_network_route", "DELETE FROM ztc_network_route WHERE network_id = $1");
+ conn.prepare("insert_network_route", "INSERT INTO ztc_network_route (network_id, address, bits, via) VALUES ($1, $2, $3, $4)");
+ conn.prepare("delete_network", "UPDATE ztc_network SET deleted = true WHERE id = $1");
+ conn.prepare("delete_member", "UPDATE ztc_member SET hidden = true, deleted = true WHERE id = $1 AND network_id = $2");
+
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);
- delete config;
- 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 identity = (*config)["identity"];
- std::string target = "NULL";
-
- if (!(*config)["remoteTraceTarget"].is_null()) {
- target = (*config)["remoteTraceTarget"];
- }
+ std::string memberId = (*config)["id"];
+ std::string networkId = (*config)["nwid"];
+ std::string target("");
+ if (!(*config)["remoteTraceTarget"].is_null()) {
+ target = (*config)["remoteTraceTarget"];
+ }
+ std::string identity = (*config)["identity"];
+ try {
std::string caps = OSUtils::jsonDump((*config)["capabilities"], -1);
- std::string lastAuthTime = std::to_string((long long)(*config)["lastAuthorizedTime"]);
- std::string lastDeauthTime = std::to_string((long long)(*config)["lastDeauthorizedTime"]);
- std::string rtraceLevel = std::to_string((int)(*config)["remoteTraceLevel"]);
- std::string rev = std::to_string((unsigned long long)(*config)["revision"]);
std::string tags = OSUtils::jsonDump((*config)["tags"], -1);
- std::string vmajor = std::to_string((int)(*config)["vMajor"]);
- std::string vminor = std::to_string((int)(*config)["vMinor"]);
- std::string vrev = std::to_string((int)(*config)["vRev"]);
- std::string vproto = std::to_string((int)(*config)["vProto"]);
- const char *values[19] = {
- memberId.c_str(),
- networkId.c_str(),
- ((*config)["activeBridge"] ? "true" : "false"),
- ((*config)["authorized"] ? "true" : "false"),
- caps.c_str(),
- identity.c_str(),
- lastAuthTime.c_str(),
- lastDeauthTime.c_str(),
- ((*config)["noAutoAssignIps"] ? "true" : "false"),
- rtraceLevel.c_str(),
- (target == "NULL") ? NULL : target.c_str(),
- rev.c_str(),
- tags.c_str(),
- vmajor.c_str(),
- vminor.c_str(),
- vrev.c_str(),
- vproto.c_str()
- };
-
- PGresult *res = PQexecParams(conn,
- "INSERT INTO ztc_member (id, network_id, active_bridge, authorized, capabilities, "
- "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, "
- "TO_TIMESTAMP($7::double precision/1000), TO_TIMESTAMP($8::double precision/1000), "
- "$9, $10, $11, $12, $13, $14, $15, $16, $17) ON CONFLICT (network_id, id) DO UPDATE SET "
- "active_bridge = EXCLUDED.active_bridge, authorized = EXCLUDED.authorized, capabilities = EXCLUDED.capabilities, "
- "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",
- 17,
- NULL,
- values,
- NULL,
- NULL,
- 0);
-
- if (PQresultStatus(res) != PGRES_COMMAND_OK) {
- fprintf(stderr, "ERROR: Error updating member: %s\n", PQresultErrorMessage(res));
- fprintf(stderr, "%s", OSUtils::jsonDump(*config, 2).c_str());
- PQclear(res);
- delete config;
- config = nullptr;
- 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);
- delete config;
- config = nullptr;
- 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"));;
- delete config;
- config = nullptr;
- continue;
- }
- PQclear(res);
+ pqxx::work w(conn);
+ pqxx::result res = w.prepared("insert_member")(memberId)(networkId)
+ ((bool)(*config)["activeBridge"])((bool)(*config)["authorized"])(caps)
+ (identity)((long long)(*config)["lastAuthorizedTime"])((long long)(*config)["lastDeauthorizedTime"])
+ ((bool)(*config)["noAutoAssignIps"])((int)(*config)["remoteTraceLevel"])
+ (target, !target.empty())
+ ((int)(*config)["revision"])(tags)((int)(*config)["vMajor"])((int)(*config)["vMinor"])
+ ((int)(*config)["vRev"])((int)(*config)["vProto"]).exec();
+ w.commit();
+ } catch (std::exception &e) {
+ fprintf(stderr, "Exception upserting member: %s\n", e.what());
+ delete config;
+ config = nullptr;
+ continue;
+ }
+ try {
+ pqxx::work w(conn);
+ pqxx::result res = w.prepared("delete_ip_assignments")(memberId)(networkId).exec();
+
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));
+ pqxx::result res2 = w.prepared("insert_ip_assignments")(memberId)(networkId)(addr).exec();
}
+ w.commit();
+ } catch (std::exception &e) {
+ fprintf(stderr, "Error assigning member IP addresses: %s\n", e.what());
+ delete config;
+ config = nullptr;
+ }
- PQclear(res);
-
- const uint64_t nwidInt = OSUtils::jsonIntHex((*config)["nwid"], 0ULL);
- const uint64_t memberidInt = OSUtils::jsonIntHex((*config)["id"], 0ULL);
- if (nwidInt && memberidInt) {
- nlohmann::json nwOrig;
- nlohmann::json memOrig;
-
- nlohmann::json memNew(*config);
-
- get(nwidInt, nwOrig, memberidInt, memOrig);
-
- _memberChanged(memOrig, memNew, (this->_ready>=2));
- } else {
- fprintf(stderr, "Can't notify of change. Error parsing nwid or memberid: %lu-%lu\n", nwidInt, memberidInt);
- }
+ const uint64_t nwidInt = OSUtils::jsonIntHex((*config)["nwid"], 0ULL);
+ const uint64_t memberidInt = OSUtils::jsonIntHex((*config)["id"], 0ULL);
+ if (nwidInt && memberidInt) {
+ nlohmann::json nwOrig;
+ nlohmann::json memOrig;
- } catch (std::exception &e) {
- fprintf(stderr, "ERROR: Error updating member: %s\n", e.what());
+ nlohmann::json memNew(*config);
+
+ get(nwidInt, nwOrig, memberidInt, memOrig);
+
+ _memberChanged(memOrig, memNew, (this->_ready>=2));
+ } else {
+ fprintf(stderr, "Can't notify of change. Error parsing nwid or memberid: %lu-%lu\n", nwidInt, memberidInt);
}
} else if (objtype == "network") {
+ std::string id = (*config)["id"];
+ std::string controllerId = _myAddressStr.c_str();
+ std::string name = (*config)["name"];
+ std::string rulesSource = (*config)["rulesSource"];
+ std::string caps = OSUtils::jsonDump((*config)["capabilitles"], -1);
+ std::string rtraceLevel = std::to_string((int)(*config)["remoteTraceLevel"]);
+ std::string rules = OSUtils::jsonDump((*config)["rules"], -1);
+ std::string tags = OSUtils::jsonDump((*config)["tags"], -1);
+ std::string v4mode = OSUtils::jsonDump((*config)["v4AssignMode"],-1);
+ std::string v6mode = OSUtils::jsonDump((*config)["v6AssignMode"], -1);
+ std::string target = "";
+ if (!(*config)["remoteTraceTarget"].is_null()) {
+ target = (*config)["remoteTraceTarget"];
+ }
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"];
- std::string caps = OSUtils::jsonDump((*config)["capabilitles"], -1);
- std::string now = std::to_string(OSUtils::now());
- std::string mtu = std::to_string((int)(*config)["mtu"]);
- std::string mcastLimit = std::to_string((int)(*config)["multicastLimit"]);
- std::string rtraceLevel = std::to_string((int)(*config)["remoteTraceLevel"]);
- std::string rules = OSUtils::jsonDump((*config)["rules"], -1);
- std::string tags = OSUtils::jsonDump((*config)["tags"], -1);
- std::string v4mode = OSUtils::jsonDump((*config)["v4AssignMode"],-1);
- std::string v6mode = OSUtils::jsonDump((*config)["v6AssignMode"], -1);
- bool enableBroadcast = (*config)["enableBroadcast"];
- bool isPrivate = (*config)["private"];
-
- const char *values[16] = {
- id.c_str(),
- controllerId.c_str(),
- caps.c_str(),
- enableBroadcast ? "true" : "false",
- now.c_str(),
- mtu.c_str(),
- mcastLimit.c_str(),
- name.c_str(),
- isPrivate ? "true" : "false",
- rtraceLevel.c_str(),
- (remoteTraceTarget == "NULL" ? NULL : remoteTraceTarget.c_str()),
- rules.c_str(),
- rulesSource.c_str(),
- tags.c_str(),
- v4mode.c_str(),
- v6mode.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);
- delete config;
- config = nullptr;
- 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);
- delete config;
- config = nullptr;
- 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"));
- delete config;
- config = nullptr;
- continue;
- }
+ pqxx::work w(conn);
+ pqxx::result res = w.prepared("update_network")(id)(controllerId)(caps)((bool)(*config)["enableBroadcast"])
+ (OSUtils::now())((int)(*config)["mtu"])((int)(*config)["multicastLimit"])(name)((bool)(*config)["private"])
+ ((int)(*config)["remoteTraceLevel"])(target, !target.empty())
+ (rules)(rulesSource)(tags)(v4mode)(v6mode).exec();
+ w.commit();
+ } catch (std::exception &e) {
+ fprintf(stderr, "Error updating network config: %s\n", e.what());
+ delete config;
+ config = nullptr;
+ continue;
+ }
- PQclear(res);
+ try {
+ pqxx::work w(conn);
+ pqxx::result res = w.prepared("delete_network_ip_pool")(id).exec();
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"));
- delete config;
- config = nullptr;
- 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"));
- delete config;
- config = nullptr;
- continue;
+ pqxx::result r2 = w.prepared("insert_nework_ip_pool")(id)(start)(end).exec();
}
+ pqxx::result res2 = w.prepared("delete_network_route")(id).exec();
+
auto routes = (*config)["routes"];
- err = false;
for (auto i = routes.begin(); i != routes.end(); ++i) {
std::string t = (*i)["target"];
std::vector<std::string> target;
@@ -1013,86 +727,42 @@ void PostgreSQL::commitThread()
continue;
}
std::string targetAddr = target[0];
- std::string targetBits = target[1];
- std::string via = "NULL";
+ int targetBits = std::stoi(target[1]);
+ std::string via = "";
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"));
- delete config;
- config = nullptr;
- continue;
+
+ pqxx::result res3 = w.prepared("insert_network_route")(id)(targetAddr)(targetBits)
+ (via, !via.empty()).exec();
}
- res = PQexec(conn, "COMMIT");
- if (PQresultStatus(res) != PGRES_COMMAND_OK) {
- fprintf(stderr, "ERROR: Error committing network update: %s\n", PQresultErrorMessage(res));
- }
- PQclear(res);
+ w.commit();
- const uint64_t nwidInt = OSUtils::jsonIntHex((*config)["nwid"], 0ULL);
- if (nwidInt) {
- nlohmann::json nwOrig;
- nlohmann::json nwNew(*config);
+ } catch (std::exception &e) {
+ fprintf(stderr, "Error updating network IP pool: %s\n", e.what());
+ }
- get(nwidInt, nwOrig);
+ const uint64_t nwidInt = OSUtils::jsonIntHex((*config)["nwid"], 0ULL);
+ if (nwidInt) {
+ nlohmann::json nwOrig;
+ nlohmann::json nwNew(*config);
- _networkChanged(nwOrig, nwNew, true);
- } else {
- fprintf(stderr, "Can't notify network changed: %lu\n", nwidInt);
- }
+ get(nwidInt, nwOrig);
- } catch (std::exception &e) {
- fprintf(stderr, "ERROR: Error updating member: %s\n", e.what());
+ _networkChanged(nwOrig, nwNew, true);
+ } else {
+ fprintf(stderr, "Can't notify network changed: %lu\n", nwidInt);
}
} 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);
+ pqxx::work w(conn);
+ pqxx::result res = w.prepared("delete_network")(networkId).exec();
+ w.commit();
} catch (std::exception &e) {
fprintf(stderr, "ERROR: Error deleting network: %s\n", e.what());
}
@@ -1101,25 +771,9 @@ void PostgreSQL::commitThread()
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);
+ pqxx::work w(conn);
+ pqxx::result res = w.prepared("delete_member")(memberId)(networkId).exec();
+ w.commit();
} catch (std::exception &e) {
fprintf(stderr, "ERROR: Error deleting member: %s\n", e.what());
}
@@ -1132,18 +786,15 @@ void PostgreSQL::commitThread()
delete config;
config = nullptr;
- std::this_thread::sleep_for(std::chrono::milliseconds(10));
+ std::this_thread::sleep_for(std::chrono::milliseconds(1));
}
-
- 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);
+ pqxx::connection conn(_connString);
+ if(!conn.is_open()) {
+ fprintf(stderr, "Connection to database failed: onlineNotificationThread\n");
exit(1);
}
_connected = 1;
@@ -1152,11 +803,6 @@ void PostgreSQL::onlineNotificationThread()
std::unordered_map< std::pair<uint64_t,uint64_t>,int64_t,_PairHasher > lastOnlineCumulative;
while (_run == 1) {
- if (PQstatus(conn) != CONNECTION_OK) {
- fprintf(stderr, "ERROR: Online Notification thread lost connection to Postgres.");
- exit(-1);
- }
-
// map used to send notifications to front end
std::unordered_map<std::string, std::vector<std::string>> updateMap;
@@ -1166,230 +812,144 @@ void PostgreSQL::onlineNotificationThread()
lastOnline.swap(_lastOnline);
}
- PGresult *res = NULL;
- int qCount = 0;
-
- 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) {
- uint64_t nwid_i = i->first.first;
- char nwidTmp[64];
- char memTmp[64];
- char ipTmp[64];
- OSUtils::ztsnprintf(nwidTmp,sizeof(nwidTmp), "%.16llx", nwid_i);
- OSUtils::ztsnprintf(memTmp,sizeof(memTmp), "%.10llx", i->first.second);
-
- auto found = _networks.find(nwid_i);
- if (found == _networks.end()) {
- continue; // skip members trying to join non-existant networks
- }
-
- lastOnlineCumulative[i->first] = i->second.first;
-
-
- std::string networkId(nwidTmp);
- std::string memberId(memTmp);
-
- std::vector<std::string> &members = updateMap[networkId];
- members.push_back(memberId);
-
- int64_t ts = i->second.first;
- std::string ipAddr = i->second.second.toIpString(ipTmp);
- std::string timestamp = std::to_string(ts);
-
- const char *values[4] = {
- networkId.c_str(),
- memberId.c_str(),
- (ipAddr.empty() ? NULL : ipAddr.c_str()),
- timestamp.c_str(),
- };
-
- res = PQexecParams(conn,
- "INSERT INTO ztc_member_status (network_id, member_id, address, last_updated) VALUES ($1, $2, $3, TO_TIMESTAMP($4::double precision/1000)) "
- "ON CONFLICT (network_id, member_id) DO UPDATE SET address = EXCLUDED.address, last_updated = EXCLUDED.last_updated",
- 4, // 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);
- PQclear(PQexec(conn, "ROLLBACK"));
- continue;
- }
-
- 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);
- PQclear(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);
+ try {
+ pqxx::work w(conn);
+ pqxx::pipeline p(w, "Member Update Pipeline");
+ for (auto i=lastOnline.begin(); i != lastOnline.end(); ++i) {
+ uint64_t nwid_i = i->first.first;
+ char nwidTmp[64];
+ char memTmp[64];
+ char ipTmp[64];
+ OSUtils::ztsnprintf(nwidTmp,sizeof(nwidTmp), "%.16llx", nwid_i);
+ OSUtils::ztsnprintf(memTmp,sizeof(memTmp), "%.10llx", i->first.second);
+
+ auto found = _networks.find(nwid_i);
+ if (found == _networks.end()) {
+ continue; // skip members trying to join non-existant networks
}
- PQclear(res);
- qCount = 0;
- }
- }
- res = PQexec(conn, "COMMIT");
- if (PQresultStatus(res) != PGRES_COMMAND_OK) {
- fprintf(stderr, "ERROR: Error on commit (onlineNotificationThread): %s\n", PQresultErrorMessage(res));
- PQclear(res);
- PQclear(PQexec(conn, "ROLLBACK"));
- exit(1);
- }
- PQclear(res);
- const int64_t now = OSUtils::now();
- if ((now - lastUpdatedNetworkStatus) > 10000) {
- lastUpdatedNetworkStatus = now;
+ lastOnlineCumulative[i->first] = i->second.first;
+
- std::vector<std::pair<uint64_t, std::shared_ptr<_Network>>> networks;
- {
- std::lock_guard<std::mutex> l(_networks_l);
- for (auto i = _networks.begin(); i != _networks.end(); ++i) {
- networks.push_back(*i);
- }
- }
+ std::string networkId(nwidTmp);
+ std::string memberId(memTmp);
- int nCount = 0;
+ std::vector<std::string> &members = updateMap[networkId];
+ members.push_back(memberId);
- 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);
+ int64_t ts = i->second.first;
+ std::string ipAddr = i->second.second.toIpString(ipTmp);
+ std::string timestamp = std::to_string(ts);
- std::string networkId(tmp);
+ std::stringstream ss;
+ ss << "INSERT INTO ztc_member_status (network_id, member_id, address, last_updated) VALUES ("
+ << "'" << w.esc(networkId) << "', "
+ << "'" << w.esc(memberId) << "', "
+ << "'" << w.esc(ipAddr) << "', "
+ << timestamp << ") "
+ << "ON CONFLICT (network_id, member_id) DO UPDATE SET address = EXCLUDED.address, last_updated = EXCLUDED.last_updated";
+ p.insert(ss.str());
+ }
+ p.complete();
+ w.commit();
+ } catch (std::exception &e) {
+ fprintf(stderr, "Error updating member status: %s\n", e.what());
+ }
- std::vector<std::string> &_notUsed = updateMap[networkId];
- (void)_notUsed;
+ try {
+ const int64_t now = OSUtils::now();
+ if ((now - lastUpdatedNetworkStatus) > 10000) {
+ pqxx::work w(conn);
+ pqxx::pipeline p(w, "Network Update Pipeline");
+ lastUpdatedNetworkStatus = now;
- uint64_t authMemberCount = 0;
- uint64_t totalMemberCount = 0;
- uint64_t onlineMemberCount = 0;
- uint64_t bridgeCount = 0;
- uint64_t ts = now;
+ std::vector<std::pair<uint64_t, std::shared_ptr<_Network>>> networks;
{
- std::lock_guard<std::mutex> 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<uint64_t,uint64_t>(i->first, m->first));
- if (lo != lastOnlineCumulative.end()) {
- if ((now - lo->second) <= (ZT_NETWORK_AUTOCONF_DELAY * 2)) {
- ++onlineMemberCount;
- } else {
- lastOnlineCumulative.erase(lo);
- }
- }
+ std::lock_guard<std::mutex> l(_networks_l);
+ for (auto i = _networks.begin(); i != _networks.end(); ++i) {
+ networks.push_back(*i);
}
}
- std::string bc = std::to_string(bridgeCount);
- std::string amc = std::to_string(authMemberCount);
- std::string omc = std::to_string(onlineMemberCount);
- std::string tmc = std::to_string(totalMemberCount);
- std::string timestamp = std::to_string(ts);
- const char *values[6] = {
- networkId.c_str(),
- bc.c_str(),
- amc.c_str(),
- omc.c_str(),
- tmc.c_str(),
- timestamp.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, TO_TIMESTAMP($6::double precision/1000)) "
- "ON CONFLICT (network_id) DO UPDATE SET bridge_count = EXCLUDED.bridge_count, "
- "authorized_member_count = EXCLUDED.authorized_member_count, online_member_count = EXCLUDED.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 Status upsert (onlineNotificationThread): %s\n", PQresultErrorMessage(res));
- PQclear(res);
- PQclear(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);
- PQclear(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);
+ for (auto i = networks.begin(); i != networks.end(); ++i) {
+ char tmp[64];
+ Utils::hex(i->first, tmp);
+
+ std::string networkId(tmp);
+
+ std::vector<std::string> &_notUsed = updateMap[networkId];
+ (void)_notUsed;
+
+ uint64_t authMemberCount = 0;
+ uint64_t totalMemberCount = 0;
+ uint64_t onlineMemberCount = 0;
+ uint64_t bridgeCount = 0;
+ uint64_t ts = now;
+ {
+ std::lock_guard<std::mutex> 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<uint64_t,uint64_t>(i->first, m->first));
+ if (lo != lastOnlineCumulative.end()) {
+ if ((now - lo->second) <= (ZT_NETWORK_AUTOCONF_DELAY * 2)) {
+ ++onlineMemberCount;
+ } else {
+ lastOnlineCumulative.erase(lo);
+ }
+ }
+ }
}
- nCount = 0;
+ std::string bc = std::to_string(bridgeCount);
+ std::string amc = std::to_string(authMemberCount);
+ std::string omc = std::to_string(onlineMemberCount);
+ std::string tmc = std::to_string(totalMemberCount);
+ std::string timestamp = std::to_string(ts);
+
+ std::stringstream ss;
+ ss << "INSERT INTO ztc_network_status (network_id, bridge_count, authorized_member_count, "
+ << "online_member_count, total_member_count, last_modified) VALUES ("
+ << "'" << w.esc(networkId) << "', "
+ << bridgeCount << ", "
+ << authMemberCount << ", "
+ << onlineMemberCount << ", "
+ << totalMemberCount << ", "
+ << "TO_TIMESTAMP(" << ts << "::double precision/1000)) "
+ << "ON CONFLICT (network_id) DO UPDATE SET bridge_count = EXCLUDED.bridge_count, "
+ << "authorized_member_count = EXCLUDED.authorized_member_count, online_member_count = EXCLUDED.online_member_count, "
+ << "total_member_count = EXCLUDED.total_member_count, last_modified = EXCLUDED.last_modified";
+ p.insert(ss.str());
}
+ p.complete();
+ w.commit();
}
- res = PQexec(conn, "COMMIT");
- if (PQresultStatus(res) != PGRES_COMMAND_OK) {
- fprintf(stderr, "ERROR: Error on COMMIT (onlineNotificationThread): %s\n", PQresultErrorMessage(res));
- PQclear(res);
- PQclear(PQexec(conn, "ROLLBACK"));
- exit(1);
- }
+ } catch (std::exception &e) {
+ fprintf(stderr, "Error updating network status: %s\n", e.what());
}
- for (auto it = updateMap.begin(); it != updateMap.end(); ++it) {
- std::string networkId = it->first;
- std::vector<std::string> members = it->second;
- std::stringstream queryBuilder;
+ try {
+ pqxx::work w(conn);
+ pqxx::pipeline p(w, "Notification Sender");
+ for (auto it = updateMap.begin(); it != updateMap.end(); ++it) {
+ std::string networkId = it->first;
+ std::vector<std::string> members = it->second;
+ std::stringstream queryBuilder;
- std::string membersStr = ::join(members, ",");
+ std::string membersStr = ::join(members, ",");
- queryBuilder << "NOTIFY controller, '" << networkId << ":" << membersStr << "'";
- std::string query = queryBuilder.str();
+ queryBuilder << "NOTIFY controller, '" << networkId << ":" << membersStr << "'";
+ std::string query = queryBuilder.str();
- PGresult *res = PQexec(conn,query.c_str());
- if (PQresultStatus(res) != PGRES_COMMAND_OK) {
- fprintf(stderr, "ERROR: Error sending NOTIFY: %s\n", PQresultErrorMessage(res));
+ p.insert(query);
}
- PQclear(res);
+ p.complete();
+ w.commit();
+ } catch (std::exception &e) {
+ fprintf(stderr, "Error notifying webapp: %s\n", e.what());
}
-
- std::this_thread::sleep_for(std::chrono::milliseconds(250));
+ std::this_thread::sleep_for(std::chrono::milliseconds(1));
}
- PQfinish(conn);
}
#endif //ZT_CONTROLLER_USE_LIBPQ
diff --git a/controller/PostgreSQL.hpp b/controller/PostgreSQL.hpp
index 36fe8c9f..86aac80d 100644
--- a/controller/PostgreSQL.hpp
+++ b/controller/PostgreSQL.hpp
@@ -23,12 +23,16 @@
#include "DB.hpp"
+#include <pqxx/pqxx>
+
extern "C" {
typedef struct pg_conn PGconn;
}
namespace ZeroTier
{
+class _MemberNotificationReceiver;
+class _NetworkNotificationReceiver;
/**
* A controller database driver that talks to PostgreSQL
@@ -56,8 +60,8 @@ protected:
};
private:
- void initializeNetworks(PGconn *conn);
- void initializeMembers(PGconn *conn);
+ void initializeNetworks(pqxx::connection &conn);
+ void initializeMembers(pqxx::connection &conn);
void heartbeat();
void membersDbWatcher();
void networksDbWatcher();
@@ -81,6 +85,9 @@ private:
mutable std::mutex _readyLock;
std::atomic<int> _ready, _connected, _run;
mutable volatile bool _waitNoticePrinted;
+
+ friend class _MemberNotificationReceiver;
+ friend class _NetworkNotificationReceiver;
};
}
diff --git a/docker/Dockerfile b/docker/Dockerfile
index ec903191..f13a2bc6 100644
--- a/docker/Dockerfile
+++ b/docker/Dockerfile
@@ -4,7 +4,7 @@ MAINTAINER Adam Ierymekno <adam.ierymenko@zerotier.com>, Grant Limberg <grant.li
RUN yum update -y
RUN yum install -y https://download.postgresql.org/pub/repos/yum/10/redhat/rhel-7-x86_64/pgdg-centos10-10-2.noarch.rpm
-RUN yum install -y postgresql10
+RUN yum install -y postgresql10 libpqxx-devel
RUN yum -y install epel-release && yum -y update && yum clean all
RUN yum -y install clang jemalloc jemalloc-devel
generated by cgit v1.2.3 (git 2.39.1) at 2025-04-07 21:59:49 +0000