void/ZeroTierOne
2
0
Fork 0
mirror of https://github.com/zerotier/ZeroTierOne.git synced 2025-06-06 20:43:44 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

Put upserts from controller into single transaction

Browse source
This commit is contained in:
Grant Limberg 2020-08-26 14:52:23 -07:00
parent 7f99c4a779
commit 45b6d11126
No known key found for this signature in database
GPG key ID: 2BA62CCABBB4095A
1 changed files with 46 additions and 29 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

75
controller/PostgreSQL.cpp
View file

@ -1079,7 +1079,17 @@ void PostgreSQL::commitThread()
vproto.c_str() vproto.c_str()
}; };
PGresult *res = PQexecParams(conn, PGresult *res = PQexec(conn, "BEGIN");
if (PQresultStatus(res) != PGRES_COMMAND_OK) {
fprintf(stderr, "ERROR: Error beginning update transaction: %s\n", PQresultErrorMessage(res));
PQclear(res);
delete config;
config = nullptr;
continue;
}
res = PQexecParams(conn,
"INSERT INTO ztc_member (id, network_id, active_bridge, authorized, capabilities, " "INSERT INTO ztc_member (id, network_id, active_bridge, authorized, capabilities, "
"identity, last_authorized_time, last_deauthorized_time, no_auto_assign_ips, " "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) " "remote_trace_level, remote_trace_target, revision, tags, v_major, v_minor, v_rev, v_proto) "
@ -1103,17 +1113,7 @@ void PostgreSQL::commitThread()
fprintf(stderr, "ERROR: Error updating member: %s\n", PQresultErrorMessage(res)); fprintf(stderr, "ERROR: Error updating member: %s\n", PQresultErrorMessage(res));
fprintf(stderr, "%s", OSUtils::jsonDump(*config, 2).c_str()); fprintf(stderr, "%s", OSUtils::jsonDump(*config, 2).c_str());
PQclear(res); PQclear(res);
delete config; PQclear(PQexec(conn, "ROLLBACK"));
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; delete config;
config = nullptr; config = nullptr;
continue; continue;
@ -1147,6 +1147,7 @@ void PostgreSQL::commitThread()
PQclear(res); PQclear(res);
std::vector<std::string> assignments; std::vector<std::string> assignments;
bool ipAssignError = false;
for (auto i = (*config)["ipAssignments"].begin(); i != (*config)["ipAssignments"].end(); ++i) { for (auto i = (*config)["ipAssignments"].begin(); i != (*config)["ipAssignments"].end(); ++i) {
std::string addr = *i; std::string addr = *i;
@ -1173,18 +1174,28 @@ void PostgreSQL::commitThread()
fprintf(stderr, "ERROR: Error setting IP addresses for member: %s\n", PQresultErrorMessage(res)); fprintf(stderr, "ERROR: Error setting IP addresses for member: %s\n", PQresultErrorMessage(res));
PQclear(res); PQclear(res);
PQclear(PQexec(conn, "ROLLBACK")); PQclear(PQexec(conn, "ROLLBACK"));
break;; ipAssignError = true;
break;
} }
PQclear(res);
assignments.push_back(addr); assignments.push_back(addr);
} }
if (ipAssignError) {
delete config;
config = nullptr;
continue;
}
res = PQexec(conn, "COMMIT"); res = PQexec(conn, "COMMIT");
if (PQresultStatus(res) != PGRES_COMMAND_OK) { if (PQresultStatus(res) != PGRES_COMMAND_OK) {
fprintf(stderr, "ERROR: Error committing ip address data: %s\n", PQresultErrorMessage(res)); fprintf(stderr, "ERROR: Error committing member transaction: %s\n", PQresultErrorMessage(res));
PQclear(res);
PQclear(PQexec(conn, "ROLLBACK"));
delete config;
config = nullptr;
continue;
} }
PQclear(res);
const uint64_t nwidInt = OSUtils::jsonIntHex((*config)["nwid"], 0ULL); const uint64_t nwidInt = OSUtils::jsonIntHex((*config)["nwid"], 0ULL);
const uint64_t memberidInt = OSUtils::jsonIntHex((*config)["id"], 0ULL); const uint64_t memberidInt = OSUtils::jsonIntHex((*config)["id"], 0ULL);
if (nwidInt && memberidInt) { if (nwidInt && memberidInt) {
@ -1247,13 +1258,24 @@ void PostgreSQL::commitThread()
v6mode.c_str(), v6mode.c_str(),
}; };
PGresult *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);
// This ugly query exists because when we want to mirror networks to/from // This ugly query exists because when we want to mirror networks to/from
// another data store (e.g. FileDB or LFDB) it is possible to get a network // another data store (e.g. FileDB or LFDB) it is possible to get a network
// that doesn't exist in Central's database. This does an upsert and sets // that doesn't exist in Central's database. This does an upsert and sets
// the owner_id to the "first" global admin in the user DB if the record // the owner_id to the "first" global admin in the user DB if the record
// did not previously exist. If the record already exists owner_id is left // did not previously exist. If the record already exists owner_id is left
// unchanged, so owner_id should be left out of the update clause. // unchanged, so owner_id should be left out of the update clause.
PGresult *res = PQexecParams(conn, res = PQexecParams(conn,
"INSERT INTO ztc_network (id, creation_time, owner_id, controller_id, capabilities, enable_broadcast, " "INSERT INTO ztc_network (id, creation_time, owner_id, controller_id, capabilities, enable_broadcast, "
"last_modified, mtu, multicast_limit, name, private, " "last_modified, mtu, multicast_limit, name, private, "
"remote_trace_level, remote_trace_target, rules, rules_source, " "remote_trace_level, remote_trace_target, rules, rules_source, "
@ -1280,24 +1302,14 @@ void PostgreSQL::commitThread()
if (PQresultStatus(res) != PGRES_COMMAND_OK) { if (PQresultStatus(res) != PGRES_COMMAND_OK) {
fprintf(stderr, "ERROR: Error updating network record: %s\n", PQresultErrorMessage(res)); fprintf(stderr, "ERROR: Error updating network record: %s\n", PQresultErrorMessage(res));
PQclear(res); PQclear(res);
PQclear(PQexec(conn, "ROLLBACK"));
delete config; delete config;
config = nullptr; config = nullptr;
continue; continue;
} }
PQclear(res); 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] = { const char *params[1] = {
id.c_str() id.c_str()
}; };
@ -1428,6 +1440,11 @@ void PostgreSQL::commitThread()
res = PQexec(conn, "COMMIT"); res = PQexec(conn, "COMMIT");
if (PQresultStatus(res) != PGRES_COMMAND_OK) { if (PQresultStatus(res) != PGRES_COMMAND_OK) {
fprintf(stderr, "ERROR: Error committing network update: %s\n", PQresultErrorMessage(res)); fprintf(stderr, "ERROR: Error committing network update: %s\n", PQresultErrorMessage(res));
PQclear(res);
PQclear(PQexec(conn, "ROLLBACK"));
delete config;
config = nullptr;
continue;
} }
PQclear(res); PQclear(res);