: DB()

, _myId(myId)

, _myAddress(myId.address())

, _ready(0)

, _connected(1)

, _run(1)

, _waitNoticePrinted(false)

, _listenPort(listenPort)

, _rc(rc)

, _redis(NULL)

, _cluster(NULL)

{

char myAddress[64];

_myAddressStr = myId.address().toString(myAddress);

_connString = std::string(path) + " application_name=controller_" + _myAddressStr;

// Database Schema Version Check

PGconn *conn = getPgConn();

if (PQstatus(conn) != CONNECTION_OK) {

fprintf(stderr, "Bad Database Connection: %s", PQerrorMessage(conn));

exit(1);

}

PGresult *res = PQexec(conn, "SELECT version FROM ztc_database");

if (PQresultStatus(res) != PGRES_TUPLES_OK) {

fprintf(stderr, "Error determining database version");

exit(1);

}

if (PQntuples(res) != 1) {

fprintf(stderr, "Invalid number of db version tuples returned.");

exit(1);

}

int dbVersion = std::stoi(PQgetvalue(res, 0, 0));

if (dbVersion < DB_MINIMUM_VERSION) {

fprintf(stderr, "Central database schema version too low. This controller version requires a minimum schema version of %d. Please upgrade your Central instance", DB_MINIMUM_VERSION);

exit(1);

}

PQclear(res);

res = NULL;

if (_rc != NULL) {

sw::redis::ConnectionOptions opts;

sw::redis::ConnectionPoolOptions poolOpts;

opts.host = _rc->hostname;

opts.port = _rc->port;

opts.password = _rc->password;

opts.db = 0;

poolOpts.size = 10;

if (_rc->clusterMode) {

fprintf(stderr, "Using Redis in Cluster Mode\n");

_cluster = std::make_shared<sw::redis::RedisCluster>(opts, poolOpts);

} else {

fprintf(stderr, "Using Redis in Standalone Mode\n");

_redis = std::make_shared<sw::redis::Redis>(opts, poolOpts);

}

}

_readyLock.lock();

fprintf(stderr, "[%s] NOTICE: %.10llx controller PostgreSQL waiting for initial data download..." ZT_EOL_S, ::_timestr(), (unsigned long long)_myAddress.toInt());

_waitNoticePrinted = true;

initializeNetworks(conn);

initializeMembers(conn);

PQfinish(conn);

conn = NULL;

_heartbeatThread = std::thread(&PostgreSQL::heartbeat, this);

_membersDbWatcher = std::thread(&PostgreSQL::membersDbWatcher, this);

_networksDbWatcher = std::thread(&PostgreSQL::networksDbWatcher, this);

for (int i = 0; i < ZT_CENTRAL_CONTROLLER_COMMIT_THREADS; ++i) {

_commitThread[i] = std::thread(&PostgreSQL::commitThread, this);

}

_onlineNotificationThread = std::thread(&PostgreSQL::onlineNotificationThread, this);

{

while (_ready < 2) {

_readyLock.lock();

_readyLock.unlock();

}

return true;

{

bool modified = false;

try {

if (!record.is_object())

return false;

const std::string objtype = record["objtype"];

if (objtype == "network") {

const uint64_t nwid = jsonIntHex(record["id"],0ULL);

if (nwid) {

nlohmann::json old;

get(nwid,old);

if ((!old.is_object())||(!_compareRecords(old,record))) {

record["revision"] = jsonInt(record["revision"],0ULL) + 1ULL;

_commitQueue.post(std::pair<nlohmann::json,bool>(record,notifyListeners));

modified = true;

}

}

} else if (objtype == "member") {

const uint64_t nwid = jsonIntHex(record["nwid"],0ULL);

const uint64_t id = jsonIntHex(record["id"],0ULL);

if ((id)&&(nwid)) {

nlohmann::json network,old;

get(nwid,network,id,old);

if ((!old.is_object())||(!_compareRecords(old,record))) {

record["revision"] = jsonInt(record["revision"],0ULL) + 1ULL;

_commitQueue.post(std::pair<nlohmann::json,bool>(record,notifyListeners));

modified = true;

}

}

}

} catch (std::exception &e) {

fprintf(stderr, "Error on PostgreSQL::save: %s\n", e.what());

} catch (...) {

fprintf(stderr, "Unknown error on PostgreSQL::save\n");

}

return modified;

{

char tmp2[24];

waitForReady();

Utils::hex(networkId, tmp2);

std::pair<nlohmann::json,bool> tmp;

tmp.first["id"] = tmp2;

tmp.first["objtype"] = "_delete_network";

tmp.second = true;

_commitQueue.post(tmp);

nlohmann::json nullJson;

_networkChanged(tmp.first, nullJson, true);

{

char tmp2[24];

waitForReady();

std::pair<nlohmann::json,bool> tmp, nw;

Utils::hex(networkId, tmp2);

tmp.first["nwid"] = tmp2;

Utils::hex(memberId, tmp2);

tmp.first["id"] = tmp2;

tmp.first["objtype"] = "_delete_member";

tmp.second = true;

_commitQueue.post(tmp);

nlohmann::json nullJson;

_memberChanged(tmp.first, nullJson, true);

{

std::lock_guard<std::mutex> l(_lastOnline_l);

std::pair<int64_t, InetAddress> &i = _lastOnline[std::pair<uint64_t,uint64_t>(networkId, memberId)];

i.first = OSUtils::now();

if (physicalAddress) {

i.second = physicalAddress;

}

{

try {

if (PQstatus(conn) != CONNECTION_OK) {

fprintf(stderr, "Bad Database Connection: %s", PQerrorMessage(conn));

exit(1);

}

std::string setKey = "networks:{" + _myAddressStr + "}";

if (_rc != NULL) {

try {

if (_rc->clusterMode) {

_cluster->del(setKey);

} else {

_redis->del(setKey);

}

} catch (sw::redis::Error &e) {

// del can throw an error if the key doesn't exist

// swallow it and move along

}

}

std::unordered_set<std::string> networkSet;

const char *params[1] = {

_myAddressStr.c_str()

};

fprintf(stderr, "Initializing Networks...\n");

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;

const char *nwidparam[1] = {

PQgetvalue(res, i, 0)

};

std::string nwid = PQgetvalue(res, i, 0);

networkSet.insert(nwid);

config["id"] = nwid;

config["nwid"] = nwid;

try {

config["creationTime"] = std::stoull(PQgetvalue(res, i, 1));

} 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);

try {

config["lastModified"] = std::stoull(PQgetvalue(res, i, 4));

} 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));

} catch (std::exception &e) {

config["mtu"] = 2800;

}

try {

config["multicastLimit"] = std::stoi(PQgetvalue(res, i, 6));

} catch (std::exception &e) {

config["multicastLimit"] = 64;

}

config["name"] = PQgetvalue(res, i, 7);

config["private"] = (strcmp(PQgetvalue(res, i, 8),"t")==0);

try {

config["remoteTraceLevel"] = std::stoi(PQgetvalue(res, i, 9));

} catch (std::exception &e) {

config["remoteTraceLevel"] = 0;

}

config["remoteTraceTarget"] = PQgetvalue(res, i, 10);

try {

config["revision"] = std::stoull(PQgetvalue(res, i, 11));

} 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["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) {

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,

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);

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(!networkSet.empty()) {

if (_rc && _rc->clusterMode) {

auto tx = _cluster->transaction(_myAddressStr, true);

tx.sadd(setKey, networkSet.begin(), networkSet.end());

tx.exec();

} else if (_rc && !_rc->clusterMode) {

auto tx = _redis->transaction(true);

tx.sadd(setKey, networkSet.begin(), networkSet.end());

tx.exec();

}

}

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 (sw::redis::Error &e) {

fprintf(stderr, "ERROR: Error initializing networks in Redis: %s\n", e.what());

exit(-1);

} catch (std::exception &e) {

fprintf(stderr, "ERROR: Error initializing networks: %s\n", e.what());

exit(-1);

}

{

try {

if (PQstatus(conn) != CONNECTION_OK) {

fprintf(stderr, "Bad Database Connection: %s", PQerrorMessage(conn));

exit(1);

}

std::string setKeyBase = "network-nodes-all:{" + _myAddressStr + "}:";

if (_rc != NULL) {

std::lock_guard<std::mutex> l(_networks_l);

std::unordered_set<std::string> deletes;

for ( auto it : _networks) {

uint64_t nwid_i = it.first;

char nwidTmp[64] = {0};

OSUtils::ztsnprintf(nwidTmp, sizeof(nwidTmp), "%.16llx", nwid_i);

std::string nwid(nwidTmp);

std::string key = setKeyBase + nwid;

deletes.insert(key);

}

if (!deletes.empty()) {

if (_rc->clusterMode) {

auto tx = _cluster->transaction(_myAddressStr, true);

for (std::string k : deletes) {

tx.del(k);

}

tx.exec();

} else {

auto tx = _redis->transaction(true);

for (std::string k : deletes) {

tx.del(k);

}

tx.exec();

}

}

}

const char *params[1] = {

_myAddressStr.c_str()

};

std::unordered_map<std::string, std::string> networkMembers;

fprintf(stderr, "Initializing Members...\n");

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));

networkMembers.insert(std::pair<std::string, std::string>(setKeyBase+networkId, memberId));

std::string ctime = PQgetvalue(res, i, 5);

config["id"] = memberId;

config["nwid"] = networkId;

config["activeBridge"] = (strcmp(PQgetvalue(res, i, 2), "t") == 0);

config["authorized"] = (strcmp(PQgetvalue(res, i, 3), "t") == 0);

try {

config["capabilities"] = json::parse(PQgetvalue(res, i, 4));

} catch (std::exception &e) {

config["capabilities"] = json::array();

}

try {

config["creationTime"] = std::stoull(PQgetvalue(res, i, 5));

} 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);

try {

config["lastAuthorizedTime"] = std::stoull(PQgetvalue(res, i, 7));

} 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"] = 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"] = 0;

}

config["remoteTraceTarget"] = PQgetvalue(res, i, 10);

try {

config["tags"] = json::parse(PQgetvalue(res, i, 11));

} catch (std::exception &e) {

config["tags"] = json::array();

}

try {

config["vMajor"] = std::stoi(PQgetvalue(res, i, 12));

} catch(std::exception &e) {

config["vMajor"] = -1;

}

try {

config["vMinor"] = std::stoi(PQgetvalue(res, i, 13));

} catch (std::exception &e) {

config["vMinor"] = -1;

}

try {

config["vRev"] = std::stoi(PQgetvalue(res, i, 14));

} catch (std::exception &e) {

config["vRev"] = -1;

}

try {

config["vProto"] = std::stoi(PQgetvalue(res, i, 15));

} catch (std::exception &e) {

config["vProto"] = -1;

}

config["noAutoAssignIps"] = (strcmp(PQgetvalue(res, i, 16), "t") == 0);

try {

config["revision"] = std::stoull(PQgetvalue(res, i, 17));

} 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 DISTINCT 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) {

std::string ipaddr = PQgetvalue(r2, j, 0);

std::size_t pos = ipaddr.find('/');

if (pos != std::string::npos) {

ipaddr = ipaddr.substr(0, pos);

}

config["ipAssignments"].push_back(ipaddr);

}

