central controller metrics & request path updates (#2012)

* internal db metrics

* use shared mutexes for read/write locks

* remove this lock. only used for a metric

* more metrics

* remove exploratory metrics

place controller request benchmarks behind ifdef
This commit is contained in:
Grant Limberg 2023-05-23 12:11:26 -07:00 committed by GitHub
parent da71e2524c
commit 17f6b3a10b
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
10 changed files with 307 additions and 80 deletions

View file

@ -460,17 +460,41 @@ static bool _parseRule(json &r,ZT_VirtualNetworkRule &rule)
} // anonymous namespace
EmbeddedNetworkController::EmbeddedNetworkController(Node *node,const char *ztPath,const char *dbPath, int listenPort, RedisConfig *rc) :
_startTime(OSUtils::now()),
_listenPort(listenPort),
_node(node),
_ztPath(ztPath),
_path(dbPath),
_sender((NetworkController::Sender *)0),
_db(this),
_ssoExpiryRunning(true),
_ssoExpiry(std::thread(&EmbeddedNetworkController::_ssoExpiryThread, this)),
_rc(rc)
EmbeddedNetworkController::EmbeddedNetworkController(Node *node,const char *ztPath,const char *dbPath, int listenPort, RedisConfig *rc)
: _startTime(OSUtils::now())
, _listenPort(listenPort)
, _node(node)
, _ztPath(ztPath)
, _path(dbPath)
, _sender((NetworkController::Sender *)0)
, _db(this)
, _ssoExpiryRunning(true)
, _ssoExpiry(std::thread(&EmbeddedNetworkController::_ssoExpiryThread, this))
, _rc(rc)
#ifdef CENTRAL_CONTROLLER_REQUEST_BENCHMARK
, _member_status_lookup{"nc_member_status_lookup",""}
, _member_status_lookup_count{"nc_member_status_lookup_count",""}
, _node_is_online{"nc_node_is_online",""}
, _node_is_online_count{"nc_node_is_online_count",""}
, _get_and_init_member{"nc_get_and_init_member",""}
, _get_and_init_member_count{"nc_get_and_init_member_count",""}
, _have_identity{"nc_have_identity",""}
, _have_identity_count{"nc_have_identity_count",""}
, _determine_auth{"nc_determine_auth",""}
, _determine_auth_count{"nc_determine_auth_count",""}
, _sso_check{"nc_sso_check",""}
, _sso_check_count{"nc_sso_check_count",""}
, _auth_check{"nc_auth_check",""}
, _auth_check_count{"nc_auth_check_count",""}
, _json_schlep{"nc_json_schlep",""}
, _json_schlep_count{"nc_json_schlep_count",""}
, _issue_certificate{"nc_issue_certificate", ""}
, _issue_certificate_count{"nc_issue_certificate_count",""}
, _save_member{"nc_save_member",""}
, _save_member_count{"nc_save_member_count",""}
, _send_netconf{"nc_send_netconf2",""}
, _send_netconf_count{"nc_send_netconf2_count",""}
#endif
{
}
@ -549,6 +573,18 @@ void EmbeddedNetworkController::request(
if (((!_signingId)||(!_signingId.hasPrivate()))||(_signingId.address().toInt() != (nwid >> 24))||(!_sender))
return;
_startThreads();
const int64_t now = OSUtils::now();
if (requestPacketId) {
std::lock_guard<std::mutex> l(_memberStatus_l);
_MemberStatus &ms = _memberStatus[_MemberStatusKey(nwid,identity.address().toInt())];
if ((now - ms.lastRequestTime) <= ZT_NETCONF_MIN_REQUEST_PERIOD) {
return;
}
ms.lastRequestTime = now;
}
_RQEntry *qe = new _RQEntry;
qe->nwid = nwid;
qe->requestPacketId = requestPacketId;
@ -1176,36 +1212,66 @@ void EmbeddedNetworkController::_request(
const Identity &identity,
const Dictionary<ZT_NETWORKCONFIG_METADATA_DICT_CAPACITY> &metaData)
{
Metrics::network_config_request++;
auto tid = std::this_thread::get_id();
std::stringstream ss; ss << tid;
std::string threadID = ss.str();
#ifdef CENTRAL_CONTROLLER_REQUEST_BENCHMARK
auto b1 = _member_status_lookup.Add({{"thread", threadID}});
auto c1 = _member_status_lookup_count.Add({{"thread", threadID}});
c1++;
b1.start();
#endif
char nwids[24];
DB::NetworkSummaryInfo ns;
json network,member;
if (((!_signingId)||(!_signingId.hasPrivate()))||(_signingId.address().toInt() != (nwid >> 24))||(!_sender))
if (((!_signingId)||(!_signingId.hasPrivate()))||(_signingId.address().toInt() != (nwid >> 24))||(!_sender)) {
return;
}
const int64_t now = OSUtils::now();
if (requestPacketId) {
std::lock_guard<std::mutex> l(_memberStatus_l);
_MemberStatus &ms = _memberStatus[_MemberStatusKey(nwid,identity.address().toInt())];
if ((now - ms.lastRequestTime) <= ZT_NETCONF_MIN_REQUEST_PERIOD)
return;
ms.lastRequestTime = now;
}
#ifdef CENTRAL_CONTROLLER_REQUEST_BENCHMARK
b1.stop();
auto b2 = _node_is_online.Add({{"thread",threadID}});
auto c2 = _node_is_online_count.Add({{"thread",threadID}});
c2++;
b2.start();
#endif
_db.nodeIsOnline(nwid,identity.address().toInt(),fromAddr);
#ifdef CENTRAL_CONTROLLER_REQUEST_BENCHMARK
b2.stop();
auto b3 = _get_and_init_member.Add({{"thread", threadID}});
auto c3 = _get_and_init_member_count.Add({{"thread",threadID}});
c3++;
b3.start();
#endif
Utils::hex(nwid,nwids);
_db.get(nwid,network,identity.address().toInt(),member,ns);
if ((!network.is_object())||(network.empty())) {
_sender->ncSendError(nwid,requestPacketId,identity.address(),NetworkController::NC_ERROR_OBJECT_NOT_FOUND, nullptr, 0);
#ifdef CENTRAL_CONTROLLER_REQUEST_BENCHMARK
b3.stop();
#endif
return;
}
const bool newMember = ((!member.is_object())||(member.empty()));
DB::initMember(member);
_MemberStatusKey msk(nwid,identity.address().toInt());
#ifdef CENTRAL_CONTROLLER_REQUEST_BENCHMARK
b3.stop();
#endif
{
#ifdef CENTRAL_CONTROLLER_REQUEST_BENCHMARK
auto b4 = _have_identity.Add({{"thread",threadID}});
auto c4 = _have_identity_count.Add({{"thread",threadID}});
c4++;
b4.start();
#endif
const std::string haveIdStr(OSUtils::jsonString(member["identity"],""));
if (haveIdStr.length() > 0) {
// If we already know this member's identity perform a full compare. This prevents
@ -1214,10 +1280,16 @@ void EmbeddedNetworkController::_request(
try {
if (Identity(haveIdStr.c_str()) != identity) {
_sender->ncSendError(nwid,requestPacketId,identity.address(),NetworkController::NC_ERROR_ACCESS_DENIED, nullptr, 0);
#ifdef CENTRAL_CONTROLLER_REQUEST_BENCHMARK
b4.stop();
#endif
return;
}
} catch ( ... ) {
_sender->ncSendError(nwid,requestPacketId,identity.address(),NetworkController::NC_ERROR_ACCESS_DENIED, nullptr, 0);
#ifdef CENTRAL_CONTROLLER_REQUEST_BENCHMARK
b4.stop();
#endif
return;
}
} else {
@ -1225,6 +1297,9 @@ void EmbeddedNetworkController::_request(
char idtmp[1024];
member["identity"] = identity.toString(false,idtmp);
}
#ifdef CENTRAL_CONTROLLER_REQUEST_BENCHMARK
b4.stop();
#endif
}
// These are always the same, but make sure they are set
@ -1237,6 +1312,12 @@ void EmbeddedNetworkController::_request(
}
// Determine whether and how member is authorized
#ifdef CENTRAL_CONTROLLER_REQUEST_BENCHMARK
auto b5 = _determine_auth.Add({{"thread",threadID}});
auto c5 = _determine_auth_count.Add({{"thread",threadID}});
c5++;
b5.start();
#endif
bool authorized = false;
bool autoAuthorized = false;
json autoAuthCredentialType,autoAuthCredential;
@ -1273,10 +1354,19 @@ void EmbeddedNetworkController::_request(
member["lastAuthorizedCredentialType"] = autoAuthCredentialType;
member["lastAuthorizedCredential"] = autoAuthCredential;
}
#ifdef CENTRAL_CONTROLLER_REQUEST_BENCHMARK
b5.stop();
#endif
// Should we check SSO Stuff?
// If network is configured with SSO, and the member is not marked exempt: yes
// Otherwise no, we use standard auth logic.
#ifdef CENTRAL_CONTROLLER_REQUEST_BENCHMARK
auto b6 = _sso_check.Add({{"thread",threadID}});
auto c6 = _sso_check_count.Add({{"thread",threadID}});
c6++;
b6.start();
#endif
AuthInfo info;
int64_t authenticationExpiryTime = -1;
bool networkSSOEnabled = OSUtils::jsonBool(network["ssoEnabled"], false);
@ -1303,10 +1393,20 @@ void EmbeddedNetworkController::_request(
}
DB::cleanMember(member);
_db.save(member,true);
#ifdef CENTRAL_CONTROLLER_REQUEST_BENCHMARK
b6.stop();
#endif
return;
}
}
#ifdef CENTRAL_CONTROLLER_REQUEST_BENCHMARK
b6.stop();
auto b7 = _auth_check.Add({{"thread",threadID}});
auto c7 = _auth_check_count.Add({{"thread",threadID}});
c7++;
b7.start();
#endif
if (authorized) {
// Update version info and meta-data if authorized and if this is a genuine request
if (requestPacketId) {
@ -1342,8 +1442,14 @@ void EmbeddedNetworkController::_request(
DB::cleanMember(member);
_db.save(member,true);
_sender->ncSendError(nwid,requestPacketId,identity.address(),NetworkController::NC_ERROR_ACCESS_DENIED, nullptr, 0);
#ifdef CENTRAL_CONTROLLER_REQUEST_BENCHMARK
b7.stop();
#endif
return;
}
#ifdef CENTRAL_CONTROLLER_REQUEST_BENCHMARK
b7.stop();
#endif
// -------------------------------------------------------------------------
// If we made it this far, they are authorized (and authenticated).
@ -1351,6 +1457,12 @@ void EmbeddedNetworkController::_request(
// Default timeout: 15 minutes. Maximum: two hours. Can be specified by an optional field in the network config
// if something longer than 15 minutes is desired. Minimum is 5 minutes since shorter than that would be flaky.
#ifdef CENTRAL_CONTROLLER_REQUEST_BENCHMARK
auto b8 = _json_schlep.Add({{"thread",threadID}});
auto c8 = _json_schlep_count.Add({{"thread", threadID}});
c8++;
b8.start();
#endif
int64_t credentialtmd = ZT_NETWORKCONFIG_DEFAULT_CREDENTIAL_TIME_DFL_MAX_DELTA;
if (network.contains("certificateTimeoutWindowSize")) {
credentialtmd = (int64_t)network["certificateTimeoutWindowSize"];
@ -1418,8 +1530,9 @@ void EmbeddedNetworkController::_request(
nc->remoteTraceLevel = (Trace::Level)OSUtils::jsonInt(network["remoteTraceLevel"],0ULL);
}
for(std::vector<Address>::const_iterator ab(ns.activeBridges.begin());ab!=ns.activeBridges.end();++ab)
for(std::vector<Address>::const_iterator ab(ns.activeBridges.begin());ab!=ns.activeBridges.end();++ab) {
nc->addSpecialist(*ab,ZT_NETWORKCONFIG_SPECIALIST_TYPE_ACTIVE_BRIDGE);
}
json &v4AssignMode = network["v4AssignMode"];
json &v6AssignMode = network["v6AssignMode"];
@ -1739,12 +1852,22 @@ void EmbeddedNetworkController::_request(
} else {
dns = json::object();
}
#ifdef CENTRAL_CONTROLLER_REQUEST_BENCHMARK
b8.stop();
#endif
// Issue a certificate of ownership for all static IPs
#ifdef CENTRAL_CONTROLLER_REQUEST_BENCHMARK
auto b9 = _issue_certificate.Add({{"thread",threadID}});
auto c9 = _issue_certificate_count.Add({{"thread",threadID}});
c9++;
b9.start();
#endif
if (nc->staticIpCount) {
nc->certificatesOfOwnership[0] = CertificateOfOwnership(nwid,now,identity.address(),1);
for(unsigned int i=0;i<nc->staticIpCount;++i)
for(unsigned int i=0;i<nc->staticIpCount;++i) {
nc->certificatesOfOwnership[0].addThing(nc->staticIps[i]);
}
nc->certificatesOfOwnership[0].sign(_signingId);
nc->certificateOfOwnershipCount = 1;
}
@ -1754,22 +1877,45 @@ void EmbeddedNetworkController::_request(
nc->com = com;
} else {
_sender->ncSendError(nwid,requestPacketId,identity.address(),NetworkController::NC_ERROR_INTERNAL_SERVER_ERROR, nullptr, 0);
#ifdef CENTRAL_CONTROLLER_REQUEST_BENCHMARK
b9.stop();
#endif
return;
}
#ifdef CENTRAL_CONTROLLER_REQUEST_BENCHMARK
b9.stop();
auto b10 = _save_member.Add({{"thread",threadID}});
auto c10 = _save_member_count.Add({{"thread",threadID}});
c10++;
b10.start();
#endif
DB::cleanMember(member);
_db.save(member,true);
#ifdef CENTRAL_CONTROLLER_REQUEST_BENCHMARK
b10.stop();
auto b11 = _send_netconf.Add({{"thread",threadID}});
auto c11 = _send_netconf_count.Add({{"thread",threadID}});
c11++;
b11.start();
#endif
_sender->ncSendConfig(nwid,requestPacketId,identity.address(),*(nc.get()),metaData.getUI(ZT_NETWORKCONFIG_REQUEST_METADATA_KEY_VERSION,0) < 6);
#ifdef CENTRAL_CONTROLLER_REQUEST_BENCHMARK
b11.stop();
#endif
}
void EmbeddedNetworkController::_startThreads()
{
std::lock_guard<std::mutex> l(_threads_l);
if (!_threads.empty())
if (!_threads.empty()) {
return;
}
const long hwc = std::max((long)std::thread::hardware_concurrency(),(long)1);
for(long t=0;t<hwc;++t) {
_threads.emplace_back([this]() {
Metrics::network_config_request_threads++;
for(;;) {
_RQEntry *qe = (_RQEntry *)0;
Metrics::network_config_request_queue_size = _queue.size();
@ -1790,6 +1936,7 @@ void EmbeddedNetworkController::_startThreads()
}
}
}
Metrics::network_config_request_threads--;
});
}
}