_memberChanged(empty, config, false);

}

PQclear(res);

if (!networkMembers.empty()) {

if (_rc != NULL) {

if (_rc->clusterMode) {

auto tx = _cluster->transaction(_myAddressStr, true);

for (auto it : networkMembers) {

tx.sadd(it.first, it.second);

}

tx.exec();

} else {

auto tx = _redis->transaction(true);

for (auto it : networkMembers) {

tx.sadd(it.first, it.second);

}

tx.exec();

}

}

}

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 (sw::redis::Error &e) {

fprintf(stderr, "ERROR: Error initializing members (redis): %s\n", e.what());

} catch (std::exception &e) {

fprintf(stderr, "ERROR: Error initializing members: %s\n", e.what());

exit(-1);

}

{

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 < (int)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 = getPgConn();

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) {

fprintf(stderr, "%s heartbeat thread lost connection to Database\n", _myAddressStr.c_str());

PQfinish(conn);

exit(6);

}

int64_t ts = OSUtils::now();

if (conn) {

std::string major = std::to_string(ZEROTIER_VERSION_MAJOR);

std::string minor = std::to_string(ZEROTIER_VERSION_MINOR);

std::string rev = std::to_string(ZEROTIER_VERSION_REVISION);

std::string build = std::to_string(ZEROTIER_VERSION_BUILD);

std::string now = std::to_string(ts);

std::string host_port = std::to_string(_listenPort);

std::string use_redis = (_rc != NULL) ? "true" : "false";

const char *values[10] = {

controllerId,

hostname,

now.c_str(),

publicIdentity,

major.c_str(),

minor.c_str(),

rev.c_str(),

build.c_str(),

host_port.c_str(),

use_redis.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, host_port, use_redis) "

"VALUES ($1, $2, TO_TIMESTAMP($3::double precision/1000), $4, $5, $6, $7, $8, $9, $10) "

"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, host_port = EXCLUDED.host_port, "

"use_redis = EXCLUDED.use_redis",

10, // 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);

}

if (_rc != NULL) {

if (_rc->clusterMode) {

_cluster->zadd("controllers", controllerId, ts);

} else {

_redis->zadd("controllers", controllerId, ts);

}

}

std::this_thread::sleep_for(std::chrono::milliseconds(1000));

}

PQfinish(conn);

conn = NULL;

fprintf(stderr, "Exited heartbeat thread\n");

{

PGconn *conn = getPgConn(NO_OVERRIDE);

if (PQstatus(conn) == CONNECTION_BAD) {

fprintf(stderr, "Connection to database failed: %s\n", PQerrorMessage(conn));

PQfinish(conn);

exit(1);

}

if (_rc) {

PQfinish(conn);

conn = NULL;

_membersWatcher_Redis();

} else {

_membersWatcher_Postgres(conn);

PQfinish(conn);

conn = NULL;

}

if (_run == 1) {

fprintf(stderr, "ERROR: %s membersDbWatcher should still be running! Exiting Controller.\n", _myAddressStr.c_str());

exit(9);

}

fprintf(stderr, "Exited membersDbWatcher\n");

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));

}

char buf[11] = {0};

std::string key = "member-stream:{" + std::string(_myAddress.toString(buf)) + "}";

while (_run == 1) {

try {

json tmp;

std::unordered_map<std::string, ItemStream> result;

if (_rc->clusterMode) {

_cluster->xread(key, "$", std::chrono::seconds(1), 0, std::inserter(result, result.end()));

} else {

_redis->xread(key, "$", std::chrono::seconds(1), 0, std::inserter(result, result.end()));

}

if (!result.empty()) {

for (auto element : result) {

#ifdef ZT_TRACE

fprintf(stdout, "Received notification from: %s\n", element.first.c_str());

#endif

for (auto rec : element.second) {

std::string id = rec.first;

auto attrs = rec.second;

#ifdef ZT_TRACE

fprintf(stdout, "Record ID: %s\n", id.c_str());

fprintf(stdout, "attrs len: %lu\n", attrs.size());

#endif

for (auto a : attrs) {

#ifdef ZT_TRACE

fprintf(stdout, "key: %s\nvalue: %s\n", a.first.c_str(), a.second.c_str());

#endif

try {

tmp = json::parse(a.second);

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 (...) {

fprintf(stderr, "json parse error in networkWatcher_Redis\n");

}

}

if (_rc->clusterMode) {

_cluster->xdel(key, id);

} else {

_redis->xdel(key, id);

}

}

}

}

} catch (sw::redis::Error &e) {

fprintf(stderr, "Error in Redis members watcher: %s\n", e.what());

}

}

fprintf(stderr, "membersWatcher ended\n");

{

PGconn *conn = getPgConn(NO_OVERRIDE);

if (PQstatus(conn) == CONNECTION_BAD) {

fprintf(stderr, "Connection to database failed: %s\n", PQerrorMessage(conn));

PQfinish(conn);

exit(1);

}

if (_rc) {

PQfinish(conn);

conn = NULL;

_networksWatcher_Redis();

} else {

_networksWatcher_Postgres(conn);

PQfinish(conn);

conn = NULL;

}

if (_run == 1) {

fprintf(stderr, "ERROR: %s networksDbWatcher should still be running! Exiting Controller.\n", _myAddressStr.c_str());

exit(8);

}

fprintf(stderr, "Exited networksDbWatcher\n");

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));

}

char buf[11] = {0};

std::string key = "network-stream:{" + std::string(_myAddress.toString(buf)) + "}";

while (_run == 1) {

try {

json tmp;

std::unordered_map<std::string, ItemStream> result;

if (_rc->clusterMode) {

_cluster->xread(key, "$", std::chrono::seconds(1), 0, std::inserter(result, result.end()));

} else {

_redis->xread(key, "$", std::chrono::seconds(1), 0, std::inserter(result, result.end()));

}

if (!result.empty()) {

for (auto element : result) {

#ifdef ZT_TRACE

fprintf(stdout, "Received notification from: %s\n", element.first.c_str());

#endif

for (auto rec : element.second) {

std::string id = rec.first;

auto attrs = rec.second;

#ifdef ZT_TRACE

fprintf(stdout, "Record ID: %s\n", id.c_str());

fprintf(stdout, "attrs len: %lu\n", attrs.size());

#endif

for (auto a : attrs) {

#ifdef ZT_TRACE

fprintf(stdout, "key: %s\nvalue: %s\n", a.first.c_str(), a.second.c_str());

#endif

try {

tmp = json::parse(a.second);

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 (...) {

fprintf(stderr, "json parse error in networkWatcher_Redis\n");